chronaxis.planning.inputs
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""检查来源和信号配置，建立静态定位与显式转换声明。23输入是当前配置的sources、signals数组和相对路径基准，输出SourcePlan4与SignalPlan元组。内存原行已在配置中，可核对原顺序和已出现字段；CSV5只保存有序路径和格式，不读取文件表头。时间使用明确单位、公共起点或时区，6原字段保持原文，整数值不先转浮点。valid_for声明有效值最多保持的时间，7省略时取本信号period或运行期DBC周期的三倍。8CAN支持既有单BLF profile与新ASC/原始CSV静态规格，完整原定位保存为规范字段键；9旧表格profile仍报告capability_unavailable。仅需补齐信号元数据时读取DBC，不读日志。10"""1112# 配置基准和时基身份使用标准库，无输入扫描。13from hashlib import sha25614from pathlib import Path15# 原模型负责声明范围/值类型/时序顺序门槛。16from ..core.binding import SignalDefinition17from ..core.unit_conversion import affine_conversion, converted_dtype, require_config_version18from ..core.models import MemoryRow, SourceSpec, TimebaseSpec19from ..core.types import SignalDType, TimeType20# CSV只构造配置，不实例化文件读取器。21from ..sources.csv_spec import SignalCSVSpec22from ..sources.can_spec import LegacyCANSpec23from ..sources.text_can_spec import CANTextSpec24from ..legacy.records import PROFILE_ID, field_from_dict25# 配置校验严格定位并深冻结元数据。26from .diagnostic_collection import usable_node, collect_node_error27from .checks import (PlanValidationError, absolute_path, array_at, canonical, choice_at, coordinate_at,28 duration_at, metadata_at, model_at, number_at, object_at, text_at, timezone_at)29from .models import SignalPlan, SourcePlan30from .signal_validity import declared_valid_for31from .input_defaults import source_defaults, source_reference32from .input_paths import expand_source_files33from .dbc_defaults import DBCDefaults343536def parse_sources(value, base: Path) -> tuple[SourcePlan, ...]:37 """按声明顺序解析内存、已解码CSV或原始CAN规格，不读取分析文件。3839 参数:40 value (list): sources配置数组，每项含名称、来源种类和时间声明。41 base (Path): 相对CSV、BLF及DBC路径的词法解释基准。42 返回:43 tuple[SourcePlan, ...]。至少含一个来源；路径转为绝对定位并保留原顺序。44 内存原行按声明顺序构造，CSV/CAN仅构造配置，文件、表头及DBC尚未核实。45 异常:46 PlanValidationError: 字段、格式、名称、时序、时间基准或来源能力错误，47 保存对应配置位置；未接入来源使用capability_unavailable错误码。48 副作用:49 不打开CSV或DBC；地区时区解析可能访问本机时区数据库。50 """51 # 来源数组保持声明顺序，只有用户明确的目录/通配符会枚举文件名。52 items = array_at(value, '$.sources')53 # dataset分析基准需要至少一个明确来源。54 if not items:55 # 空来源计划无法声明时间坐标。56 raise PlanValidationError('$.sources', '至少需要一个来源')57 # 当前计划内登记唯一来源名称。58 result, names = [], set()59 # 原文件和来源顺序保持调用者声明。60 for index, raw in enumerate(items):61 # 每个错误准确定位数组项。62 path = f'$.sources[{index}]'63 if not usable_node(path, raw):64 continue65 try:66 path = f'$.sources[{index}]'67 # 在种类分派前确认基本对象与kind，不忽略未知顶层键。68 if type(raw) is not dict:69 # 不从类名或字符串构建来源。70 raise PlanValidationError(path, 'source必须为对象')71 # 仅显式目录/通配符枚举文件名；完整路径仍不stat、不打开业务输入。72 if raw.get('source_type') != 'memory' and ('path' in raw or 'paths' in raw):73 paths, shifts = expand_source_files(raw, base, path)74 raw = dict(raw, paths=[str(item) for item in paths])75 raw.pop('path', None)76 if shifts is not None:77 raw['offsets'] = list(shifts)78 # 默认仅识别ASC/BLF，CSV/内存保持明确种类；省略时间不代表自动校时。79 raw = source_defaults(raw, index)80 # 解析补齐后的来源类型，未知值仍由统一选择检查报告。81 source_type = choice_at(raw.get('source_type'), ('memory', 'signal_csv', 'can_asc', 'asc_or_blf', 'can_blf', 'legacy_raw_csv','raw_can_csv'), path + '.source_type')82 if source_type in ('can_asc','asc_or_blf','raw_can_csv'):83 # 新路径独立声明格式和连续文件，不冒充旧profile或把原始表当信号表。84 from .text_inputs import parse_text_source85 source=parse_text_source(raw,base,path)86 if source.name in names:87 raise PlanValidationError(path+'.name','来源重复：'+source.name)88 names.add(source.name); result.append(source)89 continue90 # 其余来源只接受内存、信号CSV及受控BLF读取路径。91 if source_type not in ('memory', 'signal_csv', 'can_blf'):92 # 未支持的来源应在配置阶段拒绝。93 raise PlanValidationError(path + '.source_type', f'{source_type}不支持；原始CSV请使用raw_can_csv来源', 'capability_unavailable')94 # 格式特有字段必须按种类严格限定。95 if source_type == 'can_blf':96 # BLF来源使用固定profile、单通道DBC和有序文件路径，只接受声明的读取选项。97 raw = object_at(raw, path, ('name', 'source_type', 'time', 'dbc_by_channel'),98 ('path', 'paths', 'reader_options', 'metadata', 'decode_scope', 'input_check', 'legacy_profile', 'file_merge'))99 else:100 optional = ('metadata',) if source_type == 'memory' else ('path', 'paths', 'encoding', 'delimiter', 'missing_values', 'metadata')101 # memory行或CSV时间列不能省略。102 raw = object_at(raw, path, ('name', 'source_type', 'time', 'rows' if source_type == 'memory' else 'time_column'), optional)103 # 来源名与DSL别名采用统一安全标识规则。104 name = text_at(raw['name'], path + '.name', identifier=True)105 # 重名不能选首项/覆盖。106 if name in names:107 # 精确指出第二次定义位置。108 raise PlanValidationError(path + '.name', f'来源重复：{name}')109 # 只登记当前计划名称。110 names.add(name)111 # 不接受漂移/隐式机器时区等未定义字段。112 clock = object_at(raw['time'], path + '.time', ('time_type', 'unit'), ('origin', 'offset', 'timezone'))113 # 时间坐标种类与单位独立。114 time_type = choice_at(clock['time_type'], ('relative', 'absolute'), path + '.time.time_type')115 # ISO表示只允许absolute。116 unit = choice_at(clock['unit'], ('s', 'ms', 'us', 'ns', 'ISO8601'), path + '.time.unit')117 if source_type == 'can_blf':118 # 原CAN timestamp固定为秒；不接受不会参与该转换的timezone声明。119 choice_at(unit, ('s',), path + '.time.unit')120 if 'timezone' in clock:121 raise PlanValidationError(path + '.time.timezone', '固定CAN秒时间不接受timezone')122 # 首版relative必须有明确公共起点声明。123 if time_type == 'relative' and ('origin' not in clock or unit == 'ISO8601' or 'timezone' in clock):124 # 不把原首点0视为公共origin。125 raise PlanValidationError(path + '.time', 'relative必须origin，不接受ISO8601或timezone')126 # absolute坐标已经UTC，不接相对origin。127 if time_type == 'absolute' and 'origin' in clock:128 # 防止一来源同时声明两种坐标含义。129 raise PlanValidationError(path + '.time.origin', 'absolute不接受relative origin')130 # 同一完整relative origin生成相同ID，absolute统一UTC。131 origin = text_at(clock['origin'], path + '.time.origin') if time_type == 'relative' else None132 # ID来自声明内容，不来自机器、文件名或实际首点。133 clock_id = 'relative_' + sha256(origin.encode('utf-8')).hexdigest()[:16] if origin is not None else 'utc'134 # 固定映射坐标保持定宽纳秒，负offset允许。135 offset = coordinate_at(clock.get('offset', '0s'), path + '.time.offset')136 # naive日期是否可解析由实际显式tz决定。137 tz = timezone_at(clock.get('timezone'), path + '.time.timezone')138 # 型别与origin契约沿用已验时间模型。139 timebase = model_at(TimebaseSpec, path + '.time', clock_id, TimeType(time_type), origin)140 # 规范定位身份包括完整原格式与时间声明。141 identity_input = dict(raw)142 # memory行在配置中已经可见，允许检查原顺序但不补数据。143 if source_type == 'memory':144 # memory行time必须明确单位，不用source.unit解释裸数。145 rows = []146 # 行序号由声明原序生成，不排序。147 for row_index, row in enumerate(array_at(raw['rows'], path + '.rows')):148 # 行仅time与values，未知字段不忽略。149 row_path = f'{path}.rows[{row_index}]'150 # 保留原字段值且拒绝自由行选项。151 row = object_at(row, row_path, ('time', 'values'))152 # 原值对象不是自由元数据，但字段名称允许完整原文。153 if type(row['values']) is not dict:154 # 不接受不明列排列或假默认值。155 raise PlanValidationError(row_path + '.values', '必须为原字段对象')156 # 单位/日期与时基种类显式配合。157 time_ns = coordinate_at(row['time'], row_path + '.time', absolute=time_type == 'absolute', tz=tz)158 # 原标量模型复制只读值映射，不将None补0。159 rows.append(model_at(MemoryRow, row_path, time_ns=time_ns, values=row['values'], sequence=row_index))160 # SourceSpec检查来源内回退/offset溢出，不重排原行。161 config = model_at(SourceSpec, path, name=name, timebase=timebase, rows=tuple(rows), offset_ns=offset)162 else:163 # 文件路径必须且只能使用一种明确写法。164 if ('path' in raw) == ('paths' in raw):165 # 文件选择已在前面展开，不在这里重复扫描目录或拼接两组路径。166 raise PlanValidationError(path, 'path与paths必须且只能提供一个')167 # path单项和paths数组归一为严格原序tuple。168 raw_paths = [raw['path']] if 'path' in raw else array_at(raw['paths'], path + '.paths')169 # 词法绝对化不检查文件存在，配置目录作为明确基准。170 paths = tuple(absolute_path(item, base, f'{path}.paths[{i}]') for i, item in enumerate(raw_paths))171 if source_type == 'can_blf':172 # 每个来源按声明顺序读取1至4096个BLF文件，不扫描目录或忽略额外路径。173 if not 1 <= len(paths) <= (4096):174 raise PlanValidationError(path + '.paths', 'BLF来源需要1至4096个有序文件路径')175 profile = choice_at(raw.get('legacy_profile', PROFILE_ID), (PROFILE_ID,), path + '.legacy_profile')176 options = object_at(raw.get('reader_options', {}), path + '.reader_options', (),177 ('decode_policy',))178 dbc = raw['dbc_by_channel']179 if type(dbc) is not dict or len(dbc) != 1:180 raise PlanValidationError(path + '.dbc_by_channel', '当前CAN必须恰有一个通道到DBC路径声明')181 channel_text = next(iter(dbc))182 from .channels import channel_index183 channel = channel_index(channel_text, path + '.dbc_by_channel', key=True)184 dbc_path = absolute_path(dbc[channel_text], base, path + '.dbc_by_channel.' + channel_text)185 # model_at的path参数表示错误位置；源模型path按位置传递，避免同名冲突。186 config = model_at(LegacyCANSpec, path, name, paths[0], dbc_path,187 channel, timebase, unit, offset, profile,188 choice_at(raw.get('decode_scope', 'requested'), ('all', 'requested'), path + '.decode_scope'),189 choice_at(raw.get('input_check', 'auto'), ('auto', 'full'), path + '.input_check'),190 paths if len(paths)>1 else (),191 choice_at(options.get('decode_policy','strict'),192 ('strict','partial','legacy'),path+'.reader_options.decode_policy'),193 file_merge=choice_at(raw.get('file_merge','concatenate'),('concatenate','time_order'),path+'.file_merge'))194 # 默认按实际需求执行，不回填raw；实际执行策略还进入来源描述和运行证明。195 # 实际绝对定位进入来源声明身份，不读取路径内容或猜测DBC元数据。196 identity_input['dbc_by_channel'] = {channel_text: str(dbc_path)}197 else:198 # missing标记精确匹配，包括空字符串，不strip或猜NA。199 missing = array_at(raw.get('missing_values', ['']), path + '.missing_values')200 # SignalCSVSpec检查空paths/编码/分隔符/重复标记，无输入I/O。201 config = model_at(SignalCSVSpec, path, name=name, paths=paths, timebase=timebase,202 time_column=text_at(raw['time_column'], path + '.time_column'), unit=unit, offset_ns=offset,203 timezone=tz, encoding=raw.get('encoding', 'utf-8-sig'), delimiter=raw.get('delimiter', ','), missing_values=tuple(missing))204 # 实际定位身份使用已解析绝对路径，仍保留数组原序。205 identity_input['paths'] = [str(item) for item in paths]206 # 两种等价路径输入形式不保留冗余path键。207 identity_input.pop('path', None)208 # 自由元数据深冻结，不向reader转发未定义键。209 metadata = metadata_at(raw.get('metadata', {}), path + '.metadata')210 # 构造来源声明模型；文件存在性和实际列仍留待读取核实。211 result.append(SourcePlan(name, source_type, config, unit, clock.get('timezone'), canonical(identity_input), metadata))212 except PlanValidationError as error:213 collect_node_error(error, path, raw)214 # 不保留用户可变列表引用。215 return tuple(result)216217218def parse_signals(value, sources: tuple[SourcePlan, ...], *, config_version: int = 4,219 enum_codes=()) -> tuple[SignalPlan, ...]:220 """检查原字段到简单别名的转换与有效期声明。221222 参数:223 value (list): signals配置数组，声明来源、原字段、类型、单位及可选period/valid_for。224 sources (tuple[SourcePlan, ...]): 当前计划已经解析的来源。225 config_version (int): 当前仅支持4；只在两个系数都缺省时自动推导。226 返回:227 tuple[SignalPlan, ...]。保持信号顺序，choices原编码类型保留，228 valid_for转为正整数纳秒；缺省且等待DBC时保存None，转换和可选语义按核心模型检查。229 异常:230 PlanValidationError: 结构、来源、字段、类型、单位、有效期或转换契约错误。231 边界:232 非空内存可检查字段；CSV/CAN日志不读。缺省CAN元数据只读取DBC。233 本函数保存有效期声明，不保持信号值或执行过期判断。234 """235 # 版本在配置构建时确定，低层模型不猜系数是否被用户省略。236 require_config_version(config_version)237 # 来源定位表仅当前计划有效。238 source_map = {item.name: item for item in sources}239 defaults = DBCDefaults()240 # 信号原顺序保留，唯一性由声明绑定统一检查。241 result = []242 # 可选项来自当前配置schema，不开放自由质量或额外有效期政策。243 optional = ('valid_for', 'period', 'input_unit', 'scale', 'offset', 'sample_policy', 'same_time_policy', 'choices', 'valid_range', 'invalid_values',244 'sign_convention', 'measurement_location', 'mapping_version', 'semantic_id', 'semantic_contract_version', 'metadata')245 # 不将字符串解释为signal迭代器。246 from .signal_groups import signal_entries247 for raw, path in signal_entries(value):248 # 未登记的字段明确报错，不能静默忽略拼写错误或已删除的旧参数。249 if not usable_node(path, raw):250 continue251 try:252 raw = object_at(raw, path, ('name', 'field'), optional + ('source','dtype','unit','code_dtype'))253 if 'input_unit' in raw and raw['input_unit'] is None:254 raise PlanValidationError(path + '.conversion.input_unit',255 'input_unit不能为null；未指定输入单位时请省略此字段')256 if 'dtype' in raw:257 choice_at(raw['dtype'],tuple(item.value for item in SignalDType),path+'.dtype')258 if 'unit' in raw:259 text_at(raw['unit'],path+'.unit')260 # 来源名本身是简单标识符。261 source = source_reference(raw, sources, path)262 # 不靠相同semantic_id推测来源。263 if source not in source_map:264 # static也必须唯一定位当前来源。265 raise PlanValidationError(path + '.source', f'未知来源：{source}','invalid_reference',266 hint='填写sources中已声明的name：'+', '.join(source_map))267 # memory行已在配置内可见时检查实际字段，不为CSV额外读表头。268 config = source_map[source].config269 if isinstance(config, (LegacyCANSpec,CANTextSpec)):270 # 原CAN定位严格复用六字段模型；只保存规范键，不把点号原名变成DSL属性。271 # 当前来源只有一个DBC通道；省略时继承，显式错误仍由原模型检查。272 field_raw = raw['field']273 if type(field_raw) is dict:274 field_raw = dict(field_raw)275 from .channels import channel_index276 field_raw['channel'] = (channel_index(field_raw['channel'], path + '.field.channel')277 if 'channel' in field_raw else config.channel)278 locator = model_at(field_from_dict, path + '.field', field_raw)279 if locator.channel != config.channel:280 raise PlanValidationError(path + '.field.channel', '信号通道必须与来源唯一DBC通道一致')281 raw = defaults.fill(raw,config,locator,path,config_version=config_version,need_code=raw['name'] in enum_codes)282 try:283 # 规范定位也受CAN读取协议的有界JSON约束；生成失败仍归为当前位置的配置错误。284 field = locator.key285 except (TypeError, ValueError) as error:286 raise PlanValidationError(path + '.field', str(error)) from error287 else:288 # memory/CSV仍只接受原字段文本，不能借CAN对象扩展其他来源的定位语义。289 field = text_at(raw['field'], path + '.field')290 for key in ('dtype', 'unit'):291 if key not in raw:292 raise PlanValidationError(path + '.' + key,'已解析数据没有DBC元数据，请提供' + key, 'missing_field')293 # 空memory无实际字段证明，保留空观测状态供报告。294 if isinstance(config, SourceSpec) and config.rows and not any(field in row.values for row in config.rows):295 # 不造虚拟字段通过T39。296 raise PlanValidationError(path + '.field', f'原memory字段不存在：{field}')297 # CAN缺省值已由DBC元数据推导；不扫描日志猜数据类型。298 dtype = choice_at(raw['dtype'], tuple(item.value for item in SignalDType), path + '.dtype')299 # choices用二元JSON数组保存int/bool/str编码类型，避免对象键字符串化。300 choices = []301 # 每个明确映射均保留原序和原类型。302 for choice_index, pair in enumerate(array_at(raw.get('choices', []), path + '.choices')):303 # 不接无类型字典编码或错长度配对。304 if type(pair) is not list or len(pair) != 2:305 # 核心随后继续检查标签域与重复编码。306 raise PlanValidationError(f'{path}.choices[{choice_index}]', 'choices项必须为[原编码, 标签]二元数组')307 # 仅冻结结构，不做bool truthiness或整数float转换。308 choices.append(tuple(pair))309 # 范围若存在必须保持二端点原数值。310 valid_range = tuple(array_at(raw['valid_range'], path + '.valid_range')) if 'valid_range' in raw else None311 # 用户有效期优先；信号周期乘三只决定保持上限，不改报文实际时间。312 valid_for_ns = declared_valid_for(raw, path,313 dbc_available=isinstance(config, (LegacyCANSpec, CANTextSpec)))314 # 只在两个系数均省略时生成完整换算；显式一个系数则其余按身份默认。315 scale, offset = raw.get("scale", 1), raw.get("offset", 0)316 original_dtype = SignalDType(dtype)317 if "scale" not in raw and "offset" not in raw:318 scale, offset, _ = model_at(affine_conversion, path + ".conversion",319 raw.get("input_unit", raw["unit"]), raw["unit"])320 dtype = model_at(converted_dtype, path + ".dtype", original_dtype, scale, offset).value321 # core检查所有声明转换及语义，额外factor只在解码输出之后应用一次。322 definition = model_at(SignalDefinition, path, name=raw['name'], source=source, field=field,323 dtype=SignalDType(dtype), unit=raw['unit'], valid_for_ns=valid_for_ns,324 input_unit=raw.get('input_unit'), scale=scale, offset=offset, choices=tuple(choices),325 config_version=config_version, input_dtype=original_dtype,326 code_dtype=SignalDType(choice_at(raw['code_dtype'], ('int64','uint64'), path+'.code_dtype'))327 if 'code_dtype' in raw else None,328 valid_range=valid_range, invalid_values=tuple(array_at(raw.get('invalid_values', []), path + '.invalid_values')),329 sample_policy=raw.get('sample_policy', 'backward_hold'), same_time_policy=raw.get('same_time_policy', 'error'),330 semantic_id=raw.get('semantic_id'), semantic_contract_version=raw.get('semantic_contract_version'),331 mapping_version=raw.get('mapping_version'), sign_convention=raw.get('sign_convention'), measurement_location=raw.get('measurement_location'))332 # 自由metadata不作为隐藏转换或质量策略。333 result.append(SignalPlan(definition, metadata_at(raw.get('metadata', {}), path + '.metadata')))334 except PlanValidationError as error:335 collect_node_error(error, path, raw)336 # 全部契约已经冻结，不依赖用户列表生命周期。337 return tuple(result)338