1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
"""Automatically update xsettings based on gsettings"""
__version__ = "0.1.0"
import os
import signal
import struct
import sys
import tempfile
from collections.abc import Iterator
from pathlib import Path
from subprocess import Popen
from gi.repository import Gio, GLib
DESKTOP_INTERFACE_SCHEMA = "org.gnome.desktop.interface"
GTK_THEME_KEY = "gtk-theme"
HEADER_FORMAT = ">H"
def start_gsettings_listener() -> tuple[int, int]:
def on_change(settings, key):
if settings.props.schema_id != DESKTOP_INTERFACE_SCHEMA:
return
if key != GTK_THEME_KEY:
return
gtk_theme = settings.get_string(key).encode()
data = struct.pack(HEADER_FORMAT, len(gtk_theme))
os.write(write, data + gtk_theme)
read, write = os.pipe()
os.set_inheritable(write, True)
pid = os.fork()
if pid != 0:
os.close(write)
return read, pid
settings = Gio.Settings.new(DESKTOP_INTERFACE_SCHEMA)
settings.connect("changed", on_change)
on_change(settings, GTK_THEME_KEY)
try:
GLib.MainLoop().run()
exit(1)
except KeyboardInterrupt:
os._exit(0)
def start_xsettingsd(conf: str, args: list[str]) -> Popen:
return Popen(["xsettingsd", "--config", conf] + args)
def read_theme(fd: int) -> str:
header = os.read(fd, 2)
(length,) = struct.unpack(HEADER_FORMAT, header)
return os.read(fd, length).decode()
def xsettingsd_escape(s: str) -> str:
return s.replace("\n", r"\n").replace('"', r"\"")
def get_xsettingsd_config_paths() -> Iterator[Path]:
yield Path.home() / ".xsettingsd"
conf_home = os.environ.get("XDG_CONFIG_HOME")
conf_home = Path(conf_home) if conf_home else Path.home() / ".config"
xdg_dirs = [conf_home]
conf_dirs = os.environ.get("XDG_CONFIG_DIRS")
if conf_dirs:
for d in conf_dirs.split(":"):
xdg_dirs.append(Path(d))
for d in xdg_dirs:
yield d / "xsettingsd" / "xsettingsd.conf"
XSETTINGSD_CONFIG_PATHS = list(get_xsettingsd_config_paths())
# It would make more sense to only read the config once every HUP
def get_xsettingsd_config() -> str:
for path in XSETTINGSD_CONFIG_PATHS:
try:
with path.open() as f:
return f.read()
except FileNotFoundError:
pass
return ""
def write_merged_settings(fd: int, gtk_theme: str) -> None:
os.lseek(fd, 0, os.SEEK_SET)
os.ftruncate(fd, 0)
os.write(fd, f'Net/ThemeName "{xsettingsd_escape(gtk_theme)}"\n'.encode())
os.write(fd, get_xsettingsd_config().encode())
def start_reconfigurator(theme_fd: int, conf_fd: int, xsettingsd_pid: int) -> int:
os.set_inheritable(theme_fd, True)
os.set_inheritable(conf_fd, True)
pid = os.fork()
if pid != 0:
os.close(theme_fd)
return pid
try:
while True:
gtk_theme = read_theme(theme_fd)
write_merged_settings(conf_fd, gtk_theme)
os.kill(xsettingsd_pid, signal.SIGHUP)
except KeyboardInterrupt:
os._exit(0)
def kill_and_wait(pid: int) -> None:
os.kill(pid, signal.SIGTERM)
os.kill(pid, signal.SIGCONT)
try:
os.waitpid(pid, 0)
except ChildProcessError as e:
print(e)
pass
def main():
xsettingsd = None
gsettings_pid = None
reconf_pid = None
conf_fd, conf_name = tempfile.mkstemp(prefix="xsettings", suffix=".conf")
try:
notify_fd, gsettings_pid = start_gsettings_listener()
gtk_theme = read_theme(notify_fd)
write_merged_settings(conf_fd, gtk_theme)
xsettingsd = start_xsettingsd(conf_name, sys.argv[1:])
reconf_pid = start_reconfigurator(notify_fd, conf_fd, xsettingsd.pid)
pid, _ = os.wait()
except KeyboardInterrupt:
pid = None
pass
finally:
os.close(conf_fd)
os.unlink(conf_name)
if xsettingsd and pid != xsettingsd.pid:
xsettingsd.terminate()
if gsettings_pid and pid != gsettings_pid:
kill_and_wait(gsettings_pid)
if reconf_pid and pid != reconf_pid:
kill_and_wait(reconf_pid)
if __name__ == "__main__":
main()
|
