Add support for mqtt charging-error message (#338)

* Add support for mqtt charging-error message * Fix tests
skodaconnect · Jan 21, 2025 · fdeb1bf · fdeb1bf
1 parent 4d4c4f6
commit fdeb1bf
Show file tree

Hide file tree

Showing 5 changed files with 72 additions and 22 deletions.
diff --git a/myskoda/models/charging.py b/myskoda/models/charging.py
@@ -60,6 +60,10 @@ class PlugUnlockMode(StrEnum):
     OFF = "OFF"
 
 
+class ServiceEventChargingError(StrEnum):
+    STOPPED_DEVICE = "STOPPED_DEVICE"
+
+
 @dataclass
 class Settings(DataClassORJSONMixin):
     available_charge_modes: list[ChargeMode] = field(

diff --git a/myskoda/models/service_event.py b/myskoda/models/service_event.py
@@ -8,7 +8,7 @@
 from mashumaro import field_options
 from mashumaro.mixins.orjson import DataClassORJSONMixin
 
-from .charging import ChargeMode, ChargingState
+from .charging import ChargeMode, ChargingState, ServiceEventChargingError
 
 
 class ServiceEventName(StrEnum):
@@ -20,6 +20,7 @@ class ServiceEventName(StrEnum):
     CHANGE_REMAINING_TIME = "change-remaining-time"
     CHANGE_SOC = "change-soc"
     CHARGING_COMPLETED = "charging-completed"
+    CHARGING_ERROR = "charging-error"
     CHARGING_STATUS_CHANGED = "charging-status-changed"
     CLIMATISATION_COMPLETED = "climatisation-completed"
     DEPARTURE_READY = "departure-ready"
@@ -102,14 +103,21 @@ def _deserialize_time_to_finish(value: int | str) -> int | None:
 class ServiceEventChargingData(ServiceEventData):
     """Charging data inside charging service event change-soc."""
 
-    mode: ChargeMode = field(metadata=field_options(deserialize=_deserialize_mode))
-    state: ChargingState = field(metadata=field_options(deserialize=_deserialize_charging_state))
-    soc: int
-    charged_range: int = field(metadata=field_options(alias="chargedRange"))
+    mode: ChargeMode | None = field(
+        default=None, metadata=field_options(deserialize=_deserialize_mode)
+    )
+    state: ChargingState | None = field(
+        default=None, metadata=field_options(deserialize=_deserialize_charging_state)
+    )
+    soc: int | None = field(default=None)
+    charged_range: int | None = field(default=None, metadata=field_options(alias="chargedRange"))
     time_to_finish: int | None = field(
         default=None,
         metadata=field_options(alias="timeToFinish", deserialize=_deserialize_time_to_finish),
     )
+    error_code: ServiceEventChargingError | None = field(
+        default=None, metadata=field_options(alias="errorCode")
+    )
 
 
 @dataclass

diff --git a/tests/fixtures/events/service_event_charging_charging_error.json b/tests/fixtures/events/service_event_charging_charging_error.json
@@ -0,0 +1,12 @@
+{
+    "version": 1,
+    "traceId": "81b50439-af83-44e1-9515-5b3925d4c34a",
+    "timestamp": "2024-11-09T13:30:34Z",
+    "producer": "SKODA_MHUB",
+    "name": "charging-error",
+    "data": {
+	"errorCode": "STOPPED_DEVICE",
+        "userId": "ad0d7945-4814-43d0-801f-charging-error",
+        "vin": "TMBAXXXXXXXXXXXXX"
+    }
+}
diff --git a/tests/test_mqtt.py b/tests/test_mqtt.py
@@ -272,15 +272,21 @@ def on_event(event: Event) -> None:
             vin=VIN,
             user_id=USER_ID,
             timestamp=ANY,
-            event=ServiceEvent(
+            event=ServiceEventWithChargingData(
                 version=1,
                 trace_id=trace_id,
                 timestamp=timestamp,
                 producer="SKODA_MHUB",
                 name=ServiceEventName.CHARGING_COMPLETED,
-                data=ServiceEventData(
+                data=ServiceEventChargingData(
                     user_id=USER_ID,
                     vin=VIN,
+                    mode=None,
+                    state=None,
+                    soc=None,
+                    charged_range=None,
+                    time_to_finish=None,
+                    error_code=None,
                 ),
             ),
         ),

diff --git a/tests/test_service_event.py b/tests/test_service_event.py
@@ -13,6 +13,7 @@
     ChargingState,
     ServiceEvent,
     ServiceEventChargingData,
+    ServiceEventChargingError,
     ServiceEventData,
     ServiceEventName,
     ServiceEventWithChargingData,
@@ -28,6 +29,7 @@ def load_service_events() -> list[str]:
     for path in [
         "events/service_event_charging_change_soc.json",
         "events/service_event_charging_charging_status_changed.json",
+        "events/service_event_charging_charging_error.json",
         "events/service_event_departure_ready.json",
         "events/service_event_departure_status_changed.json",
     ]:
@@ -38,21 +40,39 @@ def load_service_events() -> list[str]:
 
 def test_parse_service_events(service_events: list[str]) -> None:
     for service_event in service_events:
-        try:
-            event = ServiceEventWithChargingData.from_json(service_event)
-        except ValueError:
-            event = ServiceEvent.from_json(service_event)
-
-        if event.name == ServiceEventName.CHANGE_SOC:
-            assert event.data == ServiceEventChargingData(
-                charged_range=195,
-                mode=ChargeMode.MANUAL,
-                soc=50,
-                state=ChargingState.CHARGING,
-                time_to_finish=440,
-                user_id="ad0d7945-4814-43d0-801f-change-soc",
-                vin="TMBAXXXXXXXXXXXXX",
-            )
+        event = ServiceEvent.from_json(service_event)
+
+        if event.name in [
+            ServiceEventName.CHANGE_SOC,
+            ServiceEventName.CHARGING_STATUS_CHANGED,
+            ServiceEventName.CHARGING_ERROR,
+        ]:
+            try:
+                event = ServiceEventWithChargingData.from_json(service_event)
+            except ValueError:
+                event = ServiceEvent.from_json(service_event)
+
+            if event.name == ServiceEventName.CHANGE_SOC:
+                assert event.data == ServiceEventChargingData(
+                    charged_range=195,
+                    mode=ChargeMode.MANUAL,
+                    soc=50,
+                    state=ChargingState.CHARGING,
+                    time_to_finish=440,
+                    user_id="ad0d7945-4814-43d0-801f-change-soc",
+                    vin="TMBAXXXXXXXXXXXXX",
+                )
+            elif event.name == ServiceEventName.CHARGING_STATUS_CHANGED:
+                assert event.data == ServiceEventChargingData(
+                    user_id=f"ad0d7945-4814-43d0-801f-{event.name.value}",
+                    vin="TMBAXXXXXXXXXXXXX",
+                )
+            elif event.name == ServiceEventName.CHARGING_ERROR:
+                assert event.data == ServiceEventChargingData(
+                    user_id=f"ad0d7945-4814-43d0-801f-{event.name.value}",
+                    vin="TMBAXXXXXXXXXXXXX",
+                    error_code=ServiceEventChargingError.STOPPED_DEVICE,
+                )
         else:
             assert event.data == ServiceEventData(
                 user_id=f"ad0d7945-4814-43d0-801f-{event.name.value}",