test_p4_m4_convergence.py 3.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980
  1. """P4-M4: convergence-chart data source (search state points_history).
  2. Run: python scripts/test_p4_m4_convergence.py (exit 0 = PASS)
  3. Verifies get_state_summary() now exposes per-evaluated-point history
  4. {id, batch_id, params, objective, feasible} consumed by the frontend
  5. convergence chart, plus the API response model carries the field.
  6. """
  7. import os
  8. import sys
  9. _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
  10. _BACKEND = os.path.join(_ROOT, "web", "backend")
  11. sys.path.insert(0, _BACKEND)
  12. sys.path.insert(0, _ROOT) # l0 re-export resolves src.afmcore.l0.prescreening
  13. from app.services.feasibility_search import ( # noqa: E402
  14. FeasibilityFirstSearch, ParameterRange,
  15. )
  16. from app.services.l0_prescreening import L0PreScreeningEngine # noqa: E402
  17. search = FeasibilityFirstSearch(
  18. parameters=[
  19. ParameterRange(name="airgap_mm", min_value=0.5, max_value=2.0),
  20. ParameterRange(name="magnet_thickness_mm", min_value=3.0, max_value=8.0),
  21. ],
  22. l0_engine=L0PreScreeningEngine(),
  23. total_budget=40,
  24. batch_size=4,
  25. initial_samples=8,
  26. objective_metric="tavg_nm",
  27. objective_direction="maximize",
  28. seed=42,
  29. )
  30. initial = search.generate_initial_batch()
  31. assert len(initial) > 0, "initial batch empty"
  32. # 1) empty points_history before results
  33. sm = search.get_state_summary()
  34. assert sm["points_history"] == [], sm["points_history"]
  35. print("[1] points_history empty before any result OK")
  36. # 2) report results for every initial point (batch 0 = initial LHS batch)
  37. for p in initial:
  38. search.report_result(p.id, {"tavg_nm": float(p.id) * 0.1 + 1.0}, "ok")
  39. sm = search.get_state_summary()
  40. ph = sm["points_history"]
  41. assert len(ph) == len(initial), (len(ph), len(initial))
  42. assert all(h["objective"] is not None for h in ph)
  43. assert all(h["batch_id"] == 0 for h in ph)
  44. assert all(h["feasible"] is True for h in ph)
  45. print("[2] points_history populated after initial-batch results OK (%d pts)" % len(ph))
  46. # 3) second batch keeps history cumulative + batch_id increments to 1
  47. batch2 = search.select_next_batch()
  48. for p in batch2:
  49. search.report_result(p.id, {"tavg_nm": 3.0}, "ok")
  50. sm = search.get_state_summary()
  51. ph2 = sm["points_history"]
  52. assert len(ph2) == len(initial) + len(batch2)
  53. assert any(h["batch_id"] == 1 for h in ph2)
  54. print("[3] history cumulative across batches OK (%d pts, batches=%s)"
  55. % (len(ph2), sorted({h["batch_id"] for h in ph2})))
  56. # 4) infeasible point still listed with feasible=False
  57. batch3 = search.select_next_batch()
  58. for i, p in enumerate(batch3):
  59. search.report_result(p.id, {"tavg_nm": 0.0}, "infeasible")
  60. sm = search.get_state_summary()
  61. ph3 = sm["points_history"]
  62. assert any(h["feasible"] is False for h in ph3), "infeasible point missing"
  63. print("[4] infeasible points flagged OK")
  64. # 5) response-model wiring carries the field (import validates schema)
  65. from app.routers.search import SearchStateResponse # noqa: E402
  66. resp = SearchStateResponse(**sm)
  67. assert isinstance(resp.points_history, list) and len(resp.points_history) == len(ph3)
  68. print("[5] SearchStateResponse carries points_history OK")
  69. print("\nALL P4-M4 CONVERGENCE-DATA TESTS PASSED")