Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
28bf64baad |
@@ -314,8 +314,12 @@ class BikeDataParser:
|
|||||||
data = {
|
data = {
|
||||||
"odometry": odometry_km,
|
"odometry": odometry_km,
|
||||||
"autonomy": autonomy_km,
|
"autonomy": autonomy_km,
|
||||||
|
"trip_odometry": None,
|
||||||
|
"trip_autonomy": None,
|
||||||
"is_light_on": is_light_on,
|
"is_light_on": is_light_on,
|
||||||
"status": status,
|
"status": status,
|
||||||
|
"accel_y": None,
|
||||||
|
"accel_z": None,
|
||||||
}
|
}
|
||||||
self.state["ebm"] = data
|
self.state["ebm"] = data
|
||||||
return data
|
return data
|
||||||
@@ -323,21 +327,47 @@ class BikeDataParser:
|
|||||||
def _parse_ebm_x20(self, message: bytes) -> Optional[Dict[str, Any]]:
|
def _parse_ebm_x20(self, message: bytes) -> Optional[Dict[str, Any]]:
|
||||||
"""Parse 20-byte EBM frame (X20 / HUS-prefixed devices).
|
"""Parse 20-byte EBM frame (X20 / HUS-prefixed devices).
|
||||||
|
|
||||||
Autonomy is a 16-bit field at offset 9 (a 32-bit decode there yields
|
The bike alternates between two slot indicators in byte 14:
|
||||||
implausible six-digit km values). Bytes 15-17 are a fixed `HIJ` marker.
|
- slot == 1 → bytes 5-9 carry the LIFETIME odometer & range
|
||||||
Byte 13 light flag is observed as 0x00 / 0x01 / 0xFF — only 0x01 means on.
|
- slot == 2 → same bytes carry the current TRIP A distance & range
|
||||||
"""
|
|
||||||
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 = {
|
Bytes 15-17 are a fixed `HIJ` (`0x48 0x49 0x4A`) marker before `#@`. A
|
||||||
"odometry": odometry_km,
|
device using protocol v200 puts an MPlatform error code and remote-SOC
|
||||||
"autonomy": autonomy_km,
|
info there instead — not yet supported.
|
||||||
"is_light_on": is_light_on,
|
"""
|
||||||
"status": status,
|
odometry_km = read24(message, 5) / 10.0
|
||||||
}
|
autonomy_km = read16(message, 8) / 10.0
|
||||||
|
is_light_on = message[10] == 1
|
||||||
|
status = read_unsigned_byte(message[11])
|
||||||
|
accel_z = read_signed_byte(message[12])
|
||||||
|
accel_y = read_signed_byte(message[13])
|
||||||
|
slot = read_unsigned_byte(message[14])
|
||||||
|
|
||||||
|
prev = self.state.get("ebm") or {}
|
||||||
|
|
||||||
|
if slot == 2:
|
||||||
|
data = {
|
||||||
|
"odometry": prev.get("odometry"),
|
||||||
|
"autonomy": prev.get("autonomy"),
|
||||||
|
"trip_odometry": odometry_km,
|
||||||
|
"trip_autonomy": autonomy_km,
|
||||||
|
}
|
||||||
|
else:
|
||||||
|
data = {
|
||||||
|
"odometry": odometry_km,
|
||||||
|
"autonomy": autonomy_km,
|
||||||
|
"trip_odometry": prev.get("trip_odometry"),
|
||||||
|
"trip_autonomy": prev.get("trip_autonomy"),
|
||||||
|
}
|
||||||
|
|
||||||
|
data.update(
|
||||||
|
{
|
||||||
|
"is_light_on": is_light_on,
|
||||||
|
"status": status,
|
||||||
|
"accel_y": accel_y,
|
||||||
|
"accel_z": accel_z,
|
||||||
|
}
|
||||||
|
)
|
||||||
self.state["ebm"] = data
|
self.state["ebm"] = data
|
||||||
return data
|
return data
|
||||||
|
|
||||||
|
|||||||
@@ -119,6 +119,25 @@ SENSORS: tuple[MySmartBikeSensorEntityDescription, ...] = (
|
|||||||
icon="mdi:map-marker-distance",
|
icon="mdi:map-marker-distance",
|
||||||
value_fn=lambda data: safe_get(data, "ebm", "autonomy"),
|
value_fn=lambda data: safe_get(data, "ebm", "autonomy"),
|
||||||
),
|
),
|
||||||
|
MySmartBikeSensorEntityDescription(
|
||||||
|
key="trip_distance",
|
||||||
|
name="Trip A Distance",
|
||||||
|
native_unit_of_measurement=UnitOfLength.KILOMETERS,
|
||||||
|
device_class=SensorDeviceClass.DISTANCE,
|
||||||
|
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||||
|
icon="mdi:bike",
|
||||||
|
value_fn=lambda data: safe_get(data, "ebm", "trip_odometry"),
|
||||||
|
),
|
||||||
|
MySmartBikeSensorEntityDescription(
|
||||||
|
key="trip_range",
|
||||||
|
name="Trip A Range",
|
||||||
|
native_unit_of_measurement=UnitOfLength.KILOMETERS,
|
||||||
|
device_class=SensorDeviceClass.DISTANCE,
|
||||||
|
state_class=SensorStateClass.MEASUREMENT,
|
||||||
|
icon="mdi:map-marker-distance",
|
||||||
|
entity_registry_enabled_default=False,
|
||||||
|
value_fn=lambda data: safe_get(data, "ebm", "trip_autonomy"),
|
||||||
|
),
|
||||||
MySmartBikeSensorEntityDescription(
|
MySmartBikeSensorEntityDescription(
|
||||||
key="light",
|
key="light",
|
||||||
name="Light",
|
name="Light",
|
||||||
|
|||||||
@@ -205,44 +205,53 @@ class TestMotorParserX20:
|
|||||||
|
|
||||||
|
|
||||||
class TestEbmParserX20:
|
class TestEbmParserX20:
|
||||||
"""20-byte EBM frame parsing (X20 / HUS-prefixed devices)."""
|
"""20-byte EBM frame parsing (X20 / HUS-prefixed devices).
|
||||||
|
|
||||||
# Real frame: odometer 0x000A0502 / 10000 = 65.7 km, autonomy 0x5C00 / 1000 = 23.55 km,
|
Field offsets verified against an app-confirmed capture:
|
||||||
# byte 13 = 0xFF (lights flag != 1 → off), byte 14 = 0x01 status, then 'HIJ' marker.
|
- Odometer 0x000084 / 10 = 13.2 km (app shows 8.08 mi = 13.005 km)
|
||||||
EBM_MESSAGE = bytes.fromhex("246a245a23000a05025c000002ff0148494a2340")
|
- Range 0x02E1 / 10 = 73.7 km (app shows 45 mi = 72.42 km)
|
||||||
|
"""
|
||||||
|
|
||||||
|
# Slot 1 frame = lifetime values
|
||||||
|
EBM_LIFETIME = bytes.fromhex("246a245a2300008402e1000001f50148494a2340")
|
||||||
|
# Slot 2 frame = trip values; same bike, same odometer/autonomy bytes →
|
||||||
|
# trip A == lifetime (no reset since first ride).
|
||||||
|
EBM_TRIP = bytes.fromhex("246a245a2300008402e1000001f50248494a2340")
|
||||||
|
|
||||||
def test_message_length(self):
|
def test_message_length(self):
|
||||||
assert len(self.EBM_MESSAGE) == 20
|
assert len(self.EBM_LIFETIME) == 20
|
||||||
|
|
||||||
def test_recognition(self):
|
def test_recognition(self):
|
||||||
parser = BikeDataParser()
|
parser = BikeDataParser()
|
||||||
assert parser.recognize_message_type(self.EBM_MESSAGE) == "ebm"
|
assert parser.recognize_message_type(self.EBM_LIFETIME) == "ebm"
|
||||||
|
|
||||||
def test_odometer(self):
|
def test_lifetime_odometer(self):
|
||||||
|
"""Odometer is a 24-bit field at offset 5 with /10 km scaling."""
|
||||||
parser = BikeDataParser()
|
parser = BikeDataParser()
|
||||||
result = parser.parse_ebm_message(self.EBM_MESSAGE)
|
result = parser.parse_ebm_message(self.EBM_LIFETIME)
|
||||||
|
|
||||||
# 0x000A0502 / 10000 = 65.7
|
# 0x000084 / 10 = 13.2 km (app: 8.08 mi = 13.005 km)
|
||||||
assert abs(result["odometry"] - 65.7) < 0.1
|
assert abs(result["odometry"] - 13.2) < 0.05
|
||||||
|
|
||||||
def test_autonomy_uses_16bit_decode(self):
|
def test_lifetime_autonomy(self):
|
||||||
"""Regression: a 32-bit decode at offset 9 yields ~154 000 km."""
|
"""Range is a 16-bit field at offset 8 with /10 km scaling."""
|
||||||
parser = BikeDataParser()
|
parser = BikeDataParser()
|
||||||
result = parser.parse_ebm_message(self.EBM_MESSAGE)
|
result = parser.parse_ebm_message(self.EBM_LIFETIME)
|
||||||
|
|
||||||
# 0x5C00 / 1000 = 23.552 km
|
# 0x02E1 / 10 = 73.7 km (app: 45 mi = 72.42 km)
|
||||||
assert 20.0 < result["autonomy"] < 30.0
|
assert abs(result["autonomy"] - 73.7) < 0.05
|
||||||
|
|
||||||
def test_lights_off_when_byte13_is_ff(self):
|
def test_lights_off_at_offset_10(self):
|
||||||
"""0xFF is observed alongside 0x00; only 0x01 maps to lights on."""
|
"""Lights flag moved from offset 13 to offset 10 in the X20 layout."""
|
||||||
parser = BikeDataParser()
|
parser = BikeDataParser()
|
||||||
result = parser.parse_ebm_message(self.EBM_MESSAGE)
|
result = parser.parse_ebm_message(self.EBM_LIFETIME)
|
||||||
|
|
||||||
|
# message[10] = 0x00 → off
|
||||||
assert result["is_light_on"] is False
|
assert result["is_light_on"] is False
|
||||||
|
|
||||||
def test_lights_on(self):
|
def test_lights_on(self):
|
||||||
msg = bytearray(self.EBM_MESSAGE)
|
msg = bytearray(self.EBM_LIFETIME)
|
||||||
msg[13] = 0x01
|
msg[10] = 0x01
|
||||||
|
|
||||||
parser = BikeDataParser()
|
parser = BikeDataParser()
|
||||||
result = parser.parse_ebm_message(bytes(msg))
|
result = parser.parse_ebm_message(bytes(msg))
|
||||||
@@ -250,10 +259,52 @@ class TestEbmParserX20:
|
|||||||
assert result["is_light_on"] is True
|
assert result["is_light_on"] is True
|
||||||
|
|
||||||
def test_status_byte(self):
|
def test_status_byte(self):
|
||||||
|
"""Status moved from offset 14 to offset 11."""
|
||||||
parser = BikeDataParser()
|
parser = BikeDataParser()
|
||||||
result = parser.parse_ebm_message(self.EBM_MESSAGE)
|
result = parser.parse_ebm_message(self.EBM_LIFETIME)
|
||||||
|
|
||||||
assert result["status"] == 0x01
|
# message[11] = 0x00
|
||||||
|
assert result["status"] == 0
|
||||||
|
|
||||||
|
def test_accelerometer_axes(self):
|
||||||
|
"""Bytes 12-13 carry accelerometer Z/Y as signed bytes."""
|
||||||
|
parser = BikeDataParser()
|
||||||
|
result = parser.parse_ebm_message(self.EBM_LIFETIME)
|
||||||
|
|
||||||
|
# message[12] = 0x01, message[13] = 0xF5 (signed = -11)
|
||||||
|
assert result["accel_z"] == 1
|
||||||
|
assert result["accel_y"] == -11
|
||||||
|
|
||||||
|
def test_slot_2_updates_trip_only(self):
|
||||||
|
"""Slot 2 frames carry trip A values; lifetime fields stay at previous."""
|
||||||
|
parser = BikeDataParser()
|
||||||
|
# First ingest a slot 1 frame so we have a previous lifetime
|
||||||
|
parser.parse_ebm_message(self.EBM_LIFETIME)
|
||||||
|
prev_odo = parser.state["ebm"]["odometry"]
|
||||||
|
|
||||||
|
# Now a slot 2 frame with different bytes (mock a real trip distance)
|
||||||
|
msg = bytearray(self.EBM_TRIP)
|
||||||
|
# Set trip odometer to 0x000050 = 80 → 8.0 km
|
||||||
|
msg[5], msg[6], msg[7] = 0x00, 0x00, 0x50
|
||||||
|
result = parser.parse_ebm_message(bytes(msg))
|
||||||
|
|
||||||
|
assert result["trip_odometry"] == 8.0
|
||||||
|
assert result["odometry"] == prev_odo # lifetime preserved
|
||||||
|
|
||||||
|
def test_slot_1_updates_lifetime_preserves_trip(self):
|
||||||
|
"""A slot 1 frame must not clobber the previously seen trip values."""
|
||||||
|
parser = BikeDataParser()
|
||||||
|
# First a slot 2 frame to populate trip_*
|
||||||
|
msg2 = bytearray(self.EBM_TRIP)
|
||||||
|
msg2[5], msg2[6], msg2[7] = 0x00, 0x00, 0x50 # trip = 8.0 km
|
||||||
|
parser.parse_ebm_message(bytes(msg2))
|
||||||
|
assert parser.state["ebm"]["trip_odometry"] == 8.0
|
||||||
|
|
||||||
|
# Now a slot 1 frame
|
||||||
|
result = parser.parse_ebm_message(self.EBM_LIFETIME)
|
||||||
|
|
||||||
|
assert result["trip_odometry"] == 8.0 # preserved
|
||||||
|
assert abs(result["odometry"] - 13.2) < 0.05
|
||||||
|
|
||||||
|
|
||||||
class TestBatteryParser:
|
class TestBatteryParser:
|
||||||
|
|||||||
Reference in New Issue
Block a user