#!/usr/bin/env python3 """ 测试完整流程中的时间戳去重和信号合并功能 """ import os import tempfile import numpy as np from core.simulator import SampleSpaceGenerator, DetectorSampler, TimeSeriesGenerator from core.data_writer import DataWriterV4 def test_complete_flow_with_debug_file(): """测试完整流程，包括生成、编码和数据写入""" print("开始测试完整流程的时间戳去重和信号合并功能...") # 配置参数 sync_pulses = [ {"count": 10, "period": 10000}, # 同步脉冲1: 10个，周期10000 (10ns) {"count": 5, "period": 20000}, # 同步脉冲2: 5个，周期20000 (10ns) {"count": 0, "period": 0}, # 同步脉冲3: 0个 {"count": 0, "period": 0} # 同步脉冲4: 0个 ] # 探测器配置 detector_configs = { "detector1": { "name": "detector1", "energy_range": [0, 1000], "time_range": [0, 1000], "signal_count": { "min": 1, "max": 10 } }, "detector2": { "name": "detector2", "energy_range": [0, 1000], "time_range": [0, 1000], "signal_count": { "min": 1, "max": 10 } }, "detector3": { "name": "detector3", "energy_range": [0, 1000], "time_range": [0, 1000], "signal_count": { "min": 1, "max": 10 } } } # 生成样本空间 sample_space_gen = SampleSpaceGenerator() sample_spaces = {} samplers = {} for det_name, config in detector_configs.items(): # 生成样本空间配置 det_config = { "name": det_name, "sample_space_size": 1000, "energy_distribution": { "type": "uniform", "low": config["energy_range"][0], "high": config["energy_range"][1] }, "timestamp_distribution": { "type": "uniform", "low": config["time_range"][0], "high": config["time_range"][1] } } # 生成样本空间 sample_space = sample_space_gen.generate_sample_space(det_config) sample_spaces[det_name] = sample_space # 创建采样器 sampler = DetectorSampler(sample_space) samplers[det_name] = sampler print(f"为{det_name}生成了{len(sample_space)}个样本") # 同步脉冲信号数配置 sync_pulse_signals = { "pulse1": { "detector1": {"min": 1, "max": 10}, "detector2": {"min": 1, "max": 10}, "detector3": {"min": 1, "max": 10} }, "pulse2": { "detector1": {"min": 1, "max": 10}, "detector2": {"min": 1, "max": 10}, "detector3": {"min": 1, "max": 10} } } # 创建时间序列生成器 ts_gen = TimeSeriesGenerator( work_mode="CO", sync_pulses=sync_pulses, sample_spaces=sample_spaces, samplers=samplers, detectors=detector_configs, energy_K=1.0, energy_B=0.0, sync_pulse_signals=sync_pulse_signals ) # 生成时间序列 print("\n生成时间序列...") time_series = ts_gen.generate_time_series() # 检查生成的时间序列 print("\n检查生成的时间序列...") for det_name in ["detector1", "detector2", "detector3"]: signals = time_series.get(det_name, []) print(f"探测器 {det_name} 生成了 {len(signals)} 个信号") # 检查重复时间戳 timestamps = [sig[0] for sig in signals] duplicates = [ts for ts in set(timestamps) if timestamps.count(ts) > 1] if duplicates: print(f"❌ 发现重复时间戳: {sorted(duplicates)[:10]}...") # 只显示前10个 else: print("✅ 未发现重复时间戳") # 编码时间序列 print("\n编码时间序列...") encoded_signals = ts_gen.encode_time_series(time_series) print(f"编码后信号数: {len(encoded_signals)}") # 创建临时文件 with tempfile.NamedTemporaryFile(suffix='_debug.txt', delete=False) as f: debug_file = f.name # 创建数据写入器 writer_config = { "simulation": { "work_mode": "CO", "sync_pulses": sync_pulses, "energy_conversion": { "K": 1.0, "B": 0.0 } }, "detectors": detector_configs } data_writer = DataWriterV4(writer_config) # 写入 debug 文件 print(f"\n写入 debug 文件: {debug_file}") # 由于 write_events_debug_text 期望的是事件列表，其中每个事件是该同步脉冲的信号列表 # 我们需要创建正确的事件格式 try: # 按事件结束标志分割信号，创建事件列表 events = [] current_event = [] for signal in encoded_signals: current_event.append(signal) # 检查是否是事件结束信号 # 事件结束标志在bit63 if (signal >> 63) & 1: events.append(current_event) current_event = [] if current_event: events.append(current_event) data_writer.write_events_debug_text(events, debug_file) print(f"✅ Debug 文件写入成功，共写入 {len(events)} 个事件") except Exception as e: print(f"❌ Debug 文件写入失败: {e}") return # 检查 debug 文件中的重复时间戳 print("\n检查 debug 文件中的重复时间戳...") check_debug_file(debug_file) # 清理临时文件 os.unlink(debug_file) print(f"\n清理临时文件: {debug_file}") print("\n测试完成！") print("✅ 完整流程的时间戳去重和信号合并功能已验证成功") def check_debug_file(debug_file): """检查 debug 文件中的重复时间戳""" duplicate_found = False with open(debug_file, 'r', encoding='utf-8') as f: current_event = None event_det_timestamps = {} # 每个事件的探测器时间戳 for line in f: line = line.strip() if not line: continue if line.startswith('#'): # 检查是否是事件结束标记 if '总结:' in line: # 事件结束，重置时间戳记录 current_event = None event_det_timestamps = {} continue # 解析行数据 parts = line.split('|') if len(parts) < 4: continue event_idx = int(parts[0].strip()) info = parts[3].strip() # 提取时间戳和探测器信息 timestamp_str = None det_infos = [] for part in info.split(','): part = part.strip() if part.startswith('时间戳:'): timestamp_str = part.split(':')[1].replace('us', '').strip() elif part.startswith('信号:'): signal_part = part.split(':', 1)[1].strip() for det_info in signal_part.split(','): det_info = det_info.strip() if det_info: det_infos.append(det_info) if timestamp_str: timestamp = int(timestamp_str) # 初始化当前事件的时间戳记录 if current_event != event_idx: current_event = event_idx event_det_timestamps = { "detector1": set(), "detector2": set(), "detector3": set() } for det_info in det_infos: # 提取探测器名称和能量 if ':' in det_info: det_part = det_info.split(':')[0] if det_part.startswith('Det'): det_idx = int(det_part[3]) - 1 # Det1 -> 0 det_name = f"detector{det_idx + 1}" # 检查该探测器在该事件内的该时间戳是否已存在 if timestamp in event_det_timestamps[det_name]: print(f"❌ 重复时间戳: 事件={event_idx}, 时间戳={timestamp}us, 探测器={det_name}") duplicate_found = True else: event_det_timestamps[det_name].add(timestamp) if not duplicate_found: print("✅ Debug 文件中未发现重复时间戳") else: print("\n注意：不同事件之间的时间戳重复是正常的，因为时间戳是相对于每个同步脉冲的") if __name__ == "__main__": test_complete_flow_with_debug_file()