Simulated interface now tracks current firmware versions and supports in-sim firmware updates

src/openlifu_sdk/ui/simulated_interface.py

@@ -23,7 +23,9 @@
 import logging
 import math
 import random
+import re
 import time
+from datetime import datetime, timezone
 from typing import List, Optional
 
 # Qt backend selection: prefer 3D Slicer's PythonQt-based ``qt`` module
@@ -68,6 +70,9 @@
 logger = logging.getLogger(__name__)
 logger.debug("openlifu_sdk.ui.simulated_interface using Qt backend: %s", _QT_BACKEND)
 
+_FALLBACK_CONSOLE_FW_VERSION = "1.2.6"
+_FALLBACK_TX_FW_VERSION = "2.0.5"
+
 # k chosen so 45 V * 0.25 duty * 600 s -> 50 deg C
 TX_HEATING_K = 50.0 / (45.0 * 45.0 * 0.25 * 600.0)
 # Newton's-law cooling time constant (seconds). 600 s ~ 10 min half-life-ish.
@@ -91,6 +96,54 @@ def _gauss(sigma: float) -> float:
     return random.gauss(0.0, sigma)
 
 
+def _normalize_semver(version: Optional[str], fallback: str) -> str:
+    text = str(version or "").strip()
+    if not text:
+        return fallback
+    base = text.lstrip("v").split("-")[0].split("+")[0]
+    if re.match(r"^\d+\.\d+\.\d+$", base):
+        return base
+    return text
+
+
+def _version_for_component(version: Optional[str], fallback: str) -> str:
+    return _normalize_semver(version, fallback)
+
+
+def _latest_console_fw_version() -> str:
+    try:
+        from openlifu_sdk.util.firmware import get_console_firmware_version
+
+        return _normalize_semver(get_console_firmware_version(), _FALLBACK_CONSOLE_FW_VERSION)
+    except Exception:
+        logger.debug("Falling back to default simulated console firmware version", exc_info=True)
+        return _FALLBACK_CONSOLE_FW_VERSION
+
+
+def _latest_tx_fw_version() -> str:
+    try:
+        from openlifu_sdk.util.firmware import get_transmitter_firmware_version
+
+        return _normalize_semver(get_transmitter_firmware_version(), _FALLBACK_TX_FW_VERSION)
+    except Exception:
+        logger.debug("Falling back to default simulated transmitter firmware version", exc_info=True)
+        return _FALLBACK_TX_FW_VERSION
+
+
+def _sim_updated_timestamp() -> str:
+    return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
+def _read_version_from_firmware_image(package_file: str, fallback: str) -> str:
+    try:
+        from openlifu_sdk.util.firmware import get_firmware_version
+
+        return _normalize_semver(get_firmware_version(package_file), fallback)
+    except Exception:
+        logger.debug("Could not parse firmware version from %s", package_file, exc_info=True)
+        return fallback
+
+
 # =============================================================================
 # Per-module thermal model
 # =============================================================================
