chronaxis.legacy.export_worker
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在独立spawn进程按PUT许可将原报文写入新的BLF文件。23有效INIT后核现有依赖版本，BLFWrite独占取得文件并把同一句柄交给python-can。4每PUT恢复原公开字段并写一次，不读取来源或DBC；FINISH仅在最终写入成功后确认。5关闭失败保留原错误与实际文件owner，第三方stop不重复执行，父侧仍核实际子退出。"""67# 模块导入不加载CAN库或打开输出；只有有效INIT之后才加载CAN依赖。8import os9from pathlib import Path10from .records import PROFILE_ID, frame_from_dict11from .worker import _exact, _load_can_runtime, _receive, _report_error, _send121314from ..can_dependencies import runtime_dependency_versions151617def _output(packet):18 """检查唯一INIT及新输出路径，不猜格式或覆盖已有文件。1920 参数: packet为已通过严格JSON边界的dict。返回: resolved新BLF Path。21 异常: TypeError/ValueError为字段、版本、profile、后缀或受保护路径；22 FileExistsError为已有目标，FileNotFoundError为尚无控制器工作根。23 副作用: 仅检查文件系统；真正独占创建仍在writer建立之前执行。24 """25 _exact(packet, ('op', 'version', 'profile', 'path'))26 if (packet['op'] != 'init' or type(packet['version']) is not int or packet['version'] != 127 or type(packet['profile']) is not str or packet['profile'] != PROFILE_ID28 or type(packet['path']) is not str):29 raise ValueError('原CAN写出需要init/version1及固定profile路径')30 path = Path(packet['path']).resolve()31 # 先取得实际目标，再检查后缀和存在性；独占创建仍由写入器执行。32 if path.suffix.lower() != '.blf':33 raise ValueError('原CAN写出仅支持新的BLF文件')34 if path.exists():35 raise FileExistsError(path)36 if not path.parent.is_dir():37 raise FileNotFoundError(path.parent)38 return path394041def _message(can, frame):42 """把全部原公开写出字段交给原can.Message构造，不添加check或归一化参数。4344 参数: can为本次已加载的python-can模块；frame为已校验LegacyRawFrame。45 返回: 新can.Message，timestamp/data/DLC及全部标志保持原参数。46 异常: 原can.Message构造错误；不以新层猜测修正旧格式限制。47 """48 if frame.is_rx is None:49 raise ValueError('BLF写出需要收发方向is_rx')50 return can.Message(timestamp=frame.timestamp, arbitration_id=frame.arbitration_id,51 is_extended_id=frame.is_extended_id, is_remote_frame=frame.is_remote_frame,52 is_error_frame=frame.is_error_frame, channel=frame.channel, dlc=frame.dlc,53 data=frame.data, is_fd=frame.is_fd, is_rx=frame.is_rx,54 bitrate_switch=frame.bitrate_switch, error_state_indicator=frame.error_state_indicator)555657def worker_main(connection):58 """作为真实spawn目标，按INIT/单PUT/FINISH顺序调用唯一BLF writer。5960 参数: connection为父传来的实际duplex端点。返回: 原stop成功并发送候选后None。61 异常: 原错误尽力发送有限ERROR，最终SystemExit(1)；管道清理也不能伪成正常退出。62 副作用: 独占创建一个新BLF、一次BLFWrite、每许可一次on_message_received，63 至多一次stop；不读取输入CAN/DBC，不调parse或为ACK额外写帧。64 """65 writer = failure = request_id = None66 phase, frame_count, stop_attempted = 'init', 0, False67 try:68 path = _output(_receive(connection))69 phase = 'dependencies'70 can, cantools = _load_can_runtime()71 versions = runtime_dependency_versions()72 # 同一独占文件对象交给python-can，不关闭后按路径重新打开。73 phase = 'build_writer'74 from ..results.blf_writer import BLFWrite75 writer = BLFWrite(path)76 _send(connection, {'op': 'ready', 'version': 1, 'profile': PROFILE_ID,77 'worker_pid': os.getpid(), 'dependency_versions': versions})78 # 仅收到一条PUT后写一帧；下一许可前不保留当前消息或构造额外帧。79 while True:80 phase, request_id = 'control', None81 packet = _receive(connection)82 if packet.get('op') == 'finish':83 _exact(packet, ('op', 'frame_count'))84 if type(packet['frame_count']) is not int or packet['frame_count'] != frame_count:85 raise ValueError('FINISH帧数与本次原writer确认数量不符')86 phase, stop_attempted = 'stop', True87 writer.finish()88 writer = None89 _send(connection, {'op': 'finished', 'frame_count': frame_count})90 break91 _exact(packet, ('op', 'request_id', 'frame'))92 candidate = packet['request_id']93 request_id = candidate if type(candidate) is int and candidate >= 0 else None94 if packet['op'] != 'put' or request_id is None or request_id != frame_count:95 raise ValueError('PUT许可必须从0严格连续且为原生int')96 phase = 'frame'97 frame = frame_from_dict(packet['frame'])98 message = _message(can, frame)99 phase = 'write'100 writer.write(message)101 frame_count += 1102 _send(connection, {'op': 'ack', 'request_id': request_id, 'frame_count': frame_count})103 frame = message = None104 except BaseException as exc:105 if writer is None:106 writer=getattr(exc,'blf_write_cleanup_owner',None)107 failure = exc108 _report_error(connection, exc, phase, request_id)109 finally:110 # 失败或父提前断管只结束已有writer；不重发PUT，不重试已经失败的原stop。111 if writer is not None:112 try:113 stop_attempted = True114 writer.close()115 except BaseException as cleanup:116 if failure is None:117 failure = cleanup118 _report_error(connection, cleanup, 'stop', request_id)119 else:120 failure.add_note('原writer.stop附加失败: ' + type(cleanup).__name__)121 for resource, label in ((connection, '子连接'),):122 if resource is not None:123 try:124 resource.close()125 except BaseException as cleanup:126 if failure is None:127 failure = cleanup128 else:129 failure.add_note(label + '关闭附加失败: ' + type(cleanup).__name__)130 if failure is not None:131 raise SystemExit(1)132