chronaxis.execution.csv_observation
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在新CSV的同一次真实扫描中保存首末原时间行事实，独立管理读取寿命。23BudgetedSignalCSVSourceAdapter的可选入口返回CSVObservationRead。它转交原4SignalBatch，原解析器只在最终协议和装箱成功后提交一次首末时间候选；最后5EOF批交付还不是完成。必须原生成器真正StopIteration、文件及本请求预算资源6全部成功关闭后，observation才给出冻结CSVObservation。原请求按对象保留，7每个read有新的独立identity，同一request可同时创建多个read，不用适配器全局8last或按request值缓存。时间为转换及offset后的整数纳秒；空投影、缺失值行9仍计原时间行，空物理行不计，不从样本、请求domain或watermark推测首端。1011read先预约请求、观测模型及有限资源责任容器的固定费用，完成事实保留到read12显式close，因此正常完成后借用adapter关闭不影响事实。提前close、失败、取消13或任何清理失败均不开放完成事实；close只释放本read，不关闭借用adapter和14其他请求。未返回构造及运行错误保留最高csv_observation_read_cleanup_owner。1516可选资源检查器继承_CSVResources的样本和装箱估算，通过事务式金额槽交接17物理行费用，并支持重试close。CSV解析、文件顺序、批次和数值仍由原读取器18负责；预算估算不代表整个进程RSS。19"""2021# 冻结事实与每次读取独立身份只依赖标准库，不保存全文件或全批次列表。22from dataclasses import dataclass23from uuid import uuid424from ..core.validation import require_identifier, require_text25from ..timebase.conversion import require_int64_ns26from ..sources.models import ReadRequest27from ..sources.signal_csv import SignalCSVSourceAdapter28from ..sources.protocol import SourceReadError29from ..results.errors import BudgetExceededError, ResultClosedError30from ..results.spool import _cleanup_owner, _reserve_owned_ram, _release_owned_ram31from .csv_reader import _CSVResources, _PhysicalLines, _CSVRecords32from .history import _metadata_bytes333435@dataclass(frozen=True)36class CSVObservation:37 """保存单个实际完成read的原行时间范围，不代替来源文件身份或同步证明。3839 request是本read实际使用的原对象，read_identity对应生产它的read.identity。40 row_count为非空且通过原时间解析的行数；完整无时间行时两个端点均为None。41 模型只校验字段，后继调度仍须借实际read及其完成事实，不能凭手造模型证明读取。42 """4344 source: str # 本次实际来源名称。45 session_id: str # 原请求所属会话。46 request: ReadRequest # 真正读取所用原请求，保持对象身份而非复制值。47 read_identity: str # 每次独立读取的新身份，同一request的两次读不会相同。48 first_ns: int | None # 第一条实际原时间行的转换后整数ns，空输入为None。49 last_ns: int | None # 最后一条实际原时间行的转换后整数ns，空输入为None。50 row_count: int # 原时间行总数，空投影也计，不按信号列样本数计算。51 source_eof: bool # 仅完整whole-source EOF的事实，完成模型必须为True。5253 def __post_init__(self):54 """校验完成事实结构，不读取文件、推测端点或修改原请求。5556 参数: 无，使用冻结字段。返回: None。异常: 类型、身份、整数时间或行数矛盾。57 """58 # 原来源和会话保持安全名称，读取身份只是非空独立文本。59 require_identifier(self.source, 'csv_observation.source')60 require_identifier(self.session_id, 'csv_observation.session_id')61 require_text(self.read_identity, 'csv_observation.read_identity')62 if not isinstance(self.request, ReadRequest) or self.request.session_id != self.session_id:63 raise ValueError('CSV观测须保留实际会话的原ReadRequest')64 if type(self.row_count) is not int or self.row_count < 0 or self.source_eof is not True:65 raise ValueError('CSV观测须为非负原生行数及完整来源EOF')66 # 空输入没有可猜测的0或请求范围；非空同刻原行允许首末相等。67 if self.row_count == 0:68 if self.first_ns is not None or self.last_ns is not None:69 raise ValueError('空CSV观测的首末时间须均为None')70 else:71 require_int64_ns(self.first_ns, 'csv_observation.first_ns')72 require_int64_ns(self.last_ns, 'csv_observation.last_ns')73 if self.first_ns > self.last_ns:74 raise ValueError('CSV观测首末时间不能回退')757677class CSVObservationRead:78 """拥有一次原生成器、可选检查器和单当前批，固定事实费到显式close释放。7980 _iterator、_resources及_handle分别保留最高真实责任，清理失败可重试。81 _candidate仅为原生成器的末端候选；_exhausted要求真正StopIteration和清理成功。82 正常完成后移除适配器活动登记，_observation可独立存活至本owner关闭。83 """8485 def __init__(self, owner, request):86 """先预约本次请求及事实容器，再创建原参数生成器，不打开输入。8788 参数: owner为实际BudgetedSignalCSVSourceAdapter；request为原ReadRequest。89 返回: 独立读取owner。异常: 原配置或预算异常；持续清理失败附最高恢复owner。90 """91 # 局部导入只作实际入口核对，避免模块载入时与预算来源形成循环。92 from .csv_reader import BudgetedSignalCSVSourceAdapter93 if not isinstance(owner, BudgetedSignalCSVSourceAdapter) or not isinstance(request, ReadRequest):94 raise TypeError('CSV观测读取需实际预算CSV来源及ReadRequest')95 owner._ensure_open()96 # 首次预算回调前安装所有责任槽；事实保留必要身份，不依赖adapter随后仍有spec。97 self._owner, self._budget, self._request = owner, owner.budget, request98 self._source, self._session_id = owner.spec.name, owner.session_id99 self._fixed_cost = self._held_cost = 0100 self._iterator = self._resources = self._handle = self._current = None101 self._identity = self._candidate = self._observation = None102 self._closed = self._closing = self._failed = self._exhausted = False103 try:104 # 原请求固定费加独立事实/身份/恢复容器上界，在UUID和原validator创建前先付。105 cost = 16384 + len(owner.signals) * 4096 + 32768 + 16 * _metadata_bytes((106 self._source, self._session_id, request))107 _reserve_owned_ram(self, cost, '_fixed_cost')108 self._identity = uuid4().hex109 self._iterator = SignalCSVSourceAdapter._request_batches(owner, request, observation_sink=self)110 except BaseException as exc:111 # 构造失败保留原预算/配置错误，清理失败不能丢失最高真实请求责任。112 self._failed = True113 self._adopt_cleanup(exc)114 _cleanup_owner(self, exc, 'csv_observation_read_cleanup_owner', 'CSV观测读取构造')115 raise116117 @property118 def identity(self):119 """读取本read独立身份。参数: 无。返回: 非空str。异常: 无。"""120 # 身份不来自request值或adapter全局状态，同一请求的并发读取可明确区分。121 return self._identity122123 @property124 def closed(self):125 """读取完整关闭事实。参数: 无。返回: bool。异常: 无。"""126 # 实际子句柄和固定费用都清完后才为True。127 return self._closed128129 @property130 def failed(self):131 """读取操作失败事实。参数: 无。返回: bool。异常: 无。"""132 # 正常提前close也不产生完成事实，但不伪称为解析失败。133 return self._failed134135 @property136 def exhausted(self):137 """读取正常耗尽事实。参数: 无。返回: bool。异常: 无。"""138 # 最后source_eof批已交付仍为False，直到下一真实StopIteration及清理成功。139 return self._exhausted140141 @property142 def observation(self):143 """借读本次已完成的冻结观测，借期到本read的close。144145 参数: 无。返回: CSVObservation。异常: ResultClosedError为未完成、失败或关闭。146 """147 # 不查询借用adapter，正常完成事实可在adapter关闭以后继续读取。148 if self._closed or self._closing or self._failed or not self._exhausted or self._observation is None:149 raise ResultClosedError('CSV观测读取尚未正常完成或已关闭')150 return self._observation151152 def _read_resources(self):153 """为本次生成器建立独立事务式检查器，在文件打开前保留最高责任。154155 参数: 无。返回: _ObservationCSVResources。异常: 原预算、登记或构造错误。156 """157 # 检查器构造与登记失败都由本read接管，不以request值查找或借其他请求资源。158 try:159 self._resources = _ObservationCSVResources(self._owner)160 self._owner._resources.add(self._resources)161 except BaseException as exc:162 self._adopt_cleanup(exc)163 raise164 return self._resources165166 def _created_iterator(self, iterator):167 """在原活动set登记前接住本次实际生成器。168169 参数: iterator为原请求包装生成器。返回: None。异常: 无预期异常。170 """171 # 原入口尚未返回也可能登记失败，构造清理须能关闭这个最高实际子对象。172 self._iterator = iterator173174 def _opened_handle(self, handle):175 """在适配器set登记前接住本次真实文件句柄。176177 参数: handle为原Path.open返回的实际只读句柄。返回: None。异常: 无预期异常。178 """179 # 原来源逐文件执行，最多一个实际文件；前一文件未关闭不能进入下一文件。180 self._handle = handle181182 def _close_handle(self, handle, primary):183 """沿原文件关闭动作保根因，并保留实际未关闭句柄供重试。184185 参数: handle为实际文件；primary为原读取错误、GeneratorExit或None。186 返回: None。异常: 无业务根因时原关闭错误传播，已有根因只补note。187 """188 # 原helper仍执行真正close及原登记处理；新增责任只补其失败后句柄可达性。189 try:190 self._owner._close_handle(handle, primary)191 except BaseException as cleanup:192 if primary is None or isinstance(primary, GeneratorExit):193 raise194 primary.add_note(f'CSV观测文件关闭附加失败: {type(cleanup).__name__}')195 finally:196 if handle.closed:197 self._handle = None198199 def _close_resources(self, primary):200 """原生成器退出时清理本次检查器，清理失败不覆盖业务或取消根因。201202 参数: primary为当前异常或None。返回: None。异常: 正常退出时的原资源关闭错误。203 """204 # 文件仍未关闭则保留其预算检查器，不能提前释放读取固定槽后丢失真实FD责任。205 if self._handle is not None:206 if primary is None:207 raise SourceReadError('CSV观测文件尚未成功关闭')208 return209 try:210 self._resources.close()211 except BaseException as cleanup:212 if primary is None or isinstance(primary, GeneratorExit):213 raise214 primary.add_note(f'CSV观测资源关闭附加失败: {type(cleanup).__name__}')215216 def _end_file(self, resources, primary):217 """清理单文件表头及行费用，保留现场正在传播的读取或关闭主异常。218219 参数: resources为本read的实际检查器；primary为原生成器finally现场的220 sys.exception()，可能为解析错误、取消、文件关闭失败、GeneratorExit或None。221 返回: None。异常: 正常文件结束或GeneratorExit时的原清理错误继续传播；222 已有其他主异常时仅给它增加note，事务式费用槽留待同一read.close重试。223 """224 # 不能事后从异常链猜主错；仅在已有真正业务或取消主异常时保护其原身份和分类。225 try:226 resources.end_file()227 except BaseException as cleanup:228 if primary is None or isinstance(primary, GeneratorExit):229 raise230 primary.add_note(f'CSV观测单文件资源清理附加失败: {type(cleanup).__name__}')231232 def _accept_candidate(self, first, last, count):233 """仅接收原生成器最终协议成功后的候选，不在此声明完成。234235 参数: first、last为原时间行首末ns或None，count为原sequence行数。236 返回: None。异常: 无预期异常，最终typed字段在真正耗尽后校验。237 """238 # 单个有限tuple早由固定费覆盖，不保存逐行回调或整个历史。239 self._candidate = (first, last, count)240241 def _adopt_cleanup(self, exc):242 """接管本路径实际未返回的最高资源检查器。243244 参数: exc为原异常。返回: None。异常: 无预期异常。245 """246 # 只接本次新资源属性，绝不接其他请求或把借用adapter当作自有资源。247 resources = getattr(exc, 'csv_observation_resources_cleanup_owner', None)248 if resources is not None and self._resources is None:249 self._resources = resources250251 def __iter__(self):252 """返回唯一读取owner。参数: 无。返回: self。异常: ResultClosedError为不可推进。253 """254 # 失败或半关闭状态只允许close，不创建带第二份同源状态的新游标。255 self._ensure_active()256 return self257258 def _ensure_active(self):259 """拒绝关闭、清理或失败后的推进。参数: 无。返回: None。异常: ResultClosedError。260 """261 # 完成后的重复next在独立检查后返回StopIteration，不再依赖借用adapter。262 if self._closed or self._closing or self._failed:263 raise ResultClosedError('CSV观测读取已关闭、正在清理或推进失败')264265 def __next__(self):266 """转交原batch，真正StopIteration并成功清理后才安装完成事实。267268 参数: 无。返回: 原SignalBatch。异常: 正常StopIteration；原读取、预算或取消错误269 保留csv_observation_read_cleanup_owner，失败后仅可close。270 """271 self._ensure_active()272 if self._exhausted:273 raise StopIteration274 try:275 self._owner._ensure_open()276 self._current = None277 try:278 self._current = next(self._iterator)279 except StopIteration:280 # EOF批交付后的生成器finally已真正成功返回，再确认全部自有读取资源关闭。281 self._close_request()282 if self._candidate is None:283 raise ValueError('CSV观测缺少实际完整EOF候选')284 first, last, count = self._candidate285 self._observation = CSVObservation(self._source, self._session_id, self._request,286 self._identity, first, last, count, True)287 self._candidate = None288 self._exhausted = True289 else:290 return self._current291 except BaseException as exc:292 # 真正正常StopIteration在外部产生；读取错误及清理取消不能被当作完成。293 self._failed = True294 self._adopt_cleanup(exc)295 exc.csv_observation_read_cleanup_owner = self296 exc.add_note('恢复后调用csv_observation_read_cleanup_owner.close()')297 raise298 raise StopIteration299300 def _close_request(self):301 """关闭本次原生成器、文件和检查器，保原首错及每个未完责任供重试。302303 参数: 无。返回: None。异常: 首个真实关闭错误，其他错误附note而不覆盖。304 """305 # 结束当前批借引用；原generator仍可能持批，必须先实际关闭再释放检查器数组槽。306 self._current = None307 failure = None308 if self._iterator is not None:309 try:310 self._iterator.close()311 except BaseException as exc:312 failure = exc313 else:314 self._owner.active.discard(self._iterator)315 self._iterator = None316 # 生成器关闭失败后仍尝试唯一真实文件，成功才丢句柄；不关闭其他请求文件。317 if self._handle is not None:318 try:319 self._handle.close()320 except BaseException as exc:321 if failure is None:322 failure = exc323 else:324 failure.add_note(f'CSV观测残留文件关闭附加失败: {type(exc).__name__}')325 finally:326 if self._handle.closed:327 self._owner.handles.discard(self._handle)328 self._handle = None329 if self._resources is not None and self._handle is None:330 try:331 self._resources.close()332 except BaseException as exc:333 if failure is None:334 failure = exc335 else:336 failure.add_note(f'CSV观测残留资源关闭附加失败: {type(exc).__name__}')337 else:338 self._resources = None339 # 原错仍需报告，即使后续尝试清了资源也不能把首次清理失败伪装成正常耗尽。340 if failure is not None:341 raise failure342 self._owner._requests.discard(self)343344 def close(self):345 """可重试释放本read及事实费，不关闭借用来源、其他请求或删除文件。346347 参数: 无。返回: None，成功后幂等。异常: 原文件、子资源或预算取消，未完责任保留。348 """349 # 提前结束只清候选，不补EOF；已完成事实到本close才停止开放。350 if self._closed:351 return352 self._closing = True353 self._candidate = self._observation = None354 self._close_request()355 self._request = None356 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')357 if self._held_cost == 0:358 self._closed = True359360361class _ObservationCSVResources(_CSVResources):362 """只为可选新路径补事务式预算责任，沿用原样本及装箱估算和金额。363364 各用途实际金额既保留在固定属性槽也同步到原charges检查接口；_held_cost365 仅为本资源的实际总额。清理取消前后均能重试，不提前closed或丢已付款槽。366 """367368 def __init__(self, owner):369 """建立实际槽后调用原检查器构造，固定读取费用仍为原65536字节。370371 参数: owner为借用预算CSV来源。返回: 独立检查器。异常: 原预算或取消错误，附恢复owner。372 """373 # 在原构造的第一次_add前安装所有有限用途槽，费用名称不随原行数增长。374 self._budget, self._held_cost = owner.budget, 0375 self._fixed_cost = self._reading_cost = self._header_cost = self._row_cost = 0376 self._pending_cost = self._work_cost = self._array_cost = self._protocol_cost = 0377 try:378 super().__init__(owner)379 except BaseException as exc:380 _cleanup_owner(self, exc, 'csv_observation_resources_cleanup_owner', 'CSV观测资源构造')381 raise382383 def _sync(self, key):384 """同步原charges视图与实际属性槽。参数: key为有限用途名。返回: None。异常: 无预期异常。385 """386 # 只在本资源预付固定容器内更新至多八个键，不产生逐样本费用字典。387 amount = getattr(self, '_' + key + '_cost')388 if amount:389 self.charges[key] = amount390 else:391 self.charges.pop(key, None)392393 def _add(self, key, amount):394 """申请成功前后取消均登记实际自有槽，不从其他owner余额推测。395396 参数: key为原用途；amount为原估算字节。返回: None。异常: 原预算或取消错误。397 """398 # helper仅围绕一笔真实预算调用判断余额，finally同步实际登记而非预先增加。399 try:400 _reserve_owned_ram(self, amount, '_' + key + '_cost')401 finally:402 self._sync(key)403404 def _release_amount(self, key, amount):405 """只抵销某用途一笔实际金额，释放前后取消都保留正确剩余。406407 参数: key为原用途；amount为本用途已有字节。返回: None。异常: 原释放或取消错误。408 """409 # 实际释放后取消也及时清槽，重试不多扣共享预算或借用来源费用。410 try:411 _release_owned_ram(self, amount, '_' + key + '_cost')412 finally:413 self._sync(key)414415 def _release(self, key):416 """抵销用途完整余额。参数: key为有限用途名。返回: None。异常: 原释放取消。417 """418 # 空用途沿helper不触发外部释放回调，不把未付款金额抵销给其他owner。419 self._release_amount(key, getattr(self, '_' + key + '_cost'))420421 def _move(self, source, target, amount):422 """在已付款用途间交接金额，不增加或释放共享预算。423424 参数: source、target为用途名，amount为已付款字节。返回: None。异常: ValueError为内部金额矛盾。425 """426 # 真实数组和当前行交接只变本资源责任槽，总_held_cost始终不变。427 source_slot, target_slot = '_' + source + '_cost', '_' + target + '_cost'428 if amount < 0 or amount > getattr(self, source_slot):429 raise ValueError('CSV观测资源交接超出原用途余额')430 setattr(self, source_slot, getattr(self, source_slot) - amount)431 setattr(self, target_slot, getattr(self, target_slot) + amount)432 self._sync(source)433 self._sync(target)434435 def lines(self, handle):436 """建立原物理行读取的事务式费用包装。参数: handle为真实文件。返回: 行游标。异常: 无预期异常。437 """438 # CSV解释仍由原csv.reader负责，只更换释放前后取消不会丢账的物理读取计费。439 self.line_source = _ObservationPhysicalLines(self, handle)440 return self.line_source441442 def records(self, reader):443 """包装原csv.reader的费用交接。参数: reader为原解释器。返回: 原记录游标。异常: 无预期异常。444 """445 # 表头、行内容及reader.line_num均来自原解释器，不复制解析规则。446 return _ObservationCSVRecords(self, reader)447448 def after_pack(self, validator):449 """沿原金额把装箱workspace交给数组与协议，余量释放取消仍保持可恢复。450451 参数: validator为已accept的原协议状态。返回: None。异常: 原账本或取消错误。452 """453 # 与原after_pack逐项金额相同，只把尚未实际释放的workspace留在真实_work_cost。454 state_cost = sum(512 + len(state[1]) * 256 for state in validator.pending.values())455 self._release('array')456 self._release('protocol')457 # 共享原值与来源引用的已有费用随批次转交，不重复计费。458 self._move('pending', 'array', self._pending_cost)459 self._move('work', 'array', self.array_cost)460 self._move('work', 'protocol', state_cost)461 self._release('work')462463 def close(self):464 """逐实际用途抵销，全部成功后才closed和移除自有登记。465466 参数: 无。返回: None，成功后幂等。异常: 原释放取消，尚未抵销槽保留供重试。467 """468 # 不在释放前pop费用或声明closed；持续故障后原对象仍保有准确金额责任。469 if self.closed:470 return471 self.line_source = None472 for key in ('fixed', 'reading', 'header', 'row', 'pending', 'work', 'array', 'protocol'):473 self._release(key)474 if self._held_cost == 0:475 self.closed = True476 self.owner._resources.discard(self)477478479class _ObservationPhysicalLines(_PhysicalLines):480 """沿原readline限长算法，在临时物理行费用释放前后取消时保留实际槽。"""481482 def __next__(self):483 """按原容量真实读取一物理行，只更换实际金额的事务式抵销。484485 参数: 无。返回: 原含换行文本，真实EOF停止。异常: 原I/O、预算或取消错误。486 """487 # 字符与字段费用沿基础读取器计算，不把截断行交给 CSV 解释器。488 budget = self.resources.budget489 capacity = (budget.memory_budget_bytes - budget.ram_bytes - 8192) // 16 - 1490 if capacity <= 0:491 raise BudgetExceededError('CSV物理行/引用记录累积超出剩余RAM估算')492 amount = 8192 + (capacity + 1) * 16493 first_line = not self.resources._reading_cost494 self.resources._add('reading', amount)495 try:496 text = self.handle.readline(capacity + 1)497 except BaseException as exc:498 try:499 self.resources._release_amount('reading', amount)500 except BaseException as cleanup:501 exc.add_note(f'CSV观测物理行清理附加失败: {type(cleanup).__name__}')502 raise503 # 原文本已受全容量预算覆盖，释放多余部分失败也不会丢失已保留的读取费用。504 retained = self._retained_line_bytes(text, first_line)505 if retained > amount:506 self.resources._add('reading', retained - amount)507 else:508 self.resources._release_amount('reading', amount - retained)509 if len(text) > capacity:510 raise BudgetExceededError('CSV物理行超过剩余RAM允许长度，未解析截断行')511 if not text:512 raise StopIteration513 return text514515516class _ObservationCSVRecords(_CSVRecords):517 """保留原csv.reader记录与物理行语义，只显式交接已付款的读取责任槽。"""518519 def __next__(self):520 """先从原解释器取完整记录，再把reading费用转给表头或当前行。521522 参数: 无。返回: 原单元格list。异常: 原EOF、CSV、I/O、预算或取消错误。523 """524 # 上一原记录仍可能由生成器引用，新读取费用与旧记录按原方式重叠计费。525 cells = next(self.reader)526 key = 'header' if self.first else 'row'527 self.resources._release(key)528 self.resources._move('reading', key, self.resources._reading_cost)529 self.first = False530 return cells531