[+] update platform

1. add few random scripts;
  1.1. systemd_gtk, oom_firefox
    are vibe coded initially;
    oom_firefox is useful,
    but migrated to ps subprocess
    manually;
    since psutils is very inefficient,
    puts too much load on CPU;
This commit is contained in:
Siarhei Siniak 2025-12-01 13:56:57 +03:00
parent b50468154f
commit 7d5868345b
4 changed files with 763 additions and 1 deletions

@ -120,11 +120,16 @@ dotfiles_put_platform:
GPG_RECIPIENTS_ARGS ?= -r 891382BEBFEFFC6729837400DA0B6C15FBB70FC9
dotfiles_fetch_platform:
mkdir -p platform_dotfiles/$(PLATFORM)
mkdir -p platform_dotfiles_gpg/$(PLATFORM)
tar -cvf - \
/etc/udev/rules.d/ \
~/.local/bin/oom_firefox \
~/.local/bin/systemd_gtk \
~/.local/bin/gnome-shortcuts-macbook-air \
/usr/local/bin \
| tar -xvf - -C platform_dotfiles/$(PLATFORM)
dotfiles_fetch_platform_gpg:
mkdir -p platform_dotfiles_gpg/$(PLATFORM)
tar -h -cvf - \
~/.sway/config.d \
~/.config/commands-status.json \

@ -0,0 +1,13 @@
#!/usr/bin/bash
commands gnome-shortcuts \
-a \
'powersave' \
'commands desktop-services --cpufreq-action powersave' \
'<Shift><Alt>1'
commands gnome-shortcuts \
-a \
'performance' \
'commands desktop-services --cpufreq-action performance' \
'<Shift><Alt>2'

