chronaxis.sources.raw_csv_layout
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""声明原始报文CSV的列和值含义，不按中文列名猜字段。23ColumnMapping选择一个列名、从1开始的列号或固定值，可明确把中文值映射到4布尔/帧类型。CANCSVLayout冻结编码、分隔符和映射，bind_header一次确定列位置，5随后只访问实际配置的列，输出规范字段；这里尚不解析时间、ID或十六进制载荷。6构造/绑定均无I/O；公共来源可先补入其唯一通道，独立布局仍须提供通道。7"""89import codecs10from dataclasses import dataclass, field11from types import MappingProxyType12from collections.abc import Mapping131415# 名称指向物理字段，frame_kind与扩展/FD标志分开，避免“帧格式/帧类型”歧义。16REQUIRED_FIELDS = frozenset(('timestamp', 'frame_id', 'channel', 'data'))17BOOL_FIELDS = frozenset(('is_rx', 'is_extended', 'is_fd', 'bitrate_switch', 'error_state_indicator'))18LENGTH_FIELDS = frozenset(('payload_length', 'dlc_code'))19ALL_FIELDS = REQUIRED_FIELDS | BOOL_FIELDS | LENGTH_FIELDS | {'frame_kind'}20MAX_COLUMNS = 409621MAX_FIXED_TEXT = 16384222324def _scalar(value, label, *, limit=MAX_FIXED_TEXT):25 """限定可静态配置的值。参数: value候选；label定位；limit文本上限。返回: 原值。异常: 类型/长度错误。26 """27 # 不允许任意对象/浮点/None进入配置，也不把bool按int处理。28 if type(value) not in (str, int, bool):29 raise TypeError(label + '需要str/int/bool')30 if type(value) is str and len(value) > limit:31 raise ValueError(label + '文本过长')32 if type(value) is int and not -(2**63) <= value <= 2**64-1:33 raise ValueError(label + '整数超出64位范围')34 return value353637def _fields(value, allowed, required, label):38 """检查严格配置键。参数: value字典、allowed允许键、required必需键、label定位。返回: 原dict。异常: 类型/未知缺失键。39 """40 # JSON形状必须明确；未知拼写不能静默变成另一种来源配置。41 if type(value) is not dict or any(type(key) is not str for key in value):42 raise TypeError(label + '需要字符串键dict')43 if set(value) - set(allowed) or set(required) - set(value):44 raise ValueError(label + '包含未知键或缺少必需键')45 return value464748def _physical_value(name, value):49 """校验已映射的字段类型，避免文本被隐式转成True。5051 参数: name规范字段、value已经取出/查表的值。返回: None。异常: 标志/帧类型或普通字段类型错误。52 """53 # 布尔和帧种类先核实实际含义；数值文字由原始CSV读取器按声明进制转换。54 if name in BOOL_FIELDS:55 if type(value) is not bool:56 raise ValueError('CSV标志需要显式bool映射: ' + name)57 elif name == 'frame_kind':58 if type(value) is not str or value not in ('data', 'remote', 'error'):59 raise ValueError('CSV帧种类需要data/remote/error映射')60 elif name == 'data':61 if type(value) is not str:62 raise ValueError('CSV载荷需要明确的十六进制文本')63 elif type(value) not in (str, int):64 raise ValueError('CSV字段需要文本或整数: ' + name)656667def _common_value(name, value):68 """只识别含义明确的常用文字；显式value_map的结果不经过此函数。"""69 if type(value) is not str:70 return value71 key = value.casefold()72 if name == 'is_rx':73 return {'rx':True, 'tx':False, '接收':True, '发送':False,74 'true':True, 'false':False}.get(key, value)75 if name == 'frame_kind':76 return {'data':'data', '数据帧':'data', 'remote':'remote',77 '远程帧':'remote', 'error':'error', '错误帧':'error'}.get(key, value)78 if name in BOOL_FIELDS:79 return {'true':True, 'false':False}.get(key, value)80 return value818283@dataclass(frozen=True)84class ColumnMapping:85 """固定一种取值来源及可选值映射；复制后不可被调用方字典修改。"""8687 column: str | int | None = None # 列名或从1开始的整数列号，字符串数字仍是列名。88 fixed: str | int | bool | None = None # 缺少实际列时声明固定值；False/0/空字符串都有效。89 value_map: Mapping[str, str | int | bool] | None = None # 严格逐值对应，未登记值报错。90 strip: bool = True # 是否去掉取出单元格两端空白，不改变载荷内部空格。9192 def __post_init__(self):93 """校验来源并冻结映射。参数: 无。返回: None。异常: 模糊来源、坏类型或超限。94 """95 # None只代表未指定；不能用真假判断而误拒固定的False、0或空载荷。96 if (self.column is None) == (self.fixed is None):97 raise ValueError('列映射必须且只能指定column或fixed')98 if self.column is not None:99 if type(self.column) is int:100 if not 1 <= self.column <= MAX_COLUMNS:101 raise ValueError('列号从1开始且不超过4096')102 elif type(self.column) is not str or not self.column or len(self.column) > 256:103 raise ValueError('列名需非空且不超过256字符')104 else:105 _scalar(self.fixed, 'fixed')106 # 查表行为提前冻结；被调用方随后改写的字典不能影响同一分析计划。107 if type(self.strip) is not bool:108 raise TypeError('strip需要bool')109 if self.value_map is not None:110 if not isinstance(self.value_map, Mapping) or not 1 <= len(self.value_map) <= 256:111 raise ValueError('value_map需要1到256项映射')112 values = {}113 for key, value in self.value_map.items():114 if type(key) is not str or len(key) > 256:115 raise ValueError('value_map键需不超过256字符的文本')116 # strip发生在查表前；拒绝永远匹配不到的边缘空白键。117 if self.strip and key != key.strip():118 raise ValueError('strip模式的映射键不能有边缘空白')119 values[key] = _scalar(value, 'value_map值', limit=256)120 object.__setattr__(self, 'value_map', MappingProxyType(values))121122 @classmethod123 def from_dict(cls, value):124 """从严格JSON形状构造。参数: value配置dict。返回: ColumnMapping。异常: 未知键/值错误。125 """126 if type(value) in (str, int):127 return cls(column=value)128 _fields(value, ('column', 'fixed', 'value_map', 'strip'), (), 'column_mapping')129 return cls(**value)130131 def to_dict(self):132 """保存明确声明，保留False/0/空值差异。参数: 无。返回: 独立dict。异常: 无。133 """134 result = {'strip':self.strip}135 result['column' if self.column is not None else 'fixed'] = self.column if self.column is not None else self.fixed136 if self.value_map is not None:137 result['value_map'] = dict(self.value_map)138 return result139140141@dataclass(frozen=True)142class CANCSVLayout:143 """声明原始报文表，不与已解码signal_csv混淆；绑定表头以前不访问文件。"""144145 fields: Mapping[str, ColumnMapping] # 长度省略时取实际载荷字节数；方向省略时保持未知。146 encoding: str = 'utf-8-sig' # 严格编码，无探测/自动重试，仓库样例明确设置gbk。147 delimiter: str = ',' # 一个明确字符，制表样例设置\t，不由扩展名推断。148 has_header: bool = True # 无表头时只允许整数列号/固定值。149 skip_rows: int = 0 # 从文件开始明确跳过的表记录数，不自动寻找“像表头”的行。150 layout_version: str = 'raw-can-csv-layout-v1' # 显式版本，不等于公共analysis schema已接通。151152 def __post_init__(self):153 """冻结布局并校验完整字段声明。参数: 无。返回: None。异常: 类型/缺失/未知字段或编码错误。154 """155 # 明确格式与编码/分隔方式，禁止依文件名猜测或遇错换编码。156 if self.layout_version != 'raw-can-csv-layout-v1':157 raise ValueError('不支持的原始报文CSV布局版本')158 if type(self.encoding) is not str or not self.encoding:159 raise TypeError('CSV编码必须明确')160 codecs.lookup(self.encoding)161 if (type(self.delimiter) is not str or len(self.delimiter) != 1162 or self.delimiter in ('\r', '\n', '\x00', '"')):163 raise ValueError('CSV分隔符需一个非换行/引号/NUL字符')164 if type(self.has_header) is not bool or type(self.skip_rows) is not int or not 0 <= self.skip_rows <= 100000:165 raise ValueError('CSV需要bool表头开关和0到100000的跳过记录数')166 # 长度可由载荷取得；显式长度和DLC仍分别检查，不能互相替代。167 if not isinstance(self.fields, Mapping):168 raise TypeError('CSV fields需要映射')169 if (any(type(key) is not str for key in self.fields) or set(self.fields) - ALL_FIELDS170 or REQUIRED_FIELDS - set(self.fields)):171 raise ValueError('CSV物理字段缺失或未知')172 copied = dict(self.fields)173 if any(type(value) is not ColumnMapping for value in copied.values()):174 raise TypeError('CSV fields值需要ColumnMapping')175 if not self.has_header and any(type(value.column) is str for value in copied.values()):176 raise ValueError('无表头CSV不能按列名取值')177 # 普通数据帧和未置位的FD选项使用默认值；收发方向不补造。178 for name in ('bitrate_switch', 'error_state_indicator'):179 copied.setdefault(name, ColumnMapping(fixed=False))180 copied.setdefault('frame_kind', ColumnMapping(fixed='data'))181 # 固定值和枚举结果在开读前就能核实，不能把配置错误拖到第一条数据才暴露。182 for name, mapping in copied.items():183 if mapping.value_map is not None:184 for value in mapping.value_map.values():185 _physical_value(name, value)186 if mapping.column is None:187 value = mapping.fixed188 if mapping.value_map is not None:189 if type(value) is not str or value not in mapping.value_map:190 raise ValueError('CSV固定值未在value_map声明: ' + name)191 value = mapping.value_map[value]192 else:193 value = _common_value(name, value)194 _physical_value(name, value)195 object.__setattr__(self, 'fields', MappingProxyType(copied))196197 @classmethod198 def from_dict(cls, value):199 """加载严格布局配置。参数: value原生dict。返回: CANCSVLayout。异常: 未知键/字段/列映射错误。200 """201 # 使用同一静态校验创建实例，解释/运行不会各用一套宽松规则。202 _fields(value, ('fields', 'encoding', 'delimiter', 'has_header', 'skip_rows', 'layout_version'), ('fields',), 'raw_csv_layout')203 if type(value['fields']) is not dict:204 raise TypeError('raw_csv_layout.fields需要dict')205 options = {key:item for key,item in value.items() if key != 'fields'}206 return cls({key:ColumnMapping.from_dict(item) for key,item in value['fields'].items()}, **options)207208 def to_dict(self):209 """输出完整已采用布局供解释/追溯。参数: 无。返回: 独立JSON字典。异常: 无。210 """211 return dict(layout_version=self.layout_version, encoding=self.encoding, delimiter=self.delimiter,212 has_header=self.has_header, skip_rows=self.skip_rows,213 fields={key:item.to_dict() for key,item in self.fields.items()})214215 def bind_header(self, header=None):216 """只解析一次表头列位置。参数: header为有表头时的str列表/tuple，无表头必须None。217218 返回: BoundCANCSVLayout。异常: 表头类型/重复/缺列或表头开关矛盾；不读任何记录。219 """220 # 表头每次来源绑定只检查一次，有界列数防止构造无限名称索引。221 if self.has_header:222 if type(header) not in (tuple, list) or not 1 <= len(header) <= MAX_COLUMNS:223 raise ValueError('CSV表头需要1到4096列')224 if any(type(item) is not str or len(item) > 256 for item in header):225 raise ValueError('CSV表头需要有限文本')226 # 允许未使用的空表头（仓库制表样例尾列）；非空重复名不能有歧义。227 names = [item for item in header if item]228 if len(names) != len(set(names)):229 raise ValueError('CSV表头存在重复列名')230 header = tuple(header)231 elif header is not None:232 raise ValueError('无表头配置不能另传表头')233 # 名称按原样匹配；整数只在这里从1基转成0基，后续逐行不再搜索表头。234 bound = []235 for name, mapping in self.fields.items():236 if type(mapping.column) is str:237 if mapping.column not in header:238 raise ValueError('CSV缺少映射列: ' + mapping.column)239 position = header.index(mapping.column)240 elif type(mapping.column) is int:241 position = mapping.column - 1242 if header is not None and position >= len(header):243 raise ValueError('CSV映射列号超出表头')244 else:245 position = None246 bound.append((name, mapping, position))247 return BoundCANCSVLayout(tuple(bound))248249250@dataclass(frozen=True)251class BoundCANCSVLayout:252 """保存有结构检查的列位置，只访问选择列；不认证实际文件表头或内容。"""253254 bindings: tuple = field(repr=False) # (规范字段名,冻结映射,0基列位置或None)。255256 def __post_init__(self):257 """校验可独立构造的绑定。参数: 无。返回: None。异常: 绑定结构、字段或列号错误。258 """259 # 公开构造也要守住字段唯一性和列范围；这只是结构检查，不冒充实际表头证据。260 if type(self.bindings) is not tuple or not len(REQUIRED_FIELDS) <= len(self.bindings) <= len(ALL_FIELDS):261 raise ValueError('CSV绑定数量错误')262 names = set()263 for entry in self.bindings:264 if type(entry) is not tuple or len(entry) != 3:265 raise ValueError('CSV绑定需要字段/映射/位置三元组')266 name, mapping, position = entry267 if type(name) is not str or name not in ALL_FIELDS or name in names or type(mapping) is not ColumnMapping:268 raise ValueError('CSV绑定字段或映射错误')269 if (mapping.column is None) != (position is None):270 raise ValueError('CSV固定值与列位置矛盾')271 if position is not None and (type(position) is not int or not 0 <= position < MAX_COLUMNS):272 raise ValueError('CSV绑定列位置错误')273 if type(mapping.column) is int and position != mapping.column - 1:274 raise ValueError('CSV绑定列号与位置矛盾')275 names.add(name)276 if REQUIRED_FIELDS - names:277 raise ValueError('CSV绑定缺少物理字段')278279 def map_row(self, row, *, names=None):280 """提取并明确映射一条表记录。参数: row为list/tuple单元格；names为内部按需字段集合，省略时提取声明全部列。281282 返回: 只读规范字段字典；时间、数值和载荷由原始CSV读取器解析。283 异常: 缺列、所需单元格非文本、未登记枚举、布尔/帧类型或值类型错误。284 """285 # 只触碰有消费者的列，未选择的列不做内容类型检查或数值转换。286 if type(row) not in (tuple, list) or len(row) > MAX_COLUMNS:287 raise ValueError('CSV记录需要不超过4096列')288 result = {}289 for name, mapping, position in self.bindings:290 if names is not None and name not in names:291 continue292 if position is None:293 value = mapping.fixed294 else:295 if position >= len(row) or type(row[position]) is not str:296 raise ValueError('CSV缺少所需文本列: ' + name)297 value = row[position].strip() if mapping.strip else row[position]298 if len(value) > MAX_FIXED_TEXT:299 raise ValueError('CSV所需单元格过长: ' + name)300 # 声明了值表就必须匹配，不能在未知中文状态上回退成某个默认身份。301 if mapping.value_map is not None:302 if type(value) is not str or value not in mapping.value_map:303 raise ValueError('CSV值未在value_map声明: ' + name)304 value = mapping.value_map[value]305 else:306 value = _common_value(name, value)307 # 标志/数据与远程类型必须明确，不能把非空中文字符串普遍当True。308 _physical_value(name, value)309 result[name] = value310 return MappingProxyType(result)311