itdominator/UTop
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Major refactor to move away from glade file

Browse Source
This commit is contained in:
itdominator
2025-11-29 23:16:01 -06:00
parent 08587c16b8
commit d2d266adda
83 changed files with 3091 additions and 1434 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/.python-version
View File

@@ -1 +0,0 @@
3.10.4

61
src/__builtins__.py
View File

@@ -1,42 +1,77 @@
import builtins, threading
# Python imports
import builtins
import traceback
import threading
import sys
# Lib imports
# Application imports
from utils.event_system import EventSystem
from utils.endpoint_registry import EndpointRegistry
from utils.keybindings import Keybindings
from utils.settings import Settings
# from libs.db import DB
from libs.event_system import EventSystem
from libs.keybindings import Keybindings
from libs.logger import Logger
from libs.settings.manager import SettingsManager
# NOTE: Threads WILL NOT die with parent's destruction.
def threaded_wrapper(fn):
def wrapper(*args, **kwargs):
threading.Thread(target=fn, args=args, kwargs=kwargs, daemon=False).start()
thread = threading.Thread(target = fn, args = args, kwargs = kwargs, daemon = False)
thread.start()
return thread
return wrapper
# NOTE: Threads WILL die with parent's destruction.
def daemon_threaded_wrapper(fn):
def wrapper(*args, **kwargs):
threading.Thread(target=fn, args=args, kwargs=kwargs, daemon=True).start()
thread = threading.Thread(target = fn, args = args, kwargs = kwargs, daemon = True)
thread.start()
return thread
return wrapper
def call_chain_wrapper(fn):
def wrapper(*args, **kwargs):
print()
print()
for line in traceback.format_stack():
print( line.strip() )
print()
print()
return fn(*args, **kwargs)
return wrapper
# NOTE: Just reminding myself we can add to builtins two different ways...
# __builtins__.update({"event_system": Builtins()})
builtins.app_name = "UTop"
builtins.APP_NAME = "UTop"
builtins.keybindings = Keybindings()
builtins.event_system = EventSystem()
builtins.endpoint_registry = EndpointRegistry()
builtins.settings = Settings()
builtins.logger = settings.get_logger()
builtins.settings_manager = SettingsManager()
settings_manager.load_settings()
builtins.settings = settings_manager.settings
builtins.logger = Logger(
settings_manager.get_home_config_path(), \
_ch_log_lvl = settings.debugging.ch_log_lvl, \
_fh_log_lvl = settings.debugging.fh_log_lvl
).get_logger()
builtins.threaded = threaded_wrapper
builtins.daemon_threaded = daemon_threaded_wrapper
builtins.event_sleep_time = 0.05
builtins.call_chain = call_chain_wrapper
def custom_except_hook(exc_type, exc_value, exc_traceback):
if issubclass(exc_type, KeyboardInterrupt):
sys.__excepthook__(exc_type, exc_value, exc_traceback)
return
logger.error("Uncaught exception", exc_info = (exc_type, exc_value, exc_traceback))
sys.excepthook = custom_except_hook

4
src/__init__.py
View File

@@ -1,3 +1,3 @@
"""
Start of package.
"""
Src Package.
"""

60
src/__main__.py
View File

@@ -9,44 +9,50 @@ from setproctitle import setproctitle
import tracemalloc
tracemalloc.start()
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from __builtins__ import *
from app import Application
def main():
setproctitle(f'{APP_NAME}')
settings_manager.set_start_load_time()
if args.debug == "true":
settings_manager.set_debug(True)
if args.trace_debug == "true":
settings_manager.set_trace_debug(True)
settings_manager.do_dirty_start_check()
app = Application()
app.run()
if __name__ == "__main__":
''' Set process title, get arguments, and create GTK main thread. '''
parser = argparse.ArgumentParser()
# Add long and short arguments
parser.add_argument("--debug", "-d", default = "false", help = "Do extra console messaging.")
parser.add_argument("--trace-debug", "-td", default = "false", help = "Disable saves, ignore IPC lock, do extra console messaging.")
parser.add_argument("--no-plugins", "-np", default = "false", help = "Do not load plugins.")
parser.add_argument("--new-tab", "-nt", default = "false", help = "Opens a 'New Tab' if a handler is set for it.")
parser.add_argument("--file", "-f", default = "default", help = "JUST SOME FILE ARG.")
# Read arguments (If any...)
args, unknownargs = parser.parse_known_args()
settings_manager.set_starting_args( args, unknownargs )
try:
setproctitle(f'{app_name}')
faulthandler.enable() # For better debug info
parser = argparse.ArgumentParser()
# Add long and short arguments
parser.add_argument("--debug", "-d", default="false", help="Do extra console messaging.")
parser.add_argument("--trace-debug", "-td", default="false", help="Disable saves, ignore IPC lock, do extra console messaging.")
parser.add_argument("--file", "-f", default="default", help="JUST SOME FILE ARG.")
parser.add_argument("--new-tab", "-t", default="", help="Open a file into new tab.")
parser.add_argument("--new-window", "-w", default="", help="Open a file into a new window.")
# Read arguments (If any...)
args, unknownargs = parser.parse_known_args()
if args.debug == "true":
settings.set_debug(True)
if args.trace_debug == "true":
settings.set_trace_debug(True)
settings.do_dirty_start_check()
Application(args, unknownargs)
Gtk.main()
main()
except Exception as e:
traceback.print_exc()
quit()
quit()

66
src/app.py
View File

@@ -1,11 +1,13 @@
# Python imports
from contextlib import suppress
import signal
import os
import time
# Lib imports
# Application imports
from utils.ipc_server import IPCServer
from libs.debugging import debug_signal_handler
from libs.ipc_server import IPCServer
from core.window import Window
@@ -13,27 +15,53 @@ from core.window import Window
class AppLaunchException(Exception):
...
class ControllerStartExceptiom(Exception):
...
class Application(IPCServer):
''' Create Settings and Controller classes. Bind signal to Builder. Inherit from Builtins to bind global methods and classes.'''
class Application:
""" docstring for Application. """
def __init__(self, args, unknownargs):
def __init__(self):
super(Application, self).__init__()
if not settings.is_trace_debug():
try:
self.create_ipc_listener()
except Exception:
...
if not self.is_ipc_alive:
for arg in unknownargs + [args.new_tab,]:
if os.path.isdir(arg):
message = f"FILE|{arg}"
self.send_ipc_message(message)
if not settings_manager.is_trace_debug():
self.load_ipc()
raise AppLaunchException(f"{app_name} IPC Server Exists: Will send path(s) to it and close...")
self.setup_debug_hook()
def run(self):
win = Window()
win.start()
def load_ipc(self):
args, \
unknownargs = settings_manager.get_starting_args()
ipc_server = IPCServer()
self.ipc_realization_check(ipc_server)
if not ipc_server.is_ipc_alive:
for arg in unknownargs + [args.new_tab,]:
if os.path.isfile(arg):
message = f"FILE|{arg}"
ipc_server.send_ipc_message(message)
raise AppLaunchException(f"{APP_NAME} IPC Server Exists: Have sent path(s) to it and closing...")
def ipc_realization_check(self, ipc_server):
try:
ipc_server.create_ipc_listener()
except Exception:
ipc_server.send_test_ipc_message()
with suppress(Exception):
ipc_server.create_ipc_listener()
def setup_debug_hook(self):
# Typically: ValueError: signal only works in main thread
with suppress(ValueError):
# kill -SIGUSR2 <pid> from Linux/Unix or SIGBREAK signal from Windows
signal.signal(
vars(signal).get("SIGBREAK") or vars(signal).get("SIGUSR2"),
debug_signal_handler
)
Window(args, unknownargs)

4
src/core/__init__.py
View File

@@ -1,3 +1,3 @@
"""
Gtk Bound Signal Module
"""
Core Package
"""

33
src/core/builder_wrapper.py Normal file
View File

@@ -0,0 +1,33 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
class BuilderWrapper(Gtk.Builder):
"""docstring for BuilderWrapper."""
def __init__(self):
super(BuilderWrapper, self).__init__()
self.objects = {}
def get_object(self, id: str, use_gtk: bool = True) -> any:
if not use_gtk:
return self.objects[id]
return super(BuilderWrapper, self).get_object(id)
def expose_object(self, id: str, object: any, use_gtk: bool = True) -> None:
if not use_gtk:
self.objects[id] = object
else:
super(BuilderWrapper, self).expose_object(id, object)
def dereference_object(self, id: str) -> None:
del self.objects[id]

3
src/core/containers/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""
Containers Package
"""

42
src/core/containers/base_container.py Normal file
View File

@@ -0,0 +1,42 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from .header_container import HeaderContainer
from .body_container import BodyContainer
from .footer_container import FooterContainer
class BaseContainer(Gtk.Box):
def __init__(self):
super(BaseContainer, self).__init__()
self.ctx = self.get_style_context()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show()
def _setup_styling(self):
self.set_orientation(Gtk.Orientation.VERTICAL)
self.ctx.add_class("base-container")
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
self.add(HeaderContainer())
self.add(BodyContainer())
self.add(FooterContainer())

43
src/core/containers/body_container.py Normal file
View File

@@ -0,0 +1,43 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from .left_container import LeftContainer
from .center_container import CenterContainer
from .right_container import RightContainer
class BodyContainer(Gtk.Box):
def __init__(self):
super(BodyContainer, self).__init__()
self.ctx = self.get_style_context()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show()
def _setup_styling(self):
self.set_orientation(Gtk.Orientation.HORIZONTAL)
self.ctx.add_class("body-container")
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
self.add(LeftContainer())
self.add(CenterContainer())
self.add(RightContainer())

43
src/core/containers/center_container.py Normal file
View File

@@ -0,0 +1,43 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from ..widgets.tab_widget import TabWidget
class CenterContainer(Gtk.Box):
def __init__(self):
super(CenterContainer, self).__init__()
self._builder = settings_manager.get_builder()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show()
def _setup_styling(self):
self.set_orientation(Gtk.Orientation.VERTICAL)
self.set_hexpand(True)
self.set_vexpand(True)
ctx = self.get_style_context()
ctx.add_class("center-container")
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
self.add(TabWidget())

42
src/core/containers/footer_container.py Normal file
View File

@@ -0,0 +1,42 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from ..widgets.clock_widget import ClockWidget
class FooterContainer(Gtk.Box):
def __init__(self):
super(FooterContainer, self).__init__()
self.ctx = self.get_style_context()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show()
def _setup_styling(self):
self.set_hexpand(True)
self.set_orientation(Gtk.Orientation.HORIZONTAL)
self.ctx.add_class("footer-container")
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
# self.add(ClockWidget())
self.pack_end(ClockWidget(), False, False, 5)

46
src/core/containers/header_container.py Normal file
View File

@@ -0,0 +1,46 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from ..widgets.pager_widget import PagerWidget
class HeaderContainer(Gtk.ButtonBox):
def __init__(self):
super(HeaderContainer, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show()
def _setup_styling(self):
self.set_hexpand(True)
self.set_vexpand(True)
self.set_orientation(Gtk.Orientation.HORIZONTAL)
ctx = self.get_style_context()
ctx.add_class("header-container")
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
pager_widget = PagerWidget().get_widget()
pager_widget.show()
self.add(pager_widget)

40
src/core/containers/left_container.py Normal file
View File

@@ -0,0 +1,40 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from ..widgets.category_list_widget import CategoryListWidget
class LeftContainer(Gtk.Box):
def __init__(self):
super(LeftContainer, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show()
def _setup_styling(self):
self.set_orientation(Gtk.Orientation.VERTICAL)
self.set_vexpand(True)
ctx = self.get_style_context()
ctx.add_class("left-container")
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
self.add(CategoryListWidget())

40
src/core/containers/right_container.py Normal file
View File

@@ -0,0 +1,40 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from ..widgets.task_list_widget import TaskListWidget
class RightContainer(Gtk.Box):
def __init__(self):
super(RightContainer, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show()
def _setup_styling(self):
self.set_orientation(Gtk.Orientation.VERTICAL)
self.set_vexpand(True)
ctx = self.get_style_context()
ctx.add_class("right-container")
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
self.add( TaskListWidget() )

182
src/core/controller.py
View File

@@ -1,182 +0,0 @@
# Python imports
import os, subprocess
# Gtk imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('Gdk', '3.0')
gi.require_version('GdkPixbuf', '2.0')
from gi.repository import Gtk
from gi.repository import Gdk
from gi.repository import Gio
from gi.repository import GLib
from gi.repository import GdkPixbuf
# Application imports
from .controller_data import ControllerData
from .core_widget import CoreWidget
class Controller(ControllerData):
def __init__(self, args, unknownargs):
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self.setup_controller_data()
def _setup_styling(self):
...
def _setup_signals(self):
...
# NOTE: To be filled out after app data is actually working...
def search_for_entry(self, widget, data = None):
...
def set_list_group(self, widget):
group = widget.get_label().strip()
group_items = self.core_widget.get_group(group)
grid = self.builder.get_object("program_list_bttns")
children = grid.get_children()
for child in children:
child.disconnect(child.sig_id)
grid.remove(child)
row = 0
col = 0
icon_theme = Gtk.IconTheme.get_default()
for item in group_items:
button = self.generate_app_button(icon_theme, item)
grid.attach(button, col, row, 1, 1)
col += 1
if col == 4:
col = 0
row += 1
def generate_app_button(self, icon_theme, item):
title = item["title"]
exec_str = item[
"exec" if not item["exec"] in ("", None) else "tryExec"
]
button = Gtk.Button(label = title)
button.sig_id = button.connect("clicked", self.test_exec, exec_str)
if self.show_image:
_icon = item["icon"]
if os.path.exists(_icon):
icon = self.get_icon_from_path(_icon)
else:
icon = self.get_icon_from_gio(icon_theme, _icon)
button.set_image(icon)
button.set_always_show_image(True)
button.show_all()
return button
def get_icon_from_path(self, path):
pixbuf = GdkPixbuf.PixbufAnimation.new_from_file(path) \
.get_static_image() \
.scale_simple(32, 32, \
GdkPixbuf.InterpType.BILINEAR)
return Gtk.Image.new_from_pixbuf(pixbuf)
def get_icon_from_gio(self, icon_theme, icon_name):
gio_icon = Gio.Icon.new_for_string(icon_name)
pixbuf = None
# Note: https://docs.gtk.org/gtk3/enum.IconSize.html
for i in [6, 5, 3, 4, 2, 1]:
icon_info = Gtk.IconTheme.lookup_by_gicon(icon_theme, gio_icon, i, Gtk.IconLookupFlags.FORCE_REGULAR)
if not icon_info: continue
pixbuf = icon_info.load_icon().scale_simple(32, 32, 2) # 2 = BILINEAR and is best by default
break
return Gtk.Image.new_from_pixbuf( pixbuf )
def test_exec(self, widget, _command):
command = _command.split("%")[0]
DEVNULL = open(os.devnull, 'w')
subprocess.Popen(command.split(), start_new_session=True, stdout=DEVNULL, stderr=DEVNULL)
def _subscribe_to_events(self):
event_system.subscribe("handle_file_from_ipc", self.handle_file_from_ipc)
def handle_file_from_ipc(self, path: str) -> None:
print(f"Path From IPC: {path}")
def load_glade_file(self):
self.builder = Gtk.Builder()
self.builder.add_from_file(settings.get_glade_file())
settings.set_builder(self.builder)
self.core_widget = CoreWidget()
settings.register_signals_to_builder([self, self.core_widget])
def get_core_widget(self):
self.setup_toggle_event()
return self.core_widget
def on_hide_window(self, data = None):
"""Handle a request to hide/show the window"""
if not self.window.get_property('visible'):
self.window.show()
self.window.grab_focus()
# NOTE: Need here to enforce sticky after hide and reshow.
self.window.stick()
else:
self.hidefunc()
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"]:
if "control" in keyname:
self.ctrl_down = False
if "shift" in keyname:
self.shift_down = False
if "alt" in keyname:
self.alt_down = False
mapping = keybindings.lookup(event)
if mapping:
getattr(self, mapping)()
return True
else:
print(f"on_global_key_release_controller > key > {keyname}")
print(f"Add logic or remove this from: {self.__class__}")
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: type) -> None:
proc = subprocess.Popen(['xclip','-selection','clipboard'], stdin=subprocess.PIPE)
proc.stdin.write(data)
proc.stdin.close()
retcode = proc.wait()

103
src/core/controller_data.py
View File

@@ -1,103 +0,0 @@
# Python imports
import os
# Lib imports
import gi
from gi.repository import GObject
# Application imports
from plugins.plugins_controller import PluginsController
try:
from gi.repository import GdkX11
except ImportError:
logger.debug("Could not import X11 gir module...")
def display_manager():
""" Try to detect which display manager we are running under... """
if os.environ.get('WAYLAND_DISPLAY'):
return 'WAYLAND'
return 'X11' # Fallback assumption of X11
if display_manager() == 'X11':
try:
#gi.require_version('Keybinder', '3.24')
gi.require_version('Keybinder', '3.0')
from gi.repository import Keybinder
Keybinder.init()
Keybinder.set_use_cooked_accelerators(False)
except (ImportError, ValueError) as e:
logger.warning(e)
logger.warning('Unable to load Keybinder module. This means the hide_window shortcut will be unavailable')
class ControllerData:
''' ControllerData contains most of the state of the app at ay given time. It also has some support methods. '''
def setup_controller_data(self) -> None:
self.builder = None
self.core_widget = None
self.load_glade_file()
self.plugins = PluginsController()
self.hidefunc = None
self.show_image = True
self.guake_key = settings.get_guake_key()
def setup_toggle_event(self) -> None:
self.window = settings.get_builder().get_object(f"{app_name.lower()}")
hidebound = None
# Attempt to grab a global hotkey for hiding the window.
# If we fail, we'll never hide the window, iconifying instead.
if self.guake_key and display_manager() == 'X11':
try:
hidebound = Keybinder.bind(self.guake_key, self.on_hide_window)
except (KeyError, NameError) as e:
logger.warning(e)
print( repr(e) )
if not hidebound:
logger.debug('Unable to bind hide_window key, another instance/window has it.')
self.hidefunc = self.window.iconify
else:
self.hidefunc = self.window.hide
def clear_console(self) -> None:
''' Clears the terminal screen. '''
os.system('cls' if os.name == 'nt' else 'clear')
def call_method(self, _method_name: str, data: type) -> type:
'''
Calls a method from scope of class.
Parameters:
a (obj): self
b (str): method name to be called
c (*): Data (if any) to be passed to the method.
Note: It must be structured according to the given methods requirements.
Returns:
Return data is that which the calling method gives.
'''
method_name = str(_method_name)
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: type, method: type) -> type:
''' Checks if a given method exists. '''
return callable(getattr(obj, method, None))
def clear_children(self, widget: type) -> None:
''' Clear children of a gtk widget. '''
for child in widget.get_children():
widget.remove(child)

3
src/core/controllers/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""
Controllers Package
"""

72
src/core/controllers/base_controller.py Normal file
View File

@@ -0,0 +1,72 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from libs.mixins.ipc_signals_mixin import IPCSignalsMixin
from libs.mixins.keyboard_signals_mixin import KeyboardSignalsMixin
from ..containers.base_container import BaseContainer
from .base_controller_data import BaseControllerData
from .bridge_controller import BridgeController
class BaseController(IPCSignalsMixin, KeyboardSignalsMixin, BaseControllerData):
""" docstring for BaseController. """
def __init__(self):
self._setup_controller_data()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_controllers()
self._load_plugins_and_files()
logger.info(f"Made it past {self.__class__} loading...")
settings_manager.set_end_load_time()
settings_manager.log_load_time()
def _setup_styling(self):
...
def _setup_signals(self):
self.window.connect("focus-out-event", self.unset_keys_and_data)
self.window.connect("key-press-event", self.on_global_key_press_controller)
self.window.connect("key-release-event", self.on_global_key_release_controller)
def _subscribe_to_events(self):
event_system.subscribe("shutting-down", lambda: print("Shutting down..."))
event_system.subscribe("handle-file-from-ipc", self.handle_file_from_ipc)
event_system.subscribe("handle-dir-from-ipc", self.handle_dir_from_ipc)
event_system.subscribe("tggl-top-main-menubar", self._tggl_top_main_menubar)
def _load_controllers(self):
BridgeController()
def _load_plugins_and_files(self):
args, unknownargs = settings_manager.get_starting_args()
if args.no_plugins == "false":
self.plugins_controller.pre_launch_plugins()
self.plugins_controller.post_launch_plugins()
for file in settings_manager.get_starting_files():
event_system.emit("post-file-to-ipc", file)
def _tggl_top_main_menubar(self):
logger.debug("_tggl_top_main_menubar > stub...")
def _load_glade_file(self):
self.builder.add_from_file( settings_manager.get_glade_file() )
self.builder.expose_object("main_window", self.window)
settings_manager.set_builder(self.builder)
self.base_container = BaseContainer()
settings_manager.register_signals_to_builder([self, self.base_container])

107
src/core/controllers/base_controller_data.py Normal file
View File

@@ -0,0 +1,107 @@
# Python imports
import os
import subprocess
from shutil import which
# Lib imports
# Application imports
from plugins.plugins_controller import PluginsController
from ..builder_wrapper import BuilderWrapper
class BaseControllerData:
''' BaseControllerData contains most of the state of the app at ay given time. It also has some support methods. '''
def _setup_controller_data(self) -> None:
self.window = settings_manager.get_main_window()
self.builder = BuilderWrapper()
self.plugins_controller = PluginsController()
self.base_container = None
self.was_midified_key = False
self.ctrl_down = False
self.shift_down = False
self.alt_down = False
self._collect_files_dirs()
self._load_glade_file()
def _collect_files_dirs(self):
args, \
unknownargs = settings_manager.get_starting_args()
files = []
for arg in unknownargs + [args.new_tab,]:
if os.path.isdir( arg.replace("file://", "") ):
files.append( f"DIR|{arg.replace('file://', '')}" )
continue
# NOTE: If passing line number with file split against :
if os.path.isfile( arg.replace("file://", "").split(":")[0] ):
files.append( f"FILE|{arg.replace('file://', '')}" )
continue
logger.info(f"Not a File: {arg}")
if len(files) == 0: return
settings_manager.set_is_starting_with_file(True)
settings_manager.set_starting_files(files)
def get_base_container(self):
return self.base_container
def clear_console(self) -> None:
''' Clears the terminal screen. '''
os.system('cls' if os.name == 'nt' else 'clear')
def call_method(self, _method_name: str, data: type) -> type:
'''
Calls a method from scope of class.
Parameters:
a (obj): self
b (str): method name to be called
c (*): Data (if any) to be passed to the method.
Note: It must be structured according to the given methods requirements.
Returns:
Return data is that which the calling method gives.
'''
method_name = str(_method_name)
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: type, method: type) -> type:
''' Checks if a given method exists. '''
return callable(getattr(obj, method, None))
def clear_children(self, widget: type) -> None:
''' Clear children of a gtk widget. '''
for child in widget.get_children():
widget.remove(child)
def get_clipboard_data(self, encoding = "utf-8") -> str:
if not which("xclip"):
logger.info('xclip not found...')
return
command = ['xclip','-selection','clipboard']
proc = subprocess.Popen(['xclip','-selection', 'clipboard', '-o'], stdout = subprocess.PIPE)
retcode = proc.wait()
data = proc.stdout.read()
return data.decode(encoding).strip()
def set_clipboard_data(self, data: type, encoding = "utf-8") -> None:
if not which("xclip"):
logger.info('xclip not found...')
return
command = ['xclip','-selection','clipboard']
proc = subprocess.Popen(command, stdin = subprocess.PIPE)
proc.stdin.write(data.encode(encoding))
proc.stdin.close()
retcode = proc.wait()

41
src/core/controllers/bridge_controller.py Normal file
View File

@@ -0,0 +1,41 @@
# Python imports
import base64
# Lib imports
# Application imports
class BridgeController:
def __init__(self):
self._setup_signals()
self._subscribe_to_events()
def _setup_signals(self):
...
def _subscribe_to_events(self):
event_system.subscribe("handle-bridge-event", self.handle_bridge_event)
def handle_bridge_event(self, event):
match event.topic:
case "save":
event_system.emit(f"handle-file-event-{event.originator}", (event,))
case "close":
event_system.emit(f"handle-file-event-{event.originator}", (event,))
case "load_buffer":
event_system.emit(f"handle-file-event-{event.originator}", (event,))
case "load_file":
event_system.emit(f"handle-file-event-{event.originator}", (event,))
case "alert":
content = base64.b64decode( event.content.encode() ).decode("utf-8")
logger.info(f"\nMessage Topic: {event.topic}\nMessage Content: {content}")
case "error":
content = base64.b64decode( event.content.encode() ).decode("utf-8")
logger.info(content)
case _:
...

193
src/core/core_widget.py
View File

@@ -1,193 +0,0 @@
# Python imports
from datetime import datetime
# Gtk imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('Wnck', '3.0')
from gi.repository import Gtk
from gi.repository import Wnck
from gi.repository import GObject
from gi.repository import GdkPixbuf
from xdg.DesktopEntry import DesktopEntry
# Application imports
from .desktop_parsing.app_finder import find_apps
class CoreWidget(Gtk.Box):
def __init__(self):
super(CoreWidget, self).__init__()
self.builder = settings.get_builder()
self.time_label = self.builder.get_object("time_lbl")
self.orientation = 1 # 0 = horizontal, 1 = vertical
self._setup_styling()
self._setup_signals()
self._load_widgets()
self.show_all()
self.menu_objects = {
"Accessories": [],
"Multimedia": [],
"Graphics": [],
"Game": [],
"Office": [],
"Development": [],
"Internet": [],
"Settings": [],
"System": [],
"Wine": [],
"Other": []
}
apps = find_apps()
self.fill_menu_objects(apps)
search_programs_entry = self.builder.get_object("search_programs_entry")
search_programs_entry.hide()
def fill_menu_objects(self, apps=[]):
for app in apps:
fPath = app.get_filename()
xdgObj = DesktopEntry( fPath )
title = xdgObj.getName()
groups = xdgObj.getCategories()
comment = xdgObj.getComment()
icon = xdgObj.getIcon()
mainExec = xdgObj.getExec()
tryExec = xdgObj.getTryExec()
group = ""
if "Accessories" in groups or "Utility" in groups:
group = "Accessories"
elif "Multimedia" in groups or "Video" in groups or "Audio" in groups:
group = "Multimedia"
elif "Development" in groups:
group = "Development"
elif "Game" in groups:
group = "Game"
elif "Internet" in groups or "Network" in groups:
group = "Internet"
elif "Graphics" in groups:
group = "Graphics"
elif "Office" in groups:
group = "Office"
elif "System" in groups:
group = "System"
elif "Settings" in groups:
group = "Settings"
elif "Wine" in groups:
group = "Wine"
else:
group = "Other"
self.menu_objects[group].append(
{
"title": title,
"groups": groups,
"comment": comment,
"exec": mainExec,
"tryExec": tryExec,
"fileName": fPath.split("/")[-1],
"filePath": fPath,
"icon": icon
}
)
def get_group(self, group):
return self.menu_objects[group]
def _setup_styling(self):
self.set_orientation(1)
self.set_vexpand(True)
self.set_hexpand(True)
def _setup_signals(self):
...
def _load_widgets(self):
widget_grid_container = self.builder.get_object("widget_grid_container")
time_lbl_eve_box = self.builder.get_object("time_lbl_eve_box")
time_lbl_eve_box.connect("button_release_event", self._toggle_cal_popover)
widget_grid_container.set_vexpand(True)
widget_grid_container.set_hexpand(True)
self.add( widget_grid_container )
self.set_pager_widget()
self.set_task_list_widget()
# Must be after pager and task list inits
self.display_clock()
self._start_clock()
def set_pager_widget(self):
pager = Wnck.Pager.new()
if self.orientation == 0:
self.builder.get_object('taskbar_workspaces_hor').add(pager)
else:
self.builder.get_object('taskbar_workspaces_ver').add(pager)
pager.set_hexpand(True)
pager.show()
def set_task_list_widget(self):
tasklist = Wnck.Tasklist.new()
tasklist.set_scroll_enabled(False)
tasklist.set_button_relief(2) # 0 = normal relief, 2 = no relief
tasklist.set_grouping(1) # 0 = mever group, 1 auto group, 2 = always group
if self.orientation == 0:
self.builder.get_object('taskbar_bttns_hor').add(tasklist)
else:
self.builder.get_object('taskbar_bttns_ver').add(tasklist)
tasklist.set_vexpand(True)
tasklist.set_include_all_workspaces(False)
tasklist.set_orientation(self.orientation)
tasklist.show()
# Displays Timer
def display_clock(self):
now = datetime.now()
hms = now.strftime("%I:%M %p")
mdy = now.strftime("%m/%d/%Y")
timeStr = hms + "\n" + mdy
self.time_label.set_label(timeStr)
return True
def _start_clock(self):
GObject.timeout_add(59000, self.display_clock)
def _close_popup(self, widget, data = None):
widget.hide()
def _toggle_cal_popover(self, widget, eve):
calendar_popup = self.builder.get_object('calendar_popup')
if (calendar_popup.get_visible() == False):
calendarWid = self.builder.get_object('calendarWid')
now = datetime.now()
timeStr = now.strftime("%m/%d/%Y")
parts = timeStr.split("/")
month = int(parts[0]) - 1
day = int(parts[1])
year = int(parts[2])
calendarWid.select_day(day)
calendarWid.select_month(month, year)
calendar_popup.popup()
else:
calendar_popup.popdown()

3
src/core/mixins/__init__.py
View File

@@ -1,3 +0,0 @@
"""
Generic Mixins Module
"""

3
src/core/widgets/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""
Widgets Package
"""

129
src/core/widgets/category_list_widget.py Normal file
View File

@@ -0,0 +1,129 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
from xdg.DesktopEntry import DesktopEntry
# Application imports
from libs.desktop_parsing.app_finder import find_apps
class CategoryListWidget(Gtk.ButtonBox):
def __init__(self):
super(CategoryListWidget, self).__init__()
self.ctx = self.get_style_context()
self.active_category: str = 'Accessories'
self.category_dict: {} = {
"Accessories": [],
"Multimedia": [],
"Graphics": [],
"Game": [],
"Office": [],
"Development": [],
"Internet": [],
"Settings": [],
"System": [],
"Wine": [],
"Other": []
}
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.fill_menu_objects()
self.show_all()
def _setup_styling(self):
self.set_orientation(Gtk.Orientation.VERTICAL)
self.ctx.add_class("category-list-widget")
def _setup_signals(self):
event_system.subscribe("refresh-active-category", self.refresh_active_category)
def _subscribe_to_events(self):
...
def _load_widgets(self):
for category in self.category_dict.keys():
button = Gtk.Button(label = category)
button.connect("clicked", self.set_active_category)
button.show()
self.add(button)
def set_active_category(self, button):
self.active_category = button.get_label()
event_system.emit(
"load-active-category",
(self.category_dict[ self.active_category ],)
)
def fill_menu_objects(self, apps: [] = []):
if not apps:
apps = find_apps()
for app in apps:
fPath = app.get_filename()
xdgObj = DesktopEntry( fPath )
title = xdgObj.getName()
groups = xdgObj.getCategories()
comment = xdgObj.getComment()
icon = xdgObj.getIcon()
mainExec = xdgObj.getExec()
tryExec = xdgObj.getTryExec()
group = ""
if "Accessories" in groups or "Utility" in groups:
group = "Accessories"
elif "Multimedia" in groups or "Video" in groups or "Audio" in groups:
group = "Multimedia"
elif "Development" in groups:
group = "Development"
elif "Game" in groups:
group = "Game"
elif "Internet" in groups or "Network" in groups:
group = "Internet"
elif "Graphics" in groups:
group = "Graphics"
elif "Office" in groups:
group = "Office"
elif "System" in groups:
group = "System"
elif "Settings" in groups:
group = "Settings"
elif "Wine" in groups:
group = "Wine"
else:
group = "Other"
self.category_dict[group].append(
{
"title": title,
"groups": groups,
"comment": comment,
"exec": mainExec,
"tryExec": tryExec,
"fileName": fPath.split("/")[-1],
"filePath": fPath,
"icon": icon
}
)
def refresh_active_category(self):
event_system.emit(
"load-active-category",
(self.category_dict[ self.active_category ],)
)

125
src/core/widgets/category_widget.py Normal file
View File

@@ -0,0 +1,125 @@
# Python imports
import os
import subprocess
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('GdkPixbuf', '2.0')
from gi.repository import Gtk
from gi.repository import Gio
from gi.repository import GdkPixbuf
# Application imports
class CategoryWidget(Gtk.Grid):
def __init__(self):
super(CategoryWidget, self).__init__()
self.ctx = self.get_style_context()
self.column_count = 4
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show()
def _setup_styling(self):
self.set_hexpand(True)
self.set_vexpand(True)
self.set_row_spacing(10)
self.set_column_spacing(10)
self.set_row_homogeneous(True)
self.set_column_homogeneous(True)
self.set_orientation(Gtk.Orientation.HORIZONTAL)
self.ctx.add_class("category-widget")
def _setup_signals(self):
event_system.subscribe("load-active-category", self.load_active_category)
def _subscribe_to_events(self):
...
def _load_widgets(self):
event_system.emit("refresh-active-category")
def load_active_category(self, app_list: [] = []):
self.clear_children()
row = 0
col = 0
icon_theme = Gtk.IconTheme.get_default()
for app in app_list:
button = self.generate_app_button(icon_theme, app)
self.attach(button, col, row, 1, 1)
col += 1
if col == self.column_count:
col = 0
row += 1
def generate_app_button(self, icon_theme, app: {} = {}):
if not app: return
title = app["title"]
exec_str = app[
"exec" if not app["exec"] in ("", None) else "tryExec"
]
button = Gtk.Button(label = title)
button.sig_id = button.connect("clicked", self._do_exec, exec_str)
icon_pth = app["icon"]
icon = self.get_icon_from_path(icon_pth) \
if os.path.exists(icon_pth) \
else \
self.get_icon_from_gio(icon_theme, icon_pth)
button.set_image(icon)
button.set_always_show_image(True)
button.show_all()
return button
def get_icon_from_path(self, path: str):
pixbuf = GdkPixbuf.PixbufAnimation.new_from_file(path) \
.get_static_image() \
.scale_simple(32, 32, \
GdkPixbuf.InterpType.BILINEAR)
return Gtk.Image.new_from_pixbuf(pixbuf)
def get_icon_from_gio(self, icon_theme, icon_name: str):
gio_icon = Gio.Icon.new_for_string(icon_name)
pixbuf = None
# Note: https://docs.gtk.org/gtk3/enum.IconSize.html
for i in [6, 5, 3, 4, 2, 1]:
icon_info = Gtk.IconTheme.lookup_by_gicon(icon_theme, gio_icon, i, Gtk.IconLookupFlags.FORCE_REGULAR)
if not icon_info: continue
pixbuf = icon_info.load_icon().scale_simple(32, 32, 2) # 2 = BILINEAR and is best by default
break
return Gtk.Image.new_from_pixbuf( pixbuf )
def _do_exec(self, widget, _command):
command = _command.split("%")[0]
subprocess.Popen(command.split(), start_new_session=True, stdout=None, stderr=None)
def clear_children(self, app_list: [] = []):
children = self.get_children()
for child in children:
child.disconnect(child.sig_id)
self.remove(child)
child.run_dispose()

110
src/core/widgets/clock_widget.py Normal file
View File

@@ -0,0 +1,110 @@
# Python imports
from datetime import datetime
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
from gi.repository import GObject
# Application imports
class CalendarWidget(Gtk.Popover):
def __init__(self):
super(CalendarWidget, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
def _setup_styling(self):
...
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
self.body = Gtk.Calendar()
self.body.show()
self.add(self.body)
class ClockWidget(Gtk.EventBox):
def __init__(self):
super(ClockWidget, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show_all()
def _setup_styling(self):
self.set_size_request(180, -1)
ctx = self.get_style_context()
ctx.add_class("clock-widget")
def _setup_signals(self):
self.connect("button_release_event", self._toggle_cal_popover)
def _subscribe_to_events(self):
...
def _load_widgets(self):
self.calendar = CalendarWidget()
self.label = Gtk.Label()
self.calendar.set_relative_to(self)
self.label.set_justify(Gtk.Justification.CENTER)
self.label.set_margin_top(15)
self.label.set_margin_bottom(15)
self.label.set_margin_left(15)
self.label.set_margin_right(15)
self._update_face()
self.add(self.label)
GObject.timeout_add(59000, self._update_face)
def _update_face(self):
dt_now = datetime.now()
hours_mins_sec = dt_now.strftime("%I:%M %p")
month_day_year = dt_now.strftime("%m/%d/%Y")
time_str = hours_mins_sec + "\n" + month_day_year
self.label.set_label(time_str)
def _toggle_cal_popover(self, widget, eve):
if (self.calendar.get_visible() == True):
self.calendar.popdown()
return
now = datetime.now()
timeStr = now.strftime("%m/%d/%Y")
parts = timeStr.split("/")
month = int(parts[0]) - 1
day = int(parts[1])
year = int(parts[2])
self.calendar.body.select_day(day)
self.calendar.body.select_month(month, year)
self.calendar.popup()

3
src/core/widgets/controls/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""
Widgets.Controls Package
"""

83
src/core/widgets/controls/open_files_button.py Normal file
View File

@@ -0,0 +1,83 @@
# Python imports
from contextlib import suppress
import os
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
from gi.repository import Gio
# Application imports
class OpenFilesButton(Gtk.Button):
"""docstring for OpenFilesButton."""
def __init__(self):
super(OpenFilesButton, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
def _setup_styling(self):
self.set_label("Open File(s)...")
self.set_image( Gtk.Image.new_from_icon_name("gtk-open", 4) )
self.set_always_show_image(True)
self.set_image_position(1) # Left - 0, Right = 1
self.set_hexpand(False)
def _setup_signals(self):
self.connect("button-release-event", self._open_files)
def _subscribe_to_events(self):
event_system.subscribe("open_files", self._open_files)
def _load_widgets(self):
...
def _open_files(self, widget = None, eve = None, gfile = None):
start_dir = None
_gfiles = []
if gfile and gfile.query_exists():
start_dir = gfile.get_parent()
chooser = Gtk.FileChooserDialog("Open File(s)...", None,
Gtk.FileChooserAction.OPEN,
(
Gtk.STOCK_CANCEL,
Gtk.ResponseType.CANCEL,
Gtk.STOCK_OPEN,
Gtk.ResponseType.OK
)
)
chooser.set_select_multiple(True)
with suppress(Exception):
folder = widget.get_current_file().get_parent() if not start_dir else start_dir
chooser.set_current_folder( folder.get_path() )
response = chooser.run()
if not response == Gtk.ResponseType.OK:
chooser.destroy()
return _gfiles
filenames = chooser.get_filenames()
if not filenames:
chooser.destroy()
return _gfiles
for file in filenames:
path = file if os.path.isabs(file) else os.path.abspath(file)
_gfiles.append( Gio.File.new_for_path(path) )
chooser.destroy()
logger.debug(_gfiles)
return _gfiles

72
src/core/widgets/controls/save_as_button.py Normal file
View File

@@ -0,0 +1,72 @@
# Python imports
import os
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
from gi.repository import Gio
# Application imports
class SaveAsButton(Gtk.Button):
def __init__(self):
super(SaveAsButton, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
def _setup_styling(self):
self.set_label("Save As")
self.set_image( Gtk.Image.new_from_icon_name("gtk-save-as", 4) )
self.set_always_show_image(True)
self.set_image_position(1) # Left - 0, Right = 1
self.set_hexpand(False)
def _setup_signals(self):
self.connect("released", self._save_as)
def _subscribe_to_events(self):
event_system.subscribe("save-as", self._save_as)
def _load_widgets(self):
...
def _save_as(self, widget = None, eve = None, gfile = None):
start_dir = None
_gfile = None
chooser = Gtk.FileChooserDialog("Save File As...", None,
Gtk.FileChooserAction.SAVE,
(
Gtk.STOCK_CANCEL,
Gtk.ResponseType.CANCEL,
Gtk.STOCK_SAVE_AS,
Gtk.ResponseType.OK
)
)
# chooser.set_select_multiple(False)
response = chooser.run()
if not response == Gtk.ResponseType.OK:
chooser.destroy()
return _gfile
file = chooser.get_filename()
if not file:
chooser.destroy()
return _gfile
path = file if os.path.isabs(file) else os.path.abspath(file)
_gfile = Gio.File.new_for_path(path)
chooser.destroy()
logger.debug(f"File To Save As: {_gfile}")
return _gfile

48
src/core/widgets/controls/transparency_scale.py Normal file
View File

@@ -0,0 +1,48 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
class TransparencyScale(Gtk.Scale):
def __init__(self):
super(TransparencyScale, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show_all()
def _setup_styling(self):
self.set_digits(0)
self.set_value_pos(Gtk.PositionType.RIGHT)
self.add_mark(50.0, Gtk.PositionType.TOP, "50%")
self.set_hexpand(True)
def _setup_signals(self):
self.connect("value-changed", self._update_transparency)
def _subscribe_to_events(self):
...
def _load_widgets(self):
adjust = self.get_adjustment()
adjust.set_lower(0)
adjust.set_upper(100)
adjust.set_value(settings.theming.transparency)
adjust.set_step_increment(1.0)
def _update_transparency(self, range):
event_system.emit("remove-transparency")
tp = int(range.get_value())
settings.theming.transparency = tp
event_system.emit("update-transparency")

35
src/core/widgets/pager_widget.py Normal file
View File

@@ -0,0 +1,35 @@
# Python imports
# Lib imports
import gi
gi.require_version('Wnck', '3.0')
from gi.repository import Wnck
# Application imports
class PagerWidget:
def __init__(self):
super(PagerWidget, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
def _setup_styling(self):
...
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
...
def get_widget(self):
return Wnck.Pager.new()

62
src/core/widgets/tab_widget.py Normal file
View File

@@ -0,0 +1,62 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from ..widgets.category_widget import CategoryWidget
from ..widgets.vte_widget import VteWidget
class TabWidget(Gtk.Notebook):
def __init__(self):
super(TabWidget, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show()
def _setup_styling(self):
self.set_hexpand(True)
self.set_vexpand(True)
ctx = self.get_style_context()
ctx.add_class("tab-widget")
def _setup_signals(self):
...
def _subscribe_to_events(self):
event_system.subscribe("focus-apps", self.focus_apps)
event_system.subscribe("focus-terminal", self.focus_terminal)
def _load_widgets(self):
scroll_view = Gtk.ScrolledWindow()
viewport = Gtk.Viewport()
viewport.add(CategoryWidget())
scroll_view.add(viewport)
scroll_view.show_all()
self.insert_page(scroll_view, Gtk.Label(label = "Apps"), 0)
self.insert_page(VteWidget(), Gtk.Label(label = "Terminal"), 1)
self.set_current_page(0)
def focus_apps(self):
widget = self.get_nth_page(0).get_children()[0].get_children()[0]
self.set_current_page(0)
widget.grab_focus()
def focus_terminal(self):
widget = self.get_nth_page(1)
self.set_current_page(1)
widget.grab_focus()

54
src/core/widgets/task_list_widget.py Normal file
View File

@@ -0,0 +1,54 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('Wnck', '3.0')
from gi.repository import Gtk
from gi.repository import Wnck
# Application imports
class TaskListWidget(Gtk.ScrolledWindow):
def __init__(self):
super(TaskListWidget, self).__init__()
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self.show_all()
def _setup_styling(self):
self.set_hexpand(False)
self.set_size_request(180, -1)
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
viewport = Gtk.Viewport()
task_list = Wnck.Tasklist.new()
vbox = Gtk.Box()
vbox.set_orientation(Gtk.Orientation.VERTICAL)
task_list.set_scroll_enabled(False)
task_list.set_button_relief(2) # 0 = normal relief, 2 = no relief
task_list.set_grouping(1) # 0 = mever group, 1 auto group, 2 = always group
task_list.set_vexpand(True)
task_list.set_include_all_workspaces(False)
task_list.set_orientation(1) # 0 = horizontal, 1 = vertical
vbox.add(task_list)
viewport.add(vbox)
self.add(viewport)

125
src/core/widgets/vte_widget.py Normal file
View File

@@ -0,0 +1,125 @@
# Python imports
import os
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('Gdk', '3.0')
gi.require_version('Vte', '2.91')
from gi.repository import Gtk
from gi.repository import Gdk
from gi.repository import GLib
from gi.repository import Vte
# Application imports
from libs.dto.event import Event
class VteWidgetException(Exception):
...
class VteWidget(Vte.Terminal):
"""
https://stackoverflow.com/questions/60454326/how-to-implement-a-linux-terminal-in-a-pygtk-app-like-vscode-and-pycharm-has
"""
def __init__(self):
super(VteWidget, self).__init__()
self.cd_cmd_prefix = ("cd".encode(), "cd ".encode())
self.dont_process = False
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
self._do_session_spawn()
self.show()
def _setup_styling(self):
ctx = self.get_style_context()
ctx.add_class("vte-widget")
self.set_clear_background(False)
self.set_enable_sixel(True)
self.set_cursor_shape( Vte.CursorShape.IBEAM )
def _setup_signals(self):
self.connect("commit", self._commit)
def _subscribe_to_events(self):
event_system.subscribe("update_term_path", self.update_term_path)
def _load_widgets(self):
...
def _do_session_spawn(self):
self.spawn_sync(
Vte.PtyFlags.DEFAULT,
settings_manager.get_home_path(),
["/bin/bash"],
[],
GLib.SpawnFlags.DEFAULT,
None, None,
)
# Note: '-->:' is used as a delimiter to split on to get command actual.
# !!! DO NOT REMOVE UNLESS CODE UPDATED ACCORDINGLY !!!
startup_cmds = [
"env -i /bin/bash --noprofile --norc\n",
"export TERM='xterm-256color'\n",
"export LC_ALL=C\n",
"export XDG_RUNTIME_DIR='/run/user/1000'\n",
"export DISPLAY=:0\n",
f"export XAUTHORITY='{settings_manager.get_home_path()}/.Xauthority'\n",
f"\nexport HOME='{settings_manager.get_home_path()}'\n",
"export PS1='\\h@\\u \\W -->: '\n",
"clear\n"
]
for i in startup_cmds:
self.run_command(i)
def _commit(self, terminal, text, size):
if self.dont_process:
self.dont_process = False
return
if not text.encode() == "\r".encode(): return
text, attributes = self.get_text()
lines = text.strip().splitlines()
command_ran = None
try:
command_ran = lines[-1].split("-->:")[1].strip()
except VteWidgetException as e:
logger.debug(e)
return
if not command_ran[0:3].encode() in self.cd_cmd_prefix:
return
target_path = command_ran.split( command_ran[0:3] )[1]
if target_path in (".", "./"): return
if not target_path:
target_path = settings_manager.get_home_path()
event = Event("pty_path_updated", "", target_path)
event_system.emit("handle_bridge_event", (event,))
def update_term_path(self, fpath: str):
self.dont_process = True
cmds = [f"cd '{fpath}'\n", "clear\n"]
for cmd in cmds:
self.run_command(cmd)
def run_command(self, cmd: str):
self.feed_child_binary(bytes(cmd, 'utf8'))

169
src/core/window.py
View File

@@ -1,88 +1,201 @@
# Python imports
import time
import signal
# Lib imports
import gi
import cairo
gi.require_version('Gtk', '3.0')
gi.require_version('Gdk', '3.0')
from gi.repository import Gtk
from gi.repository import Gdk
from gi.repository import GLib
# Application imports
from core.controller import Controller
from libs.status_icon import StatusIcon
from core.controllers.base_controller import BaseController
class ControllerStartExceptiom(Exception):
...
class Window(Gtk.ApplicationWindow):
"""docstring for Window."""
""" docstring for Window. """
def __init__(self, args, unknownargs):
def __init__(self):
super(Window, self).__init__()
settings_manager.set_main_window(self)
self._status_icon = None
self._controller = None
self.guake_key = settings_manager.get_guake_key()
self.hidefunc = None
self._set_window_data()
self._setup_styling()
self._setup_signals()
self._load_widgets(args, unknownargs)
self._subscribe_to_events()
self._load_widgets()
self._set_window_data()
self._set_size_constraints()
self._setup_window_toggle_event()
self.show()
def _setup_styling(self):
self.set_title(f"{APP_NAME}")
self.set_icon_from_file( settings_manager.get_window_icon() )
self.set_default_size(128, 720)
self.set_decorated(False)
self.set_skip_pager_hint(True)
self.set_skip_taskbar_hint(True)
self.set_gravity(5) # 5 = CENTER
self.set_position(1) # 1 = CENTER, 4 = CENTER_ALWAYS
self.set_keep_above(True)
self.stick()
self.set_default_size(1200, 600)
self.set_title(f"{app_name}")
self.set_icon_from_file( settings.get_window_icon() )
self.set_gravity(5) # 5 = CENTER
self.set_position(1) # 1 = CENTER, 4 = CENTER_ALWAYS
ctx = self.get_style_context()
ctx.add_class("main-window")
ctx.add_class(f"mw_transparency_{settings.theming.transparency}")
def _setup_signals(self):
self.connect("delete-event", self._tear_down)
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGINT, self._tear_down)
self.connect("focus-in-event", self._on_focus_in_event)
self.connect("focus-out-event", self._on_focus_out_event)
def _load_widgets(self, args, unknownargs):
controller = Controller(args, unknownargs)
self.connect("delete-event", self.stop)
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGINT, self.stop)
self.set_name(f"{app_name.lower()}")
settings.get_builder().expose_object(f"{app_name.lower()}", self)
def _subscribe_to_events(self):
event_system.subscribe("tear-down", self.stop)
event_system.subscribe("load-interactive-debug", self._load_interactive_debug)
self.add( controller.get_core_widget() )
def _load_widgets(self):
if settings_manager.is_debug():
self.set_interactive_debugging(True)
self._controller = BaseController()
self._status_icon = StatusIcon()
if not self._controller:
raise ControllerStartException("BaseController exited and doesn't exist...")
self.add( self._controller.get_base_container() )
def _display_manager(self):
""" Try to detect which display manager we are running under... """
import os
if os.environ.get('WAYLAND_DISPLAY'):
return 'WAYLAND'
return 'X11'
def _set_size_constraints(self):
# _window_x = settings.config.main_window_x
# _window_y = settings.config.main_window_y
# self.move(_window_x, _window_y - 28)
_min_width = settings.config.main_window_min_width
_min_height = settings.config.main_window_min_height
_width = settings.config.main_window_width
_height = settings.config.main_window_height
self.set_size_request(_min_width, _min_height)
self.set_default_size(_width, _height)
def _set_window_data(self) -> None:
screen = self.get_screen()
visual = screen.get_rgba_visual()
if visual != None and screen.is_composited():
if visual and screen.is_composited() and settings.config.make_transparent == 0:
self.set_visual(visual)
self.set_app_paintable(True)
self.connect("draw", self._area_draw)
# self.connect("draw", self._area_draw)
# bind css file
cssProvider = Gtk.CssProvider()
cssProvider.load_from_path( settings.get_css_file() )
cssProvider.load_from_path( settings_manager.get_css_file() )
screen = Gdk.Screen.get_default()
styleContext = Gtk.StyleContext()
styleContext.add_provider_for_screen(screen, cssProvider, Gtk.STYLE_PROVIDER_PRIORITY_USER)
def _area_draw(self, widget: Gtk.ApplicationWindow, cr: cairo.Context) -> None:
cr.set_source_rgba(0, 0, 0, 0.54)
cr.set_source_rgba( *settings_manager.get_paint_bg_color() )
cr.set_operator(cairo.OPERATOR_SOURCE)
cr.paint()
cr.set_operator(cairo.OPERATOR_OVER)
def _tear_down(self, widget=None, eve=None):
settings.clear_pid()
time.sleep(event_sleep_time)
Gtk.main_quit()
def _on_focus_in_event(self, widget, event):
event_system.emit("pause-dnd-signals")
def _on_focus_out_event(self, widget, event):
event_system.emit("listen-dnd-signals")
def _load_interactive_debug(self):
self.set_interactive_debugging(True)
def _setup_window_toggle_event(self) -> None:
hidebound = None
if not self.guake_key or not self._display_manager() == 'X11':
return
try:
import gi
gi.require_version('Keybinder', '3.0')
from gi.repository import Keybinder
Keybinder.init()
Keybinder.set_use_cooked_accelerators(False)
except (ImportError, ValueError) as e:
logger.warning(e)
logger.warning('Unable to load Keybinder module. This means the hide_window shortcut will be unavailable')
return
# Attempt to grab a global hotkey for hiding the window.
# If we fail, we'll never hide the window, iconifying instead.
try:
hidebound = Keybinder.bind(self.guake_key, self._on_toggle_window, self)
except (KeyError, NameError) as e:
logger.warning(e)
if not hidebound:
logger.debug('Unable to bind hide_window key, another instance/window has it.')
self.hidefunc = self.iconify
else:
self.hidefunc = self.hide
def _on_toggle_window(self, data, window):
"""Handle a request to hide/show the window"""
if not window.get_property('visible'):
window.show()
# Note: Needed to properly grab widget focus when set_skip_taskbar_hint set to True
window.present()
# NOTE: Need here to enforce sticky after hide and reshow.
window.stick()
else:
self.hidefunc()
def start(self):
Gtk.main()
def stop(self, widget = None, eve = None):
event_system.emit("shutting-down")
size = self.get_size()
pos = self.get_position()
settings_manager.set_main_window_width(size.width)
settings_manager.set_main_window_height(size.height)
settings_manager.set_main_window_x(pos.root_x)
settings_manager.set_main_window_y(pos.root_y)
settings_manager.save_settings()
settings_manager.clear_pid()
Gtk.main_quit()

3
src/libs/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""
Libs Package
"""

6
src/libs/db/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
"""
DB Package
"""
from .models import User
from .db import DB

42
src/libs/db/db.py Normal file
View File

@@ -0,0 +1,42 @@
# Python imports
from typing import Optional
from os import path
# Lib imports
from sqlmodel import Session, create_engine
# Application imports
from .models import SQLModel, User
class DB:
def __init__(self):
super(DB, self).__init__()
self.engine = None
self.create_engine()
def create_engine(self):
db_path = f"sqlite:///{settings_manager.get_home_config_path()}/database.db"
self.engine = create_engine(db_path)
SQLModel.metadata.create_all(self.engine)
def _add_entry(self, entry):
with Session(self.engine) as session:
session.add(entry)
session.commit()
def add_user_entry(self, name = None, password = None, email = None):
if not name or not password or not email: return
user = User()
user.name = name
user.password = password
user.email = email
self._add_entry(user)

15
src/libs/db/models.py Normal file
View File

@@ -0,0 +1,15 @@
# Python imports
from typing import Optional
# Lib imports
from sqlmodel import SQLModel, Field
# Application imports
class User(SQLModel, table = True):
id: Optional[int] = Field(default = None, primary_key = True)
name: str
password: str
email: Optional[str] = None

60
src/libs/debugging.py Normal file
View File

@@ -0,0 +1,60 @@
# Python imports
# Lib imports
# Application imports
# Break into a Python console upon SIGUSR1 (Linux) or SIGBREAK (Windows:
# CTRL+Pause/Break). To be included in all production code, just in case.
def debug_signal_handler(signal, frame):
del signal
del frame
try:
import rpdb2
logger.debug("\n\nStarting embedded RPDB2 debugger. Password is 'foobar'\n\n")
rpdb2.start_embedded_debugger("foobar", True, True)
rpdb2.setbreak(depth=1)
return
except Exception:
...
try:
from rfoo.utils import rconsole
logger.debug("\n\nStarting embedded rconsole debugger...\n\n")
rconsole.spawn_server()
return
except Exception as ex:
...
try:
from pudb import set_trace
logger.debug("\n\nStarting PuDB debugger...\n\n")
set_trace(paused = True)
return
except Exception as ex:
...
try:
import ipdb
logger.debug("\n\nStarting IPDB debugger...\n\n")
ipdb.set_trace()
return
except Exception as ex:
...
try:
import pdb
logger.debug("\n\nStarting embedded PDB debugger...\n\n")
pdb.Pdb(skip=['gi.*']).set_trace()
return
except Exception as ex:
...
try:
import code
code.interact()
except Exception as ex:
logger.debug(f"{ex}, returning to normal program flow...")

0
src/core/desktop_parsing/DesktopParser.py → src/libs/desktop_parsing/DesktopParser.py
View File

0
src/core/desktop_parsing/__init__.py → src/libs/desktop_parsing/__init__.py
View File

0
src/core/desktop_parsing/app_finder.py → src/libs/desktop_parsing/app_finder.py
View File

0
src/core/desktop_parsing/db/KeyValueDb.py → src/libs/desktop_parsing/db/KeyValueDb.py
View File

0
src/core/desktop_parsing/db/KeyValueJsonDb.py → src/libs/desktop_parsing/db/KeyValueJsonDb.py
View File

0
src/core/desktop_parsing/db/__init__.py → src/libs/desktop_parsing/db/__init__.py
View File

0
src/core/desktop_parsing/file_finder.py → src/libs/desktop_parsing/file_finder.py
View File

5
src/libs/dto/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""
Dasta Class Package
"""
from .event import Event

14
src/libs/dto/event.py Normal file
View File

@@ -0,0 +1,14 @@
# Python imports
from dataclasses import dataclass, field
# Lib imports
# Application imports
@dataclass
class Event:
topic: str
content: str
raw_content: str

4
src/utils/endpoint_registry.py → src/libs/endpoint_registry.py
View File

@@ -3,11 +3,11 @@
# Lib imports
# Application imports
from .singleton import Singleton
class EndpointRegistry():
class EndpointRegistry(Singleton):
def __init__(self):
self._endpoints = {}

73
src/libs/event_system.py Normal file
View File

@@ -0,0 +1,73 @@
# Python imports
from collections import defaultdict
# Lib imports
# Application imports
from .singleton import Singleton
class EventSystem(Singleton):
""" Create event system. """
def __init__(self):
self.subscribers = defaultdict(list)
self._is_paused = False
self._subscribe_to_events()
def _subscribe_to_events(self):
self.subscribe("pause_event_processing", self._pause_processing_events)
self.subscribe("resume_event_processing", self._resume_processing_events)
def _pause_processing_events(self):
self._is_paused = True
def _resume_processing_events(self):
self._is_paused = False
def subscribe(self, event_type, fn):
self.subscribers[event_type].append(fn)
def unsubscribe(self, event_type, fn):
self.subscribers[event_type].remove(fn)
def unsubscribe_all(self, event_type):
self.subscribers.pop(event_type, None)
def emit(self, event_type, data = None):
if self._is_paused and event_type != "resume_event_processing":
return
if event_type in self.subscribers:
for fn in self.subscribers[event_type]:
if data:
if hasattr(data, '__iter__') and not type(data) is str:
fn(*data)
else:
fn(data)
else:
fn()
def emit_and_await(self, event_type, data = None):
if self._is_paused and event_type != "resume_event_processing":
return
""" NOTE: Should be used when signal has only one listener and vis-a-vis """
if event_type in self.subscribers:
response = None
for fn in self.subscribers[event_type]:
if data:
if hasattr(data, '__iter__') and not type(data) is str:
response = fn(*data)
else:
response = fn(data)
else:
response = fn()
if not response in (None, ''):
break
return response

73
src/utils/ipc_server.py → src/libs/ipc_server.py
View File

@@ -1,26 +1,29 @@
# Python imports
import os, threading, time
from multiprocessing.connection import Listener, Client
import os
import threading
import time
from multiprocessing.connection import Client
from multiprocessing.connection import Listener
# Lib imports
# Application imports
from .singleton import Singleton
class IPCServer:
""" Create a listener so that other {app_name} instances send requests back to existing instance. """
class IPCServer(Singleton):
""" Create a listener so that other {APP_NAME} instances send requests back to existing instance. """
def __init__(self, ipc_address: str = '127.0.0.1', conn_type: str = "socket"):
self.is_ipc_alive = False
self._ipc_port = 4848
self._ipc_port = 0 # Use 0 to let Listener chose port
self._ipc_address = ipc_address
self._conn_type = conn_type
self._ipc_authkey = b'' + bytes(f'{app_name}-ipc', 'utf-8')
self._ipc_authkey = b'' + bytes(f'{APP_NAME}-ipc', 'utf-8')
self._ipc_timeout = 15.0
if conn_type == "socket":
self._ipc_address = f'/tmp/{app_name}-ipc.sock'
self._ipc_address = f'/tmp/{APP_NAME}-ipc.sock'
elif conn_type == "full_network":
self._ipc_address = '0.0.0.0'
elif conn_type == "full_network_unsecured":
@@ -32,12 +35,12 @@ class IPCServer:
self._subscribe_to_events()
def _subscribe_to_events(self):
event_system.subscribe("post_file_to_ipc", self.send_ipc_message)
event_system.subscribe("post-file-to-ipc", self.send_ipc_message)
def create_ipc_listener(self) -> None:
if self._conn_type == "socket":
if os.path.exists(self._ipc_address) and settings.is_dirty_start():
if os.path.exists(self._ipc_address) and settings_manager.is_dirty_start():
os.unlink(self._ipc_address)
listener = Listener(address=self._ipc_address, family="AF_UNIX", authkey=self._ipc_authkey)
@@ -52,23 +55,34 @@ class IPCServer:
@daemon_threaded
def _run_ipc_loop(self, listener) -> None:
while True:
conn = listener.accept()
start_time = time.perf_counter()
self._handle_ipc_message(conn, start_time)
# NOTE: Not thread safe if using with Gtk. Need to import GLib and use idle_add
while self.is_ipc_alive:
try:
conn = listener.accept()
start_time = time.perf_counter()
self._handle_ipc_message(conn, start_time)
except Exception as e:
logger.debug( repr(e) )
listener.close()
def _handle_ipc_message(self, conn, start_time) -> None:
while True:
while self.is_ipc_alive:
msg = conn.recv()
if settings.is_debug():
print(msg)
logger.debug(msg)
if "FILE|" in msg:
file = msg.split("FILE|")[1].strip()
if file:
event_system.emit("handle_file_from_ipc", file)
event_system.emit("handle-file-from-ipc", file)
conn.close()
break
if "DIR|" in msg:
file = msg.split("DIR|")[1].strip()
if file:
event_system.emit("handle-dir-from-ipc", file)
conn.close()
break
@@ -97,6 +111,25 @@ class IPCServer:
conn.send(message)
conn.close()
except ConnectionRefusedError as e:
print("Connection refused...")
logger.error("Connection refused...")
except Exception as e:
print(repr(e))
logger.error( repr(e) )
def send_test_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)
elif "unsecured" not in self._conn_type:
conn = Client((self._ipc_address, self._ipc_port), authkey=self._ipc_authkey)
else:
conn = Client((self._ipc_address, self._ipc_port))
conn.send(message)
conn.close()
except ConnectionRefusedError as e:
if self._conn_type == "socket":
logger.error("IPC Socket no longer valid.... Removing.")
os.unlink(self._ipc_address)
except Exception as e:
logger.error( repr(e) )

45
src/utils/keybindings.py → src/libs/keybindings.py
View File

@@ -1,26 +1,26 @@
# Python imports
import re
# Gtk imports
# Lib imports
import gi
gi.require_version('Gdk', '3.0')
from gi.repository import Gdk
# Application imports
from .singleton import Singleton
def err(log = ""):
"""Print an error message"""
def logger(log = ""):
print(log)
class KeymapError(Exception):
"""Custom exception for errors in keybinding configurations"""
""" Custom exception for errors in keybinding configurations """
MODIFIER = re.compile('<([^<]+)>')
class Keybindings:
"""Class to handle loading and lookup of Terminator keybindings"""
class Keybindings(Singleton):
""" Class to handle loading and lookup of Terminator keybindings """
modifiers = {
'ctrl': Gdk.ModifierType.CONTROL_MASK,
@@ -42,13 +42,24 @@ class Keybindings:
self.keymap = Gdk.Keymap.get_default()
self.configure({})
def print_keys(self):
print(self.keys)
def append_bindings(self, combos):
""" Accept new binding(s) and reload """
for item in combos:
method, keys = item.split(":")
self.keys[method] = keys
self.reload()
def configure(self, bindings):
"""Accept new bindings and reconfigure with them"""
""" Accept new bindings and reconfigure with them """
self.keys = bindings
self.reload()
def reload(self):
"""Parse bindings and mangle into an appropriate form"""
""" Parse bindings and mangle into an appropriate form """
self._lookup = {}
self._masks = 0
@@ -65,10 +76,10 @@ class Keybindings:
try:
keyval, mask = self._parsebinding(binding)
# Does much the same, but with poorer error handling.
#keyval, mask = Gtk.accelerator_parse(binding)
# Does much the same, but with worse error handling.
# keyval, mask = Gtk.accelerator_parse(binding)
except KeymapError as e:
err ("keybinding reload failed to parse binding '%s': %s" % (binding, e))
logger(f"Keybinding reload failed to parse binding '{binding}': {e}")
else:
if mask & Gdk.ModifierType.SHIFT_MASK:
if keyval == Gdk.KEY_Tab:
@@ -87,7 +98,7 @@ class Keybindings:
self._masks |= mask
def _parsebinding(self, binding):
"""Parse an individual binding using Gtk's binding function"""
""" Parse an individual binding using Gtk's binding function """
mask = 0
modifiers = re.findall(MODIFIER, binding)
@@ -102,25 +113,25 @@ class Keybindings:
keyval = Gdk.keyval_from_name(key)
if keyval == 0:
raise KeymapError("Key '%s' is unrecognised..." % key)
raise KeymapError(f"Key '{key}' is unrecognised...")
return (keyval, mask)
def _lookup_modifier(self, modifier):
"""Map modifier names to gtk values"""
""" Map modifier names to gtk values """
try:
return self.modifiers[modifier.lower()]
except KeyError:
raise KeymapError("Unhandled modifier '<%s>'" % modifier)
raise KeymapError(f"Unhandled modifier '<{modifier}>'")
def lookup(self, event):
"""Translate a keyboard event into a mapped key"""
""" Translate a keyboard event into a mapped key """
try:
_found, keyval, _egp, _lvl, consumed = self.keymap.translate_keyboard_state(
event.hardware_keycode,
Gdk.ModifierType(event.get_state() & ~Gdk.ModifierType.LOCK_MASK),
event.group)
except TypeError:
err ("Keybinding lookup failed to translate keyboard event: %s" % dir(event))
logger(f"Keybinding lookup failed to translate keyboard event: {dir(event)}")
return None
mask = (event.get_state() & ~consumed) & self._masks

9
src/utils/logger.py → src/libs/logger.py
View File

@@ -1,10 +1,15 @@
# Python imports
import os, logging
import os
import logging
# Lib imports
# Application imports
from .singleton import Singleton
class Logger:
class Logger(Singleton):
"""
Create a new logging object and return it.
:note:

3
src/libs/mixins/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""
Libs.Mixins Package
"""

70
src/libs/mixins/dnd_mixin.py Normal file
View File

@@ -0,0 +1,70 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('Gdk', '3.0')
from gi.repository import Gtk
from gi.repository import Gdk
from gi.repository import Gio
# Application imports
class DnDMixin:
def _setup_dnd(self):
flags = Gtk.DestDefaults.ALL
PLAIN_TEXT_TARGET_TYPE = 70
URI_TARGET_TYPE = 80
text_target = Gtk.TargetEntry.new('text/plain', Gtk.TargetFlags(0), PLAIN_TEXT_TARGET_TYPE)
uri_target = Gtk.TargetEntry.new('text/uri-list', Gtk.TargetFlags(0), URI_TARGET_TYPE)
# targets = [ text_target, uri_target ]
targets = [ uri_target ]
action = Gdk.DragAction.COPY
# self.drag_dest_set_target_list(targets)
self.drag_dest_set(flags, targets, action)
self._setup_dnd_signals()
def _setup_dnd_signals(self):
# self.connect("drag-motion", self._on_drag_motion)
# self.connect('drag-drop', self._on_drag_set)
self.connect("drag-data-received", self._on_drag_data_received)
def _on_drag_motion(self, widget, drag_context, x, y, time):
Gdk.drag_status(drag_context, drag_context.get_actions(), time)
return False
def _on_drag_set(self, widget, drag_context, data, info, time):
self.drag_get_data(drag_context, drag_context.list_targets()[-1], time)
return False
def _on_drag_data_received(self, widget, drag_context, x, y, data, info, time):
if info == 70: return
if info == 80:
uris = data.get_uris()
files = []
if len(uris) == 0:
uris = data.get_text().split("\n")
for uri in uris:
gfile = None
try:
gfile = Gio.File.new_for_uri(uri)
except Exception as e:
gfile = Gio.File.new_for_path(uri)
files.append(gfile)
event_system.emit('set-pre-drop-dnd', (files,))

31
src/libs/mixins/ipc_signals_mixin.py Normal file
View File

@@ -0,0 +1,31 @@
# Python imports
# Lib imports
import gi
from gi.repository import GLib
# Application imports
class IPCSignalsMixin:
""" IPCSignalsMixin handle messages from another starting {APP_NAME} process. """
def print_to_console(self, message = None):
logger.debug(message)
def handle_file_from_ipc(self, fpath: str) -> None:
logger.debug(f"File From IPC: {fpath}")
GLib.idle_add(
self.broadcast_message, "handle-file", (fpath,)
)
def handle_dir_from_ipc(self, fpath: str) -> None:
logger.debug(f"Dir From IPC: {fpath}")
GLib.idle_add(
self.broadcast_message, "handle-folder", (fpath,)
)
def broadcast_message(self, message_type: str = "none", data: () = ()) -> None:
event_system.emit(message_type, data)

96
src/libs/mixins/keyboard_signals_mixin.py Normal file
View File

@@ -0,0 +1,96 @@
# Python imports
import re
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('Gdk', '3.0')
from gi.repository import Gtk
from gi.repository import Gdk
# Application imports
valid_keyvalue_pat = re.compile(r"[a-z0-9A-Z-_\[\]\(\)\| ]")
class KeyboardSignalsMixin:
""" KeyboardSignalsMixin keyboard hooks controller. """
# TODO: Need to set methods that use this to somehow check the keybindings state instead.
def unset_keys_and_data(self, widget = None, eve = None):
self.ctrl_down = False
self.shift_down = False
self.alt_down = False
def unmap_special_keys(self, keyname):
if "control" in keyname:
self.ctrl_down = False
if "shift" in keyname:
self.shift_down = False
if "alt" in keyname:
self.alt_down = False
def on_global_key_press_controller(self, eve, user_data):
keyname = Gdk.keyval_name(user_data.keyval).lower()
modifiers = Gdk.ModifierType(user_data.get_state() & ~Gdk.ModifierType.LOCK_MASK)
self.was_midified_key = True if modifiers != 0 else False
if keyname.replace("_l", "").replace("_r", "") in ["control", "alt", "shift"]:
if "control" in keyname:
self.ctrl_down = True
if "shift" in keyname:
self.shift_down = True
if "alt" in keyname:
self.alt_down = True
def on_global_key_release_controller(self, widget, event):
""" Handler for keyboard events """
keyname = Gdk.keyval_name(event.keyval).lower()
modifiers = Gdk.ModifierType(event.get_state() & ~Gdk.ModifierType.LOCK_MASK)
if keyname.replace("_l", "").replace("_r", "") in ["control", "alt", "shift"]:
should_return = self.was_midified_key and (self.ctrl_down or self.shift_down or self.alt_down)
self.unmap_special_keys(keyname)
if should_return:
self.was_midified_key = False
return
mapping = keybindings.lookup(event)
logger.debug(f"on_global_key_release_controller > key > {keyname}")
logger.debug(f"on_global_key_release_controller > keyval > {event.keyval}")
logger.debug(f"on_global_key_release_controller > mapping > {mapping}")
if mapping:
self.handle_mapped_key_event(mapping)
else:
self.handle_as_key_event_scope(keyname)
def handle_mapped_key_event(self, mapping):
try:
self.handle_as_controller_scope(mapping)
except Exception:
self.handle_as_plugin_scope(mapping)
def handle_as_controller_scope(self, mapping):
getattr(self, mapping)()
def handle_as_plugin_scope(self, mapping):
if "||" in mapping:
sender, eve_type = mapping.split("||")
else:
sender = ""
eve_type = mapping
self.handle_key_event_system(sender, eve_type)
def handle_as_key_event_scope(self, keyname):
if self.ctrl_down and not keyname in ["1", "kp_1", "2", "kp_2", "3", "kp_3", "4", "kp_4"]:
self.handle_key_event_system(None, keyname)
def handle_key_event_system(self, sender, eve_type):
event_system.emit(eve_type)

4
src/libs/settings/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
"""
Settings Package
"""
from .manager import SettingsManager

210
src/libs/settings/manager.py Normal file
View File

@@ -0,0 +1,210 @@
# Python imports
import inspect
import time
import json
import zipfile
from os import path
from os import mkdir
# Lib imports
# Application imports
from ..singleton import Singleton
from .start_check_mixin import StartCheckMixin
from .options.settings import Settings
class MissingConfigError(Exception):
pass
class SettingsManager(StartCheckMixin, Singleton):
def __init__(self):
self._SCRIPT_PTH: str = path.dirname(path.realpath(__file__))
self._USER_HOME: str = path.expanduser('~')
self._HOME_CONFIG_PATH: str = f"{self._USER_HOME}/.config/{APP_NAME.lower()}"
self._USR_PATH: str = f"/usr/share/{APP_NAME.lower()}"
self._USR_CONFIG_FILE: str = f"{self._USR_PATH}/settings.json"
self._CONTEXT_PATH: str = f"{self._HOME_CONFIG_PATH}/context_path"
self._PLUGINS_PATH: str = f"{self._HOME_CONFIG_PATH}/plugins"
self._DEFAULT_ICONS: str = f"{self._HOME_CONFIG_PATH}/icons"
self._CONFIG_FILE: str = f"{self._HOME_CONFIG_PATH}/settings.json"
self._GLADE_FILE: str = f"{self._HOME_CONFIG_PATH}/Main_Window.glade"
self._CSS_FILE: str = f"{self._HOME_CONFIG_PATH}/stylesheet.css"
self._KEY_BINDINGS_FILE: str = f"{self._HOME_CONFIG_PATH}/key-bindings.json"
self._PID_FILE: str = f"{self._HOME_CONFIG_PATH}/{APP_NAME.lower()}.pid"
self._UI_WIDEGTS_PATH: str = f"{self._HOME_CONFIG_PATH}/ui_widgets"
self._CONTEXT_MENU: str = f"{self._HOME_CONFIG_PATH}/contexct_menu.json"
self._WINDOW_ICON: str = f"{self._DEFAULT_ICONS}/{APP_NAME.lower()}.png"
# self._USR_CONFIG_FILE: str = f"{self._USR_PATH}/settings.json"
# self._PLUGINS_PATH: str = f"plugins"
# self._CONFIG_FILE: str = f"settings.json"
# self._GLADE_FILE: str = f"Main_Window.glade"
# self._CSS_FILE: str = f"stylesheet.css"
# self._KEY_BINDINGS_FILE: str = f"key-bindings.json"
# self._PID_FILE: str = f"{APP_NAME.lower()}.pid"
# self._WINDOW_ICON: str = f"{APP_NAME.lower()}.png"
# self._UI_WIDEGTS_PATH: str = f"ui_widgets"
# self._CONTEXT_MENU: str = f"contexct_menu.json"
# self._DEFAULT_ICONS: str = f"icons"
# with zipfile.ZipFile("files.zip", mode="r", allowZip64=True) as zf:
# with io.TextIOWrapper(zf.open("text1.txt"), encoding="utf-8") as f:
if not path.exists(self._HOME_CONFIG_PATH):
mkdir(self._HOME_CONFIG_PATH)
if not path.exists(self._PLUGINS_PATH):
mkdir(self._PLUGINS_PATH)
if not path.exists(self._DEFAULT_ICONS):
self._DEFAULT_ICONS = f"{self._USR_PATH}/icons"
if not path.exists(self._DEFAULT_ICONS):
raise MissingConfigError("Unable to find the application icons directory.")
if not path.exists(self._GLADE_FILE):
self._GLADE_FILE = f"{self._USR_PATH}/Main_Window.glade"
if not path.exists(self._GLADE_FILE):
raise MissingConfigError("Unable to find the application Glade file.")
if not path.exists(self._KEY_BINDINGS_FILE):
self._KEY_BINDINGS_FILE = f"{self._USR_PATH}/key-bindings.json"
if not path.exists(self._KEY_BINDINGS_FILE):
raise MissingConfigError("Unable to find the application Keybindings file.")
if not path.exists(self._CSS_FILE):
self._CSS_FILE = f"{self._USR_PATH}/stylesheet.css"
if not path.exists(self._CSS_FILE):
raise MissingConfigError("Unable to find the application Stylesheet file.")
if not path.exists(self._WINDOW_ICON):
self._WINDOW_ICON = f"{self._USR_PATH}/icons/{APP_NAME.lower()}.png"
if not path.exists(self._WINDOW_ICON):
raise MissingConfigError("Unable to find the application icon.")
if not path.exists(self._UI_WIDEGTS_PATH):
self._UI_WIDEGTS_PATH = f"{self._USR_PATH}/ui_widgets"
if not path.exists(self._CONTEXT_MENU):
self._CONTEXT_MENU = f"{self._USR_PATH}/contexct_menu.json"
try:
with open(self._KEY_BINDINGS_FILE) as file:
bindings = json.load(file)["keybindings"]
self._guake_key = bindings["guake_key"]
keybindings.configure(bindings)
except Exception as e:
print( f"Settings Manager: {self._KEY_BINDINGS_FILE}\n\t\t{repr(e)}" )
try:
with open(self._CONTEXT_MENU) as file:
self._context_menu_data = json.load(file)
except Exception as e:
print( f"Settings Manager: {self._CONTEXT_MENU}\n\t\t{repr(e)}" )
self.settings: Settings = None
self._main_window = None
self._builder = None
self.PAINT_BG_COLOR: tuple = (0, 0, 0, 0.0)
self._trace_debug: bool = False
self._debug: bool = False
self._dirty_start: bool = False
self._passed_in_file: bool = False
self._starting_files: list = []
def register_signals_to_builder(self, classes = None):
handlers = {}
for c in classes:
methods = None
try:
methods = inspect.getmembers(c, predicate = inspect.ismethod)
handlers.update(methods)
except Exception as e:
...
self._builder.connect_signals(handlers)
def set_main_window(self, window): self._main_window = window
def set_builder(self, builder) -> any: self._builder = builder
def get_monitor_data(self) -> list:
screen = self._main_window.get_screen()
monitors = []
for m in range(screen.get_n_monitors()):
monitors.append(screen.get_monitor_geometry(m))
print("{}x{}+{}+{}".format(monitor.width, monitor.height, monitor.x, monitor.y))
return monitors
def get_main_window(self) -> any: return self._main_window
def get_builder(self) -> any: return self._builder
def get_paint_bg_color(self) -> any: return self.PAINT_BG_COLOR
def get_glade_file(self) -> str: return self._GLADE_FILE
def get_ui_widgets_path(self) -> str: return self._UI_WIDEGTS_PATH
def get_context_menu_data(self) -> str: return self._context_menu_data
def get_context_path(self) -> str: return self._CONTEXT_PATH
def get_plugins_path(self) -> str: return self._PLUGINS_PATH
def get_icon_theme(self) -> str: return self._ICON_THEME
def get_css_file(self) -> str: return self._CSS_FILE
def get_home_config_path(self) -> str: return self._HOME_CONFIG_PATH
def get_window_icon(self) -> str: return self._WINDOW_ICON
def get_home_path(self) -> str: return self._USER_HOME
def get_starting_files(self) -> list: return self._starting_files
def get_guake_key(self) -> tuple: return self._guake_key
def get_starting_args(self):
return self.args, self.unknownargs
def set_main_window_x(self, x: int = 0): self.settings.config.main_window_x = x
def set_main_window_y(self, y: int = 0): self.settings.config.main_window_y = y
def set_main_window_width(self, width: int = 800): self.settings.config.main_window_width = width
def set_main_window_height(self, height: int = 600): self.settings.config.main_window_height = height
def set_main_window_min_width(self, width: int = 720): self.settings.config.main_window_min_width = width
def set_main_window_min_height(self, height: int = 480): self.settings.config.main_window_min_height = height
def set_starting_files(self, files: list): self._starting_files = files
def set_start_load_time(self): self._start_load_time = time.perf_counter()
def set_end_load_time(self): self._end_load_time = time.perf_counter()
def set_starting_args(self, args, unknownargs):
self.args = args
self.unknownargs = unknownargs
def set_trace_debug(self, trace_debug: bool):
self._trace_debug = trace_debug
def set_debug(self, debug: bool):
self._debug = debug
def set_is_starting_with_file(self, is_passed_in_file: bool = False):
self._passed_in_file = is_passed_in_file
def is_trace_debug(self) -> str: return self._trace_debug
def is_debug(self) -> str: return self._debug
def is_starting_with_file(self) -> bool: return self._passed_in_file
def log_load_time(self): logger.info( f"Load Time: {self._end_load_time - self._start_load_time}" )
def call_method(self, target_class: any = None, _method_name: str = "", data: any = None):
method_name = str(_method_name)
method = getattr(target_class, method_name, lambda data: f"No valid key passed...\nkey={method_name}\nargs={data}")
return method(data) if data else method()
def load_settings(self):
if not path.exists(self._CONFIG_FILE):
self.settings = Settings()
return
with open(self._CONFIG_FILE) as file:
data = json.load(file)
data["load_defaults"] = False
self.settings = Settings(**data)
def save_settings(self):
with open(self._CONFIG_FILE, 'w') as outfile:
json.dump(self.settings.as_dict(), outfile, separators=(',', ':'), indent=4)

8
src/libs/settings/options/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
"""
Settings.Options Package
"""
from .settings import Settings
from .config import Config
from .filters import Filters
from .theming import Theming
from .debugging import Debugging

39
src/libs/settings/options/config.py Normal file
View File

@@ -0,0 +1,39 @@
# Python imports
from dataclasses import dataclass, field
# Lib imports
# Application imports
@dataclass
class Config:
base_of_home: str = ""
hide_hidden_files: str = "true"
thumbnailer_path: str = "ffmpegthumbnailer"
blender_thumbnailer_path: str = ""
go_past_home: str = "true"
lock_folder: str = "false"
locked_folders: list = field(default_factory=lambda: [ "venv", "flasks" ])
mplayer_options: str = "-quiet -really-quiet -xy 1600 -geometry 50%:50%",
music_app: str = "/opt/deadbeef/bin/deadbeef"
media_app: str = "mpv"
image_app: str = "mirage"
office_app: str = "libreoffice"
pdf_app: str = "evince"
code_app: str = "atom"
text_app: str = "leafpad"
file_manager_app: str = "solarfm"
terminal_app: str = "terminator"
remux_folder_max_disk_usage: str = "8589934592"
make_transparent: int = 0
main_window_x: int = 721
main_window_y: int = 465
main_window_min_width: int = 720
main_window_min_height: int = 480
main_window_width: int = 800
main_window_height: int = 600
application_dirs: list = field(default_factory=lambda: [
"/usr/share/applications",
f"{settings_manager.get_home_path()}/.local/share/applications"
])

12
src/libs/settings/options/debugging.py Normal file
View File

@@ -0,0 +1,12 @@
# Python imports
from dataclasses import dataclass
# Lib imports
# Application imports
@dataclass
class Debugging:
ch_log_lvl: int = 10
fh_log_lvl: int = 20

90
src/libs/settings/options/filters.py Normal file
View File

@@ -0,0 +1,90 @@
# Python imports
from dataclasses import dataclass, field
# Lib imports
# Application imports
@dataclass
class Filters:
meshs: list = field(default_factory=lambda: [
".blend",
".dae",
".fbx",
".gltf",
".obj",
".stl"
])
code: list = field(default_factory=lambda: [
".cpp",
".css",
".c",
".go",
".html",
".htm",
".java",
".js",
".json",
".lua",
".md",
".py",
".rs",
".toml",
".xml",
".pom"
])
videos: list = field(default_factory=lambda:[
".mkv",
".mp4",
".webm",
".avi",
".mov",
".m4v",
".mpg",
".mpeg",
".wmv",
".flv"
])
office: list = field(default_factory=lambda: [
".doc",
".docx",
".xls",
".xlsx",
".xlt",
".xltx",
".xlm",
".ppt",
".pptx",
".pps",
".ppsx",
".odt",
".rtf"
])
images: list = field(default_factory=lambda: [
".png",
".jpg",
".jpeg",
".gif",
".ico",
".tga",
".webp"
])
text: list = field(default_factory=lambda: [
".txt",
".text",
".sh",
".cfg",
".conf",
".log"
])
music: list = field(default_factory=lambda: [
".psf",
".mp3",
".ogg",
".flac",
".m4a"
])
pdf: list = field(default_factory=lambda: [
".pdf"
])

31
src/libs/settings/options/settings.py Normal file
View File

@@ -0,0 +1,31 @@
# Python imports
from dataclasses import dataclass, field
from dataclasses import asdict
# Gtk imports
# Application imports
from .config import Config
from .filters import Filters
from .theming import Theming
from .debugging import Debugging
@dataclass
class Settings:
load_defaults: bool = True
config: Config = field(default_factory=lambda: Config())
filters: Filters = field(default_factory=lambda: Filters())
theming: Theming = field(default_factory=lambda: Theming())
debugging: Debugging = field(default_factory=lambda: Debugging())
def __post_init__(self):
if not self.load_defaults:
self.load_defaults = False
self.config = Config(**self.config)
self.filters = Filters(**self.filters)
self.theming = Theming(**self.theming)
self.debugging = Debugging(**self.debugging)
def as_dict(self):
return asdict(self)

14
src/libs/settings/options/theming.py Normal file
View File

@@ -0,0 +1,14 @@
# Python imports
from dataclasses import dataclass
# Lib imports
# Application imports
@dataclass
class Theming:
transparency: int = 64
success_color: str = "#88cc27"
warning_color: str = "#ffa800"
error_color: str = "#ff0000"

3
src/libs/settings/other/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""
Settings.Other Package
"""

42
src/libs/settings/other/webkit_ui_settings.py Normal file
View File

@@ -0,0 +1,42 @@
# Python imports
# Lib imports
import gi
gi.require_version('WebKit2', '4.0')
from gi.repository import WebKit2
# Application imports
class WebkitUISettings(WebKit2.Settings):
def __init__(self):
super(WebkitUISettings, self).__init__()
self._set_default_settings()
# Note: Highly insecure setup but most "app" like setup I could think of.
# Audit heavily any scripts/links ran/clicked under this setup!
def _set_default_settings(self):
self.set_enable_xss_auditor(True)
self.set_enable_hyperlink_auditing(True)
# self.set_enable_xss_auditor(False)
# self.set_enable_hyperlink_auditing(False)
self.set_allow_file_access_from_file_urls(True)
self.set_allow_universal_access_from_file_urls(True)
self.set_enable_page_cache(False)
self.set_enable_offline_web_application_cache(False)
self.set_enable_html5_local_storage(False)
self.set_enable_html5_database(False)
self.set_enable_fullscreen(False)
self.set_print_backgrounds(False)
self.set_enable_tabs_to_links(False)
self.set_enable_developer_extras(True)
self.set_enable_webrtc(True)
self.set_enable_webaudio(True)
self.set_enable_accelerated_2d_canvas(True)
self.set_user_agent(f"{APP_NAME}")

63
src/libs/settings/start_check_mixin.py Normal file
View File

@@ -0,0 +1,63 @@
# Python imports
import os
import json
import inspect
# Lib imports
# Application imports
class StartCheckMixin:
def is_dirty_start(self) -> bool:
return self._dirty_start
def clear_pid(self):
if not self.is_trace_debug():
self._clean_pid()
def do_dirty_start_check(self):
if self.is_trace_debug():
pid = os.getpid()
self._print_pid(pid)
return
if os.path.exists(self._PID_FILE):
with open(self._PID_FILE, "r") as f:
pid = f.readline().strip()
if pid not in ("", None):
if self.is_pid_alive( int(pid) ):
print("PID file exists and PID is alive... Letting downstream errors (sans debug args) handle app closure propigation.")
return
self._write_new_pid()
""" Check For the existence of a unix pid. """
def is_pid_alive(self, pid):
print(f"PID Found: {pid}")
try:
os.kill(pid, 0)
except OSError:
print(f"{APP_NAME} PID file exists but PID is irrelevant; starting dirty...")
self._dirty_start = True
return False
return True
def _write_new_pid(self):
pid = os.getpid()
self._write_pid(pid)
self._print_pid(pid)
def _print_pid(self, pid):
print(f"{APP_NAME} PID: {pid}")
def _clean_pid(self):
os.unlink(self._PID_FILE)
def _write_pid(self, pid):
with open(self._PID_FILE, "w") as _pid:
_pid.write(f"{pid}")

22
src/libs/singleton.py Normal file
View File

@@ -0,0 +1,22 @@
# Python imports
# Lib imports
# Application imports
class SingletonError(Exception):
pass
class Singleton:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance:
raise SingletonError(f"'{cls.__name__}' is a Singleton. Cannot create a new instance...")
cls._instance = super(Singleton, cls).__new__(cls)
return cls._instance

67
src/libs/status_icon.py Normal file
View File

@@ -0,0 +1,67 @@
# Python imports
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
gi.require_version('AppIndicator3', '0.1')
from gi.repository import Gtk
from gi.repository import GLib
from gi.repository import AppIndicator3
# Application imports
class StatusIcon():
""" StatusIcon for Application to go to Status Tray. """
def __init__(self):
self._setup_styling()
self._setup_signals()
self._subscribe_to_events()
self._load_widgets()
def _setup_styling(self):
...
def _setup_signals(self):
...
def _subscribe_to_events(self):
...
def _load_widgets(self):
status_menu = Gtk.Menu()
icon_theme = Gtk.IconTheme.get_default()
check_menu_item = Gtk.CheckMenuItem.new_with_label("Update icon")
quit_menu_item = Gtk.MenuItem.new_with_label("Quit")
# Create StatusNotifierItem
self.indicator = AppIndicator3.Indicator.new(
f"{APP_NAME}-statusicon",
"gtk-info",
AppIndicator3.IndicatorCategory.APPLICATION_STATUS)
self.indicator.set_status(AppIndicator3.IndicatorStatus.ACTIVE)
check_menu_item.connect("activate", self.check_menu_item_cb)
quit_menu_item.connect("activate", self.quit_menu_item_cb)
icon_theme.connect('changed', self.icon_theme_changed_cb)
self.indicator.set_menu(status_menu)
status_menu.append(check_menu_item)
status_menu.append(quit_menu_item)
status_menu.show_all()
def update_icon(self, icon_name):
self.indicator.set_icon(icon_name)
def check_menu_item_cb(self, widget, data = None):
icon_name = "parole" if widget.get_active() else "gtk-info"
self.update_icon(icon_name)
def icon_theme_changed_cb(self, theme):
self.update_icon("gtk-info")
def quit_menu_item_cb(self, widget, data = None):
event_system.emit("tear-down")

34
src/plugins/manifest.py
View File

@@ -1,5 +1,6 @@
# Python imports
import os, json
import os
import json
from os.path import join
# Lib imports
@@ -14,13 +15,14 @@ class ManifestProcessor(Exception):
class Plugin:
path: str = None
name: str = None
author: str = None
version: str = None
support: str = None
requests:{} = None
reference: type = None
path: str = None
name: str = None
author: str = None
version: str = None
support: str = None
requests:{} = None
reference: type = None
pre_launch: bool = False
class ManifestProcessor:
@@ -45,19 +47,29 @@ class ManifestProcessor:
plugin.support = self._manifest["support"]
plugin.requests = self._manifest["requests"]
if "pre_launch" in self._manifest.keys():
plugin.pre_launch = True if self._manifest["pre_launch"] == "true" else False
return plugin
def get_loading_data(self):
loading_data = {}
requests = self._plugin.requests
keys = requests.keys()
if "pass_events" in keys:
if "pass_events" in requests:
if requests["pass_events"] in ["true"]:
loading_data["pass_events"] = True
if "bind_keys" in keys:
if "pass_ui_objects" in requests:
if isinstance(requests["pass_ui_objects"], list):
loading_data["pass_ui_objects"] = [ self._builder.get_object(obj) for obj in requests["pass_ui_objects"] ]
if "bind_keys" in requests:
if isinstance(requests["bind_keys"], list):
loading_data["bind_keys"] = requests["bind_keys"]
return self._plugin, loading_data
def is_pre_launch(self):
return self._plugin.pre_launch

50
src/plugins/plugin_base.py
View File

@@ -1,5 +1,7 @@
# Python imports
import os, time
import os
import time
import inspect
# Lib imports
@@ -11,7 +13,8 @@ class PluginBaseException(Exception):
class PluginBase:
def __init__(self):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.name = "Example Plugin" # NOTE: Need to remove after establishing private bidirectional 1-1 message bus
# where self.name should not be needed for message comms
@@ -35,13 +38,6 @@ class PluginBase:
"""
raise PluginBaseException("Method hasn't been overriden...")
def set_event_system(self, event_system):
"""
Requests Key: 'pass_events': "true"
Must define in plugin if "pass_events" is set to "true" string.
"""
self._event_system = event_system
def set_ui_object_collection(self, ui_objects):
"""
Requests Key: "pass_ui_objects": [""]
@@ -50,9 +46,45 @@ class PluginBase:
"""
self._ui_objects = ui_objects
def set_event_system(self, event_system):
"""
Requests Key: 'pass_events': "true"
Must define in plugin if "pass_events" is set to "true" string.
"""
self._event_system = event_system
def subscribe_to_events(self):
...
def _connect_builder_signals(self, caller_class, builder):
classes = [caller_class]
handlers = {}
for c in classes:
methods = None
try:
methods = inspect.getmembers(c, predicate=inspect.ismethod)
handlers.update(methods)
except Exception as e:
logger.debug(repr(e))
builder.connect_signals(handlers)
def reload_package(self, plugin_path, module_dict_main=locals()):
import importlib
from pathlib import Path
def reload_package_recursive(current_dir, module_dict):
for path in current_dir.iterdir():
if "__init__" in str(path) or path.stem not in module_dict:
continue
if path.is_file() and path.suffix == ".py":
importlib.reload(module_dict[path.stem])
elif path.is_dir():
reload_package_recursive(path, module_dict[path.stem].__dict__)
reload_package_recursive(Path(plugin_path).parent, module_dict_main["module_dict_main"])
def clear_children(self, widget: type) -> None:
""" Clear children of a gtk widget. """

85
src/plugins/plugins_controller.py
View File

@@ -1,14 +1,21 @@
# Python imports
import os, sys, importlib, traceback
from os.path import join, isdir
import os
import sys
import importlib
import traceback
from os.path import join
from os.path import isdir
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, Gio
from gi.repository import Gtk
from gi.repository import GLib
from gi.repository import Gio
# Application imports
from .manifest import Plugin, ManifestProcessor
from .manifest import Plugin
from .manifest import ManifestProcessor
@@ -24,16 +31,28 @@ class PluginsController:
path = os.path.dirname(os.path.realpath(__file__))
sys.path.insert(0, path) # NOTE: I think I'm not using this correctly...
self._builder = settings.get_builder()
self._plugins_path = settings.get_plugins_path()
self._builder = settings_manager.get_builder()
self._plugins_path = settings_manager.get_plugins_path()
self._plugins_dir_watcher = None
self._plugin_collection = []
self._plugin_manifests = {}
self._load_manifests()
def launch_plugins(self) -> None:
def _load_manifests(self):
logger.info(f"Loading manifests...")
for path, folder in [[join(self._plugins_path, item), item] if os.path.isdir(join(self._plugins_path, item)) else None for item in os.listdir(self._plugins_path)]:
manifest = ManifestProcessor(path, self._builder)
self._plugin_manifests[path] = {
"path": path,
"folder": folder,
"manifest": manifest
}
self._set_plugins_watcher()
self.load_plugins()
def _set_plugins_watcher(self) -> None:
self._plugins_dir_watcher = Gio.File.new_for_path(self._plugins_path) \
@@ -46,24 +65,50 @@ class PluginsController:
Gio.FileMonitorEvent.MOVED_OUT]:
self.reload_plugins(file)
def load_plugins(self, file: str = None) -> None:
print(f"Loading plugins...")
def pre_launch_plugins(self) -> None:
logger.info(f"Loading pre-launch plugins...")
plugin_manifests: {} = {}
for key in self._plugin_manifests:
target_manifest = self._plugin_manifests[key]["manifest"]
if target_manifest.is_pre_launch():
plugin_manifests[key] = self._plugin_manifests[key]
self._load_plugins(plugin_manifests, is_pre_launch = True)
def post_launch_plugins(self) -> None:
logger.info(f"Loading post-launch plugins...")
plugin_manifests: {} = {}
for key in self._plugin_manifests:
target_manifest = self._plugin_manifests[key]["manifest"]
if not target_manifest.is_pre_launch():
plugin_manifests[key] = self._plugin_manifests[key]
self._load_plugins(plugin_manifests)
def _load_plugins(self, plugin_manifests: {} = {}, is_pre_launch: bool = False) -> None:
parent_path = os.getcwd()
for path, folder in [[join(self._plugins_path, item), item] if os.path.isdir(join(self._plugins_path, item)) else None for item in os.listdir(self._plugins_path)]:
try:
target = join(path, "plugin.py")
manifest = ManifestProcessor(path, self._builder)
for key in plugin_manifests:
target_manifest = plugin_manifests[key]
path, folder, manifest = target_manifest["path"], target_manifest["folder"], target_manifest["manifest"]
try:
target = join(path, "plugin.py")
if not os.path.exists(target):
raise InvalidPluginException("Invalid Plugin Structure: Plugin doesn't have 'plugin.py'. Aboarting load...")
plugin, loading_data = manifest.get_loading_data()
module = self.load_plugin_module(path, folder, target)
self.execute_plugin(module, plugin, loading_data)
if is_pre_launch:
self.execute_plugin(module, plugin, loading_data)
else:
GLib.idle_add(self.execute_plugin, *(module, plugin, loading_data))
except Exception as e:
print(f"Malformed Plugin: Not loading -->: '{folder}' !")
traceback.print_exc()
logger.info(f"Malformed Plugin: Not loading -->: '{folder}' !")
logger.debug("Trace: ", traceback.print_exc())
os.chdir(parent_path)
@@ -94,13 +139,13 @@ class PluginsController:
if "ui_target" in keys:
loading_data["ui_target"].add( plugin.reference.generate_reference_ui_element() )
loading_data["ui_target"].show_all()
loading_data["ui_target"].show()
if "pass_ui_objects" in keys:
plugin.reference.set_ui_object_collection( loading_data["pass_ui_objects"] )
if "pass_events" in keys:
plugin.reference.set_fm_event_system(event_system)
plugin.reference.set_event_system(event_system)
plugin.reference.subscribe_to_events()
if "bind_keys" in keys:
@@ -110,4 +155,4 @@ class PluginsController:
self._plugin_collection.append(plugin)
def reload_plugins(self, file: str = None) -> None:
print(f"Reloading plugins... stub.")
logger.info(f"Reloading plugins... stub.")

3
src/utils/__init__.py
View File

@@ -1,3 +0,0 @@
"""
Utils module
"""

30
src/utils/event_system.py
View File

@@ -1,30 +0,0 @@
# Python imports
from collections import defaultdict
# Lib imports
# Application imports
class EventSystem:
""" Create event system. """
def __init__(self):
self.subscribers = defaultdict(list)
def subscribe(self, event_type, fn):
self.subscribers[event_type].append(fn)
def emit(self, event_type, data = None):
if event_type in self.subscribers:
for fn in self.subscribers[event_type]:
if data:
if hasattr(data, '__iter__') and not type(data) is str:
fn(*data)
else:
fn(data)
else:
fn()

152
src/utils/settings.py
View File

@@ -1,152 +0,0 @@
# Python imports
import os
import json
import inspect
# Gtk imports
# Application imports
from .logger import Logger
class Settings:
def __init__(self):
self._SCRIPT_PTH = os.path.dirname(os.path.realpath(__file__))
self._USER_HOME = os.path.expanduser('~')
self._CONFIG_PATH = f"{self._USER_HOME}/.config/{app_name.lower()}"
self._PLUGINS_PATH = f"{self._CONFIG_PATH}/plugins"
self._GLADE_FILE = f"{self._CONFIG_PATH}/Main_Window.glade"
self._KEY_BINDINGS_FILE = f"{self._CONFIG_PATH}/key-bindings.json"
self._CSS_FILE = f"{self._CONFIG_PATH}/stylesheet.css"
self._DEFAULT_ICONS = f"{self._CONFIG_PATH}/icons"
self._PID_FILE = f"{self._CONFIG_PATH}/{app_name.lower()}.pid"
self._WINDOW_ICON = f"{self._DEFAULT_ICONS}/{app_name.lower()}.png"
self._USR_PATH = f"/usr/share/{app_name.lower()}"
if not os.path.exists(self._CONFIG_PATH):
os.mkdir(self._CONFIG_PATH)
if not os.path.exists(self._PLUGINS_PATH):
os.mkdir(self._PLUGINS_PATH)
if not os.path.exists(self._GLADE_FILE):
self._GLADE_FILE = f"{self._USR_PATH}/Main_Window.glade"
if not os.path.exists(self._KEY_BINDINGS_FILE):
self._KEY_BINDINGS_FILE = f"{self._USR_PATH}/key-bindings.json"
if not os.path.exists(self._CSS_FILE):
self._CSS_FILE = f"{self._USR_PATH}/stylesheet.css"
if not os.path.exists(self._WINDOW_ICON):
self._WINDOW_ICON = f"{self._USR_PATH}/icons/{app_name.lower()}.png"
if not os.path.exists(self._DEFAULT_ICONS):
self.DEFAULT_ICONS = f"{self._USR_PATH}/icons"
# '_filters'
self._office_filter = ('.doc', '.docx', '.xls', '.xlsx', '.xlt', '.xltx', '.xlm', '.ppt', 'pptx', '.pps', '.ppsx', '.odt', '.rtf')
self._vids_filter = ('.mkv', '.avi', '.flv', '.mov', '.m4v', '.mpg', '.wmv', '.mpeg', '.mp4', '.webm')
self._txt_filter = ('.txt', '.text', '.sh', '.cfg', '.conf')
self._music_filter = ('.psf', '.mp3', '.ogg' , '.flac')
self._images_filter = ('.png', '.jpg', '.jpeg', '.gif', '.ico', '.tga')
self._pdf_filter = ('.pdf')
self._success_color = "#88cc27"
self._warning_color = "#ffa800"
self._error_color = "#ff0000"
with open(self._KEY_BINDINGS_FILE) as file:
bindings = json.load(file)["keybindings"]
keybindings.configure(bindings)
self._guake_key = bindings["guake_key"]
self._builder = None
self._logger = Logger(self._CONFIG_PATH).get_logger()
self._trace_debug = False
self._debug = False
self._dirty_start = False
def do_dirty_start_check(self):
if not os.path.exists(self._PID_FILE):
self._write_new_pid()
else:
with open(self._PID_FILE, "r") as _pid:
pid = _pid.readline().strip()
if pid not in ("", None):
self._check_alive_status(int(pid))
else:
self._write_new_pid()
""" Check For the existence of a unix pid. """
def _check_alive_status(self, pid):
print(f"PID Found: {pid}")
try:
os.kill(pid, 0)
except OSError:
print(f"{app_name} is starting dirty...")
self._dirty_start = True
self._write_new_pid()
return
print("PID is alive... Let downstream errors (sans debug args) handle app closure propigation.")
def _write_new_pid(self):
pid = os.getpid()
self._write_pid(pid)
def _clean_pid(self):
os.unlink(self._PID_FILE)
def _write_pid(self, pid):
with open(self._PID_FILE, "w") as _pid:
_pid.write(f"{pid}")
def register_signals_to_builder(self, classes=None):
handlers = {}
for c in classes:
methods = None
try:
methods = inspect.getmembers(c, predicate=inspect.ismethod)
handlers.update(methods)
except Exception as e:
print(repr(e))
self._builder.connect_signals(handlers)
def set_builder(self, builder) -> any: self._builder = builder
def get_builder(self) -> any: return self._builder
def get_glade_file(self) -> str: return self._GLADE_FILE
def get_logger(self) -> Logger: return self._logger
def get_plugins_path(self) -> str: return self._PLUGINS_PATH
def get_icon_theme(self) -> str: return self._ICON_THEME
def get_css_file(self) -> str: return self._CSS_FILE
def get_window_icon(self) -> str: return self._WINDOW_ICON
def get_home_path(self) -> str: return self._USER_HOME
# Filter returns
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_guake_key(self) -> tuple: return self._guake_key
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
def is_trace_debug(self) -> str: return self._trace_debug
def is_debug(self) -> str: return self._debug
def is_dirty_start(self) -> bool: return self._dirty_start
def clear_pid(self): self._clean_pid()
def set_trace_debug(self, trace_debug):
self._trace_debug = trace_debug
def set_debug(self, debug):
self._debug = debug