Systematize Memory Configuration bitfield constants (Make them fully cross-checkable and add related tests).

Author: Poikilos

openlcb/memoryservice.py

@@ -53,14 +53,14 @@ class MCOp(Enum):
     """Byte 1 & 0b11111100 values (assumes byte 0 is 0x20)"""
     Read_Command = 0x40  # 01000000
     Read_Reply = 0x50    # 01010000
-    Read_Reply_Failure = 0x58  # 01011000
+    Read_Reply_Failure = 0x58  # 01011000; See OP_FAILURE_BYTES
     Read_Stream_Command = 0x60  # 01100000
     Read_Stream_Reply = 0x70  # 01110000
-    Read_Stream_Reply_Failure = 0x78  # 01111000
+    Read_Stream_Reply_Failure = 0x78  # 01111000; See OP_FAILURE_BYTES
     Write_Command = 0x00
     Write_Reply = 0x10  # 00010000
     # or                  00010001 (0x11) and so on
-    Write_Reply_Failure = 0x18
+    Write_Reply_Failure = 0x18  # 00011000; See OP_FAILURE_BYTES
     Write_Under_Mask_Command = 0x08  # 00001000
     Write_Stream_Command = 0x20  # 01000000
     Write_Stream_Reply = 0x30  # 00110000
@@ -109,30 +109,42 @@ class MCOpMasks:
     #    they can be broken down with
     #    sub-checks even if Default is used
     #    (See any set with more than one entry
-    #    in MODE_BYTES):
+    #    in TWO_BIT_PARAMS):
     # Get_Configuration_Options = 0x11111110
     # Get_Address_Space_Info = 0x11111110
 
 
-MODE_BYTES = {
-    # order determines meaning for lists (See )
-    MCOp.Read_Command.value: {0x40, 0x41, 0x42, 0x43},  # pool
-    MCOp.Read_Reply.value: {0x50, 0x51, 0x52, 0x53},
-    MCOp.Read_Stream_Command.value: {0x60, 0x61, 0x62, 0x63},  # pool
-    MCOp.Read_Stream_Reply.value: {0x70, 0x71, 0x72, 0x73},  # TODO
+TWO_BIT_PARAMS = {
+    # The combined values in these lists include all 6-bit
+    # operation groups *except* those in OP_FAILURE_BYTES
+    # order determines meaning for lists:
+    # - first indicates custom space follows (in later byte)
+    # - Others are standard spaces
+    #    See MemorySpaceIndex enum for meaning of last 2 bits
+    #    (and Configuration Description Information Standard).
+    #   - does *not* apply to *sets* below!
+    # *lists* (last 2 bits are MemorySpaceIndex.fromNumber param)
+    MCOp.Read_Command.value: [0x40, 0x41, 0x42, 0x43],  # pool
+    MCOp.Read_Reply.value: [0x50, 0x51, 0x52, 0x53],
+    MCOp.Read_Stream_Command.value: [0x60, 0x61, 0x62, 0x63],  # pool
+    MCOp.Read_Stream_Reply.value: [0x70, 0x71, 0x72, 0x73],  # TODO
     MCOp.Write_Command.value: [0x00, 0x01, 0x02, 0x03],  # pool
-    MCOp.Write_Reply.value: {0x10, 0x11, 0x12, 0x13},
-    MCOp.Write_Under_Mask_Command.value: {0x08, 0x09, 0x0A, 0x0B},  # pool
-    MCOp.Write_Stream_Command.value: {0x20, 0x21, 0x22, 0x23},
-    MCOp.Write_Stream_Reply.value: {0x30, 0x31, 0x32, 0x33},  # TODO
+    MCOp.Write_Reply.value: [0x10, 0x11, 0x12, 0x13],
+    MCOp.Write_Under_Mask_Command.value: [0x08, 0x09, 0x0A, 0x0B],  # pool
+    MCOp.Write_Stream_Command.value: [0x20, 0x21, 0x22, 0x23],
+    MCOp.Write_Stream_Reply.value: [0x30, 0x31, 0x32, 0x33],  # TODO
+    # *sets* (no standard meaning to last 2 bits)
     MCOp.Get_Configuration_Options_Command.value: {0x80, },
     MCOp.Get_Configuration_Options_Reply.value: {0x82, },
     MCOp.Get_Address_Space_Info_Command.value: {
         MCOp.Get_Address_Space_Info_Command.value,
         MCOp.Get_Address_Space_Info_Reply.value,
         MCOp.Get_Address_Space_Info_Reply_Command.value,
     },
-    MCOp.Lock_or_Reserve_Command.value: {0x88, },
+    MCOp.Lock_or_Reserve_Command.value: {
+        MCOp.Lock_or_Reserve_Command.value,
+        MCOp.Lock_or_Reserve_Reply.value,
+    },
     MCOp.Get_Unique_ID_Command.value: {0x8C, },
     MCOp.Get_Unique_ID_Reply.value: {0x8D, },
     MCOp.Unfreeze_Command.value: {0xA1, 0xA0},  # unfreeze, freeze respectively
@@ -143,11 +155,21 @@ class MCOpMasks:
     }
 }
 
