227 lines
7.4 KiB
Python
227 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
测试脚本：验证完整流程中的时间戳去重功能
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import tempfile
|
||
from core.simulator import SampleSpaceGenerator, TimeSeriesGenerator, DetectorSampler
|
||
from core.data_writer import DataWriterV4
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
# 测试配置
|
||
detector_configs = {
|
||
"detector1": {
|
||
"name": "detector1",
|
||
"energy_distribution": {
|
||
"type": "uniform",
|
||
"params": {
|
||
"low": 100,
|
||
"high": 1000
|
||
}
|
||
},
|
||
"timestamp_distribution": {
|
||
"type": "uniform",
|
||
"params": {
|
||
"low": 0,
|
||
"high": 100
|
||
}
|
||
},
|
||
"sample_space_size": 1000,
|
||
"allow_replacement": True
|
||
},
|
||
"detector2": {
|
||
"name": "detector2",
|
||
"energy_distribution": {
|
||
"type": "uniform",
|
||
"params": {
|
||
"low": 100,
|
||
"high": 1000
|
||
}
|
||
},
|
||
"timestamp_distribution": {
|
||
"type": "uniform",
|
||
"params": {
|
||
"low": 0,
|
||
"high": 100
|
||
}
|
||
},
|
||
"sample_space_size": 1000,
|
||
"allow_replacement": True
|
||
},
|
||
"detector3": {
|
||
"name": "detector3",
|
||
"energy_distribution": {
|
||
"type": "uniform",
|
||
"params": {
|
||
"low": 100,
|
||
"high": 1000
|
||
}
|
||
},
|
||
"timestamp_distribution": {
|
||
"type": "uniform",
|
||
"params": {
|
||
"low": 0,
|
||
"high": 100
|
||
}
|
||
},
|
||
"sample_space_size": 1000,
|
||
"allow_replacement": True
|
||
}
|
||
}
|
||
|
||
sync_pulses = [
|
||
{
|
||
"count": 475,
|
||
"period": 16000, # 16000 * 10ns = 160us
|
||
},
|
||
{
|
||
"count": 20,
|
||
"period": 100000, # 100000 * 10ns = 1000us
|
||
},
|
||
{
|
||
"count": 1,
|
||
"period": 400000, # 400000 * 10ns = 4000us
|
||
},
|
||
{
|
||
"count": 0,
|
||
"period": 0,
|
||
}
|
||
]
|
||
|
||
# 运行测试
|
||
def test_complete_flow():
|
||
print("开始测试完整流程...")
|
||
|
||
# 创建样本空间生成器
|
||
sample_gen = SampleSpaceGenerator()
|
||
|
||
# 生成样本空间
|
||
sample_spaces = {}
|
||
samplers = {}
|
||
|
||
for det_name, det_config in detector_configs.items():
|
||
sample_space = sample_gen.generate_sample_space(det_config)
|
||
sample_spaces[det_name] = sample_space
|
||
# 创建 DetectorSampler 实例
|
||
allow_replacement = det_config.get('allow_replacement', True)
|
||
samplers[det_name] = DetectorSampler(sample_space, allow_replacement)
|
||
|
||
# 创建时间序列生成器
|
||
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
|
||
)
|
||
|
||
# 生成时间序列
|
||
time_series = ts_gen.generate_time_series()
|
||
|
||
# 检查每个探测器的信号
|
||
print("\n检查生成的时间序列...")
|
||
for det_name in ["detector1", "detector2", "detector3"]:
|
||
if det_name in time_series:
|
||
signals = time_series[det_name]
|
||
print(f"探测器 {det_name} 生成了 {len(signals)} 个信号")
|
||
|
||
# 检查是否有重复时间戳
|
||
timestamps = [signal[0] for signal in signals]
|
||
unique_timestamps = set(timestamps)
|
||
|
||
if len(timestamps) != len(unique_timestamps):
|
||
print(f"❌ 发现重复时间戳！总时间戳数: {len(timestamps)}, 唯一时间戳数: {len(unique_timestamps)}")
|
||
|
||
# 找出重复的时间戳
|
||
seen = set()
|
||
duplicates = set()
|
||
for ts in timestamps:
|
||
if ts in seen:
|
||
duplicates.add(ts)
|
||
else:
|
||
seen.add(ts)
|
||
|
||
print(f"重复的时间戳: {sorted(duplicates)[:10]}...") # 只显示前10个
|
||
else:
|
||
print("✅ 未发现重复时间戳")
|
||
|
||
# 编码时间序列
|
||
print("\n编码时间序列...")
|
||
encoded_signals = ts_gen.encode_time_series(time_series)
|
||
print(f"编码后信号数: {len(encoded_signals)}")
|
||
|
||
print("\n测试完成！")
|
||
print("✅ 时间戳去重功能已验证成功")
|
||
|
||
def check_debug_file(debug_file):
|
||
"""检查 debug 文件中的重复时间戳"""
|
||
det_timestamps = {
|
||
"detector1": {},
|
||
"detector2": {},
|
||
"detector3": {}
|
||
}
|
||
|
||
current_event = None
|
||
|
||
with open(debug_file, 'r', encoding='utf-8') as f:
|
||
for line_num, line in enumerate(f, 1):
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
|
||
# 检查是否是事件行
|
||
if '|' in line and '时间戳:' in line:
|
||
parts = line.split('|')
|
||
if len(parts) >= 4:
|
||
event_id = parts[0].strip()
|
||
timestamp_part = parts[3].strip()
|
||
signal_part = parts[3].strip()
|
||
|
||
# 提取时间戳
|
||
timestamp_start = timestamp_part.find('时间戳:') + 4
|
||
timestamp_end = timestamp_part.find('us', timestamp_start)
|
||
if timestamp_start > 4 and timestamp_end > timestamp_start:
|
||
timestamp_str = timestamp_part[timestamp_start:timestamp_end].strip()
|
||
try:
|
||
timestamp = int(timestamp_str)
|
||
except ValueError:
|
||
continue
|
||
else:
|
||
continue
|
||
|
||
# 提取探测器信号
|
||
signal_start = signal_part.find('信号:') + 3
|
||
if signal_start > 3:
|
||
signal_str = signal_part[signal_start:].strip()
|
||
# 解析探测器信号
|
||
det_signals = signal_str.split(', ')
|
||
for det_signal in det_signals:
|
||
if ':' in det_signal:
|
||
det_part, energy_part = det_signal.split(':', 1)
|
||
det_name = det_part.strip()
|
||
if det_name in det_timestamps:
|
||
# 使用事件ID作为键，记录每个事件中每个探测器的时间戳
|
||
if event_id not in det_timestamps[det_name]:
|
||
det_timestamps[det_name][event_id] = set()
|
||
|
||
# 检查是否重复
|
||
if timestamp in det_timestamps[det_name][event_id]:
|
||
print(f"❌ 发现重复时间戳: 行 {line_num}, 事件 {event_id}, 探测器 {det_name}, 时间戳 {timestamp}us")
|
||
else:
|
||
det_timestamps[det_name][event_id].add(timestamp)
|
||
|
||
# 统计结果
|
||
print("\n检查完成！")
|
||
for det_name, events in det_timestamps.items():
|
||
total_events = len(events)
|
||
print(f"探测器 {det_name}: 检查了 {total_events} 个事件")
|
||
|
||
if __name__ == "__main__":
|
||
test_complete_flow() |