Move bitfields to MemoryConfigurationHeader class and MemorySpaceIndex enum to make usage and enforcement more clear.

Author: Poikilos

openlcb/convert.py

@@ -19,45 +19,6 @@
 
 class Convert:
 
-    @staticmethod
-    def deserializeMC2ndByte(datagramByte1):
-        """Decode byte[1] (2nd) of Memory Configuration Datagram"""
-        has_byte6 = False
-        if datagramByte1 & 0x03 == 0:
-            has_byte6 = True
-        return has_byte6, datagramByte1 & 0xFC
-        # ^ 0xFC = 11111100
-
-    # formerly spaceDecode, but it serializes a space for datagram byte2
-    @staticmethod
-    def serializeSpace(space):
-        """Convert from a space number to either
-        False and control number or True and standard memory space
-        for use in a Datagram.
-
-        Args:
-            space (int): Sequential memory space identifier, where values:
-            - 0xFF to 0xFD are special spaces, and only the least significant
-              2 bits will be used in a datagram.
-            - 0x00 to 0xFC represent standard memory spaces directly.
-
-        Returns:
-            tuple(bool, byte): (is custom space, control | space)
-                - (False, control number 1 to 3 inclusive) :
-                  spaces 0xFF - 0xFD (Except bits beyond 0x00000011
-                  differ for each datagram type. See 4.2 Address
-                  Space Selection in OpenLCB Memory Configuration
-                  Standard)
-                - or (True, space number) : spaces 0 - 0xFC
-                (NOTE: type of space may affect type of output)
-        """
-        # TODO: Maybe check type of space & raise TypeError if not
-        #   something valid, whether byte, int, or what is ok [add
-        #   more _description_ to space in docstring].
-        if space >= 0xFD:
-            return (False, space & 0x03)
-        return (True, space)
-
     @staticmethod
     def arrayToInt(data: Union[bytes, bytearray, List[int]]) -> int:
         """Convert an array in MSB-first order to an integer

openlcb/memoryconfigurationheader.py

@@ -0,0 +1,81 @@
+
+from enum import Enum
+from typing import Union
+# FDI = 0xFA
+# Configuration = 0xFD
+# All = 0xFE
+# CDI = 0xFF  # ~~decodes~~ encoded (in header) as 0x03
+
+
+class MemorySpaceIndex(Enum):
+    Uninitialized = -1
+    Custom = 0
+    Configuration = 1  # 0xFD & 0x03 == 1
+    All = 2  # 0xFE & 0x03 == 2
+    CDI = 3  # 0xFF & 0x03 == 3
+
+    @classmethod
+    def fromNumber(cls, num: int):
+        """Return the MemorySpace member with the given numeric value,
+        or None if no match is found.
+        """
+        assert isinstance(num, int)
+        for member in cls:
+            if member.value == num:
+                return member
+        return cls.Custom
+
+
+class MemoryConfigurationHeader:
+    """Manage data corresponding to bitfields in Memory Configuration.
+    See OpenLCB "Memory Configuration" Standard
+
+    Arguments:
+        space (int): Space number (MemorySpaceIndex.*.Value,
+            MemorySpace.*.value, or raw number including a custom
+            space).
+            - 0xFF to 0xFD are special spaces, and only the least
+              significant 2 bits will be used in a datagram.
+            - 0x00 to 0xFC represent standard memory spaces directly.
+    """
+    def __init__(self, space: int):
+        # formerly Convert.serializeSpace
+        # formerly spaceDecode, but it serializes a space for datagram byte2
+        assert isinstance(space, int)
+        spaceIndexValue = space & 0x03
+        self.spaceIndex = \
+            MemorySpaceIndex.fromNumber(
+                spaceIndexValue)  # type: MemorySpaceIndex
+        self.customSpace = None  # type: int|None
+        if self.spaceIndex is MemorySpaceIndex.Custom:
+            self.customSpace = space
+        self.highBits = 0  # type: int
+
+    @classmethod
+    def fromMC2ndByte(cls, datagramByte1: int, space: Union[int, None] = None) -> 'MemoryConfigurationHeader':  # noqa: E501
+        """Deserialize Memory Configuration byte 1.
+
+        For serializing a space (such as packing a datagram header),
+        use constructor instead.
+
+        Args:
+            datagramByte1 (int): byte[1] (2nd) of Memory Configuration Datagram
+            space (int): Only applies for custom space (datagramByte)
+        """
+        if space is not None:
+            assert isinstance(space, int)
+            assert datagramByte1 & 0x03 == 0, \
+                'custom space requires datagramByte1 with last 2 bits 00'
+        else:
+            # space is None
+            assert datagramByte1 & 0x03 != 0, \
+                'a standard space must be in last 2 bits datagramByte1'
+            space = -1
+        # formerly deserializeMC2ndByte
+        result = cls(datagramByte1 & 0x03)
+        if datagramByte1 & 0x03 == 0:
+            # Default (-1) means not enough information
+            #    (space not known, but is MemorySpaceIndex.Custom)
+            result.customSpace = space
+        result.highBits = datagramByte1 & 0xFC  # 0xFC = 0b11111100
+        return result

openlcb/memoryservice.py

@@ -40,23 +40,30 @@
     DatagramService,
 )
 from openlcb.convert import Convert
+from openlcb.memoryconfigurationheader import MemoryConfigurationHeader, MemorySpaceIndex
+from openlcb.memorymanager import MemoryManager
 from openlcb.nodeid import NodeID
 
 logger = getLogger(__name__)
 
-
+MODE_BYTES['']
 MODE_BYTES = {
-    'Read_Command': {0x40, 0x41, 0x42, 0x43},
+    # order determines meaning for lists (See )
+    'Read_Command': {0x40, 0x41, 0x42, 0x43},  # TODO: Use memoryManagers
     'Read_Reply': {0x50, 0x51, 0x52, 0x53},
-    'Read_Stream_Command': {0x60, 0x61, 0x62, 0x63},
-    'Read_Stream_Reply': {0x70, 0x71, 0x72, 0x73},
-    'Write_Command': {0x00, 0x01, 0x02, 0x03},
+    'Read_Stream_Command': {0x60, 0x61, 0x62, 0x63},  # TODO: Use memoryManagers
+    'Read_Stream_Reply': {0x70, 0x71, 0x72, 0x73},  # TODO
+    'Write_Command': [0x00, 0x01, 0x02, 0x03],  # TODO: Use memoryManagers
     'Write_Reply': {0x10, 0x11, 0x12, 0x13},
-    'Write_Under_Mask_Command': {0x08, 0x09, 0x0A, 0x0B},
+    'Write_Under_Mask_Command': {0x08, 0x09, 0x0A, 0x0B},  # TODO: Use memoryManagers
     'Write_Stream_Command': {0x20, 0x21, 0x22, 0x23},
-    'Write_Stream_Reply': {0x30, 0x31, 0x32, 0x33},
+    'Write_Stream_Reply': {0x30, 0x31, 0x32, 0x33},  # TODO
+    'Get_Address_Space_Info_Command': {0x84, },
+    'Get_Address_Space_Info_Reply': {0x86, 0x87, },
+    'Lock_Reserve_Command': {0x88, },
 }
 
+
 MODE_ERROR_BYTES = {
     'Read_Reply': {0x58, 0x59, 0x5A, 0x5B},
     'Read_Stream_Reply': {0x78, 0x79, 0x7A, 0x7B},
@@ -85,10 +92,10 @@ class MemorySpace(Enum):
             (See OpenLCB Memory Configuration Standard 4.2).
     """
     Uninitialized = -1
