From 1ad61a56d1b73827a5961b352754e17bd6ada2d8 Mon Sep 17 00:00:00 2001 From: arduano Date: Sun, 15 Mar 2026 18:14:14 +1100 Subject: [PATCH] refactor(q7): narrow decoded command response type --- roborock/devices/rpc/b01_q7_channel.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/roborock/devices/rpc/b01_q7_channel.py b/roborock/devices/rpc/b01_q7_channel.py index fa0b5b8b..d2e373d3 100644 --- a/roborock/devices/rpc/b01_q7_channel.py +++ b/roborock/devices/rpc/b01_q7_channel.py @@ -6,7 +6,7 @@ import json import logging from collections.abc import Callable -from typing import Any, TypeVar +from typing import TypeAlias, TypeVar from roborock.devices.transport.mqtt_channel import MqttChannel from roborock.exceptions import RoborockException @@ -16,6 +16,7 @@ _LOGGER = logging.getLogger(__name__) _TIMEOUT = 10.0 _T = TypeVar("_T") +DecodedB01Response: TypeAlias = dict[str, object] | str def _matches_map_response(response_message: RoborockMessage, *, version: bytes | None) -> bytes | None: @@ -61,11 +62,11 @@ def on_message(response_message: RoborockMessage) -> None: async def send_decoded_command( mqtt_channel: MqttChannel, request_message: Q7RequestMessage, -) -> Any: +) -> DecodedB01Response: """Send a command on the MQTT channel and get a decoded response.""" _LOGGER.debug("Sending B01 MQTT command: %s", request_message) - def find_response(response_message: RoborockMessage) -> Any | None: + def find_response(response_message: RoborockMessage) -> DecodedB01Response | None: """Handle incoming messages and resolve the future.""" try: decoded_dps = decode_rpc_response(response_message)