Listing open windows on wlroots based desktops with pywayland.

[sonakinci41]

Hello, I want to write a panel with pygtk for wlroots based desktops. However, in my research, I could not see a way to list open windows. I thought I could do this with pywayland by reviewing the Waybar and logging the error on the link. I decided to open an error log as I saw a lack of documentation on this subject. I'm sorry if I'm doing something wrong. Can you give a simple example that allows you to list and manipulate existing windows? Thanks.

[nlpsuge]

Similar question here.

Can I get an opening windows list from Wayland via this projects?

[m-col]

The generic wlroots answer on how to do this is to use the wlr_foreign_toplevel_management_v1 protocol to get information about exising windows from the compositor. The compositor must support this protocol itself, but this is common on wlroots-based compositors so likely not an issue. This is the protocol that most third-party panels (and similar tools) use to get information on windows, and it also allows them to control them (e.g. focus a specific window when you click on it in the panel).

For pywlroots, handlers for this protocol are not yet part of the codebase, but I just submitted a PR for it here so this should be usable soon: flacjacket/pywlroots#63

[IamRezaMousavi]

How to use this?

[m-col]

I suggest checking out the source for wlrctl which can be used to list windows: https://git.sr.ht/~brocellous/wlrctl

[RedBearAK]

@flacjacket @m-col

To anyone reading this, please help me out if you can.

I have been attempting to use the foreign toplevel management protocol with both pywayland and pywlroots for a very long time, and have failed at every turn.

With pywayland and a generated set of protocol files that includes the protocol in question, I hit a NotImplementedError in protocol_core/message.py, because the protocol appears to use ArgumentType.NewId in the event() method decorator, and the c_to_arguments method in message.py just has a "TODO" comment and raises that error. So there's no code to handle the argument. I don't know what this means or how to fix it or get around it.

pywayland/pywayland/protocol_core/message.py

Lines 148 to 150 in 4febe9b

	elif argument.argument_type == ArgumentType.NewId:
	# TODO
	raise NotImplementedError

With pywlroots, which seems to have the protocol available to import without taking extra steps (the only example I can find of this in actual use is libqtile), I end up with a segmentation fault and core dump. Maybe I'm missing a setup step?

I'm just installing the latest pywayland and pywlroots with pip. But with pywayland I use the scanner technique from this comment:

#8 (comment)

These problems I'm running into happen either when I do an initial display.roundtrip() right after establishing a handler callback, or skip the roundtrip and go directly into a display.dispatch() loop.

I've used every scrap of the very limited information I've been able to find online about how to use either of these techniques, and have worked through a lot of initial confusion about how to use any of this stuff, but these problems are like brick walls that I don't know how to overcome.

Using other more standard protocols to get some info out of the Wayland compositor doesn't seem to be a problem. So I don't think I'm using pywayland wrong overall.

I've been testing in a VM running Fedora 40, with sway and Hyprland. The results have been the same in either environment. The registry dump shows the zwlr_foreign_toplevel_manager_v1 protocol in the list for both compositors.

Any clues to getting this working would be super helpful. I'm trying to give a keymapper the ability to do app-specific modmapping/keymapping on wlroots compositors.

The keymapper already has methods that work in several common Wayland environments that were not based on wlroots (GNOME, KDE Plasma, Cinnamon) and in some environments like sway and Hyprland where there were alternative methods available (i3ipc and hyprpy) to get app_id and title of the active window. But I'd really, really like to get wlroots in general working, which should allow the keymapper to (hopefully) support some coming updates to popular Linux desktops like the new Pop!_OS COSMIC, Budgie, LXQt 2.x, etc. Most new Wayland environments seem to be using wlroots for convenience, and should all have this foreign toplevel management protocol available. Unless I'm completely misunderstanding the situation.

I'll put the current state of the script below, in an expandable tag, in case anyone has some insights to offer. The script has the wlroots import/method uncommented currently, just because it was the last thing I tried (again). The pywayland imports and methods are commented out. The script works in general, as long as I don't try to establish the foreign toplevel manager object through bind() or create(). Information comes out for the registry, wl_seat and wl_output without any apparent issue.

Script snapshot:

#!/usr/bin/env python3


# Reference for creating the protocols with pywayland scanner:
# https://github.com/flacjacket/pywayland/issues/8#issuecomment-987040284

# Protocol documentation:
# https://wayland.app/protocols/wlr-foreign-toplevel-management-unstable-v1

# pywlroots method causes a segmentation fault
# pywayland method has a NotImplementedError
# ForeignToplevelManagerV1/ZwlrForeignToplevelManagerV1 is unusable? 

import sys
import signal
import traceback


from wlroots.wlr_types.foreign_toplevel_management_v1 import (
    ForeignToplevelManagerV1 as ZwlrForeignToplevelManagerV1,
    ForeignToplevelHandleV1 as ZwlrForeignToplevelHandleV1,
)

# from protocols.wlr_foreign_toplevel_management_unstable_v1 import (
#     ZwlrForeignToplevelManagerV1,
#     ZwlrForeignToplevelHandleV1
# )

from pywayland.client import Display
from protocols.wayland import WlOutput, WlSeat
from typing import Optional