-    CDI = 0xFF  # decodes to 0x03
     FDI = 0xFA
-    All = 0xFE
     Configuration = 0xFD
+    All = 0xFE
+    CDI = 0xFF  # decodes to 0x03
 
     @classmethod
     def fromNumber(cls, num: int):
@@ -101,6 +108,22 @@ def fromNumber(cls, num: int):
                 return member
         return None
 
+    @classmethod
+    def fromIndex(cls, msi: MemorySpaceIndex):
+        """Return the MemorySpace member with the given numeric value,
+        or None if no match is found.
+        """
+        assert isinstance(msi, MemorySpaceIndex)
+        if msi is MemorySpaceIndex.Custom:
+            return None
+        elif msi is MemorySpaceIndex.Configuration:
+            return cls.Configuration
+        elif msi is MemorySpaceIndex.All:
+            return cls.All
+        elif msi is MemorySpaceIndex.CDI:
+            return cls.CDI
+        return None
+
 
 class MemoryReadMemo:
     """Memo carries request and reply.
@@ -211,10 +234,16 @@ def parseReplyDatagram(memo: Union[MemoryReadMemo, MemoryWriteMemo],
             "Datagram is truncated to 1 byte:"
             f" it is {hex(dmemo.data[0])}")
         return
-    (hasByte6, _) = Convert.deserializeMC2ndByte(dmemo.data[1])
+    mcHeader = MemoryConfigurationHeader.fromMC2ndByte(
+        dmemo.data[1],
+        # space=memo.space,
+    )
     offset = 6
     error = None
-    if hasByte6:
+    assert mcHeader.spaceIndex is not MemorySpaceIndex.Uninitialized
+    if mcHeader.spaceIndex is MemorySpaceIndex.Custom:
+        # mcHeader.customSpace = memo.space
+        mcHeader.customSpace = dmemo.data[6]
         offset = 7
     memo.error = None
     memo.errorCode = None
@@ -256,9 +285,15 @@ def parseReplyDatagram(memo: Union[MemoryReadMemo, MemoryWriteMemo],
             error += f" ({list(dmemo.data[message_idx:])})"
     else:
         error = f"(2nd byte = {hex(dmemo.data[1])})"
-    error += f" (hasByte6={hasByte6})"
-    if hasByte6:
-        error += f" (space={hex(dmemo.data[6])})"
+    error += f" (spaceIndex={mcHeader.spaceIndex})"
+    if mcHeader.spaceIndex is mcHeader.customSpace:
+        if mcHeader.customSpace is not None:
+            if mcHeader.customSpace != dmemo.data[6]:
+                error += f" (mcHeader.customSpace={hex(mcHeader.customSpace)} != space={hex(dmemo.data[6])} !)"  # noqa: E501
+            else:
+                error += f" (mcHeader.customSpace={hex(mcHeader.customSpace)})"
+        else:
+            error += f" (space={hex(dmemo.data[6])} mcHeader.customSpace=None!)"  # noqa: E501
     memo.error = error
 
 
@@ -268,6 +303,12 @@ class MemoryService:
 
     Args:
         service (DatagramService): See DatagramService.
+
+    Attributes:
+        memoryManagers (dict[str, MemoryManager]): The storage where
+            other nodes can read and write memory. Each element can be
+            changed to a specific nodeid's memory manager. They key is
+            the NodeID in string form (dotted notation).
     """
 
     def __init__(self, service: DatagramService):
