Made match mostly new template patterns

This commit is contained in:
itdominator 2024-03-15 20:20:52 -05:00
parent b6d2e9ec54
commit 4cbdf1432d
16 changed files with 143 additions and 112 deletions

View File

@ -5,11 +5,11 @@ import threading
# Lib imports
# Application imports
from utils.event_system import EventSystem
from utils.endpoint_registry import EndpointRegistry
from utils.keybindings import Keybindings
from utils.logger import Logger
from utils.settings import Settings
from libs.event_system import EventSystem
from libs.endpoint_registry import EndpointRegistry
from libs.keybindings import Keybindings
from libs.logger import Logger
from libs.settings import Settings
@ -35,7 +35,7 @@ def sizeof_fmt_def(num, suffix="B"):
# NOTE: Just reminding myself we can add to builtins two different ways...
# __builtins__.update({"event_system": Builtins()})
builtins.app_name = "Mirage2"
builtins.APP_NAME = "Mirage2"
builtins.keybindings = Keybindings()
builtins.event_system = EventSystem()
builtins.endpoint_registry = EndpointRegistry()
@ -47,4 +47,4 @@ builtins.logger = Logger(settings.get_home_config_path(), \
builtins.threaded = threaded_wrapper
builtins.daemon_threaded = daemon_threaded_wrapper
builtins.sizeof_fmt = sizeof_fmt_def
builtins.event_sleep_time = 0.05
builtins.event_sleep_time = 0.05

View File

@ -10,9 +10,6 @@ import tracemalloc
tracemalloc.start()
# Lib imports
import gi
gi.require_version('Gtk', '3.0')
from gi.repository import Gtk
# Application imports
from __builtins__ import *
@ -20,33 +17,37 @@ from app import Application
def main(args, unknownargs):
setproctitle(f'{APP_NAME}')
if args.debug == "true":
settings_manager.set_debug(True)
if args.trace_debug == "true":
settings_manager.set_trace_debug(True)
settings.do_dirty_start_check()
Application(args, unknownargs)
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("--file", "-f", default="default", help="JUST SOME FILE ARG.")
# Read arguments (If any...)
args, unknownargs = parser.parse_known_args()
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("--no-plugins", "-np", default="false", help="Do not load plugins.")
parser.add_argument("--file", "-f", default=None, help="Pass an Image file directly.")
# 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(args, unknownargs)
except Exception as e:
traceback.print_exc()
quit()
quit()

View File

@ -5,37 +5,54 @@ import os
# Lib imports
# Application imports
from utils.debugging import debug_signal_handler
from utils.ipc_server import IPCServer
from libs.debugging import debug_signal_handler
from libs.ipc_server import IPCServer
from core.window import Window
class AppLaunchException(Exception):
...
class Application(IPCServer):
class Application:
""" docstring for Application. """
def __init__(self, args, unknownargs):
super(Application, self).__init__()
if not settings.is_trace_debug():
try:
self.create_ipc_listener()
except Exception:
...
self.load_ipc(args, unknownargs)
if not self.is_ipc_alive:
collection = unknownargs + [args.file] if args.file and os.path.isfile(args.file) else unknownargs
for arg in collection:
path = arg.replace("file://", "")
if os.path.isfile(path):
message = f"FILE|{path}"
self.send_ipc_message(message)
self.setup_debug_hook()
Window(args, unknownargs).main()
raise AppLaunchException(f"{app_name} IPC Server Exists: Will send path(s) to it and close...")
def load_ipc(self, args, unknownargs):
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()
try:
ipc_server.create_ipc_listener()
except Exception as e:
...
def setup_debug_hook(self):
try:
# kill -SIGUSR2 <pid> from Linux/Unix or SIGBREAK signal from Windows
signal.signal(
@ -44,6 +61,4 @@ class Application(IPCServer):
)
except ValueError:
# Typically: ValueError: signal only works in main thread
...
Window(args, unknownargs)
...

View File

@ -42,7 +42,7 @@ class Window(Gtk.ApplicationWindow):
def _setup_styling(self):
self.set_title(f"{app_name}")
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
@ -98,3 +98,6 @@ class Window(Gtk.ApplicationWindow):
settings.clear_pid()
time.sleep(event_sleep_time)
Gtk.main_quit()
def main(self):
Gtk.main()

View File

@ -13,17 +13,17 @@ from .singleton import Singleton
class IPCServer(Singleton):
""" Create a listener so that other {app_name} instances send requests back to existing instance. """
""" 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_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":
@ -111,4 +111,4 @@ class IPCServer(Singleton):
except ConnectionRefusedError as e:
print("Connection refused...")
except Exception as e:
print(repr(e))
print(repr(e))

View File

@ -21,8 +21,8 @@ class Settings(StartCheckMixin, Singleton):
def __init__(self):
self._SCRIPT_PTH = os.path.dirname(os.path.realpath(__file__))
self._USER_HOME = os.path.expanduser('~')
self._HOME_CONFIG_PATH = f"{self._USER_HOME}/.config/{app_name.lower()}"
self._USR_PATH = f"/usr/share/{app_name.lower()}"
self._HOME_CONFIG_PATH = f"{self._USER_HOME}/.config/{APP_NAME.lower()}"
self._USR_PATH = f"/usr/share/{APP_NAME.lower()}"
self._USR_CONFIG_FILE = f"{self._USR_PATH}/settings.json"
self._PLUGINS_PATH = f"{self._HOME_CONFIG_PATH}/plugins"
@ -31,10 +31,10 @@ class Settings(StartCheckMixin, Singleton):
self._GLADE_FILE = f"{self._HOME_CONFIG_PATH}/Main_Window.glade"
self._CSS_FILE = f"{self._HOME_CONFIG_PATH}/stylesheet.css"
self._KEY_BINDINGS_FILE = f"{self._HOME_CONFIG_PATH}/key-bindings.json"
self._PID_FILE = f"{self._HOME_CONFIG_PATH}/{app_name.lower()}.pid"
self._PID_FILE = f"{self._HOME_CONFIG_PATH}/{APP_NAME.lower()}.pid"
self._UI_WIDEGTS_PATH = f"{self._HOME_CONFIG_PATH}/ui_widgets"
self._CONTEXT_MENU = f"{self._HOME_CONFIG_PATH}/contexct_menu.json"
self._WINDOW_ICON = f"{self._DEFAULT_ICONS}/{app_name.lower()}.png"
self._WINDOW_ICON = f"{self._DEFAULT_ICONS}/{APP_NAME.lower()}.png"
if not os.path.exists(self._HOME_CONFIG_PATH):
os.mkdir(self._HOME_CONFIG_PATH)
@ -65,7 +65,7 @@ class Settings(StartCheckMixin, Singleton):
if not os.path.exists(self._CSS_FILE):
raise MissingConfigError("Unable to find the application Stylesheet file.")
if not os.path.exists(self._WINDOW_ICON):
self._WINDOW_ICON = f"{self._USR_PATH}/icons/{app_name.lower()}.png"
self._WINDOW_ICON = f"{self._USR_PATH}/icons/{APP_NAME.lower()}.png"
if not os.path.exists(self._WINDOW_ICON):
raise MissingConfigError("Unable to find the application icon.")
if not os.path.exists(self._UI_WIDEGTS_PATH):
@ -186,4 +186,4 @@ class Settings(StartCheckMixin, Singleton):
def save_settings(self):
with open(self._CONFIG_FILE, 'w') as outfile:
json.dump(self._settings, outfile, separators=(',', ':'), indent=4)
json.dump(self._settings, outfile, separators=(',', ':'), indent=4)

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}")

View File

@ -1,51 +0,0 @@
# 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): self._clean_pid()
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)
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}")