@@ -140,8 +193,12 @@ class SimulatedTxDevice:
     the emitter for unsolicited STATUS frames during sonication.
     """
 
-    def __init__(self, num_modules: int = 1):
+    def __init__(self, num_modules: int = 1, firmware_version: Optional[str] = None):
         self.num_modules = max(1, int(num_modules))
+        self._firmware_versions = [
+            _version_for_component(firmware_version, _latest_tx_fw_version())
+            for _ in range(self.num_modules)
+        ]
         self.signal_connected = OWSignal()
         self.signal_disconnected = OWSignal()
         self.signal_data_received = OWSignal()
@@ -171,9 +228,9 @@ def _default_user_config(self, idx: int) -> dict:
             "hwid": "ABCDEFGH",
             "freq": 400,
             "hw_ver": "SIM",
-            "fw_ver": "2.0.5",
+            "fw_ver": self._firmware_versions[idx],
             "sdk_ver": "1.0.7",
-            "updated": "2026-05-12 08:00:41",
+            "updated": _sim_updated_timestamp(),
             "module": {
                 "id": "txm_400_sim-400k-01",
                 "name": "TXM 400kHz (S/N SIMULATED-400K-01)",
@@ -241,7 +298,7 @@ def get_ambient_temperature(self, module: int = 0) -> float:
         return self._modules[module].read_ambient()
 
     def get_version(self, module: int = 0) -> str:
-        return "sim-1.0.7"
+        return f"v{self._firmware_versions[module]}"
 
     def get_hardware_id(self, module: int = 0, raw_hex: bool = False) -> str:
         return f"{0xA0A1A2A3A4A5A6A7B0B1B2B3B4B5B6B7 + module:032X}"
@@ -273,8 +330,14 @@ def apply_simulated_transducer(self, arr) -> None:
         modules_list = list(getattr(arr, "modules", []) or [])
         n = max(1, len(modules_list))
         if n != self.num_modules:
+            current = list(self._firmware_versions)
             self.num_modules = n
             self._modules = [_ModuleThermal(i) for i in range(n)]
+            default_fw = current[0] if current else _latest_tx_fw_version()
+            self._firmware_versions = [
+                current[i] if i < len(current) else default_fw
+                for i in range(n)
+            ]
             self._user_configs = [self._default_user_config(i) for i in range(n)]
         for i, m in enumerate(modules_list):
             cfg = self._user_configs[i]
@@ -390,9 +453,24 @@ def echo(self, echo_data: bytes, module: int = 0):
     def soft_reset(self, module: Optional[int] = None):
         return True
 
-    def update_firmware(self, *args, **kwargs):
-        # Verification tests / FW updater aren't in the simulator scope.
-        raise NotImplementedError("Firmware update not supported in simulation mode")
+    def update_firmware(self, module: int = 0, package_file: Optional[str] = None,
+                        progress_callback=None, firmware_version: Optional[str] = None,
+                        **_kwargs):
+        if module < 0 or module >= self.num_modules:
+            raise ValueError(f"Module index out of range: {module}")
+        target = _latest_tx_fw_version()
+        if firmware_version is not None:
+            target = _version_for_component(firmware_version, target)
+        elif package_file:
+            target = _read_version_from_firmware_image(package_file, target)
+        if progress_callback is not None:
+            progress_callback(0, 1, "simulated-update")
+        self._firmware_versions[module] = target
+        self._user_configs[module]["fw_ver"] = target
+        self._user_configs[module]["updated"] = _sim_updated_timestamp()
+        if progress_callback is not None:
+            progress_callback(1, 1, "simulated-update")
+        return True
 
     def close(self):
         self._connected = False
@@ -412,13 +490,14 @@ class SimulatedHVController:
     """Implements every attribute / method that a connector calls on
     ``interface.hvcontroller``."""
 
-    def __init__(self):
+    def __init__(self, firmware_version: Optional[str] = None):
         self.signal_connected = OWSignal()
         self.signal_disconnected = OWSignal()
         self.signal_data_received = OWSignal()
         self.signal_error = OWSignal()
 
         self._connected = False
+        self._firmware_version = _version_for_component(firmware_version, _latest_console_fw_version())
         self._hv_on = False
         self._v12_on = True
         self._voltage_setpoint = 0.0
@@ -465,7 +544,23 @@ def get_12v_status(self) -> bool:
         return self._v12_on
 
     def get_version(self) -> str:
-        return "sim-1.0.7"
+        return f"v{self._firmware_version}"
+
+    def update_firmware(self, package_file: Optional[str] = None,
+                        progress_callback=None,
+                        firmware_version: Optional[str] = None,
+                        **_kwargs):
+        target = _latest_console_fw_version()
+        if firmware_version is not None:
+            target = _version_for_component(firmware_version, target)
+        elif package_file:
+            target = _read_version_from_firmware_image(package_file, target)
+        if progress_callback is not None:
+            progress_callback(0, 1, "simulated-update")
+        self._firmware_version = target
+        if progress_callback is not None:
+            progress_callback(1, 1, "simulated-update")
+        return True
 
     def get_hardware_id(self, raw_hex: bool = False) -> str:
         return "C0C1C2C3C4C5C6C7D0D1D2D3D4D5D6D7"
@@ -742,6 +837,9 @@ class SimulatedLIFUInterface(QObject):
 
     def __init__(self, num_modules: int = 1,
                  transducer=None,
+                 tx_firmware_version: Optional[str] = None,
+                 hv_firmware_version: Optional[str] = None,
+                 firmware_version: Optional[str] = None,
                  voltage_table_selection: Optional[str] = None,
                  **_unused):
         # When a transducer (array) is supplied, derive num_modules from it
@@ -751,8 +849,10 @@ def __init__(self, num_modules: int = 1,
             if modules_attr is not None:
                 num_modules = max(1, len(list(modules_attr)))
         super().__init__()
-        self.txdevice = SimulatedTxDevice(num_modules=num_modules)
-        self.hvcontroller = SimulatedHVController()
+        tx_version = tx_firmware_version if tx_firmware_version is not None else firmware_version
+        hv_version = hv_firmware_version if hv_firmware_version is not None else firmware_version
+        self.txdevice = SimulatedTxDevice(num_modules=num_modules, firmware_version=tx_version)
+        self.hvcontroller = SimulatedHVController(firmware_version=hv_version)
         self.status = LIFUInterfaceStatus.STATUS_SYS_OFF
         self._engine: Optional[_SimulatedRunEngine] = None
         self.voltage_table_selection = voltage_table_selection

src/openlifu_sdk/util/firmware.py

@@ -103,6 +103,11 @@ def get_transmitter_firmware_version() -> str:
     return _get_firmware_version(get_transmitter_firmware_path())
 
 
+def get_firmware_version(path: Path | str) -> str:
+    """Version of a firmware image at *path*."""
+    return _get_firmware_version(path)
+
+
 __all__ = [
     "CONSOLE_FIRMWARE_FILENAME",
     "CONSOLE_FIRMWARE_URL",
@@ -111,6 +116,7 @@ def get_transmitter_firmware_version() -> str:
     "TRANSMITTER_FIRMWARE_URL",
     "get_console_firmware_path",
     "get_console_firmware_version",
+    "get_firmware_version",
     "get_transmitter_firmware_path",
     "get_transmitter_firmware_version",
 ]

unit-test/test_simulated_interface_firmware.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from pathlib import Path
+
+_REPO_ROOT = os.path.join(os.path.dirname(__file__), "..")
+_SRC = os.path.join(_REPO_ROOT, "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from openlifu_sdk.ui.simulated_interface import SimulatedLIFUInterface
+from openlifu_sdk.util.firmware import (
+    get_console_firmware_version,
+    get_transmitter_firmware_version,
+)
+
+
+class TestSimulatedInterfaceFirmware(unittest.TestCase):
+    def setUp(self):
+        self._objs = []
+
+    def tearDown(self):
+        for obj in self._objs:
+            try:
+                obj.close()
+            except Exception:
+                pass
+
+    def _track(self, obj):
+        self._objs.append(obj)
+        return obj
+
+    def test_default_versions_follow_bundled_firmware(self):
+        iface = self._track(SimulatedLIFUInterface())
+        expected_console = get_console_firmware_version()
+        expected_tx = get_transmitter_firmware_version()
+        self.assertEqual(iface.hvcontroller.get_version(), f"v{expected_console}")
+        self.assertEqual(iface.txdevice.get_version(), f"v{expected_tx}")
+        cfg = iface.txdevice.read_config().json_data
+        self.assertEqual(cfg["fw_ver"], expected_tx)
+
+    def test_constructor_supports_version_overrides(self):
+        iface = self._track(
+            SimulatedLIFUInterface(
+                tx_firmware_version="9.8.7",
+                hv_firmware_version="1.2.3",
+            )
+        )
+        self.assertEqual(iface.txdevice.get_version(), "v9.8.7")
+        self.assertEqual(iface.hvcontroller.get_version(), "v1.2.3")
+        self.assertEqual(iface.txdevice.read_config().json_data["fw_ver"], "9.8.7")
+
+        shared = self._track(SimulatedLIFUInterface(firmware_version="4.5.6"))
+        self.assertEqual(shared.txdevice.get_version(), "v4.5.6")
+        self.assertEqual(shared.hvcontroller.get_version(), "v4.5.6")
+
+    def test_simulated_firmware_update_changes_reported_version(self):
+        iface = self._track(SimulatedLIFUInterface(firmware_version="1.0.0"))
+        tx_progress = []
+        hv_progress = []
+
+        self.assertTrue(
+            iface.txdevice.update_firmware(
+                module=0,
+                firmware_version="2.3.4",
+                progress_callback=lambda done, total, label: tx_progress.append((done, total, label)),
+            )
+        )
+        self.assertEqual(iface.txdevice.get_version(), "v2.3.4")
+        self.assertEqual(iface.txdevice.read_config().json_data["fw_ver"], "2.3.4")
+        self.assertEqual(tx_progress, [(0, 1, "simulated-update"), (1, 1, "simulated-update")])
+
+        self.assertTrue(
+            iface.hvcontroller.update_firmware(
+                firmware_version="3.2.1",
+                progress_callback=lambda done, total, label: hv_progress.append((done, total, label)),
+            )
+        )
+        self.assertEqual(iface.hvcontroller.get_version(), "v3.2.1")
+        self.assertEqual(hv_progress, [(0, 1, "simulated-update"), (1, 1, "simulated-update")])
+
+    def test_update_without_args_applies_latest_bundled_versions(self):
+        iface = self._track(SimulatedLIFUInterface(firmware_version="0.0.1"))
+        iface.txdevice.update_firmware()
+        iface.hvcontroller.update_firmware()
+        self.assertEqual(iface.txdevice.get_version(), f"v{get_transmitter_firmware_version()}")
+        self.assertEqual(iface.hvcontroller.get_version(), f"v{get_console_firmware_version()}")
+
+    def test_progress_callback_exceptions_are_propagated(self):
+        iface = self._track(SimulatedLIFUInterface())
+
+        def _boom(*_args):
+            raise RuntimeError("callback-failed")
+
+        with self.assertRaises(RuntimeError):
+            iface.txdevice.update_firmware(progress_callback=_boom)
+        with self.assertRaises(RuntimeError):
+            iface.hvcontroller.update_firmware(progress_callback=_boom)
+
+    def test_simulated_update_can_read_version_from_firmware_file(self):
+        iface = self._track(SimulatedLIFUInterface(firmware_version="0.0.1"))
+        fw_dir = Path(_SRC) / "openlifu_sdk" / "firmware"
+        tx_fw = fw_dir / "openlifu-transmitter-fw-signed.bin"
+        hv_fw = fw_dir / "openlifu-console-fw-signed.bin"
+
+        iface.txdevice.update_firmware(module=0, package_file=str(tx_fw))
+        iface.hvcontroller.update_firmware(package_file=str(hv_fw))
+
+        self.assertEqual(iface.txdevice.get_version(), f"v{get_transmitter_firmware_version()}")
+        self.assertEqual(iface.hvcontroller.get_version(), f"v{get_console_firmware_version()}")
+
+
+if __name__ == "__main__":
+    unittest.main()