Simulation

from nexus_uav import RunError, Sim

A script reads any recorded history by indexing the run with the path of the history below the root of the run, the same path the recording shows. sim["vehicle/body/body_frd"] is the base body, sim["vehicle/joints/rotor_1_joint"] a joint, sim["vehicle/sensors/imu/imu"] the samples of a sensor, and sim["vehicle/controllers/px4/controls"] the controls the controller wrote. for path in sim lists them. Every history reads one way, as records. history() gives every row as one structured array, oldest first: t, the tick's sim time, then the signal's fields as its type declares them. rows["position"][:, 2] holds the heights over the run, and history()[-1] is the newest row. The row of an Inertial Measurement Unit (IMU) holds row["accel"], a body's row["position"], row["orientation"], row["linear_velocity"] and row["angular_velocity"], and a joint's row["q"] and row["qd"]. arrays() gives one array per quantity.

A script reads a signal's live value by the same path, typed by the signal: sim.read("vehicle/body/body_frd").position. A struct signal reads as one instance of its type, and a value signal as an array without its leading axis of one. A host signal, such as a camera's image, reads as the object its writer wrote, though it keeps no history. sim.time is the sim time of the last tick.

Sim

Sim(
    vehicle: str,
    *,
    scene: str,
    catalog: str | None = None,
    device: str = "auto",
    log: bool = False,
    view: bool = False,
    debug: bool = False,
    cache_dir: str | None = None,
    solver: str | None = None,
    max_steps: int | None = None,
    rtf: float = 0.0,
    layer: str | None = None,
    registry: Registry | None = None,
    run_dir: str | PathLike | None = None,
    clean: bool | None = None,
)

The one public entry to a Newton sim run.

Builds an Loop from a LaunchConfig and drives its loop on the caller's thread, one control tick per :meth:step, the same for a controller that takes setpoints and for PX4 on its Hardware In The Loop (HIL) lockstep. :meth:start drives setup as far as lockstep; :meth:run drives the whole run; :meth:wait_until and :meth:sleep step until a predicate or a sim-time budget. Use it as a context manager: __enter__ builds, __exit__ calls :meth:stop to tear the run down cooperatively.

Every run has a :class:~nexus_uav._src.recording.Recorder, so :attr:physics and :attr:sensors read sim ground truth, in the world frame, Z-up Forward-Left-Up (FLU), off the recorded histories.

PARAMETER	DESCRIPTION
vehicle	Catalog vehicle name, for example "astro_max_fpv", or a local .usd path. TYPE: str
catalog	Path to a catalog that extends the bundled one and those of the installed packages. None takes the nearest nexus.catalog.yaml in the working directory or a directory over it, and only the bundled and installed catalogs when no directory holds one. TYPE: str | None DEFAULT: None
scene	Catalog scene name to fly in, for example "empty" for flat ground or the "slalom" obstacle pillars, or a local scene .usd path. TYPE: str
device	Compute device for the runtime: "auto", the default, which picks CUDA when present, or an explicit "cpu", for bit-exact determinism, or "cuda". TYPE: str DEFAULT: 'auto'
log	Record the run to a .rrd; False writes no recording, at max speed. TYPE: bool DEFAULT: False
view	Serve the Rerun recording live on :9876 instead of writing a .rrd file. TYPE: bool DEFAULT: False
cache_dir	Override the asset cache directory; None = the framework default. TYPE: str | None DEFAULT: None
max_steps	Cap the run at this many control steps; None = the launch-config default. TYPE: int | None DEFAULT: None
rtf	Real-time-factor throttle. 0, the default, runs unthrottled, as fast as the controller keeps up. 1.0 paces the loop to wall-clock for human-in-the-loop flying: 1:1 stick feel, and sim-time protocol timeouts align with wall-clock peers. A vehicle that declares a free-running peer, one on its own wall clock, needs 1.0, and the build fails at any other value before a peer starts. TYPE: float DEFAULT: 0.0
layer	Path to an override layer, a local Universal Scene Description (USD) file the run composes over the vehicle and the scene: it changes a declared value, such as the scene's geodetic origin on its PhysicsScene prim, selects a variant, or drops a declaration. A layer that deactivates the scope of the PX4 Software In The Loop (SITL) peer, Px4Sitl, flies an autopilot started elsewhere, which dials port 4560. TYPE: str | None DEFAULT: None
registry	The :class:~nexus_uav.Registry the build resolves each schema with, its own rows first and the install's for the rest, so a test maps a schema to a stand-in or a peer to its fake; None takes the install alone. TYPE: Registry | None DEFAULT: None
run_dir	The folder the run writes its files to: the recording, nexus.rrd, and each peer's files under a folder named for the peer, such as PX4's console, px4_sitl/console.log, and its ULog beside it. The run creates it when missing and refuses it when it holds files. A relative path starts at the working folder. None takes a new folder under ~/.cache/nexus/logs. A peer's container mounts the folder at its own path, so a script inside a container that starts peers on the host's daemon binds it at the same path on the host. TYPE: str | PathLike | None DEFAULT: None
clean	Start every peer from nothing an earlier run stored. A peer with a store keeps it across runs, as a real vehicle does across a reboot: PX4 boots with the parameters and the mission an earlier run on the same airframe and instance stored. True empties each store first; None takes $NEXUS_CLEAN, where 1 means clean, and else keeps the store. TYPE: bool | None DEFAULT: None

