Drive your own device¶

Anything MyoGestic moves — a prosthesis on a serial port, a motor controller, a cursor, a robot arm — is a target: a plain object with three methods that you hand to a ControlBus. Nothing registers, nothing subclasses, nothing is discovered by name.

Run it first¶

examples/synthetic/my_device.py is a complete target with three lines left for you. It needs no hardware:

uv run python examples/synthetic/my_device.py

What you should see¶

a frame, as @pipeline.predict would push it:
  mydevice.grip              +0.80
  mydevice.wrist.pronation   -0.40
out of range and NaN, both handled before you see them:
  mydevice.grip              +1.00
  mydevice.wrist.pronation   +0.00
teardown, which rests every control first:
  mydevice.grip              +0.00
  mydevice.wrist.pronation   +0.00
  stopped

Three things happened without you writing them. 5.0 arrived at your device as 1.0, because it declared hi=1.0. A NaN arrived as 0.0 rather than as full deflection. And every control was returned to rest before teardown, so the device does not keep its last grip. That is ControlBus, and it is the reason a target is three methods rather than thirty.

Change these three things¶

Copy my_device.py next to your own code and edit the three marked lines:

	what	where
1	Name your controls. Two or more dotted lowercase segments; the first is your namespace. Name a direction, so `+1` means something: `grip.close`, not `grip.motor`.	`ADDRESSES`
2	Drive your hardware. Replace the `print` with your call: `self._port.write(...)`, `self._motors.set(...)`, an MQTT publish, anything.	in `send`
3	Release your hardware. Close the port, disconnect, power down.	in `stop`

Nothing else has to change. If your device only moves one way, declare lo=0.0 in capabilities and MyoGestic will refuse a map that asks for the other direction.

To check a control map against your device before running anything:

uv run python tools/inspect_control_map.py my-map.toml

It calls only capabilities() and resolve(), so it builds no target and moves nothing.

The contract, in full¶

If "control map", "address" and "alias" are new words, Concepts › Controls explains the system this page uses. A remote target is the separate case where your device is already its own program and the target is written for you.

class Target(Protocol):
    def bind(self, controls: ControlSet) -> None: ...
    def send(self, values, changed) -> None: ...
    def stop(self) -> None: ...

    claims: frozenset[str]                      # optional
    def capabilities(self) -> Sequence | None: ...   # optional

bind runs once, on the main thread, and may raise. Refuse a configuration you cannot drive there, while a human is still reading the traceback. send runs on the predict thread and must not raise; every value it gets is already finite and inside its declared range, because ControlBus sanitised the frame before fanning it out.

The two optional members are what the bus asks for by name:

member	absent means	present means
`claims`	"assume it drives everything"	the aliases it drives, so the bus can catch a control nothing drives
`capabilities()`	"the caller already knows my vocabulary"	what addresses you export, so a map can be resolved against you

A route's weight is yours to apply, and before your own range. The bus delivers the un-weighted value; compute min(hi, max(lo, weight * value)), as RemoteTarget does. Ignoring weight silently discards every gain in the map, including the weight = -1.0 that inverts an axis.

A complete target¶

The whole of a target that moves a cursor.

from myogestic.controls import Capability, ControlBus, Continuous, ControlSet


class CursorTarget:
    """Drive a 2-D cursor from two continuous controls."""

    #: What this target exports. Addresses are namespaced by their first segment, so
    #: `cursor.*` cannot collide with `vhi.*` in the same map.
    ADDRESSES = ("cursor.x", "cursor.y")

    def __init__(self) -> None:
        self.position = (0.0, 0.0)
        self._slots: dict[str, tuple[str, float]] = {}

    def capabilities(self):
        """Signed, normalised, resting at zero — the control standard's defaults."""
        return [
            Capability(address=a, kind="continuous", lo=-1.0, hi=1.0, rest=0.0)
            for a in self.ADDRESSES
        ]

    def bind(self, controls: ControlSet) -> None:
        """Refuse here, not at the first frame."""
        self._slots = {}
        for alias, refs in controls.routes.items():
            for ref in refs:
                if ref.address in self.ADDRESSES:
                    self._slots[alias] = (ref.address, ref.weight)
        if not self._slots:
            raise ValueError(f"nothing in this map targets {self.ADDRESSES}")

    @property
    def claims(self) -> frozenset[str]:
        return frozenset(self._slots)

    def send(self, values, changed) -> None:
        """One tick. Must not raise."""
        x, y = self.position
        for alias, (address, weight) in self._slots.items():
            # Weight first, then your own range. Yours to apply — the bus does not.
            value = min(1.0, max(-1.0, weight * float(values.get(alias, 0.0))))
            if address == "cursor.x":
                x = value
            else:
                y = value
        self.position = (x, y)

    def stop(self) -> None:
        """Rest. The bus sends a neutral frame first, so this is usually enough."""
        self.position = (0.0, 0.0)

