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

Introduce a library-owned QoS IntEnum and MqttConnectionHandle alias in
nwp500/mqtt/types.py, translating to/from awscrt.mqtt.QoS only at the
connection-layer boundary. Replace awscrt.mqtt.QoS and awscrt.mqtt.Connection
type references across NavienMqttClient, MqttConnection,
MqttSubscriptionManager, MqttCommandQueue and QueuedCommand. Wrap otherwise
uncaught AwsCrtError in MqttPublishError at the public publish() boundary so
awscrt exceptions no longer leak. Export QoS from nwp500 and nwp500.mqtt.
Update tests to use the library QoS enum.

Author: eman

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