Make a single and required frame handler in physicalLayer for enforcing that the stack exists (Setting physicalLayer.onReceivedFrame to CanLink's receiveListener during CanLink constructor is enforced).

Author: Poikilos

examples/example_tcp_message_interface.py

@@ -14,7 +14,7 @@
 from logging import getLogger
 # region same code as other examples
 from examples_settings import Settings
-from openlcb.rawphysicallayer import RealtimeRawPhysicalLayer  # do 1st to fix path if no pip install
+from openlcb.realtimephysicallayer import RealtimePhysicalLayer  # do 1st to fix path if no pip install
 settings = Settings()
 
 if __name__ == "__main__":
@@ -56,14 +56,14 @@
 #     assert isinstance(data, (bytes, bytearray))
 #     print("      SR: {}".format(data))
 #     sock.send(data)
-# ^ Moved to RealtimeRawPhysicalLayer sendFrameAfter override
+# ^ Moved to RealtimePhysicalLayer sendFrameAfter override
 
 
 def printMessage(msg):
     print("RM: {} from {}".format(msg, msg.source))
 
 
-physicalLayer = RealtimeRawPhysicalLayer(sock)
+physicalLayer = RealtimePhysicalLayer(sock)
 # ^ this was not in the example before
 # (just gave sendToSocket to TcpLink)
 

examples/examples_gui.py

@@ -28,7 +28,7 @@
           file=sys.stderr)
     raise
 from tkinter import ttk
-from collections import OrderedDict
+from collections import OrderedDict, deque
 
 from examples_settings import Settings
 # ^ adds parent of module to sys.path, so openlcb imports *after* this
@@ -131,7 +131,7 @@ def __init__(self, parent):
         self.zeroconf = None
         self.listener = None
         self.browser = None
-        self.errors = []
+        self.errors = deque()
         self.root = parent
         self._connect_thread = None
         try:
@@ -170,7 +170,7 @@ def on_form_loaded(self):
     def show_next_error(self):
         if not self.errors:
             return 0
-        error = self.errors.pop(0)
+        error = self.errors.popleft()
         if not error:
             return 0
         self.set_status(error)

openlcb/canbus/canlink.py

@@ -82,7 +82,8 @@ class CanLink(LinkLayer):
 
     # MIN_STATE_VALUE & MAX_STATE_VALUE are set statically below the
     #   State class declaration:
-    ALIAS_RESPONSE_DELAY = .2  # See docstring.
+    STANDARD_ALIAS_RESPONSE_DELAY = .2
+    ALIAS_RESPONSE_DELAY = 20  # See docstring.
 
     class State(Enum):
         """Used as a linux-like "runlevel"
@@ -104,7 +105,7 @@ class State(Enum):
                 determined to be success, this state triggers
                 _enqueueReserveID.
         """
-        Initial = LinkLayer.State.Undefined.value  # special case of .Inhibited
+        Initial = 1  # special case of .Inhibited
         #   where init hasn't started.
         Inhibited = 2
         EnqueueAliasAllocationRequest = 3
@@ -126,6 +127,9 @@ class State(Enum):
         Permitted = 20  # formerly 3. queued via frame
         # (formerly set at end of _notifyReservation code)
 
+    InitialState = State.Initial
+    DisconnectedState = State.Inhibited
+
     MIN_STATE_VALUE = min(entry.value for entry in State)
     MAX_STATE_VALUE = max(entry.value for entry in State)
 
@@ -230,7 +234,8 @@ def isDuplicateAlias(self, alias):
 
     def _onStateChanged(self, oldState, newState):
         # return super()._onStateChanged(oldState, newState)
-        assert isinstance(newState, CanLink.State)
+        assert isinstance(newState, CanLink.State), \
+            "expected a CanLink.State, got {}".format(emit_cast(newState))
         if newState == CanLink.State.EnqueueAliasAllocationRequest:
             self._enqueueCIDSequence()
             # - sets state to BusyLocalCIDSequence

openlcb/canbus/canphysicallayer.py

