TypeAlias for RscpMessage

Author: vchrisb

e3dc/__init__.py

@@ -8,7 +8,7 @@
 from ._e3dc import E3DC, AuthenticationError, NotAvailableError, PollError, SendError
 from ._e3dc_rscp_local import CommunicationError, RSCPAuthenticationError, RSCPKeyError
 from ._e3dc_rscp_web import RequestTimeoutError, SocketNotReady
-from ._rscpLib import FrameError
+from ._rscpLib import FrameError, RscpMessage
 from ._rscpLib import set_debug as set_rscp_debug
 
 __all__ = [
@@ -23,6 +23,7 @@
     "RequestTimeoutError",
     "SocketNotReady",
     "FrameError",
+    "RscpMessage",
     "set_rscp_debug",
 ]
 __version__ = "0.9.3"

e3dc/_e3dc_rscp_local.py

@@ -5,10 +5,9 @@
 # Licensed under a MIT license. See LICENSE for details
 
 import socket
-from typing import Any
 
 from ._RSCPEncryptDecrypt import RSCPEncryptDecrypt
-from ._rscpLib import rscpDecode, rscpEncode, rscpFrame
+from ._rscpLib import RscpMessage, rscpDecode, rscpEncode, rscpFrame
 from ._rscpTags import RscpError, RscpTag, RscpType
 
 PORT = 5033
