Add x20 parser logic

README.md

@@ -90,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
 
@@ -101,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/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/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
 

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."""