Device Control GUI and Binary Distribution

pyinstaller --add-data "logo.png:./"  scripts/gui.py -i logo.png -F --clean -y

class WorkerThread(threading.Thread):
    def __init__(self, command: list[str], update_text_ctrl):
        super().__init__()
        self.command=command
        self.update_text_ctrl= update_text_ctrl
        self.process = None

    def kill(self):
        try:
            self.process.kill()
        except (AttributeError,):
            pass

    def run(self):
        # Run the command and capture stdout
        try:
            self.process = subprocess.Popen(self.command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, universal_newlines=True)
            for line in self.process.stdout:
                wx.CallAfter(self.update_text_ctrl, line)  # Update GUI with stdout
            self.process.stdout.close()
            self.process.wait()
        except Exception as e:
            raise e

class StdoutQueue:
    def __init__(self, *args, **kwargs):
        self._queue = multiprocessing.Queue()

    def write(self,msg):
        self._queue.put(msg)

    def get(self):
        return self._queue.get()

    def close(self):
        self._queue.close()

    def flush(self):
        sys.__stdout__.flush()

    def get_nowait(self):
        return self._queue.get_nowait()


def program_bar_task(image, device, addresses, queue):
    handler = logging.StreamHandler(queue)
    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    logging.root.addHandler(handler)

    iodevice = mwir_control.cli.setup_from_ctx({'device': device, 'baud': 9600})
    mwir_control.cli.program_bar(device=iodevice, image=image, units=[int(pt) for pt in addresses])


class WorkerThread(threading.Thread):
    def __init__(self, command, update_text_ctrl):
        super().__init__()
        self.stdout_queue = StdoutQueue()
        self.command=command
        self.update_text_ctrl= update_text_ctrl
        self.process = None

    def kill(self):
        try:
            self.process.kill()
        except (AttributeError,):
            pass

    def run(self):
        # Run the command and capture stdout
        multiprocessing.freeze_support()
        try:
            self.process = multiprocessing.Process(target=self.command, kwargs={"queue": self.stdout_queue})
            self.process.start()
            while self.process.is_alive():
                try:
                    for line in self.stdout_queue.get_nowait():
                        wx.CallAfter(self.update_text_ctrl, line)  # Update GUI with stdout
                except queue.Empty:
                    pass
            self.process.join()
        except Exception as e:
            print(f"Error running command: {e}")
            raise e


class MyFrame(wx.Frame):
    def __init__(self, parent, title):
        super(MyFrame, self).__init__(parent, title=title, size=(800, 600))

        panel = wx.Panel(self)

        version_label = wx.StaticText(panel, label=f"Version: {__version__}")

        # File Input
        file_label = wx.StaticText(panel, label="Select .bin File:")
        wildcard = "Binary files (*.bin)|*.bin"
        self.file_picker = wx.FilePickerCtrl(panel, wildcard=wildcard, style=wx.FLP_OPEN|wx.FLP_FILE_MUST_EXIST)


        # Communication Channel selection (example: list serial ports)
        channel_label = wx.StaticText(panel, label="Select COM Channel:")
        com_ports = list_serial_ports()  # Fetch COM ports

        if len(com_ports) == 0:
            wx.MessageBox('No com ports found.', 'Error', wx.OK | wx.ICON_ERROR)
            raise UserWarning('No com ports found.')

        self.com_choice = wx.Choice(panel, choices=com_ports)
        self.com_choice.SetSelection(0)

        # Multi-Select ListBox
        multiselect_label = wx.StaticText(panel, label="Modbus Addresses:")
        address_choices = ["1", "2", "3", "4", "5", "6", "246"]
        self.multi_listbox = wx.ListBox(panel, choices=address_choices, style=wx.LB_EXTENDED)
        for i in range(len(address_choices)-2):
            self.multi_listbox.SetSelection(i)

        # Run Command Button
        self.run_button = wx.Button(panel, label="Run Command")
        self.run_button.Bind(wx.EVT_BUTTON, self.on_run_command)

        # Terminal Output
        terminal_label = wx.StaticText(panel, label="Terminal Output:")
        self.terminal_output = wx.TextCtrl(panel, style=wx.TE_MULTILINE|wx.TE_READONLY)

        # Layout with sizers
        sizer = wx.BoxSizer(wx.VERTICAL)
        sizer.Add(version_label, 0, wx.ALIGN_RIGHT | wx.ALL, 5)
        sizer.Add(file_label, 0, wx.ALL, 5)
        sizer.Add(self.file_picker, 0, wx.EXPAND|wx.ALL, 5)
        sizer.Add(channel_label, 0, wx.ALL, 5)
        sizer.Add(self.com_choice, 0, wx.EXPAND|wx.ALL, 5)
        sizer.Add(multiselect_label, 0, wx.ALL, 5)
        sizer.Add(self.multi_listbox, 0, wx.EXPAND|wx.ALL, 5)
        sizer.Add(terminal_label, 0, wx.ALL, 5)
        sizer.Add(self.terminal_output, 1, wx.EXPAND|wx.ALL, 5)
        sizer.Add(self.run_button, 0, wx.ALIGN_CENTER | wx.ALL, 5)

        panel.SetSizer(sizer)
        self.worker_thread = None
        self.Show()

    def get_validation_errors(self) -> list[str]:
        errors = []
        if self.com_choice.GetSelection() not in range(self.com_choice.GetCount()):
            errors.append("No com port selected")
        addresses = [self.multi_listbox.GetString(pt) for pt in self.multi_listbox.GetSelections()]

        if not len(addresses):
            errors.append("Select at least one address")

        bin = self.file_picker.GetPath()
        if not bin or not pathlib.Path(bin).exists():
            errors.append("No binary selected or file not found")

        return errors

    def on_run_command(self, event):
        """ Function of a never-ending program writing to terminal """
        if self.worker_thread is not None and self.worker_thread.is_alive():
            wx.MessageBox("Thread Already Running", 'Error', wx.OK | wx.ICON_ERROR)
            return

        errors = self.get_validation_errors()
        if len(errors):
            wx.MessageBox('\n'.join(errors), 'Validation Error', wx.OK | wx.ICON_ERROR)
            return

        com_choice = self.com_choice.GetString(self.com_choice.GetCurrentSelection())
        addresses = [self.multi_listbox.GetString(pt) for pt in self.multi_listbox.GetSelections()]
        image = self.file_picker.GetPath()
        command = functools.partial(program_bar_task, image=image, device=com_choice, addresses=addresses)

        self.worker_thread = WorkerThread(command, self.terminal_output.AppendText)
        self.worker_thread.start()


    def on_close(self, event):
        """ Clean up thread on frame close """
        if hasattr(self, 'process_thread') and self.process_thread.is_alive():
            self.process_thread.join()
        with contextlib.suppress(AttributeError):
            self.worker_thread.kill()
            self.worker_thread.join(1)
        self.Destroy()

if __name__ == '__main__':
    app = wx.App()
    frame = MyFrame(None, title='MWIR Firmware Update')
    frame.Bind(wx.EVT_CLOSE, frame.on_close)

    level = logging.INFO
    log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    logging.basicConfig(format=log_format)
    pymodbus.logging.Log.setLevel(level)
    logging.getLogger('mwir_control').setLevel(level)
    logging.getLogger('isp_programmer').setLevel(level)

    app.MainLoop()

def list_serial_ports():
    """ List available COM ports in Windows and Linux """
    ports = serial.tools.list_ports.comports()
    return [port.device for port in ports if port.description]

pyinstaller -c mwir-control/scripts/gui.py -F --clean -y --collect-all mwir_control