chronaxis.legacy.worker
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在独立spawn进程按单条PULL许可调用共用BLF读取与串行解码政策。23有效INIT后导入依赖并加载一次DBC；惰性读取器在首次PULL才打开输入，READY不取帧。4每次许可按原序读取和解码一次，STATUS/ACK/STOP不授予读取许可。只使用限长JSON5bytes协议，异常发送有限事实，实际子退出和父侧预算清理由协议控制器确认。"""67# spawn加载时不触及CAN库或输入；读取依赖只在收到有效INIT后加载。8from importlib import metadata9import os10from pathlib import Path11import platform12from ..dbc_loading import load_dbc_file13from .records import (LegacyCANRequest, LegacyDecodedFrame, LegacyRawFrame, MAX_WIRE_BYTES,14 PROFILE_ID, decode_packet, decoded_to_dict, encode_packet, field_from_dict, field_to_dict)1516def _send(connection, packet):17 """通过bytes接口发送一个有界JSON包，不调用pickle通道。1819 参数: connection为本子进程Connection；packet为协议dict。返回: None。20 异常: 原wire类型/限额或管道错误；不重试导致重复ITEM。21 """22 connection.send_bytes(encode_packet(packet))232425def _receive(connection):26 """只接收一个不超过单包上限的bytes消息。2728 参数: connection为本子进程Connection。返回: 已校验dict。29 异常: EOFError/OSError/协议错误传播，不把断管当正常来源EOF。30 """31 return decode_packet(connection.recv_bytes(MAX_WIRE_BYTES))323334def _exact(packet, names):35 """限制每种控制包的精确字段，不让多余许可或版本字段静默生效。3637 参数: packet为已解析dict；names为允许键容器。返回: None。38 异常: ValueError为字段缺失或多余。39 """40 if set(packet) != set(names):41 raise ValueError('旧CAN控制包字段缺失或多余')424344def _request(packet):45 """解析唯一INIT，只接受单个实际BLF和原DBC路径，不读取其内容摘要。4647 参数: packet为INIT dict。返回: LegacyCANRequest。48 异常: 类型/形状/版本/路径错误；不打开BLF或加载DBC。49 """50 _exact(packet, ('op', 'version', 'path', 'dbc_path', 'fields'))51 if packet['op'] != 'init' or type(packet['version']) is not int or packet['version'] != 1:52 raise ValueError('旧CAN需要init/version1')53 if type(packet['path']) is not str or type(packet['dbc_path']) is not str or type(packet['fields']) is not list:54 raise TypeError('旧CAN INIT需要文本路径与fields数组')55 request = LegacyCANRequest(Path(packet['path']), Path(packet['dbc_path']),56 tuple(field_from_dict(value) for value in packet['fields']))57 # 只做stat和扩展名边界；不额外读取原CAN或预扫消息，不暗换另一种profile。58 if request.input_path.suffix.lower() != '.blf' or not request.input_path.is_file():59 raise ValueError('首版旧CAN仅支持单个现有BLF文件')60 if not request.dbc_path.is_file():61 raise FileNotFoundError(request.dbc_path)62 return request636465def _load_can_runtime():66 """在有效INIT后导入真实CAN依赖，模块导入本身只使用标准库。6768 参数: 无。返回: python-can与cantools模块。异常: 保留缺失依赖的可执行提示。69 不读取来源、DBC或原项目文件，不安装同名vendor模块。70 """71 from ..can_dependencies import import_can_dependency72 return import_can_dependency('can'), import_can_dependency('cantools')737475def _iter_blf(path):76 """按许可惰性创建共用BLF读取器，原消息顺序交付且可靠关闭本次文件。7778 参数: 已检查的BLF路径。返回: can.Message生成器。异常: 真实读取/关闭错误。79 创建生成器不打开文件；第一次next才调用BLFRead，STOP或EOF关闭同一owner。80 """81 from ..sources.blf import BLFRead, BLFSpec82 from ..results.budget import ResourceBudget, _cleanup_owner83 reader = primary = None84 try:85 reader = BLFRead(BLFSpec(Path(path)), ResourceBudget(), frame_only=True)86 yield from reader87 except BaseException as error:88 primary = error89 if reader is None:90 pending=getattr(error,'blf_read_cleanup_owner',None)91 if pending is not None:92 _cleanup_owner(pending,error,'blf_read_cleanup_owner','协议BLF构造')93 raise94 finally:95 if reader is not None and not reader.closed:96 if primary is None or isinstance(primary,GeneratorExit):97 reader.close()98 else:99 _cleanup_owner(reader,primary,'blf_read_cleanup_owner','协议BLF读取器')100101102def _field_metadata(db, fields, *, cycle_fields=()):103 """只读同次db公开messages/signals核定位唯一，并返回原单位。104105 参数: db为同次加载的唯一DBC数据库；fields为原LegacyCANField tuple；106 cycle_fields仅指定需要周期的字段，须在fields中。默认不访问cycle_time。107 返回: 按请求顺序的字段事实；周期请求的字段另含正cycle_ns。108 仅DBC为FD而定位为经典CAN时增加固定兼容提示。109 channel保声明，不冒称DBC已测通道；不把DBC声明写回原帧。110 异常: ValueError为ID/消息/帧型/信号歧义或不符，TypeError为单位不受支持。111 """112 if cycle_fields:113 from .dbc_cycles import require_cycle_fields, cycle_time_ns114 require_cycle_fields(cycle_fields)115 if not set(cycle_fields).issubset(fields):116 raise ValueError('周期字段须包含在本次DBC定位中')117 from .binding import dbc_candidates118 periods = {} if cycle_fields else None119 result = []120 for field in fields:121 # 多定义允许由实际原帧决定；初始化所需的单位不能随候选改变。122 candidates = dbc_candidates(db, field.frame_id, (field.signal,), field.is_extended, field.is_fd)123 units = {signal.unit for message in candidates for signal in message.signals if signal.name == field.signal}124 if len(units) != 1:125 raise ValueError('候选DBC信号单位不同，请补充 is_extended 或 is_fd: ' + field.signal)126 unit = next(iter(units))127 compatible = field.is_fd is False and all(message.is_fd is True for message in candidates)128 if unit is not None and type(unit) is not str:129 raise TypeError('原DBC公开unit需要文本或None')130 item = {'locator': field_to_dict(field), 'unit': unit}131 if field.is_extended is None or field.is_fd is None:132 # 父读取器据此跳过同ID但不含所需信号的帧，不伪造一次失效更新。133 item['dbc_frames'] = [list(pair) for pair in dict.fromkeys(134 (message.is_extended_frame, message.is_fd) for message in candidates)]135 if compatible:136 item['frame_format_compatibility'] = {'code': 'dbc_fd_classic_selector',137 'dbc_is_fd': True, 'requested_is_fd': False}138 if field in cycle_fields:139 # 同一物理报文多个别名只取一次周期，后续记录不再访问DBC周期。140 values = set()141 for message in candidates:142 # 多信号来自同一消息对象时只读一次周期，同ID不同定义不能相互覆盖。143 identity = id(message)144 if identity not in periods:145 periods[identity] = cycle_time_ns(message.cycle_time)146 values.add(periods[identity])147 if len(values) != 1:148 raise ValueError('候选DBC周期不同，请补充帧类型、period 或 valid_for: ' + field.signal)149 item['cycle_ns'] = next(iter(values))150 result.append(item)151 return result152153154def _frame(msg):155 """保存当前消息的公开写出字段，不重建CAN消息或执行解码。156157 参数: msg为共用BLF读取器返回的can.Message。返回: LegacyRawFrame。158 异常: 新边界字段校验失败；data仅无损复制为bytes，不改timestamp/DLC/标志。159 """160 if len(msg.data) > 64:161 raise ValueError('原报文data超过64字节，不能先复制再拒绝')162 return LegacyRawFrame(timestamp=msg.timestamp, arbitration_id=msg.arbitration_id,163 is_extended_id=msg.is_extended_id, is_remote_frame=msg.is_remote_frame,164 is_error_frame=msg.is_error_frame, channel=msg.channel, dlc=msg.dlc, data=bytes(msg.data),165 is_fd=msg.is_fd, is_rx=msg.is_rx, bitrate_switch=msg.bitrate_switch,166 error_state_indicator=msg.error_state_indicator)167168169def _argument_text(value, limit):170 """只摘要原生异常参数，不执行任意对象str或先展开完整巨型消息。171172 参数: value为一个原args值；limit为最大字符数。返回: 有限str，短原str保持内容。173 异常: 无主动异常；未知对象仅标明类型，不调用其自定义字符串方法。174 """175 if type(value) is str:176 return value[:limit]177 if value is None or type(value) in (bool, float):178 return repr(value)[:limit]179 if type(value) is int:180 return str(value)[:limit] if value.bit_length() <= 128 else '<超过128位的整数参数>'181 return '<参数类型:' + type(value).__name__[:64] + '>'182183184def _error_packet(exc, phase, request_id):185 """保存有限原错误事实，不把远端异常对象或任意类传给父进程。186187 参数: exc为本child原错；phase为有限阶段文本；request_id为当前许可或None。188 返回: 有限ERROR dict；message是至多4个args摘要，traceback仅至多8个栈地点，189 均不是完整str(exc)或完整traceback，截断有明确note，短单str参数保持原文本。190 异常: 标准库字段访问/编码错误传播；调用方仍须确定退出，不伪报正常EOF。191 """192 # 不全量format_exception/str(exc)，仅读有限原参数及栈地点，绝不访问locals/gi_frame。193 arguments = exc.args if type(exc.args) is tuple else ()194 message = ', '.join(_argument_text(value, 1024) for value in arguments[:4])195 if not arguments:196 message = '远端异常无可安全摘要的原生args'197 places, current = [], exc.__traceback__198 for _ in range(8):199 if current is None:200 break201 code = current.tb_frame.f_code202 places.append(code.co_filename[:256] + ':' + str(current.tb_lineno) + ' ' + code.co_name[:128])203 current = current.tb_next204 trace = '\n'.join(places)205 original_notes = getattr(exc, '__notes__', ())206 notes = [note[:256] if type(note) is str else '<非文本note>' for note in207 (original_notes[:8] if type(original_notes) in (tuple, list) else ())]208 notes.append('远端message仅原args有限摘要；traceback仅有限栈地点，非完整str/trace。')209 limit = 4096210 while True:211 packet = {'op': 'error', 'request_id': request_id, 'phase': phase,212 'error': {'module': type(exc).__module__[:128], 'type': type(exc).__name__[:128],213 'message': message[:limit], 'traceback': trace[:limit],214 'notes': [note[:min(limit, 256)] for note in notes] + ['远端诊断有界；可能截断。']}}215 try:216 encode_packet(packet)217 return packet218 except ValueError:219 # 每轮减半，固定包头的诊断不受输入长度放大；最短仍明确失败类型。220 if limit <= 64:221 return {'op': 'error', 'request_id': request_id, 'phase': phase,222 'error': {'module': type(exc).__module__[:32], 'type': type(exc).__name__[:32],223 'message': '远端错误诊断超过wire上限', 'traceback': '', 'notes': ['诊断已截断。']}}224 limit //= 2225226227def _report_error(connection, exc, phase, request_id):228 """尽力发送原错误事实，断管不替代child原错或阻止退出。229230 参数: connection为本连接；exc/phase/request_id为原失败事实。返回: None。231 异常: 不传播附加报告错误；原失败仍由worker最终非零退出表达。232 """233 try:234 _send(connection, _error_packet(exc, phase, request_id))235 except BaseException as diagnostic:236 exc.add_note('旧CAN远端错误报告附加失败: ' + type(diagnostic).__name__)237238239def worker_main(connection):240 """作为spawn目标按INIT/PULL协议调用共用BLF读取与串行兼容解码，等待许可时不预取。241242 参数: connection为父传来的独立duplex Connection。返回: 正常STOP或真实EOF时None。243 异常: 原错误尽力发ERROR后以SystemExit(1)退出；连接/生成器附加关闭错误也非零。244 副作用: 本child内一次DBC加载、一次iter_frames、每许可一次next/parse；不额外扫描。245 """246 iterator, failure, request_id = None, None, None247 phase = 'init'248 pulls_received = frames_returned = parse_calls = 0249 try:250 request = _request(_receive(connection))251 phase = 'dependencies'252 can, cantools = _load_can_runtime()253 phase = 'dbc'254 # 共用安全DBC加载函数执行一次加载，不做第二次解析。255 db = load_dbc_file(str(request.dbc_path), cantools)256 phase = 'metadata'257 from .binding import FrameBindingResolver258 from .decode_policy import decode_serial_frame as _parse_selected259 from collections import Counter260 resolver = FrameBindingResolver(db, request.fields)261 duplicate_ids = {frame_id for frame_id, count in Counter(message.frame_id for message in db.messages).items()262 if count > 1}263 phase = 'metadata'264 fields = _field_metadata(db, request.fields)265 iterator = _iter_blf(str(request.input_path))266 versions = {'python': platform.python_version(), 'python-can': metadata.version('python-can'),267 'cantools': metadata.version('cantools'), 'natsort': metadata.version('natsort')}268 _send(connection, {'op': 'ready', 'version': 1, 'profile': PROFILE_ID, 'worker_pid': os.getpid(),269 'fields': fields, 'dependency_versions': versions})270 # 只在收到下一条PULL时取原下一帧；STATUS/ACK/STOP均不授予此许可。271 while True:272 phase, request_id = 'control', None273 packet = _receive(connection)274 operation = packet.get('op')275 if operation == 'stop':276 _exact(packet, ('op',))277 break278 if operation == 'status':279 _exact(packet, ('op',))280 _send(connection, {'op': 'status', 'pulls_received': pulls_received,281 'frames_returned': frames_returned, 'parse_calls': parse_calls})282 continue283 if operation == 'ack':284 expected = ('op', 'request_id') if 'request_id' in packet else ('op',)285 _exact(packet, expected)286 if 'request_id' in packet and (frames_returned == 0 or type(packet['request_id']) is not int287 or packet['request_id'] != frames_returned - 1):288 raise ValueError('ACK只允许确认最近返回ITEM')289 _send(connection, packet)290 continue291 if operation != 'pull':292 raise ValueError('未知旧CAN控制操作')293 _exact(packet, ('op', 'request_id'))294 candidate_id = packet['request_id']295 request_id = candidate_id if type(candidate_id) is int and candidate_id >= 0 else None296 if request_id is None or request_id != frames_returned:297 raise ValueError('PULL许可必须从0严格连续')298 pulls_received += 1299 phase = 'next'300 try:301 msg = next(iterator)302 except StopIteration:303 # EOF需要实际多一次许可，候选发出后仍须父确认本child退出，不能靠断管猜EOF。304 _send(connection, {'op': 'eof', 'request_id': request_id, 'frame_count': frames_returned})305 break306 phase = 'parse'307 parse_calls += 1308 fields_for_frame = resolver.fields_for(msg)309 message = resolver.message_for(msg, fields_for_frame)310 values = _parse_selected(db, msg, message, duplicate_ids)311 phase = 'item'312 record = LegacyDecodedFrame(frames_returned, _frame(msg), values)313 _send(connection, {'op': 'item', 'request_id': request_id, 'record': decoded_to_dict(record)})314 frames_returned += 1315 # 释放本层当前wire/记录引用，不给下一帧任何许可；旧generator内部状态不读取。316 msg = values = record = None317 except BaseException as exc:318 failure = exc319 _report_error(connection, exc, phase, request_id)320 finally:321 # 生成器退出时关闭共用BLF owner；父进程仍须确认实际子退出。322 if iterator is not None:323 try:324 iterator.close()325 except BaseException as cleanup:326 if failure is None:327 failure = cleanup328 _report_error(connection, cleanup, 'generator_close', request_id)329 else:330 failure.add_note('旧生成器关闭附加失败: ' + type(cleanup).__name__)331 try:332 connection.close()333 except BaseException as cleanup:334 if failure is None:335 failure = cleanup336 else:337 failure.add_note('旧CAN子连接关闭附加失败: ' + type(cleanup).__name__)338 if failure is not None:339 raise SystemExit(1)340