chronaxis.legacy.sum
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在预算内按原求值顺序累计调用方已经算好的旧口径贡献。23输入为LegacySumContribution的有限精确tuple，每项保留LegacyDecodedUpdate4身份并显式选择有限float贡献或跳过原因。输出为未舍入的冻结快照；本层不求5表达式、不再乘period/factor/dt，不把原quality推断成TTL、缓存或跳过政策。6legacy_policy必须由提供者明确填写，不证明完整复现ChrgInfo的旧行为。78累计器只保存一个来源的最后顺序、总数、计数及四种原因计数。每批先付新状态9和临时列费用，再独立核顺序及数值，本批另付的临时费用实际释放后才一次提交候选状态。10内部历史AnalysisRunner可明确预付一条的工作空间；本owner保持这笔费用直到close，11单条计算复用它，多条仅另付不足部分，不能借共享账本中其他owner的余额抵账。12参考路径逐项使用Python浮点加法；NumPy路径按需加载独立数组核，不共享参考13数值计算。输入及返回快照归调用方，本层估算不包含其已有内存或整个进程RSS。14本模块不读写文件、不调用旧reader/解码器；close只释放本owner的实际RAM槽。15"""1617# 冻结模型及候选复制只依赖标准库，数组依赖仅在明确选择且有贡献时加载。18from collections.abc import Mapping19from copy import copy20from dataclasses import dataclass21from enum import Enum22from math import isfinite23from types import MappingProxyType2425from ..core.validation import require_identifier, require_text26from ..results.budget import ResourceBudget, _cleanup_owner, _release_owned_ram, _reserve_owned_ram27from ..results.errors import ResultClosedError28from .updates import LegacyDecodedUpdate, LegacyUpdateOrder293031class LegacySkipReason(Enum):32 """提供者明确选择的一次跳过原因，不由累计器根据原值或质量推断。3334 四个固定成员覆盖输入缺项、条件不成立、求值错误和非有限输入。35 枚举无可变状态；原因仅解释该提供者为何未给出有限贡献，不补造旧缓存事实。36 """3738 MISSING_INPUT = "missing_input"39 CONDITION_FALSE = "condition_false"40 EVALUATION_ERROR = "evaluation_error"41 NONFINITE_INPUT = "nonfinite_input"424344# 原因域固定，内部计数tuple和公开冻结映射都保持此声明顺序及全部零项。45_SKIP_REASONS = tuple(LegacySkipReason)464748def _require_contribution(value: float | None, skip_reason: LegacySkipReason | None) -> None:49 """检查有限原生贡献与显式跳过原因互斥，不转换或修补数值。5051 参数: value为原生有限float或None；skip_reason为固定枚举或None。52 返回: None；合法数值和显式跳过均保持原样。53 异常: TypeError为类型不符；ValueError为非有限数值或贡献/原因关系矛盾。54 """55 # 先限制明确的原因域；None贡献只能表示提供者已选择跳过，不能省略原因。56 if skip_reason is not None and not isinstance(skip_reason, LegacySkipReason):57 raise TypeError("skip_reason必须为LegacySkipReason或None")58 if value is None:59 if skip_reason is None:60 raise ValueError("无贡献时必须显式提供skip_reason")61 return62 # int和bool即使可转float也不能隐藏提供者采用的数值政策。63 if type(value) is not float:64 raise TypeError("贡献必须为原生float或None")65 if not isfinite(value):66 raise ValueError("贡献必须为有限float；非有限输入应由提供者显式跳过")67 if skip_reason is not None:68 raise ValueError("有限贡献不能同时携带skip_reason")697071@dataclass(frozen=True)72class LegacySumContribution:73 """关联一次更新身份与提供者已求出的贡献，不保存新的信号缓存。7475 update保留原模型对象；value是有限原生float，或与明确skip_reason配对的76 None。即使update.quality非零，有限贡献也不由本层自动否决。该对象由调用方77 提供，不证明旧表达式、触发门控、缺失处理或真实日志逐次出口已经接通。78 """7980 update: LegacyDecodedUpdate # 此次求值的原身份，引用保持不变。81 value: float | None # 已计算一次的贡献，不再施加时间或单位倍率。82 skip_reason: LegacySkipReason | None = None # 仅None贡献需要明确原因。8384 def __post_init__(self) -> None:85 """验证真实更新模型及互斥的贡献/跳过结构。8687 参数: 无，读取update、value和skip_reason。88 返回: None；不复制update或修改其原值、质量。89 异常: TypeError为非模型或数值/原因类型不符；ValueError为结构或有限性错误。90 """91 # 保留已验证的原更新引用，再独立核对贡献结构，不重造或改写求值身份。92 if not isinstance(self.update, LegacyDecodedUpdate):93 raise TypeError("update必须为LegacyDecodedUpdate")94 _require_contribution(self.value, self.skip_reason)959697@dataclass(frozen=True)98class LegacySumSnapshot:99 """保存固定大小的未舍入累计事实，明确区分没有贡献与有效零和。100101 evaluation_count包括跳过，contribution_count只数实际加法项；四键102 skip_counts复制冻结且始终含零项。没有求值和全部跳过分别以reason说明，103 只有曾接收有效贡献才给float总数。快照归调用方，可在累计器关闭后保留；104 不保存逐次列表，也不声称普通时间积分的覆盖率或旧整文件行为全等。105 """106107 session_id: str # 原会话身份。108 source: str # 单一来源身份。109 legacy_policy: str # 提供者显式声明的贡献及跳过政策文本。110 value: float | None # 未舍入总数；没有有效贡献时为None。111 evaluation_count: int # 所有已接受求值，包括跳过。112 contribution_count: int # 实际有限贡献数量，包括值为0的项。113 skipped_count: int # 显式跳过的求值数量。114 skip_counts: Mapping[LegacySkipReason, int] # 固定四键的独立冻结映射。115 reason: str | None # no_evaluations/all_skipped；有贡献时为None。116117 def __post_init__(self) -> None:118 """核对有限计数关系与空结果原因，并冻结固定四键映射。119120 参数: 无，读取所有快照字段。121 返回: None；skip_counts转换成新的MappingProxyType，其他事实不改写。122 异常: TypeError为字段类型不符；ValueError为计数、原因域或结果关系矛盾；123 MemoryError为固定小映射分配失败，不返回部分快照。124 """125 # 身份和政策仍须明确，合法的数值不能掩盖缺失的来源或政策声明。126 require_identifier(self.session_id, "session_id")127 require_identifier(self.source, "source")128 require_text(self.legacy_policy, "legacy_policy")129 # 三种计数分别非负，且求值数必须由贡献和跳过完整闭合，不能遗漏已接受项。130 for count in (self.evaluation_count, self.contribution_count, self.skipped_count):131 if type(count) is not int:132 raise TypeError("快照计数必须为原生非布尔int")133 if count < 0:134 raise ValueError("快照计数必须非负")135 if self.evaluation_count != self.contribution_count + self.skipped_count:136 raise ValueError("求值数必须等于贡献数加跳过数")137 # 固定键域不能由可变Mapping实现注入额外原因，仍独立复制调用方dict。138 if type(self.skip_counts) not in (dict, MappingProxyType):139 raise TypeError("skip_counts需要dict或MappingProxyType")140 if set(self.skip_counts) != set(_SKIP_REASONS):141 raise ValueError("skip_counts必须恰好包含四个LegacySkipReason")142 copied = {}143 for reason in _SKIP_REASONS:144 count = self.skip_counts[reason]145 if type(count) is not int:146 raise TypeError("跳过原因计数必须为原生非布尔int")147 if count < 0:148 raise ValueError("跳过原因计数必须非负")149 copied[reason] = count150 if sum(copied.values()) != self.skipped_count:151 raise ValueError("四种原因计数之和必须等于skipped_count")152 # 用实际贡献数区分未求值、全跳过与有贡献，不用总数的真假判断是否为空。153 expected_reason = None if self.contribution_count else (154 "no_evaluations" if self.evaluation_count == 0 else "all_skipped"155 )156 if self.reason != expected_reason:157 raise ValueError("快照reason与实际贡献数量不一致")158 # 有贡献时零和也是有限结果；没有贡献时必须保留None，不能补成正常0。159 if self.contribution_count:160 if type(self.value) is not float:161 raise TypeError("有贡献的快照需要原生float总数")162 if not isfinite(self.value):163 raise ValueError("快照总数必须有限")164 elif self.value is not None:165 raise ValueError("没有有效贡献时value必须为None")166 object.__setattr__(self, "skip_counts", MappingProxyType(copied))167168169def _sum_reference(carry: float, values: tuple[float, ...]) -> float:170 """以独立Python浮点逐项左折叠，发现任一中间溢出立即拒绝本批。171172 参数: carry是此前有限总数；values是按原求值顺序选出的有限float tuple。173 返回: float；无舍入、补偿求和或块和重排的候选总数。174 异常: OverflowError为任一中间累计非有限；取消及内存异常原样传播。175 """176 candidate = carry177 try:178 # 每次加法承接前一项的真实浮点尾数；逐中间值检查，禁止靠后续抵消修复溢出。179 for value in values:180 candidate = candidate + value181 if not isfinite(candidate):182 raise OverflowError("legacy_sum中间累计非有限，整批拒绝")183 return candidate184 finally:185 # 异常traceback可能由调用方长期保留；不得经纯核参数保活本层选值tuple。186 values = None187 candidate = None188189190class LegacySumAccumulator:191 """拥有单来源有限累计状态和实际预算槽，不拥有调用方更新或共享账本。192193 _state一次保存顺序器、总数、贡献数及四原因计数。候选顺序器用copy.copy194 复制，共享的旧状态tuple不可变；包括skip的全部记录仍检查会话和原序。195 _fixed_cost覆盖本层有限元数据及候选重叠；_workspace_cost可预付单条工作区，196 _batch_cost只覆盖本批超过预付容量的临时费用。三者之和为_held_cost。197 费用清理未完成时拒绝新消费，close可按同owner重试。198 """199200 def __init__(self, budget: ResourceBudget, session_id: str, source: str, *,201 legacy_policy: str, backend: str = "numpy", max_batch_rows: int = 64,202 prepay_single: bool = False):203 """先安装责任槽、校验配置并预付费用，再建立空顺序及累计状态。204205 参数: budget为借用共享账本；session_id/source为固定身份；legacy_policy206 是必填非空政策；backend为reference或numpy；max_batch_rows为正原生int；207 prepay_single为原生bool，True时本owner构造预付4352字节单条工作区，208 供内部历史AnalysisRunner在共享余额用尽时继续已有单条计算，不缓存贡献；209 默认False保原批行为，多条调用仍须先预约实际所需空间的不足部分。210 返回: 新累计器；初始无求值、无贡献，不把初始加法carry当正常结果。211 异常: TypeError/ValueError为配置不符；预算及取消原错传播。构造清理持续212 失败时，原异常的legacy_sum_cleanup_owner保留真实未返回owner。213 """214 if not isinstance(budget, ResourceBudget):215 raise TypeError("budget必须为ResourceBudget")216 # 首次外部预算回调前安装全部实际费用和关闭槽，避免未返回对象失去责任。217 self._budget = budget218 self._held_cost = self._fixed_cost = self._workspace_cost = self._batch_cost = 0219 self._closed = self._closing = False220 self._state = None221 self._session_id = self._source = self._legacy_policy = None222 self._backend = None223 self._max_batch_rows = 0224 try:225 # 先检查调用配置，避免无效来源、政策或块上限进入已付费的累计状态。226 require_identifier(session_id, "session_id")227 require_identifier(source, "source")228 require_text(legacy_policy, "legacy_policy")229 if type(backend) is not str or backend not in ("reference", "numpy"):230 raise ValueError("backend仅支持reference或numpy")231 if type(max_batch_rows) is not int:232 raise TypeError("max_batch_rows必须为正原生int")233 if max_batch_rows <= 0:234 raise ValueError("max_batch_rows必须为正原生int")235 if type(prepay_single) is not bool:236 raise TypeError("prepay_single必须为原生bool")237 # 文本计费按最多UTF-32字符及候选/快照引用余量，固定开销覆盖有限状态。238 cost = 16384 + 8 * (len(session_id) + len(source) + len(legacy_policy))239 _reserve_owned_ram(self, cost, "_fixed_cost")240 # 预约前已安装独立责任槽，预约成功后取消也由本未返回owner清理。241 # 4352沿用单条原公式4096+256，不减原估算，也不替其他owner付款。242 if prepay_single:243 _reserve_owned_ram(self, 4352, "_workspace_cost")244 self._session_id, self._source, self._legacy_policy = session_id, source, legacy_policy245 self._backend, self._max_batch_rows = backend, max_batch_rows246 self._state = (LegacyUpdateOrder(session_id, source), 0.0, 0, (0, 0, 0, 0))247 except BaseException as exc:248 _cleanup_owner(self, exc, "legacy_sum_cleanup_owner", "legacy累计构造")249 raise250251 @property252 def closed(self) -> bool:253 """读取完全关闭事实；参数: 无；返回: bool；异常: 无。"""254 return self._closed255256 def _ensure_open(self) -> None:257 """拒绝关闭或有待恢复费用的owner，防止新批覆盖旧责任槽。258259 参数: 无。返回: None。异常: ResultClosedError为已关闭或正在清理。260 """261 if self._closed or self._closing:262 raise ResultClosedError("legacy累计器已关闭或正在清理")263264 def _candidate(self, contributions: tuple[LegacySumContribution, ...]) -> tuple:265 """在已预付空间内生成独立候选，返回时不保留逐次值临时列。266267 参数: contributions为本批借用tuple，长度已经检查。268 返回: tuple；新顺序器、候选总数、贡献数和四原因计数。269 异常: TypeError/ValueError为贡献或顺序错误；OverflowError为中间累计溢出；270 数组导入、分配或取消原样传播，原_state始终不被修改。271 """272 candidate_order = skips = selected = values = None273 try:274 # 只复制有限顺序状态，后续accept只改候选，保留整批失败时的原顺序事实。275 order, carry, contribution_count, previous_skips = self._state276 candidate_order = copy(order)277 skips = list(previous_skips)278 selected = []279 # 每项先核模型再推进候选，后续坏项不能让此前部分求值提前提交。280 for contribution in contributions:281 if not isinstance(contribution, LegacySumContribution):282 raise TypeError("consume仅接受LegacySumContribution")283 # 即使整批全部跳过，原更新身份和求值顺序也必须完整推进及核对。284 candidate_order.accept(contribution.update)285 if contribution.value is None:286 skips[_SKIP_REASONS.index(contribution.skip_reason)] += 1287 else:288 selected.append(contribution.value)289 # 不添加跳过占位0；全skip保持原carry，尚无贡献时快照仍由计数判为None。290 if selected:291 values = tuple(selected)292 if self._backend == "reference":293 carry = _sum_reference(carry, values)294 else:295 from .sum_numpy import sum_numpy296 carry = sum_numpy(carry, values)297 return candidate_order, carry, contribution_count + len(selected), tuple(skips)298 finally:299 # 先断开本层新建的候选、选值和原因容器；保留原输入和异常本身。300 # 否则失败traceback会在批费用释放后继续保活这些临时副本。301 candidate_order = skips = selected = values = None302303 def _release_failed_batch(self, primary: BaseException) -> None:304 """有限重试本批实际费用，清理副错只附注原异常，不关闭正常累计状态。305306 参数: primary为原校验、计算、预算或取消异常。307 返回: None；持续未释费用设置_closing，禁止覆盖该责任槽。308 异常: 清理错误只记notes；真正剩余费用由原异常携带的owner.close恢复。309 """310 # 至多重试两次已登记费用，不无限阻塞原错；剩余责任留给显式close恢复。311 for attempt in range(2):312 if not self._batch_cost:313 break314 try:315 _release_owned_ram(self, self._batch_cost, "_batch_cost")316 except BaseException as cleanup:317 primary.add_note(f"legacy批费用清理第{attempt + 1}次失败: {type(cleanup).__name__}")318 if self._batch_cost:319 self._closing = True320321 def consume(self, contributions: tuple[LegacySumContribution, ...]) -> None:322 """整批核验并累计，临时差额实际释放成功后才一次安装全部候选事实。323324 参数: contributions为精确tuple，长度不超过max_batch_rows；空tuple合法。325 已预付单条空间时仅预约超过该自有容量的差额；默认仍预约完整本批费用。326 返回: None；成功提交全部项，任何失败不推进原数值、计数或顺序。327 异常: TypeError/ValueError为类型、批上限、来源或顺序；OverflowError为累计328 非有限；ResultClosedError为关闭中。预算、导入、分配或取消原样传播并329 附legacy_sum_cleanup_owner。清理成功后普通失败可用合法后续批恢复。330 """331 self._ensure_open()332 candidate = None333 try:334 # 先限定真实有限容器和批上限；空批不建立候选，也不申请无用工作区。335 if type(contributions) is not tuple:336 raise TypeError("consume需要精确tuple")337 if len(contributions) > self._max_batch_rows:338 raise ValueError("本批超过max_batch_rows")339 if not contributions:340 return341 # 每项256覆盖选值list/tuple、seed/累计/有限性数组及验证临时量；固定342 # 4096覆盖顺序器和四计数候选。输入update.values不复制也不深遍历计费。343 extra = max(0, 4096 + len(contributions) * 256 - self._workspace_cost)344 if extra:345 _reserve_owned_ram(self, extra, "_batch_cost")346 candidate = self._candidate(contributions)347 # _candidate已退出，所有逐次临时列已丢弃；这里只保固定候选与旧状态重叠。348 _release_owned_ram(self, self._batch_cost, "_batch_cost")349 except BaseException as exc:350 # 释放预算前丢弃尚未提交的固定候选，异常frame不能继续替它保留引用。351 candidate = None352 self._release_failed_batch(exc)353 exc.legacy_sum_cleanup_owner = self354 raise355 # 提交前不再调用预算或数值代码；候选分配与临时费用收尾均已完成。356 self._state = candidate357358 def snapshot(self) -> LegacySumSnapshot:359 """返回调用方拥有的独立冻结快照，不舍入、不封口或证明来源EOF。360361 参数: 无。返回: LegacySumSnapshot；四原因键始终存在，计数为0也不省略。362 异常: ResultClosedError为关闭中；分配或取消原样传播并附真实恢复owner。363 """364 self._ensure_open()365 try:366 order, total, contribution_count, skips = self._state367 # 固定原因域每次都含全部零项，返回的小映射独立于内部计数tuple。368 counts = dict(zip(_SKIP_REASONS, skips))369 # 空结果按求值及贡献数分类，已有累计不会被后来全跳过的批次清空。370 reason = None if contribution_count else (371 "no_evaluations" if order.count == 0 else "all_skipped"372 )373 return LegacySumSnapshot(374 self._session_id, self._source, self._legacy_policy,375 total if contribution_count else None, order.count, contribution_count,376 order.count - contribution_count, counts, reason,377 )378 except BaseException as exc:379 exc.legacy_sum_cleanup_owner = self380 raise381382 def close(self) -> None:383 """释放本累计器实际费用，可在释放前后取消或持续失败后重试同一owner。384385 参数: 无。返回: None；只有所有自有槽实际归零才closed=True，重复close无操作。386 异常: 原预算或取消异常保持，附legacy_sum_cleanup_owner；半关闭时拒绝继续387 consume/snapshot，不释放调用方借款，不使已返回快照失效。388 """389 # 只有完全关闭才幂等返回，避免重复close再次触发预算回调或抵销借款。390 if self._closed:391 return392 self._closing = True393 # 本层无FD或子owner；先丢自有状态引用，释放后取消也不留下未计费的新状态。394 self._state = None395 try:396 _release_owned_ram(self, self._batch_cost, "_batch_cost")397 _release_owned_ram(self, self._workspace_cost, "_workspace_cost")398 _release_owned_ram(self, self._fixed_cost, "_fixed_cost")399 except BaseException as exc:400 exc.legacy_sum_cleanup_owner = self401 raise402 self._session_id = self._source = self._legacy_policy = None403 self._closed = True404