Writing a Driver

A driver is the only component that knows the bytes of a target. It turns the semantic telecommands of the MCS (a name, an instance, typed arguments in the unit of the catalogue) into the application units of the target's ICD, and the telemetry the target sends back into raw values of the catalogue. The MCS applies the calibration, computes derived measures and checks verifications: the driver does not.

A driver is a black box behind the NATS contract. This page shows how to write one with the Rust or the Python SDK; the NATS contract reference gives the messages for any other language.

What a driver does#

Everything else is the SDK's: registration, key pair and credentials, bindings, heartbeats, consumers, deadlines, events, echo bookkeeping, publication of values and chunks, and the decoding requests.

A first driver#

The lab-psu catalogue has one component, psu, three telecommands and three measures. Its driver turns telecommands into SCPI-like lines and status lines into samples. The same driver in both SDKs (examples/rust/drivers/lab-psu, examples/python/driver.py):

use stellar_sdk::contract::{DriverCapabilities, RawSample, SemanticTc};
use stellar_sdk::{Driver, driver_main};

/// Encodes `set_voltage 28 V` as `VOLT 28`, decodes `VOLT 28.1;CURR 0.5;OUTP 1`.
struct LabPsu;

impl Driver for LabPsu {
    fn software(&self) -> (String, String) {
        ("scpi-psu".to_owned(), "1.0.0".to_owned())
    }

    fn capabilities(&self) -> DriverCapabilities {
        let names = |items: &[&str]| items.iter().map(|i| format!("psu.{i}")).collect();
        DriverCapabilities {
            codec: "scpi-psu".to_owned(),
            catalogue: "lab-psu@^1.0".to_owned(),
            telecommands: names(&["set_voltage", "output_on", "output_off"]),
            measures: names(&["voltage", "current", "output_enabled"]),
            output: None,
            params_schema: None,
        }
    }

    fn encode(&self, tc: &SemanticTc) -> Result<Vec<u8>, String> {
        let line = match tc.telecommand.as_str() {
            "set_voltage" => {
                let volts = tc.args.get("voltage").and_then(serde_json::Value::as_f64);
                format!("VOLT {}", volts.ok_or("missing `voltage`")?)
            }
            "output_on" => "OUTP 1".to_owned(),
            "output_off" => "OUTP 0".to_owned(),
            other => return Err(format!("unknown telecommand `{other}`")),
        };
        Ok(line.into_bytes())
    }

    fn decode(&self, frame: &[u8]) -> Result<Vec<RawSample>, String> {
        let text = std::str::from_utf8(frame).map_err(|e| e.to_string())?;
        text.split(';')
            .map(|field| {
                let (key, value) = field.trim().split_once(' ').ok_or("expected `KEY value`")?;
                let number: f64 = value.parse().map_err(|_| format!("bad value `{value}`"))?;
                let (measure, value) = match key {
                    "VOLT" => ("voltage", serde_json::json!(number)),
                    "CURR" => ("current", serde_json::json!(number)),
                    "OUTP" => ("output_enabled", serde_json::json!(number != 0.0)),
                    other => return Err(format!("unknown field `{other}`")),
                };
                Ok(RawSample {
                    component: "psu".to_owned(),
                    instance: None,
                    measure: measure.to_owned(),
                    value,
                    onboard_time: None,
                })
            })
            .collect()
    }
}

fn main() -> anyhow::Result<()> {
    driver_main(LabPsu, "scpi-psu-1")
}

from collections.abc import Iterator

from stellar_mcs import Driver, RawSample, SemanticTc

driver = Driver(
    "scpi-psu",
    "1.0.0",
    # 1.0.0 and the later 1.x versions of the catalogue.
    catalogue="lab-psu@^1.0",
    telecommands=["psu.set_voltage", "psu.output_on", "psu.output_off"],
    measures=["psu.voltage", "psu.current", "psu.output_enabled"],
)


@driver.encode
def encode(tc: SemanticTc) -> str:
    """``set_voltage(voltage=28 V)`` → ``VOLT 28.0``. Raising fails it with the message."""
    match tc.telecommand:
        case "set_voltage":
            return f"VOLT {tc['voltage']}"
        case "output_on":
            return "OUTP 1"
        case "output_off":
            return "OUTP 0"
    raise ValueError(f"unknown telecommand `{tc.telecommand}`")


@driver.decode
def decode(frame: bytes) -> Iterator[RawSample]:
    """``VOLT 28.0;CURR 0.5;OUTP 1`` → three samples of the ``psu`` component."""
    for field in frame.decode().strip().split(";"):
        key, value = field.split()
        match key:
            case "VOLT":
                yield RawSample("psu", "voltage", float(value))
            case "CURR":
                yield RawSample("psu", "current", float(value))
            case "OUTP":
                yield RawSample("psu", "output_enabled", value == "1")


