Use dbus methods for konsole instead of console calls

Co-authored-by: heddxh <[email protected]>

yin_yang/plugins/konsole.py

@@ -1,23 +1,20 @@
 import logging
 import os
 import re
-import subprocess
 from configparser import ConfigParser
 from itertools import chain
 from pathlib import Path
 from shutil import copyfile
 
-import psutil
-from PySide6.QtDBus import QDBusConnection, QDBusMessage
+from PySide6.QtDBus import QDBusMessage
 
 from yin_yang import helpers
-
-from ._plugin import Plugin
+from ._plugin import DBusPlugin
 
 logger = logging.getLogger(__name__)
 
 
-class Konsole(Plugin):
+class Konsole(DBusPlugin):
     '''
     Themes are profiles. To use a color scheme,
     create a new profile or edit one to use the desired color scheme.
@@ -26,6 +23,8 @@ class Konsole(Plugin):
 
     global_path = Path(helpers.get_usr() + 'share/konsole')
     config_path = Path.home() / '.config/konsolerc'
+    # apps using konsole
+    apps_have_konsole = ["org.kde.konsole", "org.kde.yakuake", "org.kde.dolphin", "org.kde.kate"]
 
     @property
     def user_path(self) -> Path:
@@ -64,38 +63,13 @@ def set_mode(self, dark: bool) -> bool:
         # update default profile, if application is started afterward
         self.default_profile = profile + '.profile'
 
-        # Set Konsole profile for all sessions
-
-        # Get the process IDs of all running Konsole instances owned by the current user
-        process_ids = [
-            proc.pid
-            for proc in psutil.process_iter()
-            if proc.name() == 'konsole' and proc.username() == os.getlogin()
-        ]
-
-        # loop: console processes
-        for proc_id in process_ids:
-            logger.debug(f'Changing profile in konsole session {proc_id}')
-            set_profile(f'org.kde.konsole-{proc_id}', profile)
-            set_profile(f'org.kde.konsole-{proc_id}', profile, set_default_profile=True)
-
-        # konsole may don't have session dbus like above
-        set_profile('org.kde.konsole', profile)
-        set_profile('org.kde.yakuake', profile)
-        set_profile('org.kde.konsole', profile, set_default_profile=True)
-        set_profile('org.kde.yakuake', profile, set_default_profile=True)
-
-        process_ids = [
-            proc.pid
-            for proc in psutil.process_iter()
-            if proc.name() == 'dolphin' and proc.username() == os.getlogin()
-        ]
-
-        # loop: dolphin processes
-        for proc_id in process_ids:
-            logger.debug(f'Changing profile in dolphin session {proc_id}')
-            set_profile(f'org.kde.dolphin-{proc_id}', profile)
-            set_profile(f'org.kde.dolphin-{proc_id}', profile, set_default_profile=True)
+        # Find available konsole sessions, including dolphin, yakuake, kate, etc
+        services = [str(n) for n in self.connection.interface().registeredServiceNames().value()
+                    if n.split('-')[0] in self.apps_have_konsole]
+        for service in services:
+            logger.debug(f'Changing profile in konsole session {service}')
+            self.set_profile(service, profile=profile)
+            self.set_profile(service, profile, set_default_profile=True)
 
         return True
 
@@ -184,6 +158,7 @@ def default_profile(self):
     def default_profile(self, value: str):
         assert value.endswith('.profile')
 
+        match: re.Match[str] | None = None
         with self.config_path.open('r') as file:
             lines = file.readlines()
             for i, line in enumerate(lines):
@@ -195,10 +170,12 @@ def default_profile(self, value: str):
                     logger.debug(f'Changing default profile to {value}')
                     lines[i] = f'DefaultProfile={value}\n'
                     break
-                else:
-                    logger.debug('No default profile found')
-        with self.config_path.open('w') as file:
-            file.writelines(lines)
+
+        if match is None:
+            logger.error(f'No DefaultProfile field found in {self.config_path}')
+        else:
+            with self.config_path.open('w') as file:
+                file.writelines(lines)
 
     def update_profile(self, dark: bool, theme: str):
         if not self.available or theme == '':
@@ -253,42 +230,23 @@ def create_profiles(self):
             profile_config.write(file)
 
 
-def set_profile(service: str, profile: str, set_default_profile: bool = False):
-    # connect to the session bus
-    connection = QDBusConnection.sessionBus()
-    if set_default_profile:
-        path = 'Sessions/'
-        interface = 'org.kde.konsole.Session'
-        method = 'setProfile'
-    else:
-        path = 'Windows/'
-        interface = 'org.kde.konsole.Window'
-        method = 'setDefaultProfile'
-
-    # maybe it's possible with pyside6 dbus packages, but this was simpler and worked
-    try:
-        sessions = helpers.check_output(f'qdbus {service} | grep "{path}"')
-    except subprocess.CalledProcessError:
-        try:
-            sessions = helpers.check_output(
-                f'qdbus org.kde.konsole | grep "{path}"', shell=True
-            )
-            logger.debug(f'Found org.kde.konsole, use that instead')
-            service = 'org.kde.konsole'
-        except subprocess.CalledProcessError:
-            # happens when dolphins konsole is not opened
+    def set_profile(self, service: str, profile: str, set_default_profile: bool = False):
+        if set_default_profile:
+            path = 'Sessions/'
+            interface = 'org.kde.konsole.Session'
+            method = 'setProfile'
+        else:
+            path = 'Windows/'
+            interface = 'org.kde.konsole.Window'
+            method = 'setDefaultProfile'
+
+        sessions = self.list_paths(service, path)
+
+        for session in sessions:
             logger.debug(
-                f'No Konsole sessions available in service {service}, skipping'
+                f'Changing {"default" if set_default_profile else ""} profile of session {session} to {profile}'
             )
-            return
-    sessions = sessions.decode('utf-8').removesuffix('\n').split('\n')
-
-    # loop: process sessions
-    for session in sessions:
-        logger.debug(
-            f'Changing {"default" if set_default_profile else ""} profile of session {session} to {profile}'
-        )
-        # set profile
-        message = QDBusMessage.createMethodCall(service, session, interface, method)
-        message.setArguments([profile])
-        connection.call(message)
+            # set profile
+            message = QDBusMessage.createMethodCall(service, session, interface, method)
+            message.setArguments([profile])
+            self.connection.call(message)