98 lines
4.4 KiB
Python
98 lines
4.4 KiB
Python
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
|