import datetime import logging import queue import signal import threading import time import tkinter as tk from logging.handlers import RotatingFileHandler from tkinter import (HORIZONTAL, VERTICAL, E, N, S, W, filedialog, messagebox, ttk) from tkinter.scrolledtext import ScrolledText from PIL import Image, ImageTk from qhandler import QueueHandler from simsdt import runsdt fh = RotatingFileHandler( 'log/simsdt_app_monitor.log', maxBytes=1024*50, backupCount=10) fmt = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%dT%H:%M:%S') fh.setFormatter(fmt) fh.setLevel(logging.INFO) # logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', # datefmt='%Y-%m-%dT%H:%M:%S', # handlers=[RotatingFileHandler( # 'log/simsdt_app_monitor.log', maxBytes=10240, backupCount=10)], # level=logging.INFO) main_logger = logging.getLogger("simsdt") main_logger.addHandler(fh) main_logger.setLevel(logging.DEBUG) 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): now = datetime.datetime.now() runsdt.main(self.args) # time.sleep(30) main_logger.info('Fin de la función run') def stop(self): self._stop_event.set() class Clock(threading.Thread): """Class to display the time every seconds Every 5 seconds, the time is displayed using the logging.ERROR level to show that different colors are associated to the log levels """ def __init__(self): super().__init__() self._stop_event = threading.Event() def run(self): # main_logger.setLevel(logging.INFO) main_logger.debug('Clock started') previous = -1 while not self._stop_event.is_set(): now = datetime.datetime.now() if previous != now.second: previous = now.second if now.second % 5 == 0: level = logging.ERROR else: level = logging.INFO main_logger.log(level, now) time.sleep(0.2) 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 # 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)) # 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) class FormUi: runsdt_thread = None def __init__(self, frame, q): self.frame = frame self.q = q image = Image.open('img/play.png') self.photo = ImageTk.PhotoImage(image) # 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) ttk.Entry(self.frame, textvariable=self.file_path, width=50).grid(column=1, row=0, sticky=(W, E)) self.button = ttk.Button( self.frame, text='...', 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 file = filedialog.askopenfilename( filetypes=[('Excel', '.xlsx')], title='Seleccione un archivo de subasta...') main_logger.info(f'Archivo seleccionado: {file}') self.file_path.set(file) def run(self): if not self.file_path.get(): main_logger.error('Debe seleccionar un archivo para ejecutarlo') return self.button_run.configure(state='disabled') main_logger.info('Ejecutando') # self.runsdt_thread = threading.Thread( # target=runsdt.main, args=[self.file_path.get()]) # self.runsdt_thread.start() self.runsdt_thread = RunsdtThread(args=(self.file_path.get())) self.runsdt_thread.start() self.check_thread() 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 ThirdUi: 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.BOTH, 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) # max_size = 100 # print(self.q.qsize()) # while self.q.qsize(): # try: # active_tread = self.q.get(0) # print(active_tread) # main_logger.info(active_tread) # if active_tread: # step = 1 # self.progress_bar.step(step) # if step == max_size: # step = 1 # else: # step += 1 # except queue.Empty: # pass # self.frame.after(100, self.check_q) class App: def __init__(self, root): self.root = root pbqueue = queue.Queue() root.title('SimSDT') root.iconbitmap("app.ico") root.columnconfigure(0, weight=1) root.rowconfigure(0, weight=1) # Create the panes and frames vertical_pane = ttk.PanedWindow(self.root, orient=VERTICAL) vertical_pane.grid(row=0, column=0, sticky="nsew") horizontal_pane = ttk.PanedWindow(vertical_pane, orient=HORIZONTAL) vertical_pane.add(horizontal_pane) form_frame = ttk.Labelframe(horizontal_pane, text="Configuracion") form_frame.columnconfigure(1, weight=1) horizontal_pane.add(form_frame, weight=1) console_frame = ttk.Labelframe(horizontal_pane, text="Console") 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") vertical_pane.add(third_frame, weight=1) # Initialize all frames self.form = FormUi(form_frame, pbqueue) self.console = ConsoleUi(console_frame) self.third = ThirdUi(third_frame, pbqueue) self.root.protocol('WM_DELETE_WINDOW', self.quit) self.root.bind('', self.quit) signal.signal(signal.SIGINT, self.quit) def quit(self, *args): msg = 'Se esta ejecutando un proceso de optimización\n\n¿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()