diff --git a/src/openlifu_sdk/ui/simulated_interface.py b/src/openlifu_sdk/ui/simulated_interface.py index 3e330ad..5901c8b 100644 --- a/src/openlifu_sdk/ui/simulated_interface.py +++ b/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 diff --git a/src/openlifu_sdk/util/firmware.py b/src/openlifu_sdk/util/firmware.py index 8624748..d11f52b 100644 --- a/src/openlifu_sdk/util/firmware.py +++ b/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", ] diff --git a/unit-test/test_simulated_interface_firmware.py b/unit-test/test_simulated_interface_firmware.py new file mode 100644 index 0000000..d450f36 --- /dev/null +++ b/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()