Python API reference

Package version

__version__ = '0.2.17.1'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

Device and transport API

class BaseComm[source]

Bases: object

connect()[source]
disconnect()[source]
send(message)[source]
Parameters:

message (str)

receive()[source]
Return type:

str

is_connected()[source]
Return type:

bool

flush_input()[source]

Discard buffered inbound bytes so the next read starts clean.

fast_telemetry_device()[source]

Return the raw byte stream carrying fast telemetry, if supported.

class DualHandExo(left, right)[source]

Bases: object

Unified controller for a left and/or right HandExo.

Either left or right may be None if that side is not connected — the class will skip that side gracefully.

Parameters:
  • left (HandExo or None) – Left-hand exoskeleton instance (already connected).

  • right (HandExo or None) – Right-hand exoskeleton instance (already connected).

close()[source]

Close both serial connections.

info()[source]

Return merged device info with side-qualified motor names.

Left motors occupy indices 0 … N_L-1; right motors follow. Each motor dict has an extra 'side' key ('left' or 'right'). The 'name' key contains the qualified name (e.g. 'L:wrist').

Return type:

dict

send_command(cmd)[source]

Broadcast a raw serial command to all connected exos.

Parameters:

cmd (str)

get_motor_angle(motor_id='all')[source]
set_motor_angle(motor_id, angle)[source]
Parameters:

angle (float)

get_absolute_motor_angle(motor_id='all')[source]
set_absolute_motor_angle(motor_id, angle)[source]
Parameters:

angle (float)

get_motor_torque(motor_id='all')[source]
get_motor_current(motor_id='all')[source]
enable_motor(motor_id='all')[source]
disable_motor(motor_id='all')[source]
home(motor_id='all')[source]
apply_calibration(profile_or_path=None, profiles_dir=None, side=None)[source]

Apply a calibration profile to one or both exos.

Parameters:
  • profile_or_path (str or None) – Profile name (e.g. "zach left new") or full file path. None → load the side-specific default from config.json.

  • profiles_dir (str or None) – Directory containing profile JSON files. Defaults to the repo’s examples/calibration/profiles/ directory.

  • side ('left', 'right', or None) – Which exo to apply the profile to. When None, the method reads the profile’s side field; if absent, defaults to 'right'.

Raises:

RuntimeError – If the target exo is not connected.

set_zero_offset(motor_id, offset)[source]
Parameters:

offset (float)

set_flip(motor_id, flip)[source]
Parameters:

flip (bool)

set_motor_limits(motor_id, lo, hi)[source]
Parameters:
  • lo (float)

  • hi (float)

set_debug(enable)[source]
Parameters:

enable (bool)

set_exo_mode(mode)[source]
Parameters:

mode (str)

class DualSerialComm(cmd_port, telem_port, baudrate, command_delimiter=';', timeout=1, response_timeout=2.0, verbose=False, line_terminator='\r\n')[source]

Bases: BaseComm

Two USB-CDC (ACM) ports from one device on a single cable.

Commands are written on the command port and replies / telemetry are read from the telemetry port, so command writes never wait behind telemetry reads (the device-side fix for host head-of-line blocking). The ASCII line protocol and framing are identical to SerialComm; this only splits the transport across two COM ports. Requires the dual-CDC OpenRB-150 firmware (see the set_reply_route command).

connect() probes the two ports to determine direction, so it is robust to COM-number / USB-interface ordering: whichever port answers is used as the command port. It then switches the firmware to reply_route:telem so the command port carries no return traffic (full decoupling).

Reads are decoupled from writes by a background reader thread. The thread drains the telemetry port continuously and splits the stream into frames: a delimited reply goes on a queue, and any unsolicited telemetry / debug line is kept separately. Nothing on the command path ever performs an inline blocking read of the port, so a chatty device cannot stall a write and a backlog cannot accumulate. receive() waits on the queue, which returns immediately when a reply has already landed.

MAX_TELEMETRY_LINES = 1000

Cap on retained unsolicited lines, so a chatty device cannot grow memory without bound when nobody is consuming telemetry.

MAX_QUEUED_REPLIES = 256

Cap on queued reply frames. A long-running process that sends commands without reading replies would otherwise grow this queue forever; past the cap the oldest frame is dropped, since the newest is what a caller waiting on a fresh transaction actually wants.

connect()[source]
telemetry_lines()[source]

Snapshot of recent unsolicited telemetry / debug lines.

reader_alive()[source]

True while the background telemetry reader is running.

Return type:

bool

close()[source]
send(message)[source]
Parameters:

message (str)

receive(wait_until_return=False, timeout=None, *, warn_on_timeout=True)[source]

Return the next reply frame collected by the reader thread.

This never touches the serial port: the reader thread owns it. Waiting here blocks only this caller, and only until a frame is queued – the telemetry port keeps draining regardless, so a slow or absent reply can never back up the link or stall a subsequent write.

Return type:

str

is_connected()[source]
Return type:

bool

flush_input()[source]

Drop replies queued before this point.

Called before a send so a transaction cannot be handed a stale frame left over from an earlier timed-out request. The port itself is not reset – the reader thread owns it, and resetting underneath it would truncate a frame mid-parse.

fast_telemetry_device()[source]

Return the CDC stream used by the current binary-frame firmware.

Text replies follow reply_route:telem and are drained from _telem by the background reader. The current firmware writes the compact NX frame directly to the primary command CDC, which has no competing reader and can therefore be consumed synchronously here.

class FakeHandExo(comm, name='FakeHandExo', verbose=False)[source]

Bases: object

Fake HandExo class for testing purposes. This class simulates the behavior of the real HandExo device.

Parameters:
  • comm (BaseComm)

  • name (str)

  • verbose (bool)

logger(*argv, warning=False)[source]

Robust debugging print function

Parameters:
  • *argv – (str) Messages to log.

  • warning (bool) – (bool) If True, prints the message in yellow.

set_comm(comm)[source]

Sets the communication interface for the exoskeleton.

Parameters:

comm (BaseComm) – The communication interface to use.

connect()[source]

Simulate connecting to the HandExo device.

close()[source]

Simulate closing the connection to the HandExo device.

get_imu_angles()[source]

Simulate reading IMU angles.

set_imu_angles(roll, pitch, yaw)[source]

Simulate setting IMU angles.

get_motor_angles()[source]

Simulate reading motor angles.

set_motor_angles(angles)[source]

Simulate setting motor angles.

home()[source]

Simulate homing the device.

set_gesture_state(gesture_state)[source]

Simulate setting the gesture state.

Parameters:

gesture_state (str) – The gesture state to set.

class GestureController(exo, verbose=False)[source]

