张宗平
722102b335
- SUPPORTED_DATASETS registry: smd/msl/smap/swat/psm (+toy) - load_windows(name,T,stride): sliding windows + per-channel z-score norm -> shapes (N,T,C) and (N,T) - windows_to_qa: anomaly/describe QA pairs from windows (non-empty) - real loaders read data/real/<name>/test.npy (+test_label.npy) or TSMM_DATA_DIR; raise clear FileNotFoundError when absent (testable offline via 'toy') - write_eval_jsonl helper for eval_real.jsonl - tests/test_real_bench.py (7 tests, RED->GREEN) Exit criteria: >=2 datasets load (toy + registry of 5 real); window/label shapes; QA non-empty; missing real data raises clearly.
62 lines
2.4 KiB
Python
62 lines
2.4 KiB
Python
"""Tests for real-benchmark loading (T1.6)."""
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from tsmm.data import real_bench
|
|
|
|
|
|
def test_supported_dataset_registry_has_at_least_two():
|
|
regs = real_bench.SUPPORTED_DATASETS
|
|
assert len(regs) >= 2
|
|
|
|
|
|
def test_load_windows_shapes_and_dtypes():
|
|
# 'toy' is a deterministic in-memory benchmark used to exercise the pipeline
|
|
# without requiring an external download.
|
|
windows, labels, meta = real_bench.load_windows("toy", T=64, stride=32)
|
|
assert windows.ndim == 3 and labels.ndim == 2
|
|
N, T, C = windows.shape
|
|
assert labels.shape == (N, T)
|
|
assert windows.shape[0] == labels.shape[0]
|
|
assert np.isfinite(windows).all() # normalized -> finite
|
|
assert set(np.unique(labels)).issubset({0, 1})
|
|
|
|
|
|
def test_load_windows_normalizes_per_channel():
|
|
windows, _, _ = real_bench.load_windows("toy", T=64, stride=32)
|
|
# z-score normalization -> mean ~0, std ~1 per channel over the whole set
|
|
mu = windows.reshape(-1, windows.shape[-1]).mean(0)
|
|
sd = windows.reshape(-1, windows.shape[-1]).std(0)
|
|
assert np.allclose(mu, 0, atol=1e-6)
|
|
assert np.allclose(sd, 1, atol=1e-6)
|
|
|
|
|
|
def test_load_windows_stride_controls_count():
|
|
w1, _, _ = real_bench.load_windows("toy", T=64, stride=32)
|
|
w2, _, _ = real_bench.load_windows("toy", T=64, stride=64)
|
|
# smaller stride -> at least as many windows
|
|
assert w1.shape[0] >= w2.shape[0]
|
|
|
|
|
|
def test_windows_to_qa_produces_nonempty_pairs():
|
|
windows, labels, meta = real_bench.load_windows("toy", T=64, stride=32)
|
|
qa = real_bench.windows_to_qa(windows, labels, meta, seed=0)
|
|
assert isinstance(qa, list) and len(qa) > 0
|
|
for item in qa:
|
|
assert set(item.keys()) >= {"instruction", "answer", "labels", "series", "category"}
|
|
assert isinstance(item["instruction"], str) and item["instruction"].strip()
|
|
assert isinstance(item["answer"], str) and item["answer"].strip()
|
|
|
|
|
|
def test_load_windows_unknown_dataset_raises():
|
|
with pytest.raises(KeyError):
|
|
real_bench.load_windows("does-not-exist", T=64, stride=32)
|
|
|
|
|
|
def test_load_windows_real_dataset_missing_data_raises_clear_error():
|
|
# Real loaders must raise a helpful error when the local data path is absent,
|
|
# rather than silently returning empty arrays.
|
|
with pytest.raises((FileNotFoundError, RuntimeError)):
|
|
# SMD is almost certainly not present in CI; should raise.
|
|
real_bench.load_windows("smd", T=64, stride=32)
|