chronaxis.execution.batch_signal_groups

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""将已校验的解析信号批按watermark归为同刻组，不创建磁盘历史。23来源适配器负责原输入及跨批合同。本层只保尚未完整的组，最大时间组等到4后续时间或真实EOF才交付；超出预算直接失败，不拆散同刻更新。5"""67import heapq8from ..core.models import SourceRef9from ..results.errors import ResultClosedError10from ..results.spool import _reserve_owned_ram, _release_owned_ram, _cleanup_owner11from .groups import DecodedTimeGroup12from .record_sample import _record_from_validated_fields13from ..core.types import SignalDType as D14from .calculation_values import _scalar_from_validated_column15from .history import _metadata_bytes161718class BatchSignalGroups:19    """借用已验证批，拥有未完成样本与有限排序状态。"""2021    def __init__(self, budget, definitions, session_id, consumer):22        """预付状态。参数: 预算、有限定义、会话、consumer。返回: owner。异常: 预算/取消原错。"""23        self._budget, self._consumer = budget, consumer24        self._held_cost = self._fixed_cost = self._sample_cost = self._work_cost = 025        self._heap = None26        self._refs = None27        self._serial = 028        self._maximum = self._last = None29        self._closed = self._failed = self._finished = False30        try:31            _reserve_owned_ram(self, 65536 + 32*_metadata_bytes(definitions), '_fixed_cost')32            self._refs = {item.name: SourceRef(session_id, item.source, item.field) for item in definitions}33            self._heap = []34        except BaseException as error:35            _cleanup_owner(self, error, 'batch_groups_cleanup_owner', '解析信号归组')36            raise3738    def consume(self, batch):39        """保必要规范值，按完成进度交付组；丢弃原值/详细来源副本。4041        参数: 可信适配器已校验SignalBatch。返回: None。异常: 预算/模型/回退，失败后只清理。42        """43        if self._closed or self._failed or self._finished:44            raise ResultClosedError('解析信号组已关闭、失败或结束')45        sample = value = column = None46        try:47            for column in batch.series.values():48                for index in range(len(column.time_ns)):49                    time = int(column.time_ns[index])50                    if self._last is not None and time <= self._last:51                        raise ValueError('来源在已交付同刻组之后又返回旧样本')52                    _reserve_owned_ram(self, 8192, '_sample_cost')53                    # 本批列已由适配器校验并由本次运行拥有，不再重复构造校验模型。54                    value = _scalar_from_validated_column(column, index)55                    sample = _record_from_validated_fields(column.name, time, int(column.sequence[index]),56                        value, column.unit,self._refs[column.name],57                        column.values[index].item() if value.dtype is D.ENUM else None)58                    heapq.heappush(self._heap, (time, self._serial, sample))59                    self._serial += 160                    self._maximum = time if self._maximum is None else max(time, self._maximum)61            if batch.watermark_ns is not None:62                self._flush(batch.watermark_ns, keep_last=True)63        except BaseException:64            self._failed = True65            raise66        finally:67            batch = sample = value = column = None6869    def _flush(self, watermark, *, keep_last):70        """只交付完整组，费用在consumer返回后释放。参数: 进度/是否保末组。返回: None。异常: 原错。"""71        while self._heap and (watermark is None or self._heap[0][0] <= watermark):72            time = self._heap[0][0]73            if keep_last and time == self._maximum:74                return75            samples = group = None76            count = 077            primary = None78            try:79                # 当前组至多为已付费样本数量，先覆盖列表/tuple引用再构造。80                _reserve_owned_ram(self, 4096 + len(self._heap)*32, '_work_cost')81                samples = []82                while self._heap and self._heap[0][0] == time:83                    samples.append(heapq.heappop(self._heap)[2])84                    count += 185                group = DecodedTimeGroup(time, tuple(samples))86                self._consumer(group)87                self._last = time88            except BaseException as error:89                primary = error90                raise91            finally:92                samples = group = None93                try:94                    _release_owned_ram(self, count*8192, '_sample_cost')95                    _release_owned_ram(self, self._work_cost, '_work_cost')96                except BaseException as cleanup:97                    self._failed = True98                    if primary is None:99                        raise100                    primary.add_note('解析同刻组清理失败：'+type(cleanup).__name__)101102    def finish(self):103        """调用方确认真实EOF及关闭后交最后组。参数: 无。返回: None。异常: 失败拒绝完成。"""104        if self._closed or self._failed:105            raise ResultClosedError('解析信号组不能完成')106        if self._finished:107            return108        try:109            self._flush(None, keep_last=False)110            self._finished = True111        except BaseException:112            self._failed = True113            raise114115    def close(self):116        """丢待处理组并退款，不把提前close当EOF。参数: 无。返回: None。异常: 退款失败可重试。"""117        if self._closed:118            return119        self._failed = True120        self._heap = self._consumer = self._refs = None121        for slot in ('_work_cost', '_sample_cost', '_fixed_cost'):122            _release_owned_ram(self, getattr(self, slot), slot)123        self._closed = True124