chronaxis.execution.record_sample

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""记录流只保一份逻辑值；枚举单独保必要整数码，公共来源定义共享引用。"""2from dataclasses import dataclass3from ..core.models import SourceRef,SignalSample4from ..core.numeric import ScalarValue5from ..core.types import SignalDType as D67@dataclass(frozen=True,slots=True)8class RecordSample:9    """一次规范更新；没有批次副本、原文或普通值的第二份物理表示。"""10    name: str11    time_ns: int12    sequence: int13    value: ScalarValue14    unit: str15    source_ref: SourceRef16    enum_code: int | None = None1718    def __post_init__(self):19        """检查必要样本及枚举码含义。参数: 字段。返回: None。异常: 类型/值矛盾。"""20        if type(self.value) is not ScalarValue or type(self.unit) is not str:21            raise TypeError('记录样本需要规范值和单位')22        SignalSample(self.name,self.time_ns,self.value.value,self.value.valid,self.value.quality,self.source_ref,self.sequence)23        if self.value.dtype is D.ENUM:24            if type(self.enum_code) is not int:25                raise TypeError('枚举需要保留整数码')26            if self.value.valid and (not 0<=self.enum_code<len(self.value.labels) or self.value.labels[self.enum_code]!=self.value.value):27                raise ValueError('枚举码与标签不符')28        elif self.enum_code is not None:29            raise ValueError('普通值不保存第二份物理码')3031    @property32    def array_value(self):33        """按需取得数组值；无效值只返回类型占位。参数: 无。返回: 标量。异常: 无。"""34        if self.value.dtype is D.ENUM:return self.enum_code35        if self.value.valid:return self.value.value36        return False if self.value.dtype is D.BOOL else 0. if self.value.dtype is D.FLOAT64 else 03738    @property39    def original_present(self):40        """本样本不保存原文。参数: 无。返回: False。异常: 无。"""41        return False4243    @property44    def original_value(self):45        """本样本无额外原值。参数: 无。返回: None。异常: 无。"""46        return None474849def _record_from_validated_fields(name, time_ns, sequence, value, unit, source_ref, enum_code=None):50    """把已校验列或解码归一化结果包装为当前运行记录，不再创建校验样本。5152    参数: 已核别名/单位/来源、原生时间及序号、ScalarValue和已核枚举码。53    返回: 冻结RecordSample。异常: 分配错误；本函数不接外部未校验输入。54    调用方必须完成列布局/时间/枚举域验证，公开RecordSample构造保持原检查。55    """56    result = object.__new__(RecordSample)57    # 只安装原模型字段；不存在用户可启用的跳过校验选项。58    object.__setattr__(result, 'name', name)59    object.__setattr__(result, 'time_ns', time_ns)60    object.__setattr__(result, 'sequence', sequence)61    object.__setattr__(result, 'value', value)62    object.__setattr__(result, 'unit', unit)63    object.__setattr__(result, 'source_ref', source_ref)64    object.__setattr__(result, 'enum_code', enum_code)65    return result66