python-xlib example which reacts to changing the active window

x11_watch_active_window.py

#!/usr/bin/env python
"""python-xlib example which reacts to changing the active window/title.

Requires:
- Python
- python-xlib

Tested with Python 2.x because my Kubuntu 14.04 doesn't come with python-xlib
for Python 3.x.

Design:
-------

Any modern window manager that isn't horrendously broken maintains an X11
property on the root window named _NET_ACTIVE_WINDOW.

Any modern application toolkit presents the window title via a property
named _NET_WM_NAME.

This listens for changes to both of them and then hides duplicate events
so it only reacts to title changes once.

Known Bugs:
-----------

- Under some circumstances, I observed that the first window creation and last
  window deletion on on an empty desktop (ie. not even a taskbar/panel) would
  go ignored when using this test setup:

      Xephyr :3 &
      DISPLAY=:3 openbox &
      DISPLAY=:3 python3 x11_watch_active_window.py

      # ...and then launch one or more of these in other terminals
      DISPLAY=:3 leafpad
"""

# pylint: disable=unused-import
from contextlib import contextmanager
from typing import Any, Dict, Optional, Tuple, Union  # noqa

from Xlib import X
from Xlib.display import Display
from Xlib.error import XError
from Xlib.xobject.drawable import Window
from Xlib.protocol.rq import Event

# Connect to the X server and get the root window
disp = Display()
root = disp.screen().root

# Prepare the property names we use so they can be fed into X11 APIs
NET_ACTIVE_WINDOW = disp.intern_atom('_NET_ACTIVE_WINDOW')
NET_WM_NAME = disp.intern_atom('_NET_WM_NAME')  # UTF-8
WM_NAME = disp.intern_atom('WM_NAME')           # Legacy encoding

last_seen = {'xid': None, 'title': None}  # type: Dict[str, Any]


@contextmanager
def window_obj(win_id: Optional[int]) -> Window:
    """Simplify dealing with BadWindow (make it either valid or None)"""
    window_obj = None
    if win_id:
        try:
            window_obj = disp.create_resource_object('window', win_id)
        except XError:
            pass
    yield window_obj


def get_active_window() -> Tuple[Optional[int], bool]:
    """Return a (window_obj, focus_has_changed) tuple for the active window."""
    response = root.get_full_property(NET_ACTIVE_WINDOW,
                                      X.AnyPropertyType)
    if not response:
        return None, False
    win_id = response.value[0]

    focus_changed = (win_id != last_seen['xid'])
    if focus_changed:
        with window_obj(last_seen['xid']) as old_win:
            if old_win:
                old_win.change_attributes(event_mask=X.NoEventMask)

        last_seen['xid'] = win_id
        with window_obj(win_id) as new_win:
            if new_win:
                new_win.change_attributes(event_mask=X.PropertyChangeMask)

    return win_id, focus_changed


def _get_window_name_inner(win_obj: Window) -> str:
    """Simplify dealing with _NET_WM_NAME (UTF-8) vs. WM_NAME (legacy)"""
    for atom in (NET_WM_NAME, WM_NAME):
        try:
            window_name = win_obj.get_full_property(atom, 0)
        except UnicodeDecodeError:  # Apparently a Debian distro package bug
            title = "<could not decode characters>"
        else:
            if window_name:
                win_name = window_name.value  # type: Union[str, bytes]
                if isinstance(win_name, bytes):
                    # Apparently COMPOUND_TEXT is so arcane that this is how
                    # tools like xprop deal with receiving it these days
                    win_name = win_name.decode('latin1', 'replace')
                return win_name
            else:
                title = "<unnamed window>"

    return "{} (XID: {})".format(title, win_obj.id)


def get_window_name(win_id: Optional[int]) -> Tuple[Optional[str], bool]:
    """Look up the window name for a given X11 window ID"""
    if not win_id:
        last_seen['title'] = None
        return last_seen['title'], True

    title_changed = False
    with window_obj(win_id) as wobj:
        if wobj:
            try:
                win_title = _get_window_name_inner(wobj)
            except XError:
                pass
            else:
                title_changed = (win_title != last_seen['title'])
                last_seen['title'] = win_title

    return last_seen['title'], title_changed


def handle_xevent(event: Event):
    """Handler for X events which ignores anything but focus/title change"""
    if event.type != X.PropertyNotify:
        return

    changed = False
    if event.atom == NET_ACTIVE_WINDOW:
        if get_active_window()[1]:
            get_window_name(last_seen['xid'])  # Rely on the side-effects
            changed = True
    elif event.atom in (NET_WM_NAME, WM_NAME):
        changed = changed or get_window_name(last_seen['xid'])[1]

    if changed:
        handle_change(last_seen)


def handle_change(new_state: dict):
    """Replace this with whatever you want to actually do"""
    print(new_state)

if __name__ == '__main__':
    # Listen for _NET_ACTIVE_WINDOW changes
    root.change_attributes(event_mask=X.PropertyChangeMask)

    # Prime last_seen with whatever window was active when we started this
    get_window_name(get_active_window()[0])
    handle_change(last_seen)

    while True:  # next_event() sleeps until we get an event
        handle_xevent(disp.next_event())

