client.dynamic_simulations

A dynamic simulation integrates the powertrain through time: you describe a mission as a throttle/airspeed schedule plus a termination condition, and the server returns per-step samples, time series, a scorecard, and events. Use it for endurance runs, spin-up transients, mixed hover/cruise missions, and anything else a single steady operating point can't answer.

Like simulations and sweeps, dynamic runs are asynchronous — create returns status: "queued" immediately.

Methods

Two endpoints are not wrapped by the SDK: GET /v1/dynamic-simulations/{id}/export.csv (raw CSV download — fetch the URL directly) and GET /v1/dynamic-simulations/{id}/stream (SSE progress).

Request body

The body shares the simulation base (rotor_groups, battery_component_id, density_kg_m3, battery_charge_pct, ambient_temp_c) but rotor groups carry no throttle_pct — all commands come from the schedule. Two more optional fields set the starting temperatures: motor_initial_temp_c / battery_initial_temp_c (°C, −60..85, default None = cold-start at ambient).

"""Dynamic (time-domain) simulation: auth -> submit -> wait() -> read the result.

Run it:
    export THRUSTLAB_API_KEY=key_...        # never hard-code the key
    python examples/dynamic/run_and_poll.py

A dynamic run integrates the powertrain through a throttle/airspeed *schedule*
until a *termination* condition (here: a 2 s spin-up ramp, then a 20 s hover
at 70% — a fixed-duration mission). It returns per-step `samples` (each a full
steady-shaped result),
time `series`, a `scorecard`, and `events`. Swap in your own component IDs, or
resolve them by name with client.components.find(...).
"""

from thrustlab import Client

client = Client()  # reads $THRUSTLAB_API_KEY from the environment

project = client.projects.create(name="sdk dynamic example")

# Resolve by name, or paste explicit IDs: motor_id = "comp_motor_xxx"
motor = client.components.find(name="Spektrum Avian 4260 800Kv")
prop = client.components.find(name="10.5x4.5")
battery = client.components.find(name="Liperior 5000mAh 4S 35C")

dyn = client.dynamic_simulations.create(
    project_id=project["id"],
    battery_component_id=battery["id"],
    density_kg_m3=1.225,
    battery_charge_pct=100,
    ambient_temp_c=25,
    rotor_groups=[
        {
            "label": "main",
            "count": 4,
            "motor_component_id": motor["id"],
            "propeller_component_id": prop["id"],
        }
    ],
    # Ramp to 70% over 2 s, then hold for 20 s. The soft-start matters:
    # stepping 70% onto a stationary rotor draws an in-rush that can sag the
    # pack below the low-voltage cutoff and end the run immediately — exactly
    # as it would on real hardware. Endurance ("until depleted") missions work
    # the same way but integrate minutes of pack time; keep examples short.
    schedule={
        "mode": "segments",
        "segments": [
            {
                "duration_s": 2.0,
                "airspeed_target": 0.0,
                "per_group": {"main": {"throttle_target": 70, "throttle_ramp": "linear"}},
            },
            {
                "duration_s": 20.0,
                "airspeed_target": 0.0,
                "per_group": {"main": {"throttle_target": 70}},
            },
        ],
    },
    # Fixed-duration mission: run the schedule to its end.
    termination={"mode": "fixed"},
)
print(f"created {dyn['id']}, status={dyn['status']}")

result = client.dynamic_simulations.wait(dyn["id"], timeout=600)
print(f"final status: {result['status']}")

if result["status"] == "completed":
    blob = result["result"]

    # scorecard: whole-run headline metrics (canonical snake_case).
    sc = blob["scorecard"]
    print(f"peak winding temp: {sc['peak_winding_temp_c']:.1f} C")
    print(f"avg efficiency:    {sc['avg_efficiency']:.3f}")
    print(f"peak current:      {sc['peak_current_a']:.1f} A")
    print(f"min cell voltage:  {sc['min_cell_voltage_v']:.2f} V")

    # samples[]: each entry is a full steady-shaped snapshot at one time step —
    # read the same canonical per-rotor / aggregate keys as a single-point run.
    samples = blob["samples"]
    first, last = samples[0], samples[-1]
    print(f"steps: {len(samples)}")
    print(f"  t0 total thrust: {first['All']['total_thrust_n']:.2f} N")
    print(f"  tN total thrust: {last['All']['total_thrust_n']:.2f} N")

The schedule

schedule.mode picks between two ways to drive the mission: "segments" (hand-authored throttle/airspeed steps, below) or "csv" (a full time-series upload, see CSV schedules).

