Python SDK

The Python SDK (stellar-mcs, in sdk/python) implements the contracts of Stellar Control for drivers, transports, gateways and output connectors over NATS, and for pass feeders over the HTTP API. A component only declares what it does, with decorators; the SDK handles registration with the reconciler, key pair and JWT renewal, heartbeats, the bindings it receives, JetStream consumers, telecommand events (ACK 1 and ACK 2), deadlines, fragments, control verbs and reconnection.

The same package holds the client of the API: it launches runs of procedures, follows their events, answers their questions and controls them, from a script, a test bench or a notebook.

Install#

The package needs Python 3.11 or later; its dependencies are nats-py (with nkeys), httpx and websockets.

uv add stellar-mcs          # or: pip install stellar-mcs

Within the repository, examples/python uses the SDK from sdk/python directly ([tool.uv.sources] stellar-mcs = { path = "../../sdk/python", editable = true }); uv run installs it on first use.

Public API#

Everything is imported from the package root:

from stellar_mcs import (
    Driver, Transport, Gateway, Connector, PassFeeder,          # components
    SemanticTc, RawSample, FileChunk, LinkContext, Link,        # messages
    Downlink, Message, Measure, Pass, Replaced,                 # helpers
    Bindings, BoundLink, Health, TcState, Timestamp, now,
    Options, RegistrationRejected, PassRefused,
    Client, Run, RunEvent, RunResult, RunStatus, Answer, Question, ApiError,   # API client
)

Components#

instance is the default instance name, <software>-1 otherwise (the connector name for a connector); STELLAR_INSTANCE overrides it.

Declarations#

Decorated functions may be plain or async.

Messages#

Every component#

RegistrationRejected is raised when the reconciler refuses the registration: it is final.

Examples#

from stellar_mcs import Driver, RawSample

driver = Driver(
    "scpi-psu", "1.0.0",
    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):
    return f"VOLT {tc['voltage']}"

@driver.decode
def decode(frame):
    for field in frame.decode().split(";"):
        key, value = field.split()
        if key == "VOLT":
            yield RawSample("psu", "voltage", float(value))

driver.run()

examples/python carries a SCPI power supply into the MCS without hardware:

cd examples/python
uv run instrument.py --port 5025
uv run driver.py
STELLAR_INSTANCE=lab-eth-1 uv run gateway.py --target psu-lab-2 --instrument 127.0.0.1:5025
uv run connector.py --output export

stellar send psu-lab-2 psu.output_on
stellar send psu-lab-2 psu.set_voltage voltage="28 V"   # ENCODED, SENT, then VERIFIED
stellar watch psu-lab-2

The reconciler binds the driver and the gateway to psu-lab-2 of examples/config; the connector csv-export fills export/measures.csv and export/tc_events.csv, and resumes without duplicates after a restart.

A gateway that follows the mode of its target#

A gateway that must adapt the medium to the mode of its target (a data rate, a modulation) reads the parameters of the target from the context. @gateway.configure applies them when the link is bound and each time the binding changes, for instance after a configure of a procedure or stellar mode; the send can also take the context of each frame:

from stellar_mcs import Gateway, Link, LinkContext

gateway = Gateway("uhf-gateway", "1.0.0", tags=["uhf"], uplink=Link("rf", 9600))

@gateway.configure
async def configure(context: LinkContext) -> None:
    await modem.set_rate(context.parameter("rf.tx_baudrate"))

@gateway.send
async def send(target: str, frame: bytes, context: LinkContext) -> None:
    await modem.transmit(frame, power=context.parameter("rf.tx_power"))

See Modes and Parameters and Writing a Gateway.

API client#

Client talks to the HTTP API of a cell. The MCS executes a run (leases, confirmations, log, validated versions); the client only launches it, follows its events, answers its questions and decisions, and controls it. Everything is async.

from stellar_mcs import Answer, Client

async with Client("http://mcs:8080", "alice") as mcs:        # or token="…"
    result = await mcs.execute(
        "Hot standby test",
        environment="AIT",
        targets={"sat": "flatsat-1", "psu": "psu-lab-2"},
        inputs={"tcu": "TCU2", "bus_voltage": "28 V"},
        on_ask=lambda question: True,       # confirms every question
    )
    assert result.ok, result.reason

Client#

Client(url, identity=None, *, token=None, timeout=10.0, reconnections=5, transport=None)

Use it as an async context (async with Client(...) as mcs), or call await mcs.close().

Run#

A run launched on the MCS: id, ir (hash of the resolved run) and warnings (for instance a run that may outlast the pass in progress), as launch returned them.

RunResult holds run, ok, reason and every events of the log; result.failed_steps() lists the StepFinished attempts that failed.

Questions and answers#

on_ask receives a Question: run, path (the step asking), kind ("ask" or "decision"), prompt (the question, or why a decision is needed) and role (the role whose confirmation is awaited, for a hazardous telecommand). It may be plain or async, and returns:

