ota-flasher/app/main.py

import concurrent.futures
import logging
import traceback

import serial
from app.gui.action_button_panel import ActionButtonPanel
from app.release_checker import ReleaseChecker
import wx
import wx.richtext as rt

from zeroconf import ServiceBrowser, Zeroconf
import os
import webbrowser

from app import espota
from app.api import ApiHandler
from app.fw_updater import FwUpdater
from app.gui.devices_panel import DevicesPanel
from app.utils import get_app_data_folder
from app.zeroconf_listener import ZeroconfListener

from app.espota import FLASH, SPIFFS

class BTClockOTAApp(wx.App):
    def OnInit(self):
        return True
class RichTextCtrlHandler(logging.Handler):
    def __init__(self, ctrl):
        super().__init__()
        self.ctrl = ctrl

    def emit(self, record):
        msg = self.format(record)
        wx.CallAfter(self.append_text, "\n" + msg)

    def append_text(self, text):
        self.ctrl.AppendText(text)
        self.ctrl.ShowPosition(self.ctrl.GetLastPosition())
        
class SerialPortsComboBox(wx.ComboBox):
    def __init__(self, parent, fw_update):
        self.fw_update = fw_update
        self.ports = serial.tools.list_ports.comports()
        wx.ComboBox.__init__(self, parent, choices=[
                             port.device for port in self.ports])


class BTClockOTAUpdater(wx.Frame):
    updatingName = ""

    def __init__(self, parent, title):
        wx.Frame.__init__(self, parent, title=title, size=(800, 500))

        self.SetMinSize((800, 500))
        self.releaseChecker = ReleaseChecker()
        self.zeroconf = Zeroconf()
        self.listener = ZeroconfListener(self.on_zeroconf_state_change)
        self.browser = ServiceBrowser(
            self.zeroconf, "_http._tcp.local.", self.listener)
        self.api_handler = ApiHandler()
        self.fw_updater = FwUpdater(self.call_progress, self.SetStatusText)
        

        panel = wx.Panel(self)
        self.log_ctrl = rt.RichTextCtrl(panel, style=wx.TE_MULTILINE | wx.TE_READONLY | wx.TE_RICH2)
        monospace_font = wx.Font(10, wx.FONTFAMILY_TELETYPE, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL)
        self.log_ctrl.SetFont(monospace_font)

        handler = RichTextCtrlHandler(self.log_ctrl)
        handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s', '%H:%M:%S'))
        logging.getLogger().addHandler(handler)
        logging.getLogger().setLevel(logging.DEBUG)

        
        self.device_list = DevicesPanel(panel)

        vbox = wx.BoxSizer(wx.VERTICAL)
        vbox.Add(self.device_list, proportion=2,
                 flag=wx.EXPAND | wx.ALL, border=20)
        hbox = wx.BoxSizer(wx.HORIZONTAL)

        self.fw_label = wx.StaticText(
            panel, label=f"Fetching latest version from GitHub...")
        hbox.Add(self.fw_label,  1, wx.EXPAND | wx.ALL, 5)

        self.actionButtons = ActionButtonPanel(
            panel, self)
        hbox.AddStretchSpacer()

        hbox.Add(self.actionButtons, 2, wx.EXPAND | wx.ALL, 5)
        vbox.Add(hbox, 0, wx.EXPAND | wx.ALL, 20)

        self.progress_bar = wx.Gauge(panel, range=100)
        vbox.Add(self.progress_bar, 0, wx.EXPAND | wx.ALL, 20)
        vbox.Add(self.log_ctrl, 1, flag=wx.EXPAND | wx.ALL, border=20)

        panel.SetSizer(vbox)
        self.setup_ui()

        wx.CallAfter(self.fetch_latest_release_async)
        wx.YieldIfNeeded()
    def setup_ui(self):
        self.setup_menubar()
        self.status_bar = self.CreateStatusBar(2)
        self.Show(True)
        self.Centre()

    def setup_menubar(self):
        filemenu = wx.Menu()
        menuOpenDownloadDir = filemenu.Append(
            wx.ID_OPEN, "&Open Download Dir", " Open the directory with firmware files and cache")
        menuAbout = filemenu.Append(
            wx.ID_ABOUT, "&About", " Information about this program")
        menuExit = filemenu.Append(
            wx.ID_EXIT, "E&xit", " Terminate the program")

        menuBar = wx.MenuBar()
        menuBar.Append(filemenu, "&File")

        self.SetMenuBar(menuBar)
        self.Bind(wx.EVT_MENU, self.OnOpenDownloadFolder, menuOpenDownloadDir)
        self.Bind(wx.EVT_MENU, self.OnAbout, menuAbout)
        self.Bind(wx.EVT_MENU, self.OnExit, menuExit)

    def on_zeroconf_state_change(self, type, name, state, info):
        index = self.device_list.FindItem(0, name)

        if state == "Added":
            deviceSettings = self.api_handler.get_settings(
                info.parsed_addresses()[0])

            version = info.properties.get(b"rev").decode()
            fsHash = "Too old"
            hwRev = "REV_A_EPD_2_13"

            if 'gitTag' in deviceSettings:
                version = deviceSettings["gitTag"]

            if 'fsRev' in deviceSettings:
                fsHash = deviceSettings['fsRev'][:7]

            if (info.properties.get(b"hw_rev") is not None):
                hwRev = info.properties.get(b"hw_rev").decode()

            fwHash = info.properties.get(b"rev").decode()[:7]
            address = info.parsed_addresses()[0]

            if index == wx.NOT_FOUND:
                index = self.device_list.InsertItem(
                    self.device_list.GetItemCount(), type)
                self.device_list.SetItem(index, 0, name)
                self.device_list.SetItem(index, 1, version)
                self.device_list.SetItem(index, 2, fwHash)
                self.device_list.SetItem(index, 3, hwRev)
                self.device_list.SetItem(index, 4, address)

            else:
                self.device_list.SetItem(index, 0, name)
                self.device_list.SetItem(index, 1, version)
                self.device_list.SetItem(index, 2, fwHash)
                self.device_list.SetItem(index, 3, hwRev)
                self.device_list.SetItem(index, 4, address)
            self.device_list.SetItem(index, 5, fsHash)
            self.device_list.SetItemData(index, index)
            self.device_list.itemDataMap[index] = [
                name, version, fwHash, hwRev, address, fsHash]
            for col in range(0, len(self.device_list.column_headings)):
                self.device_list.SetColumnWidth(
                    col, wx.LIST_AUTOSIZE_USEHEADER)
        elif state == "Removed":
            if index != wx.NOT_FOUND:
                self.device_list.DeleteItem(index)

    def call_progress(self, progress):
        progressPerc = int(progress*100)
        self.SetStatusText(f"Progress: {progressPerc}%")
        wx.CallAfter(self.update_progress, progress)

    def update_progress(self, progress):
        progressPerc = int(progress*100)
        self.progress_bar.SetValue(progressPerc)
        wx.YieldIfNeeded()

    def fetch_latest_release_async(self):
        # Start a new thread to execute fetch_latest_release
        app_folder = get_app_data_folder()
        if not os.path.exists(app_folder):
            os.makedirs(app_folder)
        executor = concurrent.futures.ThreadPoolExecutor()
        future = executor.submit(self.releaseChecker.fetch_latest_release)
        future.add_done_callback(self.handle_latest_release)

    def handle_latest_release(self, future):
        try:
            latest_release = future.result()
            self.fw_label.SetLabelText(f"Downloaded firmware version: {
                                       latest_release}\nCommit: {self.releaseChecker.commit_hash}")
        except Exception as e:
            self.fw_label.SetLabel(f"Error occurred: {str(e)}")
            traceback.print_tb(e.__traceback__)
            
    def OnOpenDownloadFolder(self, e):
        wx.LaunchDefaultBrowser(get_app_data_folder())
            
    def OnAbout(self, e):
        dlg = wx.MessageDialog(
            self, "An updater for BTClocks", "About BTClock OTA Updater", wx.OK)
        dlg.ShowModal()
        dlg.Destroy()

    def OnExit(self, e):
        self.Close(False)