Skills › Scientific Agent Skills › simpy › resource_monitor.py
resource_monitor.py
489 lines · skills/simpy/scripts/resource_monitor.py at 5f2bc03
1#!/usr/bin/env python3
2"""Bounded resource monitoring and deterministic event tracing for SimPy."""
4from __future__ import annotations
6import argparse
7import csv
8import io
9import json
10from dataclasses import asdict, dataclass
11from functools import wraps
12from typing import TYPE_CHECKING, Any, Callable
14if TYPE_CHECKING:
15 import simpy
17from _common import (
18 MAX_TRACE_RECORDS,
19 CliError,
20 emit_json,
21 emit_text,
22 finite_number,
23 integer,
24 load_simpy,
25)
28@dataclass(frozen=True)
29class ResourceSample:
30 """One resource state observed at a simulation timestamp."""
32 time: float
33 event: str
34 count: int
35 queue_length: int
36 utilization: float
39class ResourceMonitor:
40 """Monitor one Resource/PriorityResource/PreemptiveResource instance."""
42 def __init__(
43 self,
44 env: simpy.Environment,
45 resource: simpy.Resource,
46 name: str = "resource",
47 ):
48 simpy_module = load_simpy()
49 if not isinstance(
50 resource,
51 (
52 simpy_module.Resource,
53 simpy_module.PriorityResource,
54 simpy_module.PreemptiveResource,
55 ),
56 ):
57 raise CliError("ResourceMonitor requires a SimPy Resource-like object")
58 if getattr(resource, "_simpy_skill_monitor_attached", False):
59 raise CliError("this resource already has a ResourceMonitor")
60 self.env = env
61 self.resource = resource
62 self.name = name
63 self.samples: list[ResourceSample] = []
64 self.wait_times: list[float] = []
65 self._request_times: dict[Any, float] = {}
66 self._original_request = resource.request
67 self._original_release = resource.release
68 setattr(resource, "_simpy_skill_monitor_attached", True)
69 self._record("initial")
70 self._patch()
72 def _record(self, event: str, *, at: float | None = None) -> None:
73 time = float(self.env.now if at is None else at)
74 self.samples.append(
75 ResourceSample(
76 time=time,
77 event=event,
78 count=int(self.resource.count),
79 queue_length=len(self.resource.queue),
80 utilization=float(self.resource.count / self.resource.capacity),
81 )
82 )
84 def _patch(self) -> None:
85 @wraps(self._original_request)
86 def monitored_request(*args: Any, **kwargs: Any) -> Any:
87 requested_at = float(self.env.now)
88 request = self._original_request(*args, **kwargs)
89 self._request_times[request] = requested_at
90 self._record("request")
92 def granted(_: Any) -> None:
93 start = self._request_times.pop(request, None)
94 if start is not None:
95 self.wait_times.append(float(self.env.now) - start)
96 self._record("grant")
98 if request.callbacks is not None:
99 request.callbacks.append(granted)
101 original_cancel = request.cancel
103 @wraps(original_cancel)
104 def monitored_cancel() -> None:
105 original_cancel()
106 self._request_times.pop(request, None)
107 self._record("cancel")
109 try:
110 request.cancel = monitored_cancel
111 except (AttributeError, TypeError):
112 # Current SimPy events allow instance wrappers. If a future release
113 # changes that implementation, later state samples remain valid but
114 # cancellation instants may need explicit user instrumentation.
115 pass
116 return request
118 @wraps(self._original_release)
119 def monitored_release(*args: Any, **kwargs: Any) -> Any:
120 released = self._original_release(*args, **kwargs)
121 self._record("release")
122 return released
124 self.resource.request = monitored_request
125 self.resource.release = monitored_release
127 def detach(self) -> None:
128 """Restore the resource methods patched by this monitor."""
130 self.resource.request = self._original_request
131 self.resource.release = self._original_release
132 setattr(self.resource, "_simpy_skill_monitor_attached", False)
134 def finalize(self, *, at: float | None = None) -> None:
135 """Close the final time interval with the resource's current state."""
137 final_time = float(self.env.now if at is None else at)
138 finite_number(final_time, name="monitor final time")
139 if final_time < self.samples[-1].time:
140 raise CliError("monitor final time cannot precede its latest sample")
141 self._record("final", at=final_time)
143 def _time_weighted(
144 self,
145 attribute: str,
146 *,
147 start: float,
148 end: float,
149 ) -> float:
150 finite_number(start, name="monitor window start")
151 finite_number(end, name="monitor window end")
152 if start < self.samples[0].time:
153 raise CliError("monitor window starts before monitoring began")
154 if end <= start:
155 raise CliError("monitor window end must be greater than start")
156 if end > self.samples[-1].time:
157 raise CliError("call finalize() through the requested window end first")
159 current = self.samples[0]
160 for sample in self.samples:
161 if sample.time <= start:
162 current = sample
163 else:
164 break
165 cursor = start
166 weighted = 0.0
167 for sample in self.samples:
168 if sample.time <= start:
169 continue
170 if sample.time >= end:
171 break
172 weighted += float(getattr(current, attribute)) * (sample.time - cursor)
173 cursor = sample.time
174 current = sample
175 weighted += float(getattr(current, attribute)) * (end - cursor)
176 return weighted / (end - start)
178 def average_queue_length(
179 self, *, start: float = 0.0, end: float | None = None
180 ) -> float:
181 """Return the time-weighted queue length over [start, end)."""
183 final = self.samples[-1].time if end is None else float(end)
184 return self._time_weighted(
185 "queue_length", start=float(start), end=final
186 )
188 def average_utilization(
189 self, *, start: float = 0.0, end: float | None = None
190 ) -> float:
191 """Return time-weighted allocated capacity over [start, end)."""
193 final = self.samples[-1].time if end is None else float(end)
194 return self._time_weighted("utilization", start=float(start), end=final)
196 def summary(
197 self, *, start: float = 0.0, end: float | None = None
198 ) -> dict[str, Any]:
199 """Return a JSON-compatible summary of this monitor."""
201 final = self.samples[-1].time if end is None else float(end)
202 return {
203 "average_queue_length": self.average_queue_length(
204 start=start, end=final
205 ),
206 "average_utilization": self.average_utilization(
207 start=start, end=final
208 ),
209 "capacity": self.resource.capacity,
210 "end": final,
211 "final_count": self.resource.count,
212 "final_queue_length": len(self.resource.queue),
213 "granted_requests": len(self.wait_times),
214 "maximum_queue_length": max(
215 sample.queue_length for sample in self.samples
216 ),
217 "mean_wait": (
218 sum(self.wait_times) / len(self.wait_times)
219 if self.wait_times
220 else None
221 ),
222 "name": self.name,
223 "pending_requests": len(self._request_times),
224 "samples": len(self.samples),
225 "start": start,
226 }
228 def export_csv(self, output: str, *, force: bool = False) -> None:
229 """Write state samples to a bounded, private local CSV."""
231 buffer = io.StringIO(newline="")
232 writer = csv.DictWriter(
233 buffer,
234 fieldnames=[
235 "time",
236 "event",
237 "count",
238 "queue_length",
239 "utilization",
240 ],
241 )
242 writer.writeheader()
243 for sample in self.samples:
244 writer.writerow(asdict(sample))
245 emit_text(buffer.getvalue(), output=output, suffixes={".csv"}, force=force)
248class MultiResourceMonitor:
249 """Manage uniquely named ResourceMonitor instances in one environment."""
251 def __init__(self, env: simpy.Environment):
252 load_simpy()
253 self.env = env
254 self.monitors: dict[str, ResourceMonitor] = {}
256 def add_resource(
257 self, resource: simpy.Resource, name: str
258 ) -> ResourceMonitor:
259 if not name or name in self.monitors:
260 raise CliError("resource monitor names must be non-empty and unique")
261 monitor = ResourceMonitor(self.env, resource, name)
262 self.monitors[name] = monitor
263 return monitor
265 def finalize(self, *, at: float | None = None) -> None:
266 for monitor in self.monitors.values():
267 monitor.finalize(at=at)
269 def summaries(
270 self, *, start: float = 0.0, end: float | None = None
271 ) -> dict[str, dict[str, Any]]:
272 return {
273 name: monitor.summary(start=start, end=end)
274 for name, monitor in self.monitors.items()
275 }
278class ContainerMonitor:
279 """Record completed put/get operations and time-weighted Container level."""
281 def __init__(
282 self,
283 env: simpy.Environment,
284 container: simpy.Container,
285 name: str = "container",
286 ):
287 load_simpy()
288 if getattr(container, "_simpy_skill_monitor_attached", False):
289 raise CliError("this container already has a ContainerMonitor")
290 self.env = env
291 self.container = container
292 self.name = name
293 self.samples: list[tuple[float, str, float]] = [
294 (float(env.now), "initial", float(container.level))
295 ]
296 self._original_put = container.put
297 self._original_get = container.get
298 setattr(container, "_simpy_skill_monitor_attached", True)
299 self._patch()
301 def _record(self, event: str, *, at: float | None = None) -> None:
302 self.samples.append(
303 (
304 float(self.env.now if at is None else at),
305 event,
306 float(self.container.level),
307 )
308 )
310 def _patch_operation(
311 self, operation: Callable[..., Any], event_name: str
312 ) -> Callable[..., Any]:
313 @wraps(operation)
314 def wrapper(*args: Any, **kwargs: Any) -> Any:
315 event = operation(*args, **kwargs)
316 self._record(event_name + "_call")
317 if event.callbacks is not None:
318 event.callbacks.append(lambda _: self._record(event_name))
319 return event
321 return wrapper
323 def _patch(self) -> None:
324 self.container.put = self._patch_operation(self._original_put, "put")
325 self.container.get = self._patch_operation(self._original_get, "get")
327 def finalize(self, *, at: float | None = None) -> None:
328 final_time = float(self.env.now if at is None else at)
329 finite_number(final_time, name="monitor final time")
330 if final_time < self.samples[-1][0]:
331 raise CliError("monitor final time cannot precede its latest sample")
332 self._record("final", at=final_time)
334 def detach(self) -> None:
335 self.container.put = self._original_put
336 self.container.get = self._original_get
337 setattr(self.container, "_simpy_skill_monitor_attached", False)
339 def average_level(
340 self, *, start: float = 0.0, end: float | None = None
341 ) -> float:
342 final = self.samples[-1][0] if end is None else float(end)
343 finite_number(start, name="monitor window start")
344 finite_number(final, name="monitor window end")
345 if start < self.samples[0][0]:
346 raise CliError("monitor window starts before monitoring began")
347 if final <= start:
348 raise CliError("monitor window end must be greater than start")
349 if final > self.samples[-1][0]:
350 raise CliError("call finalize() through the requested window end first")
351 current = self.samples[0]
352 for sample in self.samples:
353 if sample[0] <= start:
354 current = sample
355 else:
356 break
357 cursor = start
358 weighted = 0.0
359 for sample in self.samples:
360 if sample[0] <= start:
361 continue
362 if sample[0] >= final:
363 break
364 weighted += current[2] * (sample[0] - cursor)
365 cursor = sample[0]
366 current = sample
367 weighted += current[2] * (final - cursor)
368 return weighted / (final - start)
371class EventTraceRecorder:
372 """Trace the next queued event before each Environment.step() call."""
374 def __init__(
375 self,
376 env: simpy.Environment,
377 *,
378 max_records: int = 100_000,
379 ):
380 load_simpy()
381 self.env = env
382 self.max_records = integer(
383 max_records,
384 name="max_records",
385 minimum=1,
386 maximum=MAX_TRACE_RECORDS,
387 )
388 self.records: list[dict[str, Any]] = []
389 self.truncated = False
390 self._original_step = env.step
391 self._patch()
393 def _patch(self) -> None:
394 @wraps(self._original_step)
395 def tracing_step() -> None:
396 queue = getattr(self.env, "_queue", None)
397 if queue:
398 if len(self.records) < self.max_records:
399 time, priority, event_id, event = queue[0]
400 self.records.append(
401 {
402 "event_id": int(event_id),
403 "event_type": type(event).__name__,
404 "priority": int(priority),
405 "queue_size_before": len(queue),
406 "time": float(time),
407 }
408 )
409 else:
410 self.truncated = True
411 self._original_step()
413 self.env.step = tracing_step
415 def detach(self) -> None:
416 self.env.step = self._original_step
418 def export_jsonl(self, output: str, *, force: bool = False) -> None:
419 """Write deterministic JSON Lines without object repr or memory addresses."""
421 text = "".join(
422 json.dumps(
423 record,
424 sort_keys=True,
425 ensure_ascii=False,
426 allow_nan=False,
427 separators=(",", ":"),
428 )
429 + "\n"
430 for record in self.records
431 )
432 emit_text(text, output=output, suffixes={".jsonl"}, force=force)
435def _demo() -> tuple[dict[str, Any], ResourceMonitor]:
436 simpy_module = load_simpy()
437 env = simpy_module.Environment()
438 resource = simpy_module.Resource(env, capacity=2)
439 monitor = ResourceMonitor(env, resource, "demo_server")
441 def user(identifier: int) -> Any:
442 yield env.timeout(identifier * 0.5)
443 with resource.request() as request:
444 yield request
445 yield env.timeout(1.0 + identifier * 0.1)
447 for index in range(8):
448 env.process(user(index))
449 env.run(until=10)
450 monitor.finalize(at=10)
451 return monitor.summary(start=0, end=10), monitor
454def build_parser() -> argparse.ArgumentParser:
455 parser = argparse.ArgumentParser(
456 description=(
457 "Run a finite synthetic Resource-monitor demonstration and emit a "
458 "time-weighted summary. No network or external service is used."
459 )
460 )
461 parser.add_argument("--output", help="Optional local .json summary")
462 parser.add_argument("--samples", help="Optional local .csv state samples")
463 parser.add_argument("--force", action="store_true", help="Replace explicit outputs")
464 return parser
467def main(argv: list[str] | None = None) -> int:
468 parser = build_parser()
469 args = parser.parse_args(argv)
470 try:
471 summary, monitor = _demo()
472 if args.samples:
473 monitor.export_csv(args.samples, force=args.force)
474 emit_json(
475 {
476 "monitor": summary,
477 "network_used": False,
478 "schema_version": "1.1",
479 },
480 output=args.output,
481 force=args.force,
482 )
483 except CliError as exc:
484 parser.error(str(exc))
485 return 0
488if __name__ == "__main__":
489 raise SystemExit(main())