if __name__ == "__main__":
    driver.run()

The topology names the driver by its software name and a version requirement: scpi: {driver: scpi-psu@1, gateway: lab-eth-1, default: true}.

Life of an instance#

sequenceDiagram
    participant D as Driver
    participant R as Reconciler
    participant M as MCS (executor, compute stage)
    D->>R: stellar.ctl.register (Registration, public key U…)
    R-->>D: accepted (or rejected: final)
    R->>D: credentials verb (JWT with its permissions)
    R->>D: bindings verb (links served)
    loop every heartbeat period
      D->>R: stellar.ctl.hb.driver.<instance>
    end
    M->>D: SemanticTc on stellar.tc.encode.<codec>.<target>
    D->>M: ENCODED / ENCODE_FAILED, frame or unit

1. Start. The driver connects with its bootstrap credentials (STELLAR_NATS_CREDENTIALS), which only allow registration and heartbeats, and generates an nkey user key pair. The private seed never leaves the process.
2. Registration on stellar.ctl.register: kind driver, instance name, software name and semantic version, heartbeat period, public key, and the driver section below. Without a reply (reconciler not started, fail-over in progress), the SDK retries with an increasing delay up to 10 s. A rejected reply is final: the reason is logged and the driver stops (SessionError::Rejected in Rust, RegistrationRejected in Python).
3. Credentials. Once bound to a target, the reconciler issues a user JWT for the registered public key, with exactly the permissions of the driver, and delivers it with the credentials verb. It renews it at half-life (reconciler.jwt_ttl, 10 min by default) while the driver stays bound. The SDK reconnects with it, signing the server nonce with its own key, without losing subscriptions.
4. Bindings. The reconciler delivers the links the driver serves with the bindings verb, whenever they change and after a restart. Each link gives the target, the link name, the configuration revision (config, stamped as Stellar-Config), the codec, the subject of the telecommands to encode, the subject to publish on (the gateway's uplink, or the transport's wrap), the subject to decode (the gateway's raw telemetry, or the transport's units), and the parameters of the link.
5. Heartbeats every second on stellar.ctl.hb.driver.<instance>, with the health (healthy, or degraded with a reason). After three missed periods the instance is considered gone and its links unbound.
6. Control verbs on stellar.ctl.rpc.driver.<instance>.<verb>: status, credentials, bindings and reregister (the reconciler asks for it when it receives the heartbeat of an instance it no longer knows) are answered by the SDK; add your own with the Rpc trait (Rust) or @driver.rpc("verb") (Python). An error reply is {"error": "…"}.

Registration: catalogue and coverage#

The reconciler binds a driver to a link when:

• its software name is the driver name of the link, in a version satisfying the link (@4 means 4.x);
• the version of the catalogue of the target satisfies its catalogue requirement;
• its coverage includes every telecommand and measure of the components the link carries (the whole catalogue for the default link), except the generated measures of files and stream, which the transfer manager produces;
• the chain fits: its output is the input of the transport, or the uplink link type of the gateway without transport;
• the resolved parameters of the link, in each environment, satisfy its params_schema.

Otherwise it refuses with the reason, shown as the readiness of the target: driver … implements platform-v3@^1.4, not platform-v3@2.0.0, does not cover 1 of platform-v3@1.5.0, such as tcu.reboot.

A link that carries some components only (components: [tcu]) needs a driver covering those components: tcu-can covers tcu.* of platform-v3@^1.4 and declares nothing else.

Encoding: the telecommand chain#

The executor publishes a SemanticTc on the encode subject of the binding, in the TC_COMMANDS stream, with the headers Stellar-Config, Stellar-Correlation (the telecommand id), Stellar-Deadline and Nats-Msg-Id <tc_id>-encode. The SDK consumes it with the durable consumer encode_<codec>_<target>, one message at a time, so that the telecommands of a target keep their order.

Then the SDK:

1. refuses a telecommand received after its deadline (ENCODE_FAILED);
2. calls encode (or encode_with): an error gives ENCODE_FAILED with its message;
3. publishes ENCODED (ACK 1) on stellar.tc.evt.<target>.<tc_id>, then the frame on the uplink subject (Nats-Msg-Id <tc_id>-uplink), or the unit on the transport's wrap subject (<tc_id>-wrap), so that ACK 1 always precedes ACK 2.

In Python, encode returns bytes or str; an exception fails the telecommand with its message. tc["voltage"], tc.get("ramp", 10.0) and tc.bytes_arg("data") read the arguments; tc.name is component.telecommand.

Decoding telemetry#

The SDK subscribes to the raw telemetry of the gateway of each binding (stellar.tm.raw.<gateway>.<target>), or to the units of the transport (stellar.tm.unit.<transport>.<target>). For each frame it calls decode (or decode_with) and publishes the raw values on stellar.tm.decoded.<target> as a list of RawSample:

