Add basic tags and fix & improve type annotation

pyown/__main__.py

@@ -1,5 +1,5 @@
-import asyncio
 import argparse
+import asyncio
 
 from .client import OWNClient
 
@@ -14,11 +14,12 @@ async def main(host: str, port: int, password: int) -> None:
     await client.close()
     print("Disconnected")
 
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="OpenWebNet client")
     parser.add_argument("--host", type=str, help="The OpenWebNet gateway IP address", default="192.168.0.120")
     parser.add_argument("--port", type=int, help="The OpenWebNet gateway port", default=20000)
     parser.add_argument("--password", type=str, help="The OpenWebNet gateway password", default="12345")
     args = parser.parse_args()
-    
-    asyncio.run(main(args.host, args.port, args.password))
+
+    asyncio.run(main(args.host, args.port, args.password))

pyown/auth/open.py

@@ -1,4 +1,4 @@
-def ownCalcPass(password: str | int, nonce: str) -> str:
+def ownCalcPass(password: str | int, nonce: str | int) -> str:
     """
     Encode the password using the OPEN algorithm.
     Source: https://rosettacode.org/wiki/OpenWebNet_password#Python
@@ -17,6 +17,9 @@ def ownCalcPass(password: str | int, nonce: str) -> str:
     if isinstance(password, str):
         password = int(password)
 
+    if isinstance(nonce, int):
+        nonce = str(nonce)
+
     for c in nonce:
         if c != "0":
             if start:

pyown/client.py

@@ -1,15 +1,13 @@
 import asyncio
 from asyncio import StreamReader, StreamWriter
-
 from typing import Final, Literal
 
 from pyown.message import *
-from .auth.open import ownCalcPass
 
+from .auth.open import ownCalcPass
 
 __all__ = ["OWNClient"]
 
-
 # Use for sending commands and getting responses from the gateway
 COMMAND_SESSION: Final = "9"
 # Get every event from the gateway
@@ -44,7 +42,7 @@ async def connect(self, session_type: Literal["9", "1"] = COMMAND_SESSION):
             raise ConnectionError(f"Unexpected response: {msg}")
         # Send the session type
         await self.send(RawMessage(tags=["99", session_type]))
-        
+
     async def _open_auth(self, msg: OWNMessage):
         """
         Implement the OPEN algorithm for authentication.
@@ -81,7 +79,7 @@ async def _send(self, msg: str | bytes):
         """
         if isinstance(msg, str):
             msg = msg.encode()
-        
+
         self.writer.write(msg)
         await self.writer.drain()
 
@@ -99,7 +97,7 @@ async def _recv(self, timeout: int = 5, *, consume: bool = True) -> str:
                 msg = await self.reader.readuntil(b"##")
             else:
                 msg = await self.reader.readuntil(b"##", 1024)
-            
+
         return msg.decode(errors="ignore", encoding="ascii")  # omw uses only ascii characters
 
     async def send(self, msg: OWNMessage):

pyown/exceptions.py

@@ -14,3 +14,11 @@ class InvalidMessage(Exception):
     def __init__(self, message: str) -> None:
         self.message = message
         super().__init__(f"Invalid message: {message}")
+
+
+class InvalidTag(Exception):
+    tag: str
+
+    def __init__(self, tag: str) -> None:
+        self.tag = tag
+        super().__init__(f"Invalid tag: {tag}")

pyown/messages/__init__.py

@@ -1,4 +1,3 @@
 from .ack import ACK
-from .nack import NACK
-
 from .base import MessageType, BaseMessage, parse_message
+from .nack import NACK

pyown/messages/ack.py

@@ -1,11 +1,9 @@
 import re
-
-from typing import Self, Pattern, AnyStr
+from typing import Self, Pattern, AnyStr, Final
 
 from .base import BaseMessage, MessageType
 from ..exceptions import ParseError
 
-
 __all__ = [
     "ACK",
 ]
@@ -14,7 +12,7 @@
 class ACK(BaseMessage):
     """Represent an ACK message"""
     _type = MessageType.ACK
-    _tags: list[str] = ["#", "1"]
+    _tags: Final[tuple[str]] = ("#", "1")
 
     _regex: Pattern[AnyStr] = re.compile(r"^\*#\*1##$")
 

pyown/messages/base.py

@@ -1,9 +1,8 @@
 import abc
-import re
 import copy
-
+import re
 from enum import StrEnum
-from typing import Final, TypeVar, Type, Pattern, AnyStr
+from typing import Final, TypeVar, Type, Pattern, AnyStr, Iterable
 
 from ..exceptions import ParseError, InvalidMessage
 
@@ -29,7 +28,7 @@ class MessageType(StrEnum):
 
 class BaseMessage(abc.ABC):
     _type: MessageType = MessageType.GENERIC  # Type of the message
-    _tags: list[StrEnum, str]  # Contains the tags of the message
+    _tags: Iterable[str]  # Contains the tags of the message
 
     prefix: Final[str] = "*"  # Prefix of the message
     suffix: Final[str] = "##"  # Suffix of the message
@@ -55,7 +54,7 @@ def __repr__(self) -> str:
         Returns:
             str: The representation
         """
-        return f"<{self.__class__.__name__}: {','.join(self.tags)}>"
+        return f"<{self.__class__.__name__}: {','.join(self._tags)}>"
 
     def __hash__(self) -> int:
         return hash((self._type, self._tags))
@@ -78,7 +77,7 @@ def pattern(cls) -> Pattern[AnyStr]:
         return cls._regex
 
     @property
