Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,9 +37,12 @@
import globals as gl

if not gl.IS_MAC:
# dbus-python is used for blocking method calls only, so its GLib main loop
# is deliberately never installed. Attaching a connection to the main loop
# makes dbus-python rebuild that connection's GSources from whichever thread
# sends a message, which corrupts them once anything calls D-Bus off the
# main thread. Use GDBus (Gio) for signals and exported objects instead.
import dbus
import dbus.service
from dbus.mainloop.glib import DBusGMainLoop

# Import own modules
from src.app import App
Expand Down Expand Up @@ -548,7 +551,6 @@ def main():
'GSK_RENDERER=ngl to your "/etc/environment" file')

if not gl.IS_MAC:
DBusGMainLoop(set_as_default=True)
# Dbus
quit_running()

Expand Down
25 changes: 4 additions & 21 deletions src/backend/LockScreenManager/Detectors/Cinnamon.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,6 @@
# Import globals first to get IS_MAC
import globals as gl

if not gl.IS_MAC:
import dbus

from loguru import logger as log

class CinnamonLockScreenDetector(LockScreenDetector):
def __init__(self, lock_screen_manager: "LockScreenManager"):
self.lock_screen_manager: "LockScreenManager" = lock_screen_manager
Expand All @@ -40,19 +35,7 @@ def screen_saver_active_changed(self, active):
def setup_dbus(self):
if gl.IS_MAC:
return
try:
# Use the D-Bus MainLoop with glib integration
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

# Connect to the Session Bus
bus = dbus.SessionBus()

# Define the signal to listen to
bus.add_signal_receiver(
self.screen_saver_active_changed,
dbus_interface="org.cinnamon.ScreenSaver",
signal_name="ActiveChanged",
path="/org/cinnamon/ScreenSaver"
)
except Exception as e:
log.error(f"Failed to connect to D-Bus: {e}")
self.subscribe_to_screen_saver(
interface="org.cinnamon.ScreenSaver",
path="/org/cinnamon/ScreenSaver"
)
25 changes: 4 additions & 21 deletions src/backend/LockScreenManager/Detectors/Gnome.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,6 @@
# Import globals first to get IS_MAC
import globals as gl

if not gl.IS_MAC:
import dbus

from loguru import logger as log

class GnomeLockScreenDetector(LockScreenDetector):
def __init__(self, lock_screen_manager: "LockScreenManager"):
self.lock_screen_manager: "LockScreenManager" = lock_screen_manager
Expand All @@ -40,19 +35,7 @@ def screen_saver_active_changed(self, active):
def setup_dbus(self):
if gl.IS_MAC:
return
try:
# Use the D-Bus MainLoop with glib integration
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

# Connect to the Session Bus
bus = dbus.SessionBus()

# Define the signal to listen to
bus.add_signal_receiver(
self.screen_saver_active_changed,
dbus_interface="org.gnome.ScreenSaver",
signal_name="ActiveChanged",
path="/org/gnome/ScreenSaver"
)
except Exception as e:
log.error(f"Failed to connect to D-Bus: {e}")
self.subscribe_to_screen_saver(
interface="org.gnome.ScreenSaver",
path="/org/gnome/ScreenSaver"
)
26 changes: 4 additions & 22 deletions src/backend/LockScreenManager/Detectors/KDE.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,6 @@
# Import globals first to get IS_MAC
import globals as gl

if not gl.IS_MAC:
import dbus

from loguru import logger as log

class KDELockScreenDetector(LockScreenDetector):
def __init__(self, lock_screen_manager: "LockScreenManager"):
self.lock_screen_manager: "LockScreenManager" = lock_screen_manager
Expand All @@ -40,20 +35,7 @@ def screen_saver_active_changed(self, active):
def setup_dbus(self):
if gl.IS_MAC:
return
try:
# Use the D-Bus MainLoop with glib integration
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

# Connect to the Session Bus
bus = dbus.SessionBus()

# Define the signal to listen to
bus.add_signal_receiver(
self.screen_saver_active_changed,
dbus_interface="org.freedesktop.ScreenSaver",
signal_name="ActiveChanged",
path="/org/freedesktop/ScreenSaver"
)
except Exception as e:
log.error(f"Failed to connect to D-Bus: {e}")

self.subscribe_to_screen_saver(
interface="org.freedesktop.ScreenSaver",
path="/org/freedesktop/ScreenSaver"
)
40 changes: 27 additions & 13 deletions src/backend/LockScreenManager/Detectors/Logind.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@

import os

from gi.repository import Gio

# Import globals first to get IS_MAC
import globals as gl

Expand All @@ -20,35 +22,47 @@ def __init__(self, lock_screen_manager: "LockScreenManager"):

self.setup_dbus()

def on_lock(self):
def on_lock(self, *_args):
self.lock_screen_manager.lock(True)

def on_unlock(self):
def on_unlock(self, *_args):
self.lock_screen_manager.lock(False)

def setup_dbus(self):
if gl.IS_MAC:
return

# Use the D-Bus MainLoop with glib integration
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

# Connect to the System Bus
# Resolving the session path only needs blocking calls, which
# dbus-python supports without a main loop attached.
bus = dbus.SystemBus()

# Resolve the session path for the current user session
session_path = self._resolve_session_path(bus)

log.info(f"Logind lock detector active ({session_path})")

session_obj = bus.get_object("org.freedesktop.login1", session_path)
session_obj.connect_to_signal(
"Lock", self.on_lock,
dbus_interface="org.freedesktop.login1.Session"
# This runs on the LockScreenManager setup thread, and dbus-python's
# GLib main loop integration is not thread safe, so signals are
# subscribed to via GDBus (Gio) instead of connect_to_signal(), which
# would require installing that main loop.
self.bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
self.lock_subscription_id = self.bus.signal_subscribe(
"org.freedesktop.login1",
"org.freedesktop.login1.Session",
"Lock",
session_path,
None,
Gio.DBusSignalFlags.NONE,
self.on_lock
)
session_obj.connect_to_signal(
"Unlock", self.on_unlock,
dbus_interface="org.freedesktop.login1.Session"
self.unlock_subscription_id = self.bus.signal_subscribe(
"org.freedesktop.login1",
"org.freedesktop.login1.Session",
"Unlock",
session_path,
None,
Gio.DBusSignalFlags.NONE,
self.on_unlock
)

def _resolve_session_path(self, bus) -> str:
Expand Down
37 changes: 36 additions & 1 deletion src/backend/LockScreenManager/LockScreenDetector.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,45 @@
You should have received a copy of the GNU General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
from gi.repository import Gio

from loguru import logger as log

from typing import TYPE_CHECKING
if TYPE_CHECKING:
from src.backend.LockScreenManager.LockScreenManager import LockScreenManager

class LockScreenDetector:
# Subclasses assign self.lock_screen_manager without calling super().__init__()
bus: Gio.DBusConnection = None
subscription_id: int = None

def __init__(self, lock_screen_manager: "LockScreenManager"):
self.lock_screen_manager: "LockScreenManager" = lock_screen_manager
self.lock_screen_manager: "LockScreenManager" = lock_screen_manager

def subscribe_to_screen_saver(self, interface: str, path: str) -> None:
"""
Watch a screen saver's ActiveChanged signal.

GDBus is used rather than dbus-python because this runs on the
LockScreenManager setup thread, and dbus-python's GLib main loop
integration is not thread safe. The signal is delivered on the main
context either way.
"""
try:
self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
self.subscription_id = self.bus.signal_subscribe(
None,
interface,
"ActiveChanged",
path,
None,
Gio.DBusSignalFlags.NONE,
self.__on_active_changed
)
except Exception as e:
log.error(f"Failed to connect to D-Bus: {e}")

def __on_active_changed(self, connection, sender_name: str, object_path: str,
interface_name: str, signal_name: str, parameters) -> None:
self.screen_saver_active_changed(parameters.unpack()[0])
38 changes: 25 additions & 13 deletions src/backend/WindowGrabber/Integrations/Gnome.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,22 +27,22 @@
import globals as gl

import gi

if not gl.IS_MAC:
import dbus
from gi.repository import Gio

# Import typing
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from src.backend.WindowGrabber.WindowGrabber import WindowGrabber

class Gnome(Integration):
DBUS_NAME = "org.gnome.Shell"
DBUS_PATH = "/org/gnome/Shell/Extensions/StreamController"
DBUS_INTERFACE = "org.gnome.Shell.Extensions.StreamController"

def __init__(self, window_grabber: "WindowGrabber"):
super().__init__(window_grabber=window_grabber)

self.bus = None
self.proxy = None
self.interface = None
if not gl.IS_MAC:
self.connect_dbus()

Expand All @@ -60,14 +60,26 @@ def connect_dbus(self) -> None:
if gl.IS_MAC:
return
try:
self.bus = dbus.SessionBus()
self.proxy = self.bus.get_object("org.gnome.Shell", "/org/gnome/Shell/Extensions/StreamController")
self.interface = dbus.Interface(self.proxy, "org.gnome.Shell.Extensions.StreamController")
self.interface.connect_to_signal("FocusedWindowChanged", self.on_window_changed)
self.proxy = Gio.DBusProxy.new_for_bus_sync(
Gio.BusType.SESSION,
Gio.DBusProxyFlags.DO_NOT_LOAD_PROPERTIES | Gio.DBusProxyFlags.DO_NOT_AUTO_START,
None,
self.DBUS_NAME,
self.DBUS_PATH,
self.DBUS_INTERFACE,
None
)
self.proxy.connect("g-signal", self.on_g_signal)
except Exception as e:
log.error(f"Failed to connect to D-Bus: {e}")
pass

def on_g_signal(self, proxy, sender_name: str, signal_name: str, parameters) -> None:
if signal_name == "FocusedWindowChanged":
self.on_window_changed(parameters.unpack()[0])

def call_method(self, method_name: str) -> str:
result = self.proxy.call_sync(method_name, None, Gio.DBusCallFlags.NONE, -1, None)
return result.unpack()[0]

def on_window_changed(self, answer: str) -> None:
answer = json.loads(answer)
Expand All @@ -79,7 +91,7 @@ def get_all_windows(self) -> list[Window]:
return []

try:
answer = json.loads(self.interface.GetAllWindows())
answer = json.loads(self.call_method("GetAllWindows"))
except:
return []
windows: list[Window] = []
Expand All @@ -95,12 +107,12 @@ def get_active_window (self) -> Window:
if not self.get_is_connected():
return None
try:
answer = json.loads(self.interface.GetFocusedWindow())
answer = json.loads(self.call_method("GetFocusedWindow"))
except:
return None
wm_class = answer.get("wm_class")
title = answer.get("title")
return Window(wm_class, title)

def get_is_connected(self) -> bool:
return None not in (self.bus, self.proxy, self.interface)
return self.proxy is not None and self.proxy.get_name_owner() is not None
Empty file added tests/__init__.py
Empty file.
Loading