Add support for DPT 16.001 (#910)

farmio · web-flow · commit 4bc15ab6ba4e · 2022-04-05T16:24:14.000+02:00
diff --git a/changelog.md b/changelog.md
@@ -6,10 +6,12 @@
 
 - Add support for SearchRequestExtended to find interfaces that allow IP Secure
 - Use XKNX `state_updater` argument to set default method for StateUpdater. StateUpdater is always started - Device / RemoteValue can always opt in to use it, even if default is `False`.
+- Add support for DPT 16.001 (DPT_String_8859_1) as `DPTLatin1` with value_type "latin_1".
 
 ### Bug fixes
 
 - Stop SecureSession keepalive_task when session is stopped (and don't restart it from sending STATUS_CLOSE)
+- Fix encoding invalid characters for DPTString (value_type "string")
 
 ## 0.20.0 IP Secure 2022-03-29
 
diff --git a/test/dpt_tests/dpt_string_test.py b/test/dpt_tests/dpt_string_test.py
@@ -1,167 +1,91 @@
 """Unit test for KNX string object."""
 import pytest
 
-from xknx.dpt import DPTString
+from xknx.dpt import DPTLatin1, DPTString
 from xknx.exceptions import ConversionError
 
 
 class TestDPTString:
-    """Test class for KNX float object."""
+    """Test class for KNX ASCII string object."""
 
-    def test_value_from_documentation(self):
-        """Test parsing and streaming Example from documentation."""
-        raw = (
-            0x4B,
-            0x4E,
-            0x58,
-            0x20,
-            0x69,
-            0x73,
-            0x20,
-            0x4F,
-            0x4B,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-        )
-        string = "KNX is OK"
-        assert DPTString.to_knx(string) == raw
-        assert DPTString.from_knx(raw) == string
-
-    def test_value_empty_string(self):
-        """Test parsing and streaming empty string."""
-        raw = (
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-        )
-        string = ""
-        assert DPTString.to_knx(string) == raw
-        assert DPTString.from_knx(raw) == string
+    @pytest.mark.parametrize(
+        "string,raw",
+        [
+            (
+                "KNX is OK",
+                (75, 78, 88, 32, 105, 115, 32, 79, 75, 0, 0, 0, 0, 0),
+            ),
+            (
+                "",
+                (0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0),
+            ),
+            (
+                "AbCdEfGhIjKlMn",
+                (65, 98, 67, 100, 69, 102, 71, 104, 73, 106, 75, 108, 77, 110),
+            ),
+            (
+                ".,:;-_!?$@&#%/",
+                (46, 44, 58, 59, 45, 95, 33, 63, 36, 64, 38, 35, 37, 47),
+            ),
+        ],
+    )
+    @pytest.mark.parametrize("test_dpt", [DPTString, DPTLatin1])
+    def test_values(self, string, raw, test_dpt):
+        """Test parsing and streaming strings."""
+        assert test_dpt.to_knx(string) == raw
+        assert test_dpt.from_knx(raw) == string
 
-    def test_value_max_string(self):
-        """Test parsing and streaming large string."""
-        raw = (
-            0x41,
-            0x41,
-            0x41,
-            0x41,
-            0x41,
-            0x42,
-            0x42,
-            0x42,
-            0x42,
-            0x42,
-            0x43,
-            0x43,
-            0x43,
-            0x43,
-        )
-        string = "AAAAABBBBBCCCC"
-        assert DPTString.to_knx(string) == raw
-        assert DPTString.from_knx(raw) == string
-
-    def test_value_special_chars(self):
-        """Test parsing and streaming string with special chars."""
-        raw = (
-            0x48,
-            0x65,
-            0x79,
-            0x21,
-            0x3F,
-            0x24,
-            0x20,
-            0xC4,
-            0xD6,
-            0xDC,
-            0xE4,
-            0xF6,
-            0xFC,
-            0xDF,
-        )
-        string = "Hey!?$ ÄÖÜäöüß"
-        assert DPTString.to_knx(string) == raw
-        assert DPTString.from_knx(raw) == string
-
-    def test_to_knx_invalid_chars(self):
-        """Test streaming string with invalid chars."""
-        raw = (
-            0x4D,
-            0x61,
-            0x74,
-            0x6F,
-            0x75,
-            0x3F,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-        )
-        string = "Matouš"
-        knx_string = "Matou?"
+    @pytest.mark.parametrize(
+        "string,knx_string,raw",
+        [
+            (
+                "Matouš",
+                "Matou?",
+                (77, 97, 116, 111, 117, 63, 0, 0, 0, 0, 0, 0, 0, 0),
+            ),
+            (
+                "Gänsefüßchen",
+                "G?nsef??chen",
+                (71, 63, 110, 115, 101, 102, 63, 63, 99, 104, 101, 110, 0, 0),
+            ),
+        ],
+    )
+    def test_to_knx_ascii_invalid_chars(self, string, knx_string, raw):
+        """Test streaming ASCII string with invalid chars."""
         assert DPTString.to_knx(string) == raw
         assert DPTString.from_knx(raw) == knx_string
 
+    @pytest.mark.parametrize(
+        "string,raw",
+        [
+            (
+                "Gänsefüßchen",
+                (71, 228, 110, 115, 101, 102, 252, 223, 99, 104, 101, 110, 0, 0),
+            ),
+            (
+                "àáâãåæçèéêëìíî",
+                (224, 225, 226, 227, 229, 230, 231, 232, 233, 234, 235, 236, 237, 238),
+            ),
+        ],
+    )
+    def test_to_knx_latin_1(self, string, raw):
+        """Test streaming Latin-1 strings."""
+        assert DPTLatin1.to_knx(string) == raw
+        assert DPTLatin1.from_knx(raw) == string
+
     def test_to_knx_too_long(self):
         """Test serializing DPTString to KNX with wrong value (to long)."""
         with pytest.raises(ConversionError):
             DPTString.to_knx("AAAAABBBBBCCCCx")
 
-    def test_from_knx_wrong_parameter_too_large(self):
-        """Test parsing of KNX string with too many elements."""
-        raw = (
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-        )
-        with pytest.raises(ConversionError):
-            DPTString.from_knx(raw)
-
-    def test_from_knx_wrong_parameter_too_small(self):
-        """Test parsing of KNX string with too less elements."""
-        raw = (
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-            0x00,
-        )
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            ((0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0),),
+            ((0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0),),
+        ],
+    )
+    def test_from_knx_wrong_parameter_length(self, raw):
+        """Test parsing of KNX string with wrong elements length."""
         with pytest.raises(ConversionError):
             DPTString.from_knx(raw)
