Fixes #432

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: 2 people

src/openlifu/io/LIFUDFU.py

@@ -13,6 +13,7 @@
 import logging
 import struct
 import time
+import unittest
 from typing import TYPE_CHECKING, Callable
 
 from openlifu.io.LIFUConfig import OW_ERROR, OW_I2C_PASSTHRU
@@ -152,6 +153,134 @@ def parse_signed_package(pkg: bytes) -> dict:
     }
 
 
+# ---------------------------------------------------------------------------
+# Internal tests for DFU package parsing / CRC (deterministic, no hardware)
+# ---------------------------------------------------------------------------
+
+def _build_synthetic_signed_package(
+    fw: bytes,
+    meta: bytes,
+    fw_address: int = 0x08000000,
+    meta_address: int = 0x08008000,
+) -> bytes:
+    """Construct a minimal, self-consistent signed DFU package for testing.
+
+    This uses the module's own header layout/CRC implementation so that tests
+    validate :func:`stm32_crc32` and :func:`parse_signed_package` end to end.
+    """
+    hdr_size = struct.calcsize(_PKG_HDR_FULL)
+    payload = fw + meta
+    fw_len = len(fw)
+    meta_len = len(meta)
+
+    payload_crc = stm32_crc32(payload)
+
+    # First pack with a placeholder header CRC so we can compute the real one.
+    header_crc_placeholder = 0
+    header = struct.pack(
+        _PKG_HDR_FULL,
+        _PKG_MAGIC,
+        _PKG_VERSION,
+        hdr_size,
+        fw_address,
+        fw_len,
+        meta_address,
+        meta_len,
+        payload_crc,
+        header_crc_placeholder,
+    )
+
+    header_crc = stm32_crc32(header[:-4])
+    header = struct.pack(
+        _PKG_HDR_FULL,
+        _PKG_MAGIC,
+        _PKG_VERSION,
+        hdr_size,
+        fw_address,
+        fw_len,
+        meta_address,
+        meta_len,
+        payload_crc,
+        header_crc,
+    )
+
+    return header + payload
+
+
+class TestSignedPackage(unittest.TestCase):
+    """Unit tests for :func:`stm32_crc32` and :func:`parse_signed_package`.
+
+    These tests are deterministic and require no hardware; they can be run by
+    any standard Python test runner to guard against regressions that might
+    otherwise risk bricking devices during DFU.
+    """
+
+    def test_parse_signed_package_valid(self) -> None:
+        fw = b"\x01\x02\x03\x04"
+        meta = b"\xAA\xBB"
+        fw_addr = 0x08001000
+        meta_addr = 0x08009000
+
+        pkg = _build_synthetic_signed_package(
+            fw=fw,
+            meta=meta,
+            fw_address=fw_addr,
+            meta_address=meta_addr,
+        )
+
+        parsed = parse_signed_package(pkg)
+
+        assert parsed["fw_address"] == fw_addr
+        assert parsed["meta_address"] == meta_addr
+        assert parsed["fw"] == fw
+        assert parsed["meta"] == meta
+
+    def test_parse_signed_package_header_crc_mismatch(self) -> None:
+        """Corrupt the header so header CRC verification fails."""
+        fw = b"\x10\x20"
+        meta = b"\x30"
+        pkg = _build_synthetic_signed_package(fw=fw, meta=meta)
+
+        # Flip a bit inside the header (but keep magic/version/size plausible).
+        pkg_bytes = bytearray(pkg)
+        if len(pkg_bytes) < 8:
+            self.skipTest("synthetic package unexpectedly small")
+        pkg_bytes[4] ^= 0x01
+        corrupted = bytes(pkg_bytes)
+
+        exc_msg = None
+        try:
+            parse_signed_package(corrupted)
+            raise AssertionError("Expected ValueError was not raised")
+        except ValueError as exc:
+            exc_msg = str(exc)
+        assert exc_msg is not None
+        assert "header CRC mismatch" in exc_msg
+
+    def test_parse_signed_package_payload_crc_mismatch(self) -> None:
+        """Corrupt the payload so payload CRC verification fails."""
+        fw = b"\xDE\xAD\xBE\xEF"
+        meta = b"\x00\x01"
+        pkg = _build_synthetic_signed_package(fw=fw, meta=meta)
+
+        hdr_size = struct.calcsize(_PKG_HDR_FULL)
+        pkg_bytes = bytearray(pkg)
+        # Flip a bit in the first payload byte (after the header).
+        if len(pkg_bytes) <= hdr_size:
+            self.skipTest("synthetic package unexpectedly small")
+        pkg_bytes[hdr_size] ^= 0x01
+        corrupted = bytes(pkg_bytes)
+
+        exc_msg = None
+        try:
+            parse_signed_package(corrupted)
+            raise AssertionError("Expected ValueError was not raised")
+        except ValueError as exc:
+            exc_msg = str(exc)
+        assert exc_msg is not None
+        assert "payload CRC mismatch" in exc_msg
+
+
 # ---------------------------------------------------------------------------
 # USB DFU client  (module 0)
 # ---------------------------------------------------------------------------
