chronaxis.execution.raw_signal_groups
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""把三格式共用的已解码更新交给实时完整同刻组消费者，不写历史文件。23只归一化当前声明需要的字段；同刻组跨读取批/连续文件保持，看到下一个相关4时间后才交付上一组，EOF显式finish交付末组。每条待交付样本预付有限预算，5大同刻组超预算明确失败。跨来源调用方必须先按实际时间合并；本组件不重排6原文件、不解码DBC、不查询完整性证明，不把单个来源EOF当整个会话EOF。7"""89from ..core.models import SourceRef10from ..legacy.updates import LegacyDecodedUpdate11from ..results.budget import ResourceBudget, _reserve_owned_ram, _release_owned_ram, _cleanup_owner12from ..results.errors import ResultClosedError13from ..sources.can_spec import can_field_from_key14from ..sources.packing import pack_message_values15from .groups import DecodedTimeGroup16from .history import _metadata_bytes17from .record_sample import _record_from_validated_fields18from .calculation_values import _scalar_from_validated_column19from ..core.numeric import ScalarValue20from ..core.types import QualityFlag as Q,SignalDType as D212223class RawSignalGroups:24 """保存当前同刻的必要更新；组交付后立即释放，消费方另行承担最新值状态。"""2526 def __init__(self, budget, definitions, session_id, consumer, *, backend='numpy',clock_only=()):27 """建立所需帧/信号索引和固定来源引用，不打开原始文件。2829 参数: budget共享账本；definitions仅本次所需信号tuple；session_id会话；30 consumer同步借用完整DecodedTimeGroup；backend选择原归一化内核。31 clock_only只使用对应报文的更新时间，共用无效占位而不逐帧归一化无关值。32 返回: owner。异常: 配置、预算或取消；构造恢复句柄raw_groups_cleanup_owner。33 """34 if not isinstance(budget, ResourceBudget) or type(definitions) is not tuple or not callable(consumer):35 raise TypeError('实时原始信号组需要预算、定义tuple及消费者')36 if backend not in ('reference', 'numpy'):37 raise ValueError('实时原始信号组后端错误')38 self._budget, self._backend, self._session = budget, backend, session_id39 self._held_cost = self._fixed_cost = self._group_cost = self._work_cost = 040 self._closed = self._closing = self._failed = self._finished = False41 self._consumer = self._fields = self._samples = None42 self._time = None43 self._clock_values=None44 self._clock_candidates={}45 self._automatic_names=set()46 try:47 _reserve_owned_ram(self, 65536 + 32 * _metadata_bytes((definitions, session_id)), '_fixed_cost')48 if len({item.name for item in definitions}) != len(definitions):49 raise ValueError('实时原始信号组不能重复声明别名')50 self._fields = {}51 if not set(clock_only)<={item.name for item in definitions}:52 raise ValueError('纯时钟字段必须属于当前组定义')53 self._clock_values={item.name:ScalarValue(None,item.dtype,False,Q.MISSING,54 tuple(dict.fromkeys(label for _,label in item.choices)) if item.dtype is D.ENUM else ())55 for item in definitions if item.name in clock_only}56 for definition in definitions:57 field = can_field_from_key(definition.field)58 # 自动类型需辨别“不是本帧字段”；显式类型的低层缺键仍表示缺值。59 if field.is_extended is None or field.is_fd is None:60 self._automatic_names.add(definition.name)61 key = (definition.source, field.channel, field.frame_id, field.is_extended, field.is_fd)62 ref = SourceRef(session_id, definition.source, definition.field)63 self._fields.setdefault(key, []).append((definition, field.signal, ref))64 self._fields = {key: tuple(items) for key, items in self._fields.items()}65 self._samples, self._consumer = [], consumer66 except BaseException as exc:67 _cleanup_owner(self, exc, 'raw_groups_cleanup_owner', '实时原始信号组')68 raise69 finally:70 definitions = definition = field = ref = consumer = None7172 def _require_open(self):73 """拒绝失败/关闭后的输入。参数: 无。返回: None。异常: ResultClosedError。"""74 if self._closed or self._closing or self._failed:75 raise ResultClosedError('实时原始信号组已失败或关闭')7677 def configure_metadata(self, facts):78 """借用同次DBC候选识别纯时钟字段，不解码或读取额外记录。7980 参数: facts为已加载字段事实。返回: None。异常: 定位结构错误。81 副作用: 仅保存当前纯时钟字段的有限候选，费用包含在定义固定预算中。82 """83 self._require_open()84 from ..legacy.records import field_from_dict85 # 同物理字段可能有多个时钟别名，按别名共享候选，不把后一别名覆盖掉。86 by_field={field_from_dict(dict(fact['locator'])).key: fact['dbc_frames']87 for fact in facts if 'dbc_frames' in fact}88 self._clock_candidates={definition.name: by_field[definition.field]89 for items in self._fields.values() for definition, _, _ in items90 if definition.name in self._clock_values and definition.field in by_field}9192 def consume(self, update):93 """接收一条已投影更新，仅所需帧参与时间检查与同刻归组。9495 参数: update为LegacyDecodedUpdate；缺失的匹配字段仍生成无效更新覆盖旧值。96 返回: None。异常: 身份/时间回退、预算、归一化或消费者错误；失败后禁止finish。97 """98 self._require_open()99 if self._finished:100 raise ResultClosedError('实时原始信号组已经结束')101 column = sample = value = fields = None102 primary = None103 try:104 if not isinstance(update, LegacyDecodedUpdate) or update.source_ref.session_id != self._session:105 raise ValueError('更新不属于当前会话')106 # 每种选择器只有一个键；最多查询四种明确/省略类型组合，不扫描全部信号。107 prefix = (update.source, update.channel, update.frame_id)108 fields = tuple(item for extended in (update.is_extended, None) for fd in (update.is_fd, None)109 for item in self._fields.get((*prefix, extended, fd), ()))110 # 投影会为属于本帧但解码失败的字段保留None；未出现的键才是另一帧的字段。111 # 纯时钟不要求解码值，使用初始化时已有的DBC候选判定所属帧。112 fields = tuple(item for item in fields if113 (item[0].name not in self._clock_values and114 (item[0].name not in self._automatic_names or item[1] in update.values))115 or (item[0].name in self._clock_values and116 (item[0].name not in self._clock_candidates or any(117 extended == update.is_extended and (fd == update.is_fd or fd and not update.is_fd)118 for extended, fd in self._clock_candidates[item[0].name]))))119 if not fields or update.is_remote or update.is_error:120 return121 if update.time_ns is None:122 raise ValueError('驱动表达式需要相关报文的实际时间')123 if self._time is not None and update.time_ns < self._time:124 raise ValueError('相关报文回退；多来源须先按实际时间有界归并')125 if self._time is not None and update.time_ns > self._time:126 self._flush()127 self._time = update.time_ns128 _reserve_owned_ram(self, 16384 + 4096 * len(fields), '_work_cost')129 for definition, name, ref in fields:130 # 本条缺字段同样归一化为无效；不保原值副本或逐条原帧定位。131 _reserve_owned_ram(self, 8192, '_group_cost')132 if definition.name in self._clock_values:133 value=self._clock_values[definition.name]134 array_value=False if definition.dtype is D.BOOL else 0. if definition.dtype is D.FLOAT64 else 0135 else:136 column = pack_message_values(definition, (update.values.get(name),), self._backend)137 value = _scalar_from_validated_column(column, 0)138 array_value=column.values[0].item() if definition.dtype is D.ENUM else None139 # pack_message_values刚完成数值检查；时间/序号来自已校验更新。140 sample = _record_from_validated_fields(definition.name, update.time_ns, update.sequence,141 value, definition.unit, ref, array_value if definition.dtype is D.ENUM else None)142 self._samples.append(sample)143 except BaseException as exc:144 primary = exc145 self._failed = True146 raise147 finally:148 column = sample = value = update = fields = None149 try:150 _release_owned_ram(self, self._work_cost, '_work_cost')151 except BaseException as cleanup:152 self._failed = True153 if primary is None:154 raise155 primary.add_note('实时原始信号组临时费用清理失败：' + type(cleanup).__name__)156157 def _flush(self):158 """同步借出完整组后释放本组，不保输出副本。参数: 无。返回: None。异常: 消费/清理错误。"""159 if not self._samples:160 return161 group = None162 try:163 group = DecodedTimeGroup(self._time, tuple(self._samples))164 self._consumer(group)165 finally:166 group = None167 self._samples.clear()168 _release_owned_ram(self, self._group_cost, '_group_cost')169170 def finish(self):171 """所有输入EOF并成功关闭后交付末组，不创建末尾周期。参数: 无。返回: None。异常: 消费/清理错误。"""172 self._require_open()173 if self._finished:174 return175 try:176 self._flush()177 self._finished = True178 except BaseException:179 self._failed = True180 raise181182 def close(self):183 """仅退自有费用，不替调用方完成剩余组。参数: 无。返回: None。异常: 释放失败可重试。"""184 if self._closed:185 return186 self._closing = True187 self._samples = self._fields = self._consumer = self._clock_values = None188 self._clock_candidates={}189 self._automatic_names.clear()190 _release_owned_ram(self, self._work_cost, '_work_cost')191 _release_owned_ram(self, self._group_cost, '_group_cost')192 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')193 self._closed = True194