A Python3 Xlib script to a-b switch between recently used virtual desktops

x11-desktop-a-b-switcher.py

#!/usr/bin/env python3


"""EWMH A-B desktop switcher (aka switch to the previously selected desktop)

A Python script for X11 that listens to changes of the current virtual desktop
and remembers the previously selected virtual desktop based on the
_NET_CURRENT_DESKTOP EWMH extension in X11.
It also listens on a UNIX socket, which if sent to, will switch back to the
remembered previous desktop.
This is meant for window managers that lack a keyboard shortcut for "switch
back to previously selected desktop" (like Xfwm4 for example).

Requirements:
- python3-xlib
- netcat

Usage:
1. add the Python script to your autostart
2. add a keyboard shortcut for the following command (requires netcat):

    bash -c "nc -U /tmp/desktop-a-b-switch.socket <<< 1"

"""

from Xlib import X, display
import Xlib.protocol.event
import functools
import asyncio
import os

SOCK_PATH = "/tmp/desktop-a-b-switch.socket"

xdisplay = display.Display()
screen = xdisplay.screen()
root = screen.root

desktop_number_lock = asyncio.Lock()
current_desktop = 0
last_desktop = 0


def run_in_executor(f):
    """asyncio wrapper for legacy blocking functions
    source: https://stackoverflow.com/a/53719009
    """
    @functools.wraps(f)
    def inner(*args, **kwargs):
        loop = asyncio.get_running_loop()
        return loop.run_in_executor(None, lambda: f(*args, **kwargs))

    return inner


def change_desktop(index: int):
    data = [index, X.CurrentTime]
    data = (data+[0]*(5-len(data)))[:5]
    ev = Xlib.protocol.event.ClientMessage(
        window=root,
        client_type=xdisplay.intern_atom("_NET_CURRENT_DESKTOP"),
        data=(32, (data))
    )
    mask = (X.SubstructureRedirectMask | X.SubstructureNotifyMask)
    root.send_event(ev, event_mask=mask)
    xdisplay.flush()


@run_in_executor
def _async_get_current_desktop() -> int:
    atom = root.get_full_property(
        xdisplay.get_atom('_NET_CURRENT_DESKTOP'),
        X.AnyPropertyType
    )
    if atom:
        return atom.value[0]
    else:
        return None


@run_in_executor
def _async_get_display_event():
    return xdisplay.next_event()


async def desktop_change_listener():
    global root
    global current_desktop
    global last_desktop
    root.change_attributes(event_mask=X.PropertyChangeMask)
    while 1:
        evt = await _async_get_display_event()
        if (evt.type == X.PropertyNotify and
                evt.atom == xdisplay.intern_atom('_NET_CURRENT_DESKTOP')):
            async with desktop_number_lock:
                next_desktop = await _async_get_current_desktop()
                if (next_desktop is not None
                        and next_desktop != current_desktop):
                    last_desktop = current_desktop
                    current_desktop = next_desktop


async def switch_to_previous_desktop():
    async with desktop_number_lock:
        change_desktop(last_desktop)


async def socket_listener():

    async def handle_connection(reader, writer):
        await reader.readline()
        await switch_to_previous_desktop()
        await writer.drain()
        writer.close()
        await writer.wait_closed()

    if os.path.exists(SOCK_PATH):
        os.remove(SOCK_PATH)
    server = await asyncio.start_unix_server(handle_connection, path=SOCK_PATH)
    async with server:
        await server.serve_forever()


async def main():
    tasks = [socket_listener(), desktop_change_listener()]
    await asyncio.gather(*tasks)


asyncio.run(main())