-    def tags(self) -> list[StrEnum, str]:
+    def tags(self) -> Iterable[str]:
         """
         Return the tags of the message.
         The tags are the elements that compose the message, like the WHO, WHAT, WHERE, etc.
@@ -128,11 +127,9 @@ def parse_message(message: str) -> Type[BaseMessage]:
     if message.count(BaseMessage.suffix) != 1:
         raise InvalidMessage(message=message)
 
-    tags = (message.strip()
-            .removeprefix(BaseMessage.prefix)
-            .removesuffix(BaseMessage.suffix)
-            .split(BaseMessage.separator)
-            )
+    message = message.strip()
+
+    tags = message.removeprefix(BaseMessage.prefix).removesuffix(BaseMessage.suffix).split(BaseMessage.separator)
 
     for subclass in BaseMessage.__subclasses__():
         # noinspection PyUnresolvedReferences

pyown/messages/nack.py

@@ -1,11 +1,9 @@
 import re
-
-from typing import Self, Pattern, AnyStr
+from typing import Self, Pattern, AnyStr, Final
 
 from .base import BaseMessage, MessageType
 from ..exceptions import ParseError
 
-
 __all__ = [
     "NACK",
 ]
@@ -14,7 +12,7 @@
 class NACK(BaseMessage):
     """Represent an NACK message"""
     _type = MessageType.NACK
-    _tags: list[str] = ["#", "0"]
+    _tags: Final[tuple[str]] = ("#", "0")
 
     _regex: Pattern[AnyStr] = re.compile(r"^\*#\*0##$")
 

pyown/messages/normal.py

@@ -0,0 +1,48 @@
+import re
+from typing import Self, Pattern, AnyStr
+
+from .base import BaseMessage, MessageType
+from ..tags import Who, What, Where
+
+__all__ = [
+    "NormalMessage",
+]
+
+
+class NormalMessage(BaseMessage):
+    """Represent an NACK message"""
+    _type = MessageType.NORMAL
+    _tags: tuple[Who, What, Where]
+
+    _regex: Pattern[AnyStr] = re.compile(r"^\*[0-9#]+\*[0-9#]*\*[0-9#]*##$")
+
+    def __init__(self, tags: tuple[Who, What, Where]):
+        self._tags = tags
+
+    @property
+    def who(self) -> Who:
+        return self._tags[0]
+
+    @property
+    def what(self) -> What:
+        return self._tags[1]
+
+    @property
+    def where(self) -> Where:
+        return self._tags[2]
+
+    @property
+    def message(self) -> str:
+        return f"*{self.who}*{self.what}*{self.where}##"
+
+    @classmethod
+    def parse(cls, tags: list[str]) -> Self:
+        """Parse the tags of a message from the OpenWebNet bus."""
+
+        return cls(
+            tags=(
+                Who(tags[0]),
+                What(tags[1]),
+                Where(tags[2])
+            )
+        )

pyown/tags/__init__.py

@@ -1 +1,5 @@
-from .who import WHO
+from .base import Value
+from .dimension import Dimension
+from .what import What
+from .where import Where
+from .who import Who

pyown/tags/base.py

@@ -0,0 +1,64 @@
+from typing import Final
+
+from ..exceptions import InvalidTag
+
+__all__ = [
+    "Tag",
+    "TagWithParameters",
+    "Value",
+]
+
+VALID_TAG_CHARS: Final[str] = "0123456789#"
+
+
+class Tag(str):
+    """Tag class."""
+
+    def __init__(self, string: str):
+        # Check if the string contains only valid characters
+        if not all(c in VALID_TAG_CHARS for c in string):
+            raise InvalidTag(string)
+
+        super().__init__()
+
+    @property
+    def value(self) -> int | None:
+        """Return the value of the tag without its parameters or prefix"""
+        val = self.removeprefix("#")
+        if len(val) > 0:
+            return int(val)
+        else:
+            return None
+
+    @property
+    def parameters(self) -> list[str] | None:
+        """Return the parameters of the tag"""
+        return None
+
+    @property
+    def tag(self) -> str:
+        """Return the tag"""
+        return self
+
+
+class TagWithParameters(Tag):
+    @property
+    def value(self) -> int | None:
+        """Return the value of the tag without its parameters or prefix"""
+        val = self.split("#")[0]
+        if len(val) > 0:
+            return int(val)
+        else:
+            return None
+
+    @property
+    def parameters(self) -> list[str]:
+        """Return the parameters of the tag"""
+        return self.split("#")[1:]
+
+
+class Value(Tag):
+    """
+    Represent a value tag in a dimension response message.
+    """
+    pass

pyown/tags/dimension.py

@@ -0,0 +1,17 @@
+from .base import Tag
+
+__all__ = [
+    "Dimension",
+]
+
+
+class Dimension(Tag):
+    """
+    Represent the DIMENSION tag.
+
+    It's not clear in the official documentation what exactly is the DIMENSION tag.
+    But in many cases it's used to request information about the status of a device,
+    making it similar to the WHAT tag.
+    The difference between the two is that the DIMENSION tag does not allow parameters.
+    """
+    pass

pyown/tags/what.py

@@ -0,0 +1,15 @@
+from .base import TagWithParameters
+
+__all__ = [
+    "What",
+]
+
+
+class What(TagWithParameters):
+    """
+    Represent the WHAT tag.
+
+    The tag WHAT, identifies the action to make (ON lights, OFF lights, dimmer at 20%,
+    shutters UP, shutters DOWN, set program 1 in thermoregulation central, etc...)
+    """
+    pass

pyown/tags/where.py

@@ -0,0 +1,15 @@
+from .base import TagWithParameters
+
+__all__ = [
+    "Where",
+]
+
+
+class Where(TagWithParameters):
+    """
+    Represent the WHERE tag.
+
+    The tag WHERE detects the objects involved by the frame (environment, room, single
+    object, whole system).
+    """
+    pass