Control standard¶

A control space is a mapping, in a file: your name for a model output on the left, a control the target declares on the right. The left side is yours and arbitrary. The right side belongs to the target, which also declares what the control is: a number or a held state, its range, its states. MyoGestic hard-codes none of that, so a Virtual Hand, a keyboard, a cursor and a prosthesis each keep their own vocabulary, and a build that grows a control needs no change on this side.

New to this? Concepts › Controls explains the system this page documents: what a control is, why the standard is fixed, and whether you write the target or it is already written.

Continuous controls are normalized: +1 is the direction the control denotes, rest is 0, and the range is signed when the target says the control is. Discrete controls are separate on purpose: a held state delivered on change is a different kind of value.

There is no separate control for the other direction

A signed control already has both halves: +1 on vhi.prediction.thumb flexes the thumb and -1 extends it. There is no …thumb.extension address; a model that wants extension emits a negative value, or a mapping gives that target a negative weight.

Two things that look like exceptions:

vhi.prediction.thumb.abduction is a second axis, not the other half of the first. The thumb has two; the short …thumb is declared to mean flexion, and the other one has to be named, because silently picking one of two would be a guess. Every other digit has one axis, so its short form is all there is.
ThumbExtension is one of vhi.control.gesture's movement presets: a held state on the control hand. A preset commands a whole-hand pose in one held state, a compound shape no single continuous address expresses.

One address per control, always: the address is the control's identity and also the name of the one-channel LSL stream that carries it. A target that advertised two spellings of one control would make "these two aliases collide" impossible to decide from a manifest.

Inspect the whole control path with tools/inspect_control.py

A narrated walkthrough runs the whole path: declaration, the two kinds of control, the wire frame, and the negotiation against whatever Virtual Hand you have.

uv run --extra grpc python tools/inspect_control.py

Run it with no Virtual Hand at all and it still walks the first three steps, then shows what a target does when the far side is absent. Launch a VHI and run it again for the handshake: it prints a different step 4 for a v2 build, a v1 build, and nothing at all.

Declaring a control space¶

Write a TOML file. A ready-to-copy one ships at examples/controls/hand.toml:

[dofs]
# Left side: your model's output names. Right side: controls VHI declares.
my_thumb_spread = "vhi.prediction.thumb.abduction"

fist = [                                       # one output, fanned out
  { target = "vhi.prediction.thumb.flexion", weight = 0.6 },   # ...with a per-target gain
  { target = "vhi.prediction.index" },
  { target = "vhi.prediction.middle" },
  { target = "vhi.prediction.ring" },
  { target = "vhi.prediction.little" },
]

gesture = { target = "vhi.control.gesture", debounce_s = 0.1 }

The left side is yours. my_thumb_spread, fist, gesture, whatever your model calls its outputs. Nothing prescribes these names or reads meaning out of them. A control takes only one output, so no two entries may name the same address; a fan-out is how one output reaches several.

The right side belongs to the target. vhi.prediction.index is a name VHI declares in its own manifest, along with everything needed to send it: whether it takes a number or a held state, its range, its states. Ask a running target what it exports:

uv run --extra grpc python tools/inspect_control.py

Load it and resolve it:

import tomllib
from myogestic.controls import load_control_map, resolve

with open("hand.toml", "rb") as f:          # "rb" — tomllib requires binary
    control_map = load_control_map(tomllib.load(f))

# Resolution needs a live target: it is what declares the semantics.
controls = resolve(control_map, vhi.control_client().capabilities())

Mapping-first: the shape of each value says how a value travels. A bare string is one target control; an array is a fan-out reaching several; a table with target/targets is the explicit form, and the only place a per-target weight or a debounce_s stability gate is written. Whether a control is a number or a held state stays out of this file: the target declares it.

load_control_map takes a Mapping, not a path

The library reads no configuration files, by design, so the snippet above opens the file itself. The same call accepts JSON, a dict literal, a row from a database, or a config system you already have. TOML is what a human wants to edit, so the shipped example is TOML.

A mapping becomes a control space when a target answers

load_control_map checks structure; whether vhi.prediction.index is a number or a held state is the target's to declare. So an application that launches its own target resolves after startup rather than at import. The examples all build their bus lazily for that reason.

Classification uses the same mapping¶

A classifier produces an activation (open or closed) where a regressor produces a position, and an activation is just a control value, so it travels the same mapping. Add a threshold_fraction, the probability cutoff, to say the input is a classifier's confidence:

fist = { targets = [
  { target = "vhi.prediction.thumb.flexion", weight = 0.6 },
  { target = "vhi.prediction.index" },
], threshold_fraction = 0.5 }

Push the model's probability and the bus gates it to exactly 0.0 or 1.0 before anything else sees it: before the weights, before the wire, before the recording. From there it is an ordinary value, 0 to every listed control when inactive and 1 × weight when active. The target receives continuous per-control values either way, with no separate state command.

Drop the threshold_fraction and the identical mapping serves a regressor emitting 0..1 directly. Gating here rather than in a separate discrete path is what lets one mapping serve both: a continuous address is a position, so a raw 0.73 streamed into one says the finger is 73% curled, a different statement from 73% confident that it is closed.

Map onto a discrete address instead when the thing genuinely is a state rather than an amount: a preset, a keypress, a mode. examples/controls/classification.toml shows the activation form and examples/controls/classification_grpc.toml the discrete one.

load_control_map ¶

load_control_map(config: Mapping[str, Any]) -> ControlMap

Parse a [dofs] mapping of user aliases to target control addresses.

Takes a ~collections.abc.Mapping, not a path — parse your own TOML (or JSON, or a dict literal) and hand it over, so this library reads no configuration files.

Checks structure, not meaning: that every alias is usable and every address is address-shaped. Whether an address exists, and what it accepts, is the target's to say — see resolve.

Parameters:

Name	Type	Description	Default
`config`	`Mapping[str, Any]`	Typically `tomllib.load(f)`. The `dofs` table maps each alias to one address, a list of addresses (a broadcast), or a table with per-target weights.	required

Returns:

Type	Description
`ControlMap`	The parsed mapping, still unresolved.

Raises:

Type	Description
`ValueError`	With every fault found, not just the first.

Examples:

>>> from myogestic.controls import load_control_map
>>> cmap = load_control_map(
...     {"dofs": {"fist": ["vhi.prediction.index", "vhi.prediction.middle"]}}
... )
>>> [ref.address for ref in cmap.bindings["fist"].targets]
['vhi.prediction.index', 'vhi.prediction.middle']

resolve ¶

resolve(control_map: ControlMap, capabilities: Sequence[Capability]) -> ControlSet

Resolve a ControlMap against what a target says it can do.

Each alias becomes a myogestic.controls.Continuous or myogestic.controls.Discrete according to what its target declared. The resulting myogestic.controls.ControlSet is keyed by your aliases; the routing to addresses travels alongside in myogestic.controls.ControlSet.routes.

Parameters:

Name	Type	Description	Default
`control_map`	`ControlMap`	From `load_control_map`.	required
`capabilities`	`Sequence[Capability]`	What the target exports — for VHI, its `GetControlManifest` reply.	required

Returns:

Type	Description
`ControlSet`	Resolved, and usable.

Raises:

Type	Description
`ValueError`	For an address the target does not export (naming the near misses and the full list), for a broadcast whose members disagree about kind or states, or for a negative weight on a control that does not accept signed values.

Examples:

>>> from myogestic.controls import Capability, load_control_map, resolve
>>> caps = [Capability("cursor.x_velocity", "continuous", lo=-1.0, hi=1.0)]
>>> cmap = load_control_map({"dofs": {"drive_x": "cursor.x_velocity"}})
>>> resolved = resolve(cmap, caps)
>>> resolved.dofs["drive_x"].lo
-1.0

Capability `dataclass` ¶

Capability(address: str, kind: str, lo: float = -1.0, hi: float = 1.0, rest: float = 0.0, states: tuple[str, ...] = (), rest_state: str = '', activation_threshold: float = 0.0, description: str = '')

One control a target exports, with the semantics the target declares.

Built from a target's manifest — for VHI, from its GetControlManifest reply. Nothing in MyoGestic invents these values.

There is no transport here on purpose. A streamed control's LSL stream is named for the control's own address and is one channel wide, so the address is the transport and a separate stream_name/channel pair could only ever repeat it.

Attributes:

Name	Type	Description
`address`	`str`	The stable dotted address, e.g. `"vhi.prediction.index"`.
`kind`	`str`	`"continuous"` or `"discrete"`.
`lo, hi, rest`		Continuous only: the domain and the neutral value.
`states, rest_state`		Discrete only: the accepted states and the neutral one.
`activation_threshold`	`float`	Discrete only: the level at which a client emitting a probability should select the non-rest state. `0.0` means the target has no opinion.
`description`	`str`	Human-readable. For a log or an error message; never parsed.

Examples:

>>> from myogestic.controls import Capability
>>> Capability("cursor.x_velocity", "continuous", lo=-1.0, hi=1.0).signed
True

signed `property` ¶

signed: bool

Whether this control accepts values on both sides of its neutral value.

ControlMap `dataclass` ¶

ControlMap(bindings: Mapping[str, Binding] = dict())

A parsed but unresolved declaration: aliases bound to target addresses.

Not yet a usable control space. Whether fist is a number or a held state, and what its range is, are facts the target declares; resolve is where they arrive.

Examples:

>>> from myogestic.controls import load_control_map
>>> cmap = load_control_map({"dofs": {"my_index": "vhi.prediction.index"}})
>>> cmap.addresses()
('vhi.prediction.index',)

addresses ¶

addresses() -> tuple[str, ...]

Every distinct target address this map references, in first-seen order.

as_control_space ¶

as_control_space() -> dict[str, Any]

A persistable control space — the inverse of read_control_space.

Tagged with CONTROL_SPACE_FORMAT so a reader can name the format it found rather than guess from the shape. That tag is the whole difference between this and dump_control_map, which writes the same bindings as TOML for a human to edit; this writes them as a dict for a recording or a model sidecar to carry.

It was as_dict, which named the return type rather than the thing — nothing said it paired with read_control_space, and the pair is the point.

Binding `dataclass` ¶

Binding(alias: str, targets: tuple[TargetRef, ...], debounce_s: float = 0.0, label: str = '', threshold_fraction: float | None = None)

One [dofs] line: a user-owned alias bound to one or more target controls.

Attributes:

