chronaxis.execution.hosted_sample_reduction

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""把有限真实样本块交给明确注册的宿主归约，复用原半开选择、覆盖与独立数值后端。"""23from dataclasses import replace4from ..core.calculation_output_numpy import convert_result_numpy5from ..core.calculation_output_reference import convert_result_reference6from ..core.sample_reduction import SampleReductionResult7from ..core.sample_reduction_numpy import NumpySampleReduction8from ..core.sample_reduction_reference import ReferenceSampleReduction9from ..core.types import QualityFlag as Q10from ..extension_runtime import HostedReduction11from ..results.budget import _cleanup_owner121314class HostedSampleReduction:15    """一个会话、事件或窗口持有一个插件状态；插件不决定是否选入样本及结果显示单位。"""1617    def __init__(self, spec, input_unit, dtype, descriptor, conversion, backend, budget, partial_limit):18        """创建所选后端的覆盖计算器及宿主状态。参数: 已校验请求、描述、单位及预算。1920        返回: 独立归约owner。异常: 宿主工厂或预算失败转交可重试清理句柄。21        """22        self._host=self._coverage=None23        self._conversion=conversion24        self._convert=convert_result_reference if backend=='reference' else convert_result_numpy25        self._closed=self._finished=self.partial=False26        try:27            coverage_spec=replace(spec,operation='sample_count',output_unit='1')28            kernel=ReferenceSampleReduction if backend=='reference' else NumpySampleReduction29            self._coverage=kernel(coverage_spec,input_unit,dtype,partial_limit=partial_limit)30            self._host=HostedReduction(descriptor,backend,budget)31        except BaseException as error:32            self._host=getattr(error,'extension_cleanup_owner',self._host)33            _cleanup_owner(self,error,'hosted_sample_cleanup_owner','宿主样本归约')34            raise3536    def push(self,series):37        """沿同一选中掩码消费并返回覆盖事实。参数: 有限 CalculationSeries。返回: 本块 facts。异常: 回调或输入错误。"""38        return self._coverage.push(series,sample_consumer=self._host.consume)3940    def finish(self):41        """无参与值保留空结果，输出单位只变换一次。参数: 无。返回: SampleReductionResult。异常: 回调或转换错误。"""42        count=self._coverage.valid_sample_count43        number,quality=None,Q.NONE44        if count:45            value=self._host.finalize()46            self.partial=value.partial47            if value.valid:48                number,quality=self._convert(value.value,value.dtype,self._conversion)49            else:50                quality=value.quality51        self._finished=True52        return SampleReductionResult(number,self._conversion.unit,bool(quality),quality,count)5354    def close(self):55        """未正常结束时通知取消，释放宿主和覆盖状态。参数: 无。返回: None。异常: 清理失败可重试。"""56        if self._closed:57            return58        if self._host is not None:59            self._host.close(cancelled=not self._finished)60        if self._coverage is not None:61            self._coverage.close()62        self._closed=True63