Bases: object

A simple gesture controller for the Hand Exoskeleton.

It allows setting gestures and states (open/close) on the exoskeleton.

Parameters:
  • exo (HandExo | FakeHandExo) – An instance of HandExo or FakeHandExo.

  • verbose (bool) – If True, prints verbose output.

POSTURE_STATES = ('open', 'close')

Multi-joint postures (grasp, keygrip, pinch_*, peace) use these.

JOINT_STATES = ('extend', 'rest', 'flex')

Per-joint gestures (thumb/index/middle/ring/pinky/wrist) use these. rest needs firmware >= 0.3.0; HandExo.set_gesture enforces that.

set_gesture(gesture, state)[source]

Set the gesture and state on the exoskeleton.

Parameters:
  • gesture (str) – The gesture name (e.g., “pinch_index”, “index”, “wrist”).

  • state (str) – “open”/”close” for postures, or “extend”/”rest”/”flex” for per-joint gestures.

Raises:
  • ValueError – If the state is not one of the accepted values.

  • RuntimeError – If the state needs newer firmware than is connected.

Return type:

None

get_current_gesture()[source]

Get the current gesture and state.

Returns:

A tuple of (gesture, state).

Return type:

tuple[str, str]

class HandExo(comm, name='NMLHandExo', command_delimiter='\n', send_delay=0.01, auto_connect=False, verbose=False, side=None)[source]

Bases: object

Class to control the NML Hand Exoskeleton via serial communication.

Features:
  • Enable/disable motors

  • Move motors to specific angles

  • Query status (angle, torque, current)

  • Configure velocity and acceleration

  • Retrieve device information

  • Send low-level serial commands

Parameters:
  • comm (BaseComm)

  • command_delimiter (str)

  • send_delay (float)

  • verbose (bool)

  • side (str | None)

logger(*argv, warning=False)[source]

Robust debugging print function

Parameters:
  • *argv – (str) Messages to log.

  • warning (bool) – (bool) If True, prints the message in yellow.

detect_side()[source]

Query the firmware’s ‘info’ response to determine the hand side.

Updates self.side and returns it. Falls back to 'right' if the firmware does not include a Side field (pre-handedness firmware).

Returns:

'left' or 'right'.

Return type:

str

set_comm(comm)[source]

Sets the communication interface for the exoskeleton.

Parameters:

comm (BaseComm) – The communication interface to use.

connect()[source]

Establishes a connection to the exoskeleton device.

send_command(cmd)[source]

Sends a command to the exoskeleton over the serial connection.

Parameters:

cmd (str) – Command to send to the exoskeleton.

get_fast_telemetry(timeout=0.5, motor_ids=None)[source]

Read the firmware’s compact binary telemetry frame.

The firmware emits an NX frame with one fixed-size record per motor. Records are keyed by Dynamixel ID and include relative/absolute angles, present current, raw velocity, and raw position ticks.

Parameters:
  • timeout (float)

  • motor_ids (list[int] | tuple[int, ...] | None)

Return type:

dict[int, dict[str, float | int | bool]]

configure_shadow_telemetry(motor_ids, *, interval_ms=2)[source]

Configure read-only firmware sampling for explicit Dynamixel IDs.

This command does not enable torque, change a mode, or write a motor register. Sampling remains stopped until start_shadow_telemetry().

Parameters:
  • motor_ids (list[int] | tuple[int, ...])

  • interval_ms (int)

Return type:

str

start_shadow_telemetry()[source]

Start read-only sampling; firmware accepts this in VELOCITY mode only.

Return type:

str

stop_shadow_telemetry()[source]

Stop read-only sampling without changing any motor state.

Return type:

str

get_shadow_telemetry(timeout=0.5)[source]

Return the firmware’s buffered Phase-1 shadow evidence snapshot.

Parameters:

timeout (float)

Return type:

dict

enable_motor(motor_id='all')[source]

Enables the torque output for the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to enable.

Returns:

None

enable_motors_by_id(motor_ids)[source]

Enable torque for a list of explicit Dynamixel IDs.

Uses per-ID legacy commands for firmware compatibility.

is_enabled(motor_id='all')[source]

Checks if the specified motor is enabled.

Parameters:

motor_id (int or str) – ID of the motor to check.

Returns:

True if the motor is enabled, False otherwise.

Return type:

bool

disable_motor(motor_id='all')[source]

Disables the torque output for the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to disable.

Returns:

None

disable_motors_by_id(motor_ids)[source]

Disable torque for a list of explicit Dynamixel IDs.

Uses per-ID legacy commands for firmware compatibility.

enable_led(motor_id='all')[source]

Enables the LED for the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to enable the LED for.

Returns:

None

disable_led(motor_id='all')[source]

Disables the LED for the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to disable the LED for.

Returns:

None

help()[source]

Sends a help command to the exoskeleton to retrieve available commands.

Returns:

A string containing the help information from the exoskeleton.

Return type:

str

set_debug(enable)[source]

Enables or disables verbose debug output from the Arduino.

Parameters:

enable (bool) – True to enable debug output, False to disable.

Returns:

None

version()[source]

Gets the version of the exo

Return type:

str

firmware_version(refresh=False)[source]

Return the device firmware version as a comparable tuple, e.g. (0, 3, 0).

The result is cached because feature gates call this on every guarded command and a serial round-trip per call would dominate their cost. Returns () if the device did not answer or the reply was unparseable; that value compares less than any real version, so gates fail closed.

Parameters:

refresh (bool) – Re-query the device instead of using the cache.

Returns:

Parsed version, or () if unknown.

Return type:

tuple[int, …]

firmware_at_least(minimum)[source]

Check whether the connected firmware is at least minimum.

Parameters:

minimum (tuple[int, ...]) – Version to compare against, e.g. (0, 3, 0).

Returns:

True if the device reports a version >= minimum.

Return type:

bool

require_gesture_angle_support()[source]

Raise unless the device supports continuous gesture positioning.

Return type:

None

home(motor_id='all')[source]

Sends a home command to all motors, unless a specific motor ID is provided.

Parameters:

motor_id (int or str) – ID of the motor to home, or ‘all’ to home all motors.”

Returns:

None

info(timeout=5.0)[source]

Request and parse exoskeleton info into a structured dictionary.

Returns:

info – Keys: - name : str - version : str - n_motors : int - motors : {motor_id: {…}} - motor_<id> : per-motor dicts (back-compat)

Return type:

dict

Parameters:

timeout (float)

get_baudrate(motor_id='all')[source]

Retrieves the current baud rate of the serial connection.

Returns:

The current baud rate.

Return type:

int

