Add MQTT event classes

docs/mqtt.md

@@ -36,6 +36,7 @@ For example, if a `start-stop-air-conditioning` message is sent, the MQTT server
 * `"IN_PROGRESS"`: Operation is currently being executed by the car.
 * `"COMPLETED_SUCCESS"`: Operation completed.
 * `"ERROR"`: An error occurred. Additional information in field `"errorCode"`.
+* `"COMPLETED_WARNING"`: Extracted from analysing the smali files. We don't know the implications.
 
 ### /service-event
 

myskoda/cli.py

@@ -4,7 +4,10 @@
 poetry run python3 -m myskoda.cli
 """
 
+from logging import DEBUG, INFO
+
 import asyncclick as click
+import coloredlogs
 from aiohttp import ClientSession
 from asyncclick.core import Context
 from termcolor import colored
@@ -28,9 +31,11 @@
 @click.version_option()
 @click.option("username", "--user", help="Username used for login.", required=True)
 @click.option("password", "--password", help="Password used for login.", required=True)
+@click.option("verbose", "--verbose", help="Enable verbose logging.", is_flag=True)
 @click.pass_context
-def cli(ctx: Context, username: str, password: str) -> None:
+def cli(ctx: Context, username: str, password: str, verbose: bool) -> None:  # noqa: FBT001
     """Interact with the MySkoda API."""
+    coloredlogs.install(level=DEBUG if verbose else INFO)
     ctx.ensure_object(dict)
     ctx.obj["username"] = username
     ctx.obj["password"] = password

myskoda/const.py

@@ -6,5 +6,5 @@
 BASE_URL_SKODA = "https://mysmob.api.connect.skoda-auto.cz"
 BASE_URL_IDENT = "https://identity.vwgroup.io"
 
-MQTT_HOST = "3.72.252.203"
+MQTT_HOST = "mqtt.messagehub.de"
 MQTT_PORT = 8883

myskoda/models/mqtt.py

@@ -0,0 +1,72 @@
+"""Models relaterd to the MQTT API."""
+
+from datetime import datetime
+from enum import StrEnum
+from typing import Generic, TypeVar
+
+from pydantic import BaseModel, Field, validator
+
+
+class OperationStatus(StrEnum):
+    IN_PROGRESS = "IN_PROGRESS"
+    COMPLETED_SUCCESS = "COMPLETED_SUCCESS"
+    COMPLETED_WARNING = "COMPLETED_WARNING"
+    ERROR = "ERROR"
+
+
+class OperationRequest(BaseModel):
+    version: int
+    trace_id: str = Field(None, alias="traceId")
+    request_id: str = Field(None, alias="requestId")
+    operation: str
+    status: OperationStatus
+    error_code: str = Field(None, alias="errorCode")
+
+
+class ServiceEventData(BaseModel):
+    user_id: str = Field(None, alias="userId")
+    vin: str
+
+
+T = TypeVar("T", bound=ServiceEventData)
+
+
+class ServiceEvent(BaseModel, Generic[T]):
+    version: int
+    trace_id: str = Field(None, alias="traceId")
+    timestamp: datetime = Field(None, alias="requestId")
+    producer: str
+    name: str
+    data: T
+
+
+class ServiceEventChargingState(StrEnum):
+    CHARGING = "charging"
+    CHARGED_NOT_CONSERVING = "chargePurposeReachedAndNotConservationCharging"
+    NOT_READY = "notReadyForCharging"
+    READY = "readyForCharging"
+
+
+class ServiceEventChargingData(ServiceEventData):
+    mode: str
+    state: ServiceEventChargingState
+    soc: int
+    charged_range: str = Field(None, alias="chargedRange")
+    time_to_finish: str | None = Field(None, alias="timeToFinish")
+
+    @validator("soc")
+    def _parse_soc(cls, value: str) -> int:  # noqa: N805
+        return int(value)
+
+    @validator("charged_range")
+    def _parse_charged_range(cls, value: str) -> int:  # noqa: N805
+        return int(value)
+
+    @validator("time_to_finish")
+    def _parse_time_to_finish(cls, value: str) -> int | None:  # noqa: N805
+        if value == "null":
+            return None
+        return int(value)
+
+
+ServiceEventCharging = ServiceEvent[ServiceEventChargingData]

myskoda/mqtt.py

@@ -1,16 +1,173 @@
 """MQTT client module for the MySkoda server."""
 
+import json
 import logging
+import re
 import ssl
-from typing import Any
+from enum import StrEnum
+from typing import Literal, cast
 
 from paho.mqtt.client import Client, MQTTMessage
 
+from myskoda.models.mqtt import OperationRequest, ServiceEvent, ServiceEventCharging
+
 from .const import MQTT_HOST, MQTT_PORT
 from .models.user import User
 from .rest_api import RestApi
 
 _LOGGER = logging.getLogger(__name__)
+TOPIC_RE = re.compile("^(.*?)/(.*?)/(.*?)$")
+
+
+class Topic(StrEnum):
+    UPDATE_BATTERY_SUPPORT = "UPDATE_BATTERY_SUPPORT"
+    LOCK_VEHICLE = "LOCK_VEHICLE"
+    WAKEUP = "WAKEUP"
+    SET_TARGET_TEMPERATURE = "SET_TARGET_TEMPERATURE"
+    START_STOP_AIR_CONDITIONING = "START_STOP_AIR_CONDITIONING"
+    START_STOP_WINDOW_HEATING = "START_STOP_WINDOW_HEATING"
+    START_STOP_CHARGING = "START_STOP_CHARGING"
+    APPLY_BACKUP = "APPLY_BACKUP"
+    HONK_AND_FLASH = "HONK_AND_FLASH"
+    AIR_CONDITIONING = "AIR_CONDITIONING"
+    CHARGING = "CHARGING"
+    ACCESS = "ACCESS"
+    LIGHTS = "LIGHTS"
+
+
+class EventUpdateBatterySupport:
+    topic: Literal[Topic.UPDATE_BATTERY_SUPPORT]
+    payload: OperationRequest
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.UPDATE_BATTERY_SUPPORT
+        self.payload = OperationRequest(**payload)
+
+
+class EventLockVehicle:
+    topic: Literal[Topic.LOCK_VEHICLE]
+    payload: OperationRequest
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.LOCK_VEHICLE
+        self.payload = OperationRequest(**payload)
+
+
+class EventWakeup:
+    topic: Literal[Topic.WAKEUP]
+    payload: OperationRequest
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.WAKEUP
+        self.payload = OperationRequest(**payload)
+
+
+class EventSetTargetTemperature:
+    topic: Literal[Topic.SET_TARGET_TEMPERATURE]
+    payload: OperationRequest
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.SET_TARGET_TEMPERATURE
+        self.payload = OperationRequest(**payload)
+
+
+class EventStartStopAirConditioning:
+    topic: Literal[Topic.START_STOP_AIR_CONDITIONING]
+    payload: OperationRequest
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.START_STOP_AIR_CONDITIONING
+        self.payload = OperationRequest(**payload)
+
+
+class EventStartStopWindowHeating:
+    topic: Literal[Topic.START_STOP_WINDOW_HEATING]
+    payload: OperationRequest
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.START_STOP_WINDOW_HEATING
+        self.payload = OperationRequest(**payload)
+
+
+class EventStartStopCharging:
+    topic: Literal[Topic.START_STOP_CHARGING]
+    payload: OperationRequest
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.START_STOP_CHARGING
+        self.payload = OperationRequest(**payload)
+
+
+class EventHonkAndFlash:
+    topic: Literal[Topic.HONK_AND_FLASH]
+    payload: OperationRequest
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.HONK_AND_FLASH
+        self.payload = OperationRequest(**payload)
+
+
+class EventApplyBackup:
+    topic: Literal[Topic.APPLY_BACKUP]
+    payload: OperationRequest
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.APPLY_BACKUP
+        self.payload = OperationRequest(**payload)
+
+
+class EventAirConditioning:
+    topic: Literal[Topic.AIR_CONDITIONING]
+    payload: ServiceEvent
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.AIR_CONDITIONING
+        self.payload = ServiceEvent(**payload)
+
+
+class EventCharging:
+    topic: Literal[Topic.CHARGING]
+    payload: ServiceEventCharging
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.CHARGING
+        self.payload = ServiceEventCharging(**payload)
+
+
+class EventAccess:
+    topic: Literal[Topic.ACCESS]
+    payload: ServiceEvent
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.ACCESS
+        self.payload = ServiceEvent(**payload)
+
+
+class EventLights:
+    topic: Literal[Topic.LIGHTS]
+    payload: ServiceEvent
+
+    def __init__(self, payload: dict) -> None:  # noqa: D107
+        self.topic = Topic.LIGHTS
+        self.payload = ServiceEvent(**payload)
+
+
+Event = (
+    EventUpdateBatterySupport
+    | EventLockVehicle
+    | EventWakeup
+    | EventSetTargetTemperature
+    | EventStartStopAirConditioning
+    | EventStartStopWindowHeating
+    | EventStartStopCharging
+    | EventApplyBackup
+    | EventHonkAndFlash
+    | EventApplyBackup
+    | EventAirConditioning
+    | EventCharging
+    | EventAccess
+    | EventLights
+)
 
 
 class MQTT:
@@ -31,7 +188,6 @@ async def connect(self) -> None:
         self.client.on_connect = self._on_connect
         self.client.on_message = self._on_message
         self.client.tls_set_context(context=ssl.create_default_context())
-        self.client.tls_insecure_set(value=True)
         self.client.username_pw_set(
             self.user.id, await self.api.idk_session.get_access_token(self.api.session)
         )
@@ -41,8 +197,8 @@ def loop_forever(self) -> None:
         """Make the MQTT client process new messages until the current process is cancelled."""
         self.client.loop_forever()
 
-    def _on_connect(self, client: Client, _userdata: Any, _flags: Any, reason: int) -> None:  # noqa: ANN401
-        print(f"MQTT Connected. {reason}")
+    def _on_connect(self, client: Client, _userdata: None, _flags: dict, _reason: int) -> None:
+        _LOGGER.info("MQTT Connected.")
         user_id = self.user.id
 
         for vin in self.vehicles:
@@ -72,5 +228,51 @@ def _on_connect(self, client: Client, _userdata: Any, _flags: Any, reason: int)
                 f"{user_id}/{vin}/operation-request/vehicle-services-backup/apply-backup"
             )
 
-    def _on_message(self, _client: Client, _userdata: Any, msg: MQTTMessage) -> None:  # noqa: ANN401
-        print(f"{msg.topic}: {msg.payload}")
+    def _emit(self, event: Event) -> None:
+        print(event)
+
+    def _on_message(self, _client: Client, _userdata: None, msg: MQTTMessage) -> None:  # noqa: C901, PLR0912
+        topic_match = TOPIC_RE.match(msg.topic)
+
+        if not topic_match:
+            _LOGGER.warning("Unexpected MQTT topic encountered: %s", topic_match)
+            return
+
+        [_user_id, vin, topic] = topic_match.groups()
+        data = cast(str, msg.payload)
+
+        if len(data) == 0:
+            return
+
+        _LOGGER.debug("Message received for %s (%s): %s", vin, topic, data)
+
+        data = json.loads(msg.payload)
+
+        if topic == "account-event/privacy":
+            pass
+        elif topic == "operation-request/charging/update-battery-support":
+            self._emit(EventUpdateBatterySupport(data))
+        elif topic == "operation-request/vehicle-access/lock-vehicle":
+            self._emit(EventLockVehicle(data))
+        elif topic == "operation-request/vehicle-wakeup/wakeup":
+            self._emit(EventWakeup(data))
+        elif topic == "operation-request/air-conditioning/set-target-temperature":
+            self._emit(EventSetTargetTemperature(data))
+        elif topic == "operation-request/air-conditioning/start-stop-air-conditioning":
+            self._emit(EventStartStopAirConditioning(data))
+        elif topic == "operation-request/air-conditioning/start-stop-window-heating":
+            self._emit(EventStartStopWindowHeating(data))
+        elif topic == "operation-request/charging/start-stop-charging":
+            self._emit(EventStartStopCharging(data))
+        elif topic == "operation-request/vehicle-services-backup/apply-backup":
+            self._emit(EventApplyBackup(data))
+        elif topic == "operation-request/vehicle-access/honk-and-flash":
+            self._emit(EventHonkAndFlash(data))
+        elif topic == "service-event/air-conditioning":
+            self._emit(EventAirConditioning(data))
+        elif topic == "service-event/charging":
+            self._emit(EventCharging(data))
+        elif topic == "service-event/vehicle-status/access":
+            self._emit(EventAccess(data))
+        elif topic == "service-event/vehicle-status/lights":
+            self._emit(EventLights(data))

myskoda/rest_api.py

@@ -154,7 +154,7 @@ async def get_user(self) -> User:
             f"{BASE_URL_SKODA}/api/v1/users",
             headers=await self._headers(),
         ) as response:
-            _LOGGER.debug("vin %s: Received user")
+            _LOGGER.debug("Received user")
             return User(**await response.json())
 
     async def list_vehicles(self) -> list[str]: