"""Test analytics service logic without database.""" import sys sys.path.insert(0, '.') from app.services.analytics import ( compute_experience_stats, find_similar_cases, compute_trend_data, compute_pareto_frontier, compute_sensitivity, get_metric_defs, ) # Test data mock_cases = [ { "id": 1, "topology": "SSSR", "source_plan_id": "plan-001", "params": {"Airgap": 1.0, "RMSCurrent": 21, "MagnetThickness": 5}, "metrics": {"tavg_nm": 2.5, "efficiency_pct": 85.2, "ripple_pct": 3.5}, "conclusion": "Good baseline", "tags": ["baseline", "sssr"], "rating": 4, }, { "id": 2, "topology": "SSSR", "source_plan_id": "plan-002", "params": {"Airgap": 1.2, "RMSCurrent": 21, "MagnetThickness": 5}, "metrics": {"tavg_nm": 2.3, "efficiency_pct": 84.8, "ripple_pct": 3.2}, "conclusion": "Larger airgap reduces torque", "tags": ["airgap-study"], "rating": 3, }, { "id": 3, "topology": "DRSS", "source_plan_id": "plan-003", "params": {"Airgap": 1.0, "RMSCurrent": 25, "MagnetThickness": 6}, "metrics": {"tavg_nm": 3.1, "efficiency_pct": 87.5, "ripple_pct": 2.8}, "conclusion": "DRSS higher torque", "tags": ["drss", "high-torque"], "rating": 5, }, ] mock_results = [ {"status": "OK", "params": {"Airgap": 0.8, "RMSCurrent": 20}, "metrics": {"tavg_nm": 2.8, "efficiency_pct": 86.0, "total_losses_w": 50}}, {"status": "OK", "params": {"Airgap": 1.0, "RMSCurrent": 20}, "metrics": {"tavg_nm": 2.5, "efficiency_pct": 85.2, "total_losses_w": 48}}, {"status": "OK", "params": {"Airgap": 1.2, "RMSCurrent": 20}, "metrics": {"tavg_nm": 2.3, "efficiency_pct": 84.8, "total_losses_w": 45}}, {"status": "OK", "params": {"Airgap": 1.0, "RMSCurrent": 25}, "metrics": {"tavg_nm": 3.0, "efficiency_pct": 86.5, "total_losses_w": 55}}, {"status": "FAILED", "params": {"Airgap": 1.5, "RMSCurrent": 20}, "metrics": {}, "error_message": "convergence failed"}, ] print("=== 1. Metric Definitions ===") metrics = get_metric_defs() print(f"Total metrics: {len(metrics)}") for m in metrics[:3]: print(f" {m['key']}: {m['label']} ({m['unit']})") print("\n=== 2. Experience Stats ===") stats = compute_experience_stats(mock_cases) print(f"Total: {stats['total']}") print(f"Topology: {stats['topology_distribution']}") print(f"Avg rating: {stats['avg_rating']}") print(f"Param coverage: {stats['param_coverage']}") print(f"Metric ranges keys: {list(stats['metric_ranges'].keys())}") print("\n=== 3. Similar Case Search ===") target = {"Airgap": 1.0, "RMSCurrent": 21} similar = find_similar_cases(target, mock_cases, top_k=3, tolerance=0.5) print(f"Found {len(similar)} similar cases for {target}") for s in similar: print(f" Case #{s['id']}: similarity={s['similarity_score']}, shared={s['shared_params']}") print("\n=== 4. Trend Data ===") trend = compute_trend_data(mock_results, "Airgap", "tavg_nm") print(f"X: {trend['x_key']}, Y: {trend['y_key']}") print(f"Points: {trend['points']}") print(f"Stats: {trend['stats']}") print("\n=== 5. Pareto Frontier ===") pareto = compute_pareto_frontier(mock_results, x_metric="total_losses_w", y_metric="efficiency_pct") print(f"Total points: {pareto['total_count']}, Pareto points: {pareto['pareto_count']}") for p in pareto['pareto_points']: print(f" losses={p['x']}W, eff={p['y']}%") print("\n=== 6. Parameter Sensitivity ===") sens = compute_sensitivity(mock_results, "tavg_nm") print(f"Params analyzed: {len(sens)}") for s in sens: print(f" {s['param']}: corr={s['correlation']} ({s['direction']})") print("\n=== ALL TESTS PASSED ===")