Components, registry, and peers

from nexus_uav import Component, Stage, Tick, Signal, Registry, register, Peer

One contract serves every role: a component subclasses Component, states its per-tick work as a list of Stage, each declaring the Signals it reads and writes, and overrides the lifecycle members it has. The loop runs each stage over the shared Tick. A registry maps each schema a vehicle applies to the class that builds it, and a peer is a process the run starts and stops on the Peer contract. Bringing your own component walks through all four.

Component

The contract every component fills, whatever its role.

A component states its per-tick work in :meth:stages, each stage declaring the signals it reads and writes, which the loop wires before any stage runs. The members below are the whole of what the loop, the builder and Sim read from a component, so they find none by name: a class overrides the ones it has and inherits the rest. The role a component fills, which the schema that declares it states, fixes where its stages run in the tick and who declares it; the contract is the same for every role.

ATTRIBUTE	DESCRIPTION
name	The instance name, which ends the component's path, vehicle/sensors/<name>, and keys its recorded histories. The builder sets it for a declared component from its prim; None reads as the class's name. TYPE: str | None
requires	The names of the peers the class needs, such as ("kit",) for an RTX sensor: the builder starts each once, whatever the number of components that require it, and the loop hands the component each one's link at :meth:connect. TYPE: tuple[str, ...]
needs_instance	Whether the class takes the run's instance, the number the run's peers number their ports from, as the keyword instance: True for a controller whose peer it talks to numbers its links so. The builder passes it beside the schema's attributes. TYPE: bool
attached	Whether the component's peer has dialed in. True for a component with no peer, the default. A component that reads False before it connects waits on a peer: the loop connects it after the graph capture, so a peer that dials in early waits on no kernel load, holds the sim clock until it reads True, then ticks until the peer first answers. TYPE: bool

name class-attribute instance-attribute

name: str | None = None

requires class-attribute instance-attribute

requires: tuple[str, ...] = ()

needs_instance class-attribute instance-attribute

needs_instance: bool = False

attached class-attribute instance-attribute

attached: bool = True

stages

stages() -> list[Stage]

The component's per-tick work, in ring order: one or more stages, each declaring what it reads and writes. A device stage joins the captured graph; a host stage runs between replays. None by default, which fails the build: every component states at least one.

Source code in nexus_uav/_src/core/interfaces.py

def stages(self) -> list[Stage]:
    """The component's per-tick work, in ring order: one or more stages, each declaring what it reads and
    writes. A device stage joins the captured graph; a host stage runs between replays. None by default,
    which fails the build: every component states at least one.
    """
    return []

connect

connect(links: Mapping[str, Any]) -> None

Open what the component needs for the run: bind a port, load a model. Nothing by default. The loop connects a component before the start, or after the graph capture when it waits on a peer.

PARAMETER	DESCRIPTION
links	The link of each peer the class requires, by the peer's name, as requires lists them: the run started the peer and holds its link. Empty for a class that requires none. TYPE: Mapping[str, Any]

Source code in nexus_uav/_src/core/interfaces.py

def connect(self, links: Mapping[str, Any]) -> None:
    """Open what the component needs for the run: bind a port, load a model. Nothing by default. The loop
    connects a component before the start, or after the graph capture when it waits on a peer.

    Args:
        links: The link of each peer the class requires, by the peer's name, as ``requires`` lists them:
            the run started the peer and holds its link. Empty for a class that requires none.
    """

close

close() -> None

Release what :meth:connect opened, at the run's end. Nothing by default. Idempotent.

Source code in nexus_uav/_src/core/interfaces.py

def close(self) -> None:
    """Release what :meth:`connect` opened, at the run's end. Nothing by default. Idempotent."""

set_logger

set_logger(logger) -> None

Take the Logger scoped to the component's path, or None when the run records nothing. A component that logs a value with no fixed width live, a camera's frame or a guidance's markers, keeps it and logs only its own row's name; every fixed-width signal the Recorder keeps itself. Nothing by default.

Source code in nexus_uav/_src/core/interfaces.py