-MODE_ERROR_BYTES = {
-    MCOp.Read_Reply.value: {0x58, 0x59, 0x5A, 0x5B},
-    MCOp.Read_Stream_Reply.value: {0x78, 0x79, 0x7A, 0x7B},
-    MCOp.Write_Reply.value: {0x18, 0x19, 0x1A, 0x1B},
-    MCOp.Write_Stream_Reply.value: {0x38, 0x39, 0x3A, 0x3B},
+OP_FAILURE_BYTES = {
+    # *lists* (last 2 bits are standard memory space):
+    MCOp.Read_Reply.value: [0x58, 0x59, 0x5A, 0x5B],
+    MCOp.Read_Stream_Reply.value: [0x78, 0x79, 0x7A, 0x7B],
+    MCOp.Write_Reply.value: [0x18, 0x19, 0x1A, 0x1B],
+    MCOp.Write_Stream_Reply.value: [0x38, 0x39, 0x3A, 0x3B],
+    # *sets* (last 2 bits have no special meaning):
+    MCOp.Read_Reply_Failure.value: {
+        MCOp.Read_Reply_Failure.value},
+    MCOp.Write_Reply_Failure.value: {
+        MCOp.Write_Reply_Failure.value},
+    MCOp.Read_Stream_Reply_Failure.value: {
+        MCOp.Read_Stream_Reply_Failure.value},
+    MCOp.Write_Stream_Reply_Failure.value: {
+        MCOp.Write_Stream_Reply_Failure.value},
 }
 
 
@@ -280,7 +302,7 @@ def parseReplyDatagram(memo: Union[MemoryReadMemo, MemoryWriteMemo],
         pass
         # 0x08 (0b00001000) is error bit
         # mode = None
-        # for k, values in MODE_ERROR_BYTES.items():
+        # for k, values in OP_FAILURE_BYTES.items():
         #     if dmemo.data[1] in values:
         #         mode = k
         #         break
@@ -430,6 +452,9 @@ def datagramReceivedListener(self, dmemo: DatagramReadMemo) -> bool:
         if dmemo.data[1] in (0x50, 0x51, 0x52, 0x53, 0x58, 0x59, 0x5A, 0x5B):
             assert mcOp is MCOp.Read_Reply, \
                 "self-test failed (bad constant(s))"
+            assert (dmemo.data[1] in TWO_BIT_PARAMS[mcOp.value]
+                    or dmemo.data[1] in OP_FAILURE_BYTES[mcOp.value]), \
+                "self-test failed (bad constant(s))"
             # MCOp.Read_Reply
             # read or read-error reply
 

tests/test_memoryservice.py

@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+from collections import OrderedDict
 import os
 import struct
 import sys
@@ -30,6 +31,9 @@
 from openlcb.mti import MTI  # noqa: E402
 from openlcb.message import Message  # noqa: E402
 from openlcb.memoryservice import (  # noqa: E402
+    OP_FAILURE_BYTES,
+    TWO_BIT_PARAMS,
+    MCOp,
     MemoryReadMemo,
     MemoryWriteMemo,
     MemoryService,
@@ -183,6 +187,70 @@ def testMultipleRead(self):
         self.assertEqual(len(LinkMockLayer.sentMessages), 5)  # read reply datagram reply sent and next datagram sent  # noqa: E501
         self.assertEqual(len(self.returnedMemoryReadMemo), 2)  # memory read returned  # noqa: E501
 
+    def testProtocolGroupUniqueness(self):
+        """Ensure each 6-bit field is unique"""
+        opCounts = OrderedDict()
+
+        def incrementKey(key, counts):
+            if key not in counts:
+                counts[key] = 1
+                return
+            raise AssertionError(
+                f"Bitfield {hex(key)} applies to more than one parent op")
+            # counts[key] += 1
+
+        for op in MCOp:
+            incrementKey(op.value, opCounts)
+            # Ensure that 6-bit fields are systematized (correct constants)
+            if ((op.value in OP_FAILURE_BYTES)
+                    and (len(OP_FAILURE_BYTES[op.value]) == 1)):
+                assert op.value not in TWO_BIT_PARAMS, \
+                    (f"{op} last 2 bits are not significant"
+                     " so it should not be in TWO_BIT_PARAMS dict")
+            # elif op.value & 0b11111100 in OP_FAILURE_BYTES:
+            #     for _, failureBytes in \
+            #             OP_FAILURE_BYTES[op.value & 0b11111100].items():
+            #         for failureByte in failureBytes:
+            #             assert failureByte not in TWO_BIT_PARAMS[op.value], \
+            #                 (f"Failure bytes should not be in two-bit"
+            #                  f" params for {op}")
+            #     assert isinstance(OP_FAILURE_BYTES[op.value], list), \
+            #         (f"If {op} use last 2 bits as index,"
+            #          " it should be recorded as a list in OP_FAILURE_BYTES")
+            # elif op.value in OP_FAILURE_BYTES:
+            #     assert isinstance(OP_FAILURE_BYTES[op.value], set), \
+            #         (f"If {op} doesn't use last 2 bits as index,"
+            #          " it should be recorded as a set")
+            elif op.value in TWO_BIT_PARAMS:
+                if op.value in OP_FAILURE_BYTES:
+                    assert isinstance(OP_FAILURE_BYTES[op.value], list), \
+                        (f"If {op} use last 2 bits as index, it"
+                         " should be recorded as a list in OP_FAILURE_BYTES")
+                # elif "Reply" in str(op):
+                #     # commented since not a real problem
+                #     # MCOp.Get_Configuration_Options_Reply doesn't have a
+                #     #    corresponding error for the same 6-bit op field.
+                #     raise AssertionError(
+                #         (f"{op} is a reply but does not have an error reply."
+                #          " Does this follow the standard? If so,"
+                #          "remove this assertion."))
+            else:
+                # assert op.value in TWO_BIT_PARAMS, \
+                #     f"There is no list of two-bit params for {op}"
+                # ^ Incorrect assertion since some share 1st 6 bits and
+                #   don't have params. See below instead.
+                parentValue = op.value & 0b11111100
+                assert (op.value in TWO_BIT_PARAMS
+                        or ((parentValue in TWO_BIT_PARAMS)
+                            and (op.value in TWO_BIT_PARAMS[parentValue]))
+                        ), \
+                    f"There is no list of two-bit params for {op}"
+
+        for opValue, _ in TWO_BIT_PARAMS.items():
+            parentValue = opValue & 0b11111100
+            assert opCounts.get(parentValue) == 1, \
+                f"op {hex(opValue)} is not in MCOp parents enum"
+
 
 if __name__ == '__main__':
     unittest.main()