chronaxis.execution.can_reader
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""把同次旧CAN逐帧输出接成预算信号批，保留原始帧与真实观测证明。23adapter构造/describe只检查声明并创建新工作目录，不读BLF/DBC、不启动进程。4独立CANObservationRead首次next才启动CAN读取进程，先复制同次READY并核实际非空DBC单位，5再逐原yield读取；显式all每帧调用一次原parse，仅retain_raw_frames时将全部原帧写入6RawFrameStore。所有原帧计读取数，时间按source/selected/none分别采集，7本层收到的signals是运行实际需要的定义；requested只解码这些字段和消息指标所需ID，8只传输所需值；auto跳过无用途字段与时间，full/source保原完整检查。显式all保留原全解码检查。9当前原帧的匹配别名按全部列样本总和batch_rows拆批；其他报文10不更新此信号，匹配定位但原dict缺key则None→MISSING，不延用旧值或猜DECODE_ERROR。1112时间仅原浮点秒convert_time(...,'s')加一次offset。当前ns-1才作为非EOF watermark，13所以同刻后继不会被提前承诺；末次真实EOF交付独立空批，下一次StopIteration前须14确认CAN读取进程退出、raw writer封口、validator结束和本次输入关闭成功，才开放CANObservation。15runtime_metadata保原READY冻结键形状，None/空unit原样表示未核实，不能假verified。16take_raw_store只在完成后一次移交原store；未移交则本read.close负责，不关已移交store。17显式capture_message_updates=True时，同一record在raw追加后另存完整decoded仓，18不在别名拆批处重复捕获。消息仓保同次READY.profile/read_identity和原量化事实，19真EOF才封口；take_message_store同样先安装接收父，再撤本子责任。默认关闭时20不创建消息目录或额外预约，原PULL/parse、字段投影与批次协议保持。2122来源/请求/同次元数据、当前原记录、装箱与validator动态去重状态分别先付费。23未返回read及CAN读取进程/raw构造责任均真实安装后再推进，持续清理失败保最高恢复owner；24close可重试且不抵销借款。完成read脱离adapter活登记，事实可活到自身close。25全部只用新分析层和已有pack_decoded两后端，不修改旧链、重复读DBC或声明RSS上限。26batch_read默认自动选择：requested由can_batch_reader接有限批，all保原逐条读取；27显式bool仍覆盖自动选择。未保存raw由公开元数据None/None/0明确表达。28直接累计且没有普通样本请求时，另核两段真实输入字节的SHA256：全部原时间、29所需记录的长度与JSON；selected/none的全时间SHA为None。只借用已严格解包的原包，不重编码或保存副本；真实EOF及30进程关闭后才开放指纹。它不是普通信号样本摘要，read关闭后由接收方自付保留费用。31"""3233# 数值归一化复用既有pack；外层只处理实际来源身份、原行进度和资源责任。34from dataclasses import dataclass35import hashlib36from pathlib import Path37from uuid import uuid438from ..core.binding import SignalDefinition39from ..core.validation import require_identifier, require_text40from ..legacy.process_read import LegacyCANProcessRead41from ..legacy.raw_store import RawFrameStore42from ..legacy.message_store import DecodedMessageStore, MAX_DECODED_MESSAGE_ROW_BYTES43from ..legacy.records import LegacyCANRequest, field_to_dict44from ..legacy.batch_wire import HEADER45from ..legacy.batch_worker import REQUESTED_VALUES_POLICY46from ..results.budget import ResourceBudget47from ..results.errors import ResultClosedError48from ..results.spool import _cleanup_owner, _reserve_owned_ram, _release_owned_ram49from ..sources.can_spec import LegacyCANSpec, CANSourceRef, can_field_from_key50from ..sources.descriptions import describe_fields, source_timebase_identity51from ..sources.models import (ReadRequest, SignalBatch, SourceDescription, SourceCapabilities, freeze_metadata)52from ..sources.packing import pack_decoded53from ..sources.protocol import BatchValidator, SourceClosedError, SourceReadError54from ..timebase.conversion import checked_add_ns, convert_time, require_int64_ns55from .history import _metadata_bytes, _pending_bytes565758@dataclass(frozen=True)59class CANObservation:60 """保存本read实际全部yield的完成事实，不接受请求范围作为原行首末。"""6162 source: str # 当前来源名。63 session_id: str # 原会话。64 request: ReadRequest # 创建本read的原对象，不按值重建。65 read_identity: str # 每个read独立UUID。66 first_ns: int | None # 全部原yield经一次映射的首端，全空为None。67 last_ns: int | None # 全部原yield经一次映射的末端，全空为None。68 row_count: int # 包含未知ID/空投影的原yield数，不是样本数。69 source_eof: bool # 仅真实完整完成为True。70 time_scope: str = 'source' # source为全来源；selected仅目标；none未采集。71 retained_record_count: int = 0 # 本次实际保留记录数，不等于贡献数。72 full_check_count: int = 0 # 按需策略下非原生输入的完整检查回退数。7374 def __post_init__(self):75 """核完成事实结构。参数: 无。返回: None。异常: 类型/会话/端点或EOF矛盾。76 """77 require_identifier(self.source, 'can_observation.source')78 require_identifier(self.session_id, 'can_observation.session_id')79 require_text(self.read_identity, 'can_observation.read_identity')80 if (not isinstance(self.request, ReadRequest) or self.request.session_id != self.session_id81 or type(self.row_count) is not int or self.row_count < 0 or self.source_eof is not True):82 raise ValueError('CAN观测需原请求、非负原行数及完整EOF')83 if self.time_scope not in ('none', 'selected', 'source'):84 raise ValueError('CAN观测时间范围非法')85 if (type(self.retained_record_count) is not int or not 0 <= self.retained_record_count <= self.row_count86 or type(self.full_check_count) is not int or not 0 <= self.full_check_count <= self.row_count):87 raise ValueError('CAN观测保留/完整回退计数非法')88 if self.time_scope == 'none':89 if self.first_ns is not None or self.last_ns is not None:90 raise ValueError('未采集时间不能声明首末')91 return92 if self.row_count == 0 or self.time_scope == 'selected' and not self.retained_record_count:93 if self.first_ns is not None or self.last_ns is not None:94 raise ValueError('空CAN输入没有可猜测首末')95 else:96 require_int64_ns(self.first_ns)97 require_int64_ns(self.last_ns)98 if self.first_ns > self.last_ns:99 raise ValueError('CAN观测时间不能回退')100101102class BudgetedLegacyCANSourceAdapter:103 """拥有静态来源描述和活read登记，完成read与已移交store由新父负责。"""104105 def __init__(self, budget, spec, session_id, signals, *, backend='numpy', work_dir,106 capture_message_updates=False, batch_read=None, retain_raw_frames=True,107 input_check='full', time_scope='source',108 message_frame_ids=None,extra_fields=(), message_consumer=None, cycle_fields=(), metadata_fields=(), message_keys=(),109 input_fingerprints=True, history_integrity=True, frame_observer=None):110 """先付声明费用，建立显式新工作根但不访问原输入。111112 参数: budget为共享ResourceBudget；spec为LegacyCANSpec；session_id为会话；113 signals为SignalDefinition tuple；backend为numpy/reference；work_dir必须全新；114 capture_message_updates为严格bool，默认False，仅显式开启才存同次decoded仓；115 batch_read默认None自动选择：requested为True，all为False；显式bool保持选择，116 requested不能显式False；retain_raw_frames决定是否保存原帧；117 message_frame_ids为需要完整逐消息记录的ID，None保留所有实际返回记录。118 extra_fields只为显式报文重建传递连续量/复用选择器，不生成普通信号别名。119 message_keys为没有数值依赖也须触发的完整报文定位；120 message_consumer可选同步接收共同消息模型；只传signals所需值，保留对象须由消费者另付费用。121 cycle_fields独立请求初始化周期；metadata_fields仅请求字段候选，两者不添加解码值或普通样本。122 frame_observer仅显式导出时接收原帧身份，不保存额外载荷或扫描文件。123 input_fingerprints仅控制可选输入摘要，默认保持旧调用；必要值检查始终执行。124 input_check/time_scope为计划固定的检查和时间范围；低层默认full/source保旧入口。125 spec.decode_scope=requested使用批协议，将实际需要的字段及指标ID固定为解码集合；126 其他帧仍读取并检查原字段/时间，但不再生成或检查其解码值。127 返回: 静态adapter。异常: 原配置/目录/预算/取消；持续清理失败附can_source_cleanup_owner。128 """129 if batch_read is None and type(spec) is LegacyCANSpec:130 batch_read = (spec.decode_scope == 'requested' or len(spec.input_paths) > 1131 or spec.decode_policy != 'legacy' or bool(cycle_fields) or bool(metadata_fields))132 if (not isinstance(budget, ResourceBudget) or type(spec) is not LegacyCANSpec133 or type(signals) is not tuple or any(type(item) is not SignalDefinition for item in signals)134 or backend not in ('numpy', 'reference') or not isinstance(work_dir, (str, Path))135 or any(type(value) is not bool for value in (capture_message_updates, batch_read, retain_raw_frames, input_fingerprints, history_integrity))136 or message_frame_ids is not None and (type(message_frame_ids) is not tuple137 or any(type(value) is not int or not 0 <= value <= 0x1fffffff for value in message_frame_ids))):138 raise TypeError('CAN来源需预算、静态spec、定义tuple、后端及新工作目录')139 from ..sources.can_updates import require_message_keys140 require_message_keys(message_keys, channel=spec.channel)141 if message_keys and message_consumer is None:142 raise ValueError('消息触发定位必须有实际消费者')143 if message_keys:144 message_frame_ids = tuple(sorted(set(message_frame_ids or ()) | {key[1] for key in message_keys}))145 if message_consumer is not None and not callable(message_consumer):146 raise TypeError('message_consumer需要同步可调用对象')147 if not batch_read and (len(spec.input_paths) > 1 or spec.decode_policy != 'legacy' or cycle_fields or metadata_fields):148 raise ValueError('连续文件和新解码政策需要批次读取')149 from ..legacy.dbc_cycles import require_cycle_fields150 require_cycle_fields(cycle_fields,spec.channel)151 require_cycle_fields(metadata_fields,spec.channel)152 self._metadata_fields = metadata_fields153 self._cycle_fields = cycle_fields154 self._input_fingerprints_requested = input_fingerprints155 self._history_integrity=history_integrity156 require_identifier(session_id, 'can.session_id')157 from ..legacy.records import LegacyCANField158 if type(extra_fields) is not tuple or len(extra_fields)>4096 or any(type(f) is not LegacyCANField or f.channel!=spec.channel for f in extra_fields):159 raise ValueError('重建字段需要同通道有限LegacyCANField tuple')160 if any(item.source != spec.name for item in signals) or len({item.name for item in signals}) != len(signals):161 raise ValueError('CAN信号来源须匹配且别名唯一')162 self.budget = self._budget = budget163 self.spec, self.session_id, self.signals, self.backend = spec, session_id, signals, backend164 self._capture_message_updates = capture_message_updates165 if input_check not in ('auto', 'full') or time_scope not in ('none', 'selected', 'source'):166 raise ValueError('CAN检查范围或时间范围非法')167 if (input_check == 'full' or spec.decode_scope == 'all' or retain_raw_frames or capture_message_updates) and time_scope != 'source':168 raise ValueError('完整诊断/原帧/消息历史必须保全来源时间')169 self.input_check, self.time_scope = input_check, time_scope170 self._batch_read, self._retain_raw_frames = batch_read, retain_raw_frames171 self._message_frame_ids = message_frame_ids172 # 策略与固定集合只在创建时决定；all不构造集合或增加声明费用。173 self._decode_scope, self._decode_frame_ids = spec.decode_scope, ()174 # 当前消息消费者只接收共同update，不借用read内部的临时原帧槽。175 self._update_consumer = message_consumer176 if frame_observer is not None and not callable(frame_observer):177 raise TypeError('frame_observer需要可调用对象')178 self._frame_observer = frame_observer179 self._update_projection = None180 self.work_dir = self._description = self._definitions = self._fields = None181 self._requests = self._uninstalled_request = None182 self._fixed_cost = self._held_cost = 0183 self.closed = self._closing = False184 try:185 # 仅计静态有限树；解析field JSON/复制描述/创建登记前已覆盖所需空间。186 cost = 65536 + 32 * _metadata_bytes((spec.name, str(spec.path) if len(spec.input_paths)==1 else tuple(str(p) for p in spec.input_paths), str(spec.dbc_path),187 spec.channel, spec.timebase, spec.unit, spec.offset_ns, session_id, signals, str(work_dir)))188 if cycle_fields or metadata_fields:189 cost += 16384 + _metadata_bytes((cycle_fields,metadata_fields))*32190 _reserve_owned_ram(self, cost, '_fixed_cost')191 if extra_fields:192 _reserve_owned_ram(self,65536+64*_metadata_bytes(extra_fields),'_fixed_cost')193 if self._decode_scope == 'requested':194 if not batch_read:195 raise ValueError('requested来源需要批次读取，batch_read不能为False')196 # 每输入ID覆盖tuple/set/排序及十进制诊断文本；每别名另覆盖scope定位副本。197 # 先付保守上界再解析和创建集合，不按去重后的较小数额追补。198 _reserve_owned_ram(self, 8192 + 256 * (len(signals) + len(message_frame_ids or ()))199 + 512 * len(signals), '_fixed_cost')200 if message_keys:201 _reserve_owned_ram(self, 8192 + 2048 * len(message_keys), '_fixed_cost')202 parsed = tuple((item, can_field_from_key(item.field)) for item in signals)203 if any(field.channel != spec.channel for _, field in parsed):204 raise ValueError('CAN字段通道必须等于单声明通道')205 self._definitions = parsed206 if message_consumer is not None:207 from ..sources.can_updates import CANUpdateProjection208 self._update_projection = CANUpdateProjection(spec.name, session_id, tuple(field for _, field in parsed), message_keys=message_keys,clock_fields=metadata_fields)209 # 同一定位多别名只在声明元数据去重；默认全解码保持，requested固定集合210 # 由下一分支建立。这里不建立DBC缓存，也不共享目标帧的解码调用。211 self._fields = tuple(dict.fromkeys((*[field for _,field in parsed],*extra_fields)))212 if self._decode_scope == 'requested':213 self._decode_frame_ids = tuple(sorted({field.frame_id for field in self._fields}214 | set(message_frame_ids or ())))215 if len(self._decode_frame_ids) > 4096:216 raise ValueError('requested最多声明4096个不同报文ID')217 self._requests = set()218 path = Path(work_dir).resolve()219 path.mkdir(parents=True, exist_ok=False)220 self.work_dir = path221 locator = {'path': str(spec.path), 'dbc_path': str(spec.dbc_path), 'channel': spec.channel,222 'legacy_profile': spec.legacy_profile, 'time_unit': spec.unit, 'offset_ns': spec.offset_ns,223 'source_timebase_sha256': source_timebase_identity(spec.timebase, spec.offset_ns)}224 if len(spec.input_paths) > 1:225 locator.pop('path')226 locator['paths'] = tuple(str(p) for p in spec.input_paths)227 if spec.decode_policy != 'legacy':228 locator['decode_policy'] = spec.decode_policy229 diagnostics = (('原yield序号/完整原帧/原时间/同次DBC元数据' if retain_raw_frames else230 '全部原yield观测/所需原序号与原时间/同次DBC元数据；未保存原帧副本'),)231 if input_check == 'auto' and time_scope != 'source':232 diagnostics = (('原yield计数/所需原序号与值；未采集时间' if time_scope == 'none' else233 '原yield计数/所需原序号、值与时间；未采集非目标帧时间') + '/同次DBC元数据；未保存原帧副本',)234 if self._decode_scope == 'requested':235 locator['decode_scope'] = 'requested'236 # 即使没有signal字段，来源级描述也必须绑定完整固定集合而非PULL子投影。237 diagnostics += ('decode_scope=requested;decode_frame_ids='238 + ','.join(str(value) for value in self._decode_frame_ids),)239 if input_check == 'auto':240 locator.update(input_check=input_check, time_scope=time_scope, check_policy_version='demand-check-v1')241 diagnostics += ('按需检查：time_scope=' + time_scope + '；原生非目标帧仅保留本次要求的ID/时间处理；非原生输入完整检查回退',)242 self._description = SourceDescription(spec.name, describe_fields(spec.name, signals, locator),243 spec.timebase.timebase_id, None, None, SourceCapabilities(True, True, True, False,244 diagnostics,245 (('同格式原序BLF/单声明通道；原空dict不能区分解码失败原因' if len(spec.input_paths)>1246 else '仅单BLF/单声明通道；原空dict不能区分解码失败原因'),247 '缺DBC单位保持未核实；原始片段仅BLF且由run阶段写出；旧进程RSS未测')),248 legacy_profile=spec.legacy_profile, metadata_status='declared')249 except BaseException as exc:250 _cleanup_owner(self, exc, 'can_source_cleanup_owner', 'CAN来源构造')251 raise252253 @property254 def capture_message_updates(self):255 """读取原配置的消息捕获开关，不启动来源或开放完成证明。256257 参数: 无。返回: 原生bool，关闭adapter后仍可核原开关。258 异常: 无；只读属性不允许调用方改写活read的捕获来源。259 """260 return self._capture_message_updates261262 @property263 def batch_read(self):264 """读取本adapter冻结的批协议选择。参数: 无。返回: bool。异常: 无。"""265 return self._batch_read266267 @property268 def retain_raw_frames(self):269 """读取原始帧保存选择。参数: 无。返回: bool。异常: 无，不随关闭改变。"""270 return self._retain_raw_frames271272 @property273 def message_frame_ids(self):274 """读取实际消息保留ID。参数: 无。返回: tuple或None。异常: 无。"""275 return self._message_frame_ids276277 @property278 def decode_scope(self):279 """读取创建时的解码方式。参数: 无。返回: all/requested。异常: 无。"""280 return self._decode_scope281282 @property283 def decode_frame_ids(self):284 """借用已付费的固定解码集合。参数: 无。返回: tuple，close后为空。异常: 无。"""285 return self._decode_frame_ids286287 def _ensure_open(self):288 """拒绝关闭中/已关闭。参数: 无。返回: None。异常: SourceClosedError。289 """290 if self.closed or self._closing:291 raise SourceClosedError('CAN来源已关闭或正在清理')292293 def describe(self):294 """返回已付费静态描述，不读BLF/DBC或spawn。参数: 无。返回: SourceDescription。295 异常: SourceClosedError为关闭中/已关闭。296 """297 self._ensure_open()298 return self._description299300 def iter_batches(self, request):301 """创建独立观测read，首次next才实际输入。参数: request为原ReadRequest。302 返回: CANObservationRead。异常: 原配置/预约/登记错误，持续失败保最高真实owner。303 """304 self._ensure_open()305 if self._uninstalled_request is not None:306 raise SourceClosedError('CAN来源有未返回read待清理')307 try:308 if self.batch_read:309 from .can_batch_reader import CANBatchObservationRead310 read = CANBatchObservationRead(self, request)311 else:312 read = CANObservationRead(self, request)313 self._requests.add(read)314 self._uninstalled_request = None315 return read316 except BaseException as exc:317 read = self._uninstalled_request318 if read is not None:319 read._failed = True320 _cleanup_owner(read, exc, 'can_observation_read_cleanup_owner', 'CAN请求登记')321 if not read.closed:322 exc.can_source_cleanup_owner = self323 raise324325 def iter_batches_with_observation(self, request):326 """使用同一独立read合同。参数: request为原请求。返回: CANObservationRead。327 异常: 与iter_batches相同；不复用adapter全局最后事实或request值map。328 """329 return self.iter_batches(request)330331 def close(self):332 """关闭本来源活read及费用，不关闭已完成脱登记read或其新父raw store。333334 参数: 无。返回: None，可重试。异常: 首个实际关闭/抵销错误，最高can_source_cleanup_owner。335 """336 if self.closed:337 return338 self._closing = True339 failure = None340 for read in (() if self._uninstalled_request is None else (self._uninstalled_request,)) + tuple(self._requests or ()):341 try:342 read.close()343 except BaseException as exc:344 if failure is None:345 failure = exc346 else:347 failure.add_note('CAN其他read关闭附加失败: ' + type(exc).__name__)348 if self._uninstalled_request is None and not self._requests:349 self._description = self._definitions = self._fields = None350 self._decode_frame_ids = ()351 self._update_projection = self._update_consumer = None352 self._frame_observer = None353 self.spec, self.signals = None, ()354 try:355 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')356 except BaseException as exc:357 if failure is None:358 failure = exc359 if self._held_cost == 0:360 self.closed = True361 if failure is not None:362 failure.can_source_cleanup_owner = self363 raise failure364365366class CANObservationRead:367 """拥有同次CAN读取进程进程、原帧/可选消息仓及当前帧；事实活到自身close，移交仓归新父。"""368369 def __init__(self, owner, request):370 """先安装未返回槽并预付固定树，不实际打开来源。371372 参数: owner为预算CAN来源，request为同会话原ReadRequest。返回: 独立read。373 异常: 原请求/预算/取消；构造未返回资源由本read最高恢复owner负责。374 """375 if not isinstance(owner, BudgetedLegacyCANSourceAdapter) or not isinstance(request, ReadRequest):376 raise TypeError('CAN read需要实际来源及ReadRequest')377 owner._ensure_open()378 if request.session_id != owner.session_id or any(name not in {item.name for item in owner.signals} for name in request.signal_names):379 raise ValueError('CAN请求会话或投影不匹配')380 self._owner, self._budget, self._request, self._spec = owner, owner.budget, request, owner.spec381 self._process = self._raw_store = self._message_store = self._record = self._current = self._validator = None382 self._capture_message_updates = owner.capture_message_updates383 self._message_store_taken = False384 self._selected = self._runtime_metadata = self._observation = self._identity = self._work_path = None385 self._binding_projection = None386 self._sample_pack_costs = ()387 self._fixed_cost = self._metadata_cost = self._record_cost = self._pack_cost = 0388 self._pending_cost = self._validation_cost = self._held_cost = self._update_cost = 0389 self._first_ns = self._last_ns = self._conversion = self._mapped = None390 self._row_count = self._alias_index = 0391 self._retained_record_count = self._full_check_count = 0392 self._started = self._eof_delivered = self._finished = self._taken = False393 self._closed = self._closing = self._failed = False394 # 只有requested直接累计且无普通样本时需要输入字节证明；all保原费用和扫描证据。395 self._input_fingerprints_enabled = (owner._input_fingerprints_requested and owner.decode_scope == 'requested'396 and owner._update_consumer is not None and not request.signal_names)397 self._input_timestamp_sha = self._input_records_sha = self._input_fingerprints = None398 self._input_frame_count = self._input_retained_count = 0399 owner._uninstalled_request = self400 try:401 cost = 65536 + 16 * _metadata_bytes((request, owner._description, owner._definitions))402 # 默认费用保持；开启时先覆盖本层新子路径及固定责任槽，仓本身另按原公式付款。403 if self._capture_message_updates:404 cost += 4096 + 32 * len(str(owner.work_dir))405 if self._input_fingerprints_enabled:406 cost += 8192 # 两个SHA状态、计数、memoryview与最终64字符摘要独立先付费。407 _reserve_owned_ram(self, cost, '_fixed_cost')408 if self._input_fingerprints_enabled:409 if not owner.batch_read:410 raise ValueError('CAN真实输入指纹需要批次读取')411 self._input_timestamp_sha = hashlib.sha256()412 self._input_records_sha = hashlib.sha256()413 self._identity = uuid4().hex414 self._work_path = owner.work_dir / self._identity415 self._selected = tuple(pair for pair in owner._definitions if pair[0].name in request.signal_names)416 self._validator = BatchValidator(owner._description, request)417 except BaseException as exc:418 self._failed = True419 _cleanup_owner(self, exc, 'can_observation_read_cleanup_owner', 'CAN read构造')420 raise421422 @property423 def capture_message_updates(self):424 """读取创建本read时的原捕获开关，不因adapter关闭而失去来源。425426 参数: 无。返回: 原生bool快照。异常: 无；该标志本身不证明已经完成输入。427 """428 return self._capture_message_updates429430 @property431 def identity(self):432 """读取本次独立UUID。参数: 无。返回: str。异常: 无，不按request值复用。"""433 return self._identity434435 @property436 def request(self):437 """读取创建时原请求。参数: 无。返回: 原ReadRequest。异常: 关闭时ResultClosedError。"""438 self._require_open()439 return self._request440441 @property442 def closed(self):443 """读取真实关闭事实。参数: 无。返回: bool。异常: 无。"""444 return self._closed445446 @property447 def failed(self):448 """读取失败标记。参数: 无。返回: bool。异常: 无，不代表已清理。"""449 return self._failed450451 @property452 def finished(self):453 """读取真实完整StopIteration事实。参数: 无。返回: bool。异常: 无。"""454 return self._finished455456 @property457 def exhausted(self):458 """提供完成别名。参数: 无。返回: bool。异常: 无，EOF批交付仍为False。"""459 return self._finished460461 def _require_open(self):462 """拒绝已关闭/失败/半关闭。参数: 无。返回: None。异常: ResultClosedError。"""463 if self._closed or self._closing or self._failed:464 raise ResultClosedError('CAN read已关闭、失败或正在清理')465466 @property467 def observation(self):468 """仅完整完成后借读事实。参数: 无。返回: CANObservation。异常: ResultClosedError。"""469 self._require_open()470 if not self._finished:471 raise ResultClosedError('CAN read尚未真正完整耗尽')472 return self._observation473474 @property475 def observed_bounds(self):476 """读取已经看到的首末时间供在线规则定界，不表示输入已完整结束。477478 参数: 无。返回: 两个必要整数或None；none时间范围不补造坐标。异常: 读取器已关闭/失败。479 """480 self._require_open()481 return self._first_ns,self._last_ns482483 @property484 def input_fingerprints(self):485 """借读实际输入的两段字节指纹，不把它解释成普通信号样本摘要。486487 参数: 无。返回: (时间SHA256,记录SHA256,全原帧数,保留记录数)的tuple。488 异常: ResultClosedError为未启用、未完成、失败或已关闭；不返回半成品hash。489 指纹只活到本read关闭，接收方要继续保留须在自己的预算下复制/持有。490 """491 self._require_open()492 if not self._finished or self._input_fingerprints is None:493 raise ResultClosedError('CAN真实输入指纹尚未完成或没有启用')494 return self._input_fingerprints495496 def _observe_validated_packet(self, process, payload, batch):497 """在严格批解包后借用原字节，按原顺序增量计算时间和所需记录指纹。498499 参数: process必须是本read的实际子进程read；payload/batch为其刚核完的借用包。500 返回: None。异常: SourceReadError为归属/原序不符；原hash/取消错误保持且标记失败。501 时间区是头后每原帧8字节，记录区保每条长度及原JSON；不包含随分批改变的包头。502 任一更新中断后本read永久失败，不能重试同一批或发布部分指纹。503 """504 view = time_view = record_view = None505 try:506 self._require_open()507 if (not self._input_fingerprints_enabled or self._input_timestamp_sha is None508 or self._input_records_sha is None or self._input_fingerprints is not None509 or process is not self._process or payload is not process._pending_reply510 or batch.start_sequence != self._input_frame_count):511 raise SourceReadError('CAN输入指纹须来自同次已核批且按全原帧序号连续')512 view = memoryview(payload)513 from ..legacy.batch_wire import record_payload_offset, FILE_HEADER514 end = record_payload_offset(batch) + (FILE_HEADER.size if len(self._spec.input_paths)>1 else 0)515 record_view = view[end:]516 if self._owner.time_scope == 'source':517 # 全来源模式才生成全部原时间SHA；新头长度由记录区与帧数确定。518 time_view = view[end - 8 * batch.frame_count:end]519 self._input_timestamp_sha.update(time_view)520 self._input_records_sha.update(record_view)521 self._input_frame_count += batch.frame_count522 self._input_retained_count += len(batch.records)523 except BaseException as exc:524 self._failed = True525 exc.can_observation_read_cleanup_owner = self526 raise527 finally:528 # 即使调用方保存错误栈，也不由本层的视图继续保活整包；不复制任何原数据。529 for borrowed in (time_view, record_view, view):530 if borrowed is not None:531 borrowed.release()532 process = payload = batch = borrowed = view = time_view = record_view = None533534 @property535 def runtime_metadata(self):536 """借读同次READY原冻结映射，None/空unit仍未核实。537538 参数: 无。返回: 保operation/version/profile等原键的映射。异常: 未READY/关闭ResultClosedError。539 """540 self._require_open()541 if self._runtime_metadata is None:542 raise ResultClosedError('CAN read尚未取得同次READY')543 return self._runtime_metadata544545 def prepare_metadata(self):546 """初始化同次DBC并返回周期/单位事实，不读取首帧。547548 参数: 无。返回: 同一冻结READY。异常: 原启动错误，沿本read清理链处理。549 重复调用复用同次初始化；不把初始化成功当作输入EOF。550 """551 self._require_open()552 try:553 if self._runtime_metadata is None:554 self._start()555 return self._runtime_metadata556 except BaseException as exc:557 self._fail(exc)558 raise559560 def _adopt(self, exc):561 """接管未返回的实际CAN读取进程/raw/消息仓子owner。参数: exc为原错误。返回: None。异常: 无预期异常。"""562 if self._process is None:563 self._process = getattr(exc, 'can_process_read_cleanup_owner', None)564 if self._raw_store is None and not self._taken:565 self._raw_store = getattr(exc, 'raw_frame_store_cleanup_owner', None)566 # 消息仓构造可能已经取得真实FD/费用却未返回；只接尚未移交的本层子责任。567 if self._capture_message_updates and self._message_store is None and not self._message_store_taken:568 self._message_store = getattr(exc, 'decoded_message_store_cleanup_owner', None)569570 def _fail(self, exc):571 """保原错/取消并关闭同一最高read。参数: exc。返回: None。异常: 清理副错只note。"""572 self._failed = True573 self._adopt(exc)574 exc.can_observation_read_cleanup_owner = self575 _cleanup_owner(self, exc, 'can_observation_read_cleanup_owner', 'CAN观测read')576577 def _start(self):578 """原start仅INIT，复制同次READY并核全部声明单位后才允许第一PULL。579580 参数: 无。返回: None。异常: 原CAN读取进程/单位/预算/原始暂存错误；上层保真实子槽。581 """582 request = LegacyCANRequest(self._spec.path, self._spec.dbc_path, self._owner._fields)583 if self._owner.batch_read:584 from ..legacy.batch_read import LegacyCANBatchRead585 ids = None if (self._owner.retain_raw_frames or self._capture_message_updates586 and self._owner.message_frame_ids is None) else tuple(sorted(587 {field.frame_id for _, field in self._owner._definitions}588 | set(self._owner.message_frame_ids or ())))589 # 传输预付不能吃光下一步装箱/验证/磁盘编码需要的空间；这里只留余量，590 # 费用仍由实际消费owner在使用前逐笔入账，_record_cost已在PULL前支付。591 # 同一次read的定位/路径/定义不可变；算一次每别名装箱费，后续样本592 # 仍逐笔预付同一金额，不为相同的元数据反复遍历整个定义树。593 self._sample_pack_costs = tuple(32768 + 32 * _metadata_bytes((definition, field,594 max((str(p) for p in self._spec.input_paths),key=_metadata_bytes), self._spec.legacy_profile, self._request.session_id,595 self._spec.name)) for definition, field in self._selected)596 per_sample = 2 * max(self._sample_pack_costs, default=32768)597 scratch = (65536 + 64 * MAX_DECODED_MESSAGE_ROW_BYTES598 if self._capture_message_updates or self._owner.retain_raw_frames else 0)599 sample_bound = self._request.batch_rows if self._selected else 0600 headroom = 131072 + sample_bound * per_sample + scratch601 # 当前批交给分析层前，只在上述消费空间之外预付并授权下一批。602 # 当前处理失败/取消可能多读至多一批；预算不足时批read自动同步回退。603 source_options = {'cycle_fields':self._owner._cycle_fields} if self._owner._cycle_fields else {}604 if self._owner._metadata_fields:605 source_options['metadata_fields'] = self._owner._metadata_fields606 if len(self._spec.input_paths) > 1:607 source_options['input_paths'] = self._spec.input_paths608 if self._spec.decode_policy != 'legacy':609 source_options['decode_policy'] = self._spec.decode_policy610 if self._owner.decode_scope == 'requested':611 self._process = LegacyCANBatchRead(request, self._budget, frame_ids=ids,612 consumer_headroom_bytes=headroom, prefetch=True, decode_scope='requested',613 input_check=self._owner.input_check, time_scope=self._owner.time_scope,614 decode_frame_ids=self._owner.decode_frame_ids, **source_options)615 else:616 # 旧调用连新增默认关键字也不传，保原构造与故障注入入口。617 self._process = LegacyCANBatchRead(request, self._budget, frame_ids=ids,618 consumer_headroom_bytes=headroom, prefetch=True, **source_options)619 else:620 self._process = LegacyCANProcessRead(request, self._budget)621 if self._input_fingerprints_enabled:622 self._process._validated_packet_consumer = self._observe_validated_packet623 self._process.start()624 ready = self._process.metadata625 _reserve_owned_ram(self, 4096 + 16 * _metadata_bytes(ready), '_metadata_cost')626 self._runtime_metadata = freeze_metadata(ready)627 from ..sources.can_updates import CANUpdateProjection628 projection = self._owner._update_projection or CANUpdateProjection(self._spec.name,629 self._request.session_id, tuple(field for _, field in self._owner._definitions))630 self._binding_projection = projection.for_read(ready['fields'])631 if len(self._spec.input_paths)>1 or self._spec.decode_policy != 'legacy' or self._owner._cycle_fields or self._owner._metadata_fields:632 if (ready.get('version') != 5633 or ready.get('input_paths') != tuple(str(p) for p in self._spec.input_paths)634 or ready.get('decode_policy') != self._spec.decode_policy):635 raise SourceReadError('同次CAN READY与连续文件/解码政策不符')636 ready = dict(ready,version=ready['base_version'])637 # 同次READY先与实际adapter交叉核；版本/集合不能从缺项猜成默认成功。638 if self._owner.decode_scope == 'requested':639 fixed = ready.get('decode_frame_ids')640 expected_version = 4 if self._owner.input_check == 'auto' else 3641 if (type(ready.get('version')) is not int or ready['version'] != expected_version642 or type(ready.get('decode_scope')) is not str or ready['decode_scope'] != 'requested'643 or type(fixed) is not tuple or any(type(value) is not int for value in fixed)644 or fixed != self._owner.decode_frame_ids645 or type(ready.get('decoded_values_policy')) is not str646 or ready['decoded_values_policy'] != REQUESTED_VALUES_POLICY):647 raise SourceReadError('同次CAN READY与来源固定requested策略不符')648 if expected_version == 4 and (ready.get('input_check') != self._owner.input_check or ready.get('time_scope') != self._owner.time_scope):649 raise SourceReadError('同次CAN READY检查/时间范围与来源不同')650 elif (type(ready.get('version')) is not int or ready['version'] != 1651 or 'decode_scope' in ready or 'decode_frame_ids' in ready652 or 'decoded_values_policy' in ready):653 raise SourceReadError('全解码来源必须保原v1 READY且不带按需策略')654 for definition, field in self._owner._definitions:655 unit = next(item['unit'] for item in ready['fields'] if item['locator'] == field_to_dict(field))656 if unit not in (None, '') and unit != definition.input_unit:657 raise SourceReadError('同次DBC非空单位不等于声明input_unit: ' + definition.name)658 if self._owner.retain_raw_frames:659 self._raw_store = RawFrameStore(self._budget, self._work_path, self._spec.name, integrity=self._owner._history_integrity)660 # 原raw目录保持不变；可选子目录只在READY/单位核对完成后、首PULL之前创建。661 if self._capture_message_updates:662 self._message_store = DecodedMessageStore(self._budget, self._work_path / 'messages',663 self._spec.name, self._request.session_id, decoder_profile=self._runtime_metadata['profile'],664 read_identity=self._identity, sparse_sequences=self._owner.batch_read, integrity=self._owner._history_integrity)665 self._started = True666667 def _accept(self, batch):668 """先预约validator候选及同刻去重集合，再原子接收并留下实际保留费用。669670 参数: batch为有限SignalBatch。返回: 原batch。异常: 原协议/预算/取消，由上层poison。671 """672 _reserve_owned_ram(self, 4096 + 2 * _pending_bytes(self._validator)673 + 1024 * batch.sample_count, '_validation_cost')674 self._validator.accept(batch)675 retained = _pending_bytes(self._validator)676 _release_owned_ram(self, self._pending_cost, '_pending_cost')677 self._validation_cost -= retained678 self._pending_cost += retained679 _release_owned_ram(self, self._validation_cost, '_validation_cost')680 self._current = batch681 return batch682683 def _release_current(self):684 """下一步前停止持有原batch再归还其装箱空间。参数: 无。返回: None。异常: 原抵销/取消。"""685 self._current = None686 _release_owned_ram(self, self._pack_cost, '_pack_cost')687688 def _new_record(self):689 """原帧进入前先付本层记录/引用空间；全部yield先暂存再作分析层列选择。690691 参数: 无。返回: True有新原帧，False为CAN读取进程真实EOF。异常: 原输入/时间/暂存错误。692 """693 self._record = None694 _release_owned_ram(self, self._record_cost, '_record_cost')695 _reserve_owned_ram(self, 4259840, '_record_cost')696 record = conversion = None697 try:698 try:699 record = next(self._process)700 except StopIteration:701 _release_owned_ram(self, self._record_cost, '_record_cost')702 if not self._process.finished:703 raise SourceReadError('CAN进程没有完整EOF')704 if self._raw_store is not None:705 self._raw_store.seal()706 # 两仓仅在同一次真实CAN读取进程 EOF后封口；任一失败都不能开放finished或移交。707 if self._capture_message_updates:708 self._message_store.seal()709 return False710 conversion = convert_time(record.frame.timestamp, 's')711 mapped = checked_add_ns(conversion.time_ns, self._spec.offset_ns)712 if self._last_ns is not None and mapped < self._last_ns:713 raise SourceReadError('CAN原yield时间回退，禁止重排')714 if self._raw_store is not None:715 self._raw_store.append(record, mapped)716 # 同一个原decode对象恰追加一次，位于列选择之前，不因空投影或拆别名改变次数。717 if self._capture_message_updates:718 self._message_store.append(record, mapped, time_quantized=conversion.quantized)719 self._binding_projection.fields_for(record.frame)720 if self._owner._frame_observer is not None:721 self._owner._frame_observer(record.frame)722 if self._first_ns is None:723 self._first_ns = mapped724 self._last_ns, self._row_count = mapped, self._row_count + 1725 self._record, self._conversion, self._mapped, self._alias_index = record, conversion, mapped, 0726 self._deliver_message(record, mapped)727 return True728 finally:729 # 原异常可被保留；只断本函数临时大record/转换，不清原输入或其他模块TB。730 record = conversion = None731732 def _deliver_message(self, record, time_ns):733 """仅有明确消费者时构造共同消息，默认读取不增加逐帧对象。734735 参数: record为同次原解码帧；time_ns为已检查的时间或未采集None。736 返回: None。异常: 投影、预算或消费者错误由read统一标记失败并清理。737 """738 if self._owner._update_consumer is not None:739 from .message_delivery import deliver_message740 deliver_message(self, self._binding_projection, self._owner._update_consumer,741 record.frame, record.sequence, record.values, time_ns)742743 def _batch(self):744 """为当前原帧最多batch_rows个匹配别名装箱，缺key产生明确MISSING更新。745746 参数: 无。返回: 已校验SignalBatch。异常: 原refs/归一化/预约/协议错误。747 """748 _reserve_owned_ram(self, 65536, '_pack_cost')749 columns = {}750 while self._alias_index < len(self._selected):751 definition, field = self._selected[self._alias_index]752 self._alias_index += 1753 if field not in self._binding_projection.fields_for(self._record.frame):754 continue755 # 完整定位/定义及单样本数组均先付；不按原数值相同而合并独立别名。756 _reserve_owned_ram(self, 16384 + 32 * _metadata_bytes((definition, field, str(self._spec.path),757 self._spec.legacy_profile, self._request.session_id, self._spec.name)), '_pack_cost')758 ref = CANSourceRef(self._request.session_id, self._spec.name, definition.field,759 str(self._spec.path), self._spec.legacy_profile, field.channel, field.signal,760 field.frame_id, self._record.frame.is_extended_id, self._record.frame.is_fd, self._record.sequence,761 self._record.frame.timestamp, self._conversion.quantized)762 value = self._record.values.get(field.signal)763 columns[definition.name] = pack_decoded(definition,764 ((self._mapped, self._record.sequence, value),), (ref,), self._owner.backend)765 if len(columns) == self._request.batch_rows:766 break767 watermark = None if self._mapped == -(1 << 63) else self._mapped - 1768 return self._accept(SignalBatch(self._spec.name, self._request.session_id, columns,769 watermark, False, None))770771 def __iter__(self):772 """保持本次原序位置。参数: 无。返回: self。异常: 无，不启动读取。"""773 return self774775 def __next__(self):776 """逐次交付当前原帧分片或独立空EOF批，真正下一次耗尽才开放完成事实。777778 参数: 无。返回: SignalBatch。异常: 真StopIteration/原来源/预算/取消，最高read可恢复。779 """780 if self._finished:781 raise StopIteration782 self._require_open()783 try:784 self._release_current()785 if not self._started:786 self._start()787 if self._eof_delivered:788 self._finish()789 elif self._record is not None and self._alias_index < len(self._selected):790 return self._batch()791 elif self._new_record():792 return self._batch()793 else:794 _reserve_owned_ram(self, 65536, '_pack_cost')795 batch = SignalBatch(self._spec.name, self._request.session_id, {}, self._last_ns,796 True, self._last_ns)797 self._accept(batch)798 self._eof_delivered = True799 return batch800 except BaseException as exc:801 self._fail(exc)802 raise803 raise StopIteration804805 def _close_child(self, slot):806 """真实子owner完成关闭后才清槽。参数: slot为实际子槽。返回: None。异常: 原close错误。"""807 child = getattr(self, slot)808 if child is not None:809 try:810 child.close()811 finally:812 if child.closed:813 setattr(self, slot, None)814815 def _finish(self):816 """确认本read全部完成条件后开放事实，并解除adapter活登记。817818 参数: 无。返回: None。异常: 原关闭/协议/登记错误，不能伪造完成。819 """820 if (not self._validator.ended or not self._process.finished821 or self._owner.retain_raw_frames and (self._raw_store is None822 or self._raw_store.state != 'sealed' or self._raw_store.closed)):823 raise SourceReadError('CAN输入尚未满足真实完成条件')824 if self._capture_message_updates:825 # 完成事实必须同时属于本read的原消息仓，不能只凭原帧仓已封口推断。826 store = self._message_store827 if (store is None or store.closed or store.state != 'sealed'828 or (store.count != self._row_count if not store.sparse_sequences else829 store.count > self._row_count or store._last_sequence >= self._row_count)830 or store.source_name != self._spec.name or store.session_id != self._request.session_id831 or store.read_identity != self._identity832 or store.decoder_profile != self._runtime_metadata['profile']):833 raise SourceReadError('同次CAN消息仓尚未满足原对象及完整封口事实')834 self._close_child('_process')835 self._record = None836 _release_owned_ram(self, self._record_cost, '_record_cost')837 if self._input_fingerprints_enabled:838 if (self._input_frame_count != self._row_count839 or self._input_timestamp_sha is None or self._input_records_sha is None):840 raise SourceReadError('CAN真实输入指纹的原帧数与完成观测不符')841 self._input_fingerprints = (self._input_timestamp_sha.hexdigest() if self._owner.time_scope == 'source' else None,842 self._input_records_sha.hexdigest(), self._input_frame_count, self._input_retained_count)843 self._input_timestamp_sha = self._input_records_sha = None844 self._observation = CANObservation(self._spec.name, self._request.session_id, self._request,845 self._identity, self._first_ns, self._last_ns, self._row_count, True,846 self._owner.time_scope, self._retained_record_count, self._full_check_count)847 self._owner._requests.discard(self)848 self._finished = True849850 def take_raw_store(self, receiver, slot='_raw_store'):851 """仅完成后一次移交真实sealed store，本read后续不再关闭新父资源。852853 参数: receiver为同一_budget的可信内部owner；slot为其已初始化为None的实际接收槽。854 返回: 原RawFrameStore，先装父槽再撤本子槽。异常: 状态/预算/槽错误或原安装取消；855 安装后取消仍由接收父实际槽可达，不在返回边界丢失真实费用。856 """857 self._require_open()858 if not self._finished or self._taken or self._raw_store is None:859 raise ResultClosedError('原始store未完成或已移交')860 result = self._raw_store861 if (getattr(receiver, '_budget', None) is not self._budget or type(slot) is not str862 or not slot.startswith('_') or getattr(receiver, slot) is not None):863 raise ValueError('原始store移交须同账本且接收实际槽为空')864 try:865 # 父赋值先发生；即使setter在真实安装后取消，也至少有最高父实际责任。866 setattr(receiver, slot, result)867 except BaseException as exc:868 exc.can_observation_read_cleanup_owner = self869 raise870 finally:871 if getattr(receiver, slot) is result:872 self._raw_store, self._taken = None, True873 return result874875 def take_message_store(self, receiver, slot='_message_store'):876 """完成后一次移交同次消息仓，父实际安装成功才解除本read责任。877878 参数: receiver为同一_budget的可信owner；slot为其已存在且为空的实际槽。879 返回: 原DecodedMessageStore对象，不重建文件或复制decoded记录。880 异常: ResultClosedError为未开启、未完成或已移交；ValueError为预算/槽不符；881 父安装前后取消保原异常和can_observation_read_cleanup_owner。882 """883 self._require_open()884 if (not self._finished or not self._capture_message_updates or self._message_store_taken885 or self._message_store is None):886 raise ResultClosedError('消息store未开启、未完成或已移交')887 store = self._message_store888 if (getattr(receiver, '_budget', None) is not self._budget or type(slot) is not str889 or not slot.startswith('_') or not hasattr(receiver, slot) or getattr(receiver, slot) is not None):890 raise ValueError('消息store移交须同账本且接收实际槽为空')891 try:892 # setter可能在实际安装后取消；finally按真实父槽判断，不能先清子再return。893 setattr(receiver, slot, store)894 except BaseException as exc:895 exc.can_observation_read_cleanup_owner = self896 raise897 finally:898 if getattr(receiver, slot, None) is store:899 self._message_store, self._message_store_taken = None, True900 return store901902 def close(self):903 """只关闭仍自有的输入/raw/消息store和所有实际费用，可持续重试。904905 参数: 无。返回: None。异常: 首个子关闭/费用取消；can_observation_read_cleanup_owner保本对象。906 """907 if self._closed:908 return909 self._closing = True910 failure = None911 for slot in ('_process', '_raw_store', '_message_store'):912 try:913 self._close_child(slot)914 except BaseException as exc:915 if failure is None:916 failure = exc917 else:918 failure.add_note('CAN其他子资源关闭附加失败: ' + type(exc).__name__)919 if self._process is None and self._raw_store is None and self._message_store is None:920 self._record = self._current = self._runtime_metadata = self._observation = None921 self._validator = self._selected = None922 self._binding_projection = None923 self._sample_pack_costs = ()924 self._input_timestamp_sha = self._input_records_sha = self._input_fingerprints = None925 for slot in ('_update_cost', '_pack_cost', '_record_cost', '_validation_cost', '_pending_cost', '_metadata_cost', '_fixed_cost'):926 try:927 _release_owned_ram(self, getattr(self, slot), slot)928 except BaseException as exc:929 if failure is None:930 failure = exc931 else:932 failure.add_note('CAN费用抵销附加失败: ' + type(exc).__name__)933 if self._held_cost == 0:934 try:935 self._owner._requests.discard(self)936 if self._owner._uninstalled_request is self:937 self._owner._uninstalled_request = None938 self._closed = True939 except BaseException as exc:940 if failure is None:941 failure = exc942 else:943 failure.add_note('CAN解除read登记附加失败: ' + type(exc).__name__)944 if failure is not None:945 failure.can_observation_read_cleanup_owner = self946 raise failure947