Name	Type	Description
`alias`	`str`	The user's own name for a model output. Arbitrary.
`targets`	`tuple[TargetRef, ...]`	Where its value goes. More than one is a broadcast: the same scalar reaches every listed control, each applying its own declared range and direction.
`debounce_s`	`float`	For a binding that resolves to a discrete control: how long a state must hold before it counts as a transition. A property of this control loop, not of the target.
`threshold_fraction`	`float \| None`	For a classifier input: the probability fraction at which it counts as active. Below it the value is `0.0`; at or above it, `1.0`. On a continuous binding that gated 0/1 then travels the ordinary weighted fan-out; on a discrete one it selects the non-rest state. `None` leaves a continuous value as given and takes a discrete control's threshold from the target — override only when your model's calibration differs. Distinct from the target-declared `Capability.activation_threshold`.
`label`	`str`	Optional display label; the alias is used when empty.

TargetRef `dataclass` ¶

TargetRef(address: str, weight: float = 1.0)

One target control a binding sends to, and the gain applied on the way.

Attributes:

Name	Type	Description
`address`	`str`	The target-owned control address, e.g. `"vhi.prediction.index"`.
`weight`	`float`	Multiplied into the value before the target applies its own range, so one member of a fan-out can move less than the others: `weight = 0.6` on a thumb sends it 60% of what the fingers get. Defaults to `1.0`.

read_control_space ¶

read_control_space(raw: Mapping[str, Any]) -> ControlMap

A recording's persisted control space, as a ControlMap.

New recordings store the alias-to-address mapping and tag it with CONTROL_SPACE_FORMAT.

Parameters:

Name	Type	Description	Default
`raw`	`Mapping[str, Any]`	The `control_space` object out of a session's `meta.json` or a model's `.controls.json` sidecar.	required

Returns:

Type	Description
`ControlMap`	Still unresolved, exactly like a freshly parsed file.

Raises:

Type	Description
`ValueError`	For a control space written in the pre-alias format, naming the format it found. It is not normalised on the way in: that older shape declared its own kinds and ranges, which the target now owns. Re-record, or read the archive with the version that wrote it.

Examples:

>>> from myogestic.controls import read_control_space
>>> read_control_space(
...     {"format": "alias-address/1", "dofs": {"my_index": "vhi.prediction.index"}}
... ).addresses()
('vhi.prediction.index',)

ControlSet `dataclass` ¶

ControlSet(dofs: Mapping[str, Dof] = dict(), standard_version: str = STANDARD_VERSION, routes: Mapping[str, tuple[Any, ...]] = dict())

A validated control configuration.

dofs preserves declaration order, and that order is the wire order: it is the layout of the vector encode produces and the labels a target publishes.

Attributes:

Name	Type	Description
`dofs`	`Mapping[str, Dof]`	Resolved DOFs by alias, in declaration order.
`standard_version`	`str`	The vocabulary-format version this configuration was written against. Recorded verbatim and not validated: what a version means is settled by the target handshake, not by the loader.

Examples:

>>> from myogestic.controls import Continuous, ControlSet
>>> controls = ControlSet(dofs={"my_index": Continuous("my_index")})
>>> controls.channel_labels()
('my_index',)

continuous `property` ¶

continuous: tuple[Continuous, ...]

The continuous DOFs, in declaration order.

discrete `property` ¶

discrete: tuple[Discrete, ...]

The discrete DOFs, in declaration order.

channel_labels ¶

channel_labels() -> tuple[str, ...]

Continuous DOF names in wire order — the channel labels to publish.

rest_values ¶

rest_values() -> dict[str, float | str]

The neutral frame: every DOF at its declared rest.

Continuous `dataclass` ¶

Continuous(name: str, lo: float = -1.0, hi: float = 1.0, rest: float = 0.0, label: str = '', threshold_fraction: float | None = None)

A signed, normalized DOF. +1 is the direction the name denotes.

Attributes:

Name	Type	Description
`name`	`str`	The alias this control was declared under — the user's own name for a model output, e.g. `"my_index"`. Never interpreted.
`lo, hi`		The declared domain. Defaults to `[-1.0, 1.0]` — genuinely bidirectional, because a wrist or cursor axis needs both directions.
`rest`	`float`	The value meaning "no command". Must lie inside the domain.
`label`	`str`	Optional display label; the name is used when empty.
`threshold_fraction`	`float \| None`	Set when this control is driven by a classifier rather than a regressor: the input is a probability in `[0, 1]`, and this is the fraction at which it counts as active. Below it the value becomes `0.0`; at or above it, `1.0`. Gated here, before anything else sees the number. Distinct from a target's `Capability.activation_threshold`: this one is about the model's confidence. The gated 0/1 is an ordinary control value, fanned out and weighted like any other. `None` uses the value as given.

Examples:

>>> from myogestic.controls import Continuous
>>> Continuous("my_index").rest
0.0

Discrete `dataclass` ¶

Discrete(name: str, states: tuple[str, ...], rest: str, debounce_s: float = 0.0, label: str = '', activates: str = '', threshold_fraction: float = 0.5)

A DOF holding exactly one of states.

A discrete DOF is a held state, not an event stream: it is delivered on change, and a repeat is expressed by returning through rest.

Attributes:

Name	Type	Description
`name`	`str`	The alias this control was declared under. Never interpreted.
`states`	`tuple[str, ...]`	At least two unique state names. These double as display labels.
`rest`	`str`	The neutral state. Must be one of `states`.
`debounce_s`	`float`	Seconds a new state must hold before it is delivered. `0.0` delivers every change immediately.
`activates`	`str`	The state a numeric activation selects once it reaches `threshold_fraction`. Set when the target declares exactly two states, so a binary classifier emitting a probability in `[0, 1]` needs no thresholding of its own. Empty when a scalar cannot pick a state: with three or more states a number is ambiguous.
`threshold_fraction`	`float`	The probability fraction at which `activates` is selected. Taken from the target when it declares one (`Capability.activation_threshold`); overridable per binding.
`label`	`str`	Optional display label; the name is used when empty.

