Drive a remote target¶

A remote target is a separate application MyoGestic drives: a process of its own, reached over gRPC and driven over LSL. An in-process target is the opposite, and the simpler route: an object a ControlBus calls directly, on the thread MyoGestic already owns, with no wire between them at all. Take this page only if your device really is its own program.

If you have not chosen between the two yet, Concepts › Controls explains the control system and which route fits what you are building. This page is the contract; if you would rather arrive at it a stage at a time, with something to run and watch at each one, start from Your first remote target instead.

Which way the streams run¶

MyoGestic writes, you read. One stream per DOF (one degree of freedom), named for the DOF's own address and one channel wide:

MyoGestic  --[ vhi.prediction.index, 1 channel ]-->  your target     (you read this)

There is one shape. The address you advertise in your manifest is the stream name, so nothing further is published about the transport: there is no width to declare and no positional layout for the two of you to agree on. A DOF applies the moment its sample arrives, and the DOFs that did not deliver hold whatever they were last commanded to. These DOFs are independently actuated, possibly driven by different processes at different rates. Two programs can each drive a different DOF.

MyoGestic drives all of it with one RemoteTarget, which publishes one outlet per address the map names.

Publishing a read-back of your own is optional; VHI does it (VHI_Predict and VHI_Control, nine positional channels each, whatever the inbound shape) so a client can verify what actually moved and catch a sign error. Nothing requires it.

The contract¶

you must	why
serve `GetControlManifest`	so a control map can resolve against you. The reply is the whole contract: every address, its kind, its range, its states
report `vocabulary_version` `"2"` or newer	the compatibility gate. MyoGestic refuses an older target by name at bind, because these are separately installed applications and a mismatch is otherwise silent: a target waiting for a stream nobody publishes reports nothing at all, and the hand just never moves
advertise one address per control	the address is the control's identity and its stream name. Two spellings of one control is a second vocabulary to keep in step by hand, and it makes "these two aliases collide" undecidable from the manifest
read one stream per address, exactly one `float32` channel wide	that is the contract, so refuse anything else; do not read element zero of it. An inlet is found by its address's stream name, so a nine-channel whole-pose outlet from an out-of-date client corrupts only the one DOF it is named for, and does it quietly: element zero of somebody's pose frame is a different DOF's value, plausible and in range and wrong

you may	for
serve `SetControl`	discrete DOFs: held states and gestures, which do not belong on a per-tick stream. Both maps are keyed by address, exactly as the manifest publishes them and as your streams are named: resolve the control from the key and the state from the value, and refuse an address you do not export. Resolving on the state alone would leave two discrete controls that share a state name indistinguishable
serve `SweepControl`	letting a client sweep one DOF and read back the degrees it produced, as a direction check
serve `SetPresentation`	for a client asking you to smooth incoming poses
serve the four recording RPCs	driving a ground-truth hand through a capture session

Nothing above is about running the thing. The table below is what turns a process that satisfies the wire into an application you can leave up - a separate table only because a client cannot check any of it.

every remote target	why
check the sample, not just the width	a correctly shaped stream still carries a NaN out of a divide or a `12.0` out of an unclipped model. The range you advertised is a promise you are entitled to enforce before anything moves
refuse an address published by more than one outlet	resolving into `{info.name: info}` keeps whichever producer the sweep answered with last, so two applications driving one DOF look exactly like one, and which you obey changes between restarts
state a liveness policy	hold-last, a timed return to rest, or a hardware deadman. A `SIGKILL`ed producer sends no neutral frame, and nothing about a stream going quiet distinguishes an idle system from a dead one. `source_id` recovery is a separate concern: it decides whether the stream comes back, not what the device does meanwhile. Neither replaces an interlock the software is not in the path of
close inlets and join reader threads on shutdown	setting an event is not shutting down: it leaves the reader mid-`pull_chunk` while the caller carries on, and every inlet still open and re-connecting

The reference target below meets none of them, and that is deliberate. It is the least code that shows the transport, so it applies any sample that is the right shape, keeps {info.name: info} and cannot see a duplicate producer, is hold-last by omission and not by decision, and stop()s by setting a flag without joining its reader or closing an inlet. Read it for the contract; do not deploy it as a template. Your first remote target builds one that meets all four instead, one stage at a time. It also marks the one place its duplicate check stops looking.