schedule.mode: "segments" describes the mission as an ordered list of segments. Each segment has either a finite duration_s or until_depleted: true (a terminal hold that runs until the termination condition fires). Vehicle-level channels (airspeed_target, vertical_speed_target) sit on the segment; per-group channels (throttle_target, tilt_target) sit under per_group, keyed by rotor-group label:

schedule={
    "mode": "segments",
    "segments": [
        {   # 2 s spin-up: ramp throttle 0 -> 70%
            "duration_s": 2.0,
            "airspeed_target": 0.0,
            "per_group": {"main": {"throttle_target": 70, "throttle_ramp": "linear"}},
        },
        {   # hold 70% until the pack depletes
            "until_depleted": True,
            "airspeed_target": 0.0,
            "per_group": {"main": {"throttle_target": 70}},
        },
    ],
},
termination={"mode": "until_depleted", "soc_cutoff_pct": 20.0},

Segment-level channels:

Per-group (per_group.<label>) channels:

Rules worth knowing:

• Ramps: a "linear" ramp needs a defined end time, so it requires a finite duration_s — express ramp-then-hold as a finite ramp segment followed by the terminal hold, as above.
• The first segment starts from zero: throttle ramps up from 0, so a linear first segment is a realistic soft-start.
• Soft-start matters: stepping high throttle onto a stationary rotor draws a large in-rush current that can sag the pack below the low-voltage cutoff and end the run within milliseconds — exactly as on real hardware. If that's not the transient you're studying, ramp the first segment.
• List every group in every segment — a group omitted from a segment's per_group is commanded to 0% throttle for that segment.
• Different rotor groups get independent channels: give each group its own throttle_target / tilt_target per segment.

CSV schedules

schedule.mode: "csv" uploads the full time series as text instead of authoring segments — useful for a logged flight or an externally generated mission:

schedule={
    "mode": "csv",
    "csv_text": (
        "time_s,airspeed_ms,throttle_1,throttle_2\n"
        "0,0,0,0\n"
        "2,0,70,70\n"
        "300,15,85,85\n"
    ),
    "interpolation": "linear",  # or "hold" (left-breakpoint / zero-order-hold)
},

Header columns are matched by name (case-insensitive), not position:

• Required: time_s, airspeed_ms, throttle_1..throttle_N (one per rotor, N = total rotor count across all groups).
• Optional: vertical_speed_ms (signed, global) and tilt_deg_k (per-rotor, any subset of 1..N) — omitted columns default to 0.
• time_s must strictly increase row to row; every cell must be a finite number in range (throttle 0..100, tilt 0..90, vertical speed −30..30).

Termination

See the API reference for the full schedule and termination schema.

Wait, then read the result

dyn = client.dynamic_simulations.create(project_id="proj_xxx", ...)
result = client.dynamic_simulations.wait(dyn["id"], timeout=600)

if result["status"] == "completed":
    blob = result["result"]

Terminal states are completed / failed / canceled (one 'l' on the wire). On timeout, wait returns the dict with status="timed_out" — an SDK-side sentinel, not a server state; the run keeps computing and you can call wait() or retrieve() again.

A completed run's result has four parts:

Scorecard keys: flight_time_s, energy_wh, range_m, avg_efficiency, peak_current_a, peak_winding_temp_c, min_cell_voltage_v, depletion_criterion, terminated_early. A sibling display_labels map on the resource translates every snake_case key to its human label.

sc = blob["scorecard"]
print(f"flight time:  {sc['flight_time_s']:.1f} s")
print(f"peak current: {sc['peak_current_a']:.1f} A")
print(f"min cell V:   {sc['min_cell_voltage_v']:.2f} V")

first, last = blob["samples"][0], blob["samples"][-1]
print(f"final thrust: {last['All']['total_thrust_n']:.2f} N")

Estimate before you run

estimate projects the run's coulombic duration from the battery, schedule, and termination alone — no solver invocation, no compute-unit charge for the integration:

est = client.dynamic_simulations.estimate(
    battery_component_id="comp_batt_xxx",
    rotor_groups=[...], density_kg_m3=1.225, battery_charge_pct=100,
    schedule={...}, termination={...},
)

Long missions cost real wall-clock time (the solver integrates the full thermal/electrical state), so estimate first and keep exploratory runs short — e.g. use termination={"mode": "fixed"} with a finite schedule instead of depleting a full pack.

Cancel an in-flight run

client.dynamic_simulations.cancel("dyn_2c5tQ...")

No webhook events (yet)

Dynamic runs do not emit webhook event types — poll with wait() or retrieve(). The event catalog covers simulations and sweeps only.

See also

• Async resources — the shared poll/wait contract
• Simulations — the steady-state base body
• Sweeps — parameter studies over steady points