chronaxis.legacy.batch_worker
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在隔离子进程按有限批次读取BLF，并按显式串行政策解码。23INIT仅加载DBC和公开字段元数据；每次pull_batch明确许可原帧条数与字节上限。4共用BLFRead按原序读取；默认all范围逐帧解码，strict/partial政策跳过非数据帧。5显式decode_frame_ids限制允许解码的ID，只有其中实际请求了字段的ID才调用6decode_serial_frame；纯常量触发帧不解码。按需模式在解码后立即只取INIT所需7字段，缺失键保持缺失；v3 READY确认这一投影策略。8原帧保留集合与解码集合独立，导出原始帧可保留全部帧而不解码无需求ID。9auto模式在原生消息ID不匹配时先跳过；仅source需要读其时间。selected/none目标10只保存必要身份与所需值；非原生消息沿完整原帧检查并计入新包的回退数量。11字节不足容纳下一条最大记录时先结束当前批，不提前读取再丢弃。真正StopIteration12可与最后批数据一起报告，子实际退出及资源回收仍由父确认。没有后台预读或pickle。13"""1415from collections.abc import Mapping16from collections import Counter17from importlib import metadata18import json19from math import isfinite20import os21import platform22from ..dbc_loading import load_dbc_file2324from . import worker25from .records import (PROFILE_ID, MAX_WIRE_BYTES, LegacyDecodedFrame, decoded_to_dict,26 _decoded_values_size)27from .batch_wire import (HEADER, DEMAND_HEADER, LENGTH, MIN_BATCH_BYTES, MAX_BATCH_BYTES,28 MAX_BATCH_ROWS, FILE_HEADER, FILE_MAGIC, encode_batch)29from .batch_validation import validate_discarded, _item_base_size30from .required_records import RequiredCANFrame, RequiredDecodedFrame, required_to_dict313233# READY区分按字段投影与整条消息values，避免混淆两种检查范围。34REQUESTED_VALUES_POLICY = 'requested_fields-v1'353637from .decode_policy import decode_serial_frame as _parse_selected383940def _requested_fields(fields, decode_ids):41 """按固定报文ID整理所需信号名，不读取输入或遍历解码结果。4243 参数: fields为INIT已校验的字段tuple；decode_ids为固定解码ID集合。44 返回: ID到去重信号名字典的有限映射，名字次序保持请求顺序。45 异常: ValueError表示字段所在报文不在解码范围；原分配/取消错误直接传播。46 父策略费用为每个字段另付1024B，覆盖此表及构建临时槽；名称借自原请求。47 """48 projected = field = None49 try:50 projected = {}51 for field in fields:52 if field.frame_id not in decode_ids:53 raise ValueError('所需字段的报文ID必须属于固定解码集合')54 projected.setdefault(field.frame_id, {})[field.signal] = None55 return projected56 finally:57 # 构建失败的traceback不能在父关闭并归还策略费用后继续借用投影表。58 projected = field = fields = decode_ids = None596061def _project_values(values, names):62 """只读取本报文所需的原值，让后续完整模型只检查这些值。6364 参数: values为显式串行解码政策返回值；names为本ID需要的名字映射或空tuple。65 返回: 新dict，保留实际存在的所需键；KeyError代表缺失，不补None或零。66 异常: 有字段需求时非Mapping报TypeError；所需键访问和取消错误原样传播。67 无字段的常量触发报文直接返回空dict，不遍历或校验没有用途的原values。68 """69 projected = name = value = None70 try:71 if not names:72 return {}73 if not isinstance(values, Mapping):74 raise TypeError('解码记录需要LegacyRawFrame与数值Mapping')75 projected = {}76 for name in names:77 try:78 value = values[name]79 except KeyError:80 continue81 projected[name] = value82 return projected83 finally:84 # 后续模型负责所需值的类型、诊断浮点和ITEM上限；本层不保留原结果引用。85 values = names = projected = name = value = None868788def _encode_fresh_packet(packet):89 """编码本worker刚完成完整检查的记录，省去同一JSON树的再次遍历。9091 参数: packet只能是紧邻LegacyDecodedFrame成功构造后由原decoded_to_dict生成的92 私有dict；两者之间没有用户回调或对象交付。不是任意外部JSON的编码入口。93 返回: 与公开encode_packet相同的UTF-8 bytes，不带换行。94 异常: 原JSON/UTF-8编码错误、实际超过65536字节的ValueError或原取消异常。95 完整模型已检查全部帧字段、含ITEM外壳的长度及收到的Mapping两遍，再复制冻结数值；96 原转换只生成固定字段与非有限数值标签。沿用原批许可费用，不保留跨记录状态。97 """98 payload = None99 try:100 # 保持原JSON参数和实际字节复核；复用的是前面的检查，不放宽最终传输上限。101 payload = json.dumps(packet, ensure_ascii=False, allow_nan=False, sort_keys=True,102 separators=(',', ':')).encode('utf-8')103 if len(payload) > MAX_WIRE_BYTES:104 raise ValueError('wire包超过65536字节')105 return payload106 finally:107 # 调用方可保留取消的traceback，本层不再借用记录树或编码缓冲。108 packet = payload = None109110111def _limits(packet, frames_returned):112 """验证本次批许可及解码后投影，非法许可不能推进旧读取器。113114 参数: packet为已检查JSON；frames_returned为已发送的原帧数。115 返回: (原帧上限,字节上限,保留ID集合或None)。异常: TypeError/ValueError。116 """117 worker._exact(packet, ('op', 'request_id', 'rows', 'bytes', 'frame_ids'))118 if type(packet['request_id']) is not int or packet['request_id'] != frames_returned:119 raise ValueError('批次许可必须按原帧序号连续')120 rows, size, ids = packet['rows'], packet['bytes'], packet['frame_ids']121 if (type(rows) is not int or not 1 <= rows <= MAX_BATCH_ROWS122 or type(size) is not int or not MIN_BATCH_BYTES <= size <= MAX_BATCH_BYTES):123 raise ValueError('批次许可条数或字节上限非法')124 if ids is not None and (type(ids) is not list or len(ids) > 4096125 or any(type(value) is not int or not 0 <= value <= 0x1fffffff for value in ids)126 or len(set(ids)) != len(ids)):127 raise ValueError('解码后保留ID必须是无重复29位整数列表或None')128 return rows, size, None if ids is None else frozenset(ids)129130131def batch_worker_main(connection, *, decode_frame_ids=None, input_check='full', time_scope='source',132 input_paths=None, decode_policy='legacy', cycle_fields=(), metadata_fields=()):133 """处理一次独立批次会话，沿用原INIT、元数据和有限远端错误格式。134135 参数: connection为spawn交付的独立连接；decode_frame_ids=None沿用全解码，136 原生tuple明确本次固定解码ID，允许空集合；input_check默认full保持旧行为，137 auto必须固定解码ID；time_scope为source/selected/none，full必须source。138 input_paths为原序路径文本tuple或None；decode_policy为legacy/strict/partial。139 cycle_fields显式请求同次DBC周期，metadata_fields仅取候选；不改变request.fields或值解码的报文集合。140 新政策只核有字段需求的帧长度，非数据帧不解码；解码空值不继承前值。141 返回: 正常STOP或真实EOF时None。142 异常: 失败发送有限事实并以SystemExit(1)退出；清理副错不替代原错误。143 副作用: 子进程内加载一次DBC，每个声明文件建立一个共用BLFRead，按原序取帧；144 默认all解码全部帧，strict/partial跳过非数据帧；按需只解码固定集合中有字段需求的帧，145 并立即投影所需字段。未解码帧不产生或检查146 解码值，仍检查原帧；原始帧保留范围不扩大解码范围。147 只关闭自己连接及同次共用读取器；第三方内部RSS仍不作硬上限保证。148 """149 iterator = failure = request_id = None150 phase, frames_returned, pulls_received, parse_calls = 'init', 0, 0, 0151 timestamps = encoded = record = frame = msg = values = payload = None152 decode_ids = projected_fields = ready = None153 try:154 request = worker._request(worker._receive(connection))155 from .decode_policy import validate_decode_policy, check_decode_length156 validate_decode_policy(decode_policy)157 paths = (str(request.input_path),) if input_paths is None else input_paths158 if (type(paths) is not tuple or not 1 <= len(paths) <= 4096159 or any(type(p) is not str or not p or len(p) > 4096 for p in paths)160 or paths[0] != str(request.input_path)):161 raise ValueError('连续文件声明须与INIT首路径一致')162 input_index = 0163 envelope_size = FILE_HEADER.size if len(paths) > 1 else 0164 if (input_check not in ('auto', 'full') or time_scope not in ('none', 'selected', 'source')165 or input_check == 'full' and time_scope != 'source'166 or input_check == 'auto' and decode_frame_ids is None):167 raise ValueError('输入检查与时间范围必须匹配固定按需解码策略')168 if decode_frame_ids is not None:169 # 策略来自启动实参，不能由后续PULL更改；无策略时保持原INIT和执行先后。170 if (type(decode_frame_ids) is not tuple or len(decode_frame_ids) > 4096171 or any(type(value) is not int or not 0 <= value <= 0x1fffffff172 for value in decode_frame_ids)173 or len(set(decode_frame_ids)) != len(decode_frame_ids)):174 raise ValueError('按需解码ID必须是无重复29位整数tuple')175 decode_ids = frozenset(decode_frame_ids)176 projected_fields = _requested_fields(request.fields, decode_ids)177 phase = 'dependencies'178 can, cantools = worker._load_can_runtime()179 # 仅取初始化刚加载的真实类，不提前导入CAN库；测试替身没有此类时沿用原字段路径。180 native_message_type = getattr(can, 'Message', None)181 phase = 'dbc'182 db = load_dbc_file(str(request.dbc_path), cantools)183 phase = 'metadata'184 from .binding import FrameBindingResolver185 resolver = FrameBindingResolver(db, request.fields)186 duplicate_ids = {frame_id for frame_id, count in Counter(message.frame_id for message in db.messages).items()187 if count > 1}188 phase = 'metadata'189 if cycle_fields or metadata_fields:190 from .dbc_cycles import require_cycle_fields191 require_cycle_fields(cycle_fields)192 require_cycle_fields(metadata_fields)193 all_fields = tuple(dict.fromkeys((*request.fields,*cycle_fields,*metadata_fields)))194 fields = worker._field_metadata(db,all_fields,cycle_fields=cycle_fields)195 else:196 fields = worker._field_metadata(db, request.fields)197 # 长度表只在新政策且字段确有需求时建立，旧调用不增加DBC查询。198 if (input_paths is not None or decode_policy != 'legacy') and os.path.isdir(paths[0]):199 raise ValueError('新原始报文入口需要文件，不能隐式扫描目录')200 iterator = worker._iter_blf(paths[0])201 versions = {'python': platform.python_version(), 'python-can': metadata.version('python-can'),202 'cantools': metadata.version('cantools'), 'natsort': metadata.version('natsort')}203 ready = {'op': 'ready', 'version': 1, 'profile': PROFILE_ID,204 'worker_pid': os.getpid(), 'fields': fields, 'dependency_versions': versions}205 if decode_ids is not None:206 ready.update(version=3, decode_scope='requested', decode_frame_ids=list(decode_frame_ids),207 decoded_values_policy=REQUESTED_VALUES_POLICY)208 if input_check == 'auto':209 ready.update(version=4, input_check=input_check, time_scope=time_scope)210 if input_paths is not None or decode_policy != 'legacy' or cycle_fields or metadata_fields:211 ready.update(base_version=ready['version'], version=5, input_paths=list(paths),212 decode_policy=decode_policy)213 if cycle_fields:214 ready['cycle_fields'] = [worker.field_to_dict(f) for f in cycle_fields]215 if metadata_fields:216 ready['metadata_fields'] = [worker.field_to_dict(f) for f in metadata_fields]217 worker._send(connection, ready)218 ready = None219 while True:220 phase, request_id = 'control', None221 packet = worker._receive(connection)222 if packet.get('op') == 'stop':223 worker._exact(packet, ('op',))224 break225 if packet.get('op') == 'status':226 worker._exact(packet, ('op',))227 worker._send(connection, {'op': 'status', 'pulls_received': pulls_received,228 'frames_returned': frames_returned, 'parse_calls': parse_calls})229 continue230 if packet.get('op') != 'pull_batch':231 raise ValueError('未知批次控制操作')232 rows, byte_limit, selected = _limits(packet, frames_returned)233 byte_limit -= envelope_size234 if byte_limit < MIN_BATCH_BYTES:235 raise ValueError('连续批次字节许可不足')236 if (input_check == 'auto' and time_scope == 'source'237 and byte_limit < MIN_BATCH_BYTES+DEMAND_HEADER.size-HEADER.size):238 raise ValueError('全来源按需批次字节许可不足以容纳一条最大记录')239 request_id = frames_returned240 pulls_received += 1241 timestamps, encoded = [], []242 raw_count, full_checks = 0, 0243 encoded_size, eof = (DEMAND_HEADER.size if input_check == 'auto' else HEADER.size), False244 while raw_count < rows:245 # 预留最坏单条空间后才授权next；未知是否匹配时不能先读再丢掉。246 if encoded_size + (8 if time_scope == 'source' else 0) + LENGTH.size + MAX_WIRE_BYTES > byte_limit:247 break248 phase = 'next'249 if iterator is None:250 input_index += 1251 if os.path.isdir(paths[input_index]):252 raise ValueError('连续原始报文需要文件，不能隐式扫描目录')253 iterator = worker._iter_blf(paths[input_index])254 try:255 msg = next(iterator)256 except StopIteration:257 if input_index == len(paths)-1:258 # 最后输入保持原收尾顺序：发EOF后在finally关闭。259 eof = True260 break261 iterator.close()262 iterator = None263 if raw_count:264 # 一批只来自一个文件；下次许可才开始下个文件。265 break266 continue267 sequence = frames_returned + raw_count268 raw_count += 1269 if input_check == 'auto':270 # 先读取ID；精确原生消息的不相关帧在此跳过，不能碰时间或其他属性。271 frame_id = msg.arbitration_id272 retained = selected is None or frame_id in selected273 needed = frame_id in decode_ids274 native = native_message_type is not None and type(msg) is native_message_type275 if not native:276 # 自定义消息可能有属性副作用，沿完整旧帧检查；实际回退数写入新包。277 frame = worker._frame(msg)278 full_checks += 1279 if time_scope == 'source':280 timestamp = msg.timestamp if native else frame.timestamp281 if type(timestamp) is not float or not isfinite(timestamp):282 raise ValueError('原报文timestamp需要有限float秒')283 timestamps.append(timestamp)284 encoded_size += 8285 if not retained and not needed:286 if not native:287 # _frame只核类型/字符数，旧完整检查还核UTF-8与ITEM总字节。288 # 复用同次快照，不为此再次读取自定义消息的属性或调用解码。289 _item_base_size(frame, sequence)290 msg = frame = None291 continue292 phase = 'parse'293 if (projected_fields.get(frame_id) and (decode_policy == 'legacy'294 or not (msg.is_remote_frame or msg.is_error_frame))):295 active = resolver.fields_for(msg)296 message = resolver.message_for(msg, active)297 if message is None:298 values = {}299 else:300 check_decode_length(msg, message.length, decode_policy)301 parse_calls += 1302 values = _parse_selected(db, msg, message, duplicate_ids)303 else:304 values = {}305 phase = 'batch'306 values = _project_values(values, projected_fields.get(frame_id, ()))307 if not native:308 # 非原生完整回退按旧原帧外壳计算所需值余量；不能用较小的309 # Required外壳放过原ITEM已超限的输入，也不能重复构造_frame。310 available = MAX_WIRE_BYTES-_item_base_size(frame, sequence)+2311 _decoded_values_size(values, available)312 if time_scope == 'source':313 # source仍需完整原记录承载现有时序/导出消费者；只处理真正所需ID。314 record = LegacyDecodedFrame(sequence, worker._frame(msg) if native else frame, values)315 packet_tree = decoded_to_dict(record)316 else:317 # none不读取目标时间；selected仅读取目标时间。没有载荷/DLC副本。318 frame = RequiredCANFrame(None if time_scope == 'none' else msg.timestamp,319 frame_id, msg.channel, msg.is_extended_id, msg.is_fd,320 msg.is_remote_frame, msg.is_error_frame)321 record = RequiredDecodedFrame(sequence, frame, values)322 packet_tree = required_to_dict(record)323 if retained:324 payload = _encode_fresh_packet(packet_tree)325 encoded.append(payload)326 encoded_size += LENGTH.size + len(payload)327 if time_scope == 'selected':328 timestamps.append(record.frame.timestamp)329 msg = values = record = frame = payload = packet_tree = None330 continue331 phase = 'parse'332 if ((decode_ids is None or projected_fields.get(msg.arbitration_id))333 and (decode_policy == 'legacy' or not (msg.is_remote_frame or msg.is_error_frame))):334 active = resolver.fields_for(msg)335 message = resolver.message_for(msg, active)336 if message is not None:337 check_decode_length(msg, message.length, decode_policy)338 if message is not None or decode_ids is None:339 parse_calls += 1340 values = _parse_selected(db, msg, message, duplicate_ids)341 else:342 values = {}343 else:344 values = {}345 phase = 'batch'346 timestamp = msg.timestamp347 if type(timestamp) is not float or not isfinite(timestamp):348 raise ValueError('原报文timestamp需要有限float秒')349 timestamps.append(timestamp)350 encoded_size += 8351 retained = selected is None or msg.arbitration_id in selected352 if projected_fields is not None and (retained or msg.arbitration_id in decode_ids):353 # 先保原时间/帧字段错误，再投影解码值；不能让Mapping访问抢先失败。354 # 所需ID即使本批不保留，仍完整检查需要的值；原始导出额外帧只含空值。355 frame = worker._frame(msg)356 values = _project_values(values, projected_fields.get(msg.arbitration_id, ()))357 record = LegacyDecodedFrame(sequence, frame, values)358 if retained:359 # all仍逐项检查完整原values；requested使用刚核过的投影记录。360 if projected_fields is None:361 record = LegacyDecodedFrame(sequence, worker._frame(msg), values)362 payload = _encode_fresh_packet(decoded_to_dict(record))363 encoded.append(payload)364 encoded_size += LENGTH.size + len(payload)365 elif record is None:366 # 默认核全部原值；按需跳过解码时只存在空values，原帧检查仍执行。367 validate_discarded(msg, values, sequence, native_message_type=native_message_type)368 msg = values = record = frame = payload = None369 phase = 'send_batch'370 payload = encode_batch(frames_returned, timestamps, encoded, eof, byte_limit,371 raw_count=raw_count, time_scope=time_scope, input_check=input_check,372 full_check_count=full_checks)373 if envelope_size:374 payload = FILE_HEADER.pack(FILE_MAGIC,input_index) + payload375 connection.send_bytes(payload)376 frames_returned += raw_count377 timestamps = encoded = payload = None378 if eof:379 break380 except BaseException as exc:381 failure = exc382 worker._report_error(connection, exc, phase, request_id)383 finally:384 # 父等待实际退出才能释放资源责任；不窥原generator局部变量来关未知句柄。385 timestamps = encoded = record = frame = msg = values = payload = None386 packet_tree = None387 decode_ids = decode_frame_ids = projected_fields = ready = None388 if iterator is not None:389 try:390 iterator.close()391 except BaseException as cleanup:392 if failure is None:393 failure = cleanup394 worker._report_error(connection, cleanup, 'generator_close', request_id)395 else:396 failure.add_note('旧生成器关闭附加失败: ' + type(cleanup).__name__)397 try:398 connection.close()399 except BaseException as cleanup:400 if failure is None:401 failure = cleanup402 else:403 failure.add_note('旧CAN子连接关闭附加失败: ' + type(cleanup).__name__)404 if failure is not None:405 raise SystemExit(1)406