@@ -4,8 +4,8 @@
 This is a class because it represents a single physical connection to a layout
 and is subclassed.
 '''
-import sys
 from logging import getLogger
+import warnings
 
 from openlcb.canbus.canframe import CanFrame
 from openlcb.canbus.controlframe import ControlFrame
@@ -24,37 +24,61 @@ def __init__(self,):
         PhysicalLayer.__init__(self)
         self.listeners = []
 
+    def onReceivedFrame(self):
+        raise NotImplementedError(
+            "Your LinkLayer/subclass must set this manually (monkeypatch)"
+            " to the CanLink instance's receiveListener method.")
+
     def sendFrameAfter(self, frame: CanFrame):
-        """See sendFrameAfter documentation in PhysicalLayer.
-        This implementation behaves the same except requires
-        a specific type (CanFrame).
+        """Enqueue: *IMPORTANT* Main/other thread may have
+        called this. Any other thread sending other than the _listen
+        thread is bad, since overlapping calls to socket cause undefined
+        behavior, so this just adds to a deque (double ended queue, used
+        as FIFO).
+        - CanPhysicalLayerGridConnect formerly had canSendCallback
+          but now it uses its own frame deque, and the socket code pops
+          and sends the frames.
+          (formerly canSendCallback was set to a sendToPort function
+          which was formerly a direct call to a port which was not
+          thread-safe and could be called from anywhere in the
+          openlcb stack)
+          - Add a generalized LocalEvent queue avoid deep callstack?
+            - See issue #62 comment about a local event queue.
+              For now, CanFrame is used (improved since issue #62
+              was solved by adding more states to CanLink so it
+              can have incremental states instead of requiring two-way
+              communication [race condition] during a single
+              blocking call to defineAndReserveAlias)
         """
-        # formerly sendCan Frame, but now behavior is defined by superclass
-        #   (regardless of frame type, it is just added to self._sends)
         assert isinstance(frame, CanFrame)
+        frame.encoder = self
         PhysicalLayer.sendFrameAfter(self, frame)
 
-    def pollFrame(self) -> CanFrame:  # overloaded for type hinting.
-        """Check if there is another frame queued and get it.
-        Returns:
-            CanFrame: next frame in FIFO buffer (_sends).
-        """
-        return PhysicalLayer.pollFrame(self)
+    def pollFrame(self) -> CanFrame:
+        frame = super().pollFrame()
+        if frame is None:
+            return None
+        assert isinstance(frame, CanFrame)
+        return frame
 
     def encode(self, frame) -> str:
         '''abstract interface (encode frame to string)'''
         raise NotImplementedError("Each subclass must implement this.")
 
     def registerFrameReceivedListener(self, listener):
+        assert listener is not None
+        warnings.warn(
+            "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 self.onReceivedFrame set in LinkLayer/subclass"
+            " constructor).")
         self.listeners.append(listener)
 
     def fireListeners(self, frame):
-        if not self.listeners:
-            logger.warning(
-                "No listeners for frame received."
-                " CanLink (see LinkLayer superclass constructor)"
-                " should at least register its receiveFrame method"
-                " with a physical layer implementation.")
+        """At least the LinkLayer (CanLink in this case)
+        should register one listener."""
+
         for listener in self.listeners:
             listener(frame)
 

openlcb/canbus/canphysicallayergridconnect.py

@@ -12,19 +12,19 @@
 
 from collections import deque
 from typing import Union
-from openlcb.canbus.canlink import CanLink
 from openlcb.canbus.canphysicallayer import CanPhysicalLayer
 from openlcb.canbus.canframe import CanFrame
+from openlcb.frameencoder import FrameEncoder
 
 GC_START_BYTE = 0x3a  # :
 GC_END_BYTE = 0x3b  # ;
 
 
-class CanPhysicalLayerGridConnect(CanPhysicalLayer):
+class CanPhysicalLayerGridConnect(CanPhysicalLayer, FrameEncoder):
     """CAN physical layer subclass for GridConnect
 
     This acts as frame.encoder for canLink, and manages the packet
-    _sends queue (deque is used for speed; defined & managed in base
+    _send_frames queue (deque is used for speed; defined & managed in base
     class: PhysicalLayer)
 
     Args:
@@ -39,7 +39,7 @@ def __init__(self):
         # ^ A CanLink requires a physical layer to operate,
         #   so CanLink now requires a PhysicalLayer instance
         #   such as this in its constructor.
-        CanPhysicalLayer.__init__(self)
+        CanPhysicalLayer.__init__(self)  # creates self._send_frames
 
         # region moved to CanLink constructor
         # from canLink.linkPhysicalLayer(self)  # self.setCallBack(callback):
@@ -53,10 +53,6 @@ def __init__(self):
     #     assert callable(callback)
     #     self.canSendCallback = callback
 
-    def sendFrameAfter(self, frame: CanFrame) -> None:
-        frame.encoder = self
-        self._sends.appendleft(frame)  # self.canSendCallback(frame)
-
     def encodeFrameAsString(self, frame) -> str:
         '''Encode frame to string.'''
         output = ":X{:08X}N".format(frame.header)  # at least 8 chars, hex
@@ -65,6 +61,11 @@ def encodeFrameAsString(self, frame) -> str:
         output += ";\n"
         return output
 
+    def encodeFrameAsData(self, frame) -> Union[bytearray, bytes]:
+        # TODO: Consider doing this manually (in Python 3,
+        #   bytes/bytearray has no attribute 'format')
+        return self.encodeFrameAsString(frame).encode("utf-8")
+
     def handleDataString(self, string: str):
         '''Provide string from the outside link to be parsed
 