Parameters:

motor_id (int)

set_baudrate(motor_id, baudrate)[source]

Sets the baud rate for the specified motor.

Parameters:
  • motor_id (int or str) – ID of the motor to set the baud rate for.

  • baudrate (int) – Desired baud rate (e.g., 57600, 115200).

Returns:

None

get_motor_velocity(motor_id='all')[source]

Retrieves the current velocity of the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to query.

Returns:

Current velocity of the motor in degrees per second.

Return type:

float

set_motor_velocity(motor_id, velocity)[source]

Sets the velocity for the specified motor.

Parameters:
  • motor_id (int or str) – ID of the motor to set the velocity for.

  • velocity (float) – Desired velocity in degrees per second.

Returns:

None

get_motor_velocity_limit(motor_id='all')[source]

Read the position-profile velocity limit in rpm.

Parameters:

motor_id (int)

set_motor_velocity_limit(motor_id, velocity_rpm)[source]

Set one motor’s position-profile limit from a value expressed in rpm.

Parameters:
  • motor_id (int)

  • velocity_rpm (float)

get_present_velocity(motor_id='all')[source]

Read signed present velocity in rpm.

Parameters:

motor_id (int)

set_direct_velocity(motor_id, velocity_rpm)[source]

Command signed velocity in rpm while firmware is in velocity mode.

Parameters:
  • motor_id (int)

  • velocity_rpm (float)

get_motor_acceleration(motor_id='all')[source]

Retrieves the current acceleration of the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to query.

Returns:

Current acceleration of the motor in degrees per second squared.

Return type:

float

set_motor_acceleration(motor_id, acceleration)[source]

Sets the acceleration for the specified motor.

Parameters:
  • motor_id (int or str) – ID of the motor to set the acceleration for.

  • acceleration (float) – Desired acceleration in degrees per second squared.

Returns:

None

get_motor_angle(motor_id='all')[source]

Retrieves the current relative angle of the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to query.

Returns:

Current angle of the motor in degrees.

Return type:

float

set_motor_angle(motor_id, angle)[source]

Sets the angle for the specified motor.

Parameters:
  • motor_id (int or str) – ID of the motor to set the angle for.

  • angle (float) – Desired angle in degrees.

Returns:

None

get_absolute_motor_angle(motor_id='all')[source]

Retrieves the absolute angle of the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to query.

Returns:

Absolute angle of the motor in degrees.

Return type:

float

set_absolute_motor_angle(motor_id, angle)[source]

Sets the absolute angle for the specified motor.

Parameters:
  • motor_id (int or str) – ID of the motor to set the absolute angle for.

  • angle (float) – Desired absolute angle in degrees.

Returns:

None

get_home(motor_id='all')[source]

Retrieves the home angle of the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to query.

Returns:

Home angle of the motor in degrees.

Return type:

float

set_home(motor_id)[source]

Sets the current position as the new home/zero position for the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to set the home position for.

Returns:

None

get_motor_torque(motor_id='all')[source]

Retrieves the current torque of the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to query.

Returns:

Current torque of the motor in Newton-meters.

Return type:

float

get_motor_current(motor_id='all')[source]

Retrieves the current draw of the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to query.

Returns:

Current draw of the motor in Amperes.

Return type:

float

get_motor_current_limit(motor_id='all')[source]

Retrieves the current limit of the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to query.

Returns:

Current limit of the motor in Amperes.

Return type:

float

set_current_limit(motor_id, current_limit)[source]

Sets the per-motor current limit for the specified motor.

This is a PER-MOTOR knob and does not bound what the motors draw together. From firmware 0.4.0 it sets the motor’s nominal effort and the firmware’s budget allocator owns GOAL_CURRENT, so the value actually applied may be lower while the fleet is near its combined cap. See set_total_current_limit().

Parameters:
  • motor_id (int or str) – ID of the motor to set the current limit for.

  • current_limit (float) – Desired current limit in mA.

Returns:

None

set_total_current_limit(budget_mA)[source]

Set the COMBINED current budget across all motors, in mA.

Per-motor limits cannot protect the supply: N motors each honouring a 200 mA limit still draw up to N*200 mA together, which is what browns out the board when a posture commands every joint at once. This caps the aggregate. The firmware clamps a new budget to the range it can actually satisfy, so read it back with get_total_current_limit().

Parameters:

budget_mA (float) – Combined budget in mA.

Returns:

None

Raises:
  • RuntimeError – If the device firmware is older than 0.4.0.

  • ValueError – If budget_mA is not a positive number.

get_total_current_limit()[source]

Read the combined current budget across all motors, in mA.

Returns:

Budget in mA, or float('nan') if the device did not answer.

Return type:

float

Raises:

RuntimeError – If the device firmware is older than 0.4.0.

set_hold_current(hold_mA)[source]

Set the current a settled or load-shed motor is allowed, in mA.

Every motor may sit at this value simultaneously, so the firmware caps it at budget / n_motors.

Parameters:

hold_mA (float) – Hold current in mA.

Returns:

None

Raises:
  • RuntimeError – If the device firmware is older than 0.4.0.

  • ValueError – If hold_mA is negative or not a number.

set_current_governor(enabled)[source]

Enable or disable the closed-loop half of budget enforcement.

Disabling stops all current sampling and leaves the conservative feed-forward clamp in charge: strictly safer for the supply, but every motor is allocated its static worst-case share whether or not the fleet is actually drawing that much. It does NOT remove the budget.

Parameters:

enabled (bool) – True to run the governor, False for static clamping.

Returns:

None

Raises:

RuntimeError – If the device firmware is older than 0.4.0.

current_status(timeout=2.0)[source]

Read the budget state: cap, measured aggregate draw and per-motor allocation.

Returns:

  • dict – Keys total_budget_mA, hold_current_mA, governor (bool), measured_total_mA (None until the firmware has sampled), scale (fraction of nominal effort currently allowed), measurement_trusted (bool) and motors keyed by DXL ID with nominal_mA / applied_mA / state. applied below nominal means the budget is actively clamping.

  • Raises – RuntimeError: If the device firmware is older than 0.4.0.

Parameters:

timeout (float)

Return type:

dict

get_goal_current(motor_id='all')[source]

Read signed direct-current goal in mA.

Parameters:

motor_id (int)

set_direct_current(motor_id, current_mA)[source]

Command signed current in mA while firmware is in current mode.

Parameters:
  • motor_id (int)

  • current_mA (float)

stop_direct_control(motor_id='all')[source]

Immediately zero direct velocity and current goals.

Parameters:

motor_id (int)

hold_motor_position(motor_id, relative_angle, hold_current_mA=None)[source]

