Add x20 parser logic

README.md

@@ -1,6 +1,9 @@
 # MySmartBike BLE Integration for Home Assistant
 
-[![GitHub release](https://img.shields.io/github/release/renenulschde/ha-mysmartbike-ble.svg)](https://github.com/renenulschde/ha-mysmartbike-ble/releases)
+[![GitHub release](https://img.shields.io/github/release/renenulschde/ha-mysmartbike_ble.svg)](https://github.com/renenulschde/ha-mysmartbike_ble/releases)
+![](https://img.shields.io/github/downloads/renenulschde/ha-mysmartbike_ble/latest/total) ![Validate with HACS](https://github.com/ReneNulschDE/mbapi2020/workflows/Validate%20with%20HACS/badge.svg) 
+![HassFest tests](https://github.com/renenulschde/ha-mysmartbike_ble/workflows/Validate%20with%20hassfest/badge.svg) 
+
 
 Home Assistant custom component for E-Bikes with Mahle SmartBike systems (X25, X35+, ebikemotion, ...) via Bluetooth Low Energy (BLE).
 
@@ -12,8 +15,10 @@ This integration has been developed and tested with:
 
 | Brand | Model | Status |
 |-------|-------|--------|
+| Orbea | Vibe | Fully tested |
 | Schindelhauer | Arthur IX | Fully tested |
 
+
 **Your bike not listed?** If you have an E-Bike that uses the MySmartBike app (or ebikemotion app), it will likely work with this integration. Please open an issue to report compatibility!
 
 ## Features
@@ -85,10 +90,10 @@ The integration is configured through the Home Assistant UI:
 1. Go to **Settings** → **Devices & Services**
 2. Click **+ Add Integration**
 3. Search for **MySmartBike BLE**
-4. Select your iWoc device from the list
+4. Select your iWoc/HUS device from the list
 5. Click **Submit**
 
-The integration will automatically discover iWoc devices in range via Bluetooth.
+The integration will automatically discover iWoc and HUS devices in range via Bluetooth.
 
 ## Troubleshooting
 
@@ -96,7 +101,7 @@ The integration will automatically discover iWoc devices in range via Bluetooth.
 
 - Make sure your E-Bike is turned on and in range
 - Check that Bluetooth is enabled on your Home Assistant host
-- Verify that the device name starts with "iWoc" (please report other device names)
+- Verify that the device name starts with "iWoc" or "HUS" (please report other device names)
 
 ### Connection issues
 

custom_components/mysmartbike_ble/__init__.py

@@ -2,7 +2,6 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
 
 from homeassistant.components import bluetooth
 from homeassistant.config_entries import ConfigEntry
@@ -20,25 +19,18 @@ PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR, Platform.S
 
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up MySmartBike BLE from a config entry."""
-    _LOGGER.debug("Setting up MySmartBike BLE integration for entry_id: %s", entry.entry_id)
-
     address = entry.data[CONF_DEVICE_ADDRESS]
-    _LOGGER.debug("Device address from config: %s", address)
 
     # Get BLE device
-    _LOGGER.debug("Looking up BLE device with address: %s", address)
     ble_device = bluetooth.async_ble_device_from_address(hass, address, connectable=True)
     if not ble_device:
         # Log warning only once per config entry
-        # Use hass.data for warning flag as it's separate from coordinator runtime_data
         hass.data.setdefault(DOMAIN, {})
         warning_key = f"warned_{entry.entry_id}"
 
         if not hass.data[DOMAIN].get(warning_key):
             _LOGGER.warning(
-                "MySmartBike device with address %s not found. "
-                "Make sure the bike is powered on and in range. "
-                "Home Assistant will retry automatically",
+                "MySmartBike device %s not found - ensure bike is powered on and in range",
                 address
             )
             hass.data[DOMAIN][warning_key] = True
@@ -48,58 +40,27 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     if DOMAIN in hass.data:
         hass.data[DOMAIN].pop(f"warned_{entry.entry_id}", None)
 
-    _LOGGER.debug("Found BLE device: %s", ble_device)
-
-    # Create coordinator
-    _LOGGER.debug("Creating coordinator for device %s", address)
+    # Create and initialize coordinator
     coordinator = MySmartBikeCoordinator(hass, ble_device, entry)
-
-    # Perform first refresh
-    _LOGGER.debug("Performing first coordinator refresh")
     await coordinator.async_config_entry_first_refresh()
-    _LOGGER.debug(
-        "First refresh completed - coordinator state: is_connected=%s, manual_disconnect=%s",
-        coordinator.is_connected,
-        coordinator._manual_disconnect,
-    )
 
-    # Store coordinator in runtime_data
     entry.runtime_data = coordinator
-    _LOGGER.debug("Coordinator stored in entry.runtime_data")
-
-    # Forward entry setup to platforms
-    _LOGGER.debug("Forwarding entry setup to platforms: %s", PLATFORMS)
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
-    _LOGGER.debug("Platform setup completed")
 
-    _LOGGER.debug("MySmartBike BLE integration setup completed successfully for entry_id: %s", entry.entry_id)
+    _LOGGER.debug("MySmartBike BLE setup completed for %s", address)
     return True
 
 
 async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Unload a config entry."""
-    _LOGGER.debug("Unloading MySmartBike BLE integration for entry_id: %s", entry.entry_id)
-
-    # Unload platforms
-    _LOGGER.debug("Unloading platforms: %s", PLATFORMS)
     unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
-    _LOGGER.debug("Platform unload result: %s", unload_ok)
 
     if unload_ok:
         coordinator: MySmartBikeCoordinator = entry.runtime_data
-        _LOGGER.debug(
-            "Coordinator retrieved from runtime_data - state: is_connected=%s, manual_disconnect=%s",
-            coordinator.is_connected,
-            coordinator._manual_disconnect,
-        )
         await coordinator.async_shutdown()
-        _LOGGER.debug("Coordinator shutdown completed")
 
         # Clean up warning flag from hass.data
         if DOMAIN in hass.data:
             hass.data[DOMAIN].pop(f"warned_{entry.entry_id}", None)
-    else:
-        _LOGGER.warning("Platform unload was not successful")
 
-    _LOGGER.debug("MySmartBike BLE integration unload completed for entry_id: %s (result: %s)", entry.entry_id, unload_ok)
     return unload_ok

custom_components/mysmartbike_ble/config_flow.py

@@ -99,8 +99,7 @@ class MySmartBikeConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
             ):
                 continue
 
-            # Check if device name starts with "iWoc"
-            if discovery_info.name and discovery_info.name.startswith("iWoc"):
+            if discovery_info.name and discovery_info.name.startswith(("iWoc", "HUS")):
                 self._discovered_devices[discovery_info.address] = discovery_info
 
         if not self._discovered_devices:

custom_components/mysmartbike_ble/coordinator.py

@@ -59,13 +59,6 @@ class MySmartBikeCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._is_connected = False
         self._notify_task: asyncio.Task | None = None
         self._manual_disconnect = False  # Track if user manually disconnected
-        _LOGGER.debug(
-            "Coordinator initialized: address=%s, is_connected=%s, manual_disconnect=%s, scan_interval=%s",
-            ble_device.address,
-            self._is_connected,
-            self._manual_disconnect,
-            SCAN_INTERVAL,
-        )
 
     @property
     def address(self) -> str:
@@ -75,11 +68,6 @@ class MySmartBikeCoordinator(DataUpdateCoordinator[dict[str, Any]]):
     @property
     def is_connected(self) -> bool:
         """Return connection status."""
-        _LOGGER.debug(
-            "Coordinator.is_connected property called: returning %s (manual_disconnect: %s)",
-            self._is_connected,
-            self._manual_disconnect,
-        )
         return self._is_connected
 
     @property
@@ -92,149 +80,77 @@ class MySmartBikeCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         """Return the protocol version if available."""
         return self._parser.protocol_version
 
-    async def async_disconnect(self) -> None:
-        """Disconnect from the device (user initiated)."""
-        _LOGGER.debug(
-            "Coordinator.async_disconnect called for %s (user initiated) - current state: is_connected=%s, manual_disconnect=%s, client=%s",
-            self._ble_device.address,
-            self._is_connected,
-            self._manual_disconnect,
-            self._client is not None,
-        )
+    async def _cleanup_client(self, send_close: bool = True, wait_for_slot: bool = True) -> None:
+        """Clean up BLE client connection.
 
-        # Mark as manually disconnected to prevent auto-reconnect
-        self._manual_disconnect = True
-        _LOGGER.debug("Coordinator.async_disconnect: Set manual_disconnect=True")
+        Args:
+            send_close: Whether to send close message to bike before disconnecting.
+            wait_for_slot: Whether to wait for BLE connection slot release.
+        """
+        if not self._client:
+            return
 
-        if self._client:
-            _LOGGER.debug("Coordinator.async_disconnect: Client exists, cleaning up connection")
-            client_to_cleanup = self._client
-            self._client = None  # Clear reference immediately
-            self._is_connected = False
-
-            try:
-                # Only send close message if still connected
-                if client_to_cleanup.is_connected:
-                    # Send close message to bike before disconnecting
-                    _LOGGER.debug("Coordinator.async_disconnect: Sending close message ($D$I#@)")
-                    try:
-                        await client_to_cleanup.write_gatt_char(WRITE_UUID, CLOSE_MESSAGE)
-                        _LOGGER.debug("Coordinator.async_disconnect: Close message sent")
-                        await asyncio.sleep(0.5)
-                    except Exception as ex:
-                        _LOGGER.debug("Coordinator.async_disconnect: Error sending close message: %s", ex)
-
-                    # Stop notifications
-                    try:
-                        await client_to_cleanup.stop_notify(NOTIFY_UUID)
-                        _LOGGER.debug("Coordinator.async_disconnect: Stopped notifications")
-                    except Exception as ex:
-                        _LOGGER.debug("Coordinator.async_disconnect: Error stopping notifications: %s", ex)
-
-                    # Disconnect from device
-                    try:
-                        await client_to_cleanup.disconnect()
-                        _LOGGER.debug("Coordinator.async_disconnect: Disconnected from device")
-                    except Exception as ex:
-                        _LOGGER.debug("Coordinator.async_disconnect: Error during disconnect: %s", ex)
-                else:
-                    _LOGGER.debug("Coordinator.async_disconnect: Client exists but not connected, skipping disconnect")
-
-            except Exception as ex:
-                _LOGGER.debug("Coordinator.async_disconnect: Unexpected error during disconnect: %s", ex, exc_info=True)
-            finally:
-                # Force delete the client object to help garbage collection
-                del client_to_cleanup
-
-                # Give BLE adapter significant time to release connection slot
-                _LOGGER.debug("Coordinator.async_disconnect: Waiting for connection slot release (3 seconds)")
-                await asyncio.sleep(3.0)
-                _LOGGER.debug("Coordinator.async_disconnect: Cleaned up client (is_connected=%s)", self._is_connected)
-        else:
-            _LOGGER.debug("Coordinator.async_disconnect: No client to disconnect")
-            self._is_connected = False
-
-        _LOGGER.debug(
-            "Coordinator.async_disconnect completed - final state: is_connected=%s, manual_disconnect=%s",
-            self._is_connected,
-            self._manual_disconnect,
-        )
-
-    async def async_reconnect(self) -> None:
-        """Reconnect to the device (user initiated)."""
-        _LOGGER.debug(
-            "Coordinator.async_reconnect called for %s (user initiated) - current state: is_connected=%s, manual_disconnect=%s, client=%s",
-            self._ble_device.address,
-            self._is_connected,
-            self._manual_disconnect,
-            self._client is not None,
-        )
-
-        # Clean up any existing client first
-        if self._client:
-            _LOGGER.debug("Coordinator.async_reconnect: Found existing client, cleaning up first")
-            old_client = self._client
+        client = self._client
         self._client = None
         self._is_connected = False
 
         try:
-                if old_client.is_connected:
-                    await old_client.disconnect()
-                    _LOGGER.debug("Coordinator.async_reconnect: Disconnected existing client")
+            if client.is_connected:
+                if send_close:
+                    try:
+                        await client.write_gatt_char(WRITE_UUID, CLOSE_MESSAGE)
+                        await asyncio.sleep(0.5)
+                    except Exception:
+                        pass  # Ignore close message errors
+
+                try:
+                    await client.stop_notify(NOTIFY_UUID)
+                except Exception:
+                    pass  # Ignore notification stop errors
+
+                try:
+                    await client.disconnect()
                 except Exception as ex:
-                _LOGGER.debug("Coordinator.async_reconnect: Error disconnecting old client: %s", ex)
+                    _LOGGER.debug("Error during BLE disconnect: %s", ex)
+        except Exception as ex:
+            _LOGGER.debug("Unexpected error during client cleanup: %s", ex)
         finally:
-                del old_client
-                # Wait longer for connection slot to be released
-                _LOGGER.debug("Coordinator.async_reconnect: Waiting for connection slot release (3 seconds)")
-                await asyncio.sleep(3.0)
-                _LOGGER.debug("Coordinator.async_reconnect: Cleaned up old client and waited for slot release")
+            del client
+            if wait_for_slot:
+                await asyncio.sleep(3.0)  # Wait for BLE connection slot release
+
+    async def async_disconnect(self) -> None:
+        """Disconnect from the device (user initiated)."""
+        _LOGGER.debug("User-initiated disconnect for %s", self._ble_device.address)
+        self._manual_disconnect = True
+        await self._cleanup_client(send_close=True, wait_for_slot=True)
+
+    async def async_reconnect(self) -> None:
+        """Reconnect to the device (user initiated)."""
+        _LOGGER.debug("User-initiated reconnect for %s", self._ble_device.address)
+
+        # Clean up any existing client first
+        await self._cleanup_client(send_close=False, wait_for_slot=True)
 
         # Clear manual disconnect flag to allow auto-reconnect
         self._manual_disconnect = False
-        _LOGGER.debug("Coordinator.async_reconnect: Set manual_disconnect=False")
 
         try:
             await self._connect()
-            _LOGGER.debug(
-                "Coordinator.async_reconnect completed - final state: is_connected=%s, manual_disconnect=%s",
-                self._is_connected,
-                self._manual_disconnect,
-            )
         except Exception as ex:
-            # Only log as error if it's not a "device not reachable" issue
             error_str = str(ex).lower()
-            if "not reachable" in error_str or "turn on the bike" in error_str:
-                _LOGGER.debug("Coordinator.async_reconnect: Device not reachable, will retry later")
-            else:
-                _LOGGER.error("Coordinator.async_reconnect failed: %s", ex, exc_info=True)
+            if "not reachable" not in error_str and "turn on the bike" not in error_str:
+                _LOGGER.error("Reconnect failed: %s", ex)
             raise
 
     async def _async_update_data(self) -> dict[str, Any]:
         """Fetch data from the device."""
-        _LOGGER.debug(
-            "Coordinator._async_update_data called - current state: is_connected=%s, manual_disconnect=%s",
-            self._is_connected,
-            self._manual_disconnect,
-        )
-
-        # Don't auto-reconnect if user manually disconnected
+        # Auto-reconnect if not connected and not manually disconnected
         if not self._is_connected and not self._manual_disconnect:
-            _LOGGER.debug("Coordinator._async_update_data: Not connected and not manual disconnect, attempting auto-reconnect")
             try:
                 await self._connect()
-                _LOGGER.debug("Coordinator._async_update_data: Auto-reconnect successful (is_connected=%s)", self._is_connected)
-            except Exception as ex:
-                # Only log as warning if device is not reachable, otherwise debug
-                error_str = str(ex).lower()
-                if "not reachable" in error_str or "turn on the bike" in error_str:
-                    _LOGGER.debug("Coordinator._async_update_data: Auto-reconnect skipped - device not reachable")
-                else:
-                    _LOGGER.debug("Coordinator._async_update_data: Auto-reconnect failed: %s", ex)
-        elif not self._is_connected and self._manual_disconnect:
-            _LOGGER.debug("Coordinator._async_update_data: Not connected but manual_disconnect=True, skipping auto-reconnect")
-        else:
-            _LOGGER.debug("Coordinator._async_update_data: Already connected, no action needed")
+            except Exception:
+                pass  # Connection errors are logged in _connect()
 
         # Return current state from parser, ensure it's never None
         state = self._parser.state or {
@@ -246,106 +162,55 @@ class MySmartBikeCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         }
 
         # Add RSSI (signal strength) to state
-        # Get latest service info which contains current RSSI
         try:
             service_info = bluetooth.async_last_service_info(
                 self.hass, self._ble_device.address, connectable=True
             )
             state["rssi"] = service_info.rssi if service_info else None
-        except Exception as ex:
-            _LOGGER.debug("Could not get RSSI: %s", ex)
+        except Exception:
             state["rssi"] = None
 
-        _LOGGER.debug("Coordinator._async_update_data: Returning state (has_data=%s, rssi=%s)", self._parser.state is not None, state.get("rssi"))
         return state
 
     async def _connect(self) -> None:
         """Connect to the device and start notifications."""
-        _LOGGER.debug(
-            "Coordinator._connect: Attempting to connect to %s (current is_connected=%s, manual_disconnect=%s, client=%s)",
-            self._ble_device.address,
-            self._is_connected,
-            self._manual_disconnect,
-            self._client is not None,
-        )
-
         # Clean up any existing client before connecting
         if self._client:
-            _LOGGER.warning("Coordinator._connect: Client already exists, cleaning up before new connection")
-            old_client = self._client
-            self._client = None
-            try:
-                if old_client.is_connected:
-                    await old_client.disconnect()
-            except Exception as ex:
-                _LOGGER.debug("Coordinator._connect: Error cleaning up old client: %s", ex)
-            finally:
-                del old_client
-                _LOGGER.debug("Coordinator._connect: Waiting for connection slot release (3 seconds)")
-                await asyncio.sleep(3.0)
+            _LOGGER.debug("Cleaning up existing client before new connection")
+            await self._cleanup_client(send_close=False, wait_for_slot=True)
 
         try:
-            _LOGGER.debug("Coordinator._connect: Calling establish_connection for %s", self._ble_device.address)
-
             self._client = await establish_connection(
                 BleakClientWithServiceCache,
                 self._ble_device,
                 self._ble_device.address,
             )
 
-            _LOGGER.debug("Coordinator._connect: Successfully connected to %s, client=%s", self._ble_device.address, self._client)
-
-            # Start notifications first
-            _LOGGER.debug("Coordinator._connect: Starting notifications on UUID %s", NOTIFY_UUID)
+            # Start notifications and request device info
             await self._client.start_notify(NOTIFY_UUID, self._notification_handler)
-            _LOGGER.debug("Coordinator._connect: Started notifications successfully")
-
-            # Request VIN/serial number ($S$V#@)
-            _LOGGER.debug("Coordinator._connect: Requesting VIN/serial number")
             await self._client.write_gatt_char(WRITE_UUID, VIN_REQUEST_MESSAGE)
-            _LOGGER.debug("Coordinator._connect: VIN request sent")
-
-            # Small delay between requests
             await asyncio.sleep(0.2)
-
-            # Request protocol version ($S$P#@)
-            _LOGGER.debug("Coordinator._connect: Requesting protocol version")
             await self._client.write_gatt_char(WRITE_UUID, PROTOCOL_REQUEST_MESSAGE)
-            _LOGGER.debug("Coordinator._connect: Protocol request sent")
 
             self._is_connected = True
-            _LOGGER.debug("Coordinator._connect: Set is_connected=True")
+            _LOGGER.debug("Connected to %s", self._ble_device.address)
 
         except (BleakError, asyncio.TimeoutError) as ex:
             self._is_connected = False
-
-            # Check if error is due to device not being reachable (turned off)
             error_str = str(ex).lower()
+
             if "no longer reachable" in error_str or "out of connection slots" in error_str:
-                _LOGGER.warning(
-                    "Coordinator._connect: Device %s is not reachable or powered off. "
-                    "Turn on the bike to connect.",
-                    self._ble_device.address
-                )
-                raise UpdateFailed(
-                    f"Device {self._ble_device.address} is not reachable. "
-                    "Please turn on the bike."
-                ) from ex
+                _LOGGER.warning("Device %s not reachable - turn on the bike", self._ble_device.address)
+                raise UpdateFailed(f"Device {self._ble_device.address} is not reachable") from ex
             else:
-                _LOGGER.error(
-                    "Coordinator._connect: Failed to connect to device %s: %s (is_connected set to False)",
-                    self._ble_device.address,
-                    ex,
-                    exc_info=True,
-                )
+                _LOGGER.error("Failed to connect to %s: %s", self._ble_device.address, ex)
                 raise UpdateFailed(f"Failed to connect to device: {ex}") from ex
 
     def _notification_handler(self, sender: int, data: bytearray) -> None:
         """Handle notification data."""
-        _LOGGER.debug("Received notification from %s: %s", sender, data.hex())
-
         # Recognize message type before saving
         message_type = self._parser.recognize_message_type(bytes(data))
+        _LOGGER.debug("BLE notification [%s]: %s", message_type, data.hex())
 
         # Save BLE message to file if option is enabled (run in executor to avoid blocking)
         if self._entry.options.get(CONF_LOG_BLE_MESSAGES, False):
@@ -403,61 +268,10 @@ class MySmartBikeCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             with open(filepath, "a", encoding="utf-8") as f:
                 f.write(message_line)
 
-            _LOGGER.debug("Saved BLE message to: %s", filepath)
-
         except Exception as ex:
-            _LOGGER.error("Failed to save BLE message to file: %s", ex, exc_info=True)
+            _LOGGER.error("Failed to save BLE message to file: %s", ex)
 
     async def async_shutdown(self) -> None:
         """Shutdown the coordinator."""
-        _LOGGER.debug(
-            "Coordinator.async_shutdown called - current state: is_connected=%s, manual_disconnect=%s, client=%s",
-            self._is_connected,
-            self._manual_disconnect,
-            self._client is not None,
-        )
-
-        if self._client:
-            _LOGGER.debug("Coordinator.async_shutdown: Client exists, cleaning up connection")
-            client_to_cleanup = self._client
-            self._client = None
-            self._is_connected = False
-
-            try:
-                # Only send close message if still connected
-                if client_to_cleanup.is_connected:
-                    # Send close message to bike before disconnecting
-                    try:
-                        _LOGGER.debug("Coordinator.async_shutdown: Sending close message ($D$I#@)")
-                        await client_to_cleanup.write_gatt_char(WRITE_UUID, CLOSE_MESSAGE)
-                        _LOGGER.debug("Coordinator.async_shutdown: Close message sent")
-                        await asyncio.sleep(0.5)
-                    except Exception as ex:
-                        _LOGGER.debug("Coordinator.async_shutdown: Error sending close message: %s", ex)
-
-                    # Stop notifications
-                    try:
-                        await client_to_cleanup.stop_notify(NOTIFY_UUID)
-                        _LOGGER.debug("Coordinator.async_shutdown: Stopped notifications")
-                    except Exception as ex:
-                        _LOGGER.debug("Coordinator.async_shutdown: Error stopping notifications: %s", ex)
-
-                    # Disconnect from device
-                    try:
-                        await client_to_cleanup.disconnect()
-                        _LOGGER.debug("Coordinator.async_shutdown: Disconnected from device")
-                    except Exception as ex:
-                        _LOGGER.debug("Coordinator.async_shutdown: Error during disconnect: %s", ex)
-                else:
-                    _LOGGER.debug("Coordinator.async_shutdown: Client exists but not connected, skipping disconnect")
-
-            except Exception as ex:
-                _LOGGER.debug("Coordinator.async_shutdown: Unexpected error during shutdown: %s", ex, exc_info=True)
-            finally:
-                del client_to_cleanup
-                _LOGGER.debug("Coordinator.async_shutdown: Cleaned up client")
-        else:
-            _LOGGER.debug("Coordinator.async_shutdown: No client to clean up")
-            self._is_connected = False
-
-        _LOGGER.debug("Coordinator.async_shutdown completed")
+        _LOGGER.debug("Shutting down coordinator")
+        await self._cleanup_client(send_close=True, wait_for_slot=False)

custom_components/mysmartbike_ble/manifest.json

@@ -4,6 +4,9 @@
   "bluetooth": [
     {
       "local_name": "iWoc*"
+    },
+    {
+      "local_name": "HUS*"
     }
   ],
   "codeowners": [

custom_components/mysmartbike_ble/parsers.py

@@ -12,12 +12,25 @@ _LOGGER = logging.getLogger(__name__)
 
 
 def read16(data: bytes, offset: int) -> int:
-    """Read 16-bit value from data at offset (big-endian, as per Mahle protocol)."""
+    """Read 16-bit big-endian value from data at offset."""
     return ((data[offset] & 0xFF) << 8) | (data[offset + 1] & 0xFF)
 
 
+def read16_signed(data: bytes, offset: int) -> int:
+    """Read 16-bit big-endian value as signed int."""
+    value = read16(data, offset)
+    if value & 0x8000:
+        value -= 0x10000
+    return value
+
+
+def read_signed_byte(byte_val: int) -> int:
+    """Read byte as signed int."""
+    return byte_val - 256 if byte_val & 0x80 else byte_val
+
+
 def read24(data: bytes, offset: int) -> int:
-    """Read 24-bit value from data at offset (big-endian, as per Mahle protocol)."""
+    """Read 24-bit big-endian value from data at offset."""
     return (
         ((data[offset] & 0xFF) << 16)
         | ((data[offset + 1] & 0xFF) << 8)
@@ -26,7 +39,7 @@ def read24(data: bytes, offset: int) -> int:
 
 
 def read32(data: bytes, offset: int) -> int:
-    """Read 32-bit value from data at offset (big-endian, as per Mahle protocol)."""
+    """Read 32-bit big-endian value from data at offset."""
     return (
         ((data[offset] & 0xFF) << 24)
         | ((data[offset + 1] & 0xFF) << 16)
@@ -57,11 +70,15 @@ class BikeDataParser:
         self.protocol_version: Optional[str] = None
 
     def parse_battery_message(self, message: bytes) -> Optional[Dict[str, Any]]:
-        """Parse battery message and update state."""
-        if len(message) < BATTERY_MESSAGE_LENGTH:
+        """Parse battery frame; dispatch by length to the right layout."""
+        if len(message) == 20:
+            return self._parse_battery_x20(message)
+        if len(message) >= BATTERY_MESSAGE_LENGTH:
+            return self._parse_battery_ebm(message)
         return None
 
-        # Read values
+    def _parse_battery_ebm(self, message: bytes) -> Optional[Dict[str, Any]]:
+        """Parse 19-byte battery frame (X25 / X35+ / ebikemotion)."""
         voltage = read16(message, 5) / 10.0
         soc = read_unsigned_byte(message[7])
         temp_status = message[8]
@@ -69,66 +86,102 @@ class BikeDataParser:
         nominal_capacity = read16(message, 11) / 10.0
         remaining_wh = read16(message, 13) / 10.0
 
-        # Get battery number and cycles from combined field at offset 15
-        # Format: value = (battery_number * 10000) + cycles
-        # e.g., 10036 means battery 1, 36 cycles
         combined_raw = read16(message, 15) if len(message) >= 19 else None
         battery_number = (combined_raw // 10000) if combined_raw else 1
         cycles = (combined_raw % 10000) if combined_raw else None
 
-        # Construct battery data dictionary
         data = {
             "voltage": voltage,
             "soc": soc,
             "temperature": temp_status,
+            "temperature_mos": None,
             "current": current,
             "nominal_capacity": nominal_capacity,
             "remaining_wh": remaining_wh,
             "cycles": cycles,
+            "is_charging": False,
         }
+        self._store_battery(data, battery_number)
+        return data
 
-        # Handle secondary vs primary battery
+    def _parse_battery_x20(self, message: bytes) -> Optional[Dict[str, Any]]:
+        """Parse 20-byte battery frame (X20 / HUS-prefixed devices)."""
+        voltage = read16(message, 5) / 100.0
+        soc_raw = read_unsigned_byte(message[7])
+        # Bit 7 of the SOC byte signals charging on the newer firmware variant;
+        # safe to read unconditionally — real SOC is always ≤ 100, so the bit
+        # would never be set by accident on older firmwares.
+        is_charging = bool(soc_raw & 0x80)
+        soc = soc_raw & 0x7F
+        temp_status = read_signed_byte(message[8])
+        current = read16_signed(message, 9) / 10.0
+        nominal_capacity = read16(message, 11) / 10.0
+        remaining_wh = read16(message, 13) / 10.0
+        temperature_mos = read_signed_byte(message[15])
+
+        combined_raw = read16(message, 16)
+        battery_number = combined_raw // 10000
+        cycles = combined_raw % 10000
+
+        data = {
+            "voltage": voltage,
+            "soc": soc,
+            "temperature": temp_status,
+            "temperature_mos": temperature_mos,
+            "current": current,
+            "nominal_capacity": nominal_capacity,
+            "remaining_wh": remaining_wh,
+            "cycles": cycles,
+            "is_charging": is_charging,
+        }
+        self._store_battery(data, battery_number)
+        return data
+
+    def _store_battery(self, data: Dict[str, Any], battery_number: int) -> None:
+        """Update primary/secondary battery slots and the consecutive-primary counter."""
         if battery_number == 2:
-            # Secondary battery detected
             self.battery_packet_counter = 0
             self.state["battery_secondary"] = data
-        elif battery_number == 1:
-            # Primary battery
+            return
+
+        # Anything that isn't an explicit secondary battery (number == 2) is
+        # treated as primary — a missing/zero battery_number on a single-battery
+        # bike would otherwise leave all sensors unavailable.
         self.battery_packet_counter += 1
         self.state["battery_primary"] = data
 
-            # After 4 consecutive primary battery packets, reset secondary battery
         if self.battery_packet_counter >= 4:
             self.state["battery_secondary"] = {
                 "voltage": 0.0,
                 "soc": 0.0,
                 "temperature": 0,
+                "temperature_mos": None,
                 "current": 0.0,
                 "nominal_capacity": 0.0,
                 "remaining_wh": 0.0,
                 "cycles": None,
+                "is_charging": False,
             }
 
-        return data
-
     def parse_motor_message(self, message: bytes) -> Optional[Dict[str, Any]]:
-        """Parse motor message and update state."""
-        if len(message) < MOTOR_MESSAGE_LENGTH:
+        """Parse motor frame; dispatch by length to the right layout."""
+        if len(message) >= 20:
+            return self._parse_motor_x20(message)
+        if len(message) >= MOTOR_MESSAGE_LENGTH:
+            return self._parse_motor_ebm(message)
         return None
 
-        # Extract values from message
+    def _parse_motor_ebm(self, message: bytes) -> Optional[Dict[str, Any]]:
+        """Parse 18-byte motor frame (X25 / X35+ / ebikemotion)."""
         assist_level = message[5]
         temperature_celsius = message[6]
         power_amp = float(read16(message, 7)) / 10.0
         speed_kmh = float(read16(message, 9)) / 10.0
-
-        # Additional values
         wheel_speed = read_unsigned_byte(message[11])
         torque_pct = message[12]
         power_max = float(read16(message, 13)) / 10.0
         max_torque_pct = message[15]
 
-        # Update state with motor data
         data = {
             "assist_level": assist_level,
             "temperature_celsius": temperature_celsius,
@@ -138,8 +191,38 @@ class BikeDataParser:
             "torque_motor_pct": torque_pct,
             "power_max_amp": power_max,
             "max_torque_motor_pct": max_torque_pct,
+            "motor_power_watts": None,
+            "rider_power_watts": None,
         }
+        self.state["motor"] = data
+        return data
 
+    def _parse_motor_x20(self, message: bytes) -> Optional[Dict[str, Any]]:
+        """Parse 20-byte motor frame (X20 / HUS-prefixed devices)."""
+        assist_level = read_signed_byte(message[5])
+        # Temperature is signed: 0xD8 (= -40 °C) is the "no sensor data" sentinel
+        # the bike reports during the first packets after connect.
+        temperature_celsius = read_signed_byte(message[6])
+        motor_power_watts = read16(message, 7) / 100.0
+        speed_kmh = read16(message, 9) / 10.0
+        wheel_speed_raw = read_unsigned_byte(message[11])
+        rider_power_watts = read16(message, 12) / 10.0
+        power_max_amp = read16(message, 14) / 10.0
+        # max_torque doubles as a validity flag for wheel_speed (0 → no data).
+        max_torque = read16(message, 16)
+
+        data = {
+            "assist_level": assist_level,
+            "temperature_celsius": temperature_celsius,
+            "power_amp": None,
+            "speed_kmh": speed_kmh,
+            "wheel_speed_rpm": wheel_speed_raw if max_torque != 0 else None,
+            "torque_motor_pct": None,
+            "power_max_amp": power_max_amp,
+            "max_torque_motor_pct": max_torque,
+            "motor_power_watts": motor_power_watts,
+            "rider_power_watts": rider_power_watts,
+        }
         self.state["motor"] = data
         return data
 
@@ -214,29 +297,47 @@ class BikeDataParser:
         return None
 
     def parse_ebm_message(self, message: bytes) -> Optional[Dict[str, Any]]:
-        """Parse EBM (E-Bike Management) message."""
-        if len(message) < EBM_MESSAGE_LENGTH:
-            return None
-
-        # EbmParserEbm format: 32-bit reads directly from message (big-endian)
-        # Raw values are in decimeters, divide by 10000 to get km
-        # (Mahle code divides by 10 to get meters, then displays as km by /1000)
-        if len(message) < 15:
+        """Parse EBM (E-Bike Management) frame; dispatch by length."""
+        if len(message) >= 20:
+            return self._parse_ebm_x20(message)
+        if len(message) >= EBM_MESSAGE_LENGTH:
+            return self._parse_ebm_ebm(message)
         return None
 
+    def _parse_ebm_ebm(self, message: bytes) -> Optional[Dict[str, Any]]:
+        """Parse 17-byte EBM frame (X25 / X35+ / ebikemotion)."""
         odometry_km = read32(message, 5) / 10000.0
         autonomy_km = read32(message, 9) / 10000.0
         is_light_on = message[13] == 1
         status = read_unsigned_byte(message[14])
 
-        # EbmParserEbm only parses bytes 5-14, bytes 15-16 are suffix #@
         data = {
             "odometry": odometry_km,
             "autonomy": autonomy_km,
             "is_light_on": is_light_on,
             "status": status,
         }
+        self.state["ebm"] = data
+        return data
 
+    def _parse_ebm_x20(self, message: bytes) -> Optional[Dict[str, Any]]:
+        """Parse 20-byte EBM frame (X20 / HUS-prefixed devices).
+
+        Autonomy is a 16-bit field at offset 9 (a 32-bit decode there yields
+        implausible six-digit km values). Bytes 15-17 are a fixed `HIJ` marker.
+        Byte 13 light flag is observed as 0x00 / 0x01 / 0xFF — only 0x01 means on.
+        """
+        odometry_km = read32(message, 5) / 10000.0
+        autonomy_km = read16(message, 9) / 1000.0
+        is_light_on = message[13] == 1
+        status = read_unsigned_byte(message[14])
+
+        data = {
+            "odometry": odometry_km,
+            "autonomy": autonomy_km,
+            "is_light_on": is_light_on,
+            "status": status,
+        }
         self.state["ebm"] = data
         return data
 

custom_components/mysmartbike_ble/switch.py

@@ -22,10 +22,7 @@ async def async_setup_entry(
     async_add_entities: AddEntitiesCallback,
 ) -> None:
     """Set up MySmartBike BLE switch entities."""
-    _LOGGER.debug("Setting up switch platform for entry_id: %s", entry.entry_id)
     coordinator: MySmartBikeCoordinator = entry.runtime_data
-
-    _LOGGER.debug("Adding connection switch entity (coordinator.is_connected: %s)", coordinator.is_connected)
     async_add_entities([MySmartBikeConnectionSwitch(coordinator, entry)])
 
 
@@ -51,25 +48,11 @@ class MySmartBikeConnectionSwitch(CoordinatorEntity[MySmartBikeCoordinator], Swi
         if coordinator.protocol_version:
             self._attr_device_info["sw_version"] = coordinator.protocol_version
         self._attr_translation_key = "connection"
-        _LOGGER.debug(
-            "Switch initialized: unique_id=%s, coordinator.is_connected=%s",
-            self._attr_unique_id,
-            coordinator.is_connected,
-        )
 
     @property
     def is_on(self) -> bool:
         """Return True if connection is desired (not manually disconnected)."""
-        # Switch represents the desired state, not the actual connection status
-        # If manual_disconnect is False, user wants to be connected
-        state = not self.coordinator._manual_disconnect
-        _LOGGER.debug(
-            "Switch is_on property called: returning %s (manual_disconnect=%s, is_connected=%s)",
-            state,
-            self.coordinator._manual_disconnect,
-            self.coordinator.is_connected
-        )
-        return state
+        return not self.coordinator._manual_disconnect
 
     @property
     def icon(self) -> str:
@@ -78,32 +61,16 @@ class MySmartBikeConnectionSwitch(CoordinatorEntity[MySmartBikeCoordinator], Swi
 
     async def async_turn_on(self, **kwargs: Any) -> None:
         """Turn on the switch - request connection to the bike."""
-        _LOGGER.debug(
-            "Switch.async_turn_on called (current coordinator.is_connected: %s, manual_disconnect: %s)",
-            self.coordinator.is_connected,
-            self.coordinator._manual_disconnect,
-        )
-
-        # Update state immediately - switch is now ON (connection desired)
         self.async_write_ha_state()
 
         try:
             await self.coordinator.async_reconnect()
-            _LOGGER.debug(
-                "Switch.async_turn_on: reconnect completed (coordinator.is_connected: %s)",
-                self.coordinator.is_connected,
-            )
         except Exception as ex:
-            # Provide user-friendly error message
-            error_msg = str(ex)
-            if "not reachable" in error_msg.lower():
-                _LOGGER.warning(
-                    "Switch.async_turn_on: Cannot connect now - bike is not reachable. "
-                    "Will auto-connect when bike is powered on."
-                )
+            error_msg = str(ex).lower()
+            if "not reachable" in error_msg:
+                _LOGGER.warning("Cannot connect - bike not reachable. Will auto-connect when available.")
             else:
-                _LOGGER.error("Switch.async_turn_on: Failed to connect to bike: %s", ex, exc_info=True)
-            # Switch stays ON - coordinator will auto-reconnect when bike becomes available
+                _LOGGER.error("Failed to connect to bike: %s", ex)
 
     async def async_turn_off(self, **kwargs: Any) -> None:
         """Turn off the switch - disconnect from the bike.
@@ -111,19 +78,9 @@ class MySmartBikeConnectionSwitch(CoordinatorEntity[MySmartBikeCoordinator], Swi
         WARNING: This will turn off the bike after ~5 minutes! It must be manually
         turned on again or connected to power.
         """
-        _LOGGER.warning(
-            "Switch.async_turn_off called - Disconnecting from bike (current coordinator.is_connected: %s). "
-            "Bike will turn off after approximately 5 minutes and must be manually turned on again or connected to power",
-            self.coordinator.is_connected,
-        )
+        _LOGGER.warning("Disconnecting from bike - it will turn off after ~5 minutes")
         try:
             await self.coordinator.async_disconnect()
-            _LOGGER.debug(
-                "Switch.async_turn_off: disconnect completed (coordinator.is_connected: %s, manual_disconnect: %s)",
-                self.coordinator.is_connected,
-                self.coordinator._manual_disconnect,
-            )
             self.async_write_ha_state()
-            _LOGGER.debug("Switch.async_turn_off: state written to HA")
         except Exception as ex:
-            _LOGGER.error("Switch.async_turn_off: Failed to disconnect from bike: %s", ex, exc_info=True)
+            _LOGGER.error("Failed to disconnect from bike: %s", ex)

tests/components/mysmartbike_ble/test_parsers.py

@@ -4,8 +4,10 @@ import pytest
 from custom_components.mysmartbike_ble.parsers import (
     BikeDataParser,
     read16,
+    read16_signed,
     read24,
     read32,
+    read_signed_byte,
     read_unsigned_byte,
 )
 
@@ -35,6 +37,24 @@ class TestReadFunctions:
         assert read_unsigned_byte(0x00) == 0
         assert read_unsigned_byte(0x7F) == 127
 
+    def test_read_signed_byte(self):
+        """Test signed byte read."""
+        assert read_signed_byte(0x00) == 0
+        assert read_signed_byte(0x7F) == 127
+        assert read_signed_byte(0x80) == -128
+        assert read_signed_byte(0xFF) == -1
+
+    def test_read16_signed(self):
+        """Test 16-bit signed read (big-endian)."""
+        # Positive: 0x0001 → 1
+        assert read16_signed(bytes([0x00, 0x01]), 0) == 1
+        # Boundary: 0x7FFF → 32767
+        assert read16_signed(bytes([0x7F, 0xFF]), 0) == 32767
+        # Negative: 0x8000 → -32768
+        assert read16_signed(bytes([0x80, 0x00]), 0) == -32768
+        # -1: 0xFFFF
+        assert read16_signed(bytes([0xFF, 0xFF]), 0) == -1
+
 
 class TestEbmParser:
     """Test EBM message parsing with real data."""
@@ -110,6 +130,132 @@ class TestMotorParser:
         assert result["temperature_celsius"] == 23
 
 
+class TestMotorParserX20:
+    """20-byte motor frame parsing (X20 / HUS-prefixed devices)."""
+
+    # Real frame at rest: assist 1, 22 °C, zero power/speed, max_torque 0x03FF
+    # (the bike's idle sentinel), power_max_amp 9.0 A.
+    MOTOR_MESSAGE = bytes.fromhex("246d245a23011600000000000000005a03ff2340")
+
+    # First packet after connect: temp byte 0xD8 = -40 signed (no-sensor sentinel).
+    MOTOR_MESSAGE_BOOT = bytes.fromhex("246d245a2301d800000000000000005a03ff2340")
+
+    def test_recognition(self):
+        parser = BikeDataParser()
+        assert parser.recognize_message_type(self.MOTOR_MESSAGE) == "motor"
+
+    def test_assist_level_and_temperature(self):
+        parser = BikeDataParser()
+        result = parser.parse_motor_message(self.MOTOR_MESSAGE)
+
+        assert result["assist_level"] == 1
+        assert result["temperature_celsius"] == 22
+
+    def test_signed_temperature_handles_no_sensor_sentinel(self):
+        """0xD8 must decode as -40 °C (signed), not 216 °C (unsigned)."""
+        parser = BikeDataParser()
+        result = parser.parse_motor_message(self.MOTOR_MESSAGE_BOOT)
+
+        assert result["temperature_celsius"] == -40
+
+    def test_speed_and_power(self):
+        parser = BikeDataParser()
+        result = parser.parse_motor_message(self.MOTOR_MESSAGE)
+
+        assert result["speed_kmh"] == 0.0
+        assert result["motor_power_watts"] == 0.0
+        assert result["rider_power_watts"] == 0.0
+        # power_max_amp = 0x005A / 10 = 9.0 A
+        assert abs(result["power_max_amp"] - 9.0) < 0.01
+
+    def test_max_torque_uses_offset_16(self):
+        """max_torque is a 16-bit raw value at offset 16-17 (= 0x03FF)."""
+        parser = BikeDataParser()
+        result = parser.parse_motor_message(self.MOTOR_MESSAGE)
+
+        assert result["max_torque_motor_pct"] == 0x03FF
+
+    def test_wheel_speed_returned_when_max_torque_nonzero(self):
+        parser = BikeDataParser()
+        result = parser.parse_motor_message(self.MOTOR_MESSAGE)
+
+        # max_torque = 0x03FF != 0 → wheel_speed byte (0x00) is returned
+        assert result["wheel_speed_rpm"] == 0
+
+    def test_wheel_speed_nulled_when_max_torque_zero(self):
+        """When max_torque == 0, wheel_speed must be None."""
+        msg = bytearray(self.MOTOR_MESSAGE)
+        msg[16] = 0x00
+        msg[17] = 0x00
+
+        parser = BikeDataParser()
+        result = parser.parse_motor_message(bytes(msg))
+
+        assert result["wheel_speed_rpm"] is None
+
+    def test_x20_specific_fields_replace_legacy(self):
+        """The X20 frame doesn't carry power_amp / torque_motor_pct."""
+        parser = BikeDataParser()
+        result = parser.parse_motor_message(self.MOTOR_MESSAGE)
+
+        assert result["power_amp"] is None
+        assert result["torque_motor_pct"] is None
+        assert "motor_power_watts" in result
+        assert "rider_power_watts" in result
+
+
+class TestEbmParserX20:
+    """20-byte EBM frame parsing (X20 / HUS-prefixed devices)."""
+
+    # Real frame: odometer 0x000A0502 / 10000 = 65.7 km, autonomy 0x5C00 / 1000 = 23.55 km,
+    # byte 13 = 0xFF (lights flag != 1 → off), byte 14 = 0x01 status, then 'HIJ' marker.
+    EBM_MESSAGE = bytes.fromhex("246a245a23000a05025c000002ff0148494a2340")
+
+    def test_message_length(self):
+        assert len(self.EBM_MESSAGE) == 20
+
+    def test_recognition(self):
+        parser = BikeDataParser()
+        assert parser.recognize_message_type(self.EBM_MESSAGE) == "ebm"
+
+    def test_odometer(self):
+        parser = BikeDataParser()
+        result = parser.parse_ebm_message(self.EBM_MESSAGE)
+
+        # 0x000A0502 / 10000 = 65.7
+        assert abs(result["odometry"] - 65.7) < 0.1
+
+    def test_autonomy_uses_16bit_decode(self):
+        """Regression: a 32-bit decode at offset 9 yields ~154 000 km."""
+        parser = BikeDataParser()
+        result = parser.parse_ebm_message(self.EBM_MESSAGE)
+
+        # 0x5C00 / 1000 = 23.552 km
+        assert 20.0 < result["autonomy"] < 30.0
+
+    def test_lights_off_when_byte13_is_ff(self):
+        """0xFF is observed alongside 0x00; only 0x01 maps to lights on."""
+        parser = BikeDataParser()
+        result = parser.parse_ebm_message(self.EBM_MESSAGE)
+
+        assert result["is_light_on"] is False
+
+    def test_lights_on(self):
+        msg = bytearray(self.EBM_MESSAGE)
+        msg[13] = 0x01
+
+        parser = BikeDataParser()
+        result = parser.parse_ebm_message(bytes(msg))
+
+        assert result["is_light_on"] is True
+
+    def test_status_byte(self):
+        parser = BikeDataParser()
+        result = parser.parse_ebm_message(self.EBM_MESSAGE)
+
+        assert result["status"] == 0x01
+
+
 class TestBatteryParser:
     """Test battery message parsing with real data."""
 
@@ -195,6 +341,106 @@ class TestBatteryParser:
         assert parser.state["battery_primary"]["cycles"] == 36
 
 
+class TestBatteryParserX20:
+    """20-byte battery frame parsing (X20 / HUS-prefixed devices)."""
+
+    # Real frame from a HUS device: voltage 37.78 V, SOC 57 %, temp 22 °C,
+    # current 0 A, nominal 352.8 Wh, remaining 200.3 Wh, MOSFET temp 24 °C,
+    # combined cycles 0x2715 = 10005 → battery 1, 5 cycles.
+    BATTERY_MESSAGE = bytes.fromhex("2462245a230ec2391600000dc807d31827152340")
+
+    def test_message_length(self):
+        assert len(self.BATTERY_MESSAGE) == 20
+
+    def test_recognition(self):
+        parser = BikeDataParser()
+        assert parser.recognize_message_type(self.BATTERY_MESSAGE) == "battery"
+
+    def test_voltage_uses_centi_volt_scaling(self):
+        """Voltage on the 20-byte frame is encoded as raw / 100 (vs raw / 10 on 19-byte)."""
+        parser = BikeDataParser()
+        result = parser.parse_battery_message(self.BATTERY_MESSAGE)
+
+        assert result is not None
+        assert abs(result["voltage"] - 37.78) < 0.01
+
+    def test_soc(self):
+        """SOC is the unsigned byte at offset 7 with bit 7 masked off."""
+        parser = BikeDataParser()
+        result = parser.parse_battery_message(self.BATTERY_MESSAGE)
+
+        assert result["soc"] == 57
+        assert result["is_charging"] is False
+
+    def test_temperature(self):
+        parser = BikeDataParser()
+        result = parser.parse_battery_message(self.BATTERY_MESSAGE)
+
+        assert result["temperature"] == 22
+
+    def test_current_is_zero_at_rest(self):
+        """Current is signed read16 / 10."""
+        parser = BikeDataParser()
+        result = parser.parse_battery_message(self.BATTERY_MESSAGE)
+
+        assert result["current"] == 0.0
+
+    def test_capacity_and_remaining(self):
+        parser = BikeDataParser()
+        result = parser.parse_battery_message(self.BATTERY_MESSAGE)
+
+        assert abs(result["nominal_capacity"] - 352.8) < 0.1
+        assert abs(result["remaining_wh"] - 200.3) < 0.1
+
+    def test_temperature_mos(self):
+        """A BMS MOSFET temperature byte sits at offset 15."""
+        parser = BikeDataParser()
+        result = parser.parse_battery_message(self.BATTERY_MESSAGE)
+
+        # 0x18 = 24°C
+        assert result["temperature_mos"] == 24
+
+    def test_cycles_at_offset_16(self):
+        """(battery_number * 10000 + cycles) is read at offset 16-17, not 15-16."""
+        parser = BikeDataParser()
+        result = parser.parse_battery_message(self.BATTERY_MESSAGE)
+
+        # 0x2715 = 10005 → battery 1, 5 cycles
+        assert result["cycles"] == 5
+
+    def test_primary_state_is_set(self):
+        """Regression: reading the combined field at offset 15 would compute
+        battery_number == 0 here and silently drop the update."""
+        parser = BikeDataParser()
+        parser.parse_battery_message(self.BATTERY_MESSAGE)
+
+        assert parser.state["battery_primary"] is not None
+        assert parser.state["battery_primary"]["soc"] == 57
+
+    def test_signed_current_when_charging(self):
+        """A negative raw current value should decode as negative amps."""
+        # Replace bytes 9-10 with 0xFFEC (= -20 raw → -2.0 A)
+        msg = bytearray(self.BATTERY_MESSAGE)
+        msg[9] = 0xFF
+        msg[10] = 0xEC
+
+        parser = BikeDataParser()
+        result = parser.parse_battery_message(bytes(msg))
+
+        assert abs(result["current"] - (-2.0)) < 0.001
+
+    def test_charging_bit_in_soc_byte(self):
+        """When the SOC byte's bit 7 is set, is_charging is True and SOC is masked."""
+        msg = bytearray(self.BATTERY_MESSAGE)
+        msg[7] = 0x80 | 57  # charging flag + 57% SOC
+
+        parser = BikeDataParser()
+        result = parser.parse_battery_message(bytes(msg))
+
+        assert result["is_charging"] is True
+        assert result["soc"] == 57
+
+
 class TestVinParser:
     """Test VIN/serial number message parsing."""