Examples:

>>> from myogestic.controls import Discrete
>>> Discrete("gesture", ("rest", "fist"), "rest").states
('rest', 'fist')

Dof `module-attribute` ¶

Dof = Continuous | Discrete

Either kind of DOF. A tagged union, so ty can check a match exhaustively.

Delivering a frame¶

One bus sanitises each frame exactly once and fans it out to every target, owning an ordering that is easy to get subtly wrong per-application.

ControlBus ¶

ControlBus(controls: ControlSet, *, targets: Sequence[Target] = (), smoothing: VectorFilter | None = None, hz: float = 50.0, dead_zone: float | None = None, hysteresis: float | None = None, on_warn: Callable[[str], None] | None = None)

Sanitise a frame once, then deliver it to every target.

Owns the one ordering that must not be re-derived per application::

substitute rest -> clip -> dead zone -> smooth
                -> substitute rest -> clip -> deliver

Rest substitution comes first because min(hi, max(lo, nan)) is lo — a NaN would otherwise become full-scale deflection. It happens again after smoothing because numpy.clip passes NaN straight through and a filter carries state, so one bad sample would otherwise poison every later one. The final clip is not cosmetic either: a smoother undershoots on a falling edge, and for a one-way DOF whose rest sits at lo that undershoot is a sign flip into a direction the DOF declares does not exist.

Parameters:

Name	Type	Description	Default
`controls`	`ControlSet`	The validated configuration.	required
`targets`	`Sequence[Target]`	Targets to deliver to. Each is `bind`-ed now, so a target that cannot drive this configuration says so while a human is watching.	`()`
`smoothing`	`VectorFilter \| None`	Optional `myogestic.outputs.filters.VectorFilter` over the continuous vector, applied after mapping. Its output is re-sanitised.	`None`
`hz`	`float`	The rate `push` is expected to be called at, used to convert each discrete DOF's `debounce_s` into a tick count. A snapshot: changing the caller's rate afterwards changes the effective debounce.	`50.0`
`dead_zone`	`float \| None`	Optional symmetric dead zone in normalized units, `0 <= dead_zone < 1`. No default, because rest is interior for a signed DOF and the right value depends on a real signal — shipping a guessed constant would be a guess dressed as a safety feature.	`None`
`hysteresis`	`float \| None`	Optional threshold, `0 <= hysteresis < 1`, a value must exceed to cross rest into the opposite direction. Also without a default, and for the same reason.	`None`
`on_warn`	`Callable[[str], None] \| None`	Called the first time each distinct condition occurs — a clamp, a failed target, a failed frame. Deliberately once per condition rather than per tick: at `predict_hz` a repeated warning erases the log that would explain it. Pass `ctx.log` to surface these in the app.	`None`

Notes

push is called from the predict thread and select / rebase from the UI thread. That pairing is safe without a lock because the only shared mutable state is each EdgeTrigger's single tuple, which is replaced by one atomic assignment. Anything added here that mutates more than one field at a time needs a lock.

push ¶

push(raw: Mapping[str, Any]) -> Mapping[str, float | str]

Sanitise one frame, deliver it, and return what was delivered.

Never raises. This runs on the predict thread, where an exception is logged with a full traceback on every tick — so a failure here degrades to the neutral frame instead of burying the log that would explain it.

select ¶

select(name: str, state: str) -> bool

Command a discrete DOF from the UI, bypassing the debounce.

Use this for a manual click: it delivers immediately and rebases the trigger, so the next push carrying the same state does not fire again.

Returns:

Type	Description
`bool`	Whether the state was delivered.

rebase ¶

rebase(name: str, state: str) -> None

Accept a discrete DOF's state as current without delivering it.

For when something else already commanded the target and the bus should not repeat it.

stop ¶

stop() -> None

Deliver the neutral frame, then stop every target. Idempotent.

Rest is delivered before the targets stop: a target that is torn down while holding a non-neutral value leaves the application it drives holding it too.

connect_controls ¶

connect_controls(control_map: Any, targets: Sequence[Any], *, ctx: Any = None, hz: float = 32.0, smoothing: Any = None) -> ControlBus | None

Resolve a map against what targets export, and build the bus. None if they cannot say yet.

The bind every VHI application has to do, and had to write out: a control map names addresses, and what an address means — number or held state, its range, its neutral value — belongs to the target. So a map cannot be resolved until the targets can answer, and an application that launches its own target has nobody to ask at import.

Call it from a UI handler, or anywhere that can afford to block, and call it again until it returns something. Never from a predict callback: asking a target costs an RPC, and stalling the control loop on it is worse than a frame with no bus.

Parameters:

Name	Type	Description	Default
`control_map`	`Any`	The parsed `~myogestic.controls.ControlMap`, from `~myogestic.controls.load_control_map`.	required
`targets`	`Sequence[Any]`	Every target the map may name, already constructed. Each is asked what it exports; one that answers `None` — a remote target that has not started — makes the whole call return `None`, because a map resolved against a partial manifest would bind some aliases and silently drop the rest. The target list is therefore one atomic failure domain. Use separate calls or links for targets that should remain independently useful when another one is unavailable.	required
`ctx`	`Any`	The app's `~myogestic.Context`. Given one, the map is recorded as `ctx.control_space` so a recording carries the mapping it was made under, and the outcome is logged.	`None`
`hz`	`float`	Passed to `ControlBus`.	`32.0`
`smoothing`	`float`	Passed to `ControlBus`.	`32.0`

