enforcing type better
This commit is contained in:
parent
fdccf77367
commit
88db4171cf
@ -26,41 +26,39 @@ class EventSystem(IPCServer):
|
||||
|
||||
|
||||
# Makeshift fake "events" type system FIFO
|
||||
def _pop_gui_event(self):
|
||||
def _pop_gui_event(self) -> None:
|
||||
if len(self._gui_events) > 0:
|
||||
return self._gui_events.pop(0)
|
||||
return None
|
||||
|
||||
def _pop_module_event(self):
|
||||
def _pop_module_event(self) -> None:
|
||||
if len(self._module_events) > 0:
|
||||
return self._module_events.pop(0)
|
||||
return None
|
||||
|
||||
|
||||
def push_gui_event(self, event):
|
||||
def push_gui_event(self, event: list) -> None:
|
||||
if len(event) == 3:
|
||||
self._gui_events.append(event)
|
||||
return None
|
||||
|
||||
raise Exception("Invald event format! Please do: [type, target, (data,)]")
|
||||
|
||||
def push_module_event(self, event):
|
||||
def push_module_event(self, event: list) -> None:
|
||||
if len(event) == 3:
|
||||
self._module_events.append(event)
|
||||
return None
|
||||
|
||||
raise Exception("Invald event format! Please do: [type, target, (data,)]")
|
||||
|
||||
def read_gui_event(self):
|
||||
def read_gui_event(self) -> None:
|
||||
return self._gui_events[0]
|
||||
|
||||
def read_module_event(self):
|
||||
def read_module_event(self) -> None:
|
||||
return self._module_events[0]
|
||||
|
||||
def consume_gui_event(self):
|
||||
def consume_gui_event(self) -> None:
|
||||
return self._pop_gui_event()
|
||||
|
||||
def consume_module_event(self):
|
||||
def consume_module_event(self) -> None:
|
||||
return self._pop_module_event()
|
||||
|
||||
|
||||
@ -70,5 +68,5 @@ class EventSystem(IPCServer):
|
||||
builtins.app_name = "<change_me>"
|
||||
builtins.event_system = EventSystem()
|
||||
builtins.event_sleep_time = 0.2
|
||||
builtins.debug = False
|
||||
builtins.trace_debug = False
|
||||
builtins.debug = False
|
||||
|
||||
@ -36,7 +36,7 @@ class Controller(DummyMixin, Controller_Data):
|
||||
|
||||
|
||||
@threaded
|
||||
def gui_event_observer(self):
|
||||
def gui_event_observer(self) -> None:
|
||||
while True:
|
||||
time.sleep(event_sleep_time)
|
||||
event = event_system.consume_gui_event()
|
||||
@ -48,10 +48,10 @@ class Controller(DummyMixin, Controller_Data):
|
||||
except Exception as e:
|
||||
print(repr(e))
|
||||
|
||||
def handle_file_from_ipc(self, path):
|
||||
def handle_file_from_ipc(self, path: str) -> None:
|
||||
print(f"Path From IPC: {path}")
|
||||
|
||||
def on_global_key_release_controller(self, widget, event):
|
||||
def on_global_key_release_controller(self, widget: type, event: type) -> None:
|
||||
"""Handler for keyboard events"""
|
||||
keyname = Gdk.keyval_name(event.keyval).lower()
|
||||
if keyname.replace("_l", "").replace("_r", "") in ["control", "alt", "shift"]:
|
||||
@ -73,13 +73,13 @@ class Controller(DummyMixin, Controller_Data):
|
||||
|
||||
|
||||
|
||||
def get_clipboard_data(self):
|
||||
def get_clipboard_data(self) -> str:
|
||||
proc = subprocess.Popen(['xclip','-selection', 'clipboard', '-o'], stdout=subprocess.PIPE)
|
||||
retcode = proc.wait()
|
||||
data = proc.stdout.read()
|
||||
return data.decode("utf-8").strip()
|
||||
|
||||
def set_clipboard_data(self, data):
|
||||
def set_clipboard_data(self, data: type) -> None:
|
||||
proc = subprocess.Popen(['xclip','-selection','clipboard'], stdin=subprocess.PIPE)
|
||||
proc.stdin.write(data)
|
||||
proc.stdin.close()
|
||||
|
||||
@ -11,7 +11,7 @@ from plugins.plugins import Plugins
|
||||
class Controller_Data:
|
||||
''' Controller_Data contains most of the state of the app at ay given time. It also has some support methods. '''
|
||||
|
||||
def setup_controller_data(self, _settings):
|
||||
def setup_controller_data(self, _settings: type) -> None:
|
||||
self.plugins = Plugins(_settings)
|
||||
|
||||
self.settings = _settings
|
||||
@ -30,11 +30,11 @@ class Controller_Data:
|
||||
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGINT, self.tear_down)
|
||||
|
||||
|
||||
def clear_console(self):
|
||||
def clear_console(self) -> None:
|
||||
''' Clears the terminal screen. '''
|
||||
os.system('cls' if os.name == 'nt' else 'clear')
|
||||
|
||||
def call_method(self, _method_name, data = None):
|
||||
def call_method(self, _method_name: str, data: type) -> type:
|
||||
'''
|
||||
Calls a method from scope of class.
|
||||
|
||||
@ -51,11 +51,11 @@ class Controller_Data:
|
||||
method = getattr(self, method_name, lambda data: f"No valid key passed...\nkey={method_name}\nargs={data}")
|
||||
return method(data) if data else method()
|
||||
|
||||
def has_method(self, obj, name):
|
||||
def has_method(self, obj: type, name: type) -> type:
|
||||
''' Checks if a given method exists. '''
|
||||
return callable(getattr(obj, name, None))
|
||||
|
||||
def clear_children(self, widget):
|
||||
def clear_children(self, widget: type) -> None:
|
||||
''' Clear children of a gtk widget. '''
|
||||
for child in widget.get_children():
|
||||
widget.remove(child)
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
class DummyMixin:
|
||||
""" DummyMixin is an example of how mixins are used and structured in a project. """
|
||||
def print_hello_world(self):
|
||||
def print_hello_world(self) -> None:
|
||||
print("Hello, World!")
|
||||
|
||||
@ -11,15 +11,15 @@ from gi.repository import Gtk, Gio
|
||||
|
||||
|
||||
class Plugin:
|
||||
name = None
|
||||
module = None
|
||||
reference = None
|
||||
name: str = None
|
||||
module: str = None
|
||||
reference: type = None
|
||||
|
||||
|
||||
class Plugins:
|
||||
"""Plugins controller"""
|
||||
|
||||
def __init__(self, settings):
|
||||
def __init__(self, settings: type):
|
||||
self._settings = settings
|
||||
self._builder = self._settings.get_builder()
|
||||
self._plugins_path = self._settings.get_plugins_path()
|
||||
@ -27,11 +27,11 @@ class Plugins:
|
||||
self._plugin_collection = []
|
||||
|
||||
|
||||
def launch_plugins(self):
|
||||
def launch_plugins(self) -> None:
|
||||
self._set_plugins_watcher()
|
||||
self.load_plugins()
|
||||
|
||||
def _set_plugins_watcher(self):
|
||||
def _set_plugins_watcher(self) -> None:
|
||||
self._plugins_dir_watcher = Gio.File.new_for_path(self._plugins_path) \
|
||||
.monitor_directory(Gio.FileMonitorFlags.WATCH_MOVES, Gio.Cancellable())
|
||||
self._plugins_dir_watcher.connect("changed", self._on_plugins_changed, ())
|
||||
@ -42,7 +42,7 @@ class Plugins:
|
||||
Gio.FileMonitorEvent.MOVED_OUT]:
|
||||
self.reload_plugins(file)
|
||||
|
||||
def load_plugins(self, file=None):
|
||||
def load_plugins(self, file: str = None) -> None:
|
||||
print(f"Loading plugins...")
|
||||
parent_path = os.getcwd()
|
||||
|
||||
@ -71,8 +71,12 @@ class Plugins:
|
||||
os.chdir(parent_path)
|
||||
|
||||
|
||||
def reload_plugins(self, file=None):
|
||||
print(f"Reloading plugins...")
|
||||
def reload_plugins(self, file: str = None) -> None:
|
||||
print(f"Reloading plugins... stub.")
|
||||
|
||||
def set_message_on_plugin(self, type, data):
|
||||
def send_message_to_plugin(self, target: str , data: type) -> None:
|
||||
print("Trying to send message to plugin...")
|
||||
for plugin in self._plugin_collection:
|
||||
if target in plugin.name:
|
||||
print('Found plugin; posting message...')
|
||||
plugin.reference.set_message(data)
|
||||
|
||||
@ -17,7 +17,7 @@ def threaded(fn):
|
||||
|
||||
class IPCServer:
|
||||
''' Create a listener so that other instances send requests back to existing instance. '''
|
||||
def __init__(self, conn_type="socket"):
|
||||
def __init__(self, conn_type: str = "socket"):
|
||||
self.is_ipc_alive = False
|
||||
self._conn_type = conn_type
|
||||
self.ipc_authkey = b'app-ipc'
|
||||
@ -30,7 +30,7 @@ class IPCServer:
|
||||
self.ipc_port = 8888
|
||||
|
||||
@threaded
|
||||
def create_ipc_server(self):
|
||||
def create_ipc_server(self) -> None:
|
||||
if self._conn_type == "socket":
|
||||
if os.path.exists(self.ipc_address):
|
||||
return
|
||||
@ -74,7 +74,7 @@ class IPCServer:
|
||||
listener.close()
|
||||
|
||||
|
||||
def send_ipc_message(self, message="Empty Data..."):
|
||||
def send_ipc_message(self, message: str = "Empty Data...") -> None:
|
||||
try:
|
||||
if self._conn_type == "socket":
|
||||
conn = Client(address=self.ipc_address, family="AF_UNIX", authkey=self.ipc_authkey)
|
||||
@ -83,5 +83,6 @@ class IPCServer:
|
||||
|
||||
conn.send(message)
|
||||
conn.send('close connection')
|
||||
conn.close()
|
||||
except Exception as e:
|
||||
print(repr(e))
|
||||
|
||||
@ -5,10 +5,10 @@ import os, logging
|
||||
|
||||
|
||||
class Logger:
|
||||
def __init__(self, config_path):
|
||||
def __init__(self, config_path: str):
|
||||
self._CONFIG_PATH = config_path
|
||||
|
||||
def get_logger(self, loggerName = "NO_LOGGER_NAME_PASSED", createFile = True):
|
||||
def get_logger(self, loggerName: str = "NO_LOGGER_NAME_PASSED", createFile: bool = True) -> logging.Logger:
|
||||
"""
|
||||
Create a new logging object and return it.
|
||||
:note:
|
||||
|
||||
@ -69,12 +69,12 @@ class Settings:
|
||||
|
||||
|
||||
|
||||
def create_window(self):
|
||||
def create_window(self) -> None:
|
||||
# Get window and connect signals
|
||||
self._main_window = self._builder.get_object("Main_Window")
|
||||
self.set_window_data()
|
||||
|
||||
def set_window_data(self):
|
||||
def set_window_data(self) -> None:
|
||||
self._main_window.set_icon_from_file(self._WINDOW_ICON)
|
||||
screen = self._main_window.get_screen()
|
||||
visual = screen.get_rgba_visual()
|
||||
@ -91,13 +91,11 @@ class Settings:
|
||||
styleContext = Gtk.StyleContext()
|
||||
styleContext.add_provider_for_screen(screen, cssProvider, Gtk.STYLE_PROVIDER_PRIORITY_USER)
|
||||
|
||||
def get_monitor_data(self):
|
||||
def get_monitor_data(self) -> list:
|
||||
screen = self._builder.get_object("Main_Window").get_screen()
|
||||
monitors = []
|
||||
for m in range(screen.get_n_monitors()):
|
||||
monitors.append(screen.get_monitor_geometry(m))
|
||||
|
||||
for monitor in monitors:
|
||||
print("{}x{}|{}+{}".format(monitor.width, monitor.height, monitor.x, monitor.y))
|
||||
|
||||
return monitors
|
||||
@ -110,22 +108,21 @@ class Settings:
|
||||
|
||||
|
||||
|
||||
|
||||
def get_builder(self): return self._builder
|
||||
def get_logger(self): return self._logger
|
||||
def get_keybindings(self): return self._keybindings
|
||||
def get_main_window(self): return self._main_window
|
||||
def get_home_path(self): return self._USER_HOME
|
||||
def get_plugins_path(self): return self._PLUGINS_PATH
|
||||
def get_main_window(self) -> Gtk.ApplicationWindow: return self._main_window
|
||||
def get_builder(self) -> Gtk.Builder: return self._builder
|
||||
def get_logger(self) -> Logger: return self._logger
|
||||
def get_keybindings(self) -> Keybindings: return self._keybindings
|
||||
def get_plugins_path(self) -> str: return self._PLUGINS_PATH
|
||||
def get_home_path(self) -> str: return self._USER_HOME
|
||||
|
||||
# Filter returns
|
||||
def get_office_filter(self): return self._office_filter
|
||||
def get_vids_filter(self): return self._vids_filter
|
||||
def get_text_filter(self): return self._txt_filter
|
||||
def get_music_filter(self): return self._music_filter
|
||||
def get_images_filter(self): return self._images_filter
|
||||
def get_pdf_filter(self): return self._pdf_filter
|
||||
def get_office_filter(self) -> tuple: return self._office_filter
|
||||
def get_vids_filter(self) -> tuple: return self._vids_filter
|
||||
def get_text_filter(self) -> tuple: return self._txt_filter
|
||||
def get_music_filter(self) -> tuple: return self._music_filter
|
||||
def get_images_filter(self) -> tuple: return self._images_filter
|
||||
def get_pdf_filter(self) -> tuple: return self._pdf_filter
|
||||
|
||||
def get_success_color(self): return self._success_color
|
||||
def get_warning_color(self): return self._warning_color
|
||||
def get_error_color(self): return self._error_color
|
||||
def get_success_color(self) -> str: return self._success_color
|
||||
def get_warning_color(self) -> str: return self._warning_color
|
||||
def get_error_color(self) -> str: return self._error_color
|
||||
|
||||
Loading…
Reference in New Issue
Block a user