chronaxis.execution.record_source_cursor
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""按需读取一个来源并提供完整同刻组，供多来源按时间合并。23原始报文沿共同解码更新，解析数据沿既有SignalBatch。每次只保一个读取批4及尚未消费的同刻组，所有缓存逐笔付费；不建立磁盘历史或重复扫描输入。5"""67from collections import deque8from ..results.errors import AnalysisError, ResultClosedError9from ..results.spool import _reserve_owned_ram, _release_owned_ram, _cleanup_owner10from ..sources.can_spec import can_field_from_key11from ..sources.models import ReadRequest12from .history import _metadata_bytes13from .raw_signal_groups import RawSignalGroups14from .batch_signal_groups import BatchSignalGroups15from .text_can_reader import BudgetedTextCANSourceAdapter16from .memory_reader import BudgetedMemorySourceAdapter17from .csv_reader import BudgetedSignalCSVSourceAdapter18from .frame_capture import FrameDBCContext192021def _queued_group_cost(sample_count):22 """返回现有待消费时间组费用，同时供 CSV 判断可交接的批容量。2324 参数: sample_count 为组内样本数。返回: 字节估算。异常: 输入由内部计数保证。25 """26 return 4096 + sample_count * 8192272829class RecordSourceCursor:30 """拥有来源、实际游标、同刻归组与已付费前瞻队列。"""3132 def __init__(self, plan, source, definitions, values, clocks, budget, work_dir, *,33 group_names=None, message_consumer=None, message_keys=(), observe_bounds=False, retain_frames=False, requirements=None):34 """建立所需来源，原始报文只初始化一次DBC，不读首帧。3536 参数: 原计划/来源、该来源定义、数值和驱动别名集合、预算、专用新路径。37 retain_frames仅原帧导出请求时移交同次原帧仓；38 observe_bounds为无信号常量计算收集实际范围；group_names只含时序计算依赖；message_consumer/keys为实际逐消息消费者及定位。39 返回: 游标owner。异常: 原构造错误，恢复入口record_source_cleanup_owner。40 """41 self._budget = budget42 self.source_name = source.name43 self.frame_format_notices = ()44 self._definitions = definitions45 self._held_cost = self._fixed_cost = self._queue_cost = 046 self._adapter = self._read = self._groups = self._pending = None47 self._queue = self._current = self.metadata = None48 self._current_cost = 049 self._closed = self._closing = self._failed = self._finished = self._saw_eof = False50 self._reading_started = False51 self._raw = source.source_type in ('can_blf', 'can_asc', 'asc_or_blf', 'raw_can_csv')52 self._message_consumer = message_consumer53 self._observe_bounds=observe_bounds54 self._retain_frames=retain_frames55 self._raw_store=self._message_store=self._dbc_context=None56 self._read_identity=None57 self.source_kind=source.source_type58 self._capture_messages=False59 self._memory_first=(source.config.rows[0].time_ns+source.config.offset_ns60 if observe_bounds and source.source_type=="memory" and source.config.rows else None)61 self.observed_start = self.observed_end = None62 try:63 _reserve_owned_ram(self, 131072 + 32*_metadata_bytes(definitions), '_fixed_cost')64 self._queue = deque()65 grouped = definitions if group_names is None else tuple(item for item in definitions if item.name in group_names)66 if grouped:67 self._groups = self._new(RawSignalGroups if self._raw else BatchSignalGroups,68 'raw_groups_cleanup_owner' if self._raw else 'batch_groups_cleanup_owner',69 budget, grouped, plan.dataset.session_id, self._enqueue,70 **({'backend': plan.execution.backend,71 'clock_only':tuple(item.name for item in grouped if item.name in clocks-values)} if self._raw else {}))72 options = dict(backend=plan.execution.backend)73 if self._raw:74 output=plan.outputs.can_frames75 if retain_frames and output is not None and output.resampling is not None:76 self._dbc_context=self._new(FrameDBCContext,'frame_dbc_cleanup_owner',77 source.config.dbc_path,output.rules,budget,cycle_periods=output.resampling.period_ns is None,integrity=plan.diagnostics.integrity)78 selected = {item.name: item for item in definitions}79 cycles = tuple(dict.fromkeys(can_field_from_key(selected[choice.driver].field)80 for _, choice in plan.intervals.resolved81 if choice.mode == 'dbc' and choice.driver in selected))82 cycles = tuple(dict.fromkeys((*cycles, *(can_field_from_key(item.field)83 for item in definitions if item.valid_for_ns is None))))84 output=plan.outputs.signal_csv85 if output is not None and output.period_driver in selected:86 cycles=tuple(dict.fromkeys((*cycles,can_field_from_key(selected[output.period_driver].field))))87 fields = tuple(dict.fromkeys(can_field_from_key(item.field) for item in definitions if item.name in clocks))88 keys = tuple(dict.fromkeys((*message_keys,89 *((f.channel, f.frame_id, f.is_extended, f.is_fd) for f in fields))))90 options.update(work_dir=work_dir, retain_raw_frames=retain_frames, cycle_fields=cycles, metadata_fields=fields,91 message_keys=keys, message_consumer=self._consume_message)92 files = source.config.input_paths if source.source_type=='can_blf' else source.config.inputs93 # 进程数是上限；单文件或上限1的接续在宿主内读取，解码政策单独传递。94 parallel = min(plan.execution.process_count,len(files)) > 1 or getattr(source.config,'file_merge','concatenate') == 'time_order'95 factory, recovery = BudgetedTextCANSourceAdapter, 'text_can_source_cleanup_owner'96 options['integrity']=plan.diagnostics.integrity97 options['trace']=False98 if self._dbc_context is not None:99 options['database_context']=self._dbc_context100 if source.source_type == 'can_blf':101 full = (source.config.input_check == 'full' or source.config.decode_scope == 'all') if requirements is None else requirements.input_check=='full'102 scope = requirements.time_scope if requirements is not None else ('source' if full or grouped or observe_bounds or retain_frames else 'selected' if plan.dataset.domain is not None else 'none')103 if source.config.file_merge=='time_order':104 scope='source' if scope=='source' else 'selected'105 # 低层调用可不传requirements，沿既有值、时钟和导出声明推导目标ID。106 decode_ids = (requirements.decode_frame_ids if requirements is not None else tuple(sorted(107 {can_field_from_key(item.field).frame_id for item in definitions if item.name in values}108 | {key[1] for key in keys}109 | {field.frame_id for field in (() if self._dbc_context is None else self._dbc_context.extra_fields)})))110 options['blf_time_scope']=scope111 options['legacy_blf_decode']=source.config.decode_policy=='legacy'112 if not parallel:113 options['serial_blf_policy']=source.config.decode_scope114 if scope!='source':115 options['time_frame_ids']=decode_ids116 if not retain_frames:117 if source.config.decode_scope=='all':118 if not parallel:119 # 串行all完成全部解码/值检查后，才按旧ID范围交付；并行保持原政策。120 options['post_decode_frame_ids']=decode_ids121 else:122 options['emit_frame_ids']=decode_ids123 else:124 if (requirements is not None and requirements.input_check == 'auto'125 and source.source_type in ('can_asc','asc_or_blf') and not retain_frames):126 options['time_frame_ids'] = requirements.decode_frame_ids127 if (requirements is not None and requirements.input_check == 'auto'128 and requirements.time_scope != 'source' and not retain_frames):129 options['selected_ids'] = requirements.decode_frame_ids130 if parallel and not retain_frames and requirements is not None:131 options['emit_frame_ids']=requirements.decode_frame_ids132 if parallel:133 options['process_count']=plan.execution.process_count134 adapter_definitions = tuple(item for item in definitions if item.name in values)135 names = ()136 else:137 factory, recovery = ((BudgetedMemorySourceAdapter, 'memory_source_cleanup_owner')138 if source.source_type == 'memory' else (BudgetedSignalCSVSourceAdapter, 'csv_source_cleanup_owner'))139 adapter_definitions = definitions140 names = tuple(item.name for item in definitions)141 options.update(clock_only=tuple(item.name for item in definitions if item.name in clocks-values),142 retain_original=False)143 if source.source_type == 'signal_csv':144 # 单批可出现每样本一个时间组；给接收队列留出空间，实际费用仍由队列自己预约。145 options['consumer_sample_bytes'] = _queued_group_cost(1)146 config=source.config147 # 公共原帧来源统一使用同次文本原帧仓；旧低层仓仍由其独立API管理。148 self.legacy_frame_storage = False149 if self._raw and source.source_type=='can_blf':150 from ..sources.text_can_spec import CANTextSpec151 from ..sources.blf import BLFSpec152 config=CANTextSpec(config.name,tuple(BLFSpec(p,config.offset_ns) for p in config.input_paths),153 config.dbc_path,config.channel,config.timebase,config.decode_policy,154 'full' if config.decode_scope=='all' else config.input_check,config.file_merge)155 self._adapter = self._new(factory, recovery, budget, config,156 plan.dataset.session_id, adapter_definitions, **options)157 self._read = self._adapter.iter_batches(ReadRequest(plan.dataset.session_id, names,158 None if plan.dataset.domain is None else plan.dataset.domain.start_ns,159 None if plan.dataset.domain is None else plan.dataset.domain.end_ns, plan.execution.batch_rows),160 **({"observe_bounds":True} if source.source_type=="signal_csv" and observe_bounds else {}))161 if self._raw:162 self.metadata = self._read.prepare_metadata()163 if self._groups is not None:164 self._groups.configure_metadata(self.metadata['fields'])165 self._read_identity=self._read.identity166 except BaseException as error:167 _cleanup_owner(self, error, 'record_source_cleanup_owner', '记录流来源构造')168 raise169 finally:170 definitions = adapter_definitions = selected = fields = options = values = clocks = None171172 def _consume_message(self, update):173 """同次解码先交逐消息消费者，再按需归组。参数: update。返回: None。异常: 原消费错误。"""174 try:175 if self._message_consumer is not None:176 self._message_consumer(update)177 if self._groups is not None:178 self._groups.consume(update)179 finally:180 update = None181182 def _attach_raw_groups(self, definitions: tuple, session_id: str, backend: str) -> None:183 """候选绑定不适用时，在同一游标首帧前建立普通归组。184185 参数: 本次实际所需定义、会话和明确后端。返回: None。异常: 已开始读取或构造错误。186 只用于无纯时钟/导出的直接积分候选；复用已加载DBC，不重开适配器。187 """188 if not self._raw or self._reading_started or self._groups is not None or self._closed or self._failed:189 raise ResultClosedError('只能在原始来源首次读取前附加普通归组')190 self._groups = self._new(RawSignalGroups, 'raw_groups_cleanup_owner',191 self._budget, definitions, session_id, self._enqueue, backend=backend)192 self._groups.configure_metadata(self.metadata['fields'])193194 def _refresh_observed_bounds(self) -> None:195 """同步消息回调中取得同次读取已见范围。参数: 无。返回: None。异常: 原读取状态错误。196197 不主动读下一条；EOF关闭后保留原游标已经记录的最终范围。198 """199 if self._raw and self._read is not None:200 self.observed_start, self.observed_end = self._read.observed_bounds201202 def _new(self, factory, recovery, *args, **kwargs):203 """接实际未返回的子owner。参数: 工厂/属性/参数。返回: owner。异常: 原错。"""204 try:205 return factory(*args, **kwargs)206 except BaseException as error:207 self._pending = getattr(error, recovery, None)208 raise209210 def _enqueue(self, group):211 """在上游释放组费用前接管必要组。参数: 完整组。返回: None。异常: 预算/取消。"""212 try:213 cost = _queued_group_cost(len(group.samples))214 _reserve_owned_ram(self, cost, '_queue_cost')215 self._queue.append((group, cost))216 finally:217 group = None218219 def peek(self):220 """读取到下一个完整组或真实EOF，返回借用组。221222 参数: 无。返回: DecodedTimeGroup或EOF的None。异常: 原输入/预算错误；失败后只清理。223 """224 if self._closed or self._closing or self._failed:225 raise ResultClosedError('记录流来源已关闭或失败')226 self._reading_started = True227 batch = None228 try:229 while self._current is None and not self._queue and not self._finished:230 try:231 batch = next(self._read)232 except StopIteration:233 if (self._raw and not self._read.exhausted) or not self._raw and not self._saw_eof:234 raise AnalysisError('来源未交付真实完整EOF')235 if self._raw:236 observed = self._read.observation237 self.observed_start, self.observed_end = observed.first_ns, observed.last_ns238 self._retain_frame_notices()239 if self._retain_frames:240 self._read.take_raw_store(self,'_raw_store')241 if self._capture_messages:242 self._read.take_message_store(self,'_message_store')243 self.metadata = None244 self._close_slot('_read')245 self._close_slot('_adapter')246 if self._groups is not None:247 self._groups.finish()248 self._finished = True249 break250 self._saw_eof = batch.source_eof251 if self._raw:252 self.observed_start,self.observed_end = self._read.observed_bounds253 if not self._raw:254 for column in batch.series.values():255 if len(column.time_ns):256 first,last = int(column.time_ns[0]),int(column.time_ns[-1])257 self.observed_start = first if self.observed_start is None else min(first,self.observed_start)258 self.observed_end = last if self.observed_end is None else max(last,self.observed_end)259 if batch.observed_end_ns is not None:260 self.observed_end = int(batch.observed_end_ns)261 if self._observe_bounds:262 self.observed_start=(self._memory_first if self._memory_first is not None else263 getattr(self._read,'observed_start',self.observed_start))264 if self._groups is not None:265 self._groups.consume(batch)266 elif batch.series:267 raise AnalysisError('原始记录流不应额外构造普通信号列')268 batch = None269 if self._current is None and self._queue:270 self._current, self._current_cost = self._queue.popleft()271 return self._current272 except BaseException as error:273 self._failed = True274 error.record_source_cleanup_owner = self275 raise276 finally:277 batch = column = observed = None278279 def advance(self):280 """消费当前组并释放其费用，不主动读取下一组。参数: 无。返回: None。异常: 预算/取消。"""281 self._current = None282 cost, self._current_cost = self._current_cost, 0283 _release_owned_ram(self, cost, '_queue_cost')284285 def _retain_frame_notices(self):286 """在同次真实EOF后保留已核DBC帧型提示，不借清单声明代替读取。287288 参数: 无。返回: None。异常: 已核元数据不一致或预算失败；最高游标负责清理。289 只复制实际差异及原绑定，不保留完整READY、样本或历史仓。290 """291 from ..results.can_metadata import CANFrameFormatNotice292 facts = self._read.runtime_metadata['fields']293 notices = tuple(fact for fact in facts if 'frame_format_compatibility' in fact)294 projection = getattr(self._read, '_binding_projection', None)295 automatic = {} if projection is None else projection.compatibility_fields296 if not notices and not automatic:297 return298 _reserve_owned_ram(self, 8192 + 32 * _metadata_bytes((notices, self._definitions)), '_fixed_cost')299 result = []300 for definition in self._definitions:301 field = can_field_from_key(definition.field)302 recorded = False303 for fact in notices:304 if field != type(field)(**fact['locator']):305 continue306 notice = fact['frame_format_compatibility']307 extended = field.is_extended if field.is_extended is not None else automatic.get(field)308 if extended is None:309 continue310 result.append(CANFrameFormatNotice(definition.name, field.channel,311 field.signal, field.frame_id, extended, notice['dbc_is_fd'],312 notice['requested_is_fd'], notice['code']))313 recorded = True314 if field in automatic and not recorded:315 result.append(CANFrameFormatNotice(definition.name, field.channel, field.signal,316 field.frame_id, automatic[field], True, False, 'dbc_fd_classic_selector'))317 self.frame_format_notices = tuple(result)318319 @staticmethod320 def _require_closed(owner):321 """关闭未完成时保留槽。参数: 子owner。返回: None。异常: AnalysisError。"""322 if not (getattr(owner, 'closed', False) or getattr(owner, '_closed', False)) or getattr(owner, '_held_cost', 0):323 raise AnalysisError('来源子资源仍未关闭或持有费用')324325 def _close_slot(self, slot):326 """成功关闭后才撤槽。参数: 槽名。返回: None。异常: 原关闭错。"""327 owner = getattr(self, slot)328 if owner is not None:329 owner.close()330 self._require_closed(owner)331 setattr(self, slot, None)332333 def close(self):334 """按责任释放当前批/队列和输入；提前关闭不交付末组。335336 参数: 无。返回: None。异常: 原关闭/预算错，可从同一owner重试。337 """338 if self._closed:339 return340 self._closing = True341 self._close_slot('_pending')342 self._close_slot('_read')343 self._close_slot('_adapter')344 self._close_slot('_groups')345 self._close_slot('_raw_store')346 self._close_slot('_message_store')347 self._close_slot('_dbc_context')348 self.metadata = self._current = self._queue = self._message_consumer = None349 self._definitions = self.frame_format_notices = ()350 _release_owned_ram(self, self._queue_cost, '_queue_cost')351 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')352 self._closed = True353