From fa1de7fe16fe775d392d6894e00ceebf7dbfcd40 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Asger=20Weirs=C3=B8e?= Date: Tue, 11 Aug 2026 12:57:13 +0200 Subject: [PATCH] Add init --sync and --prune for incremental monitor setup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `init` planned every file on every run and refused if any of them existed, so attaching a new monitor could not be configured without --force — which reallocated global workspace numbers by i3's current output order and renumbered the monitors already in use. - allocate_global_starts() keeps the block each known output already holds and gives new outputs the lowest free block; blocks of inactive outputs stay reserved. - read_allocations()/parse_allocation() read those blocks back out of the generated files (inverse of render_workspaces_for_output). - --sync writes only the missing files, --prune drops configs for outputs that are gone and frees their numbers. - Files without the auto-generated header are never written or deleted, --force included. Co-Authored-By: Claude --- README.md | 32 +++- pi3/smart_workspace.py | 266 ++++++++++++++++++++++++++++++++-- pyproject.toml | 2 +- tests/test_smart_workspace.py | 253 ++++++++++++++++++++++++++++++++ 4 files changed, 535 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index e1de2bd..9686e5a 100644 --- a/README.md +++ b/README.md @@ -103,10 +103,36 @@ Useful flags: | `-n N` / `--count N` | 8 | Workspaces per output | | `--config-dir PATH` | `~/.config/i3/pi3-smart-workspace` | Where to write | | `--dry-run` | off | Print to stdout, don't touch the filesystem | -| `--force` | off | Overwrite existing files | +| `--sync` | off | Only write configs for outputs that don't have one yet | +| `--force` | off | Regenerate every file from scratch (renumbers workspaces) | +| `--prune` | off | Delete generated configs for outputs that are no longer active | -Re-run `init --force` after plugging or unplugging a monitor to -regenerate the per-output folders. +### Plugging in a new monitor + +Run: + +```sh +pi3-smart-workspace init --sync +``` + +`--sync` writes a `workspaces.conf` only for the outputs that don't have +one yet, and gives the newcomer the lowest block of global workspace +numbers that isn't already taken. The monitors you already use keep the +numbers they have, so your muscle memory survives — which a plain +`--force` regeneration would not guarantee, since that allocates numbers +by i3's current output order and can shuffle everything. + +Configs for monitors that are merely switched off are left in place and +their numbers stay reserved. Add `--prune` to delete them and free those +numbers for a new screen: + +```sh +pi3-smart-workspace init --sync --prune +``` + +Files that don't carry the `# Auto-generated by pi3-smart-workspace init` +header are never overwritten or deleted, not even with `--force` — so +anything you hand-tune in that directory is safe. ### Example config (manual) diff --git a/pi3/smart_workspace.py b/pi3/smart_workspace.py index 284cd36..1d981be 100644 --- a/pi3/smart_workspace.py +++ b/pi3/smart_workspace.py @@ -33,7 +33,7 @@ import pprint import re import sys from dataclasses import dataclass -from typing import Iterable +from typing import Iterable, Mapping import i3ipc @@ -451,6 +451,101 @@ def render_bindings(count: int, mod: str = "$mod") -> str: return "".join(out) +@dataclass(frozen=True) +class Allocation: + """A block of global workspace numbers already handed to an output. + + Attributes: + output: The output name the block belongs to. + start: First global workspace number in the block. + count: How many consecutive numbers the block covers. + """ + + output: str + start: int + count: int + + +_ALLOC_OUTPUT_RE = re.compile(r"^\s*set\s+\$out_\w+\s+(\S+)\s*$", re.MULTILINE) +_ALLOC_WS_RE = re.compile(r"^\s*set\s+\$ws_\w+\s+(\d+):\d+\s*$", re.MULTILINE) + + +def parse_allocation(config_text: str) -> Allocation | None: + """Read back the output and global-number block from a generated file. + + The inverse of :func:`render_workspaces_for_output`, used by + ``init --sync`` to keep the numbers a monitor already has instead of + reallocating them from scratch. + + Args: + config_text: Contents of a ``workspaces.conf`` previously written + by ``init``. + + Returns: + The :class:`Allocation` described by the file, or ``None`` if the + text doesn't contain both an ``$out_…`` alias and at least one + ``$ws_…`` alias (i.e. it isn't one of our generated files). + """ + output_match = _ALLOC_OUTPUT_RE.search(config_text) + numbers = [int(m.group(1)) for m in _ALLOC_WS_RE.finditer(config_text)] + if output_match is None or not numbers: + return None + return Allocation(output_match.group(1), min(numbers), len(numbers)) + + +def allocate_global_starts( + outputs: Iterable[str], + existing: Mapping[str, Allocation], + count: int, +) -> dict[str, int]: + """Decide each output's first global workspace number. + + Outputs that already hold a block of the right size keep it, so + plugging in a new monitor never renumbers the ones you already use. + New outputs get the lowest block of ``count`` free numbers. + + Blocks belonging to outputs in ``existing`` that aren't in ``outputs`` + (an unplugged monitor) stay reserved — a screen that is merely + switched off must not lose its numbers to a newcomer. Drop those + entries from ``existing`` (what ``--prune`` does) to free them. + + Args: + outputs: Names of the outputs to allocate for, in i3's order. + existing: Already-allocated blocks, keyed by output name. + count: Workspaces per output. + + Returns: + ``{output_name: global_start}`` covering every name in ``outputs``. + """ + if count < 1: + raise ValueError("count must be >= 1") + + occupied: set[int] = set() + for alloc in existing.values(): + occupied.update(range(alloc.start, alloc.start + alloc.count)) + + starts: dict[str, int] = {} + pending: list[str] = [] + for name in outputs: + alloc = existing.get(name) + if alloc is not None and alloc.count == count: + starts[name] = alloc.start + else: + if alloc is not None: + # The requested block size changed (``-n``), so the old + # block can't be kept — release it and reallocate below. + occupied.difference_update(range(alloc.start, alloc.start + alloc.count)) + pending.append(name) + + for name in pending: + start = 1 + while not occupied.isdisjoint(range(start, start + count)): + start += 1 + starts[name] = start + occupied.update(range(start, start + count)) + return starts + + @dataclass(frozen=True) class InitArgs: """Parsed CLI arguments for the ``init`` subcommand.""" @@ -459,6 +554,8 @@ class InitArgs: config_dir: pathlib.Path dry_run: bool force: bool + sync: bool + prune: bool debug: bool @@ -466,19 +563,32 @@ def _planned_files( config_dir: pathlib.Path, outputs: list[tuple[str, Rect]], count: int, + starts: Mapping[str, int] | None = None, ) -> list[tuple[pathlib.Path, str]]: """Compute the (path, content) pairs that ``init`` would write. Pure function: no filesystem or i3 access. Each output gets a folder ``//workspaces.conf``; the shared keybindings - go in ``/bindings.conf``. Output order is preserved from - ``outputs`` so global workspace numbers are deterministic. + go in ``/bindings.conf``. + + Args: + config_dir: Directory the generated files live in. + outputs: Active outputs, in i3's order. + count: Workspaces per output. + starts: Optional ``{output: global_start}`` map, typically from + :func:`allocate_global_starts`. When omitted, each output gets + a contiguous block based on its position in ``outputs`` — + deterministic, but it renumbers everything if the output order + changes. """ files: list[tuple[pathlib.Path, str]] = [ (config_dir / "bindings.conf", render_bindings(count)), ] for index, (output_name, _) in enumerate(outputs): - global_start = index * count + 1 + if starts is not None: + global_start = starts[output_name] + else: + global_start = index * count + 1 files.append( ( config_dir / output_name / "workspaces.conf", @@ -488,15 +598,73 @@ def _planned_files( return files +def _is_generated(path: pathlib.Path) -> bool: + """Return whether ``path`` carries our auto-generation header. + + Anything else is treated as hand-written and never overwritten or + deleted, not even with ``--force``. + """ + try: + return path.read_text().startswith(_AUTOGEN_HEADER) + except OSError: + return False + + +def read_allocations(config_dir: pathlib.Path) -> dict[str, Allocation]: + """Read the global-number blocks already handed out under ``config_dir``. + + Scans ``/*/workspaces.conf`` and parses each generated + file with :func:`parse_allocation`. Hand-edited or unreadable files + are skipped, so their numbers are *not* considered reserved. + + Args: + config_dir: The (possibly non-existent) init config directory. + + Returns: + ``{output_name: Allocation}``; empty if nothing has been generated + yet. + """ + allocations: dict[str, Allocation] = {} + for path in sorted(config_dir.glob("*/workspaces.conf")): + if not _is_generated(path): + continue + try: + alloc = parse_allocation(path.read_text()) + except OSError: + continue + if alloc is not None: + allocations[alloc.output] = alloc + return allocations + + +def _prune_paths(config_dir: pathlib.Path, stale: Iterable[str]) -> list[pathlib.Path]: + """Return the generated ``workspaces.conf`` files of no-longer-active outputs.""" + paths = [] + for name in stale: + path = config_dir / name / "workspaces.conf" + if path.exists() and _is_generated(path): + paths.append(path) + return paths + + def run_init(args: InitArgs) -> None: """Generate per-output i3 workspace configs for the currently-active outputs. + Three write modes: + + * default — write everything, but refuse if any target file exists. + * ``--sync`` — write only the files that are missing (the new + monitor), leaving existing outputs' numbers untouched. + * ``--force`` — regenerate everything from scratch, which reallocates + global workspace numbers by i3's current output order. + Args: args: Parsed init-subcommand arguments. Raises: - RuntimeError: If no active outputs are detected, or if files - already exist and ``--force`` was not passed. + RuntimeError: If no active outputs are detected, if files already + exist and neither ``--sync`` nor ``--force`` was passed, or if + a target file exists but wasn't generated by us. """ i3 = i3ipc.Connection() outputs = _active_outputs(i3) @@ -504,27 +672,82 @@ def run_init(args: InitArgs) -> None: raise RuntimeError("no active outputs detected by i3") config_dir = args.config_dir.expanduser() - files = _planned_files(config_dir, outputs, args.count) + active_names = [name for name, _ in outputs] + + allocations = read_allocations(config_dir) + stale = [name for name in allocations if name not in active_names] + + if args.force: + # Explicit "start over": numbers follow i3's output order. + starts = None + else: + if args.prune: + for name in stale: + allocations.pop(name) + starts = allocate_global_starts(active_names, allocations, args.count) + + files = _planned_files(config_dir, outputs, args.count, starts) + pruned = _prune_paths(config_dir, stale) if args.prune else [] + + if args.sync: + planned = [(path, content) for path, content in files if not path.exists()] + else: + planned = files if args.dry_run: - for path, content in files: + for path in pruned: + print(f"--- would remove {path} ---") + for path, content in planned: print(f"--- {path} ---") print(content, end="") + if not planned: + print("nothing to write; every active output already has a config") return - existing = [path for path, _ in files if path.exists()] + existing = [path for path, _ in planned if path.exists()] if existing and not args.force: listing = "\n ".join(str(p) for p in existing) raise RuntimeError( - "refusing to overwrite existing files (pass --force to replace):\n " - + listing + "refusing to overwrite existing files (pass --sync to only add " + "configs for new outputs, or --force to replace):\n " + listing ) - for path, content in files: + # Never clobber something the user wrote by hand — --force included. + hand_written = [path for path in existing if not _is_generated(path)] + if hand_written: + listing = "\n ".join(str(p) for p in hand_written) + raise RuntimeError( + "refusing to overwrite files that were not generated by " + "`pi3-smart-workspace init` (move them aside first):\n " + listing + ) + + for path, content in planned: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content) print(f"wrote {path}") + for path in pruned: + path.unlink() + print(f"removed {path}") + try: + path.parent.rmdir() + except OSError: + pass # directory not empty — leave whatever else is in there + + if not planned: + print("nothing to write; every active output already has a config") + if args.sync and stale and not args.prune: + listing = ", ".join(stale) + print(f"note: configs kept for inactive outputs: {listing} (--prune removes them)") + if args.sync: + bindings, bindings_content = files[0] + if bindings.exists() and _is_generated(bindings): + if bindings.read_text() != bindings_content: + print( + f"note: {bindings} differs from what -n {args.count} would " + "generate; pass --force to regenerate it" + ) + print() print("Add the following to your i3 config (typically ~/.config/i3/config):") print(f" include {config_dir}/bindings.conf") @@ -550,9 +773,22 @@ def _init_main(argv: list[str]) -> None: "--dry-run", action="store_true", help="Print generated configs to stdout instead of writing them.", ) - parser.add_argument( + mode = parser.add_mutually_exclusive_group() + mode.add_argument( + "--sync", action="store_true", + help="Only write configs for outputs that don't have one yet, keeping " + "the workspace numbers of the outputs that do. Use this after " + "plugging in a new monitor.", + ) + mode.add_argument( "--force", action="store_true", - help="Overwrite existing files in the config directory.", + help="Regenerate every file from scratch. This reallocates global " + "workspace numbers by i3's current output order.", + ) + parser.add_argument( + "--prune", action="store_true", + help="Also delete generated configs for outputs that are no longer " + "active, freeing their workspace numbers.", ) parser.add_argument( "-d", "--debug", action="store_true", @@ -565,6 +801,8 @@ def _init_main(argv: list[str]) -> None: config_dir=parsed.config_dir, dry_run=parsed.dry_run, force=parsed.force, + sync=parsed.sync, + prune=parsed.prune, debug=parsed.debug, ) diff --git a/pyproject.toml b/pyproject.toml index 5297647..3758486 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "pi3-smart-workspace" -version = "0.2.0" +version = "0.3.0" description = "Switch i3 workspaces based on which output your mouse cursor is on." readme = "README.md" requires-python = ">=3.14" diff --git a/tests/test_smart_workspace.py b/tests/test_smart_workspace.py index f02043b..35d90cc 100644 --- a/tests/test_smart_workspace.py +++ b/tests/test_smart_workspace.py @@ -1,13 +1,21 @@ from textwrap import dedent +import pytest + from pi3.smart_workspace import ( + Allocation, + InitArgs, Rect, _planned_files, + allocate_global_starts, expand_i3_includes, find_output_at_cursor, + parse_allocation, parse_workspaces_per_output, + read_allocations, render_bindings, render_workspaces_for_output, + run_init, sanitize_var_suffix, ) @@ -398,3 +406,248 @@ def test_init_output_parses_back_through_parser(): "eDP-1": ["1:1", "2:2", "3:3"], "HDMI-A-0": ["4:1", "5:2", "6:3"], } + + +# --------------------------------------------------------------------------- +# incremental allocation (init --sync) +# --------------------------------------------------------------------------- + + +def test_parse_allocation_round_trips_render(): + text = render_workspaces_for_output("HDMI-A-0", global_start=9, count=8) + + assert parse_allocation(text) == Allocation("HDMI-A-0", 9, 8) + + +def test_parse_allocation_ignores_unrelated_config(): + assert parse_allocation("set $mod Mod4\nset $ws1 1:1\n") is None + assert parse_allocation("") is None + + +def test_allocate_keeps_existing_outputs_and_appends_new(): + """The bug this fixes: a new monitor must not renumber the old ones.""" + existing = {"eDP-1": Allocation("eDP-1", 1, 8)} + + # i3 reports the newcomer *first*, which used to push eDP-1 to 9..16. + starts = allocate_global_starts(["HDMI-A-0", "eDP-1"], existing, count=8) + + assert starts == {"eDP-1": 1, "HDMI-A-0": 9} + + +def test_allocate_reserves_blocks_of_currently_inactive_outputs(): + """A switched-off monitor keeps its numbers unless it's pruned.""" + existing = { + "eDP-1": Allocation("eDP-1", 1, 8), + "HDMI-A-0": Allocation("HDMI-A-0", 9, 8), # unplugged right now + } + + starts = allocate_global_starts(["eDP-1", "DP-1"], existing, count=8) + + assert starts == {"eDP-1": 1, "DP-1": 17} + + +def test_allocate_fills_the_hole_left_by_a_pruned_output(): + existing = {"eDP-1": Allocation("eDP-1", 9, 8)} # HDMI at 1..8 was pruned + + starts = allocate_global_starts(["eDP-1", "DP-1"], existing, count=8) + + assert starts == {"eDP-1": 9, "DP-1": 1} + + +def test_allocate_reallocates_when_the_count_changes(): + """A different -n means the old block no longer fits; it's released.""" + existing = { + "eDP-1": Allocation("eDP-1", 1, 8), + "HDMI-A-0": Allocation("HDMI-A-0", 9, 8), + } + + starts = allocate_global_starts(["eDP-1", "HDMI-A-0"], existing, count=4) + + # Both 8-wide blocks are released and repacked tightly at the new width. + assert starts == {"eDP-1": 1, "HDMI-A-0": 5} + + +def test_allocate_assigns_from_scratch_when_nothing_exists(): + starts = allocate_global_starts(["eDP-1", "HDMI-A-0"], {}, count=3) + + assert starts == {"eDP-1": 1, "HDMI-A-0": 4} + + +def test_planned_files_honors_explicit_starts(): + from pathlib import Path + + outputs = [("HDMI-A-0", Rect(0, 0, 1920, 1080)), ("eDP-1", Rect(1920, 0, 1920, 1080))] + + files = _planned_files(Path("/x"), outputs, count=8, starts={"eDP-1": 1, "HDMI-A-0": 9}) + + assert "set $ws_HDMI_A_0_1 9:1" in files[1][1] + assert "set $ws_eDP_1_1 1:1" in files[2][1] + + +def test_read_allocations_skips_hand_written_files(tmp_path): + generated = tmp_path / "eDP-1" / "workspaces.conf" + generated.parent.mkdir() + generated.write_text(render_workspaces_for_output("eDP-1", 1, 8)) + + handwritten = tmp_path / "DP-1" / "workspaces.conf" + handwritten.parent.mkdir() + handwritten.write_text("set $out_DP_1 DP-1\nset $ws_DP_1_1 1:1\n") + + assert read_allocations(tmp_path) == {"eDP-1": Allocation("eDP-1", 1, 8)} + + +def test_read_allocations_on_missing_directory(tmp_path): + assert read_allocations(tmp_path / "nope") == {} + + +# --------------------------------------------------------------------------- +# run_init (with a stubbed i3 connection) +# --------------------------------------------------------------------------- + + +class _FakeOutput: + def __init__(self, name, rect): + self.name = name + self.rect = rect + self.active = True + + +class _FakeConnection: + def __init__(self, outputs): + self._outputs = outputs + + def get_outputs(self): + return self._outputs + + +@pytest.fixture +def fake_i3(monkeypatch): + """Install a stub i3 connection reporting the given active outputs.""" + import pi3.smart_workspace as sw + + def install(*names): + outputs = [ + _FakeOutput(name, Rect(i * 1920, 0, 1920, 1080)) + for i, name in enumerate(names) + ] + monkeypatch.setattr( + sw.i3ipc, "Connection", lambda *a, **kw: _FakeConnection(outputs) + ) + + return install + + +def _init_args(config_dir, **overrides): + defaults = dict( + count=8, config_dir=config_dir, dry_run=False, force=False, + sync=False, prune=False, debug=True, + ) + return InitArgs(**{**defaults, **overrides}) + + +def test_init_sync_adds_new_monitor_without_renumbering(tmp_path, fake_i3): + fake_i3("eDP-1") + run_init(_init_args(tmp_path)) + before = (tmp_path / "eDP-1" / "workspaces.conf").read_text() + + # New monitor shows up, and i3 lists it before the existing one. + fake_i3("HDMI-A-0", "eDP-1") + run_init(_init_args(tmp_path, sync=True)) + + assert (tmp_path / "eDP-1" / "workspaces.conf").read_text() == before + new = (tmp_path / "HDMI-A-0" / "workspaces.conf").read_text() + assert "set $ws_HDMI_A_0_1 9:1" in new + assert "set $ws_HDMI_A_0_8 16:8" in new + + +def test_init_sync_result_parses_back_without_number_collisions(tmp_path, fake_i3): + fake_i3("eDP-1") + run_init(_init_args(tmp_path, count=3)) + fake_i3("HDMI-A-0", "eDP-1") + run_init(_init_args(tmp_path, count=3, sync=True)) + + combined = "\n".join( + p.read_text() for p in sorted(tmp_path.glob("*/workspaces.conf")) + ) + result = parse_workspaces_per_output(combined, ["eDP-1", "HDMI-A-0"]) + + assert result == {"eDP-1": ["1:1", "2:2", "3:3"], "HDMI-A-0": ["4:1", "5:2", "6:3"]} + # No global workspace number is claimed by two outputs. + numbers = [ws.split(":")[0] for names in result.values() for ws in names] + assert len(numbers) == len(set(numbers)) + + +def test_init_sync_is_a_noop_when_everything_exists(tmp_path, fake_i3): + fake_i3("eDP-1") + run_init(_init_args(tmp_path)) + before = {p: p.read_text() for p in tmp_path.rglob("*.conf")} + + run_init(_init_args(tmp_path, sync=True)) + + assert {p: p.read_text() for p in tmp_path.rglob("*.conf")} == before + + +def test_init_without_sync_still_refuses_to_overwrite(tmp_path, fake_i3): + fake_i3("eDP-1") + run_init(_init_args(tmp_path)) + + fake_i3("HDMI-A-0", "eDP-1") + with pytest.raises(RuntimeError, match="--sync"): + run_init(_init_args(tmp_path)) + + +def test_init_force_renumbers_by_output_order(tmp_path, fake_i3): + fake_i3("eDP-1") + run_init(_init_args(tmp_path)) + + fake_i3("HDMI-A-0", "eDP-1") + run_init(_init_args(tmp_path, force=True)) + + assert "set $ws_HDMI_A_0_1 1:1" in (tmp_path / "HDMI-A-0" / "workspaces.conf").read_text() + assert "set $ws_eDP_1_1 9:1" in (tmp_path / "eDP-1" / "workspaces.conf").read_text() + + +def test_init_force_refuses_to_clobber_hand_written_files(tmp_path, fake_i3): + fake_i3("eDP-1") + target = tmp_path / "eDP-1" / "workspaces.conf" + target.parent.mkdir(parents=True) + target.write_text("# mine, hand-tuned\nset $out_eDP_1 eDP-1\n") + + with pytest.raises(RuntimeError, match="not generated"): + run_init(_init_args(tmp_path, force=True)) + + assert target.read_text() == "# mine, hand-tuned\nset $out_eDP_1 eDP-1\n" + + +def test_init_prune_removes_inactive_output_and_frees_its_numbers(tmp_path, fake_i3): + fake_i3("eDP-1", "HDMI-A-0") + run_init(_init_args(tmp_path)) + assert (tmp_path / "HDMI-A-0").exists() + + # HDMI unplugged, DP-1 plugged in instead. + fake_i3("eDP-1", "DP-1") + run_init(_init_args(tmp_path, sync=True, prune=True)) + + assert not (tmp_path / "HDMI-A-0").exists() + # DP-1 reuses the block HDMI-A-0 gave up rather than starting at 17. + assert "set $ws_DP_1_1 9:1" in (tmp_path / "DP-1" / "workspaces.conf").read_text() + + +def test_init_prune_leaves_hand_written_files_alone(tmp_path, fake_i3): + fake_i3("eDP-1") + run_init(_init_args(tmp_path)) + handwritten = tmp_path / "HDMI-A-0" / "workspaces.conf" + handwritten.parent.mkdir() + handwritten.write_text("# mine\n") + + run_init(_init_args(tmp_path, sync=True, prune=True)) + + assert handwritten.exists() + + +def test_init_dry_run_touches_nothing(tmp_path, fake_i3): + fake_i3("eDP-1") + + run_init(_init_args(tmp_path, dry_run=True)) + + assert list(tmp_path.iterdir()) == []