Merge pull request bobjacobsen#61 from Hierosoft/utf8-and-docstrings

UTF-8 and bytearray. Add some docstrings

Author: bobjacobsen

doc/Cyrillic-demo-Dmitry-Russian.png

@@ -30,7 +30,6 @@
 exclude_patterns = []
 
 
-
 # -- Options for HTML output -------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
 

doc/Cyrillic-demo-Dmitry.png

@@ -96,18 +96,19 @@ def printDatagram(memo):
 
 
 # accumulate the CDI information
-resultingCDI = []
+resultingCDI = bytearray()
 
 # callbacks to get results of memory read
 
+
 def memoryReadSuccess(memo):
     """Handle a successful read
     Invoked when the memory read successfully returns,
     this queues a new read until the entire CDI has been
     returned.  At that point, it invokes the XML processing below.
 
     Args:
-        memo (_type_): _description_
+        memo (MemoryReadMemo): Successful MemoryReadMemo
     """
     # print("successful memory read: {}".format(memo.data))
 
@@ -121,15 +122,19 @@ def memoryReadSuccess(memo):
         memo.address = memo.address+64
         # and read again
         memoryService.requestMemoryRead(memo)
+        # The last packet is not yet reached, so don't parse (However,
+        #   parser.feed could be called for realtime processing).
     else :
         # and we're done!
         # save content
         resultingCDI += memo.data
         # concert resultingCDI to a string up to 1st zero
         cdiString = ""
-        for x in resultingCDI:
-            if x == 0 : break
-            cdiString += chr(x)
+        null_i = resultingCDI.find(b'\0')
+        terminate_i = len(resultingCDI)
+        if null_i > -1:
+            terminate_i = min(null_i, terminate_i)
+        cdiString = resultingCDI[:terminate_i].decode("utf-8")
         # print (cdiString)
 
         # and process that
@@ -158,25 +163,46 @@ def memoryReadFail(memo):
 class MyHandler(xml.sax.handler.ContentHandler):
     """XML SAX callbacks in a handler object"""
     def __init__(self):
-        self._charBuffer = []
+        self._charBuffer = bytearray()
 
     def startElement(self, name, attrs):
+        """_summary_
+
+        Args:
+            name (_type_): _description_
+            attrs (_type_): _description_
+        """
         print("Start: ", name)
         if attrs is not None and attrs :
             print("  Attributes: ", attrs.getNames())
 
     def endElement(self, name):
+        """_summary_
+
+        Args:
+            name (_type_): _description_
+        """
         print(name, "content:", self._flushCharBuffer())
         print("End: ", name)
         pass
 
     def _flushCharBuffer(self):
-        s = ''.join(self._charBuffer)
-        self._charBuffer = []
+        """Decode the buffer, clear it, and return all content.
+
+        Returns:
+            str: The content of the bytes buffer decoded as utf-8.
+        """
+        s = self._charBuffer.decode("utf-8")
+        self._charBuffer.clear()
         return s
 
     def characters(self, data):
-        self._charBuffer.append(data)
+        """Received characters handler
+        Args:
+            data (Union[bytearray, bytes, list[int]]): any
+              data (any type accepted by bytearray extend).
+        """
+        self._charBuffer.extend(data)
 
 
 handler = MyHandler()
@@ -186,8 +212,12 @@ def processXML(content) :
     """process the XML and invoke callbacks
 
     Args:
-        content (_type_): _description_
+        content (str): Raw XML data
     """
+    # NOTE: The data is complete in this example since processXML is
+    #   only called when there is a null terminator, which indicates the
+    #   last packet was reached for the requested read.
+    #   - See memoryReadSuccess comments for details.
     xml.sax.parseString(content, handler)
     print("\nParser done")
 

doc/conf.py

@@ -52,7 +52,7 @@ def printFrame(frame):
 canPhysicalLayerGridConnect.registerFrameReceivedListener(printFrame)
 
 # send an AME frame with arbitrary alias to provoke response
-frame = CanFrame(ControlFrame.AME.value, 1, [])
+frame = CanFrame(ControlFrame.AME.value, 1, bytearray())
 print("SL: {}".format(frame))
 canPhysicalLayerGridConnect.sendCanFrame(frame)
 

examples/example_cdi_access.py

@@ -129,7 +129,7 @@ def displayOtherNodeIds(message) :
     """Listener to identify connected nodes
 
     Args:
-        message (_type_): _description_
+        message (Message): A response from the network
     """
     print("[displayOtherNodeIds] type(message): {}"
           "".format(type(message).__name__))

examples/example_frame_interface.py

@@ -126,9 +126,9 @@ def result(arg1, arg2=None, arg3=None, result=True) :
     Args:
         arg1: Any value.
         arg2: value to compare to arg1. Defaults to None.
