chronaxis.core.window_delta

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""声明完整窗口变化量的输入、配置和结果，不读取文件或保存触发状态。23输入是有限ZOH已经划分好的连续片段，坐标为整数纳秒，数值保留声明dtype。4片段起点的真实点质量可以不同于连续质量，例如真实EOF样本仍有效，但不能5向EOF后保持。两个后端共用结构检查；数值、覆盖率和单调判断分别实现。6结果把数值变化量与条件真假分开，缺数据不返回零，也不在这里生成事件。7"""89from dataclasses import dataclass10from math import isfinite11from .types import QualityFlag as Q, SignalDType as D, Truth12from ..timebase.conversion import require_int64_ns131415@dataclass(frozen=True)16class WindowDeltaSpec:17    """保存窗口长度、方向、阈值及单调要求；只接受完整覆盖政策。"""1819    window_ns: int  # 完整回看长度，必须为正。20    threshold: int | float  # 非负门槛，单位与输入相同，包含等于门槛。21    direction: str = 'any'  # rise末减首，fall首减末，any取绝对差。22    monotonic: bool = False  # rise非降、fall非升、any二者任一。23    coverage_policy: str = 'full'  # 样本兼容口径不在本核心隐式启用。2425    def __post_init__(self):26        """检查明确参数，不从名字猜时长或单位。2728        参数: 无，读取冻结字段。返回: None。异常: 类型、非有限或范围错误。29        """30        object.__setattr__(self, 'window_ns', require_int64_ns(self.window_ns))31        if self.window_ns <= 0 or self.direction not in ('rise', 'fall', 'any'):32            raise ValueError('window_delta窗口须为正，方向须为rise/fall/any')33        if type(self.monotonic) is not bool or type(self.threshold) not in (int, float):34            raise TypeError('monotonic需bool，threshold需数值且不能为bool')35        if self.threshold < 0 or (type(self.threshold) is float and not isfinite(self.threshold)):36            raise ValueError('threshold须为非负有限数值')37        if self.coverage_policy != 'full':38            raise ValueError('当前window_delta仅支持full覆盖政策')394041@dataclass(frozen=True)42class DeltaSpan:43    """保存一段常值和起点点值；零长度仅用于真实末点。4445    quality适用于[start_ns,end_ns)，point_quality适用于start_ns。46    值为None时两种质量都须说明无效原因。时间到期不能通过延长end_ns补齐。47    """4849    start_ns: int  # 当前更新或有效性边界。50    end_ns: int  # 连续值排除的右端。51    value: int | float | None  # 规范后的原生数值，未知可为None。52    quality: Q = Q.NONE  # 连续区间无效原因。53    point_quality: Q | None = None  # 默认与连续质量相同。54    partial: bool = False  # 连续数值依赖部分滚动结果。55    point_partial: bool = False  # 起点值的独立部分性。5657    def __post_init__(self):58        """规范纳秒和点质量，并检查有值证据。5960        参数: 无。返回: None。异常: 类型、倒序或未知无原因。61        """62        object.__setattr__(self, 'start_ns', require_int64_ns(self.start_ns))63        object.__setattr__(self, 'end_ns', require_int64_ns(self.end_ns))64        if self.end_ns < self.start_ns:65            raise ValueError('DeltaSpan终点不得早于起点')66        if not isinstance(self.quality, Q):67            raise TypeError('DeltaSpan质量需QualityFlag')68        if self.point_quality is None:69            object.__setattr__(self, 'point_quality', self.quality)70        if not isinstance(self.point_quality, Q):71            raise TypeError('DeltaSpan点质量需QualityFlag')72        if self.value is None:73            if self.quality == Q.NONE or self.point_quality == Q.NONE:74                raise ValueError('没有数值必须注明未知原因')75        elif type(self.value) not in (int, float):76            raise TypeError('DeltaSpan值需原生int/float或None')777879@dataclass(frozen=True)80class WindowDeltaResult:81    """保存一次窗口结果；value与方向一致，condition额外考虑单调要求。8283    valid_duration_ns只计完整窗口内已知连续时间；端点无效也会令条件UNKNOWN，84    即使连续覆盖率恰好为1。monotonic是对应方向的实测结果，未知时为None。85    """8687    time_ns: int  # 当前判断时刻。88    window_start_ns: int  # 精确Python整数，不因回看低于int64下限回绕。89    value: int | float | None  # 定宽差值或None，绝不以0替代未知。90    condition: Truth  # TRUE/FALSE/UNKNOWN。91    quality: Q  # UNKNOWN原因；已知为NONE。92    valid_duration_ns: int  # 已知连续时间长度。93    coverage: float  # 已知时间/完整窗口时长。94    monotonic: bool | None  # 对应方向是否单调。959697def validate_delta_input(spec, spans, time_ns, dtype):98    """检查两个数值核共用的结构及dtype，不计算变化量或单调性。99100    参数: spec配置，spans不可变有序片段，time_ns判断时间，dtype声明类型。101    返回: 规范int判断时间。异常: 类型、重叠、非法值或dtype范围错误。102    """103    time_ns = require_int64_ns(time_ns)104    if not isinstance(spec, WindowDeltaSpec) or type(spans) is not tuple:105        raise TypeError('窗口需要WindowDeltaSpec及DeltaSpan tuple')106    if dtype not in (D.INT64, D.UINT64, D.FLOAT64):107        raise TypeError('window_delta只支持数值信号')108    previous = None109    for item in spans:110        # 重叠输入会重复计算覆盖率，不能在核中悄悄排序或合并。111        if not isinstance(item, DeltaSpan):112            raise TypeError('窗口输入需DeltaSpan')113        if previous is not None and (item.start_ns < previous.end_ns or item.start_ns <= previous.start_ns):114            raise ValueError('窗口片段必须严格有序且不重叠')115        previous = item116        if item.value is None or (item.quality != Q.NONE and item.point_quality != Q.NONE):117            continue118        if dtype is D.FLOAT64:119            if type(item.value) is not float or not isfinite(item.value):120                raise ValueError('有效float64输入须为有限原生float')121        elif type(item.value) is not int:122            raise TypeError('有效整数信号不能隐式转浮点')123        elif not (-(2**63) <= item.value < 2**63 if dtype is D.INT64 else 0 <= item.value < 2**64):124            raise ValueError('有效数值超出声明dtype')125    return time_ns126