subastas-derechos-firmes/main.py

import logging
import queue
import signal
import threading
import time
import tkinter as tk
from tkinter import (HORIZONTAL, VERTICAL, E, N, S, W, filedialog, messagebox,
                     ttk)
from tkinter.scrolledtext import ScrolledText

from PIL import Image, ImageTk

from common import log
from qhandler import QueueHandler
from simsdt import runsdt
from simsdt.utils.excel_checker import check_excel_file
from simsdt.utils.solver_checker import check_solver

main_logger = logging.getLogger("simsdt.gui")


class RunsdtThread(threading.Thread):
    def __init__(self, group=None, target=None, name=None,
                 args=(), kwargs=None, *, daemon=None):
        super(RunsdtThread, self).__init__(group=group,
                                           target=target,
                                           name=name,
                                           daemon=daemon)
        self._stop_event = threading.Event()
        self.args = args
        self.kwargs = kwargs
        return

    def run(self):
        t = time.time()
        runsdt.main(self.args)
        elapsed_time = time.time() - t
        et = time.strftime("%H:%M:%S", time.gmtime(elapsed_time))
        main_logger.info('Tiempo de ejecución %s' % et)

    def stop(self):
        self._stop_event.set()


class ConsoleUi:
    """Poll messages from a logging queue and display them in a scrolled text
    widget"""

    def __init__(self, frame):
        self.frame = frame

        trash_icon = Image.open('img/icons/icons8-eliminar-16.png')
        self.trash_icon = ImageTk.PhotoImage(trash_icon)
        # Create a ScrolledText wdiget
        self.scrolled_text = ScrolledText(frame, state='disabled', height=25)
        self.scrolled_text.grid(row=0, column=0, sticky=(N, S, W, E))
        self.clear = ttk.Button(frame, text='Limpiar Log',
                                command=self.clear_console,
                                image=self.trash_icon,
                                compoun='right')
        self.clear.grid(row=1, column=0, sticky=E, pady=(5, 5))
        # Font config
        self.scrolled_text.configure(font='TkFixedFont')
        self.scrolled_text.tag_config('INFO', foreground='black')
        self.scrolled_text.tag_config('DEBUG', foreground='gray')
        self.scrolled_text.tag_config('WARNING', foreground='orange')
        self.scrolled_text.tag_config('ERROR', foreground='red')
        self.scrolled_text.tag_config(
            'CRITICAL', foreground='red', underline=1)
        # Create a logging handler using a queue
        self.log_queue = queue.Queue()
        self.queue_handler = QueueHandler(self.log_queue)
        formatter = logging.Formatter(
            '%(asctime)s: %(message)s', datefmt='%Y-%m-%dT%H:%M:%S')
        self.queue_handler.setFormatter(formatter)
        # self.queue_handler.setLevel(logging.INFO)
        main_logger.addHandler(self.queue_handler)
        # Start polling messages from the queue
        self.frame.after(100, self.poll_log_queue)

    def display(self, record):
        msg = self.queue_handler.format(record)
        self.scrolled_text.configure(state='normal')
        self.scrolled_text.insert(tk.END, msg + '\n', record.levelname)
        self.scrolled_text.configure(state='disabled')
        # Autoscroll to the bottom
        self.scrolled_text.yview(tk.END)

    def poll_log_queue(self):
        # Check every 100ms if there is a new message in the queue to display
        while True:
            try:
                record = self.log_queue.get(block=False)
            except queue.Empty:
                break
            else:
                self.display(record)
        self.frame.after(100, self.poll_log_queue)

    def clear_console(self):
        self.scrolled_text.configure(state='normal')
        self.scrolled_text.delete(1.0, tk.END)
        self.scrolled_text.configure(state='disabled')


