Virtual SCADA Simulator

What you get from one job

A single sdk.jobs.create(kind="simulate", engine="virtual_scada", ...) call produces a sealed evidence-carrying bundle with sensor telemetry, control-surface commands, alarm events, per-protocol wire traces, a pcapng capture loadable in Wireshark, and a signed sealed contract snapshot. Every byte passes the Validation Gate before delivery; blocked runs fail loudly.

Stage-2 physics realism ladder

Stage-2 ships six progressive realism mechanisms on top of the baseline AR(1) sensor simulator. Each layer is additive — pack authors opt in via declared fields; Stage-1 packs are unaffected.

• Authorable regime scheduler — LogNormal / Weibull / Gamma / Deterministic dwell distributions on RegimeDuration.distributiondrive a Hidden Semi-Markov timeline. Empirical dwell matches authored mean within 3% over 10⁴ transitions.
• Cross-signal coupling graph — pack authors msde_params.coupling_edges with target Pearson correlations; closed-form g = 2θρ / √(1 − 2ρ²) converts target ρ to drift gain (valid for |ρ| < 0.707).
• Diffusion-form aware integrators — per-signal diffusion_form ∈ {additive, multiplicative, cir} picks Euler-Maruyama, log-Euler, or Milstein; Strang splitting fires on coupled blocks.
• Adaptive dynamics calibration — auto_tune_sigmaplus newton_retune run a short pre-roll, measure observed utilisation, and correct σ in one Newton step. Resolved values stamp back into contract_k.sensor_catalog[i].msde_resolved for audit.
• ISA-18.2 alarm engine — class-default bands (safety / quality / process / equipment), MI-optimal placer with percentile fallback, BLAKE3 hash-chained operator journal. See Bundle contents.
• Unified live + batch integrator — the same integrator feeds an SSE transcript and the sealed batch bundle. Live endpoint: GET /v1/client/runs/{id}/stream.

Privacy-safe historian ingestion

Packs can author a dphe block that points at a customer historian (CSV / Parquet / OSIsoft PI JSON). The engine sanitises summary statistics under Gaussian mechanism with Rényi-DP accounting; the (ε, δ) budget and per-signal sensitivities travel with the run as privacy_evidence.json. Raw rows never leave the ingest boundary.

options = {
  "description": "WWTP aeration basin with 8 process signals",
  "dphe": {
    "enabled": True,
    "historian_source": "s3://tenant-uploads/plant_a/q1.parquet",
    "epsilon": 1.0,
    "delta": 1e-6,
    "signals": [
      {"signal_name": "flow_inlet_m3h", "sensitivity": 50.0},
      {"signal_name": "do_reactor_mgL", "sensitivity": 1.5},
      # ...
    ],
  },
}

Federated multi-site aggregation is available via kind="simulate", engine="federated_dphe". Each site runs under its own ε budget; Dwork-Roth post-processing keeps the federated ε = max(per-site ε), not the sum. Supported aggregation backends: mpc (three-server additive secret sharing), tee (SGX / SEV / Nitro), trusted.

Live streaming (SSE)

Subscribe to a running job via Server-Sent Events on /v1/client/runs/{run_id}/stream. Frame kinds: telemetry, command, alarm, attack_step, defense_alert, heartbeat, close.

curl -N \
  -H "Authorization: Bearer $RADMAH_API_KEY" \
  "https://api.radmah.ai/v1/client/runs/$RUN_ID/stream"

# Reconnect with ring-buffer replay:
curl -N -H "Authorization: Bearer $RADMAH_API_KEY" \
  "https://api.radmah.ai/v1/client/runs/$RUN_ID/stream?replay_history=true"

Replay a finished session

A finished session of any length (15 minutes to 24 hours, and 30-day Enterprise runs) plays back from a replay view derived from its sealed telemetry: every recorded sample when a window is narrow enough, otherwise buckets that keep each bucket's first, last, lowest and highest reading with their times, so no spike is dropped. The view is prepared when the session finishes; a session that finished earlier has it prepared the first time it is read. It is not part of the evidence bundle or the seal: every answer names the sealed artifact and SHA-256 it was built from, and the full telemetry file stays available for download.

