Spaces:
Running
Running
File size: 4,122 Bytes
b43adf7 6a28f91 b43adf7 f163245 b43adf7 f163245 b43adf7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 |
from __future__ import annotations
from pathlib import Path
from datetime import date
from cli.config import SimulateConfig
from src.data.case_generator import CaseGenerator
from src.simulation.engine import CourtSim, CourtSimConfig
from src.core.case import CaseStatus
from src.metrics.basic import gini
def merge_simulation_config(
default_cfg: SimulateConfig,
cases_path: str,
days: int,
start_date: date | None,
policy: str,
seed: int,
log_dir: str,
) -> SimulateConfig:
"""Merge UI inputs with default simulation config."""
return SimulateConfig(
cases=Path(cases_path) if cases_path else default_cfg.cases,
days=days or default_cfg.days,
start=start_date or default_cfg.start,
policy=policy or default_cfg.policy,
seed=seed if seed is not None else default_cfg.seed,
log_dir=Path(log_dir) if log_dir else default_cfg.log_dir,
)
def run_simulation_dashboard(scfg: SimulateConfig, run_dir: Path):
"""
Execute simulation based on the provided Streamlit configuration.
"""
# ------------------------------------------------------------------
# Load case data
# ------------------------------------------------------------------
path = scfg.cases
if path.exists():
cases = CaseGenerator.from_csv(path)
start = scfg.start or (
max(c.filed_date for c in cases) if cases else date.today()
)
else:
# Fallback (CLI fallback behaviour)
start = scfg.start or date.today().replace(day=1)
gen = CaseGenerator(start=start, end=start.replace(day=28), seed=scfg.seed)
cases = gen.generate(n_cases=5 * 151)
# ------------------------------------------------------------------
# Build CourtSimConfig
# ------------------------------------------------------------------
cfg = CourtSimConfig(
start=start,
days=scfg.days,
seed=scfg.seed,
policy=scfg.policy,
duration_percentile=scfg.duration_percentile,
log_dir=run_dir,
)
# ------------------------------------------------------------------
# Run simulation
# ------------------------------------------------------------------
sim = CourtSim(cfg, cases)
res = sim.run()
# ------------------------------------------------------------------
# Collect metrics exactly like CLI
# ------------------------------------------------------------------
disp_times = [
(c.disposal_date - c.filed_date).days
for c in cases
if c.disposal_date is not None and c.status == CaseStatus.DISPOSED
]
gini_disp = gini(disp_times) if disp_times else 0.0
summary_text = f"""
Simulation Complete!
Horizon: {cfg.start} -> {res.end_date} ({cfg.days} days)
Hearing Metrics:
Total: {res.hearings_total}
Heard: {res.hearings_heard} ({res.hearings_heard / max(1, res.hearings_total):.1%})
Adjourned: {res.hearings_adjourned} ({res.hearings_adjourned / max(1, res.hearings_total):.1%})
Disposal Metrics:
Disposed: {res.disposals} ({res.disposals / len(cases):.1%})
Gini coefficient: {gini_disp:.3f}
Efficiency:
Utilization: {res.utilization:.2%}
Avg hearings/day: {res.hearings_total / max(1, cfg.days):.2f}
"""
# Merge engine insights into report.txt
insights_text = (getattr(res, "insights_text", "") or "").strip()
if insights_text:
full_report = summary_text.rstrip() + "\n\n" + insights_text + "\n"
else:
full_report = summary_text
(run_dir / "report.txt").write_text(full_report, encoding="utf-8")
# -------------------------------------------------------
# Locate generated CSVs (if they exist)
# -------------------------------------------------------
metrics_path = run_dir / "metrics.csv"
events_path = run_dir / "events.csv"
return {
"summary": summary_text,
"insights": insights_text,
"end_date": res.end_date,
"metrics_path": metrics_path if metrics_path.exists() else None,
"events_path": events_path if events_path.exists() else None,
}
|