begin enforcing types better; glade updates

This commit is contained in:
itdominator 2022-03-24 22:15:08 -05:00
parent 2278cdc0c3
commit 0539fa41f0
9 changed files with 740 additions and 741 deletions

View File

@ -26,41 +26,41 @@ class EventSystem(IPCServer):
# Makeshift fake "events" type system FIFO # Makeshift fake "events" type system FIFO
def _pop_gui_event(self): def _pop_gui_event(self) -> None:
if len(self._gui_events) > 0: if len(self._gui_events) > 0:
return self._gui_events.pop(0) return self._gui_events.pop(0)
return None return None
def _pop_module_event(self): def _pop_module_event(self) -> None:
if len(self._module_events) > 0: if len(self._module_events) > 0:
return self._module_events.pop(0) return self._module_events.pop(0)
return None return None
def push_gui_event(self, event): def push_gui_event(self, eventevent: list) -> None:
if len(event) == 3: if len(event) == 3:
self._gui_events.append(event) self._gui_events.append(event)
return None return None
raise Exception("Invald event format! Please do: [type, target, (data,)]") 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: if len(event) == 3:
self._module_events.append(event) self._module_events.append(event)
return None return None
raise Exception("Invald event format! Please do: [type, target, (data,)]") raise Exception("Invald event format! Please do: [type, target, (data,)]")
def read_gui_event(self): def read_gui_event(self) -> list:
return self._gui_events[0] return self._gui_events[0]
def read_module_event(self): def read_module_event(self) -> list:
return self._module_events[0] return self._module_events[0]
def consume_gui_event(self): def consume_gui_event(self) -> list:
return self._pop_gui_event() return self._pop_gui_event()
def consume_module_event(self): def consume_module_event(self) -> list:
return self._pop_module_event() return self._pop_module_event()
@ -70,5 +70,5 @@ class EventSystem(IPCServer):
builtins.app_name = "SolarFM" builtins.app_name = "SolarFM"
builtins.event_system = EventSystem() builtins.event_system = EventSystem()
builtins.event_sleep_time = 0.2 builtins.event_sleep_time = 0.2
builtins.debug = False
builtins.trace_debug = False builtins.trace_debug = False
builtins.debug = False

View File