@@ -414,7 +543,6 @@ def __init__(self, uart: LIFUUart,
                  write_read_delay_s: float = 0.005):
         self._uart = uart
         self._addr = i2c_addr
-        self._wr_delay = write_read_delay_s
 
     # --- low-level transport primitives ---
 
@@ -437,11 +565,14 @@ def _write(self, payload: bytes) -> None:
 
     def _exchange(self, payload: bytes, read_len: int,
                   pre_read_delay_s: float | None = None) -> bytes:
-        """Write *payload* to the I2C slave, wait, then read *read_len* bytes back.
-
-        The firmware inserts a fixed 5 ms gap between write and read.
-        An optional extra host-side delay can be added via *pre_read_delay_s*
-        (not usually needed).
+        """Write *payload* to the I2C slave and read *read_len* bytes back.
+
+        The firmware executes a combined write+read transaction and inserts a
+        fixed 5 ms gap between the write and read phases internally.
+        The optional *pre_read_delay_s* parameter adds an extra host-side delay
+        **before** issuing the passthrough transaction (i.e. before the
+        firmware performs the write+read). This does *not* change the internal
+        5 ms gap handled by the firmware and is rarely needed.
         """
         if pre_read_delay_s and pre_read_delay_s > 0:
             time.sleep(pre_read_delay_s)
@@ -784,16 +915,35 @@ def update_module(self,
                 "Verifying I2C DFU entry (module %d, addr=0x%02X via master)...",
                 module, i2c_addr,
             )
-            try:
-                bl_version = self.get_bootloader_version_i2c(i2c_addr=i2c_addr)
-            except (RuntimeError, TimeoutError) as e:
-                raise RuntimeError(
-                    f"Module {module} did not enter I2C DFU mode at "
-                    f"0x{i2c_addr:02X}: {e}"
-                ) from e
+            start_time = time.time()
+            bl_version = None
+            last_error: Exception | None = None
+            while True:
+                elapsed = time.time() - start_time
+                if elapsed >= dfu_enum_timeout_s:
+                    break
+                try:
+                    candidate = self.get_bootloader_version_i2c(i2c_addr=i2c_addr)
+                    if candidate:
+                        bl_version = candidate
+                        break
+                    # Treat empty version string as a failure worth retrying.
+                    last_error = RuntimeError(
+                        "I2C DFU bootloader returned an empty version string"
+                    )
+                except (RuntimeError, TimeoutError) as e:
+                    last_error = e
+                # Small delay before retrying to avoid busy-waiting.
+                time.sleep(0.2)
             if not bl_version:
+                if last_error is not None:
+                    raise RuntimeError(
+                        f"Module {module} did not enter I2C DFU mode at "
+                        f"0x{i2c_addr:02X} within {dfu_enum_timeout_s}s: {last_error}"
+                    ) from last_error
                 raise RuntimeError(
-                    f"Module {module} I2C DFU bootloader returned an empty version string"
+                    f"Module {module} did not enter I2C DFU mode at "
+                    f"0x{i2c_addr:02X} within {dfu_enum_timeout_s}s"
                 )
             logger.info("I2C DFU bootloader version: %s", bl_version)
             self.program_i2c(

src/openlifu/io/LIFUTXDevice.py

@@ -397,7 +397,7 @@ def get_hardware_id(self, module:int=0) -> str:
             logger.error("Unexpected error during process: %s", e)
             raise  # Re-raise the exception for the caller to handle
 
-    def read_config(self, module:int=0) -> Optional[LifuUserConfig]:
+    def read_config(self, module:int=0) -> LifuUserConfig | None:
         """
         Read the user configuration from device flash.
 