RAISES	DESCRIPTION
FileExistsError	On entry, run_dir holds files already; raised before any peer starts.

Example

with Sim("astro_max_base", scene="empty") as sim:
    sim.start()
    body = f"vehicle/body/{sim.base_body}"
    sim.wait_until(lambda: sim.read(body).position[2] > 1.0, sim_timeout=30.0)

Source code in nexus_uav/_src/api/sim.py

def __init__(
    self,
    vehicle: str,
    *,
    scene: str,
    catalog: str | None = None,
    device: str = "auto",
    log: bool = False,
    view: bool = False,
    debug: bool = False,
    cache_dir: str | None = None,
    solver: str | None = None,
    max_steps: int | None = None,
    rtf: float = 0.0,
    layer: str | None = None,
    registry: Registry | None = None,
    run_dir: str | os.PathLike | None = None,
    clean: bool | None = None,
):
    self._registry = registry  # the registry the build resolves each schema with, or None for the default
    self._launch = LaunchConfig()
    self._launch.set_vehicle(vehicle)  # a catalog *name* or a local .usd path
    self._launch.layer = layer  # an override layer the run composes over the vehicle and the scene, or None
    self._launch.catalog = catalog  # None: the run finds its own catalog, see load_catalog
    if solver is not None:  # physics integrator override: mujoco | semi_implicit | featherstone
        self._launch.runtime.solver = solver
    self._launch.set_scene(scene)  # catalog scene, for example the 'slalom' obstacle pillars for sampling-mpc
    # 'gpu' is an alias of 'cuda'. The launch takes 'auto', 'cpu' or 'cuda', and the resolve picks the
    # device: CUDA when a CUDA device is present, else the CPU. The receipt records the pick.
    if device == "gpu":
        import warp as wp

        if not wp.is_cuda_available():
            logger.warning("device=gpu requested but no CUDA device found, falling back to CPU")
        device = "cuda"
    self._launch.runtime.device = device
    if max_steps is not None:
        self._launch.runtime.max_steps = int(max_steps)
    self._launch.runtime.rtf = float(rtf)  # 0 = unthrottled; 1.0 = wall-clock pacing
    if clean is not None:  # None keeps the launch's default, $NEXUS_CLEAN
        self._launch.runtime.clean = clean
    # Rerun, serve or file but not both: view serves live on :9876; log writes the .rrd. Omitting *both*
    # builds no Logger at all: no recording, no per-tick log fan-out → max headless speed. The two are exclusive.
    if log and view:
        raise ValueError("Sim(log=, view=): write-.rrd and serve-viewer are mutually exclusive: pick one")
    self._launch.output.log = log
    self._launch.output.view = view
    self._launch.output.debug = debug  # axes-only scene: coordinate triads, no meshes → a small .rrd
    self._cache_dir = cache_dir
    # Absolute now, so a relative folder is the working folder's at construction, and a container's mount
    # gets the absolute path docker requires.
    self._run_dir = None if run_dir is None else os.path.abspath(run_dir)
    self._orch = None
    self._guidance = None  # a launch-built run has none: a flight hands one to _from_loop
    self._ran = False
    self._stopped = False
    self._prebuilt_orch = None  # set by _from_loop, the self-assembled-example entry

