Skills › Scientific Agent Skills › simpy › validate_simulation_config.py
validate_simulation_config.py
111 lines · skills/simpy/scripts/validate_simulation_config.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate bounded queue or replication JSON without running a simulation."""
4from __future__ import annotations
6import argparse
7from dataclasses import asdict
8from typing import Any
10from _common import CliError, emit_json, load_json_object
11from basic_simulation_template import QueueConfig
12from replication_runner import (
13 MAX_TOTAL_ENTITY_BUDGET,
14 MAX_TOTAL_EVENT_BUDGET,
15 ExperimentConfig,
16)
19def infer_schema(document: dict[str, Any]) -> str:
20 """Infer only the two bundled schemas; never resolve a plugin or import path."""
22 replication_markers = {"model", "replications", "confidence"}
23 return "replication" if set(document) & replication_markers else "queue"
26def validate_document(
27 document: dict[str, Any], *, schema: str = "auto"
28) -> dict[str, Any]:
29 """Return normalized validated configuration metadata."""
31 if schema not in {"auto", "queue", "replication"}:
32 raise CliError("schema must be auto, queue, or replication")
33 resolved = infer_schema(document) if schema == "auto" else schema
34 if resolved == "queue":
35 queue = QueueConfig.from_mapping(document)
36 normalized: dict[str, Any] = asdict(queue)
37 budgets = {
38 "entities": queue.max_entities,
39 "events": queue.max_events,
40 "replications": 1,
41 "time": queue.horizon,
42 }
43 else:
44 experiment = ExperimentConfig.from_mapping(document)
45 normalized = {
46 "confidence": experiment.confidence,
47 "model": asdict(experiment.model),
48 "replications": experiment.replications,
49 }
50 budgets = {
51 "entities": experiment.model.max_entities
52 * experiment.replications,
53 "events": experiment.model.max_events * experiment.replications,
54 "replications": experiment.replications,
55 "time_per_replication": experiment.model.horizon,
56 }
57 return {
58 "budgets": budgets,
59 "limits": {
60 "total_entities_for_replication_schema": MAX_TOTAL_ENTITY_BUDGET,
61 "total_events_for_replication_schema": MAX_TOTAL_EVENT_BUDGET,
62 },
63 "network_used": False,
64 "normalized": normalized,
65 "schema": resolved,
66 "schema_version": "1.1",
67 "validation": {
68 "config_allowlist": True,
69 "finite_numeric_values": True,
70 "no_code_execution": True,
71 "valid": True,
72 },
73 }
76def build_parser() -> argparse.ArgumentParser:
77 parser = argparse.ArgumentParser(
78 description=(
79 "Validate and normalize an allowlisted local queue or replication "
80 "JSON configuration. This command does not execute a simulation."
81 )
82 )
83 parser.add_argument("config", help="Local .json input; URLs and symlinks rejected")
84 parser.add_argument(
85 "--schema",
86 choices=("auto", "queue", "replication"),
87 default="auto",
88 help="Configuration schema (default: auto)",
89 )
90 parser.add_argument("--output", help="Optional local .json validation report")
91 parser.add_argument("--force", action="store_true", help="Replace explicit output")
92 return parser
95def main(argv: list[str] | None = None) -> int:
96 parser = build_parser()
97 args = parser.parse_args(argv)
98 try:
99 document = load_json_object(args.config)
100 emit_json(
101 validate_document(document, schema=args.schema),
102 output=args.output,
103 force=args.force,
104 )
105 except CliError as exc:
106 parser.error(str(exc))
107 return 0
110if __name__ == "__main__":
111 raise SystemExit(main())