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 DualHandExo(left, right)[source]
Bases:
objectUnified controller for a left and/or right HandExo.
Either
leftorrightmay beNoneif that side is not connected — the class will skip that side gracefully.- Parameters:
- 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)
- 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 fromconfig.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’ssidefield; if absent, defaults to'right'.
- Raises:
RuntimeError – If the target exo is not connected.
- class DualSerialComm(cmd_port, telem_port, baudrate, command_delimiter=';', timeout=1, response_timeout=2.0, verbose=False, line_terminator='\r\n')[source]
Bases:
BaseCommTwo 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 theset_reply_routecommand).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 toreply_route:telemso 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.
- 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
- 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:telemand are drained from_telemby the background reader. The current firmware writes the compactNXframe 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:
objectFake 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.
- class GestureController(exo, verbose=False)[source]
Bases:
objectA 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.
restneeds 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
- class HandExo(comm, name='NMLHandExo', command_delimiter='\n', send_delay=0.01, auto_connect=False, verbose=False, side=None)[source]
Bases:
objectClass 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.sideand 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.
- 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
NXframe 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
- 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_mAis 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_mAis 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) andmotorskeyed by DXL ID withnominal_mA/applied_mA/state.appliedbelownominalmeans 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,restandflex.restneeds firmware >= 0.3.0;radexisted only between 0.3.1 and 0.6.0, where its motor was folded intowrist. 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:
percentinterpolates the gesture between its own endpoints, so0-> exactlyset_gesture(gesture, 'extend')50-> halfway between the extend and flex postures100-> exactlyset_gesture(gesture, 'flex')
Each motor the gesture names travels its own share, so a gesture driving several motors – the thumb, or the coupled
wristpair – keeps the ratio between them at every percentage. RetuningEXTEND_*/FLEX_*inconfig.hmoves the axis with them, so a host’s percentages keep meaning the same postures across a retune.Note that
0is the extend posture, not home: with a non-zeroEXTEND_*a hand parked at home sits below 0% andget_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 (
radexists only in 0.3.1 - 0.5.x).ValueError – If
percentis 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,extendreads back as 0 andflexas 100.restreads back wherever itsREST_*constants place it between the two.Values are integers so the out-of-range signals share the encoding:
0to100– position between the extend and flex posturesGESTURE_ANGLE_BELOW_RANGE(101) – past the extend endGESTURE_ANGLE_ABOVE_RANGE(102) – past the flex endGESTURE_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
wristpair, 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
reststate is 0 degrees; positions towardflexare positive and positions towardextendare negative. Multi-motor gesture percentages are still aggregated as documented byget_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.
Nonemeans 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()andget_gesture_sang(), produced from one batched motor read. Each result hasfraction(the legacy 0-100/status code) andangle_delta_deg(rest-zeroed signed degrees, orNone).- Parameters:
gesture (str) – A single angle-addressable gesture, or
"all".timeout (float) – Seconds to wait for the reply.
- Returns:
Gesture names mapped to
fractionandangle_delta_degfields.- 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_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
- 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
- class LSLClient(stream_name=None, stream_type=None, maxlen=10000, auto_start=True, verbose=False)[source]
Bases:
object
- class LSLMarkerSubscriber(stream_name=None, stream_type='Markers', timeout=5.0, buf_events=2000, recover=True, verbose=False)[source]
Bases:
objectSubscribe 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
- set_callback(func, *args, only_on_change=False, **kwargs)[source]
Register a callback invoked as
func(text, timestamp, *args, **kwargs).only_on_change=Truesuppresses repeated marker text. Positional and keyword arguments supplied here are forwarded to the callback.Pass
Noneto 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
- 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:
objectPublish 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)
- class LSLNumericSubscriber(stream_name=None, stream_type=None, timeout=5.0, buf_seconds=10.0, max_chunklen=1024, recover=True, verbose=False)[source]
Bases:
objectSubscribe to a continuous numeric stream (e.g., EMG/EEG).
Use
get_latest_window()to retrieve a channel-by-samplefloat32array 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)
- exception ProtocolResponseError(*, command, expected, raw_response)[source]
Bases:
RuntimeErrorA 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- 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
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)
- 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
- 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
- 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
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:
- 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
- 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:
- 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:
feature (ndarray)
orientation (OrientationSample)
- Return type:
- 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
- property classes: tuple[str, ...]
- 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, ...] = ()
- 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
- extract_emg_features(emg, config=None)[source]
Return a compact feature vector for 8-channel or HD-EMG windows.
- Parameters:
emg (ndarray)
config (FeatureConfig | None)
- 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:
- 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:
emg (ndarray)
config (PreprocessConfig)
- 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
- update(decision, *, open_label, close_label, rest_label='rest')[source]
- Parameters:
decision (DecoderDecision)
open_label (str)
close_label (str)
rest_label (str)
- Return type: