chronaxis.sources.can_updates
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""把三种原始报文入口的必要值交付给同一个逐消息计算模型。23投影只保存有限字段定位和可共享的来源引用。BLF交付的键是DBC信号名，4文本入口交付的键是规范字段；转换后均为原DBC信号名，不保留载荷副本、5原时间字符串或第二套时间。LegacyDecodedUpdate是已有计算模型的名称，6使用它不代表复现旧程序的缓存行为。7"""89from types import MappingProxyType1011from ..core.models import SourceRef12from ..core.validation import require_identifier13from ..legacy.records import LegacyCANField14from ..legacy.updates import LegacyDecodedUpdate, LegacyUntimedUpdate151617def require_message_keys(keys, *, channel=None):18 """校验有限触发声明。参数: keys完整定位tuple、可选来源通道。返回: 原tuple。异常: ValueError。"""19 if type(keys) is not tuple or len(keys) > 4096:20 raise ValueError('消息触发定位需要最多4096项tuple')21 for key in keys:22 if (type(key) is not tuple or len(key) != 4 or type(key[0]) is not int or key[0] < 023 or type(key[1]) is not int or key[2] is not None and type(key[2]) is not bool24 or key[3] is not None and type(key[3]) is not bool25 or not 0 <= key[1] <= (0x7ff if key[2] is False else 0x1fffffff)26 or channel is not None and key[0] != channel):27 raise ValueError('消息触发定位需要合法通道、ID、扩展及FD标志，且属于当前来源通道')28 return keys293031class CANUpdateProjection:32 """按完整报文身份索引所需字段；同一帧的多个别名只传一次物理值。"""3334 def __init__(self, source, session_id, fields, *, message_keys=(), clock_fields=()):35 """建立有限投影，不读取DBC或输入文件。3637 参数: source/session_id为来源和会话；fields为所需LegacyCANField元组；38 message_keys为无值公式仍需触发的(channel,id,extended,fd)元组；39 clock_fields只借DBC候选确定驱动帧，不加入值解码或值输出。40 返回: None；初始化实例状态。异常: TypeError/ValueError为字段、身份或同帧信号定位冲突。41 """42 if (type(fields) is not tuple or type(clock_fields) is not tuple43 or any(type(field) is not LegacyCANField for field in (*fields,*clock_fields))):44 raise TypeError('消息投影需要LegacyCANField tuple')45 require_identifier(source, 'message.source')46 require_identifier(session_id, 'message.session_id')47 groups = {}48 for key in require_message_keys(message_keys):49 groups.setdefault(key, {})50 self._value_fields = frozenset(fields)51 all_fields = tuple(dict.fromkeys((*fields, *clock_fields)))52 # 声明固定后规范键不再变化；在既有字段空间内共享，避免逐报文重新编码JSON。53 self._canonical_keys = MappingProxyType({field: field.key for field in all_fields})54 for field in all_fields:55 key = (field.channel, field.frame_id, field.is_extended, field.is_fd)56 group = groups.setdefault(key, {})57 if field.signal in group and group[field.signal] != field:58 raise ValueError('同一报文信号存在不同定位')59 group[field.signal] = field60 # 引用只标识必需的来源及代表字段，逐帧序号由已有更新模型保存一次。61 self._groups = MappingProxyType({key: (tuple(group.values()),62 SourceRef(session_id, source, self._canonical_keys[next(iter(group.values()))] if group else '__raw_message__'))63 for key, group in groups.items()})64 self.source = source65 self._seen = {}66 self._resolved = {}67 self._dbc_frames = {}68 self.compatibility_fields = {}6970 def configure_metadata(self, facts):71 """安装同次DBC的有限候选。参数: facts字段事实。返回: None。异常: 非法定位错误。72 """73 from ..legacy.records import field_from_dict74 self._dbc_frames = {field_from_dict(dict(fact['locator'])): tuple(tuple(pair) for pair in fact['dbc_frames'])75 for fact in facts if 'dbc_frames' in fact}7677 def for_read(self, facts):78 """为单次读取隔离已见状态。参数: facts同次DBC事实。返回: 新投影。异常: 元数据错误。79 """80 from copy import copy81 result = copy(self)82 result._seen = {}83 result._resolved = {}84 result.compatibility_fields = {}85 result.configure_metadata(facts)86 return result8788 def _matching(self, frame):89 """匹配四种可选选择器，记录唯一实际身份。参数: frame。返回: 字段和引用或None。异常: 迟到歧义。90 """91 if frame.is_remote_frame or frame.is_error_frame:92 return None93 frame_key = (frame.channel, frame.arbitration_id, frame.is_extended_id, frame.is_fd)94 if frame_key in self._resolved:95 return self._resolved[frame_key]96 actual = (frame.is_extended_id, frame.is_fd)97 fields, reference = {}, None98 for extended in (frame.is_extended_id, None):99 for fd in (frame.is_fd, None):100 key = (frame.channel, frame.arbitration_id, extended, fd)101 item = self._groups.get(key)102 if item is None:103 continue104 # 候选定义里没有该信号的另一种同ID帧属于无关输入，不制造失效更新。105 selected = []106 for field in item[0]:107 candidates = self._dbc_frames.get(field)108 if candidates is not None and not any(e == actual[0] and (f == actual[1] or f and not actual[1])109 for e, f in candidates):110 continue111 selected.append(field)112 if candidates is not None and not actual[1] and (actual[0], False) not in candidates:113 self.compatibility_fields[field] = actual[0]114 if item[0] and not selected:115 continue116 for identity_key in selected or (key,):117 previous = self._seen.get(identity_key)118 if previous is not None and previous != actual:119 missing = 'is_extended' if previous[0] != actual[0] else 'is_fd'120 raise ValueError('同一报文绑定匹配多种实际帧，请补充 ' + missing)121 self._seen[identity_key] = actual122 fields.update((field, None) for field in selected)123 reference = reference or item[1]124 if reference is None:125 return None126 result = (tuple(fields), reference)127 # 只缓存已声明且成功匹配的身份，最多每声明四种，不随无关报文数量增长。128 self._resolved[frame_key] = result129 return result130131 def accepts_frame(self, frame):132 """判断是否有值或触发需求。参数: 原帧。返回: bool。异常: 非原帧属性错。"""133 return self._matching(frame) is not None134135 def fields_for(self, frame):136 """定位目标数据帧，未知ID/其他通道或帧型不参与本投影。137138 参数: frame为实际读取器交付且已校验的原帧。返回: 字段元组，非目标为空。139 异常: 非原帧对象缺少必需属性时AttributeError；不吞读取错误。140 """141 item = self._matching(frame)142 return () if item is None else item[0]143144 def _canonical_key(self, field):145 """借用当前投影已编码的字段键，不按原帧增加缓存。146147 参数: field为当前匹配出的已声明LegacyCANField。148 返回: 初始化时验证并生成的规范JSON字符串。149 异常: 非本投影字段时KeyError，不猜测或临时加入新的绑定。150 """151 return self._canonical_keys[field]152153 def build(self, frame, sequence, values, time_ns, *, canonical_keys=False):154 """只拣选计算字段，复用已有有时间/无时间更新及其严格值检查。155156 参数: frame为原帧，sequence为本来源原序号，values为已解码映射，157 time_ns为实际映射时间或明确未采集的None；canonical_keys说明文本字段键。158 返回: LegacyDecodedUpdate/LegacyUntimedUpdate；非目标数据帧返回None。159 异常: 类型/数值/身份错误原样传播，不缓存前条值补齐缺失。160 """161 if type(canonical_keys) is not bool:162 raise TypeError('canonical_keys需要bool')163 item = self._matching(frame)164 if item is None:165 return None166 fields, reference = item167 selected = {field.signal: values.get(self._canonical_keys[field] if canonical_keys else field.signal)168 for field in fields if field in self._value_fields}169 update_type = LegacyUntimedUpdate if time_ns is None else LegacyDecodedUpdate170 return update_type(time_ns, self.source, frame.channel, frame.arbitration_id,171 frame.is_extended_id, frame.is_fd, frame.is_remote_frame,172 frame.is_error_frame, reference, sequence, sequence, selected)173