Fix: Add missing Node arg in TestLocalNodeProcessorClass' setUp method. Add more type hints (Including: set originalMTI to Union[int, None] instead of using hasattr; mark PIP methods as @staticmethod to match usage). Fix: Make Scanner.EOF int so doesn't overlap utf-8 & find works. Remove lint or add appropriate ignores.

Author: Poikilos

examples/tkexamples/cdiform.py

@@ -68,7 +68,7 @@ def __init__(self, *args, **kwargs):
             raise ValueError("at least one argument (parent) is required")
         self.parent = args[0]
         self.root = args[0]
-        self.ignore_non_gui_tags = None
+        self.ignore_non_gui_tags = None  # type: deque
         if hasattr(self.parent, 'root'):
             self.root = self.parent.root
         self._container = self  # where to put visible widgets

openlcb/__init__.py

@@ -23,7 +23,7 @@
 ORD_z = 0x7A
 
 
-def only_hex_pairs(value: str) -> bool:
+def only_hex_pairs(value: str) -> Union[re.Match[bytes], re.Match[str], None]:
     """Check if string contains only machine-readable hex pairs.
     See openlcb.conventions submodule for LCC ID dot notation
     functions (less restrictive).
@@ -68,7 +68,8 @@ def list_type_names(values) -> List[str]:
                     .format(type(values).__name__))
 
 
-def precise_sleep(seconds: Union[float, int], start: float = None) -> None:
+def precise_sleep(seconds: Union[float, int],
+                  start: Union[float, None] = None) -> None:
     """Wait for a precise number of seconds
     (precise to hundredths approximately, depending on accuracy of
     platform's sleep). Since time.sleep(seconds) is generally not
@@ -91,13 +92,16 @@ def formatted_ex(ex) -> str:
     return "{}: {}".format(type(ex).__name__, ex)
 
 
-def from_hex_bytes(b: bytearray, start: int, stop: int, assertValid=True) -> bytearray:
+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))
+            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):
@@ -110,12 +114,12 @@ def from_hex_bytes(b: bytearray, start: int, stop: int, assertValid=True) -> byt
     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")))
+            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)):  # noqa: E501
+                raise ValueError("Got character {}, expected hex digit".format((bytearray([x])).decode("utf-8")))  # noqa: E501
+            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)):  # noqa: E501
+                raise ValueError("Got character {}, expected hex digit".format((bytearray([y])).decode("utf-8")))  # noqa: E501
         # v =
-        # NOTE: below will still raise exception if over 255 even if assertValid is False
+        # NOTE: below will still raise exception if over 255 even if assertValid is False  # noqa: E501
         r[rel] = ((x & 15) + ((x >> 6) & 1) * 9) << 4 | \
             ((y & 15) + ((y >> 6) & 1) * 9)
         # assert v < 256, str(b[i:i+1])

openlcb/canbus/canframe.py

@@ -2,7 +2,12 @@
 
 from collections import OrderedDict
 from logging import getLogger
+from typing import Any, Union
 
+from openlcb import (
+    emit_cast,
+    list_type_names,
+)
 from openlcb.canbus.controlframe import ControlFrame
 from openlcb.nodeid import NodeID
 
@@ -125,13 +130,13 @@ def alias(self) -> int:
     def __init__(self, *args, afterSendState=None, reservation=None,
                  minimumState=None):
         self.afterSendState = afterSendState
-        self.encoder = NoEncoder()
+        self.encoder = NoEncoder()  # type: Any
         self.reservation = reservation
         self.minimumState = minimumState
         arg1 = None
         arg2 = None
         arg3 = None
-        self._alias = None
+        self._alias = None  # type: Union[int, None]  # TODO: type: int ?
         if len(args) > 0:
             arg1 = args[0]
         if len(args) > 1:
@@ -142,7 +147,7 @@ def __init__(self, *args, afterSendState=None, reservation=None,
             arg3 = bytearray()
         # There are three ctor forms.
         # - See "Args" in class for docstring.
-        self.header = 0
+        self.header = 0  # type: int
         self.direction = None  # See deque for usage
         self.data = bytearray()
         # three arguments as N_cid, nodeID, alias
@@ -153,26 +158,35 @@ def __init__(self, *args, afterSendState=None, reservation=None,
             #   (duck typing) in this case.
             if len(args) < 3:
                 args_error = "Expected alias after NodeID"
+            assert isinstance(arg3, int), \
+                ("Expected int alias after NodeID, got"
+                 f" {emit_cast(arg3)}")
             # cid must be 4 to 7 inclusive (100 to 111 binary)
             # precondition(4 <= cid && cid <= 7)
-            cid = arg1
             nodeID = arg2
             self._alias = arg3
+            if not isinstance(arg1, int):
+                args_error = \
+                    ("Expected CID 4 to 7 if arg2 is NodeID,"
+                     f" got {emit_cast(arg1)}")
+            else:
+                cid = arg1
 
-            nodeCode = ((nodeID.value >> ((cid-4)*12)) & 0xFFF)
-            # ^ cid-4 results in 0 to 3. *12 results in 0 to 36 bit shift (nodeID size)  # noqa: E501
-            self.header = ((cid << 12) | nodeCode) << 12 | (self._alias & 0xFFF) | 0x10_00_00_00  # noqa: E501
-            # self.data = bytearray()
+                nodeCode = ((nodeID.value >> ((cid-4)*12)) & 0xFFF)
+                # ^ cid-4 results in 0 to 3. *12 results in 0 to 36 bit shift (nodeID size)  # noqa: E501
+                self.header = ((cid << 12) | nodeCode) << 12 | (self._alias & 0xFFF) | 0x10_00_00_00  # noqa: E501
+                # self.data = bytearray()
 
         # two arguments as header, data
         elif isinstance(arg2, bytearray):
             # TODO: decode (header?) if self._alias is necessary in this case,
             #   otherwise is remains None!
             if not isinstance(arg1, int):
-                args_error = "Expected int(header) 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
+            self.header = arg1  # type: ignore
+            self._alias = arg1 & 0xFFF  # type: ignore
             if self._alias == 0:
                 logger.warning("Alias is {}".format(self._alias))
             self.data = arg2
@@ -190,7 +204,7 @@ def __init__(self, *args, afterSendState=None, reservation=None,
         # three arguments as control, alias, data
         elif isinstance(arg2, int):
             # Types of all 3 are enforced by usage (duck typing) in this case.
-            control = arg1
+            control = arg1  # type: int # type: ignore
             self._alias = arg2
             self.header = \
                 (control << 12) | (self._alias & 0xFFF) | 0x10_00_00_00
@@ -203,9 +217,10 @@ def __init__(self, *args, afterSendState=None, reservation=None,
 
         if args_error:
             raise TypeError(
-                args_error.rstrip(".") + ". Valid constructors:"
-                + CanFrame.constructor_help() + ". Got: "
-                + openlcb.list_type_names(args))
+                "{}. Valid constructors: {}. Got: {}".format(
+                    args_error.rstrip("."),
+                    CanFrame.constructor_help(),
+                    list_type_names(args)))
         if self._alias is not None:
             if self._alias & 0xFFF != self._alias:
                 raise ValueError(

openlcb/canbus/canlink.py

@@ -24,7 +24,7 @@
 from typing import (
     # Iterable,
     List,  # in case list doesn't support `[` in this Python version
-    # Union,  # in case `|` doesn't support 'type' in this Python version
+    Union,  # in case `|` doesn't support 'type' in this Python version
 )
 
 from openlcb import (
@@ -208,7 +208,7 @@ def isAllowed(self, frame: CanFrame) -> bool:
             return False
         return self.blockedFrameType(frame) is None
 
-    def blockedFrameType(self, frame: CanFrame) -> ControlFrame:
+    def blockedFrameType(self, frame: CanFrame) -> Union[ControlFrame, None]:
         if self._state == CanLink.State.Permitted:
             # All frame types are allowed in this state.
             return None
@@ -219,7 +219,10 @@ def blockedFrameType(self, frame: CanFrame) -> ControlFrame:
             #   - only AMR is allowed to be sent while transitioning to
             #     Inhibited state (and in this implementation, prior
             #     states occur before sending CID)
-            if self.getState().value >= frame.minimumState.value:
+            state = self.getState()
+            if state is None:
+                return control_frame
+            if state.value >= frame.minimumState.value:
                 return None
         if control_frame == ControlFrame.CID:
             return None
@@ -243,7 +246,7 @@ def isDuplicateAlias(self, alias):
     #   link layer later switched to an Inhibited state and had to
     #   generate a new alias
 
-    def blockedReason(self, frame: CanFrame) -> str:
+    def blockedReason(self, frame: CanFrame) -> Union[str, None]:
         if self.isCanceled(frame):
             return "The frame is using an alias from a previous reservation"
         blocked_type = self.blockedFrameType(frame)
@@ -619,7 +622,8 @@ def handleReceivedAMR(self, frame: CanFrame):
         alias = frame.header & 0xFFF
         if (self.checkAndHandleAliasCollision(frame)):
             pass  # return
-            logger.warning(f"Accepting AMR after collision (alias={alias:02X})")
+            logger.warning(
+                f"Accepting AMR after collision (alias={alias:02X})")
         #    Alias Map Reset - drop from maps
         if not frame.data:
             logger.warning(f"Bad AMR (no data, so no NodeID) from {alias:02X}")
@@ -645,7 +649,9 @@ def handleReceivedAMR(self, frame: CanFrame):
             except KeyError:
                 pass  # deleted by a concurrent process
         else:
-            logger.warning(f"AMR ignored: Node {nodeID} can't delete node {storedID}'s alias {alias} which is the same")
+            logger.warning(
+                f"AMR ignored: Node {nodeID} can't delete node"
+                f" {storedID}'s alias {alias} which is the same")
 
         storedAlias = self.nodeIdToAlias.get(nodeID)
         if storedAlias == alias:
@@ -658,7 +664,9 @@ def handleReceivedAMR(self, frame: CanFrame):
             except KeyError:
                 pass  # deleted by a concurrent process
         else:
-            logger.warning(f"AMR ignored: Alias {alias} can't delete alias {storedAlias}'s NodeID {nodeID} which is the same")
+            logger.warning(
+                f"AMR ignored: Alias {alias} can't delete alias"
+                f" {storedAlias}'s NodeID {nodeID} which is the same")
         try:
             self.duplicateAliases.remove(alias)
         except ValueError:
@@ -1380,9 +1388,10 @@ def waitForReady(self, device: PortInterface, mode="binary",
                 break
             # if verbose:
             #     print("  * state: {}".format(state))
-            assert self.getWaitForAliasResponseStart() is not None, \
+            responseStart = self.getWaitForAliasResponseStart()
+            assert responseStart is not None, \
                 "openlcb didn't send 7,6,5,4 CIDs (state={})".format(state)
-            if ((default_timer() - self.getWaitForAliasResponseStart())
+            if ((default_timer() - responseStart)
                     > CanLink.ALIAS_RESPONSE_DELAY):
                 # 200ms = standard wait time for responses
                 pass  # no collisions (fail collision test if doing that)

openlcb/canbus/canlinklayersimulation.py

@@ -7,4 +7,4 @@
 
 
 class CanLinkLayerSimulation(CanLink):
-    pass
+    pass

openlcb/canbus/canphysicallayer.py

@@ -5,7 +5,7 @@
 and is subclassed.
 '''
 from logging import getLogger
-from typing import Callable
+from typing import Callable, Union
 import warnings
 
 from openlcb.canbus.canframe import CanFrame
@@ -50,7 +50,7 @@ def sendFrameAfter(self, frame: CanFrame):
         frame.encoder = self
         PhysicalLayer.sendFrameAfter(self, frame)  # calls onQueuedFrame if set
 
-    def pollFrame(self) -> CanFrame:
+    def pollFrame(self) -> Union[CanFrame, None]:
         frame = super().pollFrame()
         if frame is None:
             return None
@@ -68,8 +68,8 @@ def registerFrameReceivedListener(self,
                 " You don't really need to listen to packets."
                 " Use pollFrame instead, which will collect and decode"
                 " packets into frames (this layer communicates to upper layers"
-                " using physicalLayer.onFrameReceived set by LinkLayer/subclass"
-                " constructor).")
+                " using physicalLayer.onFrameReceived set by"
+                " LinkLayer/subclass constructor).")
         self._frameReceivedListeners.append(listener)
 
     def fireFrameReceived(self, frame: CanFrame):

openlcb/canbus/canphysicallayergridconnect.py

@@ -105,7 +105,8 @@ def receiveAll(self, device: PortInterface, verbose=False) -> int:
             pass
         return count
 
-    def sendAll(self, device: PortInterface, mode="binary", verbose=False) -> int:
+    def sendAll(self, device: PortInterface, mode="binary",
+                verbose=False) -> int:
         """Send all queued frames using the given device.
 
         Args:
@@ -169,7 +170,8 @@ def handleDataString(self, string: str) -> int:
 
     @classmethod
     def nextPacketRange(cls, data: Union[bytes, bytearray],
-                        start: int = 0) -> Tuple[int, int]:
+                        start: int = 0) -> Union[Tuple[int, int],
+                                                 Tuple[None, None]]:
         """Get the packet slice if any.
         Returns:
             tuple(int, int): Position of ':' and position *after* ';' or
@@ -187,7 +189,8 @@ def nextPacketRange(cls, data: Union[bytes, bytearray],
         return (firstI, lastI+1)
 
     @classmethod
-    def nextPacket(cls, data: Union[bytes, bytearray]) -> bytearray:
+    def nextPacket(cls, data: Union[bytes, bytearray]) -> Union[bytearray,
+                                                                bytes, None]:
         """Get the packet including ':' and ';'."""
         start, end = cls.nextPacketRange(data)
         if start is None:
@@ -197,7 +200,8 @@ def nextPacket(cls, data: Union[bytes, bytearray]) -> bytearray:
     @staticmethod
     def readInt32(data, start):
         # chunk = data[start:start+8]
-        # return int.from_bytes(chunk, 'big') - 0x30303030 + ((chunk & 0x40404040) >> 6) * 9
+        # return (int.from_bytes(chunk, 'big') - 0x30303030 +
+        #         ((chunk & 0x40404040) >> 6) * 9)
         """Fast hex bytearray → 32-bit int (8 hex chars, ASCII)"""
         # branchless conversion for uppercase/lowercase or digits:
         # - (data[i] & 15): Gives the low 4 bits (0-9 for digits
@@ -320,7 +324,7 @@ def handleData(self, data: Union[bytes, bytearray],
             if first is None:
                 break
             # else last is not None guaranteed
-            semi = (end - 2) if (self.inboundBuffer[end-1] == 0x0a) else (end - 1)
+            semi = (end - 2) if (self.inboundBuffer[end-1] == 0x0a) else (end - 1)  # noqa: E501
             if semi - first < 11:  # len(":X") + 8 data + len("N")
                 logger.warning(
                     "[handleData] Skipped malformed {} "
@@ -352,17 +356,19 @@ def handleData(self, data: Union[bytes, bytearray],
             dataEnd = semi
             if self.assertValidData and (dataEnd - dataI > 0):
                 assert only_hex_pairs(self.inboundBuffer[headerI:headerEnd]), \
-                    self.inboundBuffer[headerI:headerEnd]  # show the non-hex data
+                    self.inboundBuffer[headerI:headerEnd]  # show non-hex data
                 assert only_hex_pairs(self.inboundBuffer[dataI:dataEnd]), \
                     self.inboundBuffer[dataI:dataEnd]  # show the non-hex data
-            header_bytes = from_hex_bytes(self.inboundBuffer, headerI, headerEnd,
+            header_bytes = from_hex_bytes(self.inboundBuffer, headerI,
+                                          headerEnd,
                                           assertValid=self.assertValidData)
             if dataEnd - dataI > 0:
                 if (dataEnd - dataI) % 2 > 0:
                     logger.warning(
                         "[handleData] Skipped malformed packet"
                         " (Incomplete pair in {} (range {},{}) in {})"
-                        .format(repr(self.inboundBuffer[dataI:dataEnd]), dataI, dataEnd, repr(self.inboundBuffer[first:end])))
+                        .format(repr(self.inboundBuffer[dataI:dataEnd]), dataI,
+                                dataEnd, repr(self.inboundBuffer[first:end])))
                     start = end
                     continue
                 outData = from_hex_bytes(self.inboundBuffer, dataI, dataEnd,

openlcb/canbus/canphysicallayersimulation.py

@@ -60,7 +60,7 @@ def sendAll(self, device, mode="binary", verbose=False) -> int:
                     blockedMsg = self.linkLayer.blockedReason(frame)
                     if blockedMsg:
                         if verbose:
-                            print("Skipping sending frame: {}".format(blockedMsg))
+                            print(f"Skipping sending frame: {blockedMsg}")
                         continue
                 # data = self.encodeFrameAsData(frame)
                 # device.send(data)  # commented since simulation

openlcb/canbus/seriallink.py

@@ -28,7 +28,7 @@ def _settimeout(self, seconds: float):
         logger.warning("settimeout is not implemented for SerialLink")
         pass
 
-    def _connect(self, _, port: str, device: serial.Serial = None,
+    def _connect(self, _, port: str, device: Union[serial.Serial, None] = None,
                  baudrate: int = 230400):
         """Connect to a serial port.
 
@@ -51,7 +51,7 @@ def _connect(self, _, port: str, device: serial.Serial = None,
             self._device = device
         self._device.reset_input_buffer()  # drop anything that's just sitting there already  # noqa: E501
 
-    def _send(self, msg: Union[bytes, bytearray]):
+    def _send(self, data: Union[bytes, bytearray]) -> None:
         """send bytes
 
         Args:
@@ -62,8 +62,8 @@ def _send(self, msg: Union[bytes, bytearray]):
             RuntimeError: If the string couldn't be written to the port.
         """
         total_sent = 0
-        while total_sent < len(msg[total_sent:]):
-            sent = self._device.write(msg[total_sent:])
+        while total_sent < len(data[total_sent:]):
+            sent = self._device.write(data[total_sent:])
             if sent == 0:
                 self.setOpen(False)
                 raise RuntimeError("socket connection broken")