Driving it¶

Same three lines as any other target. connect_controls asks each target what it exports, resolves the map against the answers, and builds the bus:

from myogestic.controls import connect_controls, load_control_map

control_map = load_control_map({"dofs": {"aim_x": "cursor.x", "aim_y": "cursor.y"}})

cursor = CursorTarget()
bus = connect_controls(control_map, [cursor], hz=32)

bus.push({"aim_x": 0.5, "aim_y": -0.25})
assert cursor.position == (0.5, -0.25)
bus.stop()

connect_controls answers None rather than raising while any target's capabilities() does, so a target that is not up yet defers instead of failing. ControlLink holds the arguments and asks again for you, which saves you a module-level bus and a global:

from myogestic.controls import ControlLink

link = ControlLink(control_map, [CursorTarget()], hz=32)

def on_click():                 # a button handler, or a training thread
    if link.ensure():           # idempotent, and cheap once it has bound
        link.bus.push({"aim_x": 0.5, "aim_y": -0.25})

on_click()
assert link.bus is not None     # this target answers immediately; a remote one would not
link.stop()                     # rests every target and clears the bus

Call ensure() from anywhere that can afford to block: a UI handler, a training thread. Never from @pipeline.predict, which has its own thread and a deadline - read link.bus there and no-op while it is None.

If a UI loop should reconnect automatically, wrap the link in ControlLinkConnector and call its non-blocking poll() from that loop. Resolution is atomic across all targets in one link; use separate maps and links when one output should keep working while another is unavailable.

If your target's vocabulary is fixed and you have the capabilities in hand already, build the bus directly instead; connect_controls is only the lazy-resolve convenience:

from myogestic.controls import ControlBus, resolve

controls = resolve(control_map, cursor.capabilities())
bus = ControlBus(controls, targets=[cursor], hz=32)

Addresses are yours to name¶

A control map's right-hand side is an address, and its first segment namespaces it: vhi.prediction.index, keyboard.tap.function.f1, cursor.x. Pick a segment nobody else uses and the same map can drive your device and a Virtual Hand at once, with one ControlBus and one list of targets:

bus = ControlBus(controls, targets=[cursor, vhi_target], hz=32)

The bus checks that someone claims every alias, so an address no target drives is caught at bind. Uncaught, it would look like a control that works and holds still.

A real one: a servo hand¶

A cursor has two controls and no mechanism. examples/synthetic/servo_hand.py is a real one - six servos on a serial port, runnable with no hardware:

uv run python examples/synthetic/servo_hand.py

"""A prosthetic hand on a serial port: three methods and a write.

    uv run python examples/synthetic/servo_hand.py

Runs with no hardware. Two things to notice:

**Five addresses, six servos.** `hand.thumb` drives two of them on different transfer
functions, because a real thumb opposes as it flexes. The coupling lives here, not in the
control map: an address exists so the map need not know this hand's linkage.

**Wire order is this file's.** `frame` iterates `SERVOS` and looks each fraction up by name,
so reordering a TOML cannot reorder somebody's fingers.

The bus already clamped NaN, filled missing controls, held the declared range and delivered
rest before teardown - see `myogestic.controls.ControlBus`.
"""

from __future__ import annotations

from myogestic.controls import (
    Capability,
    ControlBus,
    ControlSet,
    load_control_map,
    resolve,
)

#: One address per finger. One-way (`lo=0.0`): a servo hand cannot hyperextend, and
#: declaring a direction it does not have would clamp silently on the predict thread.
ADDRESSES = ("hand.thumb", "hand.index", "hand.middle", "hand.ring", "hand.little")

#: The firmware's channel order, and each servo's travel in degrees, `(open, closed)`.
#: **This dict's order is the wire order.**
SERVOS = {
    "thumb_flex": (10, 96),
    "thumb_rot": (0, 110),
    "index": (5, 100),
    "middle": (5, 100),
    "ring": (5, 100),
    "little": (5, 100),
}

#: How far into the thumb's travel opposition completes; past this it only flexes.
OPPOSITION_SPAN = 0.6