-        arg3: fail if arg1 not equal to arg1; arg3 is then message. Defaults to
-            None.
-        result (bool, optional): _description_. Defaults to True.
+        arg3: fail if arg1 not equal to arg1; arg3 is then message.
+            Defaults to None.
+        result (bool, optional): Expected result. Defaults to True.
 
     Raises:
         ValueError: If only arg1 was provided (undefined behavior--in other

examples/example_node_implementation.py

@@ -1,3 +1,4 @@
+from collections import OrderedDict
 import re
 
 
@@ -15,5 +16,27 @@ def only_hex_pairs(value):
 
 
 def emit_cast(value):
-    """Show type and value, such as for debug output."""
-    return "{}({})".format(type(value).__name__, repr(value))
+    """Get type and value, such as for debug output."""
+    repr_str = repr(value)
+    if repr_str.startswith(type(value).__name__):
+        return repr(value)  # type already included, such as bytearray(...)
+    return "{}({})".format(type(value).__name__, repr_str)
+
+
+def list_type_names(values):
+    """Get the type of several values, such as for debug output.
+    Args:
+        values (Union[list,tuple,dict,OrderedDict]): A collection where
+            each element's type is to be analyzed.
+    Returns:
+        list[str]: A list where each element is a type name. If
+            values argument is dict-like, each element is formatted as
+            "{key}: {type}".
+    """
+    if isinstance(values, (list, tuple)):
+        return [type(value).__name__ for value in values]
+    if isinstance(values, (dict, OrderedDict)):
+        return ["{}: {}".format(k, type(v).__name__) for k, v in values.items()]
+    raise TypeError("list_type_names is only implemented for"
+                    " list, tuple, dict, and OrderedDict, but got a(n) {}"
+                    .format(type(values).__name__))

examples/example_remote_nodes.py

@@ -1,3 +1,5 @@
+import openlcb
+from collections import OrderedDict
 from openlcb.nodeid import NodeID
 
 
@@ -36,41 +38,98 @@ class CanFrame:
             mask = 0x07FF_F000).
     """
 
-    header = 0
-    data = []
+    ARG_LISTS = [
+        OrderedDict(N_cid=int, nodeID=NodeID, alias=int),
+        OrderedDict(header=int, data=bytearray),
+        OrderedDict(control=int, alias=int, data=bytearray),
+    ]
 
-    def __str__(self):
-        return "CanFrame header: 0x{:08X} {}".format(self.header, self.data)
+    @staticmethod
+    def constructor_help():
+        result = ""
+        for d in CanFrame.ARG_LISTS:
+            result += "("
+            for k, v in d.items():
+                result += "{}: {}, ".format(k, v.__name__)
+            result = result[:-2]  # remove last ", " from this ctor
+            result += "), "
+        return result[:-2]  # -1 to remove last ", " from list
 
-    # there are three ctor forms
+    def __str__(self):
+        return "CanFrame header: 0x{:08X} {}".format(
+            self.header,
+            list(self.data),  # cast to list to format bytearray(b'') as []
+        )
 
-    def __init__(self, arg1, arg2, arg3=[]):
+    def __init__(self, *args):
+        arg1 = None
+        arg2 = None
+        arg3 = None
+        if len(args) > 0:
+            arg1 = args[0]
+        if len(args) > 1:
+            arg2 = args[1]
+        if len(args) > 2:
+            arg3 = args[2]
+        else:
+            arg3 = bytearray()
+        # There are three ctor forms.
+        # - See "Args" in class for docstring.
+        self.header = 0
+        self.data = bytearray()
         # three arguments as N_cid, nodeID, alias
+        args_error = None
         if isinstance(arg2, NodeID):
+            # Other args' types will be enforced by doing math on them
+            #   (duck typing) in this case.
+            if len(args) < 3:
+                args_error = "Expected alias after NodeID"
             # cid must be 4 to 7 inclusive (100 to 111 binary)
             # precondition(4 <= cid && cid <= 7)
             cid = arg1
             nodeID = arg2
             alias = arg3
 
             nodeCode = ((nodeID.nodeId >> ((cid-4)*12)) & 0xFFF)
-            # ^ cid-4 results in 0 to 3. *12 results in 0 to 36 bit shift (nodeID size)
+            # ^ 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 | (alias & 0xFFF) | 0x10_00_00_00  # noqa: E501
-            self.data = []
+            # self.data = bytearray()
 
         # two arguments as header, data
-        elif isinstance(arg2, list):
+        elif isinstance(arg2, bytearray):
+            if not isinstance(arg1, int):
+                args_error = "Expected int since 2nd argument is bytearray."
+            # Types of both args are enforced by this point.
             self.header = arg1
             self.data = arg2
+            if len(args) > 2:
+                args_error = "2nd argument is data, but got extra argument(s)"
+
+        # two arguments as header, data
+        elif isinstance(arg2, list):
+            args_error = ("Expected bytearray (formerly list[int])"
+                          " if data 2nd argument")
+            # self.header = arg1
+            # self.data = bytearray(arg2)
 
         # 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
             alias = arg2
             self.header = (control << 12) | (alias & 0xFFF) | 0x10_00_00_00
+            if not isinstance(arg3, bytearray):
+                args_error = ("Expected bytearray (formerly list[int])"
+                              " 2nd if 1st argument is header int")
             self.data = arg3
         else:
-            print("could not decode NodeID ctor arguments")
+            args_error = "could not decode CanFrame arguments"
+
+        if args_error:
+            raise TypeError(
+                args_error.rstrip(".") + ". Valid constructors:"
+                + CanFrame.constructor_help() + ". Got: "
+                + openlcb.list_type_names(args))
 
     def __eq__(self, other):
         if other is None: