chronaxis.core.declarations
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在读取数据之前检查来源、信号和派生名称声明，生成不可变的声明绑定。23输入为来源定位、时基、请求范围、信号定义和同步证据；输出DeclarationBindings4及其SHA-256身份。函数只检查配置关系，不证明文件、DBC字段或实际观测存在，5也不创建虚拟样本。静态表达式检查可使用这些声明；实际解释器仍需包含内存数据6的SignalBindings。本模块只依赖标准库和核心模型，没有文件操作、样本计算、7全局注册或执行计划缓存。8"""910# 冻结声明并为规范身份计算稳定摘要。11from dataclasses import dataclass, field12from hashlib import sha25613# 复用已验契约与同步证据，不放宽实际绑定校验。14from .binding import SignalDefinition, SynchronizationEvidence, _identity15from .unit_conversion import binding_version16from .unit_registry import REGISTRY_VERSION17from .lookup import LookupTablePlan, validate_lookups18# 时基和范围已经独立验证，不借空SessionSpec制造实际来源。19from .models import Interval, TimebaseSpec20from .types import TimeType21from .validation import require_identifier, require_text22from ..timebase.conversion import require_int64_ns232425@dataclass(frozen=True)26class DeclaredSource:27 """保存一个尚未读取的来源定位，以及其声明的固定时间映射。2829 input_identity包含规范配置中的路径顺序、格式和处理配置标识，不是文件内容30 哈希。offset_ns是整数纳秒偏移；模型不会估计漂移。字段实际存在性由来源执行31 接口另行检查，不能凭声明认定数据已读取。全部字段不可变。32 """3334 name: str # 当前计划来源名。35 timebase: TimebaseSpec # 原始来源坐标。36 input_identity: str # 规范声明身份，与实际数据身份分别保存。37 offset_ns: int = 0 # 显式固定偏移，不自动估计漂移。3839 def __post_init__(self) -> None:40 """校验声明结构，保持声明身份与实际内容证据保持分开。4142 参数: 无；读取name/timebase/input_identity/offset_ns。43 返回: None，冻结声明。44 异常: TypeError/ValueError/TimeOverflowError为配置错误。45 """46 # 所有名称遵守统一安全规则。47 require_identifier(self.name, 'source.name')48 # 规范身份必须明确提供，不凭文件名猜测。49 require_text(self.input_identity, 'source.input_identity')50 # offset必须保持定宽整数。51 require_int64_ns(self.offset_ns)52 # 时基不得使用未校验字典。53 if not isinstance(self.timebase, TimebaseSpec):54 # 静态接口也不接无类型占位。55 raise TypeError('声明来源需要TimebaseSpec')565758@dataclass(frozen=True)59class DeclarationBindings:60 """冻结一个计划的来源、信号名称和同步关系，供执行前检查使用。6162 signals与derived_names共用名称空间。identity只证明配置声明；来源路径、63 处理配置、转换或有效期变化会改变它。模型不验证文件内容、实际列或DBC64 元数据，不能代替执行数据绑定。同步检查只针对当前节点实际使用的来源。65 """6667 session_id: str # 当前会话声明。68 timebase: TimebaseSpec # 计划分析坐标。69 domain: Interval | None # 用户请求范围；None声明整文件无时间筛选，不表示无观测。70 sources: tuple[DeclaredSource, ...] # 保留声明原顺序。71 signals: tuple[SignalDefinition, ...] # 已校验单信号契约。72 derived_names: tuple[str, ...] = () # signals统一名称空间。73 synchronization: tuple[SynchronizationEvidence, ...] = () # 显式核实依据。74 config_version: int | None = field(default=None, kw_only=True) # 缺省继承信号；空声明可指定版本。75 extensions: object = field(default=None, kw_only=True, repr=False) # 仅接收宿主传入的不可变注册。76 lookups: tuple[LookupTablePlan, ...] = field(default=(), kw_only=True) # v4命名表，独立于信号来源。77 temporal_inputs: tuple = field(default=(), kw_only=True) # 离散序列对应的原采集时基。78 identity: str = field(init=False) # 配置身份，不是执行绑定SHA。7980 def __post_init__(self) -> None:81 """检查名称、来源和显式同步关系，并生成声明身份。8283 参数: 无；输入均为已经校验的不可变声明模型。84 返回: None；不读取任何样本、文件或DBC。85 异常: TypeError/ValueError为结构错误、名称冲突、来源错误或同步关系冲突。86 """87 # 会话名采用同一标识符规则。88 require_identifier(self.session_id, 'dataset.session')89 # 容器不得在构造后被调用者改写。90 if (not isinstance(self.timebase, TimebaseSpec)91 or self.domain is not None and not isinstance(self.domain, Interval)92 or any(type(items) is not tuple for items in (self.sources, self.signals, self.derived_names, self.synchronization))):93 # 不冻结任意迭代器或未知对象。94 raise TypeError('声明绑定需要typed时基/范围与tuple容器')95 # 单项必须由对应模型校验，不使用duck typing。96 if any(not isinstance(item, DeclaredSource) for item in self.sources) or any(not isinstance(item, SignalDefinition) for item in self.signals):97 # 静态编译也不接受伪造类名。98 raise TypeError('声明绑定需要DeclaredSource/SignalDefinition')99 # 定义版本必须一致；没有信号时仍允许明确指定新版本。100 object.__setattr__(self, "config_version", binding_version(self.signals, self.config_version))101 validate_lookups(self.lookups, self.config_version)102 if self.extensions is not None:103 from ..extensions import ExtensionRegistry104 if type(self.extensions) is not ExtensionRegistry or self.config_version != 4:105 raise ValueError('extensions需要版本4及宿主ExtensionRegistry')106 # 来源表只属于当前计划。107 sources = {item.name: item for item in self.sources}108 # 重复来源不能字典覆盖。109 if len(sources) != len(self.sources):110 # 没有自动选取第一个来源的回退。111 raise ValueError('source.name重复')112 # 所有原信号和派生名称必须唯一。113 names = set()114 # 原字段只检查已声明来源，不声称实际存在。115 for definition in self.signals:116 # 完整单信号契约已经校验。117 if definition.source not in sources or definition.name in names:118 # 定位声明中实际冲突的信号。119 raise ValueError(f'{definition.name}来源不存在或别名重复')120 # 防止派生与当前信号重名。121 names.add(definition.name)122 # 派生名称不以点号或函数名绕过白名单。123 for name in self.derived_names:124 # 与实际执行绑定使用同一规则。125 require_identifier(name, 'derived.name')126 # 声明整个名称空间，未引用也不能隐藏冲突。127 if name in names:128 # 保留用户名称，不加后缀。129 raise ValueError(f'signals/derived名称冲突：{name}')130 # 登记下一个冲突校验点。131 names.add(name)132 # 同步声明必须唯一且指向当前计划。133 evidenced = set()134 # 不从机器时钟或两文件首点为0推断同步。135 for evidence in self.synchronization:136 # 正确类型和目标是声明有效性的必要条件。137 if not isinstance(evidence, SynchronizationEvidence):138 # 不以文字字符串冒充typed同步说明。139 raise TypeError('同步需要SynchronizationEvidence')140 # 重复/未知来源及不同分析时基都拒绝。141 if evidence.source not in sources or evidence.source in evidenced or evidence.analysis_timebase != self.timebase:142 # 不静默选一个映射。143 raise ValueError('同步来源重复/未知或目标时基不一致')144 # 保留已核实来源声明集合。145 evidenced.add(evidence.source)146 # 静态身份显式使用独立版本前缀，不与真实绑定共用证明。147 # 已有显式Interval仍使用原v1字节；无筛选声明以独立v2前缀和None端点表示。148 version = 'declaration-v1' if self.domain is not None else 'declaration-v2-no-time-request'149 payload = (version, self.session_id, self.timebase.timebase_id, self.timebase.time_type.value, self.timebase.origin,150 None if self.domain is None else self.domain.start_ns,151 None if self.domain is None else self.domain.end_ns,152 tuple((item.name, item.timebase.timebase_id, item.timebase.time_type.value, item.timebase.origin, item.offset_ns, item.input_identity) for item in self.sources),153 tuple(item.identity() for item in self.signals), self.derived_names,154 tuple((item.source, item.evidence) for item in self.synchronization))155 # 当前规则版本和单位目录均属于身份。156 payload += (self.config_version, REGISTRY_VERSION)157 if self.extensions is not None:158 payload += ('host-extensions-v1', self.extensions.identity)159 if self.lookups:160 payload += ('lookup-table-v1', tuple(t.identity for t in self.lookups))161 if self.temporal_inputs:162 payload += ('temporal-inputs-v1', self.temporal_inputs)163 # 所有身份材料是已校验结构，不调用外部对象repr。164 object.__setattr__(self, 'identity', sha256(_identity(payload).encode('utf-8')).hexdigest())165166 def resolve(self, name: str) -> SignalDefinition:167 """定位当前声明中的唯一信号别名。168169 参数: name简单别名。170 返回: 当前计划SignalDefinition。171 异常: ValueError为名称非法或未知。172 """173 # 不把semantic点号转换成别名。174 require_identifier(name, 'signal alias')175 # 只查当前声明集合。176 for definition in self.signals:177 # 保留大小写敏感身份。178 if definition.name == name:179 # 不继承上个任务的输入。180 return definition181 # 未知信号不能通过空数据隐式创建。182 raise ValueError(f'未知信号别名：{name}')183184 def require_synchronized(self, names: tuple[str, ...]) -> None:185 """按实际联合依赖检查声明的公共时间基准。186187 参数: names为原信号简单别名tuple。188 返回: None，单来源或公共起点/显式同步依据齐备时通过。189 异常: TypeError/ValueError为容器、别名或同步缺项。190 """191 # 与实际绑定保持相同的依赖接口。192 if type(names) is not tuple:193 # 不接受可变或无限依赖列表。194 raise TypeError('联合依赖names必须为tuple')195 # 单来源多个信号只产生一个同步成员。196 if self.temporal_inputs:197 inputs = dict(self.temporal_inputs)198 names = tuple(alias for name in names for alias in inputs.get(name,(name,)))199 used = {self.resolve(name).source for name in names}200 # 独立来源可在自己坐标中校验。201 if len(used) <= 1:202 # 未使用来源不参与当前声明的证据检查。203 return204 # 构造已校验所有证据的来源/目标关系。205 evidenced = {item.source for item in self.synchronization}206 # 联合计算只检查实际参与来源。207 for source in self.sources:208 # 未参与或已有显式固定映射的来源无需重复声明公共起点。209 if source.name not in used or source.name in evidenced:210 # 不自动估计其他来源的漂移。211 continue212 # absolute坐标已统一为UTC；relative坐标须有相同的完整起点声明。213 same = source.timebase.time_type is self.timebase.time_type and (source.timebase.time_type is TimeType.ABSOLUTE or source.timebase == self.timebase)214 # 数值offset本身不能证明同步。215 if not same:216 # 提供具体缺项位置供配置修正。217 raise ValueError(f'联合依赖缺少同步依据：{source.name}；需要公共起点或固定offset核实证据')218