@ -0,0 +1,414 @@
#!/usr/bin/env python3
import psutil
import logging
import os
import signal
import subprocess
import threading
import time
import argparse
import re
import sys
from prompt_toolkit.application import Application
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.layout import Layout, HSplit, FloatContainer, Float
from prompt_toolkit.layout.containers import Window
from prompt_toolkit.layout.controls import FormattedTextControl
from prompt_toolkit.widgets import TextArea, Frame, Dialog, Button, Label
from prompt_toolkit.styles import Style
from typing import (TypedDict,)
logger = logging.getLogger(__name__)
__version__ = "0.6.1"
__created__ = "2025-11-21"
# — Helper for cgroup / slice matching —
def get_cgroup_path(pid):
try:
with open(f"/proc/{pid}/cgroup", "r") as f:
for line in f:
parts = line.strip().split(":", 2)
if len(parts) == 3:
return parts[2]
except Exception:
logger.exception('')
return None
return None
def slice_matches(cpath, target_slice):
if not cpath or not target_slice:
return False
comps = cpath.strip("/").split("/")
tgt = target_slice.lower()
for comp in comps:
name = comp.lower()
if name.endswith(".slice"):
name = name[:-6]
if tgt == name:
return True
return False
# — Memorymanagement logic —
class get_firefox_procs_ps_t:
class res_t:
class entry_t(TypedDict):
rss: int
pid: int
ppid: int
cgroup: str
cmd: str
def get_firefox_procs_ps(slice_name=None) -> list[get_firefox_procs_ps_t.res_t.entry_t]:
lines = subprocess.check_output([
'ps', '-ax', '-o', 'rss,pid,ppid,cgroup,cmd'
]).decode('utf-8').splitlines()[1:]
entries : list[dict[str, str]] = []
for line in lines:
r = re.compile(r'^\s*(\d+)\s+(\d+)\s+(\d+)\s+([^\s]+)\s+(.*)$')
# print([r, line])
match = r.match(line)
assert match
entry = dict(
rss=int(match[1]),
pid=int(match[2]),
ppid=int(match[3]),
cgroup=match[4],
cmd=match[5],
)
if not slice_name is None:
if not slice_name in entry['cgroup']:
continue
entries.append(entry)
return entries
def get_firefox_procs(slice_name=None):
procs = []
for p in psutil.process_iter(['pid', 'name', 'cmdline', 'memory_info']):
try:
name = p.info['name']
cmd = p.info['cmdline']
if not cmd:
continue
if 'firefox' not in name and not (cmd and 'firefox' in cmd[0]):
continue
if slice_name:
cpath = get_cgroup_path(p.pid)
if not slice_matches(cpath, slice_name):
continue
procs.append(p)
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
return procs
def total_rss_mb(procs: list['get_firefox_procs_ps_t.res_t.entry_t']):
total = 0
for p in procs:
try:
# total += p.memory_info().rss
total += p['rss']
except Exception:
logger.exception('')
pass
return total / (1024 * 1024)
def is_main_firefox(p):
try:
# for arg in p.cmdline():
for arg in p['cmd'].split():
if "contentproc" in arg:
return False
return True
except Exception:
logger.exception('')
return False
def kill_prioritized(
procs: list['get_firefox_procs_ps_t.res_t.entry_t'],
to_free_mb,
low_priority_pids
):
candidates = []
for p in procs:
if is_main_firefox(p):
continue
try:
# rss_mb = p.memory_info().rss / (1024 * 1024)
rss_mb = p['rss'] / (1024 * 1024)
candidates.append((p, rss_mb))
except Exception:
logger.exception('')
continue
candidates.sort(key=lambda x: ((x[0].pid in low_priority_pids), -x[1]))
freed = 0.0
killed = []
for p, rss in candidates:
if freed >= to_free_mb:
break
try:
os.kill(p.pid, signal.SIGTERM)
killed.append(p.pid)
freed += rss
except Exception as e:
logger.exception(f"Error killing pid {p.pid}")
# print(f"Error killing pid {p.pid}: {e}", file=sys.stderr)
return killed, freed
# — systemd-run logic —
def launch_firefox_with_limits(base_cmd, memory_high, swap_max, extra_args, unit_name):
cmd = [
"systemd-run",
"--user",
"--scope",
"-p", f"MemoryHigh={int(memory_high)}M",
]
if swap_max is not None:
cmd += ["-p", f"MemorySwapMax={int(swap_max)}M"]
if unit_name:
cmd += ["--unit", unit_name]
cmd += base_cmd
cmd += extra_args
devnull = subprocess.DEVNULL
proc = subprocess.Popen(cmd, stdin=devnull, stdout=devnull, stderr=devnull)
print("Launched Firefox via systemd-run, PID:", proc.pid, file=sys.stderr)
return proc
# — Main + TUI + Monitoring —
def main():
logging.basicConfig(level=logging.INFO)
parser = argparse.ArgumentParser(description="Firefox memory manager with slice + graceful shutdown")
parser.add_argument("--max-mb", type=float, required=True,
help="Memory threshold in MB (used for killing logic & MemoryHigh)")
parser.add_argument("--kill-percent", type=float, default=70.0,
help="If over max, kill until usage ≤ this percent of max")
parser.add_argument("--swap-max-mb", type=float, default=None,
help="MemorySwapMax (MB) for the systemd scope")
parser.add_argument("--interval", type=float, default=1.0,
help="Monitoring interval in seconds")
parser.add_argument("--slice", type=str, default=None,
help="Only monitor Firefox processes in this systemd slice")
parser.add_argument("--unit-name", type=str, default="firefox-limited",
help="Name for systemd transient unit")
parser.add_argument("--firefox-extra", action="append", default=[],
help="Extra CLI args to pass to Firefox (can repeat)")
parser.add_argument("firefox_cmd", nargs=argparse.REMAINDER,
help="Firefox command + args (if launching it)")
args = parser.parse_args()
low_priority_pids = set()
body = TextArea(focusable=False, scrollbar=True)
terminate_flag = threading.Event()
lock = threading.Lock()
firefox_proc = None
def terminate():
terminate_flag.set()
def stop():
with lock:
if firefox_proc:
try:
firefox_proc.terminate()
firefox_proc.wait(timeout=5)
except Exception:
logger.exception('')
try:
firefox_proc.kill()
except Exception:
logger.exception('')
pass
# app.exit()
# signal.signal(signal.SIGINT, lambda s, f: terminate())
# signal.signal(signal.SIGTERM, lambda s, f: terminate())
def refresh_body():
nonlocal firefox_proc
with lock:
procs = get_firefox_procs_ps(slice_name=args.slice)
total = total_rss_mb(procs)
limit = args.max_mb
kill_to = args.kill_percent / 100.0 * limit
lines = [
f"Firefox RSS (slice={args.slice}): {total:.1f} MB",
f"Threshold (max): {limit:.1f} MB",
f"Killto target: {kill_to:.1f} MB ({args.kill_percent}%)",
f"Lowpriority PIDs: {sorted(low_priority_pids)}"
]
if total > limit:
to_free = total - kill_to
killed, freed = kill_prioritized(procs, to_free, low_priority_pids)
lines.append(f"Killed: {killed}")
lines.append(f"Freed ≈ {freed:.1f} MB")
else:
lines.append("Within limit — no kill")
if firefox_proc and firefox_proc.poll() is not None:
print("Firefox died — restarting …", file=sys.stderr)
firefox_proc = launch_firefox_with_limits(
args.firefox_cmd,
memory_high=args.max_mb,
swap_max=args.swap_max_mb,
extra_args=args.firefox_extra,
unit_name=args.unit_name
)
body.text = "\n".join(lines)
dialog_float = [None]
root_floats = []
def open_pid_dialog():
ta = TextArea(text="", multiline=True, scrollbar=True)
def on_ok():
txt = ta.text
for m in re.finditer(r"\((\d+)\)", txt):
low_priority_pids.add(int(m.group(1)))
close_dialog()
refresh_body()
def on_cancel():
close_dialog()
dialog = Dialog(
title="Enter lowpriority PIDs",
body=ta,
buttons=[Button(text="OK", handler=on_ok), Button(text="Cancel", handler=on_cancel)],
width=60,
modal=True
)
f = Float(content=dialog, left=2, top=2)
dialog_float[0] = f
root_floats.append(f)
app.layout.focus(ta)
def open_message(title, message):
def on_close():
close_dialog()
dialog = Dialog(
title=title,
body=Label(text=message),
buttons=[Button(text="Close", handler=on_close)],
width=50,
modal=True
)
f = Float(content=dialog, left=4, top=4)
dialog_float[0] = f
root_floats.append(f)
app.layout.focus(dialog)
def close_dialog():
f = dialog_float[0]
if f in root_floats:
root_floats.remove(f)
dialog_float[0] = None
app.layout.focus(body)
kb = KeyBindings()
@kb.add("q")
def _(event):
terminate()
@kb.add("m")
def _(event):
open_pid_dialog()
@kb.add("h")
def _(event):
open_message("Help", "Keys: m=add PIDs, s=settings, a=about, q=quit")
@kb.add("s")
def _(event):
open_message("Settings",
f"max_mb = {args.max_mb}\n"
f"kill_percent = {args.kill_percent}\n"
f"slice = {args.slice}\n"
f"swap_max_mb = {args.swap_max_mb}\n"
f"extra firefox args = {args.firefox_extra}")
@kb.add("a")
def _(event):
open_message("About", f"Version {__version__}\nCreated {__created__}")
root = FloatContainer(
content=HSplit([
Frame(body, title="Firefox Memory Manager"),
Window(height=1, content=FormattedTextControl("q=quit, m=PID, h=help, s=setting, a=about"))
]),
floats=root_floats,
modal=True
)
style = Style.from_dict({
"frame.border": "ansicyan",
"dialog.body": "bg:#444444",
"dialog": "bg:#888888",
})
app = Application(
layout=Layout(root),
key_bindings=kb,
style=style,
full_screen=True,
refresh_interval=args.interval,
)
if args.firefox_cmd:
firefox_proc = launch_firefox_with_limits(
args.firefox_cmd,
memory_high=args.max_mb,
swap_max=args.swap_max_mb, # **fixed here**
extra_args=args.firefox_extra,
unit_name=args.unit_name
)
def monitor_loop():
nonlocal firefox_proc
while not terminate_flag.is_set():
refresh_body()
time.sleep(args.interval)
# stop()
terminate_flag = threading.Event()
t = threading.Thread(target=monitor_loop, daemon=True)
t.start()
# refresh_body()
app.run(handle_sigint=True) # from prompttoolkit API :contentReference[oaicite:0]{index=0}
t.join()
stop()
if __name__ == "__main__":
main()

@ -0,0 +1,330 @@
#!/usr/bin/env python3
# vi: filetype=python
import gi
gi.require_version("Gtk", "4.0")
from gi.repository import Gtk, Gio, GLib, GObject
import subprocess
import shlex
import threading
import uuid
import argparse
import logging
# CLI / Logging
parser = argparse.ArgumentParser(description="Systemd Scope Manager (GTK4 ColumnView)")
parser.add_argument("--log-level", "-l", choices=["DEBUG","INFO","WARNING","ERROR","CRITICAL"], default="INFO")
parser.add_argument("--values-mode", "-m", choices=["raw","human"], default="human",
help="Display memory / CPU values raw or humanreadable")
args = parser.parse_args()
logging.basicConfig(level=getattr(logging, args.log_level.upper()),
format="%(asctime)s %(name)s %(levelname)s: %(message)s")
logger = logging.getLogger(__name__)
# Helpers
def human_bytes(value: int) -> str:
for unit in ("B","KB","MB","GB","TB"):
if value < 1024:
return f"{value:.1f}{unit}"
value /= 1024
return f"{value:.1f}PB"
def human_time(nsec: int) -> str:
sec = nsec / 1_000_000_000
if sec < 60:
return f"{sec:.1f}s"
minutes = sec / 60
if minutes < 60:
return f"{minutes:.1f}m"
return f"{minutes/60:.1f}h"
def run_systemctl_show(unit: str) -> dict:
cmd = [
"systemctl", "--user", "show", unit,
"--property=MemoryCurrent,MemorySwapCurrent,CPUUsageNSec,ActiveState,Restart"
]
try:
res = subprocess.run(cmd, capture_output=True, text=True, check=True)
props = {}
for line in res.stdout.splitlines():
if "=" in line:
k, v = line.split("=", 1)
props[k.strip()] = v.strip()
return props
except Exception:
logger.exception("systemctl show failed for %s", unit)
return {}
# Data row
class ScopeRow(GObject.Object):
__gtype_name__ = "ScopeRow"
unit = GObject.Property(type=str)
cli = GObject.Property(type=str)
mem = GObject.Property(type=str)
swap = GObject.Property(type=str)
cpu = GObject.Property(type=str)
state= GObject.Property(type=str)
def __init__(self, unit, cli, mem, swap, cpu, state):
super().__init__()
self.unit = unit
self.cli = cli
self.mem = mem
self.swap = swap
self.cpu = cpu
self.state = state
# Main Window
class ScopeManagerWindow(Gtk.ApplicationWindow):
def __init__(self, app):
super().__init__(application=app, title="Systemd Scope Manager")
self.set_default_size(1000, 500)
self.values_mode = args.values_mode
self.scopes = {}
self.model = Gio.ListStore(item_type=ScopeRow)
self.sel = Gtk.SingleSelection.new(self.model)
self.view = Gtk.ColumnView.new(self.sel)
self.view.set_reorderable(True)
self.view.set_show_row_separators(True)
cols = [
("Unit", "unit"),
("CLI", "cli"),
("Memory", "mem"),
("Swap", "swap"),
("CPU", "cpu"),
("State", "state"),
]
for title, prop_name in cols:
factory = Gtk.SignalListItemFactory()
factory.connect("setup", self._factory_setup_label)
factory.connect("bind", self._make_factory_bind(prop_name))
col = Gtk.ColumnViewColumn()
col.set_title(title)
col.set_factory(factory)
col.set_resizable(True)
col.set_expand(True)
self.view.append_column(col)
# Actions column
action_factory = Gtk.SignalListItemFactory()
action_factory.connect("setup", self._factory_setup_actions)
action_factory.connect("bind", self._factory_bind_actions)
act_col = Gtk.ColumnViewColumn()
act_col.set_title("⋮")
act_col.set_factory(action_factory)
act_col.set_resizable(False)
act_col.set_expand(False)
self.view.append_column(act_col)
# Input area
self.cmd_entry = Gtk.Entry()
self.cmd_entry.set_placeholder_text("Command to run in new scope")
run_btn = Gtk.Button(label="Run")
run_btn.connect("clicked", self.on_run_clicked)
prop_btn = Gtk.Button(label="Edit Property")
prop_btn.connect("clicked", self.on_edit_property)
input_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6)
input_box.append(self.cmd_entry)
input_box.append(run_btn)
input_box.append(prop_btn)
scrolled = Gtk.ScrolledWindow()
scrolled.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
scrolled.set_propagate_natural_width(True)
scrolled.set_propagate_natural_height(True)
scrolled.set_child(self.view)
vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6)
vbox.set_vexpand(True)
vbox.append(input_box)
vbox.append(scrolled)
self.set_child(vbox)
GLib.timeout_add_seconds(2, self.refresh_scopes)
def _factory_setup_label(self, factory, list_item):
lbl = Gtk.Label()
list_item.set_child(lbl)
def _make_factory_bind(self, prop_name):
def bind(factory, list_item):
row = list_item.get_item()
lbl = list_item.get_child()
lbl.set_text(getattr(row, prop_name))
row.connect(f"notify::{prop_name}", lambda obj, pspec: lbl.set_text(getattr(obj, prop_name)))
return bind
def _factory_setup_actions(self, factory, list_item):
btn = Gtk.MenuButton()
btn.set_icon_name("open-menu-symbolic")
# style class if needed:
btn.get_style_context().add_class("flat")
list_item.set_child(btn)
def _factory_bind_actions(self, factory, list_item):
row = list_item.get_item()
btn = list_item.get_child()
menu = Gio.Menu()
menu.append("Stop", f"app.stop_{row.unit}")
menu.append("Restart", f"app.restart_{row.unit}")
menu.append("Toggle AutoRestart", f"app.toggle_{row.unit}")
btn.set_menu_model(menu)
# create actions
self._ensure_row_actions(row)
def _ensure_row_actions(self, row):
unit = row.unit
# Stop
act_stop = Gio.SimpleAction.new(f"stop_{unit}", None)
act_stop.connect("activate", lambda a, v, r=row: self.menu_action("Stop", r))
self.add_action(act_stop)
# Restart
act_restart = Gio.SimpleAction.new(f"restart_{unit}", None)
act_restart.connect("activate", lambda a, v, r=row: self.menu_action("Restart", r))
self.add_action(act_restart)
# Toggle
act_toggle = Gio.SimpleAction.new(f"toggle_{unit}", None)
act_toggle.connect("activate", lambda a, v, r=row: self.menu_action("Toggle AutoRestart", r))
self.add_action(act_toggle)
def on_run_clicked(self, button):
cmd = self.cmd_entry.get_text().strip()
if not cmd:
return
unit = f"app-{uuid.uuid4().int >> 64}.scope"
argv = ["systemd-run", "--user", "--scope", "--unit", unit,
"-p", "MemoryAccounting=yes", "-p", "CPUAccounting=yes"] + shlex.split(cmd)
logger.info("Starting scope: %s", " ".join(argv))
def worker():
try:
p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
out, err = p.communicate()
if out:
logger.debug("systemd-run stdout [%s]: %s", unit, out.strip())
if err:
logger.debug("systemd-run stderr [%s]: %s", unit, err.strip())
except Exception:
logger.exception("Failed to run systemd-run for %s", unit)
threading.Thread(target=worker, daemon=True).start()
self.scopes[unit] = cmd
self.cmd_entry.set_text("")
def on_edit_property(self, button):
idx = self.sel.get_selected()
if idx < 0:
return
row = self.model.get_item(idx)
dialog = Gtk.Dialog(transient_for=self, modal=True, title="Set systemd property")
content = dialog.get_content_area()
grid = Gtk.Grid(row_spacing=6, column_spacing=6, margin=10)
content.append(grid)
lbl1 = Gtk.Label(label="Property (e.g. MemoryMax):")
entry1 = Gtk.Entry()
lbl2 = Gtk.Label(label="Value (e.g. 512M):")
entry2 = Gtk.Entry()
runtime_chk = Gtk.CheckButton(label="Runtime only")
grid.attach(lbl1, 0,0,1,1)
grid.attach(entry1, 1,0,1,1)
grid.attach(lbl2, 0,1,1,1)
grid.attach(entry2, 1,1,1,1)
grid.attach(runtime_chk,0,2,2,1)
dialog.add_button("OK", Gtk.ResponseType.OK)
dialog.add_button("Cancel", Gtk.ResponseType.CANCEL)
dialog.show()
resp = dialog.run()
if resp == Gtk.ResponseType.OK:
prop = entry1.get_text().strip()
val = entry2.get_text().strip()
rt = runtime_chk.get_active()
dialog.destroy()
self.set_property(row.unit, prop, val, rt)
else:
dialog.destroy()
def set_property(self, unit, prop, val, runtime_flag):
cmd = ["systemctl", "--user", "set-property"]
if runtime_flag:
cmd.append("--runtime")
cmd += [unit, f"{prop}={val}"]
logger.info("Setting %s=%s on %s (runtime=%s)", prop, val, unit, runtime_flag)
try:
res = subprocess.run(cmd, capture_output=True, text=True)
if res.stdout:
logger.debug("set-property stdout: %s", res.stdout.strip())
if res.stderr:
logger.debug("set-property stderr: %s", res.stderr.strip())
except Exception:
logger.exception("Failed set-property for %s", unit)
def menu_action(self, label, row):
unit = row.unit
try:
if label == "Stop":
subprocess.run(["systemctl","--user","stop",unit])
elif label == "Restart":
subprocess.run(["systemctl","--user","restart",unit])
elif label == "Toggle AutoRestart":
props = run_systemctl_show(unit)
current = props.get("Restart","no")
new = "no" if current != "no" else "always"
subprocess.run(["systemctl","--user","set-property",unit,f"Restart={new}"])
except Exception:
logger.exception("Action %s failed on %s", label, unit)
def refresh_scopes(self):
self.model.remove_all()
for unit, cli in self.scopes.items():
props = run_systemctl_show(unit)
mem = int(props.get("MemoryCurrent", "0"))
swap = int(props.get("MemorySwapCurrent","0"))
cpu = int(props.get("CPUUsageNSec", "0"))
state = props.get("ActiveState","unknown")
if self.values_mode == "human":
mem_s = human_bytes(mem)
swap_s = human_bytes(swap)
cpu_s = human_time(cpu)
else:
mem_s = str(mem)
swap_s = str(swap)
cpu_s = str(cpu)
row = ScopeRow(unit, cli, mem_s, swap_s, cpu_s, state)
self.model.append(row)
logger.debug("Refreshed %d scopes", self.model.get_n_items())
return True
# Application
class ScopeManagerApp(Gtk.Application):
def __init__(self):
super().__init__(application_id="org.systemd.ScopeManager")
self.connect("activate", self.on_activate)
def on_activate(self, app):
win = ScopeManagerWindow(self)
win.present()
def main():
app = ScopeManagerApp()
return app.run()
if __name__ == "__main__":
import sys
sys.exit(main())