Hold one explicit DXL ID at a limit-clamped relative angle.

Parameters:
  • motor_id (int)

  • relative_angle (float)

  • hold_current_mA (float | None)

release_motor_hold(motor_id)[source]

Disable one held DXL ID and restore the current global mode.

Parameters:

motor_id (int)

set_direct_command_timeout(timeout_ms)[source]

Set the firmware direct-control watchdog timeout.

Parameters:

timeout_ms (int)

set_control_mode(mode)[source]

Set the global motor mode; firmware leaves torque disabled.

Parameters:

mode (str)

get_motor_limits(motor_id='all')[source]

Retrieves the limits for the specified motor, including minimum and maximum angles.

Parameters:

motor_id (int or str) – ID of the motor to query.

Returns:

A tuple containing the minimum and maximum angles of the motor.

Return type:

tuple

set_motor_upper_limit(motor_id, upper_limit)[source]

Sets the upper limit for the specified motor.

Parameters:
  • motor_id (int or str) – ID of the motor to set the upper limit for.

  • upper_limit (float) – Desired upper limit in degrees.

Returns:

None

set_motor_lower_limit(motor_id, lower_limit)[source]

Sets the lower limit for the specified motor.

Parameters:
  • motor_id (int or str) – ID of the motor to set the lower limit for.

  • lower_limit (float) – Desired lower limit in degrees.

Returns:

None

set_motor_limits(motor_id, lower_limit, upper_limit)[source]

Sets both the lower and upper limits for the specified motor.

Parameters:
  • motor_id (int or str) – ID of the motor to set the limits for.

  • lower_limit (float) – Desired lower limit in degrees.

  • upper_limit (float) – Desired upper limit in degrees.

Returns:

None

reboot_motor(motor_id='all')[source]

Reboots the specified motor.

Parameters:

motor_id (int or str) – ID of the motor to reboot.

Returns:

None

get_motor_mode()[source]

Retrieves the current control mode of the specified motor.

Returns:

Current mode of the motor (e.g., “position”, “velocity”, “current_position”).

Return type:

str

set_motor_mode(mode)[source]

Sets the control mode for all motors (global setting).

Parameters:

mode (str) – Desired control mode (e.g., “position”, “velocity”, “current_position”).

Returns:

None

get_exo_mode()[source]

Retrieves the current operating mode of the exoskeleton.

Returns:

Current mode of the exoskeleton (e.g., “manual”, “autonomous”).

Return type:

str

set_exo_mode(mode)[source]

Sets the operating mode for the exoskeleton.

Parameters:

mode (str) – Desired operating mode (e.g., “manual”, “autonomous”).

Returns:

None

get_gesture()[source]

Retrieves the current gesture recognized by the exoskeleton.

Returns:

Current gesture (e.g., “open”, “close”, “pinch”).

Return type:

str

set_gesture(gesture, state='default')[source]

Sets the gesture for the exoskeleton.

Per-joint gestures (thumb/index/middle/ring/pinky/wrist) accept extend, rest and flex. rest needs firmware >= 0.3.0; rad existed only between 0.3.1 and 0.6.0, where its motor was folded into wrist. Firmware ACKs a gesture or state it cannot resolve, so this raises on either side of that window instead of letting the caller believe a move happened.

Parameters:
  • gesture (str) – Desired gesture (e.g., “grasp”, “index”, “wrist”).

  • state (str) – Desired state (e.g., “open”, “close”, “extend”, “rest”, “flex”).

Returns:

None

Raises:

RuntimeError – If the gesture/state pair needs newer firmware.

set_gesture_angle(gesture, percent)[source]

Position a per-joint gesture anywhere between its two end postures.

Continuous generalization of the extend/rest/flex states: percent interpolates the gesture between its own endpoints, so

  • 0 -> exactly set_gesture(gesture, 'extend')

  • 50 -> halfway between the extend and flex postures

  • 100 -> exactly set_gesture(gesture, 'flex')

Each motor the gesture names travels its own share, so a gesture driving several motors – the thumb, or the coupled wrist pair – keeps the ratio between them at every percentage. Retuning EXTEND_*/FLEX_* in config.h moves the axis with them, so a host’s percentages keep meaning the same postures across a retune.

Note that 0 is the extend posture, not home: with a non-zero EXTEND_* a hand parked at home sits below 0% and get_gesture_angle() reports it as out of range.

get_gesture_angle() is the exact inverse (firmware >= 0.6.0).

Parameters:
  • gesture (str) – A per-joint gesture name; see ANGLE_ADDRESSABLE_GESTURES. Multi-joint postures (grasp, keygrip, pinch_*) are not addressable this way.

  • percent (float) – Position in [0, 100]. The firmware clamps values outside that range.

Returns:

None

Raises:
  • RuntimeError – If the device firmware is older than 0.3.0, or does not define the gesture (rad exists only in 0.3.1 - 0.5.x).

  • ValueError – If percent is not a number.

get_gesture_angle(gesture='all', timeout=1.0)[source]

Read where each per-joint gesture currently sits on its 0-100 axis.

The read-back half of set_gesture_angle(), on the same axis: a gesture commanded to 40 reads back 40 once it arrives, extend reads back as 0 and flex as 100. rest reads back wherever its REST_* constants place it between the two.

Values are integers so the out-of-range signals share the encoding:

  • 0 to 100 – position between the extend and flex postures

  • GESTURE_ANGLE_BELOW_RANGE (101) – past the extend end

  • GESTURE_ANGLE_ABOVE_RANGE (102) – past the flex end

  • GESTURE_ANGLE_UNAVAILABLE (255) – no position available

A hand parked at home reads 101 whenever EXTEND_* is non-zero, since home sits below the extend posture.

A gesture spanning several motors (the thumb, the coupled wrist pair, or any joint on a dual build) reports the mean of the per-motor percentages, skipping joints that cannot carry one.

Parameters:
  • gesture (str) – A single gesture name, or "all".

  • timeout (float) – Seconds to wait for the reply.

Returns:

Gesture name -> code, in firmware order. Empty if the device did not answer.

Return type:

dict[str, int]

Raises:

RuntimeError – If the device firmware is older than 0.6.0.

get_gesture_sang(gesture='all', timeout=1.0)[source]

Read OpenSim-style signed gesture angles in degrees.

The first motor named by each gesture supplies the calibrated physical degree scale. Its rest state is 0 degrees; positions toward flex are positive and positions toward extend are negative. Multi-motor gesture percentages are still aggregated as documented by get_gesture_angle(), then mapped to that first-motor scale.

