chronaxis.execution.history_records
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""定义已解码历史记录与严格JSONL格式，不再次归一化数值。23样本同时保存物理数组值、逻辑ScalarValue和可选原诊断，时间/序号保持整数。4进度记录保留空批、watermark和整个来源EOF；None观测终点不补齐。5编码复用结果层固定白名单标签，不注册动态类或允许任意对象反序列化。6格式只用于新执行层历史，不修改分析计划schema或旧读写格式。7"""89# 标准库冻结模型及严格JSON；共享既有标量验证与固定类型编码。10from dataclasses import dataclass11from math import isfinite12import json13from sys import getsizeof14from ..core.models import SourceRef, SignalSample, require_identifier, require_scalar15from ..sources.csv_spec import CSVSourceRef16from ..sources.can_spec import CANSourceRef17from ..sources.text_can_ref import CANTextSourceRef18from ..core.numeric import ScalarValue19from .record_sample import RecordSample20from ..timebase.conversion import require_int64_ns21from ..core.types import SignalDType, QualityFlag22from ..results.codec import _pack, _unpack, _unique_object, _reject_constant, _estimate23from ..results.errors import ResultFormatError242526@dataclass(frozen=True)27class HistorySample:28 """保存单个已归一化观测；array_value不被用来推断有效性。2930value携带逻辑值、类型、掩码和质量，ENUM正常值是标签文本；array_value保留整数编码。31original_present区分没有原诊断列与该列中的真实None，原诊断允许非有限浮点。32batch_index只标识传输批次，不作为事件身份或绑定身份。33"""3435 batch_index: int36 name: str37 time_ns: int38 sequence: int39 array_value: int | float | bool40 value: ScalarValue41 unit: str42 source_ref: SourceRef43 original_present: bool44 original_value: int | float | bool | str | None4546 def __post_init__(self) -> None:47 """验证物理/逻辑值、身份和原诊断，不做scale或offset转换。4849 参数: 无；读取冻结字段。返回: None。50 异常: TypeError/ValueError/OverflowError为记录矛盾或非法标量。51 """52 # 批次号为非负原生整数，不能用bool或任意精度增长隐藏开销。53 if type(self.batch_index) is not int or not 0 <= self.batch_index < 2 ** 63:54 raise ValueError('历史batch_index需要非负int64')55 if not isinstance(self.value, ScalarValue) or type(self.unit) is not str:56 raise TypeError('历史需要ScalarValue和字符串单位')57 # 核心样本检查时间、别名、原序号、质量与来源；不依赖某一数值后端。58 SignalSample(self.name, self.time_ns, self.value.value, self.value.valid,59 self.value.quality, self.source_ref, self.sequence)60 expected = {SignalDType.INT64: int, SignalDType.UINT64: int,61 SignalDType.FLOAT64: float, SignalDType.BOOL: bool, SignalDType.ENUM: int}[self.value.dtype]62 if type(self.array_value) is not expected or (type(self.array_value) is float and not isfinite(self.array_value)):63 raise TypeError('历史物理值必须符合固定数组类型且有限')64 # 数组即使无效也具有合法物理类型；无效占位仍由valid/quality明确解释。65 physical_dtype = SignalDType.INT64 if self.value.dtype is SignalDType.ENUM else self.value.dtype66 ScalarValue(self.array_value, physical_dtype)67 if self.value.valid:68 logical = self.value.labels[self.array_value] if self.value.dtype is SignalDType.ENUM and 0 <= self.array_value < len(self.value.labels) else self.array_value69 if type(logical) is not type(self.value.value) or logical != self.value.value:70 raise ValueError('历史物理值与正常逻辑值矛盾')71 # 原诊断存在性必须独立，不能把None强制解释为缺列。72 if type(self.original_present) is not bool:73 raise TypeError('original_present需要bool')74 require_scalar(self.original_value)75 if not self.original_present and self.original_value is not None:76 raise ValueError('缺原诊断列时不能附带诊断值')777879@dataclass(frozen=True)80class HistoryProgress:81 """保存一批结束时的读取事实，sample_count包含该批所有信号列。"""8283 batch_index: int84 source: str85 session_id: str86 sample_count: int87 watermark_ns: int | None88 source_eof: bool89 observed_end_ns: int | None9091 def __post_init__(self) -> None:92 """校验计数、身份和可未知的观测/进度坐标。9394 参数: 无；读取全部字段。返回: None。95 异常: TypeError/ValueError/OverflowError为记录结构或时间错误。96 """97 # 空批合法，计数不能为负，也不能把bool当成1。98 for count in (self.batch_index, self.sample_count):99 if type(count) is not int or not 0 <= count < 2 ** 63:100 raise ValueError('历史计数需要非负int64')101 require_identifier(self.source, 'history.source')102 require_identifier(self.session_id, 'history.session_id')103 if type(self.source_eof) is not bool:104 raise TypeError('source_eof需要bool')105 for time in (self.watermark_ns, self.observed_end_ns):106 if time is not None:107 require_int64_ns(time)108109110HistoryRecord = HistorySample | HistoryProgress111112113def record_values(record: HistoryRecord) -> tuple:114 """按固定字段顺序提取白名单值，不将dataclass变成动态对象格式。115116 参数: record为两种历史记录之一。返回: 字段tuple。117 异常: TypeError为未支持记录。118 """119 # 类型必须精确，不支持子类扩展隐式改变格式。120 if type(record) is HistorySample:121 # CSV位置使用历史层固定标签，基础来源仍沿用v1；不扩大结果模型注册表。122 reference = record.source_ref123 if type(reference) is CSVSourceRef:124 reference = ('csv_source_ref', SourceRef(reference.session_id, reference.source_name, reference.field),125 reference.path, reference.line_number, reference.time_text,126 reference.value_text, reference.time_quantized)127 elif type(reference) is CANSourceRef:128 # CAN v3使用固定原帧位置；不把原yield序号误当文件物理行号。129 reference = ('can_source_ref', SourceRef(reference.session_id, reference.source_name, reference.field),130 reference.path, reference.profile, reference.channel,131 reference.signal, reference.frame_id, reference.is_extended, reference.is_fd,132 reference.yield_sequence, reference.timestamp, reference.time_quantized)133 elif type(reference) is CANTextSourceRef:134 # 新文本引用单列v4，物理行与原时间不可降级成旧浮点秒定位。135 reference = ('text_can_source_ref', SourceRef(reference.session_id, reference.source_name, reference.field),136 reference.path, reference.parser_version, reference.line_number,137 reference.input_index, reference.record_sequence, reference.yield_sequence,138 reference.raw_time, reference.time_quantized)139 return (record.batch_index, record.name, record.time_ns, record.sequence,140 record.array_value, record.value, record.unit, reference,141 record.original_present, record.original_value)142 if type(record) is HistoryProgress:143 return (record.batch_index, record.source, record.session_id, record.sample_count,144 record.watermark_ns, record.source_eof, record.observed_end_ns)145 raise TypeError('未知历史记录类型')146147148# 这些是固定结构的费用，不包含任何当前样本值或来源文本。149# 规则与codec._estimate逐字段相同：tuple每槽32，模型每字段256，原生叶子另加64。150_RECORD_STRUCTURE_BYTES = getsizeof((None,) * 7) + 7 * 32 + 5 * 64151_SCALAR_STRUCTURE_BYTES = 5 * 256 + 2 * 512 + 64 + getsizeof(True) + 64 + getsizeof(())152_SOURCE_STRUCTURE_BYTES = 3 * (256 + 64)153154155def estimate_record_bytes(record: HistoryRecord) -> int:156 """保守估算一行装箱、JSON结构和UTF8副本的临时RAM。157158 参数: record为已检查记录。返回: bytes估算，非全进程RSS。159 异常: 类型或深度错误传播。160 """161 # 系数覆盖Python容器、unicode及编码副本，长文本不按固定小行估计。162 if type(record) is RecordSample:163 value, ref = record.value, record.source_ref164 # 现役普通记录只持基本来源和普通标量；固定布局无需重复dataclass反射。165 # 任何扩展引用、枚举域/原码或非原生字段仍沿通用估算，保持类型及深度错误。166 if (type(value) is ScalarValue and type(ref) is SourceRef167 and type(value.dtype) is SignalDType and type(value.quality) is QualityFlag168 and type(value.valid) is bool and type(value.partial) is bool169 and type(value.labels) is tuple and not value.labels and value.code is None170 and type(value.value) in (int, float, bool, str, type(None))171 and type(record.name) is str and type(record.unit) is str172 and type(record.time_ns) is int and type(record.sequence) is int173 and type(record.enum_code) in (int, type(None))174 and type(ref.session_id) is str and type(ref.source_name) is str and type(ref.field) is str):175 # 文本和整数仍按实际对象大小收费，包括长诊断值与大序号；不缓存整行单价。176 scalar = getsizeof(value) + _SCALAR_STRUCTURE_BYTES + getsizeof(value.value)177 if value.partial:178 scalar += 256 + getsizeof(value.partial) + 64179 source = (getsizeof(ref) + _SOURCE_STRUCTURE_BYTES + getsizeof(ref.session_id)180 + getsizeof(ref.source_name) + getsizeof(ref.field))181 return (_RECORD_STRUCTURE_BYTES + getsizeof(record.name) + getsizeof(record.time_ns)182 + getsizeof(record.sequence) + getsizeof(record.unit) + getsizeof(record.enum_code)183 + scalar + source) * 16 + 4096184 return _estimate((record.name,record.time_ns,record.sequence,record.value,record.unit,record.source_ref,record.enum_code))*16+4096185 return _estimate(record_values(record)) * 16 + 8192186187188def encode_record(record: HistoryRecord, max_bytes: int) -> bytes:189 """编码一条历史JSONL；基础v1、信号CSV v2、旧CAN v3、文本CAN v4。190191 参数: record为typed记录，max_bytes为单行上限。返回: 含换行UTF8 bytes。192 异常: ResultFormatError为超长/不支持结构。193 """194 # 固定record_type与版本，复用封闭模型标签；非有限诊断使用明确标签。195 reference_type = type(record.source_ref) if type(record) is HistorySample else None196 version = (4 if reference_type is CANTextSourceRef else 3 if reference_type is CANSourceRef197 else 2 if reference_type is CSVSourceRef else 1)198 packed = {'history_version': version, 'record_type': 'sample' if type(record) is HistorySample else 'progress',199 'values': _pack(record_values(record))}200 payload = (json.dumps(packed, ensure_ascii=False, allow_nan=False, separators=(',', ':'), sort_keys=True) + '\n').encode('utf-8')201 if len(payload) > max_bytes:202 raise ResultFormatError('历史单行超过有限读取上限')203 return payload204205206def decode_record(payload: bytes, max_bytes: int) -> HistoryRecord:207 """严格重建单行typed记录，拒绝重复键、额外字段和损坏换行。208209 参数: payload为含换行UTF8行，max_bytes为上限。返回: HistoryRecord。210 异常: ResultFormatError并保留原格式/模型异常链。211 """212 # readline由调用侧限长；这里再核对完整边界，不接收最后的半行。213 if len(payload) > max_bytes or not payload.endswith(b'\n'):214 raise ResultFormatError('历史行过长或未完整写入')215 try:216 packed = json.loads(payload, object_pairs_hook=_unique_object, parse_constant=_reject_constant)217 if type(packed) is not dict or set(packed) != {'history_version', 'record_type', 'values'} or type(packed['history_version']) is not int or packed['history_version'] not in (1, 2, 3, 4):218 raise ResultFormatError('历史版本或字段错误')219 values = _unpack(packed['values'])220 cls = {'sample': HistorySample, 'progress': HistoryProgress}.get(packed['record_type'])221 if cls is None or type(values) is not tuple:222 raise ResultFormatError('历史record_type或值容器错误')223 if packed['history_version'] == 2:224 # v2只能是明确CSV样本，未知标签/额外定位字段不能降级成基础来源。225 if cls is not HistorySample or len(values) != 10 or type(values[7]) is not tuple or len(values[7]) != 7 or values[7][0] != 'csv_source_ref' or type(values[7][1]) is not SourceRef:226 raise ResultFormatError('历史v2需要完整CSV固定定位')227 base = values[7][1]228 reference = CSVSourceRef(base.session_id, base.source_name, base.field, *values[7][2:])229 values = values[:7] + (reference,) + values[8:]230 elif packed['history_version'] == 3:231 # 固定tag、长度和基础身份齐全才恢复CAN，禁止退化或混用CSV标签。232 if (cls is not HistorySample or len(values) != 10 or type(values[7]) is not tuple233 or len(values[7]) != 12 or values[7][0] != 'can_source_ref'234 or type(values[7][1]) is not SourceRef):235 raise ResultFormatError('历史v3需要完整CAN固定定位')236 base = values[7][1]237 reference = CANSourceRef(base.session_id, base.source_name, base.field, *values[7][2:])238 values = values[:7] + (reference,) + values[8:]239 elif packed['history_version'] == 4:240 # 精确固定字段重建，禁止旧CAN/CSV引用换版本冒充文本物理行。241 if (cls is not HistorySample or len(values) != 10 or type(values[7]) is not tuple242 or len(values[7]) != 10 or values[7][0] != 'text_can_source_ref'243 or type(values[7][1]) is not SourceRef):244 raise ResultFormatError('历史v4需要完整文本CAN定位')245 base = values[7][1]246 reference = CANTextSourceRef(base.session_id, base.source_name, base.field, *values[7][2:])247 values = values[:7] + (reference,) + values[8:]248 elif cls is HistorySample and len(values) == 10 and type(values[7]) in (CANSourceRef,CANTextSourceRef):249 # 结果模型注册不能为历史v1另开CAN入口；CAN仅接受上述v3固定定位。250 raise ResultFormatError('CAN历史定位必须使用对应v3/v4固定格式')251 return cls(*values)252 except (TypeError, ValueError, OverflowError, UnicodeError, KeyError, IndexError) as exc:253 # 不把坏行当空批跳过，保留具体解析原因。254 raise ResultFormatError('历史记录结构/模型非法') from exc255