From 9e7f90577fe3fae46a29aaa4ebaf416b0efaed40 Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Wed, 3 Jun 2026 11:39:17 +0200 Subject: [PATCH 01/11] change structure to implement switches --- pyproject.toml | 2 +- src/exosuit_python/__main__.py | 23 +-- src/exosuit_python/definitions.py | 29 +++- src/exosuit_python/exosuit.py | 270 +++++++++++++++++++++--------- uv.lock | 8 +- 5 files changed, 237 insertions(+), 95 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index a0a1ced..9b126f0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ dependencies = [ "numpy>=2.2.3", "loguru>=0.7.3", "motor-python>=0.0.4", - "imu-python>=0.0.12", + "imu-python>=0.1.1", "hip-controller>=0.0.4", ] diff --git a/src/exosuit_python/__main__.py b/src/exosuit_python/__main__.py index f02b4a4..047ddc1 100644 --- a/src/exosuit_python/__main__.py +++ b/src/exosuit_python/__main__.py @@ -7,7 +7,7 @@ from exosuit_python.utils import setup_logger -def main(log_level: str, stderr_level: str) -> None: # pragma: no cover +def main(log_level: str, stderr_level: str, mock: bool) -> None: # pragma: no cover """Run the main pipeline. :param log_level: The log level to use. @@ -16,14 +16,10 @@ def main(log_level: str, stderr_level: str) -> None: # pragma: no cover """ setup_logger(log_level=log_level, stderr_level=stderr_level) - config = ExosuitConfig(frequency=100) - exosuit = Exosuit(config=config) - exosuit.turn_on_exosuit_switch() - try: - while exosuit._is_running: - pass - except KeyboardInterrupt: - exosuit._cleanup() + config = ExosuitConfig(frequency=100, mock=mock) + Exosuit(config=config) + + # Current implementation uses main thread for exosuit state handler loop if __name__ == "__main__": # pragma: no cover @@ -44,6 +40,13 @@ def main(log_level: str, stderr_level: str) -> None: # pragma: no cover required=False, type=str, ) + parser.add_argument( + "--mock", + default=False, + help="Use mock devices.", + required=False, + type=bool, + ) args = parser.parse_args() - main(log_level=args.log_level, stderr_level=args.stderr_level) + main(log_level=args.log_level, stderr_level=args.stderr_level, mock=args.mock) diff --git a/src/exosuit_python/definitions.py b/src/exosuit_python/definitions.py index 091f5a4..1ee4ebd 100644 --- a/src/exosuit_python/definitions.py +++ b/src/exosuit_python/definitions.py @@ -1,9 +1,10 @@ """Common definitions for this module.""" -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, field from pathlib import Path import numpy as np +from imu_python.definitions import IMUDescriptor np.set_printoptions(precision=3, floatmode="fixed", suppress=True) @@ -44,12 +45,36 @@ def __iter__(self): THREAD_JOIN_TIMEOUT = 2.0 +SWITCH_EVENT_HANDLER_INTERVAL = 0.5 +EXOSUIT_STANDBY_INTERVAL = 0.1 + @dataclass(frozen=True) -class ConfigTension: +class TensionConfig: """Configurations for tensioning.""" tensioning_velocity: int = 3 motor_torque_limit: float = 0.85 # tensioned when motor_torque >= 0.85 tensioning_timeout: float = 1.0 # in sec + + +@dataclass(frozen=True) +class IMUConfig: + """IMU configuration dataclass containing busID, IMU name and index for each leg.""" + + left_leg_bus: int = 7 + left_leg_descr: IMUDescriptor = field( + default_factory=lambda: IMUDescriptor(name="BNO055", index=0) + ) + right_leg_bus: int = 7 + right_leg_descr: IMUDescriptor = field( + default_factory=lambda: IMUDescriptor(name="BNO055", index=1) + ) + + +# Switch pins +POWER_SWITCH = 7 # TODO: placeholder +TENSION_SWITCH = 15 # TODO: placeholder + +GPIO_SWITCH_BOUNCETIME = 50 # TODO: test this threshold diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index 11eeb42..c90a3f2 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -3,16 +3,26 @@ import threading import time from dataclasses import dataclass +from enum import IntEnum from hip_controller.control.app import WalkOnController from hip_controller.definitions import SensorSignal -from imu_python.definitions import I2CBusID from imu_python.factory import IMUFactory from imu_python.sensor_manager import IMUManager +from Jetson import GPIO from loguru import logger from motor_python.cube_mars_motor import CubeMarsAK606v3 -from exosuit_python.definitions import THREAD_JOIN_TIMEOUT, ConfigTension +from exosuit_python.definitions import ( + EXOSUIT_STANDBY_INTERVAL, + GPIO_SWITCH_BOUNCETIME, + POWER_SWITCH, + SWITCH_EVENT_HANDLER_INTERVAL, + TENSION_SWITCH, + THREAD_JOIN_TIMEOUT, + IMUConfig, + TensionConfig, +) from exosuit_python.utils import convert_rad_per_sec_to_rpm @@ -21,6 +31,17 @@ class ExosuitConfig: """Exosuit configuration.""" frequency: float + mock: bool = False + + +class ExosuitStates(IntEnum): + """Enum for exosuit states.""" + + INITIALIZING = 0 + STANDBY = 1 + PRETENSIONING = 2 + RUNNING = 3 + STOPPED = 4 class Exosuit: @@ -31,14 +52,14 @@ def __init__(self, config: ExosuitConfig) -> None: :param config: Exosuit configuration """ - self.config = config + self.config: ExosuitConfig = config - self._is_running: bool = False - self._is_tensioning: bool = False + self._status: ExosuitStates = ExosuitStates.INITIALIZING + self._power_switch: bool = False + self._tension_switch: bool = False self.thread: threading.Thread = threading.Thread(target=self._loop, daemon=True) - self.imu_hip: IMUManager self.imu_left: IMUManager self.imu_right: IMUManager @@ -51,45 +72,106 @@ def __init__(self, config: ExosuitConfig) -> None: self.imu_initialized: bool = self._initialize_imus() self.motors_initialized: bool = self._initialize_motors() - def turn_on_exosuit_switch(self) -> None: - """Turn on the exosuit switch to start the exosuit.""" - self._is_running = True - - """Start the soft exoskeleton.""" - if not self.imu_initialized: - logger.error("IMU initialization failed. Exosuit not started.") - return - if not self.motors_initialized: - logger.error("Motor initialization failed. Exosuit not started.") - return + self._gpio_setup() self._start() + self._switch_event_handler() + + def _gpio_setup(self) -> None: + """Set up Jetson GPIO switches.""" + GPIO.setmode(GPIO.BOARD) + GPIO.setup(POWER_SWITCH, GPIO.IN, pull_up_down=GPIO.PUD_UP) + GPIO.setup(TENSION_SWITCH, GPIO.IN, pull_up_down=GPIO.PUD_UP) + + GPIO.add_event_detect( + POWER_SWITCH, + GPIO.FALLING, # TODO: wiring + callback=self._power_on_callback, + bouncetime=GPIO_SWITCH_BOUNCETIME, + ) + + GPIO.add_event_detect( + POWER_SWITCH, + GPIO.RISING, # TODO: wiring + callback=self._power_off_callback, + bouncetime=GPIO_SWITCH_BOUNCETIME, + ) + + GPIO.add_event_detect( + TENSION_SWITCH, + GPIO.FALLING, # TODO: wiring + callback=self._tension_on_callback, + bouncetime=GPIO_SWITCH_BOUNCETIME, + ) + + GPIO.add_event_detect( + TENSION_SWITCH, + GPIO.RISING, # TODO: wiring + callback=self._tension_off_callback, + bouncetime=GPIO_SWITCH_BOUNCETIME, + ) + + def _switch_event_handler(self) -> None: + """Monitor switch states and handle exosuit status changes.""" + try: + while True: + if self._status == ExosuitStates.STANDBY: + if self._power_switch and not self._tension_switch: + logger.info("State change: standby -> running") + self._status = ExosuitStates.RUNNING + elif self._tension_switch and not self._power_switch: + logger.info("State change: standby -> pretensioning") + self._status = ExosuitStates.PRETENSIONING + self._pretension() + elif self._status == ExosuitStates.RUNNING: + if not self._power_switch: + logger.info("State change: running -> standby") + self._status = ExosuitStates.STANDBY + elif self._status == ExosuitStates.PRETENSIONING: + if not self._tension_switch: + logger.info("State change: pretensioning -> standby") + self._status = ExosuitStates.STANDBY + elif self._status == ExosuitStates.STOPPED: + break + time.sleep(SWITCH_EVENT_HANDLER_INTERVAL) + finally: + self._cleanup() - def turn_off_exosuit_switch(self) -> None: - """Turn off the exosuit switch to stop the exosuit.""" - self._is_running = False - self._cleanup() + def _pretension(self) -> None: + """Pretension the tendons by turning the motors.""" + self.motor_left.set_velocity(TensionConfig.tensioning_velocity) + # TODO: right motor + + def _power_on_callback(self, channel) -> None: + self._power_switch = True + + def _power_off_callback(self, channel) -> None: + self._power_switch = False - def turn_on_tension_switch(self) -> None: - """Start tensioning process of the exosuit.""" - self._is_tensioning = True - self.motor_left.set_velocity(ConfigTension.tensioning_velocity) + def _tension_on_callback(self, channel) -> None: + self._tension_switch = True - # TODO get motor_torque from motor and loop to see if >= 0.85 the velocity should be zero. After one second the process should be stopped + def _tension_off_callback(self, channel) -> None: + self._tension_switch = False def _start(self) -> None: """Start the IMUs and Motors.""" - self._is_running = True + if not self.imu_initialized: + logger.error("IMU initialization failed. Exosuit not started.") + return + if not self.motors_initialized: + logger.error("Motor initialization failed. Exosuit not started.") + return logger.info(f"Starting Exosuit at '{self.config.frequency}' Hz.") try: logger.debug("Starting IMUs") - self.imu_hip.start() self.imu_left.start() self.imu_right.start() logger.debug("Starting Motors") logger.debug("Starting Controller") + self._status = ExosuitStates.STANDBY # Start main control loop self.thread.start() @@ -100,55 +182,70 @@ def _start(self) -> None: def _cleanup(self) -> None: """Clean up the soft exoskeleton.""" logger.info("Cleaning up exosuit.") - self.imu_hip.stop() + self._status = ExosuitStates.STOPPED + GPIO.cleanup() self.imu_left.stop() self.imu_right.stop() self.motor_left.close() if self.thread is not None and self.thread.is_alive(): self.thread.join(timeout=THREAD_JOIN_TIMEOUT) - self._is_running = False logger.success("Exosuit shutdown.") def _loop(self) -> None: """Run main control loop.""" - while self._is_running and not self._is_tensioning: - try: - data_right = self.imu_right.get_data() - data_left = self.imu_left.get_data() - - if data_right is None or data_left is None: - raise TypeError - - timestamp_right = data_right.timestamp - signal_right = SensorSignal( - angle_rad=data_right.quat.to_euler(seq="xyz").z, - velocity_rad_per_sec=data_right.device_data.gyro.z, - ) - command_right = self.controller_right.step( - timestamp=timestamp_right, curr_signal=signal_right - ) - - timestamp_left = data_left.timestamp - signal_left = SensorSignal( - angle_rad=data_left.quat.to_euler(seq="xyz").z, - velocity_rad_per_sec=data_left.device_data.gyro.z, - ) - command_left = self.controller_left.step( - timestamp=timestamp_left, curr_signal=signal_left - ) - - self.motor_left.set_velocity(convert_rad_per_sec_to_rpm(command_left)) - - # place holder - self.motor_right.append( - f"Motor command at timestamp {timestamp_right} is {convert_rad_per_sec_to_rpm(command_right)} ERPM.)" - ) - - time.sleep(1 / self.config.frequency) - except TypeError as err: - logger.error(f"Failed getting data from the IMU: '{err}'.") - except Exception as err: - logger.error(f"Exosuit control loop exception: '{err}'.") + while True: + while self._status == ExosuitStates.RUNNING: + try: + data_right = self.imu_right.get_data() + data_left = self.imu_left.get_data() + + if data_right is None or data_left is None: + raise TypeError + + timestamp_right = data_right.timestamp + signal_right = SensorSignal( + angle_rad=data_right.quat.to_euler(seq="xyz").z, + velocity_rad_per_sec=data_right.device_data.gyro.z, + ) + command_right = self.controller_right.step( + timestamp=timestamp_right, curr_signal=signal_right + ) + + timestamp_left = data_left.timestamp + signal_left = SensorSignal( + angle_rad=data_left.quat.to_euler(seq="xyz").z, + velocity_rad_per_sec=data_left.device_data.gyro.z, + ) + command_left = self.controller_left.step( + timestamp=timestamp_left, curr_signal=signal_left + ) + + self.motor_left.set_velocity( + convert_rad_per_sec_to_rpm(command_left) + ) + + # place holder + self.motor_right.append( + f"Motor command at timestamp {timestamp_right} is {convert_rad_per_sec_to_rpm(command_right)} ERPM.)" + ) + + time.sleep(1 / self.config.frequency) + except TypeError as err: + logger.error(f"Failed getting data from the IMU: '{err}'.") + except Exception as err: + logger.error(f"Exosuit control loop exception: '{err}'.") + + while self._status == ExosuitStates.PRETENSIONING: + try: + # TODO right motor + left_motor_torque = 0.85 # place holder, get real torque here + if left_motor_torque >= TensionConfig.motor_torque_limit: + self.motor_left.set_velocity(0) + time.sleep(TensionConfig.tensioning_timeout) + except Exception as err: + logger.error(f"Exosuit control loop exception: '{err}'.") + + time.sleep(EXOSUIT_STANDBY_INTERVAL) def _initialize_imus(self) -> bool: """Initialize IMUs. @@ -156,18 +253,35 @@ def _initialize_imus(self) -> bool: :return: True if successful, False otherwise """ try: - # TODO changes - sensor_managers_hip = IMUFactory.detect_and_create( - i2c_id=I2CBusID.bus_1, - log_data=False, - ) - sensor_managers_legs = IMUFactory.detect_and_create( - i2c_id=I2CBusID.bus_7, + left_init: bool = False + right_init: bool = False + sensor_managers = IMUFactory.detect_and_create( + free_threading=True, log_data=False, ) - self.imu_hip = sensor_managers_hip[0] - self.imu_left = sensor_managers_legs[0] - self.imu_right = sensor_managers_legs[1] + detected_imus = len(sensor_managers) + if detected_imus != 2: + raise ValueError(f"Wrong number of IMUs detected: {detected_imus} != 2") + for idx in range(2): # Match each leg with the IMU according to IMUConfig + manager = sensor_managers[idx] + if ( + not left_init + and manager.i2c_id == IMUConfig.left_leg_bus + and manager.imu_descriptor == IMUConfig.left_leg_descr + ): + self.imu_left = manager + left_init = True + continue + if ( + manager.i2c_id == IMUConfig.right_leg_bus + and manager.imu_descriptor == IMUConfig.right_leg_descr + ): + self.imu_right = manager + right_init = True + if not left_init: + raise RuntimeError("No detected IMUs matches left leg config") + if not right_init: + raise RuntimeError("No detected IMUs matches right leg config") return True except Exception as err: logger.error(f"Exosuit exception: '{err}'. Check IMU connections.") diff --git a/uv.lock b/uv.lock index 0ed818c..91ad473 100644 --- a/uv.lock +++ b/uv.lock @@ -570,7 +570,7 @@ hw = [ [package.metadata] requires-dist = [ { name = "hip-controller", specifier = ">=0.0.4" }, - { name = "imu-python", specifier = ">=0.0.12" }, + { name = "imu-python", specifier = ">=0.1.1" }, { name = "loguru", specifier = ">=0.7.3" }, { name = "motor-python", specifier = ">=0.0.4" }, { name = "numpy", specifier = ">=2.2.3" }, @@ -689,7 +689,7 @@ wheels = [ [[package]] name = "imu-python" -version = "0.1.0" +version = "0.1.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "ahrs" }, @@ -701,9 +701,9 @@ dependencies = [ { name = "py-imu" }, { name = "scipy" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/8f/d5/43da3175a9c92da28fcf30e4da6eae81a7c189f526809d4c07ba86bea01f/imu_python-0.1.0.tar.gz", hash = "sha256:1f95381e3c0d44470614911d279e3ef7294495b52e1df894d88bd08eded37477", size = 30447, upload-time = "2026-03-18T10:24:27.387Z" } +sdist = { url = "https://files.pythonhosted.org/packages/90/06/d414bcaefd6a9dfa9954d2692df3c897df0aa684a72a5fdf7e36c781f86c/imu_python-0.1.1.tar.gz", hash = "sha256:e48e641a8420b3718aa3dc39e870b93a2984acc434e7c52c3642ddbe0dc7cfcf", size = 32569, upload-time = "2026-04-21T09:22:23.323Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/34/62/664b7daec8672713dd39df57bb56be16f613027f0941e7c79b9dfc212cfd/imu_python-0.1.0-py3-none-any.whl", hash = "sha256:ad10a34122445b3f8e70fe601edefa7dbcc3d9beb2a5573468cb773b5b4ccf26", size = 36544, upload-time = "2026-03-18T10:24:25.79Z" }, + { url = "https://files.pythonhosted.org/packages/38/62/e64a4204f6abb091a415ff5b95aef2bfffa47e48686addeb365ae62686e8/imu_python-0.1.1-py3-none-any.whl", hash = "sha256:3bca7535f6c888313c550fb329714949acd145980bde9ba33470a8aec35ea167", size = 39263, upload-time = "2026-04-21T09:22:21.983Z" }, ] [[package]] From 551154374b4c317f830807cb8fd4100f5ae383dc Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Fri, 5 Jun 2026 15:30:23 +0200 Subject: [PATCH 02/11] add mock for GPIO and motor --- Makefile | 2 +- src/exosuit_python/__main__.py | 47 ++++++-- src/exosuit_python/definitions.py | 12 +- src/exosuit_python/exosuit.py | 188 +++++++++++++++++------------- src/exosuit_python/gpio.py | 86 ++++++++++++++ src/exosuit_python/motor.py | 25 ++++ 6 files changed, 271 insertions(+), 89 deletions(-) create mode 100644 src/exosuit_python/gpio.py create mode 100644 src/exosuit_python/motor.py diff --git a/Makefile b/Makefile index da74968..5617b89 100644 --- a/Makefile +++ b/Makefile @@ -7,7 +7,7 @@ init: # ENV SETUP test: uv run pytest --cov=src --cov-report=term-missing --no-cov-on-fail --cov-report=xml --cov-fail-under=30 - rm .coverage + rm -f .coverage lint: uv run ruff format src/ tests/ diff --git a/src/exosuit_python/__main__.py b/src/exosuit_python/__main__.py index 047ddc1..8e94f30 100644 --- a/src/exosuit_python/__main__.py +++ b/src/exosuit_python/__main__.py @@ -1,9 +1,18 @@ """Sample doc string.""" import argparse +import time -from exosuit_python.definitions import DEFAULT_LOG_LEVEL, LogLevel -from exosuit_python.exosuit import Exosuit, ExosuitConfig +from loguru import logger + +from exosuit_python.definitions import ( + DEFAULT_LOG_LEVEL, + POWER_SWITCH, + TENSION_SWITCH, + LogLevel, +) +from exosuit_python.exosuit import GPIO, Exosuit, ExosuitConfig +from exosuit_python.gpio import MockGPIO from exosuit_python.utils import setup_logger @@ -17,9 +26,35 @@ def main(log_level: str, stderr_level: str, mock: bool) -> None: # pragma: no c setup_logger(log_level=log_level, stderr_level=stderr_level) config = ExosuitConfig(frequency=100, mock=mock) - Exosuit(config=config) + exosuit = Exosuit(config=config) + try: + while True: + test_pipeline(exosuit) + except KeyboardInterrupt: + exosuit._cleanup() + - # Current implementation uses main thread for exosuit state handler loop +def test_pipeline(exosuit: Exosuit) -> None: + """Run the test pipeline.""" + if isinstance(GPIO, MockGPIO): + # wait for initialization + time.sleep(5) + # activate pre-tensioning + logger.info("Simulating tensioning switch ON...") + GPIO.simulate_switch(TENSION_SWITCH, exosuit.on_signal) + time.sleep(1) + # deactivate pre-tensioning + logger.info("Simulating tensioning switch OFF...") + GPIO.simulate_switch(TENSION_SWITCH, exosuit.off_signal) + time.sleep(2) + # power on + logger.info("Simulating power switch ON...") + GPIO.simulate_switch(POWER_SWITCH, exosuit.on_signal) + time.sleep(10) + # power off + logger.info("Simulating power switch OFF...") + GPIO.simulate_switch(POWER_SWITCH, exosuit.off_signal) + time.sleep(4) if __name__ == "__main__": # pragma: no cover @@ -42,10 +77,8 @@ def main(log_level: str, stderr_level: str, mock: bool) -> None: # pragma: no c ) parser.add_argument( "--mock", - default=False, help="Use mock devices.", - required=False, - type=bool, + action="store_true", ) args = parser.parse_args() diff --git a/src/exosuit_python/definitions.py b/src/exosuit_python/definitions.py index 1ee4ebd..f0826a3 100644 --- a/src/exosuit_python/definitions.py +++ b/src/exosuit_python/definitions.py @@ -1,5 +1,7 @@ """Common definitions for this module.""" +from __future__ import annotations + from dataclasses import asdict, dataclass, field from pathlib import Path @@ -51,7 +53,7 @@ def __iter__(self): @dataclass(frozen=True) class TensionConfig: - """Configurations for tensioning.""" + """Configurations for tensioning.""" # TODO: unit tensioning_velocity: int = 3 motor_torque_limit: float = 0.85 # tensioned when motor_torque >= 0.85 @@ -65,11 +67,11 @@ class IMUConfig: left_leg_bus: int = 7 left_leg_descr: IMUDescriptor = field( - default_factory=lambda: IMUDescriptor(name="BNO055", index=0) + default_factory=lambda: IMUDescriptor(name="MOCK", index=0) ) right_leg_bus: int = 7 right_leg_descr: IMUDescriptor = field( - default_factory=lambda: IMUDescriptor(name="BNO055", index=1) + default_factory=lambda: IMUDescriptor(name="MOCK", index=1) ) @@ -78,3 +80,7 @@ class IMUConfig: TENSION_SWITCH = 15 # TODO: placeholder GPIO_SWITCH_BOUNCETIME = 50 # TODO: test this threshold + +# switch signals +SWITCH_ON = "FALLING" +SWITCH_OFF = "RISING" diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index c90a3f2..e0b9fd1 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -2,15 +2,22 @@ import threading import time -from dataclasses import dataclass +from dataclasses import dataclass, field from enum import IntEnum +from loguru import logger + +try: + from Jetson import GPIO +except Exception: + logger.warning("Jetson GPIO import failed. Are you running on the Jetson?") + from exosuit_python.gpio import MockGPIO + + GPIO = MockGPIO() from hip_controller.control.app import WalkOnController from hip_controller.definitions import SensorSignal from imu_python.factory import IMUFactory from imu_python.sensor_manager import IMUManager -from Jetson import GPIO -from loguru import logger from motor_python.cube_mars_motor import CubeMarsAK606v3 from exosuit_python.definitions import ( @@ -18,11 +25,14 @@ GPIO_SWITCH_BOUNCETIME, POWER_SWITCH, SWITCH_EVENT_HANDLER_INTERVAL, + SWITCH_OFF, + SWITCH_ON, TENSION_SWITCH, THREAD_JOIN_TIMEOUT, IMUConfig, TensionConfig, ) +from exosuit_python.motor import MockMotor from exosuit_python.utils import convert_rad_per_sec_to_rpm @@ -32,6 +42,7 @@ class ExosuitConfig: frequency: float mock: bool = False + imu_cfg: IMUConfig = field(default_factory=IMUConfig) class ExosuitStates(IntEnum): @@ -59,6 +70,9 @@ def __init__(self, config: ExosuitConfig) -> None: self._tension_switch: bool = False self.thread: threading.Thread = threading.Thread(target=self._loop, daemon=True) + self.switch_thread: threading.Thread = threading.Thread( + target=self._switch_event_handler, daemon=True + ) self.imu_left: IMUManager self.imu_right: IMUManager @@ -66,75 +80,95 @@ def __init__(self, config: ExosuitConfig) -> None: self.controller_left = WalkOnController(reverse=False) self.controller_right = WalkOnController(reverse=True) - self.motor_left: CubeMarsAK606v3 = CubeMarsAK606v3() - self.motor_right = ["Motor right."] # place holder + self.motor_left: CubeMarsAK606v3 | MockMotor + self.motor_right: CubeMarsAK606v3 | MockMotor - self.imu_initialized: bool = self._initialize_imus() - self.motors_initialized: bool = self._initialize_motors() + if not self.config.mock: + self.motor_left = CubeMarsAK606v3() + self.motor_right = CubeMarsAK606v3() + else: + self.motor_left = MockMotor() + self.motor_right = MockMotor() + if not self._initialize_imus(): + logger.error("IMU initialization failed. Exosuit not started.") + return + if not self._initialize_motors(): + logger.error("Motor initialization failed. Exosuit not started.") + return + if hasattr(GPIO, SWITCH_ON) and hasattr(GPIO, SWITCH_OFF): + self.on_signal = getattr(GPIO, SWITCH_ON) + self.off_signal = getattr(GPIO, SWITCH_OFF) + else: + logger.error( + f"GPIO signal attribute '{SWITCH_ON}' or '{SWITCH_OFF}' not found." + ) + return self._gpio_setup() self._start() - self._switch_event_handler() - def _gpio_setup(self) -> None: + def _gpio_setup(self) -> bool: """Set up Jetson GPIO switches.""" - GPIO.setmode(GPIO.BOARD) - GPIO.setup(POWER_SWITCH, GPIO.IN, pull_up_down=GPIO.PUD_UP) - GPIO.setup(TENSION_SWITCH, GPIO.IN, pull_up_down=GPIO.PUD_UP) - - GPIO.add_event_detect( - POWER_SWITCH, - GPIO.FALLING, # TODO: wiring - callback=self._power_on_callback, - bouncetime=GPIO_SWITCH_BOUNCETIME, - ) + try: + GPIO.setmode(GPIO.BOARD) + GPIO.setup(POWER_SWITCH, GPIO.IN, pull_up_down=GPIO.PUD_UP) + GPIO.setup(TENSION_SWITCH, GPIO.IN, pull_up_down=GPIO.PUD_UP) + + GPIO.add_event_detect( + POWER_SWITCH, + self.on_signal, + callback=self._power_on_callback, + bouncetime=GPIO_SWITCH_BOUNCETIME, + ) - GPIO.add_event_detect( - POWER_SWITCH, - GPIO.RISING, # TODO: wiring - callback=self._power_off_callback, - bouncetime=GPIO_SWITCH_BOUNCETIME, - ) + GPIO.add_event_detect( + POWER_SWITCH, + self.off_signal, + callback=self._power_off_callback, + bouncetime=GPIO_SWITCH_BOUNCETIME, + ) - GPIO.add_event_detect( - TENSION_SWITCH, - GPIO.FALLING, # TODO: wiring - callback=self._tension_on_callback, - bouncetime=GPIO_SWITCH_BOUNCETIME, - ) + GPIO.add_event_detect( + TENSION_SWITCH, + self.on_signal, + callback=self._tension_on_callback, + bouncetime=GPIO_SWITCH_BOUNCETIME, + ) - GPIO.add_event_detect( - TENSION_SWITCH, - GPIO.RISING, # TODO: wiring - callback=self._tension_off_callback, - bouncetime=GPIO_SWITCH_BOUNCETIME, - ) + GPIO.add_event_detect( + TENSION_SWITCH, + self.off_signal, + callback=self._tension_off_callback, + bouncetime=GPIO_SWITCH_BOUNCETIME, + ) + return True + except Exception as err: + logger.error(f"Jetson GPIO init failure: {err}") + GPIO.cleanup() + return False - def _switch_event_handler(self) -> None: + def _switch_event_handler( + self, + ) -> None: # TODO: implement proper state machine if needed """Monitor switch states and handle exosuit status changes.""" - try: - while True: - if self._status == ExosuitStates.STANDBY: - if self._power_switch and not self._tension_switch: - logger.info("State change: standby -> running") - self._status = ExosuitStates.RUNNING - elif self._tension_switch and not self._power_switch: - logger.info("State change: standby -> pretensioning") - self._status = ExosuitStates.PRETENSIONING - self._pretension() - elif self._status == ExosuitStates.RUNNING: - if not self._power_switch: - logger.info("State change: running -> standby") - self._status = ExosuitStates.STANDBY - elif self._status == ExosuitStates.PRETENSIONING: - if not self._tension_switch: - logger.info("State change: pretensioning -> standby") - self._status = ExosuitStates.STANDBY - elif self._status == ExosuitStates.STOPPED: - break - time.sleep(SWITCH_EVENT_HANDLER_INTERVAL) - finally: - self._cleanup() + while self._status != ExosuitStates.STOPPED: + if self._status == ExosuitStates.STANDBY: + if self._power_switch and not self._tension_switch: + logger.info("State change: standby -> running") + self._status = ExosuitStates.RUNNING + elif self._tension_switch and not self._power_switch: + logger.info("State change: standby -> pretensioning") + self._status = ExosuitStates.PRETENSIONING + self._pretension() + elif self._status == ExosuitStates.RUNNING: + if not self._power_switch: + logger.info("State change: running -> standby") + self._status = ExosuitStates.STANDBY + elif self._status == ExosuitStates.PRETENSIONING: + if not self._tension_switch: + logger.info("State change: pretensioning -> standby") + self._status = ExosuitStates.STANDBY + time.sleep(SWITCH_EVENT_HANDLER_INTERVAL) def _pretension(self) -> None: """Pretension the tendons by turning the motors.""" @@ -155,13 +189,6 @@ def _tension_off_callback(self, channel) -> None: def _start(self) -> None: """Start the IMUs and Motors.""" - if not self.imu_initialized: - logger.error("IMU initialization failed. Exosuit not started.") - return - if not self.motors_initialized: - logger.error("Motor initialization failed. Exosuit not started.") - return - logger.info(f"Starting Exosuit at '{self.config.frequency}' Hz.") try: logger.debug("Starting IMUs") @@ -172,8 +199,10 @@ def _start(self) -> None: logger.debug("Starting Controller") self._status = ExosuitStates.STANDBY + logger.info("Exosuit status: standby") # Start main control loop self.thread.start() + self.switch_thread.start() except Exception as err: logger.info(f"Exosuit exception: '{err}'.") @@ -189,6 +218,8 @@ def _cleanup(self) -> None: self.motor_left.close() if self.thread is not None and self.thread.is_alive(): self.thread.join(timeout=THREAD_JOIN_TIMEOUT) + if self.switch_thread is not None and self.switch_thread.is_alive(): + self.switch_thread.join(timeout=THREAD_JOIN_TIMEOUT) logger.success("Exosuit shutdown.") def _loop(self) -> None: @@ -224,9 +255,8 @@ def _loop(self) -> None: convert_rad_per_sec_to_rpm(command_left) ) - # place holder - self.motor_right.append( - f"Motor command at timestamp {timestamp_right} is {convert_rad_per_sec_to_rpm(command_right)} ERPM.)" + self.motor_left.set_velocity( + convert_rad_per_sec_to_rpm(command_right) ) time.sleep(1 / self.config.frequency) @@ -260,21 +290,23 @@ def _initialize_imus(self) -> bool: log_data=False, ) detected_imus = len(sensor_managers) - if detected_imus != 2: - raise ValueError(f"Wrong number of IMUs detected: {detected_imus} != 2") - for idx in range(2): # Match each leg with the IMU according to IMUConfig + if detected_imus < 2: + raise ValueError(f"Wrong number of IMUs detected: {detected_imus} < 2") + for idx in range( + detected_imus + ): # Match each leg with the IMU according to IMUConfig manager = sensor_managers[idx] if ( not left_init - and manager.i2c_id == IMUConfig.left_leg_bus - and manager.imu_descriptor == IMUConfig.left_leg_descr + and manager.i2c_id == self.config.imu_cfg.left_leg_bus + and manager.imu_descriptor == self.config.imu_cfg.left_leg_descr ): self.imu_left = manager left_init = True continue if ( - manager.i2c_id == IMUConfig.right_leg_bus - and manager.imu_descriptor == IMUConfig.right_leg_descr + manager.i2c_id == self.config.imu_cfg.right_leg_bus + and manager.imu_descriptor == self.config.imu_cfg.right_leg_descr ): self.imu_right = manager right_init = True diff --git a/src/exosuit_python/gpio.py b/src/exosuit_python/gpio.py new file mode 100644 index 0000000..bd41496 --- /dev/null +++ b/src/exosuit_python/gpio.py @@ -0,0 +1,86 @@ +"""Mock GPIO for CI testing.""" + +import threading +import time +from collections.abc import Callable + +from exosuit_python.definitions import THREAD_JOIN_TIMEOUT + + +class MockGPIO: + """Mock GPIO for CI testing.""" + + OUT = 0 + IN = 1 + RISING = 31 + FALLING = 32 + BOARD = 10 + PUD_UP = 22 + + def __init__(self) -> None: + self.gpio_running: bool = False + self._threads: list[threading.Thread] = [] + self._detected_events: dict[int, int | None] = {} + + def setmode(self, mode: int) -> None: + """Set mode.""" + pass + + def setup(self, channels: int, direction: int, pull_up_down: int = 0) -> None: + """Set up GPIO.""" + self.gpio_running = True + + def add_event_detect( + self, + channel: int, + edge: int, + callback: Callable | None = None, + bouncetime: int | None = None, + polltime: float = 0.2, + ) -> None: + """Add detection for the given channel's event and call the callback function.""" + thread = threading.Thread( + target=self._loop_event_detect_thread, + kwargs={ + "channel": channel, + "edge": edge, + "callback": callback, + "bouncetime": bouncetime, + "polltime": polltime, + }, + daemon=True, + ) + self._threads.append(thread) + thread.start() + + def cleanup(self): + """Clean up GPIO.""" + self.gpio_running = False + for thread in self._threads: + thread.join(timeout=THREAD_JOIN_TIMEOUT) + + def simulate_switch(self, channel: int, event: int) -> None: + """Simulate switches for event detection.""" + self._detected_events[channel] = event + + def _loop_event_detect_thread( + self, + channel: int, + edge: int, + callback: Callable | None = None, + bouncetime: int | None = None, + polltime: float = 0.2, + ) -> None: + while self.gpio_running: + # Check if an event was simulated for this channel + if ( + channel in self._detected_events + and self._detected_events[channel] == edge + ): + # Call the callback + if callback: + callback(channel) + # Clear the event to prevent duplicate triggers + self._detected_events[channel] = None + + time.sleep(polltime) diff --git a/src/exosuit_python/motor.py b/src/exosuit_python/motor.py new file mode 100644 index 0000000..f651692 --- /dev/null +++ b/src/exosuit_python/motor.py @@ -0,0 +1,25 @@ +"""Mock motor for CI testing.""" + + +class MockMotor: + """Mock motor class.""" + + def __init__(self) -> None: + pass + + def set_velocity(self, velocity_erpm: int) -> None: + """Set mock motor velocity.""" + pass + + def close(self) -> None: + """Close mock motor connection.""" + pass + + def check_communication(self) -> bool: + """Check if motor can communicate.""" + return True + + def get_torque(self) -> int: + # TODO: make this function name/param identical to the motor implementation + """Get the torque of the motor.""" + return 0 From 00bcc877252d1785101a8adea0ea3b1f79eb1651 Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Mon, 8 Jun 2026 12:17:02 +0200 Subject: [PATCH 03/11] add exosuit tests --- Makefile | 2 +- src/exosuit_python/__main__.py | 14 +++---- src/exosuit_python/definitions.py | 12 +++--- src/exosuit_python/exosuit.py | 54 +++++++++++++++------------ tests/exosuit_test.py | 61 +++++++++++++++++++++++++++++++ 5 files changed, 107 insertions(+), 36 deletions(-) diff --git a/Makefile b/Makefile index 5617b89..593ff9a 100644 --- a/Makefile +++ b/Makefile @@ -6,7 +6,7 @@ init: # ENV SETUP @echo "Environment initialized with uv." test: - uv run pytest --cov=src --cov-report=term-missing --no-cov-on-fail --cov-report=xml --cov-fail-under=30 + uv run pytest --cov=src --cov-report=term-missing --no-cov-on-fail --cov-report=xml --cov-fail-under=80 rm -f .coverage lint: diff --git a/src/exosuit_python/__main__.py b/src/exosuit_python/__main__.py index 8e94f30..419633a 100644 --- a/src/exosuit_python/__main__.py +++ b/src/exosuit_python/__main__.py @@ -11,7 +11,7 @@ TENSION_SWITCH, LogLevel, ) -from exosuit_python.exosuit import GPIO, Exosuit, ExosuitConfig +from exosuit_python.exosuit import Exosuit, ExosuitConfig from exosuit_python.gpio import MockGPIO from exosuit_python.utils import setup_logger @@ -36,24 +36,24 @@ def main(log_level: str, stderr_level: str, mock: bool) -> None: # pragma: no c def test_pipeline(exosuit: Exosuit) -> None: """Run the test pipeline.""" - if isinstance(GPIO, MockGPIO): + if isinstance(exosuit.gpio, MockGPIO): # wait for initialization - time.sleep(5) + time.sleep(2) # activate pre-tensioning logger.info("Simulating tensioning switch ON...") - GPIO.simulate_switch(TENSION_SWITCH, exosuit.on_signal) + exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.on_signal) time.sleep(1) # deactivate pre-tensioning logger.info("Simulating tensioning switch OFF...") - GPIO.simulate_switch(TENSION_SWITCH, exosuit.off_signal) + exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.off_signal) time.sleep(2) # power on logger.info("Simulating power switch ON...") - GPIO.simulate_switch(POWER_SWITCH, exosuit.on_signal) + exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.on_signal) time.sleep(10) # power off logger.info("Simulating power switch OFF...") - GPIO.simulate_switch(POWER_SWITCH, exosuit.off_signal) + exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.off_signal) time.sleep(4) diff --git a/src/exosuit_python/definitions.py b/src/exosuit_python/definitions.py index f0826a3..c8b32b4 100644 --- a/src/exosuit_python/definitions.py +++ b/src/exosuit_python/definitions.py @@ -6,7 +6,7 @@ from pathlib import Path import numpy as np -from imu_python.definitions import IMUDescriptor +from imu_python.definitions import I2CBusID, IMUDescriptor np.set_printoptions(precision=3, floatmode="fixed", suppress=True) @@ -45,6 +45,8 @@ def __iter__(self): DEFAULT_LOG_LEVEL = LogLevel.info DEFAULT_LOG_FILENAME = "log_file" +DEFAULT_EXOSUIT_FREQUENCY_HZ = 100 + THREAD_JOIN_TIMEOUT = 2.0 SWITCH_EVENT_HANDLER_INTERVAL = 0.5 @@ -65,14 +67,14 @@ class TensionConfig: class IMUConfig: """IMU configuration dataclass containing busID, IMU name and index for each leg.""" - left_leg_bus: int = 7 + left_leg_bus: int = I2CBusID.bus_7 left_leg_descr: IMUDescriptor = field( default_factory=lambda: IMUDescriptor(name="MOCK", index=0) ) - right_leg_bus: int = 7 + right_leg_bus: int = I2CBusID.bus_7 right_leg_descr: IMUDescriptor = field( default_factory=lambda: IMUDescriptor(name="MOCK", index=1) - ) + ) # TODO: set actual IMUs # Switch pins @@ -82,5 +84,5 @@ class IMUConfig: GPIO_SWITCH_BOUNCETIME = 50 # TODO: test this threshold # switch signals -SWITCH_ON = "FALLING" +SWITCH_ON = "FALLING" # TODO: wiring SWITCH_OFF = "RISING" diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index e0b9fd1..2f054c1 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -11,9 +11,7 @@ from Jetson import GPIO except Exception: logger.warning("Jetson GPIO import failed. Are you running on the Jetson?") - from exosuit_python.gpio import MockGPIO - - GPIO = MockGPIO() + GPIO = None from hip_controller.control.app import WalkOnController from hip_controller.definitions import SensorSignal from imu_python.factory import IMUFactory @@ -32,6 +30,7 @@ IMUConfig, TensionConfig, ) +from exosuit_python.gpio import MockGPIO from exosuit_python.motor import MockMotor from exosuit_python.utils import convert_rad_per_sec_to_rpm @@ -65,6 +64,13 @@ def __init__(self, config: ExosuitConfig) -> None: """ self.config: ExosuitConfig = config + if self.config.mock: + self.gpio = MockGPIO() + elif GPIO is None: + self.gpio = MockGPIO() + else: + self.gpio = GPIO + self._status: ExosuitStates = ExosuitStates.INITIALIZING self._power_switch: bool = False self._tension_switch: bool = False @@ -83,12 +89,12 @@ def __init__(self, config: ExosuitConfig) -> None: self.motor_left: CubeMarsAK606v3 | MockMotor self.motor_right: CubeMarsAK606v3 | MockMotor - if not self.config.mock: - self.motor_left = CubeMarsAK606v3() - self.motor_right = CubeMarsAK606v3() - else: + if self.config.mock: self.motor_left = MockMotor() self.motor_right = MockMotor() + else: + self.motor_left = CubeMarsAK606v3() + self.motor_right = CubeMarsAK606v3() if not self._initialize_imus(): logger.error("IMU initialization failed. Exosuit not started.") @@ -96,46 +102,48 @@ def __init__(self, config: ExosuitConfig) -> None: if not self._initialize_motors(): logger.error("Motor initialization failed. Exosuit not started.") return - if hasattr(GPIO, SWITCH_ON) and hasattr(GPIO, SWITCH_OFF): - self.on_signal = getattr(GPIO, SWITCH_ON) - self.off_signal = getattr(GPIO, SWITCH_OFF) + if hasattr(self.gpio, SWITCH_ON) and hasattr(self.gpio, SWITCH_OFF): + self.on_signal = getattr(self.gpio, SWITCH_ON) + self.off_signal = getattr(self.gpio, SWITCH_OFF) else: logger.error( f"GPIO signal attribute '{SWITCH_ON}' or '{SWITCH_OFF}' not found." ) return - self._gpio_setup() - self._start() + if self._gpio_initialized(): + self._start() + else: + logger.error("GPIO initialization failed. Exosuit not started.") - def _gpio_setup(self) -> bool: + def _gpio_initialized(self) -> bool: """Set up Jetson GPIO switches.""" try: - GPIO.setmode(GPIO.BOARD) - GPIO.setup(POWER_SWITCH, GPIO.IN, pull_up_down=GPIO.PUD_UP) - GPIO.setup(TENSION_SWITCH, GPIO.IN, pull_up_down=GPIO.PUD_UP) + self.gpio.setmode(self.gpio.BOARD) + self.gpio.setup(POWER_SWITCH, self.gpio.IN, pull_up_down=self.gpio.PUD_UP) + self.gpio.setup(TENSION_SWITCH, self.gpio.IN, pull_up_down=self.gpio.PUD_UP) - GPIO.add_event_detect( + self.gpio.add_event_detect( POWER_SWITCH, self.on_signal, callback=self._power_on_callback, bouncetime=GPIO_SWITCH_BOUNCETIME, ) - GPIO.add_event_detect( + self.gpio.add_event_detect( POWER_SWITCH, self.off_signal, callback=self._power_off_callback, bouncetime=GPIO_SWITCH_BOUNCETIME, ) - GPIO.add_event_detect( + self.gpio.add_event_detect( TENSION_SWITCH, self.on_signal, callback=self._tension_on_callback, bouncetime=GPIO_SWITCH_BOUNCETIME, ) - GPIO.add_event_detect( + self.gpio.add_event_detect( TENSION_SWITCH, self.off_signal, callback=self._tension_off_callback, @@ -143,8 +151,8 @@ def _gpio_setup(self) -> bool: ) return True except Exception as err: - logger.error(f"Jetson GPIO init failure: {err}") - GPIO.cleanup() + logger.error(f"GPIO init failure: {err}") + self.gpio.cleanup() return False def _switch_event_handler( @@ -212,7 +220,7 @@ def _cleanup(self) -> None: """Clean up the soft exoskeleton.""" logger.info("Cleaning up exosuit.") self._status = ExosuitStates.STOPPED - GPIO.cleanup() + self.gpio.cleanup() self.imu_left.stop() self.imu_right.stop() self.motor_left.close() diff --git a/tests/exosuit_test.py b/tests/exosuit_test.py index cc22dcf..ef74496 100644 --- a/tests/exosuit_test.py +++ b/tests/exosuit_test.py @@ -1 +1,62 @@ """Test the main program.""" + +import time + +from exosuit_python.definitions import ( + DEFAULT_EXOSUIT_FREQUENCY_HZ, + EXOSUIT_STANDBY_INTERVAL, + POWER_SWITCH, + SWITCH_EVENT_HANDLER_INTERVAL, + TENSION_SWITCH, + IMUConfig, +) +from exosuit_python.exosuit import Exosuit, ExosuitConfig, ExosuitStates +from exosuit_python.gpio import MockGPIO + + +def test_exosuit_initialization(): + """Test if exosuit is initialized with the set config.""" + imu_config = IMUConfig() + exosuit_config = ExosuitConfig( + frequency=DEFAULT_EXOSUIT_FREQUENCY_HZ, mock=True, imu_cfg=imu_config + ) + exosuit = Exosuit(exosuit_config) + + assert exosuit.config == exosuit_config + exosuit._cleanup() + + +def test_exosuit_switches(): + """Test if exosuit's state changes correctly upon switch triggers.""" + imu_config = IMUConfig() + exosuit_config = ExosuitConfig( + frequency=DEFAULT_EXOSUIT_FREQUENCY_HZ, mock=True, imu_cfg=imu_config + ) + exosuit = Exosuit(exosuit_config) + + # wait for initialization + time.sleep(1) + assert isinstance(exosuit.gpio, MockGPIO) + assert exosuit._status == ExosuitStates.STANDBY + # time for the event handler to register changes + state_wait = EXOSUIT_STANDBY_INTERVAL + SWITCH_EVENT_HANDLER_INTERVAL + 0.1 + # simulate tension switch ON + exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.on_signal) + time.sleep(state_wait) + assert exosuit._status == ExosuitStates.PRETENSIONING + # simulate tension switch OFF + exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.off_signal) + time.sleep(state_wait) + assert exosuit._status == ExosuitStates.STANDBY + # simulate power switch ON + exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.on_signal) + time.sleep(state_wait) + assert exosuit._status == ExosuitStates.RUNNING + # simulate power switch OFF + exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.off_signal) + time.sleep(state_wait) + assert exosuit._status == ExosuitStates.STANDBY + + # stop exosuit + exosuit._cleanup() + assert exosuit._status == ExosuitStates.STOPPED From 37f11100a6ceeaaca0cba972f7ade97a234ac62a Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Mon, 8 Jun 2026 13:12:43 +0200 Subject: [PATCH 04/11] add docstring --- src/exosuit_python/__main__.py | 2 +- src/exosuit_python/exosuit.py | 74 ++++++++++++++++++++++++---------- src/exosuit_python/motor.py | 3 ++ 3 files changed, 56 insertions(+), 23 deletions(-) diff --git a/src/exosuit_python/__main__.py b/src/exosuit_python/__main__.py index 419633a..5703464 100644 --- a/src/exosuit_python/__main__.py +++ b/src/exosuit_python/__main__.py @@ -42,7 +42,7 @@ def test_pipeline(exosuit: Exosuit) -> None: # activate pre-tensioning logger.info("Simulating tensioning switch ON...") exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.on_signal) - time.sleep(1) + time.sleep(2) # deactivate pre-tensioning logger.info("Simulating tensioning switch OFF...") exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.off_signal) diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index 2f054c1..ac56bed 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -37,7 +37,14 @@ @dataclass class ExosuitConfig: - """Exosuit configuration.""" + """Exosuit configuration. + + Attributes: + frequency: Exosuit frequency in Hz. + mock: flag to use mock devices. + imu_cfg: IMU config that defines the IMU to use for each leg. + + """ frequency: float mock: bool = False @@ -63,18 +70,27 @@ def __init__(self, config: ExosuitConfig) -> None: :param config: Exosuit configuration """ self.config: ExosuitConfig = config + self._status: ExosuitStates = ExosuitStates.INITIALIZING - if self.config.mock: - self.gpio = MockGPIO() - elif GPIO is None: + # GPIO for switches + if self.config.mock or GPIO is None: self.gpio = MockGPIO() else: self.gpio = GPIO - self._status: ExosuitStates = ExosuitStates.INITIALIZING + if hasattr(self.gpio, SWITCH_ON) and hasattr(self.gpio, SWITCH_OFF): + self.on_signal = getattr(self.gpio, SWITCH_ON) + self.off_signal = getattr(self.gpio, SWITCH_OFF) + else: + logger.error( + f"GPIO signal attribute '{SWITCH_ON}' or '{SWITCH_OFF}' not found." + ) + return + self._power_switch: bool = False self._tension_switch: bool = False + # main loop thread and switch handler thread self.thread: threading.Thread = threading.Thread(target=self._loop, daemon=True) self.switch_thread: threading.Thread = threading.Thread( target=self._switch_event_handler, daemon=True @@ -96,27 +112,24 @@ def __init__(self, config: ExosuitConfig) -> None: self.motor_left = CubeMarsAK606v3() self.motor_right = CubeMarsAK606v3() + # initialization calls if not self._initialize_imus(): logger.error("IMU initialization failed. Exosuit not started.") return if not self._initialize_motors(): logger.error("Motor initialization failed. Exosuit not started.") return - if hasattr(self.gpio, SWITCH_ON) and hasattr(self.gpio, SWITCH_OFF): - self.on_signal = getattr(self.gpio, SWITCH_ON) - self.off_signal = getattr(self.gpio, SWITCH_OFF) - else: - logger.error( - f"GPIO signal attribute '{SWITCH_ON}' or '{SWITCH_OFF}' not found." - ) - return - if self._gpio_initialized(): + + if self._initialize_gpio(): self._start() else: logger.error("GPIO initialization failed. Exosuit not started.") - def _gpio_initialized(self) -> bool: - """Set up Jetson GPIO switches.""" + def _initialize_gpio(self) -> bool: + """Initialize and set up Jetson GPIO switches. + + :return: True if successful, False otherwise + """ try: self.gpio.setmode(self.gpio.BOARD) self.gpio.setup(POWER_SWITCH, self.gpio.IN, pull_up_down=self.gpio.PUD_UP) @@ -158,7 +171,10 @@ def _gpio_initialized(self) -> bool: def _switch_event_handler( self, ) -> None: # TODO: implement proper state machine if needed - """Monitor switch states and handle exosuit status changes.""" + """Monitor switch states and handle exosuit status changes. + + :return: None + """ while self._status != ExosuitStates.STOPPED: if self._status == ExosuitStates.STANDBY: if self._power_switch and not self._tension_switch: @@ -179,7 +195,10 @@ def _switch_event_handler( time.sleep(SWITCH_EVENT_HANDLER_INTERVAL) def _pretension(self) -> None: - """Pretension the tendons by turning the motors.""" + """Pretension the tendons by turning the motors. + + :return: None + """ self.motor_left.set_velocity(TensionConfig.tensioning_velocity) # TODO: right motor @@ -196,7 +215,10 @@ def _tension_off_callback(self, channel) -> None: self._tension_switch = False def _start(self) -> None: - """Start the IMUs and Motors.""" + """Start the IMUs and Motors. + + :return: None + """ logger.info(f"Starting Exosuit at '{self.config.frequency}' Hz.") try: logger.debug("Starting IMUs") @@ -217,7 +239,10 @@ def _start(self) -> None: self._cleanup() def _cleanup(self) -> None: - """Clean up the soft exoskeleton.""" + """Clean up the soft exoskeleton. + + :return: None + """ logger.info("Cleaning up exosuit.") self._status = ExosuitStates.STOPPED self.gpio.cleanup() @@ -231,7 +256,10 @@ def _cleanup(self) -> None: logger.success("Exosuit shutdown.") def _loop(self) -> None: - """Run main control loop.""" + """Run main control loop. + + :return: None + """ while True: while self._status == ExosuitStates.RUNNING: try: @@ -276,7 +304,9 @@ def _loop(self) -> None: while self._status == ExosuitStates.PRETENSIONING: try: # TODO right motor - left_motor_torque = 0.85 # place holder, get real torque here + left_motor_torque = ( + 0.85 # TODO place holder, get actual torque here + ) if left_motor_torque >= TensionConfig.motor_torque_limit: self.motor_left.set_velocity(0) time.sleep(TensionConfig.tensioning_timeout) diff --git a/src/exosuit_python/motor.py b/src/exosuit_python/motor.py index f651692..30c5a01 100644 --- a/src/exosuit_python/motor.py +++ b/src/exosuit_python/motor.py @@ -1,5 +1,7 @@ """Mock motor for CI testing.""" +from loguru import logger + class MockMotor: """Mock motor class.""" @@ -9,6 +11,7 @@ def __init__(self) -> None: def set_velocity(self, velocity_erpm: int) -> None: """Set mock motor velocity.""" + logger.debug(f"Velocity set: {velocity_erpm}") pass def close(self) -> None: From 8a9ff535c7d7b53b089db59c9ec059ad07f24433 Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Tue, 9 Jun 2026 15:21:53 +0200 Subject: [PATCH 05/11] bug fixes based on Claude feedback --- src/exosuit_python/__main__.py | 4 +- src/exosuit_python/definitions.py | 7 +- src/exosuit_python/exosuit.py | 190 ++++++++++++++++-------------- src/exosuit_python/gpio.py | 53 ++++++--- src/exosuit_python/motor.py | 9 +- tests/exosuit_test.py | 3 +- 6 files changed, 156 insertions(+), 110 deletions(-) diff --git a/src/exosuit_python/__main__.py b/src/exosuit_python/__main__.py index 5703464..99333b9 100644 --- a/src/exosuit_python/__main__.py +++ b/src/exosuit_python/__main__.py @@ -42,11 +42,11 @@ def test_pipeline(exosuit: Exosuit) -> None: # activate pre-tensioning logger.info("Simulating tensioning switch ON...") exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.on_signal) - time.sleep(2) + time.sleep(1) # deactivate pre-tensioning logger.info("Simulating tensioning switch OFF...") exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.off_signal) - time.sleep(2) + time.sleep(3) # power on logger.info("Simulating power switch ON...") exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.on_signal) diff --git a/src/exosuit_python/definitions.py b/src/exosuit_python/definitions.py index c8b32b4..af7ae7d 100644 --- a/src/exosuit_python/definitions.py +++ b/src/exosuit_python/definitions.py @@ -59,8 +59,8 @@ class TensionConfig: tensioning_velocity: int = 3 motor_torque_limit: float = 0.85 # tensioned when motor_torque >= 0.85 - tensioning_timeout: float = 1.0 # in sec + torque_check_interval: float = 0.1 # in sec @dataclass(frozen=True) @@ -84,5 +84,6 @@ class IMUConfig: GPIO_SWITCH_BOUNCETIME = 50 # TODO: test this threshold # switch signals -SWITCH_ON = "FALLING" # TODO: wiring -SWITCH_OFF = "RISING" +SWITCH_ON = "HIGH" # TODO: wiring +SWITCH_OFF = "LOW" +BOTH = "BOTH" diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index ac56bed..c3219f6 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -19,6 +19,7 @@ from motor_python.cube_mars_motor import CubeMarsAK606v3 from exosuit_python.definitions import ( + BOTH, EXOSUIT_STANDBY_INTERVAL, GPIO_SWITCH_BOUNCETIME, POWER_SWITCH, @@ -78,12 +79,17 @@ def __init__(self, config: ExosuitConfig) -> None: else: self.gpio = GPIO - if hasattr(self.gpio, SWITCH_ON) and hasattr(self.gpio, SWITCH_OFF): + if ( + hasattr(self.gpio, SWITCH_ON) + and hasattr(self.gpio, SWITCH_OFF) + and hasattr(self.gpio, BOTH) + ): self.on_signal = getattr(self.gpio, SWITCH_ON) self.off_signal = getattr(self.gpio, SWITCH_OFF) + self.both_signal = getattr(self.gpio, BOTH) else: logger.error( - f"GPIO signal attribute '{SWITCH_ON}' or '{SWITCH_OFF}' not found." + f"GPIO signal attribute '{SWITCH_ON}', '{SWITCH_OFF}', or '{BOTH}' not found." ) return @@ -137,31 +143,18 @@ def _initialize_gpio(self) -> bool: self.gpio.add_event_detect( POWER_SWITCH, - self.on_signal, - callback=self._power_on_callback, - bouncetime=GPIO_SWITCH_BOUNCETIME, - ) - - self.gpio.add_event_detect( - POWER_SWITCH, - self.off_signal, - callback=self._power_off_callback, + self.both_signal, + callback=self._power_callback, bouncetime=GPIO_SWITCH_BOUNCETIME, ) self.gpio.add_event_detect( TENSION_SWITCH, - self.on_signal, - callback=self._tension_on_callback, + self.both_signal, + callback=self._tension_callback, bouncetime=GPIO_SWITCH_BOUNCETIME, ) - self.gpio.add_event_detect( - TENSION_SWITCH, - self.off_signal, - callback=self._tension_off_callback, - bouncetime=GPIO_SWITCH_BOUNCETIME, - ) return True except Exception as err: logger.error(f"GPIO init failure: {err}") @@ -183,36 +176,33 @@ def _switch_event_handler( elif self._tension_switch and not self._power_switch: logger.info("State change: standby -> pretensioning") self._status = ExosuitStates.PRETENSIONING - self._pretension() elif self._status == ExosuitStates.RUNNING: if not self._power_switch: logger.info("State change: running -> standby") self._status = ExosuitStates.STANDBY elif self._status == ExosuitStates.PRETENSIONING: - if not self._tension_switch: - logger.info("State change: pretensioning -> standby") - self._status = ExosuitStates.STANDBY + pass # state change handled in _loop() time.sleep(SWITCH_EVENT_HANDLER_INTERVAL) - def _pretension(self) -> None: - """Pretension the tendons by turning the motors. - - :return: None - """ - self.motor_left.set_velocity(TensionConfig.tensioning_velocity) - # TODO: right motor - - def _power_on_callback(self, channel) -> None: - self._power_switch = True - - def _power_off_callback(self, channel) -> None: - self._power_switch = False - - def _tension_on_callback(self, channel) -> None: - self._tension_switch = True - - def _tension_off_callback(self, channel) -> None: - self._tension_switch = False + def _power_callback(self, channel) -> None: + """Handle power switch states triggered by signal events.""" + power_state = self.gpio.input(POWER_SWITCH) + if power_state == self.on_signal: + self._power_switch = True + elif power_state == self.off_signal: + self._power_switch = False + else: + logger.warning(f"Unrecognized power switch state: {power_state}") + + def _tension_callback(self, channel) -> None: + """Handle tension switch states triggered by signal events.""" + tension_state = self.gpio.input(TENSION_SWITCH) + if tension_state == self.on_signal: + self._tension_switch = True + elif tension_state == self.off_signal: + self._tension_switch = False + else: + logger.warning(f"Unrecognized tension switch state: {tension_state}") def _start(self) -> None: """Start the IMUs and Motors. @@ -249,6 +239,7 @@ def _cleanup(self) -> None: self.imu_left.stop() self.imu_right.stop() self.motor_left.close() + self.motor_right.close() if self.thread is not None and self.thread.is_alive(): self.thread.join(timeout=THREAD_JOIN_TIMEOUT) if self.switch_thread is not None and self.switch_thread.is_alive(): @@ -260,61 +251,83 @@ def _loop(self) -> None: :return: None """ - while True: + while self._status != ExosuitStates.STOPPED: while self._status == ExosuitStates.RUNNING: try: - data_right = self.imu_right.get_data() - data_left = self.imu_left.get_data() - - if data_right is None or data_left is None: - raise TypeError - - timestamp_right = data_right.timestamp - signal_right = SensorSignal( - angle_rad=data_right.quat.to_euler(seq="xyz").z, - velocity_rad_per_sec=data_right.device_data.gyro.z, - ) - command_right = self.controller_right.step( - timestamp=timestamp_right, curr_signal=signal_right - ) - - timestamp_left = data_left.timestamp - signal_left = SensorSignal( - angle_rad=data_left.quat.to_euler(seq="xyz").z, - velocity_rad_per_sec=data_left.device_data.gyro.z, - ) - command_left = self.controller_left.step( - timestamp=timestamp_left, curr_signal=signal_left - ) - - self.motor_left.set_velocity( - convert_rad_per_sec_to_rpm(command_left) - ) - - self.motor_left.set_velocity( - convert_rad_per_sec_to_rpm(command_right) - ) - - time.sleep(1 / self.config.frequency) + self._control() except TypeError as err: logger.error(f"Failed getting data from the IMU: '{err}'.") except Exception as err: logger.error(f"Exosuit control loop exception: '{err}'.") + time.sleep(1 / self.config.frequency) + while self._status == ExosuitStates.PRETENSIONING: try: - # TODO right motor - left_motor_torque = ( - 0.85 # TODO place holder, get actual torque here - ) - if left_motor_torque >= TensionConfig.motor_torque_limit: - self.motor_left.set_velocity(0) - time.sleep(TensionConfig.tensioning_timeout) + self._pretension() except Exception as err: logger.error(f"Exosuit control loop exception: '{err}'.") + time.sleep(TensionConfig.torque_check_interval) + time.sleep(EXOSUIT_STANDBY_INTERVAL) + def _control(self) -> None: + """Execute one iteration of control loop.""" + data_right = self.imu_right.get_data() + data_left = self.imu_left.get_data() + + if data_right is None or data_left is None: + raise TypeError + + timestamp_right = data_right.timestamp + signal_right = SensorSignal( + angle_rad=data_right.quat.to_euler(seq="xyz").z, + velocity_rad_per_sec=data_right.device_data.gyro.z, + ) + command_right = self.controller_right.step( + timestamp=timestamp_right, curr_signal=signal_right + ) + + timestamp_left = data_left.timestamp + signal_left = SensorSignal( + angle_rad=data_left.quat.to_euler(seq="xyz").z, + velocity_rad_per_sec=data_left.device_data.gyro.z, + ) + command_left = self.controller_left.step( + timestamp=timestamp_left, curr_signal=signal_left + ) + + self.motor_left.set_velocity(convert_rad_per_sec_to_rpm(command_left)) + + self.motor_right.set_velocity(convert_rad_per_sec_to_rpm(command_right)) + + def _pretension(self) -> None: + """Execute one iteration of pretensioning loop.""" + left_motor_torque = 0.85 # TODO place holder, get actual torque here + right_motor_torque = 0.85 # TODO place holder, get actual torque here + + # Only apply velocity if motor hasn't reached threshold + if left_motor_torque < TensionConfig.motor_torque_limit: + self.motor_left.set_velocity(TensionConfig.tensioning_velocity) + else: + self.motor_left.set_velocity(0) + + if right_motor_torque < TensionConfig.motor_torque_limit: + self.motor_right.set_velocity(TensionConfig.tensioning_velocity) + else: + self.motor_right.set_velocity(0) + + # Exit pretensioning when both motors reach threshold and switch is released + if ( + left_motor_torque >= TensionConfig.motor_torque_limit + and right_motor_torque >= TensionConfig.motor_torque_limit + and not self._tension_switch + ): + time.sleep(TensionConfig.tensioning_timeout) + logger.info("State change: pretensioning -> standby") + self._status = ExosuitStates.STANDBY + def _initialize_imus(self) -> bool: """Initialize IMUs. @@ -362,12 +375,15 @@ def _initialize_motors(self) -> bool: :return: True if successful, False otherwise """ - # TODO: right motor try: + communication_status = True if not self.motor_left.check_communication(): - logger.error("Motor not responding. Check power and connections.") - return False - return True + logger.error("Left Motor not responding. Check power and connections.") + communication_status = False + if not self.motor_right.check_communication(): + logger.error("Right Motor not responding. Check power and connections.") + communication_status = False + return communication_status except Exception as err: logger.error(f"Exosuit exception: '{err}'. Check Motor connections.") return False diff --git a/src/exosuit_python/gpio.py b/src/exosuit_python/gpio.py index bd41496..5adbcb7 100644 --- a/src/exosuit_python/gpio.py +++ b/src/exosuit_python/gpio.py @@ -10,17 +10,23 @@ class MockGPIO: """Mock GPIO for CI testing.""" + LOW = 0 + HIGH = 1 OUT = 0 IN = 1 RISING = 31 FALLING = 32 + BOTH = 33 BOARD = 10 PUD_UP = 22 def __init__(self) -> None: self.gpio_running: bool = False self._threads: list[threading.Thread] = [] - self._detected_events: dict[int, int | None] = {} + self._channel_states: dict[int, int] = {} # Track current state (0=LOW, 1=HIGH) + self._previous_states: dict[ + int, int | None + ] = {} # Track previous state for edge detection def setmode(self, mode: int) -> None: """Set mode.""" @@ -29,6 +35,14 @@ def setmode(self, mode: int) -> None: def setup(self, channels: int, direction: int, pull_up_down: int = 0) -> None: """Set up GPIO.""" self.gpio_running = True + # Initialize channel state(s) when set up + if isinstance(channels, int): + self._channel_states[channels] = 0 + self._previous_states[channels] = None + else: + for ch in channels: + self._channel_states[ch] = 0 + self._previous_states[ch] = None def add_event_detect( self, @@ -59,9 +73,9 @@ def cleanup(self): for thread in self._threads: thread.join(timeout=THREAD_JOIN_TIMEOUT) - def simulate_switch(self, channel: int, event: int) -> None: - """Simulate switches for event detection.""" - self._detected_events[channel] = event + def simulate_switch(self, channel: int, state: int) -> None: + """Simulate a switch state change (HIGH or LOW) to trigger edge detection.""" + self._channel_states[channel] = state def _loop_event_detect_thread( self, @@ -72,15 +86,26 @@ def _loop_event_detect_thread( polltime: float = 0.2, ) -> None: while self.gpio_running: - # Check if an event was simulated for this channel - if ( - channel in self._detected_events - and self._detected_events[channel] == edge - ): - # Call the callback - if callback: - callback(channel) - # Clear the event to prevent duplicate triggers - self._detected_events[channel] = None + current_state = self._channel_states.get(channel, 0) + previous_state = self._previous_states.get(channel) + # Detect edge transitions if we have a previous state + if previous_state is not None: + # RISING edge: LOW → HIGH + if previous_state == self.LOW and current_state == self.HIGH: + if edge in {self.RISING, self.BOTH}: + if callback: + callback(channel) + # FALLING edge: HIGH → LOW + elif previous_state == self.HIGH and current_state == self.LOW: + if edge in {self.FALLING, self.BOTH}: + if callback: + callback(channel) + + # Update previous state for next iteration + self._previous_states[channel] = current_state time.sleep(polltime) + + def input(self, channel) -> int: + """Read the current state of a channel (HIGH or LOW).""" + return self._channel_states.get(channel, 0) diff --git a/src/exosuit_python/motor.py b/src/exosuit_python/motor.py index 30c5a01..19b8e06 100644 --- a/src/exosuit_python/motor.py +++ b/src/exosuit_python/motor.py @@ -22,7 +22,10 @@ def check_communication(self) -> bool: """Check if motor can communicate.""" return True - def get_torque(self) -> int: - # TODO: make this function name/param identical to the motor implementation + def get_torque( + self, + ) -> ( + int + ): # TODO: make this function name/param identical to the motor implementation """Get the torque of the motor.""" - return 0 + return 0 # TODO: simulate real motor behavior diff --git a/tests/exosuit_test.py b/tests/exosuit_test.py index ef74496..dca966d 100644 --- a/tests/exosuit_test.py +++ b/tests/exosuit_test.py @@ -9,6 +9,7 @@ SWITCH_EVENT_HANDLER_INTERVAL, TENSION_SWITCH, IMUConfig, + TensionConfig, ) from exosuit_python.exosuit import Exosuit, ExosuitConfig, ExosuitStates from exosuit_python.gpio import MockGPIO @@ -46,7 +47,7 @@ def test_exosuit_switches(): assert exosuit._status == ExosuitStates.PRETENSIONING # simulate tension switch OFF exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.off_signal) - time.sleep(state_wait) + time.sleep(state_wait + TensionConfig.tensioning_timeout) assert exosuit._status == ExosuitStates.STANDBY # simulate power switch ON exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.on_signal) From 3b87f8340194b517ec42e1e40d0739c3152b94dc Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Tue, 9 Jun 2026 15:26:47 +0200 Subject: [PATCH 06/11] minor style changes --- src/exosuit_python/definitions.py | 2 +- src/exosuit_python/exosuit.py | 4 ++-- src/exosuit_python/gpio.py | 3 ++- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/src/exosuit_python/definitions.py b/src/exosuit_python/definitions.py index af7ae7d..b7e1c79 100644 --- a/src/exosuit_python/definitions.py +++ b/src/exosuit_python/definitions.py @@ -83,7 +83,7 @@ class IMUConfig: GPIO_SWITCH_BOUNCETIME = 50 # TODO: test this threshold -# switch signals +# switch signals - used as 'getattr' keys for GPIO SWITCH_ON = "HIGH" # TODO: wiring SWITCH_OFF = "LOW" BOTH = "BOTH" diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index c3219f6..792a887 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -184,7 +184,7 @@ def _switch_event_handler( pass # state change handled in _loop() time.sleep(SWITCH_EVENT_HANDLER_INTERVAL) - def _power_callback(self, channel) -> None: + def _power_callback(self, channel: int) -> None: """Handle power switch states triggered by signal events.""" power_state = self.gpio.input(POWER_SWITCH) if power_state == self.on_signal: @@ -194,7 +194,7 @@ def _power_callback(self, channel) -> None: else: logger.warning(f"Unrecognized power switch state: {power_state}") - def _tension_callback(self, channel) -> None: + def _tension_callback(self, channel: int) -> None: """Handle tension switch states triggered by signal events.""" tension_state = self.gpio.input(TENSION_SWITCH) if tension_state == self.on_signal: diff --git a/src/exosuit_python/gpio.py b/src/exosuit_python/gpio.py index 5adbcb7..a43b96f 100644 --- a/src/exosuit_python/gpio.py +++ b/src/exosuit_python/gpio.py @@ -67,7 +67,7 @@ def add_event_detect( self._threads.append(thread) thread.start() - def cleanup(self): + def cleanup(self) -> None: """Clean up GPIO.""" self.gpio_running = False for thread in self._threads: @@ -85,6 +85,7 @@ def _loop_event_detect_thread( bouncetime: int | None = None, polltime: float = 0.2, ) -> None: + """Check for edge event and call callback function.""" while self.gpio_running: current_state = self._channel_states.get(channel, 0) previous_state = self._previous_states.get(channel) From b175dda1cb40000f507440bdb9e20c43149f11cc Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Tue, 9 Jun 2026 15:38:01 +0200 Subject: [PATCH 07/11] use mock flag for IMU --- pyproject.toml | 2 +- src/exosuit_python/exosuit.py | 3 +-- uv.lock | 8 ++++---- 3 files changed, 6 insertions(+), 7 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 9b126f0..56b96f1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ dependencies = [ "numpy>=2.2.3", "loguru>=0.7.3", "motor-python>=0.0.4", - "imu-python>=0.1.1", + "imu-python>=0.1.2", "hip-controller>=0.0.4", ] diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index 792a887..f01e349 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -337,8 +337,7 @@ def _initialize_imus(self) -> bool: left_init: bool = False right_init: bool = False sensor_managers = IMUFactory.detect_and_create( - free_threading=True, - log_data=False, + free_threading=True, log_data=False, create_mock=self.config.mock ) detected_imus = len(sensor_managers) if detected_imus < 2: diff --git a/uv.lock b/uv.lock index 91ad473..70e279a 100644 --- a/uv.lock +++ b/uv.lock @@ -570,7 +570,7 @@ hw = [ [package.metadata] requires-dist = [ { name = "hip-controller", specifier = ">=0.0.4" }, - { name = "imu-python", specifier = ">=0.1.1" }, + { name = "imu-python", specifier = ">=0.1.2" }, { name = "loguru", specifier = ">=0.7.3" }, { name = "motor-python", specifier = ">=0.0.4" }, { name = "numpy", specifier = ">=2.2.3" }, @@ -689,7 +689,7 @@ wheels = [ [[package]] name = "imu-python" -version = "0.1.1" +version = "0.1.2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "ahrs" }, @@ -701,9 +701,9 @@ dependencies = [ { name = "py-imu" }, { name = "scipy" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/90/06/d414bcaefd6a9dfa9954d2692df3c897df0aa684a72a5fdf7e36c781f86c/imu_python-0.1.1.tar.gz", hash = "sha256:e48e641a8420b3718aa3dc39e870b93a2984acc434e7c52c3642ddbe0dc7cfcf", size = 32569, upload-time = "2026-04-21T09:22:23.323Z" } +sdist = { url = "https://files.pythonhosted.org/packages/be/d0/788112fffcce7d570a5c8e264073a2239aae1f20582282e91cac6efab322/imu_python-0.1.2.tar.gz", hash = "sha256:c25a7eeb71527927dbf549fce35aa16facf22a07a1a48b05512224099d6e9455", size = 32663, upload-time = "2026-06-09T13:34:48.676Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/38/62/e64a4204f6abb091a415ff5b95aef2bfffa47e48686addeb365ae62686e8/imu_python-0.1.1-py3-none-any.whl", hash = "sha256:3bca7535f6c888313c550fb329714949acd145980bde9ba33470a8aec35ea167", size = 39263, upload-time = "2026-04-21T09:22:21.983Z" }, + { url = "https://files.pythonhosted.org/packages/04/4c/1f033b89e05b864719ec46301852b0cb775fe8143cf7284517f95a2c6c37/imu_python-0.1.2-py3-none-any.whl", hash = "sha256:a5b8e173701a001eff1bf3bacc10280a5ed04fef455026f94cb9d95ed70ac874", size = 39387, upload-time = "2026-06-09T13:34:47.399Z" }, ] [[package]] From 8fd0b8ccda275f9207e01d0796eca6010509b22b Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Wed, 10 Jun 2026 11:13:33 +0200 Subject: [PATCH 08/11] update controller version, fix jetson bugs --- pyproject.toml | 2 +- src/exosuit_python/exosuit.py | 86 ++++++++++---------- tests/csv_writer_test.py | 144 ++++++++++++++++++++++------------ uv.lock | 81 +++++++++---------- 4 files changed, 177 insertions(+), 136 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 56b96f1..a8798d0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,7 +13,7 @@ dependencies = [ "loguru>=0.7.3", "motor-python>=0.0.4", "imu-python>=0.1.2", - "hip-controller>=0.0.4", + "hip-controller>=0.1.1", ] [dependency-groups] diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index f01e349..4c2c0b3 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -2,6 +2,7 @@ import threading import time +import warnings from dataclasses import dataclass, field from enum import IntEnum @@ -235,9 +236,15 @@ def _cleanup(self) -> None: """ logger.info("Cleaning up exosuit.") self._status = ExosuitStates.STOPPED - self.gpio.cleanup() - self.imu_left.stop() - self.imu_right.stop() + with warnings.catch_warnings(): + # suppress warning from GPIO when no channels has been set up + warnings.simplefilter("ignore", RuntimeWarning) + self.gpio.cleanup() + try: # imu attributes can be unassigned in case of failure + self.imu_left.stop() + self.imu_right.stop() + except AttributeError: + pass self.motor_left.close() self.motor_right.close() if self.thread is not None and self.thread.is_alive(): @@ -284,19 +291,17 @@ def _control(self) -> None: signal_right = SensorSignal( angle_rad=data_right.quat.to_euler(seq="xyz").z, velocity_rad_per_sec=data_right.device_data.gyro.z, + timestamp=timestamp_right, ) - command_right = self.controller_right.step( - timestamp=timestamp_right, curr_signal=signal_right - ) + command_right = self.controller_right.step(curr_signal=signal_right) timestamp_left = data_left.timestamp signal_left = SensorSignal( angle_rad=data_left.quat.to_euler(seq="xyz").z, velocity_rad_per_sec=data_left.device_data.gyro.z, + timestamp=timestamp_left, ) - command_left = self.controller_left.step( - timestamp=timestamp_left, curr_signal=signal_left - ) + command_left = self.controller_left.step(curr_signal=signal_left) self.motor_left.set_velocity(convert_rad_per_sec_to_rpm(command_left)) @@ -333,41 +338,40 @@ def _initialize_imus(self) -> bool: :return: True if successful, False otherwise """ - try: - left_init: bool = False - right_init: bool = False + left_init: bool = False + right_init: bool = False + + with warnings.catch_warnings(): + warnings.simplefilter("ignore", RuntimeWarning) sensor_managers = IMUFactory.detect_and_create( free_threading=True, log_data=False, create_mock=self.config.mock ) - detected_imus = len(sensor_managers) - if detected_imus < 2: - raise ValueError(f"Wrong number of IMUs detected: {detected_imus} < 2") - for idx in range( - detected_imus - ): # Match each leg with the IMU according to IMUConfig - manager = sensor_managers[idx] - if ( - not left_init - and manager.i2c_id == self.config.imu_cfg.left_leg_bus - and manager.imu_descriptor == self.config.imu_cfg.left_leg_descr - ): - self.imu_left = manager - left_init = True - continue - if ( - manager.i2c_id == self.config.imu_cfg.right_leg_bus - and manager.imu_descriptor == self.config.imu_cfg.right_leg_descr - ): - self.imu_right = manager - right_init = True - if not left_init: - raise RuntimeError("No detected IMUs matches left leg config") - if not right_init: - raise RuntimeError("No detected IMUs matches right leg config") - return True - except Exception as err: - logger.error(f"Exosuit exception: '{err}'. Check IMU connections.") - return False + detected_imus = len(sensor_managers) + if detected_imus < 2: + logger.error(f"Wrong number of IMUs detected: {detected_imus} < 2") + for idx in range( + detected_imus + ): # Match each leg with the IMU according to IMUConfig + manager = sensor_managers[idx] + if ( + not left_init + and manager.i2c_id == self.config.imu_cfg.left_leg_bus + and manager.imu_descriptor == self.config.imu_cfg.left_leg_descr + ): + self.imu_left = manager + left_init = True + continue + if ( + manager.i2c_id == self.config.imu_cfg.right_leg_bus + and manager.imu_descriptor == self.config.imu_cfg.right_leg_descr + ): + self.imu_right = manager + right_init = True + if not left_init: + logger.error("No detected IMUs matches left leg config") + if not right_init: + logger.error("No detected IMUs matches right leg config") + return left_init and right_init def _initialize_motors(self) -> bool: """Initialize Motors. diff --git a/tests/csv_writer_test.py b/tests/csv_writer_test.py index 8220738..81022bb 100644 --- a/tests/csv_writer_test.py +++ b/tests/csv_writer_test.py @@ -33,13 +33,17 @@ def test_reset_clears_rows(self) -> None: # Add some data data = RecordData( timestamp=1.0, - raw_signal_left=SensorSignal(angle_rad=0.1, velocity_rad_per_sec=0.2), + raw_signal_left=SensorSignal( + timestamp=1.0, angle_rad=0.1, velocity_rad_per_sec=0.2 + ), filtered_signal_left=SensorSignal( - angle_rad=0.15, velocity_rad_per_sec=0.25 + timestamp=1.0, angle_rad=0.15, velocity_rad_per_sec=0.25 + ), + raw_signal_right=SensorSignal( + timestamp=1.0, angle_rad=0.3, velocity_rad_per_sec=0.4 ), - raw_signal_right=SensorSignal(angle_rad=0.3, velocity_rad_per_sec=0.4), filtered_signal_right=SensorSignal( - angle_rad=0.35, velocity_rad_per_sec=0.45 + timestamp=1.0, angle_rad=0.35, velocity_rad_per_sec=0.45 ), motor_torque_nm_per_kg_left=1.5, motor_speed_rad_per_sec_left=2.0, @@ -65,13 +69,17 @@ class TestCSVWriterAppendData: # Basic zero values RecordData( timestamp=0.0, - raw_signal_left=SensorSignal(angle_rad=0.0, velocity_rad_per_sec=0.0), + raw_signal_left=SensorSignal( + timestamp=0.0, angle_rad=0.0, velocity_rad_per_sec=0.0 + ), filtered_signal_left=SensorSignal( - angle_rad=0.0, velocity_rad_per_sec=0.0 + timestamp=0.0, angle_rad=0.0, velocity_rad_per_sec=0.0 + ), + raw_signal_right=SensorSignal( + timestamp=0.0, angle_rad=0.0, velocity_rad_per_sec=0.0 ), - raw_signal_right=SensorSignal(angle_rad=0.0, velocity_rad_per_sec=0.0), filtered_signal_right=SensorSignal( - angle_rad=0.0, velocity_rad_per_sec=0.0 + timestamp=0.0, angle_rad=0.0, velocity_rad_per_sec=0.0 ), motor_torque_nm_per_kg_left=0.0, motor_speed_rad_per_sec_left=0.0, @@ -83,13 +91,17 @@ class TestCSVWriterAppendData: # Small positive values RecordData( timestamp=1.5, - raw_signal_left=SensorSignal(angle_rad=0.1, velocity_rad_per_sec=0.2), + raw_signal_left=SensorSignal( + timestamp=1.5, angle_rad=0.1, velocity_rad_per_sec=0.2 + ), filtered_signal_left=SensorSignal( - angle_rad=0.15, velocity_rad_per_sec=0.25 + timestamp=1.5, angle_rad=0.15, velocity_rad_per_sec=0.25 + ), + raw_signal_right=SensorSignal( + timestamp=1.5, angle_rad=0.3, velocity_rad_per_sec=0.4 ), - raw_signal_right=SensorSignal(angle_rad=0.3, velocity_rad_per_sec=0.4), filtered_signal_right=SensorSignal( - angle_rad=0.35, velocity_rad_per_sec=0.45 + timestamp=1.5, angle_rad=0.35, velocity_rad_per_sec=0.45 ), motor_torque_nm_per_kg_left=1.5, motor_speed_rad_per_sec_left=2.0, @@ -101,15 +113,17 @@ class TestCSVWriterAppendData: # Large values RecordData( timestamp=1000.5, - raw_signal_left=SensorSignal(angle_rad=3.14, velocity_rad_per_sec=6.28), + raw_signal_left=SensorSignal( + timestamp=1000.5, angle_rad=3.14, velocity_rad_per_sec=6.28 + ), filtered_signal_left=SensorSignal( - angle_rad=2.71, velocity_rad_per_sec=5.42 + timestamp=1000.5, angle_rad=2.71, velocity_rad_per_sec=5.42 ), raw_signal_right=SensorSignal( - angle_rad=1.41, velocity_rad_per_sec=2.82 + timestamp=1000.5, angle_rad=1.41, velocity_rad_per_sec=2.82 ), filtered_signal_right=SensorSignal( - angle_rad=1.73, velocity_rad_per_sec=3.46 + timestamp=1000.5, angle_rad=1.73, velocity_rad_per_sec=3.46 ), motor_torque_nm_per_kg_left=100.0, motor_speed_rad_per_sec_left=200.0, @@ -121,15 +135,17 @@ class TestCSVWriterAppendData: # Negative values RecordData( timestamp=2.0, - raw_signal_left=SensorSignal(angle_rad=-0.5, velocity_rad_per_sec=-1.0), + raw_signal_left=SensorSignal( + timestamp=2.0, angle_rad=-0.5, velocity_rad_per_sec=-1.0 + ), filtered_signal_left=SensorSignal( - angle_rad=-0.4, velocity_rad_per_sec=-0.9 + timestamp=2.0, angle_rad=-0.4, velocity_rad_per_sec=-0.9 ), raw_signal_right=SensorSignal( - angle_rad=-0.2, velocity_rad_per_sec=-0.3 + timestamp=2.0, angle_rad=-0.2, velocity_rad_per_sec=-0.3 ), filtered_signal_right=SensorSignal( - angle_rad=-0.15, velocity_rad_per_sec=-0.25 + timestamp=2.0, angle_rad=-0.15, velocity_rad_per_sec=-0.25 ), motor_torque_nm_per_kg_left=-2.0, motor_speed_rad_per_sec_left=-3.0, @@ -142,14 +158,16 @@ class TestCSVWriterAppendData: RecordData( timestamp=5.5, raw_signal_left=SensorSignal( - angle_rad=-0.82, velocity_rad_per_sec=-7.3 + timestamp=5.5, angle_rad=-0.82, velocity_rad_per_sec=-7.3 ), filtered_signal_left=SensorSignal( - angle_rad=0.14, velocity_rad_per_sec=3.78 + timestamp=5.5, angle_rad=0.14, velocity_rad_per_sec=3.78 + ), + raw_signal_right=SensorSignal( + timestamp=5.5, angle_rad=0.21, velocity_rad_per_sec=2.9 ), - raw_signal_right=SensorSignal(angle_rad=0.21, velocity_rad_per_sec=2.9), filtered_signal_right=SensorSignal( - angle_rad=-0.12, velocity_rad_per_sec=5.1 + timestamp=5.5, angle_rad=-0.12, velocity_rad_per_sec=5.1 ), motor_torque_nm_per_kg_left=2.3, motor_speed_rad_per_sec_left=-1.0, @@ -162,16 +180,16 @@ class TestCSVWriterAppendData: RecordData( timestamp=0.001, raw_signal_left=SensorSignal( - angle_rad=0.0001, velocity_rad_per_sec=0.0002 + timestamp=0.001, angle_rad=0.0001, velocity_rad_per_sec=0.0002 ), filtered_signal_left=SensorSignal( - angle_rad=0.00015, velocity_rad_per_sec=0.00025 + timestamp=0.001, angle_rad=0.00015, velocity_rad_per_sec=0.00025 ), raw_signal_right=SensorSignal( - angle_rad=0.0003, velocity_rad_per_sec=0.0004 + timestamp=0.001, angle_rad=0.0003, velocity_rad_per_sec=0.0004 ), filtered_signal_right=SensorSignal( - angle_rad=0.00035, velocity_rad_per_sec=0.00045 + timestamp=0.001, angle_rad=0.00035, velocity_rad_per_sec=0.00045 ), motor_torque_nm_per_kg_left=0.0015, motor_speed_rad_per_sec_left=0.002, @@ -256,16 +274,20 @@ def test_append_data_multiple_times(self, num_appends) -> None: data = RecordData( timestamp=float(i), raw_signal_left=SensorSignal( - angle_rad=0.1 * i, velocity_rad_per_sec=0.2 * i + timestamp=float(i), angle_rad=0.1 * i, velocity_rad_per_sec=0.2 * i ), filtered_signal_left=SensorSignal( - angle_rad=0.15 * i, velocity_rad_per_sec=0.25 * i + timestamp=float(i), + angle_rad=0.15 * i, + velocity_rad_per_sec=0.25 * i, ), raw_signal_right=SensorSignal( - angle_rad=0.3 * i, velocity_rad_per_sec=0.4 * i + timestamp=float(i), angle_rad=0.3 * i, velocity_rad_per_sec=0.4 * i ), filtered_signal_right=SensorSignal( - angle_rad=0.35 * i, velocity_rad_per_sec=0.45 * i + timestamp=float(i), + angle_rad=0.35 * i, + velocity_rad_per_sec=0.45 * i, ), motor_torque_nm_per_kg_left=1.5 * i, motor_speed_rad_per_sec_left=2.0 * i, @@ -295,13 +317,17 @@ class TestCSVWriterSaveData: [ RecordData( timestamp=1.0, - raw_signal_left=SensorSignal(angle_rad=0.1, velocity_rad_per_sec=0.2), + raw_signal_left=SensorSignal( + timestamp=1.0, angle_rad=0.1, velocity_rad_per_sec=0.2 + ), filtered_signal_left=SensorSignal( - angle_rad=0.15, velocity_rad_per_sec=0.25 + timestamp=1.0, angle_rad=0.15, velocity_rad_per_sec=0.25 + ), + raw_signal_right=SensorSignal( + timestamp=1.0, angle_rad=0.3, velocity_rad_per_sec=0.4 ), - raw_signal_right=SensorSignal(angle_rad=0.3, velocity_rad_per_sec=0.4), filtered_signal_right=SensorSignal( - angle_rad=0.35, velocity_rad_per_sec=0.45 + timestamp=1.0, angle_rad=0.35, velocity_rad_per_sec=0.45 ), motor_torque_nm_per_kg_left=1.5, motor_speed_rad_per_sec_left=2.0, @@ -312,15 +338,17 @@ class TestCSVWriterSaveData: ), RecordData( timestamp=2.5, - raw_signal_left=SensorSignal(angle_rad=-0.5, velocity_rad_per_sec=-1.0), + raw_signal_left=SensorSignal( + timestamp=2.5, angle_rad=-0.5, velocity_rad_per_sec=-1.0 + ), filtered_signal_left=SensorSignal( - angle_rad=-0.4, velocity_rad_per_sec=-0.9 + timestamp=2.5, angle_rad=-0.4, velocity_rad_per_sec=-0.9 ), raw_signal_right=SensorSignal( - angle_rad=-0.2, velocity_rad_per_sec=-0.3 + timestamp=2.5, angle_rad=-0.2, velocity_rad_per_sec=-0.3 ), filtered_signal_right=SensorSignal( - angle_rad=-0.15, velocity_rad_per_sec=-0.25 + timestamp=2.5, angle_rad=-0.15, velocity_rad_per_sec=-0.25 ), motor_torque_nm_per_kg_left=-2.0, motor_speed_rad_per_sec_left=-3.0, @@ -358,16 +386,20 @@ def test_save_data_multiple_rows(self) -> None: data = RecordData( timestamp=float(i), raw_signal_left=SensorSignal( - angle_rad=0.1 * i, velocity_rad_per_sec=0.2 * i + timestamp=float(i), angle_rad=0.1 * i, velocity_rad_per_sec=0.2 * i ), filtered_signal_left=SensorSignal( - angle_rad=0.15 * i, velocity_rad_per_sec=0.25 * i + timestamp=float(i), + angle_rad=0.15 * i, + velocity_rad_per_sec=0.25 * i, ), raw_signal_right=SensorSignal( - angle_rad=0.3 * i, velocity_rad_per_sec=0.4 * i + timestamp=float(i), angle_rad=0.3 * i, velocity_rad_per_sec=0.4 * i ), filtered_signal_right=SensorSignal( - angle_rad=0.35 * i, velocity_rad_per_sec=0.45 * i + timestamp=float(i), + angle_rad=0.35 * i, + velocity_rad_per_sec=0.45 * i, ), motor_torque_nm_per_kg_left=1.5 * i, motor_speed_rad_per_sec_left=2.0 * i, @@ -388,13 +420,17 @@ def test_save_data_preserves_all_columns(self) -> None: writer = CSVWriter() data = RecordData( timestamp=1.0, - raw_signal_left=SensorSignal(angle_rad=0.1, velocity_rad_per_sec=0.2), + raw_signal_left=SensorSignal( + timestamp=1.0, angle_rad=0.1, velocity_rad_per_sec=0.2 + ), filtered_signal_left=SensorSignal( - angle_rad=0.15, velocity_rad_per_sec=0.25 + timestamp=1.0, angle_rad=0.15, velocity_rad_per_sec=0.25 + ), + raw_signal_right=SensorSignal( + timestamp=1.0, angle_rad=0.3, velocity_rad_per_sec=0.4 ), - raw_signal_right=SensorSignal(angle_rad=0.3, velocity_rad_per_sec=0.4), filtered_signal_right=SensorSignal( - angle_rad=0.35, velocity_rad_per_sec=0.45 + timestamp=1.0, angle_rad=0.35, velocity_rad_per_sec=0.45 ), motor_torque_nm_per_kg_left=1.5, motor_speed_rad_per_sec_left=2.0, @@ -419,13 +455,17 @@ def test_save_data_with_varying_row_counts(self, num_rows) -> None: for i in range(num_rows): data = RecordData( timestamp=float(i), - raw_signal_left=SensorSignal(angle_rad=0.1, velocity_rad_per_sec=0.2), + raw_signal_left=SensorSignal( + timestamp=float(i), angle_rad=0.1, velocity_rad_per_sec=0.2 + ), filtered_signal_left=SensorSignal( - angle_rad=0.15, velocity_rad_per_sec=0.25 + timestamp=float(i), angle_rad=0.15, velocity_rad_per_sec=0.25 + ), + raw_signal_right=SensorSignal( + timestamp=float(i), angle_rad=0.3, velocity_rad_per_sec=0.4 ), - raw_signal_right=SensorSignal(angle_rad=0.3, velocity_rad_per_sec=0.4), filtered_signal_right=SensorSignal( - angle_rad=0.35, velocity_rad_per_sec=0.45 + timestamp=float(i), angle_rad=0.35, velocity_rad_per_sec=0.45 ), motor_torque_nm_per_kg_left=1.5, motor_speed_rad_per_sec_left=2.0, diff --git a/uv.lock b/uv.lock index 70e279a..3816ede 100644 --- a/uv.lock +++ b/uv.lock @@ -569,7 +569,7 @@ hw = [ [package.metadata] requires-dist = [ - { name = "hip-controller", specifier = ">=0.0.4" }, + { name = "hip-controller", specifier = ">=0.1.1" }, { name = "imu-python", specifier = ">=0.1.2" }, { name = "loguru", specifier = ">=0.7.3" }, { name = "motor-python", specifier = ">=0.0.4" }, @@ -653,10 +653,11 @@ wheels = [ [[package]] name = "hip-controller" -version = "0.0.5" +version = "0.1.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "loguru" }, + { name = "matplotlib" }, { name = "numpy" }, { name = "pandas" }, { name = "pyqt6" }, @@ -664,9 +665,9 @@ dependencies = [ { name = "pyside6" }, { name = "scipy" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/f8/95/31df9d66861e4acb7bcce98ff2c19b657f68c05ca5995cd064ad487b5c88/hip_controller-0.0.5.tar.gz", hash = "sha256:1f8dcc6758f4c9986bce855fb127d6de377964b30bb20e525f676c9ed1e93f99", size = 24463, upload-time = "2026-03-11T14:42:48.343Z" } +sdist = { url = "https://files.pythonhosted.org/packages/e3/ea/4f7a57c169b2eb0d9937252e46bd8d4e96f51961116dace825dbae546849/hip_controller-0.1.1.tar.gz", hash = "sha256:c7022eb2ebdf71a0cbfec39cd3c464643ce14f296f3e4137ca88c24969950343", size = 48883, upload-time = "2026-05-13T14:32:37.423Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/b3/d4/f4559dd7aa8eeba1441deb3805e677dc0b30f01ceaea0f2d1c9e1a9999ef/hip_controller-0.0.5-py3-none-any.whl", hash = "sha256:24a27317ff16207dc019d139247861228007670cc1d52fffae75830fddf15232", size = 31605, upload-time = "2026-03-11T14:42:46.968Z" }, + { url = "https://files.pythonhosted.org/packages/f9/cc/94c9baffb2b2473b3ebb3201c6d102f83f0fcb52f8d42cc51271891b7635/hip_controller-0.1.1-py3-none-any.whl", hash = "sha256:6e61227b1b2133a029ab27342feae5b199cddfb134b27811723f3486fb63d266", size = 53941, upload-time = "2026-05-13T14:32:35.968Z" }, ] [[package]] @@ -1284,32 +1285,32 @@ wheels = [ [[package]] name = "pyqt6" -version = "6.10.2" +version = "6.7.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "pyqt6-qt6" }, { name = "pyqt6-sip" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/96/03/e756f52e8b0d7bb5527baf8c46d59af0746391943bdb8655acba22ee4168/pyqt6-6.10.2.tar.gz", hash = "sha256:6c0db5d8cbb9a3e7e2b5b51d0ff3f283121fa27b864db6d2f35b663c9be5cc83", size = 1085573, upload-time = "2026-01-08T16:40:00.244Z" } +sdist = { url = "https://files.pythonhosted.org/packages/d1/f9/b0c2ba758b14a7219e076138ea1e738c068bf388e64eee68f3df4fc96f5a/PyQt6-6.7.1.tar.gz", hash = "sha256:3672a82ccd3a62e99ab200a13903421e2928e399fda25ced98d140313ad59cb9", size = 1051212, upload-time = "2024-07-19T08:49:58.247Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/fb/3f/f073a980969aa485ef288eb2e3b94c223ba9c7ac9941543f19b51659b98d/pyqt6-6.10.2-cp39-abi3-macosx_10_14_universal2.whl", hash = "sha256:37ae7c1183fe4dd0c6aefd2006a35731245de1cb6f817bb9e414a3e4848dfd6d", size = 60244482, upload-time = "2026-01-08T16:38:50.837Z" }, - { url = "https://files.pythonhosted.org/packages/ec/3e/9a015651ec71cea2e2f960c37edeb21623ba96a74956c0827def837f7c6b/pyqt6-6.10.2-cp39-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:78e1b3d5763e4cbc84485aef600e0aba5e1932fd263b716f92cd1a40dfa5e924", size = 37899440, upload-time = "2026-01-08T16:39:09.027Z" }, - { url = "https://files.pythonhosted.org/packages/51/74/a88fec2b99700270ca5d7dc7d650236a4990ed6fc88e055ca0fc8a339ee3/pyqt6-6.10.2-cp39-abi3-manylinux_2_39_aarch64.whl", hash = "sha256:bbc3af541bbecd27301bfe69fe445aa1611a9b490bd3de77306b12df632f7ec6", size = 40748467, upload-time = "2026-01-08T16:39:29.551Z" }, - { url = "https://files.pythonhosted.org/packages/75/34/be7a55529607b21db00a49ca53cb07c3092d2a5a95ea19bb95cfa0346904/pyqt6-6.10.2-cp39-abi3-win_amd64.whl", hash = "sha256:bd328cb70bc382c48861cd5f0a11b2b8ae6f5692d5a2d6679ba52785dced327b", size = 26015391, upload-time = "2026-01-08T16:39:42.946Z" }, - { url = "https://files.pythonhosted.org/packages/af/de/d9c88f976602b7884fec4ad54a4575d48e23e4f390e5357ea83917358846/pyqt6-6.10.2-cp39-abi3-win_arm64.whl", hash = "sha256:7901ba1df024b7ee9fdacfb2b7661aeb3749ae8b0bef65428077de3e0450eabb", size = 26208415, upload-time = "2026-01-08T16:39:57.751Z" }, + { url = "https://files.pythonhosted.org/packages/45/b0/20a05cfe287a1bc5a034cfed002bb1999f71c15e53a6ab7886c010ea0ba3/PyQt6-6.7.1-1-cp38-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:7f397f4b38b23b5588eb2c0933510deb953d96b1f0323a916c4839c2a66ccccc", size = 8020146, upload-time = "2024-07-31T09:50:04.992Z" }, + { url = "https://files.pythonhosted.org/packages/e4/d3/8789879c05cfe06127c4b59258632bd175fcdd9eaaadaf0c897b458fb91d/PyQt6-6.7.1-1-cp38-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:c2f202b7941aa74e5c7e1463a6f27d9131dbc1e6cabe85571d7364f5b3de7397", size = 8227345, upload-time = "2024-07-31T09:50:13.511Z" }, + { url = "https://files.pythonhosted.org/packages/15/2b/a0c516931697214dcb93b24a62f54b7467194ba1c76f3f7a55cb3a120cc9/PyQt6-6.7.1-cp38-abi3-macosx_11_0_universal2.whl", hash = "sha256:f053378e3aef6248fa612c8afddda17f942fb63f9fe8a9aeb2a6b6b4cbb0eba9", size = 11871174, upload-time = "2024-07-19T08:49:21.831Z" }, + { url = "https://files.pythonhosted.org/packages/59/8c/3b528f5fa8dfc3d0ba07d8da37ea72dfc59352d80804a12507d7080efb30/PyQt6-6.7.1-cp38-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:0adb7914c732ad1dee46d9cec838a98cb2b11bc38cc3b7b36fbd8701ae64bf47", size = 7999939, upload-time = "2024-07-19T08:49:30.82Z" }, + { url = "https://files.pythonhosted.org/packages/d8/58/5082dd3654da2b17de19057f181526df566f38af90f517cb8a541bea0890/PyQt6-6.7.1-cp38-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:2d771fa0981514cb1ee937633dfa64f14caa902707d9afffab66677f3a73e3da", size = 8177790, upload-time = "2024-07-19T08:49:48.394Z" }, + { url = "https://files.pythonhosted.org/packages/a3/69/99d22ee685c08a99fcf2048d366fe6173ba6e43ee13b95a3a2ac2911c52c/PyQt6-6.7.1-cp38-abi3-win_amd64.whl", hash = "sha256:fa3954698233fe286a8afc477b84d8517f0788eb46b74da69d3ccc0170d3714c", size = 6596360, upload-time = "2024-07-19T08:49:55.594Z" }, ] [[package]] name = "pyqt6-qt6" -version = "6.10.2" +version = "6.7.3" source = { registry = "https://pypi.org/simple" } wheels = [ - { url = "https://files.pythonhosted.org/packages/9a/eb/f04d547d8ed9f20c7b246db4ef5d93b49cab4692009a10652ed0a8b9d2aa/pyqt6_qt6-6.10.2-py3-none-macosx_10_14_x86_64.whl", hash = "sha256:5761cfccc721da2311c3f1213577f0ff1df07bbbbe3fa3a209a256b82cf057e3", size = 68688870, upload-time = "2026-01-29T12:26:48.619Z" }, - { url = "https://files.pythonhosted.org/packages/ce/c8/d99e65ab01c2402fb6bc4f77abef7244f7d5fb2f2e6d5b0abdf71bb2e4fc/pyqt6_qt6-6.10.2-py3-none-macosx_11_0_arm64.whl", hash = "sha256:6dda853a8db1b8d1a2ddbbe76cc6c3aa86614cad14056bd3c0435d8feea73b2d", size = 62512013, upload-time = "2026-01-29T12:27:24.642Z" }, - { url = "https://files.pythonhosted.org/packages/d5/fe/01fd9b9d2ca139ef61582f2e2da249fa169229144294c1bb27db59ad8420/pyqt6_qt6-6.10.2-py3-none-manylinux_2_34_x86_64.whl", hash = "sha256:19c10b5f0806e9f9bac2c9759bd5d7d19a78967f330fd60a2db409177fa76e49", size = 84028760, upload-time = "2026-01-29T12:28:03.267Z" }, - { url = "https://files.pythonhosted.org/packages/f4/20/a0d027ebb267d3afaf319d94efe1ff4d667004ee83b96701329a4d11fb95/pyqt6_qt6-6.10.2-py3-none-manylinux_2_39_aarch64.whl", hash = "sha256:2e60d616861ca4565cd295418d605975aa2dc407ba4b94c1586a70c92e9cb052", size = 83063975, upload-time = "2026-01-29T12:28:48.928Z" }, - { url = "https://files.pythonhosted.org/packages/06/8e/595f215876d507417cc8565e05519916d3b0b76baedea6a1e4e5105633fc/pyqt6_qt6-6.10.2-py3-none-win_amd64.whl", hash = "sha256:c4b7f7d66cc58bddf1bc1ca28dfcf7a45f58cfcb11d81d13a0510409dd4957ac", size = 78433821, upload-time = "2026-01-29T12:29:35.493Z" }, - { url = "https://files.pythonhosted.org/packages/50/5f/2196e2b536217b87cb3d2ce13ef8f7607d08b02f1990a4bd84a88d293a3c/pyqt6_qt6-6.10.2-py3-none-win_arm64.whl", hash = "sha256:7164a6f0c1335358a3026df9865c8f75395b01f60f0dcd2f66c029ec16fc83d2", size = 58354426, upload-time = "2026-01-29T12:30:02.95Z" }, + { url = "https://files.pythonhosted.org/packages/25/a2/9ef7c001068da2d3c8c37fe0e1e0451b1073d47c6ef4e44abf5883559963/PyQt6_Qt6-6.7.3-py3-none-macosx_10_14_x86_64.whl", hash = "sha256:f517a93b6b1a814d4aa6587adc312e812ebaf4d70415bb15cfb44268c5ad3f5f", size = 49136114, upload-time = "2024-09-29T16:25:15.055Z" }, + { url = "https://files.pythonhosted.org/packages/ec/63/a85bdd7c66800208f0af417bb4d07cb1543a75384021e4594e66d919f855/PyQt6_Qt6-6.7.3-py3-none-macosx_11_0_arm64.whl", hash = "sha256:8551732984fb36a5f4f3db51eafc4e8e6caf18617365830285306f2db17a94c2", size = 45762813, upload-time = "2024-09-29T16:25:20.956Z" }, + { url = "https://files.pythonhosted.org/packages/8a/6c/4f329f83a6082a7b4c1dc6046e2c48edb72e0d6d0ca3f8d0701fe134dccf/PyQt6_Qt6-6.7.3-py3-none-manylinux_2_28_aarch64.whl", hash = "sha256:50c7482bcdcf2bb78af257fb10ed8b582f8daf91d829782393bc50ac5a0a900c", size = 63801442, upload-time = "2024-09-29T16:25:26.637Z" }, + { url = "https://files.pythonhosted.org/packages/88/4d/26ca7239f7223e5b95b58a58537a09b069582ebb4dfa38234113a9f898ab/PyQt6_Qt6-6.7.3-py3-none-manylinux_2_28_x86_64.whl", hash = "sha256:cb525fdd393332de60887953029276a44de480fce1d785251ae639580f5e7246", size = 74366973, upload-time = "2024-09-29T16:25:33.595Z" }, + { url = "https://files.pythonhosted.org/packages/7e/57/3b44f6af1020fa543bd564c5bd346ba4aab1f1be0b861c2e8a0ad88cf3ca/PyQt6_Qt6-6.7.3-py3-none-win_amd64.whl", hash = "sha256:36ea0892b8caeb983af3f285f45fb8dfbb93cfd972439f4e01b7efb2868f6230", size = 58467498, upload-time = "2024-09-29T16:25:39.569Z" }, ] [[package]] @@ -1376,7 +1377,7 @@ wheels = [ [[package]] name = "pyside6" -version = "6.10.2" +version = "6.7.3" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "pyside6-addons" }, @@ -1384,42 +1385,39 @@ dependencies = [ { name = "shiboken6" }, ] wheels = [ - { url = "https://files.pythonhosted.org/packages/35/0f/5736889fc850794623692cb369e295a994175e51295fa52134626f486296/pyside6-6.10.2-cp39-abi3-macosx_13_0_universal2.whl", hash = "sha256:4b084293caa7845d0064aaf6af258e0f7caae03a14a33537d0a552131afddaf0", size = 563185, upload-time = "2026-02-02T08:50:47.161Z" }, - { url = "https://files.pythonhosted.org/packages/35/d3/ab5cd2fac3d34469c7376e0cd18eec92905dbe44748c70bda7699a2a7206/pyside6-6.10.2-cp39-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:1b89ce8558d4b4f35b85bff1db90d680912e4d3ce9e79ff804d6fef1d1a151ef", size = 563357, upload-time = "2026-02-02T08:50:48.919Z" }, - { url = "https://files.pythonhosted.org/packages/ea/8c/55bbd50c138c8dc12edc9f25e9d94760a33e574905468e98dff399094baa/pyside6-6.10.2-cp39-abi3-manylinux_2_39_aarch64.whl", hash = "sha256:0439f5e9b10ebe6177981bac9e219096ec970ac6ec215bef055279802ba50601", size = 563357, upload-time = "2026-02-02T08:50:50.077Z" }, - { url = "https://files.pythonhosted.org/packages/4f/d4/673b8112b4a260377f760be835c4e357163fdaf68a56a1aec59aeb8e584b/pyside6-6.10.2-cp39-abi3-win_amd64.whl", hash = "sha256:032bad6b18a17fcbf4dddd0397f49b07f8aae7f1a45b7e4de7037bf7fd6e0edf", size = 569554, upload-time = "2026-02-02T08:50:51.147Z" }, - { url = "https://files.pythonhosted.org/packages/14/95/bda648fcccf61fe58cb417284716ae30acdddd44f7d4cbad6eea4ccaa872/pyside6-6.10.2-cp39-abi3-win_arm64.whl", hash = "sha256:65a59ad0bc92525639e3268d590948ce07a80ee97b55e7a9200db41d493cac31", size = 553828, upload-time = "2026-02-02T08:50:52.244Z" }, + { url = "https://files.pythonhosted.org/packages/d4/34/86b0fd9b5ce8eee35bd1311a620e8cdfd56d0ca36edca314a6189336682d/PySide6-6.7.3-cp39-abi3-macosx_11_0_universal2.whl", hash = "sha256:1c21c4cf6cdd29bd13bbd7a2514756a19188eab992b92af03e64bf06a9b33d5b", size = 532108, upload-time = "2024-09-27T07:54:02.107Z" }, + { url = "https://files.pythonhosted.org/packages/36/8a/af0c2c91bcd8a0003e4233a0c74dd6b0e5af5534e96185d48a2c03a8f359/PySide6-6.7.3-cp39-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:a21480cc746358f70768975fcc452322f03b3c3622625bfb1743b40ce4e24beb", size = 532727, upload-time = "2024-09-27T07:54:04.758Z" }, + { url = "https://files.pythonhosted.org/packages/8a/56/a5347e273de35bd63fd3204bd9d80a134db959ff6f2bbeb299867dbe83db/PySide6-6.7.3-cp39-abi3-manylinux_2_31_aarch64.whl", hash = "sha256:c2a1313296c0088d1c3d231d0a8ccf0eda52b84139d0c4065fded76e4a4378f4", size = 532630, upload-time = "2024-09-27T07:54:07.145Z" }, + { url = "https://files.pythonhosted.org/packages/69/f0/18c6b7f5087eec0d673eb8703c3e9eb76d62cfc427b015d4fc833287c1c4/PySide6-6.7.3-cp39-abi3-win_amd64.whl", hash = "sha256:3ac8dcb4ca82d276e319f89dd99098b01061f255a2b9104216504aece5e0faf8", size = 539981, upload-time = "2024-09-27T07:54:10.354Z" }, ] [[package]] name = "pyside6-addons" -version = "6.10.2" +version = "6.7.3" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "pyside6-essentials" }, { name = "shiboken6" }, ] wheels = [ - { url = "https://files.pythonhosted.org/packages/61/06/c283567628ffa2cefc3c72374ad607f1dfc9842a03db65f1347b9ae52bee/pyside6_addons-6.10.2-cp39-abi3-macosx_13_0_universal2.whl", hash = "sha256:0de7d0c9535e17d5e3b634b61314a1867f3b0f6d35c3d7cdc99efc353192faff", size = 322745605, upload-time = "2026-02-02T08:39:19.929Z" }, - { url = "https://files.pythonhosted.org/packages/a5/69/e1ab8c756fd3984b1fd7b186446227f524f6b561160bfbfdba8874b4709a/pyside6_addons-6.10.2-cp39-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:030a851163b51dbf0063be59e9ddb6a9e760bde89a28e461ccc81a224d286eaf", size = 170718434, upload-time = "2026-02-02T08:40:55.989Z" }, - { url = "https://files.pythonhosted.org/packages/df/e5/18ba86ba86d1231c486d36f9accfe862ed6eb52ca0b698aeaf6e837a87ca/pyside6_addons-6.10.2-cp39-abi3-manylinux_2_39_aarch64.whl", hash = "sha256:fcee0373e3fd7b98f014094e5e37b4a39e4de7c5a47c13f654a7d557d4a426ad", size = 166423836, upload-time = "2026-02-02T08:42:44.918Z" }, - { url = "https://files.pythonhosted.org/packages/99/13/503bec9201881968c372cb634069535e80aec2489f3907d676e151a1023f/pyside6_addons-6.10.2-cp39-abi3-win_amd64.whl", hash = "sha256:c20150068525a17494f3b6576c5d61c417cf9a5870659e29f5ebd83cd20a78ea", size = 164712775, upload-time = "2026-02-02T08:43:23.729Z" }, - { url = "https://files.pythonhosted.org/packages/b6/39/44d6710b4dd18d745077b5fc6ded4ba6f32987a6e49c5834529e50f02155/pyside6_addons-6.10.2-cp39-abi3-win_arm64.whl", hash = "sha256:3d18db739b46946ba7b722d8ad4cc2097135033aa6ea57076e64d591e6a345f3", size = 34041396, upload-time = "2026-02-02T08:43:31.246Z" }, + { url = "https://files.pythonhosted.org/packages/03/35/e4b69e36d3ffdb6818e546188a99ab2c2fe9834f7d06d57a47d9bfce8fd3/PySide6_Addons-6.7.3-cp39-abi3-macosx_11_0_universal2.whl", hash = "sha256:3174cb3a373c09c98740b452e8e8f4945d64cfa18ed8d43964111d570f0dc647", size = 292945102, upload-time = "2024-09-27T07:41:48.313Z" }, + { url = "https://files.pythonhosted.org/packages/bb/57/1b16719360c23ddf4aafffbb281fc5cd8949db42c868d0df1ac466814508/PySide6_Addons-6.7.3-cp39-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:bde1eb03dbffd089b50cd445847aaecaf4056cea84c49ea592d00f84f247251e", size = 138279265, upload-time = "2024-09-27T07:42:11.497Z" }, + { url = "https://files.pythonhosted.org/packages/77/4d/547d8d5d7471a725b7221925266f5e8075bfd495f4d39156efdad4f59fa4/PySide6_Addons-6.7.3-cp39-abi3-manylinux_2_31_aarch64.whl", hash = "sha256:5a9e0df31345fe6caea677d916ea48b53ba86f95cc6499c57f89e392447ad6db", size = 123148721, upload-time = "2024-09-27T07:42:24.729Z" }, + { url = "https://files.pythonhosted.org/packages/3e/ca/1b34d0785298f86bed27582103760f67f7f78d9e2006405e256e9c770993/PySide6_Addons-6.7.3-cp39-abi3-win_amd64.whl", hash = "sha256:d8a19c2b2446407724c81c33ebf3217eaabd092f0f72da8130c17079e04a7813", size = 123578170, upload-time = "2024-09-27T07:42:41.047Z" }, ] [[package]] name = "pyside6-essentials" -version = "6.10.2" +version = "6.7.3" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "shiboken6" }, ] wheels = [ - { url = "https://files.pythonhosted.org/packages/1d/2e/5f18a77f5e0bd730bacec93a690d0ef3c96a9711d213653eacecbf241b8d/pyside6_essentials-6.10.2-cp39-abi3-macosx_13_0_universal2.whl", hash = "sha256:1dee2cb9803ff135f881dadeb5c0edcef793d1ec4f8a9140a1348cecb71074e1", size = 105913067, upload-time = "2026-02-02T08:45:37.508Z" }, - { url = "https://files.pythonhosted.org/packages/99/20/3a6ca95052e1744b5a3eba164e2dd451d358a3dcaf78179de4b45c8e3f47/pyside6_essentials-6.10.2-cp39-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:660aea45bfa36f1e06f799b934c2a7df963bd31abc5083e8bb8a5bfaef45686b", size = 77027153, upload-time = "2026-02-02T08:45:53.09Z" }, - { url = "https://files.pythonhosted.org/packages/93/a6/6073e4ddc2a5c7b3941606e4bc8bbaadcf0737f57450620b0793041c8d22/pyside6_essentials-6.10.2-cp39-abi3-manylinux_2_39_aarch64.whl", hash = "sha256:c2b028e4c6f8047a02c31f373408e23b4eedfd405f56c6aba8d0525c29472835", size = 76114242, upload-time = "2026-02-02T08:46:07.184Z" }, - { url = "https://files.pythonhosted.org/packages/22/a8/616bbbd009efd3e17bf9a2db09d90c6764c010565cd2bdea2a240bfd18f7/pyside6_essentials-6.10.2-cp39-abi3-win_amd64.whl", hash = "sha256:0741018c2b6395038cad4c41775cfae3f13a409e87995ac9f7d89e5b1fb6b22a", size = 74546490, upload-time = "2026-02-02T08:46:26.395Z" }, - { url = "https://files.pythonhosted.org/packages/b9/f9/c9757a984c4ffb6d12fab69e966d95dfc862a5d44e12b7900f3a03780b76/pyside6_essentials-6.10.2-cp39-abi3-win_arm64.whl", hash = "sha256:db5f4913648bb6afddb8b347edae151ee2378f12bceb03c8b2515a530a4b38d9", size = 55258626, upload-time = "2026-02-02T08:46:36.788Z" }, + { url = "https://files.pythonhosted.org/packages/bf/d8/053a3a95454ebc5313b3a4a9167b2f959fc6eaee3536b6e4269b75ce9716/PySide6_Essentials-6.7.3-cp39-abi3-macosx_11_0_universal2.whl", hash = "sha256:f9e08a4e9e7dc7b5ab72fde20abce8c97df7af1b802d9743f098f577dfe1f649", size = 158595022, upload-time = "2024-09-27T07:50:20.107Z" }, + { url = "https://files.pythonhosted.org/packages/8d/d8/a08ca6eca6839be5d604e4035ce899f9d6df0f612191f4ed4d03e51f1e0e/PySide6_Essentials-6.7.3-cp39-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:cda6fd26aead48f32e57f044d18aa75dc39265b49d7957f515ce7ac3989e7029", size = 90396416, upload-time = "2024-09-27T07:50:32.635Z" }, + { url = "https://files.pythonhosted.org/packages/5c/4d/00a5716e56c35c5621368c89abb360f6e953e2be639cd5f11e8075894e41/PySide6_Essentials-6.7.3-cp39-abi3-manylinux_2_31_aarch64.whl", hash = "sha256:acdde06b74f26e7d26b4ae1461081b32a6cb17fcaa2a580050b5e0f0f12236c9", size = 87264397, upload-time = "2024-09-27T07:50:43.557Z" }, + { url = "https://files.pythonhosted.org/packages/0b/18/8679adff0b7a6ccacc2e0febd15c1b78d02abc0180b56007c24e94c9d582/PySide6_Essentials-6.7.3-cp39-abi3-win_amd64.whl", hash = "sha256:f0950fcdcbcd4f2443336dc6a5fe692172adc225f876839583503ded0ab2f2a7", size = 68887979, upload-time = "2024-09-27T07:50:52.56Z" }, ] [[package]] @@ -1676,14 +1674,13 @@ wheels = [ [[package]] name = "shiboken6" -version = "6.10.2" +version = "6.7.3" source = { registry = "https://pypi.org/simple" } wheels = [ - { url = "https://files.pythonhosted.org/packages/fb/38/3912eb08a3b865b5fcdb4bdce8076cacc211986cee587f5cb62e637791af/shiboken6-6.10.2-cp39-abi3-macosx_13_0_universal2.whl", hash = "sha256:3bd4e94e9a3c8c1fa8362fd752d399ef39265d5264e4e37bae61cdaa2a00c8c7", size = 479829, upload-time = "2026-02-02T08:50:22.495Z" }, - { url = "https://files.pythonhosted.org/packages/52/88/292e0576489c46624ab419ee284ac5a59ae10e2eb34a58b6abca51dfd290/shiboken6-6.10.2-cp39-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:ace0790032d9cb0adda644b94ee28d59410180d9773643bb6cf8438c361987ad", size = 273052, upload-time = "2026-02-02T08:50:24.539Z" }, - { url = "https://files.pythonhosted.org/packages/06/c2/03d44d34e8264e1f25671677fece95b414c70fd85dcc2be8d5e821ee2628/shiboken6-6.10.2-cp39-abi3-manylinux_2_39_aarch64.whl", hash = "sha256:f74d3ed1f92658077d0630c39e694eb043aeb1d830a5d275176c45d07147427f", size = 269868, upload-time = "2026-02-02T08:50:25.662Z" }, - { url = "https://files.pythonhosted.org/packages/71/5d/5ca52c0ef86b3d01572131b6709bd531a080995f7e680720e9424328ce1d/shiboken6-6.10.2-cp39-abi3-win_amd64.whl", hash = "sha256:10f3c8c5e1b8bee779346f21c10dbc14cff068f0b0b4e62420c82a6bf36ac2e7", size = 1222052, upload-time = "2026-02-02T08:50:27.502Z" }, - { url = "https://files.pythonhosted.org/packages/46/52/421fd378313c89b67ee7d584bf4e9ec088fa1804891b8d74e02b16703457/shiboken6-6.10.2-cp39-abi3-win_arm64.whl", hash = "sha256:20c671645d70835af212ee05df60361d734c5305edb2746e9875c6a31283f963", size = 1784089, upload-time = "2026-02-02T08:50:29.069Z" }, + { url = "https://files.pythonhosted.org/packages/ef/7f/85d2234371fd260393760656e1f5977c0b6e270a0a6aca6bf7231a3e60de/shiboken6-6.7.3-cp39-abi3-macosx_11_0_universal2.whl", hash = "sha256:285fe3cf79be3135fe1ad1e2b9ff6db3a48698887425af6aa6ed7a05a9abc3d6", size = 388568, upload-time = "2024-09-27T07:51:06.762Z" }, + { url = "https://files.pythonhosted.org/packages/b2/e9/5c0c67de510da7818703d01123f7998cc495e99c0e6979a6ed6eb2ee09d0/shiboken6-6.7.3-cp39-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:f0852e5781de78be5b13c140ec4c7fb9734e2aaf2986eb2d6a224363e03efccc", size = 189890, upload-time = "2024-09-27T07:51:08.983Z" }, + { url = "https://files.pythonhosted.org/packages/ef/3c/bb524437df11ae79845b230d682b162935c7226b1fb7be2544dceca01e0f/shiboken6-6.7.3-cp39-abi3-manylinux_2_31_aarch64.whl", hash = "sha256:f0dd635178e64a45be2f84c9f33dd79ac30328da87f834f21a0baf69ae210e6e", size = 178147, upload-time = "2024-09-27T07:51:10.836Z" }, + { url = "https://files.pythonhosted.org/packages/3e/cc/e2b95aedb5e5f900c20325ef347e51c7750d18369d157f9b7da99943dae1/shiboken6-6.7.3-cp39-abi3-win_amd64.whl", hash = "sha256:5f29325dfa86fde0274240f1f38e421303749d3174ce3ada178715b5f4719db9", size = 1136685, upload-time = "2024-09-27T07:51:12.899Z" }, ] [[package]] From c92de4c872a3515062eb2087f07fccd66f31dab4 Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Thu, 11 Jun 2026 15:53:38 +0200 Subject: [PATCH 09/11] add flag to test gpio on Jetson --- src/exosuit_python/__main__.py | 27 +++++++++++++++++++++++---- src/exosuit_python/definitions.py | 4 ++-- src/exosuit_python/exosuit.py | 18 +++++++++++------- tests/exosuit_test.py | 4 ++-- 4 files changed, 38 insertions(+), 15 deletions(-) diff --git a/src/exosuit_python/__main__.py b/src/exosuit_python/__main__.py index 99333b9..59d0317 100644 --- a/src/exosuit_python/__main__.py +++ b/src/exosuit_python/__main__.py @@ -11,12 +11,14 @@ TENSION_SWITCH, LogLevel, ) -from exosuit_python.exosuit import Exosuit, ExosuitConfig +from exosuit_python.exosuit import Exosuit, ExosuitConfig, ExosuitStates from exosuit_python.gpio import MockGPIO from exosuit_python.utils import setup_logger -def main(log_level: str, stderr_level: str, mock: bool) -> None: # pragma: no cover +def main( + log_level: str, stderr_level: str, mock_devices: bool, test_gpio: bool +) -> None: # pragma: no cover """Run the main pipeline. :param log_level: The log level to use. @@ -25,7 +27,9 @@ def main(log_level: str, stderr_level: str, mock: bool) -> None: # pragma: no c """ setup_logger(log_level=log_level, stderr_level=stderr_level) - config = ExosuitConfig(frequency=100, mock=mock) + config = ExosuitConfig( + frequency=100, mock_devices=mock_devices, test_gpio=test_gpio + ) exosuit = Exosuit(config=config) try: while True: @@ -55,6 +59,11 @@ def test_pipeline(exosuit: Exosuit) -> None: logger.info("Simulating power switch OFF...") exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.off_signal) time.sleep(4) + else: + if exosuit._status not in [ExosuitStates.INITIALIZING, ExosuitStates.STOPPED]: + power_state = exosuit.gpio.input(POWER_SWITCH) + logger.info(power_state) + time.sleep(0.2) if __name__ == "__main__": # pragma: no cover @@ -80,6 +89,16 @@ def test_pipeline(exosuit: Exosuit) -> None: help="Use mock devices.", action="store_true", ) + parser.add_argument( + "--gpio", + help="Test GPIO on the Jetson.", + action="store_true", + ) args = parser.parse_args() - main(log_level=args.log_level, stderr_level=args.stderr_level, mock=args.mock) + main( + log_level=args.log_level, + stderr_level=args.stderr_level, + mock_devices=args.mock, + test_gpio=args.gpio, + ) diff --git a/src/exosuit_python/definitions.py b/src/exosuit_python/definitions.py index b7e1c79..6f56a30 100644 --- a/src/exosuit_python/definitions.py +++ b/src/exosuit_python/definitions.py @@ -78,8 +78,8 @@ class IMUConfig: # Switch pins -POWER_SWITCH = 7 # TODO: placeholder -TENSION_SWITCH = 15 # TODO: placeholder +POWER_SWITCH = 29 +TENSION_SWITCH = 31 GPIO_SWITCH_BOUNCETIME = 50 # TODO: test this threshold diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index 4c2c0b3..98ba251 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -43,13 +43,15 @@ class ExosuitConfig: Attributes: frequency: Exosuit frequency in Hz. - mock: flag to use mock devices. + mock_devices: flag to use mock devices. + test_gpio: flag to use Jetson GPIO (for switch testing on the Jetson) and use mock_devices. imu_cfg: IMU config that defines the IMU to use for each leg. """ frequency: float - mock: bool = False + mock_devices: bool = False + test_gpio: bool = False imu_cfg: IMUConfig = field(default_factory=IMUConfig) @@ -75,7 +77,7 @@ def __init__(self, config: ExosuitConfig) -> None: self._status: ExosuitStates = ExosuitStates.INITIALIZING # GPIO for switches - if self.config.mock or GPIO is None: + if GPIO is None or (not self.config.test_gpio and self.config.mock_devices): self.gpio = MockGPIO() else: self.gpio = GPIO @@ -112,7 +114,7 @@ def __init__(self, config: ExosuitConfig) -> None: self.motor_left: CubeMarsAK606v3 | MockMotor self.motor_right: CubeMarsAK606v3 | MockMotor - if self.config.mock: + if self.config.mock_devices or self.config.test_gpio: self.motor_left = MockMotor() self.motor_right = MockMotor() else: @@ -139,8 +141,8 @@ def _initialize_gpio(self) -> bool: """ try: self.gpio.setmode(self.gpio.BOARD) - self.gpio.setup(POWER_SWITCH, self.gpio.IN, pull_up_down=self.gpio.PUD_UP) - self.gpio.setup(TENSION_SWITCH, self.gpio.IN, pull_up_down=self.gpio.PUD_UP) + self.gpio.setup(POWER_SWITCH, self.gpio.IN) + self.gpio.setup(TENSION_SWITCH, self.gpio.IN) self.gpio.add_event_detect( POWER_SWITCH, @@ -344,7 +346,9 @@ def _initialize_imus(self) -> bool: with warnings.catch_warnings(): warnings.simplefilter("ignore", RuntimeWarning) sensor_managers = IMUFactory.detect_and_create( - free_threading=True, log_data=False, create_mock=self.config.mock + free_threading=True, + log_data=False, + create_mock=self.config.mock_devices or self.config.test_gpio, ) detected_imus = len(sensor_managers) if detected_imus < 2: diff --git a/tests/exosuit_test.py b/tests/exosuit_test.py index dca966d..2263e41 100644 --- a/tests/exosuit_test.py +++ b/tests/exosuit_test.py @@ -19,7 +19,7 @@ def test_exosuit_initialization(): """Test if exosuit is initialized with the set config.""" imu_config = IMUConfig() exosuit_config = ExosuitConfig( - frequency=DEFAULT_EXOSUIT_FREQUENCY_HZ, mock=True, imu_cfg=imu_config + frequency=DEFAULT_EXOSUIT_FREQUENCY_HZ, mock_devices=True, imu_cfg=imu_config ) exosuit = Exosuit(exosuit_config) @@ -31,7 +31,7 @@ def test_exosuit_switches(): """Test if exosuit's state changes correctly upon switch triggers.""" imu_config = IMUConfig() exosuit_config = ExosuitConfig( - frequency=DEFAULT_EXOSUIT_FREQUENCY_HZ, mock=True, imu_cfg=imu_config + frequency=DEFAULT_EXOSUIT_FREQUENCY_HZ, mock_devices=True, imu_cfg=imu_config ) exosuit = Exosuit(exosuit_config) From 4bfbbf0eab2fd10cd07bed29047f473e4687685f Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Fri, 12 Jun 2026 14:22:43 +0200 Subject: [PATCH 10/11] add mode switch --- src/exosuit_python/__main__.py | 3 +- src/exosuit_python/definitions.py | 56 ++++++++++++++++++++++++++- src/exosuit_python/exosuit.py | 63 ++++++++++++++++++++++--------- tests/exosuit_test.py | 27 +++++++++++++ 4 files changed, 128 insertions(+), 21 deletions(-) diff --git a/src/exosuit_python/__main__.py b/src/exosuit_python/__main__.py index 59d0317..444fccd 100644 --- a/src/exosuit_python/__main__.py +++ b/src/exosuit_python/__main__.py @@ -9,9 +9,10 @@ DEFAULT_LOG_LEVEL, POWER_SWITCH, TENSION_SWITCH, + ExosuitStates, LogLevel, ) -from exosuit_python.exosuit import Exosuit, ExosuitConfig, ExosuitStates +from exosuit_python.exosuit import Exosuit, ExosuitConfig from exosuit_python.gpio import MockGPIO from exosuit_python.utils import setup_logger diff --git a/src/exosuit_python/definitions.py b/src/exosuit_python/definitions.py index 6f56a30..ab22667 100644 --- a/src/exosuit_python/definitions.py +++ b/src/exosuit_python/definitions.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import asdict, dataclass, field +from enum import IntEnum, StrEnum from pathlib import Path import numpy as np @@ -80,10 +81,61 @@ class IMUConfig: # Switch pins POWER_SWITCH = 29 TENSION_SWITCH = 31 +MODE_SWITCH_1 = 32 +MODE_SWITCH_2 = 33 GPIO_SWITCH_BOUNCETIME = 50 # TODO: test this threshold + # switch signals - used as 'getattr' keys for GPIO -SWITCH_ON = "HIGH" # TODO: wiring -SWITCH_OFF = "LOW" +class SwitchStates(StrEnum): + """Enum to match switch states with electrical signals.""" + + ON = "HIGH" + OFF = "LOW" + + BOTH = "BOTH" + + +class ExosuitStates(IntEnum): + """Enum for exosuit states.""" + + INITIALIZING = 0 + STANDBY = 1 + PRETENSIONING = 2 + RUNNING = 3 + STOPPED = 4 + + +class InclinationModes(IntEnum): + """Enum for operation modes for different inclinations.""" + + LEVEL_GROUND = 0 + UPHILL = 1 + DOWNHILL = 2 + + +@dataclass +class ModeSwitchStates: + """Data class representing the states of the mode switch.""" + + switch_1: SwitchStates + switch_2: SwitchStates + + +# mode switch wiring: +MODE_SWITCH_LOGIC: dict[InclinationModes, ModeSwitchStates] = { + InclinationModes.UPHILL: ModeSwitchStates( + switch_1=SwitchStates.ON, + switch_2=SwitchStates.OFF, + ), + InclinationModes.DOWNHILL: ModeSwitchStates( + switch_1=SwitchStates.OFF, + switch_2=SwitchStates.ON, + ), + InclinationModes.LEVEL_GROUND: ModeSwitchStates( + switch_1=SwitchStates.OFF, + switch_2=SwitchStates.OFF, + ), +} diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index 98ba251..d286d8d 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -4,7 +4,6 @@ import time import warnings from dataclasses import dataclass, field -from enum import IntEnum from loguru import logger @@ -23,13 +22,17 @@ BOTH, EXOSUIT_STANDBY_INTERVAL, GPIO_SWITCH_BOUNCETIME, + MODE_SWITCH_1, + MODE_SWITCH_2, + MODE_SWITCH_LOGIC, POWER_SWITCH, SWITCH_EVENT_HANDLER_INTERVAL, - SWITCH_OFF, - SWITCH_ON, TENSION_SWITCH, THREAD_JOIN_TIMEOUT, + ExosuitStates, IMUConfig, + InclinationModes, + SwitchStates, TensionConfig, ) from exosuit_python.gpio import MockGPIO @@ -55,16 +58,6 @@ class ExosuitConfig: imu_cfg: IMUConfig = field(default_factory=IMUConfig) -class ExosuitStates(IntEnum): - """Enum for exosuit states.""" - - INITIALIZING = 0 - STANDBY = 1 - PRETENSIONING = 2 - RUNNING = 3 - STOPPED = 4 - - class Exosuit: """Tendon-based soft exoskeleton.""" @@ -83,16 +76,16 @@ def __init__(self, config: ExosuitConfig) -> None: self.gpio = GPIO if ( - hasattr(self.gpio, SWITCH_ON) - and hasattr(self.gpio, SWITCH_OFF) + hasattr(self.gpio, SwitchStates.ON) + and hasattr(self.gpio, SwitchStates.OFF) and hasattr(self.gpio, BOTH) ): - self.on_signal = getattr(self.gpio, SWITCH_ON) - self.off_signal = getattr(self.gpio, SWITCH_OFF) + self.on_signal = getattr(self.gpio, SwitchStates.ON) + self.off_signal = getattr(self.gpio, SwitchStates.OFF) self.both_signal = getattr(self.gpio, BOTH) else: logger.error( - f"GPIO signal attribute '{SWITCH_ON}', '{SWITCH_OFF}', or '{BOTH}' not found." + f"GPIO signal attribute '{SwitchStates.ON}', '{SwitchStates.OFF}', or '{BOTH}' not found." ) return @@ -104,6 +97,7 @@ def __init__(self, config: ExosuitConfig) -> None: self.switch_thread: threading.Thread = threading.Thread( target=self._switch_event_handler, daemon=True ) + self.inclination_mode: InclinationModes self.imu_left: IMUManager self.imu_right: IMUManager @@ -143,6 +137,8 @@ def _initialize_gpio(self) -> bool: self.gpio.setmode(self.gpio.BOARD) self.gpio.setup(POWER_SWITCH, self.gpio.IN) self.gpio.setup(TENSION_SWITCH, self.gpio.IN) + self.gpio.setup(MODE_SWITCH_1, self.gpio.IN) + self.gpio.setup(MODE_SWITCH_2, self.gpio.IN) self.gpio.add_event_detect( POWER_SWITCH, @@ -185,6 +181,21 @@ def _switch_event_handler( self._status = ExosuitStates.STANDBY elif self._status == ExosuitStates.PRETENSIONING: pass # state change handled in _loop() + + switch_1 = ( + SwitchStates.ON + if self.gpio.input(MODE_SWITCH_1) == self.on_signal + else SwitchStates.OFF + ) + switch_2 = ( + SwitchStates.ON + if self.gpio.input(MODE_SWITCH_2) == self.on_signal + else SwitchStates.OFF + ) + mode = self._get_mode(switch_1=switch_1, switch_2=switch_2) + if mode is not None: + self.inclination_mode = mode + time.sleep(SWITCH_EVENT_HANDLER_INTERVAL) def _power_callback(self, channel: int) -> None: @@ -221,6 +232,7 @@ def _start(self) -> None: logger.debug("Starting Motors") logger.debug("Starting Controller") + # TODO: get mode self._status = ExosuitStates.STANDBY logger.info("Exosuit status: standby") # Start main control loop @@ -394,3 +406,18 @@ def _initialize_motors(self) -> bool: except Exception as err: logger.error(f"Exosuit exception: '{err}'. Check Motor connections.") return False + + def _get_mode( + self, switch_1: SwitchStates, switch_2: SwitchStates + ) -> InclinationModes | None: + """Get inclination mode based on the current switch states. Return None if unrecognized. + + :param switch_1: state of the first channel of the switch. + :param switch_2: state of the second channel of the switch. + :return: the matched inclination mode, or None if unrecognized. + """ + for mode, state in MODE_SWITCH_LOGIC.items(): + if switch_1 == state.switch_1 and switch_2 == state.switch_2: + return mode + + return None diff --git a/tests/exosuit_test.py b/tests/exosuit_test.py index 2263e41..29b1925 100644 --- a/tests/exosuit_test.py +++ b/tests/exosuit_test.py @@ -5,6 +5,9 @@ from exosuit_python.definitions import ( DEFAULT_EXOSUIT_FREQUENCY_HZ, EXOSUIT_STANDBY_INTERVAL, + MODE_SWITCH_1, + MODE_SWITCH_2, + MODE_SWITCH_LOGIC, POWER_SWITCH, SWITCH_EVENT_HANDLER_INTERVAL, TENSION_SWITCH, @@ -61,3 +64,27 @@ def test_exosuit_switches(): # stop exosuit exosuit._cleanup() assert exosuit._status == ExosuitStates.STOPPED + + +def test_exosuit_inclination_mode_switch(): + """Test if exosuit's mode changes correctly upon switch triggers.""" + imu_config = IMUConfig() + exosuit_config = ExosuitConfig( + frequency=DEFAULT_EXOSUIT_FREQUENCY_HZ, mock_devices=True, imu_cfg=imu_config + ) + exosuit = Exosuit(exosuit_config) + + # wait for initialization + time.sleep(1) + assert isinstance(exosuit.gpio, MockGPIO) + + # test each mode + for mode, state in MODE_SWITCH_LOGIC.items(): + switch_1 = getattr(exosuit.gpio, state.switch_1) + switch_2 = getattr(exosuit.gpio, state.switch_2) + exosuit.gpio.simulate_switch(MODE_SWITCH_1, switch_1) + exosuit.gpio.simulate_switch(MODE_SWITCH_2, switch_2) + time.sleep(SWITCH_EVENT_HANDLER_INTERVAL + 0.1) + assert exosuit.inclination_mode == mode + + exosuit._cleanup() From 5179985cd458277f45909312f94bb41906d4f467 Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Wed, 17 Jun 2026 14:25:08 +0200 Subject: [PATCH 11/11] switch modes in controller, rename switch --- src/exosuit_python/__main__.py | 30 ++++++++++++++------- src/exosuit_python/definitions.py | 14 +++++++++- src/exosuit_python/exosuit.py | 45 +++++++++++++++++++------------ tests/exosuit_test.py | 10 +++---- 4 files changed, 66 insertions(+), 33 deletions(-) diff --git a/src/exosuit_python/__main__.py b/src/exosuit_python/__main__.py index 444fccd..c4d0e27 100644 --- a/src/exosuit_python/__main__.py +++ b/src/exosuit_python/__main__.py @@ -7,7 +7,10 @@ from exosuit_python.definitions import ( DEFAULT_LOG_LEVEL, - POWER_SWITCH, + MODE_SWITCH_1, + MODE_SWITCH_2, + MODE_SWITCH_LOGIC, + OPERATION_SWITCH, TENSION_SWITCH, ExosuitStates, LogLevel, @@ -52,18 +55,25 @@ def test_pipeline(exosuit: Exosuit) -> None: logger.info("Simulating tensioning switch OFF...") exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.off_signal) time.sleep(3) - # power on - logger.info("Simulating power switch ON...") - exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.on_signal) - time.sleep(10) - # power off - logger.info("Simulating power switch OFF...") - exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.off_signal) + # start operation + logger.info("Simulating operation switch ON...") + exosuit.gpio.simulate_switch(OPERATION_SWITCH, exosuit.on_signal) + time.sleep(2) + for mode, state in MODE_SWITCH_LOGIC.items(): + logger.info(f"Simulating mode {mode.name}...") + switch_1 = getattr(exosuit.gpio, state.switch_1) + switch_2 = getattr(exosuit.gpio, state.switch_2) + exosuit.gpio.simulate_switch(MODE_SWITCH_1, switch_1) + exosuit.gpio.simulate_switch(MODE_SWITCH_2, switch_2) + time.sleep(2) + # stop operation + logger.info("Simulating operation switch OFF...") + exosuit.gpio.simulate_switch(OPERATION_SWITCH, exosuit.off_signal) time.sleep(4) else: if exosuit._status not in [ExosuitStates.INITIALIZING, ExosuitStates.STOPPED]: - power_state = exosuit.gpio.input(POWER_SWITCH) - logger.info(power_state) + operation_state = exosuit.gpio.input(OPERATION_SWITCH) + logger.info(operation_state) time.sleep(0.2) diff --git a/src/exosuit_python/definitions.py b/src/exosuit_python/definitions.py index ab22667..639be45 100644 --- a/src/exosuit_python/definitions.py +++ b/src/exosuit_python/definitions.py @@ -7,6 +7,12 @@ from pathlib import Path import numpy as np +from hip_controller.control.motor_reference_control.amplitude_modulation import ( + AscendStairsMode, + DescendStairsMode, + LevelGroundMode, + ModeStrategy, +) from imu_python.definitions import I2CBusID, IMUDescriptor np.set_printoptions(precision=3, floatmode="fixed", suppress=True) @@ -79,7 +85,7 @@ class IMUConfig: # Switch pins -POWER_SWITCH = 29 +OPERATION_SWITCH = 29 TENSION_SWITCH = 31 MODE_SWITCH_1 = 32 MODE_SWITCH_2 = 33 @@ -124,6 +130,12 @@ class ModeSwitchStates: switch_2: SwitchStates +controller_modes: dict[InclinationModes, ModeStrategy] = { + InclinationModes.UPHILL: AscendStairsMode(), + InclinationModes.DOWNHILL: DescendStairsMode(), + InclinationModes.LEVEL_GROUND: LevelGroundMode(), +} + # mode switch wiring: MODE_SWITCH_LOGIC: dict[InclinationModes, ModeSwitchStates] = { InclinationModes.UPHILL: ModeSwitchStates( diff --git a/src/exosuit_python/exosuit.py b/src/exosuit_python/exosuit.py index d286d8d..04063d0 100644 --- a/src/exosuit_python/exosuit.py +++ b/src/exosuit_python/exosuit.py @@ -25,7 +25,7 @@ MODE_SWITCH_1, MODE_SWITCH_2, MODE_SWITCH_LOGIC, - POWER_SWITCH, + OPERATION_SWITCH, SWITCH_EVENT_HANDLER_INTERVAL, TENSION_SWITCH, THREAD_JOIN_TIMEOUT, @@ -34,6 +34,7 @@ InclinationModes, SwitchStates, TensionConfig, + controller_modes, ) from exosuit_python.gpio import MockGPIO from exosuit_python.motor import MockMotor @@ -89,7 +90,7 @@ def __init__(self, config: ExosuitConfig) -> None: ) return - self._power_switch: bool = False + self._operation_switch: bool = False self._tension_switch: bool = False # main loop thread and switch handler thread @@ -97,7 +98,8 @@ def __init__(self, config: ExosuitConfig) -> None: self.switch_thread: threading.Thread = threading.Thread( target=self._switch_event_handler, daemon=True ) - self.inclination_mode: InclinationModes + self.inclination_mode: InclinationModes = InclinationModes.LEVEL_GROUND + self._prev_inclination_mode: InclinationModes = InclinationModes.LEVEL_GROUND self.imu_left: IMUManager self.imu_right: IMUManager @@ -135,15 +137,15 @@ def _initialize_gpio(self) -> bool: """ try: self.gpio.setmode(self.gpio.BOARD) - self.gpio.setup(POWER_SWITCH, self.gpio.IN) + self.gpio.setup(OPERATION_SWITCH, self.gpio.IN) self.gpio.setup(TENSION_SWITCH, self.gpio.IN) self.gpio.setup(MODE_SWITCH_1, self.gpio.IN) self.gpio.setup(MODE_SWITCH_2, self.gpio.IN) self.gpio.add_event_detect( - POWER_SWITCH, + OPERATION_SWITCH, self.both_signal, - callback=self._power_callback, + callback=self._operation_callback, bouncetime=GPIO_SWITCH_BOUNCETIME, ) @@ -169,14 +171,14 @@ def _switch_event_handler( """ while self._status != ExosuitStates.STOPPED: if self._status == ExosuitStates.STANDBY: - if self._power_switch and not self._tension_switch: + if self._operation_switch and not self._tension_switch: logger.info("State change: standby -> running") self._status = ExosuitStates.RUNNING - elif self._tension_switch and not self._power_switch: + elif self._tension_switch and not self._operation_switch: logger.info("State change: standby -> pretensioning") self._status = ExosuitStates.PRETENSIONING elif self._status == ExosuitStates.RUNNING: - if not self._power_switch: + if not self._operation_switch: logger.info("State change: running -> standby") self._status = ExosuitStates.STANDBY elif self._status == ExosuitStates.PRETENSIONING: @@ -198,15 +200,15 @@ def _switch_event_handler( time.sleep(SWITCH_EVENT_HANDLER_INTERVAL) - def _power_callback(self, channel: int) -> None: - """Handle power switch states triggered by signal events.""" - power_state = self.gpio.input(POWER_SWITCH) - if power_state == self.on_signal: - self._power_switch = True - elif power_state == self.off_signal: - self._power_switch = False + def _operation_callback(self, channel: int) -> None: + """Handle operation switch states triggered by signal events.""" + opration_state = self.gpio.input(OPERATION_SWITCH) + if opration_state == self.on_signal: + self._operation_switch = True + elif opration_state == self.off_signal: + self._operation_switch = False else: - logger.warning(f"Unrecognized power switch state: {power_state}") + logger.warning(f"Unrecognized operation switch state: {opration_state}") def _tension_callback(self, channel: int) -> None: """Handle tension switch states triggered by signal events.""" @@ -298,6 +300,15 @@ def _control(self) -> None: data_right = self.imu_right.get_data() data_left = self.imu_left.get_data() + if self._prev_inclination_mode != self.inclination_mode: + logger.info( + f"Mode change:{self._prev_inclination_mode.name} -> {self.inclination_mode.name}" + ) + self.controller_left.amplitude_modulation.set_mode( + controller_modes[self.inclination_mode] + ) + self._prev_inclination_mode = self.inclination_mode + if data_right is None or data_left is None: raise TypeError diff --git a/tests/exosuit_test.py b/tests/exosuit_test.py index 29b1925..00aac86 100644 --- a/tests/exosuit_test.py +++ b/tests/exosuit_test.py @@ -8,7 +8,7 @@ MODE_SWITCH_1, MODE_SWITCH_2, MODE_SWITCH_LOGIC, - POWER_SWITCH, + OPERATION_SWITCH, SWITCH_EVENT_HANDLER_INTERVAL, TENSION_SWITCH, IMUConfig, @@ -52,12 +52,12 @@ def test_exosuit_switches(): exosuit.gpio.simulate_switch(TENSION_SWITCH, exosuit.off_signal) time.sleep(state_wait + TensionConfig.tensioning_timeout) assert exosuit._status == ExosuitStates.STANDBY - # simulate power switch ON - exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.on_signal) + # simulate operation switch ON + exosuit.gpio.simulate_switch(OPERATION_SWITCH, exosuit.on_signal) time.sleep(state_wait) assert exosuit._status == ExosuitStates.RUNNING - # simulate power switch OFF - exosuit.gpio.simulate_switch(POWER_SWITCH, exosuit.off_signal) + # simulate operation switch OFF + exosuit.gpio.simulate_switch(OPERATION_SWITCH, exosuit.off_signal) time.sleep(state_wait) assert exosuit._status == ExosuitStates.STANDBY