chronaxis.execution.can_batch_reader
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""将有界CAN批次接成按需历史，保留每条原yield的时间及必要记录的原序。23旧读取/解码仍由批进程逐条执行。这里只省去无关帧的重复装箱和历史：全部原时间4用于首末/计数/倒退检查，目标记录保存完整values，信号按样本总数分批。5无BLF导出不建立原帧仓。所有批次、转换、引用和数组在共享预算内预付；不保整文件。6"""78from math import isfinite9from ..legacy.records import LegacyDecodedFrame, LegacyRawFrame10from ..results.spool import _reserve_owned_ram, _release_owned_ram11from ..sources.can_spec import CANSourceRef12from ..sources.models import SignalBatch13from ..sources.packing import pack_decoded14from ..sources.protocol import SourceReadError15from ..timebase.conversion import checked_add_ns, INT64_MIN, INT64_MAX, TimeOverflowError, require_int64_ns16from .can_reader import CANObservationRead17from .history import _pending_bytes181920def _float_seconds_ns(value):21 """精确把已有二进制float秒舍入至纳秒，不经Decimal文本和逐位展开。2223 参数: value为原生有限float秒。返回: (int64纳秒, 是否量化)，最近偶数舍入。24 异常: TypeError/ValueError为输入类型/非有限；TimeOverflowError为舍入后越界。25 边界: 整数比精确保留float已有位值；负零、负半值和次正规数均不先乘float。26 """27 if type(value) is not float:28 raise TypeError('原CAN时间需要float')29 if not isfinite(value):30 raise ValueError('原CAN时间必须有限')31 numerator, denominator = value.as_integer_ratio()32 quotient, remainder = divmod(abs(numerator) * 1000000000, denominator)33 if remainder * 2 > denominator or remainder * 2 == denominator and quotient & 1:34 quotient += 135 # 上面的原生float整数比与divmod只产生Python int，无需重复Integral类型分流。36 ns = -quotient if numerator < 0 else quotient37 if not INT64_MIN <= ns <= INT64_MAX:38 raise TimeOverflowError('time_ns超出int64纳秒范围')39 return ns, remainder != 0404142def _dense_time_tail(owner, timestamps, offset, mapped_times):43 """首条已按原规则提交后，继续原顺序校验余下时间，不提前报告末点错误。4445 参数: owner为当前read；timestamps为同批原时间；offset已核int64；mapped_times只含首条。46 返回: 完整映射tuple。异常: 原时间/偏移溢出、倒退或取消；保实际已完成观测前缀。47 流程: 稀疏证明不成立时逐项换算，微小float回退仍允许量化成同一纳秒。48 """49 try:50 for index, timestamp in enumerate(timestamps):51 if index == 0:52 continue53 ns, quantized = _float_seconds_ns(timestamp)54 mapped = ns + offset55 if not INT64_MIN <= mapped <= INT64_MAX:56 raise TimeOverflowError('shifted_time_ns超出int64纳秒范围')57 if mapped < owner._last_ns:58 raise SourceReadError('CAN原yield时间回退，禁止重排')59 owner._last_ns, owner._row_count = mapped, owner._row_count + 160 mapped_times.append((mapped, quantized))61 return tuple(mapped_times)62 finally:63 owner = timestamps = mapped_times = None646566def _retained_time_indices(batch):67 """确认保留记录能按原序号使用稀疏时间表，不替代传输包的完整认证。6869 参数: batch的时间已确认是原生有限非降float。返回: 索引列表，或None表示走完整换算。70 异常: 取消/分配错误原样传播；非典型记录回原路径，不新增业务拒绝条件。71 """72 indices = record = None73 try:74 indices, previous = [], -175 for record in batch.records:76 if (type(record) is not LegacyDecodedFrame or type(record.sequence) is not int77 or type(record.frame) is not LegacyRawFrame or type(record.frame.timestamp) is not float):78 return None79 index = record.sequence - batch.start_sequence80 if (not previous < index < len(batch.timestamps)81 or record.frame.timestamp != batch.timestamps[index]):82 return None83 indices.append(index)84 previous = index85 return indices86 finally:87 batch = record = indices = None888990def _sparse_batch_times(owner, batch):91 """证明全部原时间合法后，仅换算首末和实际消费记录，保完整观测事实。9293 参数: owner为已支付转换空间的read；batch为16至4096帧、目标不超过四分之一的批。94 返回: 按原序号索引的tuple；无需消费的位置为None，证明失败则返回完整映射。95 异常: 业务错误沿原顺序产生；取消/MemoryError保原异常，只保已提交的真实前缀。96 依据: 有限float非降时，精确乘十亿后最近偶数舍入仍单调；固定int偏移也保持顺序。97 首末原纳秒与偏移后纳秒均在int64范围，即证明所有内部时间有界。保留点仍精确98 换算其纳秒和量化标志。末点失败不能抢先报错，必须回原后缀寻找第一处错误。99 费用: 最多4096个槽和1024个保留索引，含回退表仍小于原4MiB预付，无缓存或新owner。100 """101 timestamps = batch.timestamps102 mapped_times = indices = sparse = result = None103 try:104 # 第一条保原检查/提交顺序，坏时间先于坏偏移，跨批回退也在这里拒绝。105 first = timestamps[0]106 ns, quantized = _float_seconds_ns(first)107 offset = require_int64_ns(owner._spec.offset_ns, 'offset_ns')108 mapped = ns + offset109 if not INT64_MIN <= mapped <= INT64_MAX:110 raise TimeOverflowError('shifted_time_ns超出int64纳秒范围')111 if owner._last_ns is not None and mapped < owner._last_ns:112 raise SourceReadError('CAN原yield时间回退，禁止重排')113 if owner._first_ns is None:114 owner._first_ns = mapped115 owner._last_ns, owner._row_count = mapped, owner._row_count + 1116 mapped_times, previous = [(mapped, quantized)], first117 for timestamp in timestamps:118 if type(timestamp) is not float or not isfinite(timestamp) or timestamp < previous:119 return _dense_time_tail(owner, timestamps, offset, mapped_times)120 previous = timestamp121 indices = _retained_time_indices(batch)122 if indices is None:123 return _dense_time_tail(owner, timestamps, offset, mapped_times)124 try:125 last_ns, last_quantized = _float_seconds_ns(timestamps[-1])126 except TimeOverflowError:127 return _dense_time_tail(owner, timestamps, offset, mapped_times)128 last_mapped = last_ns + offset129 if not INT64_MIN <= last_mapped <= INT64_MAX:130 return _dense_time_tail(owner, timestamps, offset, mapped_times)131 sparse = [None] * len(timestamps)132 sparse[0], sparse[-1] = mapped_times[0], (last_mapped, last_quantized)133 for index in indices:134 if index == 0 or index == len(timestamps) - 1:135 continue136 ns, quantized = _float_seconds_ns(timestamps[index])137 sparse[index] = (ns + offset, quantized)138 result = tuple(sparse)139 # 全部所需映射完成后才提交其余原观测；取消不能冒充整批已经处理完成。140 owner._last_ns, owner._row_count = last_mapped, owner._row_count + len(timestamps) - 1141 return result142 finally:143 owner = batch = timestamps = mapped_times = indices = sparse = result = None144145146class CANBatchObservationRead(CANObservationRead):147 """复用真实read责任槽与完成协议，追加一个有界批次及其游标。148149 _incoming由进程保活原记录，转换表单独预付。原yield总数与消息仓保存条数分开；150 同刻记录跨SignalBatch不合并，尚未输出的同刻别名不提前推进watermark。151 """152153 def __init__(self, owner, request):154 """安装批游标后调用原责任初始化。参数: owner/request同原read。返回: read。异常: 原错误。"""155 self._incoming = self._mapped_times = None156 self._incoming_index = 0157 self._input_ended = False158 self._progress_ns = None159 super().__init__(owner, request)160161 def _load_batch(self):162 """读取一批并检查所有原时间，记录全部观测，EOF才封口实际仓。163164 参数: 无。返回: bool是否取到批次，空时间批也算真实返回。165 异常: 原进程、时间倒退、预算/仓错误；调用者统一保留最高恢复owner。166 """167 try:168 self._record = self._incoming = self._mapped_times = None169 _release_owned_ram(self, self._record_cost, '_record_cost')170 # 4096项时间转换及当前引用副本的保守空间，进程另支付完整records树。171 _reserve_owned_ram(self, 4259840, '_record_cost')172 if getattr(self._process, '_prefetch', False):173 aliases = len(self._selected)174 # 当前原批最多R*A个样本，跨原批尚未accept的装箱最多S个。175 # pending每新增身份计256字节、每列512；留存最多增长G，之后176 # accept还会预约2*(P+G)，相对当前已付P共需额外2*P+3*G。177 # 固定余量另覆盖装箱和4096+1024*S；动态只约束预取，不误拒同步。178 growth = (512 * aliases + 256 * (self._process._batch_rows * aliases179 + self._request.batch_rows)) if aliases else 0180 self._process._set_prefetch_headroom(181 2 * _pending_bytes(self._validator) + 3 * growth)182 try:183 batch = next(self._process)184 except StopIteration:185 if not self._process.finished:186 raise SourceReadError('CAN批进程未完成真实EOF及关闭')187 self._input_ended = True188 if self._raw_store is not None:189 self._raw_store.seal()190 if self._message_store is not None:191 self._message_store.seal()192 _release_owned_ram(self, self._record_cost, '_record_cost')193 return False194 self._incoming = batch195 if batch.start_sequence != self._row_count:196 raise SourceReadError('CAN批次原序跨度不连续')197 if batch.time_scope != self._owner.time_scope:198 raise SourceReadError('CAN批次时间范围与实际来源不同')199 self._retained_record_count += len(batch.records)200 self._full_check_count += batch.full_check_count or 0201 if batch.time_scope != 'source':202 # 无关帧不进入时间循环。所需时间仅以原记录序号索引；none保存明确None。203 mapped_times = {}204 for record in batch.records:205 if batch.time_scope == 'none':206 mapped_times[record.sequence - batch.start_sequence] = (None, None)207 continue208 ns, quantized = _float_seconds_ns(record.frame.timestamp)209 mapped = checked_add_ns(ns, self._spec.offset_ns)210 if self._last_ns is not None and mapped < self._last_ns:211 raise SourceReadError('CAN所需记录时间回退，禁止重排')212 if self._first_ns is None:213 self._first_ns = mapped214 self._last_ns = mapped215 mapped_times[record.sequence - batch.start_sequence] = (mapped, quantized)216 self._row_count += batch.frame_count217 self._mapped_times, self._incoming_index = mapped_times, 0218 return True219 if (type(batch.timestamps) is tuple and type(batch.records) is tuple220 and type(batch.start_sequence) is int and 16 <= len(batch.timestamps) <= 4096221 and len(batch.records) * 4 <= len(batch.timestamps)):222 self._mapped_times, self._incoming_index = _sparse_batch_times(self, batch), 0223 return True224 # 密集/短批保留原循环，避免为没有可省换算的输入增加逐帧分支。225 mapped_times, offset = [], None226 for timestamp in batch.timestamps:227 ns, quantized = _float_seconds_ns(timestamp)228 # 保原顺序：时间换算成功才核偏移。冻结spec的偏移每批核一次；229 # 空时间批仍不触碰偏移。此循环没有会改变spec的外部回调。230 if offset is None:231 offset = require_int64_ns(self._spec.offset_ns, 'offset_ns')232 mapped = ns + offset233 if not INT64_MIN <= mapped <= INT64_MAX:234 raise TimeOverflowError('shifted_time_ns超出int64纳秒范围')235 if self._last_ns is not None and mapped < self._last_ns:236 raise SourceReadError('CAN原yield时间回退，禁止重排')237 if self._first_ns is None:238 self._first_ns = mapped239 self._last_ns, self._row_count = mapped, self._row_count + 1240 mapped_times.append((mapped, quantized))241 self._mapped_times, self._incoming_index = tuple(mapped_times), 0242 return True243 finally:244 batch = mapped_times = None245246247 def _advance_record(self):248 """推进到下一保留原记录，最多追加一次raw及消息历史。249250 参数: 无。返回: bool是否有记录，跳过无目标批但保其全部观测。251 异常: 原记录/定位/仓/预算/取消；不在别名分片处重复保存。252 """253 try:254 self._record = None255 while self._incoming is None or self._incoming_index == len(self._incoming.records):256 if self._incoming is not None and self._incoming.timestamps:257 self._progress_ns = self._last_ns258 if not self._load_batch():259 return False260 record = self._incoming.records[self._incoming_index]261 self._incoming_index += 1262 mapped, quantized = self._mapped_times[record.sequence - self._incoming.start_sequence]263 if self._raw_store is not None:264 self._raw_store.append(record, mapped)265 ids = self._owner.message_frame_ids266 if self._message_store is not None and (ids is None or record.frame.arbitration_id in ids):267 self._message_store.append(record, mapped, time_quantized=quantized)268 self._binding_projection.fields_for(record.frame)269 if self._owner._frame_observer is not None:270 self._owner._frame_observer(record.frame)271 self._record, self._mapped, self._conversion, self._alias_index = record, mapped, quantized, 0272 self._progress_ns = mapped273 # 在实际定位/时间检查成功后交给新分析层；不改变旧解码或原记录顺序。274 self._deliver_message(record, mapped)275 return True276 finally:277 record = None278279280 def _packed_batch(self):281 """跨保留记录装箱至总样本上限，不为每条非目标原帧生成空历史。282283 参数: 无。返回: 已校验SignalBatch，或者None表示真实输入耗尽且无剩余样本。284 异常: 原模型/归一化/预算/IO错误；样本时间和sequence均来自同次原记录。285 """286 try:287 _reserve_owned_ram(self, 65536, '_pack_cost')288 records, references, count, delivered = {}, {}, 0, 0289 # 记录流只订阅消息时没有信号列，样本数永远为零。每256条消息让主290 # 状态机消费已完成时间组，避免把整文件的组堆在来源队列中。291 while count < self._request.batch_rows and (self._selected or delivered < 256):292 if self._record is None or self._alias_index >= len(self._selected):293 if not self._advance_record():294 break295 delivered += 1296 while self._alias_index < len(self._selected) and count < self._request.batch_rows:297 definition, field = self._selected[self._alias_index]298 self._alias_index += 1299 if self._spec.decode_policy != 'legacy' and (self._record.frame.is_remote_frame or self._record.frame.is_error_frame):300 continue301 if field not in self._binding_projection.fields_for(self._record.frame):302 continue303 # 包含完整定位引用、列表/tuple转换及pack数组，而非只付标量大小。304 _reserve_owned_ram(self, self._sample_pack_costs[self._alias_index - 1], '_pack_cost')305 ref = CANSourceRef(self._request.session_id, self._spec.name, definition.field,306 str(self._spec.input_paths[self._incoming.input_index]), self._spec.legacy_profile, field.channel,307 field.signal, field.frame_id, self._record.frame.is_extended_id, self._record.frame.is_fd, self._record.sequence,308 self._record.frame.timestamp, self._conversion)309 records.setdefault(definition.name, []).append((self._mapped, self._record.sequence,310 self._record.values.get(field.signal)))311 references.setdefault(definition.name, []).append(ref)312 count += 1313 if not count:314 if delivered:315 # 这只是有界读取进度，不是文件/来源结束；最后同刻组仍由316 # RawSignalGroups保留，下一次不同时间或真实EOF才封口。317 progress = self._progress_ns318 watermark = None if progress is None or progress == -(1 << 63) else progress - 1319 return self._accept(SignalBatch(self._spec.name, self._request.session_id,320 {}, watermark, False, None))321 return None322 columns = {definition.name: pack_decoded(definition, tuple(records[definition.name]),323 tuple(references[definition.name]), self._owner.backend)324 for definition, _ in self._selected if definition.name in records}325 progress = self._progress_ns326 watermark = None if progress == -(1 << 63) else progress - 1327 return self._accept(SignalBatch(self._spec.name, self._request.session_id,328 columns, watermark, False, None))329 finally:330 records = references = columns = ref = None331332333 def __next__(self):334 """交付有样本批或唯一EOF空批，下一次耗尽才开放观测与移交。335336 参数: 无。返回: SignalBatch。异常: StopIteration为真实完成；其他原错误poison本read。337 """338 if self._finished:339 raise StopIteration340 self._require_open()341 try:342 self._release_current()343 if not self._started:344 self._start()345 if self._eof_delivered:346 self._finish()347 else:348 batch = None if self._input_ended else self._packed_batch()349 if batch is not None:350 return batch351 if not self._pack_cost:352 _reserve_owned_ram(self, 65536, '_pack_cost')353 batch = SignalBatch(self._spec.name, self._request.session_id, {}, self._last_ns,354 True, self._last_ns)355 self._accept(batch)356 self._eof_delivered = True357 return batch358 except BaseException as exc:359 self._fail(exc)360 raise361 raise StopIteration362363 def close(self):364 """释放本批引用后按原真实owner顺序关闭。参数: 无。返回: None。异常: 原可恢复错误。"""365 self._incoming = self._mapped_times = None366 super().close()367