Skills › Scientific Agent Skills › simpy › replication_runner.py
replication_runner.py
197 lines · skills/simpy/scripts/replication_runner.py at 5f2bc03
1#!/usr/bin/env python3
2"""Run independent bounded replications and compute Student-t intervals."""
4from __future__ import annotations
6import argparse
7from dataclasses import asdict, dataclass
8from typing import Any
10from _common import (
11 MAX_REPLICATIONS,
12 CliError,
13 emit_json,
14 finite_number,
15 integer,
16 load_json_object,
17 mean_confidence_interval,
18 validate_keys,
19)
20from basic_simulation_template import QueueConfig, run_simulation
23MAX_TOTAL_EVENT_BUDGET = 5_000_000
24MAX_TOTAL_ENTITY_BUDGET = 2_000_000
25EXPERIMENT_KEYS = {"confidence", "model", "replications"}
28@dataclass(frozen=True)
29class ExperimentConfig:
30 model: QueueConfig
31 replications: int = 20
32 confidence: float = 0.95
34 @classmethod
35 def from_mapping(cls, value: dict[str, Any] | None) -> "ExperimentConfig":
36 value = {} if value is None else dict(value)
37 validate_keys(
38 value, allowed=EXPERIMENT_KEYS, context="replication configuration"
39 )
40 model_value = value.get("model", {})
41 if not isinstance(model_value, dict):
42 raise CliError("model must be a JSON object")
43 model = QueueConfig.from_mapping(model_value)
44 replications = integer(
45 value.get("replications", 20),
46 name="replications",
47 minimum=2,
48 maximum=MAX_REPLICATIONS,
49 )
50 confidence = finite_number(
51 value.get("confidence", 0.95),
52 name="confidence",
53 minimum=0.50,
54 maximum=0.999,
55 minimum_inclusive=False,
56 )
57 if replications * model.max_events > MAX_TOTAL_EVENT_BUDGET:
58 raise CliError(
59 "replications * model.max_events exceeds the total event budget "
60 f"({MAX_TOTAL_EVENT_BUDGET})"
61 )
62 if replications * model.max_entities > MAX_TOTAL_ENTITY_BUDGET:
63 raise CliError(
64 "replications * model.max_entities exceeds the total entity budget "
65 f"({MAX_TOTAL_ENTITY_BUDGET})"
66 )
67 return cls(
68 model=model,
69 replications=replications,
70 confidence=confidence,
71 )
74def _interval_or_unavailable(
75 reports: list[dict[str, Any]],
76 *,
77 metric: str,
78 confidence: float,
79) -> dict[str, Any]:
80 values = [report["metrics"][metric] for report in reports]
81 missing = sum(value is None for value in values)
82 if missing:
83 return {
84 "missing_replications": missing,
85 "reason": "metric was undefined in one or more replications",
86 "status": "unavailable",
87 }
88 interval = mean_confidence_interval(values, confidence=confidence)
89 interval["status"] = "ok"
90 return interval
93def run_experiment(config: ExperimentConfig) -> dict[str, Any]:
94 """Run independent replication streams and summarize replication estimates."""
96 config = ExperimentConfig.from_mapping(asdict(config))
97 reports = [
98 run_simulation(config.model, replication=index)
99 for index in range(config.replications)
100 ]
101 metrics = (
102 "average_queue_length",
103 "average_system_time",
104 "average_wait",
105 "loss_probability",
106 "server_utilization",
107 "throughput_per_time_unit",
108 )
109 intervals = {
110 metric: _interval_or_unavailable(
111 reports, metric=metric, confidence=config.confidence
112 )
113 for metric in metrics
114 }
115 compact_runs = [
116 {
117 "counters": report["counters"],
118 "metrics": report["metrics"],
119 "replication": report["replication"],
120 "seed_manifest": report["seed_manifest"],
121 "warnings": report["warnings"],
122 }
123 for report in reports
124 ]
125 return {
126 "analysis": {
127 "analysis_mode": config.model.analysis_mode,
128 "confidence_method": (
129 "two-sided Student-t interval across independent "
130 "replication-level estimates"
131 ),
132 "confidence_level": config.confidence,
133 "independence_unit": "replication",
134 "single_run_interval": False,
135 "warm_up": config.model.warm_up,
136 },
137 "intervals": intervals,
138 "metric_windows": reports[0]["metric_windows"],
139 "simpy_version": reports[0]["simpy_version"],
140 "model": {
141 key: getattr(config.model, key)
142 for key in config.model.__dataclass_fields__
143 },
144 "replications": compact_runs,
145 "safety": {
146 "network_used": False,
147 "replication_limit": MAX_REPLICATIONS,
148 "total_entity_budget": MAX_TOTAL_ENTITY_BUDGET,
149 "total_event_budget": MAX_TOTAL_EVENT_BUDGET,
150 },
151 "schema_version": "1.2",
152 "warnings": [
153 "Intervals quantify Monte Carlo uncertainty under the configured model; "
154 "they do not validate the model or establish causality.",
155 "For steady-state analysis, warm-up and run length require a separate "
156 "transient-bias assessment and sensitivity analysis.",
157 "Do not treat within-run customer observations as independent replications.",
158 ],
159 }
162def build_parser() -> argparse.ArgumentParser:
163 parser = argparse.ArgumentParser(
164 description=(
165 "Run 2-1000 deterministic, independent queue replications from an "
166 "allowlisted local JSON configuration and report Student-t confidence "
167 "intervals across replication estimates."
168 )
169 )
170 parser.add_argument(
171 "--config",
172 help="Optional local .json experiment configuration; URLs/symlinks rejected",
173 )
174 parser.add_argument("--output", help="Optional local .json report")
175 parser.add_argument("--force", action="store_true", help="Replace explicit output")
176 return parser
179def main(argv: list[str] | None = None) -> int:
180 parser = build_parser()
181 args = parser.parse_args(argv)
182 try:
183 config = ExperimentConfig.from_mapping(
184 load_json_object(args.config) if args.config else None
185 )
186 emit_json(
187 run_experiment(config),
188 output=args.output,
189 force=args.force,
190 )
191 except CliError as exc:
192 parser.error(str(exc))
193 return 0
196if __name__ == "__main__":
197 raise SystemExit(main())