ports property

ports: Mapping[str, dict]

The run's port map: each link that leaves the run, by name, to the address a script opens its client on. The run owns every address. The builder builds a link's end inside the run from them, and names here each link whose other end a script holds. An entry is a plain mapping of what that client takes, such as {"protocol": "udp", "port": 14540, "system_id": 1} for the "offboard" link of a PX4 run, which the PX4 page of the reference documents. Open a client after sim.start(), and step the sim while it waits for an autopilot that runs on the sim's clock. A link with nothing behind it stays out of the map, and its lookup raises with the reason.

RETURNS	DESCRIPTION
Mapping[str, dict]	The port map, link name to entry.

RAISES	DESCRIPTION
RuntimeError	Accessed before entering the Sim context.

time property

time: float

The sim time of the last tick, seconds: 0.0 before the first tick.

Read off the loop's host clock, so it costs no device copy, and :meth:wait_until and :meth:sleep read it every tick. Every history's newest row carries the same time.

RAISES	DESCRIPTION
RuntimeError	Read before entering the Sim context.

base_body property

base_body: str

The base body's label, discovered from the model, not hardcoded: the body under vehicle/body/ a script reads the vehicle's pose from, and the one the sim clock :meth:wait_until and :meth:sleep use.

from_args classmethod

from_args(args: Namespace, **overrides) -> Sim

Construct a Sim from a :func:sim_argparser namespace, ignoring script-specific extras.

overrides win over the namespace, for the Sim kwargs the shared parser doesn't cover. So a script does Sim.from_args(args, max_steps=500).

Source code in nexus_uav/_src/api/sim.py

@classmethod
def from_args(cls, args: argparse.Namespace, **overrides) -> Sim:
    """Construct a ``Sim`` from a :func:`sim_argparser` namespace, ignoring script-specific extras.

    ``overrides`` win over the namespace, for the ``Sim`` kwargs the shared parser doesn't cover.
    So a script does ``Sim.from_args(args, max_steps=500)``.
    """
    from nexus_uav._src.diagnostics import diagnostics

    diagnostics.configure(args)  # the shared --profile/--trace/--benchmark flags, process-wide
    kw = {
        "vehicle": args.vehicle,
        "catalog": getattr(args, "catalog", None),
        "device": getattr(args, "device", "auto"),
        "scene": args.scene,
        "solver": getattr(args, "solver", None),
        "log": getattr(args, "log", False),
        "view": getattr(args, "view", False),
        "debug": getattr(args, "debug", False),
        "max_steps": getattr(args, "max_steps", None),
        "rtf": getattr(args, "rtf", 0.0),
        "layer": getattr(args, "layer", None),
        "clean": True if getattr(args, "clean", False) else None,
    }
    kw.update(overrides)
    return cls(**kw)

start

start(timeout: float | None = None) -> None

Drive the run's setup, returning once the first control tick has completed.

One step: the loop's first step() runs physics reset, the wait for the Kit render peer to boot and warm its stage when the vehicle renders, the start, which records the row at sim time zero, and the graph capture, controller.connect(), which binds the HIL port PX4 dials, the wait for the peer to dial in, and the preroll ticks while it boots, then flies one tick. So for a PX4 sim this is the "lockstep is up" verb, and it returns with a row recorded for every tick so far and the captured graph replayed.

Idempotent: a second call is a no-op once the run has started. Optional for a run with a setpoint controller, where :meth:run and :meth:step drive the same setup.