Parameters:
  • gesture (str) – A single angle-addressable gesture, or "all".

  • timeout (float) – Seconds to wait for the reply.

Returns:

Gesture name -> signed degree delta from rest. None means no signed angle was available.

Return type:

dict[str, float | None]

Raises:

RuntimeError – If the device firmware is older than 0.6.1.

get_gesture_angles(gesture='all', timeout=1.0)[source]

Read percentage codes and signed degree deltas together.

This is the combined form of get_gesture_angle() and get_gesture_sang(), produced from one batched motor read. Each result has fraction (the legacy 0-100/status code) and angle_delta_deg (rest-zeroed signed degrees, or None).

Parameters:
  • gesture (str) – A single angle-addressable gesture, or "all".

  • timeout (float) – Seconds to wait for the reply.

Returns:

Gesture names mapped to fraction and angle_delta_deg fields.

Return type:

dict

Raises:

RuntimeError – If the device firmware is older than 0.6.1.

set_gesture_state(state)[source]

Sets the state of the current gesture for the exoskeleton.

Parameters:

state (str) – Desired state of the gesture (e.g., “default”, “active”).

Returns:

None

get_gesture_list()[source]

Retrieves the list of available gestures for the exoskeleton.

Returns:

A list of available gestures.

Return type:

list

cycle_gesture()[source]

Cycles through the available gestures for the exoskeleton.

Returns:

None

cycle_gesture_state()[source]

Cycles through the states of the current gesture for the exoskeleton.

Returns:

None

set_zero_offset(motor_id, offset)[source]

Sets the zero offset for the specified motor to an arbitrary value.

Parameters:
  • motor_id (int or str) – ID or name of the motor.

  • offset (float) – Zero offset in degrees (absolute angle of the open/home position).

set_flip(motor_id, flip)[source]

Sets the direction flip flag for a motor.

Parameters:
  • motor_id (int or str) – ID or name of the motor.

  • flip (bool) – True to invert direction, False for normal.

apply_calibration(profile_or_path=None, profiles_dir=None, name_to_id=None)[source]

Loads a calibration profile and pushes all values to the device.

Can be called with a profile name (e.g. “zach”), a full file path, or with no arguments to load the default profile.

Parameters:
  • profile_or_path (str or None) – One of: - A profile name (e.g. “zach”) → loads profiles/zach.json - A full file path to a calibration JSON - None → loads the default profile from profiles/config.json

  • profiles_dir (str or None) – Directory containing profile JSONs. Defaults to examples/calibration/profiles/ relative to the repo root.

  • name_to_id (dict[str, int] or None) – Optional mapping of bare motor name → Dynamixel ID. When provided, calibration commands use the explicit integer ID instead of the bare name string so that duplicate motor names in dual firmware (e.g. two “wrist” motors on left ID 1 and right ID 11) are resolved to the correct side without ambiguity. When None, bare names are used (legacy behaviour, safe only in single-exo firmware builds).

calibrate_exo(mode='timed', duration=10.0)[source]

Starts the calibration routine for the exoskeleton. Note: This feature may not be fully implemented in the firmware.

Parameters:
  • mode (str) – Calibration mode (default: “timed”).

  • duration (float) – Duration in seconds for timed calibration (default: 10.0).

Returns:

None

enable_oled()[source]

Enables the OLED display on the exoskeleton.

Returns:

Response from the device.

Return type:

str

disable_oled()[source]

Disables the OLED display on the exoskeleton.

Returns:

Response from the device.

Return type:

str

get_oled_status()[source]

Gets the current status of the OLED display.

Returns:

OLED status (“OLED ENABLED” or “OLED DISABLED”).

Return type:

str

close()[source]

Closes the serial connection to the exoskeleton.

Returns:

None

get_imu_data()[source]

Retrieves the IMU data from the exoskeleton.

Receives a serial message with contents, as an example:

“Received: get_imu Heading: 0.00, Pitch: 0.00, Roll: 0.00”

Returns:

A dictionary containing IMU data (e.g., accelerometer, gyroscope, magnetometer).

Return type:

dict

get_imu_angles()[source]

Retrieves the current roll, pitch, and yaw of the IMU.

Returns:

[Roll, Pitch, Yaw]

Return type:

list

get_imu_heading()[source]

Retrieves the current yaw of the IMU.

Returns:

Current yaw in degrees.

Return type:

float

get_imu_roll()[source]

Retrieves the current roll of the IMU.

Returns:

Current roll in degrees.

Return type:

float

get_imu_pitch()[source]

Retrieves the current pitch of the IMU.

Returns:

Current pitch in degrees.

Return type:

float

set_yaw_angle(motor_id, target_angle, direction)[source]

Sets the wrist angle (IMU yaw) for the specified motor.

Parameters:
  • motor_id (int or str) – ID of the motor to set the angle for.

  • target_angle (float) – Desired angle in degrees.

  • direction (str) – Desired direction of motion (“flex” or “extend”, or single char ‘f’/’e’)

Returns:

None

get_gesture_state()[source]

Retrieves the current state of the gesture.

Returns:

Current gesture state (e.g., “default”, “active”).

Return type:

str

class LSLClient(stream_name=None, stream_type=None, maxlen=10000, auto_start=True, verbose=False)[source]

Bases: object

start_streaming()[source]
stop_streaming()[source]
get_latest_window(window_ms)[source]
get_connection_status()[source]
get_metadata()[source]

Return dictionary of all available stream metadata.

class LSLMarkerSubscriber(stream_name=None, stream_type='Markers', timeout=5.0, buf_events=2000, recover=True, verbose=False)[source]

Bases: object

Subscribe to a string-valued marker stream (e.g., type=”Markers”).

Usage

def on_marker(text: str, ts: float, exo):

exo.set_gesture(text)

with LSLMarkerSubscriber(stream_type=”Markers”, verbose=True) as sub:

sub.set_callback(on_marker, exo, only_on_change=True) # … do other work; callbacks run in the background thread

start()[source]
stop()[source]
set_callback(func, *args, only_on_change=False, **kwargs)[source]

Register a callback invoked as func(text, timestamp, *args, **kwargs).

only_on_change=True suppresses repeated marker text. Positional and keyword arguments supplied here are forwarded to the callback.

Pass None to clear the callback.

Parameters:
  • func (Callable[[...], Any] | None)

  • only_on_change (bool)

Return type:

None

get_recent_markers(k=1)[source]

Return the last k markers as (timestamp, text).

Parameters:

k (int)

Return type:

List[Tuple[float, str]]

get_markers_since(since_ts)[source]
Parameters:

since_ts (float)

Return type:

List[Tuple[float, str]]

poll_latest()[source]

Non-blocking: return the most recent (ts, text) or None.

Return type:

Tuple[float, str] | None

pull(timeout=0.0)[source]

Wait up to timeout seconds for a new marker (compares against the last one). Returns (ts, text) or None if nothing new appears.

Parameters:

timeout (float)

Return type:

Tuple[float, str] | None

metadata()[source]
Return type:

dict

Parameters:
  • stream_name (Optional[str])

  • stream_type (Optional[str])

  • timeout (float)

  • buf_events (int)

  • recover (bool)

  • verbose (bool)

class LSLMessagePublisher(name='Message', stream_type='Markers', source_id=None, metadata=None, only_on_change=True, chunk_size=1, max_buffered=360)[source]

Bases: object

Publish messages over Lab Streaming Layer (LSL) as a string Marker stream.

  • One channel (string), nominal_srate=0 (irregular), type=’Markers’ by default.

  • Call publish(label) to send ‘pinkyflexion’, ‘thumbflexion’, etc.

  • Optional: pass metadata to embed model/config info in the stream description.

  • Optional: de-duplicate consecutive identical labels.

Example

lsl = GestureLSLBroadcaster(name=”EMGGesture”, metadata={“model”:”128_channels”}) lsl.publish(“pinkyflexion”) lsl.close()

Parameters:
  • name (str)

  • stream_type (str)

  • source_id (Optional[str])

  • metadata (Optional[Dict[str, Any]])

  • only_on_change (bool)

  • chunk_size (int)

  • max_buffered (int)

publish(label, timestamp=None)[source]

Push a single gesture label as a string sample.

Parameters:
  • label (str) – Your gesture string, e.g. “pinkyflexion”.

  • timestamp (float | None) – LSL timestamp; if None, uses local_clock().

Return type:

None

close()[source]

Close the outlet (optional; GC will handle it too).

Return type:

None

class LSLNumericSubscriber(stream_name=None, stream_type=None, timeout=5.0, buf_seconds=10.0, max_chunklen=1024, recover=True, verbose=False)[source]

Bases: object

Subscribe to a continuous numeric stream (e.g., EMG/EEG).

Use get_latest_window() to retrieve a channel-by-sample float32 array from the rolling buffer.

Parameters:
  • stream_name (Optional[str])

  • stream_type (Optional[str])

  • timeout (float)

  • buf_seconds (float)

  • max_chunklen (int)

  • recover (bool)

  • verbose (bool)

start()[source]
stop()[source]
get_latest_window(window_ms)[source]

Return last window_ms of data as (C, N) float32 (zero-padded if not enough).

Parameters:

window_ms (int)

Return type:

ndarray

metadata()[source]
Return type:

dict

exception ProtocolResponseError(*, command, expected, raw_response)[source]

Bases: RuntimeError

A device reply did not match the response required by a command.

Parameters:
  • command (str)

  • expected (str)

  • raw_response (str)

class SerialComm(port, baudrate, command_delimiter=';', timeout=1, response_timeout=2.0, verbose=False)[source]

Bases: BaseComm

connect()[source]
close()[source]
send(message)[source]
Parameters:

message (str)

receive(wait_until_return=False, timeout=None, *, warn_on_timeout=True)[source]

Reads data from the serial device. If wait_until_return is True, waits for a command delimiter until the timeout is reached.

Parameters:
  • wait_until_return (bool) – Whether to wait for a full response ending with delimiter.

  • timeout (float) – Maximum time to wait (in seconds) for complete response.

Returns:

Decoded and cleaned response string.

Return type:

str

is_connected()[source]
Return type:

bool

flush_input()[source]

Discard buffered inbound bytes so the next read starts clean.

fast_telemetry_device()[source]

Return the raw byte stream carrying fast telemetry, if supported.

class TCPComm(ip, port=5001, timeout=5, verbose=False)[source]

Bases: BaseComm

connect()[source]
close()[source]
send(message)[source]
Parameters:

message (str)

receive()[source]
Return type:

str

is_connected()[source]
Return type:

bool

Calibration

Calibration profile storage and range-of-motion utilities.

class CalibrationProfileStore(profiles_dir=PosixPath('/home/runner/work/NML_Hand_Exo/NML_Hand_Exo/examples/calibration/profiles'))[source]

Read and write named, side-aware calibration profiles.

Parameters:

profiles_dir (str | Path)

profile_path(name)[source]
Parameters:

name (str)

Return type:

Path

list_profiles(side=None)[source]

Return sorted profile names, optionally filtered by hand side.

Parameters:

side (str | None)

Return type:

list[str]

load_profile(name, side='right')[source]

Load a profile by name, or the side default when name is None.

Parameters:
  • name (str | None)

  • side (str)

Return type:

dict | None

get_default_profile_name(side='right')[source]

Return the configured default, preserving legacy right-hand fallback.

Parameters:

side (str)

Return type:

str | None

save_profile(name, data, side='right')[source]

Save a named profile and ensure its side metadata is authoritative.

Parameters:
  • name (str)

  • data (dict)

  • side (str)

Return type:

Path

set_default_profile(name, side='right')[source]

Set a side-specific default and maintain the legacy right key.

Parameters:
  • name (str)

  • side (str)

Return type:

None

build_motor_orientation(profile, motor_names)[source]

Return home/flip metadata for every requested motor.

Parameters:
  • profile (dict | None)

  • motor_names (list[str])

Return type:

dict[str, dict[str, float | bool]]

determine_run_number(participant, date_str, output_dir=PosixPath('/home/runner/work/NML_Hand_Exo/NML_Hand_Exo/output_data'))[source]

Return the next run number for a participant and date.

Parameters:
  • participant (str)

  • date_str (str)

  • output_dir (str | Path)

Return type:

int

get_default_profile_name(side='right')[source]
Parameters:

side (str)

Return type:

str | None

list_profiles(side=None)[source]
Parameters:

side (str | None)

Return type:

list[str]

load_profile(name, side='right')[source]
Parameters:
  • name (str | None)

  • side (str)

Return type:

dict | None

normalize_angle(absolute, home, flip)[source]

Normalize an encoder angle to zero-at-home, positive-flexion space.

Parameters:
  • absolute (float)

  • home (float)

  • flip (bool)

Return type:

float

profile_path(name)[source]
Parameters:

name (str)

Return type:

Path

save_profile(name, data, side='right')[source]
Parameters:
  • name (str)

  • data (dict)

  • side (str)

Return type:

Path

set_default_profile(name, side='right')[source]
Parameters:
  • name (str)

  • side (str)

Return type:

None

Intent decoding

Modular EMG intent-decoding components.