def set_logger(self, logger) -> None:
    """Take the Logger scoped to the component's path, or ``None`` when the run records nothing. A component
    that logs a value with no fixed width live, a camera's frame or a guidance's markers, keeps it and logs
    only its own row's name; every fixed-width signal the Recorder keeps itself. Nothing by default.
    """

artifacts

artifacts() -> dict

The files the component leaves for a reader after the run, by name, which Sim.artifacts collects beside the recording. Empty by default.

Source code in nexus_uav/_src/core/interfaces.py

def artifacts(self) -> dict:
    """The files the component leaves for a reader after the run, by name, which ``Sim.artifacts`` collects
    beside the recording. Empty by default.
    """
    return {}

ports

ports() -> Mapping[str, dict]

The links the component knows of that leave the run, by name, each to the address a script opens its client on, which the run's port map, Sim.ports, lists. Empty by default.

Source code in nexus_uav/_src/core/interfaces.py

def ports(self) -> Mapping[str, dict]:
    """The links the component knows of that leave the run, by name, each to the address a script opens its
    client on, which the run's port map, ``Sim.ports``, lists. Empty by default.
    """
    return {}

Stage dataclass

Stage(
    name: str,
    kind: Literal["device", "host"],
    run: Callable[[Tick], Any],
    reads: tuple[Signal, ...] = (),
    writes: tuple[Signal, ...] = (),
)

One unit of a component's per-tick work, the shape every loop component states in its stages(). A device stage joins the CUDA graph; a host stage runs between graph replays. run(tick) does the work; a host stage returns False when it produced nothing because its peer didn't answer, which a peer's preroll ticks on through and the steady loop ends the run on. Every stage but the clock's, the physics substeps' and a sensor's host stage also runs once in the start, over the settled state at sim time zero, before any capture. reads and writes are the signals the stage reads and writes, which the loop wires before any stage runs.

name instance-attribute

name: str

kind instance-attribute

kind: Literal['device', 'host']

run instance-attribute

run: Callable[[Tick], Any]

reads class-attribute instance-attribute

reads: tuple[Signal, ...] = ()

writes class-attribute instance-attribute

writes: tuple[Signal, ...] = ()

Tick dataclass

Tick(
    state: Any,
    t: SimTime,
    dt: float,
    timeout: float | None = None,
    base: int = 0,
    done: bool = False,
    control: Any = None,
)

The context one control tick hands every stage: the loop's own state, which a stage reads and writes in place. A value between two components is a signal the stages declare, not a field here, a sensor's sample among them. dt is one physics step, the control timestep over physics_substeps. timeout bounds a host stage's wait on its peer. base is the index of the vehicle's base body, its airframe, in state: a stage that reads the vehicle's true state reads that row. control is Newton's control the physics steps with, whose inputs a command element writes.

A stage sets done to end the run, a guidance's when its mission is over: the loop completes the tick and takes no further one.

state instance-attribute

state: Any

t instance-attribute

t: SimTime

dt instance-attribute

dt: float

timeout class-attribute instance-attribute

timeout: float | None = None

base class-attribute instance-attribute

base: int = 0

done class-attribute instance-attribute

done: bool = False

control class-attribute instance-attribute

control: Any = None

Signal

Signal(
    name: str,
    type: type,
    *,
    shape: tuple[int, ...] | None = None,
    default: Any = None,
)

One value a stage reads or writes: a name, a type and, for a device type, a shape.

PARAMETER	DESCRIPTION
name	The name that wires a reader to its writer. TYPE: str
type	A Warp struct or value type, whose buffer lives on the device, or any other class, whose buffer holds one object on the host. A reader that takes every writer of the signal declares list[T] and reads a list, one value per writer. TYPE: type
shape	The device buffer's shape. A reader reads the leading part of a wider buffer. TYPE: tuple[int, ...] | None DEFAULT: None
default	The value a reader takes until a component writes the signal, and for good when no component writes it. None gives no default. TYPE: Any DEFAULT: None

Source code in nexus_uav/_src/core/signals.py

def __init__(self, name: str, type: type, *, shape: tuple[int, ...] | None = None, default: Any = None):
    self.name = name
    self.type = type
    self.shape = None if shape is None else tuple(int(n) for n in shape)
    self.default = default
    self.buffer = None
    self.writers = ()

