SimForge

High-throughput batch simulation framework for engineers and researchers.

SimForge automates the execution, monitoring, and debugging of large-scale computational workloads — circuit simulations, numerical analysis sweeps, ML experiments, or any batch of independent compute-heavy tasks. It improves runtime efficiency through concurrency and surfaces failures fast.

┌─────────────────────────────────────────────────────────────────┐
│  Jobs (JSON / Python)                                           │
│      ↓                                                          │
│  TaskScheduler  ──priority queue + dependency DAG──────────┐   │
│      ↓                                                      ↓   │
│  BatchRunner  ──N worker threads──────────────────────────────  │
│      ↓                                                          │
│  OutputParsers  ──KeyValue / JSON / Regex──────────────────     │
│      ↓                                                          │
│  ResultAggregator  ──JSON / CSV export + statistics────────     │
│      ↓                                                          │
│  FailureTracer  ──pattern classification + per-job traces──     │
└─────────────────────────────────────────────────────────────────┘

---

Features

Component	What it does
BatchRunner	Multithreaded execution engine with configurable worker pool, timeouts, and per-job retries with exponential backoff
TaskScheduler	Priority queue with full dependency-graph support (DAG), automatic cycle detection, and cascade cancellation on upstream failure
ResultAggregator	Ingests completed jobs, runs output parsers, exports JSON/CSV, and computes descriptive statistics (mean, median, p95, etc.)
FailureTracer	Classifies failures by root cause (OOM, segfault, NaN, timeout, …), writes per-job trace files, and produces a prioritized failure digest
Output Parsers	KeyValueParser for key=value outputs, JSONOutputParser for embedded JSON blobs, RegexParser for arbitrary named-capture patterns
ProgressReporter	Live terminal progress bar with ETA, throughput rate, and per-status counts
CLI	simforge run jobs.json — run a JSON manifest with full logging, aggregation, and failure reporting

---

Installation

# Clone and install (stdlib-only, no external dependencies)
git clone https://github.com/yourname/simforge.git
cd simforge
pip install -e .

# Verify
simforge --help

Requires Python 3.10+. Zero external runtime dependencies — the entire core uses the standard library.

---

Quick Start

Python API

from simforge import BatchRunner, ResultAggregator, Job
from simforge.utils.logger import setup_logging
from simforge.parsers import KeyValueParser

setup_logging(level="INFO", log_dir="logs")

# Define jobs (any shell command)
jobs = [
    Job(
        command=f"python3 simulate.py --param {p}",
        name=f"sim_p{p}",
        tags={"param": p},
        timeout=120,
        retries=1,
    )
    for p in [0.1, 0.5, 1.0, 2.0, 5.0]
]

# Run with 8 parallel workers
runner = BatchRunner(workers=8)
runner.submit(jobs)
completed = runner.run()

# Parse and aggregate results
parser = KeyValueParser(cast_nums=True)
agg = ResultAggregator(output_dir="results", parser=parser)
agg.ingest(completed)
agg.export_json("results.json")
agg.export_csv("results.csv")

print(agg.status_summary())
print(agg.compute_stats("elapsed_s"))

CLI

# Run a JSON manifest
simforge run examples/jobs.json --workers 16 --output results/

# Dry run (plan without executing)
simforge run examples/jobs.json --dry-run

# Custom logging
simforge run examples/jobs.json --log-level DEBUG --log-dir logs/

---

Job Manifest Format

Jobs can be defined in a JSON file and run via the CLI:

[
  {
    "command": "python3 simulate.py --alpha 0.01 --beta 0.5",
    "name": "sim_001",
    "tags": {"alpha": 0.01, "beta": 0.5, "group": "sweep_v2"},
    "timeout": 300,
    "retries": 2,
    "priority": 0,
    "depends_on": []
  },
  {
    "command": "python3 postprocess.py --input sim_001_output.dat",
    "name": "postprocess_001",
    "depends_on": ["sim_001"],
    "priority": 1
  }
]

Field	Type	Default	Description
command	string	required	Shell command to execute
name	string	auto	Human-readable label
tags	object	{}	Arbitrary metadata for grouping/filtering
timeout	float	null	Max seconds before job is killed
retries	int	0	Retry attempts on failure
priority	int	0	Lower = higher urgency
depends_on	list	[]	Job IDs that must succeed first

---

Dependency Graphs

SimForge supports arbitrarily complex DAG pipelines. Jobs that list depends_on are held until all upstream jobs succeed.

from simforge import BatchRunner, Job

# preprocess → [sim_A, sim_B, sim_C] → postprocess → report

preprocess = Job(command="...", job_id="pre")

sims = [
    Job(command=f"...", job_id=f"sim_{i}", depends_on=["pre"])
    for i in range(3)
]

postprocess = Job(
    command="...",
    job_id="post",
    depends_on=[f"sim_{i}" for i in range(3)],
)

report = Job(command="...", job_id="report", depends_on=["post"])

runner = BatchRunner(workers=4, cancel_on_dep_failure=True)
runner.submit([preprocess] + sims + [postprocess, report])
completed = runner.run()

If any simulation fails, postprocess and report are automatically cancelled — no wasted compute.

