pubsub works

TODO.md

@@ -2,12 +2,17 @@
 
 ## Before initial release
 
-- make pubsub work
 - make docs decent
 - document properties/fields
 - include examples in readme. hello.py, textchat.py, replymessage.py
 - DONE use port enumeration to find ports https://pyserial.readthedocs.io/en/latest/shortintro.html
 - DONE make serial debug output optional (by providing a null stream)
+- DONE make pubsub work
+
+## Soon after initial release
+
+- keep nodedb up-to-date based on received MeshPackets
+- handle radio reboots and redownload db when that happens. Look for a special FromRadio.rebooted packet
 
 ## Eventual
 

bin/run.sh

@@ -1 +1 @@
-python3 -m meshtastic "$@"
+python3 -m meshtastic "$@" --seriallog none 

meshtastic/__main__.py

@@ -4,6 +4,30 @@ import argparse
 from .interface import StreamInterface
 import logging
 import sys
+from pubsub import pub
+import google.protobuf.json_format
+
+
+def onReceive(packet):
+    """Callback invoked when a packet arrives"""
+    print(f"Received: {packet}")
+
+
+def onConnection(topic=pub.AUTO_TOPIC):
+    """Callback invoked when we connect/disconnect from a radio"""
+    print(f"Connection changed: {topic.getName()}")
+
+
+def onNode(node):
+    """Callback invoked when the node DB changes"""
+    print(f"Node changed: {node}")
+
+
+def subscribe():
+    """Subscribe to the topics the user probably wants to see, prints output to stdout"""
+    pub.subscribe(onReceive, "meshtastic.receive")
+    pub.subscribe(onConnection, "meshtastic.connection")
+    pub.subscribe(onNode, "meshtastic.node")
 
 
 def main():
@@ -35,6 +59,7 @@ def main():
         logging.info(f"Logging serial output to {args.seriallog}")
         logfile = open(args.seriallog, 'w+', buffering=1)  # line buffering
 
+    subscribe()
     client = StreamInterface(args.device, debugOut=logfile)
 
 

meshtastic/interface.py

@@ -7,6 +7,7 @@ import logging
 import sys
 import traceback
 from . import mesh_pb2
+from pubsub import pub
 
 START1 = 0x94
 START2 = 0xc3
@@ -34,12 +35,12 @@ Use a pubsub model to communicate events [https://pypubsub.readthedocs.io/en/v4.
 
 - meshtastic.connection.established - published once we've successfully connected to the radio and downloaded the node DB
 - meshtastic.connection.lost - published once we've lost our link to the radio
-- meshtastic.receive.position(MeshPacket)
-- meshtastic.receive.user(MeshPacket)
-- meshtastic.receive.data(MeshPacket)
-- meshtastic.node.updated(NodeInfo) - published when a node in the DB changes (appears, location changed, username changed, etc...)
-- meshtastic.debug(string)
-- meshtastic.send(MeshPacket) - Not yet implemented, instead call sendPacket(...) on MeshInterface
+- meshtastic.receive.position(packet = MeshPacket)
+- meshtastic.receive.user(packet = MeshPacket)
+- meshtastic.receive.data(packet = MeshPacket)
+- meshtastic.node.updated(node = NodeInfo) - published when a node in the DB changes (appears, location changed, username changed, etc...)
+- meshtastic.debug(message = string)
+- meshtastic.send(packet = MeshPacket) - Not yet implemented, instead call sendPacket(...) on MeshInterface
 
 """
 
@@ -76,6 +77,10 @@ class MeshInterface:
         toRadio.packet.CopyFrom(meshPacket)
         self._sendToRadio(toRadio)
 
+    def _disconnected(self):
+        """Called by subclasses to tell clients this interface has disconnected"""
+        pub.sendMessage("meshtastic.connection.lost")
+
     def _startConfig(self):
         """Start device packets flowing"""
         self.myInfo = None
@@ -102,7 +107,7 @@ class MeshInterface:
         logging.debug(f"Received: {json}")
         if fromRadio.HasField("my_info"):
             self.myInfo = fromRadio.my_info
-        if fromRadio.HasField("radio"):
+        elif fromRadio.HasField("radio"):
             self.radioConfig = fromRadio.radio
         elif fromRadio.HasField("node_info"):
             node = fromRadio.node_info
@@ -110,10 +115,31 @@ class MeshInterface:
             self.nodes[node.user.id] = node
         elif fromRadio.config_complete_id == MY_CONFIG_ID:
             # we ignore the config_complete_id, it is unneeded for our stream API fromRadio.config_complete_id
-            pass
+            pub.sendMessage("meshtastic.connection.established")
+        elif fromRadio.HasField("packet"):
+            self._handlePacketFromRadio(self, fromRadio.packet)
         else:
             logging.warn("Unexpected FromRadio payload")
 
+    def _handlePacketFromRadio(self, meshPacket):
+        """Handle a MeshPacket that just arrived from the radio
+
+        Will publish one of the following events:
+        - meshtastic.receive.position(packet = MeshPacket dictionary)
+        - meshtastic.receive.user(packet = MeshPacket dictionary)
+        - meshtastic.receive.data(packet = MeshPacket dictionary)
+        """
+        # FIXME, update node DB as needed
+        json = google.protobuf.json_format.MessageToDict(meshPacket)
+        if meshPacket.payload.HasField("position"):
+            pub.sendMessage("meshtastic.receive.position", packet=json)
+        if meshPacket.payload.HasField("user"):
+            pub.sendMessage("meshtastic.receive.user",
+                            packet=json)
+        if meshPacket.payload.HasField("data"):
+            pub.sendMessage("meshtastic.receive.data",
+                            packet=json)
+
 
 class StreamInterface(MeshInterface):
     """Interface class for meshtastic devices over a stream link(serial, TCP, etc)"""
@@ -201,3 +227,4 @@ class StreamInterface(MeshInterface):
                         self._rxBuf = empty
         logging.debug("reader is exiting")
         self.stream.close()
+        self._disconnected()

setup.py

@@ -25,7 +25,7 @@ setup(
     ],
     packages=["meshtastic"],
     include_package_data=True,
-    install_requires=["pyserial>=3.4"],
+    install_requires=["pyserial>=3.4", "protobuf>=3.6.1", "pypubsub>=4.0.3"],
     python_requires='>=3',
     entry_points={
         "console_scripts": [