chronaxis.execution.record_value_history
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""只为需要回看范围的消费者保存计算值，一份实际时间和必要有效截止。23同一表达式/条件/驱动可共享此文件。最后一点留在内存，便于非驱动失效更新4它的截止；其余记录定长保存，可二分定位而无需全量核验或JSON重建。5"""6from struct import Struct7from tempfile import TemporaryFile8from ..core.numeric import ScalarValue9from ..core.types import SignalDType as D,QualityFlag as Q10from ..results.errors import ResultClosedError11from ..results.spool import _reserve_owned_ram,_release_owned_ram,_cleanup_owner121314class RecordValueHistory:15 """拥有单个必要临时文件和最后一个计算点；没有原帧、原文或第二时间轴。"""1617 def __init__(self,budget,path,dtype,conditioned=False,*,sample_updates=False,labels=()):18 """预付有限读写状态，不立即建文件。1920 参数: 预算/专用路径/值类型/条件开关；sample_updates按需存真实更新标志，labels共享枚举域。21 返回: owner。异常: 原构造错。22 """23 self._budget,self._path,self._dtype,self._conditioned=budget,path,dtype,conditioned24 self._sample_updates,self._labels=sample_updates,labels25 self._held_cost=self._fixed_cost=026 self._file=self._last=None27 self._count=028 self._closed=self._failed=False29 try:30 _reserve_owned_ram(self,65536,'_fixed_cost')31 if type(sample_updates) is not bool:raise TypeError('样本标志开关需要bool')32 if dtype is D.ENUM:33 if type(labels) is not tuple or not labels:raise ValueError('枚举历史需要共享标签域')34 _reserve_owned_ram(self,4096+sum(1024+64*len(label) for label in labels),'_fixed_cost')35 code={D.FLOAT64:'d',D.INT64:'q',D.UINT64:'Q',D.BOOL:'?',D.ENUM:'I'}[dtype]36 self._codec=Struct('>q'+code+'?Iq?'+('??Iq?' if conditioned else '')+('?' if sample_updates else ''))37 except BaseException as error:38 _cleanup_owner(self,error,'record_value_history_cleanup_owner','必要计算历史构造')39 raise4041 @property42 def count(self):43 """含内存末点的记录数。参数: 无。返回: int。异常: 无。"""44 return self._count+(self._last is not None)4546 @property47 def last_time(self):48 """最后真实驱动时刻。参数: 无。返回: int或None。异常: 无。"""49 return None if self._last is None else self._last[0]5051 def _open(self):52 """拒绝失败/关闭后使用。参数: 无。返回: None。异常: ResultClosedError。"""53 if self._closed or self._failed:54 raise ResultClosedError('必要计算历史已关闭或失败')5556 def append(self,row,condition=None,*,actual_update=False):57 """写前一点并留下新点；输入是本执行器已验证的求值结果。5859 参数: row(time,ScalarValue,deadline)，condition为同样三元组或None；actual_update标识真实值更新。60 返回: None。异常: 顺序/短写/预算错误，失败后只能清理。61 """62 self._open()63 try:64 if type(actual_update) is not bool:raise TypeError('真实更新标志需要bool')65 time,value,deadline=row66 if self._last is not None and time<=self._last[0]:67 raise ValueError('必要计算历史时间须递增')68 if self._last is not None:69 if self._file is None:70 self._file=TemporaryFile(mode='w+b',dir=self._path,prefix='calculation-values-')71 t,v,d,c,cd=self._last[:5]72 number=self._labels.index(v.value) if self._dtype is D.ENUM and v.valid else v.value if v.valid else 073 # 本执行私有暂存中的高位携带部分状态，永不混入 ScalarValue 的有效性质量。74 fields=(t,number,v.valid,int(v.quality) | (int(v.partial)<<31),d or 0,d is not None)75 if self._conditioned:76 fields+=(bool(c.value) if c.valid else False,c.valid,int(c.quality) | (int(c.partial)<<31),cd or 0,cd is not None)77 if self._sample_updates:fields+=(self._last[5],)78 payload=self._codec.pack(*fields)79 self._budget.reserve_spool(len(payload))80 self._file.seek(0,2)81 if self._file.write(payload)!=len(payload):82 raise OSError('必要计算历史短写')83 self._count+=184 self._last=(time,value,deadline,None if condition is None else condition[1],85 None if condition is None else condition[2])+((actual_update,) if self._sample_updates else ())86 except BaseException:87 self._failed=True88 raise8990 def expire_at(self,time_ns,*,condition=False):91 """非驱动失效只裁最后点，不添加假记录。参数: 实际截止/条件标识。返回: None。异常: 时间回退。"""92 self._open()93 if self._last is None:94 return95 if time_ns<self._last[0]:96 raise ValueError('必要计算历史截止回退')97 index=4 if condition else 298 old=self._last[index]99 self._last=self._last[:index]+(time_ns if old is None else min(old,time_ns),)+self._last[index+1:]100101 def at(self,index):102 """读取一个实际记录，不全量扫描。参数: 已登记索引。返回: 点及条件五元组。异常: 越界/短读。"""103 self._open()104 if not 0<=index<self.count:105 raise IndexError('必要计算历史索引越界')106 if index==self._count:107 return self._last108 self._file.seek(index*self._codec.size)109 payload=self._file.read(self._codec.size)110 if len(payload)!=self._codec.size:111 raise ValueError('必要计算历史短读')112 fields=self._codec.unpack(payload)113 t,v,valid,quality,deadline,has_deadline=fields[:6]114 value=ScalarValue((self._labels[v] if self._dtype is D.ENUM else v) if valid else None,115 self._dtype,valid,Q(quality & 0x7fffffff),self._labels,partial=bool(quality>>31))116 selected=selected_deadline=None117 if self._conditioned:118 c,valid,quality,selected_deadline,has_end=fields[6:11]119 selected=ScalarValue(c if valid else None,D.BOOL,valid,Q(quality & 0x7fffffff),partial=bool(quality>>31))120 if not has_end:121 selected_deadline=None122 return (t,value,deadline if has_deadline else None,selected,selected_deadline)+((fields[-1],) if self._sample_updates else ())123124 def lower(self,time_ns):125 """二分首个不早于time_ns的索引。参数: 实际时间。返回: int。异常: 原读取错。"""126 lo,hi=0,self.count127 while lo<hi:128 mid=(lo+hi)//2129 if self.at(mid)[0]<time_ns:130 lo=mid+1131 else:132 hi=mid133 return lo134135 def rows(self,start_ns,end_ns):136 """只借出范围两侧及其中的记录，保留裁剪分母。参数: 实际起止。返回: 迭代器。异常: 原读取错。"""137 self._open()138 if not self.count:139 return140 first=self.lower(start_ns)141 if first==self.count or first>0 and self.at(first)[0]>start_ns:142 first-=1143 last=min(self.count-1,self.lower(end_ns))144 for index in range(first,last+1):145 yield self.at(index)146147 @property148 def closed(self):149 """返回自有资源关闭状态。参数: 无。返回: bool。异常: 无。"""150 return self._closed151152 def close(self):153 """关闭必要文件并退款，不碰原输入。参数: 无。返回: None。异常: 原关闭错，可重试。"""154 if self._closed:155 return156 if self._file is not None:157 self._file.close()158 self._file=None159 self._last=None160 _release_owned_ram(self,self._fixed_cost,'_fixed_cost')161 self._closed=True162