Returns:

Type	Description
`ControlBus or None`	`None` while any target is unreachable. Try again later; nothing is left half-built.

ControlLink ¶

ControlLink(control_map: Any, targets: Sequence[Any], *, ctx: Any = None, hz: float = 32.0, smoothing: Any = None)

Hold connect_controls's retry, so an application does not carry it as a global.

connect_controls answers None while a target cannot yet say what it exports, which is the normal state for an application that launches its own target: it necessarily binds before that target exists. That leaves every such application holding the same three things — a nullable bus, a guard, and a re-try — and every one of them wrote it out. This is those three things and nothing else.

Parameters:

Name	Type	Description	Default
`control_map`	`Any`	Exactly `connect_controls`'s arguments, kept for every attempt. The targets are constructed once by the caller and reused: a failed attempt asks each target for its capabilities and stops there, so nothing is bound and no target is left part-way.	required
`targets`	`Any`	Exactly `connect_controls`'s arguments, kept for every attempt. The targets are constructed once by the caller and reused: a failed attempt asks each target for its capabilities and stops there, so nothing is bound and no target is left part-way.	required
`ctx`	`Any`	Exactly `connect_controls`'s arguments, kept for every attempt. The targets are constructed once by the caller and reused: a failed attempt asks each target for its capabilities and stops there, so nothing is bound and no target is left part-way.	required
`hz`	`Any`	Exactly `connect_controls`'s arguments, kept for every attempt. The targets are constructed once by the caller and reused: a failed attempt asks each target for its capabilities and stops there, so nothing is bound and no target is left part-way.	required
`smoothing`	`Any`	Exactly `connect_controls`'s arguments, kept for every attempt. The targets are constructed once by the caller and reused: a failed attempt asks each target for its capabilities and stops there, so nothing is bound and no target is left part-way.	required

Examples:

>>> from myogestic.controls import ControlLink, load_control_map
>>> class NotStartedYet:
...     def capabilities(self):
...         return None                   # the target is not up
>>> control_map = load_control_map({"dofs": {"aim": "cursor.x"}})
>>> link = ControlLink(control_map, [NotStartedYet()])
>>> link.ensure() is None                 # call it again on the next click
True
>>> link.bus is None
True

Notes

Call ensure from a UI handler or a training thread — anywhere that can afford to block — and never from @pipeline.predict: asking a target what it exports costs a blocking RPC, and that callback has a deadline. predict reads bus and no-ops while it is None. One link is one atomic failure domain: if independently useful targets may start or fail separately, give each its own map and link.

bus `property` ¶

bus: ControlBus | None

The bus, or None while no target has answered. Read-only.

ensure ¶

ensure() -> ControlBus | None

Bind if it is not bound yet, and return the bus. Idempotent and cheap once bound.

Returns:

Type	Description
`ControlBus or None`	`None` while any target is still unreachable — try again later. Safe to call on every click; once it has answered, this is one attribute read.

stop ¶

stop() -> None

Rest and tear down the bus, and forget it. Idempotent.

The link is reusable afterwards: the next ensure binds the same targets again.

ControlLinkConnector ¶

ControlLinkConnector(link: ControlLink, *, retry_s: float = 2.0)

Resolve a deferred ControlLink without blocking UI or prediction frames.

A target launched from an application's process panel cannot answer its capability request when the application first opens. ControlLink deliberately keeps the blocking retry explicit; this coordinator supplies a rate-limited, single-flight background retry for UI loops.

Parameters:

Name	Type	Description	Default
`link`	`ControlLink`	The `ControlLink` to resolve. It remains the owner of the eventual bus.	required
`retry_s`	`float`	Minimum time between background attempts. `poll(force=True)` bypasses this interval, but never starts a second attempt while one is in flight.	`2.0`

Notes

Call poll from a UI loop. Prediction code only reads link.bus or connected; it never calls ControlLink.ensure, which may block on a remote procedure call.

connected `property` ¶

connected: bool

Whether the link has resolved to a bus.

busy `property` ¶

busy: bool

Whether a capability request is currently in flight.

status `property` ¶

status: str

A short status suitable for an application's process panel.

last_error `property` ¶

last_error: Exception | None

The most recent unexpected connection error, or None.

poll ¶

poll(*, force: bool = False) -> bool

Start one non-blocking connection attempt when it is due.

Returns:

Type	Description
`bool`	`True` only when this call started a worker.

ensure_now ¶

ensure_now() -> ControlBus | None

Make one synchronous attempt unless a background worker owns it.

stop ¶

stop(*, timeout_s: float = 4.0) -> None

Stop retrying, tear down the bus, and briefly join an in-flight request.

Targets¶

Anything that moves is a target: three methods and a list of Capability. The protocol is structural, so there is no base class to inherit and nothing to register — an object with these methods is a target. myogestic.remote.RemoteTarget and myogestic.keyboard.KeyboardTarget are the two this project ships; Drive your own device writes a third.

Target ¶

Bases: Protocol

Drive some control DOFs. One protocol for every application.

A target is user-owned, exactly like myogestic.outputs.Outlet: construct it, hand it to a ControlBus, and register teardown with app.cleanup_hooks. The framework does not track it.

Notes

bind runs on the main thread at construction and may raise — that is the place to reject a configuration this target cannot drive, while there is still a human reading the traceback. send runs on the predict thread and must not raise; the bus already guarantees every value it delivers is finite and inside its declared range.