PARAMETER	DESCRIPTION
timeout	Override the assembly's preroll_timeout: seconds to wait for the peer to establish lockstep, 30 s by default. None keeps it. This is where the unbounded time is: physics reset and the Kit peer's start have bounds of their own. TYPE: float | None DEFAULT: None

RAISES	DESCRIPTION
RuntimeError	Called outside the Sim context manager, or the run ended during setup with no error, for example on stop().
RunError	The run's setup failed, for example PX4 never dialed in, never answered within the preroll timeout or exited, or the Kit render peer failed to start; the message names the peers' logs and quotes their error lines, and the run's teardown has run.

Source code in nexus_uav/_src/api/sim.py

def start(self, timeout: float | None = None) -> None:
    """Drive the run's setup, returning once the first control tick has completed.

    One step: the loop's first ``step()`` runs physics reset, the wait for the Kit
    render peer to boot and warm its stage when the vehicle renders, the start, which records the
    row at sim time zero, and the graph capture, ``controller.connect()``, which binds the HIL port
    PX4 dials, the wait for the peer to dial in, and the preroll ticks while it boots, then flies one tick.
    So for a PX4 sim this is the "lockstep is up" verb, and it returns with a row recorded for every
    tick so far and the captured graph replayed.

    Idempotent: a second call is a no-op once the run has started. Optional for a run with
    a setpoint controller, where :meth:`run` and :meth:`step` drive the same setup.

    Args:
        timeout: Override the assembly's ``preroll_timeout``: seconds to wait for the peer
            to establish lockstep, 30 s by default. ``None`` keeps it. This is where the
            unbounded time is: physics reset and the Kit peer's start have bounds of their own.

    Raises:
        RuntimeError: Called outside the ``Sim`` context manager, or the run ended during
            setup with no error, for example on ``stop()``.
        RunError: The run's setup failed, for example PX4 never dialed in, never answered within the
            preroll timeout or exited, or the Kit render peer failed to start; the message names the
            peers' logs and quotes their error lines, and the run's teardown has run.
    """
    if self._orch is None:
        raise RuntimeError("Sim.start() called outside the context manager (use `with nx.Sim(...) as sim:`)")
    if self._ran:
        return
    if timeout is not None:
        self._orch.preroll_timeout = float(timeout)
    if not self.step():
        raise RuntimeError("the sim ended during setup, before its first tick")

run

run() -> None

Run the sim synchronously to completion on the calling thread: exhaust the tick generator, setup, loop and teardown. Bounded by max_steps or the guidance's mission end; stop() ends it early, and a lost peer fails it.

Read any history, sim[path], with .history(), after it returns. A script that wants to observe or command mid-flight uses :meth:step or :meth:wait_until instead.

RAISES	DESCRIPTION
RuntimeError	Called outside the Sim context manager.
RunError	The run's setup failed, or the run lost a peer, PX4 or the Kit render peer, before it ended; the message names the peers' logs and quotes their error lines, and the run's teardown has run and closed the recording.

Source code in nexus_uav/_src/api/sim.py

def run(self) -> None:
    """Run the sim **synchronously** to completion on the calling thread: exhaust the tick
    generator, setup, loop and teardown. Bounded by ``max_steps`` or the guidance's mission end;
    ``stop()`` ends it early, and a lost peer fails it.

    Read any history, ``sim[path]``, with ``.history()``, after it returns. A script
    that wants to observe or command mid-flight uses :meth:`step` or :meth:`wait_until` instead.

    Raises:
        RuntimeError: Called outside the ``Sim`` context manager.
        RunError: The run's setup failed, or the run lost a peer, PX4 or the Kit render peer, before
            it ended; the message names the peers' logs and quotes their error lines, and the run's
            teardown has run and closed the recording.
    """
    if self._orch is None:
        raise RuntimeError("Sim.run() called outside the context manager (use `with nx.Sim(...) as sim:`)")
    self._ran = True
    # Blocks: runs every stage, the guidance's among them, each step; closes the recorder plus controller in its finally.
    self._orch.run()