Events are listed a page at a time, in time order, with the exact count of the window (by default the whole session, events outside the telemetry's span included). A page's next_cursor is null only at the end of the list; a page the API ends early says complete: false and its cursor continues it, so read until the cursor is null — iter_replay_events does, on the synchronous and the asynchronous client alike. An ICS session's security events are its own (ics_security_events), malicious first with flagged=True. Replay reads are limited per API key or signed-in person (a few hundred a minute for each route); a window is sent one signal at a time.

status = sdk.wait_for_replay(job.id)            # final: ready, failed or not_available
window = sdk.get_replay_window(job.id, max_points=800)          # the whole session
zoomed = sdk.get_replay_window(job.id, start_ms=t0, end_ms=t0 + 600_000, signals=[0, 3])
raised = list(sdk.iter_replay_events(job.id, kind="alarm", flagged=True))
attacks = list(sdk.iter_replay_events(job.id, kind="security", flagged=True))   # every malicious event
async for event in async_sdk.iter_replay_events(job.id, kind="security"):   # the async client pages the same
    ...

# the same from the CLI
rady scada replay <job-id> --wait
rady scada replay-window <job-id> --signal 0 --start 2026-01-01T01:00:00Z --end 2026-01-01T01:10:00Z --csv window.csv
rady scada replay-events <job-id> --kind alarm --flagged --all --json

Live OT sessions (Sovereign and Enterprise)

A session started with ot_server_enabled serves its plant live while it runs: on Sovereign over Modbus TCP (port 5502), OPC UA (4840) and BACnet/IP (47808/udp), and MQTT published to a broker you provide (mqtt_broker_host), for up to 24 hours; Enterprise adds a DNP3 outstation (20000) and runs up to 30 days. IEC 61850 is recorded frames and an SCL export only: a live request that names it — or MQTT without a broker — is refused before it is quoted, with LIVE_PROTOCOL_NOT_SERVED. A session is priced for exactly the protocols it serves (none named: Modbus). Free runs are recorded only. get_session_health reports the live servers (state, protocols, connection strings, end time); stop_session ends the session early and gracefully: the servers close, what was recorded is sealed, the session completes with it, and the unused part of the time it was requested for is refunded at the session's hourly rate, rounded up, keeping at least 1 credit (Commerce 7.1.4; the charge itself is per started hour, 7.1). Cancelling a live session is an early end too and is refunded the same way; a session cancelled, stopped or failed before its servers began serving is refunded in full; a session stopped because the account's balance ran out is not refunded (Commerce 7.1.5). A recorded run has no live server: its health says so and a stop answers 409 SESSION_NOT_LIVE.

health = sdk.get_session_health(job.id)   # kind "live_session" or "not_applicable"
sdk.stop_session(job.id)                  # 202: stopping (graceful)

rady scada session-health <job-id>
rady scada stop <job-id>

Deep-dive references

• Engine architecture — three-tier dispatch, a broad library of building blocks, and the determinism guarantee.
• Bundle contents — every artifact the customer receives, its format, and how to download it through the SDK.
• Commands stream — row schema, the four write_source roles, baseline cadences, and a worked customer example.
• Validation Gate — the four scoring dimensions (schedule, protocol, physics, label), their thresholds, and the structural checks that block degraded data from release.
• Protocol reference — Modbus TCP, OPC-UA, BACnet/IP, MQTT, DNP3, and IEC 61850 stack details, default port map, and address conventions.
• ICS Security overlay — how MITRE ATT&CK ICS techniques layer onto a Virtual-SCADA run to produce labelled attack evidence for SOC / IDS training.

Quickstart

from radmah_sdk import RadMah

sdk = RadMah(api_key="...", base_url="https://api.radmah.ai")

job = sdk.jobs.create(
    kind="simulate",
    engine="virtual_scada",
    seed=42,
    options={
        "description": (
            "11 kV distribution substation with two protective relays, "
            "one RTU, one breaker, and a metering panel"
        ),
        "total_seconds": 120,
        "protocols": ["iec61850", "dnp3", "modbus"],
    },
)

# Poll until job.state == "succeeded", then:
artifacts = sdk.list_artifacts(job.id)
for art in artifacts:
    body = sdk.download_artifact(job.id, art.id)
    open(art.name, "wb").write(body if isinstance(body, bytes) else body.encode())

Enterprise tier

Enterprise deployments get container-isolated Virtual PLC simulation with Docker isolation, real-time-capable Linux scheduling, and live OT protocol servers on deterministic ports — suitable for true hardware-in-the-loop IDS realism. See Deployment Models.