chronaxis.execution.text_can_reader
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""把ASC、BLF、原始CSV的同次解码接到公共信号批次和真实观测协议。23适配器声明字段和时基，不读文件；每个TextCANObservationRead拥有独立解析器、4有限原帧块和预算。每次交付同时段各别名批，总样本数不超过batch_rows；同刻别名未全部5交付前不越过其watermark。完整EOF并关闭解析器后才开放观测证明和可选原帧仓。6原帧仓仅显式导出时创建，共用已解码值，不另扫描原文件。7"""89from pathlib import Path10from sys import getsizeof11from uuid import uuid41213from ..core.validation import require_identifier14from ..sources.text_can_spec import CANTextSpec15from ..sources.text_can_decode import CANTextDecodeRead,pack_text_records16from ..sources.descriptions import describe_fields,source_timebase_identity17from ..sources.models import ReadRequest,SignalBatch,SourceDescription,SourceCapabilities,freeze_metadata18from ..sources.protocol import BatchValidator,SourceClosedError19from ..results.budget import _reserve_owned_ram,_release_owned_ram,_cleanup_owner20from ..results.errors import ResultClosedError21from .can_reader import CANObservation22from .frame_capture import TextFrameStore23from .history import _metadata_bytes,_pending_bytes242526def _record_metadata_bytes(record):27 """按已检查记录的固定结构计算与通用估算器相同的费用。2829 参数: record为当前解码器交付的冻结DecodedCANRecord，键和值均已检查。30 返回: values、重建值/布局、路径及原时间文本的基础字节费用，尚未乘32。31 异常: 传播内存分配错误；不接受任意外部对象树。32 每个引用位置仍计费，仅省去通用元数据的递归类型分流；不降低预算金额。33 """34 # 外层五元组逐项收32字节；每个键值对的两个引用同样各收32字节。35 total = getsizeof((None,) * 5) + 5 * 3236 for pairs in (record.values, record.frame_values, record.frame_layout):37 total += getsizeof(pairs)38 for pair in pairs:39 total += 32 + getsizeof(pair) + 2 * 3240 total += getsizeof(pair[0]) + 64 + getsizeof(pair[1]) + 6441 total += getsizeof(str(record.raw.path)) + 6442 total += getsizeof(record.raw.raw_time) + 6443 return total444546class BudgetedTextCANSourceAdapter:47 """拥有静态描述和活read集合，借用导出DBC上下文；关闭不会释放借用上下文。"""4849 def __init__(self,budget,spec,session_id,signals,*,backend='numpy',work_dir=None,retain_raw_frames=False,database_context=None,message_consumer=None,cycle_fields=(), metadata_fields=(),message_keys=(),integrity=True,trace=True,selected_ids=None,time_frame_ids=None,process_count=1,emit_frame_ids=None,blf_time_scope='source',legacy_blf_decode=False,serial_blf_policy=None,post_decode_frame_ids=None):50 """构造声明，不读取输入。参数: 预算/spec/会话/定义、后端、可选暂存目录及DBC上下文；message_keys为无值公式的完整触发定位；message_consumer同步接收必要消息，保留需自付预算；cycle_fields显式请求初始化DBC周期；metadata_fields仅确认驱动候选，两者不增加值解码。返回: adapter。异常: 配置/预算错。5152 time_frame_ids将所需通道/ID的单调性检查与全来源范围统计分开，不改变原帧时间。53 selected_ids是纯消息计算需要的有限ID集合；仅在不要求全来源时间或原帧导出时54 下推给原始CSV读取器。此时观测范围是保留记录的范围，不能冒称全部来源范围。55 legacy_blf_decode仅由纯can_blf转接设置；serial_blf_policy仅串行使用，保原异常及all/requested范围。56 post_decode_frame_ids为串行all完成解码后交付的ID集合，None全部交付；不下推并行。57 这些是内部转接参数，不增加用户配置项。58 """59 if type(spec) is not CANTextSpec or backend not in ('reference','numpy') or type(retain_raw_frames) is not bool:60 raise ValueError('文本CAN来源配置或后端非法')61 from ..sources.can_updates import require_message_keys62 require_message_keys(message_keys, channel=spec.channel)63 if message_keys and message_consumer is None:64 raise ValueError('消息触发定位必须有实际消费者')65 if message_consumer is not None and not callable(message_consumer):66 raise TypeError('message_consumer需要同步可调用对象')67 if retain_raw_frames and work_dir is None:68 raise ValueError('显式原帧保存需要新工作目录')69 from ..legacy.dbc_cycles import require_cycle_fields70 require_cycle_fields(cycle_fields,spec.channel)71 require_cycle_fields(metadata_fields,spec.channel)72 if retain_raw_frames and time_frame_ids is not None:73 raise ValueError('原帧导出需要完整时间顺序检查')74 self._process_count = process_count75 self._emit_frame_ids = emit_frame_ids76 self._blf_time_scope=blf_time_scope77 self._legacy_blf_decode=legacy_blf_decode78 self._serial_blf_policy=serial_blf_policy79 self._post_decode_frame_ids=post_decode_frame_ids80 self._metadata_fields = metadata_fields81 self._selected_ids = selected_ids82 self._time_frame_ids = time_frame_ids83 self._cycle_fields=cycle_fields84 self._integrity=integrity85 self._trace=trace86 require_identifier(session_id,'text_can.session_id')87 self.budget=self._budget=budget88 self.spec=spec89 self.session_id=session_id90 self.signals=signals91 self.backend=backend92 self.work_dir=None if work_dir is None else Path(work_dir)93 self._retain_raw_frames=retain_raw_frames94 self.database_context=database_context95 self._update_consumer=message_consumer96 self._update_projection=None97 self._held_cost=self._fixed_cost=098 self._closed=False99 self._requests=set()100 self._pending=None101 try:102 cost=131072+_metadata_bytes((signals,selected_ids,time_frame_ids))*32+sum(1024+len(str(i.path).encode('utf-8'))*32 for i in spec.inputs)103 if message_keys:104 cost += 8192 + 2048 * len(message_keys)105 if cycle_fields or metadata_fields:106 cost += 16384 + _metadata_bytes((cycle_fields,metadata_fields))*32107 _reserve_owned_ram(self,cost,'_fixed_cost')108 if message_consumer is not None:109 from ..sources.can_updates import CANUpdateProjection110 from ..sources.can_spec import can_field_from_key111 self._update_projection=CANUpdateProjection(spec.name,session_id,tuple(can_field_from_key(s.field) for s in signals),message_keys=message_keys,clock_fields=metadata_fields)112 locator={'paths':tuple(str(i.path) for i in spec.inputs),'dbc_path':str(spec.dbc_path),113 'channel':spec.channel,'profile':'can-text-v1','offset_ns':0,'time_unit':'ns',114 'source_timebase_sha256':source_timebase_identity(spec.timebase,0)}115 self._description=SourceDescription(spec.name,describe_fields(spec.name,signals,locator),116 spec.timebase.timebase_id,None,None,SourceCapabilities(True,True,True,False,117 ('ASC/CSV按文本转换时间并保原行；混合BLF沿用库浮点秒转纳秒并以记录序号定位；CSV按ID筛选时仅观测保留记录',),118 (('所需通道/ID按声明顺序时间非递减；其他帧仅参与来源范围统计；DBC缺单位仍未核实'119 if time_frame_ids is not None else '输入必须按声明顺序时间非递减；DBC缺单位仍未核实'),)),legacy_profile='can-text-v1')120 except BaseException as error:121 _cleanup_owner(self,error,'text_can_source_cleanup_owner','文本CAN来源构造')122 raise123124 @property125 def retain_raw_frames(self):126 """返回本来源冻结的原帧保存选择。参数: 无。返回: bool。异常: 无。127 """128 return self._retain_raw_frames129130 def _ensure_open(self):131 """拒绝使用已关闭适配器。参数: 无。返回: None。异常: SourceClosedError。132 """133 if self._closed:134 raise SourceClosedError('文本CAN来源已关闭')135136 def describe(self):137 """返回静态字段及时间声明。参数: 无。返回: SourceDescription。异常: 已关闭。138 """139 self._ensure_open()140 return self._description141142 def iter_batches_with_observation(self,request):143 """创建具有同次观测的独立read。参数: request。返回: read。异常: 合同/预算错及可恢复owner。144 """145 self._ensure_open()146 if self._pending is not None:147 raise ResultClosedError('文本CAN仍有未返回read待清理')148 try:149 self._pending=TextCANObservationRead(self,request)150 read=self._pending151 self._requests.add(read)152 self._pending=None153 return read154 except BaseException as error:155 self._pending=self._pending or getattr(error,'text_can_observation_cleanup_owner',None)156 _cleanup_owner(self._pending,error,'text_can_observation_cleanup_owner','文本CAN读取登记') if self._pending is not None else None157 if self._pending is not None and self._pending.closed:158 self._pending=None159 if self._pending is not None:160 error.text_can_source_cleanup_owner=self161 raise162163 def iter_batches(self,request):164 """沿同一批次合同创建read，不另读输入。参数: request。返回: read。异常: 原工厂错。165 """166 return self.iter_batches_with_observation(request)167168 def close(self):169 """关闭所有本来源活read，再释放描述费用。参数: 无。返回: None。异常: 关闭/费用错可重试。170 """171 if self._closed:172 return173 if self._pending is not None:174 self._pending.close()175 self._pending=None176 while self._requests:177 next(iter(self._requests)).close()178 self._description=None179 self._update_projection=self._update_consumer=None180 _release_owned_ram(self,self._fixed_cost,'_fixed_cost')181 self._closed=True182183184class TextCANObservationRead:185 """拥有一个有限原帧块和一个归一化批，完整EOF前不开放完成事实。"""186187 def __init__(self,owner,request):188 """验证原会话/投影并构造未开始的解码器。参数: adapter/request。返回: read。异常: 合同/预算错。189 """190 owner._ensure_open()191 if type(request) is not ReadRequest or request.session_id!=owner.session_id:192 raise ValueError('文本CAN读取需要原会话ReadRequest')193 definitions={s.name:s for s in owner.signals}194 if not set(request.signal_names)<=set(definitions):195 raise ValueError('文本CAN请求包含未声明信号')196 self._owner=owner197 self._request=request198 self._budget=owner.budget199 self.identity=uuid4().hex200 self._definitions=tuple(definitions[n] for n in request.signal_names)201 self._held_cost=self._fixed_cost=self._records_cost=self._batch_cost=self._pending_cost=self._validation_cost=self._runtime_cost=self._update_cost=0202 self._decoder=self._raw_store=self._current=self._validator=self._runtime_metadata=self._observation=None203 self._binding_projection=None204 self._records=[]205 self._column=0206 self._raw_eof=self._eof_sent=False207 self._closed=self._exhausted=self._failed=False208 try:209 _reserve_owned_ram(self,131072+_metadata_bytes((request,owner.signals))*32,'_fixed_cost')210 self._validator=BatchValidator(owner.describe(),request)211 try:212 factory = CANTextDecodeRead213 parallel_options = dict(emit_frame_ids=owner._emit_frame_ids,blf_time_scope=owner._blf_time_scope,serial_blf_policy=owner._serial_blf_policy,post_decode_frame_ids=owner._post_decode_frame_ids)214 if owner._process_count > 1 or owner.spec.file_merge == 'time_order':215 from .parallel_can_decode import ParallelCANDecodeRead216 factory = ParallelCANDecodeRead217 parallel_options = dict(process_count=owner._process_count,work_dir=owner.work_dir,218 emit_frame_ids=owner._emit_frame_ids,blf_time_scope=owner._blf_time_scope)219 self._decoder=factory(owner.spec,self._budget,220 owner.signals if owner._update_consumer is not None else self._definitions,221 **parallel_options,222 database_context=owner.database_context, integrity=owner._integrity, trace=owner._trace,retain_source_time=owner.retain_raw_frames,223 legacy_blf_decode=owner._legacy_blf_decode,224 classify_unselected=owner._update_consumer is None,225 **({'cycle_fields':owner._cycle_fields} if owner._cycle_fields else {}),226 **({'metadata_fields':owner._metadata_fields} if owner._metadata_fields else {}),227 **({'selected_ids':owner._selected_ids} if owner._selected_ids is not None else {}),228 **({'time_frame_ids':owner._time_frame_ids} if owner._time_frame_ids is not None else {}))229 except BaseException as error:230 self._decoder=getattr(error,'text_can_cleanup_owner',None)231 raise232 if owner.retain_raw_frames:233 try:234 self._raw_store=TextFrameStore(owner.work_dir/('frames-'+self.identity),self._budget,235 owner.spec.name,owner.session_id,self.identity,database_context=owner.database_context,channel=owner.spec.channel,integrity=owner._integrity,trace=owner._trace)236 except BaseException as error:237 self._raw_store=getattr(error,'text_frame_store_cleanup_owner',None)238 raise239 except BaseException as error:240 _cleanup_owner(self,error,'text_can_observation_cleanup_owner','文本CAN观测构造')241 raise242243 @property244 def closed(self):245 """返回真实关闭状态。参数: 无。返回: bool。异常: 无。246 """247 return self._closed248249 @property250 def failed(self):251 """返回当前读取是否失败。参数: 无。返回: bool。异常: 无。252 """253 return self._failed254255 @property256 def exhausted(self):257 """返回完整EOF及解析器关闭事实。参数: 无。返回: bool。异常: 无。258 """259 return self._exhausted260261 @property262 def observation(self):263 """仅交付本read已完成的真实观测。参数: 无。返回: CANObservation。异常: 未完成/失败。264 """265 if not self._exhausted or self._failed or self._observation is None:266 raise ResultClosedError('文本CAN观测尚未完成')267 return self._observation268269 @property270 def observed_bounds(self):271 """返回同次解析已经看到的首末时间，不额外扫描或冒充EOF。272273 参数: 无。返回: (first,last)，空输入为None。异常: 失败/关闭状态拒绝借读。274 """275 if self._closed or self._failed:276 raise ResultClosedError('文本CAN观测已关闭或失败')277 if self._observation is not None:278 return self._observation.first_ns,self._observation.last_ns279 return self._decoder.first_ns,self._decoder.last_ns280281 @property282 def runtime_metadata(self):283 """交付同次DBC身份与解码次数。参数: 无。返回: 冻结元数据。异常: 未完成。284 """285 self.observation286 return self._runtime_metadata287288 def prepare_metadata(self):289 """在读取前取得同次DBC周期，不冒充最终解码计数。290291 参数: 无。返回: 冻结DBC事实。异常: 关闭/失败或启动原错；清理由本read负责。292 """293 if self._closed or self._failed or self._decoder is None:294 raise ResultClosedError('文本CAN读取已关闭或失败')295 try:296 facts = self._decoder.prepare_metadata()297 if self._binding_projection is None and self._owner._update_projection is not None:298 self._binding_projection = self._owner._update_projection.for_read(facts['fields'])299 return facts300 except BaseException as error:301 self._failed = True302 _cleanup_owner(self, error, 'text_can_observation_cleanup_owner', '文本CAN初始化')303 raise304305 def __iter__(self):306 """返回独立游标。参数: 无。返回: self。异常: 无。307 """308 return self309310 def _accept(self,batch):311 """按既有批合同核同刻组，保留所需pending费用。参数: batch。返回: 原批。异常: 校验/预算错。312 """313 _reserve_owned_ram(self,4096+2*_pending_bytes(self._validator)+1024*batch.sample_count,'_validation_cost')314 self._validator.accept(batch)315 retained=_pending_bytes(self._validator)316 _release_owned_ram(self,self._pending_cost,'_pending_cost')317 self._validation_cost-=retained318 self._pending_cost+=retained319 _release_owned_ram(self,self._validation_cost,'_validation_cost')320 self._current=batch321 return batch322323 def _fill(self):324 """按全部别名的样本容量取有限原记录，所有值只解码一次。参数: 无。返回: None。异常: 原读/预算/暂存错。325 """326 self._records.clear()327 self._column=0328 if self._binding_projection is None:329 self.prepare_metadata()330 _release_owned_ram(self,self._records_cost,'_records_cost')331 raw_rows=min(256,max(1,self._request.batch_rows//max(1,len(self._definitions))))332 for _ in range(raw_rows):333 try:334 record=next(self._decoder)335 except StopIteration:336 self._raw_eof=True337 break338 cost=16384+_record_metadata_bytes(record)*32339 _reserve_owned_ram(self,cost,'_records_cost')340 self._records.append(record)341 if self._raw_store is not None:342 self._raw_store.append(record)343 if self._owner._update_consumer is not None:344 from .message_delivery import deliver_message345 deliver_message(self,self._binding_projection,self._owner._update_consumer,346 record.raw.frame,record.sequence,dict(record.values),record.raw.time_ns,canonical_keys=True)347348 def __next__(self):349 """交付全局时间非递减的有界批，最终EOF后保完成事实。参数: 无。返回: SignalBatch。异常: EOF/原读/预算/合同错。350 """351 if self._closed or self._exhausted:352 raise StopIteration353 if self._failed:354 raise ResultClosedError('失败的文本CAN读取只能清理')355 try:356 self._owner._ensure_open()357 self._current=None358 _release_owned_ram(self,self._batch_cost,'_batch_cost')359 if self._eof_sent:360 self._finish()361 raise StopIteration362 if not self._records or self._column>=max(1,len(self._definitions)):363 self._fill()364 if not self._records and self._raw_eof:365 self._eof_sent=True366 return self._accept(SignalBatch(self._owner.spec.name,self._request.session_id,{},367 self._decoder.last_ns,True,self._decoder.last_ns))368 column_count=min(self._request.batch_rows,max(1,len(self._definitions)-self._column))369 _reserve_owned_ram(self,65536+(2*self._records_cost+16384)*column_count,'_batch_cost')370 columns={}371 if self._definitions:372 for definition in self._definitions[self._column:self._column+column_count]:373 columns[definition.name]=pack_text_records(definition,tuple(self._records),self._request.session_id,self._owner.backend)374 self._column+=column_count375 # watermark含端点；尚待交付的别名可能从块首开始，因此只能推进到首点前1ns。376 progress=self._records[-1].raw.time_ns if self._column>=max(1,len(self._definitions)) else self._records[0].raw.time_ns377 watermark=None if progress is None or progress==-(1<<63) else progress-1378 return self._accept(SignalBatch(self._owner.spec.name,self._request.session_id,columns,watermark,False,None))379 except StopIteration:380 raise381 except BaseException as error:382 self._failed=True383 _cleanup_owner(self,error,'text_can_observation_cleanup_owner','文本CAN观测读取')384 raise385386 def _finish(self):387 """封口可选原帧仓，解析器真正关闭后提交完整事实。参数: 无。返回: None。异常: 封口/关闭/预算错。388 """389 if not self._decoder.exhausted:390 raise ValueError('文本原始解析器没有真正到达EOF')391 raw=self._decoder392 fact=CANObservation(self._owner.spec.name,self._request.session_id,self._request,self.identity,393 raw.first_ns,raw.last_ns,raw.count,True,394 self._owner._blf_time_scope if self._owner._blf_time_scope!='source' else395 'selected' if self._owner._selected_ids is not None and self._owner.spec.source_type=='raw_can_csv' else 'source',396 raw.retained_count,getattr(raw,'full_check_count',0))397 metadata=dict(raw.metadata,decode_calls=raw.decode_calls,record_count=raw.count,source_kind='can_blf' if self._owner._serial_blf_policy is not None else self._owner.spec.source_type)398 _reserve_owned_ram(self,65536+_metadata_bytes(metadata)*32,'_runtime_cost')399 self._runtime_metadata=freeze_metadata(metadata)400 if self._raw_store is not None:401 self._raw_store.seal()402 self._decoder.close()403 self._decoder=None404 self._observation=fact405 self._exhausted=True406407 def take_raw_store(self,receiver,slot):408 """移交本read的同次封口原帧仓。参数: receiver空槽及slot名。返回: None。异常: 未完成/已有责任。409 """410 self.observation411 if self._raw_store is None or self._raw_store.state!='sealed' or getattr(receiver,slot,None) is not None:412 raise ValueError('文本原帧仓不能移交')413 setattr(receiver,slot,self._raw_store)414 self._raw_store=None415416 def close(self):417 """关闭实际解析器/原帧仓并释放自身费用，不释放共享DBC。参数: 无。返回: None。异常: 原清理错可恢复。418 """419 if self._closed:420 return421 self._current=None422 self._records.clear()423 self._validator=None424 if self._decoder is not None:425 self._decoder.close()426 self._decoder=None427 if self._raw_store is not None:428 self._raw_store.close()429 self._raw_store=None430 self._runtime_metadata=self._observation=self._binding_projection=None431 for slot in ('_update_cost','_batch_cost','_records_cost','_pending_cost','_validation_cost','_runtime_cost','_fixed_cost'):432 _release_owned_ram(self,getattr(self,slot),slot)433 self._owner._requests.discard(self)434 self._closed=True435