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/src/imu_python/__main__.py b/src/imu_python/__main__.py index 31dc361..cde7225 100644 --- a/src/imu_python/__main__.py +++ b/src/imu_python/__main__.py @@ -11,14 +11,15 @@ def main( - log_level: str, stderr_level: str, freq: float, record_imu: bool + log_level: str, stderr_level: str, freq: float, record_imu: bool, use_mock: bool ) -> None: # pragma: no cover """Run the main pipeline. :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 use_mock: Flag to create mock IMUs. :return: None """ setup_logger(log_level=log_level, stderr_level=stderr_level) @@ -26,6 +27,7 @@ def main( free_threading=True, log_data=record_imu, calibration_mode=False, + create_mock=use_mock, ) time.sleep(1) for manager in imu_managers: @@ -74,6 +76,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 +89,5 @@ def main( stderr_level=args.stderr_level, freq=args.freq, record_imu=args.record, + use_mock=args.mock, ) diff --git a/src/imu_python/factory.py b/src/imu_python/factory.py index 42022c6..b182346 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) + i2c_bus = None if create_mock else 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..798803b 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,12 +12,14 @@ 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 + # populate the wrapper device dict + imu_manager.imu_wrapper.reload() # The manager should have at least one device assert len(config.devices) > 0 @@ -26,11 +30,49 @@ 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 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() 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" },