class WaylandClient:
    def __init__(self):
        """Initialize the WaylandClient."""
        signal.signal(signal.SIGINT, self.signal_handler)
        self.display: Optional[Display] = None
        self.registry = None
        self.toplevel_manager: Optional[ZwlrForeignToplevelManagerV1] = None
        self.forn_topl_mgr_prot_supported = False
        self.window_handles_dct = {}
        self.active_window_handle = None
        self.outputs = {}

    def signal_handler(self, signal, frame):
        print("\nSignal received, shutting down.")
        self.cleanup()
        sys.exit(0)

    def cleanup(self):
        if self.display is not None:
            print("Disconnecting from Wayland display...")
            self.display.disconnect()
            print("Disconnected from Wayland display.")

    def registry_global_handler(self, registry, id_, interface_name, version):
        """Handle registry events."""
        # print(f"Registry event: id={id_}, interface={interface_name}, version={version}")
        if interface_name == 'zwlr_foreign_toplevel_manager_v1':
            print()
            print(f"Protocol '{interface_name}' version {version} is SUPPORTED.")
            self.forn_topl_mgr_prot_supported = True
            print(f"Creating toplevel manager by binding protocol to registry")

            # # pywayland version:
            # self.toplevel_manager = registry.bind(id_, ZwlrForeignToplevelManagerV1, version)

            # pywlroots version:
            self.toplevel_manager = ZwlrForeignToplevelManagerV1.create(self.display)

            print(f"Subscribing to 'toplevel' events from toplevel manager")
            self.toplevel_manager.dispatcher['toplevel'] = self.handle_toplevel_event
            print()
            self.display.roundtrip()
        elif interface_name == 'wl_seat':
            print(f"Binding to wl_seat interface.")
            seat = registry.bind(id_, WlSeat, version)
            seat.dispatcher['capabilities'] = self.handle_seat_capabilities
            seat.dispatcher['name'] = self.handle_seat_name
            self.display.roundtrip()
        elif interface_name == 'wl_output':
            print(f"Binding to wl_output interface.")
            output = registry.bind(id_, WlOutput, version)
            self.outputs[id_] = output
            output.dispatcher['geometry'] = self.handle_output_geometry
            output.dispatcher['mode'] = self.handle_output_mode
            output.dispatcher['done'] = self.handle_output_done
            output.dispatcher['scale'] = self.handle_output_scale
            self.display.roundtrip()

    def handle_seat_capabilities(self, seat, capabilities):
        print(f"Seat {seat} capabilities changed: {capabilities}")

    def handle_seat_name(self, seat, name):
        print(f"Seat {seat} name set to: {name}")

    def handle_output_geometry(self, output, x, y, physical_width, physical_height, subpixel, make, model, transform):
        print(f"Output {output} geometry updated: {make} {model}, {physical_width}x{physical_height}")

    def handle_output_mode(self, output, flags, width, height, refresh):
        print(f"Output {output} mode changed: {width}x{height}@{refresh/1000}Hz")

    def handle_output_done(self, output):
        print(f"Output {output} configuration done.")

    def handle_output_scale(self, output, factor):
        print(f"Output {output} scale set to {factor}.")

    def handle_toplevel_event(self, toplevel_handle: ZwlrForeignToplevelHandleV1):
        """Handle events for new toplevel windows."""
        print(f"New toplevel window created: {toplevel_handle}")
        
        # Subscribe to title and app_id changes as well as close event
        toplevel_handle.dispatcher['title']             = self.handle_title_change
        toplevel_handle.dispatcher['app_id']            = self.handle_app_id_change
        toplevel_handle.dispatcher['closed']            = self.handle_window_closed
        toplevel_handle.dispatcher['state']             = self.handle_state_change

    def handle_title_change(self, handle, title):
        """Update title in local state."""
        if handle.id not in self.window_handles_dct:
            self.window_handles_dct[handle.id] = {}
        self.window_handles_dct[handle.id]['title'] = title
        print(f"Title updated for window {handle.id}: {title}")

    def handle_app_id_change(self, handle, app_id):
        """Update app_id in local state."""
        if handle.id not in self.window_handles_dct:
            self.window_handles_dct[handle.id] = {}
        self.window_handles_dct[handle.id]['app_id'] = app_id
        print(f"App ID updated for window {handle.id}: {app_id}")

    def handle_window_closed(self, handle):
        """Remove window from local state."""
        if handle.id in self.window_handles_dct:
            del self.window_handles_dct[handle.id]
        print(f"Window {handle.id} has been closed.")

    def handle_state_change(self, handle, states):
        """Track active window based on state changes."""
        if 'activated' in states:
            if self.active_window is not None and self.active_window != handle.id:
                print(f"Window {self.active_window} is no longer active.")
            self.active_window = handle.id
            print(f"Window {handle.id} is now active.")
        elif self.active_window == handle.id:
            # If the currently active window reports any state change that might imply it is no longer active,
            # you would handle that here. Since there's no 'deactivated' state, this part might adjust based on your app's logic.
            print(f"Window {handle.id} is no longer active.")
            self.active_window = None

    def run(self):
        """Run the Wayland client."""
        try:
            print("Connecting to Wayland display...")
            self.display = Display()
            self.display.connect()
            print("Connected to Wayland display")

            print("Getting registry...")
            self.registry = self.display.get_registry()
            print("Registry obtained")

            print("Subscribing to 'global' events from registry")
            self.registry.dispatcher["global"] = self.registry_global_handler

            print("Running roundtrip to process registry events...")
            self.display.roundtrip()

            if self.forn_topl_mgr_prot_supported and self.toplevel_manager:
                print()
                print("Protocol is supported, waiting for events...")

                while self.display.dispatch() != -1:
                    pass

                self.cleanup()
            else:
                print()
                print("Protocol 'zwlr_foreign_toplevel_manager_v1' is NOT supported.")

        except Exception as e:
            print()
            print(f"An error occurred: {e}")
            print(traceback.format_exc())
        finally:
            if self.display is not None:
                self.cleanup()

