Implement My Open Web Net Introduction document

- Add support for the various types of messages
- Various improvements

README.md

@@ -10,7 +10,7 @@ OpenWebNet is a home automation protocol developed by Bticino (now part of Legra
 lights, shutters, heating, etc.
 It was developed around 2000, and it's still used today in many installations.
 It does not implement any encryption, so it's not secure to use it over the internet.
-Also many devices implement only the old password algorithm, which is easily bruteforceable.
+Also, many devices implement only the old password algorithm, which is easily bruteforceable.
 So, when using OpenWebNet, be sure to use it only in a trusted network and taking security measures, like vlan
 separation.
 

pyown/messages/__init__.py

@@ -1,3 +1,6 @@
 from .ack import ACK
 from .base import MessageType, BaseMessage, parse_message
 from .nack import NACK
+from .status import *
+from .dimension import *
+from .normal import *

pyown/messages/ack.py

@@ -10,8 +10,12 @@
 
 
 class ACK(BaseMessage):
-    """Represent an ACK message"""
-    _type = MessageType.ACK
+    """
+    Represent an ACK message
+
+    Syntax: *#*1##
+    """
+    _type: Final[MessageType] = MessageType.ACK
     _tags: Final[tuple[str]] = ("#", "1")
 
     _regex: Pattern[AnyStr] = re.compile(r"^\*#\*1##$")

pyown/messages/base.py

@@ -20,8 +20,8 @@ class MessageType(StrEnum):
     NACK = "NACK"
     NORMAL = "NORMAL"
     STATUS_REQUEST = "STATUS REQUEST"
-    STATUS_RESPONSE = "STATUS RESPONSE"
     DIMENSION_REQUEST = "DIMENSION REQUEST"
+    DIMENSION_WRITING = "DIMENSION WRITING"
     DIMENSION_RESPONSE = "DIMENSION RESPONSE"
     GENERIC = "GENERIC"
 

pyown/messages/dimension.py

@@ -0,0 +1,125 @@
+import re
+from typing import Self, Pattern, AnyStr, Final
+
+from .base import BaseMessage, MessageType
+from ..tags import Who, What, Where, Dimension, Value
+
+__all__ = [
+    "DimensionRequest",
+    "DimensionWriting",
+    "DimensionResponse",
+]
+
+
+class DimensionRequest(BaseMessage):
+    """
+    Represent a dimension request message
+
+    Syntax: *#who*where*dimension##
+
+    This is sent by the client to the server
+    """
+    _type: Final[MessageType] = MessageType.DIMENSION_REQUEST
+    _tags: Final[tuple[Who, Where, Dimension]]
+
+    _regex: Pattern[AnyStr] = re.compile(r"^\*#[0-9#]+\*[0-9#]*\*[0-9]*##$")
+
+    def __init__(self, tags: tuple[Who, Where, Dimension]):
+        self._tags = tags
+
+    @property
+    def who(self) -> Who:
+        return self._tags[0]
+
+    @property
+    def where(self) -> Where:
+        return self._tags[1]
+
+    @property
+    def dimension(self) -> Dimension:
+        return self._tags[2]
+
+    @property
+    def message(self) -> str:
+        return f"*#{self.who}*{self.where}*{self.dimension}##"
+
+    @classmethod
+    def parse(cls, tags: list[str]) -> Self:
+        """Parse the tags of a message from the OpenWebNet bus."""
+
+        return cls(
+            tags=(
+                Who(tags[0].removeprefix("#")),
+                Where(tags[1]),
+                Dimension(tags[2])
+            )
+        )
+
+
+class DimensionWriting(BaseMessage):
+    """
+    Represent a dimension writing message
+
+    Syntax: *#who*where*#dimension*value1*value2*...*valueN##
+
+    This is sent by the client to the server
+    """
+    _type: MessageType = MessageType.DIMENSION_WRITING
+    _tags: Final[tuple[Who, Where, Dimension, Value, ...]]
+
+    _regex: Pattern[AnyStr] = re.compile(r"^\*#[0-9#]+\*[0-9#]*\*#[0-9]*(?:\*[0-9#]*)*##$")
+
+    def __init__(self, tags: tuple[Who, Where, Dimension, Value, ...]):
+        self._tags = tags
+
+    @property
+    def who(self) -> Who:
+        return self._tags[0]
+
+    @property
+    def where(self) -> Where:
+        return self._tags[1]
+
+    @property
+    def dimension(self) -> Dimension:
+        return self._tags[2]
+
+    @property
+    def values(self) -> tuple[Value]:
+        return self._tags[3:]
+
+    @property
+    def message(self) -> str:
+        return f"*#{self.who}*{self.where}*#{self.dimension}*{'*'.join(self.values)}##"
+
+    @classmethod
+    def parse(cls, tags: list[str]) -> Self:
+        """Parse the tags of a message from the OpenWebNet bus."""
+
+        values: list[Value] = [Value(t) for t in tags[3:]]
+
+        # noinspection PyTypeChecker
+        return cls(
+            tags=(
+                Who(tags[0].removeprefix("#")),
+                Where(tags[1]),
+                Dimension(tags[2]).removeprefix("#"),
+                *values
+            )
+        )
+
+
+class DimensionResponse(DimensionWriting, BaseMessage):
+    """
+    Represent a dimension writing message
+
+    Syntax: *#who*where*dimension*value1*value2*...*valueN##
+
+    This is sent by the server to the client
+    """
+    _type: Final[MessageType] = MessageType.DIMENSION_RESPONSE
+    _regex: Pattern[AnyStr] = re.compile(r"^\*#[0-9#]+\*[0-9#]*\*[0-9]*(?:\*[0-9#]*)*##$")
+
+    @property
+    def message(self) -> str:
+        return f"*#{self.who}*{self.where}*{self.dimension}*{'*'.join(self.values)}##"

pyown/messages/nack.py

@@ -10,8 +10,12 @@
 
 
 class NACK(BaseMessage):
-    """Represent an NACK message"""
-    _type = MessageType.NACK
+    """
+    Represent an NACK message
+
+    Syntax: *#*0##
+    """
+    _type: Final[MessageType] = MessageType.NACK
     _tags: Final[tuple[str]] = ("#", "0")
 
     _regex: Pattern[AnyStr] = re.compile(r"^\*#\*0##$")

pyown/messages/normal.py

@@ -1,5 +1,5 @@
 import re
-from typing import Self, Pattern, AnyStr
+from typing import Self, Pattern, AnyStr, Final
 
 from .base import BaseMessage, MessageType
 from ..tags import Who, What, Where
@@ -10,9 +10,14 @@
 
 
 class NormalMessage(BaseMessage):
-    """Represent an NACK message"""
-    _type = MessageType.NORMAL
-    _tags: tuple[Who, What, Where]
+    """
+    Represent a Normal message
+
+    Syntax: *who*what*where##
+
+    """
+    _type: Final[MessageType] = MessageType.NORMAL
+    _tags: Final[tuple[Who, What, Where]]
 
     _regex: Pattern[AnyStr] = re.compile(r"^\*[0-9#]+\*[0-9#]*\*[0-9#]*##$")
 

pyown/messages/status.py

@@ -0,0 +1,48 @@
+import re
+from typing import Self, Pattern, AnyStr, Final
+
+from .base import BaseMessage, MessageType
+from ..tags import Who, What, Where
+
+__all__ = [
+    "StatusRequest"
+]
+
+
+class StatusRequest(BaseMessage):
+    """
+    Represent a status request message
+
+    Syntax: *#who*where##
+    """
+    _type: Final[MessageType] = MessageType.STATUS_REQUEST
+    _tags: Final[tuple[Who, Where]]
+
+    _regex: Pattern[AnyStr] = re.compile(r"^\*#[0-9#]+\*[0-9#]*##$")
+
+    def __init__(self, tags: tuple[Who, Where]):
+        self._tags = tags
+
+    @property
+    def who(self) -> Who:
+        return self._tags[0]
+
+    @property
+    def where(self) -> Where:
+        return self._tags[1]
+
+    @property
+    def message(self) -> str:
+        return f"*#{self.who}*{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].removeprefix("#")),
+                Where(tags[1])
+            )
+        )
+

pyown/tags/base.py

@@ -22,7 +22,7 @@ def __init__(self, string: str):
         super().__init__()
 
     @property
-    def value(self) -> int | None:
+    def tag(self) -> int | None:
         """Return the value of the tag without its parameters or prefix"""
         val = self.removeprefix("#")
         if len(val) > 0:
@@ -35,15 +35,10 @@ 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:
+    def tag(self) -> int | None:
         """Return the value of the tag without its parameters or prefix"""
         val = self.split("#")[0]
         if len(val) > 0:

pyown/tags/who.py

@@ -28,6 +28,9 @@ class Who(Tag, enum.Enum):
     DEVICE_DIAGNOSTICS: str = "1013"
     ENERGY_DIAGNOSTICS: str = "1018"
 
+    def __str__(self) -> str:
+        return self.value
+
     @property
     def name(self) -> str:
         return who_map[self]

tests/messages/test_ack.py

@@ -6,6 +6,7 @@ def test_nack():
 
     ack = parse_message(message)
 
+    assert isinstance(ack, ACK)
     assert str(ack) == message
     assert ack.tags == ("#", "1")
     assert ack.type == MessageType.ACK

tests/messages/test_dimension.py

@@ -0,0 +1,91 @@
+from pyown.messages import DimensionRequest, DimensionWriting, DimensionResponse, MessageType, parse_message
+
+
+def test_dimension_request():
+    msg = "*#13**1##"
+
+    dimension_request = parse_message(msg)
+
+    assert isinstance(dimension_request, DimensionRequest)
+    assert str(dimension_request) == msg
+    assert dimension_request.who == "13"
+    assert dimension_request.where == ""
+    assert dimension_request.dimension == "1"
+    assert dimension_request.type == MessageType.DIMENSION_REQUEST
+
+
+def test_dimension_request_with_params():
+    msg = "*#13*7#3*1##"  # not a real message
+
+    dimension_request = parse_message(msg)
+
+    assert isinstance(dimension_request, DimensionRequest)
+    assert str(dimension_request) == msg
+    assert dimension_request.who == "13"
+    assert dimension_request.where == "7#3"
+    assert dimension_request.where.tag == 7
+    assert dimension_request.where.parameters == ["3"]
+    assert dimension_request.dimension == "1"
+    assert dimension_request.type == MessageType.DIMENSION_REQUEST
+
+
+def test_dimension_writing():
+    msg = "*#13**#0*21*10*00*01##"
+
+    dimension_writing = parse_message(msg)
+
+    assert isinstance(dimension_writing, DimensionWriting)
+    assert str(dimension_writing) == msg
+    assert dimension_writing.who == "13"
+    assert dimension_writing.where == ""
+    assert dimension_writing.dimension == "0"
+    assert dimension_writing.values == ("21", "10", "00", "01")
+    assert dimension_writing.type == MessageType.DIMENSION_WRITING
+
+
+# noinspection DuplicatedCode
+def test_dimension_writing_with_params():
+    msg = "*#13*7#3*#0*21*10*00*01##"
+
+    dimension_writing = parse_message(msg)
+
+    assert isinstance(dimension_writing, DimensionWriting)
+    assert str(dimension_writing) == msg
+    assert dimension_writing.who == "13"
+    assert dimension_writing.where == "7#3"
+    assert dimension_writing.where.tag == 7
+    assert dimension_writing.where.parameters == ["3"]
+    assert dimension_writing.dimension == "0"
+    assert dimension_writing.values == ("21", "10", "00", "01")
+    assert dimension_writing.type == MessageType.DIMENSION_WRITING
+
+
+def test_dimension_response():
+    msg = "*#13**1*1*1*1*2012##"
+
+    dimension_response = parse_message(msg)
+
+    assert isinstance(dimension_response, DimensionResponse)
+    assert str(dimension_response) == msg
+    assert dimension_response.who == "13"
+    assert dimension_response.where == ""
+    assert dimension_response.dimension == "1"
+    assert dimension_response.values == ("1", "1", "1", "2012")
+    assert dimension_response.type == MessageType.DIMENSION_RESPONSE
+
+
+# noinspection DuplicatedCode
+def test_dimension_response_with_params():
+    msg = "*#13*7#3*1*1*1*1*2012##"
+
+    dimension_response = parse_message(msg)
+
+    assert isinstance(dimension_response, DimensionResponse)
+    assert str(dimension_response) == msg
+    assert dimension_response.who == "13"
+    assert dimension_response.where == "7#3"
+    assert dimension_response.where.tag == 7
+    assert dimension_response.where.parameters == ["3"]
+    assert dimension_response.dimension == "1"
+    assert dimension_response.values == ("1", "1", "1", "2012")
+    assert dimension_response.type == MessageType.DIMENSION_RESPONSE

tests/messages/test_nack.py

@@ -6,6 +6,7 @@ def test_nack():
 
     nack = parse_message(message)
 
+    assert isinstance(nack, NACK)
     assert str(nack) == message
     assert nack.tags == ("#", "0")
     assert nack.type == MessageType.NACK