# -*- coding: utf-8 -*- """Ti-1 质量预测特征工程测试（Issue #68）。 覆盖： 1. 点位字典加载（CSV 解析、检索、存在性校验、缺列报错）； 2. FeatureSpec 校验（非法算子/未知点位/负窗口/ratio 缺 denominator）； 3. 各 transform 算子（raw/mean/std/min/max/range/diff/slope/ratio）数值正确； 4. 缺失点位 → NaN 占位； 5. 滚动窗：window 外的样本不参与； 6. 声明式加载（YAML 子集 + JSON）； 7. 重复特征名报错； 8. 模板资产 features.template.yaml 可加载并通过校验（对齐默认点位集）。 """ import math import os import sys import unittest HERE = os.path.dirname(os.path.abspath(__file__)) # .../quality-forecast/tests PKG_DIR = os.path.dirname(HERE) # .../quality-forecast TI_CL4_DIR = os.path.dirname(PKG_DIR) # .../ti-cl4 sys.path.insert(0, HERE) import _bootstrap # noqa: F401,E402 挂载 quality_forecast 包 from quality_forecast import features as F # noqa: E402 PDICT_DEFAULT = os.path.join( TI_CL4_DIR, "point-dict", "point_dict.default.csv") FEATURES_TPL = os.path.join( PKG_DIR, "config", "features.template.yaml") def _pdict(): return F.PointDict.from_csv(PDICT_DEFAULT) def _samples(values, ts0=0.0, step=10.0): """构造样本序列：values 是 [{point_id: v}, ...]。""" out = [] for i, vmap in enumerate(values): out.append(F.Sample(ts=ts0 + i * step, values=dict(vmap))) return out class TestPointDict(unittest.TestCase): def test_load_default_csv(self): pd = _pdict() self.assertTrue(pd.has("CLF-01.TEMP")) self.assertIn("CLF-01", {p.device_id for p in pd.points}) def test_by_point_id_unknown_raises(self): pd = _pdict() with self.assertRaises(F.FeatureError): pd.by_point_id("NOPE") def test_by_device(self): pd = _pdict() clf = pd.by_device("CLF-01") self.assertTrue(all(p.device_id == "CLF-01" for p in clf)) self.assertGreater(len(clf), 0) class TestFeatureSpecValidate(unittest.TestCase): def test_bad_transform(self): s = F.FeatureSpec(name="x", source="CLF-01.TEMP", transform="bogus") self.assertIn("非法 transform", "\n".join(s.validate())) def test_negative_window(self): s = F.FeatureSpec(name="x", source="CLF-01.TEMP", window=-1) self.assertIn("window 不能为负", "\n".join(s.validate())) def test_ratio_needs_denominator(self): s = F.FeatureSpec(name="x", source="CLF-01.CL2", transform="ratio") self.assertIn("denominator", "\n".join(s.validate())) def test_unknown_point_with_dict(self): pd = _pdict() s = F.FeatureSpec(name="x", source="UNKNOWN.PT") errs = s.validate(pd) self.assertTrue(any("不在点位字典" in e for e in errs)) def test_constant_source_ok(self): pd = _pdict() s = F.FeatureSpec(name="x", source="1.5") self.assertEqual(s.validate(pd), []) class TestTransforms(unittest.TestCase): def setUp(self): self.pd = _pdict() # 4 个样本，TEMP 单调上升 self.samples = _samples([ {"CLF-01.TEMP": 100.0}, {"CLF-01.TEMP": 110.0}, {"CLF-01.TEMP": 120.0}, {"CLF-01.TEMP": 130.0}, ], step=10.0) def test_raw(self): ext = F.FeatureExtractor( [F.FeatureSpec("t", "CLF-01.TEMP", "raw")], self.pd) m = ext.extract(self.samples) self.assertEqual(m.column("t")[-1], 130.0) def test_mean(self): ext = F.FeatureExtractor( [F.FeatureSpec("t", "CLF-01.TEMP", "mean", window=1000)], self.pd) m = ext.extract(self.samples) self.assertAlmostEqual(m.column("t")[-1], 115.0) def test_min_max_range(self): ext = F.FeatureExtractor([ F.FeatureSpec("mn", "CLF-01.TEMP", "min", window=1000), F.FeatureSpec("mx", "CLF-01.TEMP", "max", window=1000), F.FeatureSpec("rg", "CLF-01.TEMP", "range", window=1000), ], self.pd) m = ext.extract(self.samples) last = m.rows[-1] self.assertEqual(last[0], 100.0) # min self.assertEqual(last[1], 130.0) # max self.assertEqual(last[2], 30.0) # range def test_std(self): ext = F.FeatureExtractor( [F.FeatureSpec("s", "CLF-01.TEMP", "std", window=1000)], self.pd) m = ext.extract(self.samples) # 无偏样本标准差：100,110,120,130 → 12.9099... self.assertAlmostEqual(m.column("s")[-1], math.sqrt(500.0 / 3), places=4) def test_diff(self): ext = F.FeatureExtractor( [F.FeatureSpec("d", "CLF-01.TEMP", "diff")], self.pd) m = ext.extract(self.samples) self.assertEqual(m.column("d")[-1], 10.0) def test_slope(self): ext = F.FeatureExtractor( [F.FeatureSpec("sl", "CLF-01.TEMP", "slope", window=1000)], self.pd) m = ext.extract(self.samples) # 每 10s +10 → 斜率 1.0 self.assertAlmostEqual(m.column("sl")[-1], 1.0, places=6) def test_ratio(self): ext = F.FeatureExtractor([ F.FeatureSpec("r", "CLF-01.CL2", "ratio", denominator="CLF-01.FEED"), ], self.pd) samples = _samples([ {"CLF-01.CL2": 30.0, "CLF-01.FEED": 10.0}, {"CLF-01.CL2": 60.0, "CLF-01.FEED": 20.0}, ], step=10.0) m = ext.extract(samples) self.assertAlmostEqual(m.column("r")[-1], 3.0) class TestMissingAndWindow(unittest.TestCase): def test_missing_point_is_nan(self): ext = F.FeatureExtractor( [F.FeatureSpec("t", "CLF-01.TEMP", "raw")], _pdict()) # 样本里没有 TEMP → NaN samples = _samples([{"CLF-01.PRES": 1.0}]) m = ext.extract(samples) self.assertTrue(math.isnan(m.column("t")[0])) def test_window_excludes_old(self): ext = F.FeatureExtractor( [F.FeatureSpec("t", "CLF-01.TEMP", "mean", window=15)], _pdict()) # window=15s 只含最近 ≤2 个样本（step=10） samples = _samples([{"CLF-01.TEMP": 0.0}, {"CLF-01.TEMP": 100.0}, {"CLF-01.TEMP": 200.0}], step=10.0) m = ext.extract(samples) # 最后时刻 window=15 → 含 ts=20(100) 与 ts=30(200) → 均值 150 self.assertAlmostEqual(m.column("t")[-1], 150.0) def test_drop_nan_rows(self): ext = F.FeatureExtractor( [F.FeatureSpec("t", "CLF-01.TEMP", "raw")], _pdict()) samples = _samples([ {"CLF-01.PRES": 1.0}, # TEMP 缺失 → NaN {"CLF-01.TEMP": 50.0}, ]) m = ext.extract(samples).drop_nan_rows() self.assertEqual(len(m.rows), 1) class TestLoading(unittest.TestCase): def test_load_template_yaml(self): pd = _pdict() with open(FEATURES_TPL, "r", encoding="utf-8") as fh: text = fh.read() ext = F.load_feature_specs(text, pd) self.assertGreater(len(ext.names), 0) # 抽取一次能跑通（合成样本） samples = _samples([{"CLF-01.TEMP": 850.0, "CLF-01.CL2": 120.0, "CLF-01.FEED": 4.0, "CLF-01.CO": 2.0, "CLF-01.BED": 60.0, "RF-01.PURITY": 99.0, "RF-01.IMP": 0.3}]) m = ext.extract(samples) self.assertEqual(len(m.names), len(ext.names)) self.assertEqual(len(m.rows), 1) def test_load_json(self): import json text = json.dumps({"features": [ {"name": "t", "source": "CLF-01.TEMP", "transform": "raw"}]}) ext = F.load_feature_specs(text, _pdict()) self.assertEqual(ext.names, ["t"]) def test_duplicate_names_raise(self): with self.assertRaises(F.FeatureError): F.FeatureExtractor([ F.FeatureSpec("dup", "CLF-01.TEMP"), F.FeatureSpec("dup", "CLF-01.PRES"), ], _pdict()) def test_empty_features_raise(self): with self.assertRaises(F.FeatureError): F.load_feature_specs("features: []", _pdict()) if __name__ == "__main__": unittest.main(verbosity=2)