From 8cc8cbb2000f3a9b6f76576b66eb7fa814935720 Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Thu, 4 Jun 2026 10:54:16 +0200 Subject: [PATCH 1/4] add flag to use mock --- src/imu_python/__main__.py | 10 ++++++++- src/imu_python/factory.py | 11 +++++++++- src/imu_python/i2c_bus.py | 2 +- tests/factory_test.py | 45 ++++++++++++++++++++++++++++++++++++-- 4 files changed, 63 insertions(+), 5 deletions(-) diff --git a/src/imu_python/__main__.py b/src/imu_python/__main__.py index 31dc361..f37c6af 100644 --- a/src/imu_python/__main__.py +++ b/src/imu_python/__main__.py @@ -11,7 +11,7 @@ def main( - log_level: str, stderr_level: str, freq: float, record_imu: bool + log_level: str, stderr_level: str, freq: float, record_imu: bool, mock: bool ) -> None: # pragma: no cover """Run the main pipeline. @@ -26,6 +26,7 @@ def main( free_threading=True, log_data=record_imu, calibration_mode=False, + create_mock=mock, ) time.sleep(1) for manager in imu_managers: @@ -74,6 +75,12 @@ def main( help="Record IMU data.", action="store_true", ) + parser.add_argument( + "--mock", + "-m", + help="Use Mock IMUs.", + action="store_true", + ) args = parser.parse_args() main( @@ -81,4 +88,5 @@ def main( stderr_level=args.stderr_level, freq=args.freq, record_imu=args.record, + mock=args.mock, ) diff --git a/src/imu_python/factory.py b/src/imu_python/factory.py index 42022c6..98374f1 100644 --- a/src/imu_python/factory.py +++ b/src/imu_python/factory.py @@ -20,12 +20,14 @@ def detect_and_create( free_threading: bool = True, log_data: bool = False, calibration_mode: bool = False, + create_mock: bool = False, ) -> list[IMUManager]: """Automatically detect addresses on all buses defined in I2CBUSID and create sensor managers. :param free_threading: Flag to enable free threading. :param log_data: Flag to record the IMU data. :param calibration_mode: Flag to use calibration mode. + :param create_mock: Flag to create mock instead of real IMUs. :return: list of IMUManager instances. """ # GIL enabled or core_count == 0 means no free threading @@ -43,6 +45,7 @@ def detect_and_create( i2c_id=bus, log_data=log_data, calibration_mode=calibration_mode, + create_mock=create_mock, ) if free_threading: if len(managers) > CORE_COUNT: @@ -62,6 +65,7 @@ def _detect_and_create_per_bus( i2c_id: I2CBusID | None = None, log_data: bool = False, calibration_mode: bool = False, + create_mock: bool = False, ) -> list[IMUManager]: """Automatically detect addresses on the given bus and create sensor managers. @@ -69,13 +73,18 @@ def _detect_and_create_per_bus( :param i2c_id: I2C bus identifier. If None, attempt to use board.I2C(). :param log_data: Flag to record the IMU data. :param calibration_mode: Flag to use calibration mode. + :param create_mock: Flag to create mock instead of real IMUs. :return: list of IMUManager instances. """ imu_managers: list[IMUManager] = [] i2c_bus = JetsonBus.get(bus_id=i2c_id) - addresses = IMUFactory.scan_i2c_bus(i2c=i2c_bus) + if create_mock: + _, mock = get_mock() + addresses = [a for d in mock.devices.values() for a in d.addresses] + else: + addresses = IMUFactory.scan_i2c_bus(i2c=i2c_bus) detected_configs = get_config(addresses=addresses) diff --git a/src/imu_python/i2c_bus.py b/src/imu_python/i2c_bus.py index b597bea..e5b0e04 100644 --- a/src/imu_python/i2c_bus.py +++ b/src/imu_python/i2c_bus.py @@ -47,7 +47,7 @@ def _initialize(cls) -> None: def get(cls, bus_id: I2CBusID | None) -> ExtendedI2C | None: """Return the Jetson I2C bus for a given ID. - :param bus_id: One of I2CBusID.left, I2CBusID.right, or None. + :param bus_id: One of I2CBusID.bus_1, I2CBusID.bus_7, or None. :return: I2C bus instance or None. """ if bus_id is None: diff --git a/tests/factory_test.py b/tests/factory_test.py index ed2f0db..778d45a 100644 --- a/tests/factory_test.py +++ b/tests/factory_test.py @@ -1,5 +1,7 @@ """Test the IMUFactory class.""" +from unittest.mock import MagicMock + from src.imu_python.devices import IMU_DEVICES, get_mock from src.imu_python.factory import IMUFactory @@ -10,10 +12,10 @@ def test_imu_factory() -> None: mock_imu_name, mock_imu_config = get_mock() # Act - imu_managers = IMUFactory.detect_and_create() + imu_managers = IMUFactory.detect_and_create(create_mock=True) # Assert - assert len(imu_managers) >= 0 + assert len(imu_managers) > 0 for imu_manager in imu_managers: config = imu_manager.imu_wrapper.config @@ -34,3 +36,42 @@ def test_imu_factory() -> None: # check that config matches the expected IMU name assert mock_imu_name in IMU_DEVICES assert mock_imu_config == IMU_DEVICES[mock_imu_name] + + +def test_scan_i2c_bus_success() -> None: + """Test scan_i2c_bus with successful scan.""" + # Arrange + mock_i2c = MagicMock() + mock_i2c.try_lock.return_value = True + expected_addresses = [0x28, 0x6A, 0x1C] + mock_i2c.scan.return_value = expected_addresses + + # Act + addresses = IMUFactory.scan_i2c_bus(i2c=mock_i2c) + + # Assert + assert addresses == expected_addresses + mock_i2c.try_lock.assert_called_once() + mock_i2c.scan.assert_called_once() + mock_i2c.unlock.assert_called_once() + + +def test_scan_i2c_bus_exception_returns_mock_addresses() -> None: + """Test scan_i2c_bus returns mock addresses on exception.""" + # Arrange + mock_i2c = MagicMock() + mock_i2c.try_lock.return_value = True + mock_i2c.scan.side_effect = OSError("I2C communication error") + + # Act + addresses = IMUFactory.scan_i2c_bus(i2c=mock_i2c) + + # Assert + # Should return mock addresses + _, mock_config = get_mock() + expected_addresses = [a for d in mock_config.devices.values() for a in d.addresses] + assert addresses == expected_addresses + mock_i2c.try_lock.assert_called() + mock_i2c.scan.assert_called_once() + # unlock should not be called when scan() raises exception + mock_i2c.unlock.assert_not_called() From 9a8dcb9dcc738ffa763d808bf7207a02f588ee75 Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Mon, 8 Jun 2026 10:40:45 +0200 Subject: [PATCH 2/4] fix docstring --- src/imu_python/__main__.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/imu_python/__main__.py b/src/imu_python/__main__.py index f37c6af..181e80a 100644 --- a/src/imu_python/__main__.py +++ b/src/imu_python/__main__.py @@ -18,7 +18,8 @@ def main( :param log_level: The log level to use. :param stderr_level: The std err level to use. :param freq: The frequency to use. - :param record_imu: Flag to record the IMU data + :param record_imu: Flag to record the IMU data. + :param mock: Flag to create mock IMUs. :return: None """ setup_logger(log_level=log_level, stderr_level=stderr_level) From b367c13f6ece637b1ecfe5042cc320eccd47c2c8 Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Mon, 8 Jun 2026 11:07:44 +0200 Subject: [PATCH 3/4] bump version number --- pyproject.toml | 2 +- uv.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 0f04c81..dfa0d62 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "imu_python" -version = "0.1.1" +version = "0.1.2" description = "IMU sensor codes in Python for the exosuit" readme = "README.md" authors = [ diff --git a/uv.lock b/uv.lock index 18608b7..e00f349 100644 --- a/uv.lock +++ b/uv.lock @@ -644,7 +644,7 @@ wheels = [ [[package]] name = "imu-python" -version = "0.1.1" +version = "0.1.2" source = { editable = "." } dependencies = [ { name = "ahrs" }, From 29c12ce20dd121cf19117609b821c5746c8c7941 Mon Sep 17 00:00:00 2001 From: CatYang3 Date: Tue, 9 Jun 2026 11:02:24 +0200 Subject: [PATCH 4/4] minor changes based on Claude feedback --- src/imu_python/__main__.py | 8 ++++---- src/imu_python/factory.py | 2 +- tests/factory_test.py | 9 +++++---- 3 files changed, 10 insertions(+), 9 deletions(-) diff --git a/src/imu_python/__main__.py b/src/imu_python/__main__.py index 181e80a..cde7225 100644 --- a/src/imu_python/__main__.py +++ b/src/imu_python/__main__.py @@ -11,7 +11,7 @@ def main( - log_level: str, stderr_level: str, freq: float, record_imu: bool, mock: bool + log_level: str, stderr_level: str, freq: float, record_imu: bool, use_mock: bool ) -> None: # pragma: no cover """Run the main pipeline. @@ -19,7 +19,7 @@ def main( :param stderr_level: The std err level to use. :param freq: The frequency to use. :param record_imu: Flag to record the IMU data. - :param mock: Flag to create mock IMUs. + :param use_mock: Flag to create mock IMUs. :return: None """ setup_logger(log_level=log_level, stderr_level=stderr_level) @@ -27,7 +27,7 @@ def main( free_threading=True, log_data=record_imu, calibration_mode=False, - create_mock=mock, + create_mock=use_mock, ) time.sleep(1) for manager in imu_managers: @@ -89,5 +89,5 @@ def main( stderr_level=args.stderr_level, freq=args.freq, record_imu=args.record, - mock=args.mock, + use_mock=args.mock, ) diff --git a/src/imu_python/factory.py b/src/imu_python/factory.py index 98374f1..b182346 100644 --- a/src/imu_python/factory.py +++ b/src/imu_python/factory.py @@ -78,7 +78,7 @@ def _detect_and_create_per_bus( """ imu_managers: list[IMUManager] = [] - i2c_bus = JetsonBus.get(bus_id=i2c_id) + i2c_bus = None if create_mock else JetsonBus.get(bus_id=i2c_id) if create_mock: _, mock = get_mock() diff --git a/tests/factory_test.py b/tests/factory_test.py index 778d45a..798803b 100644 --- a/tests/factory_test.py +++ b/tests/factory_test.py @@ -18,6 +18,8 @@ def test_imu_factory() -> None: assert len(imu_managers) > 0 for imu_manager in imu_managers: config = imu_manager.imu_wrapper.config + # populate the wrapper device dict + imu_manager.imu_wrapper.reload() # The manager should have at least one device assert len(config.devices) > 0 @@ -28,10 +30,9 @@ def test_imu_factory() -> None: # Each role attribute should exist in the device driver device = imu_manager.imu_wrapper._devices.get(device_id) - if device: # device may not exist if not reloaded - attr_name = role.value - # getattr should succeed without error - getattr(device, attr_name, None) + attr_name = role.value + # getattr should succeed without error + getattr(device, attr_name, None) # check that config matches the expected IMU name assert mock_imu_name in IMU_DEVICES