@@ -64,9 +63,7 @@ def __init__(
         self.encdec: RSCPEncryptDecrypt
         self.processedData = None
 
-    def _send(
-        self, plainMsg: tuple[str | int | RscpTag, str | int | RscpType, Any]
-    ) -> None:
+    def _send(self, plainMsg: RscpMessage) -> None:
         sendData = rscpFrame(rscpEncode(plainMsg))
         encData = self.encdec.encrypt(sendData)
         self.socket.send(encData)
@@ -78,19 +75,15 @@ def _receive(self):
         decData = rscpDecode(self.encdec.decrypt(data))[0]
         return decData
 
-    def sendCommand(
-        self, plainMsg: tuple[str | int | RscpTag, str | int | RscpType, Any]
-    ) -> None:
+    def sendCommand(self, plainMsg: RscpMessage) -> None:
         """Sending RSCP command.
 
         Args:
             plainMsg (tuple): plain message
         """
         self.sendRequest(plainMsg)  # same as sendRequest but doesn't return a value
 
-    def sendRequest(
-        self, plainMsg: tuple[str | int | RscpTag, str | int | RscpType, Any]
-    ) -> tuple[str | int | RscpTag, str | int | RscpType, Any]:
+    def sendRequest(self, plainMsg: RscpMessage) -> RscpMessage:
         """Sending RSCP request.
 
         Args:

e3dc/_e3dc_rscp_web.py

@@ -8,12 +8,13 @@
 import struct
 import threading
 import time
-from typing import Any, Callable
+from typing import Callable
 
 import tzlocal
 from websocket import ABNF, WebSocketApp
 
 from ._rscpLib import (
+    RscpMessage,
     rscpDecode,
     rscpEncode,
     rscpFindTag,
@@ -132,11 +133,9 @@ def reset(self):
         self.virtConId = None
         self.virtAuthLevel = None
         self.webSerialno = None
-        self.responseCallback: Callable[
-            [tuple[str | int | RscpTag, str | int | RscpType, Any]], None
-        ]
+        self.responseCallback: Callable[[RscpMessage], None]
         self.responseCallbackCalled = False
-        self.requestResult: tuple[str | int | RscpTag, str | int | RscpType, Any]
+        self.requestResult: RscpMessage
 
     def buildVirtualConn(self):
         """Method to create Virtual Connection."""
@@ -161,9 +160,7 @@ def buildVirtualConn(self):
         # print("--------------------- Sending virtual conn")
         self.ws.send(virtualConn, ABNF.OPCODE_BINARY)
 
-    def respondToINFORequest(
-        self, decoded: tuple[str | int | RscpTag, str | int | RscpType, Any]
-    ):
+    def respondToINFORequest(self, decoded: RscpMessage):
         """Create Response to INFO request."""
         TIMEZONE_STR, utcDiffS = calcTimeZone()
 
@@ -228,9 +225,7 @@ def respondToINFORequest(
             return ""
         return None  # this is no standard request
 
-    def registerConnectionHandler(
-        self, decodedMsg: tuple[str | int | RscpTag, str | int | RscpType, Any]
-    ):
+    def registerConnectionHandler(self, decodedMsg: RscpMessage):
         """Registering Connection Handler."""
         if self.conId == 0:
             self.conId = rscpFindTagIndex(decodedMsg, RscpTag.SERVER_CONNECTION_ID)
@@ -324,14 +319,10 @@ def on_message(self, message: bytes):
                 ABNF.OPCODE_BINARY,
             )
 
-    def _defaultRequestCallback(
-        self, msg: tuple[str | int | RscpTag, str | int | RscpType, Any]
-    ):
+    def _defaultRequestCallback(self, msg: RscpMessage):
         self.requestResult = msg
 
-    def sendRequest(
-        self, message: tuple[str | int | RscpTag, str | int | RscpType, Any]
-    ) -> tuple[str | int | RscpTag, str | int | RscpType, Any]:
+    def sendRequest(self, message: RscpMessage) -> RscpMessage:
         """Send a request and wait for a response."""
         self._sendRequest_internal(rscpFrame(rscpEncode(message)))
         for _ in range(self.TIMEOUT * 10):
@@ -343,24 +334,19 @@ def sendRequest(
 
         return self.requestResult
 
-    def sendCommand(
-        self, message: tuple[str | int | RscpTag, str | int | RscpType, Any]
-    ):
+    def sendCommand(self, message: RscpMessage):
         """Send a command."""
         return self._sendRequest_internal(rscpFrame(rscpEncode(message)))
 
     def _sendRequest_internal(
         self,
-        innerFrame: bytes | tuple[str | int | RscpTag, str | int | RscpType, Any],
-        callback: (
-            Callable[[tuple[str | int | RscpTag, str | int | RscpType, Any]], None]
-            | None
-        ) = None,
+        innerFrame: bytes | RscpMessage,
+        callback: Callable[[RscpMessage], None] | None = None,
     ):
         """Internal send request method.
 
         Args:
-            innerFrame (Union[tuple, <RSCP encoded frame>]): inner frame
+            innerFrame (tuple | bytes): inner frame
             callback (str): callback method
             synchronous (bool): If True, the method waits for a response (i.e. exits after calling callback).
                 If True and callback = None, the method returns the (last) response message

e3dc/_rscpLib.py

@@ -8,7 +8,7 @@
 import struct
 import time
 import zlib
-from typing import Any, cast
+from typing import Any, TypeAlias, cast
 
 from ._rscpTags import (
     RscpTag,
@@ -21,6 +21,9 @@
     getStrRscpType,
 )
 
+# Type alias for RSCP messages
+RscpMessage: TypeAlias = tuple[str | int | RscpTag, str | int | RscpType, Any]
+
 DEBUG_DICT = {"print_rscp": False}
 
 
@@ -60,9 +63,9 @@ def set_debug(debug: bool):
 
 
 def rscpFindTag(
-    decodedMsg: tuple[str | int | RscpTag, str | int | RscpType, Any] | None,
+    decodedMsg: RscpMessage | None,
     tag: int | str | RscpTag,
-) -> tuple[str | int | RscpTag, str | int | RscpType, Any] | None:
+) -> RscpMessage | None:
     """Finds a submessage with a specific tag.
 
     Args:
@@ -83,9 +86,7 @@ def rscpFindTag(
     if decodedMsg[0] == tagStr:
         return decodedMsg
     if isinstance(decodedMsg[2], list):
-        msgList: list[tuple[str | int | RscpTag, str | int | RscpType, Any]] = cast(
-            list[tuple[str | int | RscpTag, str | int | RscpType, Any]], decodedMsg[2]
-        )
+        msgList: list[RscpMessage] = cast(list[RscpMessage], decodedMsg[2])
         for msg in msgList:
             msgValue = rscpFindTag(msg, tag)
             if msgValue is not None:
@@ -94,7 +95,7 @@ def rscpFindTag(
 
 
 def rscpFindTagIndex(
-    decodedMsg: tuple[str | int | RscpTag, str | int | RscpType, Any] | None,
+    decodedMsg: RscpMessage | None,
     tag: int | str | RscpTag,
     index: int = 2,
 ) -> Any:
@@ -127,7 +128,7 @@ class FrameError(Exception):
 
 
 def rscpEncode(
-    tag: int | str | RscpTag | tuple[str | int | RscpTag, str | int | RscpType, Any],
+    tag: int | str | RscpTag | RscpMessage,
     rscptype: int | str | RscpType | None = None,
     data: Any = None,
 ) -> bytes:
@@ -172,9 +173,7 @@ def rscpEncode(
     elif rscptype == RscpType.Container:
         if isinstance(data, list):
             newData = b""
-            dataList: list[tuple[str | int | RscpTag, str | int | RscpType, Any]] = (
-                cast(list[tuple[str | int | RscpTag, str | int | RscpType, Any]], data)
-            )
+            dataList: list[RscpMessage] = cast(list[RscpMessage], data)
             for dataChunk in dataList:
                 newData += rscpEncode(
                     dataChunk[0], dataChunk[1], dataChunk[2]
@@ -248,7 +247,7 @@ def rscpFrameDecode(frameData: bytes, returnFrameLen: bool = False):
 
 def rscpDecode(
     data: bytes,
-) -> tuple[tuple[str | int | RscpTag, str | int | RscpType, Any], int]:
+) -> tuple[RscpMessage, int]:
     """Decodes RSCP data."""
     headerFmt = (
         "<IBH"  # format of header: little-endian, Uint32 tag, Uint8 type, Uint16 length
@@ -273,7 +272,7 @@ def rscpDecode(
 
     if type_ == RscpType.Container:
         # this is a container: parse the inside
-        dataList: list[tuple[str | int | RscpTag, str | int | RscpType, Any]] = []
+        dataList: list[RscpMessage] = []
         curByte = headerSize
         while curByte < headerSize + length:
             innerData, usedLength = rscpDecode(data[curByte:])