chronaxis.legacy.export_write
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在独立进程把原始帧交给共用BLF写入器，真实收尾后才确认完成。23LegacyCANProcessWrite接收共享ResourceBudget和新输出路径，构造只计有限状态、4核路径而不spawn或打开输出。start加载依赖并由共用BLFWrite取得文件和写入器；每次5write只发一个有界PUT，收到同序ACK才承认该帧，FINISH没有额外帧写入许可。6finish须BLFWrite完成收尾、实际exitcode0、所有Pipe/Process及自有RAM清完，并核输出7stat不超过本writer直接预留，才公开finished。提前close、失败或取消不能伪完成。89进程/管道/wire/READY/当前帧分别先付并记真实槽；释放前后取消可重试，最高10can_process_write_cleanup_owner始终指向同一实际owner。磁盘在start前保守预留1165536 bytes、每PUT前4096 bytes，累计保留且不倒扣；不凭共享余额差接管费用。12原BLF压缩、库内部缓存及整进程RSS未测，也不承诺文件与源BLF字节一致。13这里只写控制器新工作根下的新文件，不读取来源CAN或DBC。14"""1516# 仅导入标准库和既有轻量协议；CAN依赖留在独立spawn子内。17import multiprocessing18from pathlib import Path19from .export_worker import worker_main20from ..can_dependencies import runtime_dependency_versions21from .process_read import RemoteLegacyError, _freeze, _wire_metadata_bytes22from .records import MAX_WIRE_BYTES, PROFILE_ID, LegacyRawFrame, decode_packet, encode_packet, frame_to_dict23from ..results.budget import ResourceBudget24from ..results.errors import AnalysisError, ResultClosedError25from ..results.spool import _cleanup_owner, _release_owned_ram, _reserve_owned_ram262728class LegacyCANWriteError(AnalysisError):29 """报告新外部writer协议或原子进程失败；远端原错仅作为明确事实cause。"""303132class LegacyCANProcessWrite:33 """拥有一个真实writer子进程与独立费用，借用账本但不关闭其他owner。3435 _waiting阻止第二个在途控制包；_process_stopped只由实际exitcode确认。36 _closed表示自有责任结束，_finished还须本次FINISH及输出大小验证成功。37 输出路径和累计spool事实在close后保留，READY借引用在close时解除。38 """3940 def __init__(self, budget, output_path):41 """预约固定协调状态并检查新路径，不启动进程或创建目标文件。4243 参数: budget为借用ResourceBudget；output_path为控制器已有新工作根下的44 str/Path目标，须为.blf后缀，不能已存在。45 返回: 未启动的独立writer。异常: 类型、路径或预算原错；未返回构造的持续46 清理失败附can_process_write_cleanup_owner，同对象恢复后close。47 """48 if not isinstance(budget, ResourceBudget) or not isinstance(output_path, (str, Path)):49 raise TypeError('原CAN写出需要ResourceBudget和新输出路径')50 # 所有真实资源和交易槽先安装；后续任何预算取消都有本owner可达责任。51 self._budget, self._output_path = budget, None52 self._connection = self._child_connection = self._process = None53 self._worker_pid = self._exit_code = self._metadata = self._pending_reply = self._frame = None54 self._fixed_cost = self._wire_cost = self._pipe_cost = self._process_cost = 055 self._metadata_cost = self._frame_cost = self._held_cost = self._spool_bytes = 056 self._puts_issued = self._acks_received = self._frames_written = 057 self._actual_bytes = None58 self._process_stopped = self._waiting = self._started = self._finish_received = False59 self._closed = self._closing = self._failed = self._finished = False60 try:61 _reserve_owned_ram(self, 1048576, '_fixed_cost')62 if len(str(output_path)) > 8192:63 raise ValueError('原CAN输出路径超过本层有限能力')64 self._output_path = Path(output_path).resolve()65 self._check_path()66 except BaseException as exc:67 self._failed = True68 _cleanup_owner(self, exc, 'can_process_write_cleanup_owner', '原CAN writer构造')69 raise7071 def _check_path(self):72 """拒绝已有目标及缺少的新工作根，不创建目录或文件。7374 参数: 无，使用已保存resolved路径。返回: None。75 异常: ValueError为后缀错误，FileExistsError为已有目标，76 FileNotFoundError为父目录不存在；真正独占创建由child执行。77 """78 path = self._output_path79 if path.suffix.lower() != '.blf':80 raise ValueError('原CAN写出仅支持新的BLF文件')81 if path.exists():82 raise FileExistsError(path)83 if not path.parent.is_dir():84 raise FileNotFoundError(path.parent)8586 @property87 def output_path(self):88 """读取实际新目标。参数: 无。返回: resolved Path或构造前失败时None。异常: 无。"""89 return self._output_path9091 @property92 def closed(self):93 """读取全部自有责任关闭状态。参数: 无。返回: bool。异常: 无。"""94 return self._closed9596 @property97 def finished(self):98 """读取完整FINISH事实。参数: 无。返回: bool，失败/提前close为False。异常: 无。"""99 return self._finished100101 @property102 def failed(self):103 """读取失败诊断状态。参数: 无。返回: bool。异常: 无。"""104 return self._failed105106 @property107 def worker_pid(self):108 """读取本次真实启动PID。参数: 无。返回: int或None，关闭后保留。异常: 无。"""109 return self._worker_pid110111 @property112 def spool_bytes(self):113 """读取本writer直接累计磁盘预留。参数: 无。返回: int bytes，不倒扣。异常: 无。"""114 return self._spool_bytes115116 @property117 def counters(self):118 """区分PUT尝试、合法ACK和本父确认帧数，不用断管推断写入成功。119120 参数: 无。返回: 新dict，取消时puts_issued仅为许可尝试上界。121 异常: 原小容器分配错误；不向child发送额外控制或写入。122 """123 return {'puts_issued': self._puts_issued, 'acks_received': self._acks_received,124 'frames_written': self._frames_written}125126 @property127 def metadata(self):128 """借读同次READY事实，不隐式start。参数: 无。返回: 冻结映射，借用至close。129 异常: ResultClosedError为关闭/失败，RuntimeError为尚无实际READY。130 """131 self._require_open()132 if self._metadata is None:133 raise RuntimeError('原CAN writer尚未READY')134 return self._metadata135136 def _require_open(self):137 """阻止关闭或失败后继续许可。参数: 无。返回: None。138 异常: ResultClosedError为不可推进；close仍可重复恢复。139 """140 if self._closed or self._closing or self._failed:141 raise ResultClosedError('原CAN writer已关闭、清理中或失败')142143 def _reserve_spool(self, amount):144 """逐笔记录本次直接磁盘预留，回调执行后取消也保留真实归属。145146 参数: amount为非负int bytes。返回: None。异常: 原配额/取消，文件及累计值不倒扣。147 """148 before = self._budget.spool_bytes149 try:150 self._budget.reserve_spool(amount)151 finally:152 if self._budget.spool_bytes == before + amount:153 self._spool_bytes += amount154155 def _fail(self, exc, phase):156 """以实际父错误为主收尾，普通底层错误包装并保cause及最高恢复owner。157158 参数: exc为原异常；phase为有限业务位置。返回: 不返回。159 异常: KI/SystemExit/AnalysisError原对象，其余LegacyCANWriteError保原cause；160 两次有限cleanup副错只附note，绝不伪称写出完成。161 """162 self._failed = True163 failure = exc if isinstance(exc, (KeyboardInterrupt, SystemExit, AnalysisError)) else LegacyCANWriteError(164 '原CAN writer' + phase + '失败')165 _cleanup_owner(self, failure, 'can_process_write_cleanup_owner', '原CAN writer')166 if failure is exc:167 raise failure168 raise failure from exc169170 def start(self):171 """实际创建独立spawn及原writer，READY不包含任何帧许可。172173 参数: 无。返回: self，重复成功调用不重建或重复预留。174 异常: 路径/依赖/管道/协议/配额/预算或父原取消，最高owner按实际槽可恢复。175 """176 self._require_open()177 if self._started:178 return self179 try:180 self._check_path()181 # 磁盘容量在spawn/原build_writer前预留，空BLF也保65536而非事后认领。182 self._reserve_spool(65536)183 _reserve_owned_ram(self, 128 * MAX_WIRE_BYTES, '_wire_cost')184 _reserve_owned_ram(self, 65536, '_pipe_cost')185 context = multiprocessing.get_context('spawn')186 self._connection, self._child_connection = context.Pipe(duplex=True)187 _reserve_owned_ram(self, 65536, '_process_cost')188 self._process = context.Process(target=worker_main, args=(self._child_connection,),189 name='chronaxis-legacy-can-export')190 try:191 self._process.start()192 finally:193 self._worker_pid = self._process.pid194 self._close_connection('_child_connection')195 reply = self._exchange({'op': 'init', 'version': 1, 'profile': PROFILE_ID,196 'path': str(self._output_path)})197 self._check_ready(reply)198 _reserve_owned_ram(self, 4096 + 16 * _wire_metadata_bytes(reply), '_metadata_cost')199 self._metadata = _freeze(reply)200 self._pending_reply = None201 self._started = True202 return self203 except BaseException as exc:204 self._fail(exc, '初始化')205206 def _check_ready(self, packet):207 """核同次实际PID及现有依赖，不接受bool伪版本或自称READY。208209 参数: packet为decode_packet得到的dict。返回: None。210 异常: LegacyCANWriteError为结构/类型/profile/PID/版本/摘要不符。211 """212 if (set(packet) != {'op', 'version', 'profile', 'worker_pid', 'dependency_versions'}213 or packet.get('op') != 'ready' or type(packet.get('version')) is not int or packet['version'] != 1214 or type(packet.get('profile')) is not str or packet['profile'] != PROFILE_ID215 or type(packet.get('worker_pid')) is not int or packet['worker_pid'] != self._worker_pid216 or packet['dependency_versions'] != runtime_dependency_versions()):217 raise LegacyCANWriteError('原CAN writer READY身份、版本或实际PID不符')218219 def _exchange(self, packet):220 """同步交换唯一控制包，禁止第二个在途PUT或把子退出当成ACK。221222 参数: packet为有界控制dict。返回: 严格JSON dict。223 异常: 父原取消/管道/编码错误或LegacyCANWriteError；远端失败以明确224 RemoteLegacyError事实为cause，不能声称跨进程保留异常对象身份。225 """226 if self._connection is None or self._waiting:227 raise LegacyCANWriteError('原CAN writer连接不可用或已有在途许可')228 payload = encode_packet(packet)229 self._waiting = True230 self._connection.send_bytes(payload)231 while not self._connection.poll(0.05):232 if self._process is not None and self._process.exitcode is not None:233 raise LegacyCANWriteError('原CAN writer未交付完整回应便退出')234 self._pending_reply = self._connection.recv_bytes(MAX_WIRE_BYTES)235 reply = decode_packet(self._pending_reply)236 self._waiting = False237 if reply.get('op') == 'error':238 facts = reply.get('error')239 if (set(reply) != {'op', 'request_id', 'phase', 'error'}240 or type(reply.get('phase')) is not str241 or reply.get('request_id') is not None and type(reply.get('request_id')) is not int242 or reply.get('request_id') != packet.get('request_id')243 or type(facts) is not dict or set(facts) != {'module', 'type', 'message', 'traceback', 'notes'}244 or any(type(facts[key]) is not str for key in ('module', 'type', 'message', 'traceback'))245 or type(facts['notes']) is not list or any(type(note) is not str for note in facts['notes'])):246 raise LegacyCANWriteError('原CAN writer远端错误包缺少明确事实或请求不符')247 remote = RemoteLegacyError(facts, reply['phase'], reply['request_id'])248 raise LegacyCANWriteError('原CAN writer子进程失败') from remote249 return reply250251 def write(self, frame):252 """完整发送一个原始帧并等唯一ACK，不缓存后继或按ID筛帧。253254 参数: frame为原生LegacyRawFrame，保全部公开字段及原浮点timestamp。255 返回: None，合法ACK后counters增加；无额外on_message_received。256 异常: TypeError为非原模型；原预算/磁盘/取消或writer/协议失败，之后只能close。257 """258 self._require_open()259 if type(frame) is not LegacyRawFrame:260 raise TypeError('原CAN writer.write需要LegacyRawFrame')261 if frame.is_rx is None:262 raise ValueError('BLF写出需要收发方向is_rx')263 try:264 self.start()265 # 当前记录、hex和wire容器均在复制前付款；只有本owner持有时才保留该槽。266 _reserve_owned_ram(self, 4259840, '_frame_cost')267 self._frame = frame268 self._reserve_spool(4096)269 request_id = self._frames_written270 self._puts_issued += 1271 reply = self._exchange({'op': 'put', 'request_id': request_id, 'frame': frame_to_dict(frame)})272 if (set(reply) != {'op', 'request_id', 'frame_count'} or reply.get('op') != 'ack'273 or type(reply.get('request_id')) is not int or reply['request_id'] != request_id274 or type(reply.get('frame_count')) is not int or reply['frame_count'] != request_id + 1):275 raise LegacyCANWriteError('原CAN writer ACK不属于本次唯一PUT')276 self._acks_received += 1277 self._frames_written += 1278 self._frame = self._pending_reply = None279 _release_owned_ram(self, self._frame_cost, '_frame_cost')280 except BaseException as exc:281 self._fail(exc, '逐帧写出')282283 def finish(self):284 """完成原stop、真实正常退出及全部费用，再核实际文件未超过预留。285286 参数: 无。返回: self，成功后重复调用不重发FINISH或重写文件。287 异常: 原取消/预算/关闭错误，或LegacyCANWriteError为不完整回应、异常退出288 或实际stat超本writer累计预留；任何失败都不置finished。289 """290 if self._finished:291 return self292 self._require_open()293 try:294 self.start()295 reply = self._exchange({'op': 'finish', 'frame_count': self._frames_written})296 if (set(reply) != {'op', 'frame_count'} or reply.get('op') != 'finished'297 or type(reply.get('frame_count')) is not int or reply['frame_count'] != self._frames_written):298 raise LegacyCANWriteError('原CAN writer FINISH未确认全部原帧')299 self._finish_received = True300 self.close()301 if self._exit_code != 0:302 raise LegacyCANWriteError('原CAN writer虽发完成候选但未正常退出')303 self._actual_bytes = self._output_path.stat().st_size304 if self._actual_bytes > self._spool_bytes:305 raise LegacyCANWriteError('原BLF实际大小超过本writer保守磁盘预留')306 self._finished = True307 return self308 except BaseException as exc:309 self._fail(exc, '封口')310311 def _close_connection(self, slot):312 """实际关闭端点后才撤槽。参数: slot为本owner自有端点名。返回: None。313 异常: 原close/取消；执行后异常也按真实closed确认，不丢未关句柄。314 """315 connection = getattr(self, slot)316 if connection is not None:317 try:318 connection.close()319 finally:320 if connection.closed:321 setattr(self, slot, None)322323 def _close_process(self):324 """有限等待自己实际进程退出再关句柄，绝不以终止请求代替退出证明。325326 参数: 无。返回: None。异常: 原join/terminate/kill/close或LegacyCANWriteError；327 未确认时保原进程/费用并附稳定退出诊断。显式join上限为.5/.5/5秒，328 Windows kill仍是TerminateProcess，必须实际exitcode非None。329 """330 process = self._process331 if process is None:332 return333 if not self._process_stopped:334 if process.pid is not None:335 process.join(timeout=.5)336 if process.exitcode is None:337 process.terminate()338 process.join(timeout=.5)339 if process.exitcode is None:340 process.kill()341 process.join(timeout=5.0)342 code = process.exitcode343 if code is None:344 failure = LegacyCANWriteError('尚未确认原CAN writer子进程退出')345 failure.can_process_write_exit_state = {'pid': self._worker_pid, 'exit_code': None,346 'join_wait_limits_seconds': (.5, .5, 5.0)}347 raise failure348 self._exit_code = code349 self._process_stopped = True350 # stopped先存，再实际close；close之后取消也不会再次读取已关闭Process属性。351 process.close()352 self._process = None353354 def close(self):355 """只关本writer实际IPC/进程和费用，产物与累计磁盘预留保留。356357 参数: 无。返回: None，完成后重复调用幂等。358 异常: 原关闭/预算/取消附can_process_write_cleanup_owner，当前槽供持续恢复；359 提前关闭不发送额外帧或FINISH，不设置finished，不修改借用预算的其他费用。360 """361 if self._closed:362 return363 self._closing = True364 try:365 self._close_connection('_child_connection')366 self._close_connection('_connection')367 self._close_process()368 self._metadata = self._frame = self._pending_reply = None369 self._waiting = False370 for slot in ('_frame_cost', '_metadata_cost', '_wire_cost', '_pipe_cost', '_process_cost', '_fixed_cost'):371 _release_owned_ram(self, getattr(self, slot), slot)372 self._closed = True373 except BaseException as exc:374 self._failed = True375 exc.can_process_write_cleanup_owner = self376 raise377