chronaxis.results.codec
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""编码和严格解码当前扁平结果JSONL；内部叶子格式另用于历史和导出。23公开结果只登记当前行类型和编号。内部_pack/_unpack保留整数、tuple、枚举、4来源与非有限诊断标记；它们不提供旧结果包装或公开v1—v6读取。5最大行字节与递归深度显式限制；调用方须先预留编码与解析临时空间。"""67# 只访问登记dataclass字段，不开放表达式对象访问或反射导入。8from dataclasses import fields9from enum import Enum10import json11from math import isfinite12from sys import getsizeof13from ..core.models import Interval, SourceRef14from ..core.numeric import ScalarValue15from ..core.calculations import CalculationResult, CalculationDefinition16from ..core.selection import SelectionRow17from ..core.temporal import Event, TemporalRule18from ..core.sequence import SequenceRule, SequenceResult19from ..core.windows import Segment, WindowSpec20from ..core.types import QualityFlag, SignalDType, Truth21# CSV实际定位需要完整具体字段；只登记冻结协议，不打开来源或引入reader。22from ..sources.csv_spec import CSVSourceRef23# CAN引用只导入冻结位置模型，不加载原reader/DBC或产生额外扫描。24from ..sources.can_spec import CANSourceRef25from ..sources.text_can_ref import CANTextSourceRef26from .rows import ResultRow, row_table27from .errors import ResultFormatError28from .summary import SummaryCalculationRow, SUMMARY_FIELDS, summary_fields, SUMMARY_MODELS29# 行类型各有固定编号；只接受当前登记的格式，不提供旧版本解码。30_SUMMARY_VERSIONS = {version:(model,names) for model,(_,names,version) in SUMMARY_MODELS.items()}31_MODELS = {cls.__name__: cls for cls in (Interval, SourceRef, CSVSourceRef, CANSourceRef, CANTextSourceRef, ScalarValue, CalculationDefinition,32 CalculationResult, SelectionRow, Event, TemporalRule, Segment, WindowSpec,33 SequenceRule, SequenceResult)}34_ENUMS = {cls.__name__: cls for cls in (QualityFlag, SignalDType, Truth)}35def _pack(value, depth: int = 0):36 """递归编码固定模型、枚举、tuple和原生标量。3738 参数: value为白名单树，depth为递归深度。返回: 仅JSON结构的树。39 异常: ResultFormatError表示类型或深度不受支持。40 """41 # 深度限制避免异常嵌套占用无界栈；序列长度由调用侧RAM预估限制。42 try:43 if depth > 32:44 raise ResultFormatError('结果行嵌套超过32层')45 value_type = type(value)46 # 原生叶子直接处理；无效数值仍保留诊断标签，不能变成普通JSON非有限值。47 if value_type is float and not isfinite(value):48 return {'diagnostic_float': 'nan' if value != value else 'inf' if value > 0 else '-inf'}49 if value is None or value_type in (str, int, bool, float):50 return value51 if isinstance(value, Enum) and type(value).__name__ in _ENUMS and type(value) is _ENUMS[type(value).__name__]:52 return {'enum': type(value).__name__, 'value': value.value}53 if type(value).__name__ in _MODELS and type(value) is _MODELS[type(value).__name__]:54 packed = {}55 for field in fields(value):56 if value_type is ScalarValue and field.name == 'code' and value.code is None:57 continue # 未请求的码视图不改变原始记录格式。58 if value_type is ScalarValue and field.name == 'partial' and not value.partial:59 continue # 完整标量仍写原格式，只有部分估算需要额外标记。60 if field.init:61 packed[field.name] = _pack(getattr(value, field.name), depth + 1)62 return {'model': type(value).__name__, 'fields': packed}63 if type(value) is tuple:64 packed = []65 for item in value:66 packed.append(_pack(item, depth + 1))67 return {'tuple': packed}68 raise ResultFormatError(f'结果行含未注册类型{type(value).__name__}')69 finally:70 # 显式循环不留下comprehension迭代器；失败TB只断本层参数/新容器。71 value = packed = items = kwargs = field = key = item = None72 value_type = None737475def _unpack(value, depth: int = 0):76 """只从已登记标签重建冻结对象。7778 参数: value为解析结构，depth为深度。返回: typed值。79 异常: ResultFormatError/TypeError/ValueError为结构或模型校验错误。80 """81 # 未标记数组/对象不得绕过tuple和模型校验。82 try:83 if depth > 32:84 raise ResultFormatError('结果行嵌套超过32层')85 # JSON的1e999也会被解析为Inf，不能绕过明确诊断标签。86 if type(value) is float and not isfinite(value):87 raise ResultFormatError('非有限结果必须使用诊断标签')88 if value is None or type(value) in (str, int, bool, float):89 return value90 if type(value) is not dict:91 raise ResultFormatError('结果结构需要明确类型标签')92 if set(value) == {'tuple'} and type(value['tuple']) is list:93 items = []94 for item in value['tuple']:95 items.append(_unpack(item, depth + 1))96 return tuple(items)97 if set(value) == {'diagnostic_float'} and value['diagnostic_float'] in ('nan', 'inf', '-inf'):98 return float(value['diagnostic_float'])99 if set(value) == {'enum', 'value'} and value['enum'] in _ENUMS:100 # enum原始值不能用bool冒充质量位整数。101 expected = int if value['enum'] == 'QualityFlag' else str102 if type(value['value']) is not expected:103 raise ResultFormatError('枚举底层值类型错误')104 return _ENUMS[value['enum']](value['value'])105 if set(value) == {'model', 'fields'} and value['model'] in _MODELS and type(value['fields']) is dict:106 cls = _MODELS[value['model']]107 expected = {field.name for field in fields(cls) if field.init}108 if cls is ScalarValue and 'code' not in value['fields']:109 expected.discard('code')110 if cls is ScalarValue and 'partial' not in value['fields']:111 expected.discard('partial')112 if set(value['fields']) != expected:113 raise ResultFormatError('结果模型字段缺失或出现未知字段')114 kwargs = {}115 for key, item in value['fields'].items():116 kwargs[key] = _unpack(item, depth + 1)117 return cls(**kwargs)118 raise ResultFormatError('未知或多余结果类型标签')119 finally:120 # 显式循环不留下comprehension迭代器；失败TB只断本层参数/新容器。121 value = packed = items = kwargs = field = key = item = None122123124def _unique_object(pairs):125 """拒绝重复JSON键，避免解析时后值覆盖前值。126127 参数: pairs为JSON对象键值对。返回: dict。128 异常: ResultFormatError为重复字段。129 """130 # 精确键检查覆盖外层版本和所有嵌套对象。131 try:132 result = {}133 for key, value in pairs:134 if key in result:135 raise ResultFormatError(f'结果JSON重复键{key}')136 result[key] = value137 return result138 finally:139 # 显式循环不留下comprehension迭代器；失败TB只断本层参数/新容器。140 pairs = result = key = value = None141142143144145146147148149150151def _reject_constant(value):152 """拒绝非标准JSON NaN/Infinity，诊断只能使用明确标签。153154 参数: value为解析器识别的常量名。返回: 无。155 异常: ResultFormatError始终抛出。156 """157 raise ResultFormatError(f'结果JSON含非法数值{value}')158159160def estimate_row_bytes(row: ResultRow) -> int:161 """保守估计结果树、JSON转换与编码期间的新层RAM开销。162163 参数: row为冻结行。返回: 正整数bytes估算，含容器/标签和字符串转义余量。164 异常: ResultFormatError为未登记类型或过深结构；不测量全进程RSS。165 """166 # 先确认根类型；递归只观察既有输入，不复制完整树。167 row_table(row)168 if type(row) is SummaryCalculationRow:169 return (getsizeof(row) + sum(256 + getsizeof(getattr(row, name)) for name in SUMMARY_FIELDS))*16 + 8192170 return _estimate(row) * 16 + 8192171172173def _estimate(value, depth: int = 0) -> int:174 """累计原生对象及字段容器成本，重复引用也保守重复计数。175176 参数: value为固定树，depth为深度。返回: 基础bytes估算。177 异常: ResultFormatError为类型/深度不支持。178 """179 if depth > 32:180 raise ResultFormatError('结果行嵌套超过32层')181 value_type = type(value)182 if value is None or value_type in (str, int, bool, float):183 return getsizeof(value) + 64184 # 字段键与临时dict的开销按每字段256bytes留余量。185 if type(value) in SUMMARY_MODELS or (type(value).__name__ in _MODELS and type(value) is _MODELS[type(value).__name__]):186 return getsizeof(value) + sum(256 + _estimate(getattr(value, field.name), depth + 1)187 for field in fields(value) if field.init and188 not (value_type is ScalarValue and189 (field.name=='code' and value.code is None or field.name=='partial' and not value.partial)))190 if type(value) is tuple:191 return getsizeof(value) + sum(_estimate(item, depth + 1) + 32 for item in value)192 if isinstance(value, Enum) and type(value).__name__ in _ENUMS:193 return 512194 raise ResultFormatError('不能估算未登记结果对象')195196197def encode_row(row: ResultRow, max_bytes: int) -> bytes:198 """生成一条确定JSONL，包含行结束符且限制字节数。199200 参数: row为typed行；max_bytes为正原生int。201 返回: UTF-8 bytes，整数不经过浮点数转换。202 异常: ResultFormatError/ValueError为类型、大小或格式错误。203 """204 # 限额由资源账本给出；此函数不自行声称RAM已预留。205 try:206 row_table(row)207 if type(max_bytes) is not int or max_bytes <= 0:208 raise ValueError('max_bytes需正原生int')209 packed = {'row_schema_version': SUMMARY_MODELS[type(row)][2], 'row': summary_fields(row)}210 payload = (json.dumps(packed, ensure_ascii=False, allow_nan=False, sort_keys=True,211 separators=(',', ':')) + '\n').encode('utf-8')212 if len(payload) > max_bytes:213 raise ResultFormatError('结果行超过max_bytes')214 return payload215 finally:216 # 原异常/TB和返回值保持；只断本层临时引用，不清调用方帧。217 packed = payload = row = None218219220def decode_row(payload: bytes, max_bytes: int) -> ResultRow:221 """从单行版本化JSONL恢复并重新校验所有模型。222223 参数: payload为UTF-8 bytes；max_bytes为正原生int。224 返回: 当前登记的平坦结果行；旧版本1—6不再读取。225 异常: ResultFormatError为解析/版本/模型错误，ValueError为限额配置错误。226 """227 try:228 if type(max_bytes) is not int or max_bytes <= 0:229 raise ValueError('max_bytes需正原生int')230 if type(payload) is not bytes or len(payload) > max_bytes:231 raise ResultFormatError('结果行类型或字节限额错误')232 try:233 # 禁止重复键、隐式非有限值和未知外层字段。234 packed = json.loads(payload, object_pairs_hook=_unique_object, parse_constant=_reject_constant)235 if type(packed) is not dict or set(packed) != {'row_schema_version', 'row'}:236 raise ResultFormatError('结果行外层字段错误')237 version = packed['row_schema_version']238 if type(version) is int and version in _SUMMARY_VERSIONS:239 model, names = _SUMMARY_VERSIONS[version]240 fields_ = packed['row']241 if (type(fields_) is not dict or set(fields_) != set(names)242 or type(fields_['quality']) is not int or fields_['quality'] < 0):243 raise ResultFormatError('结果字段集合或质量类型错误')244 fields_['quality'] = QualityFlag(fields_['quality'])245 if version == 15:246 for name in ('group_keys','group_values'):247 if type(fields_[name]) is not list:248 raise ResultFormatError('聚合分组字段需要JSON数组')249 fields_[name]=tuple(fields_[name])250 if version==11:251 if type(fields_['parent_event_ids']) is not list:252 raise ResultFormatError('片段父ID需要JSON数组')253 fields_['parent_event_ids']=tuple(fields_['parent_event_ids'])254 return model(**fields_)255 raise ResultFormatError('不支持的结果行版本')256 except (ValueError, TypeError, OverflowError, RecursionError, UnicodeError) as exc:257 # 保留原模型报错原因，不吞异常或给出正常零。258 raise ResultFormatError(f'结果行无法解析: {exc}') from exc259 finally:260 # 原异常/TB和返回值保持；只断本层临时引用，不清调用方帧。261 packed = payload = row = fields_ = None262