Replace fixed decodings, like 'latin-1', with a chardet based function.

pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
     "toml<1.0.0,>=0.10.2",
     "python-can>=4.0.0",
     "uptime>=3.0.1",
+    "chardet>=5.2.0",
 ]
 name = "pyxcp"
 version = "0.21.2"

pyxcp/examples/xcphello.py

@@ -5,6 +5,7 @@
 from pprint import pprint
 
 from pyxcp.cmdline import ArgumentParser
+from pyxcp.utils import decode_bytes
 
 daq_info = False
 
@@ -49,7 +50,7 @@ def callout(master, args):
         for idx in range(dqp.maxEventChannel):
             evt = x.getDaqEventInfo(idx)
             length = evt.eventChannelNameLength
-            name = x.pull(length).decode("utf-8")
+            name = decode_bytes(x.pull(length))
             dq = "DAQ" if evt.daqEventProperties.daq else ""
             st = "STIM" if evt.daqEventProperties.stim else ""
             dq_st = dq + " " + st

pyxcp/examples/xcphello_recorder.py

@@ -7,6 +7,7 @@
 from pyxcp.cmdline import ArgumentParser
 from pyxcp.recorder import XcpLogFileReader
 from pyxcp.transport import FrameRecorderPolicy
+from pyxcp.utils import decode_bytes
 
 daq_info = False
 
@@ -55,7 +56,7 @@ def callout(master, args):
         for idx in range(dqp.maxEventChannel):
             evt = x.getDaqEventInfo(idx)
             length = evt.eventChannelNameLength
-            name = x.pull(length).decode("utf-8")
+            name = decode_bytes(x.pull(length))
             dq = "DAQ" if evt.daqEventProperties.daq else ""
             st = "STIM" if evt.daqEventProperties.stim else ""
             dq_st = dq + " " + st

pyxcp/master/master.py

@@ -35,6 +35,7 @@
 from pyxcp.master.errorhandler import disable_error_handling
 from pyxcp.master.errorhandler import wrapped
 from pyxcp.transport.base import createTransport
+from pyxcp.utils import decode_bytes
 from pyxcp.utils import delay
 from pyxcp.utils import SHORT_SLEEP
 
@@ -1676,8 +1677,10 @@ def getDaqInfo(self):
         for ecn in range(dpi.maxEventChannel):
             eci = self.getDaqEventInfo(ecn)
             name = self.fetch(eci.eventChannelNameLength)
+            if name:
+                name = decode_bytes(name)
             channel = {
-                "name": name.decode("latin-1"),
+                "name": name,
                 "priority": eci["eventChannelPriority"],
                 "unit": eci["eventChannelTimeUnit"],
                 "cycle": eci["eventChannelTimeCycle"],
@@ -1806,7 +1809,7 @@ def identifier(self, id_value: int) -> str:
             value = bytes(gid.identification or b"")
         else:
             value = self.fetch(gid.length)
-        return value.decode("utf-8")
+        return decode_bytes(value)
 
     def id_scanner(self, scan_ranges: Optional[Collection[Collection[int]]] = None) -> Dict[str, str]:
         """Scan for available standard identification types (GET_ID).

pyxcp/utils.py

@@ -6,6 +6,8 @@
 from time import perf_counter
 from time import time
 
+import chardet
+
 
 def hexDump(arr):
     if isinstance(arr, (bytes, bytearray)):
@@ -48,6 +50,15 @@ def getPythonVersion():
     return sys.version_info
 
 
+def decode_bytes(byte_str: bytes) -> str:
+    """Decode bytes with the help of ``chardet"""
+    encoding = chardet.detect(byte_str).get("encoding")
+    if not encoding:
+        return byte_str.decode("ascii", "ignore")
+    else:
+        return byte_str.decode(encoding)
+
+
 PYTHON_VERSION = getPythonVersion()
 SHORT_SLEEP = 0.0005
 
@@ -58,4 +69,3 @@ def delay(amount: float):
     start = perf_counter()
     while perf_counter() < start + amount:
         pass
-

setup.py

@@ -41,7 +41,6 @@
 EXT_NAMES = ["rekorder"]
 
 if has_pybind11:
-
     ext_modules = [
         Pybind11Extension(
             EXT_NAMES[0],
@@ -55,7 +54,7 @@
 else:
     ext_modules = []
 
-install_reqs = ["pybind11", "pyusb", "construct >= 2.9.0", "mako", "pyserial", "toml", "python-can", "uptime"]
+install_reqs = ["pybind11", "pyusb", "construct >= 2.9.0", "mako", "pyserial", "toml", "python-can", "uptime", "chardet"]
 
 
 class AsamKeyDllAutogen(setuptools.Command):