buffer instance-attribute

buffer: Any = None

The one buffer the builder hands the writer and every reader, before the capture; for a list input, the buffer of each writer.

writers instance-attribute

writers: tuple[str, ...] = ()

For a list input, the instance name of each writer whose buffer it holds, in the same order, which the builder sets when it wires the ring, so a reader of two or more writers has each one's name; empty otherwise.

name instance-attribute

name = name

type instance-attribute

type = type

shape instance-attribute

shape = (
    None
    if shape is None
    else tuple((int(n)) for n in shape)
)

default instance-attribute

default = default

device property

device: bool

Whether the signal's buffer lives on the device: its type is a Warp struct or value type.

read

read() -> Any

The signal's value on the host, typed by the signal: a device buffer without its leading axis of one, as an instance of its struct, read().position, or as an array of its other axes, and the object a host buffer holds. A device value is a copy, which later writes leave alone. A list input reads one value per writer, or its default when no component writes the signal.

Source code in nexus_uav/_src/core/signals.py

def read(self) -> Any:
    """The signal's value on the host, typed by the signal: a device buffer without its leading axis of one,
    as an instance of its struct, ``read().position``, or as an array of its other axes, and the object a host
    buffer holds. A device value is a copy, which later writes leave alone. A list input reads one value per
    writer, or its default when no component writes the signal.
    """
    if typing.get_origin(self.type) is list:
        return [_value(buffer) for buffer in self.buffer] if self.buffer else self.default
    return _value(self.buffer)

write

write(value: Any) -> None

Write the signal's value from the host, in place: a device buffer takes a copy of value, an array of the signal's shape, so the next graph replay reads it, and a host buffer holds value. A struct's value is a record per element, its fields in order, such as [(time, accel, gyro)].

Source code in nexus_uav/_src/core/signals.py

def write(self, value: Any) -> None:
    """Write the signal's value from the host, in place: a device buffer takes a copy of ``value``, an
    array of the signal's shape, so the next graph replay reads it, and a host buffer holds ``value``.
    A struct's value is a record per element, its fields in order, such as ``[(time, accel, gyro)]``.
    """
    if self.device:
        dtype = self.type.numpy_dtype() if wp.types.type_is_struct(self.type) else None
        self.buffer.assign(np.asarray(value, dtype=dtype))
    else:
        self.buffer.value = value
        self.buffer.writes += 1

Registry

Registry(entries: dict[str, type | str] | None = None)

The classes a run resolves its names to: this registry's own rows, then the install's.

PARAMETER	DESCRIPTION
entries	The rows of this registry: each name to a class or an import path, module:Class. A row here answers before the install does, so a test maps a schema to a stand-in and the install answers the rest. Registry() is the install alone. TYPE: dict[str, type | str] | None DEFAULT: None

Source code in nexus_uav/_src/core/registry.py

def __init__(self, entries: dict[str, type | str] | None = None):
    self._entries: dict[str, Target] = dict(entries or {})

add

add(name: str, target: type | str) -> None

Map name to target, a class or an import path, module:Class, replacing any row it had.

Source code in nexus_uav/_src/core/registry.py

def add(self, name: str, target: type | str) -> None:
    """Map `name` to `target`, a class or an import path, `module:Class`, replacing any row it had."""
    self._entries[name] = target

resolve

resolve(name: str) -> type | None

The class name maps to: this registry's row, else the install's, or None when neither claims it.

An import path, or the install's entry point, imports its module here, not before, so a class with heavy imports costs nothing until a run declares its schema.

Source code in nexus_uav/_src/core/registry.py

def resolve(self, name: str) -> type | None:
    """The class `name` maps to: this registry's row, else the install's, or ``None`` when neither claims it.

    An import path, or the install's entry point, imports its module here, not before, so a class with heavy
    imports costs nothing until a run declares its schema.
    """
    target = self._entries.get(name)
    if target is None:
        target = _installed().get(name)
    if target is None:
        return None
    if isinstance(target, str):
        target = EntryPoint(name=name, value=target, group=ENTRY_POINT_GROUP)
    return target.load() if isinstance(target, EntryPoint) else target

register

register(name: str, target: type | str) -> None

