Skills › Scientific Agent Skills › simpy › event_trace_summary.py
event_trace_summary.py
300 lines · skills/simpy/scripts/event_trace_summary.py at 5f2bc03
1#!/usr/bin/env python3
2"""Summarize bounded event-trace JSONL or ResourceMonitor CSV artifacts."""
4from __future__ import annotations
6import argparse
7import csv
8from collections import Counter
9from pathlib import Path
10from typing import Any
12from _common import (
13 MAX_TRACE_RECORDS,
14 CliError,
15 checked_input_file,
16 emit_json,
17 finite_number,
18 integer,
19 parse_json_line,
20 validate_keys,
21)
24EVENT_FIELDS = {
25 "event_id",
26 "event_type",
27 "priority",
28 "queue_size_before",
29 "time",
30}
31RESOURCE_FIELDS = {
32 "count",
33 "event",
34 "queue_length",
35 "time",
36 "utilization",
37}
40def _csv_number(value: str | None, *, name: str) -> float:
41 if value is None:
42 raise CliError(f"{name} is missing")
43 try:
44 parsed = float(value)
45 except ValueError as exc:
46 raise CliError(f"{name} must be numeric") from exc
47 return finite_number(parsed, name=name)
50def summarize_event_trace(path: Path, *, max_records: int) -> dict[str, Any]:
51 """Validate and summarize deterministic JSONL event records."""
53 counts: Counter[str] = Counter()
54 first_time: float | None = None
55 last_time: float | None = None
56 ordering_violations = 0
57 maximum_queue_size = 0
58 records = 0
59 try:
60 with path.open("r", encoding="utf-8") as handle:
61 for line_number, raw_line in enumerate(handle, start=1):
62 if not raw_line.strip():
63 raise CliError(f"blank JSONL record on line {line_number}")
64 records += 1
65 if records > max_records:
66 raise CliError(
67 f"trace exceeds the configured record limit ({max_records})"
68 )
69 record = parse_json_line(raw_line, line_number=line_number)
70 validate_keys(
71 record,
72 allowed=EVENT_FIELDS,
73 required=EVENT_FIELDS,
74 context=f"trace line {line_number}",
75 )
76 time = finite_number(
77 record["time"], name=f"line {line_number} time", minimum=0
78 )
79 integer(
80 record["priority"],
81 name=f"line {line_number} priority",
82 minimum=-1_000_000,
83 maximum=1_000_000,
84 )
85 integer(
86 record["event_id"],
87 name=f"line {line_number} event_id",
88 minimum=0,
89 maximum=2**63 - 1,
90 )
91 queue_size = integer(
92 record["queue_size_before"],
93 name=f"line {line_number} queue_size_before",
94 minimum=1,
95 maximum=MAX_TRACE_RECORDS,
96 )
97 event_type = record["event_type"]
98 if (
99 not isinstance(event_type, str)
100 or not event_type.isidentifier()
101 or len(event_type) > 128
102 ):
103 raise CliError(
104 f"line {line_number} event_type must be a short identifier"
105 )
106 # A callback may schedule an urgent event at the same time.
107 # Heap ordering applies to the current queue, not to the full
108 # processing trace. Only backwards model time is detectable here.
109 if last_time is not None and time < last_time:
110 ordering_violations += 1
111 first_time = time if first_time is None else first_time
112 last_time = time
113 maximum_queue_size = max(maximum_queue_size, queue_size)
114 counts[event_type] += 1
115 except (OSError, UnicodeError) as exc:
116 raise CliError(f"cannot read trace {path.name}: {exc}") from exc
117 if records == 0:
118 raise CliError("event trace contains no records")
119 return {
120 "event_type_counts": dict(sorted(counts.items())),
121 "first_time": first_time,
122 "last_time": last_time,
123 "maximum_queue_size_before_step": maximum_queue_size,
124 "ordering_violations": ordering_violations,
125 "records": records,
126 "semantics": (
127 "Each row is the next private event-queue entry immediately before "
128 "Environment.step(); ordering_violations counts decreasing times only. "
129 "Same-time priority/event_id decreases can be valid; validating heap "
130 "selection requires schedule/insertion history absent from this trace."
131 ),
132 }
135def summarize_resource_csv(path: Path, *, max_records: int) -> dict[str, Any]:
136 """Validate and summarize ResourceMonitor state-change samples."""
138 samples: list[dict[str, Any]] = []
139 event_counts: Counter[str] = Counter()
140 try:
141 with path.open("r", encoding="utf-8", newline="") as handle:
142 reader = csv.DictReader(handle)
143 if reader.fieldnames is None:
144 raise CliError("resource CSV has no header")
145 if len(reader.fieldnames) != len(set(reader.fieldnames)):
146 raise CliError("resource CSV has duplicate columns")
147 if set(reader.fieldnames) != RESOURCE_FIELDS:
148 raise CliError(
149 "resource CSV columns must be exactly: "
150 + ", ".join(sorted(RESOURCE_FIELDS))
151 )
152 previous_time: float | None = None
153 for row_number, row in enumerate(reader, start=2):
154 if None in row or any(value is None for value in row.values()):
155 raise CliError(f"resource CSV row {row_number} has the wrong field count")
156 if len(samples) >= max_records:
157 raise CliError(
158 f"resource CSV exceeds the record limit ({max_records})"
159 )
160 time = finite_number(
161 _csv_number(row["time"], name=f"row {row_number} time"),
162 name=f"row {row_number} time",
163 minimum=0,
164 )
165 if previous_time is not None and time < previous_time:
166 raise CliError("resource sample times must be nondecreasing")
167 previous_time = time
168 count = integer(
169 int(row["count"]),
170 name=f"row {row_number} count",
171 minimum=0,
172 maximum=MAX_TRACE_RECORDS,
173 )
174 queue_length = integer(
175 int(row["queue_length"]),
176 name=f"row {row_number} queue_length",
177 minimum=0,
178 maximum=MAX_TRACE_RECORDS,
179 )
180 utilization = finite_number(
181 _csv_number(
182 row["utilization"],
183 name=f"row {row_number} utilization",
184 ),
185 name=f"row {row_number} utilization",
186 minimum=0,
187 maximum=1,
188 )
189 event = row["event"]
190 if (
191 not event
192 or len(event) > 64
193 or any(character in "\r\n," for character in event)
194 ):
195 raise CliError(f"row {row_number} has an invalid event label")
196 samples.append(
197 {
198 "count": count,
199 "event": event,
200 "queue_length": queue_length,
201 "time": time,
202 "utilization": utilization,
203 }
204 )
205 event_counts[event] += 1
206 except (OSError, UnicodeError, csv.Error) as exc:
207 raise CliError(f"cannot read resource CSV {path.name}: {exc}") from exc
208 except ValueError as exc:
209 raise CliError(f"resource CSV contains an invalid integer: {exc}") from exc
210 if len(samples) < 2:
211 raise CliError("resource CSV needs at least an initial and final sample")
212 start = samples[0]["time"]
213 end = samples[-1]["time"]
214 if end <= start:
215 raise CliError("resource CSV must span positive simulation time")
216 queue_area = 0.0
217 utilization_area = 0.0
218 for current, following in zip(samples, samples[1:]):
219 duration = following["time"] - current["time"]
220 queue_area += current["queue_length"] * duration
221 utilization_area += current["utilization"] * duration
222 return {
223 "average_queue_length": queue_area / (end - start),
224 "average_utilization": utilization_area / (end - start),
225 "end": end,
226 "event_counts": dict(sorted(event_counts.items())),
227 "maximum_count": max(sample["count"] for sample in samples),
228 "maximum_queue_length": max(
229 sample["queue_length"] for sample in samples
230 ),
231 "records": len(samples),
232 "start": start,
233 "semantics": (
234 "Time-weighted values carry each post-transition state forward "
235 "until the next sample. Same-time pre/post samples have zero duration."
236 ),
237 }
240def summarize(path: Path, *, max_records: int) -> dict[str, Any]:
241 if path.suffix.lower() == ".jsonl":
242 kind = "event_trace"
243 summary = summarize_event_trace(path, max_records=max_records)
244 elif path.suffix.lower() == ".csv":
245 kind = "resource_monitor"
246 summary = summarize_resource_csv(path, max_records=max_records)
247 else:
248 raise CliError("input must end in .jsonl or .csv")
249 return {
250 "input": {"kind": kind, "name": path.name},
251 "network_used": False,
252 "schema_version": "1.2",
253 "summary": summary,
254 }
257def build_parser() -> argparse.ArgumentParser:
258 parser = argparse.ArgumentParser(
259 description=(
260 "Summarize a local event-trace .jsonl or ResourceMonitor .csv file. "
261 "Inputs are bounded and validated; no model code is executed."
262 )
263 )
264 parser.add_argument("input", help="Local .jsonl or .csv input")
265 parser.add_argument(
266 "--max-records",
267 type=int,
268 default=100_000,
269 help="Input record cap from 1 through 1,000,000 (default: 100000)",
270 )
271 parser.add_argument("--output", help="Optional local .json summary")
272 parser.add_argument("--force", action="store_true", help="Replace explicit output")
273 return parser
276def main(argv: list[str] | None = None) -> int:
277 parser = build_parser()
278 args = parser.parse_args(argv)
279 try:
280 max_records = integer(
281 args.max_records,
282 name="max_records",
283 minimum=1,
284 maximum=MAX_TRACE_RECORDS,
285 )
286 path = checked_input_file(
287 args.input, suffixes={".csv", ".jsonl"}
288 )
289 emit_json(
290 summarize(path, max_records=max_records),
291 output=args.output,
292 force=args.force,
293 )
294 except CliError as exc:
295 parser.error(str(exc))
296 return 0
299if __name__ == "__main__":
300 raise SystemExit(main())