openlcb/canbus/canphysicallayersimulation.py

@@ -8,8 +8,8 @@
 class CanPhysicalLayerSimulation(CanPhysicalLayer):
 
     def __init__(self):
-        self.receivedFrames = []
+        self.receivedPackets = []
         CanPhysicalLayer.__init__(self)
 
-    def sendFrameAfter(self, frame):
-        self.receivedFrames.append(frame)
+    def handlePacket(self, frame):
+        self.receivedPackets.append(frame)

openlcb/dispatcher.py

@@ -171,7 +171,7 @@ def start_listening(self, connected_port, localNodeID):
         # self._physicalLayer.registerFrameReceivedListener(
         #     self._printFrame
         # )
-        # ^ Commented since canlink already adds CanLink's default
+        # ^ Commented since CanLink constructor now registers its default
         #   receiveListener to CanLinkPhysicalLayer & that's all we need
         #   for this application.
 
@@ -203,18 +203,21 @@ def start_listening(self, connected_port, localNodeID):
         self.listen()  # Must listen for alias reservation responses
         #   (sendAliasConnectionSequence will occur for another 200ms
         #   once, then another 200ms on each alias collision if any)
+        #   - must also keep doing frame = pollFrame() and sending
+        #     if not None.
 
         self._callback_status("physicalLayerUp...")
         self._physicalLayer.physicalLayerUp()
-        while canLink.pollState() != CanLink.State.Permitted:
+        self._callback_status("Waiting for alias reservation...")
+        while self._canLink.pollState() != CanLink.State.Permitted:
             precise_sleep(.02)
-
         # ^ triggers fireListeners which calls CanLink's default
         #   receiveListener by default since added on CanPhysicalLayer
         #   arg of linkPhysicalLayer.
         #   - Must happen *after* listen thread starts, since
         #     generates ControlFrame.LinkUp and calls fireListeners
         #     which calls sendAliasConnectionSequence on this thread!
