File size: 9,691 Bytes
0e6887b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 | """描述性梳理统计器测试（intent-understanding-layer 任务 9）。
覆盖 Property 8（不外推）/ Property 10（语言无关）与需求 7.2–7.4、7.7。
"""
from __future__ import annotations
import math
from skills.descriptive_summary.summarizer import summarize
def test_groups_by_strength_batch_attribute():
obs = [
{"strength": "20μg", "batch": "B1", "attribute": "总杂", "value": "0.00"},
{"strength": "40μg", "batch": "B2", "attribute": "总杂", "value": "0.00"},
]
out = summarize(obs)
assert out["n_groups"] == 2
assert set(out["strengths"]) == {"20μg", "40μg"}
def test_mean_and_rsd_math():
obs = [
{"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "99.19"},
{"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "100.00"},
{"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "97.68"},
]
out = summarize(obs)
g = out["groups"][0]
assert g["n"] == 3
assert math.isclose(g["mean"], (99.19 + 100.0 + 97.68) / 3, rel_tol=1e-9)
assert g["rsd_pct"] is not None and g["rsd_pct"] > 0
def test_rsd_none_for_single_point():
out = summarize([{"strength": "x", "batch": "B", "attribute": "a", "value": "5"}])
assert out["groups"][0]["rsd_pct"] is None
def test_conformance_upper_limit():
obs = [{"strength": "20μg", "batch": "B1", "attribute": "总杂", "value": "0.5"}]
out = summarize(obs, spec_limits={"总杂": "总杂≤2.0%"})
assert out["groups"][0]["within_spec"] is True
obs2 = [{"strength": "20μg", "batch": "B1", "attribute": "总杂", "value": "3.0"}]
out2 = summarize(obs2, spec_limits={"总杂": "总杂≤2.0%"})
assert out2["groups"][0]["within_spec"] is False
def test_conformance_range_assay():
obs = [{"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "98.92"}]
out = summarize(obs, spec_limits={"含量": "90.0~110.0%"})
assert out["groups"][0]["within_spec"] is True
def test_conformance_none_without_spec():
out = summarize([{"strength": "x", "batch": "B", "attribute": "总杂", "value": "0.5"}])
assert out["groups"][0]["within_spec"] is None
def test_no_extrapolation_fields_property():
out = summarize([{"strength": "x", "batch": "B", "attribute": "a", "value": "1"}])
forbidden = {"shelf_life", "k", "r2", "model", "extrapolation", "target_timepoints", "arrhenius"}
blob = str(out).lower()
for f in forbidden:
assert f not in blob
def test_multi_document_aggregation():
obs = [
{"strength": "20μg", "batch": "B1", "attribute": "总杂", "value": "0.1", "source": "a.xlsx"},
{"strength": "20μg", "batch": "B1", "attribute": "总杂", "value": "0.2", "source": "b.xlsx"},
]
out = summarize(obs)
g = out["groups"][0]
assert g["n"] == 2
assert set(g["sources"]) == {"a.xlsx", "b.xlsx"}
def test_non_numeric_values_skipped():
obs = [
{"strength": "x", "batch": "B", "attribute": "a", "value": "样品名称"},
{"strength": "x", "batch": "B", "attribute": "a", "value": "1.0"},
]
out = summarize(obs)
assert out["groups"][0]["n"] == 1
def test_conformance_less_than_operators():
# ＜ / < 上限型
obs = [{"strength": "20μg", "batch": "B1", "attribute": "溶化时限", "value": "60"}]
out = summarize(obs, spec_limits={"溶化时限": "＜120s"})
assert out["groups"][0]["within_spec"] is True
out2 = summarize([{"strength": "20μg", "batch": "B1", "attribute": "溶化时限", "value": "130"}],
spec_limits={"溶化时限": "<120s"})
assert out2["groups"][0]["within_spec"] is False
def test_conformance_greater_than_lower_limit():
obs = [{"strength": "20μg", "batch": "B1", "attribute": "抗拉强度", "value": "16.98"}]
out = summarize(obs, spec_limits={"抗拉强度": "≥1"})
assert out["groups"][0]["within_spec"] is True
out2 = summarize([{"strength": "20μg", "batch": "B1", "attribute": "抗拉强度", "value": "0.5"}],
spec_limits={"抗拉强度": ">1"})
assert out2["groups"][0]["within_spec"] is False
def test_same_attribute_different_tables_not_merged():
"""不同来源表的同名属性不应被合并（修复过度合并）。"""
obs = [
{"strength": "20μg", "batch": "B1", "attribute": "含量mg", "value": "0.834", "table": "含量均匀度"},
{"strength": "20μg", "batch": "B1", "attribute": "含量mg", "value": "0.826", "table": "含量均匀度"},
{"strength": "20μg", "batch": "B1", "attribute": "含量mg", "value": "0.826", "table": "含量测定"},
]
out = summarize(obs)
groups = [g for g in out["groups"] if g["attribute"] == "含量mg"]
# 两张表 → 两个分组，而非合并成一个 n=3
assert len(groups) == 2
tables = {g["table"] for g in groups}
assert tables == {"含量均匀度", "含量测定"}
def test_spec_limit_normalized_match_attribute_with_unit_suffix():
"""限度键与观测属性名存在单位后缀差异时，归一化回退仍能挂载限度。"""
# 观测属性 "总杂%"，限度键 "总杂" —— 精确匹配失败，归一化匹配成功。
obs = [{"strength": "20μg", "batch": "B1", "attribute": "总杂%", "value": "0.5"}]
out = summarize(obs, spec_limits={"总杂": "总杂≤2.0%"})
g = out["groups"][0]
assert g["spec_limit"] == "总杂≤2.0%"
assert g["within_spec"] is True
def test_spec_limit_normalized_match_assay_mg_per_g():
"""'含量mg/g' 观测可匹配 '含量' 限度键。"""
obs = [{"strength": "20μg", "batch": "B1", "attribute": "含量mg/g", "value": "98.0"}]
out = summarize(obs, spec_limits={"含量": "90.0~110.0%"})
assert out["groups"][0]["within_spec"] is True
def test_exact_match_still_preferred_over_normalized():
"""精确键存在时优先精确匹配，不被归一化覆盖。"""
obs = [{"strength": "x", "batch": "B", "attribute": "总杂", "value": "0.5"}]
out = summarize(obs, spec_limits={"总杂": "总杂≤2.0%"})
assert out["groups"][0]["spec_limit"] == "总杂≤2.0%"
def test_display_fields_consistent_format():
"""mean_display / rsd_display 为统一定点串，供表格与叙述共用。"""
obs = [
{"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "99.19"},
{"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "100.00"},
{"strength": "20μg", "batch": "B1", "attribute": "含量", "value": "97.68"},
]
g = summarize(obs)["groups"][0]
assert g["mean_display"] == f"{g['mean']:.4g}"
assert g["rsd_display"] == f"{g['rsd_pct']:.2f}"
def test_single_point_flag():
"""n=1 分组带 single_point 标注；n≥2 不带。"""
out = summarize([
{"strength": "x", "batch": "B", "attribute": "a", "value": "5"},
{"strength": "y", "batch": "B", "attribute": "a", "value": "5"},
{"strength": "y", "batch": "B", "attribute": "a", "value": "6"},
])
by_strength = {g["strength"]: g for g in out["groups"]}
assert by_strength["x"]["single_point"] is True
assert by_strength["y"]["single_point"] is False
assert by_strength["y"]["rsd_display"] is not None
def test_censored_value_excluded_from_mean_but_counted():
"""删失值（＞100）不参与均值/RSD，但计入观测总数并单独保留。"""
out = summarize([
{"strength": "20μg", "batch": "B1", "attribute": "耐折度", "value": "＞100"},
])
g = out["groups"][0]
assert g["mean"] is None # 无真实点测定 → 无均值
assert g["rsd_pct"] is None
assert g["n"] == 1 # 仍计为一个观测
assert g["n_numeric"] == 0
assert g["censored_values"] == ["＞100"]
def test_censored_gt_satisfies_lower_limit():
"""＞100 对下限 ≥50 → 必然符合。"""
out = summarize(
[{"strength": "20μg", "batch": "B1", "attribute": "耐折度", "value": "＞100"}],
spec_limits={"耐折度": "≥50"},
)
assert out["groups"][0]["within_spec"] is True
def test_censored_lt_satisfies_upper_limit():
"""<0.05 对上限 ≤1.0 → 必然符合。"""
out = summarize(
[{"strength": "20μg", "batch": "B1", "attribute": "单杂", "value": "<0.05"}],
spec_limits={"单杂": "≤1.0"},
)
assert out["groups"][0]["within_spec"] is True
def test_censored_gt_violates_upper_limit():
"""＞100 对上限 ≤80 → 必然违反。"""
out = summarize(
[{"strength": "20μg", "batch": "B1", "attribute": "x", "value": "＞100"}],
spec_limits={"x": "≤80"},
)
assert out["groups"][0]["within_spec"] is False
def test_censored_indeterminate_returns_none():
"""＞100 对上限 ≤120 → 真值可能超限，无法确定 → None（不臆断）。"""
out = summarize(
[{"strength": "20μg", "batch": "B1", "attribute": "x", "value": "＞100"}],
spec_limits={"x": "≤120"},
)
assert out["groups"][0]["within_spec"] is None
def test_mixed_numeric_and_censored_conformance():
"""数值点 + 删失值混合：全部确定满足 → True。"""
out = summarize(
[
{"strength": "20μg", "batch": "B1", "attribute": "耐折度", "value": "120"},
{"strength": "20μg", "batch": "B1", "attribute": "耐折度", "value": "＞100"},
],
spec_limits={"耐折度": "≥50"},
)
g = out["groups"][0]
assert g["n"] == 2 and g["n_numeric"] == 1
assert g["mean"] == 120.0 # 仅数值点参与均值
assert g["within_spec"] is True
|