Who may answer is decided by the MCS, as for any answer: see Questions, Decisions and Control and Hazardous Confirmations.

Events#

run.events() and on_event give typed events, from stellar_mcs.client. Every event has run, seq (position in the log, from 1), at, event (its kind) and data (the event as the API sent it).

ModeChanged is published when a configure sets the mode of a target:

from stellar_mcs.client import ModeChanged

if isinstance(event, ModeChanged):
    print(f"{event.target} is now in {event.mode}")

An event of a kind the SDK has no class for yet comes as a plain RunEvent, its fields in data (event.event gives its kind).

See Running Procedures for the meaning of each event.

Errors#

A request the API refuses raises ApiError: status (HTTP), errors (the [{code, message, help}] of the error body) and codes:

from stellar_mcs import ApiError

try:
    run = await mcs.launch("Hot standby test", environment="in_orbit", targets={"sat": "sat1-fm"})
except ApiError as refused:
    if "run::must-be-scheduled" in refused.codes:
        ...

run.events() raises ApiError too when the WebSocket is refused (an unknown run).

A command-line runner#

examples/python/run_procedure.py runs any procedure, prints its events as they come and answers at the terminal (--yes confirms every question). It exits with 0 when the run succeeds, 1 when it fails, 2 when the API refuses it:

cd examples/python
uv run run_procedure.py "Hot standby test" --environment AIT \
    --target sat=sim-1 --target psu=psu-sim-1 --input tcu=TCU2 --input "bus_voltage=28 V"

--api defaults to STELLAR_API_URL (http://127.0.0.1:8080), --identity to STELLAR_USER (the login name). Its core:

from stellar_mcs.client import Checked, Finished, StepStarted, TelecommandSent

def show(event: RunEvent) -> None:
    match event:
        case StepStarted(path=path, attempt=attempt):
            print(f"▶ {path}" + (f" (attempt {attempt})" if attempt > 1 else ""))
        case TelecommandSent(telecommand=telecommand, target=target):
            print(f"  → {telecommand} to {target}")
        case Checked(condition=condition, ok=ok):
            print(f"  {'✓' if ok else '✗'} {condition}")
        case Finished(ok=ok, reason=reason):
            print("run succeeded" if ok else f"run failed: {reason}")

async with Client(args.api, args.identity) as mcs:
    run = await mcs.launch(args.procedure, environment=args.environment,
                           targets=targets, inputs=inputs)
    result = await run.wait(on_ask=ask, on_event=show)

Typed functions per procedure#

stellar generate client writes, for a library, one pair of functions per procedure with its roles, inputs and environments typed: <procedure>(client, …) follows the run to its end, launch_<procedure>(client, …) returns the Run once accepted.

from platform_v3_steps import TcuId, hot_standby_test

async with Client("http://127.0.0.1:8080", "operator") as mcs:
    result = await hot_standby_test(
        mcs, sat="sim-1", psu="psu-sim-1", tcu=TcuId.TCU2, bus_voltage="28 V",
        environment="AIT",                  # Literal["AIT", "IVV"]: where the procedure may run
        on_ask=lambda question: True,
    )

See Code Generators. stellar generate driver writes the typed skeleton of a driver in the same way: see Writing a Driver.

Running#

component.run() reads its settings from the environment (Options.from_env()):

Within an application, build Options yourself and use the async context:

from stellar_mcs import Options

async with driver.running(Options(nats_url="nats://mcs:4222", instance="scpi-psu-2")):
    ...

Options also has heartbeat_period (1 s), register_timeout (2 s) and metrics_period (5 s, throughput reports of a gateway). Without CA nor client certificate, a tls:// URL checks the server against the system roots.

Protocol helpers#

stellar_mcs.csp (CSP 1 and CSP 2 packets, CRC32), stellar_mcs.can (CAN frames on NATS) and stellar_mcs.cfp (CSP over CAN, the CFP of libcsp): see Writing a Transport.

What the Python SDK does not cover#

• the CFDP entity of drivers, and COP-1 in transports: they are in the Rust SDK only;
• a Prometheus endpoint.

A Python gateway still carries the frames of no telecommand (CFDP PDUs of a Rust driver) like the others, without event.

Development#

cd sdk/python
uv sync
uv run pytest                                               # unit tests
STELLAR_TEST_NATS_URL=nats://127.0.0.1:4222 uv run pytest   # and against a NATS server
uv run ruff check && uv run ruff format --check && uv run mypy src tests
uv build                                                    # wheel and sdist in dist/

The integration tests (tests/test_components.py) play the reconciler themselves: a driver, a gateway and a connector registered, bound, a telecommand to ACK 2 and its echo, a frame to decoded values, a connector's message. They need a NATS server with JetStream (nats-server -js) and nothing else of the MCS. tests/test_contract.py, tests/test_passes.py and tests/test_tls.py cover the messages, the pass feeder and TLS; tests/test_client.py covers the API client against a fake API, and tests/test_client_live.py runs it against a running MCS when STELLAR_TEST_API_URL is set.