@ -2,6 +2,7 @@
import sys, os, signal import sys, os, signal
# Lib imports # Lib imports
import gi
from gi.repository import GLib from gi.repository import GLib
# Application imports # Application imports
@ -11,17 +12,17 @@ from plugins.plugins import Plugins
class State: class State:
wid = None wid: int = None
tid = None tid: int = None
tab = None tab: type = None
icon_grid = None icon_grid: gi.overrides.Gtk.IconView = None
store = None store: gi.overrides.Gtk.ListStore = None
class Controller_Data: class Controller_Data:
""" Controller_Data contains most of the state of the app at ay given time. It also has some support methods. """ """ 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.settings = _settings self.settings = _settings
self.builder = self.settings.get_builder() self.builder = self.settings.get_builder()
self.logger = self.settings.get_logger() self.logger = self.settings.get_logger()
@ -58,6 +59,7 @@ class Controller_Data:
self.trash_files_path = f"{GLib.get_user_data_dir()}/Trash/files" self.trash_files_path = f"{GLib.get_user_data_dir()}/Trash/files"
self.trash_info_path = f"{GLib.get_user_data_dir()}/Trash/info" self.trash_info_path = f"{GLib.get_user_data_dir()}/Trash/info"
self.icon_theme = self.settings.get_icon_theme()
# In compress commands: # In compress commands:
# %n: First selected filename/dir to archive # %n: First selected filename/dir to archive
@ -117,7 +119,7 @@ class Controller_Data:
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGINT, self.tear_down) GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGINT, self.tear_down)
def get_current_state(self): def get_current_state(self) -> State:
''' '''
Returns the state info most useful for any given context and action intent. Returns the state info most useful for any given context and action intent.
@ -132,14 +134,15 @@ class Controller_Data:
state.tab = self.get_fm_window(state.wid).get_tab_by_id(state.tid) state.tab = self.get_fm_window(state.wid).get_tab_by_id(state.tid)
state.icon_grid = self.builder.get_object(f"{state.wid}|{state.tid}|icon_grid") state.icon_grid = self.builder.get_object(f"{state.wid}|{state.tid}|icon_grid")
state.store = state.icon_grid.get_model() state.store = state.icon_grid.get_model()
return state return state
def clear_console(self): def clear_console(self) -> None:
''' Clears the terminal screen. ''' ''' Clears the terminal screen. '''
os.system('cls' if os.name == 'nt' else 'clear') 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 = None) -> type:
''' '''
Calls a method from scope of class. Calls a method from scope of class.
@ -156,11 +159,11 @@ class Controller_Data:
method = getattr(self, method_name, lambda data: f"No valid key passed...\nkey={method_name}\nargs={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() return method(data) if data else method()
def has_method(self, obj, name): def has_method(self, obj, name) -> type:
''' Checks if a given method exists. ''' ''' Checks if a given method exists. '''
return callable(getattr(obj, name, None)) return callable(getattr(obj, name, None))
def clear_children(self, widget): def clear_children(self, widget: type) -> None:
''' Clear children of a gtk widget. ''' ''' Clear children of a gtk widget. '''
for child in widget.get_children(): for child in widget.get_children():
widget.remove(child) widget.remove(child)

View File

@ -27,8 +27,10 @@ class GridMixin:
dir = tab.get_current_directory() dir = tab.get_current_directory()
files = tab.get_files() files = tab.get_files()
for i, file in enumerate(files): for file in files:
store.append([None, file[0]]) store.append([None, file[0]])
for i, file in enumerate(files):
self.create_icon(i, tab, store, dir, file[0]) self.create_icon(i, tab, store, dir, file[0])
# NOTE: Not likely called often from here but it could be useful # NOTE: Not likely called often from here but it could be useful
@ -41,40 +43,30 @@ class GridMixin:
GLib.idle_add(self.update_store, *(i, store, icon, tab, dir, file,)) GLib.idle_add(self.update_store, *(i, store, icon, tab, dir, file,))
def update_store(self, i, store, icon, tab, dir, file): def update_store(self, i, store, icon, tab, dir, file):
fpath = f"{dir}/{file}" while True:
loop = True
while loop:
try: try:
itr = store.get_iter(i) itr = store.get_iter(i)
loop = False break
except: except:
pass pass
if not icon: if not icon:
icon = self.get_system_thumbnail(fpath, tab.SYS_ICON_WH[0]) path = f"{dir}/{file}"
if not icon: icon = self.get_system_thumbnail(path, tab.SYS_ICON_WH[0])
icon = GdkPixbuf.Pixbuf.new_from_file(tab.DEFAULT_ICON)
if not icon:
icon = GdkPixbuf.Pixbuf.new_from_file(tab.DEFAULT_ICON)
store.set_value(itr, 0, icon) store.set_value(itr, 0, icon)
def get_system_thumbnail(self, filename, size): def get_system_thumbnail(self, filename, size):
try: try:
if os.path.exists(filename): gio_file = Gio.File.new_for_path(filename)
gioFile = Gio.File.new_for_path(filename) info = gio_file.query_info('standard::icon' , 0, None)
info = gioFile.query_info('standard::icon' , 0, Gio.Cancellable()) icon = info.get_icon().get_names()[0]
icon = info.get_icon().get_names()[0] icon_path = self.icon_theme.lookup_icon(icon , size , 0).get_filename()
iconTheme = Gtk.IconTheme.get_default() return GdkPixbuf.Pixbuf.new_from_file(icon_path)
iconData = iconTheme.lookup_icon(icon , size , 0)
if iconData:
iconPath = iconData.get_filename()
return GdkPixbuf.Pixbuf.new_from_file(iconPath)
else:
return None
else:
return None
except Exception as e: except Exception as e:
print("System icon generation issue:")
return None return None

View File

@ -11,15 +11,15 @@ from gi.repository import Gtk, Gio
class Plugin: class Plugin:
name = None name: str = None
module = None module: str = None
reference = None reference: type = None
class Plugins: class Plugins:
"""Plugins controller""" """Plugins controller"""
def __init__(self, settings): def __init__(self, settings: type):
self._settings = settings self._settings = settings
self._builder = self._settings.get_builder() self._builder = self._settings.get_builder()
self._plugins_path = self._settings.get_plugins_path() self._plugins_path = self._settings.get_plugins_path()
@ -27,11 +27,11 @@ class Plugins:
self._plugin_collection = [] self._plugin_collection = []
def launch_plugins(self): def launch_plugins(self) -> None:
self._set_plugins_watcher() self._set_plugins_watcher()
self.load_plugins() 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) \ self._plugins_dir_watcher = Gio.File.new_for_path(self._plugins_path) \
.monitor_directory(Gio.FileMonitorFlags.WATCH_MOVES, Gio.Cancellable()) .monitor_directory(Gio.FileMonitorFlags.WATCH_MOVES, Gio.Cancellable())
self._plugins_dir_watcher.connect("changed", self._on_plugins_changed, ()) self._plugins_dir_watcher.connect("changed", self._on_plugins_changed, ())
@ -43,7 +43,7 @@ class Plugins:
self.reload_plugins(file) self.reload_plugins(file)
# @threaded # @threaded
def load_plugins(self, file=None): def load_plugins(self, file: str = None) -> None:
print(f"Loading plugins...") print(f"Loading plugins...")
parent_path = os.getcwd() parent_path = os.getcwd()
@ -72,12 +72,12 @@ class Plugins:
os.chdir(parent_path) os.chdir(parent_path)
def reload_plugins(self, file=None): def reload_plugins(self, file: str = None) -> None:
print(f"Reloading plugins... stub.") print(f"Reloading plugins... stub.")
def send_message_to_plugin(self, type, data): def send_message_to_plugin(self, target: str , data: type) -> None:
print("Trying to send message to plugin...") print("Trying to send message to plugin...")
for plugin in self._plugin_collection: for plugin in self._plugin_collection:
if type in plugin.name: if target in plugin.name:
print('Found plugin; posting message...') print('Found plugin; posting message...')
plugin.reference.set_message(data) plugin.reference.set_message(data)

View File

@ -59,7 +59,7 @@ class Settings:
subpath = settings["base_of_home"] subpath = settings["base_of_home"]
HIDE_HIDDEN_FILES = True if settings["hide_hidden_files"] == "true" else False HIDE_HIDDEN_FILES = True if settings["hide_hidden_files"] == "true" else False
FFMPG_THUMBNLR = FFMPG_THUMBNLR if settings["thumbnailer_path"] == "" else settings["thumbnailer_path"] FFMPG_THUMBNLR = FFMPG_THUMBNLR if settings["thumbnailer_path"] == "" else settings["thumbnailer_path"]
go_past_home = True if settings["go_past_home"] == "" else settings["go_past_home"] go_past_home = True if settings["go_past_home"] == "" else settings["go_past_home"]
lock_folder = True if settings["lock_folder"] == "true" else False lock_folder = True if settings["lock_folder"] == "true" else False
locked_folders = settings["locked_folders"].split("::::") locked_folders = settings["locked_folders"].split("::::")
mplayer_options = settings["mplayer_options"].split() mplayer_options = settings["mplayer_options"].split()
@ -76,7 +76,7 @@ class Settings:
# Filters # Filters
fvideos = ('.mkv', '.avi', '.flv', '.mov', '.m4v', '.mpg', '.wmv', '.mpeg', '.mp4', '.webm') fvideos = ('.mkv', '.avi', '.flv', '.mov', '.m4v', '.mpg', '.wmv', '.mpeg', '.mp4', '.webm')
foffice = ('.doc', '.docx', '.xls', '.xlsx', '.xlt', '.xltx', '.xlm', '.ppt', 'pptx', '.pps', '.ppsx', '.odt', '.rtf') foffice = ('.doc', '.docx', '.xls', '.xlsx', '.xlt', '.xltx', '.xlm', '.ppt', 'pptx', '.pps', '.ppsx', '.odt', '.rtf')
fimages = ('.png', '.jpg', '.jpeg', '.gif', '.ico', '.tga') fimages = ('.png', '.jpg', '.jpeg', '.gif', '.ico', '.tga', '.webp')
ftext = ('.txt', '.text', '.sh', '.cfg', '.conf') ftext = ('.txt', '.text', '.sh', '.cfg', '.conf')
fmusic = ('.psf', '.mp3', '.ogg', '.flac', '.m4a') fmusic = ('.psf', '.mp3', '.ogg', '.flac', '.m4a')
fpdf = ('.pdf') fpdf = ('.pdf')

View File

@ -17,7 +17,7 @@ def threaded(fn):
class IPCServer: class IPCServer:
""" Create a listener so that other SolarFM instances send requests back to existing instance. """ """ Create a listener so that other SolarFM 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.is_ipc_alive = False
self._conn_type = conn_type self._conn_type = conn_type
self.ipc_authkey = b'solarfm-ipc' self.ipc_authkey = b'solarfm-ipc'
@ -31,7 +31,7 @@ class IPCServer:
@threaded @threaded
def create_ipc_server(self): def create_ipc_server(self) -> None:
if self._conn_type == "socket": if self._conn_type == "socket":
if os.path.exists(self.ipc_address): if os.path.exists(self.ipc_address):
return return
@ -44,7 +44,7 @@ class IPCServer:
self.is_ipc_alive = True self.is_ipc_alive = True
while True: while True:
conn = listener.accept() conn = listener.accept()
start_time = time.time() start_time = time.perf_counter()
print(f"New Connection: {listener.last_accepted}") print(f"New Connection: {listener.last_accepted}")
while True: while True:
@ -69,14 +69,14 @@ class IPCServer:
break break
# NOTE: Not perfect but insures we don't lock up the connection for too long. # NOTE: Not perfect but insures we don't lock up the connection for too long.
end_time = time.time() end_time = time.perf_counter()
if (end - start) > self.ipc_timeout: if (end - start) > self.ipc_timeout:
conn.close() conn.close()
listener.close() listener.close()
def send_ipc_message(self, message="Empty Data..."): def send_ipc_message(self, message: str = "Empty Data...") -> None:
try: try:
if self._conn_type == "socket": if self._conn_type == "socket":
conn = Client(address=self.ipc_address, family="AF_UNIX", authkey=self.ipc_authkey) conn = Client(address=self.ipc_address, family="AF_UNIX", authkey=self.ipc_authkey)
@ -86,5 +86,6 @@ class IPCServer:
conn.send(message) conn.send(message)
conn.send('close connection') conn.send('close connection')
conn.close()
except Exception as e: except Exception as e:
print(repr(e)) print(repr(e))

View File

@ -5,10 +5,10 @@ import os, logging
class Logger: class Logger:
def __init__(self, config_path): def __init__(self, config_path: str):
self._CONFIG_PATH = config_path 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. Create a new logging object and return it.
:note: :note:

View File

@ -31,6 +31,7 @@ class Settings:
self._KEY_BINDINGS = f"{self._CONFIG_PATH}/key-bindings.json" self._KEY_BINDINGS = f"{self._CONFIG_PATH}/key-bindings.json"
self._DEFAULT_ICONS = f"{self._CONFIG_PATH}/icons" self._DEFAULT_ICONS = f"{self._CONFIG_PATH}/icons"
self._WINDOW_ICON = f"{self._DEFAULT_ICONS}/{app_name.lower()}.png" self._WINDOW_ICON = f"{self._DEFAULT_ICONS}/{app_name.lower()}.png"
self._ICON_THEME = Gtk.IconTheme.get_default()
if not os.path.exists(self._CONFIG_PATH): if not os.path.exists(self._CONFIG_PATH):
os.mkdir(self._CONFIG_PATH) os.mkdir(self._CONFIG_PATH)
@ -63,12 +64,12 @@ class Settings:
self._builder.add_from_file(self._GLADE_FILE) self._builder.add_from_file(self._GLADE_FILE)
def create_window(self): def create_window(self) -> None:
# Get window and connect signals # Get window and connect signals
self._main_window = self._builder.get_object("Main_Window") self._main_window = self._builder.get_object("Main_Window")
self._set_window_data() 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) self._main_window.set_icon_from_file(self._WINDOW_ICON)
screen = self._main_window.get_screen() screen = self._main_window.get_screen()
visual = screen.get_rgba_visual() visual = screen.get_rgba_visual()
@ -85,29 +86,28 @@ class Settings:
styleContext = Gtk.StyleContext() styleContext = Gtk.StyleContext()
styleContext.add_provider_for_screen(screen, cssProvider, Gtk.STYLE_PROVIDER_PRIORITY_USER) styleContext.add_provider_for_screen(screen, cssProvider, Gtk.STYLE_PROVIDER_PRIORITY_USER)
def _area_draw(self, widget, cr): def _area_draw(self, widget: Gtk.ApplicationWindow, cr: cairo.Context) -> None:
cr.set_source_rgba(0, 0, 0, 0.54) cr.set_source_rgba(0, 0, 0, 0.54)
cr.set_operator(cairo.OPERATOR_SOURCE) cr.set_operator(cairo.OPERATOR_SOURCE)
cr.paint() cr.paint()
cr.set_operator(cairo.OPERATOR_OVER) cr.set_operator(cairo.OPERATOR_OVER)
def get_monitor_data(self): def get_monitor_data(self) -> list:
screen = self._builder.get_object("Main_Window").get_screen() screen = self._builder.get_object("Main_Window").get_screen()
monitors = [] monitors = []
for m in range(screen.get_n_monitors()): for m in range(screen.get_n_monitors()):
monitors.append(screen.get_monitor_geometry(m)) monitors.append(screen.get_monitor_geometry(m))
for monitor in monitors:
print("{}x{}+{}+{}".format(monitor.width, monitor.height, monitor.x, monitor.y)) print("{}x{}+{}+{}".format(monitor.width, monitor.height, monitor.x, monitor.y))
return monitors return monitors
def get_builder(self): return self._builder def get_main_window(self) -> Gtk.ApplicationWindow: return self._main_window
def get_logger(self): return self._logger def get_builder(self) -> Gtk.Builder: return self._builder
def get_keybindings(self): return self._keybindings def get_logger(self) -> Logger: return self._logger
def get_main_window(self): return self._main_window def get_keybindings(self) -> Keybindings: return self._keybindings
def get_plugins_path(self): return self._PLUGINS_PATH def get_plugins_path(self) -> str: return self._PLUGINS_PATH
def get_icon_theme(self) -> str: return self._ICON_THEME
def get_success_color(self): return self._success_color def get_success_color(self) -> str: return self._success_color
def get_warning_color(self): return self._warning_color def get_warning_color(self) -> str: return self._warning_color
def get_error_color(self): return self._error_color def get_error_color(self) -> str: return self._error_color

File diff suppressed because it is too large Load Diff