Adapt init consume str

Author: Breina

custom_components/dmx/client/__init__.py

@@ -448,10 +448,44 @@ def _append_str(packet: bytearray, text: str, length: int):
         packet.extend(map(ord, padded_text))
 
     @staticmethod
-    def _consume_str(packet: bytearray, index: int, length: int) -> (str, int):
-        str_bytes = str(packet[index:index + length - 1], "ASCII")
-        string = str_bytes.split('\0')[0]
-        return string, index + length
+    def _consume_str(packet: bytearray, index: int, length: int) -> (Optional[str], int):
+        decoded_str: Optional[str] = None
+        raw_string_from_packet = packet[index:index + length]
+
+        # assume the data is ascii
+        try:
+            # if there is a NUL character in the bytearry, it terminates earlier
+            nul_terminator = re.compile(b"([^\\x00]+)")
+
+            terminated_string = nul_terminator.match(raw_string_from_packet).group(1)  \
+                if nul_terminator.match(raw_string_from_packet) else raw_string_from_packet
+
+            # remove every other control character
+            sanitized_str = re.sub(b"[^\x00-\x7F]",b"", terminated_string)
+
+            # decode
+            decoded_str = sanitized_str.decode('ascii')
+
+        except UnicodeDecodeError:
+            # data not ascii, try to use the decoding shotgun
+            decoded_str = ArtBase._decode_bytes(raw_string_from_packet)
+
+        # check if decoding has failed
+        if decoded_str is None:
+            log.error("Unable to convert bytes to string: {raw_hex}".format(raw_hex=bytes(raw_string_from_packet).hex()))
+
+        return decoded_str, index + length
+
+    @staticmethod
+    def _decode_bytes(byte_str: bytearray) -> Optional[str]:
+        for encoding in ArtBase.__ENCODINGS__:
+            try:
+                decoded_str = byte_str.decode(encoding)
+                sanitized_str = str(decoded_str).strip().strip('\x00')
+                log.debug(f"decoded as {encoding}: {sanitized_str}")
+                return sanitized_str
+            except UnicodeDecodeError as e:
+                log.warning(f"failed ({e.reason}) to decode as {encoding}: {bytes(byte_str).hex()}")
 
     @staticmethod
     def peek_opcode(packet: bytearray) -> OpCode | None: