| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980 |
- """P4-M4: convergence-chart data source (search state points_history).
- Run: python scripts/test_p4_m4_convergence.py (exit 0 = PASS)
- Verifies get_state_summary() now exposes per-evaluated-point history
- {id, batch_id, params, objective, feasible} consumed by the frontend
- convergence chart, plus the API response model carries the field.
- """
- import os
- import sys
- _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
- _BACKEND = os.path.join(_ROOT, "web", "backend")
- sys.path.insert(0, _BACKEND)
- sys.path.insert(0, _ROOT) # l0 re-export resolves src.afmcore.l0.prescreening
- from app.services.feasibility_search import ( # noqa: E402
- FeasibilityFirstSearch, ParameterRange,
- )
- from app.services.l0_prescreening import L0PreScreeningEngine # noqa: E402
- search = FeasibilityFirstSearch(
- parameters=[
- ParameterRange(name="airgap_mm", min_value=0.5, max_value=2.0),
- ParameterRange(name="magnet_thickness_mm", min_value=3.0, max_value=8.0),
- ],
- l0_engine=L0PreScreeningEngine(),
- total_budget=40,
- batch_size=4,
- initial_samples=8,
- objective_metric="tavg_nm",
- objective_direction="maximize",
- seed=42,
- )
- initial = search.generate_initial_batch()
- assert len(initial) > 0, "initial batch empty"
- # 1) empty points_history before results
- sm = search.get_state_summary()
- assert sm["points_history"] == [], sm["points_history"]
- print("[1] points_history empty before any result OK")
- # 2) report results for every initial point (batch 0 = initial LHS batch)
- for p in initial:
- search.report_result(p.id, {"tavg_nm": float(p.id) * 0.1 + 1.0}, "ok")
- sm = search.get_state_summary()
- ph = sm["points_history"]
- assert len(ph) == len(initial), (len(ph), len(initial))
- assert all(h["objective"] is not None for h in ph)
- assert all(h["batch_id"] == 0 for h in ph)
- assert all(h["feasible"] is True for h in ph)
- print("[2] points_history populated after initial-batch results OK (%d pts)" % len(ph))
- # 3) second batch keeps history cumulative + batch_id increments to 1
- batch2 = search.select_next_batch()
- for p in batch2:
- search.report_result(p.id, {"tavg_nm": 3.0}, "ok")
- sm = search.get_state_summary()
- ph2 = sm["points_history"]
- assert len(ph2) == len(initial) + len(batch2)
- assert any(h["batch_id"] == 1 for h in ph2)
- print("[3] history cumulative across batches OK (%d pts, batches=%s)"
- % (len(ph2), sorted({h["batch_id"] for h in ph2})))
- # 4) infeasible point still listed with feasible=False
- batch3 = search.select_next_batch()
- for i, p in enumerate(batch3):
- search.report_result(p.id, {"tavg_nm": 0.0}, "infeasible")
- sm = search.get_state_summary()
- ph3 = sm["points_history"]
- assert any(h["feasible"] is False for h in ph3), "infeasible point missing"
- print("[4] infeasible points flagged OK")
- # 5) response-model wiring carries the field (import validates schema)
- from app.routers.search import SearchStateResponse # noqa: E402
- resp = SearchStateResponse(**sm)
- assert isinstance(resp.points_history, list) and len(resp.points_history) == len(ph3)
- print("[5] SearchStateResponse carries points_history OK")
- print("\nALL P4-M4 CONVERGENCE-DATA TESTS PASSED")
|