chronaxis.execution.record_session_calculations
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在原AnalysisRunner会话内合并所需来源并执行会话时间指标。23共享同刻最新状态；每项指标只保自己的驱动点和归约状态。输入只读一次，4不创建普通历史、消息历史或完整贡献。均值仅保其算法必需的值与权重暂存。5每来源只保有界前瞻，同刻跨来源更新收齐后求值。所有输入真正EOF并关闭6后才交付最终结果，失败不发布已累计的前缀。7"""89from ..core.moments import SAMPLE_MOMENTS10from ..core.quantiles import SAMPLE_QUANTILES11from dataclasses import replace12from ..core.models import Interval13from functools import partial14from ..core.calculations import CalculationDefinition15from ..legacy.records import field_to_dict16from ..results.errors import ResultClosedError, AnalysisError17from ..results.spool import _reserve_owned_ram, _release_owned_ram, _cleanup_owner18from ..results.summary import SummaryCalculationRow19from ..results.summary import SummaryCoverageCalculationRow, SummaryHostedCalculationRow20from ..results.summary import SummaryEventRow21from ..core.types import QualityFlag as Q22from ..results.summary import SummarySequenceRow23from ..sources.can_spec import can_field_from_key24from .history import _metadata_bytes25from .state import LatestSignalState26from .driver_expression import DriverExpression27from .record_integral import RecordIntegral28from .record_time_calculation import RecordTimeCalculation29from .record_source_cursor import RecordSourceCursor30from .groups import DecodedTimeGroup31from .current_message_calculation import CurrentMessageCalculation32from ..legacy.sum import LegacySkipReason33from .record_rule_timeline import RecordRuleTimeline34from .record_sequences import RecordSequences35from ..planning.intervals import IntervalChoice36from .record_event_durations import RecordEventDurations,is_event_duration37from ..core.counts_v4 import OBJECT_COUNTS38from .record_object_counts import object_count_results39from .record_scope_calculations import RecordScopeCalculations40from .record_time_indexes import RecordTimeIndexes41from .record_windows import RecordWindows42from .record_exports import RecordExports43from ..results.summary import SummarySegmentRow44from ..results.summary import SummaryContributionRow45from ..results.summary import SummaryInputRow46from .record_diagnostics import RecordDiagnostics47from .record_selection import RecordSelection48from ..results.summary import SummarySelectionRow495051class RecordSessionCalculations:52 """拥有实时计算和读取子资源；计划与最终结果仓只借用。"""5354 def __init__(self, plan, budget, work_dir, *, requirements=None, declaration_owner=None, direct_integrals: bool=False):55 """预约有限声明空间，不启动读取。5657 参数: plan已校验计划、budget共享预算、work_dir已有私有运行目录；58 declaration_owner为持有共享声明的父RunSession，独立使用时省略；59 direct_integrals仅由公共入口的auto选择启用，仍需首帧前证明适用。60 返回: 新owner。异常: 构造原错附record_session_cleanup_owner供父会话接管。61 """62 if type(direct_integrals) is not bool:63 raise TypeError('直接积分内部选择需要bool')64 from ..planning.backend_selection import execution_plan65 plan = execution_plan(plan)66 self._plan, self._budget, self._path = plan, budget, work_dir67 self._declaration_owner = declaration_owner68 self._full_domain = plan.dataset.domain is None69 self._observed_domain = None70 if requirements is None:71 from ..planning.input_requirements import collect_input_requirements72 requirements=collect_input_requirements(plan)73 self._requirements=requirements74 self._held_cost = self._fixed_cost = self._group_cost = 075 self._state = self._groups = self._adapter = self._read = self._pending = None76 self._calculations = self._expressions = self._cursors = None77 self._messages = None78 # 汇总请求共享一份有界目标消息批；完整贡献保原逐消息输出次序。79 self._message_buffer = []80 self._message_cost = 081 self._message_batch_rows = (1 if plan.outputs.contributions == 'full'82 else min(plan.execution.batch_rows, 4096))83 self._source_ends = None84 self._rules = None85 self._sequences = None86 self._event_durations=None87 self._scope_calculations=None88 self._segment_counts = {}89 self._transform_stages = []90 self._transform_aliases = frozenset()91 self._time_indexes=self._windows=self._exports=None92 self._diagnostics=None93 self._direct_requested = direct_integrals94 self._direct_binding = self._direct_owner = None95 self._rules_begun = False96 self._constant_calculations=frozenset()97 self._spool = None98 self._aggregations = self._append_result = None99 self._next_sequence = 0100 # 指标状态随原写出计数；不为了概况回读结果仓。101 self._outcome_counts = dict(complete=0, partial=0, indeterminate=0, no_observation=0)102 self._closed = self._closing = self._failed = self._started = False103 try:104 # 本层只创建子资源槽、来源索引和计数；表达式/信号声明仍借用原计划。105 count = (len(plan.calculations) + len(plan.message_calculations) + len(plan.rules)106 + len(plan.windows) + len(plan.sources) + len(plan.signals))107 _reserve_owned_ram(self, 131584 + 4096 * count, '_fixed_cost')108 self._calculations, self._expressions, self._cursors, self._messages = [], [], [], []109 self._source_ends = {}110 self._rules = []111 except BaseException as error:112 _cleanup_owner(self, error, 'record_session_cleanup_owner', '记录流会话构造')113 raise114115 def _new(self, factory, recovery, *args, **kwargs):116 """保留尚未交付的真实子owner。参数: 工厂/恢复属性/参数。返回: owner。异常: 原错。"""117 try:118 return factory(*args, **kwargs)119 except BaseException as error:120 self._pending = getattr(error, recovery, None)121 raise122123 def _consume_group(self, group):124 """先原子提交同刻组，再分别按所选报文推进各指标。125126 参数: 完整DecodedTimeGroup。返回: None。异常: 表达式/预算/累计原错。127 非驱动无效更新仅截断待算值，不新增周期；条件截止独立处理。128 """129 try:130 if self._diagnostics is not None:131 raw_sources={c.source_name for c in self._cursors if c._raw}132 for sample in group.samples:133 if sample.name in self._transform_aliases:134 continue # 网格样本不是实际来源的报文或原字段诊断。135 if sample.source_ref.source_name not in raw_sources:136 values_needed=set(self._requirements.ordinary_sample_aliases)|set(self._requirements.message_aliases)137 clock_only=sample.name not in values_needed138 self._diagnostics.consume(sample.source_ref.source_name,sample.sequence,sample.time_ns,139 'signal_clock' if clock_only else 'signal_sample',sample.source_ref.field,140 {} if clock_only else {sample.name:sample.value.value},sample.value.quality)141 if self._time_indexes is not None:142 self._time_indexes.observe_group(group)143 if self._event_durations is not None:144 self._event_durations.observe_group(group)145 for owner in self._rules:146 ends=[self._source_ends[source] for source in owner.sources if source in self._source_ends]147 if len(ends)==len(owner.sources) and ends and max(ends)<group.time_ns:148 owner.finish(max(ends))149 owner.before_group(group.time_ns)150 owner.clock_group(group)151 if self._sequences is not None:152 self._sequences.flush_before(group.time_ns)153 if self._exports is not None and self._exports._signal is not None:154 self._exports._signal.observe(group)155 if self._state is not None:156 selected=tuple(s for s in group.samples if s.name in self._state.definitions)157 if selected:158 self._state.apply(group if len(selected)==len(group.samples) else DecodedTimeGroup(group.time_ns,selected))159 if self._scope_calculations is not None:160 self._scope_calculations.observe_group()161 if self._sequences is not None:162 self._sequences.drivers(group)163 for index,(calculation, expressions) in enumerate(zip(self._calculations, self._expressions)):164 if index in self._constant_calculations:165 continue166 value, condition = expressions167 row = value.evaluate()168 selected = condition.evaluate() if condition is not None else None169 if value.invalidated_at is not None:170 calculation.expire_at(value.invalidated_at)171 if condition is not None and condition.invalidated_at is not None:172 calculation.expire_at(condition.invalidated_at, condition=True)173 if row is not None:174 updates=({'actual_value_updates':(value.actual_update_at(row[0]),)}175 if getattr(calculation,'_samples',False) else {})176 if condition is None:177 calculation.consume((row,),**updates)178 else:179 if selected is None or selected[0] != row[0]:180 raise AnalysisError('值和条件未按同一驱动组求值')181 calculation.consume((row,), conditions=(selected[1],),182 condition_valid_until_ns=(selected[2],),**updates)183 for owner in self._rules:184 owner.after_group(group.time_ns)185 ends=[self._source_ends[source] for source in owner.sources if source in self._source_ends]186 if len(ends)==len(owner.sources) and ends and max(ends)==group.time_ns:187 owner.finish(max(ends))188 finally:189 group = calculation = expressions = value = condition = row = selected = None190191 def _emit_selection(self,request,parent,row):192 """按请求保存一份选择值及状态。参数: 请求/父ID/旧核借入行。返回: None。异常: 暂存错误。"""193 dataset=self._plan.dataset194 self._spool.append(SummarySelectionRow(self._next_sequence,dataset.dataset_id,dataset.entity,195 dataset.session_id,request.calculation_id,request.scope.split(':',1)[0],parent,row.time_ns,196 row.value.value if row.value.valid else None,row.value.dtype.value,row.unit,row.value.valid,197 row.selection.value,row.quality))198 self._next_sequence+=1199200 def _emit_input_trace(self,source,sequence,time_ns,record_type,locator,values,quality):201 """仅按trace请求写最终追溯表。参数: 本次必要输入定位。返回: None。异常: 暂存错误。"""202 dataset=self._plan.dataset203 self._spool.append(SummaryInputRow(self._next_sequence,dataset.dataset_id,dataset.entity,204 dataset.session_id,source,sequence,time_ns,record_type,locator,values,quality))205 self._next_sequence+=1206207 def _emit_contribution(self, request, contribution, quality):208 """仅按请求写计算定位和值/跳过。参数: 请求、借入贡献及本次质量。返回: None。异常: 暂存原错。"""209 update=contribution.update210 dataset=self._plan.dataset211 self._spool.append(SummaryContributionRow(self._next_sequence,dataset.dataset_id,dataset.entity,212 dataset.session_id,request.calculation_id,update.source,update.sequence,update.evaluation_index,213 contribution.value,contribution.skip_reason.value if contribution.skip_reason is not None else 'contributed',214 quality))215 self._next_sequence+=1216217 def execute_into(self, spool):218 """准备输入与算子，推进一次输入，并把已完成结果写入原结果仓。219220 参数: spool为本次ResultSpool。返回: None，不在这里发布文件。221 异常: 原始准备、读取或计算错误；保留可重试的record_session_cleanup_owner。222 """223 if self._closed or self._closing or self._failed or self._started:224 raise ResultClosedError('记录流会话只能执行一次')225 self._started = True226 self._spool = spool227 session_values = choices = definitions = source_cursors = state = None228 try:229 session_values, choices = self._prepare_requests(spool)230 definitions, source_cursors, state = self._prepare_input_state(choices)231 self._prepare_operators(session_values, choices, definitions, source_cursors, state)232 self._merge_sources()233 if self._direct_owner is not None:234 self._direct_owner.finish(self._source_ends.get(self._direct_binding.source))235 self._complete_results(session_values, spool)236 except BaseException as error:237 self._failed = True238 error.record_session_cleanup_owner = self239 raise240 finally:241 # 只解除编排帧对已交付声明和资源的借用，不关闭调用方结果仓。242 session_values = choices = definitions = source_cursors = state = spool = None243 self._spool = self._append_result = None244245 def _prepare_requests(self, spool):246 """从已验证计划取得请求和间隔选择；只为实际输出建立汇总状态。247248 参数: 结果仓。返回: 会话数值请求与间隔选择。异常: 汇总状态构造错误。"""249 try:250 plan = self._plan251 self._append_result = spool.append252 if plan.aggregations:253 from .result_aggregations import ResultAggregations254 self._aggregations=self._new(ResultAggregations,'aggregation_cleanup_owner',plan,self._budget)255 self._append_result=self._append_aggregate_input256 choices = dict(plan.intervals.resolved)257 hosted_reductions=plan.extensions.reductions if plan.extensions is not None else {}258 # 仅有数值输入的会话指标建立表达式和归约资源；对象计数在规则、窗口封口后产生。259 session_values = tuple(m for m in plan.calculations if m.scope == 'session'260 and m.operation not in OBJECT_COUNTS)261 self._segment_counts = {m.count_target: 0 for m in plan.calculations if m.operation == 'segment_count'}262 for request in plan.calculations:263 if request.operation in ('sample_count','sample_minimum','sample_maximum','sample_mean','select', *SAMPLE_MOMENTS, *SAMPLE_QUANTILES, *hosted_reductions):264 choices[request.calculation_id]=IntervalChoice('actual')265 self._constant_calculations=frozenset(index for index,calculation in266 enumerate(session_values)267 if choices[calculation.calculation_id].driver is None and not any(268 expression.dependencies for expression in (calculation.signal,calculation.condition) if expression is not None))269 return session_values, choices270 finally:271 # 仅放开本栈帧的临时引用；返回值已交付，子资源仍由会话所有者关闭。272 spool = plan = choices = hosted_reductions = session_values = request = None273274 def _prepare_input_state(self, choices):275 """按共用需求打开来源并建立最新信号状态，不提前读完文件。276277 参数: 已解析间隔选择。返回: 实际定义、来源游标映射和状态。278 异常: 来源、DBC周期或资源构造错误；子资源立即登记到本会话。"""279 try:280 plan = self._plan281 needed=self._requirements282 active_transforms=tuple(t for t in plan.transforms if t.name in needed.transform_names)283 self._transform_aliases=frozenset(t.name for t in active_transforms)284 if plan.diagnostics.trace or plan.diagnostics.integrity:285 self._diagnostics=self._new(RecordDiagnostics,'record_diagnostics_cleanup_owner',286 plan,needed,self._budget,self._emit_input_trace)287 observe_bounds=needed.constant_observation288 values=set(needed.ordinary_sample_aliases)|set(needed.message_aliases)289 clocks={name for source in needed.sources for name in source.clock_aliases}290 group_names=set(needed.ordinary_sample_aliases)|clocks291 state_names=set(needed.analysis_sample_aliases)|clocks|self._transform_aliases292 export_names=set() if plan.outputs.signal_csv is None else set(plan.outputs.signal_csv.spec.signals)293 export_sources={item.definition.source for item in plan.signals if item.definition.name in export_names}294 for request in plan.message_calculations:295 values.update(request.expr.dependencies)296 if request.disjunction is not None:297 values.add(request.disjunction.signal)298 owner = self._new(CurrentMessageCalculation, 'current_message_calculation_cleanup_owner',299 request, plan.sources, plan.bindings, plan.dataset.session_id, self._budget,300 backend=plan.execution.backend, batch_rows=self._message_batch_rows, domain=plan.dataset.domain,301 contribution_consumer=(partial(self._emit_contribution,request)302 if plan.outputs.contributions=='full' else None))303 self._messages.append((request, owner))304 from ..core.enum_codes import code_dependencies305 expressions = [expr for rule in plan.rules for expr in rule.expressions]306 expressions.extend(expr for calc in plan.calculations for expr in (calc.signal, calc.condition)307 if expr is not None)308 expressions.extend(t.expression for t in active_transforms)309 requested_codes = {name for expr in expressions for name in code_dependencies(expr.root)}310 definitions = tuple(replace(item.definition, capture_code=True) if item.definition.name in requested_codes311 else item.definition for item in plan.signals312 if item.definition.name in values | clocks)313 from .same_frame_integrals import direct_integral_inputs, same_frame_binding314 direct = self._direct_requested and direct_integral_inputs(plan) is not None315 state = None316 source_cursors = {}317 for source in plan.sources:318 source_definitions = tuple(item for item in definitions if item.source == source.name)319 selectors = tuple(item.update_message for item in plan.message_calculations if item.source == source.name)320 keys = tuple(dict.fromkeys((item.channel, item.frame_id, item.is_extended, item.is_fd) for item in selectors))321 retain_frames=plan.outputs.can_frames is not None and plan.outputs.can_frames.source==source.name322 source_requirement=needed.for_source(source.name)323 if source_requirement.active:324 cursor = self._new(RecordSourceCursor, 'record_source_cleanup_owner',325 plan, source, source_definitions, values, clocks, self._budget,326 self._path / ('input-' + source.name), group_names=() if direct else group_names,327 message_consumer=self._consume_direct_update if direct else self._consume_message if keys or self._diagnostics is not None else None, message_keys=keys,observe_bounds=observe_bounds or source.name in export_sources,retain_frames=retain_frames,requirements=source_requirement)328 self._cursors.append(cursor)329 source_cursors[source.name] = cursor330 # 复用本次来源初始化得到的DBC周期，首帧前补齐默认有效期。331 # 不提前构造带假有效期的计算状态；无关信号不进入definitions。332 from .signal_validity import resolve_signal_validity333 definitions = resolve_signal_validity(definitions, source_cursors)334 if direct:335 cursor = source_cursors[definitions[0].source]336 self._direct_binding = same_frame_binding(definitions, cursor.metadata['fields'])337 if self._direct_binding is None:338 # 首帧前决定普通路径；原DBC和游标保留，不再次加载或读取。339 cursor._message_consumer = None340 cursor._attach_raw_groups(definitions, plan.dataset.session_id, plan.execution.backend)341 definitions += tuple(t.output for t in active_transforms)342 # 由最后一层向前接回调。每层只打开自己所需状态，原Reader仍只消费一次。343 if active_transforms:344 from .transform_stream import TransformStage345 emit=self._consume_group346 for t in reversed(active_transforms):347 choice=choices[t.name]348 period=choice.step_ns349 if choice.mode=='dbc':350 definition=next(d for d in definitions if d.name==choice.driver)351 locator=field_to_dict(can_field_from_key(definition.field))352 period=next(f['cycle_ns'] for f in source_cursors[definition.source].metadata['fields']353 if f['locator']==locator)354 stage=self._new(TransformStage,'transform_cleanup_owner',plan,t,definitions,choice,period,355 self._source_ends,source_cursors,self._budget,emit)356 self._transform_stages.insert(0,stage)357 emit=stage.push358 if self._full_domain:359 # 零长度只供尚未收到输入的数值owner构造，不作为观测或读取过滤条件。360 # 游标已用原始None请求创建；每轮前瞻后改成真实已读范围。361 plan = replace(plan, dataset=replace(plan.dataset, domain=Interval(0, 0)))362 self._plan = plan363 if self._direct_binding is None and (state_names or observe_bounds):364 state = self._state = self._new(LatestSignalState, 'state_cleanup_owner',365 self._budget, tuple(item for item in definitions if item.name in state_names), plan.dataset.session_id,366 declaration_owner=self._declaration_owner)367 if plan.outputs.can_frames is not None or plan.outputs.signal_csv is not None:368 self._exports=self._new(RecordExports,'record_exports_cleanup_owner',369 plan,self._budget,self._path/'exports',self._cursors)370 return definitions, source_cursors, state371 finally:372 # 仅放开本栈帧的临时引用；返回值已交付，子资源仍由会话所有者关闭。373 choices = plan = needed = active_transforms = values = clocks = group_names = state_names = export_names = export_sources = None374 request = owner = expressions = requested_codes = definitions = state = source_cursors = source = source_definitions = selectors = keys = source_requirement = cursor = None375 emit = t = choice = definition = locator = stage = None376377 def _prepare_operators(self, session_values, choices, definitions, source_cursors, state):378 """依次建立实际请求的计算、事件、窗口和序列算子。379380 参数: 会话请求、间隔选择、实际信号定义及本次来源与状态。381 返回: None。异常: 单位、周期或资源构造错误，已构造对象仍由本会话持有。"""382 try:383 plan = self._plan384 hosted_reductions = plan.extensions.reductions if plan.extensions is not None else {}385 derived = tuple(item.definition for item in plan.derived)386 for request in session_values:387 choice = choices[request.calculation_id]388 step = choice.step_ns389 if choice.mode == 'dbc':390 locator = field_to_dict(can_field_from_key(state.definitions[choice.driver].field))391 metadata = source_cursors[state.definitions[choice.driver].source].metadata392 step = next(item['cycle_ns'] for item in metadata['fields'] if item['locator'] == locator)393 expression = request.signal if request.signal is not None else request.condition394 condition = request.condition if request.signal is not None else None395 # 先登记可达槽，再构造下一个子资源，避免后续构造失败遗失前者。396 if self._direct_binding is None:397 self._expressions.append([None, None])398 for index, static in enumerate((expression, condition)):399 if static is not None:400 self._expressions[-1][index] = self._new(DriverExpression,401 'driver_expression_cleanup_owner', static, state, choice.driver,402 backend=plan.execution.backend, derived=derived, observation_ends=self._source_ends)403 descriptor=hosted_reductions.get(request.operation)404 spec = CalculationDefinition(request.calculation_id, 'sample_minimum' if descriptor else request.operation, plan.dataset.domain,405 request.partial_policy, request.unit, minimum_coverage=request.minimum_coverage,406 config_version=plan.config_version, ddof=request.ddof, q=request.q)407 factory = RecordIntegral if request.operation == 'integral' else RecordTimeCalculation408 recovery = 'record_integral_cleanup_owner' if request.operation == 'integral' else 'record_time_cleanup_owner'409 extra = {'work_dir': self._path} if request.operation == 'time_weighted_mean' else {}410 if descriptor:411 extra={'hosted':descriptor}412 if request.operation=='select':413 factory,recovery=RecordSelection,'record_selection_cleanup_owner'414 extra={'consumer':partial(self._emit_selection,request,None)}415 self._calculations.append(self._new(factory, recovery, self._budget, spec,416 expression.root.unit, expression.root.dtype, step, backend=plan.execution.backend,417 batch_intervals=min(plan.execution.batch_rows, 4096), conditioned=condition is not None, **extra))418 if self._direct_binding is not None:419 from .same_frame_integrals import SameFrameIntegrals420 self._direct_owner = self._new(SameFrameIntegrals, 'same_frame_cleanup_owner',421 self._budget, definitions, tuple(item.signal for item in session_values),422 tuple(self._calculations), self._direct_binding, plan.dataset.session_id,423 self._before_direct_group, derived=derived)424 metadata = None425 event_requests=tuple(item for item in plan.calculations if is_event_duration(item))426 index_ids={item.calculation_id for item in event_requests}|{item.window_id for item in plan.windows}427 drivers=tuple(dict.fromkeys(choices[name].driver for name in index_ids if choices[name].mode!='actual'))428 if drivers:429 self._time_indexes=self._new(RecordTimeIndexes,'record_time_indexes_cleanup_owner',430 self._budget,self._path,drivers)431 if plan.windows:432 steps={}433 for request in plan.windows:434 choice=choices[request.window_id]435 step=choice.step_ns436 if choice.mode=='dbc':437 definition=state.definitions[choice.driver]438 locator=field_to_dict(can_field_from_key(definition.field))439 step=next(item['cycle_ns'] for item in source_cursors[definition.source].metadata['fields']440 if item['locator']==locator)441 steps[request.window_id]=step442 self._windows=self._new(RecordWindows,'record_windows_cleanup_owner',443 plan,self._budget,self._time_indexes,steps,self._emit_segment)444 if event_requests:445 steps={}446 for request in event_requests:447 choice=choices[request.calculation_id]448 step=choice.step_ns449 if choice.mode=='dbc':450 definition=state.definitions[choice.driver]451 locator=field_to_dict(can_field_from_key(definition.field))452 step=next(item['cycle_ns'] for item in source_cursors[definition.source].metadata['fields']453 if item['locator']==locator)454 steps[request.calculation_id]=step455 self._event_durations=self._new(RecordEventDurations,'record_event_duration_cleanup_owner',456 plan,self._budget,self._path,steps,self._emit_event_duration,indexes=self._time_indexes)457 scope_requests=tuple(item for item in plan.calculations if item.scope!='session' and not is_event_duration(item))458 if scope_requests:459 steps={}460 for request in scope_requests:461 choice=choices[request.calculation_id]462 step=choice.step_ns463 if choice.mode=='dbc':464 definition=state.definitions[choice.driver]465 locator=field_to_dict(can_field_from_key(definition.field))466 step=next(item['cycle_ns'] for item in source_cursors[definition.source].metadata['fields']467 if item['locator']==locator)468 steps[request.calculation_id]=step469 self._scope_calculations=self._new(RecordScopeCalculations,'record_scope_calculations_cleanup_owner',470 plan,state,self._budget,self._path,steps,scope_requests,self._source_ends,self._emit_event_duration,471 selection_consumer=self._emit_selection)472 if plan.sequences:473 steps={}474 for request in plan.sequences:475 choice=choices[request.sequence_id]476 step=choice.step_ns477 if choice.mode=='dbc':478 definition=state.definitions[choice.driver]479 locator=field_to_dict(can_field_from_key(definition.field))480 step=next(item['cycle_ns'] for item in source_cursors[definition.source].metadata['fields']481 if item['locator']==locator)482 steps[request.sequence_id]=step483 self._sequences=self._new(RecordSequences,'record_sequences_cleanup_owner',484 plan,state,self._budget,steps,self._emit_sequence)485 for request in plan.rules:486 if any(p.preset_type=='threshold_excursion' and request.rule_id==p.id+'_activity' for p in plan.presets):487 request=replace(request,stability=replace(request.stability,track_duration=True))488 choice=choices.get(request.rule_id,IntervalChoice('actual'))489 step=choice.step_ns490 if choice.mode=='dbc':491 definition=state.definitions[choice.driver]492 locator=field_to_dict(can_field_from_key(definition.field))493 step=next(item['cycle_ns'] for item in source_cursors[definition.source].metadata['fields']494 if item['locator']==locator)495 self._rules.append(self._new(RecordRuleTimeline,'record_rule_cleanup_owner',496 plan,request,state,self._budget,step,choice.driver,self._source_ends,self._emit_event,497 step_consumer=None if self._sequences is None else self._sequences.observe))498 finally:499 # 仅放开本栈帧的临时引用；返回值已交付，子资源仍由会话所有者关闭。500 session_values = choices = definitions = source_cursors = state = plan = hosted_reductions = derived = request = choice = metadata = expression = condition = static = None501 descriptor = spec = factory = extra = event_requests = index_ids = drivers = steps = definition = locator = scope_requests = None502503 def _complete_results(self, session_values, spool):504 """输入真正EOF后完成尾部、统计和结果汇总，不发布未完成前缀。505506 参数: 会话请求和原结果仓。返回: None。异常: 尾部计算、封存或写入错误。"""507 try:508 plan = self._plan509 # 真正EOF并关闭来源后，提交不足一批的尾部；失败不发布前缀。510 self._flush_messages()511 if self._exports is not None:512 self._exports.session()513 self._consume_constants()514 self._finish_actual_tails()515 if self._scope_calculations is not None:516 self._scope_calculations.finish(self._cursors)517 if self._event_durations is not None:518 self._event_durations.finish()519 if self._windows is not None:520 self._windows.finish(self._cursors,spool)521 if self._exports is not None:522 self._exports.seal()523 for request, calculation in zip(session_values, self._calculations):524 result = calculation.finish()525 if request.operation=='select':continue526 if self._full_domain and self._observed_domain is None:527 result = replace(result, value=None, status='indeterminate', reason='no_observation')528 self._emit_event_duration(request,None,result)529 # 所有片段已完成合并，此时只读取事件计数及按需记录的最终片段身份数。530 for request, result in object_count_results(self._plan, self._rules, self._segment_counts):531 self._emit_event_duration(request, None, result)532 for sequence, (request, owner) in enumerate(self._messages, len(plan.calculations)):533 result = owner.snapshot()534 incomplete = any(count for reason, count in result.skip_counts.items()535 if reason is not LegacySkipReason.CONDITION_FALSE)536 status = 'indeterminate' if result.value is None else 'partial' if incomplete else 'complete'537 self._append_result(SummaryCalculationRow(self._next_sequence, plan.dataset.dataset_id, plan.dataset.entity,538 plan.dataset.session_id, request.calculation_id, 'legacy_sum', 'session', None,539 None if result.value is None else round(result.value, 6), request.unit, status,540 result.reason or ('partial_input' if incomplete else 'ok'), owner.quality))541 self._count_outcome(status,result.reason)542 self._next_sequence+=1543 if self._aggregations is not None:544 for row in self._aggregations.rows(self._rules,self._next_sequence):545 spool.append(row)546 self._next_sequence+=1547 finally:548 # 仅放开本栈帧的临时引用；返回值已交付，子资源仍由会话所有者关闭。549 session_values = spool = plan = request = calculation = result = owner = row = None550551 def _append_aggregate_input(self,row):552 """从同一源行同时得到公开记录和聚合输入，不复制或重扫源表。553554 参数: row 为当前行。返回: None。异常: 原写出/归约错误终止整个会话。555 """556 self._spool.append(row)557 self._aggregations.accept(row)558559 def _finish_actual_tails(self):560 """以实际来源EOF结算连续计算和样本统计，固定周期不凭空加报文。561562 参数: 无，读取已耗尽游标、最新状态及尚待结算的末点。563 返回: None。异常: 表达式、预算或累计错误原样传播。564 不新增样本、贡献或追溯；观察结束和TTL都来自真实输入，domain仅负责裁剪。565 """566 for index, (calculation, expressions) in enumerate(zip(self._calculations, self._expressions)):567 samples = getattr(calculation, '_samples', False)568 if (index in self._constant_calculations569 or not samples and calculation._step_ns is not None or not calculation._rows):570 continue571 value, condition = expressions572 sources = {definition.source for expression in expressions if expression is not None573 for _, definition in expression._definitions}574 ends = [self._source_ends[source] for source in sources if source in self._source_ends]575 if not ends or max(ends) <= calculation._last_time:576 continue577 start, end = calculation._last_time, max(ends)578 # 最后组当时可能尚未知另一来源EOF；现在按右侧保持重算，不能沿用已结束来源的点值。579 current, deadline = value.evaluate_at(start)580 selected, condition_deadline = (None, None) if condition is None else condition.evaluate_at(start)581 if samples:582 # 保留真实末点及其更新标志；EOF查询只补齐有效截止，不把保持值写成新样本。583 # 某依赖来源已在末点结束时，其右侧保持无效，仍沿用既有半开EOF政策。584 previous = calculation._rows[-1]585 value_ends = tuple(t for t in (previous[2], deadline,586 start if not current.valid else None) if t is not None)587 condition_ends = tuple(t for t in (previous[4], condition_deadline,588 start if selected is not None and not selected.valid else None) if t is not None)589 calculation._rows[-1] = (start, previous[1], min(value_ends) if value_ends else None,590 previous[3], min(condition_ends) if condition_ends else None) + previous[5:]591 else:592 calculation._rows[-1] = (start, current, deadline, selected, condition_deadline)593 calculation.consume(((end, current, deadline),),594 **({'actual_value_updates': (False,)} if samples else {}),595 **({} if condition is None else596 {'conditions': (selected,), 'condition_valid_until_ns': (condition_deadline,)}))597598 def _consume_constants(self):599 """常量只需真实首末范围，EOF后复用相同统计核，不创建假信号或逐行值。600601 参数: 无。返回: None。异常: 求值/预算/计算原错；无观测不喂任何点。602 """603 if not self._constant_calculations:604 return605 starts=[c.observed_start for c in self._cursors if c.observed_start is not None]606 ends=[c.observed_end for c in self._cursors if c.observed_start is not None and c.observed_end is not None]607 if not starts or not ends:608 return609 first,last=min(starts),max(ends)610 for index in self._constant_calculations:611 calculation=self._calculations[index]612 value,condition=self._expressions[index]613 result,_=value.evaluate_at(last)614 selected=None if condition is None else condition.evaluate_at(last)[0]615 for time in dict.fromkeys((first,last)):616 # 常量的两个观测边界只说明保持范围，没有真实值更新可供计数。617 updates = {'actual_value_updates': (False,)} if calculation._spec.operation in ('sample_count', *SAMPLE_MOMENTS, *SAMPLE_QUANTILES) else {}618 calculation.consume(((time,result,None),),**updates,**({} if condition is None else619 {'conditions':(selected,),'condition_valid_until_ns':(None,)}))620621 def _emit_event(self,event):622 """将必要事件字段直接写一次最终结果仓。参数: 原核心事件。返回: None。异常: 原写入错。"""623 try:624 plan=self._plan625 self._append_result(SummaryEventRow(self._next_sequence,plan.dataset.dataset_id,plan.dataset.entity,626 plan.dataset.session_id,event.rule.rule_id,event.event_id,event.start_ns,event.confirmed_ns,627 event.end_ns,event.status,event.trigger_kind,event.quality | (event.dependency_quality & Q.ESTIMATED)))628 self._next_sequence+=1629 if self._event_durations is not None:630 self._event_durations.observe_event(event)631 if self._scope_calculations is not None:632 self._scope_calculations.observe_event(event)633 finally:634 event=None635636 def _emit_event_duration(self,request,event_id,result):637 """把每段事件指标直接写最终指标表。参数: 请求/父ID/结果。返回: None。异常: 原发布错。"""638 plan=self._plan639 model=SummaryCalculationRow if result.coverage_status is None else SummaryCoverageCalculationRow640 extra={} if result.coverage_status is None else {'coverage_status':result.coverage_status}641 if plan.extensions is not None and request.operation in plan.extensions.reductions:642 model=SummaryHostedCalculationRow643 extra['extension_version']=plan.extensions.reductions[request.operation].version644 self._append_result(model(self._next_sequence,plan.dataset.dataset_id,plan.dataset.entity,645 plan.dataset.session_id,request.calculation_id,request.operation,request.scope.split(':',1)[0],event_id,646 result.value,result.unit,result.status,result.reason,result.quality,**extra))647 self._count_outcome(result.status,result.reason)648 self._next_sequence+=1649650 def _emit_segment(self,segment):651 """只写实际片段与父关联，后续指标借用同一片段。参数: 核心片段。返回: None。异常: 原发布/计算错。"""652 if self._exports is not None:653 self._exports.segment(segment)654 plan=self._plan655 actual=segment.actual656 self._append_result(SummarySegmentRow(self._next_sequence,plan.dataset.dataset_id,plan.dataset.entity,657 plan.dataset.session_id,segment.window_id,segment.segment_id,658 None if actual is None else actual.start_ns,None if actual is None else actual.end_ns,659 segment.status,segment.parent_event_ids,segment.quality))660 self._next_sequence+=1661 if segment.window_id in self._segment_counts:662 self._segment_counts[segment.window_id] += 1663 if self._scope_calculations is not None:664 self._scope_calculations.observe_segment(segment)665666 def _count_outcome(self,status,reason):667 """随已写出的指标计数。参数: 行状态/原因。返回: None。异常: 未支持的内部状态。"""668 self._outcome_counts[status] += 1669 if reason in ('no_observation','no_evaluations'):670 self._outcome_counts['no_observation'] += 1671672 def analysis_outcome(self):673 """取得同次规则和指标概况。参数: 无。返回: AnalysisOutcome。异常: 规则未完成。"""674 from ..results.analysis_outcome import AnalysisOutcome675 return AnalysisOutcome(tuple(owner.outcome() for owner in self._rules),**self._outcome_counts)676677 def threshold_summaries(self):678 """生成明确预设请求的有限汇总，不复制状态轨迹。参数: 无。返回: tuple。异常: 未完成统计错误。"""679 from ..results.threshold_metadata import SummaryThresholdDuration680 presets={preset.id:preset for preset in self._plan.presets if preset.preset_type=='threshold_excursion'}681 result=[]682 for owner in self._rules:683 name=owner._request.rule_id684 preset_id=name.removesuffix('_activity')685 if preset_id in presets:686 value,status=owner.duration_result687 result.append(SummaryThresholdDuration(preset_id,name,value,status))688 return tuple(result)689690 def _emit_sequence(self,result):691 """只写最终配对结论，不复制事件全文。参数: 核心结果。返回: None。异常: 原写入错。"""692 try:693 plan=self._plan694 self._append_result(SummarySequenceRow(self._next_sequence,plan.dataset.dataset_id,plan.dataset.entity,695 plan.dataset.session_id,result.rule.rule_id,result.result_id,result.request.event_id,696 None if result.response is None else result.response.event_id,result.decided_ns,697 result.delay_ns,result.status,result.observation_quality))698 self._next_sequence+=1699 finally:700 result=None701702 def _consume_message(self, update):703 """按原序收集有限目标批，供同来源各累计器共用。704705 参数: update是只借用到本回调结束的原消息；延长寿命前先付费。706 返回: None。异常: 预算、计算或取消原错；完整明细仍逐消息交付。707 """708 try:709 if self._diagnostics is not None:710 self._diagnostics.consume(update.source,update.sequence,update.time_ns,711 'decoded_message',f'can:{update.channel}:{update.frame_id}:{int(update.is_extended)}:{int(update.is_fd)}',712 update.values,update.quality)713 if not any(request.source == update.source for request, _ in self._messages):714 return715 if self._message_batch_rows == 1:716 for request, owner in self._messages:717 if request.source == update.source:718 owner.consume((update,))719 return720 # 来源切换时先提交上一批，绝不将不同来源输入混入同一求和序列。721 if self._message_buffer and self._message_buffer[0].source != update.source:722 self._flush_messages()723 cost = 16384 + 16 * _metadata_bytes(update)724 # 为下一次实际计算留出空间，不能只按缓存本身是否放得下填满预算。725 row_cost = max(owner._row_cost for request, owner in self._messages726 if request.source == update.source)727 compute_cost = 16384 + (len(self._message_buffer) + 1) * row_cost728 # 读取器还可能构造本批返回列，不能把它尚未申请的费用当成空闲缓存。729 # 消息延长持有最多占总预算的1/16，其余仍供读取、计算和结果共用。730 buffer_limit = self._budget.memory_budget_bytes // 16731 if (self._message_buffer and (self._message_cost + cost > buffer_limit732 or self._budget.ram_bytes + cost + compute_cost733 > self._budget.memory_budget_bytes)):734 self._flush_messages()735 _reserve_owned_ram(self, cost, '_message_cost')736 self._message_buffer.append(update)737 if (len(self._message_buffer) >= self._message_batch_rows738 or self._message_cost >= buffer_limit):739 self._flush_messages()740 finally:741 update = None742743 def _flush_messages(self):744 """共享当前目标批完成各汇总计算，不创建历史或逐次输出。745746 参数: 无，使用已付费的有限消息缓冲。返回: None。747 异常: 原计算/取消错误优先；清理失败保留费用，供close重试。748 """749 if not self._message_buffer:750 return751 updates = tuple(self._message_buffer)752 primary = None753 try:754 for request, owner in self._messages:755 if request.source == updates[0].source:756 owner.consume(updates)757 except BaseException as error:758 primary = error759 raise760 finally:761 # 即使计算失败也不重放可能已经累计过的批；只允许资源清理。762 updates = None763 self._message_buffer.clear()764 try:765 _release_owned_ram(self, self._message_cost, '_message_cost')766 except BaseException as cleanup:767 if primary is None:768 raise769 primary.add_note('目标消息批费用清理失败：' + type(cleanup).__name__)770771 def _consume_direct_update(self, update: 'LegacyDecodedUpdate') -> None:772 """向已完成首帧前检查的同帧owner交付更新。参数: 更新。返回: None。异常: 原消费错误。"""773 self._direct_owner.consume(update)774775 def _before_direct_group(self) -> None:776 """结算前更新省略domain的实际范围；显式裁剪不额外查询。参数: 无。返回: None。异常: 原读取状态错误。"""777 if not self._full_domain:778 return779 for cursor in self._cursors:780 cursor._refresh_observed_bounds()781 self._advance_observed_domain()782783 def _advance_observed_domain(self):784 """用已付费游标的真实首末更新在线范围，不读取更多输入或保存时间序列。785786 参数: 无。返回: None。显式范围保持原裁剪规则；省略范围只保一个当前并集。787 异常: 非法时间或已封闭检测器的推进错误沿原清理路径传播。788 """789 if not self._full_domain:790 return791 starts = [c.observed_start for c in self._cursors if c.observed_start is not None]792 ends = [c.observed_end for c in self._cursors if c.observed_end is not None]793 if not starts or not ends:794 return795 domain = Interval(min(starts), max(ends))796 if domain == self._observed_domain:797 return798 self._observed_domain = domain799 self._plan = replace(self._plan, dataset=replace(self._plan.dataset, domain=domain))800 for calculation in self._calculations:801 calculation._spec = replace(calculation._spec, window=domain)802 reduction = getattr(calculation, '_reduction', None)803 if reduction is not None:804 reduction._spec = calculation._spec805 for owner in self._rules:806 owner._plan = self._plan807 owner.extend_observation(domain.end_ns)808 for owner in (self._sequences, self._windows, self._scope_calculations,809 self._event_durations, self._exports):810 if owner is not None:811 owner._plan = self._plan812 if self._exports is not None and self._exports._signal is not None:813 self._exports._signal.update_observation(self._plan)814815 def _merge_sources(self):816 """每来源只取已付费前瞻，收齐同刻更新再求值。817818 参数: 无。返回: None，所有实际来源EOF才返回。异常: 原读取/计算/预算错误。819 不按文件顺序拼接不同来源，不为跨源合并建磁盘历史。820 """821 heads = group = samples = None822 primary = None823 try:824 while True:825 heads = tuple(cursor.peek() for cursor in self._cursors)826 self._advance_observed_domain()827 for cursor in self._cursors:828 if cursor._finished and cursor.observed_end is not None:829 self._source_ends[cursor.source_name] = cursor.observed_end830 if self._sequences is not None:831 self._sequences.bounds(self._cursors,self._source_ends)832 if not self._rules_begun:833 for owner in self._rules:834 starts=[cursor.observed_start for cursor in self._cursors835 if cursor.source_name in owner.sources and cursor.observed_start is not None]836 if starts:837 owner.begin(min(starts))838 self._rules_begun=True839 live = tuple(head for head in heads if head is not None)840 if not live:841 for stage in self._transform_stages:842 stage.finish()843 for owner in self._rules:844 ends=[self._source_ends[source] for source in owner.sources if source in self._source_ends]845 if ends:846 owner.finish(max(ends))847 if self._sequences is not None:848 self._sequences.flush_before()849 return850 time = min(head.time_ns for head in live)851 count = sum(len(head.samples) for head in live if head.time_ns == time)852 _reserve_owned_ram(self, 16384 + count*1024, '_group_cost')853 samples = tuple(sample for head in live if head.time_ns == time for sample in head.samples)854 group = DecodedTimeGroup(time, samples)855 if self._transform_stages:856 self._transform_stages[0].push(group)857 else:858 self._consume_group(group)859 group = samples = live = None860 for cursor, head in zip(self._cursors, heads):861 if head is not None and head.time_ns == time:862 cursor.advance()863 heads = head = None864 _release_owned_ram(self, self._group_cost, '_group_cost')865 except BaseException as error:866 primary = error867 raise868 finally:869 heads = group = samples = live = head = None870 try:871 _release_owned_ram(self, self._group_cost, '_group_cost')872 except BaseException as cleanup:873 if primary is None:874 raise875 primary.add_note('跨来源同刻组清理失败：'+type(cleanup).__name__)876877 def _close_slot(self, slot):878 """关闭成功才解除责任。参数: 槽名。返回: None。异常: 原关闭错，槽保留以便重试。"""879 owner = getattr(self, slot)880 if owner is not None:881 owner.close()882 self._require_closed(owner)883 setattr(self, slot, None)884885 @staticmethod886 def _require_closed(owner):887 """核实子资源真实关闭再撤槽。参数: owner。返回: None。异常: AnalysisError保留原责任。"""888 if not (getattr(owner, 'closed', False) or getattr(owner, '_closed', False)) or getattr(owner, '_held_cost', 0):889 raise AnalysisError('记录流子资源仍未关闭或持有费用')890891 def close(self):892 """反向释放自有资源，保留失败槽供父会话重试。893894 参数: 无。返回: None。异常: 原关闭/预算错误；不会抵销其他owner的费用。895 """896 if self._closed:897 return898 self._closing = True899 self._message_buffer.clear()900 self._segment_counts.clear()901 _release_owned_ram(self, self._message_cost, '_message_cost')902 self._close_slot('_pending')903 for stage in self._transform_stages:904 stage.close()905 self._transform_stages.clear()906 self._close_slot('_aggregations')907 self._close_slot('_diagnostics')908 self._close_slot('_direct_owner')909 self._close_slot('_exports')910 for index, owner in enumerate(self._cursors or ()):911 if owner is not None:912 owner.close()913 self._require_closed(owner)914 self._cursors[index] = None915 self._close_slot('_read')916 self._close_slot('_adapter')917 self._close_slot('_groups')918 self._close_slot('_sequences')919 self._close_slot('_event_durations')920 self._close_slot('_scope_calculations')921 self._close_slot('_windows')922 self._close_slot('_time_indexes')923 for index,owner in enumerate(self._rules or ()):924 if owner is not None:925 owner.close()926 self._require_closed(owner)927 self._rules[index]=None928 for index, pair in enumerate(self._messages or ()):929 if pair is not None:930 pair[1].close()931 self._require_closed(pair[1])932 self._messages[index] = None933 for index, owner in enumerate(self._calculations or ()):934 if owner is not None:935 owner.close()936 self._require_closed(owner)937 self._calculations[index] = None938 for pair in self._expressions or ():939 for index, owner in enumerate(pair):940 if owner is not None:941 owner.close()942 self._require_closed(owner)943 pair[index] = None944 self._close_slot('_state')945 self._calculations = self._expressions = self._cursors = self._messages = self._plan = self._source_ends = None946 self._declaration_owner = None947 self._direct_binding = None948 self._rules=self._spool=self._requirements=None949 _release_owned_ram(self, self._group_cost, '_group_cost')950 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')951 self._closed = True952