Main.py

#!/usr/bin/env python

import wx
import wx.adv
import wx.lib.inspection
import wx.lib.mixins.inspection

import sys
import os
import esptool
import json
import spiffsgen
import tempfile
import threading
import images as images
from AppRepository import AppRepository
from enum import Enum
from serial import SerialException
from serial.tools import list_ports

__version__ = "3.0"
__app_name__ = "ThingPulse App Fairy"
__auto_select__ = "Auto-select"
__auto_select_explanation__ = "(first port with Espressif device)"
__supported_baud_rates__ = [9600, 57600, 74880, 115200, 230400, 460800, 921600]
__spiffs_build_dir__ = "{0}/thingpulse-spiffs".format(tempfile.gettempdir())

# ---------------------------------------------------------------------------


class DeviceType(Enum):
    ICON64 = "icon64", "Icon64", "qio"
    COLOR_KIT = "color-kit", "Color Kit (w/ touch screen)", "dio"

    def __init__(self, identifier, label, flash_mode):
        self.identifier = identifier
        self.label = label
        self.flash_mode = flash_mode
# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# See discussion at http://stackoverflow.com/q/41101897/131929
class RedirectText:
    def __init__(self, text_ctrl):
        self.__out = text_ctrl

    def write(self, string):
        if string.startswith("\r"):
            # carriage return -> remove last line i.e. reset position to start of last line
            current_value = self.__out.GetValue()
            last_newline = current_value.rfind("\n")
            new_value = current_value[:last_newline + 1]  # preserve \n
            new_value += string[1:]  # chop off leading \r
            wx.CallAfter(self.__out.SetValue, new_value)
        else:
            wx.CallAfter(self.__out.AppendText, string)

    # noinspection PyMethodMayBeStatic
    def flush(self):
        # noinspection PyStatementEffect
        None

    # esptool >=3 handles output differently if the output stream is not a TTY
    # noinspection PyMethodMayBeStatic
    def isatty(self):
        return True

# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
class FlashingThread(threading.Thread):
    def __init__(self, parent, config, callback):
        threading.Thread.__init__(self)
        self.daemon = True
        self._parent = parent
        self._config = config
        self._callback = callback

    def run(self):
        try:
            spiffs_image_path = self.create_spiffs_image()

            command = []

            if not self._config.port.startswith(__auto_select__):
                command.append("--port")
                command.append(self._config.port)

            command.extend(["--baud", str(self._config.baud),
                            # these are the default values
                            # "--before", "default_reset",
                            # "--after", "hard_reset",
                            "write_flash",
                            # https://github.com/espressif/esptool/issues/599
                            "--flash_size", "detect",
                            "--flash_mode", self._config.device_type.flash_mode,
                            "0x00000", self._config.app_path,
                            self._config.spiffs.offset, spiffs_image_path])

            if self._config.erase_before_flash:
                command.append("--erase-all")

            print("\nInstalling application...")
            print("Command: esptool.py %s\n" % " ".join(command))

            esptool.main(command)

            # The "success" message depends on the esptool post-write behavior. This one is in line with the
            # --after hard_reset configuration about.
            print("\nApplication successfully installed. Device will reset.")
        except SerialException as e:
            self._parent.report_error(e.strerror)
            raise e
        finally:
            self._callback()

    def create_spiffs_image(self):
        print("Building app configuration file...", end="")
        os.makedirs(__spiffs_build_dir__, exist_ok=True)

        # 1. write the app config values into a properties file
        file = open("{0}/application.properties".format(__spiffs_build_dir__), "w")
        for key, value in self._config.app_properties.items():
            effective_value = value
            if isinstance(value, bool):
                effective_value = str(value).lower()
            file.write("{0}={1}\n".format(key, effective_value))
        file.close()

        # 2. build a SPIFFS image of the directory the properties file resides in
        spiffs_image_path = "{0}/thingpulse-app.spiffs".format(tempfile.gettempdir())
        conf = self._config.spiffs
        args = []
        args.extend(["spiffsgen.py",
                     "--block-size", str(conf.block_size),
                     "--page-size", str(conf.page_size),
                     "--meta-len", str(conf.meta_len)])

        if conf.aligned_obj_ix_tables:
            args.append("--aligned-obj-ix-tables")

        if conf.use_magic_len:
            args.append("--use-magic-len")
        else:
            args.append("--no-magic-len")

        args.extend([conf.size, __spiffs_build_dir__, spiffs_image_path])

        sys.argv = args
        spiffsgen.main()
        print("ok")
        return spiffs_image_path


# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
# DTOs between GUI and flashing thread


class SpiffsConfig:
    def __init__(self, props=None):
        self.page_size = self.extract_value(props, "page-size")
        self.block_size = self.extract_value(props, "block-size")
        self.use_magic_len = self.extract_value(props, "use-magic-len")
        self.aligned_obj_ix_tables = self.extract_value(props, "aligned-obj-ix-tables")
        self.meta_len = self.extract_value(props, "meta-len")
        self.offset = self.extract_value(props, "offset")
        self.size = self.extract_value(props, "size")

    @staticmethod
    def extract_value(props, param):
        return props.get(param) if props else None


class FlashConfig:
    def __init__(self):
        self.baud = 115200
        self.erase_before_flash = False
        self.device_type = None
        self.port = None
        # below fields are not persisted
        self.properties_definitions = None
        self.app_properties = {}
        self.app_path = None
        self.spiffs = SpiffsConfig()

    @classmethod
    def load(cls, file_path):
        conf = cls()
        if os.path.exists(file_path):
            with open(file_path, 'r') as f:
                data = json.load(f)
            device_type_ = data['deviceType']
            conf.port = data.get('port', "")
            conf.baud = data['baud']
            conf.device_type = DeviceType[device_type_] if device_type_ and device_type_ in DeviceType.__members__ else None
            conf.erase_before_flash = data['erase']
        return conf

    def safe(self, file_path):
        data = {
            'port': self.port,
            'baud': self.baud,
            'deviceType': self.device_type.name if self.device_type else '',
            'erase': self.erase_before_flash,
        }
        with open(file_path, 'w') as f:
            json.dump(data, f)


# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
class NodeMcuFlasher(wx.Frame):

    def __init__(self, parent, title):
        wx.Frame.__init__(self, parent, -1, title, size=(725, 650),
                          style=wx.DEFAULT_FRAME_STYLE | wx.NO_FULL_REPAINT_ON_RESIZE)
        self._config = FlashConfig.load(self._get_config_file_path())
        self._app_repository = AppRepository()

        self._build_status_bar()
        self._set_icons()
        self._build_menu_bar()
        self._init_ui()

        sys.stdout = RedirectText(self.console_ctrl)

        self.Centre(wx.BOTH)
        self.Show(True)

        if self._config.device_type:
            self.populate_apps_choice_ctrl()
        else:
            # no configured device type -> maybe first-time user -> give USB-to-serial hints
            print("Connect your device")
            print("\nIf you chose the serial port auto-select feature you might need to ")
            print("turn off Bluetooth")

    def _init_ui(self):
        def on_reload(event):
            self.port_choice.SetItems(self._get_serial_ports())

        def on_baud_changed(event):
            radio_button = event.GetEventObject()

            if radio_button.GetValue():
                self._config.baud = radio_button.rate

        def on_erase_changed(event):
            radio_button = event.GetEventObject()

            if radio_button.GetValue():
                self._config.erase_before_flash = radio_button.erase

        def on_installation_done():
            self._install_button.Enable()

        def on_install_clicked(event):
            worker = FlashingThread(self, self._config, callback=on_installation_done)
            self._install_button.Disable()
            worker.start()

        def on_select_port(event):
            choice = event.GetEventObject()
            self._config.port = choice.GetString(choice.GetSelection())

        def on_select_device_type(event):
            choice = event.GetEventObject()
            self._config.device_type = choice.GetClientData(choice.GetSelection())
            app_choice.Set([])
            if self._config.device_type:
                self.populate_apps_choice_ctrl()

        def on_select_app(event):
            self._configure_button.Disable()
            choice = event.GetEventObject()
            app_release = choice.GetClientData(choice.GetSelection())
            self._config.app_path = None
            self._config.spiffs = None
            if app_release:
                app_release_path = app_release["path"]
                device_type_id = self._config.device_type.identifier
                props = self._app_repository.load_app_properties(device_type_id, app_release_path)
                self._config.app_path = self._app_repository.download_app(device_type_id, app_release_path)
                self._config.spiffs = SpiffsConfig(props["spiffs"])

                app_release_label = choice.GetString(choice.GetSelection())
                self._config.properties_definitions = props["application"]
                if self._config.properties_definitions and len(self._config.properties_definitions) > 0:
                    for property_definition in props["application"]:
                        if "default" in property_definition:
                            self._config.app_properties[property_definition["prop"]] = property_definition["default"]
                    self._configure_button.Enable()
                    self.show_configuration_dialog(app_release_label, self._config.properties_definitions)

        def on_configure_clicked(event):
            app_release_label = self._app_choice_ctrl.GetString(self._app_choice_ctrl.GetSelection())
            self.show_configuration_dialog(app_release_label, self._config.properties_definitions)

        panel = wx.Panel(self)

        hbox = wx.BoxSizer(wx.HORIZONTAL)

        fgs = wx.FlexGridSizer(7, 2, 10, 10)

        self.port_choice = wx.Choice(panel, choices=self._get_serial_ports())
        self.port_choice.Bind(wx.EVT_CHOICE, on_select_port)
        self._select_configured_port()

        reload_button = wx.Button(panel, label="Reload")
        reload_button.Bind(wx.EVT_BUTTON, on_reload)
        reload_button.SetToolTip("Reload serial device list")

        serial_boxsizer = wx.BoxSizer(wx.HORIZONTAL)
        serial_boxsizer.Add(self.port_choice, 1, wx.EXPAND)
        serial_boxsizer.Add(reload_button, flag=wx.LEFT, border=10)

        device_type_choice = wx.Choice(panel, size=wx.DefaultSize)
        device_type_choice.Append("", None)
        for device_type in DeviceType:
            device_type_choice.Append(device_type.label, device_type)
        device_type_choice.SetLabel("" if self._config.device_type is None else self._config.device_type.label)
        device_type_choice.Bind(wx.EVT_CHOICE, on_select_device_type)

        app_choice = wx.Choice(panel, size=wx.DefaultSize)
        app_choice.Bind(wx.EVT_CHOICE, on_select_app)
        self._app_choice_ctrl = app_choice
        configure_button = wx.Button(panel, label="Configure")
        configure_button.Bind(wx.EVT_BUTTON, on_configure_clicked)
        configure_button.SetToolTip("Configure the selected app")
        configure_button.Disable()
        self._configure_button = configure_button

        app_boxsizer = wx.BoxSizer(wx.HORIZONTAL)
        app_boxsizer.Add(app_choice, flag=wx.EXPAND)
        app_boxsizer.Add(configure_button, flag=wx.LEFT, border=10)

        baud_boxsizer = wx.BoxSizer(wx.HORIZONTAL)

        def add_baud_radio_button(sizer, index, baud_rate):
            style = wx.RB_GROUP if index == 0 else 0
            radio_button = wx.RadioButton(panel, name="baud-%d" % baud_rate, label="%d" % baud_rate, style=style)
            radio_button.rate = baud_rate
            # sets default value
            radio_button.SetValue(baud_rate == self._config.baud)
            radio_button.Bind(wx.EVT_RADIOBUTTON, on_baud_changed)
            sizer.Add(radio_button)
            sizer.AddSpacer(10)

        for idx, rate in enumerate(__supported_baud_rates__):
            add_baud_radio_button(baud_boxsizer, idx, rate)

        erase_boxsizer = wx.BoxSizer(wx.HORIZONTAL)

        def add_erase_radio_button(sizer, index, erase_before_flash, label, value):
            style = wx.RB_GROUP if index == 0 else 0
            radio_button = wx.RadioButton(panel, name="erase-%s" % erase_before_flash, label="%s" % label, style=style)
            radio_button.Bind(wx.EVT_RADIOBUTTON, on_erase_changed)
            radio_button.erase = erase_before_flash
            radio_button.SetValue(value)
            sizer.Add(radio_button)
            sizer.AddSpacer(10)

        erase = self._config.erase_before_flash
        add_erase_radio_button(erase_boxsizer, 0, False, "no", erase is False)
        add_erase_radio_button(erase_boxsizer, 1, True, "yes, wipes all data", erase is True)

        button = wx.Button(panel, -1, "Install Application")
        button.Bind(wx.EVT_BUTTON, on_install_clicked)
        self._install_button = button

        self.console_ctrl = wx.TextCtrl(panel, style=wx.TE_MULTILINE | wx.TE_READONLY | wx.HSCROLL)
        self.console_ctrl.SetFont(wx.Font((0, 13), wx.FONTFAMILY_TELETYPE, wx.FONTSTYLE_NORMAL,
                                          wx.FONTWEIGHT_NORMAL))
        self.console_ctrl.SetBackgroundColour(wx.WHITE)
        self.console_ctrl.SetForegroundColour(wx.BLUE)
        self.console_ctrl.SetDefaultStyle(wx.TextAttr(wx.BLUE))

        port_label = wx.StaticText(panel, label="Serial port")
        device_type_label = wx.StaticText(panel, label="Device type")
        app_label = wx.StaticText(panel, label="Application")
        baud_label = wx.StaticText(panel, label="Baud rate")

        erase_label = wx.StaticText(panel, label="Erase memory")
        console_label = wx.StaticText(panel, label="Console")

        fgs.AddMany([
                    port_label, serial_boxsizer,
                    device_type_label, device_type_choice,
                    app_label, app_boxsizer,
                    baud_label, baud_boxsizer,
                    erase_label, erase_boxsizer,
                    (wx.StaticText(panel, label="")), (button, 1, wx.EXPAND),
                    (console_label, 1, wx.EXPAND), (self.console_ctrl, 1, wx.EXPAND)])
        fgs.AddGrowableRow(6, 1)
        fgs.AddGrowableCol(1, 1)
        hbox.Add(fgs, proportion=2, flag=wx.ALL | wx.EXPAND, border=15)
        panel.SetSizer(hbox)

    def _select_configured_port(self):
        count = 0
        for item in self.port_choice.GetItems():
            if item == self._config.port:
                self.port_choice.Select(count)
                break
            count += 1

    @staticmethod
    def _get_serial_ports():
        ports = [__auto_select__ + " " + __auto_select_explanation__]
        for port, desc, hwid in sorted(list_ports.comports()):
            ports.append(port)
        return ports

    def _set_icons(self):
        self.SetIcon(images.Icon.GetIcon())

    def _build_status_bar(self):
        self.statusBar = self.CreateStatusBar(2, wx.STB_SIZEGRIP)
        self.statusBar.SetStatusWidths([-2, -1])
        status_text = "Welcome to " + __app_name__ + " %s" % __version__
        self.statusBar.SetStatusText(status_text, 0)

    def _build_menu_bar(self):
        self.menuBar = wx.MenuBar()

        # File menu
        file_menu = wx.Menu()
        wx.App.SetMacExitMenuItemId(wx.ID_EXIT)
        exit_item = file_menu.Append(wx.ID_EXIT, "E&xit\tCtrl-Q", "Exit " + __app_name__)
        exit_item.SetBitmap(images.Exit.GetBitmap())
        self.Bind(wx.EVT_MENU, self._on_exit_app, exit_item)
        self.menuBar.Append(file_menu, "&File")

        # Help menu
        help_menu = wx.Menu()
        help_item = help_menu.Append(wx.ID_ABOUT, '&About ' + __app_name__, 'About')
        self.Bind(wx.EVT_MENU, self._on_help_about, help_item)
        self.menuBar.Append(help_menu, '&Help')

        self.SetMenuBar(self.menuBar)

    @staticmethod
    def _get_config_file_path():
        return wx.StandardPaths.Get().GetUserConfigDir() + "/thingpulse-app-fairy.json"

    # Menu methods
    def _on_exit_app(self, event):
        self._config.safe(self._get_config_file_path())
        self.Close(True)

    def _on_help_about(self, event):
        from About import AboutDlg
        about = AboutDlg(self)
        about.ShowModal()
        about.Destroy()

    def report_error(self, message):
        self.console_ctrl.SetValue(message)

    def log_message(self, message):
        self.console_ctrl.AppendText(message)

    def populate_apps_choice_ctrl(self):
        # reset console content
        self.console_ctrl.SetValue("")
        apps = self._app_repository.load_app_definitions(self._config.device_type.identifier)
        if apps:
            self._app_choice_ctrl.Append("", None)
            for app in apps:
                for release in app["releases"]:
                    self._app_choice_ctrl.Append("{0} {1}".format(app["name"], release["version"]), release)
            self._app_choice_ctrl.InvalidateBestSize()
            self._app_choice_ctrl.SetSize(self._app_choice_ctrl.GetBestSize())
            self._app_choice_ctrl.GetParent().Layout()

    def show_configuration_dialog(self, app_release_label, properties_definitions):
        from AppConfigDialog import AppConfigDialog
        config_dialog = AppConfigDialog(app_release_label, properties_definitions, self._config.app_properties)
        if config_dialog.ShowModal() == wx.ID_OK:
            self._config.app_properties = config_dialog.get_properties()
        config_dialog.Destroy()

# ---------------------------------------------------------------------------


# ---------------------------------------------------------------------------
class MySplashScreen(wx.adv.SplashScreen):
    def __init__(self):
        wx.adv.SplashScreen.__init__(self, images.Splash.GetBitmap(),
                                     wx.adv.SPLASH_CENTRE_ON_SCREEN | wx.adv.SPLASH_TIMEOUT, 2500, None, -1)
        self.Bind(wx.EVT_CLOSE, self._on_close)
        self.__fc = wx.CallLater(2000, self._show_main)

    def _on_close(self, evt):
        # Make sure the default handler runs too so this window gets
        # destroyed
        evt.Skip()
        self.Hide()

        # if the timer is still running then go ahead and show the
        # main frame now
        if self.__fc.IsRunning():
            self.__fc.Stop()
            self._show_main()

    def _show_main(self):
        frame = NodeMcuFlasher(None, __app_name__)
        frame.Show()
        if self.__fc.IsRunning():
            self.Raise()

# ---------------------------------------------------------------------------


# ----------------------------------------------------------------------------
class App(wx.App, wx.lib.mixins.inspection.InspectionMixin):
    def OnInit(self):
        wx.SystemOptions.SetOption("mac.window-plain-transition", 1)
        self.SetAppName(__app_name__)

        # Create and show the splash screen.  It will then create and
        # show the main frame when it is time to do so.  Normally when
        # using a SplashScreen you would create it, show it and then
        # continue on with the application's initialization, finally
        # creating and showing the main application window(s).  In
        # this case we have nothing else to do so we'll delay showing
        # the main frame until later (see ShowMain above) so the users
        # can see the SplashScreen effect.
        splash = MySplashScreen()
        splash.Show()

        return True


# ---------------------------------------------------------------------------
def main():
    app = App(False)
    app.MainLoop()
# ---------------------------------------------------------------------------


if __name__ == '__main__':
    __name__ = 'Main'
    main()