chronaxis.legacy.binding
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用报文 ID、所需信号和实际帧类型选择 DBC 定义。23本模块只使用调用方已加载的数据库，不打开 DBC 或日志。候选表仅覆盖当前4请求字段；每次运行保存有限的已见身份，避免省略帧型时混合两种同 ID 报文。5"""678def field_matches(field, frame):9 """检查声明是否接受当前实际帧。参数: field定位、frame原帧。返回: bool。异常: 属性错误。10 """11 # 只把未填写的类型当作开放条件；False是明确的标准帧或经典CAN约束。12 return (field.channel == frame.channel and field.frame_id == frame.arbitration_id13 and (field.is_extended is None or field.is_extended == frame.is_extended_id)14 and (field.is_fd is None or field.is_fd == frame.is_fd))151617def dbc_candidates(db, frame_id, signal_names=(), is_extended=None, is_fd=None):18 """筛出定义候选，保留同 ID 不同帧型，不采用数据库的最后一项覆盖。1920 参数: db为已加载数据库；frame_id整数；signal_names所需名称；两个帧型可省略。21 返回: 候选消息tuple。异常: ValueError为报文或信号不存在；不读取日志。22 """23 # 同数值ID可以同时有标准帧和扩展帧定义，不能使用只保留其中一项的ID字典。24 messages = tuple(message for message in db.messages if message.frame_id == frame_id)25 if not messages:26 raise ValueError('DBC中没有报文ID: ' + str(frame_id))27 # 先区分缺失信号，避免把错误名称误报为缺少帧型。28 named = tuple(message for message in messages29 if all(sum(signal.name == name for signal in message.signals) == 130 for name in signal_names))31 if not named:32 raise ValueError('DBC报文中没有唯一的所需信号: ' + ', '.join(signal_names))33 # 保留用户批准的单向兼容：DBC标为FD、实际选择经典CAN；反方向不放宽。34 selected = tuple(message for message in named35 if (is_extended is None or message.is_extended_frame == is_extended)36 and (is_fd is None or message.is_fd == is_fd37 or message.is_fd is True and is_fd is False))38 if not selected:39 raise ValueError('DBC报文与指定标准/扩展帧或CAN/FD不匹配: ' + str(frame_id))40 # 同一帧型仍有两份定义时，之后读取实际帧也无法消除歧义，须现在拒绝。41 kinds = [(message.is_extended_frame, message.is_fd) for message in selected]42 if len(set(kinds)) != len(kinds):43 raise ValueError('DBC报文在相同ID、信号和帧型下仍不唯一，请修正DBC')44 return selected454647def select_message(candidates, frame):48 """按实际类型选唯一定义，允许已批准的 DBC FD 与实际经典 CAN 差异。4950 参数: candidates消息tuple、frame实际帧。返回: 唯一消息或None表示不匹配。51 异常: ValueError为实际类型仍不能消除多定义歧义。52 """53 # 按文件中真实帧型筛选，不用载荷长度或ID大小推测类型。54 typed = tuple(message for message in candidates55 if message.is_extended_frame == frame.is_extended_id56 and (message.is_fd == frame.is_fd or message.is_fd and not frame.is_fd))57 # 同型定义优先，只有没有同型定义时才使用已批准的单向兼容。58 exact = tuple(message for message in typed if message.is_fd == frame.is_fd)59 selected = exact or typed60 if len(selected) > 1:61 raise ValueError('DBC中同ID和帧型仍有多个定义，请修正DBC或信号绑定')62 return selected[0] if selected else None636465class FrameBindingResolver:66 """持有本次所需定义及已见身份；不保存报文载荷或无关信号。67 """6869 def __init__(self, db, fields):70 """建立有限候选表。参数: db已有DBC、fields字段tuple。返回: None。异常: 定位错误。71 """72 # 候选与缓存只属于本次读取；按ID缩小逐帧工作范围，不保存原帧内容。73 self.by_id = {}74 self.database = db75 self.candidates = {}76 self.seen = {}77 self.resolved = {}78 for field in dict.fromkeys(fields):79 self.candidates[field] = dbc_candidates(db, field.frame_id, (field.signal,),80 field.is_extended, field.is_fd)81 self.by_id.setdefault(field.frame_id, []).append(field)8283 def fields_for(self, frame):84 """取得真实匹配字段，并拒绝后续帧改变同一省略型绑定的身份。8586 参数: frame实际数据帧。返回: 匹配字段tuple。异常: ValueError为运行中出现歧义。87 """88 # 远程帧和错误帧没有普通信号更新；成功匹配的身份可在后续同型帧复用。89 if frame.is_remote_frame or frame.is_error_frame:90 return ()91 key = (frame.channel, frame.arbitration_id, frame.is_extended_id, frame.is_fd)92 if key in self.resolved:93 return self.resolved[key][0]94 # 对每个实际包含目标信号的候选检查身份；无关类型不清掉原信号的有效值。95 result = []96 for field in self.by_id.get(frame.arbitration_id, ()):97 if not field_matches(field, frame) or select_message(self.candidates[field], frame) is None:98 continue99 identity = (frame.is_extended_id, frame.is_fd)100 previous = self.seen.get(field)101 if previous is not None and previous != identity:102 missing = 'is_extended' if previous[0] != identity[0] else 'is_fd'103 raise ValueError('同一绑定匹配多种实际帧，请补充 ' + missing + ': ' + field.signal)104 self.seen[field] = identity105 result.append(field)106 # 只缓存匹配到已声明字段的帧，避免无关ID使缓存随日志规模增长。107 fields = tuple(result)108 if fields:109 self.resolved[key] = (fields, self.message_for(frame, fields))110 return fields111112 def message_for(self, frame, fields):113 """核同一帧的字段来自同一解码定义。参数: frame原帧、fields匹配字段。返回: 消息或None。异常: 歧义。114 """115 # 同一帧的多个目标信号必须共用同一DBC消息对象，不能拼接不同定义的结果。116 key = (frame.channel, frame.arbitration_id, frame.is_extended_id, frame.is_fd)117 if key in self.resolved:118 return self.resolved[key][1]119 messages = []120 for field in fields:121 message = select_message(self.candidates[field], frame)122 if message is not None and all(message is not other for other in messages):123 messages.append(message)124 if len(messages) > 1:125 raise ValueError('同一帧的信号指向多个DBC定义，无法唯一解码')126 return messages[0] if messages else None127128 def missing_flags(self, channel, frame_id, extended, fd):129 """为缺少帧型列的原始CSV寻找唯一声明，绝不按小ID或短载荷猜经典帧。130131 参数: channel/frame_id为已解析身份；extended/fd为实际列值或None。132 返回: 两个已确定bool。异常: ValueError提示缺失的源列或绑定帧型。133 """134 # 已绑定字段先提供候选；DBC为FD时不能据此猜实际CSV一定是FD。135 choices = set()136 fields = tuple(field for field in self.by_id.get(frame_id, ()) if field.channel == channel)137 for field in fields:138 for message in self.candidates[field]:139 for candidate_fd in ((field.is_fd,) if field.is_fd is not None140 else (False, True) if message.is_fd else (False,)):141 choices.add((message.is_extended_frame, candidate_fd))142 if not fields:143 # 导出未绑定的原帧仍需完整身份，不能替无关帧伪造标准/经典标记。144 for message in self.database.messages:145 if message.frame_id == frame_id:146 for candidate_fd in ((False, True) if message.is_fd else (False,)):147 choices.add((message.is_extended_frame, candidate_fd))148 # 有列值时以实际列值收窄；只有唯一剩余组合才能补齐缺列。149 choices = {pair for pair in choices if (extended is None or pair[0] == extended)150 and (fd is None or pair[1] == fd)}151 if len(choices) != 1:152 missing = 'is_extended_id / is_extended' if extended is None and len({p[0] for p in choices}) != 1 else 'is_fd'153 raise ValueError('CSV缺少可唯一确定的帧型，请补充列映射或绑定参数 ' + missing)154 return next(iter(choices))155