Map name to target, a class or an import path, module:Class, in the install's table, for a class a script defines rather than installs.

Every registry that maps name itself keeps its own row.

Source code in nexus_uav/_src/core/registry.py

def register(name: str, target: type | str) -> None:
    """Map `name` to `target`, a class or an import path, `module:Class`, in the install's table, for a class a
    script defines rather than installs.

    Every registry that maps `name` itself keeps its own row.
    """
    _installed()[name] = target

Peer

Bases: Protocol

A process the run starts, stops and can ask after.

start blocks until the process is running, or raises naming why it couldn't start. stop is idempotent: a peer that never started, or stopped already, stays stopped. alive says whether the process still runs; a peer that exited on its own reads dead, and a run waiting on it fails at once. A peer that writes files a reader wants after the run, a console log, names them in artifacts, which is optional. A peer that can tell its own error lines from the rest of its console, such as the line PX4 prints for an airframe its tree lacks, returns them from errors, which is optional too, and a run that fails quotes them in its RunError.

clock states how the peer keeps time, one of three ways. stepped: a component steps it each tick, as PX4 SITL runs on the sim time the lockstep link carries. on_request: it answers a request for a given sim time, as the Kit peer renders the frame a tick asks for. free: it runs on its own wall clock, as a companion computer under an emulator does. Sim time and a free-running peer's clock agree only while the run paces at a real-time factor of 1, so the build fails a run that declares one at any other factor, before any peer starts, and the loop warns when the live factor falls below 0.95.

A peer's schema can declare a serial line, a relationship nexus:serial:<line> whose target is the prim of the peer at the line's other end, as a companion computer's line to the flight controller. The run owns the link, :mod:.serial: it makes one per line before any peer starts and hands each of the two peers the path of its end in the keyword serial, the declaring peer by the line's name and the other by the declaring prim's name. A peer with an end takes serial beside the run's other values and opens the path it names.

A peer's schema can declare an input, a relationship nexus:inputs:<signal> whose targets are the prims whose components write that signal, one or more, as a camera's USB cable plugs into a companion computer. The loop end of that link is the run's: once per tick, after the writers, it hands the peer each writer's value through take(signal, writer, value), with the writer's instance name, which a peer with an input states. The peer decides what it makes of a value and how it carries it. The loop hands a host value once, on the write that produced it, so a frame reaches the peer at the camera's rate, and a device value every tick, since a kernel writes it every tick. A take that raises fails the run, and the error names the peer.

clock class-attribute

clock: Literal['stepped', 'on_request', 'free']

start

start() -> None

Source code in nexus_uav/_src/peers/__init__.py

def start(self) -> None: ...

stop

stop() -> None

Source code in nexus_uav/_src/peers/__init__.py

def stop(self) -> None: ...

alive

alive() -> bool

Source code in nexus_uav/_src/peers/__init__.py

def alive(self) -> bool: ...

ContainerPeer

ContainerPeer(*, instance: int, log_dir: str | PathLike)

Bases: Peer

A peer whose process is one container, on the Peer contract: the PX4 SITL and Kit peers subclass it, and so does a project's.

A subclass states name, the peer's kind, and run_options, what its container runs. The class does the rest. start removes a leftover container of this peer's name, runs the container under that name with the label nexus.peer=<name>, and streams its console into console.log under the peer's folder. stop removes the container, running or exited, and is idempotent. alive reads from the daemon whether the container still runs.

The run hands every peer its folder, and the peer's files live under <log_dir>/<name>/, dir: its console at log_path, console.log there, and whatever else it writes. The container's name, container, is nexus-<name>-<pid>-<instance>, so two runs in one process keep both containers.

PARAMETER	DESCRIPTION
instance	The run's instance, which names the container. TYPE: int
log_dir	The run's folder, which holds the peer's folder. TYPE: str | PathLike

Example

A project's peer that runs the image built from a folder it ships:

class Radio(ContainerPeer):
    name = "radio"

    def __init__(self, *, instance, log_dir, **run):
        super().__init__(instance=instance, log_dir=log_dir)

    def run_options(self):
        return {"image": ensure_image(IMAGE_DIR, "radio"), "command": ["radio", "--serve"]}