claims `instance-attribute` ¶

claims: frozenset[str]

Which aliases this target drives, for the bus's coverage check.

Absent means "assume it takes everything" — right for a recorder or a test double that does not know. But if every target reports and a control appears in none of them, nothing drives it, which looks exactly like a control that works and holds still. Report it if you can.

bind ¶

bind(controls: ControlSet) -> None

Accept (or refuse) a configuration, before anything is running.

send ¶

send(values: Mapping[str, float | str], changed: Mapping[str, str]) -> None

Actuate one tick.

Parameters:

Name	Type	Description	Default
`values`	`Mapping[str, float \| str]`	Every declared DOF, sanitised: continuous names map to finite floats inside their declared range, discrete names to a valid state.	required
`changed`	`Mapping[str, str]`	Only the discrete DOFs whose state settled this tick — the edges. A continuous DOF is always in `values`; a discrete one is only in `changed` when it just changed, because re-sending a keystroke is not the same as re-sending a pose.	required

stop ¶

stop() -> None

Release whatever this target owns. Must be idempotent.

capabilities ¶

capabilities() -> Sequence[Any] | None

What this target exports, as myogestic.controls.Capability values.

What myogestic.controls.connect_controls asks so it can resolve a map before anything is bound. Return None — not an empty sequence — while the target cannot answer, e.g. a remote target that has not started: empty reads as "drives nothing" and would resolve to a bus that silently drives nothing.

Absent is fine for a target whose vocabulary is fixed and known to the caller.

A target for a separate program¶

RemoteTarget ¶

RemoteTarget(*, client: Any = None, interface: Any = None)

Drive a target in another process from control values.

Requires a remote target that speaks the v2 control contract: it asks what that target exports and refuses anything it cannot place. A build older than the vocabulary this client needs is refused by version at bind, not driven on a guess — see myogestic.remote._control.RemoteClient.capabilities.

One target drives a whole control map. It owns one LSL outlet per address it drives, each named for that address and one channel wide, all built after negotiation has resolved which addresses those are.

Parameters:

Name	Type	Description	Default
`client`	`Any`	A `myogestic.remote._control.RemoteClient` — `spec.control_client()`. Required: the control space is negotiated over it, and it carries discrete state, which the pose streams cannot express.	`None`
`interface`	`Any`	The `~myogestic.remote.InterfaceSpec` to build streams from — for the Virtual Hand, `virtual_hand()`. The target calls `stream_outlet(address, n_channels=1)` on it once per address it drives, so the application never states a stream name or a width. Anything with that method serves: a recorder or a test double substitutes here.	`None`

Notes

bind refuses a configuration it cannot drive rather than driving part of it: a silently-dropped control is indistinguishable from one holding still.

Binding is deferred rather than decided when the far side is silent, because an application that launches its own remote target from its UI necessarily binds before that target exists. Call negotiate once it is up. A remote target that answers and does not speak the current vocabulary raises.

On shutdown every stream's declared rest is pushed and flushed: the send loop is paced, so a pushed-only value would sit unsent while the process exits, leaving the hand at its last commanded pose.

Examples:

>>> from myogestic.controls import ControlBus, load_control_map, resolve
>>> from myogestic.remote import RemoteTarget
>>> from myogestic.vhi import virtual_hand
>>>
>>> vhi = virtual_hand()
>>> client = vhi.control_client()
>>> control_map = load_control_map({"dofs": {"my_index": "vhi.prediction.index"}})
>>> controls = resolve(control_map, client.capabilities())   # needs VHI running
>>> bus = ControlBus(controls, targets=[RemoteTarget(client=client, interface=vhi)])
>>> _ = bus.push({"my_index": 0.8})        # sanitised on the way to the wire

claims `property` ¶

claims: frozenset[str]

Which aliases this target actually drives.

ControlBus reads this to check that every control was claimed by someone — a map may also name controls another target drives, a keyboard's for instance.

negotiated `property` ¶

negotiated: bool

Whether the contract has been settled with the remote target.

bind ¶

bind(controls: ControlSet) -> None

Negotiate the control space with the remote target, or defer until it is reachable.

Raises:

Type	Description
`ValueError`	When no client was given, when the remote target is too old for the vocabulary this client speaks, or when it answers and the configuration cannot be driven — an address it does not export, or two aliases aimed at one control. A remote target too old to answer `capabilities()` at all looks identical to one that is simply not up yet, so that case is deferred, never raised.
`Exception`	Whatever an outlet raises: a settled negotiation puts each declared rest value on the wire, so a stream that cannot be published fails here rather than per tick. Deliberate — this is a main-thread setup call where a traceback is visible, and a target that cannot write at bind will not write later either.

negotiate ¶

negotiate(*, force: bool = False) -> bool

Retry the handshake now — for an application that launches its own remote target.

Cheap and idempotent when already settled, so it is fine to call from a button handler. Never call it from the predict thread — it blocks on an RPC.

Parameters:

Name	Type	Description	Default
`force`	`bool`	Re-run the handshake even if one already succeeded. Useful after the remote target restarts or its manifest changes — this target caches what it resolved and does not notice on its own. There is no automatic detection of that today.	`False`

Returns:

Type	Description
`bool`	Whether the contract is settled. `False` means the remote target has not answered yet, so nothing is being driven and this will keep retrying.

Raises:

Type	Description
`ValueError`	If the remote target answers and the configuration cannot be driven. Call this from a setup path or a button handler where a traceback is visible, never from the predict thread.
`Exception`	Whatever an outlet raises when the rest values that settle a negotiation are pushed onto a dead one — the same reason `bind` can, and the same remedy.

