chronaxis.results.frame_patch
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用已有DBC位域工具更新明确的连续信号，其他位沿用同一基础帧。23调用者传入本次已加载的cantools Message及本次已解码的物理值，不再次读取DBC4或解码原帧。整数按nearest_even-v1量化，浮点沿DBC编码后核实际表示误差。5码值或物理值越界立即拒绝。只更新选中位的掩码，计数、校验及未知位保持；6这些生成帧不承诺校验码有效或可用于车辆回放。7"""89from dataclasses import dataclass, replace10from fractions import Fraction11from math import isfinite1213from .budget import _reserve_owned_ram, _release_owned_ram, _cleanup_owner14from ..legacy.records import normalize_frame_id151617@dataclass(frozen=True)18class FrameRebuildRule:19 """明确报文身份、连续字段及禁止插值的字段；声明不能从名字猜业务含义。"""2021 channel: int22 frame_id: int23 continuous: tuple[str,...]24 is_extended: bool | None = None # 省略时按实际帧与DBC确定，不等同标准帧。25 is_fd: bool | None = None # 省略时保留输入帧的实际CAN/FD类型。26 protected: tuple[str,...] = ()27 same_time_policy: str = 'error'2829 def __post_init__(self):30 """核身份和字段分类。参数: 无。返回: None。异常: 非法/重复/保护字段冲突。31 """32 # ID只在规则建立时归一化，避免每个输出帧重新解释进制。33 object.__setattr__(self,'frame_id',normalize_frame_id(self.frame_id))34 if (type(self.channel) is not int or self.channel<035 or self.is_extended is False and self.frame_id>0x7ff36 or self.is_extended is not None and type(self.is_extended) is not bool37 or self.is_fd is not None and type(self.is_fd) is not bool38 or self.same_time_policy not in ('error','last_by_sequence')):39 raise ValueError('报文重建身份或同刻策略非法')40 for names in (self.continuous,self.protected):41 if type(names) is not tuple or len(names)>4096 or any(type(n) is not str or not n for n in names) or len(set(names))!=len(names):42 raise ValueError('报文重建字段需要唯一非空名称tuple')43 if not self.continuous or set(self.continuous)&set(self.protected):44 raise ValueError('必须选择连续量，且不能插值保护的计数或校验字段')4546 def matches(self,frame):47 """按实际帧身份匹配。参数: frame原帧。返回: bool。异常: 原模型访问错。48 """49 return (frame.channel==self.channel and frame.arbitration_id==self.frame_id50 and (self.is_extended is None or frame.is_extended_id==self.is_extended)51 and (self.is_fd is None or frame.is_fd==self.is_fd))525354class DBCFramePatch:55 """借用一条已加载DBC定义，拥有选中位格式与预算；只缓存当前复用布局。"""5657 def __init__(self,rule,message,budget):58 """核DBC身份与连续字段并预约格式空间。参数: rule/message/budget。返回: patcher。异常: 不匹配或预算错。59 """60 if type(rule) is not FrameRebuildRule:61 raise TypeError('需要FrameRebuildRule')62 self.rule=rule; self.message=message; self._budget=budget63 self._held_cost=self._fixed_cost=0; self._closed=False64 self.signals={}; self.formats=None; self.active=(); self.mask=065 self._identity=None66 try:67 _reserve_owned_ram(self,262144+len(message.signals)*16384,'_fixed_cost')68 if (message.frame_id!=rule.frame_id69 or rule.is_extended is not None and message.is_extended_frame!=rule.is_extended70 or rule.is_fd is True and not message.is_fd71 or not 0<message.length<=64 or message.is_container):72 raise ValueError('重建规则与DBC报文身份/载荷不符或为容器报文')73 self.signals={signal.name:signal for signal in message.signals}74 for name in (*rule.continuous,*rule.protected):75 if name not in self.signals:76 raise ValueError('DBC缺少重建声明字段: '+name)77 for name in rule.continuous:78 signal=self.signals[name]79 if signal.is_multiplexer or signal.choices or not signal.scale or signal.length==1:80 raise ValueError('复用选择器、枚举/布尔或零倍率不能当连续量: '+name)81 except BaseException as error:82 _cleanup_owner(self,error,'frame_patch_cleanup_owner','DBC位域重建构造')83 raise8485 def accept_frame(self,frame):86 """检查实际基础帧与选定DBC，并识别自动规则后续出现的帧型冲突。8788 参数: frame为同次输入原帧。返回: None。异常: ValueError为类型不匹配或歧义。89 副作用: 只记录首次匹配的两个类型标志，后续同型帧不扩大状态。90 """91 if (not self.rule.matches(frame) or frame.is_extended_id!=self.message.is_extended_frame92 or frame.is_fd and not self.message.is_fd):93 raise ValueError('重建实际帧与已选DBC类型不符')94 identity=(frame.is_extended_id,frame.is_fd)95 if self._identity is not None and self._identity!=identity:96 flag='is_extended' if self._identity[0]!=identity[0] else 'is_fd'97 raise ValueError('重建规则匹配多种实际帧类型，请补充'+flag)98 self._identity=identity99100 def selected(self,layout):101 """只取当前复用分支适用的连续字段。参数: layout原生选择器码值tuple。返回: 名称tuple。异常: 布局非法。102 """103 selectors=dict(layout)104 if len(selectors)!=len(layout) or any(name not in self.signals or not self.signals[name].is_multiplexer105 or type(value) is not int for name,value in layout):106 raise ValueError('报文复用布局不合法')107 selected=[]108 for name in self.rule.continuous:109 signal=self.signals[name]; visited=set(); applicable=True110 # 沿真实父选择器判断，避免将另一分支同位置的位当作本连续量。111 while signal.multiplexer_signal is not None:112 parent=signal.multiplexer_signal113 if parent in visited or parent not in self.signals:114 raise ValueError('DBC复用父链非法')115 visited.add(parent)116 if parent not in selectors or selectors[parent] not in signal.multiplexer_ids:117 applicable=False; break118 signal=self.signals[parent]119 if applicable:120 selected.append(name)121 return tuple(selected)122123 def patch(self,base,values,layout,*,exact_values=None,trace=True):124 """仅按允许掩码更新载荷并记录量化误差。125126 参数: base同一基础LegacyRawFrame；values所需物理估计dict；layout实际复用布局；127 exact_values可选的有理数编码目标，用于避免中间float改变半码决策。128 返回: (新帧, 每字段目标/码值/实际值/误差tuple)。异常: 缺值、范围/格式/编码错误。129 """130 if self._closed or not self.rule.matches(base) or base.dlc!=self.message.length or len(base.data)!=self.message.length:131 raise ValueError('重建基础帧已关闭、身份或长度不符')132 self.accept_frame(base)133 selected=self.selected(layout)134 if not selected or set(values)!=set(selected):135 raise ValueError('当前复用布局缺少必须的连续值')136 if exact_values is not None and (set(exact_values)!=set(selected) or any(type(v) is not Fraction for v in exact_values.values())):137 raise ValueError('精确量化目标需要同组Fraction字段')138 from cantools.database.utils import create_encode_decode_formats,encode_data,decode_data139 if selected!=self.active:140 # 编解码格式来自现有库；仅所选信号的非padding位允许改变。141 self.formats=create_encode_decode_formats([self.signals[n] for n in selected],self.message.length)142 self.mask=((1<<(self.message.length*8))-1)^self.formats.padding_mask143 self.active=selected144 raw={}145 for name in selected:146 signal=self.signals[name]; value=values[name]147 if type(value) not in (int,float) or not isfinite(value):148 raise ValueError('重建物理值必须有限: '+name)149 physical=Fraction(str(value)) if exact_values is None else exact_values[name]150 scale=Fraction(str(signal.scale)); offset=Fraction(str(signal.offset))151 if ((signal.minimum is not None and physical<Fraction(str(signal.minimum)))152 or (signal.maximum is not None and physical>Fraction(str(signal.maximum)))):153 raise ValueError('重建目标物理值越界: '+name)154 code=(physical-offset)/scale155 if signal.is_float:156 raw[name]=float(code)157 else:158 # Fraction的round精确实现最近整数/恰半取偶，负倍率和负值同样适用。159 integer=round(code)160 low=-(1<<(signal.length-1)) if signal.is_signed else 0161 high=(1<<(signal.length-int(signal.is_signed)))-1162 if not low<=integer<=high:163 raise ValueError('重建码值超出位宽: '+name)164 raw[name]=integer165 fields=[self.signals[n] for n in selected]166 encoded=encode_data(raw,fields,self.formats,scaling=False)167 payload=((int.from_bytes(base.data,'big')&~self.mask)|(encoded&self.mask)).to_bytes(self.message.length,'big')168 decoded=decode_data(payload,self.message.length,fields,self.formats,False,True,False,False)169 facts=[]170 for name in selected:171 signal=self.signals[name]; actual=decoded[name]172 if (not isfinite(actual) or signal.minimum is not None and actual<signal.minimum173 or signal.maximum is not None and actual>signal.maximum):174 raise ValueError('重建量化后物理值越界: '+name)175 if trace:176 target=Fraction(str(values[name])) if exact_values is None else exact_values[name]177 facts.append((name,float(target),raw[name],actual,float(Fraction(str(actual))-target)))178 return replace(base,data=payload),tuple(facts)179180 def close(self):181 """释放本格式缓存，不关闭借用DBC。参数: 无。返回: None。异常: 费用释放错可重试。182 """183 if self._closed:184 return185 self.signals.clear(); self.formats=None; self.message=None; self.active=()186 _release_owned_ram(self,self._fixed_cost,'_fixed_cost'); self._closed=True187