if __name__ == "__main__":
    print("Starting Wayland client...")
    client = WaylandClient()
    client.run()
    print("Wayland client finished")

[heuer]

Please try PR #64

It attempts to fix the "new id" problem and I used it with the foreign toplevel manager protocol.

[RedBearAK]

@heuer

Definitely gets me past the initial problem, but then reveals that apparently all my handlers aren't set up quite right. Or the dispatchers. Or something.

Do you have a working example of actually using the protocol to get stuff like title, app_id and state events?

Also, I'm still hitting my dispatch() loop and then never seeing any new events show up in the terminal after that. So I have lots of fundamental misunderstandings of things, and not a lot of working examples to draw on.

But the finish line is already much closer.

[heuer]

I haven't studied your code in detail, to be honest, but you'll have to import the generated protocols. I.e. if the generated protocols are located in a module relative to the current module:

from .protocol.wlr_foreign_toplevel_management_unstable_v1.zwlr_foreign_toplevel_manager_v1 import \
    ZwlrForeignToplevelManagerV1, ZwlrForeignToplevelManagerV1Proxy, ZwlrForeignToplevelHandleV1

The protocol implementations in pywlroots are for compositors which want to provide the messages, your client has to use the generated protocols.

I hope this gives a hint into the right direction. I am on vacation and I'll come back later to this issue if you get no other help.

Maybe PR #55 helps regarding the dispatching problem, although it does not implement the foreign toplevel protocol.

[RedBearAK]

@heuer

Thanks for your time. Enjoy your vacation. Even that import example is very helpful. I'll come back if I can't get things working, or let you know if I break through.

The link is interesting. Much different than any other example I've seen. Dispatch comes before the loop even, and then it goes roundtripping if it needs to... Makes more sense to me already.

[RedBearAK]

This is mostly doing what I wanted it to do (showing me app class and window title whenever the focused app changes), but I still have a problem with not really understanding how to do the dispatch loop without saturating a CPU core (or thread) constantly. I have to use time.sleep() to slow down the loop, and I'm pretty sure that's not how it's supposed to work. But without the extra roundtrip() in the loop, I get absolutely nothing coming out after the initial setup phase.

I'll have to take another look at it tomorrow.

Mostly working script (click to expand)

#!/usr/bin/env python3


# Reference for creating the protocols with pywayland scanner:
# https://github.com/flacjacket/pywayland/issues/8#issuecomment-987040284

# Protocol documentation:
# https://wayland.app/protocols/wlr-foreign-toplevel-management-unstable-v1

# pywayland method has a NotImplementedError for NewId argument (Use PR #64 branch or commit)
# "git+https://github.com/heuer/pywayland@issue_33_newid",
# "git+https://github.com/flacjacket/pywayland@db8fb1c3a29761a014cfbb57f84025ddf3882c3c",


import sys
import signal
import traceback

from protocols.wlr_foreign_toplevel_management_unstable_v1.zwlr_foreign_toplevel_manager_v1 import (
    ZwlrForeignToplevelManagerV1,
    ZwlrForeignToplevelManagerV1Proxy,
    ZwlrForeignToplevelHandleV1,
)
from pywayland.client import Display
from protocols.wayland import WlOutput, WlSeat
from typing import Optional
from time import sleep