The message carries Stellar-Ground-Time and Stellar-Delivery of the frame, Stellar-Link, Stellar-Config (revision of the binding), Stellar-Driver (<software>@<version>, so that an archive can be decoded again with the same driver) and Stellar-Environment when known. A frame that cannot be decoded is logged; it stays archived as received in TM_RAW.

Links of a target that go through the same gateway share their driver: it decodes the frames of that gateway once, without duplicated samples.

The link context#

The parameters of a link (params in the topology, overridden key by key by environments.<env>.params) reach the driver with every telecommand and frame, resolved for an environment, as a LinkContext: target, link, environment, params, and the current mode of the target with its parameters in that mode (see Modes and Parameters).

• The environment of a telecommand is its environment; a telecommand without one (sent by the MCS itself, such as a transfer) takes that of the last telecommand of its link.
• A frame received is decoded in the environment of the last telecommand of its link, or with the base parameters before any.

The obc-csp driver reads the CSP address of the ground from the link:

fn encode(&self, _tc: &SemanticTc) -> Result<Vec<u8>, String> {
    Err("the CSP address of the ground comes from the link (`csp_address`)".to_owned())
}

fn encode_with(&self, tc: &SemanticTc, context: &LinkContext) -> Result<Vec<u8>, String> {
    let src = u16::try_from(context.param_u64("csp_address")?)
        .map_err(|_| "parameter `csp_address` out of range".to_owned())?;
    self.packet(tc, src)?.encode(self.version)
}

@driver.encode
def encode(tc, context):
    return csp(src=context.param("csp_address"), dst=1, dport=4, payload=tc["text"])

In Rust, LinkContext::param, param_u64 and param_str read a parameter of the link (the last two with an error naming the link), and LinkContext::parameter("tcu[TCU1].anode_voltage") a parameter of the target in its mode, falling back on the value given for every instance (tcu.anode_voltage); in Python, context.parameter(name, default) does the same. In Python, a decorated encode or decode that takes a second argument receives the context; context.param(name, default) reads a parameter. Declare the parameters in params_schema, so that the reconciler refuses a link that misses or mistypes them.

Echoes#

For verify: [echo], the driver compares each frame received with the telecommands it encoded in the last 60 s. echo(tc, frame) returns Some(true) (Python True) for a conforming echo of tc, Some(false) for a non-conforming one, None when the frame is not its echo. The SDK publishes an ECHO event with conforming; a non-conforming echo gives VERIFY_FAILED.

The platform-v3 driver recognises the PUS verification reports of the packets it sent:

fn echo(&self, tc: &SemanticTc, unit: &[u8]) -> Option<bool> {
    let (_, identification, control) = *lock(&self.sent)
        .iter()
        .rev()
        .find(|(id, ..)| *id == tc.id)?;
    tm::decode(unit)
        .ok()?
        .iter()
        .find_map(|report| match report {
            Report::Verification { kind, identification: i, control: c }
                if *i == identification && *c == control => match kind {
                    Verification::CompletionSuccess => Some(true),
                    Verification::AcceptanceFailure | Verification::CompletionFailure => Some(false),
                    Verification::AcceptanceSuccess => None,
                },
            _ => None,
        })
}

Files#

When the catalogue declares a files component (see Files and Streams):

• Directory. The listing telecommand (transfer.list) brings the on-board directory down. Decode it into RawSamples of the files component, one instance per file named by its decimal identifier: file_type, size, checksum (hexadecimal) and generation.
• Chunks. The read telecommand (transfer.read) brings chunks down. chunks(frame) returns the FileChunks a frame carries (file_id, generation, offset, data); the SDK publishes each raw on stellar.file.chunk.<target>.<file_id> with Stellar-File-Generation, Stellar-File-Offset, Stellar-Link and Stellar-Config.
• Uploads. The write telecommand (transfer.write) is encoded like any telecommand, its data in hexadecimal. A chunk counts as written once its telecommand is VERIFIED (declare verify: [echo] and recognise the on-board acknowledgement in echo) or COMPLETE.
• Recorded telemetry. Once a file is verified, the transfer manager asks the drivers of its target to decode it (stellar.file.decode.<target>, queue group). The SDK reads the file from the stellar_files object store and calls decode_file(file_type, content). Return None for a type the driver does not decode (the transfer stays VERIFIED); the samples, each with its onboard_time, are published with Stellar-Delivery: deferred, one message per on-board time, and the transfer becomes PROCESSED.

fn decode_file(&self, file_type: &str, content: &[u8]) -> Option<Result<Vec<RawSample>, String>> {
    (file_type == "lttm").then(|| tm::decode_lttm(content))
}