+        self._callback_status("Alias reservation complete.")
 
     def listen(self):
         self._listen_thread = threading.Thread(
@@ -244,19 +247,49 @@ def _listen(self):
             #   the alias from each node (collision or not)
             #   to has to get the expected replies to the alias
             #   reservation sequence below.
-            precise_sleep(.05)   # Wait for physicalLayerUp to complete
+            precise_sleep(.05)   # Wait for physicalLayerUp non-network Message
             while True:
                 # Wait 200 ms for all nodes to announce (and for alias
                 #   reservation to complete), as per section 6.2.1 of CAN
                 #   Frame Transfer Standard (sendMessage requires )
                 logger.debug("[_listen] _receive...")
                 try:
-                    sends = self._physicalLayer.popFrames()
-                    while sends:
+                    # Receive mode (switches to write mode on BlockingIOError
+                    #   which is expected and used on purpose)
+                    # print("Waiting for _receive")
+                    received = self._receive()  # requires setblocking(False)
+                    print("[_listen] received {} byte(s)".format(len(received)),
+                          file=sys.stderr)
+                    # print("      RR: {}".format(received.strip()))
+                    # pass to link processor
+                    self._physicalLayer.handleData(received)
+                    # ^ will trigger self._printFrame if that was added
+                    #   via registerFrameReceivedListener during connect.
+                    precise_sleep(.01)  # let processor sleep before read
+                    if time.perf_counter() - self._connecting_t > .2:
+                        if self._canLink._state != CanLink.State.Permitted:
+                            if ((self._message_t is None)
+                                    or (time.perf_counter() - self._message_t
+                                        > 1)):
+                                logger.warning(
+                                    "CanLink is not ready yet."
+                                    " There must have been a collision"
+                                    "--processCollision increments node alias"
+                                    " in this case and tries again.")
+                        # else _on_link_state_change will be called
+                    # TODO: move *all* send calls to this loop.
+                except BlockingIOError:
+                    # Nothing to receive right now, so perform all sends
+                    #   This *must* occur (require socket.setblocking(False))
+                    # sends = self._physicalLayer.popFrames()
+                    # while sends:
+                    while True:
                         # *Always* do send in the receive thread to
                         #   avoid overlapping calls to socket
                         #   (causes undefined behavior)!
-                        frame = sends.pop()
+                        frame = self._physicalLayer.pollFrame()
+                        if frame is None:
+                            break  # allow receive to run!
                         if isinstance(frame, CanFrame):
                             if self._canLink.isDuplicateAlias(frame.alias):
                                 logger.warning(
@@ -266,43 +299,23 @@ def _listen(self):
                                     .format(frame.alias))
                                 continue
                             logger.debug("[_listen] _sendString...")
-                            self._port.sendString(frame.encodeAsString())
+                            packet = frame.encodeAsString()
+                            assert isinstance(packet, str)
+                            print("Sending {}".format(packet))
+                            self._port.sendString(packet)
                             if frame.afterSendState:
                                 self._canLink.setState(frame.afterSendState)
                         else:
                             raise NotImplementedError(
                                 "Event type {} is not handled."
                                 .format(type(frame).__name__))
-                    received = self._receive()  # requires setblocking(False)
                     #   so that it doesn't block (or occur during) recv
                     #   (overlapping calls would cause undefined behavior)!
-                    # TODO: move *all* send calls to this loop.
-                except BlockingIOError:
                     # delay = random.uniform(.005,.02)
                     # ^ random delay may help if send is on another thread
                     #   (but avoid that for stability and speed)
                     precise_sleep(.01)
-                    continue
-                print("[_listen] received {} byte(s)".format(len(received)),
-                      file=sys.stderr)
-                # print("      RR: {}".format(received.strip()))
-                # pass to link processor
-                self._physicalLayer.handleData(received)
-                # ^ will trigger self._printFrame if that was added
-                #   via registerFrameReceivedListener during connect.
-                precise_sleep(.01)  # let processor sleep briefly before read
-                if time.perf_counter() - self._connecting_t > .2:
-                    if self._canLink._state != CanLink.State.Permitted:
-                        if ((self._message_t is None)
-                                or (time.perf_counter() - self._message_t
-                                    > 1)):
-                            logger.warning(
-                                "CanLink is not ready yet."
-                                " There must have been a collision"
-                                "--processCollision increments node alias"
-                                " in this case and tries again.")
-                    # else _on_link_state_change will be called
-
+            # raise RuntimeError("We should never get here")
         except RuntimeError as ex:
             caught_ex = ex
             # If _port is a TcpSocket:

openlcb/frameencoder.py

@@ -0,0 +1,10 @@
+from typing import Union
+
+
+class FrameEncoder:
+    def encodeFrameAsString(self, frame) -> str:
+        '''Encode frame to string.'''
+        raise NotImplementedError("Implement this in each subclass.")
+
+    def encodeFrameAsData(self, frame) -> Union[bytearray, bytes]:
+        raise NotImplementedError("Implement this in each subclass.")