DRAFT: Delegate state to link and flow to port (do not merge here, but to upstream)

.github/workflows/run-tests.yml

@@ -13,4 +13,4 @@ jobs:
     - uses: actions/checkout@v3
     - name: check tests run
       run: |
-        python3 test_all.py
+        python3 -m unittest discover -s tests -p 'test_*.py'

.gitignore

@@ -183,3 +183,5 @@ cython_debug/
 /build
 /doc/_autosummary
 /examples/settings.json
+/.venv-3.12/
+/cached-cdi.xml

doc/conf.py

@@ -8,6 +8,7 @@
 
 import os
 import sys
+
 sys.path.insert(0, os.path.abspath('..'))
 
 project = 'python-openlcb'

examples/example_cdi_access.py

@@ -11,24 +11,26 @@
                           address and port.
 '''
 # region same code as other examples
+import copy
+# from xml.sax.expatreader import AttributesImpl  # only for IDE autocomplete
 from examples_settings import Settings  # do 1st to fix path if no pip install
+from openlcb import precise_sleep
+from openlcb.tcplink.tcpsocket import TcpSocket
 settings = Settings()
 
 if __name__ == "__main__":
     settings.load_cli_args(docstring=__doc__)
 # endregion same code as other examples
 
-from openlcb.canbus.tcpsocket import TcpSocket
-
-from openlcb.canbus.canphysicallayergridconnect import (
+from openlcb.canbus.canphysicallayergridconnect import (  # noqa:E402
     CanPhysicalLayerGridConnect,
 )
-from openlcb.canbus.canlink import CanLink
-from openlcb.nodeid import NodeID
-from openlcb.datagramservice import (
+from openlcb.canbus.canlink import CanLink  # noqa:E402
+from openlcb.nodeid import NodeID  # noqa:E402
+from openlcb.datagramservice import (  # noqa:E402
     DatagramService,
 )
-from openlcb.memoryservice import (
+from openlcb.memoryservice import (  # noqa:E402
     MemoryReadMemo,
     MemoryService,
 )
@@ -42,18 +44,20 @@
 # farNodeID = "02.01.57.00.04.9C"
 # endregion moved to settings
 
-s = TcpSocket()
+sock = TcpSocket()
 # s.settimeout(30)
-s.connect(settings['host'], settings['port'])
+sock.connect(settings['host'], settings['port'])
 
 
 # print("RR, SR are raw socket interface receive and send;"
 #      " RL, SL are link interface; RM, SM are message interface")
 
 
-def sendToSocket(string):
-    # print("      SR: {}".format(string.strip()))
-    s.send(string)
+# def sendToSocket(frame: CanFrame):
+#     string = frame.encodeAsString()
+#     # print("      SR: {}".format(string.strip()))
+#     sock.sendString(string)
+#     physicalLayer.onFrameSent(frame)
 
 
 def printFrame(frame):
@@ -80,11 +84,10 @@ def printDatagram(memo):
     return False
 
 
-canPhysicalLayerGridConnect = CanPhysicalLayerGridConnect(sendToSocket)
-canPhysicalLayerGridConnect.registerFrameReceivedListener(printFrame)
+physicalLayer = CanPhysicalLayerGridConnect()
+physicalLayer.registerFrameReceivedListener(printFrame)
 
-canLink = CanLink(NodeID(settings['localNodeID']))
-canLink.linkPhysicalLayer(canPhysicalLayerGridConnect)
+canLink = CanLink(physicalLayer, NodeID(settings['localNodeID']))
 canLink.registerMessageReceivedListener(printMessage)
 
 datagramService = DatagramService(canLink)
@@ -100,6 +103,9 @@ def printDatagram(memo):
 
 # callbacks to get results of memory read
 
+complete_data = False
+read_failed = False
+
 
 def memoryReadSuccess(memo):
     """Handle a successful read
@@ -113,6 +119,7 @@ def memoryReadSuccess(memo):
     # print("successful memory read: {}".format(memo.data))
 
     global resultingCDI
+    global complete_data
 
     # is this done?
     if len(memo.data) == 64 and 0 not in memo.data:
@@ -139,12 +146,15 @@ def memoryReadSuccess(memo):
 
         # and process that
         processXML(cdiString)
+        complete_data = True
 
         # done
 
 
 def memoryReadFail(memo):
+    global read_failed
     print("memory read failed: {}".format(memo.data))
+    read_failed = True
 
 
 #######################
@@ -161,54 +171,57 @@ def memoryReadFail(memo):
 
 
 class MyHandler(xml.sax.handler.ContentHandler):
-    """XML SAX callbacks in a handler object"""
-    def __init__(self):
-        self._charBuffer = bytearray()
+    """XML SAX callbacks in a handler object
 
-    def startElement(self, name, attrs):
-        """_summary_
+    Attributes:
+        _chunks (list[str]): Collects chunks of data.
+            This is implementation-specific, and not
+            required if streaming (parser.feed).
+    """
 
-        Args:
-            name (_type_): _description_
-            attrs (_type_): _description_
-        """
+    def __init__(self):
+        self._chunks = []
+
+    def startElement(self, name: str, attrs):
+        """See xml.sax.handler.ContentHandler documentation."""
         print("Start: ", name)
         if attrs is not None and attrs :
             print("  Attributes: ", attrs.getNames())
 
-    def endElement(self, name):
-        """_summary_
-
-        Args:
-            name (_type_): _description_
-        """
+    def endElement(self, name: str):
+        """See xml.sax.handler.ContentHandler documentation."""
         print(name, "content:", self._flushCharBuffer())
         print("End: ", name)
         pass
 
     def _flushCharBuffer(self):
         """Decode the buffer, clear it, and return all content.
+        See xml.sax.handler.ContentHandler documentation.
 
         Returns:
             str: The content of the bytes buffer decoded as utf-8.
         """
-        s = self._charBuffer.decode("utf-8")
-        self._charBuffer.clear()
+        s = ''.join(self._chunks)
+        self._chunks.clear()
         return s
 
-    def characters(self, data):
-        """Received characters handler
+    def characters(self, data: str):
+        """Received characters handler.
+        See xml.sax.handler.ContentHandler documentation.
+
         Args:
             data (Union[bytearray, bytes, list[int]]): any
               data (any type accepted by bytearray extend).
         """
-        self._charBuffer.extend(data)
+        if not isinstance(data, str):
+            raise TypeError("Expected str, got {}".format(type(data).__name__))
+        self._chunks.append(data)
 
 
 handler = MyHandler()
 
 
-def processXML(content) :
+def processXML(content: str) :
     """process the XML and invoke callbacks
 
     Args:
@@ -218,15 +231,37 @@ def processXML(content) :
     #   only called when there is a null terminator, which indicates the
     #   last packet was reached for the requested read.
     #   - See memoryReadSuccess comments for details.
+    with open("cached-cdi.xml", 'w') as stream:
+        # NOTE: Actual caching should key by all SNIP info that could
+        #   affect CDI/FDI: manufacturer, model, and version. Without
+        #   all 3 being present in SNIP, the cache may be incorrect.
+        stream.write(content)
     xml.sax.parseString(content, handler)
     print("\nParser done")
 
 
 #######################
 
 # have the socket layer report up to bring the link layer up and get an alias
-# print("      SL : link up")
-canPhysicalLayerGridConnect.physicalLayerUp()
+print("      QUEUE frames : link up...")
+physicalLayer.physicalLayerUp()
+print("      QUEUED frames : link up...waiting...")
+while canLink.pollState() != CanLink.State.Permitted:
+    # provides incoming data to physicalLayer & sends queued:
+    physicalLayer.receiveAll(sock, verbose=True)
+    physicalLayer.sendAll(sock)
+
+    if canLink.getState() == CanLink.State.WaitForAliases:
+        # physicalLayer.receiveAll(sock, verbose=True)
+        physicalLayer.sendAll(sock)
+        # ^ prevent assertion error below, proceed to send.
+    if canLink.pollState() == CanLink.State.Permitted:
+        break
+    assert canLink.getWaitForAliasResponseStart() is not None, \
+        ("openlcb didn't send the 7,6,5,4 CID frames (state={})"
+         .format(canLink.getState()))
+    precise_sleep(.02)
+print("      SENT frames : link up")
 
 
 def memoryRead():
@@ -236,8 +271,16 @@ def memoryRead():
     to AME
     """
     import time
-    time.sleep(1)
-
+    time.sleep(.21)
+    # ^ 200ms: See section 6.2.1 of CAN Frame Transfer Standard
+    #   (CanLink.State.Permitted will only occur after that, but waiting
+    #   now will reduce output & delays below in this example).
+    while canLink.getState() != CanLink.State.Permitted:
+        print("Waiting for connection sequence to complete...")
+        # This delay could be .2 (per alias collision), but longer to
+        #   reduce console messages:
+        time.sleep(.5)
+    print("Requesting memory read. Please wait...")
     # read 64 bytes from the CDI space starting at address zero
     memMemo = MemoryReadMemo(NodeID(settings['farNodeID']), 64, 0xFF, 0,
                              memoryReadFail, memoryReadSuccess)
@@ -247,10 +290,30 @@ def memoryRead():
 import threading  # noqa E402
 thread = threading.Thread(target=memoryRead)
 thread.start()
-
+previous_nodes = copy.deepcopy(canLink.nodeIdToAlias)
 # process resulting activity
-while True:
-    received = s.receive()
-    # print("      RR: {}".format(received.strip()))
-    # pass to link processor
-    canPhysicalLayerGridConnect.receiveString(received)
+print()
+print("This example will exit on failure or complete data.")
+while not complete_data and not read_failed:
+    # In this example, requests are initiate by the
+    #   memoryRead thread, and receiveAll actually
+    #   receives the data from the requested memory space (CDI in this
+    #   case) and offset (incremental position in the file/data,
+    #   incremented by this example's memoryReadSuccess handler).
+    count = 0
+    count += physicalLayer.receiveAll(sock)
+    count += physicalLayer.sendAll(sock)
+    if canLink.nodeIdToAlias != previous_nodes:
+        print("nodeIdToAlias updated: {}".format(canLink.nodeIdToAlias))
+    if count < 1:
+        precise_sleep(.01)
+    # else skip sleep to avoid latency (port already delayed)
+    if canLink.nodeIdToAlias != previous_nodes:
+        previous_nodes = copy.deepcopy(canLink.nodeIdToAlias)
+
+physicalLayer.physicalLayerDown()
+
+if read_failed:
+    print("Read complete (FAILED)")
+else:
+    print("Read complete (OK)")