test_analytics.py 3.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283
  1. """Test analytics service logic without database."""
  2. import sys
  3. sys.path.insert(0, '.')
  4. from app.services.analytics import (
  5. compute_experience_stats,
  6. find_similar_cases,
  7. compute_trend_data,
  8. compute_pareto_frontier,
  9. compute_sensitivity,
  10. get_metric_defs,
  11. )
  12. # Test data
  13. mock_cases = [
  14. {
  15. "id": 1, "topology": "SSSR", "source_plan_id": "plan-001",
  16. "params": {"Airgap": 1.0, "RMSCurrent": 21, "MagnetThickness": 5},
  17. "metrics": {"tavg_nm": 2.5, "efficiency_pct": 85.2, "ripple_pct": 3.5},
  18. "conclusion": "Good baseline", "tags": ["baseline", "sssr"], "rating": 4,
  19. },
  20. {
  21. "id": 2, "topology": "SSSR", "source_plan_id": "plan-002",
  22. "params": {"Airgap": 1.2, "RMSCurrent": 21, "MagnetThickness": 5},
  23. "metrics": {"tavg_nm": 2.3, "efficiency_pct": 84.8, "ripple_pct": 3.2},
  24. "conclusion": "Larger airgap reduces torque", "tags": ["airgap-study"], "rating": 3,
  25. },
  26. {
  27. "id": 3, "topology": "DRSS", "source_plan_id": "plan-003",
  28. "params": {"Airgap": 1.0, "RMSCurrent": 25, "MagnetThickness": 6},
  29. "metrics": {"tavg_nm": 3.1, "efficiency_pct": 87.5, "ripple_pct": 2.8},
  30. "conclusion": "DRSS higher torque", "tags": ["drss", "high-torque"], "rating": 5,
  31. },
  32. ]
  33. mock_results = [
  34. {"status": "OK", "params": {"Airgap": 0.8, "RMSCurrent": 20}, "metrics": {"tavg_nm": 2.8, "efficiency_pct": 86.0, "total_losses_w": 50}},
  35. {"status": "OK", "params": {"Airgap": 1.0, "RMSCurrent": 20}, "metrics": {"tavg_nm": 2.5, "efficiency_pct": 85.2, "total_losses_w": 48}},
  36. {"status": "OK", "params": {"Airgap": 1.2, "RMSCurrent": 20}, "metrics": {"tavg_nm": 2.3, "efficiency_pct": 84.8, "total_losses_w": 45}},
  37. {"status": "OK", "params": {"Airgap": 1.0, "RMSCurrent": 25}, "metrics": {"tavg_nm": 3.0, "efficiency_pct": 86.5, "total_losses_w": 55}},
  38. {"status": "FAILED", "params": {"Airgap": 1.5, "RMSCurrent": 20}, "metrics": {}, "error_message": "convergence failed"},
  39. ]
  40. print("=== 1. Metric Definitions ===")
  41. metrics = get_metric_defs()
  42. print(f"Total metrics: {len(metrics)}")
  43. for m in metrics[:3]:
  44. print(f" {m['key']}: {m['label']} ({m['unit']})")
  45. print("\n=== 2. Experience Stats ===")
  46. stats = compute_experience_stats(mock_cases)
  47. print(f"Total: {stats['total']}")
  48. print(f"Topology: {stats['topology_distribution']}")
  49. print(f"Avg rating: {stats['avg_rating']}")
  50. print(f"Param coverage: {stats['param_coverage']}")
  51. print(f"Metric ranges keys: {list(stats['metric_ranges'].keys())}")
  52. print("\n=== 3. Similar Case Search ===")
  53. target = {"Airgap": 1.0, "RMSCurrent": 21}
  54. similar = find_similar_cases(target, mock_cases, top_k=3, tolerance=0.5)
  55. print(f"Found {len(similar)} similar cases for {target}")
  56. for s in similar:
  57. print(f" Case #{s['id']}: similarity={s['similarity_score']}, shared={s['shared_params']}")
  58. print("\n=== 4. Trend Data ===")
  59. trend = compute_trend_data(mock_results, "Airgap", "tavg_nm")
  60. print(f"X: {trend['x_key']}, Y: {trend['y_key']}")
  61. print(f"Points: {trend['points']}")
  62. print(f"Stats: {trend['stats']}")
  63. print("\n=== 5. Pareto Frontier ===")
  64. pareto = compute_pareto_frontier(mock_results, x_metric="total_losses_w", y_metric="efficiency_pct")
  65. print(f"Total points: {pareto['total_count']}, Pareto points: {pareto['pareto_count']}")
  66. for p in pareto['pareto_points']:
  67. print(f" losses={p['x']}W, eff={p['y']}%")
  68. print("\n=== 6. Parameter Sensitivity ===")
  69. sens = compute_sensitivity(mock_results, "tavg_nm")
  70. print(f"Params analyzed: {len(sens)}")
  71. for s in sens:
  72. print(f" {s['param']}: corr={s['correlation']} ({s['direction']})")
  73. print("\n=== ALL TESTS PASSED ===")