diff --git a/xknx/dpt/__init__.py b/xknx/dpt/__init__.py
@@ -172,7 +172,7 @@
 from .dpt_datetime import DPTDateTime
 from .dpt_hvac_mode import DPTControllerStatus, DPTHVACContrMode, DPTHVACMode
 from .dpt_scaling import DPTAngle, DPTScaling
-from .dpt_string import DPTString
+from .dpt_string import DPTLatin1, DPTString
 from .dpt_time import DPTTime
 
 __all__ = [
@@ -249,6 +249,7 @@
     "DPTHVACMode",
     "DPTImpedance",
     "DPTKelvinPerPercent",
+    "DPTLatin1",
     "DPTLength",
     "DPTLengthMm",
     "DPTLightQuantity",
diff --git a/xknx/dpt/dpt_string.py b/xknx/dpt/dpt_string.py
@@ -8,7 +8,7 @@
 
 class DPTString(DPTBase):
     """
-    Abstraction for KNX 14 Octet ASCII String.
+    Abstraction for KNX 14 Octet ASCII string.
 
     DPT 16.000
     """
@@ -19,15 +19,15 @@ class DPTString(DPTBase):
     value_type = "string"
     unit = ""
 
+    _encoding = "ascii"
+
     @classmethod
     def from_knx(cls, raw: tuple[int, ...]) -> str:
         """Parse/deserialize from KNX/IP raw data."""
         cls.test_bytesarray(raw)
-        value = ""
-        for byte in raw:
-            if byte != 0x00:
-                value += chr(byte)
-        return value
+        return bytes(byte for byte in raw if byte != 0x00).decode(
+            cls._encoding, errors="replace"
+        )
 
     @classmethod
     def to_knx(cls, value: str) -> tuple[int, ...]:
@@ -36,15 +36,26 @@ def to_knx(cls, value: str) -> tuple[int, ...]:
             knx_value = str(value)
             if not cls._test_boundaries(knx_value):
                 raise ValueError
-            raw = [ord(character) for character in knx_value]
-            raw.extend([0] * (cls.payload_length - len(raw)))
-            # replace invalid characters with question marks
-            # bytes(knx_value, 'ascii') would raise UnicodeEncodeError
-            return tuple(map(lambda char: char if char <= 0xFF else ord("?"), raw))
         except ValueError:
             raise ConversionError(f"Could not serialize {cls.__name__}", value=value)
+        # replace invalid characters with question marks
+        raw_bytes = knx_value.encode(cls._encoding, errors="replace")
+        padding = bytes(cls.payload_length - len(raw_bytes))
+        return tuple(raw_bytes + padding)
 
     @classmethod
     def _test_boundaries(cls, value: str) -> bool:
         """Test if value is within defined range for this object."""
         return len(value) <= cls.payload_length
+
+
+class DPTLatin1(DPTString):
+    """
+    Abstraction for KNX 14 Octet Latin-1 (ISO 8859-1) string.
+
+    DPT 16.001
+    """
+
+    dpt_sub_number = 1
+    value_type = "latin_1"
+    _encoding = "latin_1"
