Mock flag

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 = [

src/imu_python/__main__.py

@@ -11,21 +11,23 @@
 
 
 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)
     imu_managers = IMUFactory.detect_and_create(
         free_threading=True,
         log_data=record_imu,
         calibration_mode=False,
+        create_mock=use_mock,
     )
     time.sleep(1)
     for manager in imu_managers:
@@ -74,11 +76,18 @@ 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(
         log_level=args.log_level,
         stderr_level=args.stderr_level,
         freq=args.freq,
         record_imu=args.record,
+        use_mock=args.mock,
     )

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,20 +65,26 @@ 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.
 
         :param i2c_lock: a shared i2c lock among IMU managers on this 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)
 

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:

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()