class FormUi:

    runsdt_thread = None

    def __init__(self, frame, q):
        self.frame = frame
        self.q = q
        image = Image.open('img/icons/icons8-play-16.png')
        self.photo = ImageTk.PhotoImage(image)

        opfile = Image.open('img/icons/icons8-abrir-carpeta-16.png')
        self.opfile = ImageTk.PhotoImage(opfile)

        # Add a text field for file path
        self.file_path = tk.StringVar()
        ttk.Label(self.frame, text='Archivo:').grid(
            column=0, row=0, sticky=W, pady=(5, 5), padx=(0, 10))
        self.entry = ttk.Entry(self.frame, textvariable=self.file_path,
                               width=50)
        self.entry.grid(column=1, row=0, sticky=(W, E), pady=(5, 5))
        self.button = ttk.Button(
            self.frame, image=self.opfile, compound='center',
            command=self.open_file_dialog, width=3)
        self.button.grid(column=2, row=0, sticky=W)
        self.button_run = ttk.Button(self.frame, image=self.photo,
                                     compound="right", text='Ejecutar',
                                     command=self.run)
        self.button_run.grid(column=1, row=1, sticky=W)

    def open_file_dialog(self):
        # Open file dialog
        try:
            file = filedialog.askopenfilename(
                filetypes=[('Excel', '.xlsx')],
                title='Seleccione un archivo de subasta...')
            check_excel_file(file)
        except Exception as e:
            main_logger.error(e)
            self.file_path.set('')
        else:
            main_logger.info(f'Archivo seleccionado: {file}')
            self.file_path.set(file)

    def run(self):
        try:

            if not self.file_path.get():
                main_logger.error(
                    'Debe seleccionar un archivo para ejecutarlo')
                return

            check_solver()

            self.button_run.configure(state='disabled')
            main_logger.info('Ejecutando')

            self.runsdt_thread = RunsdtThread(args=(self.file_path.get()))
            self.runsdt_thread.start()

            self.check_thread()

        except Exception as e:
            main_logger.error(e)
            return

    def check_thread(self):

        if not self.runsdt_thread.is_alive():
            self.button_run.configure(state='normal')
            self.q.put(False)
        else:
            self.q.put(True)
            self.frame.after(100, self.check_thread)


class StatusUi:

    def __init__(self, frame, q):
        self.q = q
        self.frame = frame
        self.pb_max = 100
        # ttk.Label(self.frame, text='This is just an example of a third frame').grid(
        #     column=0, row=1, sticky=W)
        self.progress_bar = ttk.Progressbar(
            self.frame, mode='determinate')
        self.progress_bar.pack(expand=True, fill=tk.X, side=tk.TOP)

        self.frame.after(100, self.check_q)

    def check_q(self):
        # Check every 100ms if there is a new message in the queue to display
        while True:
            try:
                record = self.q.get(block=False)
            except queue.Empty:
                break
            else:
                if record:
                    step = 1
                    self.progress_bar.step(step)
                    if step >= self.pb_max:
                        step = 1
                    else:
                        step += 1
                else:
                    self.progress_bar.config(value=0)
        self.frame.after(100, self.check_q)


class App:

    def __init__(self, root):
        self.root = root
        # Queue a nivel de App para manejar el progress bar
        pbqueue = queue.Queue()

        root.title('SimSDT')
        root.iconbitmap("app.ico")
        root.columnconfigure(0, weight=1)
        root.rowconfigure(0, weight=1)
        # root.geometry("1080x960")

        # Create the panes and frames
        vertical_pane = ttk.PanedWindow(
            self.root, orient=VERTICAL)
        vertical_pane.grid(row=0, column=0, sticky="nsew", padx=10, pady=10)

        main_title = ttk.Label(
            vertical_pane,
            text='Simulación de Subastas de Derechos de Transmisión',
            font=('', '20', 'bold'), padding=(0, 5))
        vertical_pane.add(main_title, weight=0)

        horizontal_pane = ttk.PanedWindow(vertical_pane, orient=HORIZONTAL)
        vertical_pane.add(horizontal_pane, weight=1)

        form_frame = ttk.Labelframe(
            horizontal_pane, text="Configuración", padding=(5, 5))
        form_frame.columnconfigure(1, weight=1)
        horizontal_pane.add(form_frame, weight=1)

        console_frame = ttk.Labelframe(
            horizontal_pane, text="Consola", padding=(5, 5))
        console_frame.columnconfigure(0, weight=1)
        console_frame.rowconfigure(0, weight=1)

        horizontal_pane.add(console_frame, weight=1)
        third_frame = ttk.Labelframe(
            vertical_pane, text="Estado", padding=(5, 5))
        vertical_pane.add(third_frame, weight=0)
        # Initialize all frames
        self.form = FormUi(form_frame, pbqueue)
        self.console = ConsoleUi(console_frame)
        self.third = StatusUi(third_frame, pbqueue)
        self.root.protocol('WM_DELETE_WINDOW', self.quit)
        self.root.bind('<Control-q>', self.quit)
        signal.signal(signal.SIGINT, self.quit)

        print(root.geometry())

    def quit(self, *args):

        msg = 'Se esta ejecutando un proceso de optimización\n\n'
        msg += '¿Desea salir de la aplicación?'

        if self.form.runsdt_thread and self.form.runsdt_thread.is_alive():
            if messagebox.askokcancel("Salir de SimSDT", msg):
                main_logger.info('Cerrando aplicacion SimSDT')
                self.form.runsdt_thread.stop()
                self.root.destroy()
        else:
            main_logger.info('Saliendo de la aplicacion')
            self.root.destroy()


def main():
    main_logger.info('SimSDT ha iniciado')
    root = tk.Tk()
    app = App(root)
    app.root.mainloop()


if __name__ == '__main__':
    main()