@@ -280,6 +321,7 @@ def __init__(self, service: DatagramService):
         self.service.registerDatagramReceivedListener(
             self.datagramReceivedListener
         )
+        self.memoryManagers = {}  # type: dict[str, MemoryManager]
 
     def requestMemoryRead(self, memo, stream: bool = False):
         # type: (MemoryReadMemo, Optional[bool]) -> None
@@ -308,16 +350,10 @@ def requestMemoryReadNext(self, memo, stream: bool = False):
             memo (MemoryReadMemo): Request to send.
         """
         assert isinstance(stream, bool)
-        hasByte6 = False  # if custom space is defined in byte 6
-        flag = 0
-        (hasByte6, flag) = Convert.serializeSpace(memo.space)
-        if stream:
-            # Encoding: 0x60=custom, 0x61=0xFD, 0x62=0xFE, 0x63=0xFF
-            spaceFlag = 0x60 if hasByte6 else (flag | 0x60)
-        else:
-            # Encoding: 0x40=custom, 0x41=0xFD, 0x42=0xFE, 0x43=0xFF
-            spaceFlag = 0x40 if hasByte6 else (flag | 0x40)  # | 0b11111100
-        # ^ In else case, flag is 1-3, so re-add 0xFC (0b11111100)
+        mcHeader = MemoryConfigurationHeader(memo.space)
+        assert mcHeader.spaceIndex is not None
+        spaceFlag = (0x60 if stream else 0x40) | mcHeader.spaceIndex.value
+        # NOTE: Why was there commented: | 0xFC (0b11111100) if not stream?
         addr2 = ((memo.address >> 24) & 0xFF)
         addr3 = ((memo.address >> 16) & 0xFF)
         addr4 = ((memo.address >> 8) & 0xFF)
@@ -326,7 +362,7 @@ def requestMemoryReadNext(self, memo, stream: bool = False):
             DatagramService.ProtocolID.MemoryOperation.value, spaceFlag,
             addr2, addr3, addr4, addr5])
         # NOTE: list[int] is ok for bytearray extend (`+` requires cast)
-        if hasByte6:
+        if mcHeader.customSpace is not None:
             assert memo.space <= 0xFF, f"Space {memo.space} out of byte range"
             data.extend([(memo.space & 0xFF)])
         data.extend([memo.size])
@@ -352,7 +388,6 @@ def datagramReceivedListener(self, dmemo: DatagramReadMemo) -> bool:
         if self.service.datagramType(dmemo.data) \
                 != DatagramService.ProtocolID.MemoryOperation :
             return False
-
         # datagram must has a command value
         if len(dmemo.data) < 2:
             logger.error("Memory service datagram too short: {}"
@@ -449,14 +484,8 @@ def requestMemoryWrite(self, memo: MemoryWriteMemo, stream: bool = False):
         self.writeMemos.append(memo)
         # create & send a write datagram
         hasByte6 = False  # if custom space is defined in byte 6
-        flag = 0
-        (hasByte6, flag) = Convert.serializeSpace(memo.space)
-        if stream:
-            # Encoding: 0x20=custom, 0x21=0xFD, 0x22=0xFE, 0x23=0xFF
-            spaceFlag = 0x20 if hasByte6 else (flag | 0x20)
-        else:
-            # Encoding: 0x00=custom, 0x01=0xFD, 0x02=0xFE, 0x03=0xFF
-            spaceFlag = 0x00 if hasByte6 else (flag | 0x00)
+        header = MemoryConfigurationHeader(memo.space)
+        spaceFlag = (0x20 if stream else 0) | header.spaceIndex.value
         addr2 = ((memo.address >> 24) & 0xFF)
         addr3 = ((memo.address >> 16) & 0xFF)
         addr4 = ((memo.address >> 8) & 0xFF)

python-openlcb.code-workspace

@@ -68,6 +68,7 @@
 			"localoverrides",
 			"MDNS",
 			"mdnsconventions",
+			"memoryconfigurationheader",
 			"memorymanager",
 			"memoryservice",
 			"metas",

tests/test_convert.py

@@ -3,6 +3,7 @@
 import unittest
 
 from openlcb.convert import Convert
+from openlcb.memoryconfigurationheader import MemoryConfigurationHeader, MemorySpaceIndex
 
 
 class TestConvertClass(unittest.TestCase):
@@ -88,17 +89,19 @@ def testIntToArrayFail(self):
                 Convert.intToArray(value, length)
 
     def testSerializeSpace(self):
-        byte6 = False
-        space = 0x00
-
-        (byte6, space) = Convert.serializeSpace(0xF8)
-        self.assertEqual(space, 0xF8)
-        self.assertTrue(byte6)
-
-        (byte6, space) = Convert.serializeSpace(0xFF)
-        self.assertEqual(space, 0x03)
-        self.assertFalse(byte6)
-
-        (byte6, space) = Convert.serializeSpace(0xFD)
-        self.assertEqual(space, 0x01)
-        self.assertFalse(byte6)
+        # byte6 = False
+        # space = 0x00
+
+        mcHeader = MemoryConfigurationHeader(0xF8)
+        self.assertEqual(mcHeader.customSpace, 0xF8)
+        self.assertEqual(mcHeader.spaceIndex, MemorySpaceIndex.Custom)
+
+        mcHeader = MemoryConfigurationHeader(0xFF)
+        self.assertIs(mcHeader.spaceIndex, MemorySpaceIndex.CDI)
+        self.assertEqual(mcHeader.spaceIndex.value, 0x03)
+        self.assertIsNone(mcHeader.customSpace)
+
+        mcHeader = MemoryConfigurationHeader(0xFD)
+        self.assertIs(mcHeader.spaceIndex, MemorySpaceIndex.Configuration)
+        self.assertEqual(mcHeader.spaceIndex.value, 0x01)
+        self.assertIsNone(mcHeader.customSpace)