chronaxis.fast.stream
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""流式读取单文件并产出当前帧贡献，不维护跨帧信号或最终累计。23输入为已编译声明和DBC；输出(time_ns, 原记录序号, 计算项贡献tuple)。4只保存当前帧，文件关闭后才提供最终计数。父会话负责顺序和结果发布。5"""6from pathlib import Path7from time import perf_counter8import can9import cantools1011from ..results.budget import _cleanup_owner12from ..legacy.decode_policy import check_decode_length13from ..timebase.conversion import convert_time, checked_add_ns14from .can_io import _prepare, _UNSUPPORTED_FD151617class FileContributions:18 """持有一个读取器和有限公式组；同帧仅解码一次，计数按计算项隔离。"""1920 def __init__(self, spec, database, cancel=None, budget=None, *, _programs_checked=False):21 """打开单文件。参数: spec编译声明，database已载DBC，cancel可选事件，budget账本。22 返回: 迭代器。异常: 配置、文件或DBC错误；已取得的读取器在后续构造失败时清理。23 ASC/BLF共用各自文件owner；第三方BLF文件头构造失败也保留实际文件关闭责任。24 _programs_checked只供已完成公式准备的内部调用；普通调用仍检查原公式。25 """26 # 在任何配置准备或实际打开前安装最小关闭字段，半构造对象也可恢复。27 self.reader, self._pending_reader = None, None28 self.closed, self.text = False, False29 try:30 self.spec, self.cancel = spec, cancel31 self.groups, self.metrics = _prepare(spec, database, _programs_checked=_programs_checked)32 self.names = tuple(self.metrics)33 self.ids = {name: i for i, name in enumerate(self.names)}34 self.raw_count = 035 self.targets = [0] * len(self.names)36 self.valid = [0] * len(self.names)37 self.selected = [0] * len(self.names)38 self.kinds = [0] * len(self.names)39 self.compatible = set()40 self.last_time = None41 self.reader = None42 self.finished = False43 self.closed = False44 self.domain = spec.get('domain_ns')45 self.profile = spec.get('_profile', False)46 self.timings = dict(read_seconds=0.0, decode_seconds=0.0, formula_seconds=0.0)47 self.ordered = spec.get('file_merge', 'concatenate') == 'time_order'48 self.channel = spec['channel'] - 149 self.path = Path(spec['path'])50 self.text = self.path.suffix.lower() == '.asc'51 if self.text:52 from ..sources.asc import ASCRead, ASCSpec53 from ..results.budget import ResourceBudget54 self.reader = ASCRead(ASCSpec(self.path, **spec.get('asc_options', {})),55 budget if budget is not None else ResourceBudget(),56 retain_original=False, retain_source_time=False,57 time_selector=self._selected_time)58 elif self.path.suffix.lower() == '.blf':59 from ..sources.blf import BLFRead, BLFSpec60 from ..results.budget import ResourceBudget61 self.reader = BLFRead(BLFSpec(self.path),62 budget if budget is not None else ResourceBudget(), frame_only=True)63 else:64 raise ValueError('fast仅支持ASC/BLF')65 self.iterator = iter(self.reader)66 if self.profile:67 self.iterator = self._measured(self.iterator)68 except BaseException as error:69 # 来源构造器提供实际未关闭对象；不从总账余额猜测应退金额。70 self._pending_reader = getattr(error, 'asc_read_cleanup_owner', None)71 if self._pending_reader is None:72 self._pending_reader = getattr(error, 'blf_read_cleanup_owner', None)73 _cleanup_owner(self, error, 'fast_reader_cleanup_owner', 'fast贡献读取器')74 raise7576 def _measured(self, iterator):77 """仅诊断运行计量解析迭代。参数: 原读取迭代器。返回: 原记录生成器。异常: 原错误。78 正式执行不使用此包装，避免每条原始报文计时开销。79 """80 while True:81 started = perf_counter()82 try:83 record = next(iterator)84 except StopIteration:85 self.timings['read_seconds'] += perf_counter() - started86 return87 self.timings['read_seconds'] += perf_counter() - started88 yield record8990 def _selected_time(self, frame):91 """确定ASC精确时间检查集合。参数: 原帧。返回: bool。异常: 无。92 范围过滤前选择，零贡献和方向不影响集合；无时间消费者不加强检查。93 """94 return ((self.ordered or self.domain is not None) and frame.channel == self.channel95 and not frame.is_error_frame and not frame.is_remote_frame96 and (frame.is_extended_id, frame.is_fd) in self.groups.get(frame.arbitration_id, {}))9798 def __iter__(self):99 """返回自身。参数: 无。返回: 当前迭代器。异常: 无。"""100 return self101102 def __next__(self):103 """跳过无关帧并返回一条目标贡献。参数: 无。返回: 时间/序号/贡献元组。104 异常: 解析、时间、解码、公式错误或取消；EOF在读取资源成功关闭后抛出。105 """106 if self.finished or self.closed:107 raise StopIteration108 try:109 for record in self.iterator:110 index = self.raw_count111 self.raw_count += 1112 if not (index & 4095) and self.cancel is not None and self.cancel.is_set():113 raise InterruptedError('读取贡献时取消')114 frame = record.frame if self.text else record115 candidates = self.groups.get(frame.arbitration_id)116 if (candidates is None or frame.channel != self.channel117 or frame.is_error_frame or frame.is_remote_frame):118 continue119 kind = frame.is_extended_id, frame.is_fd120 group = candidates.get(kind)121 if group is None:122 continue123 timestamp = 0124 if self.ordered or self.domain is not None:125 timestamp = record.time_ns if self.text else checked_add_ns(126 convert_time(frame.timestamp, 's').time_ns, self.spec.get('offset_ns', 0))127 if self.ordered and self.last_time is not None and timestamp < self.last_time:128 raise ValueError('目标报文在文件内部时间回退')129 self.last_time = timestamp130 if group is not _UNSUPPORTED_FD:131 for name, _, _ in group.entries:132 self.targets[self.ids[name]] += 1133 # 不提前break：concatenate的后续记录可能重新落回分析范围。134 if self.domain is not None and not self.domain[0] <= timestamp < self.domain[1]:135 continue136 if group is _UNSUPPORTED_FD:137 raise ValueError('实录为FD但DBC只定义经典CAN')138 if not frame.is_fd and frame.dlc > 8:139 raise ValueError('经典CAN载荷不能超过8字节')140 check_decode_length(frame, group.message.length, 'strict')141 started = perf_counter() if self.profile else 0142 values = group.message.decode(frame.data, decode_choices=False, scaling=True,143 allow_truncated=False, allow_excess=False)144 if self.profile:145 self.timings['decode_seconds'] += perf_counter() - started146 started = perf_counter()147 if group.message.is_fd and not frame.is_fd:148 self.compatible.add(frame.arbitration_id)149 entries = []150 for name, metric, observed in group.entries:151 number = self.ids[name]152 mask = 1 << (int(kind[0]) * 2 + int(kind[1]))153 self.kinds[number] |= mask154 if observed is not None and self.kinds[number].bit_count() > 1:155 raise ValueError(name + '匹配多种实际帧型')156 value, keep = metric.contribution(values)157 self.valid[number] += 1158 self.selected[number] += int(keep)159 entries.append((number, value, keep))160 if self.profile:161 self.timings['formula_seconds'] += perf_counter() - started162 return timestamp, index, tuple(entries)163 self.close()164 self.finished = True165 raise StopIteration166 except (ArithmeticError, ValueError, TypeError, KeyError, cantools.database.DecodeError) as error:167 raise ValueError(f'{self.path.name} 第 {self.raw_count} 条记录：{error}') from error168169 def summary(self):170 """取得关闭后的文件事实。参数: 无。返回: 有限dict。异常: 未成功EOF时拒绝。"""171 if not self.finished or not self.closed:172 raise RuntimeError('尚未成功关闭文件')173 return dict(raw=self.raw_count, targets=self.targets, valid=self.valid,174 selected=self.selected, kinds=self.kinds, compatible=sorted(self.compatible))175176177 def close(self):178 """关闭本读取器及未返回的实际来源子所有者，失败保留原对象供重试。179180 参数: 无。返回: None。异常: 原关闭错误；失败继续保留实际owner供重试。181 """182 if self.closed:183 return184 if self._pending_reader is not None:185 self._pending_reader.close()186 if not self._pending_reader.closed:187 raise RuntimeError('来源构造恢复对象未完成关闭')188 self._pending_reader = None189 if self.reader is not None:190 self.reader.close()191 self.closed = True192