smartwatch.py

#Reference for Threading and GUI architecture: https://maldus512.medium.com/how-to-setup-correctly-an-application-with-python-and-tkinter-107c6bc5a45
#Reference for function animate https://towardsdatascience.com/plotting-live-data-with-matplotlib-d871fac7500b
import queue
import tkinter as tk
import matplotlib.pyplot as plt
import numpy as np
import collections

from tkinter import TOP, BOTH, X, LEFT, RIGHT
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
from matplotlib.animation import FuncAnimation
from types import SimpleNamespace
from enum import Enum

from threading import Thread

from hardware_interface import Hardware_interface

TIME_OUT = 0.5  #Time out used for refreshing the User Interface
PPG_QUEUE_MAX_LENGTH = 1000 #1000 points will be displayed per PPG (10mS/point x 1000 points = 10 seconds )
CHART_MARGIN_PERCENTAGE = 1.1     #Increase Y Lims of the chart to improve the visualization. 10% in this case.

class Messages(Enum):
    HRSETALARM = 0

#Method in charge of updating the UI. It is called from another thread and uses queue for communication
def refreshUI(guiRef, model, guiQueue, ppgQueue):
    msg = None
    while True:
        try:
            msg = guiQueue.get(timeout = TIME_OUT)

            if msg == Messages.HRSETALARM:
                model.count += 1
                guiRef.entry["state"] = 'normal'
                guiRef.entry.delete(0,"end")
                guiRef.entry.insert(0,"Valor")
                guiRef.entry["state"] = 'disabled'

                device.set_alarm(Hardware_interface.MESSAGE_ID_HR_ALARM, 80, 100)

        except queue.Empty:
            pass 

        #Other tasks...
        #Getting data from device object
        dataAvailable = device.new_data_available()

        if(dataAvailable == True):
            #Process message
            #If message adheres to protocol, update queues or instant values
            guiRef.entry_two.delete(0,"end")
            guiRef.entry_two.insert(0, "Dato nuevo")
            #TODO Do Shallow copies of the queues
            #guiRef.ppgQueue.put()


#Class for the User Interface
class Application(tk.Frame):
    def __init__(self, master=None):
        super().__init__(master)
        self.master = master
        #self.master.geometry('1000x1000')
        self.master.title('Smart Watch')
        self.pack(fill=BOTH, expand=True)
        
        #Animate needs to be created here and declared self. If not, will be gargabe collected
        self.ppgQueue = queue.Queue()
        self.chartQueue = collections.deque(np.zeros(PPG_QUEUE_MAX_LENGTH), maxlen=PPG_QUEUE_MAX_LENGTH)
        self.figure, self.ax = plt.subplots(1, 1, figsize=(6,5))
        self.create_widgets()
        self.animate = FuncAnimation(self.figure, self.refreshChart, interval=500)

    def create_widgets(self):
    
    #FRAME 1 
        self.frame_one = tk.Frame(self.master)
        self.frame_one.pack(fill=X)

        self.label_title = tk.Label(self.frame_one, text = "Smart Watch")
        self.label_title.pack(side=LEFT)

    #FRAME 2 - GRAPH
        self.frame_two = tk.Frame(self.master)
        self.frame_two.pack(fill=X)
        chart_type = FigureCanvasTkAgg(self.figure, self.frame_two)
        chart_type.get_tk_widget().pack()

    #FRAME 3
        self.frame_three = tk.Frame(self.master)
        self.frame_three.pack(fill=X)

        self.label_subtitle_two = tk.Label(self.frame_three, text = "Valores Instantáneos")
        self.label_subtitle_two.pack(side=LEFT)

    #FRAME 4
        self.frame_four = tk.Frame(self.master)
        self.frame_four.pack(fill=X)
        
        self.label_heartrate = tk.Label(self.frame_four, text = "Heart Rate")
        self.label_heartrate.pack(side=LEFT)

        #TODO Entry must be enabled before updating it

        self.entry_heartrate = tk.Entry(self.frame_four, state='disabled')
        self.entry_heartrate.pack(side=LEFT)

        self.label_heartrate_units_one = tk.Label(self.frame_four, text = "bpm")
        self.label_heartrate_units_one.pack(side=LEFT)

        self.label_heartrate_alarm = tk.Label(self.frame_four, text = "ALARMA")
        self.label_heartrate_alarm.pack(side=LEFT)

        self.label_heartrate_minimum = tk.Label(self.frame_four, text = "Mínimo")
        self.label_heartrate_minimum.pack(side=LEFT)

        self.entry_heartrate_minimum = tk.Entry(self.frame_four)
        self.entry_heartrate_minimum.pack(side=LEFT)

        self.label_heartrate_units_two = tk.Label(self.frame_four, text = "bpm")
        self.label_heartrate_units_two.pack(side=LEFT)

        self.label_heartrate_maximum = tk.Label(self.frame_four, text = "Máximo")
        self.label_heartrate_maximum.pack(side=LEFT)

        self.entry_heartrate_maximum = tk.Entry(self.frame_four)
        self.entry_heartrate_maximum.pack(side=LEFT)

        self.label_units_heartrate_three = tk.Label(self.frame_four, text = "bpm")
        self.label_units_heartrate_three.pack(side=LEFT)

        self.button_heartrate = tk.Button(self.frame_four, text = "Configurar alarma")
        self.button_heartrate.pack(side=LEFT)

    #FRAME 5
        self.frame_five = tk.Frame(self.master)
        self.frame_five.pack(fill=X)

        self.quit = tk.Button(self.frame_five, text="QUIT", fg="red", command=self.master.destroy)
        self.quit.pack(side=RIGHT)

    #Method that provides reference to the user interface. It uses a queue to communicate the UI with the exterior. It returns references to 
    #inner objetcs so the exterior can communicate with it. More references can be given in "SimpleNameSpace"
    def gui_reference(self, root, queue, ppgQueue):
        entry = self.entry_heartrate
        entry["state"] = 'normal'
        entry.delete(0,"end")
        entry.insert(0,"10")
        entry["state"] = 'disabled'

        self.button_heartrate["command"] = lambda : queue.put(Messages.HRSETALARM)
        
        ppgQueue = self.ppgQueue

        #Just for testing purposes to emit reference to another widget
        entry_two = self.entry_heartrate_minimum
        return SimpleNamespace(entry=entry, entry_two=entry_two, ppgQueue=ppgQueue)

    #Function used to update the chart with external data. It is called every 500 mseconds. Data needs to be passed by reference using "gui_reference"
    def refreshChart(self, x):
        #get data
        while(self.ppgQueue.empty() == False):
            self.chartQueue.popleft()
            #TODO Verify with index that there is no missing value. Add None if there are any...
            self.chartQueue.append(self.ppgQueue.get())
        
        #Get limits to update chart
        copyOfChartQueue = self.chartQueue.copy()
        maxValue = 0
        minValue = 0
        for p in copyOfChartQueue:
            if(type(p) == int):
                if p > maxValue:
                    maxValue = p
                if p < minValue:
                    minValue = p


        self.ax.cla()
        self.ax.plot(self.chartQueue, color='tab:blue')

        #print("Chart refreshed")
        #set limits and title
        self.ax.set_xlim([0, PPG_QUEUE_MAX_LENGTH])
        self.ax.set_ylim([minValue * CHART_MARGIN_PERCENTAGE, maxValue * CHART_MARGIN_PERCENTAGE])
        self.ax.set_title('PPG - RED')

if __name__ == '__main__':

    root = tk.Tk()
    app = Application(master=root)

    #Queue used to communicate the UI with the exterior
    guiQueue = queue.Queue()

    #Queue used to store PPG data
    ppgQueue = queue.Queue(maxsize=PPG_QUEUE_MAX_LENGTH)

    guiRef = app.gui_reference(root, guiQueue, ppgQueue)
    model = SimpleNamespace(count=0)

    #Object for communication with MSP432 and for processing values
    device = Hardware_interface()
    
    #Thread used to separate the use interface with the front end. It is used to not slow down the GUI when processing information
    t = Thread(target=refreshUI, args=(guiRef, model, guiQueue, ppgQueue, ))
    t.daemon = True
    t.start()
    
    app.mainloop()