Implement CRC as class with byte-wise update.

diff --git a/controlpi_plugins/modbus.py b/controlpi_plugins/modbus.py
index dad43c780dd6d55f634e1e2ec44ac45f041ffaba..f9b0f41ed04536fb273bcb147c9fcba945af9d0c 100644 (file)
--- a/controlpi_plugins/modbus.py
+++ b/controlpi_plugins/modbus.py
@@ -12,6 +12,94 @@ from controlpi import BasePlugin, Message, MessageTemplate
 from typing import Dict
 
 
+class CRC:
+    """Calculate CRC for message.
+
+    The CRC as specified for Modbus is computed using a precomputed table
+    for its inner loop.
+
+    We define 12 test messages with their expected results from the manual
+    of the Hitachi PJ series Modbus devices:
+    >>> tests = [(bytes.fromhex('08 01 0006 0006'), bytes.fromhex('5C90')),
+    ...          (bytes.fromhex('08 01 01 17'), bytes.fromhex('121A')),
+    ...          (bytes.fromhex('05 03 03E8 0003'), bytes.fromhex('843F')),
+    ...          (bytes.fromhex('05 03 06 0007 0000 1770'),
+    ...           bytes.fromhex('A861')),
+    ...          (bytes.fromhex('0A 05 0000 FF00'), bytes.fromhex('8D41')),
+    ...          (bytes.fromhex('01 06 2F4D 1388'), bytes.fromhex('1C5F')),
+    ...          (bytes.fromhex('05 0F 0006 0006 02 1700'),
+    ...           bytes.fromhex('DB3E')),
+    ...          (bytes.fromhex('05 0F 0006 0006'), bytes.fromhex('344C')),
+    ...          (bytes.fromhex('01 10 2B01 0002 04 0004 93E0'),
+    ...           bytes.fromhex('F42B')),  # error in manual
+    ...          (bytes.fromhex('01 10 2B01 0002'),
+    ...           bytes.fromhex('19EC')),  # error in manual
+    ...          (bytes.fromhex('01 17 2710 0002 2AF8 0002 04 0000 1388'),
+    ...           bytes.fromhex('964D')),  # error in manual
+    ...          (bytes.fromhex('01 17 04 0000 1388'),
+    ...           bytes.fromhex('F471'))]
+
+    Now, we test the crc function against this list:
+    >>> crc_zero = bytes.fromhex('0000')
+    >>> for (message, expected) in tests:
+    ...     crc = CRC(message)
+    ...     assert bytes(crc) == expected
+    ...     assert not bool(crc)
+    ...     for byte in bytes(crc):
+    ...         crc.update(byte)
+    ...     assert bool(crc)
+    """
+
+    def __init__(self, message: bytes = b'') -> None:
+        self._crc = 0xFFFF
+        self._message = b''
+        for byte in message:
+            self.update(byte)
+
+    def update(self, byte: int) -> None:
+        """Update CRC with one additional byte."""
+        if not self._precomputed:
+            print("Precomputing CRC values.")
+            self.precompute()
+        self._message += bytes([byte])
+        self._crc ^= byte
+        key = self._crc & 0xFF
+        self._crc >>= 8
+        self._crc ^= self._precomputed[key]
+
+    _precomputed: Dict[int, int] = {}
+
+    @classmethod
+    def precompute(cls) -> None:
+        """Precompute inner loop of CRC algorithm."""
+        for key in range(256):
+            value = key
+            for bit in range(8):
+                lsb = value & 1
+                value >>= 1
+                if lsb:
+                    value ^= 0xA001
+            cls._precomputed[key] = value
+
+    def __repr__(self) -> str:
+        return f"CRC(bytes.fromhex({self._message.hex().upper()}))"
+
+    def __str__(self) -> str:
+        return f"crc({self._message.hex().upper()}) = 0x{self._crc:04X}"
+
+    def __bool__(self) -> bool:
+        return self._crc == 0
+
+    def __int__(self) -> int:
+        return self._crc
+
+    def __bytes__(self) -> bytes:
+        return bytes((self._crc & 0xFF, self._crc >> 8))
+
+
+CRC.precompute()
+
+
 def crc(message: bytes) -> bytes:
     """Calculate CRC for message.