Merge pull request #39 from Overdrivr/transport-monitoring

Health monitoring

Author: Overdrivr

.travis.yml

@@ -34,4 +34,4 @@ install:
 
 # command to run tests
 script:
-  - py.test --cov
+  - py.test -v --cov

README.md

@@ -93,6 +93,15 @@ Plots <topic> in a graph window.
 Usage: plot <topic>
 ```
 
+### stats
+```bash
+Displays different metrics about the active transport (ex : serial port).
+This allows you to know if for instance corrupted frames are received, what fraction
+of the maximum baudrate is being used, etc.
+
+Usage: stats
+```
+
 ### disconnect
 ```bash
 Disconnects from any open connection.

appveyor.yml

@@ -43,4 +43,4 @@ install:
   - "cd %APPVEYOR_BUILD_FOLDER%"
   - "SET PYTHONPATH=%PYTHONPATH%;%APPVEYOR_BUILD_FOLDER%"
 test_script:
-  - "%MINICONDA%\\Scripts\\py.test --cov"
+  - "%MINICONDA%\\Scripts\\py.test -v --cov"

pytelemetrycli/cli.py

@@ -57,6 +57,7 @@ def __init__(self, transport=None, stdout=None):
         else:
             self.transport = transport
         self.telemetry = Pytelemetry(self.transport)
+
         self.topics = Topics()
         self.plots = []
         self.plotsLock = Lock()
@@ -65,6 +66,7 @@ def __init__(self, transport=None, stdout=None):
                              self.plots,
                              self.plotsLock,
                              self.topics)
+
         self.telemetry.subscribe(None,self.topics.process)
 
         self.types_lookup = {'--s'    :  'string',
@@ -115,6 +117,10 @@ def do_serial(self, arg):
             logger.info(s)
             self.topics.clear()
             logger.info("Cleared all topics for new session.")
+            self.transport.resetStats()
+            self.runner.resetStats()
+            self.telemetry.resetStats()
+            logger.info("Cleared all stats for new session.")
 
     @docopt_cmd
     def do_print(self, arg):
@@ -146,7 +152,7 @@ def do_print(self, arg):
 
         if s is not None:
             for i in s:
-                self.stdout.write("%i\n" % i)
+                self.stdout.write("{0}\n".format(i))
         else:
             logger.error("Could not retrieve {0} sample(s) under topic '{1}'.\n".format(amount,topic))
 
@@ -274,13 +280,35 @@ def do_disconnect(self, arg):
             self.runner.disconnect()
             self.stdout.write("Disconnected.\n")
             logger.info("Disconnected.")
+
+            measures = self.transport.stats()
+
+            for key,item in measures.items():
+                logger.info("Raw IO : %s : %s" % (key,item))
+
+            measures = self.runner.stats()
+
+            for key,item in measures.items():
+                logger.info("IO speeds : %s : %s" % (key,item))
+
+            measures = self.telemetry.stats()
+
+            for key,item in measures['framing'].items():
+                logger.info("Framing : %s : %s" % (key,item))
+
+            for key,item in measures['protocol'].items():
+                logger.info("Protocol : %s : %s" % (key,item))
+
+            logger.info("Logged session statistics.")
+
+
         except:
             logger.warn("Already disconnected. Continuing happily.")
 
     @docopt_cmd
     def do_info(self, arg):
         """
-Disconnects from any open connection.
+Prints out cli.py full path, module version.
 
 Usage: info
         """
@@ -290,13 +318,45 @@ def do_info(self, arg):
         except:
             self.stdout.write("- version : not found.\n")
 
+    @docopt_cmd
+    def do_stats(self, arg):
+        """
+Displays different metrics about the active transport (ex : serial port).
+This allows you to know if for instance corrupted frames are received, what fraction
+of the maximum baudrate is being used, etc.
+
+Usage: stats
+        """
+        measures = self.transport.stats()
+
+        self.stdout.write("Raw IO:\n")
+        for key,item in measures.items():
+            self.stdout.write("\t%s : %s\n" % (key,item))
+
+        measures = self.runner.stats()
+
+        self.stdout.write("IO speeds:\n")
+        for key,item in measures.items():
+            self.stdout.write("\t%s : %s\n" % (key,item))
+
+        measures = self.telemetry.stats()
+
+        self.stdout.write("Framing:\n")
+        for key,item in measures['framing'].items():
+            self.stdout.write("\t%s : %s\n" % (key,item))
+
+        self.stdout.write("Protocol:\n")
+        for key,item in measures['protocol'].items():
+            self.stdout.write("\t%s : %s\n" % (key,item))
+
     def do_quit(self, arg):
         """
 Exits the terminal application.
 
 Usage: quit
         """
         self.runner.terminate()
+        self.do_disconnect("")
         self.stdout.write("Good Bye!\n")
         logger.info("Application quit.")
         exit()

pytelemetrycli/runner.py

@@ -18,10 +18,13 @@ def __init__(self, transport, telemetry, plots, plotsLock, topics):
         self.connected = threading.Event()
         self.connected.clear()
 
+        self.resetStats()
+
     def connect(self,port,bauds):
         options = dict()
         options['port'] = port
         options['baudrate'] = bauds
+        self.baudrate = bauds
         self.transport.connect(options)
         self.connected.set()
         self.thread = threading.Thread(target=self.run)
@@ -40,6 +43,21 @@ def terminate(self):
         except:
             pass # Already disconnected
 
+    def stats(self):
+        return {
+            "baudspeed" : self.baudspeed,
+            "baudspeed_avg" : self.baudspeed_avg,
+            "baudratio" : self.baudspeed / self.baudrate,
+            "baudratio_avg" : self.baudspeed_avg / self.baudrate
+        }
+
+    def resetStats(self):
+        self.baudrate = 1.0
+        self.baudspeed = 0.0
+        self.lasttime = time.time()
+        self.lastamount = 0.0
+        self.baudspeed_avg = 0.0
+
     def update(self):
         # Update protocol decoding
         self.telemetryWrapper.update()
@@ -48,6 +66,26 @@ def update(self):
         # being modified from the main thread
         self.plotsLock.acquire()
 
+        # Update baudspeed value
+        current = time.time()
+        difft = current - self.lasttime
+
+        if difft > 0.05 :
+            self.lasttime = current
+
+            current = self.transport.stats()['rx_bytes']
+            diff = current - self.lastamount
+            self.lastamount = current
+
+            self.baudspeed = diff / difft
+
+            # Compute rolling average baud speed on about 1 second window
+            n = 20
+            self.baudspeed_avg = (self.baudspeed + n * self.baudspeed_avg) / (n + 1)
+            # Need a dedicated flag with ls to display program parameters
+            #self.topics.process("baudspeed",self.baudspeed)
+            #self.topics.process("baudspeed_avg",self.baudspeed_avg)
+
         # Poll each poll pipe to see if user closed them
         plotToDelete = None
         for p, i in zip(self.plots,range(len(self.plots))):

pytelemetrycli/test/test_cli.py

@@ -11,6 +11,7 @@ class TransportMock:
     def __init__(self):
         self.q = queue.Queue()
         self.canConnect = False
+        self.counter = 0
     def authorizeConnect(self,value):
         self.canConnect = value
     def connect(self,options):
@@ -20,6 +21,7 @@ def disconnect(self):
         pass
     def read(self, maxbytes=1):
         amount = maxbytes if self.q.qsize() > maxbytes else self.q.qsize()
+        self.counter += amount
         data = []
         for i in range(amount):
             data.append(self.q.get())
@@ -31,6 +33,12 @@ def write(self, data):
             self.q.put(c)
     def writeable(self):
         return True
+    def resetStats(self):
+        self.counter = 0
+    def stats(self):
+        return {
+        "rx_bytes": self.counter
+        }
 
 # To be done
 class SuperplotMock:
@@ -122,6 +130,12 @@ def test_print(fixturefortests):
 
     clear(outstream)
 
+    tlm.onecmd("pub --s hello world")
+    tlm.runner.update()
+    assert outstream.getvalue() == "Published on topic 'hello' : world [string]\n"
+
+    clear(outstream)
+
     tlm.onecmd("pub --i32 foo 4")
     tlm.runner.update()
     assert outstream.getvalue() == "Published on topic 'foo' : 4 [int32]\n"
@@ -164,6 +178,12 @@ def test_print(fixturefortests):
 
     clear(outstream)
 
+    tlm.onecmd("print hello")
+    tlm.runner.update()
+    assert outstream.getvalue() == "world\n"
+
+    clear(outstream)
+
     tlm.onecmd("print foo -a 2.3")
     tlm.runner.update()
     assert outstream.getvalue() == "Could not cast --amount = '2.3' to integer. Using 1.\n4\n"
@@ -228,7 +248,7 @@ def test_disconnect_quit(fixturefortests):
     clear(outstream)
 
     pytest.raises(SystemExit, tlm.onecmd, "quit")
-    assert outstream.getvalue() == "Good Bye!\n"
+    assert outstream.getvalue() == "Disconnected.\nGood Bye!\n"
 
     clear(outstream)
 
@@ -311,3 +331,107 @@ def test_topics_are_cleared_after_reconnect(fixturefortests):
     assert outstream.getvalue() == ""
 
     clear(outstream)
+
+
+def test_stats(fixturefortests):
+    tr, outstream, tlm = fixturefortests
+    clear(outstream)
+    tlm.topics.clear() # Clear all topics
+
+    tr.resetStats()
+    tlm.runner.resetStats()
+    tlm.telemetry.resetStats()
+    tlm.onecmd("stats")
+
+    assert "Raw IO:\n" in outstream.getvalue()
+    assert "\trx_bytes : 0\n" in outstream.getvalue()
+    assert "IO speeds:\n" in outstream.getvalue()
+    assert "\tbaudspeed : 0.0\n" in outstream.getvalue()
+    assert "\tbaudratio : 0.0\n" in outstream.getvalue()
+    assert "\tbaudratio_avg : 0.0\n" in outstream.getvalue()
+    assert "\tbaudspeed_avg : 0.0\n" in outstream.getvalue()
+    assert "Framing:\n" in outstream.getvalue()
+    assert "\ttx_encoded_frames : 0\n" in outstream.getvalue()
+    assert "\trx_uncomplete_frames : 0\n" in outstream.getvalue()
+    assert "\ttx_processed_bytes : 0\n" in outstream.getvalue()
+    assert "\trx_complete_frames : 0\n" in outstream.getvalue()
+    assert "\ttx_escaped_bytes : 0\n" in outstream.getvalue()
+    assert "\trx_discarded_bytes : 0\n" in outstream.getvalue()
+    assert "\trx_processed_bytes : 0\n" in outstream.getvalue()
+    assert "\trx_escaped_bytes : 0\n" in outstream.getvalue()
+    assert "Protocol:\n" in outstream.getvalue()
+    assert "\ttx_encoded_frames : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_header : 0\n" in outstream.getvalue()
+    assert "\trx_decoded_frames : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_payload : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_crc : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_eol : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_topic : 0\n" in outstream.getvalue()
+
+    tlm.onecmd("pub --i32 foo 2")
+
+    clear(outstream)
+
+    tlm.runner.update()
+
+    tlm.onecmd("stats")
+
+    speeds = tlm.runner.stats()
+
+    assert "Raw IO:\n" in outstream.getvalue()
+    assert "\trx_bytes : 14\n" in outstream.getvalue()
+    assert "IO speeds:\n" in outstream.getvalue()
+    assert "\tbaudspeed : {0}\n".format(speeds['baudspeed']) in outstream.getvalue()
+    assert "\tbaudratio : {0}\n".format(speeds['baudratio']) in outstream.getvalue()
+    assert "\tbaudratio_avg : {0}\n".format(speeds['baudratio_avg']) in outstream.getvalue()
+    assert "\tbaudspeed_avg : {0}\n".format(speeds['baudspeed_avg']) in outstream.getvalue()
+    assert "Framing:\n" in outstream.getvalue()
+    assert "\ttx_encoded_frames : 1\n" in outstream.getvalue()
+    assert "\trx_uncomplete_frames : 0\n" in outstream.getvalue()
+    assert "\ttx_processed_bytes : 12\n" in outstream.getvalue()
+    assert "\trx_complete_frames : 1\n" in outstream.getvalue()
+    assert "\ttx_escaped_bytes : 0\n" in outstream.getvalue()
+    assert "\trx_discarded_bytes : 0\n" in outstream.getvalue()
+    assert "\trx_processed_bytes : 14\n" in outstream.getvalue()
+    assert "\trx_escaped_bytes : 0\n" in outstream.getvalue()
+    assert "Protocol:\n" in outstream.getvalue()
+    assert "\ttx_encoded_frames : 1\n" in outstream.getvalue()
+    assert "\trx_corrupted_header : 0\n" in outstream.getvalue()
+    assert "\trx_decoded_frames : 1\n" in outstream.getvalue()
+    assert "\trx_corrupted_payload : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_crc : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_eol : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_topic : 0\n" in outstream.getvalue()
+
+    # Check stats are cleaned after restart
+    tr.authorizeConnect(True)
+    tlm.onecmd("serial com123")
+
+    clear(outstream)
+
+    tlm.onecmd("stats")
+
+    assert "Raw IO:\n" in outstream.getvalue()
+    assert "\trx_bytes : 0\n" in outstream.getvalue()
+    assert "IO speeds:\n" in outstream.getvalue()
+    assert "\tbaudspeed : 0.0\n" in outstream.getvalue()
+    assert "\tbaudratio : 0.0\n" in outstream.getvalue()
+    assert "\tbaudratio_avg : 0.0\n" in outstream.getvalue()
+    assert "\tbaudspeed_avg : 0.0\n" in outstream.getvalue()
+    assert "Framing:\n" in outstream.getvalue()
+    assert "\ttx_encoded_frames : 0\n" in outstream.getvalue()
+    assert "\trx_uncomplete_frames : 0\n" in outstream.getvalue()
+    assert "\ttx_processed_bytes : 0\n" in outstream.getvalue()
+    assert "\trx_complete_frames : 0\n" in outstream.getvalue()
+    assert "\ttx_escaped_bytes : 0\n" in outstream.getvalue()
+    assert "\trx_discarded_bytes : 0\n" in outstream.getvalue()
+    assert "\trx_processed_bytes : 0\n" in outstream.getvalue()
+    assert "\trx_escaped_bytes : 0\n" in outstream.getvalue()
+    assert "Protocol:\n" in outstream.getvalue()
+    assert "\ttx_encoded_frames : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_header : 0\n" in outstream.getvalue()
+    assert "\trx_decoded_frames : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_payload : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_crc : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_eol : 0\n" in outstream.getvalue()
+    assert "\trx_corrupted_topic : 0\n" in outstream.getvalue()