send ¶

send(values: Mapping[str, float | str], changed: Mapping[str, str]) -> None

Encode one sample per driven control for the negotiated contract.

Only the names bind accepted are read, and each falls back to its own declared rest, so neither a stray key nor a missing one can move a joint or raise on the predict thread.

changed carries discrete edges, which go over gRPC rather than onto a stream.

capabilities ¶

capabilities() -> tuple[Capability, ...] | None

What the remote target exports, or None while it cannot be reached.

The same question KeyboardTarget.capabilities answers off a local list, so a caller can ask a mixed set of targets what they drive without knowing which of them needs a live connection. None is the honest answer for a remote target that has not started: not an empty manifest, which would read as "drives nothing".

stop ¶

stop() -> None

Return the hand to its declared rest pose, and take every stream down.

Each outlet is this target's own — it built them all — so each is stopped here. Nothing else can: see _retire, which every one of them goes through.

Per outlet, not per target. Teardown is the path most likely to meet an outlet that is already dead, and a single push or flush raising used to abandon every outlet after it — still published, still discoverable, and still in _outlets, so a retry double-stopped the ones it had already released. Each is rested and released on its own now, the state is cleared up front so a second stop is a genuine no-op, and the first failure is re-raised once they are all down.

Negotiating with the target¶

Hand RemoteTarget a control client and it asks what the target drives at bind time, then encodes according to the answer.

One target drives the whole map. It owns one LSL outlet per address it drives, each named for that address and one channel wide, all built after negotiation has resolved which addresses those are:

vhi = virtual_hand()
client = vhi.control_client()
# No stream is named here and none is counted. `interface=` is why: which controls exist
# is the manifest's answer, and each one's stream is named for its own address, so both
# facts arrive together once `bind` has something to ask.
target = RemoteTarget(client=client, interface=vhi)
bus = connect_controls(control_map, [target])   # None while the far side is unreachable

An application that launches its own target binds before that target exists, so it holds a ControlLink instead and calls ensure() from each handler that needs the hand. A UI loop that should reconnect automatically wraps the link in ControlLinkConnector and calls poll(); the blocking capability request then runs in one rate-limited background worker.

The client is required: every address, range and state comes from that answer. A target that answers but reports an older vocabulary_version is refused by name, since it would be listening for a stream layout this client no longer publishes and would report nothing while the rig stayed still.

Four things are refused rather than half-driven: a target too old for the vocabulary this client speaks, an address the target does not export, one it exports as something other than a number, and two aliases aimed at one control. A partly understood negotiation leaves some controls believed driven and others quietly dropped, and a dropped control is indistinguishable from one that is working and holding still.

One case is deferred instead: a target that has not answered at all. Nothing is decided until negotiate settles it. A Virtual Hand older than 2.0 has no manifest to answer with, and MyoGestic 2.x carries no fallback table to drive one from, but it answers capabilities() with None exactly as a target that is simply not up yet does, so it is retried the same way.

control_client ¶

control_client() -> RemoteClient

Construct a client for this target's control service.

Hand it to myogestic.remote.RemoteTarget and it asks the remote target which named DOFs it drives, refusing a configuration it cannot place. Required: without one there is nothing to negotiate the control space against.

Imported lazily — a plain install has no [grpc] extra, and stream_outlet / launcher must keep working without it.

Examples:

>>> from myogestic.controls import ControlBus, Continuous, ControlSet
>>> from myogestic.remote import RemoteTarget
>>> from myogestic.vhi import virtual_hand
>>>
>>> vhi = virtual_hand()
>>> controls = ControlSet(dofs={"my_index": Continuous("my_index")})
>>> target = RemoteTarget(client=vhi.control_client(), interface=vhi)
>>> bus = ControlBus(controls, targets=[target])
>>> target.negotiated     # True once the remote end has answered, False until then
False

Three layers of smoothing¶

Three separate mechanisms sit at three different places, and collapsing any two of them is a bug:

Layer	Where	Applies to	Authoritative?
1. Continuous smoothing	`ControlBus(smoothing=...)`, MyoGestic	continuous DOFs	Yes - it decides the value that is commanded
2. Debounce	`Discrete.debounce_s`, MyoGestic	discrete DOFs	Yes - it decides when a state transition happens
2b. Dead zone + hysteresis	`ControlBus(dead_zone=...)`, `ControlBus(hysteresis=...)`, MyoGestic	continuous DOFs	Yes - they decide the value that is commanded
3. Presentation blending	the target (`control_client().set_presentation`)	how a commanded value looks	No - appearance only

Layer 1 runs before any target sees a frame, and that is what makes it authoritative: smoothing after delivery would mean different targets acted on different values.

Layer 2 is the one people reach for layer 1 to solve. Low-pass filtering a discrete control as though it were an axis averages "rest" and "fist" and interpolates through states nobody selected. What a noisy classifier needs is a stability gate: hold the new state for debounce_s before it counts, with optional hysteresis so a value hovering near a boundary settles on one side. debounce_s is declared on the DOF for that reason, rather than configured on the filter.

Layer 3 affects presentation only. With blending on and no debounce every accepted transition is still applied; blending smooths how the change is drawn, not whether it happened.

Recording is not control¶

A discrete DOF is a held state: ask for a grip, hold a grip. Collecting regression training data wants the opposite, a control hand that keeps moving, so the recorded kinematics sweep a continuous range for EMG windows to align against.

