chronaxis.planning.export_options
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""解释独立的信号CSV和CAN报文输出选项，默认均关闭。23只构造已实现低层导出的类型化规格；不加载DBC、不读取输入或输出文件。4信号分类使用明确别名，报文重建用明确原字段，范围引用既有会话或窗口。5日期规则按用户确认检查，未支持的选项严格拒绝，不让schema接受空实现。6"""78from dataclasses import dataclass910from ..core.types import SignalDType11from ..results.signal_exports import SignalCSVExportSpec12from ..results.signal_grid import SignalGridSpec13from ..results.frame_exports import FrameExportSpec14from ..results.frame_rebuild import FrameGridSpec15from ..results.frame_patch import FrameRebuildRule16from .checks import (PlanValidationError,array_at,choice_at,coordinate_at,duration_at,model_at,object_at,text_at)17from ..core.validation import require_identifier181920@dataclass(frozen=True)21class SignalCSVOutput:22 """保存信号表输出规格及范围，不拥有运行时历史或文件。"""2324 spec: SignalCSVExportSpec25 scope: str = 'session'26 window: str | None = None27 period_driver: str | None = None # 仅DBC信号网格，已解析的原始报文别名。2829 def __post_init__(self):30 """校验冻结声明，typed构造同样受约束。参数: 字段。返回: None。异常: 类型/范围错。31 """32 if type(self.spec) is not SignalCSVExportSpec:33 raise TypeError('signal_csv需要SignalCSVExportSpec')34 _typed_scope(self.scope,self.window)35 if self.period_driver is not None:36 require_identifier(self.period_driver,'signal_csv.resampling.period.driver')37 if (self.period_driver is not None)!=(self.spec.resampling is not None and self.spec.resampling.period_ns is None):38 raise ValueError('DBC信号网格需要已解析驱动，固定网格不能附带驱动')394041@dataclass(frozen=True)42class CANFramesOutput:43 """保存一个来源的原始/重建帧输出规格，与信号表的网格和开关独立。"""4445 source: str46 spec: FrameExportSpec47 scope: str = 'session'48 window: str | None = None49 resampling: FrameGridSpec | None = None50 rules: tuple[FrameRebuildRule,...] = ()5152 def __post_init__(self):53 """校验来源、独立规格和重建规则。参数: 字段。返回: None。异常: 类型/范围错。54 """55 require_identifier(self.source,'can_frames.source'); _typed_scope(self.scope,self.window,allow_all=True)56 if type(self.spec) is not FrameExportSpec or self.resampling is not None and type(self.resampling) is not FrameGridSpec:57 raise TypeError('can_frames需要冻结帧规格与可选网格')58 if type(self.rules) is not tuple or len(self.rules)>4096 or any(type(r) is not FrameRebuildRule for r in self.rules):59 raise TypeError('重建规则需要有限FrameRebuildRule tuple')60 if self.rules and self.resampling is None:61 raise ValueError('未启用重采样不能配置重建规则')62 identities={(r.channel,r.frame_id,r.is_extended,r.is_fd) for r in self.rules}63 if len(identities)!=len(self.rules) or self.resampling is not None and not self.rules and self.resampling.unknown_policy!='nearest':64 raise ValueError('重建规则重复或没有可执行规则')656667def _typed_scope(scope,window,*,allow_all=False):68 """核typed范围与窗口配对。参数: scope/window。返回: None。异常: 不支持或组合错误。69 """70 if scope not in ('session','segments') or scope=='session' and window is not None:71 raise ValueError('导出范围错误')72 if scope=='segments' and not (allow_all and window is None):73 require_identifier(window,'exports.window')747576def require_export_options(plan):77 """核typed选项引用当前来源/信号/窗口，不读DBC。参数: plan。返回: None。异常: 声明或引用错误。78 """79 windows={w.window_id for w in plan.windows}; definitions={s.definition.name:s.definition for s in plan.signals}80 for export_type,expected in (('signal_csv',SignalCSVOutput),('can_frames',CANFramesOutput)):81 option=getattr(plan.outputs,export_type)82 if option is None:83 continue84 if type(option) is not expected:85 raise TypeError(export_type+'需要冻结typed输出声明')86 option.__post_init__()87 if option.scope=='segments' and not ((export_type=='can_frames') and (option.window is None)) and option.window not in windows:88 raise ValueError('导出窗口未在当前计划声明')89 if export_type=='signal_csv':90 if any(name not in definitions for name in option.spec.signals):91 raise ValueError('导出信号未在当前计划声明')92 if option.spec.resampling is not None and any(definitions[name].dtype not in93 (SignalDType.FLOAT64,SignalDType.INT64,SignalDType.UINT64) for name in option.spec.resampling.continuous):94 raise ValueError('状态/枚举信号不能线性插值')95 if option.period_driver is not None:96 _signal_grid_driver(definitions,plan.sources,option.spec.resampling.continuous,option.period_driver,'$.outputs.signal_csv')97 else:98 source=next((s for s in plan.sources if s.name==option.source),None)99 if source is None or source.source_type not in ('can_blf','can_asc','asc_or_blf','raw_can_csv'):100 raise ValueError('报文导出需要本计划CAN来源')101 _require_frame_direction(source)102 if any(rule.channel!=source.config.channel for rule in option.rules):103 raise ValueError('重建报文通道与来源不同')104105106def _scope(raw,path,windows,*,allow_all=False):107 """核会话或单个显式窗口引用。参数: raw/path/windows。返回: (scope,window)。异常: 配置错。108 """109 scope=choice_at(raw.get('scope','session'),('session','segments'),path+'.scope')110 window=raw.get('window')111 if scope=='session' and window is not None:112 raise PlanValidationError(path+'.window','session范围不能同时指定窗口')113 if scope=='segments' and not (allow_all and 'window' not in raw):114 window=text_at(window,path+'.window',identifier=True)115 if window not in {item.window_id for item in windows}:116 raise PlanValidationError(path+'.window','未知窗口：'+window)117 return scope,window118119120def _grid(raw,path):121 """解析显式单位的全局网格。参数: raw/path。返回: period/origin/max_gap整数ns。异常: 时间错。122 """123 return (duration_at(raw['period'],path+'.period',positive=True),124 coordinate_at(raw.get('origin','0s'),path+'.origin'),duration_at(raw['max_gap'],path+'.max_gap',positive=True))125126127def _signal_grid_driver(definitions,sources,continuous,driver,path):128 """同一信号网格使用一个明确DBC周期；多报文不能擅自选最快周期。129130 参数: 有限定义/来源/连续量/可选驱动/路径。返回: 唯一原始别名。异常: 类型/格式/歧义。131 """132 from ..sources.can_spec import can_field_from_key133 source_types={source.name:source.source_type for source in sources}134 candidates={}135 for name in continuous:136 definition=definitions[name]137 if source_types[definition.source] not in ('can_blf','can_asc','asc_or_blf','raw_can_csv'):138 raise PlanValidationError(path+'.resampling.period','已解析信号不能选择DBC周期')139 field=can_field_from_key(definition.field)140 candidates.setdefault((definition.source,field.channel,field.frame_id,field.is_extended,field.is_fd),name)141 if driver is not None:142 if type(driver) is not str or driver not in continuous:143 raise PlanValidationError(path+'.resampling.period.driver','驱动须属于本次连续原始信号')144 return driver145 if len(candidates)!=1:146 raise PlanValidationError(path+'.resampling.period.driver','多个报文周期需要明确指定驱动信号')147 return next(iter(candidates.values()))148149150def parse_signal_output(value,signals,windows,*,sources=()):151 """核显式信号选择及连续量分类。参数: value、当前信号/窗口。返回: SignalCSVOutput或None。异常: 定位配置错。152 """153 if value is None:154 return None155 path='$.outputs.signal_csv'156 raw=object_at(value,path,('signals',),('scope','window','resampling'))157 names=tuple(array_at(raw['signals'],path+'.signals'))158 definitions={s.definition.name:s.definition for s in signals}159 if any(type(n) is not str or n not in definitions for n in names):160 raise PlanValidationError(path+'.signals','导出信号必须为已定义的原信号别名')161 grid=None;driver=None162 if raw.get('resampling') is not None:163 res=object_at(raw['resampling'],path+'.resampling',('continuous','period','max_gap'),('origin',))164 continuous=tuple(array_at(res['continuous'],path+'.resampling.continuous'))165 if any(type(n) is not str or n not in definitions or definitions[n].dtype not in166 (SignalDType.FLOAT64,SignalDType.INT64,SignalDType.UINT64) for n in continuous):167 raise PlanValidationError(path+'.resampling.continuous','连续量必须为当前已定义的数值信号，状态/枚举不能插值')168 if type(res['period']) is dict:169 period_choice=object_at(res['period'],path+'.resampling.period',('mode',),('driver',))170 if 'driver' in period_choice and type(period_choice['driver']) is not str:171 raise PlanValidationError(path+'.resampling.period.driver','显式driver必须为信号名称')172 if period_choice['mode']!='dbc':raise PlanValidationError(path+'.resampling.period.mode','网格对象需dbc，固定周期使用带单位数值')173 driver=_signal_grid_driver(definitions,sources,continuous,period_choice.get('driver'),path)174 period=None175 origin=coordinate_at(res.get('origin','0s'),path+'.resampling.origin')176 gap=duration_at(res['max_gap'],path+'.resampling.max_gap',positive=True)177 else:178 period,origin,gap=_grid(res,path+'.resampling')179 grid=model_at(SignalGridSpec,path+'.resampling',continuous,period,origin,gap)180 spec=model_at(SignalCSVExportSpec,path,names,grid)181 scope,window=_scope(raw,path,windows)182 return SignalCSVOutput(spec,scope,window,driver)183184185def _require_frame_direction(source):186 """只有原帧导出/重建需要收发方向；分析可保留未知方向。"""187 if source.source_type == 'raw_can_csv' and any(188 'is_rx' not in item.layout.fields for item in source.config.inputs):189 raise PlanValidationError('$.outputs.can_frames',190 '原始报文导出或重建需要收发方向，请在CSV字段映射中提供is_rx；普通分析可以省略')191192193def parse_frame_output(value,sources,windows):194 """核帧来源、日期和可选重建规则。参数: value、当前来源/窗口。返回: CANFramesOutput或None。异常: 定位配置错。195 """196 if value is None:197 return None198 path='$.outputs.can_frames'199 raw=object_at(value,path,('format',),('source','scope','window','resampling','output_offset',200 'source_time_reference','capture_date','timezone_offset_minutes'))201 from .input_defaults import source_reference202 source=source_reference(raw,sources,path,can_only=True)203 declared=next((s for s in sources if s.name==source),None)204 if declared is None or declared.source_type not in ('can_blf','can_asc','asc_or_blf','raw_can_csv'):205 raise PlanValidationError(path+'.source','报文导出需要当前已支持的CAN来源')206 _require_frame_direction(declared)207 reference=raw.get('source_time_reference','epoch' if declared.config.timebase.time_type.value=='absolute' else 'relative')208 spec=model_at(FrameExportSpec,path,choice_at(raw['format'],('asc','blf'),path+'.format'),209 coordinate_at(raw.get('output_offset','0s'),path+'.output_offset'),reference,210 raw.get('capture_date'),raw.get('timezone_offset_minutes'))211 scope,window=_scope(raw,path,windows,allow_all=True)212 grid=None; rules=[]213 if raw.get('resampling') is not None:214 res=object_at(raw['resampling'],path+'.resampling',('period','max_gap','messages'),215 ('origin','unknown_policy','max_distance'))216 if type(res['period']) is dict:217 choice=object_at(res['period'],path+'.resampling.period',('mode',),())218 if (choice['mode']!='dbc'):219 raise PlanValidationError(path+'.resampling.period','按各报文DBC周期重采样时，period.mode必须为dbc')220 period=None221 origin=coordinate_at(res.get('origin','0s'),path+'.resampling.origin')222 gap=duration_at(res['max_gap'],path+'.resampling.max_gap',positive=True)223 else:224 period,origin,gap=_grid(res,path+'.resampling')225 distance=duration_at(res['max_distance'],path+'.resampling.max_distance',positive=False) if 'max_distance' in res else None226 grid=model_at(FrameGridSpec,path+'.resampling',period,origin,gap,res.get('unknown_policy','passthrough'),distance)227 for index,item in enumerate(array_at(res['messages'],path+'.resampling.messages')):228 where=f'{path}.resampling.messages[{index}]'229 item=object_at(item,where,('frame_id','continuous'),('channel','is_extended','is_fd','protected','same_time_policy'))230 # 省略表示自动识别；显式null不能悄悄取得默认行为。231 for flag in ('is_extended','is_fd'):232 if flag in item and type(item[flag]) is not bool:233 raise PlanValidationError(where+'.'+flag,'填写时需要bool；自动识别请省略该项')234 from .channels import channel_index235 channel = channel_index(item['channel'],where+'.channel') if 'channel' in item else declared.config.channel236 rule=model_at(FrameRebuildRule,where,channel,item['frame_id'],237 tuple(array_at(item['continuous'],where+'.continuous')),item.get('is_extended'),item.get('is_fd'),238 tuple(array_at(item.get('protected',[]),where+'.protected')),item.get('same_time_policy','error'))239 if rule.channel!=declared.config.channel:240 raise PlanValidationError(where+'.channel','重建规则通道必须匹配来源的DBC通道')241 rules.append(rule)242 identities=[(r.channel,r.frame_id,r.is_extended,r.is_fd) for r in rules]243 if len(rules)>4096 or len(set(identities))!=len(rules):244 raise PlanValidationError(path+'.resampling.messages','报文规则重复或超过4096项')245 if not rules and grid.unknown_policy!='nearest':246 raise PlanValidationError(path+'.resampling.messages','重建需明确报文清单或未知报文最近邻策略')247 return CANFramesOutput(source,spec,scope,window,grid,tuple(rules))248