Wrap awscrt types out of public MQTT signatures (#101)

CHANGELOG.rst

@@ -25,6 +25,29 @@ Changed
   is byte-for-byte unchanged (verified by golden capture and the existing
   CLI tests); energy output now renders a summary table plus a breakdown
   table entirely in Rich rather than printing plain text and a Rich table.
+- **awscrt types wrapped out of public MQTT signatures** (`#101
+  <https://github.com/eman/nwp500-python/issues/101>`_): the library no
+  longer exposes ``awscrt`` SDK types on its own public/semi-public API
+  surface. A library-owned ``nwp500.mqtt.QoS`` ``IntEnum`` (also exported
+  as ``nwp500.QoS``) now replaces ``awscrt.mqtt.QoS`` on the ``publish``
+  and ``subscribe`` methods of ``NavienMqttClient``, ``MqttConnection``
+  and ``MqttSubscriptionManager``, on ``MqttCommandQueue.enqueue`` and on
+  the ``QueuedCommand`` dataclass. ``awscrt.mqtt.Connection`` handles are
+  typed behind the ``MqttConnectionHandle`` alias, and translation to/from
+  ``awscrt`` happens only at the connection-layer boundary
+  (``nwp500/mqtt/types.py``). ``NavienMqttClient.publish`` now wraps
+  otherwise-uncaught ``AwsCrtError`` in ``MqttPublishError`` so ``awscrt``
+  exceptions no longer leak out of the public boundary.
+
+  .. code-block:: python
+
+     # OLD
+     from awscrt import mqtt
+     await client.publish(topic, payload, qos=mqtt.QoS.AT_LEAST_ONCE)
+
+     # NEW
+     from nwp500 import QoS
+     await client.publish(topic, payload, qos=QoS.AT_LEAST_ONCE)
 
 Fixed
 -----

src/nwp500/__init__.py

@@ -110,6 +110,7 @@
     MqttMetrics,
     NavienMqttClient,
     PeriodicRequestType,
+    QoS,
 )
 from nwp500.mqtt_events import (
     MqttClientEvents,
@@ -221,6 +222,7 @@
     "MqttMetrics",
     "ConnectionDropEvent",
     "ConnectionEvent",
+    "QoS",
     # Event Emitter
     "EventEmitter",
     "EventListener",

src/nwp500/mqtt/__init__.py

@@ -9,6 +9,7 @@
 - PeriodicRequestType: Enum for periodic request types
 - MqttDiagnosticsCollector: Metrics and diagnostics collector
 - MqttMetrics, ConnectionDropEvent, ConnectionEvent: Diagnostic types
+- QoS: Library-owned MQTT Quality of Service enum
 """
 
 from .client import NavienMqttClient
@@ -18,6 +19,7 @@
     MqttDiagnosticsCollector,
     MqttMetrics,
 )
+from .types import QoS
 from .utils import MqttConnectionConfig, PeriodicRequestType
 
 __all__ = [
@@ -28,4 +30,5 @@
     "MqttMetrics",
     "ConnectionDropEvent",
     "ConnectionEvent",
+    "QoS",
 ]

src/nwp500/mqtt/client.py

@@ -16,7 +16,6 @@
 from collections.abc import Callable, Sequence
 from typing import TYPE_CHECKING, Any, cast
 
-from awscrt import mqtt
 from awscrt.exceptions import AwsCrtError
 
 from ..auth import NavienAuthClient
@@ -41,6 +40,7 @@
 from .periodic import MqttPeriodicRequestManager
 from .reconnection import MqttReconnectionHandler
 from .subscriptions import MqttSubscriptionManager
+from .types import MqttConnectionHandle, QoS
 from .utils import (
     MqttConnectionConfig,
     PeriodicRequestType,
@@ -246,7 +246,7 @@ def __init__(
         self._diagnostics = MqttDiagnosticsCollector()
 
         # Connection state (simpler than checking _connection_manager)
-        self._connection: mqtt.Connection | None = None
+        self._connection: MqttConnectionHandle | None = None
         self._connected = False
         # Guards _active_reconnect / _deep_reconnect against re-entrancy.
         # While True, _on_connection_interrupted_internal will not forward
@@ -294,7 +294,7 @@ def _schedule_coroutine(self, coro: Any) -> None:
             future.add_done_callback(_log_scheduled_coroutine_result)
 
     def _on_connection_interrupted_internal(
-        self, connection: mqtt.Connection, error: AwsCrtError, **kwargs: Any
+        self, connection: MqttConnectionHandle, error: Exception, **kwargs: Any
     ) -> None:
         """Internal handler for connection interruption.
 
@@ -350,7 +350,7 @@ def _on_connection_interrupted_internal(
 
     def _on_connection_resumed_internal(
         self,
-        connection: mqtt.Connection,
+        connection: MqttConnectionHandle,
         return_code: Any,
         session_present: Any,
         **kwargs: Any,
@@ -885,7 +885,7 @@ async def subscribe(
         self,
         topic: str,
         callback: Callable[[str, dict[str, Any]], None],
-        qos: mqtt.QoS = mqtt.QoS.AT_LEAST_ONCE,
+        qos: QoS = QoS.AT_LEAST_ONCE,
     ) -> int:
         """
         Subscribe to an MQTT topic.
@@ -930,7 +930,7 @@ async def publish(
         self,
         topic: str,
         payload: dict[str, Any],
-        qos: mqtt.QoS = mqtt.QoS.AT_LEAST_ONCE,
+        qos: QoS = QoS.AT_LEAST_ONCE,
     ) -> int:
         """
         Publish a message to an MQTT topic.
@@ -990,9 +990,13 @@ async def publish(
                     retriable=True,
                 ) from e
 
-            # Other AWS CRT errors
+            # Other AWS CRT errors: wrap so awscrt exceptions never leak out
+            # of the public publish() boundary.
             _logger.error(f"Failed to publish to topic: {e}")
-            raise
+            raise MqttPublishError(
+                f"Failed to publish to MQTT topic: {e}",
+                retriable=True,
+            ) from e
 
     # Navien-specific convenience methods
 

src/nwp500/mqtt/command_queue.py

@@ -11,8 +11,7 @@
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any
 
-from awscrt import mqtt
-
+from .types import QoS
 from .utils import QueuedCommand, redact_topic
 
 if TYPE_CHECKING:
@@ -54,9 +53,7 @@ def __init__(self, config: MqttConnectionConfig):
             maxlen=config.max_queued_commands
         )
 
-    def enqueue(
-        self, topic: str, payload: dict[str, Any], qos: mqtt.QoS
-    ) -> None:
+    def enqueue(self, topic: str, payload: dict[str, Any], qos: QoS) -> None:
         """
         Add a command to the queue.
 

src/nwp500/mqtt/connection.py

@@ -14,14 +14,14 @@
 from collections.abc import Callable
 from typing import TYPE_CHECKING, Any
 
-from awscrt import mqtt
 from awscrt.exceptions import AwsCrtError
 from awsiot import mqtt_connection_builder
 
 from ..exceptions import (
     MqttCredentialsError,
     MqttNotConnectedError,
 )
+from .types import MqttConnectionHandle, QoS, to_awscrt_qos
 
 if TYPE_CHECKING:
     from ..auth import NavienAuthClient
@@ -50,10 +50,10 @@ def __init__(
         config: MqttConnectionConfig,
         auth_client: NavienAuthClient,
         on_connection_interrupted: (
-            Callable[[mqtt.Connection, AwsCrtError], None] | None
+            Callable[[MqttConnectionHandle, Exception], None] | None
         ) = None,
         on_connection_resumed: (
-            Callable[[mqtt.Connection, Any, Any | None], None] | None
+            Callable[[MqttConnectionHandle, Any, Any | None], None] | None
         ) = None,
     ):
         """
@@ -87,7 +87,7 @@ def __init__(
 
         self.config = config
         self._auth_client = auth_client
-        self._connection: mqtt.Connection | None = None
+        self._connection: MqttConnectionHandle | None = None
         self._connected = False
         self._on_connection_interrupted = on_connection_interrupted
         self._on_connection_resumed = on_connection_resumed
@@ -350,7 +350,7 @@ async def close(self) -> None:
     async def subscribe(
         self,
         topic: str,
-        qos: mqtt.QoS,
+        qos: QoS,
         callback: Callable[..., None] | None = None,
     ) -> tuple[Any, int]:
         """
@@ -376,7 +376,7 @@ async def subscribe(
         # Use shield to prevent cancellation from propagating to
         # underlying future
         subscribe_future_raw, packet_id_raw = self._connection.subscribe(
-            topic=topic, qos=qos, callback=callback
+            topic=topic, qos=to_awscrt_qos(qos), callback=callback
         )
         subscribe_future = subscribe_future_raw
         packet_id = packet_id_raw
@@ -422,7 +422,7 @@ async def publish(
         self,
         topic: str,
         payload: str | dict[str, Any],
-        qos: mqtt.QoS = mqtt.QoS.AT_LEAST_ONCE,
+        qos: QoS = QoS.AT_LEAST_ONCE,
     ) -> int:
         """
         Publish a message to an MQTT topic.
@@ -453,7 +453,7 @@ async def publish(
 
         # Publish and get the concurrent.futures.Future
         publish_future_raw, packet_id_raw = self._connection.publish(
-            topic=topic, payload=payload_bytes, qos=qos
+            topic=topic, payload=payload_bytes, qos=to_awscrt_qos(qos)
         )
         publish_future = publish_future_raw
         packet_id = int(packet_id_raw)
@@ -488,7 +488,7 @@ def is_connected(self) -> bool:
         return self._connected
 
     @property
-    def connection(self) -> mqtt.Connection | None:
+    def connection(self) -> MqttConnectionHandle | None:
         """Get the underlying MQTT connection.
 
         Returns:

src/nwp500/mqtt/subscriptions.py

@@ -16,7 +16,6 @@
 from collections.abc import Callable
 from typing import TYPE_CHECKING, Any, cast
 
-from awscrt import mqtt
 from awscrt.exceptions import AwsCrtError
 from pydantic import ValidationError
 
@@ -35,6 +34,7 @@
 from ..mqtt_events import FeatureReceivedEvent, StatusReceivedEvent
 from ..topic_builder import MqttTopicBuilder
 from .state_tracker import DeviceStateTracker
+from .types import QoS, to_awscrt_qos
 from .utils import get_response_data, redact_topic, topic_matches_pattern
 
 if TYPE_CHECKING:
@@ -87,7 +87,7 @@ def __init__(
         self._operation_timeout = operation_timeout
 
         # Track subscriptions and handlers
-        self._subscriptions: dict[str, mqtt.QoS] = {}
+        self._subscriptions: dict[str, QoS] = {}
         self._message_handlers: dict[
             str, list[Callable[[str, dict[str, Any]], None]]
         ] = {}
@@ -96,7 +96,7 @@ def __init__(
         self._state_tracker = DeviceStateTracker(event_emitter)
 
     @property
-    def subscriptions(self) -> dict[str, mqtt.QoS]:
+    def subscriptions(self) -> dict[str, QoS]:
         """Get current subscriptions."""
         return self._subscriptions.copy()
 
@@ -205,7 +205,7 @@ async def subscribe(
         self,
         topic: str,
         callback: Callable[[str, dict[str, Any]], None],
-        qos: mqtt.QoS = mqtt.QoS.AT_LEAST_ONCE,
+        qos: QoS = QoS.AT_LEAST_ONCE,
     ) -> int:
         """
         Subscribe to an MQTT topic.
@@ -247,7 +247,9 @@ async def subscribe(
             # callbacks (avoids InvalidStateError); the underlying
             # subscribe completes independently.
             subscribe_future, packet_id = self._connection.subscribe(
-                topic=topic, qos=qos, callback=self._on_message_received
+                topic=topic,
+                qos=to_awscrt_qos(qos),
+                callback=self._on_message_received,
             )
             awaitable_future = asyncio.wrap_future(subscribe_future)
             try:
@@ -457,7 +459,7 @@ async def resubscribe_all(self) -> None:
             qos_map = dict(subscriptions_to_restore)
             for topic in failed_subscriptions:
                 self._subscriptions[topic] = qos_map.get(
-                    topic, mqtt.QoS.AT_LEAST_ONCE
+                    topic, QoS.AT_LEAST_ONCE
                 )
                 self._message_handlers[topic] = handlers_to_restore.get(
                     topic, []

src/nwp500/mqtt/types.py

@@ -0,0 +1,50 @@
+"""Library-owned MQTT transport types.
+
+This module defines transport-level types that the library exposes on its own
+public and internal signatures instead of leaking ``awscrt`` SDK types. It is
+the single boundary where the library's own :class:`QoS` enum is translated
+to and from :class:`awscrt.mqtt.QoS`, keeping ``awscrt`` an implementation
+detail that could be swapped for a different MQTT transport in the future.
+"""
+
+from enum import IntEnum
+
+from awscrt import mqtt
+
+__author__ = "Emmanuel Levijarvi"
+__copyright__ = "Emmanuel Levijarvi"
+__license__ = "MIT"
+
+
+class QoS(IntEnum):
+    """MQTT Quality of Service level.
+
+    Mirrors the MQTT specification's QoS levels. Integer values match the
+    wire protocol (and :class:`awscrt.mqtt.QoS`) so translation is a direct
+    value mapping.
+    """
+
+    AT_MOST_ONCE = 0
+    """QoS 0: fire-and-forget delivery with no acknowledgement."""
+
+    AT_LEAST_ONCE = 1
+    """QoS 1: acknowledged delivery; duplicates possible."""
+
+    EXACTLY_ONCE = 2
+    """QoS 2: acknowledged, exactly-once delivery."""
+
+
+# Opaque handle for the underlying MQTT connection object. Typed as an alias so
+# callers and internal code refer to the library's name rather than the
+# concrete ``awscrt`` type, keeping the transport swappable.
+type MqttConnectionHandle = mqtt.Connection
+
+
+def to_awscrt_qos(qos: QoS) -> mqtt.QoS:
+    """Translate a library :class:`QoS` to :class:`awscrt.mqtt.QoS`."""
+    return mqtt.QoS(int(qos))
+
+
+def from_awscrt_qos(qos: mqtt.QoS) -> QoS:
+    """Translate an :class:`awscrt.mqtt.QoS` to a library :class:`QoS`."""
+    return QoS(int(qos))

src/nwp500/mqtt/utils.py

@@ -12,9 +12,8 @@
 from enum import StrEnum
 from typing import Any
 
-from awscrt import mqtt
-
 from ..config import AWS_IOT_ENDPOINT, AWS_REGION
+from .types import QoS
 
 __author__ = "Emmanuel Levijarvi"
 __copyright__ = "Emmanuel Levijarvi"
@@ -286,7 +285,7 @@ class QueuedCommand:
 
     topic: str
     payload: dict[str, Any]
-    qos: mqtt.QoS
+    qos: QoS
     timestamp: datetime