Feature: Export a session as a runnable Python file

[jaywonchung]

Depends on #3. The exported file's value comes from the richness of the scenario it captures; without #3 it can only serialize a feeder plus one model, which barely justifies the feature.

The user actions list (from #4) is enrichment, not a dependency. If #4 isn't in, actions = [] and the script still runs deterministically. Once #4 lands, the action list grows from empty to whatever the user did; the template doesn't change shape.

The streaming session concept (from #2) is also not a dependency. Export can hang off the form-submit flow: "download a runnable .py of the scenario you've configured" works without any WebSocket session.

A live session is fully described by three values: a scenario config, a seed, and an ordered list of (t_s, command) actions. openg2g simulations are deterministic given those three, so serializing them and replaying yields the same trajectory. The "Export scenario" button should produce a single .py file that captures the session, depends only on openg2g, and runs standalone on any machine.

With this, users can use the OpenG2G live web app to find and construct scenarios that fit their needs or are interesting, and then export them into an OpenG2G simulation scenario that they can play with, be it implementing and testing a new controller, further tweaking configurations, or programmatically analyzing intermediate/final simulation metrics.

What the exported file looks like

Everything is constructed inline. No helper function reads a JSON dict and builds the coordinator; the user can edit any field of any constructor.

"""openg2g scenario, exported on 2026-05-11T01:23:45Z.

Run with:
    pip install "openg2g[opendss]"
    python scenario_e4f7c9.py
"""
from fractions import Fraction

from openg2g.controller.tap_schedule import TapScheduleController
from openg2g.coordinator import Coordinator
from openg2g.datacenter.command import SetBatchSize
from openg2g.datacenter.config import (
    DatacenterConfig, InferenceModelSpec, ModelDeployment, ReplicaSchedule,
)
from openg2g.datacenter.offline import OfflineDatacenter, OfflineWorkload
from openg2g.datacenter.workloads.inference import InferenceData
from openg2g.grid.command import SetTapPosition
from openg2g.grid.config import TapPosition, TapSchedule
from openg2g.grid.opendss import OpenDSSGrid


SEED = 42
TOTAL_DURATION_S = 3600

LLAMA_8B = InferenceModelSpec(
    model_label="Llama-3.1-8B",
    model_id="meta-llama/Llama-3.1-8B-Instruct",
    gpu_model="H100",
    task="lm-arena-chat",
    precision="bfloat16",
    gpus_per_replica=1,
    tensor_parallel=1,
    itl_deadline_s=0.08,
    batch_sizes=(8, 16, 32, 64, 128, 256, 512),
    feasible_batch_sizes=(8, 16, 32, 64, 128, 256, 512),
)


def main() -> None:
    grid = OpenDSSGrid(
        dss_case_dir="data/grid/ieee13",
        dss_master_file="IEEE13Bus.dss",
        dt_s=Fraction(1),
        initial_tap_position=TapPosition(regulators={
            "creg1a": 1.0875, "creg1b": 1.0375, "creg1c": 1.09375,
        }),
    )

    inference_data = InferenceData.ensure("data/specs", (LLAMA_8B,), plot=False)
    dc = OfflineDatacenter(
        DatacenterConfig(),
        OfflineWorkload(
            inference_data=inference_data,
            replica_schedules={"Llama-3.1-8B": ReplicaSchedule(initial=720)},
        ),
        name="dc",
        dt_s=Fraction(1),
        total_gpu_capacity=720,
        seed=SEED,
    )
    grid.attach_dc(dc, bus="671")

    controllers = [TapScheduleController(schedule=TapSchedule(()), dt_s=Fraction(1))]

    coord = Coordinator(
        datacenters=[dc], grid=grid, controllers=controllers,
        total_duration_s=TOTAL_DURATION_S,
    )

    # Captured user actions. Each entry is (sim_time_s, list of openg2g commands).
    actions: list[tuple[float, list]] = [
        (120.0, [SetBatchSize(batch_size_by_model={"Llama-3.1-8B": 256}, target=dc)]),
        (240.0, [SetBatchSize(batch_size_by_model={"Llama-3.1-8B": 64},  target=dc)]),
        (300.0, [SetTapPosition(regulators={"creg1a": 1.05625})]),
    ]

    pending = list(actions)
    for tick in coord.run_iter():
        while pending and pending[0][0] <= tick.t_s:
            _, cmds = pending.pop(0)
            coord.dispatch_commands(cmds)


if __name__ == "__main__":
    main()

Requirements the exporter has to enforce:

• Every openg2g object is constructed inline. No helper that takes a JSON dict.
• Actions are real command-constructor calls, not opaque dicts decoded at runtime. They reference dc/grid via closure inside main().
• Imports are only from openg2g. No live module, no from .helpers import ....
• Top-level constants (SEED, TOTAL_DURATION_S, model specs) are at module scope so they're trivial to edit.

Backend

@app.get("/sessions/{sid}/export")
def export_session(sid: str) -> Response:
    s = sessions[sid]
    code = render_template(
        feeder=s.feeder, dc_layout=s.dc_layout, models=s.models,
        controllers=s.controllers, actions=s.actions, seed=s.seed,
    )
    return Response(
        content=code, media_type="text/x-python",
        headers={"Content-Disposition": f'attachment; filename="scenario_{sid}.py"'},
    )

The real work is render_template. It needs to emit each openg2g object's constructor call as readable Python. Python's built-in repr won't do (class names need to be qualified with their imports, and openg2g classes don't all round-trip through repr). Write per-class repr_* helpers:

• repr_inference_model_spec(spec) -> str
• repr_command(cmd) -> str
• repr_controller(ctrl) -> str

Roughly 20-30 lines each, ~6 helpers total. Many openg2g classes are dataclass or pydantic.BaseModel; for those, .model_dump() (pydantic) or dataclasses.asdict() (dataclasses) gives a dict you can format.

Sessions are held in a dict[session_id, Session] on the server. No DB. Server restart → in-flight sessions disappear; the user kept their scenario by downloading the file.