step

step() -> bool

Advance the sim one control tick on the calling thread. Deterministic: the predicate or state you read between steps lands at exact tick boundaries, with no wall-clock.

The one driving verb for every controller: one that takes setpoints and PX4 alike advance one tick per call. Step, then read sim[path] between steps. Returns False when the run has ended: mission complete, max_steps or stopped. A lost peer raises instead.

RAISES	DESCRIPTION
RuntimeError	Called outside the Sim context manager.
RunError	The run's setup failed, or the run lost a peer, PX4 or the Kit render peer, before it ended; the message names the peers' logs and quotes their error lines, and the run's teardown has run and closed the recording.

Source code in nexus_uav/_src/api/sim.py

def step(self) -> bool:
    """Advance the sim one control tick on the calling thread. Deterministic: the predicate or state
    you read between steps lands at exact tick boundaries, with no wall-clock.

    The one driving verb for every controller: one that takes setpoints and PX4 alike advance
    one tick per call. Step, then read ``sim[path]`` between steps. Returns ``False`` when the run
    has ended: mission complete, ``max_steps`` or stopped. A lost peer raises instead.

    Raises:
        RuntimeError: Called outside the ``Sim`` context manager.
        RunError: The run's setup failed, or the run lost a peer, PX4 or the Kit render peer, before
            it ended; the message names the peers' logs and quotes their error lines, and the run's
            teardown has run and closed the recording.
    """
    if self._orch is None:
        raise RuntimeError("sim.step() called outside the context manager (use `with nx.Sim(...) as sim:`)")
    self._ran = True
    return self._orch.step()

read

read(path: str) -> Any

The live value of the signal at path, typed by the signal: what its writer wrote on the last tick.

path is the writer's path below the run's root then the signal's name, as a history's key, and a signal that keeps no history, a camera's frame say, has one too. sim.read(f"vehicle/body/{sim.base_body}") is the base body's :class:~nexus_uav.PoseTwist, so .position[2] is its height, and sim.read("vehicle/controllers/px4/controls") the 16 controls PX4 wrote, one array. A device signal reads without its leading axis of one, as an instance of its struct or an array of its other axes, and a host signal as the object its writer wrote. The value is a copy, which later ticks leave alone.

RAISES	DESCRIPTION
RuntimeError	Called before entering the Sim context.
KeyError	No component writes a signal at path.

Source code in nexus_uav/_src/api/sim.py

def read(self, path: str) -> Any:
    """The live value of the signal at ``path``, typed by the signal: what its writer wrote on the last tick.

    ``path`` is the writer's path below the run's root then the signal's name, as a history's key, and a
    signal that keeps no history, a camera's frame say, has one too. ``sim.read(f"vehicle/body/{sim.base_body}")``
    is the base body's :class:`~nexus_uav.PoseTwist`, so ``.position[2]`` is its height, and
    ``sim.read("vehicle/controllers/px4/controls")`` the 16 controls PX4 wrote, one array. A device signal
    reads without its leading axis of one, as an instance of its struct or an array of its other axes, and
    a host signal as the object its writer wrote. The value is a copy, which later ticks leave alone.

    Raises:
        RuntimeError: Called before entering the ``Sim`` context.
        KeyError: No component writes a signal at ``path``.
    """
    return self._loop().signals[path].read()

wait_until

wait_until(
    predicate: Callable[[], bool], sim_timeout: float
) -> None

Step the sim until predicate() is true, or sim_timeout sim seconds elapse.

Deterministic: this drives the sim itself, so it evaluates the predicate at exact tick boundaries and there is no wall-clock in the loop. Budgets in sim time, the lockstep clock, so a driver's timeouts are invariant to how fast the sim actually runs.

PARAMETER	DESCRIPTION
predicate	Zero-arg callable evaluated after each control tick. TYPE: Callable[[], bool]
sim_timeout	Budget in sim seconds, measured by advance of the sim clock, the lockstep, before giving up. TYPE: float

