Read-Sensor-Resistances/serial_plotter.py

from datetime import datetime
from matplotlib.animation import FuncAnimation
import os, json, traceback, wx
import numpy as np
import matplotlib
import matplotlib.pyplot as plt
import decimal
from itertools import islice
matplotlib.use("WXAgg")  # for JetBrains IDE to force use wxPython as backend UI for plotting


class SerialPlotter:
    def __init__(self, parent: wx.Frame = None) -> None:
        """
        Dynamically plot a graph with moving window, if a finite window size set by the user. It reads the .csv file
        read_arduino generated, use the last line as the latest data and update the graph accordingly. It will plot
        multiple lines on the graph if there are more than one values on the last row of the .csv file
        :param parent: optional, parent frame
        """
        self.parent = parent
        self.settings = json.load(open('settings.json', 'r'))
        self.sensors = len(self.settings['sensor_ports'])
        self.windowsize = self.settings['winSize']
        self.delay = self.settings["delay"] / 1000

        self.colors = ['blue', 'orange', 'green', 'yellow']

        # TODO: make the figure size an UI option and pass into the settings.json
        self.fig, self.axs = plt.subplots(1, 1, figsize=(9, 6))

        self.fig.canvas.mpl_connect('close_event', self.event_close)

        self.timeStamps = {}
        self.sensorsData = {}
        self.timeElapsed = 0  # time have passed since the graph started, in seconds

        for i in range(self.sensors):
            self.timeStamps[i] = ['']
            self.sensorsData[i] = [0]

    def event_close(self, event) -> None:
        """
        Actiions need to be executed when the graph has closed. Start a new .csv file to get read for new graph and
        bring back the UI, if exist
        """
        file = self.settings["file_name"]
        wx.MessageBox(f"File has saved as {os.path.split(file)[1]} under {os.path.split(file)[0]} directory!\n")
        if self.parent:
            self.parent.Show()

    def animation(self, t: int) -> None:
        """
        render a frame of the animated graph
        """
        try:
            plt.cla()  # clear previous frame
            # read the last line from the .csv file, the data start from the second column so omit index #0
            file = open(self.settings["file_name"], "r")
            ndata = np.array([np.asarray(line.split(", ")[1:], dtype=np.float32) for line in islice(file, 1, None)])#read from the second row of the file ignoring headers
            if len(ndata) > 0:
                row = ndata[-1]
                i = 0
                while i < self.sensors:
                    # shift all data left by 1 index, pop out the leftmost value, if windowsize is not 0 (infinite)
                    # TODO: make sure the two lists have the same size
                    # if self.windowsize > 0 and len(self.timeStamps[i]) > self.windowsize
                    if 0 < self.windowsize < len(self.timeStamps[i]):
                        self.timeStamps[i].pop(0)
                        self.sensorsData[i].pop(0)

                    # self.timeStamps[i].append(datetime.now().strftime('%H:%M:%S'))  # version 1

                    # version 2, if we decide to go with this one change the list type to list[int] from list[str]
                    self.timeStamps[i].append(str(self.timeElapsed))

                    self.sensorsData[i].append(row[i])
                    # plot a line
                    # round the number to scientific notation
                    self.axs.plot(self.timeStamps[i],self.sensorsData[i], color=self.colors[i],
                                  label=f'sensor {i + 1}, latest: {np.format_float_scientific(self.sensorsData[i][-1], precision = 2)} $\Omega$')
                    
                    self.axs.set_xlabel('Time (seconds)')
                    self.axs.set_ylabel(u'Resistance ($\Omega$)')
                    i += 1
            self.timeElapsed += 1  # increment time

            # Acknowledgement: https://stackoverflow.com/a/13589144
            handles, labels = self.axs.get_legend_handles_labels()
            by_label = dict(zip(labels, handles))
            self.axs.legend(by_label.values(), by_label.keys(), loc='best')
        except:
            traceback.print_exc()

    def plotting(self) -> FuncAnimation:
        """ animate the dynamic plot """
        ani = FuncAnimation(self.fig, self.animation, blit=False, interval=self.delay * 1000)
        return ani