Bugfix for checksum calculations

Author: newAM

CHANGELOG.md

@@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html),
 as of version 2.1.1.
 
+## [3.0.1] - 2022-05-11
+### Fixed
+- Fixed checksum calculation on Linux.
+
 ## [3.0.0] - 2022-04-09
 ### Added
 - Added support for python 3.10.
@@ -70,7 +74,8 @@ as of version 2.1.1.
 - Added a changelog.
 
 
-[Unreleased]: https://github.com/newAM/monitorcontrol/compare/3.0.0...HEAD
+[Unreleased]: https://github.com/newAM/monitorcontrol/compare/3.0.1...HEAD
+[3.0.1]: https://github.com/newAM/monitorcontrol/compare/3.0.0...3.0.1
 [3.0.0]: https://github.com/newAM/monitorcontrol/compare/2.5.1...3.0.0
 [2.5.1]: https://github.com/newAM/monitorcontrol/compare/2.5.0...2.5.1
 [2.5.0]: https://github.com/newAM/monitorcontrol/compare/2.4.2...2.5.0

flake.lock

@@ -38,7 +38,7 @@ class LinuxVCP(VCP):
 
     # addresses
     DDCCI_ADDR = 0x37  # DDC-CI command address on the I2C bus
-    HOST_ADDRESS = 0x50  # virtual I2C slave address of the host
+    HOST_ADDRESS = 0x51  # virtual I2C slave address of the host
     I2C_SLAVE = 0x0703  # I2C bus slave address
 
     GET_VCP_RESULT_CODES = {
@@ -110,9 +110,12 @@ def set_vcp_feature(self, code: int, value: int):
         # add headers and footers
         data.insert(0, (len(data) | self.PROTOCOL_FLAG))
         data.insert(0, self.HOST_ADDRESS)
-        data.append(self.get_checksum(data))
+        data.append(
+            self.get_checksum(bytearray([self.DDCCI_ADDR << 1]) + data)
+        )
 
         # write data
+        self.logger.debug("data=" + " ".join([f"{x:02X}" for x in data]))
         self.write_bytes(data)
 
         # store time of last set VCP
@@ -141,24 +144,26 @@ def get_vcp_feature(self, code: int) -> Tuple[int, int]:
         # add headers and footers
         data.insert(0, (len(data) | self.PROTOCOL_FLAG))
         data.insert(0, self.HOST_ADDRESS)
-        data.append(self.get_checksum(data))
-        self.logger.debug(f"data={data}")
+        data.append(
+            self.get_checksum(bytearray([self.DDCCI_ADDR << 1]) + data)
+        )
 
         # write data
+        self.logger.debug("data=" + " ".join([f"{x:02X}" for x in data]))
         self.write_bytes(data)
 
         time.sleep(self.GET_VCP_TIMEOUT)
 
         # read the data
         header = self.read_bytes(self.GET_VCP_HEADER_LENGTH)
-        self.logger.debug(f"header={header}")
-        source, length = struct.unpack("BB", header)
+        self.logger.debug("header=" + " ".join([f"{x:02X}" for x in header]))
+        source, length = struct.unpack("=BB", header)
         length &= ~self.PROTOCOL_FLAG  # clear protocol flag
         payload = self.read_bytes(length + 1)
-        self.logger.debug(f"payload={payload}")
+        self.logger.debug("payload=" + " ".join([f"{x:02X}" for x in payload]))
 
         # check checksum
-        payload, checksum = struct.unpack(f"{length}sB", payload)
+        payload, checksum = struct.unpack(f"={length}sB", payload)
         calculated_checksum = self.get_checksum(header + payload)
         checksum_xor = checksum ^ calculated_checksum
         if checksum_xor:
@@ -245,11 +250,15 @@ def get_vcp_capabilities(self):
 
             # read the data
             header = self.read_bytes(self.GET_VCP_HEADER_LENGTH)
-            self.logger.debug(f"response header={header}")
+            self.logger.debug(
+                "header=" + " ".join([f"{x:02X}" for x in header])
+            )
             source, length = struct.unpack("BB", header)
             length &= ~self.PROTOCOL_FLAG  # clear protocol flag
             payload = self.read_bytes(length + 1)
-            self.logger.debug(f"payload={payload}")
+            self.logger.debug(
+                "payload=" + " ".join([f"{x:02X}" for x in payload])
+            )
 
             # check if length is valid
             if length < 3 or length > 35:
@@ -298,19 +307,19 @@ def get_vcp_capabilities(self):
 
         return caps_str
 
-    def get_checksum(self, data: List, prime: bool = False) -> int:
+    @staticmethod
+    def get_checksum(data: bytearray) -> int:
         """
         Computes the checksum for a set of data, with the option to
         use the virtual host address (per the DDC-CI specification).
 
         Args:
-            data: data array to transmit
-            prime: compute checksum using the 0x50 virtual host address
+            data: Data array to transmit.
 
         Returns:
-            checksum for the data
+            Checksum for the data.
         """
-        checksum = self.HOST_ADDRESS
+        checksum = 0x00
         for data_byte in data:
             checksum ^= data_byte
         return checksum

monitorcontrol/vcp/vcp_linux.py

@@ -5,7 +5,7 @@ target-version = ["py36", "py37", "py38"]
 [tool.poetry]
 name = "monitorcontrol"
 description = "Monitor controls using MCCS over DDC-CI."
-version = "3.0.0"
+version = "3.0.1"
 authors = ["Alex Martens <alex@thinglab.org>"]
 license = "MIT"
 readme = "README.rst"

poetry.lock

@@ -0,0 +1,20 @@
+import pytest
+from monitorcontrol.vcp.vcp_linux import LinuxVCP
+
+
+@pytest.mark.parametrize(
+    "data, checksum",
+    [
+        (bytearray([0x6E, 0x51, 0x82, 0x01, 0x10]), 0xAC),
+        (bytearray([0xF0, 0xF1, 0x81, 0xB1]), 0x31),
+        (bytearray([0x6E, 0xF1, 0x81, 0xB1]), 0xAF),
+    ],
+)
+def test_get_checksum(data: bytearray, checksum: int):
+    computed = LinuxVCP.get_checksum(data)
+    xor = checksum ^ computed
+    assert computed == checksum, (
+        f"computed=0x{computed:02X} 0b{computed:08b} "
+        f"checksum=0x{checksum:02X} 0b{checksum:08b} "
+        f"xor=0x{xor:02X} 0b{xor:08b}"
+    )