This package contains no exoskeleton hardware commands. Decoders produce a guarded intent decision that can be published through the existing LSL contract and consumed by the hand-exo GUI.

class ContinuousRestAdapter(coefficients: 'np.ndarray | None' = None, scale: 'np.ndarray | None' = None, global_baseline: 'np.ndarray | None' = None, ridge: 'float' = 0.001)[source]
Parameters:
  • coefficients (ndarray | None)

  • scale (ndarray | None)

  • global_baseline (ndarray | None)

  • ridge (float)

coefficients: ndarray | None = None
scale: ndarray | None = None
global_baseline: ndarray | None = None
ridge: float = 0.001
fit(features, labels, roll_deg, pitch_deg, rest_label='rest')[source]
Parameters:
  • features (ndarray)

  • labels (ndarray)

  • roll_deg (ndarray)

  • pitch_deg (ndarray)

  • rest_label (str)

Return type:

ContinuousRestAdapter

baseline(roll_deg, pitch_deg)[source]
Parameters:
  • roll_deg (ndarray)

  • pitch_deg (ndarray)

Return type:

ndarray

transform(features, roll_deg, pitch_deg)[source]
Parameters:
  • features (ndarray)

  • roll_deg (ndarray)

  • pitch_deg (ndarray)

Return type:

ndarray

class DecoderDecision(state: 'str', signed_intent: 'float', confidence: 'float', rejected: 'bool', reason: 'str' = '', probabilities: 'dict[str, float]'=<factory>, open_activation: 'float' = 0.0, close_activation: 'float' = 0.0, raw_signed_projection: 'float | None' = None)[source]
Parameters:
  • state (str)

  • signed_intent (float)

  • confidence (float)

  • rejected (bool)

  • reason (str)

  • probabilities (dict[str, float])

  • open_activation (float)

  • close_activation (float)

  • raw_signed_projection (float | None)

state: str
signed_intent: float
confidence: float
rejected: bool
reason: str = ''
probabilities: dict[str, float]
open_activation: float = 0.0
close_activation: float = 0.0
raw_signed_projection: float | None = None
class FeatureConfig(common_mode: 'str' = 'median', log_compress: 'bool' = True, include_waveform_length: 'bool' = False)[source]
Parameters:
  • common_mode (str)

  • log_compress (bool)

  • include_waveform_length (bool)

common_mode: str = 'median'
log_compress: bool = True
include_waveform_length: bool = False
class IntentCaptureSession(participant_id: 'str' = '', device_name: 'str' = '', channel_count: 'int' = 0, features: 'list[np.ndarray]' = <factory>, labels: 'list[str]' = <factory>, groups: 'list[str]' = <factory>, roll_deg: 'list[float]' = <factory>, pitch_deg: 'list[float]' = <factory>, emg_windows: 'list[np.ndarray]' = <factory>, comfort: 'dict[str, float]'=<factory>)[source]
Parameters:
  • participant_id (str)

  • device_name (str)

  • channel_count (int)

  • features (list[ndarray])

  • labels (list[str])

  • groups (list[str])

  • roll_deg (list[float])

  • pitch_deg (list[float])

  • emg_windows (list[ndarray])

  • comfort (dict[str, float])

participant_id: str = ''
device_name: str = ''
channel_count: int = 0
features: list[ndarray]
labels: list[str]
groups: list[str]
roll_deg: list[float]
pitch_deg: list[float]
emg_windows: list[ndarray]
comfort: dict[str, float]
add(feature, label, group, roll_deg=None, pitch_deg=None, emg_window=None)[source]
Parameters:
  • feature (ndarray)

  • label (str)

  • group (str)

  • roll_deg (float | None)

  • pitch_deg (float | None)

  • emg_window (ndarray | None)

Return type:

None

arrays()[source]
class_counts()[source]
Return type:

dict[str, int]

save(path)[source]
Parameters:

path (str | Path)

Return type:

None

classmethod load(path)[source]
Parameters:

path (str | Path)

Return type:

IntentCaptureSession

class IntentDecoderPipeline(rest_label: 'str' = 'rest', open_label: 'str' = 'open', close_label: 'str' = 'close', confidence_threshold: 'float' = 0.6, effort_deadband: 'float' = 0.05, active_reference_quantile: 'float' = 0.9, require_orientation: 'bool' = False, adapter: 'ContinuousRestAdapter' = <factory>, model: 'ShrinkageLDAIntentModel' = <factory>)[source]
Parameters:
  • rest_label (str)

  • open_label (str)

  • close_label (str)

  • confidence_threshold (float)

  • effort_deadband (float)

  • active_reference_quantile (float)

  • require_orientation (bool)

  • adapter (ContinuousRestAdapter)

  • model (ShrinkageLDAIntentModel)

rest_label: str = 'rest'
open_label: str = 'open'
close_label: str = 'close'
confidence_threshold: float = 0.6
effort_deadband: float = 0.05
active_reference_quantile: float = 0.9
require_orientation: bool = False
adapter: ContinuousRestAdapter
model: ShrinkageLDAIntentModel
fit(features, labels, roll_deg, pitch_deg)[source]
Parameters:
  • features (ndarray)

  • labels (ndarray)

  • roll_deg (ndarray)

  • pitch_deg (ndarray)

Return type:

IntentDecoderPipeline

project_continuous(features, roll_deg, pitch_deg)[source]

Project one or more samples onto the normalized continuous control axis.

Parameters:
  • features (ndarray)

  • roll_deg (ndarray)

  • pitch_deg (ndarray)

Return type:

dict[str, ndarray]

predict(feature, orientation)[source]
Parameters:
Return type:

DecoderDecision

class OrientationSample(roll_deg: 'float | None' = None, pitch_deg: 'float | None' = None, gyro_norm: 'float | None' = None, accel_norm: 'float | None' = None)[source]
Parameters:
  • roll_deg (float | None)

  • pitch_deg (float | None)

  • gyro_norm (float | None)

  • accel_norm (float | None)

roll_deg: float | None = None
pitch_deg: float | None = None
gyro_norm: float | None = None
accel_norm: float | None = None
property is_available: bool
class PairEvaluation(open_label: 'str', close_label: 'str', balanced_accuracy: 'float', balanced_accuracy_std: 'float', rest_false_activation_rate: 'float', reject_false_activation_rate: 'float', direction_confusion_rate: 'float', composite_score: 'float', folds: 'int')[source]
Parameters:
  • open_label (str)

  • close_label (str)

  • balanced_accuracy (float)

  • balanced_accuracy_std (float)

  • rest_false_activation_rate (float)

  • reject_false_activation_rate (float)

  • direction_confusion_rate (float)

  • composite_score (float)

  • folds (int)

open_label: str
close_label: str
balanced_accuracy: float
balanced_accuracy_std: float
rest_false_activation_rate: float
reject_false_activation_rate: float
direction_confusion_rate: float
composite_score: float
folds: int
class PreprocessConfig(sample_rate_hz: 'float', highpass_hz: 'float' = 20.0, lowpass_hz: 'float' = 200.0, notch_hz: 'float' = 60.0, notch_quality: 'float' = 30.0)[source]
Parameters:
  • sample_rate_hz (float)

  • highpass_hz (float)

  • lowpass_hz (float)

  • notch_hz (float)

  • notch_quality (float)

sample_rate_hz: float
highpass_hz: float = 20.0
lowpass_hz: float = 200.0
notch_hz: float = 60.0
notch_quality: float = 30.0
class ShrinkageLDAIntentModel(scaler: 'StandardScaler | None' = None, classifier: 'LinearDiscriminantAnalysis | None' = None)[source]
Parameters:
  • scaler (StandardScaler | None)

  • classifier (LinearDiscriminantAnalysis | None)

scaler: StandardScaler | None = None
classifier: LinearDiscriminantAnalysis | None = None
fit(features, labels)[source]
Parameters:
  • features (ndarray)

  • labels (ndarray)

Return type:

ShrinkageLDAIntentModel

property classes: tuple[str, ...]
predict(features)[source]
Parameters:

features (ndarray)

Return type:

ndarray

predict_proba(features)[source]
Parameters:

features (ndarray)

Return type:

ndarray

class StreamLayout(emg_channels: 'tuple[int, ...]', accel_channels: 'tuple[int, ...]' = (), gyro_channels: 'tuple[int, ...]' = ())[source]
Parameters:
  • emg_channels (tuple[int, ...])

  • accel_channels (tuple[int, ...])

  • gyro_channels (tuple[int, ...])

emg_channels: tuple[int, ...]
accel_channels: tuple[int, ...] = ()
gyro_channels: tuple[int, ...] = ()
classmethod from_specs(channel_count, emg, accel='', gyro='')[source]
Parameters:
  • channel_count (int)

  • emg (str)

  • accel (str)

  • gyro (str)

Return type:

StreamLayout

select_emg(data)[source]
Parameters:

data (ndarray)

Return type:

ndarray

mean_accel(data)[source]
Parameters:

data (ndarray)

Return type:

ndarray | None

mean_gyro(data)[source]
Parameters:

data (ndarray)

Return type:

ndarray | None

class SignalQuality(channel_rms: 'np.ndarray', flat_channels: 'tuple[int, ...]', saturated_channels: 'tuple[int, ...]', noisy_channels: 'tuple[int, ...]')[source]
Parameters:
  • channel_rms (ndarray)

  • flat_channels (tuple[int, ...])

  • saturated_channels (tuple[int, ...])

  • noisy_channels (tuple[int, ...])

channel_rms: ndarray
flat_channels: tuple[int, ...]
saturated_channels: tuple[int, ...]
noisy_channels: tuple[int, ...]
property usable_fraction: float
assess_signal_quality(emg)[source]
Parameters:

emg (ndarray)

Return type:

SignalQuality

canonical_intent_label(value)[source]
Parameters:

value (str)

Return type:

str

extract_emg_features(emg, config=None)[source]

Return a compact feature vector for 8-channel or HD-EMG windows.

Parameters:
Return type:

ndarray

import_xdf_file(session, path, *, window_s=0.25, step_s=0.12, start_offset_s=0.5, end_offset_s=0.25)[source]
Parameters:
  • session (IntentCaptureSession)

  • path (str | Path)

  • window_s (float)

  • step_s (float)

  • start_offset_s (float)

  • end_offset_s (float)

Return type:

tuple[int, int]

import_xdf_session(paths, *, participant_id='', progress=None)[source]
Parameters:
  • paths (Iterable[str | Path])

  • participant_id (str)

  • progress (Callable[[int, int, Path], None] | None)

Return type:

tuple[IntentCaptureSession, dict[str, object]]

orientation_from_accel(accel_xyz, gyro_xyz=None)[source]
Parameters:
  • accel_xyz (ndarray)

  • gyro_xyz (ndarray | None)

Return type:

OrientationSample

parse_channel_spec(specification, channel_count)[source]

Parse comma-separated indices and inclusive ranges such as 1-8,12.

Parameters:
  • specification (str)

  • channel_count (int)

Return type:

tuple[int, …]

preprocess_emg(emg, config)[source]

Band-pass and notch a channels-by-samples EMG window.

Filters are omitted individually when the stream rate cannot represent their requested frequency. This keeps low-rate envelope streams usable while applying the full path to conventional raw EMG.

Parameters:
Return type:

ndarray

rank_intent_pairs(features, labels, groups, roll_deg, pitch_deg, rest_label='rest', reject_label='reject', folds=5, use_orientation=True)[source]
Parameters:
  • features (ndarray)

  • labels (ndarray)

  • groups (ndarray)

  • roll_deg (ndarray)

  • pitch_deg (ndarray)

  • rest_label (str)

  • reject_label (str)

  • folds (int)

  • use_orientation (bool)

Return type:

list[PairEvaluation]

class IntentOutputStabilizer(ema_alpha=0.25, max_step=0.18, enter_threshold=0.08, open_enter_threshold=None, close_enter_threshold=None, release_threshold=0.04, switch_samples=3, output_gain=1.0, response_exponent=1.0, value=0.0, direction=0, pending_direction=0, pending_count=0)[source]

Smooth continuous intent and reject single-window direction reversals.

Parameters:
  • ema_alpha (float)

  • max_step (float)

  • enter_threshold (float)

  • open_enter_threshold (float | None)

  • close_enter_threshold (float | None)

  • release_threshold (float)

  • switch_samples (int)

  • output_gain (float)

  • response_exponent (float)

  • value (float)

  • direction (int)

  • pending_direction (int)

  • pending_count (int)

ema_alpha: float = 0.25
max_step: float = 0.18
enter_threshold: float = 0.08
open_enter_threshold: float | None = None
close_enter_threshold: float | None = None
release_threshold: float = 0.04
switch_samples: int = 3
output_gain: float = 1.0
response_exponent: float = 1.0
value: float = 0.0
direction: int = 0
pending_direction: int = 0
pending_count: int = 0
reset()[source]
Return type:

None

update(decision, *, open_label, close_label, rest_label='rest')[source]
Parameters:
  • decision (DecoderDecision)

  • open_label (str)

  • close_label (str)

  • rest_label (str)

Return type:

DecoderDecision