chronaxis.legacy.raw_store
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""按首次CAN读取的原序暂存完整报文，供片段发布回读而不重扫原输入。23RawFrameStore借用ResourceBudget，append接收同次LegacyDecodedFrame和已显式映射的4int64纳秒，仅保存sequence/time_ns/原frame，不存decoded.values。每行是有界records5JSON包加一个换行；未知ID、同刻帧及全部公开写出字段保留，不排序、筛选或去重。6完整写入后才推进计数和流SHA；部分写入、预算、编码或取消错误均poison，禁止seal。7seal须真实writer关闭成功才开放原序reader；reader逐行校验结构、原序和时间，真实8EOF核条数及完整SHA，关闭也成功后才verified_eof。提前close不产生完整读取证明。910本层只使用标准库、冻结records和新分析层预算工具，不导入旧CAN库或调用解码器。11固定、append工作空间与各活reader分别预付；reader数量受共享预算约束，未保存12全部帧或全部文件路径。账本是保守估算而非整进程RSS；调用方长期保留已返回行须13自行承担空间。close只关闭自有真实FD/费用，保留文件及累计spool预留，不关借用14budget。全部槽先初始化，关闭失败可重试；原异常/KI不被清理覆盖，未返回reader15及持续失败资源从最高raw_frame_store_cleanup_owner恢复，不凭余额差补账。16"""1718# 只封装新暂存和typed记录；原reader/parse/writer实现及调用次数由外层保持。19from hashlib import sha25620from pathlib import Path21from ..core.validation import require_identifier22from ..results.budget import ResourceBudget23from ..results.errors import ResultClosedError, ResultFormatError24from ..results.spool import _cleanup_owner, _reserve_owned_ram, _release_owned_ram25from ..timebase.conversion import require_int64_ns26from .records import (MAX_WIRE_BYTES, LegacyDecodedFrame, StoredRawFrame,27 frame_to_dict, frame_from_dict, encode_packet, decode_packet)282930class RawFrameStore:31 """拥有一个原序JSONL和所有独立reader，完成事实不等于FD已经全部关闭。3233 state为building/sealed/failed/cancelled；closed另记所有自有资源实际释放完成。34 _pending_reader先安装未返回实例，_readers保存已注册实例；两者均由本父close。35 固定/当前append费用是实际交易槽，累计spool只增加，不从共享账本推测归属。36 """3738 def __init__(self, budget, work_dir, source_name, *, max_row_bytes=MAX_WIRE_BYTES, integrity=True):39 """预付有限状态，在显式新目录独占打开原始帧暂存。4041 参数: budget为借用ResourceBudget；work_dir为必须不存在的str/Path；source_name42 为安全来源名；max_row_bytes为含换行行上限，1到MAX_WIRE_BYTES，默认65536。43 返回: building存储。异常: 类型/路径/预算/文件/取消原错；持续清理失败保最高owner。44 """45 if not isinstance(budget, ResourceBudget) or not isinstance(work_dir, (str, Path)):46 raise TypeError('原始帧暂存需ResourceBudget及显式新目录')47 if type(max_row_bytes) is not int or not 1 <= max_row_bytes <= MAX_WIRE_BYTES:48 raise ValueError('原始帧行上限需1到MAX_WIRE_BYTES的原生int')49 require_identifier(source_name, 'raw_store.source_name')50 # 先装全部关闭责任，首次reserve更新后取消也不会留下未归属费用或FD。51 if type(integrity) is not bool: raise TypeError('integrity需要bool')52 self.integrity=integrity53 self.budget = self._budget = budget54 self.source_name, self.max_row_bytes = source_name, max_row_bytes55 self.path = self.work_dir = self.file_path = None56 self._writer = self._pending_reader = self._pending_payload = None57 self._readers = self._digest = None58 self._fixed_cost = self._append_cost = self._held_cost = self._spool_bytes = 059 self.count, self.sha256, self._last_time_ns = 0, None, None60 self.state, self._closed, self._closing = 'building', False, False61 try:62 # 路径字符串只作有限身份，不深遍历来源配置或已解码values。63 requested = Path(work_dir)64 cost = 65536 + 32 * (len(str(requested)) + len(source_name))65 _reserve_owned_ram(self, cost, '_fixed_cost')66 resolved = requested.resolve()67 if resolved.exists():68 raise FileExistsError(resolved)69 self.path = self.work_dir = resolved70 self._readers, self._digest = set(), sha256() if integrity else None71 resolved.mkdir(parents=True, exist_ok=False)72 self.file_path = resolved / 'raw_frames.jsonl'73 self._writer = self.file_path.open('xb', buffering=0)74 except BaseException as exc:75 self._fail(exc)76 raise7778 @property79 def closed(self):80 """读取真实关闭事实。参数: 无。返回: bool。异常: 无；费用/FD未清不能为True。"""81 return self._closed8283 @property84 def spool_bytes(self):85 """读取本owner直接预留的累计暂存金额。参数: 无。返回: int字节。异常: 无。"""86 return self._spool_bytes8788 def _require(self, state):89 """拒绝半关闭或错误状态，不能把poison前缀当完整输入。9091 参数: state为所需building/sealed。返回: None。异常: ResultClosedError/RuntimeError。92 """93 if self._closed or self._closing:94 raise ResultClosedError('原始帧暂存已关闭或正在清理')95 if self.state != state:96 raise RuntimeError('原始帧暂存状态不允许此操作: ' + self.state)9798 def _fail(self, exc):99 """保原根因、安装最高恢复owner，再有限尝试真实资源关闭。100101 参数: exc为实际主异常。返回: None。异常: 清理副错只加note，不替换原错或取消对象。102 """103 self.state = 'cancelled' if isinstance(exc, (KeyboardInterrupt, SystemExit)) else 'failed'104 exc.raw_frame_store_cleanup_owner = self105 _cleanup_owner(self, exc, 'raw_frame_store_cleanup_owner', '原始帧暂存')106107 def _reserve_spool(self, amount):108 """写入前逐笔预留累计磁盘，成功后取消也记录本笔，不倒扣保留文件。109110 参数: amount为本行bytes。返回: None。异常: 原配额/取消；未成功不增加本owner金额。111 """112 before = self._budget.spool_bytes113 try:114 self._budget.reserve_spool(amount)115 finally:116 if self._budget.spool_bytes == before + amount:117 self._spool_bytes += amount118119 def append(self, record, time_ns):120 """追加同次原帧，保0起连续sequence、非递减ns及全部同刻/未知ID。121122 参数: record为LegacyDecodedFrame；time_ns为显式映射后int64整数，非原浮点秒。123 返回: None。异常: 原类型/顺序/编码/行长/预算/写入/取消；任何失败poison且不可seal。124 """125 self._require('building')126 try:127 if type(record) is not LegacyDecodedFrame:128 raise TypeError('原始帧暂存仅接受LegacyDecodedFrame')129 mapped = require_int64_ns(time_ns, 'raw_store.time_ns')130 if record.sequence != self.count or self._last_time_ns is not None and mapped < self._last_time_ns:131 raise ValueError('原始帧sequence必须从0连续且时间不能回退')132 # 编码前付完整wire工作空间；小max_row_bytes也不能漏算超长channel的拒绝路径。133 _reserve_owned_ram(self, 65536 + 64 * MAX_WIRE_BYTES, '_append_cost')134 body = encode_packet({'sequence': record.sequence, 'time_ns': mapped,135 'frame': frame_to_dict(record.frame)})136 self._pending_payload = body + b'\n'137 if len(self._pending_payload) > self.max_row_bytes:138 raise ValueError('原始帧编码行超过max_row_bytes')139 self._reserve_spool(len(self._pending_payload))140 # 真实unbuffered写可能只完成前缀；只要未完整成功，文件保留且流永久poison。141 offset = 0142 while offset < len(self._pending_payload):143 written = self._writer.write(self._pending_payload[offset:])144 if type(written) is not int or written <= 0:145 raise OSError('原始帧写入未推进')146 offset += written147 if self._digest is not None:148 self._digest.update(self._pending_payload)149 self.count += 1150 self._last_time_ns = mapped151 self._pending_payload = None152 _release_owned_ram(self, self._append_cost, '_append_cost')153 except BaseException as exc:154 self._fail(exc)155 raise156157 def _close_writer(self):158 """真实FD关闭后才清writer槽，close前持续失败保留同一handle。159160 参数: 无。返回: None。异常: 原close/取消；close后抛错也据真实closed清槽。161 """162 if self._writer is not None:163 try:164 self._writer.close()165 finally:166 if self._writer.closed:167 self._writer = None168169 def seal(self):170 """成功关闭writer后固定整流SHA和条数，空来源亦可完整封口。171172 参数: 无。返回: None。异常: 原flush/close/摘要/取消，失败不能读或再次seal。173 """174 self._require('building')175 try:176 self._writer.flush()177 self._close_writer()178 digest = self._digest.hexdigest() if self._digest is not None else None179 self.sha256, self.state = digest, 'sealed'180 except BaseException as exc:181 self._fail(exc)182 raise183184 def iter_frames(self):185 """为已封口文件创建独立有限reader，不复用其他reader位置或原CAN输入。186187 参数: 无。返回: 逐行StoredRawFrame迭代器。异常: 状态/预算/打开/注册/取消；188 未返回reader先在本父槽可达，持续关闭失败从最高raw_frame_store_cleanup_owner恢复。189 """190 self._require('sealed')191 try:192 reader = _RawFrameReader(self)193 self._pending_reader = None194 return reader195 except BaseException as exc:196 self._fail(exc)197 raise198199 def close(self):200 """关闭本store的所有实际reader/writer及自有费用，文件和累计磁盘保留。201202 参数: 无。返回: None，完成后重复调用安全。异常: 首个实际关闭/抵销错误原样传播，203 其他清理错仅note；未完槽不清，最高恢复owner就是本store，不接管借用账本。204 """205 if self._closed:206 return207 self._closing = True208 if self.state == 'building':209 self.state = 'cancelled'210 first_error = None211 # 未返回子对象先处理；close会解除自身登记，再处理其余活reader避免重复责任。212 if self._pending_reader is not None:213 try:214 self._pending_reader.close()215 except BaseException as exc:216 first_error = exc217 for reader in () if self._readers is None else tuple(self._readers):218 try:219 reader.close()220 except BaseException as exc:221 if first_error is None:222 first_error = exc223 else:224 first_error.add_note('原始帧其他reader关闭附加失败: ' + type(exc).__name__)225 try:226 self._close_writer()227 except BaseException as exc:228 if first_error is None:229 first_error = exc230 else:231 first_error.add_note('原始帧writer关闭附加失败: ' + type(exc).__name__)232 # 活stream数量的元数据和此关闭快照由每个reader费用覆盖，不与原行总数挂钩。233 if self._writer is None and self._pending_reader is None and not self._readers:234 self._pending_payload = None235 for slot in ('_append_cost', '_fixed_cost'):236 try:237 _release_owned_ram(self, getattr(self, slot), slot)238 except BaseException as exc:239 if first_error is None:240 first_error = exc241 else:242 first_error.add_note('原始帧费用抵销附加失败: ' + type(exc).__name__)243 if self._held_cost == 0:244 self._closed = True245 if first_error is not None:246 first_error.raw_frame_store_cleanup_owner = self247 raise first_error248249250class _RawFrameReader:251 """持有一个文件位置、摘要和有界读槽，父只保实际活stream引用。252253 关闭中禁止继续读；真实FD/本笔费用/父登记全部成功后closed。254 verified_eof须真实EOF、完整SHA/count和成功关闭，不从最后一条交付推断。255 """256257 def __init__(self, owner):258 """先安装本父未返回槽，再预约/打开/登记真实reader。259260 参数: owner为活sealed RawFrameStore。返回: 独立reader。异常: 原预算/打开/注册/取消；261 本构造不遗失self，由父工厂统一清理并附最高store恢复属性。262 """263 self.owner, self._budget = owner, owner.budget264 self.handle = self._digest = None265 self._fixed_cost = self._held_cost = self._count = 0266 self._last_time_ns = None267 self.closed = self._closing = self.verified_eof = False268 owner._pending_reader = self269 # 仅新原帧暂存预读8192字节，另预付16384；不改变旧reader或原逻辑行/EOF验证。270 # 同一handle持缓冲及raw FD，关闭失败继续由原pending/readers槽保责。271 _reserve_owned_ram(self, 65536 + 64 * owner.max_row_bytes + 16384, '_fixed_cost')272 self._digest = sha256() if owner.integrity else None273 self.handle = owner.file_path.open('rb', buffering=8192)274 owner._readers.add(self)275276 def __iter__(self):277 """返回原位置。参数: 无。返回: self。异常: 无，不重新打开文件。"""278 return self279280 def __next__(self):281 """限长读取并核原序/时间，完整EOF核流身份再关闭。282283 参数: 无。返回: StoredRawFrame。异常: StopIteration仅完整EOF；损坏/预算/关闭/取消284 保原错且poison父，最高raw_frame_store_cleanup_owner可清全部真实资源。285 """286 if self.closed:287 raise StopIteration288 self.owner._require('sealed')289 if self._closing:290 raise ResultClosedError('原始帧reader正在清理')291 try:292 payload = self.handle.readline(self.owner.max_row_bytes + 1)293 if payload:294 if len(payload) > self.owner.max_row_bytes or not payload.endswith(b'\n'):295 raise ResultFormatError('原始帧行超限或缺少完整换行')296 if self._count >= self.owner.count:297 raise ResultFormatError('原始帧封口后出现额外后缀')298 packet = decode_packet(payload[:-1])299 if set(packet) != {'sequence', 'time_ns', 'frame'}:300 raise ResultFormatError('原始帧暂存字段不完整或多余')301 record = StoredRawFrame(packet['sequence'], packet['time_ns'], frame_from_dict(packet['frame']))302 if record.sequence != self._count or self._last_time_ns is not None and record.time_ns < self._last_time_ns:303 raise ResultFormatError('原始帧原序或纳秒时间被破坏')304 if self._digest is not None:305 self._digest.update(payload)306 self._count += 1307 self._last_time_ns = record.time_ns308 return record309 # 仅实际文件EOF比较完整hash/count；最后一条已交付仍不能提前证明完整。310 if self._count != self.owner.count or self._digest is not None and self._digest.hexdigest() != self.owner.sha256:311 raise ResultFormatError('原始帧EOF条数或SHA不匹配')312 self.close()313 except BaseException as exc:314 self.owner._fail(exc)315 raise316 self.verified_eof = True317 raise StopIteration318319 def close(self):320 """可重试关闭本reader，不关闭父或其他独立reader。321322 参数: 无。返回: None。异常: 原FD/预算/解除登记错误，最高父恢复属性保持同一责任；323 真实FD关闭后抛取消也不重关，已释放费用不重复抵销。324 """325 if self.closed:326 return327 self._closing = True328 try:329 if self.handle is not None and not self.handle.closed:330 self.handle.close()331 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')332 if self.owner._readers is not None:333 self.owner._readers.discard(self)334 if self.owner._pending_reader is self:335 self.owner._pending_reader = None336 self.closed = True337 except BaseException as exc:338 exc.raw_frame_store_cleanup_owner = self.owner339 raise340