Migrate to aiomqtt (rebased)

pyproject.toml

@@ -17,10 +17,10 @@ requires-python = '>=3.12,<4.0'
 dependencies = [
     "saic-ismart-client-ng (>=0.9.3,<0.10.0)",
     'httpx (>=0.28.1,<0.29.0)',
-    'gmqtt (>=0.7.0,<0.8.0)',
     'inflection (>=0.5.1,<0.6.0)',
     'apscheduler (>=3.11.0,<4.0.0)',
     'python-dotenv (>=1.1.1,<2.0.0)',
+    "aiomqtt (>=2.4.0,<3.0.0)",
 ]
 
 [project.urls]

src/configuration/__init__.py

@@ -1,16 +1,19 @@
 from __future__ import annotations
 
 from enum import Enum
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Literal
 
 if TYPE_CHECKING:
     from zoneinfo import ZoneInfo
 
     from integrations.openwb.charging_station import ChargingStation
 
 
+Transport = Literal["tcp", "websockets"]
+
+
 class TransportProtocol(Enum):
-    def __init__(self, transport_mechanism: str, with_tls: bool) -> None:
+    def __init__(self, transport_mechanism: Transport, with_tls: bool) -> None:
         self.transport_mechanism = transport_mechanism
         self.with_tls = with_tls
 

src/configuration/parser.py

@@ -105,15 +105,18 @@ def __parse_mqtt_transport(args: Namespace, config: Configuration) -> None:
                     args.tls_server_cert_check_hostname
                 )
         else:
-            msg = f"Invalid MQTT URI scheme: {parse_result.scheme}, use tcp or ws"
+            msg = f"Invalid MQTT URI scheme: {parse_result.scheme}, use tls, tcp or ws"
             raise SystemExit(msg)
 
         if parse_result.port:
             config.mqtt_port = parse_result.port
-        elif config.mqtt_transport_protocol == TransportProtocol.TCP:
-            config.mqtt_port = 1883
-        else:
+        elif config.mqtt_transport_protocol == TransportProtocol.TLS:
+            config.mqtt_port = 8883
+        elif config.mqtt_transport_protocol == TransportProtocol.WS:
             config.mqtt_port = 9001
+        else:
+            # fallback to default mqtt port
+            config.mqtt_port = 1883
         config.mqtt_host = str(parse_result.hostname)
 
 

src/log_config.py

@@ -7,7 +7,7 @@
 
 MODULES_DEFAULT_LOG_LEVEL = {
     "asyncio": "WARNING",
-    "gmqtt": "WARNING",
+    "aiomqtt": "WARNING",
     "httpcore": "WARNING",
     "httpx": "WARNING",
     "saic_ismart_client_ng": "WARNING",

src/main.py

@@ -27,4 +27,5 @@
     configuration = process_command_line()
 
     mqtt_gateway = MqttGateway(configuration)
+
     asyncio.run(mqtt_gateway.run(), debug=debug_log_enabled())

src/publisher/core.py

@@ -102,30 +102,55 @@ def publish_json(
         no_prefix: bool = False,
         *,
         retain: bool = True,
+        qos: int = 0,
     ) -> None:
         raise NotImplementedError
 
     @abstractmethod
     def publish_str(
-        self, key: str, value: str, no_prefix: bool = False, *, retain: bool = True
+        self,
+        key: str,
+        value: str,
+        no_prefix: bool = False,
+        *,
+        retain: bool = True,
+        qos: int = 0,
     ) -> None:
         raise NotImplementedError
 
     @abstractmethod
     def publish_int(
-        self, key: str, value: int, no_prefix: bool = False, *, retain: bool = True
+        self,
+        key: str,
+        value: int,
+        no_prefix: bool = False,
+        *,
+        retain: bool = True,
+        qos: int = 0,
     ) -> None:
         raise NotImplementedError
 
     @abstractmethod
     def publish_bool(
-        self, key: str, value: bool, no_prefix: bool = False, *, retain: bool = True
+        self,
+        key: str,
+        value: bool,
+        no_prefix: bool = False,
+        *,
+        retain: bool = True,
+        qos: int = 0,
     ) -> None:
         raise NotImplementedError
 
     @abstractmethod
     def publish_float(
-        self, key: str, value: float, no_prefix: bool = False, *, retain: bool = True
+        self,
+        key: str,
+        value: float,
+        no_prefix: bool = False,
+        *,
+        retain: bool = True,
+        qos: int = 0,
     ) -> None:
         raise NotImplementedError
 
@@ -173,7 +198,7 @@ def publish(
             raise TypeError(msg)
 
     @abstractmethod
-    def clear_topic(self, key: str, no_prefix: bool = False) -> None:
+    def clear_topic(self, key: str, no_prefix: bool = False, qos: int = 0) -> None:
         raise NotImplementedError
 
     def get_mqtt_account_prefix(self) -> str:
@@ -249,7 +274,9 @@ def __anonymize(self, data: T) -> T:
         return data
 
     def keepalive(self) -> None:
-        self.publish_str(mqtt_topics.INTERNAL_LWT, "online", False)
+        self.publish_str(
+            mqtt_topics.INTERNAL_LWT, "online", no_prefix=False, retain=True, qos=1
+        )
 
     @staticmethod
     def anonymize_str(value: str) -> str:

src/publisher/log_publisher.py

@@ -30,36 +30,61 @@ def publish_json(
         no_prefix: bool = False,
         *,
         retain: bool = True,
+        qos: int = 0,
     ) -> None:
         anonymized_json = self.dict_to_anonymized_json(data)
         self.internal_publish(key, anonymized_json, retain=retain)
 
     @override
     def publish_str(
-        self, key: str, value: str, no_prefix: bool = False, *, retain: bool = True
+        self,
+        key: str,
+        value: str,
+        no_prefix: bool = False,
+        *,
+        retain: bool = True,
+        qos: int = 0,
     ) -> None:
         self.internal_publish(key, value, retain=retain)
 
     @override
     def publish_int(
-        self, key: str, value: int, no_prefix: bool = False, *, retain: bool = True
+        self,
+        key: str,
+        value: int,
+        no_prefix: bool = False,
+        *,
+        retain: bool = True,
+        qos: int = 0,
     ) -> None:
         self.internal_publish(key, value, retain=retain)
 
     @override
     def publish_bool(
-        self, key: str, value: bool, no_prefix: bool = False, *, retain: bool = True
+        self,
+        key: str,
+        value: bool,
+        no_prefix: bool = False,
+        *,
+        retain: bool = True,
+        qos: int = 0,
     ) -> None:
         self.internal_publish(key, value, retain=retain)
 
     @override
     def publish_float(
-        self, key: str, value: float, no_prefix: bool = False, *, retain: bool = True
+        self,
+        key: str,
+        value: float,
+        no_prefix: bool = False,
+        *,
+        retain: bool = True,
+        qos: int = 0,
     ) -> None:
         self.internal_publish(key, value, retain=retain)
 
     @override
-    def clear_topic(self, key: str, no_prefix: bool = False) -> None:
+    def clear_topic(self, key: str, no_prefix: bool = False, qos: int = 0) -> None:
         self.internal_publish(key, None)
 
     def internal_publish(