class WaylandClient:
    def __init__(self):
        """Initialize the WaylandClient."""
        signal.signal(signal.SIGINT, self.signal_handler)
        self.display                            = None
        self.registry                           = None
        self.forn_topl_mgr_prot_supported       = False
        self.toplevel_manager                   = None

        self.wdw_handles_dct                    = {}
        self.active_app_class                   = None
        self.active_wdw_title                   = None

    def signal_handler(self, signal, frame):
        print("\nSignal received, shutting down.")
        self.cleanup()
        sys.exit(0)

    def cleanup(self):
        if self.display is not None:
            print("Disconnecting from Wayland display...")
            self.display.disconnect()
            print("Disconnected from Wayland display.")

    def handle_toplevel_event(self, 
            toplevel_manager: ZwlrForeignToplevelManagerV1Proxy, 
            toplevel_handle: ZwlrForeignToplevelHandleV1):
        """Handle events for new toplevel windows."""
        print(f"New toplevel window created: {toplevel_handle}")
        # Subscribe to title and app_id changes as well as close event
        toplevel_handle.dispatcher['title']             = self.handle_title_change
        toplevel_handle.dispatcher['app_id']            = self.handle_app_id_change
        toplevel_handle.dispatcher['closed']            = self.handle_window_closed
        toplevel_handle.dispatcher['state']             = self.handle_state_change

    def handle_title_change(self, handle, title):
        """Update title in local state."""
        if handle not in self.wdw_handles_dct:
            self.wdw_handles_dct[handle] = {}
        self.wdw_handles_dct[handle]['title'] = title
        print(f"Title updated for window {handle}: '{title}'")

    def handle_app_id_change(self, handle, app_id):
        """Update app_id in local state."""
        if handle not in self.wdw_handles_dct:
            self.wdw_handles_dct[handle] = {}
        self.wdw_handles_dct[handle]['app_id'] = app_id
        print(f"App ID updated for window {handle}: '{app_id}'")

    def handle_window_closed(self, handle):
        """Remove window from local state."""
        if handle in self.wdw_handles_dct:
            del self.wdw_handles_dct[handle]
        print(f"Window {handle} has been closed.")

    def handle_state_change(self, handle, states_bytes):
        """Track active window based on state changes."""
        states = []
        if isinstance(states_bytes, bytes):
            states = list(states_bytes)
        if ZwlrForeignToplevelHandleV1.state.activated.value in states:
            self.active_app_class = self.wdw_handles_dct[handle]['app_id']
            self.active_wdw_title = self.wdw_handles_dct[handle]['title']
            print()
            print(f"Active app class: '{self.active_app_class}'")
            print(f"Active window title: '{self.active_wdw_title}'")

    def registry_global_handler(self, registry, id_, interface_name, version):
        """Handle registry events."""
        # print(f"Registry event: id={id_}, interface={interface_name}, version={version}")
        if interface_name == 'zwlr_foreign_toplevel_manager_v1':
            print()
            print(f"Protocol '{interface_name}' version {version} is _SUPPORTED_.")
            self.forn_topl_mgr_prot_supported = True
            print(f"Creating toplevel manager...")

            # pywayland version:
            self.toplevel_manager = registry.bind(id_, ZwlrForeignToplevelManagerV1, version)

            print(f"Subscribing to 'toplevel' events from toplevel manager...")
            self.toplevel_manager.dispatcher['toplevel'] = self.handle_toplevel_event
            print()
            self.display.roundtrip()

    def run(self):
        """Run the Wayland client."""
        try:
            print("Connecting to Wayland display...")
            with Display() as self.display:
                self.display.connect()
                print("Connected to Wayland display")

                print("Getting registry...")
                self.registry = self.display.get_registry()
                print("Registry obtained")

                print("Subscribing to 'global' events from registry")
                self.registry.dispatcher["global"] = self.registry_global_handler

                print("Running roundtrip to process registry events...")
                self.display.roundtrip()

                if self.forn_topl_mgr_prot_supported and self.toplevel_manager:
                    print()
                    print("Protocol is supported, starting dispatch loop...")

                    # Can't get this to stop pegging a core (or thread) without sleep()
                    # TODO: Need to properly use a lightweight event-driven loop
                    while self.display.dispatch() != -1:
                        sleep(0.05)
                        # seems to be necessary to trigger roundtrip() in a loop,
                        # or no further events will ever print out in the terminal
                        self.display.roundtrip()

                else:
                    print()
                    print("Protocol 'zwlr_foreign_toplevel_manager_v1' is _NOT_ supported.")

        except Exception as e:
            print()
            print(f"An error occurred: {e}")
            print(traceback.format_exc())

if __name__ == "__main__":
    print("Starting Wayland client...")
    client = WaylandClient()
    client.run()
    print("Wayland client finished")

[RedBearAK]

I found a way to keep the event loop from going out of control and using CPU constantly, but it looks nothing like any example I've seen. It involves getting the Wayland file descriptor after connecting to the display object, and then using that as a way to control the loop (instead of just using display.dispatch() != -1).

In one script:

                    while True:
                        # Wait for the Wayland file descriptor to be ready
                        rlist, wlist, xlist = select.select([self.wl_fd], [], [])

                        if self.wl_fd in rlist:
                            # self.display.dispatch()   # won't show me new events
                            self.display.roundtrip()

In another script:

def wayland_event_callback(fd, condition, display):
    if condition & GLib.IO_IN:
        # display.dispatch()    # dispatch() fails to prompt new events to appear
        # dispatch() also seems to trigger the callback to get called many times in a loop,
        # but without any new useful events appearing, while roundtrip() just shows
        # the new events that I need to see, as they happen.
        display.roundtrip()     # gets new events to appear immediately
    return True

def main():

    ... (other code)

    # GLib.idle_add(wayland_event_callback)
    GLib.io_add_watch(wl_fd, GLib.IO_IN, wayland_event_callback, display)

    display.roundtrip() # get the event cycle started (callback never gets called without this)

    # Run the main loop
    # dbus.mainloop.glib.DBusGMainLoop().run()
    GLib.MainLoop().run()