class ServoHand:
    """Drive six servos over a serial port from a control map.

    Parameters
    ----------
    port
        Anything with ``write(bytes)`` and ``close()``::

            ServoHand(serial.Serial("/dev/ttyACM0", 115200))

        ``None`` computes frames and sends nothing, so this file runs without hardware.
    """

    def __init__(self, port=None) -> None:
        self._port = port
        self._routed: tuple[tuple[str, str, float], ...] = ()

    def capabilities(self) -> tuple[Capability, ...]:
        """What a control map may name."""
        return tuple(
            Capability(
                address=address,
                kind="continuous",
                lo=0.0,
                hi=1.0,
                rest=0.0,
                description=f"{address.rsplit('.', 1)[-1]} curls from open (0) to closed (1)",
            )
            for address in ADDRESSES
        )

    def bind(self, controls: ControlSet) -> None:
        """Accept a configuration, or refuse it while a human is still reading.

        Raises
        ------
        ValueError
            Nothing in the map reaches this hand.
        """
        self._routed = tuple(
            (ref.address, alias, ref.weight)
            for alias, refs in controls.routes.items()
            for ref in refs
            if ref.address in ADDRESSES
        )
        if not self._routed:
            raise ValueError(
                f"nothing in this map reaches this hand. It drives {', '.join(ADDRESSES)} — "
                f"check the namespace on the right-hand side of your [dofs] table."
            )

    @property
    def claims(self) -> frozenset[str]:
        """Which aliases this hand drives."""
        return frozenset(alias for _, alias, _ in self._routed)

    def send(self, values, changed) -> None:
        """Actuate one tick. Never raises: the bus guarantees finite, in-range values."""
        levels = {
            # Weight first, then this hand's range: a gain must not exceed what we accept.
            address: min(1.0, max(0.0, weight * float(values.get(alias, 0.0))))
            for address, alias, weight in self._routed
        }
        if self._port is not None:
            self._port.write(self.frame(levels))

    def stop(self) -> None:
        """Open the hand, then close the port. Idempotent.

        The bus delivers rest before calling this; repeated because a target can also be
        stopped directly, and a missed open-hand frame leaves a hand closed.
        """
        port, self._port = self._port, None
        if port is None:
            return
        try:
            port.write(self.frame({}))
        finally:
            port.close()

    @staticmethod
    def frame(levels) -> bytes:
        """The bytes for one pose. Pure, so a test can read it without a port.

        An address nobody drives is held open: a servo has no third state.
        """
        thumb = levels.get("hand.thumb", 0.0)
        fraction = {
            "thumb_flex": thumb,
            "thumb_rot": min(1.0, thumb / OPPOSITION_SPAN),
            "index": levels.get("hand.index", 0.0),
            "middle": levels.get("hand.middle", 0.0),
            "ring": levels.get("hand.ring", 0.0),
            "little": levels.get("hand.little", 0.0),
        }
        angles = (
            round(lo + fraction[servo] * (hi - lo)) for servo, (lo, hi) in SERVOS.items()
        )
        return (",".join(str(a) for a in angles) + "\n").encode()


if __name__ == "__main__":

    class _Log:
        """Stands in for `serial.Serial`, keeping what would have gone down the wire."""

        def __init__(self) -> None:
            self.lines: list[str] = []

        def write(self, payload: bytes) -> None:
            self.lines.append(payload.decode().strip())

        def close(self) -> None:
            self.lines.append("<closed>")

    port = _Log()
    hand = ServoHand(port)

    # The left side is yours; the right side is what the hand declared above.
    control_map = load_control_map(
        {
            "dofs": {
                "thumb": "hand.thumb",
                "index": "hand.index",
                "close": ["hand.middle", "hand.ring", "hand.little"],
            }
        }
    )
    bus = ControlBus(resolve(control_map, hand.capabilities()), targets=[hand], hz=32)

    bus.push({"thumb": 0.0, "index": 0.0, "close": 0.0})
    assert port.lines[-1] == "10,0,5,5,5,5", port.lines[-1]

    bus.push({"thumb": 1.0, "index": 1.0, "close": 1.0})
    assert port.lines[-1] == "96,110,100,100,100,100", port.lines[-1]

    # Half a thumb is *most* of the rotator's travel, not half of it: the coupling.
    bus.push({"thumb": 0.5, "index": 0.0, "close": 0.0})
    assert port.lines[-1] == "53,92,5,5,5,5", port.lines[-1]

    # Out of range never reaches a servo: the bus clipped to the declared domain first.
    bus.push({"thumb": 5.0, "index": -3.0, "close": 0.0})
    assert port.lines[-1] == "96,110,5,5,5,5", port.lines[-1]

    bus.stop()  # rest, then teardown, in that order
    assert port.lines[-2] == "10,0,5,5,5,5", port.lines[-2]
    assert port.lines[-1] == "<closed>", "the port was left open"

    print("\n".join(port.lines))

Three things in it carry the weight: hand.thumb drives two servos, so the coupling stays out of the map; frame iterates SERVOS by name, so the wire order is the device's; and stop rests before it closes.

What the standard asks of you¶

A control value is signed and normalised: [-1, 1], 0 at rest, and +1 means the direction the name denotes. cursor.x at +1 should move right if you called it right. A one-way control declares lo=0.0 instead.

Getting a sign backwards is the one mistake that survives every test you are likely to write - Concepts › Controls has the reason. Check it against something outside the loop: a person looking at the device.