Source code in nexus_uav/_src/peers/containers.py

def __init__(self, *, instance: int, log_dir: str | os.PathLike) -> None:
    self.instance = instance
    self.dir = Path(log_dir) / self.name
    self.log_path = self.dir / "console.log"
    self.container = f"nexus-{self.name}-{os.getpid()}-{instance}"
    self._container: Container | None = None

name class-attribute instance-attribute

name: str = ''

The peer's kind: the value of its label, the stem of its container's name and the name of its folder.

instance instance-attribute

instance = instance

dir instance-attribute

dir = Path(log_dir) / self.name

log_path instance-attribute

log_path = self.dir / 'console.log'

container instance-attribute

container = f'nexus-{self.name}-{os.getpid()}-{instance}'

run_options

run_options() -> dict[str, Any]

What the container runs: image, command and the other keywords of the docker SDK's containers.run, such as volumes, environment or labels. Called by start, so an image a peer builds or pulls at start resolves here.

Source code in nexus_uav/_src/peers/containers.py

def run_options(self) -> dict[str, Any]:
    """What the container runs: ``image``, ``command`` and the other keywords of the docker SDK's
    ``containers.run``, such as ``volumes``, ``environment`` or ``labels``. Called by ``start``, so an image a
    peer builds or pulls at start resolves here.
    """
    raise NotImplementedError(f"{type(self).__name__} states no run_options")

start

start() -> None

Run the container, after removing a leftover of its name; its console streams to log_path.

RAISES	DESCRIPTION
RuntimeError	The daemon is unreachable; see :func:client.
DockerException	The daemon refused the run.

Source code in nexus_uav/_src/peers/containers.py

def start(self) -> None:
    """Run the container, after removing a leftover of its name; its console streams to ``log_path``.

    Raises:
        RuntimeError: The daemon is unreachable; see :func:`client`.
        docker.errors.DockerException: The daemon refused the run.
    """
    self.dir.mkdir(parents=True, exist_ok=True)
    options = self.run_options()
    labels = {**options.pop("labels", {}), LABEL: self.name}  # the peer label last, so no subclass unsets it
    self._container = run_container(name=self.container, log_path=str(self.log_path), labels=labels, **options)

stop

stop() -> None

Remove the container this peer started, running or exited. Idempotent.

Source code in nexus_uav/_src/peers/containers.py

def stop(self) -> None:
    """Remove the container this peer started, running or exited. Idempotent."""
    if self._container is None:
        return
    stop_container(self.container)
    self._container = None

alive

alive() -> bool

Whether the container still runs; one that exited, or that the daemon no longer holds, reads dead.

Source code in nexus_uav/_src/peers/containers.py

def alive(self) -> bool:
    """Whether the container still runs; one that exited, or that the daemon no longer holds, reads dead."""
    from docker.errors import NotFound

    if self._container is None:
        return False
    try:
        self._container.reload()
    except NotFound:
        return False
    return self._container.status in ("created", "running")

ensure_image

ensure_image(root: Path, name: str) -> str

Build the image from root when this machine lacks its tag, and return the tag.

The build prints its own output as it goes.

RAISES	DESCRIPTION
RuntimeError	The build failed, with the build's own error, or the daemon is unreachable.

Source code in nexus_uav/_src/peers/containers.py

def ensure_image(root: Path, name: str) -> str:
    """Build the image from ``root`` when this machine lacks its tag, and return the tag.

    The build prints its own output as it goes.

    Raises:
        RuntimeError: The build failed, with the build's own error, or the daemon is unreachable.
    """
    from docker.errors import ImageNotFound

    tag = image_tag(root, name)
    c = client()
    try:
        c.images.get(tag)
        return tag
    except ImageNotFound:
        pass
    logger.info(f"building the image {tag} from {root}: once per machine and per change to that folder")
    for chunk in c.api.build(path=str(root), tag=tag, rm=True, decode=True):
        if "error" in chunk:
            raise RuntimeError(f"building the image {tag} failed: {chunk['error'].strip()}")
        line = chunk.get("stream")
        if line:
            sys.stderr.write(line)
            sys.stderr.flush()
    logger.info(f"built the image {tag}")
    return tag