Optimizations, QoL improvements

Author: SystemXFiles

process-governor.py

@@ -3,7 +3,7 @@
 
 import pyuac
 
-from constants.app_info import APP_NAME, APP_NAME_WITH_VERSION
+from constants.app_info import APP_NAME
 from constants.log import LOG
 from main_loop import start_app
 from util.lock_instance import create_lock_file, remove_lock_file
@@ -19,7 +19,7 @@
                    "Please run the program as an administrator to ensure proper functionality.")
 
         LOG.error(message)
-        show_error(f"Error Detected - {APP_NAME_WITH_VERSION}", message)
+        show_error(message)
         sys.exit(1)
 
     create_lock_file()

src/configuration/migration/all_migration.py

@@ -4,7 +4,6 @@
 from configuration.migration.m0_rules_to_split_rules_config import MigrationRules2SplitRulesConfig
 from configuration.migration.m1_new_fields_in_rule import NewFieldsInRule
 from configuration.migration.m2_remove_high_io_priority import RemoveHighIoPriority
-from constants.app_info import APP_NAME_WITH_VERSION
 from constants.log import LOG
 from service.config_service import ConfigService
 from util.messages import show_error
@@ -55,14 +54,12 @@ def run_all_migration():
         except Exception as e:
             has_error = True
             LOG.exception(f"[{migration_name}] Migration failed.")
-            show_error(
-                f"Error Detected - {APP_NAME_WITH_VERSION}",
-                f"Migration `{migration_name}` failed: \n{str(e)}"
-            )
+            show_error(f"Migration `{migration_name}` failed: \n{str(e)}")
             break
 
     if not has_error:
         ConfigService.save_config_raw(config)
 
+
 if __name__ == '__main__':
     run_all_migration()

src/constants/app_info.py

@@ -11,6 +11,7 @@
 
 CURRENT_TAG: Final[str] = f"v{APP_VERSION}"
 APP_NAME_WITH_VERSION: Final[str] = f"{APP_NAME} {CURRENT_TAG}"
+TITLE_ERROR: Final[str] = f"Error Detected - {APP_NAME_WITH_VERSION}"
 
 if is_portable():
     APP_PATH: Final[str] = sys._MEIPASS

src/main_loop.py

@@ -8,7 +8,7 @@
 
 from configuration.config import Config
 from configuration.migration.all_migration import run_all_migration
-from constants.app_info import APP_NAME_WITH_VERSION, APP_NAME
+from constants.app_info import APP_NAME
 from constants.files import LOG_FILE_NAME
 from constants.log import LOG
 from constants.threads import THREAD_SETTINGS, THREAD_TRAY
@@ -20,6 +20,7 @@
 from util.messages import yesno_error_box, show_error
 from util.scheduler import TaskScheduler
 from util.startup import update_startup
+from util.updates import check_updates
 
 
 def priority_setup():
@@ -96,6 +97,7 @@ def start_app():
         run_all_migration()
         update_startup()
         priority_setup()
+        check_updates(True)
 
         tray: Icon = init_tray()
         main_loop(tray)
@@ -110,26 +112,24 @@ def start_app():
 
 
 def show_rules_error_message():
-    title = f"Error Detected - {APP_NAME_WITH_VERSION}"
     message = "An error has occurred while loading or applying the rules.\n"
 
     if TaskScheduler.check_task(THREAD_SETTINGS):
         message += "Please check the correctness of the rules."
-        show_error(title, message)
+        show_error(message)
     else:
         message += f"Would you like to open the {SETTINGS_TITLE} to review and correct the rules?"
-        if yesno_error_box(title, message):
+        if yesno_error_box(message):
             open_settings()
 
 
 def show_abstract_error_message(will_closed: bool):
-    title = f"Error Detected - {APP_NAME_WITH_VERSION}"
     will_closed_text = 'The application will now close.' if will_closed else ''
     message = (
         f"An error has occurred in the {APP_NAME} application. {will_closed_text}\n"
         f"To troubleshoot, please check the log file `{LOG_FILE_NAME}` for details.\n\n"
         f"Would you like to open the log file?"
     )
 
-    if yesno_error_box(title, message):
+    if yesno_error_box(message):
         os.startfile(LOG_FILE_NAME)

src/model/process.py

@@ -82,6 +82,8 @@ class Process(BaseModel):
         exclude=True
     )
 
+    is_new: bool = Field(exclude=True)
+
     def __hash__(self):
         return hash((self.pid, self.bin_path, self.process_name, self.cmd_line))
 

src/service/processes_info_service.py

@@ -15,41 +15,46 @@ class ProcessesInfoService(ABC):
     It is an abstract base class (ABC) to be subclassed by specific implementation classes.
     """
 
-    __prev_pids: set[int] = []
+    _cache: dict[int, Process] = {}
 
     @classmethod
-    def get_processes(
-            cls,
-            only_new: bool,
-            force_return_pids: set[int] = None
-    ) -> dict[int, Process]:
+    def get_processes(cls) -> dict[int, Process]:
         """
-        Retrieves a dictionary of running processes and their information.
+        Returns a dictionary with information about running processes.
 
-        Args:
-            only_new (bool): If True, returns only processes that have started since the last call.
-                             If False, returns all currently running processes.
-            force_return_pids (set[int], optional): A set of process IDs that should be included in the result
-                                                    even if they are not new. Defaults to an empty set if not provided.
         Returns:
-            dict[int, Process]: A dictionary where keys are process IDs and values are Process objects,
-                                containing detailed information about each running process.
+            dict[int, Process]: A dictionary with information about running processes.
         """
-        services = ServicesInfoService.get_services()
-        result: dict[int, Process] = {}
-        current_pids: list[int] = psutil.pids()
-        force_return_pids = force_return_pids or set()
 
-        for pid in current_pids:
-            if only_new and pid in cls.__prev_pids and pid not in force_return_pids:
-                continue
+        cache = cls._cache
+        services = ServicesInfoService.get_services()
+        pids = set(psutil.pids())
 
+        for pid in pids:
             try:
-                process = psutil.Process(pid)
+                process_info = psutil.Process(pid)
+                info = process_info.as_dict(attrs=[
+                    'exe',
+                    'nice', 'ionice', 'cpu_affinity'
+                ])
+
+                if pid in cache:
+                    process = cache[pid]
+
+                    if process.bin_path == info['exe']:
+                        process.priority = none_int(info['nice'])
+                        process.io_priority = none_int(info['ionice'])
+                        process.affinity = info['cpu_affinity']
+                        process.is_new = False
+                        continue
+
                 service = services.get(pid)
+                info = process_info.as_dict(attrs=[
+                    'name', 'exe', 'cmdline',
+                    'nice', 'ionice', 'cpu_affinity'
+                ])
 
-                info = process.as_dict(attrs=['name', 'exe', 'nice', 'ionice', 'cpu_affinity', 'cmdline'])
-                result[pid] = Process.model_construct(
+                cache[pid] = Process.model_construct(
                     pid=pid,
                     process_name=info['name'],
                     service_name=getattr(service, 'name', None),
@@ -58,16 +63,19 @@ def get_processes(
                     priority=none_int(info['nice']),
                     io_priority=none_int(info['ionice']),
                     affinity=info['cpu_affinity'],
-                    process=process,
-                    service=service
+                    process=process_info,
+                    service=service,
+                    is_new=True
                 )
             except NoSuchProcess:
                 pass
 
-        if only_new:
-            cls.__prev_pids = set(current_pids)
+        deleted_pids = cache.keys() - pids
+
+        for pid in deleted_pids:
+            del cache[pid]
 
-        return result
+        return cache.copy()
 
     @staticmethod
     def _get_command_line(pid, info):

src/service/rules_service.py

@@ -29,7 +29,6 @@ class RulesService(ABC):
 
     __ignore_pids: set[int] = {0, os.getpid()}
     __ignored_process_parameters: dict[Process, set[ProcessParameter]] = {}
-    __force_pids: set[int] = set()
 
     @classmethod
     def apply_rules(cls, config: Config, only_new: bool):
@@ -47,15 +46,14 @@ def apply_rules(cls, config: Config, only_new: bool):
             return
 
         cls.__light_gc_ignored_process_parameters()
-        cls.__force_pids = cls.__handle_processes(
+        cls.__handle_processes(
             config,
-            ProcessesInfoService.get_processes(only_new, cls.__force_pids)
+            ProcessesInfoService.get_processes(),
+            only_new
         )
 
     @classmethod
-    def __handle_processes(cls, config: Config, processes: dict[int, Process]) -> set[int]:
-        force_pids: set[int] = set()
-
+    def __handle_processes(cls, config: Config, processes: dict[int, Process], only_new: bool):
         for pid, process in processes.items():
             if pid in cls.__ignore_pids:
                 continue
@@ -65,16 +63,14 @@ def __handle_processes(cls, config: Config, processes: dict[int, Process]) -> se
             if not rule:
                 continue
 
-            if rule.force == BoolStr.YES:
-                force_pids.add(pid)
+            if rule.force == BoolStr.NO and only_new and not process.is_new:
+                continue
 
             if rule.delay > 0:
                 TaskScheduler.schedule_task(process, cls.__handle_process, rule.delay, process, rule)
             else:
                 cls.__handle_process(process, rule)
 
-        return force_pids
-
     @classmethod
     def __handle_process(cls, process: Process, rule: ProcessRule | ServiceRule):
         parameter_methods: dict[ProcessParameter, tuple[Callable[[Process, ProcessRule | ServiceRule], bool], str]] = {

src/ui/settings.py

@@ -3,7 +3,7 @@
 from typing import Optional
 
 from configuration.migration.all_migration import run_all_migration
-from constants.app_info import APP_NAME_WITH_VERSION, APP_NAME
+from constants.app_info import APP_NAME_WITH_VERSION, APP_NAME, TITLE_ERROR
 from constants.files import LOG_FILE_NAME
 from constants.log import LOG
 from constants.resources import APP_ICON, UI_SAVE, UI_LOG, UI_CONFIG
@@ -41,7 +41,7 @@ def __init__(self):
     def _setup_window(self):
         self._center_window()
 
-        self.protocol("WM_DELETE_WINDOW", self._on_window_closing)
+        self.protocol("WM_DELETE_WINDOW", self.close)
         self.iconbitmap(APP_ICON)
         self.title(APP_NAME_WITH_VERSION)
         self.minsize(*RC_WIN_SIZE)
@@ -145,38 +145,45 @@ def _save(self):
         self._update_actions_state()
         return result
 
-    def _on_window_closing(self):
+    def close(self):
         has_error = self._tabs.has_error()
 
         if self._tabs.has_unsaved_changes():
             if has_error:
+                self.to_front()
+
                 message = ("There are errors in the rules, and they can't be saved. "
                            "Do you want to DISCARD them and exit?")
-                result = messagebox.askyesno(f"{APP_NAME_WITH_VERSION}", message)
+                result = messagebox.askyesno(TITLE_ERROR, message)
 
                 if not result:
-                    return
+                    return False
             else:
+                self.to_front()
+
                 message = ("There are unsaved changes. "
                            "Do you want to save them before exiting?")
-                result = messagebox.askyesnocancel(f"{APP_NAME_WITH_VERSION}", message)
+                result = messagebox.askyesnocancel(APP_NAME_WITH_VERSION, message)
 
                 if result is None:
-                    return
+                    return False
 
                 if result and not self._save():
-                    return
+                    return False
         else:
             if has_error:
+                self.to_front()
+
                 message = (
                     f"There are errors in the rules, and they must be corrected before the application can work properly. "
                     f"Do you still want to close the {SETTINGS_TITLE}?")
-                result = messagebox.askyesno(f"{APP_NAME_WITH_VERSION}", message)
+                result = messagebox.askyesno(APP_NAME_WITH_VERSION, message)
 
                 if not result:
-                    return
+                    return False
 
         self.destroy()
+        return True
 
     def _update_actions_state(self, _=None):
         tabs = self._tabs
@@ -331,6 +338,12 @@ def _search_focus(self):
 
         tab.actions.search.focus_set()
 
+    def to_front(self):
+        self.deiconify()
+        self.lift()
+        self.attributes('-topmost', True)
+        self.after_idle(self.attributes, '-topmost', False)
+
 
 class SettingsActions(ttk.Frame):
     def __init__(self, *args, **kwargs):
@@ -367,27 +380,50 @@ def _setup_btn(self):
         save.pack(**RIGHT_PACK)
 
 
+__app: Optional[Settings] = None
+
+
 def open_settings():
+    global __app
+
+    if __app is not None:
+        __app.after_idle(__app.to_front)
+        return
+
     def settings():
         try:
-            app = Settings()
-            app.mainloop()
+            global __app
+
+            try:
+                __app = Settings()
+                __app.mainloop()
+            finally:
+                __app = None
         except:
             LOG.exception(f"An unexpected error occurred in the {SETTINGS_TITLE} of {APP_NAME}.")
             show_settings_error_message()
 
     TaskScheduler.schedule_task(THREAD_SETTINGS, settings)
 
 
+def is_opened_settings() -> bool:
+    global __app
+    return __app is not None
+
+
+def get_settings() -> Settings:
+    global __app
+    return __app
+
+
 def show_settings_error_message():
-    title = f"Error Detected - {APP_NAME_WITH_VERSION}"
     message = (
         f"An error has occurred in the {SETTINGS_TITLE} of {APP_NAME}.\n"
         f"To troubleshoot, please check the log file `{LOG_FILE_NAME}` for details.\n\n"
         f"Would you like to open the log file?"
     )
 
-    if yesno_error_box(title, message):
+    if yesno_error_box(message):
         os.startfile(LOG_FILE_NAME)