chronaxis.core.sequence_reference
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""按时间组将请求与响应FIFO配对，区分未响应和无法确定。23输入为同一组完整有序的SequenceTick，输出为每次push新确定的SequenceResult。4检测器保存尚未决定的请求；返回的结果不再在内部累积。候选数有明确上限，5任何失败均封闭实例，调用者不能把部分结果宣告为完整任务成功。6响应可判断性来自既有ConditionTick的连续/点视图，TTL等边界必须由上游提供。7状态循环只依赖标准库，不冒称NumPy数组优化，不读取来源或写盘。8"""910# FIFO候选入队出队为常数时间；身份使用稳定内容，避免批次影响结果ID。11from collections import deque12from hashlib import sha25613from .binding import _identity14from .models import Interval15from .sequence import SequenceResult, SequenceRule, SequenceTick16from .temporal import ConditionTimeline, Event17from .temporal_reference import require_batch18from .types import QualityFlag as Q, Truth19from ..timebase.conversion import checked_add_ns2021# 固定质量位前缀计数避免每次输入都遍历全部待响应请求。22_QUALITY_BITS = tuple(Q)232425class SequenceDetector:26 """维护一个dataset/session/group/rule的等待队列与观测状态。2728候选按A时间和同刻稳定ID排序。B只消费最早的合格A；截止点先匹配再超时。29输入跨批连续、批尾不当EOF；每次push立即转交已决定结果。max_pending限制30同时存在的请求数量，不是整进程RAM保证，公共AnalysisRunner的SequenceRows另行计费。31 """3233 def __init__(self, rule: SequenceRule, start_ns: int, end_ns: int, *, max_pending: int = 65536):34 """建立独立状态，不从别的组或会话复制候选。3536 参数: rule；start_ns/end_ns为明确会话范围；max_pending为正原生整数。37 返回: 新检测器，尚未消费任何时间组。38 异常: TypeError/ValueError/OverflowError为规则、范围或容量错误。39 """40 # 会话范围和容量在任何状态分配前校验。41 if not isinstance(rule, SequenceRule) or type(max_pending) is not int or max_pending <= 0:42 raise TypeError('需要SequenceRule及正int max_pending')43 domain = Interval(start_ns, end_ns)44 self.rule, self.start_ns, self.end_ns = rule, domain.start_ns, domain.end_ns45 self.max_pending = max_pending46 self._pending = deque()47 self._counts = [0] * len(_QUALITY_BITS)48 self._quality = Q.NONE49 self._cursor = self.start_ns50 self._last_time = None51 self._record_elapsed = 052 self._ended = self._finished = self._failed = False5354 @property55 def pending_count(self):56 """返回尚未决定的请求数量，不暴露可修改队列。5758 参数: 无。59 返回: 非负整数；已结束或失败时为0。60 异常: 无主动异常。61 """62 # 不为查看数量复制全部事件。63 return len(self._pending)6465 def _note(self, quality):66 """累计本段未知原因，供各请求按自己的起点查询。6768 参数: quality为已校验的质量位。69 返回: None，更新固定长度计数器。70 异常: 无主动异常；只由内部协议调用。71 """72 # 每段最多检查固定9位，不按候选数做重复状态更新。73 for index, bit in enumerate(_QUALITY_BITS):74 if quality & bit:75 self._counts[index] += 17677 def _advance(self, time_ns):78 """结算上个时间组至指定时刻的连续观测，不推断额外响应。7980 参数: time_ns为不早于内部游标的时间。81 返回: None；正长度未知区间计入质量，零长度不增加区间证据。82 异常: 无主动异常，外层已验证时间顺序。83 """84 # 同刻处理不伪造时长；此段状态由上个完整输入组确定。85 if time_ns > self._cursor and self._quality != Q.NONE:86 self._note(self._quality)87 self._cursor = time_ns8889 def _result(self, pending, time_ns, response=None, *, early_end=False):90 """把一个请求的最终事实转换为稳定结果，保留原事件引用。9192 参数: pending含请求/质量快照/deadline；time_ns为决定时刻；response可空；early_end表示提前EOF。93 返回: SequenceResult，匹配或有质量依据的未匹配结果。94 异常: 模型校验或内存异常传播，不输出伪成功。95 """96 # 只累积请求之后出现的未知位，之前的缺失不能污染新请求。97 request, snapshot, _deadline = pending98 quality = request.dependency_quality & Q.ESTIMATED99 if response is not None:100 quality |= response.dependency_quality & Q.ESTIMATED101 for bit, before, after in zip(_QUALITY_BITS, snapshot, self._counts):102 if after > before:103 quality |= bit104 if early_end:105 quality |= Q.MISSING106 status = 'matched' if response is not None else 'indeterminate' if quality != Q.NONE else 'no_response'107 delay = None if response is None else (self._record_elapsed - _deadline + self.rule.max_delay_ns108 if self.rule.record_clock else time_ns-int(request.confirmed_ns))109 # 身份包括组、规则参数及原事件；批大小/候选容量不影响业务身份。110 identity = (self.rule.dataset_id, self.rule.session_id, self.rule.group_id, self.rule.rule_id,111 self.rule.start_rule_id, self.rule.response_rule_id, self.rule.min_delay_ns,112 self.rule.max_delay_ns, request.event_id, int(request.confirmed_ns),113 None if response is None else response.event_id, time_ns, status)114 if self.rule.record_clock:115 identity += ('record_clock', delay)116 result_id = sha256(_identity(identity).encode('utf-8')).hexdigest()117 return SequenceResult(result_id, self.rule, request, response, status, time_ns, delay, quality)118119 def _consume(self, tick, output, advance=None):120 """按先前截止点、同刻配对、同刻超时、EOF的顺序处理一组。121122 参数: tick为已检查的SequenceTick，output是本批局部输出列表。123 返回: None；更新候选和游标，不存储已经转交的结果。124 异常: MemoryError为候选上限；结果构造失败传播。125 """126 # 上游允许NumPy整数；这里规范容器，避免后续运算/身份编码继承数组标量。127 time = int(tick.condition.time_ns)128 if self.rule.record_clock:129 self._record_elapsed += advance130 self._advance(time)131 clock = self._record_elapsed if self.rule.record_clock else time132 # 无新消息也在精确deadline结算；严格小于当前时间的候选已无匹配机会。133 while self._pending and self._pending[0][2] < clock:134 deadline = self._pending[0][2]135 if not self.rule.record_clock:136 self._advance(deadline)137 output.append(self._result(self._pending.popleft(), time if self.rule.record_clock else deadline))138 self._advance(time)139 # 同刻A必须先收齐，B的输入顺序不能影响零延迟匹配。140 if len(self._pending) + len(tick.starts) > self.max_pending:141 raise MemoryError('sequence待响应请求超过max_pending')142 for event in sorted(tick.starts, key=lambda item: item.event_id):143 self._pending.append((event, tuple(self._counts), clock + self.rule.max_delay_ns144 if self.rule.record_clock else checked_add_ns(time, self.rule.max_delay_ns)))145 # 独立点UNKNOWN也不能被当作完整观察；terminal人工连续UNKNOWN不计入。146 if tick.condition.point_state is Truth.UNKNOWN:147 self._note(tick.condition.point_quality)148 for response in sorted(tick.responses, key=lambda item: item.event_id):149 # 所有请求使用同一min_delay；最老的还太新时，其余也不能匹配。150 if self._pending and (clock - self._pending[0][2] + self.rule.max_delay_ns151 if self.rule.record_clock else time-int(self._pending[0][0].confirmed_ns)) >= self.rule.min_delay_ns:152 output.append(self._result(self._pending.popleft(), time, response))153 # 只有同刻合格B都使用过后，才决定截止点仍剩余的请求。154 while self._pending and self._pending[0][2] == clock:155 output.append(self._result(self._pending.popleft(), time))156 if tick.condition.terminal:157 # 观察先结束不能证明设备未响应；不把deadline裁短成正常超时。158 while self._pending:159 output.append(self._result(self._pending.popleft(), time, early_end=True))160 self._ended = True161 self._quality = tick.condition.quality | (tick.condition.diagnostic_quality & Q.ESTIMATED)162 self._last_time = time163164 def push(self, ticks: tuple[SequenceTick, ...], *, advances=None) -> tuple[SequenceResult, ...]:165 """消费一批完整时间组并仅返回本批确定的结果。166167 参数: ticks为有序tuple，可为空，首末与声明会话范围一致；advances仅规则168 record_clock模式使用，保存一个必要累计计数而不存逐点时间对照。169 返回: 新SequenceResult tuple；调用者负责消费，不在实例中重复保留。170 异常: TypeError/ValueError/OverflowError为协议错误；MemoryError为容量不足。171 副作用: 失败后释放候选并封闭实例，已转交的早先批次不能视为完整成功。172 """173 # 失败或finish后的实例不允许继续使用，避免部分状态拼成另一个任务。174 if self._failed or self._finished:175 raise ValueError('sequence实例已失败或finish')176 try:177 if not isinstance(ticks, tuple) or any(not isinstance(tick, SequenceTick) for tick in ticks):178 raise TypeError('push需要SequenceTick tuple')179 require_batch(tuple(tick.condition for tick in ticks), self.start_ns, self.end_ns, self._last_time, self._ended)180 from .temporal_clock import require_clock_advances181 require_clock_advances(tuple(t.condition for t in ticks), advances, self.rule.record_clock, self._last_time)182 # 整批先检查归属及deadline溢出，再推进；错误事件不能污染另一组。183 for tick in ticks:184 for event in tick.starts:185 self.rule.check_event(event, response=False)186 if not self.rule.record_clock:187 checked_add_ns(event.confirmed_ns, self.rule.max_delay_ns)188 for event in tick.responses:189 self.rule.check_event(event, response=True)190 output = []191 for index, tick in enumerate(ticks):192 self._consume(tick, output, None if advances is None else advances[index])193 return tuple(output)194 except Exception:195 # 超限/坏输入均使任务失败，不能跳过该批继续生产正常结果。196 self._failed = True197 self._pending.clear()198 raise199200 def finish(self):201 """确认已经收到真正EOF，不制造额外响应或超时。202203 参数: 无。204 返回: None；结果已经由各次push转交。205 异常: ValueError为缺EOF、失败或重复finish。206 """207 # 与批尾区分，只有terminal之后可以声明消费完成。208 if self._failed or self._finished or not self._ended:209 raise ValueError('finish需要完整terminal且实例未关闭')210 self._finished = True211212213def pair_sequence(rule: SequenceRule, starts: tuple[Event, ...], responses: tuple[Event, ...],214 timeline: ConditionTimeline, *, batch_rows: int = 65536, max_pending: int = 65536):215 """将既有事件与完整响应条件轴接到顺序核心，适合可控内存分析。216217 参数: rule；starts/responses为已确认事件tuple；timeline含全部条件变化、TTL及EOF；218 batch_rows为每批完整时间组数，max_pending限制等待队列。219 返回: SequenceResult tuple；只为此内存便利入口收集全部输出。220 异常: TypeError/ValueError为类型/归属/缺失事件时刻；容量与时间溢出传播。221 边界: 不补采样、解析文件或推断同步；大数据上层直接分批使用SequenceDetector。222 """223 # 完整轴定义了可判断区间，来源范围不允许继承其他会话。224 if not isinstance(timeline, ConditionTimeline) or type(batch_rows) is not int or batch_rows <= 0:225 raise TypeError('需要ConditionTimeline及正int batch_rows')226 detector = SequenceDetector(rule, timeline.start_ns, timeline.end_ns, max_pending=max_pending)227 if any(ref.session_id != rule.session_id for ref in timeline.source_refs):228 raise ValueError('响应观测来源不属于当前会话')229 groups = ({}, {})230 for events, group, response in ((starts, groups[0], False), (responses, groups[1], True)):231 if not isinstance(events, tuple):232 raise TypeError('完整事件输入需要tuple')233 seen = set()234 for event in events:235 rule.check_event(event, response=response)236 if event.event_id in seen:237 raise ValueError('同一角色内事件ID重复')238 seen.add(event.event_id)239 group.setdefault(event.confirmed_ns, []).append(event)240 # 事件时刻必须有明确条件点证据，不由本入口伪造TTL、点状态或EOF。241 times = {tick.time_ns for tick in timeline.ticks}242 if any(time not in times for group in groups for time in group):243 raise ValueError('事件确认时刻不在响应条件时间轴中')244 results = []245 for offset in range(0, len(timeline.ticks), batch_rows):246 ticks = tuple(SequenceTick(tick, tuple(groups[0].get(tick.time_ns, ())), tuple(groups[1].get(tick.time_ns, ())))247 for tick in timeline.ticks[offset:offset+batch_rows])248 results.extend(detector.push(ticks))249 detector.finish()250 return tuple(results)251