What calls it¶

You write none of the client side. RemoteTarget drives the wire through two clients, RemoteClient and RecordingClient. Both are built from the InterfaceSpec that says where your target listens, and both are reached as spec.control_client() and spec.recording_client(). Absence is reported, not raised: every call answers None or False while you are not up, so an application that launches its target from its own button stays responsive while it starts.

the client calls	when, and what it needs back
`control_client().capabilities()`	at bind, and on every retry until you answer. Your manifest is the whole reply. `None` means "not up yet" and is deferred, never failed
`control_client().set_control(...)`	on a discrete edge, from the predict thread. Queued and latest-wins. It never blocks and never raises, so a slow handler of yours cannot stall a 60 fps loop, and your `rejected` reasons reach a log, not a caller
`control_client().sweep(name)`	never automatically. Verification only, by a human or a test: it animates one DOF and reads back the degrees your rig produced
`control_client().set_presentation(blend=…)`	when a client asks you to smooth what you show. Appearance only: it must not change what was commanded
`recording.set_recording_session(True/False)`	around a capture, to gate any local input of your own so the recording has one movement source
`recording.start_trajectory(movement, frequency_hz=…)`	to sweep a ground-truth control through a range while training data is collected
`recording.state()`	to show a client which movements you have, which one is current, and whether a trajectory is running

Everything except set_control is synchronous: setup, teardown and verification, where the caller cannot go on without your answer. Only the per-edge call is queued, and only because it is the one on a deadline.

The whole target¶

The reference target, complete. Every method is the real one; nothing is elided. It talks to the service defined in myogestic/remote/_proto/remote_control.proto; generate stubs from that file for any language other than Python:

"""The smallest thing MyoGestic can drive.

Serve one RPC, read your streams. That is the whole contract — everything else in
`remote_control.proto` is an extra a target may offer and a client may use.

Run it, then point any control map at `vhi.prediction.*`:

    uv run --extra grpc python examples/synthetic/reference_target.py
"""

from __future__ import annotations

import threading
from concurrent import futures
from contextlib import suppress

import grpc
from mne_lsl.lsl import StreamInlet, resolve_streams

from myogestic.remote._proto import remote_control_pb2 as pb2
from myogestic.remote._proto import remote_control_pb2_grpc as pb2_grpc

#: What this target exports. The first segment is the namespace, so `vhi.*` means a map
#: written for a Virtual Hand drives this too. **One stream per address, named after the
#: address, one channel wide** — each applied as its sample arrives, and one nobody
#: drives keeps its last value. `+1` is always the direction the address name denotes, a
#: convention a target does not redefine: a fist is `[1, -1, 1, 1, 1, 1, 0, 0, 0]`.
ADDRESSES = [
    "vhi.prediction.thumb.flexion",
    "vhi.prediction.thumb.abduction",
    "vhi.prediction.index",
    "vhi.prediction.middle",
    "vhi.prediction.ring",
    "vhi.prediction.little",
]


