chronaxis.legacy.required_records
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""保存按需分析使用的CAN身份和值，允许明确不采集时间。23输入是已匹配原消息的必要身份和所需解码值；输出为冻结模型与有界JSON记录。4时间为None表示没有采集，不能代替零秒。记录不保存载荷、DLC和写出标志，不能5用于原始BLF导出。复用既有严格数值、非有限标签和64KiB限制；无I/O或跨请求状态。6"""78from collections.abc import Mapping9from dataclasses import dataclass10import json11from math import isfinite12from types import MappingProxyType1314from .records import (MAX_WIRE_BYTES, _integer, _shape, _wire_tree, _decoded_values_size,15 _decoded_entry_size, _value_to_wire, _value_from_wire, _unique_object, _reject_constant)161718_FRAME_FIELDS = ('timestamp', 'arbitration_id', 'channel', 'is_extended_id', 'is_fd',19 'is_remote_frame', 'is_error_frame')202122@dataclass(frozen=True)23class RequiredCANFrame:24 """仅持有所需匹配身份，时间None明确表示未采集；模型不具备原始写出能力。"""2526 timestamp: float | None # 有限原float秒，或本任务未采集。27 arbitration_id: int # 原29位以内ID，不含通道。28 channel: int | str | None # 原通道身份，保持原类型，不猜测。29 is_extended_id: bool # 原标准/扩展标志。30 is_fd: bool # 原经典CAN/FD标志。31 is_remote_frame: bool # 匹配时排除远程帧所需。32 is_error_frame: bool # 匹配时排除错误帧所需。3334 def __post_init__(self):35 """检查身份及可选时间。参数: 无。返回: None。异常: 原类型或边界错误。"""36 if self.timestamp is not None and (type(self.timestamp) is not float or not isfinite(self.timestamp)):37 raise ValueError('所需报文timestamp需要有限float秒或None')38 _integer(self.arbitration_id, 'arbitration_id', maximum=0x1fffffff)39 if self.channel is not None and type(self.channel) not in (int, str):40 raise TypeError('所需报文channel需要int/str/None')41 if type(self.channel) is str and len(self.channel) > MAX_WIRE_BYTES:42 raise ValueError('所需报文channel文本过长')43 for name in ('is_extended_id', 'is_fd', 'is_remote_frame', 'is_error_frame'):44 if type(getattr(self, name)) is not bool:45 raise TypeError(name + '需要bool')464748def required_frame_to_dict(frame):49 """输出固定身份字段。参数: frame为RequiredCANFrame。返回: dict。异常: TypeError。"""50 if type(frame) is not RequiredCANFrame:51 raise TypeError('需要RequiredCANFrame')52 return {name: getattr(frame, name) for name in _FRAME_FIELDS}535455@dataclass(frozen=True)56class RequiredDecodedFrame:57 """持有一次目标原序号和冻结所需值；不根据未采集的时间建立时序坐标。"""5859 sequence: int # 原yield序号，可有无关帧形成的间隙。60 frame: RequiredCANFrame # 必要身份与可选时间。61 values: Mapping[str, int | float | bool] # 缺键保持缺失，诊断非有限值保留。6263 def __post_init__(self):64 """先检查上界再复制外部Mapping。参数: 无。返回: None。异常: 类型、编码或超限错误。"""65 _integer(self.sequence, 'sequence', maximum=(1 << 64)-1)66 if type(self.frame) is not RequiredCANFrame or not isinstance(self.values, Mapping):67 raise TypeError('按需记录需要RequiredCANFrame与数值Mapping')68 # 外壳与值共用64KiB上限；复制时继续以这次精确余量防止可变Mapping扩大前缀。69 base = _wire_tree({'op': 'item', 'request_id': self.sequence,70 'record': {'sequence': self.sequence, 'frame': required_frame_to_dict(self.frame), 'values': {}}})71 available = MAX_WIRE_BYTES-base+272 _decoded_values_size(self.values, available)73 copied, used = {}, 274 for name, value in self.values.items():75 # 外部Mapping两次遍历可变化，每个插入仍须验证；常见worker原生字典复用上界。76 separator = 1 if copied else 077 used += separator + _decoded_entry_size(name, value, available-used-separator)78 copied[name] = value79 object.__setattr__(self, 'values', MappingProxyType(copied))808182def required_to_dict(record):83 """编码固定按需记录。参数: record为RequiredDecodedFrame。返回: dict。异常: 类型或值错误。"""84 if type(record) is not RequiredDecodedFrame:85 raise TypeError('需要RequiredDecodedFrame')86 return {'sequence': record.sequence, 'frame': required_frame_to_dict(record.frame),87 'values': {name: _value_to_wire(value) for name, value in record.values.items()}}888990def required_from_dict(value):91 """从独立JSON树恢复按需记录。参数: value为固定dict。返回: 冻结模型。异常: 形状、类型或限额错误。"""92 _shape(value, ('sequence', 'frame', 'values'), 'required decoded frame')93 _wire_tree(value)94 _shape(value['frame'], _FRAME_FIELDS, 'required CAN frame')95 if type(value['values']) is not dict:96 raise TypeError('required decoded values需要dict')97 return RequiredDecodedFrame(value['sequence'], RequiredCANFrame(**value['frame']),98 {name: _value_from_wire(item) for name, item in value['values'].items()})99100101def required_from_packet(payload):102 """严格解析单条按需JSON，固定模型检查代替同一树的重复通用遍历。103104 参数: payload为64KiB内bytes。返回: 独立冻结模型。105 异常: UTF-8、重复键、深度、裸非有限常量或模型错误；不解任意对象标签。106 """107 parsed = values = frame = None108 try:109 if type(payload) is not bytes or not 0 < len(payload) <= MAX_WIRE_BYTES:110 raise ValueError('按需记录需要65536字节内bytes')111 parsed = json.loads(payload.decode('utf-8'), object_pairs_hook=_unique_object,112 parse_constant=_reject_constant)113 _shape(parsed, ('sequence', 'frame', 'values'), 'required decoded frame')114 _shape(parsed['frame'], _FRAME_FIELDS, 'required CAN frame')115 if type(parsed['values']) is not dict:116 raise TypeError('required decoded values需要dict')117 frame = RequiredCANFrame(**parsed['frame'])118 values = {name: _value_from_wire(item) for name, item in parsed['values'].items()}119 # 所有可接受容器都是固定浅层结构；未知嵌套先拒绝，不需要再次遍历JSON树。120 return RequiredDecodedFrame(parsed['sequence'], frame, values)121 finally:122 payload = parsed = values = frame = None123