chronaxis.legacy.batch_read
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在共享预算内按批取得BLF解码结果，按配置保留观测时间与所需原始记录。23复用逐条read的INIT/READY验证、实际进程关闭和失败恢复；子进程使用共用BLF读取器。4父在授予一批许可前预付两端缓冲、解析模型和时间数组的保守费用；同时只一批在途。5每批最多4096原帧且256KiB，预算较小时缩字节上限。默认同步；显式开启预取时，6当前批交付前可再预付并授权一批，使子读取与父计算重叠。取消或当前批处理失败时7可能额外读出这一批；预约/发送失败则当前批也不交付。预算不足时同步读取。8最后数据批可含真实EOF，下一次next完成退出/关闭才finished。9低层默认full/source保旧协议；auto结合固定requested范围使用v4 READY与CXB2，10原帧数独立于时间数量，计数和预取不会因跳过无用时间而丢失读取进度。11"""1213from functools import partial14from dataclasses import replace15from pathlib import Path1617from .batch_wire import (BATCH_MAGIC, DEMAND_MAGIC, HEADER, DEMAND_HEADER, MAX_BATCH_BYTES, MAX_BATCH_ROWS, MIN_BATCH_BYTES,18 FILE_HEADER, FILE_MAGIC, decode_batch)19from .batch_worker import REQUESTED_VALUES_POLICY, batch_worker_main20from .process_read import LegacyCANProcessRead21from .records import MAX_WIRE_BYTES, decode_packet22from ..results.budget import _reserve_owned_ram, _release_owned_ram23from ..results.errors import BudgetExceededError24from ..sources.protocol import SourceReadError252627class LegacyCANBatchRead(LegacyCANProcessRead):28 """持有当前借用结果及可选下一批许可，全部费用由共享关闭流程归还。2930 _current保存已验证批次，_current_cost预付两端全部批次工作空间；31 _prefetch_packet/_prefetch_cost独立持有至多一个下一批许可及其完整费用，32 所有共享账本操作均在父调用线程完成，没有后台线程预约或无界队列。33 _batch_eof仅表示收到EOF候选，实际.finished仍要求子正常退出和父关闭成功。34 """3536 def __init__(self, request, budget, *, frame_ids=None, batch_rows=MAX_BATCH_ROWS,37 consumer_headroom_bytes=0, prefetch=False, decode_scope='all',38 decode_frame_ids=None, input_check='full', time_scope='source', _worker_target=None,39 input_paths=None, decode_policy='legacy', cycle_fields=(), metadata_fields=()):40 """保存批次上限与解码后选择，不打开文件或启动子进程。4142 参数: request/budget同逐条read；frame_ids=None保全部原记录，tuple仅保这些ID；43 batch_rows为1至4096原帧上限；consumer_headroom_bytes为调用方处理本批时44 仍需预约的工作空间，此处只留余量、不代其付款；prefetch为显式bool，45 True允许额外一个已付批次在途；_worker_target仅供真实spawn故障测试。46 decode_scope='all'沿用原全解码与v1 READY，此时decode_frame_ids须为None；47 'requested'须给固定原生tuple ID，fields只能属于这些ID，使用严格v3 READY；48 此时frame_ids仍可为None或其他ID，保留原始帧不扩大固定解码范围。49 input_paths为原序BLF Path tuple或None；decode_policy默认legacy，strict/partial50 采用新长度政策。两者启用时v5 READY确认文件和政策，多文件批带文件位置。51 cycle_fields为另行请求周期的有限字段tuple；metadata_fields仅取候选。两者不加入值解码，v5确认声明。52 input_check默认full/source保旧行为；auto启用ID优先和v4 READY，53 time_scope为source/selected/none，后两者只传所需身份不能供原始片段导出。54 返回: 尚未启动read。异常: 原配置/预算错误，未返回owner仍可恢复close。55 """56 self._frame_ids, self._batch_rows = frame_ids, batch_rows57 self._consumer_headroom_bytes = consumer_headroom_bytes58 self._batch_limit = MAX_WIRE_BYTES59 self._batch_eof = False60 self._frames_authorized = self._records_returned = 061 self._prefetch = prefetch62 self._prefetch_packet = None63 self._prefetch_cost = 064 self._prefetch_headroom_bytes = 065 # 内部来源证明可借用刚核完的原包；默认无观察者，不复制包或另收取散列费用。66 self._validated_packet_consumer = None67 # 基类构造取消会动态调用本类close，先建立策略引用与独立费用责任槽。68 self._decode_scope, self._decode_frame_ids = decode_scope, decode_frame_ids69 self._input_check, self._time_scope = input_check, time_scope70 self._minimum_batch_bytes = MIN_BATCH_BYTES + (71 DEMAND_HEADER.size-HEADER.size if input_check == 'auto' and time_scope == 'source' else 0)72 self._cycle_fields = cycle_fields73 self._metadata_fields = metadata_fields74 self._input_paths, self._decode_policy = input_paths, decode_policy75 self._input_index_accepted = 076 self._source_policy_cost = 077 self._file_envelope = False78 self._full_checks_accepted = 079 self._decode_policy_cost = 080 super().__init__(request, budget,81 _worker_target=batch_worker_main if _worker_target is None else _worker_target)82 try:83 if type(prefetch) is not bool:84 raise TypeError('CAN批次预取开关必须是bool')85 if type(batch_rows) is not int or not 1 <= batch_rows <= MAX_BATCH_ROWS:86 raise ValueError('批次需要1至4096条原帧上限')87 if type(consumer_headroom_bytes) is not int or consumer_headroom_bytes < 0:88 raise ValueError('消费批次的后续工作余量必须是非负整数')89 if frame_ids is not None and (type(frame_ids) is not tuple or len(frame_ids) > 409690 or any(type(value) is not int or not 0 <= value <= 0x1fffffff for value in frame_ids)91 or len(set(frame_ids)) != len(frame_ids)):92 raise ValueError('解码后保留ID必须是无重复29位整数tuple或None')93 if type(decode_scope) is not str or decode_scope not in ('all', 'requested'):94 raise ValueError('CAN解码范围需要all或requested')95 if type(input_check) is not str or input_check not in ('auto', 'full'):96 raise ValueError('CAN输入检查需要auto/full')97 if type(time_scope) is not str or time_scope not in ('source', 'selected', 'none'):98 raise ValueError('CAN时间范围需要source/selected/none')99 if input_check == 'full' and time_scope != 'source':100 raise ValueError('完整检查必须保留全来源时间')101 if input_check == 'auto' and decode_scope != 'requested':102 raise ValueError('自动按需检查需要固定requested解码范围')103 if decode_scope == 'all':104 if decode_frame_ids is not None:105 raise ValueError('全解码模式不能另设固定解码ID')106 else:107 if (type(decode_frame_ids) is not tuple or len(decode_frame_ids) > 4096108 or any(type(value) is not int or not 0 <= value <= 0x1fffffff109 for value in decode_frame_ids)):110 raise ValueError('按需解码ID必须是无重复29位整数tuple')111 # 独立预付父/子tuple、集合、partial及策略元数据；不挪用消费批次余量。112 # 每ID 256B覆盖两侧容器槽，8192B覆盖空容器/函数描述；每字段1024B113 # 另覆盖子侧投影表和构建临时槽，名称借自已付请求，不复制其文本。114 _reserve_owned_ram(self, 8192 + 256 * len(decode_frame_ids)115 + 1024 * len(request.fields), '_decode_policy_cost')116 if len(set(decode_frame_ids)) != len(decode_frame_ids):117 raise ValueError('按需解码ID必须是无重复29位整数tuple')118 if any(field.frame_id not in decode_frame_ids for field in request.fields):119 raise ValueError('所需字段的报文ID必须属于固定解码集合')120 if _worker_target is None:121 self._worker_target = partial(batch_worker_main, decode_frame_ids=decode_frame_ids,122 input_check=input_check, time_scope=time_scope)123 from .decode_policy import validate_decode_policy124 validate_decode_policy(decode_policy)125 from .dbc_cycles import require_cycle_fields126 require_cycle_fields(cycle_fields)127 require_cycle_fields(metadata_fields)128 if input_paths is not None:129 if (type(input_paths) is not tuple or not 1 <= len(input_paths) <= 4096130 or any(not isinstance(p,Path) or p.suffix.lower() != '.blf'131 or len(str(p)) > 4096 for p in input_paths)132 or input_paths[0] != request.input_path):133 raise ValueError('连续输入需要原序BLF Path tuple，首路径须与INIT一致')134 if input_paths is not None or decode_policy != 'legacy' or cycle_fields or metadata_fields:135 paths = input_paths or (request.input_path,)136 _reserve_owned_ram(self,16384 + sum(512+len(str(p))*32 for p in paths)137 + len(request.fields)*512138 + sum(8192+128*len(f.signal) for f in (*cycle_fields,*metadata_fields)),'_source_policy_cost')139 self._file_envelope = len(paths) > 1140 self._minimum_batch_bytes += FILE_HEADER.size if self._file_envelope else 0141 if _worker_target is None:142 self._worker_target = partial(batch_worker_main,143 decode_frame_ids=decode_frame_ids,input_check=input_check,time_scope=time_scope,144 input_paths=None if input_paths is None else tuple(str(p) for p in input_paths),145 decode_policy=decode_policy,cycle_fields=cycle_fields,metadata_fields=metadata_fields)146 except BaseException as exc:147 self._fail(exc, '批次构造')148149 def _check_ready(self, packet):150 """默认沿用原v1验证；requested核v3，auto核v4及检查/时间范围。151152 参数: packet为严格wire解码后的READY。返回: None。异常: SourceReadError。153 ID先逐项核原生int再比较tuple，拒绝bool/int混同、缺项、额外字段或顺序改变；154 显式测试worker也必须确认同策略，不能用旧v1/v2应答冒充字段已裁剪。155 """156 if self._input_paths is not None or self._decode_policy != 'legacy' or self._cycle_fields or self._metadata_fields:157 expected_paths = [str(p) for p in (self._input_paths or (self._request.input_path,))]158 base_version = 1 if self._decode_scope == 'all' else (4 if self._input_check == 'auto' else 3)159 if (type(packet.get('version')) is not int or packet['version'] != 5160 or type(packet.get('base_version')) is not int or packet['base_version'] != base_version161 or packet.get('input_paths') != expected_paths162 or packet.get('decode_policy') != self._decode_policy):163 raise SourceReadError('连续CAN READY文件或解码政策不符')164 if self._cycle_fields:165 from .records import field_from_dict166 actual = packet.get('cycle_fields')167 if type(actual) is not list or tuple(field_from_dict(f) for f in actual) != self._cycle_fields:168 raise SourceReadError('CAN READY周期请求与声明不同')169 if self._metadata_fields:170 from .records import field_from_dict171 actual = packet.get('metadata_fields')172 if type(actual) is not list or tuple(field_from_dict(f) for f in actual) != self._metadata_fields:173 raise SourceReadError('CAN READY字段候选请求与声明不同')174 packet = dict(packet)175 if self._metadata_fields:176 packet.pop('metadata_fields')177 if self._cycle_fields:178 packet.pop('cycle_fields')179 packet['version'] = packet.pop('base_version')180 packet.pop('input_paths'); packet.pop('decode_policy')181 if self._decode_scope == 'all':182 return super()._check_ready(packet)183 required_keys = {'op', 'version', 'profile', 'worker_pid', 'fields',184 'dependency_versions', 'decode_scope', 'decode_frame_ids',185 'decoded_values_policy'}186 if self._input_check == 'auto':187 required_keys.update(('input_check', 'time_scope'))188 if (set(packet) != required_keys189 or packet.get('op') != 'ready' or type(packet.get('version')) is not int190 or packet.get('version') != (4 if self._input_check == 'auto' else 3)191 or packet.get('profile') != self._request.profile192 or type(packet.get('worker_pid')) is not int193 or packet.get('worker_pid') != self._worker_pid):194 raise SourceReadError('按需CAN READY身份、版本或实际PID不符')195 ids = packet['decode_frame_ids']196 if (type(packet['decode_scope']) is not str or packet['decode_scope'] != 'requested'197 or type(ids) is not list or len(ids) != len(self._decode_frame_ids)198 or any(type(value) is not int for value in ids)199 or tuple(ids) != self._decode_frame_ids200 or type(packet['decoded_values_policy']) is not str201 or packet['decoded_values_policy'] != REQUESTED_VALUES_POLICY):202 raise SourceReadError('按需CAN READY解码策略或固定ID与启动请求不同')203 if self._input_check == 'auto' and (packet['input_check'] != 'auto'204 or packet['time_scope'] != self._time_scope):205 raise SourceReadError('按需CAN READY检查或时间范围与启动请求不同')206 self._check_ready_metadata(packet)207208 def _check_ready_metadata(self, packet):209 """周期只为显式请求字段核验；默认维持原元数据契约。210211 参数: packet为已核READY。返回: None。异常: 原字段、周期或依赖校验错误。212 """213 return super()._check_ready_metadata(packet,cycle_fields=self._cycle_fields,metadata_fields=self._metadata_fields)214215 @property216 def counters(self):217 """区分批次许可上界、收到的原帧与保留记录，不把未消费批次伪称已完成。218219 参数: 无。返回: 独立dict；frames_authorized包含许可但可能没读到的余量。220 异常: 原小字典分配错误。边界: items计全部原帧；accepted为接收校验完成，221 returned为已交付，records_returned计已交付的保留记录，预取失败不会虚增交付数。222 """223 # 完整保留历史诊断字典形状，新检查事实通过独立属性与版本化批次发布。224 return dict(super().counters, frames_authorized=self._frames_authorized,225 records_returned=self._records_returned)226227 @property228 def full_check_count(self):229 """读取已验收批次中非原生输入执行完整检查的实际回退数。230231 参数: 无。返回: 非负int；不计尚在预取中的包，关闭后仍保留统计。232 异常: 无。此属性不把完整检查默认模式中的原帧数误作按需回退数量。233 """234 return self._full_checks_accepted235236 def _receive_reply(self):237 """在已预付的上限内接收批数据或原严格控制JSON，拒绝错误版本和超长包。238239 参数: 无。返回: 批次回应dict或原READY/ERROR/STATUS。异常: 原管道/格式/取消错误。240 内部观察者仅在整包严格解码成功后调用一次；只借用本槽中的payload与batch，241 不改变所有权。回调失败沿next的同一read清理，不交付或再次处理该批。242 """243 batch = None244 try:245 self._pending_reply = self._connection.recv_bytes(self._batch_limit)246 if self._pending_reply.startswith((BATCH_MAGIC, DEMAND_MAGIC, FILE_MAGIC)):247 if self._current_cost == 0:248 raise SourceReadError('未授予/付费的批次数据')249 payload = self._pending_reply250 index, allowance = 0, self._batch_limit251 if self._file_envelope:252 if len(payload) < FILE_HEADER.size or not payload.startswith(FILE_MAGIC):253 raise SourceReadError('连续CAN缺少文件位置')254 _, index = FILE_HEADER.unpack_from(payload)255 if not self._input_index_accepted <= index < len(self._input_paths):256 raise SourceReadError('连续CAN文件位置越界或倒退')257 payload = payload[FILE_HEADER.size:]258 allowance -= FILE_HEADER.size259 elif payload.startswith(FILE_MAGIC):260 raise SourceReadError('单文件CAN收到未声明的连续包')261 batch, eof = decode_batch(payload, expected_start=self._items_accepted,262 row_limit=self._batch_rows, byte_limit=allowance, frame_ids=self._frame_ids,263 time_scope=self._time_scope, input_check=self._input_check)264 if self._file_envelope:265 if eof and index != len(self._input_paths)-1:266 raise SourceReadError('连续CAN在最后文件前报告EOF')267 batch = replace(batch,input_index=index)268 self._input_index_accepted = index269 if self._validated_packet_consumer is not None:270 self._validated_packet_consumer(self, self._pending_reply, batch)271 return {'op': 'batch', 'request_id': batch.start_sequence, 'batch': batch, 'eof': eof}272 # 控制包即使在批读取时也保持原64KiB与严格JSON边界。273 return decode_packet(self._pending_reply)274 finally:275 # 原错误可留在调用方；已结束调用的临时引用不继续占有整批对象。276 batch = payload = None277278 def __next__(self):279 """授予一批有界许可并交付原序数据；确认真实EOF及退出后才结束迭代。280281 参数: 无。返回: LegacyDecodedBatch，借至下次next或close。282 异常: 真StopIteration、原预算/来源/取消错误；持续清理失败保同一owner。283 """284 if self._finished:285 raise StopIteration286 self._require_open()287 reply = batch = packet = None288 try:289 self.start()290 # 同步入口保持原归还时机；预取入口的最后批费用在实际关闭后归还。291 if not (self._prefetch and self._batch_eof):292 self._current = self._pending_reply = None293 _release_owned_ram(self, self._current_cost, '_current_cost')294 if self._batch_eof:295 self._eof_received = True296 self.close()297 if self._exit_code != 0:298 raise SourceReadError('旧CAN批次发EOF但没有正常退出')299 self._finished = True300 else:301 if self._prefetch_packet is not None:302 packet = self._prefetch_packet303 if packet['request_id'] != self._items_accepted:304 raise SourceReadError('预取许可与实际已消费位置不符')305 self._batch_limit = packet['bytes']306 # 同一owner的两槽之间移交已付责任，没有新的预算回调或线程。307 self._current_cost, self._prefetch_cost = self._prefetch_cost, 0308 self._prefetch_packet = None309 reply = self._receive_response(packet)310 else:311 self._batch_limit = self._available_limit()312 if self._batch_limit < self._minimum_batch_bytes:313 raise BudgetExceededError('CAN批次预算不足以保留一条最大记录及两端工作空间')314 _reserve_owned_ram(self, 128*self._batch_limit, '_current_cost')315 packet = self._new_packet(self._batch_limit)316 self._pulls_issued += 1317 self._frames_authorized += self._batch_rows318 reply = self._exchange(packet)319 if reply.get('op') != 'batch' or reply.get('request_id') != packet['request_id']:320 raise SourceReadError('批次许可未得到对应批次回应')321 batch = reply['batch']322 self._batch_eof = reply['eof']323 self._items_accepted += batch.frame_count324 self._full_checks_accepted += batch.full_check_count325 self._current = batch326 self._pending_reply = None327 if self._prefetch:328 self._try_prefetch()329 # 预取预约/发送失败时当前批未交付；accepted保接收事实，returned不提前增加。330 self._items_returned += batch.frame_count331 self._records_returned += len(batch.records)332 return batch333 except BaseException as exc:334 # 清掉本frame对大批对象的额外引用；真正持有关系由父_current负责清理。335 reply = batch = packet = None336 self._fail(exc, '批次读取')337 finally:338 reply = batch = packet = None339 raise StopIteration340341 def _new_packet(self, byte_limit):342 """为下一原序号建立有限批次许可，不触碰管道。343344 参数: 已确定并付费的字节上限。返回: 小dict。异常: 原分配错误。345 """346 return {'op': 'pull_batch', 'request_id': self._items_accepted,347 'rows': self._batch_rows, 'bytes': byte_limit,348 'frame_ids': None if self._frame_ids is None else list(self._frame_ids)}349350 def _set_prefetch_headroom(self, extra_bytes):351 """更新仅供下一次额外预取的动态消费余量，不收费且不改变同步许可。352353 参数: extra_bytes为调用方证明所需的非负原生int，在每次next前更新。354 返回: None。异常: TypeError/ValueError为非法余量；关闭后拒绝更新。355 边界: 本线程直接设置有限整数，没有后台预算回调；不足时只停预取。356 """357 self._require_open()358 if type(extra_bytes) is not int:359 raise TypeError('预取额外消费余量必须是原生int')360 if extra_bytes < 0:361 raise ValueError('预取额外消费余量不能为负数')362 self._prefetch_headroom_bytes = extra_bytes363364 def _available_limit(self, *, additional_headroom=0):365 """在当前共享余额中留下消费者余量，计算完整可付款的批次上限。366367 参数: additional_headroom为预取专用额外余量，原同步读取固定为0。368 返回: 不超过MAX_BATCH_BYTES的整数，可能不足一条。369 异常: 原账本读取错误；这里只看余额，不取得额外空间。370 """371 available = (self._budget.memory_budget_bytes - self._budget.ram_bytes372 - self._consumer_headroom_bytes - additional_headroom)373 return min(MAX_BATCH_BYTES, available // 128)374375 def _try_prefetch(self):376 """余量足够时全额预付并发送下一批；否则留到下一次同步读取。377378 参数: 无，当前批已完成全部接包校验。返回: None。379 异常: 原预约/发送/取消错误；失败不重发，也不交付当前批，由next统一清理。380 边界: EOF不再发许可；至多一个在途批，消费余量不用于预付。381 """382 if self._batch_eof:383 return384 if self._waiting or self._prefetch_packet is not None or self._prefetch_cost:385 raise SourceReadError('预取已有一个在途请求或费用责任')386 limit = self._available_limit(additional_headroom=self._prefetch_headroom_bytes)387 if limit < self._minimum_batch_bytes:388 return389 # 只有付款前的余额不足可以回退；预约实际入账或发送中断不可猜测成功程度。390 _reserve_owned_ram(self, 128*limit, '_prefetch_cost')391 self._prefetch_packet = self._new_packet(limit)392 self._pulls_issued += 1393 self._frames_authorized += self._batch_rows394 self._send_request(self._prefetch_packet)395396 def _release_closed_state(self):397 """共享close确认实际资源全关后清预取责任，再按原槽逐笔归还。398399 参数: 无。返回: None。异常: 原退费错误；同一close可从实际剩余槽恢复。400 """401 self._prefetch_packet = self._frame_ids = None402 self._validated_packet_consumer = None403 # 实际进程已结束才清策略与partial；退款失败仍保本owner费用槽供重复close。404 self._decode_frame_ids = None405 self._worker_target = None406 self._input_paths = None407 self._cycle_fields = ()408 self._metadata_fields = ()409 _release_owned_ram(self,self._source_policy_cost,'_source_policy_cost')410 _release_owned_ram(self, self._decode_policy_cost, '_decode_policy_cost')411 _release_owned_ram(self, self._prefetch_cost, '_prefetch_cost')412 super()._release_closed_state()413