@driver.decode_file
def decode_file(file_type, content):
    if file_type != "lttm":
        return None
    return list(lttm_samples(content))   # your decoder: RawSample with onboard_time

CFDP#

A platform declaring transfer: {protocol: cfdp} transfers its files with CFDP class 2 (CCSDS 727.0). The ground entity runs in the driver, on the fork of cfdp-rs in vendor/cfdp-rs; only the Rust SDK provides it.

The transfer manager drives the entity through the cfdp verb (stellar.ctl.rpc.driver.<instance>.cfdp) with a CfdpRequest ({target, file_id, generation, direction, file_type, size, source}), every 5 s while the transfer is active and the link available. Downloads use a Proxy Put Request; the data received is published as chunks, exactly like the chunked protocol. Uploads read the content from stellar_uploads. Transactions live in memory: after a restart, the next request starts them again. See File Transfers.

Testing: vectors from the ICD#

A driver is validated against its ICD, not against itself. The example drivers keep test vectors: telecommand → bytes pairs and frame → values pairs, generated from the ICD by an independent reference implementation and replayed in CI.

• examples/rust/drivers/platform-v3/vectors/ holds telecommands.json, telemetry.json and files.json, written by generate.py, a dependency-free Python script that implements the ICD (examples/icd/platform-v3.md) on its own. The CI checks that they are up to date, then tests/vectors.rs replays them bit for bit through the driver and the ccsds-tc transport: telecommands through the FOP from the V(S) of the vector, telemetry frames into packets, values, verification reports, chunks and CLCWs, files into values.
• examples/rust/drivers/tcu-can/vectors/ does the same for the CAN bench bus.
• tests/catalogue.rs of platform-v3 and obc-csp checks that the coverage of the driver matches the catalogue of examples/config.

The COP-1 state (V(S), CLCW) is part of a vector, so that frames are reproducible bit for bit.

Checking a driver on a running MCS#

Every driver answers the control verbs encode and decode, provided by both SDKs: the MCS asks it what it makes of a telecommand or a frame, in the context of a link, and nothing is sent or published. The API and the CLI call them, on the driver bound to the link or on one named with driver (a driver being developed, registered but not bound):

# The frame of a telecommand, resolved as a direct telecommand (units, ranges, defaults).
stellar encode psu-lab-2 set_voltage 'voltage=28 V' --environment AIT
# psu.set_voltage to psu-lab-2 via scpi (driver scpi-psu-1, environment AIT)
# arguments: voltage=28.0
# frame, 9 B:
#   0000  56 4f 4c 54 20 32 38 2e 30                       VOLT 28.0
# text: "VOLT 28.0"

# The samples of a frame, calibrated by the catalogue.
stellar decode psu-lab-2 --text 'VOLT 27.9;CURR 0.5;OUTP 1' --driver scpi-psu-dev
stellar decode sat1-fm 1801c00000050a02... --link nominal
stellar decode sat1-fm --file frame.bin

The context is the one the link has in the current configuration: its parameters in the environment, the mode of the target and its parameters. On a link with a transport, encode gives the unit the driver hands to the transport, and decode takes the unit the transport hands to the driver. A driver keeping a state, such as a packet sequence counter, may advance it when asked to encode. The routes are POST /v1/encode and POST /v1/decode (see the API reference); driver::unbound when no driver is bound to the link and none is named, driver::no-answer when the driver does not answer, driver::encode-failed and driver::decode-failed with the reason of the driver.

Generate a typed skeleton#

For a Python driver, stellar generate driver writes the skeleton from the catalogue:

stellar generate driver lab-psu --repository examples/config -o lab_psu --software scpi-psu

_generated.py holds what the catalogue says (coverage and requirement, enums, a typed class of arguments per telecommand, builders of raw samples per component, and a base class dispatching each telecommand to its encode_<component>_<telecommand> method); it is generated again with each version of the catalogue, and --check tells in CI when it is stale. driver.py and test_driver.py are yours, written once: the encoding and decoding from the ICD, and a test per telecommand. See Code Generators.

Running a driver#

driver_main(driver, default_instance) (Rust) and driver.run() (Python) read their settings from the environment and stop cleanly on Ctrl-C or SIGTERM:

STELLAR_NATS_URL=tls://nats.lab:4222 \
STELLAR_NATS_CREDENTIALS=/etc/stellar/bootstrap.creds \
STELLAR_NATS_CA=/etc/stellar/ca.pem \
STELLAR_INSTANCE=platform-v3-ccsds-1 \
platform-v3-ccsds

Run several instances under different names for availability: a new link goes to the eligible instance that serves the fewest, and an instance that leaves hands its links over. To embed a driver in a larger program, use run_driver(driver, &options, shutdown) (Rust) or async with driver.running(options): (Python). See Rust SDK and Python SDK.