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

Updates, additions, minor improvements

Browse Source
This commit is contained in:
itdominator 2022-09-05 18:01:39 -05:00
parent 477ba79f93
commit e1c42d9839
12 changed files with 306 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
plugins/template/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""
Pligin Module
"""

56
plugins/template/__main__.py
View File

@ -1,53 +1,3 @@
# Python imports
import sys, threading, subprocess, time
# Gtk imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
def threaded(fn):
def wrapper(*args, **kwargs):
threading.Thread(target=fn, args=args, kwargs=kwargs, daemon=False).start()
return wrapper
class Plugin:
def __init__(self, builder, event_system):
self._plugin_name = "Example Plugin"
self._builder = builder
self._event_system = event_system
self._message = None
self._time_out = 5
button = Gtk.Button(label=self._plugin_name)
button.connect("button-release-event", self._do_action)
plugin_list = self._builder.get_object("plugin_socket")
plugin_list.add(button)
plugin_list.show_all()
@threaded
def _do_action(self, widget=None, eve=None):
message = "Hello, World!"
self._event_system.push_gui_event(["some_type", "display_message", ("warning", message, None)])
def set_message(self, data):
self._message = data
def get_plugin_name(self):
return self._plugin_name
def get_socket_id(self):
return self._socket_id
def _run_timeout(self):
timeout = 0
while not self._message and timeout < self._time_out:
time.sleep(1)
timeout += 1
"""
Pligin Package
"""

13
plugins/template/manifest.json Normal file
View File

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

76
plugins/template/plugin.py Normal file
View File

@ -0,0 +1,76 @@
# Python imports
import os, threading, subprocess, time
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
# 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:
def __init__(self):
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
self._event_system = None
self._event_sleep_time = .5
self._event_message = None
def get_ui_element(self):
button = Gtk.Button(label=self.name)
button.connect("button-release-event", self.send_message)
return button
def set_fm_event_system(self, fm_event_system):
self._event_system = fm_event_system
def run(self):
self._module_event_observer()
def send_message(self, widget=None, eve=None):
message = "Hello, World!"
self._event_system.push_gui_event([self.name, "display_message", ("warning", message, None)])
def wait_for_fm_message(self):
while not self._event_message:
pass
@daemon_threaded
def _module_event_observer(self):
while True:
time.sleep(self._event_sleep_time)
event = self._event_system.read_module_event()
if event:
try:
if event[0] == self.name:
target_id, method_target, data = self._event_system.consume_module_event()
if not method_target:
self._event_message = data
else:
method = getattr(self.__class__, f"{method_target}")
if data:
data = method(*(self, *data))
else:
method(*(self,))
except Exception as e:
print(repr(e))

73
src/__builtins__.py
View File

@ -1,4 +1,4 @@
import builtins
import builtins, threading
# Python imports
import builtins
@ -6,60 +6,38 @@ import builtins
# Lib imports
# Application imports
from utils.ipc_server import IPCServer
from utils.event_system import EventSystem
class EventSystem(IPCServer):
""" Inheret IPCServerMixin. Create an pub/sub systems. """
# 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()
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()
return wrapper
class EndpointRegistry():
def __init__(self):
super(EventSystem, self).__init__()
self._endpoints = {}
# NOTE: The format used is list of ['who', target, (data,)] Where:
# who is the sender or target ID and is used for context and control flow,
# method_target is the method to call,
# data is the method parameters OR message data to give
# Where data may be any kind of data
self._gui_events = []
self._module_events = []
def register(self, rule, **options):
def decorator(f):
self._endpoints[rule] = f
return f
return decorator
# Makeshift "events" system FIFO
def _pop_gui_event(self) -> None:
if len(self._gui_events) > 0:
return self._gui_events.pop(0)
def get_endpoints(self):
return self._endpoints
def _pop_module_event(self) -> None:
if len(self._module_events) > 0:
return self._module_events.pop(0)
def push_gui_event(self, event: list) -> None:
if len(event) == 3:
self._gui_events.append(event)
return None
raise Exception("Invald event format! Please do: ['target_id': str, method_target: method, (data,): any]")
def push_module_event(self, event: list) -> None:
if len(event) == 3:
self._module_events.append(event)
return None
raise Exception("Invald event format! Please do: [type, target, (data,)]")
def read_gui_event(self) -> list:
return self._gui_events[0] if self._gui_events else None
def read_module_event(self) -> list:
return self._module_events[0] if self._module_events else None
def consume_gui_event(self) -> None:
return self._pop_gui_event()
def consume_module_event(self) -> None:
return self._pop_module_event()
@ -67,6 +45,9 @@ class EventSystem(IPCServer):
# __builtins__.update({"event_system": Builtins()})
builtins.app_name = "<change_me>"
builtins.event_system = EventSystem()
builtins.endpoint_registry = EndpointRegistry()
builtins.threaded = threaded_wrapper
builtins.daemon_threaded = daemon_threaded_wrapper
builtins.event_sleep_time = 0.05
builtins.trace_debug = False
builtins.debug = False

25
src/app.py
View File

@ -4,30 +4,37 @@ import os, inspect, time
# Lib imports
# Application imports
from __builtins__ import *
from utils.ipc_server import IPCServer
from utils.settings import Settings
from context.controller import Controller
from __builtins__ import EventSystem
class App_Launch_Exception(Exception):
...
class Controller_Start_Exceptio(Exception):
...
class Application(EventSystem):
class Application(IPCServer):
''' Create Settings and Controller classes. Bind signal to Builder. Inherit from Builtins to bind global methods and classes.'''
def __init__(self, args, unknownargs):
if not debug:
event_system.create_ipc_listener()
time.sleep(0.1)
super(Application, self).__init__()
if not trace_debug:
if not event_system.is_ipc_alive:
self.create_ipc_listener()
time.sleep(0.05)
if not self.is_ipc_alive:
if unknownargs:
for arg in unknownargs:
if os.path.isdir(arg):
message = f"FILE|{arg}"
event_system.send_ipc_message(message)
self.send_ipc_message(message)
raise Exception("IPC Server Exists: Will send data to it and close...")
raise App_Launch_Exception(f"IPC Server Exists: Will send path(s) to it and close...\nNote: If no fm exists, remove /tmp/{app_name}-ipc.sock")
settings = Settings()
@ -35,7 +42,7 @@ class Application(EventSystem):
controller = Controller(settings, args, unknownargs)
if not controller:
raise Exception("Controller exited and doesn't exist...")
raise Controller_Start_Exceptio("Controller exited and doesn't exist...")
# Gets the methods from the classes and sets to handler.
# Then, builder from settings will connect to any signals it needs.

17
src/context/controller.py
View File

@ -1,5 +1,5 @@
# Python imports
import threading, subprocess, time
import subprocess, time
# Gtk imports
@ -13,19 +13,6 @@ from .mixins.dummy_mixin import DummyMixin
from .controller_data import Controller_Data
# 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: Insure threads 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 Controller(DummyMixin, Controller_Data):
@ -36,8 +23,6 @@ class Controller(DummyMixin, Controller_Data):
def tear_down(self, widget=None, eve=None):
event_system.send_ipc_message("close server")
time.sleep(event_sleep_time)
Gtk.main_quit()

1
src/context/mixins/dummy_mixin.py
View File

@ -1,4 +1,5 @@
class DummyMixin:
""" DummyMixin is an example of how mixins are used and structured in a project. """
def print_hello_world(self) -> None:
print("Hello, World!")

63
src/plugins/manifest.py Normal file
View File

@ -0,0 +1,63 @@
# Python imports
import os, json
from os.path import join
# Lib imports
# Application imports
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
class ManifestProcessor:
def __init__(self, path, builder):
manifest = join(path, "manifest.json")
if not os.path.exists(manifest):
raise Exception("Invalid Plugin Structure: Plugin doesn't have 'manifest.json'. Aboarting load...")
self._path = path
self._builder = builder
with open(manifest) as f:
data = json.load(f)
self._manifest = data["manifest"]
self._plugin = self.collect_info()
def collect_info(self) -> Plugin:
plugin = Plugin()
plugin.path = self._path
plugin.name = self._manifest["name"]
plugin.author = self._manifest["author"]
plugin.version = self._manifest["version"]
plugin.support = self._manifest["support"]
plugin.requests = self._manifest["requests"]
return plugin
def get_loading_data(self):
loading_data = {}
requests = self._plugin.requests
keys = requests.keys()
if "pass_fm_events" in keys:
if requests["pass_fm_events"] in ["true"]:
loading_data["pass_fm_events"] = True
if "bind_keys" in keys:
if isinstance(requests["bind_keys"], list):
loading_data["bind_keys"] = requests["bind_keys"]
return self._plugin, loading_data

66
src/plugins/plugins.py
View File

@ -8,14 +8,13 @@ gi.require_version('Gtk', '3.0')
from gi.repository import Gtk, Gio
# Application imports
from .manifest import Plugin, ManifestProcessor
class Plugin:
name: str = None
author: str = None
version: str = None
module: str = None
reference: type = None
class InvalidPluginException(Exception):
...
class Plugins:
@ -25,6 +24,8 @@ class Plugins:
self._settings = settings
self._builder = self._settings.get_builder()
self._plugins_path = self._settings.get_plugins_path()
self._keybindings = self._settings.get_keybindings()
self._plugins_dir_watcher = None
self._plugin_collection = []
@ -48,33 +49,48 @@ class Plugins:
print(f"Loading plugins...")
parent_path = os.getcwd()
for file in os.listdir(self._plugins_path):
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:
path = join(self._plugins_path, file)
if isdir(path):
os.chdir(path)
target = join(path, "plugin.py")
manifest = ManifestProcessor(path, self._builder)
sys.path.insert(0, path)
spec = importlib.util.spec_from_file_location(file, join(path, "__main__.py"))
app = importlib.util.module_from_spec(spec)
spec.loader.exec_module(app)
if not os.path.exists(target):
raise InvalidPluginException("Invalid Plugin Structure: Plugin doesn't have 'plugin.py'. Aboarting load...")
plugin_reference = app.Plugin(self._builder, event_system)
plugin = Plugin()
plugin.name = plugin_reference.get_plugin_name()
plugin.author = plugin_reference.get_plugin_author()
plugin.version = plugin_reference.get_plugin_version()
plugin.module = path
plugin.reference = plugin_reference
self._plugin_collection.append(plugin)
plugin, loading_data = manifest.get_loading_data()
module = self.load_plugin_module(path, folder, target)
self.execute_plugin(module, plugin, loading_data)
except Exception as e:
print("Malformed plugin! Not loading!")
print(f"Malformed Plugin: Not loading -->: '{folder}' !")
traceback.print_exc()
os.chdir(parent_path)
def load_plugin_module(self, path, folder, target):
os.chdir(path)
sys.path.insert(0, path) # NOTE: I think I'm not using this correctly...
# The folder and target aren't working to create parent package references, so using as stopgap.
# The above is probably polutling import logic and will cause unforseen import issues.
spec = importlib.util.spec_from_file_location(folder, target)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def execute_plugin(self, module: type, plugin: Plugin, loading_data: []):
plugin.reference = module.Plugin()
keys = loading_data.keys()
if "pass_fm_events" in keys:
plugin.reference.set_fm_event_system(event_system)
if "bind_keys" in keys:
self._keybindings.append_bindings( loading_data["bind_keys"] )
plugin.reference.run()
self._plugin_collection.append(plugin)
def reload_plugins(self, file: str = None) -> None:
print(f"Reloading plugins... stub.")

59
src/utils/event_system.py Normal file
View File

@ -0,0 +1,59 @@
# Python imports
# Lib imports
# Application imports
class EventSystem:
""" Inheret IPCServerMixin. Create an pub/sub systems. """
def __init__(self):
# NOTE: The format used is list of ['who', target, (data,)] Where:
# who is the sender or target ID and is used for context and control flow,
# method_target is the method to call,
# data is the method parameters OR message data to give
# Where data may be any kind of data
self._gui_events = []
self._module_events = []
# Makeshift "events" system FIFO
def _pop_gui_event(self) -> None:
if len(self._gui_events) > 0:
return self._gui_events.pop(0)
return None
def _pop_module_event(self) -> None:
if len(self._module_events) > 0:
return self._module_events.pop(0)
return None
def push_gui_event(self, event: list) -> None:
if len(event) == 3:
self._gui_events.append(event)
return None
raise Exception("Invald event format! Please do: ['sender_id': str, method_target: method, (data,): any]")
def push_module_event(self, event: list) -> None:
if len(event) == 3:
self._module_events.append(event)
return None
raise Exception("Invald event format! Please do: ['target_id': str, method_target: method, (data,): any]")
def read_gui_event(self) -> list:
return self._gui_events[0] if self._gui_events else None
def read_module_event(self) -> list:
return self._module_events[0] if self._module_events else None
def consume_gui_event(self) -> list:
return self._pop_gui_event()
def consume_module_event(self) -> list:
return self._pop_module_event()

12
src/utils/ipc_server.py
View File

@ -7,12 +7,6 @@ from multiprocessing.connection import Listener, Client
# Application imports
def threaded(fn):
def wrapper(*args, **kwargs):
threading.Thread(target=fn, args=args, kwargs=kwargs, daemon=True).start()
return wrapper
class IPCServer:
@ -22,11 +16,11 @@ class IPCServer:
self._ipc_port = 4848
self._ipc_address = ipc_address
self._conn_type = conn_type
self._ipc_authkey = b'app-ipc'
self._ipc_authkey = b'' + bytes(f'{app_name}-ipc', 'utf-8')
self._ipc_timeout = 15.0
if conn_type == "socket":
self._ipc_address = '/tmp/app-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":
@ -36,7 +30,7 @@ class IPCServer:
self._ipc_authkey = None
@threaded
@daemon_threaded
def create_ipc_listener(self) -> None:
if self._conn_type == "socket":
if os.path.exists(self._ipc_address):