Two different jobs, so two vocabularies. The sweep lives in the recording aid, never in the control standard, because bending a held state to accommodate data collection would make hand.grip mean "grip, unless someone is recording".

While a recording trajectory runs it owns the control hand, and discrete DOFs are refused with a reason rather than silently interrupting the trajectory a recording is aligned against. Continuous DOFs keep flowing.

recording_client ¶

recording_client() -> RecordingClient

Construct a client for this target's recording session gate.

Not a control plane, and nothing it does is a control DOF. It carries the two things a recording session needs: the gate that stops the target's own local input competing as a movement source, and trajectories that cycle its control rig so the recorded kinematics sweep a continuous range.

Imported lazily, like the other gRPC client, so a plain install without the [grpc] extra can still use stream_outlet / launcher.

Examples:

>>> from myogestic.vhi import virtual_hand
>>> aid = virtual_hand().recording_client()
>>> aid.set_recording_session(True)   # False when the target is unreachable
False

Encoding helpers¶

Wire-level helpers, for a target that needs a vector rather than a mapping.

encode ¶

encode(controls: ControlSet, values: Mapping[str, Any]) -> ndarray

Continuous DOF values as one float32 vector in declaration order.

Discrete DOFs are absent by construction — they travel as edges, not as a per-tick frame. Call substitute_rest (and usually clip) first.

Examples:

>>> from myogestic.controls import Continuous, ControlSet, encode
>>> controls = ControlSet(dofs={"a": Continuous("a"), "b": Continuous("b")})
>>> encode(controls, {"a": 0.5, "b": -0.25}).tolist()
[0.5, -0.25]

decode ¶

decode(controls: ControlSet, frame: Sequence[float]) -> dict[str, float]

Inverse of encode — a wire frame back to named continuous values.

Raises:

Type	Description
`ValueError`	If `frame` is not as wide as the declared continuous DOFs.

Examples:

>>> from myogestic.controls import ControlSet, Continuous, decode
>>> controls = ControlSet(dofs={"a": Continuous("a")})
>>> decode(controls, [0.75])
{'a': 0.75}

clip ¶

clip(controls: ControlSet, values: Mapping[str, Any]) -> tuple[dict[str, float | str], tuple[str, ...]]

Clamp every value into its DOF's declared range.

Clamping is to the declared domain, never to a global rail: a DOF declared [0.0, 1.0] must not be able to emit -1. Unknown discrete states snap to rest. Returns the frame and the names that were actually clamped, so a caller can report them once instead of every tick.

Never raises. Call substitute_rest first — this assumes finite numbers.

Examples:

>>> from myogestic.controls import Continuous, ControlSet, clip
>>> controls = ControlSet(dofs={"g": Continuous("g", lo=0.0, hi=1.0)})
>>> clip(controls, {"g": 2.5})
({'g': 1.0}, ('g',))

substitute_rest ¶

substitute_rest(controls: ControlSet, values: Mapping[str, Any]) -> dict[str, float | str]

Fill a full frame, replacing anything unusable with the DOF's rest value.

A missing key, a non-finite number and an unknown discrete state all become rest. Keys that are not declared DOFs are dropped.

Never raises: this runs on the predict thread, where an exception is logged with a full traceback on every tick.

Examples:

>>> import numpy as np
>>> from myogestic.controls import Continuous, ControlSet, substitute_rest
>>> controls = ControlSet(dofs={"a": Continuous("a")})
>>> substitute_rest(controls, {"a": np.nan, "unrelated": "x"})
{'a': 0.0}

Control standard¶

Declaring a control space¶

Classification uses the same mapping¶

load_control_map ¶

resolve ¶

Capability dataclass ¶

signed property ¶

ControlMap dataclass ¶

addresses ¶

as_control_space ¶

Binding dataclass ¶

TargetRef dataclass ¶

read_control_space ¶

ControlSet dataclass ¶

continuous property ¶

discrete property ¶

channel_labels ¶

rest_values ¶

Continuous dataclass ¶

Discrete dataclass ¶

Dof module-attribute ¶

Delivering a frame¶

ControlBus ¶

push ¶

select ¶

rebase ¶

stop ¶

connect_controls ¶

ControlLink ¶

bus property ¶

ensure ¶

stop ¶

ControlLinkConnector ¶

connected property ¶

busy property ¶

status property ¶

last_error property ¶

poll ¶

ensure_now ¶

stop ¶

Targets¶

Target ¶

claims instance-attribute ¶

bind ¶

send ¶

stop ¶

capabilities ¶

A target for a separate program¶

RemoteTarget ¶

claims property ¶

negotiated property ¶

bind ¶

negotiate ¶

send ¶

capabilities ¶

stop ¶

Negotiating with the target¶

control_client ¶

Three layers of smoothing¶

Recording is not control¶

recording_client ¶

Encoding helpers¶

encode ¶

decode ¶

clip ¶

substitute_rest ¶

Capability `dataclass` ¶

signed `property` ¶

ControlMap `dataclass` ¶

Binding `dataclass` ¶

TargetRef `dataclass` ¶

ControlSet `dataclass` ¶

continuous `property` ¶

discrete `property` ¶

Continuous `dataclass` ¶

Discrete `dataclass` ¶

Dof `module-attribute` ¶

bus `property` ¶

connected `property` ¶

busy `property` ¶

status `property` ¶

last_error `property` ¶

claims `instance-attribute` ¶

claims `property` ¶

negotiated `property` ¶