chronaxis.extension_runtime
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""托管可信回调的后端调用、有限资源、返回形状及取消和关闭责任。23逐点调用只拥有本次工作区；样本归约和时序状态分别活到当前统计范围或4会话结束。共享ResourceBudget只为请求的实现付费，回调可在分配额外内存前5预约并归还自己的金额。宿主代码必须遵守该协议，这不是Python代码沙箱。6"""78from .extensions import _BUDGET, FunctionSpec, ReductionSpec, TransformSpec9from .core.numeric import ScalarValue10from .results.budget import ResourceBudget, _reserve_owned_ram, _release_owned_ram, _cleanup_owner11from .results.errors import ResultClosedError121314class ExtensionResources:15 """一次调用或状态所有的预算视图；动态费用与固定状态费用分开登记。"""1617 def __init__(self, budget, fixed=0):18 """参数: 共享budget或None及固定状态字节。返回: 资源所有者。异常: 预约失败可清理。"""19 self._budget = budget if budget is not None else _BUDGET.get()20 if self._budget is None:21 self._budget = ResourceBudget()22 self._held_cost = self._fixed_cost = self._dynamic_cost = 023 self._closed = self._closing = False24 try:25 if fixed:26 _reserve_owned_ram(self,fixed,'_fixed_cost')27 except BaseException as error:28 _cleanup_owner(self,error,'extension_cleanup_owner','扩展状态预约')29 raise3031 def reserve(self, amount):32 """参数: 即将分配的非负字节数。返回: None。异常: 预算超额、取消或已关闭。"""33 if self._closed or self._closing:34 raise ResultClosedError('扩展资源已关闭')35 _reserve_owned_ram(self,amount,'_dynamic_cost')3637 def release(self, amount):38 """参数: 本回调已预付的动态字节数。返回: None。异常: 不允许退其他所有者或固定额度。"""39 if self._closed or self._closing:40 raise ResultClosedError('扩展资源已关闭')41 if type(amount) is not int or amount < 0 or amount > self._dynamic_cost:42 raise ValueError('扩展只能归还自己实际申请的动态费用')43 _release_owned_ram(self,amount,'_dynamic_cost')4445 def close(self,cancelled=False):46 """退回全部自有费用。参数: cancelled为上游取消提示。返回: None。异常: 账本异常后可重试。"""47 if self._closed:48 return49 self._closing = True50 try:51 _release_owned_ram(self,self._dynamic_cost,'_dynamic_cost')52 _release_owned_ram(self,self._fixed_cost,'_fixed_cost')53 finally:54 self._closed = self._held_cost == 0555657def _check_scalar(value, port):58 """验证回调输出模型。参数: value与静态输出端口。返回: 原ScalarValue。异常: 类型错误。"""59 if not isinstance(value,ScalarValue) or value.dtype is not port.dtype:60 raise TypeError('宿主扩展返回的ScalarValue与声明dtype不符')61 return value626364def _check_values(values, port, backend):65 """验证已选样本的布局和类型。参数: 输入、静态端口及后端。返回: 个数。异常: 布局或质量异常。"""66 if backend == 'reference':67 if type(values) is not tuple or any(not _check_scalar(v,port).valid for v in values):68 raise ValueError('标量归约/网格扩展只接已选有效样本tuple')69 return len(values)70 from .core.numeric_numpy import NumericArray71 if not isinstance(values,NumericArray) or values.dtype is not port.dtype or not values.valid.all():72 raise ValueError('数组归约/网格扩展只接已选有效的同型NumericArray')73 return len(values.values)747576def point_call(spec, backend, args):77 """按声明后端运行有限逐点函数，任何退出路径都清理本次工作区。7879 参数: spec为编译好的FunctionSpec；args为typed值tuple，按静态端口数量排列。80 返回: ScalarValue或NumericArray。异常: 能力、结构、回调、取消或额度错误原样传播。81 """82 if type(spec) is not FunctionSpec or len(args) != len(spec.inputs):83 raise ValueError('逐点扩展实参与声明不同')84 implementation = spec.require_backend(backend)85 resources = ExtensionResources(None)86 try:87 if backend == 'reference':88 # 任一必要输入无效时直接传播全部质量原因，不要求回调解释无效占位。89 from .core.types import QualityFlag as Q90 quality = Q.NONE91 for value,port in zip(args,spec.inputs):92 quality |= _check_scalar(value,port).quality93 result = (ScalarValue(None,spec.output.dtype,False,quality) if quality else94 _check_scalar(implementation(args,resources),spec.output))95 else:96 import numpy as np97 from .core.numeric_numpy import DTYPES, NumericArray, _output98 size = len(args[0].values)99 if any(not isinstance(v,NumericArray) or v.dtype is not p.dtype or len(v.values) != size100 for v,p in zip(args,spec.inputs)):101 raise TypeError('数组逐点扩展的输入dtype或行数错误')102 quality = np.zeros(size,dtype=np.uint16)103 for value in args:104 quality |= value.quality105 selected = quality == 0106 valid_count = int(np.count_nonzero(selected))107 out = np.zeros(size,dtype=DTYPES[spec.output.dtype])108 if valid_count:109 # 只把全部必要值已知的行交给宿主，输出仍须回到原批坐标。110 operands = args if valid_count == size else tuple(v.take(selected) for v in args)111 column = implementation(operands,resources)112 if not isinstance(column,NumericArray) or column.dtype is not spec.output.dtype or len(column.values) != valid_count:113 raise TypeError('数组逐点扩展返回的dtype或行数错误')114 out[selected] = column.values115 quality[selected] = column.quality116 result = _output(spec.output.dtype,out,quality)117 resources.close()118 return result119 except BaseException as error:120 _cleanup_owner(resources,error,'extension_cleanup_owner','逐点扩展')121 raise122123124class _HostedState:125 """持有一个工厂状态及其独立资源账本；失败后只允许取消和关闭。"""126127 def __init__(self,spec,backend,budget,methods):128 """参数: typed描述、后端、共享预算、必需方法。返回: owner。异常: 工厂或协议错误。"""129 factory = spec.require_backend(backend)130 self.spec,self.backend = spec,backend131 self.state = None132 self._closed = self._closing = self._failed = self._cancelled = False133 self.resources = ExtensionResources(budget,spec.state_bytes)134 try:135 self.state = factory(self.resources)136 if any(not callable(getattr(self.state,m,None)) for m in (*methods,'cancel','close')):137 raise TypeError(f'扩展 {spec.name} 的工厂状态缺少协议方法')138 except BaseException as error:139 self._failed = True140 # 不合格状态也尽量归还可关闭的宿主资源，不能因缺cancel遗失预算。141 if not callable(getattr(self.state,'close',None)):142 self.state = None143 _cleanup_owner(self,error,'extension_cleanup_owner','宿主扩展构造')144 raise145146 def _require_open(self):147 """参数: 无。返回: None。异常: 拒绝使用已失败或已关闭状态。"""148 if self._closed or self._closing or self._failed:149 raise ResultClosedError('宿主扩展已经失败或关闭')150151 def close(self,cancelled=False):152 """通知取消、关闭状态、归还费用。参数: cancelled为上游取消。返回: None。异常: 原清理错误可重试。"""153 if self._closed:154 return155 self._closing = True156 if self.state is not None:157 if (cancelled or self._failed) and not self._cancelled:158 self._cancelled = True159 cancel = getattr(self.state,'cancel',None)160 if callable(cancel):161 cancel()162 self.state.close()163 self.state = None164 self.resources.close()165 self._closed = True166167168class HostedReduction(_HostedState):169 """每个统计范围独享一次样本归约，已结束的状态不会再次消费。"""170171 def __init__(self,spec,backend,budget):172 """参数: ReductionSpec、声明后端和共享账本。返回: owner。异常: 工厂、预算或能力错。"""173 if type(spec) is not ReductionSpec:174 raise TypeError('归约需要ReductionSpec')175 self._result = None176 super().__init__(spec,backend,budget,('consume','finalize','merge') if spec.mergeable else ('consume','finalize'))177178 def consume(self,values):179 """参数: 本块已选真实有效值。返回: None。异常: 错类型、已结束或回调/预算失败。"""180 self._require_open()181 if self._result is not None:182 raise ResultClosedError('宿主归约已结算')183 try:184 if _check_values(values,self.spec.input,self.backend):185 self.state.consume(values)186 except BaseException:187 self._failed = True188 raise189190 def merge(self,other):191 """参数: 同一描述/后端的另一未结算状态。返回: None。异常: 不可合并或回调失败。"""192 self._require_open()193 if (not self.spec.mergeable or not isinstance(other,HostedReduction) or other is self194 or other.spec is not self.spec or other.backend != self.backend):195 raise ValueError('只能合并明确支持且描述/后端相同的不同归约状态')196 other._require_open()197 if self._result is not None or other._result is not None:198 raise ResultClosedError('已结算归约不能合并')199 try:200 self.state.merge(other.state)201 except BaseException:202 self._failed = True203 raise204205 def finalize(self):206 """参数: 无。返回: 静态声明dtype的最终ScalarValue。异常: 错误输出或回调失败。"""207 self._require_open()208 if self._result is None:209 try:210 self._result = _check_scalar(self.state.finalize(),self.spec.output)211 except BaseException:212 self._failed = True213 raise214 return self._result215216217class HostedTransform(_HostedState):218 """只持有此时序请求的状态，按原时基的实际网格递增推进。"""219220 def __init__(self,spec,backend,budget):221 """参数: TransformSpec、后端和共享账本。返回: owner。异常: 工厂、预算或能力错。"""222 if type(spec) is not TransformSpec:223 raise TypeError('时序扩展需要TransformSpec')224 self._last = None225 super().__init__(spec,backend,budget,('advance',))226227 def advance(self,time_ns,window_ns,values,weights_ns):228 """参数: 实际递增网格、已声明窗口及等长有效值/正纳秒权重。返回: ScalarValue。异常: 协议或回调失败。"""229 self._require_open()230 if (type(time_ns) is not int or self._last is not None and time_ns <= self._last231 or type(window_ns) is not int or not 0 < window_ns <= self.spec.history_ns):232 raise ValueError('时序扩展需要递增网格及声明历史内的窗口')233 try:234 if len(weights_ns) != _check_values(values,self.spec.input,self.backend):235 raise ValueError('时序扩展的有效区间权重与值长度不同')236 if self.backend == 'reference':237 from fractions import Fraction238 if type(weights_ns) is not tuple or any(type(w) not in (int,Fraction) or w <= 0 for w in weights_ns):239 raise ValueError('参考时序权重须为正int/Fraction纳秒tuple')240 else:241 import numpy as np242 if (type(weights_ns) is not np.ndarray or weights_ns.ndim != 1243 or weights_ns.dtype not in (np.dtype('int64'),np.dtype('float64'))244 or np.any(weights_ns <= 0) or not np.isfinite(weights_ns).all()):245 raise ValueError('数组时序权重须为正的一维有限int64/float64纳秒列')246 result = _check_scalar(self.state.advance(time_ns,window_ns,values,weights_ns),self.spec.output)247 self._last = time_ns248 return result249 except BaseException:250 self._failed = True251 raise252253254def advance_grid(host, time_ns, window_ns, latest_first, weights, conversion):255 """将已覆盖区间按正序交给宿主，并保留实际或所选周期的纳秒权重。256257 参数: host为当前有状态网格，time_ns/window_ns为实际网格/窗口，latest_first为258 从右向左读到的有效原值，weights为相应纳秒，conversion为已验证的目标单位换算。259 返回: ScalarValue，保留宿主自身失败质量或部分状态。异常: 权重/回调或协议异常。260 """261 integral = all(w.denominator == 1 for w in weights)262 if host.backend == 'reference':263 from .core.calculation_output_reference import convert_result_reference as convert264 inputs = tuple(ScalarValue(v,host.spec.input.dtype) for v in reversed(latest_first))265 durations = tuple(int(w) if w.denominator == 1 else w for w in reversed(weights))266 else:267 import numpy as np268 from .core.numeric_numpy import DTYPES,_output269 from .core.calculation_output_numpy import convert_result_numpy as convert270 array = np.asarray(latest_first,dtype=DTYPES[host.spec.input.dtype])[::-1]271 inputs = _output(host.spec.input.dtype,array,np.zeros(len(array),dtype=np.uint16))272 # 只为数值加权转换分数，时间线的窗口裁剪与驱动位置仍保持精确有理数。273 durations = np.fromiter(reversed(weights),dtype=np.int64 if integral else np.float64,count=len(weights))274 result = host.advance(time_ns,window_ns,inputs,durations)275 target = conversion.dtype(result.dtype)276 if not result.valid:277 return ScalarValue(None,target,False,result.quality)278 value, quality = convert(result.value,result.dtype,conversion)279 return ScalarValue(value,target,not quality,quality,partial=result.partial and not quality)280