From 251056a8d0c20b2a139656f7b3ac1d6b45c9e3d8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 25 Jul 2026 17:54:43 +0000 Subject: [PATCH 1/5] feat: add MQTT-over-WebSocket support with device telemetry decoding --- docs/API_EXAMPLES.md | 108 ++++ pyproject.toml | 1 + src/pymbrewclient/__init__.py | 8 +- src/pymbrewclient/client.py | 42 +- src/pymbrewclient/mqtt/__init__.py | 44 ++ src/pymbrewclient/mqtt/client.py | 471 +++++++++++++++++ src/pymbrewclient/mqtt/models.py | 152 ++++++ src/pymbrewclient/mqtt/proto.py | 226 ++++++++ src/pymbrewclient/rest/client.py | 19 +- src/pymbrewclient/rest/models.py | 9 + tests/fixtures/device_log.bin | Bin 0 -> 29 bytes tests/test_mqtt.py | 806 +++++++++++++++++++++++++++++ 12 files changed, 1883 insertions(+), 3 deletions(-) create mode 100644 src/pymbrewclient/mqtt/__init__.py create mode 100644 src/pymbrewclient/mqtt/client.py create mode 100644 src/pymbrewclient/mqtt/models.py create mode 100644 src/pymbrewclient/mqtt/proto.py create mode 100644 tests/fixtures/device_log.bin create mode 100644 tests/test_mqtt.py diff --git a/docs/API_EXAMPLES.md b/docs/API_EXAMPLES.md index 0c1caee..ce67caf 100644 --- a/docs/API_EXAMPLES.md +++ b/docs/API_EXAMPLES.md @@ -164,3 +164,111 @@ The response is organized into different categories based on device state: - `stage`: "Serving" - Contains beer information - Temperature maintained at serving temperature + +--- + +## MQTT-over-WebSocket + +`pymbrewclient` supports real-time device telemetry over MQTT-over-WebSocket using +`create_mqtt_client()`. The underlying broker is `broker.minibrew.io:15675/ws` (TLS). + +> **Security warning:** The MiniBrew REST API token is used as the MQTT +> password. Do **not** log the `MqttClient` object, its `repr()`, or any +> callback data in contexts where credentials could be exposed. + +### Connection example + +```python +from pymbrewclient import BreweryClient + +client = BreweryClient(username="you@example.com", *****) + +with client.create_mqtt_client() as mqtt: + mqtt.on_connected(lambda: print("Connected")) + mqtt.on_disconnected(lambda: print("Disconnected")) + mqtt.on_reconnecting(lambda: print("Reconnecting...")) + mqtt.on_error(lambda err: print(f"Error: {err}")) + + mqtt.subscribe_device_logs("7391Q4827-5NZC8R2M") + + import time + time.sleep(30) +``` + +### Raw-message example + +```python +from pymbrewclient.mqtt.models import MqttMessage + +with client.create_mqtt_client() as mqtt: + def handle(msg: MqttMessage) -> None: + print(msg.topic, msg.received_at, msg.payload.hex()) + + mqtt.on_message(handle) + mqtt.subscribe_device_logs("7391Q4827-5NZC8R2M") + import time; time.sleep(30) +``` + +### Decoded telemetry example + +```python +from pymbrewclient.mqtt.models import DeviceLogMessage + +with client.create_mqtt_client() as mqtt: + def handle_log(msg: DeviceLogMessage) -> None: + if msg.decode_error: + print(f"Decode error: {msg.decode_error}") + return + print(f"Session {msg.session_id}") + print(f"Process state: {msg.process_state}") + print(f"Current temp: {msg.current_temperature}°C") + print(f"Target temp: {msg.target_temperature}°C") + + mqtt.on_device_log(handle_log) + mqtt.subscribe_device_logs("7391Q4827-5NZC8R2M") + import time; time.sleep(30) +``` + +### `next_action_at` example + +```python +with client.create_mqtt_client() as mqtt: + def handle_log(msg: DeviceLogMessage) -> None: + if msg.next_action_at: + print(f"Next action at (UTC): {msg.next_action_at.isoformat()}") + # Home Assistant can display this UTC timestamp in the user's local timezone. + + mqtt.on_device_log(handle_log) + mqtt.subscribe_device_logs("7391Q4827-5NZC8R2M") + import time; time.sleep(30) +``` + +`next_action_at` is calculated as: + +```python +next_action_at = message_timestamp + timedelta(seconds=seconds_until_next_action) +``` + +It is always a timezone-aware UTC `datetime`. No local timezone conversion is +applied inside the library; downstream consumers (such as Home Assistant) should +handle display timezone conversion. + +### Limitations: protobuf schema + +MiniBrew's official protobuf schema (`minibrew/minibrew-protobuf`) is a private +repository that is not publicly accessible. The field numbers used in +`DeviceLogMessage` were **reconstructed** from: + +- The MiniBrew REST API response structure (`Device` dataclass field names) +- The community-maintained `minibrew/enduser-docker-server` project documentation + +**Until official field numbers are confirmed, decoded telemetry values may be +incorrect.** The raw `payload` bytes are always preserved in `MqttMessage.payload` +and `DeviceLogMessage.payload`. To inspect a live message yourself: + +```bash +protoc --decode_raw < captured_payload.bin +``` + +The raw field numbers are also exposed via `DeviceLogMessage.raw_fields` for +inspection without any mapping. diff --git a/pyproject.toml b/pyproject.toml index ff8c2ce..5f7541c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ dependencies = [ "requests>=2.32.3,<3", "typer>=0.12.5,<1", "rich>=13.9.4,<16", + "paho-mqtt>=2.0.0,<3", ] dynamic = ["version"] diff --git a/src/pymbrewclient/__init__.py b/src/pymbrewclient/__init__.py index b194ec1..a449f19 100644 --- a/src/pymbrewclient/__init__.py +++ b/src/pymbrewclient/__init__.py @@ -35,12 +35,17 @@ # # Disclaimer: This software is an independent project and is not affiliated with, endorsed by, or associated with MiniBrew. MiniBrew's trademarks, logos, API, and other intellectual property are owned by MiniBrew and are not included in this software. Users are responsible for complying with MiniBrew's terms of service when using this software. from .client import BreweryClient, BreweryClientError, DeviceLookupError -from .rest.models import Device, TokenResponse, ApiResponse, BreweryOverview, Session, DeviceDetails, Beer +from .mqtt.client import MqttClient +from .mqtt.models import DeviceLogMessage, MqttMessage +from .rest.models import Device, TokenResponse, ApiResponse, BreweryOverview, Session, DeviceDetails, Beer, UserProfile __all__ = [ "BreweryClient", "BreweryClientError", "DeviceLookupError", + "MqttClient", + "MqttMessage", + "DeviceLogMessage", "Device", "TokenResponse", "ApiResponse", @@ -48,4 +53,5 @@ "Session", "DeviceDetails", "Beer", + "UserProfile", ] diff --git a/src/pymbrewclient/client.py b/src/pymbrewclient/client.py index 6b9c96a..81f102e 100644 --- a/src/pymbrewclient/client.py +++ b/src/pymbrewclient/client.py @@ -37,7 +37,7 @@ from datetime import datetime from pymbrewclient.rest.client import RestApiClient -from pymbrewclient.rest.models import BreweryOverview, Device, Session, TokenResponse +from pymbrewclient.rest.models import BreweryOverview, Device, Session, TokenResponse, UserProfile class BreweryClientError(ValueError): @@ -127,3 +127,43 @@ def get_process_estimate_remaining_seconds( """Return a locally calculated remaining duration for a selected device.""" device = self.get_device(device_uuid=device_uuid, session_id=session_id) return device.process_estimate_remaining_seconds + + def get_user_profile(self) -> UserProfile: + """ + Fetch and return the authenticated user's profile. + + The user UUID is required to construct MQTT credentials. + + :return: A UserProfile object containing the user UUID and profile data. + """ + return self.client.get_user_profile() + + def create_mqtt_client(self) -> "MqttClient": + """ + Create and return a configured MQTT-over-WebSocket client. + + The current REST API token is reused as the MQTT password. A fresh + token is obtained if the current one has expired. A new random + ``client_uuid`` is generated for each call, so multiple independent + MQTT clients can be created from the same :class:`BreweryClient`. + + .. warning:: + + The API token is used as the MQTT password. Do not log the + returned :class:`~pymbrewclient.mqtt.MqttClient` instance in + contexts that would reveal sensitive credentials. + + :return: A ready-to-connect :class:`~pymbrewclient.mqtt.MqttClient`. + """ + from pymbrewclient.mqtt.client import MqttClient + + self.client._ensure_token() + profile = self.get_user_profile() + return MqttClient(api_token=self.client.token, user_uuid=profile.uuid) + + +# Local alias for the forward reference in create_mqtt_client's type hint +try: + from pymbrewclient.mqtt.client import MqttClient # noqa: E402, F401 +except ImportError: + pass diff --git a/src/pymbrewclient/mqtt/__init__.py b/src/pymbrewclient/mqtt/__init__.py new file mode 100644 index 0000000..a0985b5 --- /dev/null +++ b/src/pymbrewclient/mqtt/__init__.py @@ -0,0 +1,44 @@ +# "Commons Clause" License Condition v1.0 +# +# The Software is provided to you by the Licensor under the License, as defined below, subject to the following condition. +# +# Without limiting other conditions in the License, the grant of rights under the License will not include, and the License does not grant to you, the right to Sell the Software. +# +# For purposes of the foregoing, "Sell" means practicing any or all of the rights granted to you under the License to provide to third parties, for a fee or other consideration (including without limitation fees for hosting or consulting/ support services related to the Software), a product or service whose value derives, entirely or substantially, from the functionality of the Software. Any license notice or attribution required by the License must also include this Commons Clause License Condition notice. +# +# Software: pymbrewclient +# License: MIT License +# Licensor: Stuart Pearson +# +# +# MIT License +# +# Copyright (c) 2024 Stuart Pearson +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# +# Disclaimer: This software is an independent project and is not affiliated with, endorsed by, or associated with MiniBrew. MiniBrew's trademarks, logos, API, and other intellectual property are owned by MiniBrew and are not included in this software. Users are responsible for complying with MiniBrew's terms of service when using this software. +from .client import MqttClient +from .models import DeviceLogMessage, MqttMessage + +__all__ = [ + "MqttClient", + "MqttMessage", + "DeviceLogMessage", +] diff --git a/src/pymbrewclient/mqtt/client.py b/src/pymbrewclient/mqtt/client.py new file mode 100644 index 0000000..dd6f582 --- /dev/null +++ b/src/pymbrewclient/mqtt/client.py @@ -0,0 +1,471 @@ +# "Commons Clause" License Condition v1.0 +# +# The Software is provided to you by the Licensor under the License, as defined below, subject to the following condition. +# +# Without limiting other conditions in the License, the grant of rights under the License will not include, and the License does not grant to you, the right to Sell the Software. +# +# For purposes of the foregoing, "Sell" means practicing any or all of the rights granted to you under the License to provide to third parties, for a fee or other consideration (including without limitation fees for hosting or consulting/ support services related to the Software"), a product or service whose value derives, entirely or substantially, from the functionality of the Software. Any license notice or attribution required by the License must also include this Commons Clause License Condition notice. +# +# Software: pymbrewclient +# License: MIT License +# Licensor: Stuart Pearson +# +# +# MIT License +# +# Copyright (c) 2024 Stuart Pearson +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# +# Disclaimer: This software is an independent project and is not affiliated with, endorsed by, or associated with MiniBrew. MiniBrew's trademarks, logos, API, and other intellectual property are owned by MiniBrew and are not included in this software. Users are responsible for complying with MiniBrew's terms of service when using this software. +""" +MQTT-over-WebSocket client for the MiniBrew Brewery Portal. + +Security note +------------- +The MiniBrew REST API token is used as the MQTT password. **Never log the +token**, include it in exception messages, or pass it to external systems. +This module takes care to keep the token out of ``__repr__``, log output, +and callback arguments. +""" +import logging +import threading +import uuid +from collections.abc import Callable +from datetime import datetime, timezone +from types import TracebackType +from typing import Any + +import paho.mqtt.client as _paho + +from .models import DeviceLogMessage, MqttMessage +from .proto import decode_device_log + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +BROKER_HOST: str = "broker.minibrew.io" +BROKER_PORT: int = 15675 +WS_PATH: str = "/ws" +KEEPALIVE: int = 60 +RECONNECT_DELAY: int = 5 + +_DEVICE_LOG_TOPIC_PREFIX = "devices/logs/" +_DEVICE_TOPIC_PREFIX_PARTS = ("devices",) + + +def _extract_device_uuid(topic: str) -> str | None: + """Return the device UUID from a ``devices/{type}/{uuid}`` topic, or ``None``.""" + parts = topic.split("/") + if len(parts) == 3 and parts[0] == "devices": + return parts[2] + return None + + +# --------------------------------------------------------------------------- +# Callback type aliases +# --------------------------------------------------------------------------- + +ConnectedCallback = Callable[[], None] +DisconnectedCallback = Callable[[], None] +ReconnectingCallback = Callable[[], None] +ErrorCallback = Callable[[Exception], None] +RawMessageCallback = Callable[[MqttMessage], None] +DeviceLogCallback = Callable[[DeviceLogMessage], None] + + +# --------------------------------------------------------------------------- +# MqttClient +# --------------------------------------------------------------------------- + + +class MqttClient: + """MQTT-over-WebSocket client for the MiniBrew Brewery Portal. + + Do not instantiate directly; use + :meth:`~pymbrewclient.client.BreweryClient.create_mqtt_client` instead. + + Example:: + + client = BreweryClient(username="you@example.com", ******) + + with client.create_mqtt_client() as mqtt: + mqtt.on_connected(lambda: print("connected")) + mqtt.subscribe_device_logs("7391Q4827-5NZC8R2M") + + .. warning:: + + The MiniBrew REST API token is used as the MQTT password. + **Do not log, print, or expose the MqttClient instance in contexts that + would reveal the token.** This class's ``__repr__`` deliberately omits + sensitive fields. + """ + + def __init__(self, api_token: str, user_uuid: str) -> None: + """ + :param api_token: A valid MiniBrew REST API token. Used as the MQTT + password. **Never log this value.** + :param user_uuid: The authenticated user's UUID, used to construct the + MQTT username (``breweryportal-{user_uuid}``). + """ + self._api_token = api_token # intentionally private — never logged + self._user_uuid = user_uuid + + self._client_uuid = str(uuid.uuid4()) + self._client_id = f"breweryportal-{self._client_uuid}" + self._username = f"breweryportal-{self._user_uuid}" + + self._subscriptions: set[str] = set() + self._subscription_lock = threading.Lock() + + self._connected = False + self._ever_connected = False + + # Callbacks + self._on_connected_callbacks: list[ConnectedCallback] = [] + self._on_disconnected_callbacks: list[DisconnectedCallback] = [] + self._on_reconnecting_callbacks: list[ReconnectingCallback] = [] + self._on_error_callbacks: list[ErrorCallback] = [] + self._on_raw_message_callbacks: list[RawMessageCallback] = [] + self._on_device_log_callbacks: list[DeviceLogCallback] = [] + + self._paho_client = self._build_paho_client() + + # ------------------------------------------------------------------ + # Internal paho-mqtt construction + # ------------------------------------------------------------------ + + def _build_paho_client(self) -> _paho.Client: + """Construct and configure the underlying paho-mqtt client.""" + will_topic = f"apps/lastwill/{self._client_id}" + + client = _paho.Client( + callback_api_version=_paho.CallbackAPIVersion.VERSION2, + client_id=self._client_id, + protocol=_paho.MQTTv311, + transport="websockets", + clean_session=True, + ) + client.ws_set_options(path=WS_PATH) + client.tls_set() # TLS with system CA bundle; certificate verification enabled + client.username_pw_set(self._username, self._api_token) + client.will_set(topic=will_topic, payload="offline", qos=0, retain=False) + client.reconnect_delay_set(min_delay=RECONNECT_DELAY, max_delay=RECONNECT_DELAY) + + client.on_connect = self._on_paho_connect + client.on_disconnect = self._on_paho_disconnect + client.on_message = self._on_paho_message + client.on_pre_connect = self._on_paho_pre_connect + + return client + + # ------------------------------------------------------------------ + # paho callbacks + # ------------------------------------------------------------------ + + def _on_paho_pre_connect(self, client: _paho.Client, userdata: Any) -> None: # noqa: ANN401 + """Called by paho just before a (re)connection attempt.""" + if self._ever_connected: + logger.debug("MQTT reconnecting to %s:%d", BROKER_HOST, BROKER_PORT) + self._fire_reconnecting() + + def _on_paho_connect( + self, + client: _paho.Client, + userdata: Any, # noqa: ANN401 + connect_flags: _paho.ConnectFlags, + reason_code: _paho.ReasonCode, + properties: Any, # noqa: ANN401 + ) -> None: + """Called when paho establishes or re-establishes the connection.""" + if reason_code.is_failure: + logger.warning("MQTT connect failed: %s", reason_code) + self._fire_error(ConnectionError(f"MQTT connect failed: {reason_code}")) + return + + logger.debug("MQTT connected (session_present=%s)", connect_flags.session_present) + self._connected = True + self._ever_connected = True + + # Resubscribe to all topics (handles reconnect + clean_session=True) + with self._subscription_lock: + topics = list(self._subscriptions) + for topic in topics: + logger.debug("MQTT resubscribing to %s", topic) + client.subscribe(topic) + + self._fire_connected() + + def _on_paho_disconnect( + self, + client: _paho.Client, + userdata: Any, # noqa: ANN401 + disconnect_flags: _paho.DisconnectFlags, + reason_code: _paho.ReasonCode, + properties: Any, # noqa: ANN401 + ) -> None: + """Called when paho loses the connection.""" + self._connected = False + logger.debug("MQTT disconnected: %s", reason_code) + self._fire_disconnected() + + def _on_paho_message(self, client: _paho.Client, userdata: Any, msg: _paho.MQTTMessage) -> None: # noqa: ANN401 + """Called for every incoming MQTT PUBLISH.""" + received_at = datetime.now(tz=timezone.utc) + device_uuid = _extract_device_uuid(msg.topic) + + raw_msg = MqttMessage( + topic=msg.topic, + payload=bytes(msg.payload), + received_at=received_at, + device_uuid=device_uuid, + ) + + self._fire_raw_message(raw_msg) + + if msg.topic.startswith(_DEVICE_LOG_TOPIC_PREFIX): + decoded = decode_device_log(raw_msg) + if decoded.decode_error: + logger.debug("MQTT device-log decode error on %s: %s", msg.topic, decoded.decode_error) + self._fire_device_log(decoded) + + # ------------------------------------------------------------------ + # Callback firing helpers + # ------------------------------------------------------------------ + + def _fire_connected(self) -> None: + for cb in self._on_connected_callbacks: + try: + cb() + except Exception as exc: # noqa: BLE001 + logger.debug("Exception in on_connected callback: %s", exc) + + def _fire_disconnected(self) -> None: + for cb in self._on_disconnected_callbacks: + try: + cb() + except Exception as exc: # noqa: BLE001 + logger.debug("Exception in on_disconnected callback: %s", exc) + + def _fire_reconnecting(self) -> None: + for cb in self._on_reconnecting_callbacks: + try: + cb() + except Exception as exc: # noqa: BLE001 + logger.debug("Exception in on_reconnecting callback: %s", exc) + + def _fire_error(self, error: Exception) -> None: + for cb in self._on_error_callbacks: + try: + cb(error) + except Exception as exc: # noqa: BLE001 + logger.debug("Exception in on_error callback: %s", exc) + + def _fire_raw_message(self, msg: MqttMessage) -> None: + for cb in self._on_raw_message_callbacks: + try: + cb(msg) + except Exception as exc: # noqa: BLE001 + logger.debug("Exception in on_message callback: %s", exc) + + def _fire_device_log(self, msg: DeviceLogMessage) -> None: + for cb in self._on_device_log_callbacks: + try: + cb(msg) + except Exception as exc: # noqa: BLE001 + logger.debug("Exception in on_device_log callback: %s", exc) + + # ------------------------------------------------------------------ + # Public connection API + # ------------------------------------------------------------------ + + def connect(self) -> None: + """Open the WebSocket connection to the MiniBrew MQTT broker. + + Starts a background network thread. Returns immediately; connection + events are delivered via :meth:`on_connected` / :meth:`on_disconnected` + callbacks. + """ + logger.debug("MQTT connecting to %s:%d%s", BROKER_HOST, BROKER_PORT, WS_PATH) + self._paho_client.connect(host=BROKER_HOST, port=BROKER_PORT, keepalive=KEEPALIVE) + self._paho_client.loop_start() + + def disconnect(self) -> None: + """Disconnect from the broker and stop the background network thread. + + Safe to call even when not connected. + """ + logger.debug("MQTT disconnecting") + self._paho_client.disconnect() + self._paho_client.loop_stop() + self._connected = False + + # ------------------------------------------------------------------ + # Subscription API + # ------------------------------------------------------------------ + + def subscribe(self, topic: str, qos: int = 0) -> None: + """Subscribe to an arbitrary MQTT topic. + + The topic is remembered and resubscribed automatically after each + reconnect. + + :param topic: Full MQTT topic string (wildcards ``+`` and ``#`` are + supported). + :param qos: Quality of Service level (0 or 1). + """ + with self._subscription_lock: + self._subscriptions.add(topic) + if self._connected: + self._paho_client.subscribe(topic, qos=qos) + logger.debug("MQTT subscribed to %s (qos=%d)", topic, qos) + + def unsubscribe(self, topic: str) -> None: + """Unsubscribe from an arbitrary MQTT topic. + + The topic is also removed from the auto-resubscribe set. + + :param topic: Full MQTT topic string, as previously passed to + :meth:`subscribe`. + """ + with self._subscription_lock: + self._subscriptions.discard(topic) + if self._connected: + self._paho_client.unsubscribe(topic) + logger.debug("MQTT unsubscribed from %s", topic) + + @staticmethod + def device_log_topic(device_uuid: str) -> str: + """Return the ``devices/logs/{device_uuid}`` topic for a device. + + :param device_uuid: Device serial number or UUID. + """ + return f"{_DEVICE_LOG_TOPIC_PREFIX}{device_uuid}" + + def subscribe_device_logs(self, device_uuid: str, qos: int = 0) -> None: + """Subscribe to real-time telemetry logs for a specific device. + + Incoming messages are delivered as :class:`~pymbrewclient.mqtt.models.DeviceLogMessage` + instances to callbacks registered with :meth:`on_device_log`. + + :param device_uuid: Device serial number or UUID (e.g. ``"7391Q4827-5NZC8R2M"``). + :param qos: Quality of Service level (0 or 1). + """ + self.subscribe(self.device_log_topic(device_uuid), qos=qos) + + def unsubscribe_device_logs(self, device_uuid: str) -> None: + """Unsubscribe from device log telemetry for a specific device. + + :param device_uuid: Device serial number or UUID. + """ + self.unsubscribe(self.device_log_topic(device_uuid)) + + # ------------------------------------------------------------------ + # Callback registration + # ------------------------------------------------------------------ + + def on_connected(self, callback: ConnectedCallback) -> None: + """Register a callback invoked when the connection is established. + + For reconnect events the callback fires after all subscriptions are + restored. + + :param callback: A zero-argument callable. + """ + self._on_connected_callbacks.append(callback) + + def on_disconnected(self, callback: DisconnectedCallback) -> None: + """Register a callback invoked when the connection is lost. + + :param callback: A zero-argument callable. + """ + self._on_disconnected_callbacks.append(callback) + + def on_reconnecting(self, callback: ReconnectingCallback) -> None: + """Register a callback invoked just before a reconnection attempt. + + :param callback: A zero-argument callable. + """ + self._on_reconnecting_callbacks.append(callback) + + def on_error(self, callback: ErrorCallback) -> None: + """Register a callback invoked when a connection-level error occurs. + + The callback receives the exception as its sole argument. The token + is never included in the exception object. + + :param callback: A callable accepting one :class:`Exception` argument. + """ + self._on_error_callbacks.append(callback) + + def on_message(self, callback: RawMessageCallback) -> None: + """Register a callback invoked for every incoming MQTT message. + + The callback receives a :class:`~pymbrewclient.mqtt.models.MqttMessage` + containing the raw payload bytes. + + :param callback: A callable accepting one + :class:`~pymbrewclient.mqtt.models.MqttMessage` argument. + """ + self._on_raw_message_callbacks.append(callback) + + def on_device_log(self, callback: DeviceLogCallback) -> None: + """Register a callback invoked for decoded device-log messages. + + The callback receives a :class:`~pymbrewclient.mqtt.models.DeviceLogMessage`. + If decoding fails, ``decode_error`` is set on the message; the raw + payload is always preserved. + + :param callback: A callable accepting one + :class:`~pymbrewclient.mqtt.models.DeviceLogMessage` argument. + """ + self._on_device_log_callbacks.append(callback) + + # ------------------------------------------------------------------ + # Context manager + # ------------------------------------------------------------------ + + def __enter__(self) -> "MqttClient": + self.connect() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + self.disconnect() + + # ------------------------------------------------------------------ + # Representation — token intentionally excluded + # ------------------------------------------------------------------ + + def __repr__(self) -> str: + return ( + f"MqttClient(" + f"client_id={self._client_id!r}, " + f"username={self._username!r}, " + f"broker={BROKER_HOST!r}:{BROKER_PORT}, " + f"connected={self._connected!r}" + f")" + ) diff --git a/src/pymbrewclient/mqtt/models.py b/src/pymbrewclient/mqtt/models.py new file mode 100644 index 0000000..0116275 --- /dev/null +++ b/src/pymbrewclient/mqtt/models.py @@ -0,0 +1,152 @@ +# "Commons Clause" License Condition v1.0 +# +# The Software is provided to you by the Licensor under the License, as defined below, subject to the following condition. +# +# Without limiting other conditions in the License, the grant of rights under the License will not include, and the License does not grant to you, the right to Sell the Software. +# +# For purposes of the foregoing, "Sell" means practicing any or all of the rights granted to you under the License to provide to third parties, for a fee or other consideration (including without limitation fees for hosting or consulting/ support services related to the Software), a product or service whose value derives, entirely or substantially, from the functionality of the Software. Any license notice or attribution required by the License must also include this Commons Clause License Condition notice. +# +# Software: pymbrewclient +# License: MIT License +# Licensor: Stuart Pearson +# +# +# MIT License +# +# Copyright (c) 2024 Stuart Pearson +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# +# Disclaimer: This software is an independent project and is not affiliated with, endorsed by, or associated with MiniBrew. MiniBrew's trademarks, logos, API, and other intellectual property are owned by MiniBrew and are not included in this software. Users are responsible for complying with MiniBrew's terms of service when using this software. +""" +Typed message models for the MiniBrew MQTT client. +""" +from dataclasses import dataclass, field +from datetime import datetime + + +@dataclass +class MqttMessage: + """A raw MQTT message received from the broker.""" + + topic: str + """The full MQTT topic the message was published on.""" + + payload: bytes + """The raw message payload bytes, always preserved.""" + + received_at: datetime + """Timezone-aware UTC datetime when this message was received by the client.""" + + device_uuid: str | None = None + """ + Device UUID extracted from the topic when the topic follows the + ``devices/{message_type}/{device_uuid}`` pattern; ``None`` otherwise. + """ + + +@dataclass +class DeviceLogMessage: + """ + A decoded MiniBrew device-log telemetry message from the + ``devices/logs/{SerialNumber}`` MQTT topic. + + **Schema caveat**: MiniBrew's official ``.proto`` source + (``minibrew/minibrew-protobuf``) is a private repository and is not + publicly available. The field numbers used for decoding were + reconstructed from the MiniBrew REST API response structure and the + community-maintained ``minibrew/enduser-docker-server`` project. + Fields may be mapped to wrong values until the official schema is + confirmed. The raw :attr:`payload` is always retained. + + All decoded fields default to ``None`` when absent from the message + or when decoding fails. + """ + + topic: str + """The full MQTT topic.""" + + payload: bytes + """Raw protobuf bytes — always present regardless of decode success.""" + + received_at: datetime + """Timezone-aware UTC datetime when this message was received.""" + + device_uuid: str | None = None + """Device serial / UUID extracted from the topic path.""" + + # --- Decoded telemetry fields ---------------------------------------- + + session_id: int | None = None + """Active brewing session ID (protobuf field 1, best-effort).""" + + device_timestamp: datetime | None = None + """ + Timezone-aware UTC datetime from the device's own clock + (protobuf field 2, best-effort). Derived from a Unix epoch integer. + """ + + process_state: int | None = None + """ + Process state integer (protobuf field 3, best-effort). + See ``PROCESS_STATE_LABELS`` in the MiniBrew community docs for a + complete mapping (e.g. 80 → FERMENTATION_TEMP_CONTROL). + """ + + user_action: int | None = None + """ + User-action state integer (protobuf field 4, best-effort). + See ``USER_ACTION_LABELS`` in the MiniBrew community docs. + """ + + current_temperature: float | None = None + """Current temperature in °C (protobuf field 5, best-effort).""" + + target_temperature: float | None = None + """Target temperature in °C (protobuf field 6, best-effort).""" + + remaining_duration_seconds: int | None = None + """Remaining process duration in seconds (protobuf field 7, best-effort).""" + + seconds_until_next_action: int | None = None + """Seconds until the next scheduled action (protobuf field 8, best-effort).""" + + next_action_at: datetime | None = None + """ + Calculated UTC datetime of the next scheduled action:: + + next_action_at = device_timestamp + timedelta(seconds=seconds_until_next_action) + + ``None`` when either :attr:`device_timestamp` or + :attr:`seconds_until_next_action` is absent. + """ + + decode_error: str | None = None + """ + A brief description of any error encountered during protobuf decoding. + When set, some or all decoded telemetry fields may be ``None``. + The raw :attr:`payload` is unaffected. + """ + + raw_fields: dict[int, list[object]] = field(default_factory=dict) + """ + All raw protobuf field numbers and their decoded wire values, as + returned by the wire-format decoder. Useful for inspecting an + unrecognised schema or verifying field-number assignments. + """ diff --git a/src/pymbrewclient/mqtt/proto.py b/src/pymbrewclient/mqtt/proto.py new file mode 100644 index 0000000..db181c9 --- /dev/null +++ b/src/pymbrewclient/mqtt/proto.py @@ -0,0 +1,226 @@ +# "Commons Clause" License Condition v1.0 +# +# The Software is provided to you by the Licensor under the License, as defined below, subject to the following condition. +# +# Without limiting other conditions in the License, the grant of rights under the License will not include, and the License does not grant to you, the right to Sell the Software. +# +# For purposes of the foregoing, "Sell" means practicing any or all of the rights granted to you under the License to provide to third parties, for a fee or other consideration (including without limitation fees for hosting or consulting/ support services related to the Software), a product or service whose value derives, entirely or substantially, from the functionality of the Software. Any license notice or attribution required by the License must also include this Commons Clause License Condition notice. +# +# Software: pymbrewclient +# License: MIT License +# Licensor: Stuart Pearson +# +# +# MIT License +# +# Copyright (c) 2024 Stuart Pearson +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# +# Disclaimer: This software is an independent project and is not affiliated with, endorsed by, or associated with MiniBrew. MiniBrew's trademarks, logos, API, and other intellectual property are owned by MiniBrew and are not included in this software. Users are responsible for complying with MiniBrew's terms of service when using this software. +""" +Pure-Python protobuf wire-format decoder for MiniBrew device-log messages. + +Why pure Python? + MiniBrew's official ``.proto`` source (``minibrew/minibrew-protobuf``) is + private. The field numbers here are RECONSTRUCTED from community + documentation and the MiniBrew REST API; they are not guaranteed to match + the real schema. Using a pure wire-format decoder avoids a hard dependency + on a specific ``google.protobuf`` version and makes the mapping explicit. + +Schema caveat + Field numbers are labelled "best-effort". Until an official schema is + published, decoded values should be treated as approximate. Use + ``DeviceLogMessage.raw_fields`` or ``MqttMessage.payload`` to inspect the + raw wire data. + + To dump a live message: ``protoc --decode_raw < captured_payload.bin`` + +Field mapping (best-effort) + ========= ============================== ========== + Field No. Name Wire type + ========= ============================== ========== + 1 session_id varint (int32) + 2 device_timestamp varint (int64, Unix epoch s) + 3 process_state varint (int32) + 4 user_action varint (int32) + 5 current_temperature 32-bit (float32) + 6 target_temperature 32-bit (float32) + 7 remaining_duration_seconds varint (int32) + 8 seconds_until_next_action varint (int32) + ========= ============================== ========== +""" +import struct +from datetime import datetime, timedelta, timezone + +from .models import DeviceLogMessage, MqttMessage + +# --------------------------------------------------------------------------- +# Wire-type constants +# --------------------------------------------------------------------------- +_WIRE_VARINT = 0 +_WIRE_64BIT = 1 +_WIRE_LEN_DELIM = 2 +_WIRE_32BIT = 5 + + +# --------------------------------------------------------------------------- +# Low-level wire decoder +# --------------------------------------------------------------------------- + + +def _decode_varint(data: bytes, pos: int) -> tuple[int, int]: + """Decode a protobuf base-128 varint starting at *pos*. + + Returns ``(value, new_position)``. Raises ``ValueError`` for truncated or + over-long varints. + """ + result = 0 + shift = 0 + while pos < len(data): + byte = data[pos] + pos += 1 + result |= (byte & 0x7F) << shift + if not (byte & 0x80): + return result, pos + shift += 7 + if shift >= 64: + raise ValueError("Varint exceeds 64 bits — data may be corrupt") + raise ValueError("Truncated varint — unexpected end of data") + + +def decode_raw_fields(data: bytes) -> dict[int, list[object]]: + """Decode raw protobuf wire format into ``{field_number: [values]}``. + + All fields are decoded regardless of whether their numbers are known. + Varint fields are returned as ``int``, 32-bit fields as ``bytes`` (4 + bytes, little-endian), 64-bit fields as ``bytes`` (8 bytes, + little-endian), and length-delimited fields as ``bytes``. + + :param data: Raw protobuf payload bytes. + :returns: A dict mapping each field number to the list of values seen for + that field (repeated fields produce multiple entries). + :raises ValueError: If the data contains a structurally invalid wire + encoding (truncated field, unknown wire type). + """ + result: dict[int, list[object]] = {} + pos = 0 + while pos < len(data): + tag, pos = _decode_varint(data, pos) + field_number = tag >> 3 + wire_type = tag & 0x7 + + if wire_type == _WIRE_VARINT: + value, pos = _decode_varint(data, pos) + result.setdefault(field_number, []).append(value) + + elif wire_type == _WIRE_64BIT: + if pos + 8 > len(data): + raise ValueError(f"Truncated 64-bit field {field_number}") + raw = data[pos : pos + 8] + pos += 8 + result.setdefault(field_number, []).append(raw) + + elif wire_type == _WIRE_LEN_DELIM: + length, pos = _decode_varint(data, pos) + if pos + length > len(data): + raise ValueError(f"Truncated length-delimited field {field_number}: expected {length} bytes") + value = data[pos : pos + length] + pos += length + result.setdefault(field_number, []).append(value) + + elif wire_type == _WIRE_32BIT: + if pos + 4 > len(data): + raise ValueError(f"Truncated 32-bit field {field_number}") + raw = data[pos : pos + 4] + pos += 4 + result.setdefault(field_number, []).append(raw) + + else: + raise ValueError(f"Unknown wire type {wire_type} for field {field_number}") + + return result + + +# --------------------------------------------------------------------------- +# High-level decoder +# --------------------------------------------------------------------------- + + +def _first_varint(raw: dict[int, list[object]], field_num: int) -> int | None: + """Return the first varint value for *field_num*, or ``None``.""" + values = raw.get(field_num) + if values and isinstance(values[0], int): + return values[0] + return None + + +def _first_float32(raw: dict[int, list[object]], field_num: int) -> float | None: + """Return the first 32-bit float value for *field_num*, or ``None``.""" + values = raw.get(field_num) + if values and isinstance(values[0], (bytes, bytearray)) and len(values[0]) == 4: + return struct.unpack(" DeviceLogMessage: + """Attempt to decode a ``devices/logs/`` MQTT message as a DeviceLog. + + On any decoding error the returned :class:`~pymbrewclient.mqtt.models.DeviceLogMessage` + will have :attr:`~pymbrewclient.mqtt.models.DeviceLogMessage.decode_error` set and all + telemetry fields will be ``None``. The raw :attr:`~pymbrewclient.mqtt.models.MqttMessage.payload` + is always preserved. + + :param msg: The raw :class:`~pymbrewclient.mqtt.models.MqttMessage` to decode. + :returns: A populated :class:`~pymbrewclient.mqtt.models.DeviceLogMessage`. + """ + base = DeviceLogMessage( + topic=msg.topic, + payload=msg.payload, + received_at=msg.received_at, + device_uuid=msg.device_uuid, + ) + + try: + raw = decode_raw_fields(msg.payload) + except (ValueError, Exception) as exc: # noqa: BLE001 + base.decode_error = f"Wire decode failed: {exc}" + return base + + base.raw_fields = raw + + # --- Best-effort field mapping (field numbers not officially confirmed) -- + + base.session_id = _first_varint(raw, 1) + + ts_int = _first_varint(raw, 2) + if ts_int is not None: + base.device_timestamp = datetime.fromtimestamp(ts_int, tz=timezone.utc) + + base.process_state = _first_varint(raw, 3) + base.user_action = _first_varint(raw, 4) + base.current_temperature = _first_float32(raw, 5) + base.target_temperature = _first_float32(raw, 6) + base.remaining_duration_seconds = _first_varint(raw, 7) + base.seconds_until_next_action = _first_varint(raw, 8) + + if base.device_timestamp is not None and base.seconds_until_next_action is not None: + base.next_action_at = base.device_timestamp + timedelta(seconds=base.seconds_until_next_action) + + return base diff --git a/src/pymbrewclient/rest/client.py b/src/pymbrewclient/rest/client.py index 1bf0d81..d1043fe 100644 --- a/src/pymbrewclient/rest/client.py +++ b/src/pymbrewclient/rest/client.py @@ -40,7 +40,7 @@ import requests -from .models import BreweryOverview, Device, Session, TokenResponse, coerce_device_payload +from .models import BreweryOverview, Device, Session, TokenResponse, UserProfile, coerce_device_payload logger = logging.getLogger(__name__) @@ -196,3 +196,20 @@ def get_session_info(self, sessionid: int) -> Session: logger.debug(f"Fetching session info for session ID: {sessionid}") response = self.get(f"v1/sessions/{sessionid}") return Session(**response.json()) + + def get_user_profile(self) -> UserProfile: + """ + Fetch the authenticated user's profile from the API. + + :return: A UserProfile object containing the user UUID and profile data. + """ + logger.debug("Fetching user profile...") + response = self.get("v1/profile") + data = response.json() + return UserProfile( + uuid=data["uuid"], + email=data.get("email"), + username=data.get("username"), + first_name=data.get("first_name"), + last_name=data.get("last_name"), + ) diff --git a/src/pymbrewclient/rest/models.py b/src/pymbrewclient/rest/models.py index e742afe..6c8d048 100644 --- a/src/pymbrewclient/rest/models.py +++ b/src/pymbrewclient/rest/models.py @@ -265,6 +265,15 @@ def __repr__(self) -> str: ) +@dataclass +class UserProfile: + uuid: str + email: str | None = None + username: str | None = None + first_name: str | None = None + last_name: str | None = None + + @dataclass class TokenResponse: token: str diff --git a/tests/fixtures/device_log.bin b/tests/fixtures/device_log.bin new file mode 100644 index 0000000000000000000000000000000000000000..3a40fb4d99bfd2c04e1411d608a1a32c1c845562 GIT binary patch literal 29 lcmd bytes: + result = [] + while value > 127: + result.append((value & 0x7F) | 0x80) + value >>= 7 + result.append(value) + return bytes(result) + + +def _encode_float32(value: float) -> bytes: + return struct.pack(" bytes: + """Build a minimal valid protobuf payload matching the best-effort schema.""" + data = b"" + data += _encode_varint((1 << 3) | 0) + _encode_varint(session_id) + data += _encode_varint((2 << 3) | 0) + _encode_varint(device_timestamp) + data += _encode_varint((3 << 3) | 0) + _encode_varint(process_state) + data += _encode_varint((4 << 3) | 0) + _encode_varint(user_action) + data += _encode_varint((5 << 3) | 5) + _encode_float32(current_temperature) + data += _encode_varint((6 << 3) | 5) + _encode_float32(target_temperature) + data += _encode_varint((7 << 3) | 0) + _encode_varint(remaining_duration_seconds) + data += _encode_varint((8 << 3) | 0) + _encode_varint(seconds_until_next_action) + return data + + +def _make_mqtt_message(topic: str, payload: bytes) -> MqttMessage: + return MqttMessage( + topic=topic, + payload=payload, + received_at=datetime(2024, 7, 26, 12, 0, 0, tzinfo=timezone.utc), + device_uuid=_extract_device_uuid(topic), + ) + + +# --------------------------------------------------------------------------- +# MqttClient construction tests +# --------------------------------------------------------------------------- + + +class TestMqttClientConstruction(unittest.TestCase): + """Tests for client_id and username construction, credential wiring.""" + + def _make_client(self) -> MqttClient: + with ( + patch("pymbrewclient.mqtt.client._paho.Client") as mock_paho_cls, + ): + mock_paho = MagicMock() + mock_paho_cls.return_value = mock_paho + return MqttClient(api_token="test-token-xyz", user_uuid="user-uuid-abc") + + def test_client_id_uses_breweryportal_prefix(self) -> None: + client = self._make_client() + self.assertTrue(client._client_id.startswith("breweryportal-")) + + def test_client_id_contains_uuid4(self) -> None: + + client = self._make_client() + uuid_part = client._client_id.removeprefix("breweryportal-") + self.assertRegex(uuid_part, r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$") + + def test_client_uuid_is_unique_per_instance(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client"): + c1 = MqttClient(api_token="token", user_uuid="u1") + c2 = MqttClient(api_token="token", user_uuid="u1") + self.assertNotEqual(c1._client_id, c2._client_id) + + def test_username_uses_breweryportal_prefix(self) -> None: + client = self._make_client() + self.assertEqual(client._username, "breweryportal-user-uuid-abc") + + def test_api_token_used_as_mqtt_password(self) -> None: + """The REST API token must be passed as the MQTT password.""" + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_paho_cls: + mock_paho = MagicMock() + mock_paho_cls.return_value = mock_paho + MqttClient(api_token="secret-api-token", user_uuid="user-uuid-123") + + args, kwargs = mock_paho.username_pw_set.call_args + # paho-mqtt called with positional args: (username, password) + self.assertEqual(args[0], "breweryportal-user-uuid-123") + self.assertEqual(args[1], "secret-api-token") + + def test_token_absent_from_repr(self) -> None: + """The API token must never appear in __repr__.""" + with patch("pymbrewclient.mqtt.client._paho.Client"): + client = MqttClient(api_token="super-secret-token", user_uuid="uid") + self.assertNotIn("super-secret-token", repr(client)) + + def test_repr_contains_expected_fields(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client"): + client = MqttClient(api_token="token", user_uuid="uid") + r = repr(client) + self.assertIn("client_id=", r) + self.assertIn("username=", r) + self.assertIn("broker=", r) + self.assertIn("connected=", r) + + +# --------------------------------------------------------------------------- +# Broker and TLS configuration tests +# --------------------------------------------------------------------------- + + +class TestMqttBrokerConfiguration(unittest.TestCase): + def test_broker_constants(self) -> None: + self.assertEqual(BROKER_HOST, "broker.minibrew.io") + self.assertEqual(BROKER_PORT, 15675) + self.assertEqual(WS_PATH, "/ws") + self.assertEqual(KEEPALIVE, 60) + + def test_paho_client_uses_websocket_transport(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_cls.return_value = MagicMock() + MqttClient(api_token="t", user_uuid="u") + _, kwargs = mock_cls.call_args + self.assertEqual(kwargs["transport"], "websockets") + + def test_paho_client_uses_mqtt_v311(self) -> None: + import paho.mqtt.client as paho + + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_cls.return_value = MagicMock() + MqttClient(api_token="t", user_uuid="u") + _, kwargs = mock_cls.call_args + self.assertEqual(kwargs["protocol"], paho.MQTTv311) + + def test_paho_client_uses_clean_session(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_cls.return_value = MagicMock() + MqttClient(api_token="t", user_uuid="u") + _, kwargs = mock_cls.call_args + self.assertTrue(kwargs["clean_session"]) + + def test_tls_is_enabled(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + MqttClient(api_token="t", user_uuid="u") + mock_paho.tls_set.assert_called_once_with() + + def test_ws_path_is_set(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + MqttClient(api_token="t", user_uuid="u") + mock_paho.ws_set_options.assert_called_once_with(path="/ws") + + def test_keepalive_passed_to_connect(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + client = MqttClient(api_token="t", user_uuid="u") + client.connect() + mock_paho.connect.assert_called_once_with( + host=BROKER_HOST, port=BROKER_PORT, keepalive=KEEPALIVE + ) + + def test_reconnect_delay_is_five_seconds(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + MqttClient(api_token="t", user_uuid="u") + mock_paho.reconnect_delay_set.assert_called_once_with( + min_delay=RECONNECT_DELAY, max_delay=RECONNECT_DELAY + ) + + +# --------------------------------------------------------------------------- +# Last Will tests +# --------------------------------------------------------------------------- + + +class TestLastWill(unittest.TestCase): + def test_last_will_topic_format(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + client = MqttClient(api_token="t", user_uuid="u") + + expected_topic = f"apps/lastwill/{client._client_id}" + mock_paho.will_set.assert_called_once_with( + topic=expected_topic, payload="offline", qos=0, retain=False + ) + + def test_last_will_payload_is_offline(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + MqttClient(api_token="t", user_uuid="u") + _, kwargs = mock_paho.will_set.call_args + self.assertEqual(kwargs["payload"], "offline") + self.assertEqual(kwargs["qos"], 0) + self.assertFalse(kwargs["retain"]) + + +# --------------------------------------------------------------------------- +# Device topic construction +# --------------------------------------------------------------------------- + + +class TestDeviceTopicConstruction(unittest.TestCase): + def test_device_log_topic(self) -> None: + self.assertEqual(MqttClient.device_log_topic("7391Q4827-5NZC8R2M"), "devices/logs/7391Q4827-5NZC8R2M") + + def test_extract_device_uuid_from_device_topic(self) -> None: + self.assertEqual(_extract_device_uuid("devices/logs/7391Q4827-5NZC8R2M"), "7391Q4827-5NZC8R2M") + self.assertEqual(_extract_device_uuid("devices/events/ABC-123"), "ABC-123") + + def test_extract_device_uuid_returns_none_for_other_topics(self) -> None: + self.assertIsNone(_extract_device_uuid("apps/lastwill/some-id")) + self.assertIsNone(_extract_device_uuid("backend/notifications/user-id")) + + +# --------------------------------------------------------------------------- +# Subscribe / unsubscribe / auto-resubscribe tests +# --------------------------------------------------------------------------- + + +class TestSubscriptions(unittest.TestCase): + def _make_client(self) -> tuple[MqttClient, MagicMock]: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + client = MqttClient(api_token="t", user_uuid="u") + return client, mock_paho + + def test_subscribe_adds_to_subscription_set(self) -> None: + client, _ = self._make_client() + client._connected = False + client.subscribe("devices/logs/my-device") + self.assertIn("devices/logs/my-device", client._subscriptions) + + def test_subscribe_calls_paho_when_connected(self) -> None: + client, mock_paho = self._make_client() + client._connected = True + client.subscribe("devices/logs/my-device", qos=1) + mock_paho.subscribe.assert_called_with("devices/logs/my-device", qos=1) + + def test_unsubscribe_removes_from_set(self) -> None: + client, _ = self._make_client() + client._subscriptions.add("devices/logs/my-device") + client._connected = False + client.unsubscribe("devices/logs/my-device") + self.assertNotIn("devices/logs/my-device", client._subscriptions) + + def test_subscribe_device_logs_uses_correct_topic(self) -> None: + client, mock_paho = self._make_client() + client._connected = True + client.subscribe_device_logs("7391Q4827-5NZC8R2M") + mock_paho.subscribe.assert_called_with("devices/logs/7391Q4827-5NZC8R2M", qos=0) + + def test_auto_resubscribe_on_connect(self) -> None: + """All remembered topics are resubscribed on each connect callback.""" + client, mock_paho = self._make_client() + client._subscriptions = {"devices/logs/dev-1", "devices/logs/dev-2"} + + # Simulate paho on_connect with a successful reason code + mock_reason_code = MagicMock() + mock_reason_code.is_failure = False + mock_connect_flags = MagicMock() + mock_connect_flags.session_present = False + + client._on_paho_connect(mock_paho, None, mock_connect_flags, mock_reason_code, None) + + subscribed_topics = {c[0][0] for c in mock_paho.subscribe.call_args_list} + self.assertIn("devices/logs/dev-1", subscribed_topics) + self.assertIn("devices/logs/dev-2", subscribed_topics) + + def test_auto_resubscribe_skipped_on_connect_failure(self) -> None: + client, mock_paho = self._make_client() + client._subscriptions = {"devices/logs/dev-1"} + + mock_reason_code = MagicMock() + mock_reason_code.is_failure = True + mock_connect_flags = MagicMock() + + client._on_paho_connect(mock_paho, None, mock_connect_flags, mock_reason_code, None) + + mock_paho.subscribe.assert_not_called() + + +# --------------------------------------------------------------------------- +# Connection lifecycle +# --------------------------------------------------------------------------- + + +class TestConnectionLifecycle(unittest.TestCase): + def _make_client(self) -> tuple[MqttClient, MagicMock]: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + client = MqttClient(api_token="t", user_uuid="u") + return client, mock_paho + + def test_connect_calls_loop_start(self) -> None: + client, mock_paho = self._make_client() + client.connect() + mock_paho.loop_start.assert_called_once() + + def test_disconnect_calls_loop_stop(self) -> None: + client, mock_paho = self._make_client() + client.disconnect() + mock_paho.loop_stop.assert_called_once() + + def test_disconnect_calls_paho_disconnect(self) -> None: + client, mock_paho = self._make_client() + client.disconnect() + mock_paho.disconnect.assert_called_once() + + def test_disconnect_clears_connected_flag(self) -> None: + client, mock_paho = self._make_client() + client._connected = True + client.disconnect() + self.assertFalse(client._connected) + + def test_context_manager_connects_and_disconnects(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + client = MqttClient(api_token="t", user_uuid="u") + + with client: + mock_paho.connect.assert_called_once() + mock_paho.loop_start.assert_called_once() + + mock_paho.disconnect.assert_called_once() + mock_paho.loop_stop.assert_called_once() + + def test_connected_callback_fires_on_successful_connect(self) -> None: + client, mock_paho = self._make_client() + fired = [] + client.on_connected(lambda: fired.append(True)) + + mock_reason = MagicMock() + mock_reason.is_failure = False + mock_flags = MagicMock() + mock_flags.session_present = False + + client._on_paho_connect(mock_paho, None, mock_flags, mock_reason, None) + self.assertEqual(fired, [True]) + + def test_disconnected_callback_fires(self) -> None: + client, mock_paho = self._make_client() + fired = [] + client.on_disconnected(lambda: fired.append(True)) + + mock_reason = MagicMock() + mock_flags = MagicMock() + client._on_paho_disconnect(mock_paho, None, mock_flags, mock_reason, None) + self.assertEqual(fired, [True]) + + def test_reconnecting_callback_fires_on_second_pre_connect(self) -> None: + client, mock_paho = self._make_client() + fired = [] + client.on_reconnecting(lambda: fired.append(True)) + + # First pre-connect: _ever_connected is False → no reconnecting event + client._on_paho_pre_connect(mock_paho, None) + self.assertEqual(fired, []) + + # Mark as previously connected, then fire again + client._ever_connected = True + client._on_paho_pre_connect(mock_paho, None) + self.assertEqual(fired, [True]) + + def test_error_callback_fires_on_connect_failure(self) -> None: + client, mock_paho = self._make_client() + errors = [] + client.on_error(lambda e: errors.append(e)) + + mock_reason = MagicMock() + mock_reason.is_failure = True + mock_flags = MagicMock() + + client._on_paho_connect(mock_paho, None, mock_flags, mock_reason, None) + self.assertEqual(len(errors), 1) + self.assertIsInstance(errors[0], ConnectionError) + + def test_token_absent_from_error_exception(self) -> None: + """The API token must not appear in error callback exceptions.""" + client, mock_paho = self._make_client() + errors: list[Exception] = [] + client.on_error(lambda e: errors.append(e)) + + mock_reason = MagicMock() + mock_reason.is_failure = True + mock_reason.__str__ = lambda self: "auth failure" + mock_flags = MagicMock() + + client._on_paho_connect(mock_paho, None, mock_flags, mock_reason, None) + + self.assertTrue(errors) + self.assertNotIn("test-token-xyz", str(errors[0])) + + +# --------------------------------------------------------------------------- +# Raw message callback +# --------------------------------------------------------------------------- + + +class TestRawMessageCallback(unittest.TestCase): + def _make_client(self) -> tuple[MqttClient, MagicMock]: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + client = MqttClient(api_token="t", user_uuid="u") + return client, mock_paho + + def test_on_message_delivers_raw_message(self) -> None: + client, mock_paho = self._make_client() + received: list[MqttMessage] = [] + client.on_message(received.append) + + mock_msg = MagicMock() + mock_msg.topic = "devices/events/my-dev" + mock_msg.payload = b"\x01\x02\x03" + + client._on_paho_message(mock_paho, None, mock_msg) + + self.assertEqual(len(received), 1) + self.assertEqual(received[0].topic, "devices/events/my-dev") + self.assertEqual(received[0].payload, b"\x01\x02\x03") + self.assertIsNotNone(received[0].received_at.tzinfo) + + def test_on_message_extracts_device_uuid(self) -> None: + client, mock_paho = self._make_client() + received: list[MqttMessage] = [] + client.on_message(received.append) + + mock_msg = MagicMock() + mock_msg.topic = "devices/logs/7391Q4827-5NZC8R2M" + mock_msg.payload = b"\x00" + + client._on_paho_message(mock_paho, None, mock_msg) + + self.assertEqual(received[0].device_uuid, "7391Q4827-5NZC8R2M") + + +# --------------------------------------------------------------------------- +# Protobuf fixture parsing +# --------------------------------------------------------------------------- + + +class TestDeviceLogFixtureParsing(unittest.TestCase): + """Parse the captured binary device-log fixture from tests/fixtures/.""" + + def setUp(self) -> None: + fixture_path = FIXTURES_DIR / "device_log.bin" + self.payload = fixture_path.read_bytes() + + def test_fixture_file_exists_and_is_nonempty(self) -> None: + self.assertGreater(len(self.payload), 0) + + def test_decode_session_id(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + self.assertEqual(decoded.session_id, 12345) + + def test_decode_device_timestamp(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + expected = datetime.fromtimestamp(1721993600, tz=timezone.utc) + self.assertEqual(decoded.device_timestamp, expected) + + def test_decode_process_state(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + self.assertEqual(decoded.process_state, 80) # FERMENTATION_TEMP_CONTROL + + def test_decode_user_action(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + self.assertEqual(decoded.user_action, 0) + + def test_decode_current_temperature(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + self.assertIsNotNone(decoded.current_temperature) + self.assertAlmostEqual(decoded.current_temperature, 15.1, places=2) + + def test_decode_target_temperature(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + self.assertIsNotNone(decoded.target_temperature) + self.assertAlmostEqual(decoded.target_temperature, 14.91, places=2) + + def test_decode_remaining_duration_seconds(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + self.assertEqual(decoded.remaining_duration_seconds, 4607) + + def test_decode_seconds_until_next_action(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + self.assertEqual(decoded.seconds_until_next_action, 300) + + def test_raw_payload_is_preserved(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + self.assertEqual(decoded.payload, self.payload) + + def test_no_decode_error(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + self.assertIsNone(decoded.decode_error) + + +# --------------------------------------------------------------------------- +# Next-action timestamp calculation +# --------------------------------------------------------------------------- + + +class TestNextActionTimestamp(unittest.TestCase): + def test_next_action_at_is_calculated_correctly(self) -> None: + payload = _build_device_log_payload( + device_timestamp=1721993600, + seconds_until_next_action=300, + ) + msg = _make_mqtt_message("devices/logs/dev", payload) + decoded = decode_device_log(msg) + + expected_ts = datetime.fromtimestamp(1721993600, tz=timezone.utc) + timedelta(seconds=300) + self.assertEqual(decoded.next_action_at, expected_ts) + + def test_next_action_at_is_none_when_timestamp_absent(self) -> None: + """next_action_at requires both device_timestamp and seconds_until_next_action.""" + # Encode only seconds_until_next_action, omit device_timestamp + payload = _encode_varint((8 << 3) | 0) + _encode_varint(300) + msg = _make_mqtt_message("devices/logs/dev", payload) + decoded = decode_device_log(msg) + self.assertIsNone(decoded.next_action_at) + + def test_next_action_at_is_utc(self) -> None: + payload = _build_device_log_payload(device_timestamp=1721993600, seconds_until_next_action=60) + msg = _make_mqtt_message("devices/logs/dev", payload) + decoded = decode_device_log(msg) + self.assertIsNotNone(decoded.next_action_at) + self.assertEqual(decoded.next_action_at.tzinfo, timezone.utc) + + +# --------------------------------------------------------------------------- +# Graceful error handling for malformed messages +# --------------------------------------------------------------------------- + + +class TestMalformedProtoHandling(unittest.TestCase): + def test_empty_payload_decodes_without_error(self) -> None: + msg = _make_mqtt_message("devices/logs/dev", b"") + decoded = decode_device_log(msg) + self.assertIsNone(decoded.decode_error) + self.assertIsNone(decoded.session_id) + + def test_truncated_payload_sets_decode_error(self) -> None: + # Start a valid varint tag but truncate mid-payload + truncated = b"\x08" # tag for field 1 varint, but value bytes missing + msg = _make_mqtt_message("devices/logs/dev", truncated) + decoded = decode_device_log(msg) + self.assertIsNotNone(decoded.decode_error) + + def test_truncated_payload_preserves_raw_bytes(self) -> None: + truncated = b"\x08" + msg = _make_mqtt_message("devices/logs/dev", truncated) + decoded = decode_device_log(msg) + self.assertEqual(decoded.payload, truncated) + + def test_unknown_fields_are_captured_in_raw_fields(self) -> None: + """Future fields with unknown numbers are preserved in raw_fields.""" + payload = _encode_varint((99 << 3) | 0) + _encode_varint(42) + msg = _make_mqtt_message("devices/logs/dev", payload) + decoded = decode_device_log(msg) + self.assertIn(99, decoded.raw_fields) + self.assertIsNone(decoded.decode_error) + + def test_malformed_message_does_not_raise(self) -> None: + """Decoding errors must never propagate as exceptions.""" + bad_payloads = [ + b"\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff", # overlong varint + b"\x0d\x00\x00", # truncated 32-bit field + b"\x09\x00\x00\x00\x00\x00\x00\x00", # truncated 64-bit field + ] + for payload in bad_payloads: + with self.subTest(payload=payload.hex()): + msg = _make_mqtt_message("devices/logs/dev", payload) + decoded = decode_device_log(msg) + self.assertIsNotNone(decoded.decode_error) + + def test_device_log_callback_fires_even_on_decode_error(self) -> None: + """The on_device_log callback must fire for malformed messages.""" + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + client = MqttClient(api_token="t", user_uuid="u") + + received: list[DeviceLogMessage] = [] + client.on_device_log(received.append) + + mock_msg = MagicMock() + mock_msg.topic = "devices/logs/dev" + mock_msg.payload = b"\x08" # truncated + client._on_paho_message(mock_paho, None, mock_msg) + + self.assertEqual(len(received), 1) + self.assertIsNotNone(received[0].decode_error) + + +# --------------------------------------------------------------------------- +# decode_raw_fields unit tests +# --------------------------------------------------------------------------- + + +class TestDecodeRawFields(unittest.TestCase): + def test_decode_single_varint(self) -> None: + # field 1, wire type 0 (varint), value 42 + data = _encode_varint((1 << 3) | 0) + _encode_varint(42) + raw = decode_raw_fields(data) + self.assertEqual(raw[1], [42]) + + def test_decode_single_float32(self) -> None: + data = _encode_varint((5 << 3) | 5) + _encode_float32(3.14) + raw = decode_raw_fields(data) + result = struct.unpack(" None: + data = ( + _encode_varint((2 << 3) | 0) + _encode_varint(10) + + _encode_varint((2 << 3) | 0) + _encode_varint(20) + ) + raw = decode_raw_fields(data) + self.assertEqual(raw[2], [10, 20]) + + def test_decode_length_delimited_field(self) -> None: + value = b"hello" + data = _encode_varint((3 << 3) | 2) + _encode_varint(len(value)) + value + raw = decode_raw_fields(data) + self.assertEqual(raw[3], [b"hello"]) + + def test_empty_bytes_returns_empty_dict(self) -> None: + self.assertEqual(decode_raw_fields(b""), {}) + + +# --------------------------------------------------------------------------- +# Token redaction in logs +# --------------------------------------------------------------------------- + + +class TestTokenRedaction(unittest.TestCase): + def test_token_not_logged_at_debug(self) -> None: + """Confirm the MQTT client module does not log the raw token.""" + import io + import logging + + stream = io.StringIO() + handler = logging.StreamHandler(stream) + logger = logging.getLogger("pymbrewclient.mqtt.client") + original_handlers = logger.handlers[:] + original_level = logger.level + + try: + logger.addHandler(handler) + logger.setLevel(logging.DEBUG) + handler.setLevel(logging.DEBUG) + + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + client = MqttClient(api_token="very-secret-api-token", user_uuid="u") + client.connect() + client.subscribe("test/topic") + client.disconnect() + + log_output = stream.getvalue() + self.assertNotIn("very-secret-api-token", log_output) + finally: + logger.handlers = original_handlers + logger.setLevel(original_level) + + +# --------------------------------------------------------------------------- +# BreweryClient.create_mqtt_client integration +# --------------------------------------------------------------------------- + + +class TestBreweryClientCreateMqttClient(unittest.TestCase): + def test_create_mqtt_client_returns_mqtt_client(self) -> None: + from pymbrewclient.client import BreweryClient + from pymbrewclient.rest.models import UserProfile + + with patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), patch( + "pymbrewclient.rest.client.RestApiClient.get_user_profile" + ) as mock_profile, patch( + "pymbrewclient.mqtt.client._paho.Client" + ) as mock_paho_cls: + mock_paho = MagicMock() + mock_paho_cls.return_value = mock_paho + mock_profile.return_value = UserProfile(uuid="profile-uuid-xyz") + + bc = BreweryClient("u", "p", base_url="https://api.example.com") + bc.client.token = "rest-token-abc" + + mqtt = bc.create_mqtt_client() + + self.assertIsInstance(mqtt, MqttClient) + + def test_create_mqtt_client_reuses_rest_token(self) -> None: + from pymbrewclient.client import BreweryClient + from pymbrewclient.rest.models import UserProfile + + with patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), patch( + "pymbrewclient.rest.client.RestApiClient.get_user_profile" + ) as mock_profile, patch( + "pymbrewclient.mqtt.client._paho.Client" + ) as mock_paho_cls: + mock_paho = MagicMock() + mock_paho_cls.return_value = mock_paho + mock_profile.return_value = UserProfile(uuid="profile-uuid-xyz") + + bc = BreweryClient("u", "p", base_url="https://api.example.com") + bc.client.token = "the-rest-token" + + bc.create_mqtt_client() + + # REST token used as MQTT password + args, kwargs = mock_paho.username_pw_set.call_args + self.assertEqual(args[0], "breweryportal-profile-uuid-xyz") + self.assertEqual(args[1], "the-rest-token") + + def test_create_mqtt_client_uses_user_uuid_for_username(self) -> None: + from pymbrewclient.client import BreweryClient + from pymbrewclient.rest.models import UserProfile + + with patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), patch( + "pymbrewclient.rest.client.RestApiClient.get_user_profile" + ) as mock_profile, patch( + "pymbrewclient.mqtt.client._paho.Client" + ) as mock_paho_cls: + mock_paho = MagicMock() + mock_paho_cls.return_value = mock_paho + mock_profile.return_value = UserProfile(uuid="my-user-uuid-123") + + bc = BreweryClient("u", "p", base_url="https://api.example.com") + bc.client.token = "tok" + + mqtt = bc.create_mqtt_client() + + self.assertEqual(mqtt._username, "breweryportal-my-user-uuid-123") + + def test_multiple_mqtt_clients_have_unique_client_ids(self) -> None: + from pymbrewclient.client import BreweryClient + from pymbrewclient.rest.models import UserProfile + + with patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), patch( + "pymbrewclient.rest.client.RestApiClient.get_user_profile" + ) as mock_profile, patch( + "pymbrewclient.mqtt.client._paho.Client" + ) as mock_paho_cls: + mock_paho_cls.return_value = MagicMock() + mock_profile.return_value = UserProfile(uuid="uid") + + bc = BreweryClient("u", "p", base_url="https://api.example.com") + bc.client.token = "tok" + + m1 = bc.create_mqtt_client() + m2 = bc.create_mqtt_client() + + self.assertNotEqual(m1._client_id, m2._client_id) + + +if __name__ == "__main__": + unittest.main() From 366ad120f409a29a848e001c584e4993ea5a5d33 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 25 Jul 2026 17:55:45 +0000 Subject: [PATCH 2/5] fix: simplify exception catch in proto decoder per code review --- src/pymbrewclient/mqtt/proto.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pymbrewclient/mqtt/proto.py b/src/pymbrewclient/mqtt/proto.py index db181c9..193ac4b 100644 --- a/src/pymbrewclient/mqtt/proto.py +++ b/src/pymbrewclient/mqtt/proto.py @@ -199,7 +199,7 @@ def decode_device_log(msg: MqttMessage) -> DeviceLogMessage: try: raw = decode_raw_fields(msg.payload) - except (ValueError, Exception) as exc: # noqa: BLE001 + except Exception as exc: # noqa: BLE001 base.decode_error = f"Wire decode failed: {exc}" return base From 76f7a0472dd43b6d1ebc4f542bc39ed67fb40cc6 Mon Sep 17 00:00:00 2001 From: Stuart Pearson <1926002+stuartp44@users.noreply.github.com> Date: Sat, 25 Jul 2026 20:16:52 +0200 Subject: [PATCH 3/5] style: apply black formatting to mqtt module and tests Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/pymbrewclient/mqtt/client.py | 1 + src/pymbrewclient/mqtt/models.py | 1 + src/pymbrewclient/mqtt/proto.py | 1 + tests/test_mqtt.py | 61 +++++++++++++------------------- 4 files changed, 28 insertions(+), 36 deletions(-) diff --git a/src/pymbrewclient/mqtt/client.py b/src/pymbrewclient/mqtt/client.py index dd6f582..9f1c641 100644 --- a/src/pymbrewclient/mqtt/client.py +++ b/src/pymbrewclient/mqtt/client.py @@ -44,6 +44,7 @@ This module takes care to keep the token out of ``__repr__``, log output, and callback arguments. """ + import logging import threading import uuid diff --git a/src/pymbrewclient/mqtt/models.py b/src/pymbrewclient/mqtt/models.py index 0116275..5e30417 100644 --- a/src/pymbrewclient/mqtt/models.py +++ b/src/pymbrewclient/mqtt/models.py @@ -37,6 +37,7 @@ """ Typed message models for the MiniBrew MQTT client. """ + from dataclasses import dataclass, field from datetime import datetime diff --git a/src/pymbrewclient/mqtt/proto.py b/src/pymbrewclient/mqtt/proto.py index 193ac4b..8bf6ea4 100644 --- a/src/pymbrewclient/mqtt/proto.py +++ b/src/pymbrewclient/mqtt/proto.py @@ -66,6 +66,7 @@ 8 seconds_until_next_action varint (int32) ========= ============================== ========== """ + import struct from datetime import datetime, timedelta, timezone diff --git a/tests/test_mqtt.py b/tests/test_mqtt.py index 6d258eb..4bd005e 100644 --- a/tests/test_mqtt.py +++ b/tests/test_mqtt.py @@ -77,9 +77,7 @@ class TestMqttClientConstruction(unittest.TestCase): """Tests for client_id and username construction, credential wiring.""" def _make_client(self) -> MqttClient: - with ( - patch("pymbrewclient.mqtt.client._paho.Client") as mock_paho_cls, - ): + with (patch("pymbrewclient.mqtt.client._paho.Client") as mock_paho_cls,): mock_paho = MagicMock() mock_paho_cls.return_value = mock_paho return MqttClient(api_token="test-token-xyz", user_uuid="user-uuid-abc") @@ -187,18 +185,14 @@ def test_keepalive_passed_to_connect(self) -> None: mock_cls.return_value = mock_paho client = MqttClient(api_token="t", user_uuid="u") client.connect() - mock_paho.connect.assert_called_once_with( - host=BROKER_HOST, port=BROKER_PORT, keepalive=KEEPALIVE - ) + mock_paho.connect.assert_called_once_with(host=BROKER_HOST, port=BROKER_PORT, keepalive=KEEPALIVE) def test_reconnect_delay_is_five_seconds(self) -> None: with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: mock_paho = MagicMock() mock_cls.return_value = mock_paho MqttClient(api_token="t", user_uuid="u") - mock_paho.reconnect_delay_set.assert_called_once_with( - min_delay=RECONNECT_DELAY, max_delay=RECONNECT_DELAY - ) + mock_paho.reconnect_delay_set.assert_called_once_with(min_delay=RECONNECT_DELAY, max_delay=RECONNECT_DELAY) # --------------------------------------------------------------------------- @@ -214,9 +208,7 @@ def test_last_will_topic_format(self) -> None: client = MqttClient(api_token="t", user_uuid="u") expected_topic = f"apps/lastwill/{client._client_id}" - mock_paho.will_set.assert_called_once_with( - topic=expected_topic, payload="offline", qos=0, retain=False - ) + mock_paho.will_set.assert_called_once_with(topic=expected_topic, payload="offline", qos=0, retain=False) def test_last_will_payload_is_offline(self) -> None: with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: @@ -658,10 +650,7 @@ def test_decode_single_float32(self) -> None: self.assertAlmostEqual(result, 3.14, places=4) def test_decode_repeated_field(self) -> None: - data = ( - _encode_varint((2 << 3) | 0) + _encode_varint(10) - + _encode_varint((2 << 3) | 0) + _encode_varint(20) - ) + data = _encode_varint((2 << 3) | 0) + _encode_varint(10) + _encode_varint((2 << 3) | 0) + _encode_varint(20) raw = decode_raw_fields(data) self.assertEqual(raw[2], [10, 20]) @@ -722,11 +711,11 @@ def test_create_mqtt_client_returns_mqtt_client(self) -> None: from pymbrewclient.client import BreweryClient from pymbrewclient.rest.models import UserProfile - with patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), patch( - "pymbrewclient.rest.client.RestApiClient.get_user_profile" - ) as mock_profile, patch( - "pymbrewclient.mqtt.client._paho.Client" - ) as mock_paho_cls: + with ( + patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), + patch("pymbrewclient.rest.client.RestApiClient.get_user_profile") as mock_profile, + patch("pymbrewclient.mqtt.client._paho.Client") as mock_paho_cls, + ): mock_paho = MagicMock() mock_paho_cls.return_value = mock_paho mock_profile.return_value = UserProfile(uuid="profile-uuid-xyz") @@ -742,11 +731,11 @@ def test_create_mqtt_client_reuses_rest_token(self) -> None: from pymbrewclient.client import BreweryClient from pymbrewclient.rest.models import UserProfile - with patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), patch( - "pymbrewclient.rest.client.RestApiClient.get_user_profile" - ) as mock_profile, patch( - "pymbrewclient.mqtt.client._paho.Client" - ) as mock_paho_cls: + with ( + patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), + patch("pymbrewclient.rest.client.RestApiClient.get_user_profile") as mock_profile, + patch("pymbrewclient.mqtt.client._paho.Client") as mock_paho_cls, + ): mock_paho = MagicMock() mock_paho_cls.return_value = mock_paho mock_profile.return_value = UserProfile(uuid="profile-uuid-xyz") @@ -765,11 +754,11 @@ def test_create_mqtt_client_uses_user_uuid_for_username(self) -> None: from pymbrewclient.client import BreweryClient from pymbrewclient.rest.models import UserProfile - with patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), patch( - "pymbrewclient.rest.client.RestApiClient.get_user_profile" - ) as mock_profile, patch( - "pymbrewclient.mqtt.client._paho.Client" - ) as mock_paho_cls: + with ( + patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), + patch("pymbrewclient.rest.client.RestApiClient.get_user_profile") as mock_profile, + patch("pymbrewclient.mqtt.client._paho.Client") as mock_paho_cls, + ): mock_paho = MagicMock() mock_paho_cls.return_value = mock_paho mock_profile.return_value = UserProfile(uuid="my-user-uuid-123") @@ -785,11 +774,11 @@ def test_multiple_mqtt_clients_have_unique_client_ids(self) -> None: from pymbrewclient.client import BreweryClient from pymbrewclient.rest.models import UserProfile - with patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), patch( - "pymbrewclient.rest.client.RestApiClient.get_user_profile" - ) as mock_profile, patch( - "pymbrewclient.mqtt.client._paho.Client" - ) as mock_paho_cls: + with ( + patch("pymbrewclient.rest.client.RestApiClient._ensure_token"), + patch("pymbrewclient.rest.client.RestApiClient.get_user_profile") as mock_profile, + patch("pymbrewclient.mqtt.client._paho.Client") as mock_paho_cls, + ): mock_paho_cls.return_value = MagicMock() mock_profile.return_value = UserProfile(uuid="uid") From 5e0a92590e3b9cb7de1fcd59a389e4d14eb9a71f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 25 Jul 2026 19:26:45 +0000 Subject: [PATCH 4/5] feat: add watch-device-logs CLI command for real-time MQTT telemetry --- src/pymbrewclient/cli.py | 63 ++++++++++++++++++++ tests/test_client.py | 124 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 187 insertions(+) diff --git a/src/pymbrewclient/cli.py b/src/pymbrewclient/cli.py index 6b80e1d..06b5f17 100644 --- a/src/pymbrewclient/cli.py +++ b/src/pymbrewclient/cli.py @@ -38,6 +38,7 @@ from datetime import datetime import json import logging +import threading from typing import Annotated import typer @@ -74,6 +75,8 @@ def serialize_output(data: object) -> object: return serialize_output(data.dict()) if isinstance(data, datetime): return datetime_to_api_string(data) + if isinstance(data, bytes): + return data.hex() if isinstance(data, Device): return data.to_dict() if is_dataclass(data): @@ -328,5 +331,65 @@ def process_estimate( raise typer.Exit(code=1) +@app.command() +def watch_device_logs( + username: str = typer.Option(..., "--username", help="The username for authentication."), + password: str = typer.Option(..., "--password", help="The password for authentication."), + base_url: str = typer.Option(base_url, "--base-url", help="The base URL for the API."), + serials: Annotated[ + list[str], + typer.Option("--serial", help="Device serial number to watch (can be specified multiple times)."), + ] = [], + output_format: str = typer.Option("pretty", "--format", help="Output format: pretty or json."), + duration: int | None = typer.Option( + None, "--duration", help="Stop automatically after this many seconds (default: run until Ctrl+C)." + ), +) -> None: + """Stream real-time MQTT device telemetry logs for one or more devices.""" + if not serials: + typer.echo("Error: at least one --serial is required.") + raise typer.Exit(code=1) + + try: + client = initialize_brewery_client(base_url, username, password) + stop_event = threading.Event() + + with client.create_mqtt_client() as mqtt: + + def on_connected() -> None: + typer.echo(f"Connected. Watching: {', '.join(serials)}") + + def on_disconnected() -> None: + typer.echo("Disconnected.") + + def on_error(exc: Exception) -> None: + typer.echo(f"MQTT error: {exc}") + + def on_device_log(msg: object) -> None: + print_output(msg, output_format.lower()) + + mqtt.on_connected(on_connected) + mqtt.on_disconnected(on_disconnected) + mqtt.on_error(on_error) + mqtt.on_device_log(on_device_log) + + for serial in serials: + mqtt.subscribe_device_logs(serial) + + try: + stop_event.wait(timeout=duration) + except KeyboardInterrupt: + typer.echo("\nStopping...") + + except BreweryClientError as e: + logger.error(f"Error: {e}") + typer.echo(f"Error: {e}") + raise typer.Exit(code=1) + except Exception as e: + logger.error(f"Error: {e}") + typer.echo(f"Error: {e}") + raise typer.Exit(code=1) + + if __name__ == "__main__": app() diff --git a/tests/test_client.py b/tests/test_client.py index 564fb50..9ef3d29 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -453,6 +453,130 @@ def test_process_estimate_cli_outputs_json(self) -> None: self.assertEqual(payload["process_estimate_remaining_seconds"], 4607) self.assertEqual(payload["process_estimate_remaining_formatted"], "1:16:47") + def test_watch_device_logs_requires_serial(self) -> None: + runner = CliRunner() + result = runner.invoke( + app, + ["watch-device-logs", "--username", "user", "--password", "pass"], + ) + self.assertNotEqual(result.exit_code, 0) + self.assertIn("--serial", result.stdout) + + def test_watch_device_logs_streams_messages_as_json(self) -> None: + """Connect, receive one device-log message, then stop via --duration 0.""" + from datetime import datetime, timezone + + from pymbrewclient.mqtt.models import DeviceLogMessage + + runner = CliRunner() + + sample_msg = DeviceLogMessage( + topic="devices/logs/SER-001", + payload=b"\x08\x01", + received_at=datetime(2024, 7, 26, 12, 0, 0, tzinfo=timezone.utc), + device_uuid="SER-001", + session_id=1, + process_state=80, + current_temperature=15.0, + ) + + mock_mqtt = MagicMock() + mock_mqtt.__enter__ = MagicMock(return_value=mock_mqtt) + mock_mqtt.__exit__ = MagicMock(return_value=False) + + # Capture the on_device_log callback so we can invoke it synchronously + device_log_callbacks: list = [] + + def capture_on_device_log(cb: object) -> None: + device_log_callbacks.append(cb) + + mock_mqtt.on_device_log.side_effect = capture_on_device_log + + mock_client = MagicMock() + mock_client.create_mqtt_client.return_value = mock_mqtt + + def fake_wait(event: object, timeout: object = None) -> bool: # type: ignore[override] + for cb in device_log_callbacks: + cb(sample_msg) + return True + + with ( + patch("pymbrewclient.cli.initialize_brewery_client", return_value=mock_client), + patch("threading.Event.wait", fake_wait), + ): + result = runner.invoke( + app, + [ + "watch-device-logs", + "--username", + "user", + "--password", + "pass", + "--serial", + "SER-001", + "--format", + "json", + "--duration", + "0", + ], + ) + + self.assertEqual(result.exit_code, 0, result.stdout) + # Find the JSON object in stdout (there may be preceding non-JSON lines) + decoder = json.JSONDecoder() + json_start = result.stdout.find("{") + self.assertGreater(json_start, -1, f"No JSON object found in: {result.stdout!r}") + payload, _ = decoder.raw_decode(result.stdout, json_start) + self.assertEqual(payload["device_uuid"], "SER-001") + self.assertEqual(payload["session_id"], 1) + self.assertEqual(payload["process_state"], 80) + self.assertEqual(payload["payload"], b"\x08\x01".hex()) + + def test_watch_device_logs_subscribes_all_serials(self) -> None: + """Each --serial value is subscribed on the MQTT client.""" + runner = CliRunner() + + mock_mqtt = MagicMock() + mock_mqtt.__enter__ = MagicMock(return_value=mock_mqtt) + mock_mqtt.__exit__ = MagicMock(return_value=False) + + mock_client = MagicMock() + mock_client.create_mqtt_client.return_value = mock_mqtt + + with ( + patch("pymbrewclient.cli.initialize_brewery_client", return_value=mock_client), + patch("threading.Event.wait", return_value=True), + ): + result = runner.invoke( + app, + [ + "watch-device-logs", + "--username", + "user", + "--password", + "pass", + "--serial", + "SER-001", + "--serial", + "SER-002", + "--duration", + "0", + ], + ) + + self.assertEqual(result.exit_code, 0, result.stdout) + subscribed = {call.args[0] for call in mock_mqtt.subscribe_device_logs.call_args_list} + self.assertIn("SER-001", subscribed) + self.assertIn("SER-002", subscribed) + + def test_serialize_output_converts_bytes_to_hex(self) -> None: + from pymbrewclient.cli import serialize_output + + self.assertEqual(serialize_output(b"\xde\xad\xbe\xef"), "deadbeef") + self.assertEqual(serialize_output(b""), "") + result = serialize_output({"payload": b"\x01\x02"}) + self.assertEqual(result, {"payload": "0102"}) + if __name__ == "__main__": unittest.main() From c118bcea6fe575e89c6f8a421fab5c66e59db633 Mon Sep 17 00:00:00 2001 From: Stuart Pearson <1926002+stuartp44@users.noreply.github.com> Date: Sat, 25 Jul 2026 22:53:33 +0200 Subject: [PATCH 5/5] Enhance device log handling and telemetry decoding - Updated the MQTT client to include Wi-Fi RSSI and next action timestamps in device log messages. - Improved the decoding of telemetry fields, including measurements and nested state values. - Added a curated output function for device logs to simplify the displayed information. - Adjusted the CLI to support a debug mode that reveals all fields in the device log output. - Modified the REST API client to fetch user profiles from the correct endpoint. - Removed outdated binary fixture and replaced it with hex representation for better readability. - Enhanced unit tests to cover new telemetry fields and ensure accurate decoding of device logs. --- docs/API_EXAMPLES.md | 73 +++++----- pyproject.toml | 2 +- src/pymbrewclient/cli.py | 34 ++++- src/pymbrewclient/mqtt/client.py | 3 +- src/pymbrewclient/mqtt/models.py | 60 +++++--- src/pymbrewclient/mqtt/proto.py | 128 +++++++++++++---- src/pymbrewclient/rest/client.py | 4 +- tests/fixtures/device_log.bin | Bin 29 -> 0 bytes tests/fixtures/device_log.hex | 1 + tests/fixtures/device_log_envelope.hex | 1 + tests/test_client.py | 140 +++++++++++++++++- tests/test_mqtt.py | 192 ++++++++++++++++++------- 12 files changed, 490 insertions(+), 148 deletions(-) delete mode 100644 tests/fixtures/device_log.bin create mode 100644 tests/fixtures/device_log.hex create mode 100644 tests/fixtures/device_log_envelope.hex diff --git a/docs/API_EXAMPLES.md b/docs/API_EXAMPLES.md index ce67caf..4628da1 100644 --- a/docs/API_EXAMPLES.md +++ b/docs/API_EXAMPLES.md @@ -223,52 +223,61 @@ with client.create_mqtt_client() as mqtt: print(f"Process state: {msg.process_state}") print(f"Current temp: {msg.current_temperature}°C") print(f"Target temp: {msg.target_temperature}°C") + print(f"Wi-Fi RSSI: {msg.wifi_rssi_dbm} dBm") + if msg.next_action_at: + print(f"Next action: {msg.next_action_at.isoformat()}") + for measurement_id, value in msg.measurements.items(): + print(f"Measurement {measurement_id}: {value}") mqtt.on_device_log(handle_log) mqtt.subscribe_device_logs("7391Q4827-5NZC8R2M") import time; time.sleep(30) ``` -### `next_action_at` example - -```python -with client.create_mqtt_client() as mqtt: - def handle_log(msg: DeviceLogMessage) -> None: - if msg.next_action_at: - print(f"Next action at (UTC): {msg.next_action_at.isoformat()}") - # Home Assistant can display this UTC timestamp in the user's local timezone. +### Limitations: protobuf schema - mqtt.on_device_log(handle_log) - mqtt.subscribe_device_logs("7391Q4827-5NZC8R2M") - import time; time.sleep(30) -``` +MiniBrew's official protobuf schema (`minibrew/minibrew-protobuf`) is a private +repository that is not publicly accessible. The fields exposed directly on +`DeviceLogMessage` were reconstructed from captured device traffic and compared +with the MiniBrew REST API response structure. -`next_action_at` is calculated as: +Unknown nested state values are exposed through `DeviceLogMessage.state_fields`. +Confirmed measurement ID 24 is also exposed as `DeviceLogMessage.wifi_rssi_dbm`; +all readings, including measurements whose semantics remain unconfirmed, stay +available by numeric ID in `DeviceLogMessage.measurements`. The raw `payload` +bytes are always preserved in `MqttMessage.payload` and +`DeviceLogMessage.payload`. To inspect a live message: -```python -next_action_at = message_timestamp + timedelta(seconds=seconds_until_next_action) +```bash +protoc --decode_raw < captured_payload.bin ``` -It is always a timezone-aware UTC `datetime`. No local timezone conversion is -applied inside the library; downstream consumers (such as Home Assistant) should -handle display timezone conversion. +Live MQTT messages contain an envelope around the telemetry message. +`DeviceLogMessage.raw_fields` exposes that envelope, while +`DeviceLogMessage.telemetry_fields` exposes the nested telemetry fields. +Telemetry field 26 is the countdown in seconds to the next required action; +`DeviceLogMessage.next_action_at` adds it to the device timestamp and returns a +timezone-aware UTC datetime. Unconfirmed fields, including telemetry fields 8 +and 30, remain available without speculative semantic names. -### Limitations: protobuf schema +The CLI prints curated decoded telemetry by default, excluding the binary +payload, numeric measurement map, and protobuf field maps. Confirmed named +measurements such as `wifi_rssi_dbm` remain visible: -MiniBrew's official protobuf schema (`minibrew/minibrew-protobuf`) is a private -repository that is not publicly accessible. The field numbers used in -`DeviceLogMessage` were **reconstructed** from: - -- The MiniBrew REST API response structure (`Device` dataclass field names) -- The community-maintained `minibrew/enduser-docker-server` project documentation +```bash +pymbrewclient watch-device-logs \ + --username you@example.com \ + --password 'your-password' \ + --serial 7391Q4827-5NZC8R2M +``` -**Until official field numbers are confirmed, decoded telemetry values may be -incorrect.** The raw `payload` bytes are always preserved in `MqttMessage.payload` -and `DeviceLogMessage.payload`. To inspect a live message yourself: +Add `--debug` to include the complete numeric `measurements` map, raw payload, +`raw_fields`, `telemetry_fields`, and `state_fields`: ```bash -protoc --decode_raw < captured_payload.bin +pymbrewclient watch-device-logs \ + --username you@example.com \ + --password 'your-password' \ + --serial 7391Q4827-5NZC8R2M \ + --debug ``` - -The raw field numbers are also exposed via `DeviceLogMessage.raw_fields` for -inspection without any mapping. diff --git a/pyproject.toml b/pyproject.toml index 5f7541c..198da0b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,7 +23,7 @@ classifiers = [ dependencies = [ "pydantic>=1.10.17,<3", "requests>=2.32.3,<3", - "typer>=0.12.5,<1", + "typer>=0.16.0,<1", "rich>=13.9.4,<16", "paho-mqtt>=2.0.0,<3", ] diff --git a/src/pymbrewclient/cli.py b/src/pymbrewclient/cli.py index 06b5f17..0c59865 100644 --- a/src/pymbrewclient/cli.py +++ b/src/pymbrewclient/cli.py @@ -47,6 +47,7 @@ from rich.pretty import Pretty from pymbrewclient.client import BreweryClient, BreweryClientError +from pymbrewclient.mqtt.models import DeviceLogMessage from pymbrewclient.rest.client import RestApiClient from pymbrewclient.rest.models import Device, datetime_to_api_string, format_duration @@ -102,6 +103,29 @@ def print_output(data: BaseModel | dict | list, format: str) -> None: rich_print(Pretty(serialized_data)) +def curate_device_log_output(msg: DeviceLogMessage) -> dict[str, object]: + """Return useful decoded telemetry without verbose protobuf internals.""" + curated = { + "topic": msg.topic, + "received_at": msg.received_at, + "device_uuid": msg.device_uuid, + "sequence_number": msg.sequence_number, + "session_id": msg.session_id, + "device_timestamp": msg.device_timestamp, + "current_state": msg.current_state, + "process_type": msg.process_type, + "process_state": msg.process_state, + "user_action": msg.user_action, + "current_temperature": msg.current_temperature, + "target_temperature": msg.target_temperature, + "wifi_rssi_dbm": msg.wifi_rssi_dbm, + "seconds_until_next_action": msg.seconds_until_next_action, + "next_action_at": msg.next_action_at, + "decode_error": msg.decode_error, + } + return {key: value for key, value in curated.items() if value is not None} + + def setup_logging(level: str) -> None: """Configure standard logging with the specified log level.""" logging.basicConfig(level=getattr(logging, level.upper(), logging.INFO), format="%(message)s", force=True) @@ -344,6 +368,11 @@ def watch_device_logs( duration: int | None = typer.Option( None, "--duration", help="Stop automatically after this many seconds (default: run until Ctrl+C)." ), + debug: bool = typer.Option( + False, + "--debug", + help="Include the raw payload and all decoded protobuf fields.", + ), ) -> None: """Stream real-time MQTT device telemetry logs for one or more devices.""" if not serials: @@ -365,8 +394,9 @@ def on_disconnected() -> None: def on_error(exc: Exception) -> None: typer.echo(f"MQTT error: {exc}") - def on_device_log(msg: object) -> None: - print_output(msg, output_format.lower()) + def on_device_log(msg: DeviceLogMessage) -> None: + output = msg if debug else curate_device_log_output(msg) + print_output(output, output_format.lower()) mqtt.on_connected(on_connected) mqtt.on_disconnected(on_disconnected) diff --git a/src/pymbrewclient/mqtt/client.py b/src/pymbrewclient/mqtt/client.py index 9f1c641..02fc909 100644 --- a/src/pymbrewclient/mqtt/client.py +++ b/src/pymbrewclient/mqtt/client.py @@ -54,6 +54,7 @@ from typing import Any import paho.mqtt.client as _paho +from requests.certs import where as requests_ca_bundle from .models import DeviceLogMessage, MqttMessage from .proto import decode_device_log @@ -167,7 +168,7 @@ def _build_paho_client(self) -> _paho.Client: clean_session=True, ) client.ws_set_options(path=WS_PATH) - client.tls_set() # TLS with system CA bundle; certificate verification enabled + client.tls_set(ca_certs=requests_ca_bundle()) client.username_pw_set(self._username, self._api_token) client.will_set(topic=will_topic, payload="offline", qos=0, retain=False) client.reconnect_delay_set(min_delay=RECONNECT_DELAY, max_delay=RECONNECT_DELAY) diff --git a/src/pymbrewclient/mqtt/models.py b/src/pymbrewclient/mqtt/models.py index 5e30417..56984ce 100644 --- a/src/pymbrewclient/mqtt/models.py +++ b/src/pymbrewclient/mqtt/models.py @@ -94,48 +94,53 @@ class DeviceLogMessage: # --- Decoded telemetry fields ---------------------------------------- + sequence_number: int | None = None + """Per-device message sequence number from live envelope field 1.""" + session_id: int | None = None - """Active brewing session ID (protobuf field 1, best-effort).""" + """Active brewing session ID (live envelope field 4 or telemetry field 11).""" device_timestamp: datetime | None = None """ Timezone-aware UTC datetime from the device's own clock - (protobuf field 2, best-effort). Derived from a Unix epoch integer. + (live envelope field 5 or telemetry field 1). Derived from Unix epoch milliseconds. """ + current_state: int | None = None + """Current device-state integer (nested state field 1, observed).""" + + process_type: int | None = None + """Process-type integer (nested state field 2, observed).""" + process_state: int | None = None """ - Process state integer (protobuf field 3, best-effort). + Process-state integer (nested state field 3, observed). See ``PROCESS_STATE_LABELS`` in the MiniBrew community docs for a complete mapping (e.g. 80 → FERMENTATION_TEMP_CONTROL). """ user_action: int | None = None """ - User-action state integer (protobuf field 4, best-effort). + User-action state integer (nested state field 8, observed). See ``USER_ACTION_LABELS`` in the MiniBrew community docs. """ current_temperature: float | None = None - """Current temperature in °C (protobuf field 5, best-effort).""" + """Current temperature in °C (telemetry field 19, observed).""" target_temperature: float | None = None - """Target temperature in °C (protobuf field 6, best-effort).""" + """Target temperature in °C (telemetry field 18, observed).""" - remaining_duration_seconds: int | None = None - """Remaining process duration in seconds (protobuf field 7, best-effort).""" + wifi_rssi_dbm: float | None = None + """Wi-Fi received signal strength in dBm (measurement ID 24, confirmed).""" seconds_until_next_action: int | None = None - """Seconds until the next scheduled action (protobuf field 8, best-effort).""" + """Seconds until the next required user action (telemetry field 26, observed).""" next_action_at: datetime | None = None """ - Calculated UTC datetime of the next scheduled action:: - - next_action_at = device_timestamp + timedelta(seconds=seconds_until_next_action) - - ``None`` when either :attr:`device_timestamp` or - :attr:`seconds_until_next_action` is absent. + Timezone-aware UTC datetime calculated from :attr:`device_timestamp` plus + :attr:`seconds_until_next_action`. """ decode_error: str | None = None @@ -147,7 +152,26 @@ class DeviceLogMessage: raw_fields: dict[int, list[object]] = field(default_factory=dict) """ - All raw protobuf field numbers and their decoded wire values, as - returned by the wire-format decoder. Useful for inspecting an - unrecognised schema or verifying field-number assignments. + Raw fields from the MQTT payload's outermost protobuf message. + For live wrapped messages these are the envelope fields. + """ + + telemetry_fields: dict[int, list[object]] = field(default_factory=dict) + """ + Raw fields from the decoded telemetry message. For an unwrapped telemetry + payload this is equal to :attr:`raw_fields`; for a live envelope it is + decoded from envelope field 3. + """ + + state_fields: dict[int, list[object]] = field(default_factory=dict) + """ + Raw fields decoded from the nested state message in telemetry field 2. + This preserves observed but not yet semantically identified state values. + """ + + measurements: dict[int, float] = field(default_factory=dict) + """ + Measurement ID to float value from repeated telemetry field 3 entries. + Confirmed IDs are also exposed as named fields; all readings remain here + so unknown measurements are preserved. """ diff --git a/src/pymbrewclient/mqtt/proto.py b/src/pymbrewclient/mqtt/proto.py index 8bf6ea4..efbb1b8 100644 --- a/src/pymbrewclient/mqtt/proto.py +++ b/src/pymbrewclient/mqtt/proto.py @@ -52,19 +52,26 @@ To dump a live message: ``protoc --decode_raw < captured_payload.bin`` -Field mapping (best-effort) - ========= ============================== ========== +Observed telemetry structure + ========= ============================== ============================ Field No. Name Wire type - ========= ============================== ========== - 1 session_id varint (int32) - 2 device_timestamp varint (int64, Unix epoch s) - 3 process_state varint (int32) - 4 user_action varint (int32) - 5 current_temperature 32-bit (float32) - 6 target_temperature 32-bit (float32) - 7 remaining_duration_seconds varint (int32) - 8 seconds_until_next_action varint (int32) - ========= ============================== ========== + ========= ============================== ============================ + 1 device_timestamp varint (Unix epoch ms) + 2 state nested message + 3 measurements repeated nested ``{id,float}`` + 11 session_id varint + 18 target_temperature 32-bit float + 19 current_temperature 32-bit float + 26 seconds_until_next_action varint + ========= ============================== ============================ + +The nested state message has observed fields 1 (current state), 2 (process +type), 3 (process state), and 8 (user action). Unknown outer, state, and +measurement fields remain available without speculative names. + +Live broker messages wrap this telemetry message in an envelope with field 1 +(sequence number), field 3 (nested telemetry), field 4 (session ID), and field +5 (Unix epoch milliseconds). """ import struct @@ -176,17 +183,47 @@ def _first_float32(raw: dict[int, list[object]], field_num: int) -> float | None """Return the first 32-bit float value for *field_num*, or ``None``.""" values = raw.get(field_num) if values and isinstance(values[0], (bytes, bytearray)) and len(values[0]) == 4: - return struct.unpack(" bytes | None: + """Return the first length-delimited value for *field_num*, or ``None``.""" + values = raw.get(field_num) + if values and isinstance(values[0], bytes): + return values[0] + return None + + +def _decode_measurements(raw: dict[int, list[object]]) -> dict[int, float]: + """Decode repeated outer field 3 entries as measurement ID/value pairs.""" + measurements: dict[int, float] = {} + for value in raw.get(3, []): + if not isinstance(value, bytes): + continue + entry = decode_raw_fields(value) + measurement_id = _first_varint(entry, 1) + measurement_value = _first_float32(entry, 2) + if measurement_id is not None and measurement_value is not None: + measurements[measurement_id] = measurement_value + return measurements + + +def _unwrap_telemetry(raw: dict[int, list[object]]) -> tuple[dict[int, list[object]], bool]: + """Return telemetry fields and whether the payload used the live envelope.""" + nested = _first_bytes(raw, 3) + if _first_varint(raw, 5) is None or nested is None: + return raw, False + return decode_raw_fields(nested), True + + def decode_device_log(msg: MqttMessage) -> DeviceLogMessage: """Attempt to decode a ``devices/logs/`` MQTT message as a DeviceLog. - On any decoding error the returned :class:`~pymbrewclient.mqtt.models.DeviceLogMessage` - will have :attr:`~pymbrewclient.mqtt.models.DeviceLogMessage.decode_error` set and all - telemetry fields will be ``None``. The raw :attr:`~pymbrewclient.mqtt.models.MqttMessage.payload` - is always preserved. + On a decoding error the returned :class:`~pymbrewclient.mqtt.models.DeviceLogMessage` + has :attr:`~pymbrewclient.mqtt.models.DeviceLogMessage.decode_error` set. Fields decoded + before a nested-message error remain available. The raw + :attr:`~pymbrewclient.mqtt.models.MqttMessage.payload` is always preserved. :param msg: The raw :class:`~pymbrewclient.mqtt.models.MqttMessage` to decode. :returns: A populated :class:`~pymbrewclient.mqtt.models.DeviceLogMessage`. @@ -200,27 +237,56 @@ def decode_device_log(msg: MqttMessage) -> DeviceLogMessage: try: raw = decode_raw_fields(msg.payload) - except Exception as exc: # noqa: BLE001 + except ValueError as exc: base.decode_error = f"Wire decode failed: {exc}" return base base.raw_fields = raw - # --- Best-effort field mapping (field numbers not officially confirmed) -- - - base.session_id = _first_varint(raw, 1) - - ts_int = _first_varint(raw, 2) - if ts_int is not None: - base.device_timestamp = datetime.fromtimestamp(ts_int, tz=timezone.utc) + try: + telemetry, is_wrapped = _unwrap_telemetry(raw) + except ValueError as exc: + base.decode_error = f"Telemetry decode failed: {exc}" + return base - base.process_state = _first_varint(raw, 3) - base.user_action = _first_varint(raw, 4) - base.current_temperature = _first_float32(raw, 5) - base.target_temperature = _first_float32(raw, 6) - base.remaining_duration_seconds = _first_varint(raw, 7) - base.seconds_until_next_action = _first_varint(raw, 8) + base.telemetry_fields = telemetry + + # Confirmed from captured device traffic. + if is_wrapped: + base.sequence_number = _first_varint(raw, 1) + base.session_id = _first_varint(raw, 4) + timestamp_ms = _first_varint(raw, 5) + else: + base.session_id = _first_varint(telemetry, 11) + timestamp_ms = _first_varint(telemetry, 1) + + if timestamp_ms is not None: + try: + base.device_timestamp = datetime.fromtimestamp(timestamp_ms / 1000, tz=timezone.utc) + except (OSError, OverflowError, ValueError) as exc: + base.decode_error = f"Invalid device timestamp: {exc}" + + state_payload = _first_bytes(telemetry, 2) + if state_payload is not None: + try: + base.state_fields = decode_raw_fields(state_payload) + except ValueError as exc: + base.decode_error = f"Nested state decode failed: {exc}" + else: + base.current_state = _first_varint(base.state_fields, 1) + base.process_type = _first_varint(base.state_fields, 2) + base.process_state = _first_varint(base.state_fields, 3) + base.user_action = _first_varint(base.state_fields, 8) + try: + base.measurements = _decode_measurements(telemetry) + except ValueError as exc: + base.decode_error = f"Measurement decode failed: {exc}" + + base.wifi_rssi_dbm = base.measurements.get(24) + base.current_temperature = _first_float32(telemetry, 19) + base.target_temperature = _first_float32(telemetry, 18) + base.seconds_until_next_action = _first_varint(telemetry, 26) if base.device_timestamp is not None and base.seconds_until_next_action is not None: base.next_action_at = base.device_timestamp + timedelta(seconds=base.seconds_until_next_action) diff --git a/src/pymbrewclient/rest/client.py b/src/pymbrewclient/rest/client.py index d1043fe..e1a10a8 100644 --- a/src/pymbrewclient/rest/client.py +++ b/src/pymbrewclient/rest/client.py @@ -204,12 +204,12 @@ def get_user_profile(self) -> UserProfile: :return: A UserProfile object containing the user UUID and profile data. """ logger.debug("Fetching user profile...") - response = self.get("v1/profile") + response = self.get("v1/users/me") data = response.json() return UserProfile( uuid=data["uuid"], email=data.get("email"), - username=data.get("username"), + username=data.get("username") or data.get("display_name"), first_name=data.get("first_name"), last_name=data.get("last_name"), ) diff --git a/tests/fixtures/device_log.bin b/tests/fixtures/device_log.bin deleted file mode 100644 index 3a40fb4d99bfd2c04e1411d608a1a32c1c845562..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 29 lcmd None: + mock_response = MagicMock() + mock_response.json.return_value = { + "uuid": "user-uuid-1", + "display_name": "test-brewer", + "first_name": "Test", + "last_name": "Brewer", + } + mock_get.return_value = mock_response + + profile = self.client.get_user_profile() + + mock_ensure_token.assert_called_once() + mock_get.assert_called_once_with( + f"{self.client.base_url}/v1/users/me/", + params=None, + headers=self.client.headers, + ) + self.assertEqual( + profile, + UserProfile( + uuid="user-uuid-1", + username="test-brewer", + first_name="Test", + last_name="Brewer", + ), + ) + @patch("pymbrewclient.rest.client.requests.get") @patch("pymbrewclient.rest.client.RestApiClient._ensure_token") def test_get_brewery_overview_with_unknown_fields(self, mock_ensure_token: MagicMock, mock_get: MagicMock) -> None: @@ -463,7 +497,7 @@ def test_watch_device_logs_requires_serial(self) -> None: self.assertIn("--serial", result.stdout) def test_watch_device_logs_streams_messages_as_json(self) -> None: - """Connect, receive one device-log message, then stop via --duration 0.""" + """Default watch output includes curated telemetry without protobuf internals.""" from datetime import datetime, timezone from pymbrewclient.mqtt.models import DeviceLogMessage @@ -475,9 +509,17 @@ def test_watch_device_logs_streams_messages_as_json(self) -> None: payload=b"\x08\x01", received_at=datetime(2024, 7, 26, 12, 0, 0, tzinfo=timezone.utc), device_uuid="SER-001", - session_id=1, + sequence_number=24098, + session_id=80851, + current_state=1, + process_type=4, process_state=80, - current_temperature=15.0, + current_temperature=19.3, + target_temperature=19.0, + wifi_rssi_dbm=-45.0, + seconds_until_next_action=851142, + next_action_at=datetime(2026, 8, 4, 16, 37, 6, 734000, tzinfo=timezone.utc), + measurements={0: 27.3, 3: 19.3, 24: -45.0}, ) mock_mqtt = MagicMock() @@ -528,9 +570,93 @@ def fake_wait(event: object, timeout: object = None) -> bool: # type: ignore[ov self.assertGreater(json_start, -1, f"No JSON object found in: {result.stdout!r}") payload, _ = decoder.raw_decode(result.stdout, json_start) self.assertEqual(payload["device_uuid"], "SER-001") - self.assertEqual(payload["session_id"], 1) + self.assertEqual(payload["sequence_number"], 24098) + self.assertEqual(payload["session_id"], 80851) + self.assertEqual(payload["current_state"], 1) + self.assertEqual(payload["process_type"], 4) self.assertEqual(payload["process_state"], 80) - self.assertEqual(payload["payload"], b"\x08\x01".hex()) + self.assertEqual(payload["current_temperature"], 19.3) + self.assertEqual(payload["target_temperature"], 19.0) + self.assertEqual(payload["wifi_rssi_dbm"], -45.0) + self.assertEqual(payload["seconds_until_next_action"], 851142) + self.assertEqual(payload["next_action_at"], "2026-08-04T16:37:06.734000Z") + self.assertNotIn("measurements", payload) + self.assertNotIn("payload", payload) + self.assertNotIn("raw_fields", payload) + self.assertNotIn("telemetry_fields", payload) + self.assertNotIn("state_fields", payload) + + def test_watch_device_logs_debug_output_preserves_every_field(self) -> None: + from pymbrewclient.mqtt.models import DeviceLogMessage + + message = DeviceLogMessage( + topic="devices/logs/SER-001", + payload=b"\x08\x01", + received_at=datetime(2024, 7, 26, 12, 0, 0, tzinfo=timezone.utc), + device_uuid="SER-001", + raw_fields={1: [1]}, + telemetry_fields={26: [851142]}, + state_fields={1: [1]}, + measurements={0: 27.3, 24: -45.0}, + ) + mock_mqtt = MagicMock() + mock_mqtt.__enter__ = MagicMock(return_value=mock_mqtt) + mock_mqtt.__exit__ = MagicMock(return_value=False) + callbacks: list = [] + mock_mqtt.on_device_log.side_effect = callbacks.append + mock_client = MagicMock() + mock_client.create_mqtt_client.return_value = mock_mqtt + + def fake_wait(event: object, timeout: object = None) -> bool: # type: ignore[override] + callbacks[0](message) + return True + + with ( + patch("pymbrewclient.cli.initialize_brewery_client", return_value=mock_client), + patch("threading.Event.wait", fake_wait), + ): + result = CliRunner().invoke( + app, + [ + "watch-device-logs", + "--username", + "user", + "--password", + "pass", + "--serial", + "SER-001", + "--format", + "json", + "--duration", + "0", + "--debug", + ], + ) + + self.assertEqual(result.exit_code, 0, result.stdout) + json_start = result.stdout.find("{") + payload, _ = json.JSONDecoder().raw_decode(result.stdout, json_start) + self.assertEqual(payload["payload"], "0801") + self.assertEqual(payload["raw_fields"], {"1": [1]}) + self.assertEqual(payload["telemetry_fields"], {"26": [851142]}) + self.assertEqual(payload["state_fields"], {"1": [1]}) + self.assertEqual(payload["measurements"], {"0": 27.3, "24": -45.0}) + + def test_curated_device_log_output_omits_absent_values(self) -> None: + from pymbrewclient.mqtt.models import DeviceLogMessage + + message = DeviceLogMessage( + topic="devices/logs/SER-001", + payload=b"", + received_at=datetime(2024, 7, 26, 12, 0, 0, tzinfo=timezone.utc), + ) + + output = curate_device_log_output(message) + + self.assertEqual(output["topic"], "devices/logs/SER-001") + self.assertNotIn("measurements", output) + self.assertNotIn("session_id", output) + self.assertNotIn("decode_error", output) def test_watch_device_logs_subscribes_all_serials(self) -> None: """Each --serial value is subscribed on the MQTT client.""" diff --git a/tests/test_mqtt.py b/tests/test_mqtt.py index 4bd005e..d3c605b 100644 --- a/tests/test_mqtt.py +++ b/tests/test_mqtt.py @@ -1,6 +1,6 @@ import struct import unittest -from datetime import datetime, timedelta, timezone +from datetime import datetime, timezone from pathlib import Path from unittest.mock import MagicMock, patch @@ -15,6 +15,7 @@ ) from pymbrewclient.mqtt.models import DeviceLogMessage, MqttMessage from pymbrewclient.mqtt.proto import decode_device_log, decode_raw_fields +from requests.certs import where as requests_ca_bundle # --------------------------------------------------------------------------- # Helpers @@ -38,24 +39,31 @@ def _encode_float32(value: float) -> bytes: def _build_device_log_payload( session_id: int = 12345, - device_timestamp: int = 1721993600, + device_timestamp_ms: int = 1721993600000, + current_state: int = 1, + process_type: int = 4, process_state: int = 80, user_action: int = 0, current_temperature: float = 15.1, target_temperature: float = 14.91, - remaining_duration_seconds: int = 4607, - seconds_until_next_action: int = 300, + unknown_field_8: int = 4607, + unknown_field_30: int = 300, ) -> bytes: - """Build a minimal valid protobuf payload matching the best-effort schema.""" + """Build a minimal valid protobuf payload matching the observed schema.""" + state = b"" + state += _encode_varint((1 << 3) | 0) + _encode_varint(current_state) + state += _encode_varint((2 << 3) | 0) + _encode_varint(process_type) + state += _encode_varint((3 << 3) | 0) + _encode_varint(process_state) + state += _encode_varint((8 << 3) | 0) + _encode_varint(user_action) + data = b"" - data += _encode_varint((1 << 3) | 0) + _encode_varint(session_id) - data += _encode_varint((2 << 3) | 0) + _encode_varint(device_timestamp) - data += _encode_varint((3 << 3) | 0) + _encode_varint(process_state) - data += _encode_varint((4 << 3) | 0) + _encode_varint(user_action) - data += _encode_varint((5 << 3) | 5) + _encode_float32(current_temperature) - data += _encode_varint((6 << 3) | 5) + _encode_float32(target_temperature) - data += _encode_varint((7 << 3) | 0) + _encode_varint(remaining_duration_seconds) - data += _encode_varint((8 << 3) | 0) + _encode_varint(seconds_until_next_action) + data += _encode_varint((1 << 3) | 0) + _encode_varint(device_timestamp_ms) + data += _encode_varint((2 << 3) | 2) + _encode_varint(len(state)) + state + data += _encode_varint((8 << 3) | 0) + _encode_varint(unknown_field_8) + data += _encode_varint((11 << 3) | 0) + _encode_varint(session_id) + data += _encode_varint((18 << 3) | 5) + _encode_float32(target_temperature) + data += _encode_varint((19 << 3) | 5) + _encode_float32(current_temperature) + data += _encode_varint((30 << 3) | 0) + _encode_varint(unknown_field_30) return data @@ -170,7 +178,8 @@ def test_tls_is_enabled(self) -> None: mock_paho = MagicMock() mock_cls.return_value = mock_paho MqttClient(api_token="t", user_uuid="u") - mock_paho.tls_set.assert_called_once_with() + mock_paho.tls_set.assert_called_once_with(ca_certs=requests_ca_bundle()) + mock_paho.tls_insecure_set.assert_not_called() def test_ws_path_is_set(self) -> None: with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: @@ -470,11 +479,11 @@ def test_on_message_extracts_device_uuid(self) -> None: class TestDeviceLogFixtureParsing(unittest.TestCase): - """Parse the captured binary device-log fixture from tests/fixtures/.""" + """Parse a captured device-log payload from tests/fixtures/.""" def setUp(self) -> None: - fixture_path = FIXTURES_DIR / "device_log.bin" - self.payload = fixture_path.read_bytes() + fixture_path = FIXTURES_DIR / "device_log.hex" + self.payload = bytes.fromhex(fixture_path.read_text().strip()) def test_fixture_file_exists_and_is_nonempty(self) -> None: self.assertGreater(len(self.payload), 0) @@ -482,14 +491,23 @@ def test_fixture_file_exists_and_is_nonempty(self) -> None: def test_decode_session_id(self) -> None: msg = _make_mqtt_message("devices/logs/test-dev", self.payload) decoded = decode_device_log(msg) - self.assertEqual(decoded.session_id, 12345) + self.assertEqual(decoded.session_id, 80851) def test_decode_device_timestamp(self) -> None: msg = _make_mqtt_message("devices/logs/test-dev", self.payload) decoded = decode_device_log(msg) - expected = datetime.fromtimestamp(1721993600, tz=timezone.utc) + expected = datetime(2026, 7, 25, 19, 58, 43, 644000, tzinfo=timezone.utc) self.assertEqual(decoded.device_timestamp, expected) + def test_decode_nested_state(self) -> None: + msg = _make_mqtt_message("devices/logs/test-dev", self.payload) + decoded = decode_device_log(msg) + self.assertEqual(decoded.current_state, 1) + self.assertEqual(decoded.process_type, 4) + self.assertEqual(decoded.process_state, 80) + self.assertEqual(decoded.user_action, 0) + self.assertEqual(decoded.state_fields[7], [b"-"]) + def test_decode_process_state(self) -> None: msg = _make_mqtt_message("devices/logs/test-dev", self.payload) decoded = decode_device_log(msg) @@ -504,23 +522,43 @@ def test_decode_current_temperature(self) -> None: msg = _make_mqtt_message("devices/logs/test-dev", self.payload) decoded = decode_device_log(msg) self.assertIsNotNone(decoded.current_temperature) - self.assertAlmostEqual(decoded.current_temperature, 15.1, places=2) + self.assertAlmostEqual(decoded.current_temperature, 19.3, places=2) def test_decode_target_temperature(self) -> None: msg = _make_mqtt_message("devices/logs/test-dev", self.payload) decoded = decode_device_log(msg) self.assertIsNotNone(decoded.target_temperature) - self.assertAlmostEqual(decoded.target_temperature, 14.91, places=2) + self.assertAlmostEqual(decoded.target_temperature, 19.0, places=2) - def test_decode_remaining_duration_seconds(self) -> None: - msg = _make_mqtt_message("devices/logs/test-dev", self.payload) - decoded = decode_device_log(msg) - self.assertEqual(decoded.remaining_duration_seconds, 4607) + def test_decode_next_action(self) -> None: + decoded = decode_device_log(_make_mqtt_message("devices/logs/test-dev", self.payload)) + self.assertEqual(decoded.seconds_until_next_action, 851903) + expected = datetime(2026, 8, 4, 16, 37, 6, 644000, tzinfo=timezone.utc) + self.assertEqual(decoded.next_action_at, expected) - def test_decode_seconds_until_next_action(self) -> None: + def test_decode_measurements_preserves_all_sensor_ids(self) -> None: msg = _make_mqtt_message("devices/logs/test-dev", self.payload) decoded = decode_device_log(msg) - self.assertEqual(decoded.seconds_until_next_action, 300) + expected = { + 0: 27.3, + 3: 19.3, + 4: 18.2, + 13: 121.0, + 19: 0.0, + 21: 0.0, + 22: 0.0, + 23: 0.0, + 24: -45.0, + 26: 100.0, + 27: 0.0, + } + self.assertEqual(decoded.measurements.keys(), expected.keys()) + for measurement_id, value in expected.items(): + self.assertAlmostEqual(decoded.measurements[measurement_id], value, places=2) + + def test_decode_wifi_rssi(self) -> None: + decoded = decode_device_log(_make_mqtt_message("devices/logs/test-dev", self.payload)) + self.assertEqual(decoded.wifi_rssi_dbm, -45.0) def test_raw_payload_is_preserved(self) -> None: msg = _make_mqtt_message("devices/logs/test-dev", self.payload) @@ -532,38 +570,58 @@ def test_no_decode_error(self) -> None: decoded = decode_device_log(msg) self.assertIsNone(decoded.decode_error) + def test_mqtt_callback_delivers_decoded_captured_payload(self) -> None: + with patch("pymbrewclient.mqtt.client._paho.Client") as mock_cls: + mock_paho = MagicMock() + mock_cls.return_value = mock_paho + client = MqttClient(api_token="t", user_uuid="u") -# --------------------------------------------------------------------------- -# Next-action timestamp calculation -# --------------------------------------------------------------------------- - + received: list[DeviceLogMessage] = [] + client.on_device_log(received.append) + mock_msg = MagicMock(topic="devices/logs/test-dev", payload=self.payload) + client._on_paho_message(mock_paho, None, mock_msg) -class TestNextActionTimestamp(unittest.TestCase): - def test_next_action_at_is_calculated_correctly(self) -> None: - payload = _build_device_log_payload( - device_timestamp=1721993600, - seconds_until_next_action=300, - ) - msg = _make_mqtt_message("devices/logs/dev", payload) - decoded = decode_device_log(msg) + self.assertEqual(len(received), 1) + self.assertEqual(received[0].session_id, 80851) + self.assertEqual(received[0].process_state, 80) + self.assertAlmostEqual(received[0].measurements[3], 19.3, places=2) - expected_ts = datetime.fromtimestamp(1721993600, tz=timezone.utc) + timedelta(seconds=300) - self.assertEqual(decoded.next_action_at, expected_ts) - def test_next_action_at_is_none_when_timestamp_absent(self) -> None: - """next_action_at requires both device_timestamp and seconds_until_next_action.""" - # Encode only seconds_until_next_action, omit device_timestamp - payload = _encode_varint((8 << 3) | 0) + _encode_varint(300) - msg = _make_mqtt_message("devices/logs/dev", payload) - decoded = decode_device_log(msg) - self.assertIsNone(decoded.next_action_at) +class TestLiveDeviceLogEnvelopeParsing(unittest.TestCase): + """Parse the envelope received by the live MQTT watch command.""" - def test_next_action_at_is_utc(self) -> None: - payload = _build_device_log_payload(device_timestamp=1721993600, seconds_until_next_action=60) - msg = _make_mqtt_message("devices/logs/dev", payload) - decoded = decode_device_log(msg) - self.assertIsNotNone(decoded.next_action_at) - self.assertEqual(decoded.next_action_at.tzinfo, timezone.utc) + def setUp(self) -> None: + fixture_path = FIXTURES_DIR / "device_log_envelope.hex" + self.payload = bytes.fromhex(fixture_path.read_text().strip()) + self.decoded = decode_device_log(_make_mqtt_message("devices/logs/test-dev", self.payload)) + + def test_decodes_envelope_metadata(self) -> None: + self.assertEqual(self.decoded.sequence_number, 24098) + self.assertEqual(self.decoded.session_id, 80851) + expected = datetime(2026, 7, 25, 20, 6, 0, 644000, tzinfo=timezone.utc) + self.assertEqual(self.decoded.device_timestamp, expected) + + def test_decodes_nested_telemetry(self) -> None: + self.assertEqual(self.decoded.current_state, 1) + self.assertEqual(self.decoded.process_type, 4) + self.assertEqual(self.decoded.process_state, 80) + self.assertEqual(self.decoded.user_action, 0) + self.assertEqual(self.decoded.current_temperature, 19.2) + self.assertEqual(self.decoded.target_temperature, 19.0) + self.assertEqual(self.decoded.wifi_rssi_dbm, -36.0) + self.assertEqual(self.decoded.measurements[3], 19.2) + self.assertEqual(self.decoded.measurements[24], -36.0) + + def test_decodes_next_action_matching_portal(self) -> None: + self.assertEqual(self.decoded.seconds_until_next_action, 851466) + expected = datetime(2026, 8, 4, 16, 37, 6, 644000, tzinfo=timezone.utc) + self.assertEqual(self.decoded.next_action_at, expected) + + def test_preserves_envelope_and_telemetry_fields(self) -> None: + self.assertEqual(self.decoded.raw_fields[1], [24098]) + self.assertIn(3, self.decoded.raw_fields) + self.assertEqual(self.decoded.telemetry_fields[11], [80851]) + self.assertIsNone(self.decoded.decode_error) # --------------------------------------------------------------------------- @@ -599,6 +657,32 @@ def test_unknown_fields_are_captured_in_raw_fields(self) -> None: self.assertIn(99, decoded.raw_fields) self.assertIsNone(decoded.decode_error) + def test_malformed_nested_state_sets_decode_error(self) -> None: + state = b"\x08" + payload = _encode_varint((2 << 3) | 2) + _encode_varint(len(state)) + state + decoded = decode_device_log(_make_mqtt_message("devices/logs/dev", payload)) + self.assertIsNotNone(decoded.decode_error) + self.assertIn("Nested state decode failed", decoded.decode_error) + + def test_malformed_measurement_sets_decode_error(self) -> None: + measurement = b"\x08" + payload = _encode_varint((3 << 3) | 2) + _encode_varint(len(measurement)) + measurement + decoded = decode_device_log(_make_mqtt_message("devices/logs/dev", payload)) + self.assertIsNotNone(decoded.decode_error) + self.assertIn("Measurement decode failed", decoded.decode_error) + + def test_malformed_live_envelope_sets_decode_error(self) -> None: + payload = ( + _encode_varint((3 << 3) | 2) + + _encode_varint(1) + + b"\x08" + + _encode_varint((5 << 3) | 0) + + _encode_varint(1721993600000) + ) + decoded = decode_device_log(_make_mqtt_message("devices/logs/dev", payload)) + self.assertIsNotNone(decoded.decode_error) + self.assertIn("Telemetry decode failed", decoded.decode_error) + def test_malformed_message_does_not_raise(self) -> None: """Decoding errors must never propagate as exceptions.""" bad_payloads = [