RAISES	DESCRIPTION
TimeoutError	sim_timeout sim seconds elapsed without the predicate becoming true, or the run ended first: mission complete or max_steps.
RunError	The run lost a peer before the predicate became true; the message names it.

Source code in nexus_uav/_src/api/sim.py

def wait_until(self, predicate: Callable[[], bool], sim_timeout: float) -> None:
    """Step the sim until ``predicate()`` is true, or ``sim_timeout`` sim seconds elapse.

    Deterministic: this drives the sim itself, so it evaluates the predicate at exact tick
    boundaries and there is no wall-clock in the loop. Budgets in sim time, the lockstep clock,
    so a driver's timeouts are invariant to how fast the sim actually runs.

    Args:
        predicate: Zero-arg callable evaluated after each control tick.
        sim_timeout: Budget in sim seconds, measured by advance of the sim clock, the lockstep,
            before giving up.

    Raises:
        TimeoutError: ``sim_timeout`` sim seconds elapsed without the predicate becoming
            true, or the run ended first: mission complete or ``max_steps``.
        RunError: The run lost a peer before the predicate became true; the message names it.
    """
    t0 = self.time
    while True:
        ran = self.step()
        if predicate():
            return
        if not ran:
            raise TimeoutError("the run ended before the condition was met")
        if self.time - t0 >= sim_timeout:
            raise TimeoutError(f"condition not met within {sim_timeout}s sim-time")

sleep

sleep(sim_seconds: float) -> None

Step the sim until sim_seconds of sim time elapse, not wall-clock.

Measures elapsed time by the advance of the sim clock, the lockstep, so the wait is invariant to the sim's real-time factor. Returns early if the run ends: mission complete or max_steps.

PARAMETER	DESCRIPTION
sim_seconds	Amount of sim time to wait, in seconds. TYPE: float

RAISES	DESCRIPTION
RunError	The run lost a peer before the time elapsed; the message names it.

Source code in nexus_uav/_src/api/sim.py

def sleep(self, sim_seconds: float) -> None:
    """Step the sim until ``sim_seconds`` of sim time elapse, not wall-clock.

    Measures elapsed time by the advance of the sim clock, the lockstep, so the wait is
    invariant to the sim's real-time factor. Returns early if the run ends: mission complete or
    ``max_steps``.

    Args:
        sim_seconds: Amount of sim time to wait, in seconds.

    Raises:
        RunError: The run lost a peer before the time elapsed; the message names it.
    """
    t0 = self.time
    while self.time - t0 < sim_seconds:
        if not self.step():
            break

results

results() -> dict

Run stats from the last run(): the loop's run_stats, which are control_steps, steady rtf and full_rtf. Empty before a run completes.

RETURNS	DESCRIPTION
dict	The run-stats mapping, for example control_steps, rtf and full_rtf; empty if no run has completed. TYPE: dict

Source code in nexus_uav/_src/api/sim.py

def results(self) -> dict:
    """Run stats from the last ``run()``: the loop's ``run_stats``, which are
    ``control_steps``, steady ``rtf`` and ``full_rtf``. Empty before a run completes.

    Returns:
        dict: The run-stats mapping, for example ``control_steps``, ``rtf`` and ``full_rtf``;
            empty if no run has completed.
    """
    return dict(self._orch.run_stats) if self._orch is not None else {}

artifacts

artifacts() -> dict

Return the artifacts produced by this run.

The Rerun .rrd this run wrote, merged with what every component and every peer of the run contributes through its artifacts(): the PX4 peer names its console log and its .ulg. Stays generic: it names no artifact itself.

RETURNS	DESCRIPTION
dict	At least {"rrd": <path or None>}, the .rrd path, or None if no recorder ran, updated with the entries the components and the peers contribute. TYPE: dict

Source code in nexus_uav/_src/api/sim.py

def artifacts(self) -> dict:
    """Return the artifacts produced by this run.

    The Rerun ``.rrd`` this run wrote, merged with what every component and every peer of the run
    contributes through its ``artifacts()``: the PX4 peer names its console log and its ``.ulg``. Stays
    generic: it names no artifact itself.

    Returns:
        dict: At least ``{"rrd": <path or None>}``, the ``.rrd`` path, or ``None``
            if no recorder ran, updated with the entries the components and the peers contribute.
    """
    if self._orch is None:
        return {"rrd": None}
    out = {"rrd": self._orch.logger.rrd_path if self._orch.logger is not None else None}
    for source in self._orch.components():
        out.update(source.artifacts())
    for peer in self._orch.peers:  # a peer's artifacts are optional, see Peer
        out.update(getattr(peer, "artifacts", dict)())
    return out

stop

stop() -> None

Tear the run down cooperatively: one path for every controller.

Signals the loop to stop, then closes the tick generator so its teardown runs: the Real-Time Factor (RTF) stamp, controller.close(), the stop of every peer the build started, the PX4 container, and the logging flush. Idempotent: repeat calls are no-ops. __exit__ calls it automatically.

Source code in nexus_uav/_src/api/sim.py

def stop(self) -> None:
    """Tear the run down cooperatively: one path for every controller.

    Signals the loop to stop, then closes the tick generator so its teardown runs:
    the Real-Time Factor (RTF) stamp, ``controller.close()``, the stop of every peer the build
    started, the PX4 container, and the logging flush. Idempotent: repeat calls are no-ops.
    ``__exit__`` calls it automatically.
    """
    if self._stopped:
        return
    self._stopped = True
    if self._orch is None:
        return
    self._orch.stop()
    # Step/run-driven, this closes the tick generator so its teardown, the RTF stamp plus renderer,
    # controller, peers and logs close, runs; a no-op if run() drove to completion. Entered but
    # never driven, it stops the peers started at build, the renderer's and PX4, and closes the
    # Logger built at construction, so the .rrd flushes or the :9876 server releases.
    self._orch.close()

A run that never starts, or that loses a peer before it ends, raises one error from whichever verb the script called: start, step, run, wait_until or sleep. Its message names each peer's console log and quotes the peer's own error lines. A script catches it to report an error, not a failed check. A run that ends on max_steps, on stop() or at its mission's end raises nothing.

RunError

Bases: RuntimeError

The run failed for a reason outside the script's own check: it never started, or it lost a peer before it ended.

A component never connected or never attached, a peer never answered or exited during the setup, or a peer stopped answering mid-run. The message names the cause, then each peer's files, such as its console log, and the peer's own error lines, so a script's output says why even after the machine goes away. A run that ends on max_steps, on stop() or at its mission's end raises nothing.

A script separates this error from its own check failing by catching it::

try:
    sim.wait_until(op.at_target, sim_timeout=60.0)
except nexus_uav.RunError:
    ...  # the run broke: report an error, not a failed check

The scripts share one argument parser, and save a run's artifacts with one call:

sim_argparser

sim_argparser(
    description: str | None = None,
) -> ArgumentParser

The shared argument parser for Sim-driven scripts and the nexus command-line tool.

Carries the common flags :meth:Sim.from_args reads, --vehicle / --device / --scene / --layer / --max-steps / --log / --view, so the command-line tool and any Sim-driven script share one arg surface. The bundled examples are zero-arg by design, and their configuration lives in the script; this parser serves the nexus command-line tool.

Source code in nexus_uav/_src/api/args.py

