#!/usr/bin/env python3 import threading from os import path import tkinter as tk from tkinter import Text import configparser from sys import platform import pystray from pystray import MenuItem as item from PIL import Image from zope import component import interfaces from interfaces import endpoint from interfaces.message import IMessage, Debug, Error import consumer from configuration import Mapping script_path = path.dirname(path.realpath(__file__)) config_file = path.join(script_path, "config.ini") config = configparser.ConfigParser(delimiters="=") config.read(config_file) # # How to connect to the peripherical # component.provideAdapter(interfaces.endpoint.EndPoint) if config.has_section("connection.serial"): from serial_conn import SerialConnection endpoint = component.queryAdapter( SerialConnection(config["connection.serial"]), endpoint.IEndpoint) elif config.has_section("connection.socket"): from socket_conn import SocketConnection endpoint = component.queryAdapter( SocketConnection(config["connection.socket"]), endpoint.IEndpoint) component.provideUtility(endpoint, interfaces.endpoint.IEndpoint) endpoint.connect() if config.has_section("socket.serve"): import socketserver server = socketserver.Handler(config["socket.serve"]) else: server = None handler = consumer.SocketMessageConsumer() handler.start() class Icon(): """Icon displayed in the notification bar.""" def __init__(self, image): menu=( item('Quit', self.quit_window), item('Show', self.show_window, default=True), item('Reset',self.reset), ) self.icon=pystray.Icon("name", image, "Macropad companion", menu) self.stop = threading.Event() self.show_hide = threading.Event() # Start the icon into a new thread in order to keep the main loop control icon_thread = threading.Thread(target=self.icon.run) self.icon_thread = icon_thread def start(self): """ Start the icon. Handler is runned in a dedicated thread to avoid blocking the events from the main loop. """ self.icon_thread.start() def quit(self): self.icon.stop() def quit_window(self): self.stop.set() def show_window(self): self.show_hide.set() def reset(self): """ Read the configuration file again, and update the mapping list. """ config.read(config_file) mapping.set(config) class Application(): """ The main application. """ def __init__(self): # Override the default function called when exception are reported tk.Tk.report_callback_exception = self.report_callback_exception self.window = tk.Tk() self.text = Text(self.window, height=8) self.text.tag_config("error", background="red", foreground="white") ## State of the #pplication self.running = True self.visible = False self.focused_window = None self.last_layout = None component.provideHandler(self.log) # Window property self.window.withdraw() self.window.title("Macropad companion") icon = path.join(script_path, "favicon.ico") try: self.window.iconbitmap(icon) except: pass self.text.pack() # When closing, return back to the iconified mode self.window.protocol("WM_DELETE_WINDOW", self.hide) # Start the application in iconified mode image=Image.open(icon) self.icon = Icon(image) self.icon.start() component.handle(Debug("Started")) def connect_desktop(self): """ Launch the thread listening events from the desktop """ component.handle(Debug(platform)) listener = None if platform == "win32": import win32 listener = win32.Listener(mapping) elif platform == 'linux': import xlib listener = xlib.Listener(mapping) if listener is not None: component.provideUtility(listener, interfaces.desktopEvent.IDesktop) listener.start() else: component.handle(Error("Unsupported system %s" % platform)) def report_callback_exception(self, exc, val, tb): """ Handle exception reported inside the Tk application. This method overrid the default Tk.tk.report_callback_exception method. """ import traceback traceback.print_exception(exc, value=val, tb=tb) self.icon.stop.set() self.icon.quit() self.running = False self.window.destroy() component.queryUtility(interfaces.desktopEvent.IDesktop).stop() return def hide(self): self.icon.show_hide.clear() self.visible = False self.window.withdraw() self.update() def update(self): if self.icon.stop.is_set(): print("stopping") self.icon.quit() self.running = False self.window.destroy() component.queryUtility(interfaces.desktopEvent.IDesktop).stop() return if self.icon.show_hide.is_set(): if not self.visible: self.window.deiconify() else: self.window.withdraw() self.icon.show_hide.clear() self.visible = not self.visible @component.adapter(IMessage) def log(self, message : IMessage): """ Log a message. The message is printed in the console, and displayed in the application window. """ print(message.content) try: self.text.insert("1.0", "\n") self.text.insert("1.0", message.content) self.text.delete("200.0", "end") if message.level == 0: self.text.tag_add("error", "1.0", "1.end") self.icon.show_hide.set() except Exception as e: print(e) def exec(self): try: self.update() if server is not None: server.update() except BaseException as e: component.handle(Debug( str(e) )) print(e) # Got any error, stop the application properly self.icon.stop.set() if app.running: self.window.after(200, self.exec) if __name__ == '__main__': # Start the main application, Initializing the message listener before # listening desktop events app = Application() # The application is started, ready to display any error message. mapping = Mapping(config) component.provideUtility(mapping, interfaces.configuration.IConfiguration) app.connect_desktop() # Initialize the main loop app.exec() try: app.window.mainloop() except BaseException: app.running = False app.window.destroy() app.icon.quit() desktop = component.queryUtility(interfaces.desktopEvent.IDesktop) if desktop is not None: desktop.stop()