chronaxis.results.manifest
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""定义结果清单中的运行事实和已发布输出身份。23输入为执行侧明确提供的时间/身份/后端，输出为冻结元数据模型。4不会推断CAN、DBC或同步关系；软件/Python版本、资源统计和输出由结果层补齐。5调用者须提供真实信息，模型校验不能证明真实日志已经验收。6"""78# 清单字段冻结；带时区时间及int64会话边界与核心协议一致。9from dataclasses import dataclass, asdict, field, fields10from datetime import datetime11from math import isfinite, isnan12from pathlib import Path13from sys import getsizeof14from .._version import software_version15from ..core.validation import require_text, require_identifier16from ..core.models import Interval17from ..core.binding import SignalDefinition18from ..core.types import QualityFlag19from .binding_metadata import PresetBindingMetadata, PresetRoleMetadata20from .can_metadata import CANSourceMetadata, CANUnitCheck, CANFrameFormatNotice21from .text_can_metadata import TextCANSourceMetadata22from .threshold_metadata import ThresholdDurationMetadata,SummaryThresholdDuration23from .input_diagnostics import InputDiagnostic24from .analysis_outcome import AnalysisOutcome, RuleOutcome25from ..legacy.records import LegacyCANField262728@dataclass(frozen=True)29class RunMetadata:30 """保存运行身份、实际后端、时间解释、会话边界与兼容限制。3132 started_at须含时区；session_boundaries为(session_id,start_ns,end_ns)tuple。33 映射/DBC/依赖事实未涉及时使用空tuple，不能把空值当成验证通过。34 字段只保存明确输入；新层资源/输出/结束时间由实际执行结果补充。35 signal_bindings保本次实际定义，preset_bindings保最终角色；可选语义仍是声明，36 角色契约通过也不表示物理事实已测量。can_sources保存同次真实CAN读取/暂存/37 单位核对的结构化事实；text_can_sources保存新文本解析/DBC/观测事实。38 未提供的详细事实不作为当前运行的验证证明。39 """4041 run_id: str42 input_identity: str | None43 plan_hash: str44 backend: str45 config_version: int46 source_time_interpretation: str47 session_boundaries: tuple[tuple[str, int, int], ...]48 started_at: str49 mapping_hashes: tuple[str, ...] = ()50 dbc_hashes: tuple[str, ...] = ()51 parsing_policy: str = 'decoded-results'52 worker_config: str = 'single-thread'53 dependency_versions: tuple[tuple[str, str], ...] = ()54 limitations: tuple[str, ...] = ()55 software_version: str = field(default_factory=software_version) # 构造本次元数据时读取版本；显式历史值保持。56 device_label: str = 'pending'57 signal_bindings: tuple[SignalDefinition, ...] = () # 本次实际完整定义，不含样本或输入行。58 preset_bindings: tuple[PresetBindingMetadata, ...] = () # 原预设版本及最终角色。59 can_sources: tuple[CANSourceMetadata, ...] = () # 同次实际CAN read事实，不含样本或伪造DBC哈希。60 session_ids: tuple[str, ...] = () # 无时间请求仍明确会话身份，绝不伪造边界。61 text_can_sources: tuple[TextCANSourceMetadata,...] = () # 新文本入口同次DBC/解析/观测事实。62 threshold_durations: tuple[SummaryThresholdDuration, ...] = () # 仅显式阈值预设的同次状态累计。63 detailed_diagnostics: bool = False # 按显式请求输出额外质量明细和摘要。64 retain_work: bool = False # 显式现场留存请求，不默认保留工作文件。65 input_diagnostics: tuple[InputDiagnostic,...] = () # 仅按请求记录同次所需输入范围。66 trace_requested: bool = False # input_trace表独立于贡献和哈希。67 integrity_diagnostics: bool = False # 摘要独立于逐条追溯。68 frame_format_notices: tuple[tuple[str, CANFrameFormatNotice], ...] = () # 同次已核差异始终提示，独立于可选诊断。6970 analysis_outcome: AnalysisOutcome | None = None # 同次推进的轻量业务结果概况，非逐条追溯。71 aggregation_definitions: tuple = () # 本次组级总体、单位及排除政策，每项只登记一次。7273 def __post_init__(self) -> None:74 """验证清单文本与明确会话范围，不访问外部输入。7576 参数: 无。返回: None。异常: TypeError/ValueError为结构或事实格式错误。77 """78 if self.analysis_outcome is not None and type(self.analysis_outcome) is not AnalysisOutcome:79 raise TypeError('分析结果概况需要AnalysisOutcome或None')80 if self.aggregation_definitions:81 from ..planning.aggregations import AggregationPlan82 if type(self.aggregation_definitions) is not tuple or any(type(a) is not AggregationPlan for a in self.aggregation_definitions):83 raise TypeError('聚合清单需要已冻结的聚合定义')84 # 必填文本不能用空字符串表示已经取得的运行身份。85 for name in ('run_id', 'plan_hash', 'source_time_interpretation',86 'parsing_policy', 'worker_config', 'software_version', 'device_label'):87 require_text(getattr(self, name), name)88 if self.backend not in ('reference', 'numpy') or type(self.config_version) is not int or self.config_version != 4:89 raise ValueError('结果清单需实际reference/numpy及配置版本4')90 if (self.input_identity is not None):91 require_text(self.input_identity, 'input_identity')92 if type(self.detailed_diagnostics) is not bool or type(self.retain_work) is not bool:93 raise TypeError('诊断/现场保留需要bool')94 if type(self.trace_requested) is not bool or type(self.integrity_diagnostics) is not bool:95 raise TypeError('输入诊断开关需要bool')96 if (type(self.input_diagnostics) is not tuple97 or any(type(item) is not InputDiagnostic for item in self.input_diagnostics)):98 raise TypeError('输入诊断需InputDiagnostic tuple')99 if self.input_diagnostics and not (self.trace_requested or self.integrity_diagnostics):100 raise ValueError('未请求不能附输入诊断')101 if (type(self.frame_format_notices) is not tuple or any(type(pair) is not tuple or len(pair) != 2102 or type(pair[1]) is not CANFrameFormatNotice for pair in self.frame_format_notices)):103 raise TypeError('帧型提示需要来源与CANFrameFormatNotice的tuple')104 for source, notice in self.frame_format_notices:105 require_identifier(source, 'frame_format.source')106 if len({(source, notice.alias) for source, notice in self.frame_format_notices}) != len(self.frame_format_notices):107 raise ValueError('同来源信号的帧型提示不能重复')108 require_text(self.started_at, 'started_at')109 if datetime.fromisoformat(self.started_at).utcoffset() is None:110 raise ValueError('started_at必须带时区')111 # 会话边界只验证半开范围，空会话允许start=end；不自动联合或平移。112 if type(self.session_boundaries) is not tuple:113 raise TypeError('session_boundaries需tuple')114 for boundary in self.session_boundaries:115 if type(boundary) is not tuple or len(boundary) != 3:116 raise TypeError('会话边界需(session_id,start,end)')117 require_identifier(boundary[0], 'session_id')118 # 清单直接JSON序列化，固定原生整数，bool及NumPy容器不能偷偷换类型。119 if type(boundary[1]) is not int or type(boundary[2]) is not int:120 raise TypeError('清单会话边界需原生int纳秒')121 Interval(boundary[1], boundary[2])122 for values in (self.mapping_hashes, self.dbc_hashes, self.limitations):123 if type(values) is not tuple or any(type(value) is not str or not value for value in values):124 raise TypeError('清单映射/限制需要非空文本tuple')125 if type(self.dependency_versions) is not tuple or any(type(item) is not tuple or len(item) != 2126 or any(type(value) is not str or not value for value in item) for item in self.dependency_versions):127 raise TypeError('依赖版本需(name,version)tuple')128129 # 新快照只接不可变定义和有限角色记录，拒绝自由字典或原样本模型。130 if (type(self.signal_bindings) is not tuple131 or any(type(item) is not SignalDefinition for item in self.signal_bindings)132 or type(self.preset_bindings) is not tuple133 or any(type(item) is not PresetBindingMetadata for item in self.preset_bindings)):134 raise TypeError('运行绑定清单需SignalDefinition及PresetBindingMetadata tuple')135 names = {item.name for item in self.signal_bindings}136 if len(names) != len(self.signal_bindings):137 raise ValueError('运行绑定清单不能重复别名')138 if len({item.preset_id for item in self.preset_bindings}) != len(self.preset_bindings):139 raise ValueError('运行绑定清单不能重复预设ID')140 if any(role.alias not in names for preset in self.preset_bindings for role in preset.roles):141 raise ValueError('预设最终角色必须引用本清单实际信号别名')142 if (type(self.threshold_durations) is not tuple143 or any(type(item) is not (SummaryThresholdDuration)144 for item in self.threshold_durations)):145 raise TypeError('阈值汇总需ThresholdDurationMetadata tuple')146 preset_ids = {item.preset_id for item in self.preset_bindings if item.preset_type == 'threshold_excursion'}147 if (len({item.preset_id for item in self.threshold_durations}) != len(self.threshold_durations)148 or any((item.rule_id != item.preset_id + '_activity')149 for item in self.threshold_durations)):150 raise ValueError('阈值汇总必须对应本次预设及生成规则，不能重复')151 if type(self.session_ids) is not tuple or len(set(self.session_ids)) != len(self.session_ids):152 raise TypeError('独立会话身份需要无重复tuple')153 for session_id in self.session_ids:154 require_identifier(session_id, 'session_id')155 # CAN仅接受登记冻结模型；会话、来源和单位事实必须关联本次原绑定，不能借旧任务别名。156 if type(self.text_can_sources) is not tuple or any(type(item) is not TextCANSourceMetadata for item in self.text_can_sources):157 raise TypeError('文本CAN清单需要TextCANSourceMetadata tuple')158 if len({item.source for item in self.text_can_sources})!=len(self.text_can_sources):159 raise ValueError('文本CAN清单来源重复')160 for source in self.text_can_sources:161 if source.session_id not in ({b[0] for b in self.session_boundaries}|set(self.session_ids)):162 raise ValueError('文本CAN清单会话错误')163 expected={s.name:s.input_unit for s in self.signal_bindings if s.source==source.source}164 if {c.alias:c.input_unit for c in source.unit_checks}!=expected or len(source.unit_checks)!=len(expected):165 raise ValueError('文本CAN单位清单必须覆盖实际绑定')166 if type(self.can_sources) is not tuple or any(type(item) is not CANSourceMetadata for item in self.can_sources):167 raise TypeError('CAN运行清单需要CANSourceMetadata tuple')168 if len({item.source for item in self.can_sources}) != len(self.can_sources):169 raise ValueError('CAN运行清单来源重复')170 # 空默认不为旧运行再构造来源/单位索引，实际CAN存在时才核本次对应关系。171 if self.can_sources:172 sessions = {boundary[0] for boundary in self.session_boundaries} | set(self.session_ids)173 definitions = {item.name: item for item in self.signal_bindings}174 for source in self.can_sources:175 if source.session_id not in sessions:176 raise ValueError('CAN运行清单会话不属于本次session_boundaries')177 expected = {name for name, definition in definitions.items() if definition.source == source.source}178 if {check.alias for check in source.unit_checks} != expected:179 raise ValueError('CAN单位清单必须恰好覆盖同来源本次实际绑定别名')180 for check in source.unit_checks:181 definition = definitions.get(check.alias)182 if (definition is None or definition.source != source.source183 or definition.input_unit != check.input_unit):184 raise ValueError('CAN单位核对别名/来源/input_unit与本次原声明不同')185 if source.decode_scope == 'requested':186 # requested可服务普通/混合/导出计划；只核实际需求，不由结果模式猜范围。187 # 仍要求原完整CAN定位和通道吻合，固定ID必须覆盖每个实际所需字段。188 # 实际需要CAN字段才导入来源包，保持纯预算/旧默认元数据的轻量导入。189 from ..sources.can_spec import can_field_from_key190 selected_field = can_field_from_key(definition.field)191 if (selected_field.channel != source.channel192 or selected_field.frame_id not in source.decode_frame_ids):193 raise ValueError('requested固定解码集合须覆盖本次实际绑定字段及同来源通道')194 for notice in source.frame_format_notices:195 # 提示必须满足本次信号定位；省略的帧型允许由同次实际读取补齐。196 from ..sources.can_spec import can_field_from_key197 definition = definitions.get(notice.alias)198 field = None if definition is None else can_field_from_key(definition.field)199 if (definition is None or definition.source != source.source200 or (field.channel,field.frame_id,field.signal)!=(notice.channel,notice.frame_id,notice.signal)201 or field.is_extended is not None and field.is_extended!=notice.is_extended202 or field.is_fd is not None and field.is_fd!=notice.requested_is_fd):203 raise ValueError('CAN帧型提示必须对应本次原信号完整定位')204205 def estimated_bytes(self) -> int:206 """在asdict/JSON复制前估算元数据与其临时对象开销。207208 参数: 无。返回: 保守bytes估算。异常: 无，类型已经校验。209 """210 # 文本/tuple结构只读遍历，转换副本成本以16倍和固定余量保守覆盖。211 # 空默认新字段不增加旧金额；非空定义/角色/CAN事实在asdict复制前递归覆盖。212 return 8192 + 16 * sum(_metadata_size(getattr(self, field.name)) + 256 for field in fields(self)213 if field.name not in ('signal_bindings', 'preset_bindings', 'can_sources', 'session_ids','text_can_sources','threshold_durations','detailed_diagnostics','retain_work','input_diagnostics','trace_requested','integrity_diagnostics','frame_format_notices','analysis_outcome','aggregation_definitions')214 or getattr(self, field.name))215216 def to_dict(self) -> dict:217 """返回独立JSON兼容元数据，调用侧必须先预留estimated_bytes。218219 参数: 无。220 返回: 独立JSON兼容dict。detailed_diagnostics=False时仅输出运行身份、221 后端、配置版本、时间解释、运行设置及适用的边界/限制/请求诊断；222 signal_bindings、preset_bindings、can_sources等即使非空也不输出。223 detailed_diagnostics=True才展开详细模型；空的可选明细继续按分支省略。224 模型持有字段不等于默认序列化一定包含该键。225 异常: 无预期异常；调用方须在复制前承担estimated_bytes预算。226 """227 if (not self.detailed_diagnostics):228 # 从生成处只复制必要运行信息，不先展开完整绑定/来源后再隐藏。229 result = {name: getattr(self, name) for name in ('run_id', 'plan_hash', 'backend',230 'config_version', 'started_at', 'device_label', 'software_version',231 'source_time_interpretation', 'parsing_policy', 'worker_config', 'retain_work', 'detailed_diagnostics')}232 for name in ('input_identity', 'session_boundaries', 'session_ids', 'limitations', 'dependency_versions'):233 if getattr(self, name):234 result[name] = getattr(self, name)235 if self.threshold_durations:236 result['threshold_durations']=tuple(item.to_dict() for item in self.threshold_durations)237 if self.trace_requested or self.integrity_diagnostics:238 result['diagnostics']={'trace':self.trace_requested,'integrity':self.integrity_diagnostics}239 result['input_diagnostics']=[item.to_dict() for item in self.input_diagnostics]240 if self.frame_format_notices:241 result['frame_format_notices']=[dict(source=source, **asdict(notice)) for source, notice in self.frame_format_notices]242 if self.analysis_outcome is not None:243 result['analysis_outcome']=self.analysis_outcome.to_dict()244 if self.aggregation_definitions:245 result['aggregation_definitions']=[asdict(a) for a in self.aggregation_definitions]246 return result247 result = asdict(self)248 if not self.aggregation_definitions:249 result.pop('aggregation_definitions')250 if self.analysis_outcome is None:251 result.pop('analysis_outcome')252 else:253 result['analysis_outcome']=self.analysis_outcome.to_dict()254 if self.frame_format_notices:255 result['frame_format_notices']=[dict(source=source, **asdict(notice)) for source, notice in self.frame_format_notices]256 else:257 result.pop('frame_format_notices')258 if self.input_identity is None:259 result.pop('input_identity') # 未计算的全量输入摘要不以占位文字冒充。260 if self.threshold_durations:261 result['threshold_durations'] = tuple(item.to_dict() for item in self.threshold_durations)262 else:263 result.pop('threshold_durations')264 if not self.text_can_sources:265 result.pop('text_can_sources')266 if not self.session_ids:267 result.pop('session_ids')268 # 只有实际存在的详细事实才展开对应字段。269 if self.signal_bindings:270 for row, definition in zip(result['signal_bindings'], self.signal_bindings):271 # dtype转明确文本；无效值里的非有限诊断用既有显式标签，不输出裸NaN。272 row['dtype'] = definition.dtype.value273 row['invalid_values'] = tuple({'diagnostic_float': 'nan' if isnan(value) else274 'inf' if value > 0 else '-inf'} if type(value) is float and not isfinite(value) else value275 for value in definition.invalid_values)276 row['semantic_status'] = 'not_provided' if definition.semantic_id is None else 'declared'277 # 原基础策略拒绝全部非零质量位；声明名称不代表物理语义已核实。278 row['quality_policy'] = {'mode': 'reject_nonzero',279 'forbidden_flags': tuple(flag.name for flag in QualityFlag),280 'forbidden_mask': sum(int(flag) for flag in QualityFlag)}281 else:282 result.pop('signal_bindings')283 if not self.preset_bindings:284 result.pop('preset_bindings')285 if not self.can_sources:286 result.pop('can_sources')287 else:288 # 仅实际差异增加提示；原无差异CAN清单的键和值完全保持。289 for row, source in zip(result['can_sources'], self.can_sources):290 if not source.frame_format_notices:291 row.pop('frame_format_notices')292 if source.input_check == 'full' and source.check_policy_version is None:293 for key in ('input_check', 'time_scope', 'check_policy_version', 'retained_record_count',294 'full_check_count', 'selected_first_ns', 'selected_last_ns'):295 row.pop(key)296 if source.decode_scope == 'all':297 # 新默认字段既不改变旧JSON，也不增加旧估算费用。298 row.pop('decode_scope')299 row.pop('decode_frame_ids')300 if source.decoded_values_policy == 'all_values-v1':301 # 兼容旧all与旧requested全字段事实，默认值不改变清单键或费用。302 row.pop('decoded_values_policy')303 return result304305306def _metadata_size(value) -> int:307 """统计已校验元数据树的基础对象大小，不产生转换树。308309 参数: value为元数据字段。返回: 基础bytes估算。异常: 无。310 """311 # 仅遍历已验证的有限模型，不泛化为raw_config或任意dataclass序列化。312 from ..planning.aggregations import AggregationPlan313 if type(value) is AggregationPlan:314 return getsizeof(value)+sum(_metadata_size(getattr(value,key)) for key in value.__dataclass_fields__)315 if type(value) in (SignalDefinition, PresetBindingMetadata, PresetRoleMetadata,316 CANSourceMetadata, CANUnitCheck, CANFrameFormatNotice,TextCANSourceMetadata,InputDiagnostic,317 AnalysisOutcome, RuleOutcome):318 return getsizeof(value) + sum(_metadata_size(getattr(value, field.name)) + 32 for field in fields(value)319 if type(value) is not CANSourceMetadata or (320 (field.name != 'frame_format_notices' or value.frame_format_notices)321 and (field.name not in ('decode_scope', 'decode_frame_ids') or value.decode_scope != 'all')322 and (field.name != 'decoded_values_policy' or value.decoded_values_policy != 'all_values-v1')323 and (field.name not in ('input_check', 'time_scope', 'check_policy_version',324 'retained_record_count', 'full_check_count', 'selected_first_ns', 'selected_last_ns')325 or value.check_policy_version is not None or value.input_check != 'full')))326327 if type(value) is tuple:328 return getsizeof(value) + sum(_metadata_size(item) + 32 for item in value)329 return getsizeof(value)330331332@dataclass(frozen=True)333class OutputManifest:334 """保存成功发布目录、清单和固定白名单表行数；legacy只在启用时出现。"""335336 run_id: str337 output_dir: Path338 manifest_path: Path339 output_files: tuple[Path, ...]340 row_counts: tuple[tuple[str, int], ...]341 status: str = 'completed'342343 def __post_init__(self) -> None:344 """校验返回清单身份，不凭构造调用证明目录已发布。345346 参数: 无。返回: None。异常: TypeError/ValueError为字段错误。347 """348 require_text(self.run_id, 'run_id')349 if self.status != 'completed' or not isinstance(self.output_dir, Path) or not isinstance(self.manifest_path, Path):350 raise ValueError('OutputManifest需completed和Path')351 if type(self.output_files) is not tuple or any(not isinstance(path, Path) for path in self.output_files):352 raise TypeError('输出路径需tuple Path')353 if type(self.row_counts) is not tuple or any(type(item) is not tuple or len(item) != 2354 or item[0] not in ('events', 'segments', 'calculations', 'message_calculations', 'legacy_contributions', 'sequences','contributions','input_trace','selections','aggregations')355 or type(item[1]) is not int or item[1] < 0 for item in self.row_counts):356 raise ValueError('输出行数需固定表及非负原生int')357358359@dataclass(frozen=True)360class ExportManifest:361 """独立信号或原帧导出的实际输出路径；调用者无需猜测时间文件名。"""362363 output_dir: Path364 manifest_path: Path365 output_files: tuple[Path, ...]366367 def __post_init__(self):368 """核对实际路径类型和所属目录。参数: 无。返回: None。异常: 类型或路径错，不访问文件。"""369 if not isinstance(self.output_dir, Path) or not isinstance(self.manifest_path, Path):370 raise TypeError('导出返回目录和清单必须为Path')371 if type(self.output_files) is not tuple or any(not isinstance(p, Path) for p in self.output_files):372 raise TypeError('导出文件必须为Path tuple')373 if any(p.parent != self.output_dir for p in (self.manifest_path, *self.output_files)):374 raise ValueError('导出返回路径必须属于本次输出根')375