Another aspect is avoiding the use of dispatch() entirely, which seems to either cause me problems with a non-blocking event loop, and/or just never gets any new events to print out. So I'm using only roundtrip() calls. Neither I nor any version of ChatGPT no matter how "smart" can explain why I have to do things this way. Doesn't seem to make much sense. Any input on this is welcome.

Anyhoo, here are a couple of scripts that are working for me, and I added the printing out of the whole list of running applications in the output, even though I'm only interested in the class/title of the focused app window.

---

This first script is just a straight establishing of the foreign toplevel management protocol being supported, and then a continuous printing out of the "active" application class and window title (and the app list).

Simple Python script to show active app info (click to expand)

#!/usr/bin/env python3


# Reference for generating the protocol modules with pywayland scanner:
# https://github.com/flacjacket/pywayland/issues/8#issuecomment-987040284

# Protocol documentation:
# https://wayland.app/protocols/wlr-foreign-toplevel-management-unstable-v1

# pywayland method has a NotImplementedError for NewId argument
# (Use GitHub repo PR #64 branch or commit to install with 'pip')
# git+https://github.com/heuer/pywayland@issue_33_newid
# git+https://github.com/flacjacket/pywayland@db8fb1c3a29761a014cfbb57f84025ddf3882c3c


import sys
import select
import signal
import traceback


from protocols.wlr_foreign_toplevel_management_unstable_v1.zwlr_foreign_toplevel_manager_v1 import (
    ZwlrForeignToplevelManagerV1,
    ZwlrForeignToplevelManagerV1Proxy,
    ZwlrForeignToplevelHandleV1,
)

from pywayland.client import Display
from time import sleep

ERR_NO_WLR_APP_CLASS = "ERR_no_wlr_app_class"
ERR_NO_WLR_WDW_TITLE = "ERR_no_wlr_wdw_title"

class WaylandClient:
    def __init__(self):
        """Initialize the WaylandClient."""
        signal.signal(signal.SIGINT, self.signal_handler)
        self.display                            = None
        self.wl_fd                              = None
        self.registry                           = None
        self.forn_topl_mgr_prot_supported       = False
        self.toplevel_manager                   = None

        self.wdw_handles_dct                    = {}
        self.active_app_class                   = ERR_NO_WLR_APP_CLASS
        self.active_wdw_title                   = ERR_NO_WLR_WDW_TITLE

    def signal_handler(self, signal, frame):
        print(f"\nSignal {signal} received, shutting down.")
        self.cleanup()
        sys.exit(0)

    def cleanup(self):
        if self.display is not None:
            print("Disconnecting from Wayland display...")
            self.display.disconnect()
            print("Disconnected from Wayland display.")

    def handle_toplevel_event(self, 
            toplevel_manager: ZwlrForeignToplevelManagerV1Proxy, 
            toplevel_handle: ZwlrForeignToplevelHandleV1):
        """Handle events for new toplevel windows."""
        # print(f"New toplevel window created: {toplevel_handle}")
        # Subscribe to title and app_id changes as well as close event
        toplevel_handle.dispatcher['title']             = self.handle_title_change
        toplevel_handle.dispatcher['app_id']            = self.handle_app_id_change
        toplevel_handle.dispatcher['closed']            = self.handle_window_closed
        toplevel_handle.dispatcher['state']             = self.handle_state_change

    def handle_title_change(self, handle, title):
        """Update title in local state."""
        if handle not in self.wdw_handles_dct:
            self.wdw_handles_dct[handle] = {}
        self.wdw_handles_dct[handle]['title'] = title
        # print(f"Title updated for window {handle}: '{title}'")

    def handle_app_id_change(self, handle, app_id):
        """Update app_id in local state."""
        if handle not in self.wdw_handles_dct:
            self.wdw_handles_dct[handle] = {}
        self.wdw_handles_dct[handle]['app_id'] = app_id
        # print(f"App ID updated for window {handle}: '{app_id}'")

    def handle_window_closed(self, handle):
        """Remove window from local state."""
        if handle in self.wdw_handles_dct:
            del self.wdw_handles_dct[handle]
        print(f"Window {handle} has been closed.")

    def handle_state_change(self, handle, states_bytes):
        """Track active window based on state changes."""
        states = []
        if isinstance(states_bytes, bytes):
            states = list(states_bytes)
        if ZwlrForeignToplevelHandleV1.state.activated.value in states:
            self.active_app_class = self.wdw_handles_dct[handle]['app_id']
            self.active_wdw_title = self.wdw_handles_dct[handle]['title']
            print()
            print(f"Active app class: '{self.active_app_class}'")
            print(f"Active window title: '{self.active_wdw_title}'")
            self.print_running_applications()  # Print the list of running applications

    def print_running_applications(self):
        """Print a complete list of running applications."""
        print("\nList of running applications:")
        print(f"{'App ID':<30} {'Title':<50}")
        print("-" * 80)
        for handle, info in self.wdw_handles_dct.items():
            app_id = info.get('app_id', ERR_NO_WLR_APP_CLASS)
            title = info.get('title', ERR_NO_WLR_WDW_TITLE)
            print(f"{app_id:<30} {title:<50}")
        print()

    def registry_global_handler(self, registry, id_, interface_name, version):
        """Handle registry events."""
        # print(f"Registry event: id={id_}, interface={interface_name}, version={version}")
        if interface_name == 'zwlr_foreign_toplevel_manager_v1':
            print()
            print(f"Protocol '{interface_name}' version {version} is _SUPPORTED_.")
            self.forn_topl_mgr_prot_supported = True
            print(f"Creating toplevel manager...")

            # pywayland version:
            self.toplevel_manager = registry.bind(id_, ZwlrForeignToplevelManagerV1, version)

            print(f"Subscribing to 'toplevel' events from toplevel manager...")
            self.toplevel_manager.dispatcher['toplevel'] = self.handle_toplevel_event
            print()
            self.display.roundtrip()

    def run(self):
        """Run the Wayland client."""
        try:
            print("Connecting to Wayland display...")
            with Display() as self.display:
                self.display.connect()
                print("Connected to Wayland display")

                self.wl_fd = self.display.get_fd()
                print("Got Wayland file descriptor")

                print("Getting registry...")
                self.registry = self.display.get_registry()
                print("Registry obtained")

                print("Subscribing to 'global' events from registry")
                self.registry.dispatcher["global"] = self.registry_global_handler

                print("Running roundtrip to process registry events...")
                self.display.roundtrip()

                # After initial events are processed, we should know if the right
                # protocol is supported, and have a toplevel_manager object.
                if self.forn_topl_mgr_prot_supported and self.toplevel_manager:
                    print()
                    print("Protocol is supported, starting dispatch loop...")

                    # # Can't get this to stop pegging a core (or thread) without sleep()
                    # # It is as if dispatch() does not block at all
                    # # TODO: Need to properly use a lightweight event-driven loop, but how? 
                    # while self.display.dispatch() != -1:
                    #     sleep(0.05)
                    #     # seems to be necessary to trigger roundtrip() in a loop,
                    #     # or no further events will ever print out in the terminal
                    #     self.display.roundtrip()

                    while True:
                        # Wait for the Wayland file descriptor to be ready
                        rlist, wlist, xlist = select.select([self.wl_fd], [], [])

                        if self.wl_fd in rlist:
                            # self.display.dispatch()   # won't show me new events
                            self.display.roundtrip()

                else:
                    print()
                    print("Protocol 'zwlr_foreign_toplevel_manager_v1' is _NOT_ supported.")

        except Exception as e:
            print()
            print(f"An error occurred: {e}")
            print(traceback.format_exc())

if __name__ == "__main__":
    print("Starting Wayland client...")
    client = WaylandClient()
    client.run()
    print("Wayland client finished")

---

(Edited this to be the version I converted to encapsulate the Wayland-related functions in a class, like the other one.)

This other script is more complicated, integrating the same kind of extraction of the active window info into a module that also creates a D-Bus service (with GLib) that can be queried to get the info:

Script to make app info available via D-Bus (click to expand)

#!/usr/bin/env python3

# Reference for generating the protocol modules with pywayland scanner:
# https://github.com/flacjacket/pywayland/issues/8#issuecomment-987040284

# Protocol documentation:
# https://wayland.app/protocols/wlr-foreign-toplevel-management-unstable-v1

# pywayland method has a NotImplementedError for NewId argument
# (Use GitHub repo PR #64 branch or commit to install with 'pip')
# git+https://github.com/heuer/pywayland@issue_33_newid
# git+https://github.com/flacjacket/pywayland@db8fb1c3a29761a014cfbb57f84025ddf3882c3c


print("(--) Starting D-Bus service to monitor wlr-foreign-toplevel-management protocol...")

import os
import sys
import dbus
import time
import signal
import platform
import dbus.service
import dbus.mainloop.glib
import xwaykeyz.lib.logger

from pywayland.client import Display
from gi.repository import GLib
from dbus.exceptions import DBusException
from subprocess import DEVNULL
from typing import Dict
from xwaykeyz.lib.logger import debug, error

xwaykeyz.lib.logger.VERBOSE = True


# Independent module/script to create a D-Bus window context service in 
# a wlroots Wayland environment, which will be notified of window 
# focus changes by the Wayland compositor, as long as the compositor 
# implements the `wlr_foreign_toplevel_management_unstable_v1` protocol.

# Add paths to avoid errors like ModuleNotFoundError or ImportError
home_dir            = os.path.expanduser("~")
run_tmp_dir         = os.environ.get('XDG_RUNTIME_DIR') or '/tmp'
parent_folder_path  = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
current_folder_path = os.path.abspath(os.path.dirname(__file__))

sys.path.insert(0, current_folder_path)
sys.path.insert(0, parent_folder_path)

existing_path = os.environ.get('PYTHONPATH', '')
os.environ['PYTHONPATH'] = f'{parent_folder_path}:{current_folder_path}:{existing_path}'

# local imports now that path is prepped
import lib.env as env

from protocols.wlr_foreign_toplevel_management_unstable_v1.zwlr_foreign_toplevel_manager_v1 import (
    ZwlrForeignToplevelManagerV1,
    ZwlrForeignToplevelManagerV1Proxy,
    ZwlrForeignToplevelHandleV1
)

if os.name == 'posix' and os.geteuid() == 0:
    error("This app should not be run as root/superuser.")
    sys.exit(1)


# Establish our Wayland client global variable
wl_client = None

def signal_handler(sig, frame):
    """handle signals like Ctrl+C"""
    if sig in (signal.SIGINT, signal.SIGQUIT):
        # Perform any cleanup code here before exiting
        # traceback.print_stack(frame)
        debug(f'\nSIGINT or SIGQUIT received. Exiting.\n')
        clean_shutdown()

def clean_shutdown():
    if wl_client and wl_client.display:  # Check if the display is globally defined and initialized
        try:
            wl_client.display.disconnect()
        except Exception as e:
            error(f"Error disconnecting display: {e}")
    GLib.MainLoop().quit()  # Stop the GLib main loop if it's running
    sys.exit(0)


if platform.system() != 'Windows':
    signal.signal(signal.SIGINT,    signal_handler)
    signal.signal(signal.SIGQUIT,   signal_handler)
else:
    error(f'This is only meant to run on Linux. Exiting...')
    sys.exit(1)


sep_reps        = 80
sep_char        = '='
separator       = sep_char * sep_reps

LOG_PFX = 'TOSHY_WLR_DBUS_SVC'

DISTRO_ID       = None
DISTRO_VER      = None
VARIANT_ID      = None
SESSION_TYPE    = None
DESKTOP_ENV     = None
DE_MAJ_VER      = None


def check_environment():
    """Retrieve the current environment from env module"""
    env_info: Dict[str, str] = env.get_env_info()   # Returns a dict
    global DISTRO_ID, DISTRO_VER, VARIANT_ID, SESSION_TYPE, DESKTOP_ENV, DE_MAJ_VER
    DISTRO_ID       = env_info.get('DISTRO_ID')
    DISTRO_VER      = env_info.get('DISTRO_VER')
    VARIANT_ID      = env_info.get('VARIANT_ID')
    SESSION_TYPE    = env_info.get('SESSION_TYPE')
    DESKTOP_ENV     = env_info.get('DESKTOP_ENV')
    DE_MAJ_VER      = env_info.get('DE_MAJ_VER')


check_environment()

# TODO: put the DE restriction back in place, find a way to identify wlroots compositors
if SESSION_TYPE == 'wayland': # and DESKTOP_ENV not in ['kde', 'plasma', 'gnome', 'cinnamon']:
    pass
else:
    debug(f'{LOG_PFX}: Probably not a wlroots environment. Exiting.')
    time.sleep(2)
    sys.exit(0)


debug("")
debug(  f'Toshy KDE D-Bus service script sees this environment:'
        f'\n\t{DISTRO_ID        = }'
        f'\n\t{DISTRO_VER       = }'
        f'\n\t{VARIANT_ID       = }'
        f'\n\t{SESSION_TYPE     = }'
        f'\n\t{DESKTOP_ENV      = }'
        f'\n\t{DE_MAJ_VER       = }\n', ctx="CG")


TOSHY_WLR_DBUS_SVC_PATH         = '/org/toshy/Wlroots'
TOSHY_WLR_DBUS_SVC_IFACE        = 'org.toshy.Wlroots'

ERR_NO_WLR_APP_CLASS = "ERR_no_wlr_app_class"
ERR_NO_WLR_WDW_TITLE = "ERR_no_wlr_wdw_title"


class WaylandClient:
    def __init__(self):
        self.display = None
        self.registry = None
        self.toplevel_manager = None
        self.wl_fd = None
        self.wdw_handles_dct = {}
        self.active_app_class = ERR_NO_WLR_APP_CLASS
        self.active_wdw_title = ERR_NO_WLR_WDW_TITLE

    def connect(self):
        try:
            self.display = Display()
            self.display.connect()
            self.wl_fd = self.display.get_fd()
            self.registry = self.display.get_registry()
            self.registry.dispatcher["global"] = self.handle_registry_global
            self.display.roundtrip()
        except Exception as e:
            print(f"Failed to connect to the Wayland display: {e}")
            clean_shutdown()

    def handle_registry_global(self, registry, id_, interface_name, version):
        if interface_name == 'zwlr_foreign_toplevel_manager_v1':
            self.toplevel_manager = registry.bind(id_, ZwlrForeignToplevelManagerV1, version)
            self.toplevel_manager.dispatcher["toplevel"] = self.handle_toplevel_event

    def handle_toplevel_event(self, toplevel_manager, toplevel_handle):
        toplevel_handle.dispatcher["app_id"] = self.handle_app_id_change
        toplevel_handle.dispatcher["title"] = self.handle_title_change
        toplevel_handle.dispatcher['closed'] = self.handle_window_closed
        toplevel_handle.dispatcher['state'] = self.handle_state_change

    def handle_app_id_change(self, handle, app_id):
        if handle not in self.wdw_handles_dct:
            self.wdw_handles_dct[handle] = {}
        self.wdw_handles_dct[handle]['app_id'] = app_id

    def handle_title_change(self, handle, title):
        if handle not in self.wdw_handles_dct:
            self.wdw_handles_dct[handle] = {}
        self.wdw_handles_dct[handle]['title'] = title

    def handle_window_closed(self, handle):
        if handle in self.wdw_handles_dct:
            del self.wdw_handles_dct[handle]
        print(f"Window {handle} has been closed.")

    def handle_state_change(self, handle, states_bytes):
        states = []
        if isinstance(states_bytes, bytes):
            states = list(states_bytes)
        if ZwlrForeignToplevelHandleV1.state.activated.value in states:
            self.active_app_class = self.wdw_handles_dct[handle]['app_id']
            self.active_wdw_title = self.wdw_handles_dct[handle]['title']
            print()
            print(f"Active app class: '{self.active_app_class}'")
            print(f"Active window title: '{self.active_wdw_title}'")
            self.print_running_applications()

    def print_running_applications(self):
        print("\nList of running applications:")
        print(f"{'App ID':<30} {'Title':<50}")
        print("-" * 80)
        for handle, info in self.wdw_handles_dct.items():
            app_id = info.get('app_id', ERR_NO_WLR_APP_CLASS)
            title = info.get('title', ERR_NO_WLR_WDW_TITLE)
            print(f"{app_id:<30} {title:<50}")
        print()