---

Output Parsers

Plug parsers into ResultAggregator to automatically extract metrics from job stdout.

KeyValueParser — key=value or key: value lines

# Simulation output:
#   result=0.942478
#   converged=true
#   iterations=34

from simforge.parsers import KeyValueParser
parser = KeyValueParser(cast_nums=True, prefix="sim.")
# Produces: {"sim.result": 0.942478, "sim.converged": True, "sim.iterations": 34}

JSONOutputParser — embedded JSON in stdout

# Simulation output:
#   Running solver...
#   {"loss": 0.0034, "epoch": 100, "accuracy": 0.987}
#   Done.

from simforge.parsers import JSONOutputParser
parser = JSONOutputParser(keys=["loss", "epoch"])
# Produces: {"loss": 0.0034, "epoch": 100}

RegexParser — arbitrary named-capture patterns

from simforge.parsers import RegexParser
parser = RegexParser(cast_nums=True)
parser.add(r"Final loss:\s+(?P<loss>[\d.eE+\-]+)")
parser.add(r"Epoch (?P<epoch>\d+) complete")
# Produces: {"loss": 1.23e-4, "epoch": 100}

Chaining parsers

from simforge.parsers import KeyValueParser, RegexParser

def combined_parser(job):
    out = {}
    out.update(KeyValueParser(prefix="kv.").parse(job))
    out.update(RegexParser().add(r"score=(?P<score>[\d.]+)").parse(job))
    return out

agg = ResultAggregator(output_dir="results", parser=combined_parser)

---

Failure Tracing

from simforge.utils.failure_tracer import FailureTracer

tracer = FailureTracer(trace_dir="logs/traces")

# Register custom patterns
tracer.add_pattern("convergence_fail", r"(?i)(did not converge|diverged)")
tracer.add_pattern("license_error",    r"(?i)(license.*expired|no license)")

report = tracer.analyze(completed_jobs)
tracer.write_report(report, "results/failure_report.json")

# report["top_causes"] → [("oom", 12), ("timeout", 5), ("convergence_fail", 3)]

Each failed job also gets an individual .trace file in trace_dir/ with full stdout, stderr, and traceback.

Built-in patterns: oom, timeout, segfault, nan_value, file_error, import_err, assert_err, cuda_err.

---

Statistics & Export

agg = ResultAggregator(output_dir="results")
agg.ingest(completed)

# Export
agg.export_json("results.json")         # Full record array
agg.export_csv("results.csv")           # Flattened CSV, ready for pandas/Excel
agg.export_failure_report("fail.json")  # Failed jobs only

# Stats for any numeric column
stats = agg.compute_stats("elapsed_s")
# → {"count": 500, "mean": 1.24, "median": 0.98, "stdev": 0.44, "p95": 2.81, ...}

# Filter by tag
group_a = agg.get_by_tag("group", "sweep_v1")

# Status breakdown
print(agg.status_summary())
# → {"SUCCESS": 487, "FAILED": 11, "TIMEOUT": 2}

---

Project Structure

simforge/
├── simforge/
│   ├── __init__.py
│   ├── cli.py                  # Command-line interface
│   ├── core/
│   │   ├── job.py              # Job dataclass + JobStatus enum
│   │   ├── scheduler.py        # Priority queue + dependency DAG
│   │   ├── runner.py           # Multithreaded batch executor
│   │   └── aggregator.py       # Result collection + export + stats
│   ├── parsers/
│   │   ├── base.py             # BaseParser ABC
│   │   ├── key_value.py        # key=value / key: value parser
│   │   ├── json_parser.py      # Embedded JSON extractor
│   │   └── regex_parser.py     # Named-group regex parser
│   └── utils/
│       ├── logger.py           # Colored console + JSON file logging
│       ├── failure_tracer.py   # Pattern-based failure classifier
│       └── progress.py         # Live terminal progress bar
├── tests/
│   └── test_core.py            # 24 unit tests (stdlib only)
├── examples/
│   ├── basic_sweep.py          # Parameter sweep example
│   ├── dependency_dag.py       # Multi-stage pipeline example
│   └── jobs.json               # Sample manifest
├── pyproject.toml
└── README.md

---

Running Tests

python -m unittest tests/test_core.py -v
# Ran 24 tests in ~0.8s — OK

---

Configuration Reference

BatchRunner

BatchRunner(
    workers=8,                   # Thread pool size
    max_retries_override=None,   # Override all per-job retry counts
    poll_interval=0.05,          # Scheduler drain interval (seconds)
    on_job_done=None,            # Callback: fn(Job) -> None
    dry_run=False,               # Log commands without executing
    cancel_on_dep_failure=True,  # Cascade cancel downstream on failure
)

Job

Job(
    command="...",               # Shell command (required)
    name="",                     # Human label (auto-generated if omitted)
    tags={},                     # Arbitrary metadata dict
    timeout=None,                # Seconds before SIGKILL (None = unlimited)
    retries=0,                   # Retry attempts on non-zero exit
    priority=0,                  # Lower = higher scheduling priority
    depends_on=[],               # List of upstream job_ids
)

License

MIT