From 52178a219baca03ae5eeeda885737cd6cd25621d Mon Sep 17 00:00:00 2001 From: Rene Nulsch <33263735+ReneNulschDE@users.noreply.github.com> Date: Thu, 30 Apr 2026 10:00:59 +0200 Subject: [PATCH] Add x20 parser logic --- custom_components/mysmartbike_ble/parsers.py | 205 +++++++++++---- .../mysmartbike_ble/test_parsers.py | 246 ++++++++++++++++++ 2 files changed, 399 insertions(+), 52 deletions(-) diff --git a/custom_components/mysmartbike_ble/parsers.py b/custom_components/mysmartbike_ble/parsers.py index 50c3b07..a2c96d7 100644 --- a/custom_components/mysmartbike_ble/parsers.py +++ b/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: - return None + """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, } - - # Handle secondary vs primary battery - if battery_number == 2: - # Secondary battery detected - self.battery_packet_counter = 0 - self.state["battery_secondary"] = data - elif battery_number == 1: - # Primary battery - 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, - "current": 0.0, - "nominal_capacity": 0.0, - "remaining_wh": 0.0, - "cycles": None, - } - + self._store_battery(data, battery_number) 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: - return None + 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]) - # Extract values from message + 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: + self.battery_packet_counter = 0 + self.state["battery_secondary"] = data + 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 + + 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, + } + + def parse_motor_message(self, message: bytes) -> Optional[Dict[str, Any]]: + """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 + + 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: - return None + """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 diff --git a/tests/components/mysmartbike_ble/test_parsers.py b/tests/components/mysmartbike_ble/test_parsers.py index a3ce4e9..85e8afe 100644 --- a/tests/components/mysmartbike_ble/test_parsers.py +++ b/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."""