improved examples

Author: Marc Hofmann

.vscode/launch.json

@@ -9,7 +9,7 @@
       "type": "debugpy",
       "request": "launch",
       // "program": "src/__main__.py",
-      "module": "examples.low_level.example_low_level_plot",
+      "module": "examples.dyscom.example_dyscom_fastplotlib",
       "justMyCode": false,
       // "args": ["COM3"],
       "console": "integratedTerminal"

examples/dyscom/example_dyscom_fastplotlib.py

@@ -62,14 +62,14 @@ async def device_communication() -> int:
         await dyscom.start()
 
         # loop for some time
-        for x in range(1000):
+        for x in range(5000):
             # check if we closed window
             if not is_window_open:
                 break
 
             # check operation mode from time to time, this function is not waiting for response
             # so we have to handle it by ourself later
-            if x % 100 == 0:
+            if x % 500 == 0:
                 dyscom.send_get_operation_mode()
 
             live_data_counter = 0

examples/dyscom/example_dyscom_pyplot.py

@@ -49,9 +49,10 @@ async def main() -> int:
     # start dyscom measurement
     await dyscom.start()
 
-    for x in range(1000):
+    # loop for some time
+    for x in range(5000):
         # check operation mode from time to time
-        if x % 100 == 0:
+        if x % 500 == 0:
             dyscom.send_get_operation_mode()
 
         live_data_counter = 0
@@ -70,7 +71,7 @@ async def main() -> int:
                         break
 
                     # reduce framerate further
-                    if sld.number % 60 == 0:
+                    if sld.number % 10 == 0:
                         plot_helper.append_value(0, sld.samples[0].value)
                         plot_helper.update()
 
@@ -88,6 +89,8 @@ async def main() -> int:
     # close serial port connection
     connection.close()
 
+    print("Close plot window to quit")
+    plot_helper.loop()
     return 0
 
 

examples/dyscom/example_dyscom_write_csv.py

@@ -64,10 +64,10 @@ async def device_communication() -> int:
         total_count = 0
 
         # loop for some time
-        for x in range(1000):
+        for x in range(5000):
             # check operation mode from time to time, this function is not waiting for response
             # so we have to handle it by ourself later
-            if x % 100 == 0:
+            if x % 500 == 0:
                 dyscom.send_get_operation_mode()
 
             live_data_counter = 0

examples/low_level/example_low_level_plot.py

@@ -37,7 +37,7 @@ def get_channel_color(channel: Channel) -> str:
     """Retrieves color from channel"""
     color = channel.name
     if color == "WHITE":
-        color = "PINK"
+        color = "PURPLE"
     return color
 
 
@@ -48,7 +48,7 @@ async def main() -> int:
     for connector in Connector:
         for channel in Channel:
             plots_info[calc_plot_index(connector, channel)] = f"Connector {connector.name}, channel {channel.name}", get_channel_color(channel)
-    plot_helper = PyPlotHelper(plots_info, 2500)
+    plot_helper = PyPlotHelper(plots_info, 500)
 
     # get comport from command line argument
     com_port = ExampleUtils.get_comport_from_commandline_argument()
@@ -74,27 +74,31 @@ async def main() -> int:
         send_channel_config(low_level_layer)
 
         # wait for stimulation to happen
-        await asyncio.sleep(0.25)
+        await asyncio.sleep(1.0)
 
         # process all acknowledges and append values to plot data
         while True:
             ack = low_level_layer.packet_buffer.get_packet_from_buffer()
             if ack:
                 if ack.command == Commands.LOW_LEVEL_CHANNEL_CONFIG_ACK:
                     ll_config_ack: PacketLowLevelChannelConfigAck = ack
+                    # update plot with measured values
                     plot_helper.append_values(calc_plot_index(ll_config_ack.connector, ll_config_ack.channel),
                                               ll_config_ack.measurement_samples)
+                    plot_helper.update()
             else:
                 break
 
+        await asyncio.sleep(0.1)
+
     # call stop low level
     await low_level_layer.stop()
-    # update plot with measured values
-    plot_helper.update()
 
     # close serial port connection
     connection.close()
 
+    print("Close plot window to quit")
+    plot_helper.loop()
     return 0
 
 

examples/utils/example_utils.py

@@ -3,6 +3,7 @@
 import sys
 import threading
 import os
+import asyncio
 from typing import Callable
 from getch import getch
 
@@ -56,3 +57,21 @@ def get_comport_from_commandline_argument() -> str:
 
         com_port = sys.argv[1]
         return com_port
+
+
+    @staticmethod
+    async def wait_for_any_key_pressed(cb: Callable[[], None] | None):
+        # keyboard func
+        def input_callback(input_value: str) -> bool:
+            """Callback call from keyboard input thread"""
+            # quit on any key
+            return True
+
+        # create keyboard input thread for non blocking console input
+        keyboard_input_thread = KeyboardInputThread(input_callback)
+
+        # now we can start stimulation
+        while keyboard_input_thread.is_alive():
+            if cb is not None:
+                cb()
+            await asyncio.sleep(0.1)

examples/utils/plot_base.py

@@ -65,6 +65,11 @@ def update(self):
         """Update plot"""
 
 
+    def loop(self):
+        """Run event loop until plot window closed"""
+        pass
+
+
     def _calc_layout_dimension(self, channel_count: int) -> tuple[int, int]:
         """Calculates layout for a specific number of channels, tries to grow
         equal in both directions"""

examples/utils/pyplot_utils.py

@@ -71,25 +71,39 @@ class PyPlotHelper(PlotHelper):
     def __init__(self, channels: dict[int, tuple[str, str]], max_value_count: int):
         super().__init__()
 
+        self._window_closed = False
         x_dimension, y_dimension = self._calc_layout_dimension(len(channels))
         self._figure, self._axes = plt.subplots(y_dimension, x_dimension, constrained_layout=True, squeeze=False)
+        self._figure.canvas.mpl_connect("close_event", self._on_close)
 
         for (key, value), sub_plot in zip(channels.items(), self._axes.flat):
             sub_plot.set(xlabel="Samples", ylabel=value[0], title=value[0])
             self._data[key] = PyPlotValueChannel(sub_plot, max_value_count, value[1])
 
         # interactive mode and show plot
-        self._animation_result = animation.FuncAnimation(self._figure, self._animation, interval=100)
-        # plt.ion()
+        self._animation_result = animation.FuncAnimation(self._figure, self._animation, 
+                                                         interval=100, save_count=max_value_count)
+        plt.ion()
         plt.show(block=False)
+        self.update()
 
 
     def update(self):
         """Wait for a short time to process events"""
         plt.pause(0.0001)
 
 
+    def loop(self):
+        """Run event loop until plot window closed"""
+        while not self._window_closed:
+            self.update()
+
+
     def _animation(self, frame: int, *fargs: tuple): # pylint:disable=unused-argument
         """This function is call in context of main thread"""
         for x in self._data.values():
             x.update_plot()
+
+
+    def _on_close(self, event):
+        self._window_closed = True