@@ -450,7 +450,7 @@ def read_config(self, module:int=0) -> Optional[LifuUserConfig]:
             logger.exception("Unexpected error reading config")
             raise
 
-    def write_config(self, config: LifuUserConfig, module:int=0) -> Optional[LifuUserConfig]:
+    def write_config(self, config: LifuUserConfig, module:int=0) -> LifuUserConfig | None:
         """
         Write user configuration to device flash.
 
@@ -517,7 +517,7 @@ def write_config(self, config: LifuUserConfig, module:int=0) -> Optional[LifuUse
             logger.exception("Unexpected error writing config")
             raise
 
-    def write_config_json(self, json_str: str, module:int=0) -> Optional[LifuUserConfig]:
+    def write_config_json(self, json_str: str, module:int=0) -> LifuUserConfig | None:
         """
         Write user configuration from a JSON string.
 
@@ -540,6 +540,13 @@ def write_config_json(self, json_str: str, module:int=0) -> Optional[LifuUserCon
             return self.write_config(module=module, config=config)
         except json.JSONDecodeError as e:
             logger.error(f"Invalid JSON: {e}")
+            raise ValueError(f"Invalid JSON: {e}") from e
+        except ValueError as v:
+            logger.error("ValueError: %s", v)
+            raise
+        except (OSError, RuntimeError, AttributeError) as e:
+            logger.exception("Unexpected error writing config from JSON")
+            raise
 
     def get_temperature(self, module:int=1) -> float:
         """
@@ -935,7 +942,7 @@ def enter_dfu(self, module:int=0) -> bool:
             if not self.uart.is_connected():
                 raise ValueError("TX Device not connected")
 
-            r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_DFU, addr=module)
+            r = self.uart.send_packet(id=None, packetType=OW_CMD, command=OW_CMD_DFU, addr=module)
             self.uart.clear_buffer()
             if r is None:
                 # Device disconnected immediately after reset — expected for DFU entry
@@ -982,7 +989,7 @@ def async_mode(self, enable: bool | None = None) -> bool:
             else:
                 payload = None
 
-            r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_ASYNC, addr=0, data=payload)
+            r = self.uart.send_packet(id=None, packetType=OW_CMD, command=OW_CMD_ASYNC, addr=0, data=payload)
             self.uart.clear_buffer()
             # r.print_packet()
             if r.packet_type == OW_ERROR:
@@ -1329,11 +1336,10 @@ def write_block(self, identifier: int, start_address: int, reg_values: List[int]
             raise  # Re-raise the exception for the caller to handle
 
         except Exception as e:
-            logger.error("Unexpected error during process: %s", e)
-            raise  # Re-raise the exception for the caller to handleected error in write_block: {e}")
-            return False
+            logger.error("Unexpected error in write_block: %s", e)
+            raise  # Re-raise the exception for the caller to handle
 
-    def read_block(self, identifier: int, start_address: int, count: int) -> Optional[List[int]]:
+    def read_block(self, identifier: int, start_address: int, count: int) -> List[int] | None:
         """
         Read a block of consecutive register values from the TX device.
 

src/openlifu/io/LIFUUserConfig.py

@@ -3,14 +3,15 @@
 import json
 import logging
 import struct
+import zlib
 from dataclasses import dataclass
 from typing import Any, Dict
 
 logger = logging.getLogger(__name__)
 
 # Constants from C code
 LIFU_MAGIC = 0x4C494655  # 'LIFU'
-LIFU_VER = 0x00010002    # v1.0.0
+LIFU_VER = 0x00010002    # v1.0.2
 
 @dataclass
 class LifuUserConfigHeader:
@@ -139,6 +140,9 @@ def to_wire_bytes(self) -> bytes:
         # Update header with JSON length
         self.header.json_len = len(json_bytes)
 
+        # Update header CRC based on JSON payload (16-bit from CRC32)
+        self.header.crc = zlib.crc32(json_bytes) & 0xFFFF
+
         # Build wire format
         return self.header.to_bytes() + json_bytes