itdominator/Python-With-Gtk-Template
Template
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Pligins refactor with new context and controller integration

Browse Source
This commit is contained in:
itdominator
2026-01-18 20:33:49 -06:00
parent e2f29207ba
commit b8ce6e160a
13 changed files with 181 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
plugins/template/manifest.json
View File

@@ -1,13 +1,9 @@
{
"manifest": {
"name": "Example Plugin",
"author": "John Doe",
"version": "0.0.1",
"support": "",
"requests": {
"ui_target": "plugin_control_list",
"pass_events": true,
"bind_keys": ["Example Plugin||send_message:<Control>f"]
}
}
}

41
plugins/template/plugin.py
View File

@@ -1,8 +1,4 @@
# Python imports
import os
import threading
import subprocess
import time
# Lib imports
import gi
@@ -14,38 +10,27 @@ from plugins.plugin_base import PluginBase
# NOTE: Threads WILL NOT die with parent's destruction.
def threaded(fn):
def wrapper(*args, **kwargs):
threading.Thread(target=fn, args=args, kwargs=kwargs, daemon=False).start()
return wrapper
# NOTE: Threads WILL die with parent's destruction.
def daemon_threaded(fn):
def wrapper(*args, **kwargs):
threading.Thread(target=fn, args=args, kwargs=kwargs, daemon=True).start()
return wrapper
class Plugin(PluginBase):
def __init__(self):
super().__init__()
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
def generate_reference_ui_element(self):
button = Gtk.Button(label=self.name)
button.connect("button-release-event", self.send_message)
return button
def load(self):
ui_element = self.requests_ui_element("plugin_control_list")
ui_element.add( self.generate_plugin_element() )
def run(self):
...
def generate_plugin_element(self):
button = Gtk.Button(label = self.name)
button.connect("button-release-event", self.send_message)
button.show()
return button
def send_message(self, widget = None, eve = None):
message = "Hello, World!"
event_system.emit("display_message", ("warning", message, None))
self.emit("display_message", ("warning", message, None))

4
src/libs/dto/plugins/requests.py
View File

@@ -8,8 +8,4 @@ from dataclasses import dataclass, field
@dataclass
class Requests:
ui_target: str = ""
ui_target_id: str = ""
pass_events: bool = False
pass_ui_objects: list = field(default_factory = lambda: [])
bind_keys: list = field(default_factory = lambda: [])

10
src/libs/event_system.py
View File

@@ -28,16 +28,16 @@ class EventSystem(Singleton):
def _resume_processing_events(self):
self._is_paused = False
def subscribe(self, event_type, fn):
def subscribe(self, event_type: str, fn: callable):
self.subscribers[event_type].append(fn)
def unsubscribe(self, event_type, fn):
def unsubscribe(self, event_type: str, fn: callable):
self.subscribers[event_type].remove(fn)
def unsubscribe_all(self, event_type):
def unsubscribe_all(self, event_type: str):
self.subscribers.pop(event_type, None)
def emit(self, event_type, data = None):
def emit(self, event_type: str, data: tuple = ()):
if self._is_paused and event_type != "resume_event_processing":
return
@@ -51,7 +51,7 @@ class EventSystem(Singleton):
else:
fn()
def emit_and_await(self, event_type, data = None):
def emit_and_await(self, event_type: str, data: tuple = ()):
if self._is_paused and event_type != "resume_event_processing":
return

3
src/plugins/dto/__init__.py
View File

@@ -1,3 +0,0 @@
"""
Gtk Plugins DTO Module
"""

27
src/plugins/dto/manifest.py
View File

@@ -1,27 +0,0 @@
# Python imports
from dataclasses import dataclass, field
from dataclasses import asdict
# Gtk imports
# Application imports
from .requests import Requests
@dataclass
class Manifest:
name: str = ""
author: str = ""
credit: str = ""
version: str = "0.0.1"
support: str = "support@mail.com"
pre_launch: bool = False
requests: Requests = field(default_factory = lambda: Requests())
def __post_init__(self):
if isinstance(self.requests, dict):
self.requests = Requests(**self.requests)
def as_dict(self):
return asdict(self)

19
src/plugins/dto/manifest_meta.py
View File

@@ -1,19 +0,0 @@
# Python imports
from dataclasses import dataclass, field
from dataclasses import asdict
# Gtk imports
# Application imports
from .manifest import Manifest
@dataclass
class ManifestMeta:
folder: str = ""
path: str = ""
manifest: Manifest = field(default_factory = lambda: Manifest())
def as_dict(self):
return asdict(self)

16
src/plugins/dto/requests.py
View File

@@ -1,16 +0,0 @@
# Python imports
from dataclasses import dataclass, field
# Lib imports
# Application imports
@dataclass
class Requests:
ui_target: str = ""
ui_target_id: str = ""
pass_events: bool = False
pass_fm_events: bool = False
pass_ui_objects: list = field(default_factory = lambda: [])
bind_keys: list = field(default_factory = lambda: [])

10
src/plugins/manifest_manager.py
View File

@@ -6,8 +6,8 @@ from os.path import join
# Lib imports
# Application imports
from .dto.manifest_meta import ManifestMeta
from .dto.manifest import Manifest
from libs.dto.plugins.manifest_meta import ManifestMeta
from libs.dto.plugins.manifest import Manifest
@@ -21,8 +21,8 @@ class ManifestManager:
self._plugins_path = settings_manager.path_manager.get_plugins_path()
self.pre_launch_manifests = []
self.post_launch_manifests = []
self.pre_launch_manifests: list = []
self.post_launch_manifests: list = []
self.load_manifests()
@@ -60,7 +60,7 @@ class ManifestManager:
else:
self.post_launch_manifests.append(manifest_meta)
def get_pre_launch_manifests(self) -> dict:
def get_pre_launch_plugins(self) -> dict:
return self.pre_launch_manifests
def get_post_launch_plugins(self) -> None:

97
src/plugins/plugin_base.py
View File

@@ -1,92 +1,49 @@
# Python imports
import os
import time
import inspect
# Lib imports
# Application imports
from libs.dto.base_event import BaseEvent
from .plugin_context import PluginContext
class PluginBaseException(Exception):
...
class PluginBase:
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
def __init__(self, *args, **kwargs):
super(PluginBase, self).__init__(*args, **kwargs)
self._builder = None
self._ui_objects = None
self._event_system = None
self.plugin_context: PluginContext = None
def _controller_message(self):
raise PluginBaseException("Plugin Base '_controller_message' must be overriden by Plugin")
def load(self):
raise PluginBaseException("Plugin Base 'load' must be overriden by Plugin")
def run(self):
"""
Must define regardless if needed and can 'pass' if plugin doesn't need it.
Is intended to be used to setup internal signals or custom Gtk Builders/UI logic.
"""
raise PluginBaseException("Method hasn't been overriden...")
raise PluginBaseException("Plugin Base 'run' must be overriden by Plugin")
def generate_reference_ui_element(self):
"""
Requests Key: 'ui_target': "plugin_control_list",
Must define regardless if needed and can 'pass' if plugin doesn't use it.
Must return a widget if "ui_target" is set.
"""
raise PluginBaseException("Method hasn't been overriden...")
def requests_ui_element(self, element_id: str):
return self.plugin_context.requests_ui_element(element_id)
def set_ui_object_collection(self, ui_objects):
"""
Requests Key: "pass_ui_objects": [""]
Request reference to a UI component. Will be passed back as array to plugin.
Must define in plugin if set and an array of valid glade UI IDs is given.
"""
self._ui_objects = ui_objects
def message(self, event: BaseEvent):
return self.plugin_context.message(event)
def set_event_system(self, event_system):
"""
Requests Key: 'pass_events': true
Must define in plugin if "pass_events" is set to true.
"""
self._event_system = event_system
def message_to(self, name: str, event: BaseEvent):
return self.plugin_context.message_to(name, event)
def subscribe_to_events(self):
...
def message_to_selected(self, names: list[str], event: BaseEvent):
return self.plugin_context.message_to_selected(names, event)
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))
def emit(self, event_type: str, data: tuple = ()):
self.plugin_context.emit(event_type, data)
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. """
for child in widget.get_children():
widget.remove(child)
def emit_and_await(self, event_type: str, data: tuple = ()):
self.plugin_context.emit_and_await(event_type, data)

40
src/plugins/plugin_context.py Normal file
View File

@@ -0,0 +1,40 @@
# Python imports
# Lib imports
# Application imports
from libs.dto.base_event import BaseEvent
class PluginContextException(Exception):
...
class PluginContext:
""" PluginContext """
def __init__(self):
super(PluginContext, self).__init__()
def requests_ui_element(self, element_id: str):
raise PluginContextException("Plugin Context 'requests_ui_element' must be overridden...")
def _controller_message(self, event: BaseEvent):
raise PluginContextException("Plugin Context '_controller_message' must be overridden...")
def message(self, event: BaseEvent):
raise PluginContextException("Plugin Context 'message' must be overridden...")
def message_to(self, name: str, event: BaseEvent):
raise PluginContextException("Plugin Context 'message_to' must be overridden...")
def message_to_selected(self, names: list[str], event: BaseEvent):
raise PluginContextException("Plugin Context 'message_to_selected' must be overridden...")
def emit(self, event_type: str, data: tuple = ()):
raise PluginContextException("Plugin Context 'emit' must be overridden...")
def emit_and_await(self, event_type: str, data: tuple = ()):
raise PluginContextException("Plugin Context 'emit_and_await' must be overridden...")

119
src/plugins/plugins_controller.py
View File

@@ -11,47 +11,56 @@ import gi
from gi.repository import GLib
# Application imports
from .dto.manifest_meta import ManifestMeta
from .plugin_reload_mixin import PluginReloadMixin
from libs.controllers.controller_base import ControllerBase
from libs.dto.plugins.manifest_meta import ManifestMeta
from libs.dto.base_event import BaseEvent
from .manifest_manager import ManifestManager
from .plugins_controller_mixin import PluginsControllerMixin
from .plugin_reload_mixin import PluginReloadMixin
from .plugin_context import PluginContext
class InvalidPluginException(Exception):
class PluginsControllerException(Exception):
...
class PluginsController(PluginReloadMixin):
class PluginsController(ControllerBase, PluginsControllerMixin, PluginReloadMixin):
""" PluginsController controller """
def __init__(self):
super(PluginsController, self).__init__()
# path = os.path.dirname(os.path.realpath(__file__))
# sys.path.insert(0, path) # NOTE: I think I'm not using this correctly...
self._plugin_collection = []
self._plugin_collection: list = []
self._plugins_path = settings_manager.path_manager.get_plugins_path()
self._manifest_manager = ManifestManager()
self._plugins_path: str = settings_manager.path_manager.get_plugins_path()
self._manifest_manager: ManifestManager = ManifestManager()
self._set_plugins_watcher()
def pre_launch_plugins(self) -> None:
logger.info(f"Loading pre-launch plugins...")
manifest_metas: [] = self._manifest_manager.get_pre_launch_manifests()
self._load_plugins(manifest_metas, is_pre_launch = True)
def _controller_message(self, event: BaseEvent):
...
def post_launch_plugins(self) -> None:
logger.info(f"Loading post-launch plugins...")
manifest_metas: [] = self._manifest_manager.get_post_launch_plugins()
self._load_plugins(manifest_metas)
def _collect_search_locations(self, path: str, locations: list):
locations.append(path)
for file in os.listdir(path):
_path = os.path.join(path, file)
if os.path.isdir(_path):
self.collect_search_locations(_path, locations)
def _load_plugins(
self,
manifest_metas: [] = [],
manifest_metas: list = [],
is_pre_launch: bool = False
) -> None:
):
parent_path = os.getcwd()
for manifest_meta in manifest_metas:
@@ -62,7 +71,7 @@ class PluginsController(PluginReloadMixin):
if not os.path.exists(target):
raise InvalidPluginException("Invalid Plugin Structure: Plugin doesn't have 'plugin.py'. Aboarting load...")
module = self.load_plugin_module(path, folder, target)
module = self._load_plugin_module(path, folder, target)
if is_pre_launch:
self.execute_plugin(module, manifest_meta)
@@ -70,15 +79,15 @@ class PluginsController(PluginReloadMixin):
GLib.idle_add(self.execute_plugin, module, manifest_meta)
except Exception as e:
logger.info(f"Malformed Plugin: Not loading -->: '{folder}' !")
logger.debug("Trace: ", traceback.print_exc())
logger.debug(f"Trace: {traceback.print_exc()}")
os.chdir(parent_path)
def load_plugin_module(self, path, folder, target):
def _load_plugin_module(self, path, folder, target):
os.chdir(path)
locations = []
self.collect_search_locations(path, locations)
self._collect_search_locations(path, locations)
spec = importlib.util.spec_from_file_location(folder, target, submodule_search_locations = locations)
module = importlib.util.module_from_spec(spec)
@@ -87,48 +96,44 @@ class PluginsController(PluginReloadMixin):
return module
def collect_search_locations(self, path: str, locations: list):
locations.append(path)
for file in os.listdir(path):
_path = os.path.join(path, file)
if os.path.isdir(_path):
self.collect_search_locations(_path, locations)
def create_plugin_context(self):
plugin_context: PluginContext = PluginContext()
plugin_context.requests_ui_element: callable = self.requests_ui_element
plugin_context.message: callable = self.message
plugin_context.message_to: callable = self.message_to
plugin_context.message_to_selected: callable = self.message_to_selected
plugin_context.emit: callable = event_system.emit
plugin_context.emit_and_await: callable = event_system.emit_and_await
return plugin_context
def pre_launch_plugins(self) -> None:
logger.info(f"Loading pre-launch plugins...")
manifest_metas: list = self._manifest_manager.get_pre_launch_plugins()
self._load_plugins(manifest_metas, is_pre_launch = True)
def post_launch_plugins(self) -> None:
logger.info(f"Loading post-launch plugins...")
manifest_metas: list = self._manifest_manager.get_post_launch_plugins()
self._load_plugins(manifest_metas)
def execute_plugin(self, module: type, manifest_meta: ManifestMeta):
plugin = module.Plugin()
plugin.plugin_context: PluginContext = self.create_plugin_context()
plugin._controller_message: callable = self._controller_message
manifest = manifest_meta.manifest
manifest_meta.instance = module.Plugin()
if manifest.requests.ui_target:
builder = settings_manager.get_builder()
ui_target = manifest.requests.ui_target
ui_target_id = manifest.requests.ui_target_id
if not ui_target == "other":
ui_target = builder.get_object(ui_target)
else:
if not ui_target_id:
raise InvalidPluginException('Invalid "ui_target_id" given in requests. Must have one if setting "ui_target" to "other"...')
ui_target = builder.get_object(ui_target_id)
if not ui_target:
raise InvalidPluginException('Unknown "ui_target" given in requests.')
ui_element = manifest_meta.instance.generate_reference_ui_element()
ui_target.add(ui_element)
if manifest.requests.pass_ui_objects:
manifest_meta.instance.set_ui_object_collection(
[ builder.get_object(obj) for obj in manifest.requests.pass_ui_objects ]
)
if manifest.requests.pass_events:
manifest_meta.instance.set_event_system(event_system)
manifest_meta.instance.subscribe_to_events()
manifest_meta.instance = plugin
if manifest.requests.bind_keys:
keybindings.append_bindings( manifest.requests.bind_keys )
manifest_meta.instance.load()
manifest_meta.instance.run()
self._plugin_collection.append(manifest_meta)
plugins_controller = PluginsController()

19
src/plugins/plugins_controller_mixin.py Normal file
View File

@@ -0,0 +1,19 @@
# Python imports
# Lib imports
# Application imports
class PluginsControllerMixin:
def requests_ui_element(self, target_id: str):
builder = settings_manager.get_builder()
ui_target = builder.get_object(target_id)
if not ui_target:
raise InvalidPluginException('Unknown "target_id" given in requests.')
return ui_target