gib/Panama
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add a Storage page

Browse Source 
Nothing showed what was using the drive, and removable media was handled
by a tray helper with no surface in Settings at all.

One scroll rather than tabs: space above, the device below. Every other
settings page is a scrolling card stack, and a tab would not be
deep-linkable from the launcher command or from search.

Three things the page has to get right, each now pinned by a contract,
because each is a way it could quietly lie. / and /home are one btrfs
filesystem sharing one pool of free space, and a page that copies df
shows double the free space that exists. zram is a block device and is
not storage; counting it as a drive overstates this machine by 8 GB.
Unmount and eject refuse anything not on a removable drive, because the
UI is what asks and a UI can be wrong.

The cheap read -- layout, usage, health -- runs when the page opens, at
around 90ms. Measuring what is filling the drive means walking every
file, so it happens on request and says so rather than showing an empty
list that reads as "nothing here".

Partitioning and formatting are deliberately absent. A settings pane is
the wrong place to put erasing a disk two clicks deep; the page opens
GNOME Disks for that.

Adding the page found a fourth hard-coded page list in ShellState. A
page missing from it does not error -- openSettings() falls back to
"home", so the launcher opens the wrong page and logs nothing. A
registry contract now holds the three lists together.

Claude-Session: https://claude.ai/code/session_01BRvzt4H8XXLPVH5MyYdk9L
This commit is contained in:
Gabriel Brown
2026-08-19 10:52:10 -04:00
parent a68e4f6dcd
commit 99433c0e8e
12 changed files with 1141 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/dot/quickshell/scripts/panama-disks
Executable
+372
View File
@@ -0,0 +1,372 @@
#!/usr/bin/env python3
"""Read storage layout, usage, and drive health for Panama's Storage page.
Two boundaries, deliberately separate:
snapshot topology, usage, health, removable media. Cheap -- lsblk and a
single udisks call -- so the page can open with it.
scan what is actually filling the drive. Expensive: measuring a folder
means walking it, and this machine has a 1.2 TiB Steam library.
The page asks for this on demand and remembers the answer.
unmount PATH / eject PATH removable media only, by explicit request.
Deliberately absent: partitioning and formatting. A settings pane is the wrong
place to hand someone a way to erase a disk in two clicks; GNOME Disks is one
button away on the page for that.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import time
# Bounded so a pathological tree cannot hang the page. A folder that cannot be
# measured in this long is reported as unmeasured rather than silently omitted,
# because a missing row reads as "this folder is small".
SCAN_TIMEOUT_SECONDS = 90
# Where "what is filling my drive" actually gets answered. Ordered so the most
# likely culprits are measured first; the walk stops when the budget runs out.
SCAN_TARGETS = [
("Steam library", "~/.local/share/Steam"),
("Documents", "~/Documents"),
("Downloads", "~/Downloads"),
("Videos", "~/Videos"),
("Pictures", "~/Pictures"),
("Music", "~/Music"),
("Flatpak applications", "~/.var/app"),
("Caches", "~/.cache"),
("Trash", "~/.local/share/Trash"),
]
class BoundaryError(RuntimeError):
"""A user-visible validation or command failure."""
def run(command: list[str], timeout: float = 15.0) -> str:
try:
completed = subprocess.run(command, check=False, capture_output=True,
text=True, timeout=timeout)
except (OSError, subprocess.TimeoutExpired) as error:
raise BoundaryError(f"{command[0]} did not answer") from error
if completed.returncode != 0:
raise BoundaryError(completed.stderr.strip() or f"{command[0]} failed")
return completed.stdout
def lsblk() -> list[dict]:
"""The block device tree.
PANAMA_DISKS_LSBLK substitutes a recorded tree for the real one. It exists
so the refusal rules below can be tested against a removable drive that is
not plugged in -- the alternative is a test that runs unmount against
whatever is actually mounted, which is not a test anyone should write.
"""
fixture = os.environ.get("PANAMA_DISKS_LSBLK")
if fixture:
try:
return json.loads(Path(fixture).read_text(encoding="utf-8"))["blockdevices"]
except (OSError, json.JSONDecodeError, KeyError) as error:
raise BoundaryError("could not read the recorded block device layout") from error
fields = ("NAME,PATH,SIZE,TYPE,FSTYPE,MOUNTPOINTS,MODEL,SERIAL,RM,HOTPLUG,"
"ROTA,FSSIZE,FSUSED,FSAVAIL")
try:
return json.loads(run(["lsblk", "-J", "-b", "-o", fields]))["blockdevices"]
except (json.JSONDecodeError, KeyError) as error:
raise BoundaryError("could not read the block device layout") from error
def udisks_objects() -> dict:
"""Every udisks object in one call, or nothing if udisks is not running.
Health is a nice-to-have: a machine without udisks still gets its layout and
usage, which is most of the page. Failing the whole snapshot because a
temperature is unavailable would be the wrong trade.
"""
if not shutil.which("busctl"):
return {}
try:
raw = run(["busctl", "--system", "--json=short", "call",
"org.freedesktop.UDisks2", "/org/freedesktop/UDisks2",
"org.freedesktop.DBus.ObjectManager", "GetManagedObjects"])
payload = json.loads(raw)
except (BoundaryError, json.JSONDecodeError):
return {}
# busctl wraps the reply as {"type": "...", "data": [ {path: {iface: {prop: {"type","data"}}}} ]}
data = payload.get("data") or []
return data[0] if data else {}
def unwrap(value):
"""busctl encodes every value as {"type": t, "data": v}."""
if isinstance(value, dict) and "data" in value and "type" in value:
return value["data"]
return value
def drive_health(objects: dict) -> dict[str, dict]:
"""Health keyed by drive serial, which is the only stable join to lsblk."""
health: dict[str, dict] = {}
for path, interfaces in objects.items():
if "/drives/" not in path:
continue
drive = {key: unwrap(value)
for key, value in (interfaces.get("org.freedesktop.UDisks2.Drive") or {}).items()}
serial = str(drive.get("Serial") or "")
if not serial:
continue
entry = {
"ejectable": bool(drive.get("Ejectable")),
"removable": bool(drive.get("Removable")),
"temperatureC": None,
"powerOnHours": None,
"healthy": None,
"warnings": [],
"selfTest": "",
}
nvme = {key: unwrap(value) for key, value in
(interfaces.get("org.freedesktop.UDisks2.NVMe.Controller") or {}).items()}
ata = {key: unwrap(value) for key, value in
(interfaces.get("org.freedesktop.UDisks2.Drive.Ata") or {}).items()}
if nvme:
kelvin = nvme.get("SmartTemperature")
if isinstance(kelvin, (int, float)) and kelvin > 0:
entry["temperatureC"] = round(kelvin - 273.15, 1)
hours = nvme.get("SmartPowerOnHours")
if isinstance(hours, (int, float)):
entry["powerOnHours"] = int(hours)
warnings = nvme.get("SmartCriticalWarning") or []
entry["warnings"] = [str(item) for item in warnings]
entry["healthy"] = not entry["warnings"]
entry["selfTest"] = str(nvme.get("SmartSelftestStatus") or "")
elif ata:
kelvin = ata.get("SmartTemperature")
if isinstance(kelvin, (int, float)) and kelvin > 0:
entry["temperatureC"] = round(kelvin - 273.15, 1)
seconds = ata.get("SmartPowerOnSeconds")
if isinstance(seconds, (int, float)) and seconds > 0:
entry["powerOnHours"] = int(seconds // 3600)
failing = ata.get("SmartFailing")
if isinstance(failing, bool):
entry["healthy"] = not failing
entry["warnings"] = ["failing"] if failing else []
entry["selfTest"] = str(ata.get("SmartSelftestStatus") or "")
health[serial] = entry
return health
def walk(node: dict, depth: int = 0):
yield node, depth
for child in node.get("children") or []:
yield from walk(child, depth + 1)
def mountpoints(node: dict) -> list[str]:
return [point for point in (node.get("mountpoints") or []) if point and point != "[SWAP]"]
def snapshot() -> dict:
tree = lsblk()
health = drive_health(udisks_objects())
drives = []
swap = []
filesystems: dict[str, dict] = {}
for root in tree:
if root.get("type") != "disk":
continue
# zram is compressed swap in RAM. It is a block device and it is not
# storage; listing it as a drive would be actively misleading about how
# much of this machine is disk.
if str(root.get("name", "")).startswith("zram"):
swap.append({
"name": root.get("name"),
"sizeBytes": root.get("size") or 0,
"kind": "zram",
})
continue
serial = str(root.get("serial") or "")
drive_health_entry = health.get(serial, {})
partitions = []
encrypted = False
for node, depth in walk(root):
if depth == 0:
continue
if node.get("fstype") == "crypto_LUKS":
encrypted = True
if node.get("type") in ("part", "crypt"):
partitions.append({
"name": node.get("name"),
"path": node.get("path"),
"sizeBytes": node.get("size") or 0,
"fstype": node.get("fstype") or "",
"type": node.get("type"),
"mountpoints": mountpoints(node),
})
for point in mountpoints(node):
if node.get("fssize") is None:
continue
# Keyed by device: btrfs subvolumes mounted at / and /home are
# ONE filesystem with one pool of free space. Reporting them as
# two independent bars, which is what df does, doubles the free
# space on screen.
key = str(node.get("path"))
entry = filesystems.setdefault(key, {
"device": node.get("path"),
"fstype": node.get("fstype") or "",
"sizeBytes": int(node.get("fssize") or 0),
"usedBytes": int(node.get("fsused") or 0),
"availBytes": int(node.get("fsavail") or 0),
"mountpoints": [],
"encrypted": node.get("type") == "crypt",
})
entry["mountpoints"].append(point)
entry["mountpoints"].sort(key=lambda mount: (mount != "/", mount))
drives.append({
"name": root.get("name"),
"path": root.get("path"),
"model": (root.get("model") or "").strip() or str(root.get("name")),
"serial": serial,
"sizeBytes": root.get("size") or 0,
"rotational": bool(root.get("rota")),
"removable": bool(root.get("rm")) or bool(root.get("hotplug"))
or bool(drive_health_entry.get("removable")),
"ejectable": bool(drive_health_entry.get("ejectable")),
"encrypted": encrypted,
"partitions": partitions,
"temperatureC": drive_health_entry.get("temperatureC"),
"powerOnHours": drive_health_entry.get("powerOnHours"),
"healthy": drive_health_entry.get("healthy"),
"warnings": drive_health_entry.get("warnings", []),
"selfTest": drive_health_entry.get("selfTest", ""),
})
ordered = sorted(filesystems.values(),
key=lambda entry: (0 if "/" in entry["mountpoints"] else 1,
entry["mountpoints"][0] if entry["mountpoints"] else ""))
return {"drives": drives, "filesystems": ordered, "swap": swap}
def measure(path: Path, budget: float) -> tuple[int | None, float]:
"""Bytes used by a folder, and what is left of the time budget."""
if not path.is_dir():
return None, budget
started = time.monotonic()
try:
completed = subprocess.run(["du", "-sxb", str(path)], check=False,
capture_output=True, text=True, timeout=budget)
except (OSError, subprocess.TimeoutExpired):
# A folder that ran out of time has consumed the whole budget by
# definition; the caller stops rather than starting another walk.
return None, 0.0
left = max(budget - (time.monotonic() - started), 0.0)
if completed.returncode != 0:
return None, left
match = re.match(r"^(\d+)", completed.stdout)
return (int(match.group(1)) if match else None), left
def container_reclaimable() -> dict | None:
"""Container images nothing is using. Absent when podman is not installed."""
if not shutil.which("podman"):
return None
try:
raw = run(["podman", "system", "df", "--format", "json"], timeout=20.0)
entries = json.loads(raw)
except (BoundaryError, json.JSONDecodeError):
return None
for entry in entries if isinstance(entries, list) else []:
if not str(entry.get("Type", "")).lower().startswith("image"):
continue
# Raw* are integers; Size and Reclaimable are display strings like
# "10.8GB" and "5.31GB (49%)". Reading the display strings is how this
# first crashed, so only the raw fields are trusted, and an old podman
# that lacks them reports nothing rather than a wrong number.
total = entry.get("RawSize")
reclaimable = entry.get("RawReclaimable")
if not isinstance(total, int) or not isinstance(reclaimable, int):
return None
return {"totalBytes": total, "reclaimableBytes": reclaimable}
return None
def scan() -> dict:
folders = []
remaining = float(SCAN_TIMEOUT_SECONDS)
truncated = False
for label, target in SCAN_TARGETS:
path = Path(os.path.expanduser(target))
if remaining <= 1.0:
truncated = True
break
size, remaining = measure(path, remaining)
if size is None:
continue
folders.append({"label": label, "path": str(path), "bytes": size})
folders.sort(key=lambda item: item["bytes"], reverse=True)
return {
"folders": folders,
"truncated": truncated,
"containers": container_reclaimable(),
}
def removable_device(path: str) -> dict:
"""Resolve a device path, refusing anything that is not removable.
Unmounting the root filesystem is not a feature. The check is on the device
rather than the button because the caller is a UI that can be wrong.
"""
if not re.fullmatch(r"/dev/[A-Za-z0-9/_-]+", path or ""):
raise BoundaryError("That is not a device path.")
for root in lsblk():
for node, _ in walk(root):
if node.get("path") != path:
continue
top = root
if not (bool(top.get("rm")) or bool(top.get("hotplug"))):
raise BoundaryError("That drive is not removable.")
return node
raise BoundaryError("No such device.")
def main(arguments: list[str]) -> int:
try:
if arguments == ["snapshot"]:
print(json.dumps(snapshot(), separators=(",", ":")))
elif arguments == ["scan"]:
print(json.dumps(scan(), separators=(",", ":")))
elif len(arguments) == 2 and arguments[0] in ("unmount", "eject"):
removable_device(arguments[1])
action = "unmount" if arguments[0] == "unmount" else "power-off"
flag = "-b" if arguments[0] == "unmount" else "-b"
run(["udisksctl", action, flag, arguments[1]], timeout=30.0)
else:
raise BoundaryError(
"Usage: panama-disks snapshot | scan | unmount DEVICE | eject DEVICE")
except BoundaryError as error:
print(str(error), file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))