test_downsample.py

"""测试降采样的均匀分布"""

import time
from main import state, _query_data_impl

# 添加2000条测试数据
print("Adding 2000 test data entries...")
base_time = time.time()
test_data = []
for i in range(2000):
    ts = base_time + i * 0.001
    time_str = time.strftime("%H:%M:%S", time.localtime(ts))
    ms = int((ts % 1) * 1000)
    formatted = f"{time_str}.{ms:03d}"
    test_data.append((ts, formatted, f"{1000 + i}, {i}, {i * 2}"))

with state.buffer_lock:
    state.data_buffer.clear()
    for item in test_data:
        state.data_buffer.append(item)

print(f"Added {len(test_data)} entries")

# 测试降采样到500点
print("\n=== Testing downsample: 2000 -> 500 ===")
result = _query_data_impl(max_points=500)
lines = [l for l in result.split("\n") if l and not l.startswith("-") and ":" not in l]
data_rows = len(lines) - 1  # -1 for header
print(f"Returned data rows: {data_rows}")
assert data_rows == 500, f"Expected 500 rows, got {data_rows}"

# 检查是否均匀分布
indices = []
for i, line in enumerate(lines[1:]):  # 跳过header
    # 从数据中提取tick号来判断分布
    parts = line.split(",")
    if len(parts) >= 2:
        tick = int(parts[1].strip())
        indices.append(tick - 1000)  # 获取原始索引

print(f"First 10 indices: {indices[:10]}")
print(f"Last 10 indices: {indices[-10:]}")

# 检查步长是否均匀
if len(indices) > 1:
    steps = [indices[i + 1] - indices[i] for i in range(len(indices) - 1)]
    avg_step = sum(steps) / len(steps)
    print(f"Average step: {avg_step:.2f}")
    print(f"Step variance: {max(steps) - min(steps)}")

    # 步长应该在3或4（2000/500=4）
    assert 3 <= avg_step <= 4, f"Step size {avg_step} not in expected range"
    print("Distribution is uniform!")

print("\n=== Downsample test PASSED! ===")