chronaxis.sources.signal_csv
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""读取明确格式的已解码CSV或TSV信号表，逐批传送归一化信号。23输入是SignalCSVSpec和显式SignalDefinition。构造与describe只检查声明；4实际迭代时才按paths原顺序打开文件，检查表头、列数和来源内时间顺序。5每行每个请求列都是一次观测，缺失文本转None并覆盖旧值，区别于旧Filter6工件中空白表示未更新的政策。数字先保留精确整数，再由选定后端归一化。78输出SignalBatch保留文件、物理行、原文本和时间量化诊断。9batch_rows限制全部列的样本总数；同时间组允许跨文件或跨批，只有更晚10原行或整个来源结束才确认进度。全部paths耗尽才报告source_eof。11请求分析域不用于裁剪原行，开始前的更新仍保留。1213本实例管理活动生成器和文件句柄。正常耗尽、读取错误和显式close均尝试14释放资源，清理失败保留读取根因。本模块不调用旧CAN解码，也不重排文件。15"""1617# 只使用标准CSV解析，不自动嗅探原始报文格式。18import csv19# 显式类型联合与迭代接口说明资源生命周期。20from collections.abc import Iterator21# Decimal避免整型文本先经过float而失真。22from decimal import Decimal, InvalidOperation23# sys.exception提供当前根因用于清理，不使用frame反射。24import sys25# 文件句柄类型不是无类型object占位。26from typing import TextIO27# 明确的转换定义属于解码之后的分析层。28from ..core.binding import SignalDefinition29# Bool/Enum有不同文本政策，不能Python truthiness。30from ..core.types import SignalDType31from ..core.models import SourceRef32# 当前会话和来源名是安全标识。33from ..core.validation import require_identifier34# 时间字符串统一精确转换并保留量化，offset另显式施加。35from ..timebase.conversion import checked_add_ns, convert_time, parse_datetime_ns36# 来源声明与样本CSV位置是实际公开类型。37from .csv_spec import CSVSourceRef, SignalCSVSpec38# 与memory使用相同字段/转换描述，不复制不同政策。39from .descriptions import describe_fields, source_timebase_identity40# 来源批次/请求与能力严格有类型。41from .models import ReadRequest, SignalBatch, SourceCapabilities, SourceDescription42# 每个有限批次实际调用独立数值后端。43from .packing import pack_decoded44# 跨批协议和根因/关闭错误由公共层定义。45from .protocol import BatchValidator, SourceClosedError, SourceReadError464748def choice_text(value: int | bool | str) -> str:49 """生成声明编码对应的CSV文本，区分布尔与整数。5051 参数:52 value (int | bool | str): SignalDefinition已检查的原编码。53 返回:54 str。布尔使用'true'或'false'，整数使用十进制文本，字符串保持原样。55 异常:56 无主动异常；调用者须先保证原编码属于允许类型。57 """58 # 布尔文本使用明确小写协议，不使用任意非空字符串真值。59 if type(value) is bool:60 # 不把true写成整数1，编码域保持可区分。61 return 'true' if value else 'false'62 # 已声明整数或字符串按明确文本形式输出，不猜进制。63 return str(value)646566class SignalCSVSourceAdapter:67 """按文件声明顺序读取已解码表，每次请求独立保存进度。6869 spec固定格式和路径，session_id确定会话，signals与backend决定列转换。70 choices保存声明原编码的文本映射，不缓存随时间变化的值或DBC消息绑定。71 closed、active和handles分别记录关闭标记、活动请求及文件句柄。72 close后拒绝描述和读取。可重复扫描只表示新CSV能再次打开，不授权增加旧链扫描。73 """7475 def __init__(self, spec: SignalCSVSpec, session_id: str, signals: tuple[SignalDefinition, ...], *, backend: str = 'numpy', clock_only=(), retain_original=True) -> None:76 """检查来源绑定和声明编码的文本歧义，不读取实际表头。7778 参数:79 spec (SignalCSVSpec): 已检查的表格格式、时钟和有序路径声明。80 session_id (str): 本来源所属的会话名称。81 signals (tuple[SignalDefinition, ...]): 当前来源的定义，输出别名须唯一。82 backend (str): 默认'numpy'；也接受'reference'，不自动回退。83 clock_only: 只取更新时间的别名，其单元格不作数值解析。84 retain_original: 默认True保留旧逐行引用；False只使用每列共用身份，不构造原文副本。85 返回:86 None。保存输入和编码映射，初始化活动资源集合。87 异常:88 TypeError: 配置、信号容器或后端类型不合法。89 ValueError: 会话、信号归属、别名冲突，或原码文本与缺失标记有歧义。90 副作用:91 不打开文件；实际字段是否存在在迭代读取时核实。92 """93 # 原始报文CSV不能只有后缀就自动接到此接口。94 if not isinstance(spec, SignalCSVSpec) or not isinstance(signals, tuple) or backend not in ('reference', 'numpy'):95 # 所有输入都先显式解析，未知后端拒绝。96 raise TypeError('signal_csv需要typed spec/signals和明确backend')97 # 会话身份不从文件名或旧缓存继承。98 require_identifier(session_id, 'csv.session_id')99 # 当前来源全部定义必须唯一，其他来源不能借用表列。100 if any(not isinstance(item, SignalDefinition) or item.source != spec.name for item in signals) or len({item.name for item in signals}) != len(signals):101 # 不能取第一个名称冲突或静默跳过他源定义。102 raise ValueError(f'{spec.name}:CSV绑定来源/别名冲突')103 if type(retain_original) is not bool or not set(clock_only)<={item.name for item in signals}:104 raise ValueError('纯时钟字段必须是当前投影，原值保留需要bool')105 self._clock_only,self._retain_original=frozenset(clock_only),retain_original106 self._compact_refs={item.name:SourceRef(session_id,spec.name,item.field) for item in signals} if not retain_original else {}107 # 静态choices文本表只映射已解码值，不涉及DBC消息绑定缓存。108 choices = {}109 # 每个声明编码保持原类型，不对每行重新猜业务标签。110 for definition in signals:111 if definition.name in self._clock_only:112 choices[definition.name]={}113 continue114 # 输入文本必须能唯一对应声明原码。115 pairs = tuple((choice_text(code), code) for code, _ in definition.choices)116 # int1与str1在CSV中不可区分，不能自动挑第一个。117 if len({text for text, _ in pairs}) != len(pairs) or any(text in spec.missing_values for text, _ in pairs):118 # 也不能同时把一个文本声明为正常码和缺失标识。119 raise ValueError(f'{spec.name}:{definition.name}:CSV choices文本歧义/缺失冲突')120 # 保存配置解析结果，不共享任何随时间变化的信号值。121 choices[definition.name] = dict(pairs)122 # 已解析输入与当前后端明确固定。123 self.spec, self.session_id, self.signals, self.backend, self.choices = spec, session_id, signals, backend, choices124 # 当前生命周期与活跃资源只属于本实例。125 self.closed, self.active, self.handles = False, set(), set()126127 def _ensure_open(self) -> None:128 """拒绝关闭后操作，重复close本身仍安全。129130 参数: 无；读取closed。131 返回: None；开放时无副作用。132 异常: SourceClosedError包含当前来源名。133 """134 # 不能在close后自动重新打开输入延续任务。135 if self.closed:136 # 同步入口直接报告，而不是等下次next。137 raise SourceClosedError(f'{self.spec.name}:source已关闭')138139 def describe(self) -> SourceDescription:140 """返回CSV配置描述，保留尚未读取的表头和观测范围状态。141142 参数:143 无。读取spec和signals，不打开文件或查询文件状态。144 返回:145 SourceDescription。定位保存有序路径、时间列、单位、偏移和时区，146 观测端点均为None，metadata_status为'declared'。147 异常:148 SourceClosedError: 本实例已经关闭。149 TypeError或ValueError: 字段元数据或描述模型结构错误。150 """151 # 静态查询遵守已关闭状态。152 self._ensure_open()153 # 来源原定位包括有序文件列表，不能从目标列首末伪造范围。154 fields = describe_fields(self.spec.name, self.signals, {'paths': tuple(str(path) for path in self.spec.paths), 'time_column': self.spec.time_column,155 'time_unit': self.spec.unit, 'offset_ns': self.spec.offset_ns,156 'source_timebase_sha256': source_timebase_identity(self.spec.timebase, self.spec.offset_ns),157 'encoding': self.spec.encoding, 'delimiter': self.spec.delimiter, 'missing_values': self.spec.missing_values,158 'timezone': None if self.spec.timezone is None else str(self.spec.timezone)})159 # 正常/错误行诊断来自本parser，raw报文能力明确不支持。160 return SourceDescription(self.spec.name, fields, self.spec.timebase.timebase_id, None, None,161 SourceCapabilities(True, False, True, False, ('CSV文件/物理行/原文本/时间量化',),162 ('未读取前表头/观测范围仅声明', 'signal_csv缺失标记为明确MISSING，不用于旧filter未更新工件')), metadata_status='declared')163164 def iter_batches(self, request: ReadRequest) -> Iterator[SignalBatch]:165 """创建读取请求生成器，第一次迭代时才打开文件。166167 参数:168 request (ReadRequest): 当前会话、投影别名、分析范围及总样本上限。169 返回:170 Iterator[SignalBatch]。保留前置更新；所有文件耗尽后最后一批报告来源结束。171 异常:172 SourceClosedError: 实例已经关闭。173 ValueError: 请求类型、会话或投影别名错误。174 SourceReadError: 实际迭代时失败，附来源、文件、物理行及原异常链。175 副作用:176 登记活动请求；创建迭代器本身不读取表头。177 """178 # 默认不采集首端事实；委托仍执行原同步验证、批次和清理，预算入口不变。179 return self._request_batches(request)180181 def _request_batches(self, request: ReadRequest, observation_sink=None, resource_sink=None) -> Iterator[SignalBatch]:182 """创建独立原请求，可选向执行层传递首次扫描的最终观测候选。183184 参数: request为原ReadRequest；observation_sink默认None，不增加默认采集；185 非None时为执行层本次读取owner，负责独立资源及最终候选的接收。186 resource_sink默认None，仅为默认预算请求接住生成器和真实清理责任，不采集观测。187 返回: 原SignalBatch生成器；构造只验证并登记，不打开CSV。188 异常: 原SourceClosedError、配置和后续SourceReadError保持；可选sink原错传播。189 副作用: 仅登记当前生成器，其他请求及默认预算协议保持。190 """191 # 生命周期在同步调用边界验证，不隐藏延迟失败。192 self._ensure_open()193 # 本实例只接受ReadRequest，且请求会话必须与构造时一致。194 if not isinstance(request, ReadRequest) or request.session_id != self.session_id:195 # 相同时间数值也不证明会话身份相同。196 raise ValueError(f'{self.spec.name}:CSV请求会话错误')197 # 在打开文件之前拒绝未知请求列。198 validator = BatchValidator(self.describe(), request)199 # wrapper负责清理活动登记，不通过反射查找generator。200 def iterate() -> Iterator[SignalBatch]:201 """转发本次请求，并在任意退出路径移除活动登记。202203 参数:204 无。闭包保存当前request、validator及iterator。205 返回:206 每次产生内部读取器提供的SignalBatch，不重复读取或改动进度。207 异常:208 原读取错误传播；finally只移除本请求的登记。209 """210 # 异常、EOF和generator.close共用生命周期。211 try:212 # 保持私有源实际顺序与watermark。213 if observation_sink is None and resource_sink is None:214 yield from self._iterate(request, validator)215 elif observation_sink is None:216 yield from self._iterate(request, validator, resource_sink=resource_sink)217 else:218 yield from self._iterate(request, validator, observation_sink=observation_sink)219 finally:220 # 移除本请求，其他活动请求不受影响。221 self.active.discard(iterator)222 # 私有来源是真正惰性，不在创建请求时读取表头。223 iterator = iterate()224 # 可选owner先接住实际未返回生成器，active登记前后失败仍可关闭同一对象。225 if observation_sink is not None:226 observation_sink._created_iterator(iterator)227 elif resource_sink is not None:228 resource_sink._created_iterator(iterator)229 # close可统一中止已经启动或尚未启动的请求。230 self.active.add(iterator)231 # 交付有实际文件接线的公开迭代器。232 return iterator233234 def _parse_value(self, text: str, definition: SignalDefinition) -> int | float | bool | str | None:235 """按缺失标记、原编码和逻辑类型解释单元格文本。236237 参数:238 text (str): 原单元格文本，缺失标记精确匹配，不先去除空白。239 definition (SignalDefinition): 当前字段的目标类型与显式转换声明。240 返回:241 int、float、bool、str或None。缺失为None；数学整数先精确转int，242 其他数值用float表示；未声明的字符串编码可交后续归一化层判无效。243 异常:244 ValueError: 布尔文本不合法、数值语法错误、解析资源超限、245 整型转换前小数无法精确表示，或非零小数解析下溢。246 边界:247 非有限数值留给质量检查，不伪装成0。文本和整数展开分别限制4096字符或位。248 本函数只解析原值，不提前做工程缩放或有效期保持。249 """250 # 缺失精确匹配原cell，不strip文本扩大标记范围。251 if text in self.spec.missing_values:252 # 该行明确缺失观测会覆盖旧值，不同于无字段更新。253 return None254 # choices先按明确文本映射原码，随后数值核按原类型映射逻辑值。255 if text in self.choices[definition.name]:256 # 不在此提前做工程缩放或映射到目标标签。257 return self.choices[definition.name][text]258 # 未配置choices的bool只接受明确true/false协议。259 if definition.dtype is SignalDType.BOOL and not definition.choices:260 # 非空字符串和整数1不能靠truthiness变正常True。261 if text not in ('true', 'false'):262 # 任意其他布尔编码必须显式choices。263 raise ValueError(f'{definition.name}:bool文本须为true/false或显式choices')264 # 两个已声明文本保留真正False，不当缺失。265 return text == 'true'266 # 字符串原码的enum/布尔不把未声明文本猜成数值编码。267 if definition.dtype in (SignalDType.ENUM, SignalDType.BOOL) and any(type(code) is str for code, _ in definition.choices):268 # 原文本交给normalize产生OUT_OF_RANGE，而不是自动选首标签。269 return text270 # 有限文本长度限制解析本身，长输入不能构造无界整数。271 if len(text) > 4096:272 # 4096字符上限用于约束数值解析资源，与信号业务阈值无关。273 raise ValueError(f'{definition.name}:数值文本超过4096字符解析限额')274 # Decimal保留整型有效位；无穷/NaN作为无效观测原值后续诊断。275 try:276 # 文本转换不经过float秒或先损失大整数位值。277 number = Decimal(text)278 except InvalidOperation as error:279 # 错误列语法带别名，外层增加文件/物理行。280 raise ValueError(f'{definition.name}:非法数值文本') from error281 # 非有限值不是正常0，由质量核明确NONFINITE。282 if not number.is_finite():283 # 原CSV cell仍在CSVSourceRef保存，诊断不能丢原文。284 return float(number)285 # 数学整数不先转float，含.0/指数的完整整数同样精确。286 if number == number.to_integral_value():287 # 极大指数不能导致无界int分配。288 if number and number.adjusted() >= 4096:289 # 超解析资源直接失败，不生成wrap后的正常样本。290 raise ValueError(f'{definition.name}:整数展开超过4096位解析限额')291 # 原输入是否超64位由独立归一化核附OVERFLOW。292 return int(number)293 # 声明float64的小数按正常binary64表示，原文本定位保留。294 result = float(number)295 # 整型目标仍可显式转换精确小数（如0.5*2），但不可先损失原输入。296 if definition.dtype in (SignalDType.INT64, SignalDType.UINT64) and Decimal.from_float(result) != number:297 # 防止大整数加小数先舍成另一整数后被误标valid。298 raise ValueError(f'{definition.name}:整型转换前小数不能被float64精确表示')299 # 非零小数不可因float下溢变成有效零。300 if result == 0 and number:301 # 不默默补0或放宽normalizer契约。302 raise ValueError(f'{definition.name}:float64解析下溢，不能当有效0')303 # 后续有限性/类型/范围/转换由当前显式归一化政策统一处理。304 return result305306 def _close_handle(self, handle: TextIO, primary: BaseException | None) -> None:307 """关闭一个已登记句柄，保留读取根因和清理失败证据。308309 参数:310 handle (TextIO): 本实例打开的文件句柄。311 primary (BaseException | None): 已有读取根因；没有时为None。312 返回:313 None。无论关闭成功与否都移除本句柄登记，不删除文件。314 异常:315 关闭错误: 没有根因或primary为GeneratorExit时原样传播；316 已有读取根因时添加附注，保留最初异常。317 """318 # 关闭动作必须执行，不能仅移除登记。319 try:320 # TextIO关闭也可能失败，不能声明已成功完成。321 handle.close()322 except Exception as cleanup:323 # generator正常中止不是业务根因，关闭失败仍应报告。324 if primary is None or isinstance(primary, GeneratorExit):325 # 外层包装来源信息，不能吞掉正常关闭错误。326 raise327 # 错误链保留最初解析/读取异常，附加清理证据。328 primary.add_note(f'{self.spec.name}:文件关闭同时失败：{cleanup}')329 finally:330 # 即使关闭报错也不让实例无界保留引用。331 self.handles.discard(handle)332333 def _read_resources(self, request: ReadRequest):334 """为执行层封装提供独立请求的内部资源检查入口。335336 参数: request为当前已校验请求。返回: 默认None，直接CSV入口行为不变。337 异常: 执行层子类可在申请预算失败时抛错；不涉及旧报文reader或DBC。338 """339 # 原独立入口不默认创建预算或读取文件；执行层只在子类启用检查。340 return None341342 def _iterate(self, request: ReadRequest, validator: BatchValidator, observation_sink=None, resource_sink=None) -> Iterator[SignalBatch]:343 """按原文件顺序逐行读取，保持跨文件的时间组和原行序号。344345 参数:346 request (ReadRequest): 当前投影和有限批次样本上限。347 validator (BatchValidator): 本次请求独立的跨批协议检查状态。348 observation_sink: 默认None；执行层可选的本次独立观测接收者，不逐行回调。349 resource_sink: 默认None；只接默认预算请求的资源，不改变first或最终观测候选。350 返回:351 每次产生SignalBatch。最后一批可为空，且source_eof为True。352 异常:353 SourceReadError: 文件、表头、行、时间、数值或协议校验失败；354 保存来源、路径、记录结束物理行及原异常链，清理错误不覆盖根因。355 处理流程:356 每个文件打开后验证表头，按请求列解析原值并记录逐样本位置。357 缓冲满时先发布已有样本；更晚原行证明上一组完整。358 单文件结束不重置时间、序号或缓冲；所有文件结束后确认最后时间组。359 退出时关闭句柄和清空缓冲，不修改已经返回的数组。360 """361 # 显式投影保持请求顺序，分析层列选择不改变旧路径。362 selected = tuple(next(item for item in self.signals if item.name == name) for name in request.signal_names)363 # 有限pending包含记录与逐样本物理位置，不存全文件rows。364 pending, count, previous, watermark, sequence = {}, 0, None, None, 0365 # 只有可选采集保存首端；末端和原时间行数复用原previous、sequence，不看投影样本。366 first = None367 # 路径/行号即使open失败也有可读位置。368 path, line, reader, resources = self.spec.paths[0], 0, None, None369 # 读取/转换错误统一带当前真实来源位置。370 try:371 # 检查器在文件打开前申请固定读取/协议开销；每个请求独立关闭。372 if observation_sink is not None:373 resources = observation_sink._read_resources()374 elif resource_sink is not None:375 resources = resource_sink._read_resources()376 else:377 resources = self._read_resources(request)378 # 配置paths原序，不能排序、预读或重排文件。379 for path in self.spec.paths:380 # 新文件即使open失败，也不沿用前一个文件的错误行号。381 line, reader = 0, None382 # 关闭之后不能开始下一文件。383 self._ensure_open()384 # newline空串由csv解释换行，编码显式，只读输入。385 handle = path.open('r', encoding=self.spec.encoding, newline='')386 # 可选owner在活动set登记前接住真实句柄，登记取消也有恢复责任。387 if observation_sink is not None:388 observation_sink._opened_handle(handle)389 elif resource_sink is not None:390 resource_sink._opened_handle(handle)391 # 活跃句柄在close/cancellation时可追踪。392 self.handles.add(handle)393 # 单文件生命周期不重置协议时钟/sequence/pending。394 try:395 # strict拒绝损坏引号，原表头文本完整保留。396 reader = csv.reader(handle if resources is None else resources.lines(handle), delimiter=self.spec.delimiter, strict=True)397 # 预算包装只控制原reader的next，不替换CSV解释或记录物理行计数。398 records = reader if resources is None else resources.records(reader)399 # 空文件无表头不能证明声明字段，header-only表合法。400 header = next(records, None)401 # 最新物理行用于错误定位与引用。402 line = reader.line_num403 # 导出表末尾多余分隔符形成的匿名列可忽略；保留原宽度，逐行确认这些列确实没有数据。404 header_width = 0 if header is None else len(header)405 named_width = header_width406 while header is not None and named_width and header[named_width - 1] == '':407 named_width -= 1408 header = None if header is None else header[:named_width]409 # 中间空名、仅空白的名字和重名仍拒绝；已声明但未请求的字段也必须能够定位。410 if header is None or any(not name.strip() for name in header) or len(set(header)) != len(header) or self.spec.time_column not in header or any(item.field not in header for item in self.signals):411 # 不凭字段下标猜含义或把缺列当全UNKNOWN。412 hint = '；文件含BOM时请将reader.encoding设为utf-8-sig' if header and header[0].startswith('\ufeff') else ''413 raise ValueError('CSV表头缺失/重名/声明字段无法定位' + hint)414 # 静态列索引只作用已解码表，不是DBC绑定缓存。415 indexes = {name: position for position, name in enumerate(header)}416 # 每文件实际逐行读取，没有第二次输入扫描。417 for cells in records:418 # 保存记录结束的物理行号，使引号内多行记录仍可追溯。419 line = reader.line_num420 # 空物理行没有时间观测，不伪造新样本。421 if not cells:422 # 不reset前一状态或whole-source进度。423 continue424 # 允许省去表尾匿名空单元格；有名列不能缺失，也不能多出表头未声明的列。425 if not named_width <= len(cells) <= header_width:426 # 原cell错误明确报当前文件/行。427 raise ValueError('CSV行列数与表头不同')428 if any(value != '' for value in cells[named_width:]):429 raise ValueError('CSV尾部匿名列包含非空数据，需要填写列名')430 # 关闭可中止当前源，不继续读取后续数据。431 self._ensure_open()432 # 时间先精确转换，Date必须明确offset或source_timezone。433 time_text = cells[indexes[self.spec.time_column]]434 # 只有明确absolute ISO单位使用日期解析。435 conversion = parse_datetime_ns(time_text, self.spec.timezone) if self.spec.unit == 'ISO8601' else convert_time(time_text, self.spec.unit)436 # 固定offset在整数ns域安全施加，不推断同步。437 time = checked_add_ns(conversion.time_ns, self.spec.offset_ns)438 # 跨文件回退也拒绝，不重排旧输入以掩盖错误。439 if previous is not None and time < previous:440 # 单文件EOF不能允许时间重置。441 raise ValueError('来源时间回退（含连续文件边界）')442 # 必须经过转换、offset和顺序检查，空投影与缺失值原行仍算观测。443 if first is None and (observation_sink is not None or444 resource_sink is not None and resource_sink.observe_bounds):445 first = time446 if resource_sink is not None:447 resource_sink.observed_start=time448 # 下一真实时间更晚才完整越过上一组。449 if previous is not None and time > previous:450 # 观测事实与进度保证分开保存。451 watermark = previous452 # 每请求列在该行是实际观测，缺失标记也传MISSING更新。453 for definition in selected:454 # 原文本只定位一次；预算层可在数组峰值达到容量前提前交批。455 text = '' if definition.name in self._clock_only else cells[indexes[definition.field]]456 if count == request.batch_rows or (resources is not None and457 resources.should_flush(time_text, text, definition, validator)):458 # 数组/字符串定宽提升和协议集合副本必须在分配之前预留。459 if resources is not None:460 resources.before_pack(pending, validator)461 # 当前组未完整时watermark仍留在上一组。462 batch = self._make_batch(pending, watermark, False, None)463 # 跨批结构/身份/承诺验证后才交付。464 validator.accept(batch)465 # 校验成功后转为当前owned数组和当前组身份的持有预算。466 if resources is not None:467 resources.after_pack(validator)468 # 返回owned列，下一缓冲与当前批断开。469 yield batch470 # 只释放传输原记录，不reset持续/进度状态。471 pending, count = {}, 0472 # yield恢复后才释放原缓冲；交接时当前数组仍由检查器持有。473 if resources is not None:474 resources.clear_pending()475 # CSV引用会保留原文本，在解析、构造引用和append之前申请。476 if resources is not None:477 resources.before_sample(time_text, text, path, definition)478 # 原生标量随后只应用一次当前额外转换。479 value = None if definition.name in self._clock_only else self._parse_value(text, definition)480 # 每样本保留物理行/原字段和原文本/时间量化。481 reference = (CSVSourceRef(self.session_id, self.spec.name, definition.field, str(path), line, time_text, text, conversion.quantized)482 if self._retain_original else self._compact_refs[definition.name])483 # 原序号在整个配置来源内递增，单文件结束时不重置。484 pending.setdefault(definition.name, []).append((time, sequence, value, reference))485 # 上限口径为所有列样本总数。486 count += 1487 # 即使空投影，该原行也推进源实际时间/原序号。488 previous, sequence = time, sequence + 1489 finally:490 # 根因可能为读取错误或正常generator.close，分别处理关闭失败。491 try:492 if observation_sink is None and resource_sink is None:493 self._close_handle(handle, sys.exception())494 elif observation_sink is None:495 resource_sink._close_handle(handle, sys.exception())496 else:497 observation_sink._close_handle(handle, sys.exception())498 finally:499 # 预算模式保留最新实际物理行，再丢文件局部引用后释放表头/行。500 if resources is not None:501 line = reader.line_num if reader is not None else line502 header = indexes = cells = reader = records = None503 if observation_sink is None and resource_sink is None:504 resources.end_file()505 elif observation_sink is None:506 resource_sink._end_file(resources, sys.exception())507 else:508 # 此处捕获正在传播的解析、取消或文件关闭主错，清理附加错不能盖掉它。509 observation_sink._end_file(resources, sys.exception())510 # 所有paths才whole-source EOF；实际末点不补周期。511 watermark = previous if previous is not None else watermark512 # header-only或空投影末点仍是源实际事实，空来源可以None。513 if resources is not None:514 # EOF也可能有剩余样本，不能绕过数组/协议检查。515 resources.before_pack(pending, validator)516 batch = self._make_batch(pending, watermark, True, previous)517 # EOF承诺通过同一个跨batch validator。518 validator.accept(batch)519 if resources is not None:520 resources.after_pack(validator)521 # 最终validator与after_pack均通过才交候选；接收者仍须等真正StopIteration及清理成功。522 if observation_sink is not None:523 observation_sink._accept_candidate(first, previous, sequence)524 if resource_sink is not None and resource_sink.observe_bounds:525 resource_sink.observed_end=previous526 # 最后批次可含剩余样本或为空。527 yield batch528 except Exception as error:529 # csv.reader在取下一record时也会抛错，line应反映已消费物理行。530 line = reader.line_num if reader is not None else line531 # 保留最初cause/关闭附注，位置绝不使用当前机器时间代替。532 raise SourceReadError(f'{self.spec.name}:{path}:{line}:signal_csv读取失败：{error}') from error533 finally:534 # 释放有限原记录，已返回owned输出不受影响。535 pending.clear()536 # 原错误或取消都触发本请求预算抵销，不删除输入文件。537 if resources is not None:538 if observation_sink is None and resource_sink is None:539 resources.close()540 elif observation_sink is None:541 resource_sink._close_resources(sys.exception())542 else:543 observation_sink._close_resources(sys.exception())544545 def _make_batch(self, pending, watermark: int | None, eof: bool, observed_end: int | None) -> SignalBatch:546 """归一化当前缓冲中的原值，保留逐样本CSV位置。547548 参数:549 pending (dict): 别名到(time_ns, sequence, 原值, CSVSourceRef)记录列表。550 watermark (int | None): 完整时间组的进度坐标，单位为纳秒。551 eof (bool): 整个来源是否耗尽。552 observed_end (int | None): 实际观测终点纳秒坐标，未知时为None。553 返回:554 SignalBatch。每列与当前位置记录对应，未知端点不补0。555 异常:556 TypeError、ValueError或OverflowError: 数值、时间或列结构校验失败；外层附读取位置。557 副作用:558 分配批次列，不额外读取输入，也不继承其他请求的信号值。559 """560 # 当前声明不共享原信号值/跨任务绑定缓存。561 definitions = {item.name: item for item in self.signals}562 # 三元原值记录与四元CSV位置分开装箱，位置不会混入数值矩阵。563 columns = {name: pack_decoded(definitions[name], tuple((time, sequence, value) for time, sequence, value, _ in records),564 tuple(reference for _, _, _, reference in records), self.backend,565 retain_original=self._retain_original,clock_only=name in self._clock_only) for name, records in pending.items()}566 # 输出真实会话/来源/watermark/终点，未知不补0。567 return SignalBatch(self.spec.name, self.session_id, columns, watermark, eof, observed_end)568569 def close(self) -> None:570 """尝试关闭全部活动请求和句柄，最后报告最初清理失败。571572 参数:573 无。读取closed、active和handles。574 返回:575 None。重复调用直接返回；原输入文件保留。576 异常:577 SourceReadError: 至少一次清理失败，原异常链保存首错，其他失败作为附注。578 副作用:579 先标记关闭，再逐个清理生成器和残留句柄，最后清空资源登记。580 """581 # 幂等关闭不会重复删除或执行输入读取。582 if self.closed:583 # 已清理实例不自动重开。584 return585 # 先拒绝后续调用，再逐个尝试所有资源清理。586 self.closed, failure = True, None587 # generator.close触发单文件finally，未启动generator也安全。588 for iterator in tuple(self.active):589 # 一个关闭失败也不能跳过其他活动请求。590 try:591 # 释放读取进度和有限原值缓冲。592 iterator.close()593 except Exception as error:594 # 首错作为根因，其他错误另作附注。595 if failure is None:596 # 不用最后一个错误覆盖最初错误。597 failure = error598 else:599 # 保留所有清理失败但最终只抛一个带来源的错误链。600 failure.add_note(f'其他CSV请求关闭失败：{error}')601 # 残留句柄也尝试关闭，不依赖生成器已经启动。602 for handle in tuple(self.handles):603 # _close_handle用相同根因保护政策。604 try:605 # 没有根因时close失败应成为failure。606 self._close_handle(handle, failure)607 except Exception as error:608 # 原failure为None才会从helper抛错。609 failure = error610 # 所有引用清空，输入文件本身保留。611 self.active.clear()612 # 活跃句柄登记不延续到其他session。613 self.handles.clear()614 # 若任一资源清理失败，明确报告错误，不能声称清理成功。615 if failure is not None:616 # 将清理失败交给调用方，后续执行器须据此判断整体任务状态。617 raise SourceReadError(f'{self.spec.name}:signal_csv关闭失败') from failure618