Fix bobjacobsen#81: Clear others from maps if AME has no data, even if not Permitted, as per Standard; Test exact length of data when matching AME as per Standard. Move decodeControlFrameFormat to CanFrame as static method for modularity (no CanLink instance is required). Add noise rejection (move old handleData code to handleDataOptimized) (fix bobjacobsen#82 and fix bobjacobsen#83 before committing). Add reusable from*_hex_bytes functions. Make handleData and CanFrame constructor messages more explicit. Fix bobjacobsen#86: Add minimumState option to CanFrame so transition to Inhibited can send AMR after CIDs are queued but not sent, since enqueuing since enqueuing changes CanLink state immediately. Fix bobjacobsen#77: Always increment nextInternallyAssignedNodeID when used. Add explicit blockedReason function for tracing.

Author: Poikilos

openlcb/__init__.py

@@ -8,17 +8,29 @@
     Union,  # in case `|` doesn't support 'type' in this Python version
 )
 
-
-hex_pairs_rc = re.compile(r"^([0-9A-Fa-f]{2})+$")
+hex_pairs_re = r"^([0-9A-Fa-f]{2})+$"
+hex_pairs_rc = re.compile(hex_pairs_re)
+hex_pairs_brc = re.compile(hex_pairs_re.encode("utf-8"))
 # {2}: Exactly two characters found (only match if pair)
 # +: at least one match plus 0 or more additional matches
+ORD_0 = 0x30
+ORD_9 = 0x39
+ORD_A = 0x41
+ORD_F = 0x46
+ORD_Z = 0x5A
+ORD_a = 0x61
+ORD_f = 0x66
+ORD_z = 0x7A
 
 
 def only_hex_pairs(value: str) -> bool:
     """Check if string contains only machine-readable hex pairs.
     See openlcb.conventions submodule for LCC ID dot notation
     functions (less restrictive).
     """
+    if isinstance(value, (bytearray, bytes)):
+        return hex_pairs_brc.fullmatch(value)
+    assert isinstance(value, str)
     return hex_pairs_rc.fullmatch(value)
 
 
@@ -77,3 +89,41 @@ def precise_sleep(seconds: Union[float, int], start: float = None) -> None:
 
 def formatted_ex(ex) -> str:
     return "{}: {}".format(type(ex).__name__, ex)
+
+
+def from_hex_bytes(b: bytearray, start: int, stop: int, assertValid=True) -> bytearray:
+    """ASCII hex bytearray (even length) → binary bytearray"""
+    # like bytearray.fromhex, except accepts bytes rather than str only
+    r = bytearray((stop-start) // 2)
+    if assertValid:
+        if (stop-start) % 2 > 0:
+            raise IndexError("Only hex pairs are accepted, got odd count: start={} stop={}".format(start, stop))
+        if start < 0 or start > len(b):
+            raise IndexError("start={} len={}".format(start, len(b)))
+        if stop < 0 or stop > len(b):
+            raise IndexError("stop={} len={}".format(start, len(b)))
+        if stop - start < 2:
+            raise IndexError("start={} stop={}".format(start, stop))
+        assert len(r) == (stop - start) // 2
+    i = start
+    rel = 0
+    while i < stop:
+        x, y = b[i], b[i+1]
+        if assertValid:
+            if not ((x >= ORD_A and x <= ORD_F) or (x >= ORD_a and x <= ORD_f) or (x >= ORD_0 and x <= ORD_9)):
+                raise ValueError("Got character {}, expected hex digit".format((bytearray([x])).decode("utf-8")))
+            if not ((y >= ORD_A and y <= ORD_F) or (y >= ORD_a and y <= ORD_f) or (y >= ORD_0 and y <= ORD_9)):
+                raise ValueError("Got character {}, expected hex digit".format((bytearray([y])).decode("utf-8")))
+        # v =
+        # NOTE: below will still raise exception if over 255 even if assertValid is False
+        r[rel] = ((x & 15) + ((x >> 6) & 1) * 9) << 4 | \
+            ((y & 15) + ((y >> 6) & 1) * 9)
+        # assert v < 256, str(b[i:i+1])
+        # r[rel] = v
+        i += 2
+        rel += 1
+    return r
+
+
+def from_all_hex_bytes(b: bytearray) -> bytearray:
+    return from_hex_bytes(b, 0, len(b))

openlcb/canbus/canframe.py

@@ -3,6 +3,7 @@
 from collections import OrderedDict
 from logging import getLogger
 
+from openlcb.canbus.controlframe import ControlFrame
 from openlcb.nodeid import NodeID
 
 logger = getLogger(__name__)
@@ -83,6 +84,28 @@ def constructor_help():
             result += "), "
         return result[:-2]  # -1 to remove last ", " from list
 
+    @staticmethod
+    def decodeControlFrameFormat(frame) -> ControlFrame:
+        # type: (CanFrame) -> ControlFrame
+        if (frame.header & 0x0800_0000) == 0x0800_0000:
+            # data case; not checking leading 1 bit
+            # NOTE: handleReceivedData can get all header bits via frame
+            return ControlFrame.Data
+        if (frame.header & 0x4_000_000) != 0:  # CID case
+            # NOTE: handleReceivedCID can get all header bits via frame
+            return ControlFrame.CID
+
+        try:
+            retval = ControlFrame((frame.header >> 12) & 0x2_FF_FF)
+            return retval  # top 1 bit for out-of-band messages
+        except KeyboardInterrupt:
+            raise
+        except:
+            logger.warning(
+                "Could not decode header 0x{:08X}"
+                .format(frame.header))
+            return ControlFrame.UnknownFormat
+
     def __str__(self):
         return "CanFrame header: 0x{:08X} {}".format(
             self.header,
@@ -99,10 +122,12 @@ def encodeAsBytes(self) -> bytes:
     def alias(self) -> int:
         return self._alias
 
-    def __init__(self, *args, afterSendState=None, reservation=None):
+    def __init__(self, *args, afterSendState=None, reservation=None,
+                 minimumState=None):
         self.afterSendState = afterSendState
         self.encoder = NoEncoder()
         self.reservation = reservation
+        self.minimumState = minimumState
         arg1 = None
         arg2 = None
         arg3 = None
@@ -144,9 +169,12 @@ def __init__(self, *args, afterSendState=None, reservation=None):
             # TODO: decode (header?) if self._alias is necessary in this case,
             #   otherwise is remains None!
             if not isinstance(arg1, int):
-                args_error = "Expected int since 2nd argument is bytearray."
+                args_error = "Expected int(header) since 2nd argument is bytearray."
             # Types of both args are enforced by this point.
             self.header = arg1
+            self._alias = arg1 & 0xFFF
+            if self._alias == 0:
+                logger.warning("Alias is {}".format(self._alias))
             self.data = arg2
             if len(args) > 2:
                 args_error = "2nd argument is data, but got extra argument(s)"