sgeboers/motief
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(diagnostics): enhance trajectory diagnostic script with real data mode

Browse Source
This commit is contained in:
sgeboers
2026-04-01 01:59:41 +02:00
parent 7e202e15be
commit c9c59dd166
1 changed files with 156 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/diagnose_trajectories_cli.py
+142 -26
View File
@@ -1,10 +1,13 @@
#!/usr/bin/env python3
# HL REV:NEWFILE
"""
Automated probes for Trajectories tab diagnostics.
This script runs several simulated scenarios by monkeypatching explorer.load_positions,
explorer.load_party_map and explorer.select_trajectory_plot_data to reproduce common
failure modes that lead to "no plot at all" and prints the module-level diagnostics.
This script runs several simulated scenarios to diagnose the trajectories tab:
1. NORMAL: Uses real data from data/motions.db
2. EMPTY_POSITIONS: Simulates load_positions returning empty (realistic failure mode)
3. EMPTY_PARTY_MAP: Simulates load_party_map returning empty (realistic failure mode)
4. BOTH_EMPTY: Simulates both returning empty (severe failure mode)
Run: python scripts/diagnose_trajectories_cli.py
"""
@@ -13,6 +16,13 @@ import os
import importlib
import traceback
import sys
import json
from datetime import datetime
# Real DB path relative to project root
DB_PATH = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "data", "motions.db")
)
def run():
@@ -21,26 +31,76 @@ def run():
root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
if root not in sys.path:
sys.path.insert(0, root)
# Import explorer fresh so env var reads take effect
import explorer
# Results collection for all scenarios
all_results = {
"generated_at": datetime.now().isoformat(),
"db_path": DB_PATH,
"db_exists": os.path.exists(DB_PATH),
"scenarios": {},
}
def run_scenario(
name,
load_positions_ret=None,
load_party_map_ret=None,
select_helper_behavior=None,
use_real_data=False,
):
print("\n" + "=" * 80)
print("SCENARIO:", name)
print("=" * 80)
scenario_result = {
"name": name,
"use_real_data": use_real_data,
"monkeypatched": {},
}
# Clear previous diagnostics
explorer._last_trajectories_diagnostics.clear()
# Backup originals
orig_load_positions = getattr(explorer, "load_positions", None)
orig_load_party_map = getattr(explorer, "load_party_map", None)
orig_select_helper = getattr(explorer, "select_trajectory_plot_data", None)
# If using real data, call the real functions once to get actual values
if use_real_data and os.path.exists(DB_PATH):
print(f"Loading real data from: {DB_PATH}")
try:
real_positions = explorer.load_positions(DB_PATH, "annual")
real_party_map = explorer.load_party_map(DB_PATH)
print(
f" Real load_positions: {len(real_positions[0]) if real_positions[0] else 0} windows"
)
print(f" Real party_map: {len(real_party_map)} entries")
scenario_result["real_data"] = {
"positions_windows": len(real_positions[0])
if real_positions[0]
else 0,
"party_map_count": len(real_party_map),
}
except Exception as e:
print(f" ERROR loading real data: {e}")
scenario_result["real_data_error"] = str(e)
if load_positions_ret is not None:
explorer.load_positions = lambda db, ws: load_positions_ret
scenario_result["monkeypatched"]["load_positions"] = "ARTIFICIAL_EMPTY"
print(
" Monkeypatched: load_positions -> ARTIFICIAL_EMPTY (for comparison)"
)
if load_party_map_ret is not None:
explorer.load_party_map = lambda db: load_party_map_ret
scenario_result["monkeypatched"]["load_party_map"] = "ARTIFICIAL_EMPTY"
print(
" Monkeypatched: load_party_map -> ARTIFICIAL_EMPTY (for comparison)"
)
if select_helper_behavior == "raise":
@@ -48,6 +108,9 @@ def run():
raise ValueError("simulated crash from select_trajectory_plot_data")
explorer.select_trajectory_plot_data = raising
scenario_result["monkeypatched"]["select_helper"] = "RAISE_ERROR"
print(" Monkeypatched: select_trajectory_plot_data -> RAISE_ERROR")
elif select_helper_behavior == "zero_traces":
class DummyFig:
@@ -58,24 +121,32 @@ def run():
return DummyFig(), 0, None
explorer.select_trajectory_plot_data = zero
scenario_result["monkeypatched"]["select_helper"] = "ZERO_TRACES"
print(" Monkeypatched: select_trajectory_plot_data -> ZERO_TRACES")
try:
# Call the UI function; it's import-safe and uses a dummy st when streamlit is absent
explorer.build_trajectories_tab(db_path="dummy", window_size=1)
explorer.build_trajectories_tab(db_path=DB_PATH, window_size="annual")
except Exception as e:
print("build_trajectories_tab RAISED:", type(e), e)
print(traceback.format_exc())
finally:
scenario_result["exception"] = str(e)
scenario_result["traceback"] = traceback.format_exc()
# Capture diagnostics
diag = getattr(explorer, "_last_trajectories_diagnostics", None)
scenario_result["diagnostics"] = dict(diag) if diag else {}
print("module _last_trajectories_diagnostics:", diag)
sh = None
if hasattr(explorer, "select_trajectory_plot_data"):
sh = getattr(
explorer.select_trajectory_plot_data, "_last_diagnostics", None
)
scenario_result["select_helper_diagnostics"] = dict(sh) if sh else {}
print("select_helper _last_diagnostics:", sh)
# restore
# restore originals
if orig_load_positions is not None:
explorer.load_positions = orig_load_positions
if orig_load_party_map is not None:
@@ -83,36 +154,81 @@ def run():
if orig_select_helper is not None:
explorer.select_trajectory_plot_data = orig_select_helper
# Scenario 1: load_positions returns empty
all_results["scenarios"][name] = scenario_result
# Scenario 1: NORMAL - Uses real data
print("\n" + "=" * 80)
print("SCENARIO 1: NORMAL (using real data)")
print("=" * 80)
run_scenario("normal", use_real_data=True)
# Scenario 2: Empty positions (ARTIFICIAL - for comparison)
print("\n" + "=" * 80)
print("SCENARIO 2: EMPTY_POSITIONS (ARTIFICIAL - for comparison)")
print("=" * 80)
run_scenario(
"load_positions_empty", load_positions_ret=({}, None), load_party_map_ret={}
"empty_positions_ARTIFICIAL",
load_positions_ret=({}, {}),
load_party_map_ret=None, # Use real party_map
)
# Scenario 2: positions present but MP coords malformed -> mp_positions empty
positions_malformed = {"W1": {"mp1": ("bad", "bad")}}
# Scenario 3: Empty party_map (ARTIFICIAL - for comparison)
print("\n" + "=" * 80)
print("SCENARIO 3: EMPTY_PARTY_MAP (ARTIFICIAL - for comparison)")
print("=" * 80)
run_scenario(
"mp_positions_malformed",
load_positions_ret=(positions_malformed, {}),
"empty_party_map_ARTIFICIAL",
load_positions_ret=None, # Use real positions
load_party_map_ret={},
)
# Scenario 3: select_trajectory_plot_data raises an exception
positions_valid = {"W1": {"mp1": (0.1, 0.2)}}
# Scenario 4: Both empty (ARTIFICIAL - for comparison)
print("\n" + "=" * 80)
print("SCENARIO 4: BOTH_EMPTY (ARTIFICIAL - for comparison)")
print("=" * 80)
run_scenario(
"select_helper_raise",
load_positions_ret=(positions_valid, {}),
load_party_map_ret={},
select_helper_behavior="raise",
"both_empty_ARTIFICIAL", load_positions_ret=({}, {}), load_party_map_ret={}
)
# Scenario 4: helper returns a fig with zero traces
run_scenario(
"helper_zero_traces",
load_positions_ret=(positions_valid, {}),
load_party_map_ret={},
select_helper_behavior="zero_traces",
)
# Write results to diagnostics file
output_dir = os.path.join(root, "thoughts", "shared", "diagnostics")
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "2026-03-31-trajectories-diagnostics.json")
with open(output_path, "w") as f:
json.dump(all_results, f, indent=2, default=str)
print("\n" + "=" * 80)
print(f"DIAGNOSTICS SAVED TO: {output_path}")
print("=" * 80)
# Summary
print("\n" + "=" * 80)
print("SUMMARY")
print("=" * 80)
normal_scenario = all_results["scenarios"].get("normal", {})
real_data = normal_scenario.get("real_data", {})
party_map_count = real_data.get("party_map_count", 0)
print(f"DB Path: {DB_PATH}")
print(f"DB Exists: {all_results['db_exists']}")
print(f"Real data party_map_count: {party_map_count}")
if party_map_count > 0:
print("\n✅ SUCCESS: Real data scenario shows party_map_count > 0")
print(f" Found {party_map_count} party entries (expected ~1000+)")
else:
print("\n❌ ISSUE: Real data scenario shows party_map_count = 0")
# Check diagnostics from normal scenario
normal_diag = normal_scenario.get("diagnostics", {})
if normal_diag:
print(f"\nNormal scenario diagnostics keys: {list(normal_diag.keys())}")
print(f"Normal scenario stage: {normal_diag.get('stage', 'N/A')}")
return all_results
if __name__ == "__main__":
run()
results = run()