Wonderful implementation, thank you for the detailed comments!

Does anyone know how to turn code above into rust code ? this is what I did.

use std::collections::HashMap;
use std::error::Error;

use x11rb::connection::Connection;

use x11rb::protocol::xproto::{Atom, AtomEnum, ConnectionExt, GetPropertyReply, Window};

use x11rb::protocol::Event;
use x11rb::x11_utils::TryParse;
use x11rb::xcb_ffi::XCBConnection;

fn find_active_window(
conn: &impl Connection,
root: Window,
net_active_window: Atom,
hash_map: &mut HashMap<&str, Option>,
) -> Result<(Window, bool), Box> {
let window: Window = AtomEnum::ANY.into();
let active_window = conn
.get_property(false, root, net_active_window, window, 0, 1)?
.reply()?;
if active_window.format == 32 && active_window.length == 1 {
// Things will be so much easier with the next release:
let widnow_id = u32::try_parse(&active_window.value)?.0;
let focus_changed = widnow_id != hash_map["xid"].unwrap();
hash_map.insert("xid", Some(widnow_id));
Ok((u32::try_parse(&active_window.value)?.0, focus_changed))
} else {
// Query the input focus
Ok((conn.get_input_focus()?.reply()?.focus, false))
}
}
fn parse_string_property(property: &GetPropertyReply) -> &str {
std::str::from_utf8(&property.value).unwrap_or("Invalid utf8")
}
fn parse_wm_class(property: &GetPropertyReply) -> (&str, &str) {
if property.format != 8 {
return (
"Malformed property: wrong format",
"Malformed property: wrong format",
);
}
let value = &property.value;
// The property should contain two null-terminated strings. Find them.
if let Some(middle) = value.iter().position(|&b| b == 0) {
let (instance, class) = value.split_at(middle);
// Skip the null byte at the beginning
let mut class = &class[1..];
// Remove the last null byte from the class, if it is there.
if class.last() == Some(&0) {
class = &class[..class.len() - 1];
}
let instance = std::str::from_utf8(instance);
let class = std::str::from_utf8(class);
(
instance.unwrap_or("Invalid utf8"),
class.unwrap_or("Invalid utf8"),
)
} else {
("Missing null byte", "Missing null byte")
}
}

pub fn start() -> Result<(), Box> {
let mut last_seen = HashMap::new();
last_seen.insert("xid", Some(10000000));
// Set up our state
let (conn, screen) = XCBConnection::connect(None)?;
let root = conn.setup().roots[screen].root;
let net_activate_win = conn.intern_atom(false, b"_NET_ACTIVE_WINDOW").unwrap();
let net_wm_name = conn.intern_atom(false, b"_NET_WM_NAME").unwrap();
let utf8_string = conn.intern_atom(false, b"UTF8_STRING").unwrap();
let net_activate_win = net_activate_win.reply().unwrap().atom;
let net_wm_name = net_wm_name.reply().unwrap().atom;
let utf8_string = utf8_string.reply().unwrap().atom;
let (focus, _) = find_active_window(&conn, root, net_activate_win, &mut last_seen)?;
println!("XID {:?}", focus);
// Collect the replies to the atoms
let (net_wm_name, utf8_string) = (net_wm_name, utf8_string);
let (wm_class, string): (AtomEnum, AtomEnum) =
(AtomEnum::WM_CLASS.into(), AtomEnum::STRING.into());
// Get the property from the window that we need
let name = conn.get_property(false, focus, net_wm_name, utf8_string, 0, u32::max_value())?;
let class = conn.get_property(false, focus, wm_class, string, 0, u32::max_value())?;
let (name, class) = (name.reply()?, class.reply()?);
println!("Window name: {:?}", parse_string_property(&name));
let (instance, class) = parse_wm_class(&class);
println!("Window instance: {:?}", instance);
println!("Window class: {:?}", class);
// Print out the result

loop {
    conn.flush().unwrap();
    match conn.wait_for_event() {
        Ok(event) => match event {
            Event::PropertyNotify(e) => {
                println!("Event: {:?}", e);
            }
            _ => {}
        },
        Err(e) => println!("Error: {:?}", e),
    }
}

}

You should package the main into its own function and have a callback so this can be used fully as a lib instead of modifying it, it can be used directly as a submodule/its own dependency

@ShayBox The problem is that, when you start to think about how to integrate it into programs which may already have an Xlib connection for other purposes, it's not as simple to offer it up structured that way. Unnecessarily pushing a program to open more X sockets is a code smell.

Can it react to change in position rather than title?

@vednig I've never tried it, so I can't be sure. What you want to do is check xprop's output to identify a property you can listen for changes to.

_NET_WM_OPAQUE_REGION looks appealing, but it's actually an optimization hint that not all applications set (eg. Firefox does but FeatherPad doesn't).

Since you just need to detect change without directly parsing what you're listening on, I'd suggest watching WM_NORMAL_HINTS.

Is anybody interested in make a script to save and restore window layout?
I have a feature request on another project
joshgoebel/keyszer#70