class ReferenceTarget(pb2_grpc.RemoteControlServicer):
    """A remote target in eighty lines. Holds the last value it was sent, per address."""

    def __init__(self, port: int = 50051) -> None:
        self.pose = dict.fromkeys(ADDRESSES, 0.0)
        self._port = port
        self._server: grpc.Server | None = None
        self._stop = threading.Event()

    # --- the one required RPC -------------------------------------------------

    def GetControlManifest(self, request, context):
        """What this target exports. The only thing a client must be able to ask.

        `vocabulary_version` is the compatibility gate: a client refuses a target
        reporting less than the vocabulary it speaks. Report ``"2"`` — one stream per
        DOF — or a current MyoGestic refuses this target instead of driving it.
        """
        manifest = pb2.ControlManifest(target_name="reference", vocabulary_version="2")
        for address in ADDRESSES:
            manifest.capabilities.append(
                pb2.ControlCapability(
                    address=address,
                    kind=pb2.CONTINUOUS,
                    lo=-1.0,
                    hi=1.0,
                    rest=0.0,
                )
            )
        return manifest

    # --- lifecycle ------------------------------------------------------------

    def serve(self) -> None:
        """Start the gRPC server and the inlet reader."""
        self._server = grpc.server(futures.ThreadPoolExecutor(max_workers=4))
        pb2_grpc.add_RemoteControlServicer_to_server(self, self._server)
        self._server.add_insecure_port(f"127.0.0.1:{self._port}")
        self._server.start()
        threading.Thread(target=self._read, name="reference-inlet", daemon=True).start()

    def stop(self) -> None:
        self._stop.set()
        if self._server is not None:
            self._server.stop(grace=None)

    def _read(self) -> None:
        """Read every stream and apply each value as it arrives.

        **One thread, not one per stream.** `resolve_streams` is a multicast sweep of the
        whole network; one sweep already answers for every stream still missing an inlet.
        """
        inlets: dict[str, StreamInlet] = {}
        while not self._stop.is_set():
            # Resolve inside the try: an outlet vanishing mid-open raises, and outside it
            # that race kills this thread while gRPC still answers the manifest.
            try:
                missing = [a for a in ADDRESSES if a not in inlets]
                if missing:
                    found = {s.name: s for s in resolve_streams(timeout=1.0)}
                    for address in missing:
                        info = found.get(address)
                        if info is None:
                            continue
                        if info.n_channels != 1:
                            # Refused, not read: element zero of somebody's pose frame is a
                            # different DOF, and this address would track it all session.
                            print(
                                f"reference target: {address} is published "
                                f"{info.n_channels} channels wide, and this contract is "
                                f"one address per stream, one channel. Not opening it."
                            )
                            continue
                        inlets[address] = StreamInlet(info)
                        inlets[address].open_stream()
                for address, inlet in inlets.items():
                    chunk, _ = inlet.pull_chunk(timeout=0.0)
                    if chunk is not None and len(chunk):
                        self.pose[address] = float(chunk[-1][0])
                self._stop.wait(0.005)
            except Exception as exc:
                print(f"reference target: lost an inlet ({type(exc).__name__}: {exc}), re-resolving")
                for inlet in inlets.values():
                    # Closing a *broken* inlet is exactly the case that raises, and a raise
                    # here would kill the thread this handler exists to keep alive.
                    with suppress(Exception):
                        inlet.close_stream()
                # All of them: telling which inlet raised costs a try per inlet per tick.
                inlets.clear()
                # `resolve_streams` raises too, and with no inlet to lose that is a tight loop.
                self._stop.wait(1.0)


if __name__ == "__main__":
    target = ReferenceTarget()
    target.serve()
    print("reference target on 127.0.0.1:50051 — Ctrl-C to stop")
    try:
        threading.Event().wait()
    except KeyboardInterrupt:
        target.stop()

The standard¶

Values are [-1, 1], 0 is rest, +1 is the direction the DOF's name denotes. Across the nine addresses of a VHI-shaped hand a fist is [1, -1, 1, 1, 1, 1, 0, 0, 0]: five flexions and an adducted thumb, because that is what a fist does with a thumb. The claim is about the addresses, so it reads the same whether those nine values travel as one nine-channel sample or as nine one-channel ones.

The warning¶

A backwards sign survives every test you would think to write - see the one convention a device may not redefine for why no suite can reach it. Check against something outside the loop. Look at the device yourself. This repo shipped that bug and its own contract suite passed throughout.

Design notes¶

Detail a target author needs once, usually while debugging, rather than while building.

A liveness timeout is not a safety interlock. A target that has itself crashed, hung or been swapped out cannot time anything out. Anything that can injure a person needs a release the software is not in the path of.

Stream recovery and the liveness timeout are different jobs. Recovery decides whether the stream comes back; the timeout decides what the device does meanwhile. A rig can want both.

SetControl delivery timing is not guaranteed. set_control queues a frame for the client's worker thread, which drains one blocking RPC at a time with a two-second deadline. A later state can be queued before an earlier one is delivered, and two can land inside one tick of a status poller. So verify discrete state from a handler that prints what it applied, not from a poller — the poller is a matter of luck.