class DBUS_Object(dbus.service.Object):
    """Class to handle D-Bus interactions"""
    def __init__(self, session_bus, object_path, interface_name):
        super().__init__(session_bus, object_path)
        self.interface_name     = interface_name
        self.dbus_svc_bus_name  = dbus.service.BusName(interface_name, bus=session_bus)

    @dbus.service.method(TOSHY_WLR_DBUS_SVC_IFACE, out_signature='a{sv}')
    def GetActiveWindow(self):
        debug(f'{LOG_PFX}: GetActiveWindow() called...')
        return {'app_id':           wl_client.active_app_class,
                'title':            wl_client.active_wdw_title}


def wayland_event_callback(fd, condition, display: Display):
    if condition & GLib.IO_IN:
        # display.dispatch()    # dispatch() fails to prompt new events to appear
        # dispatch() also seems to trigger the callback to get called many times in a loop,
        # but without any new useful events appearing, while roundtrip() just shows
        # the new events that I need to see, as they happen.
        display.roundtrip()     # gets new events to appear immediately
    return True


def main():

    # Initialize the D-Bus main loop
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    # Connect to the session bus
    session_bus = dbus.SessionBus()

    # Create the DBUS_Object
    try:
        DBUS_Object(session_bus, TOSHY_WLR_DBUS_SVC_PATH, TOSHY_WLR_DBUS_SVC_IFACE)
    except DBusException as dbus_error:
        error(f"{LOG_PFX}: Error occurred while creating D-Bus service object:\n\t{dbus_error}")
        sys.exit(1)

    global wl_client        # Is this necessary?
    wl_client = WaylandClient()
    wl_client.connect()     # This connects display, gets registry, and also gets file descriptor

    GLib.io_add_watch(wl_client.wl_fd, GLib.IO_IN, wayland_event_callback, wl_client.display)

    wl_client.display.roundtrip() # get the event cycle started (callback never gets called without this)

    # Run the main loop
    # dbus.mainloop.glib.DBusGMainLoop().run()
    GLib.MainLoop().run()


if __name__ == "__main__":
    main()
    # After main() is done:
    clean_shutdown()

---

Tip

FYI, for these hidden "spoiler" expandable things you need a blank line between the </summary> HTML tag and the "```py" tag that starts the code block, or everything after that gets misinterpreted for some reason.

[RedBearAK]

@heuer

Your hints and examples allowed me to put together window context providers for both the zwlr_foreign_toplevel_manager_v1 interface in compositors that use the wlr_foreign_toplevel_management_unstable_v1 protocol, and a closely related method for the upcoming Pop!_OS COSMIC desktop environment. Just needed some additional XML files from their cosmic-protocols GitHub repo:

cosmic-toplevel-info-unstable-v1.xml
cosmic-workspace-unstable-v1.xml

There's some kind of dependency on the "workspace" protocol, so there's an error when generating the Python protocol files with it.

And then the object names were a bit different (and might change when they implement a new draft "state" protocol, we'll see how that goes):

# from protocols.wlr_foreign_toplevel_management_unstable_v1.zwlr_foreign_toplevel_manager_v1 import (
#     ZwlrForeignToplevelManagerV1,
#     ZwlrForeignToplevelManagerV1Proxy,
#     ZwlrForeignToplevelHandleV1,
# )

# COSMIC-specific protocol module, using their own namespace
# XML files downloaded from `cosmic-protocols` GitHub repo, in 'unstable' folder
from protocols.cosmic_toplevel_info_unstable_v1.zcosmic_toplevel_info_v1 import (
    ZcosmicToplevelInfoV1,
    ZcosmicToplevelInfoV1Proxy,
    ZcosmicToplevelHandleV1
)

The code is otherwise basically identical to the more generic wlroots method, which has been confirmed to work on multiple Wayland compositors so far (Niri, Qtile, Hyprland, sway).

Weirdly the COSMIC protocol that seemed closest in naming turned out to not be the one I needed to look at:

cosmic-toplevel-management-unstable-v1.xml

Thanks for your help with this. It was instrumental.