chronaxis.execution.record_diagnostics

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""仅在请求追溯或完整性诊断时观察同次输入，不增加来源扫描。23trace逐条保存所需输入值和来源序号；integrity随这些记录计算摘要，两者独立。4来源配置只在清单保存一次。借入值在调用结束释放，固定哈希状态按来源付费；5最终明细使用已有结果仓的有界外排和生命周期，不建立第二份历史。6"""7import hashlib8import json9from math import isfinite10from ..results.input_diagnostics import InputDiagnostic11from ..results.budget import _reserve_owned_ram,_release_owned_ram,_cleanup_owner12from ..results.errors import ResultClosedError13from .history import _metadata_bytes141516class RecordDiagnostics:17    """保存每来源一个计数及可选摘要；逐条明细同步交给最终结果消费者。"""18    def __init__(self,plan,requirements,budget,consumer):19        """预约实际来源状态。参数: 计划/需求/预算/追溯回调。返回: owner。异常: 预算错误。"""20        self._plan,self._requirements,self._budget=plan,requirements,budget21        self._consumer=consumer if plan.diagnostics.trace else None22        self._held_cost=self._fixed_cost=self._row_cost=023        self._counts=self._hashes=None24        self._closed=False25        try:26            names=tuple(s.source for s in requirements.sources if s.active)27            _reserve_owned_ram(self,65536+4096*len(names),'_fixed_cost')28            # 来源清单只计路径/需求，不遍历memory来源的全部数据；覆盖最终清单交接副本。29            extra=32*_metadata_bytes(requirements)30            for source in plan.sources:31                if source.name not in names:continue32                config=source.config33                paths=(config.input_paths if source.source_type=='can_blf' else34                    (item.path for item in config.inputs) if source.source_type in ('can_asc','asc_or_blf','raw_can_csv') else35                    config.paths if source.source_type=='signal_csv' else ())36                extra+=sum(4096+256*len(str(path)) for path in paths)37            _reserve_owned_ram(self,extra,'_fixed_cost')38            self._counts=dict.fromkeys(names,0)39            self._hashes={name:hashlib.sha256() for name in names} if plan.diagnostics.integrity else {}40        except BaseException as error:41            _cleanup_owner(self,error,'record_diagnostics_cleanup_owner','输入诊断');raise4243    def consume(self,source,sequence,time_ns,record_type,locator,values,quality):44        """借入必要输入，按请求追溯/取摘要。参数: 定位/类型/值/质量。返回: None。异常: 编码/预算/取消。"""45        if self._closed:raise ResultClosedError('输入诊断已关闭')46        payload=normalized=text=None47        try:48            # 先覆盖编码及非有限标签空间，非法值不偷偷转成有效0。49            _reserve_owned_ram(self,16384+64*_metadata_bytes(values),'_row_cost')50            normalized={name:({'nonfinite':'nan' if value!=value else 'inf' if value>0 else '-inf'}51                if type(value) is float and not isfinite(value) else value) for name,value in values.items()}52            text=json.dumps(normalized,ensure_ascii=False,allow_nan=False,sort_keys=True,separators=(',',':'))53            if source in self._hashes:54                payload=json.dumps((sequence,time_ns,record_type,locator,int(quality),text),ensure_ascii=False,55                    allow_nan=False,separators=(',',':')).encode('utf-8')56                self._hashes[source].update(len(payload).to_bytes(8,'big'))57                self._hashes[source].update(payload)58            if self._consumer is not None:59                self._consumer(source,sequence,time_ns,record_type,locator,text,quality)60            self._counts[source]+=161        finally:62            payload=normalized=values=text=None63            _release_owned_ram(self,self._row_cost,'_row_cost')6465    def summaries(self,cursors):66        """仅EOF后交付有限来源清单。参数: 实际游标。返回: InputDiagnostic tuple。异常: 不完整来源。"""67        result=[]68        for cursor in cursors:69            source=next(s for s in self._plan.sources if s.name==cursor.source_name)70            config=source.config71            paths=(tuple(str(p) for p in config.input_paths) if source.source_type=='can_blf' else72                tuple(str(i.path) for i in config.inputs) if source.source_type in ('can_asc','asc_or_blf','raw_can_csv') else73                tuple(str(path) for path in config.paths) if source.source_type=='signal_csv' else ())74            req=self._requirements.for_source(source.name)75            result.append(InputDiagnostic(source.name,source.source_type,paths,req.required_aliases,76                tuple((name,self._plan.bindings.resolve(name).field) for name in req.required_aliases),77                req.input_check,req.time_scope,self._counts[source.name],cursor._finished,78                self._hashes[source.name].hexdigest() if source.name in self._hashes else None))79        return tuple(result)8081    def close(self):82        """断开哈希/回调后归还本owner费用。参数: 无。返回: None。异常: 预算释放错误。"""83        if self._closed:return84        self._counts=self._hashes=self._consumer=None85        _release_owned_ram(self,self._row_cost,'_row_cost')86        _release_owned_ram(self,self._fixed_cost,'_fixed_cost')87        self._closed=True88