chronaxis.execution.record_point_buffer

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""共享有限驱动点的接收与失效处理，不执行数值统计或创建历史。23积分和时间统计沿用同一同刻/递增/条件门槛，具体owner负责预付缓冲费用并实现_flush。4该内部基类不能独立构造使用；没有独立预算或第二份输入缓冲。5"""67from ..core.numeric import ScalarValue8from ..core.types import SignalDType as D9from ..results.errors import ResultClosedError10from ..timebase.conversion import require_int64_ns111213class RecordPointBuffer:14    """复用一份有限点缓冲，子类初始化字段并拥有数值状态及关闭过程。"""1516    def _require_open(self):17        """失败/清理/关闭后拒绝读取与推进。参数: 无。返回: None。异常: ResultClosedError。"""18        if self._closed or self._closing or self._failed:19            raise ResultClosedError('记录流统计已关闭、失败或正在清理')2021    def consume(self, rows, *, conditions=None, condition_valid_until_ns=None, actual_value_updates=None):22        """接收已按所选驱动求值的有限点，不先截掉窗口两侧记录。2324        参数: rows为不超过batch_intervals+1的tuple，每项(time_ns, ScalarValue)，25            或附第三项实际有效截止valid_until_ns；到截止即失效，None表示本点无额外截止。26            conditioned时另传同长conditions及可选实际条件截止tuple；TRUE/FALSE/UNKNOWN分开处理。27            actual_value_updates仅样本统计/选择使用，明确区分真实值更新与条件更新，不提供时保持原五字段缓冲。28        返回: None。异常: 类型、时间倒退/同刻、预算/取消；任何错误后不交付前缀结果。29        约束: 上游处理同刻、条件及TTL；无效值必须传入，不能跳过后连接有效值。30        """31        self._require_open()32        if self._finished:33            raise ResultClosedError('记录流统计已结束，不能继续追加')34        item = value = None35        try:36            if type(rows) is not tuple or len(rows) > self._limit + 1:37                raise ValueError('驱动点需要不超批上限的tuple')38            if actual_value_updates is not None and (type(actual_value_updates) is not tuple39                    or len(actual_value_updates)!=len(rows) or any(type(flag) is not bool for flag in actual_value_updates)):40                raise ValueError('真实更新标志需要同长bool tuple')41            if self._conditioned:42                if type(conditions) is not tuple or len(conditions) != len(rows):43                    raise ValueError('条件统计需要同长BOOL条件tuple')44                if condition_valid_until_ns is not None and (type(condition_valid_until_ns) is not tuple45                        or len(condition_valid_until_ns) != len(rows)):46                    raise ValueError('条件截止需要同长tuple')47            elif conditions is not None or condition_valid_until_ns is not None:48                raise ValueError('无条件统计不能隐式接入条件')49            for index, item in enumerate(rows):50                if type(item) is not tuple or len(item) not in (2, 3):51                    raise TypeError('驱动点需要时间、数值及可选实际有效截止')52                time, value = item[:2]53                deadline = item[2] if len(item) == 3 else None54                if deadline is not None:55                    deadline = require_int64_ns(deadline, 'record_calculation.valid_until_ns')56                time = require_int64_ns(time, 'record_calculation.time_ns')57                if not isinstance(value, ScalarValue) or value.dtype is not self._dtype:58                    raise TypeError('驱动点数值与声明dtype不同')59                if self._last_time is not None and time <= self._last_time:60                    raise ValueError('驱动点必须严格递增；同刻更新应先按既有政策归组')61                condition = conditions[index] if self._conditioned else None62                condition_deadline = None if condition_valid_until_ns is None else condition_valid_until_ns[index]63                if self._conditioned and (not isinstance(condition, ScalarValue) or condition.dtype is not D.BOOL):64                    raise TypeError('统计选择条件需要BOOL ScalarValue')65                if condition_deadline is not None:66                    condition_deadline = require_int64_ns(condition_deadline, 'record_calculation.condition_valid_until_ns')67                # 只保计算必需的冻结数值与实际时间，不留下原消息或来源对象。68                self._rows.append((time, value, deadline, condition, condition_deadline)69                    +(() if actual_value_updates is None else (actual_value_updates[index],)))70                self._last_time = time71                if len(self._rows) == self._limit + 1:72                    self._flush()73        except BaseException:74            self._failed = True75            raise76        finally:77            rows = item = value = conditions = condition = condition_valid_until_ns = None7879    def expire_at(self, time_ns, *, condition=False):80        """依赖中途失效时截断最后待结算驱动点，不新增驱动点或补计周期。8182        参数: time_ns为已按序处理的失效实际时刻；condition=True只截断选择条件。返回: None。83        异常: 非法/回退时间或对象关闭失败；调用方必须在下一驱动点进入前通知。84        """85        self._require_open()86        if self._finished:87            raise ResultClosedError('记录流统计已结束，不能修改有效截止')88        try:89            if type(condition) is not bool or condition and not self._conditioned:90                raise ValueError('条件失效通知只能用于条件统计')91            time = require_int64_ns(time_ns, 'record_calculation.expire_at')92            if self._last_time is not None and time < self._last_time:93                raise ValueError('失效通知不能回退到已处理驱动点之前')94            if self._rows:95                index = 4 if condition else 296                previous = self._rows[-1]97                deadline = previous[index]98                self._rows[-1] = previous[:index] + (time if deadline is None else min(time, deadline),) + previous[index+1:]99        except BaseException:100            self._failed = True101            raise102