Skills › Scientific Agent Skills › simpy › basic_simulation_template.py
basic_simulation_template.py
440 lines · skills/simpy/scripts/basic_simulation_template.py at 5f2bc03
1#!/usr/bin/env python3
2"""Bounded, reproducible SimPy queue template with independent random streams."""
4from __future__ import annotations
6import argparse
7import random
8from dataclasses import asdict, dataclass, field
9from statistics import fmean
10from typing import TYPE_CHECKING, Any
12if TYPE_CHECKING:
13 import simpy
15from _common import (
16 DEFAULT_SEED,
17 MAX_ENTITIES,
18 MAX_EVENTS,
19 MAX_QUEUE_CAPACITY,
20 MAX_SIM_TIME,
21 CliError,
22 derive_seed,
23 emit_json,
24 finite_number,
25 integer,
26 load_json_object,
27 load_simpy,
28 validate_keys,
29)
30from resource_monitor import EventTraceRecorder, ResourceMonitor
33_SIMPY = load_simpy(required=False)
36CONFIG_KEYS = {
37 "analysis_mode",
38 "base_seed",
39 "horizon",
40 "max_entities",
41 "max_events",
42 "mean_interarrival",
43 "mean_service",
44 "queue_capacity",
45 "servers",
46 "warm_up",
47}
50class EventLimitExceeded(RuntimeError):
51 """Raised when a simulation attempts to process too many events."""
54if _SIMPY is not None:
56 class BoundedEnvironment(_SIMPY.Environment):
57 """Environment that enforces a public, explicit event-processing budget."""
59 def __init__(self, *, max_events: int):
60 super().__init__()
61 self.max_events = max_events
62 self.processed_events = 0
64 def step(self) -> None:
65 if self.peek() == float("inf"):
66 super().step() # Preserve EmptySchedule without consuming budget.
67 return
68 if self.processed_events >= self.max_events:
69 raise EventLimitExceeded(
70 f"simulation reached the max_events limit ({self.max_events})"
71 )
72 self.processed_events += 1
73 super().step() # Include the event whose callback stops run().
75else:
77 class BoundedEnvironment:
78 """Unavailable placeholder used only when SimPy is not installed."""
80 def __init__(self, *, max_events: int):
81 del max_events
82 load_simpy()
83 raise AssertionError("unreachable")
86@dataclass(frozen=True)
87class QueueConfig:
88 """Validated configuration for a finite-horizon multi-server queue."""
90 analysis_mode: str = "terminating"
91 horizon: float = 480.0
92 warm_up: float = 0.0
93 servers: int = 2
94 queue_capacity: int = 20
95 mean_interarrival: float = 4.0
96 mean_service: float = 6.0
97 base_seed: int = DEFAULT_SEED
98 max_entities: int = 10_000
99 max_events: int = 200_000
101 @classmethod
102 def from_mapping(cls, value: dict[str, Any] | None) -> "QueueConfig":
103 """Build a strict configuration; unknown JSON keys are rejected."""
105 value = {} if value is None else dict(value)
106 validate_keys(value, allowed=CONFIG_KEYS, context="queue configuration")
107 defaults = cls()
108 analysis_mode = value.get("analysis_mode", defaults.analysis_mode)
109 if analysis_mode not in {"terminating", "steady_state"}:
110 raise CliError(
111 "analysis_mode must be either 'terminating' or 'steady_state'"
112 )
113 horizon = finite_number(
114 value.get("horizon", defaults.horizon),
115 name="horizon",
116 minimum=0.0,
117 maximum=MAX_SIM_TIME,
118 minimum_inclusive=False,
119 )
120 warm_up = finite_number(
121 value.get("warm_up", defaults.warm_up),
122 name="warm_up",
123 minimum=0.0,
124 maximum=horizon,
125 )
126 if warm_up >= horizon:
127 raise CliError("warm_up must be strictly less than horizon")
128 if analysis_mode == "terminating" and warm_up != 0:
129 raise CliError("terminating analyses must use warm_up=0")
130 if analysis_mode == "steady_state" and warm_up <= 0:
131 raise CliError("steady_state analyses require a positive warm_up")
132 return cls(
133 analysis_mode=analysis_mode,
134 horizon=horizon,
135 warm_up=warm_up,
136 servers=integer(
137 value.get("servers", defaults.servers),
138 name="servers",
139 minimum=1,
140 maximum=10_000,
141 ),
142 queue_capacity=integer(
143 value.get("queue_capacity", defaults.queue_capacity),
144 name="queue_capacity",
145 minimum=0,
146 maximum=MAX_QUEUE_CAPACITY,
147 ),
148 mean_interarrival=finite_number(
149 value.get("mean_interarrival", defaults.mean_interarrival),
150 name="mean_interarrival",
151 minimum=0.0,
152 maximum=MAX_SIM_TIME,
153 minimum_inclusive=False,
154 ),
155 mean_service=finite_number(
156 value.get("mean_service", defaults.mean_service),
157 name="mean_service",
158 minimum=0.0,
159 maximum=MAX_SIM_TIME,
160 minimum_inclusive=False,
161 ),
162 base_seed=integer(
163 value.get("base_seed", defaults.base_seed),
164 name="base_seed",
165 minimum=0,
166 maximum=2**63 - 1,
167 ),
168 max_entities=integer(
169 value.get("max_entities", defaults.max_entities),
170 name="max_entities",
171 minimum=1,
172 maximum=MAX_ENTITIES,
173 ),
174 max_events=integer(
175 value.get("max_events", defaults.max_events),
176 name="max_events",
177 minimum=10,
178 maximum=MAX_EVENTS,
179 ),
180 )
183@dataclass
184class QueueStatistics:
185 """Per-run counters and post-warm-up observations."""
187 arrivals: int = 0
188 admitted: int = 0
189 rejected: int = 0
190 completed: int = 0
191 observed_completed: int = 0
192 window_arrivals: int = 0
193 window_rejected: int = 0
194 window_departures: int = 0
195 wait_times: list[float] = field(default_factory=list)
196 service_times: list[float] = field(default_factory=list)
197 system_times: list[float] = field(default_factory=list)
198 entity_limit_hit: bool = False
201def _optional_mean(values: list[float]) -> float | None:
202 return fmean(values) if values else None
205def _customer(
206 env: BoundedEnvironment,
207 *,
208 resource: simpy.Resource,
209 stats: QueueStatistics,
210 service_rng: random.Random,
211 mean_service: float,
212 arrival_time: float,
213 warm_up: float,
214) -> Any:
215 """Request one server, receive service, and record a completed observation."""
217 with resource.request() as request:
218 yield request
219 service_start = env.now
220 service_time = service_rng.expovariate(1.0 / mean_service)
221 yield env.timeout(service_time)
222 stats.completed += 1
223 if env.now >= warm_up:
224 stats.window_departures += 1
225 if arrival_time >= warm_up:
226 stats.observed_completed += 1
227 stats.wait_times.append(service_start - arrival_time)
228 stats.service_times.append(service_time)
229 stats.system_times.append(env.now - arrival_time)
231def _arrivals(
232 env: BoundedEnvironment,
233 *,
234 resource: simpy.Resource,
235 config: QueueConfig,
236 stats: QueueStatistics,
237 arrival_rng: random.Random,
238 service_rng: random.Random,
239) -> Any:
240 """Generate at most max_entities arrivals strictly before the horizon."""
242 while stats.arrivals < config.max_entities:
243 delay = arrival_rng.expovariate(1.0 / config.mean_interarrival)
244 if env.now + delay >= config.horizon:
245 return
246 yield env.timeout(delay)
247 stats.arrivals += 1
248 if env.now >= config.warm_up:
249 stats.window_arrivals += 1
250 system_capacity = config.servers + config.queue_capacity
251 if resource.count + len(resource.queue) >= system_capacity:
252 stats.rejected += 1
253 if env.now >= config.warm_up:
254 stats.window_rejected += 1
255 continue
256 stats.admitted += 1
257 env.process(
258 _customer(
259 env,
260 resource=resource,
261 stats=stats,
262 service_rng=service_rng,
263 mean_service=config.mean_service,
264 arrival_time=env.now,
265 warm_up=config.warm_up,
266 )
267 )
268 stats.entity_limit_hit = True
271def run_simulation(
272 config: QueueConfig,
273 *,
274 replication: int = 0,
275 trace_max_records: int | None = None,
276) -> dict[str, Any]:
277 """Run one bounded replication and return a strict-JSON-compatible report."""
279 # Dataclasses can be constructed directly; library callers need the same
280 # validation and execution bounds as JSON callers.
281 config = QueueConfig.from_mapping(asdict(config))
282 simpy_module = load_simpy()
283 replication = integer(
284 replication,
285 name="replication",
286 minimum=0,
287 maximum=999,
288 )
289 arrival_seed = derive_seed(config.base_seed, replication, "arrivals")
290 service_seed = derive_seed(config.base_seed, replication, "service")
291 arrival_rng = random.Random(arrival_seed)
292 service_rng = random.Random(service_seed)
293 env = BoundedEnvironment(max_events=config.max_events)
294 trace = (
295 EventTraceRecorder(env, max_records=trace_max_records)
296 if trace_max_records is not None
297 else None
298 )
299 resource = simpy_module.Resource(env, capacity=config.servers)
300 monitor = ResourceMonitor(env, resource, name="server")
301 stats = QueueStatistics()
302 env.process(
303 _arrivals(
304 env,
305 resource=resource,
306 config=config,
307 stats=stats,
308 arrival_rng=arrival_rng,
309 service_rng=service_rng,
310 )
311 )
312 try:
313 env.run(until=config.horizon)
314 except EventLimitExceeded as exc:
315 raise CliError(str(exc)) from exc
316 if trace is not None:
317 trace.detach()
318 monitor.finalize(at=config.horizon)
320 observed_duration = config.horizon - config.warm_up
321 warnings: list[str] = []
322 unfinished = stats.admitted - stats.completed
323 if stats.entity_limit_hit:
324 warnings.append(
325 "max_entities was reached before the horizon; arrivals were truncated"
326 )
327 if unfinished:
328 warnings.append(
329 "some admitted entities were unfinished at the horizon; completed-customer "
330 "metrics exclude those right-censored paths"
331 )
332 if stats.observed_completed == 0:
333 warnings.append("no completed entities were observed in the analysis window")
334 if config.analysis_mode == "steady_state":
335 warnings.append(
336 "warm-up deletion is user-specified and does not prove steady state"
337 )
339 loss_probability = (
340 stats.window_rejected / stats.window_arrivals if stats.window_arrivals else None
341 )
342 report = {
343 "config": asdict(config),
344 "counters": {
345 "admitted": stats.admitted,
346 "arrivals": stats.arrivals,
347 "completed": stats.completed,
348 "observed_completed": stats.observed_completed,
349 "rejected": stats.rejected,
350 "unfinished_at_horizon": unfinished,
351 "window_arrivals": stats.window_arrivals,
352 "window_rejected": stats.window_rejected,
353 "window_departures": stats.window_departures,
354 },
355 "event_processing": {
356 "limit": config.max_events,
357 "processed": env.processed_events,
358 },
359 "metrics": {
360 "average_queue_length": monitor.average_queue_length(
361 start=config.warm_up, end=config.horizon
362 ),
363 "average_system_time": _optional_mean(stats.system_times),
364 "average_wait": _optional_mean(stats.wait_times),
365 "loss_probability": loss_probability,
366 "mean_service_time": _optional_mean(stats.service_times),
367 "server_utilization": monitor.average_utilization(
368 start=config.warm_up, end=config.horizon
369 ),
370 "throughput_per_time_unit": (
371 stats.window_departures / observed_duration
372 ),
373 },
374 "replication": replication,
375 "metric_windows": {
376 "time_weighted": "[warm_up, horizon)",
377 "throughput": "departures in [warm_up, horizon), regardless of arrival time",
378 "loss_probability": "rejected / all arrivals in [warm_up, horizon)",
379 "customer_means": "arrived at/after warm_up and completed before horizon",
380 },
381 "schema_version": "1.2",
382 "simpy_version": simpy_module.__version__,
383 "seed_manifest": {
384 "algorithm": "BLAKE2b-derived Python random.Random streams",
385 "arrival_seed": arrival_seed,
386 "base_seed": config.base_seed,
387 "service_seed": service_seed,
388 },
389 "warnings": warnings,
390 }
391 if trace is not None:
392 report["event_trace"] = {
393 "records": trace.records,
394 "truncated": trace.truncated,
395 }
396 return report
399def build_parser() -> argparse.ArgumentParser:
400 parser = argparse.ArgumentParser(
401 description=(
402 "Run a deterministic, finite-horizon queue template. JSON input is "
403 "allowlisted and all time, event, entity, and queue bounds are enforced."
404 )
405 )
406 parser.add_argument(
407 "--config",
408 help="Optional local .json queue configuration; URLs and symlinks are rejected",
409 )
410 parser.add_argument(
411 "--replication",
412 type=int,
413 default=0,
414 help="Deterministic replication index from 0 through 999 (default: 0)",
415 )
416 parser.add_argument("--output", help="Optional local .json report")
417 parser.add_argument("--force", action="store_true", help="Replace explicit output")
418 return parser
421def main(argv: list[str] | None = None) -> int:
422 parser = build_parser()
423 args = parser.parse_args(argv)
424 try:
425 config = QueueConfig.from_mapping(
426 load_json_object(args.config) if args.config else None
427 )
428 report = run_simulation(config, replication=args.replication)
429 report["source"] = {
430 "config": None if args.config is None else "local_json",
431 "network_used": False,
432 }
433 emit_json(report, output=args.output, force=args.force)
434 except CliError as exc:
435 parser.error(str(exc))
436 return 0
439if __name__ == "__main__":
440 raise SystemExit(main())