def sim_argparser(description: str | None = None) -> argparse.ArgumentParser:
    """The shared argument parser for ``Sim``-driven scripts and the ``nexus`` command-line tool.

    Carries the common flags :meth:`Sim.from_args` reads, ``--vehicle`` / ``--device``
    / ``--scene`` / ``--layer`` / ``--max-steps`` / ``--log`` / ``--view``, so the
    command-line tool and any Sim-driven script share one arg surface. The bundled examples are zero-arg by design, and their configuration
    lives in the script; this parser serves the ``nexus`` command-line tool.
    """
    p = argparse.ArgumentParser(description=description, formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    p.add_argument(
        "--vehicle", required=True,
        help="catalog vehicle NAME (e.g. astro_max_fpv) or a local .usd path",
    )  # fmt: skip
    p.add_argument(
        "--catalog", default=None,
        help="path to a catalog that extends the bundled and installed ones; omit to take the nearest\n"
        "nexus.catalog.yaml at or above the working directory, else only the bundled and installed ones",
    )  # fmt: skip
    p.add_argument(
        "--device", default="auto", choices=["cpu", "cuda", "auto"], help="compute device (cpu=deterministic)"
    )
    p.add_argument(
        "--scene", required=True,
        help="catalog scene NAME (e.g. 'empty' or 'slalom') or a local scene .usdz path (a converted\n"
        "mesh/splat, flown as the visual world)",
    )  # fmt: skip
    p.add_argument(
        "--solver", default=None, choices=["mujoco", "semi_implicit", "featherstone"],
        help="physics integrator (default: catalog default, mujoco)",
    )  # fmt: skip
    p.add_argument("--max-steps", type=int, default=None, help="cap the run at N control steps")
    p.add_argument(
        "--rtf", type=float, default=0.0,
        help="real-time-factor throttle: 0 = unthrottled (default); 1.0 = pace to wall-clock for interactive flying, "
        "which a vehicle with a free-running peer needs",
    )  # fmt: skip
    p.add_argument(
        "--layer", default=None, metavar="PATH",
        help="an override layer, a local USD file composed over the vehicle and the scene: it changes a\n"
        "declared value, such as the scene's nexus:geodeticOrigin, selects a variant, or drops a declaration;\n"
        f"one that deactivates the PX4 SITL peer's scope flies an autopilot started elsewhere, which dials the HIL port {HIL_PORT}",
    )  # fmt: skip
    p.add_argument(
        "--clean", action="store_true",
        help="start every peer from nothing an earlier run stored, such as PX4's parameters and mission; without it\n"
        "PX4 boots with what the last run on its airframe and instance stored, unless $NEXUS_CLEAN is 1",
    )  # fmt: skip
    p.add_argument("--log", action="store_true", help="write the Rerun .rrd to disk")
    p.add_argument("--view", action="store_true", help="serve the live Rerun viewer on :9876")
    p.add_argument(
        "--debug", action="store_true",
        help="axes-only scene: log each body's coordinate-frame triad instead of its mesh (much smaller .rrd)",
    )  # fmt: skip
    p.add_argument("--stats-json", default=None, help="write the run's stats dict to this JSON path")
    p.add_argument("--rrd-out", default=None, help="copy the run's .rrd here (e.g. for a docs page)")
    add_diagnostics_args(p)  # the shared --profile/--trace/--benchmark; Sim.from_args reads them
    return p

save_run_artifacts

save_run_artifacts(
    sim: Sim, args: Namespace, stats: dict | None = None
) -> None

Honor the shared --stats-json / --rrd-out flags after a run.

Writes stats, if given, to args.stats_json and copies the run's .rrd to args.rrd_out. Call it after the Sim context exits: the close flushes the .rrd, and sim.artifacts() then reports its path. A no-op for flags the caller didn't set.

Source code in nexus_uav/_src/api/args.py

def save_run_artifacts(sim: Sim, args: argparse.Namespace, stats: dict | None = None) -> None:
    """Honor the shared ``--stats-json`` / ``--rrd-out`` flags after a run.

    Writes ``stats``, if given, to ``args.stats_json`` and copies the run's ``.rrd`` to ``args.rrd_out``.
    Call it after the ``Sim`` context exits: the close flushes the ``.rrd``, and ``sim.artifacts()``
    then reports its path. A no-op for flags the caller didn't set.
    """
    import json
    import shutil

    if stats is not None and getattr(args, "stats_json", None):
        with open(args.stats_json, "w") as f:
            json.dump(stats, f, indent=2)
    dest = getattr(args, "rrd_out", None)
    if dest:
        rrd = (sim.artifacts() or {}).get("rrd")
        if rrd:
            shutil.copy2(rrd, dest)