feat(desktop): consolidate declarative workstation profile
Capture desktop-specific system, session, Noctalia, and Pi integration alongside shared laptop support. Retire obsolete X11 configuration and make package ownership explicit for repeatable Decman deployments.
This commit is contained in:
@@ -0,0 +1,139 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Focus the most recently focused Obsidian window under Hyprland."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote, urlencode
|
||||
|
||||
OBSIDIAN_CLASS = "md.obsidian.Obsidian"
|
||||
VAULT_ROOT: Path
|
||||
OBSIDIAN_VAULT: str
|
||||
|
||||
|
||||
def hyprctl(*arguments: str) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
["hyprctl", *arguments],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
|
||||
def focus_expression(field: str, value: str) -> str:
|
||||
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
|
||||
return f'hl.dispatch(hl.dsp.focus({{ {field} = "{escaped}" }}))'
|
||||
|
||||
|
||||
def note_uri() -> str | None:
|
||||
if len(sys.argv) != 4:
|
||||
print("usage: focus-obsidian.py VAULT VAULT_ID NOTE_PATH", file=sys.stderr)
|
||||
return None
|
||||
note = Path(sys.argv[3]).resolve()
|
||||
try:
|
||||
relative = note.relative_to(VAULT_ROOT).as_posix()
|
||||
except ValueError as error:
|
||||
print(f"note escaped the knowledge vault: {error}", file=sys.stderr)
|
||||
return None
|
||||
return "obsidian://open?" + urlencode(
|
||||
{"vault": OBSIDIAN_VAULT, "file": relative}, quote_via=quote
|
||||
)
|
||||
|
||||
|
||||
def dispatch_uri(uri: str) -> bool:
|
||||
try:
|
||||
subprocess.Popen(
|
||||
["obsidian", uri],
|
||||
stdin=subprocess.DEVNULL,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
start_new_session=True,
|
||||
)
|
||||
except OSError as error:
|
||||
print(f"cannot open Obsidian: {error}", file=sys.stderr)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def main() -> int:
|
||||
global VAULT_ROOT, OBSIDIAN_VAULT
|
||||
if len(sys.argv) != 4:
|
||||
print("usage: focus-obsidian.py VAULT VAULT_ID NOTE_PATH", file=sys.stderr)
|
||||
return 2
|
||||
VAULT_ROOT = Path(sys.argv[1]).expanduser().resolve()
|
||||
OBSIDIAN_VAULT = sys.argv[2]
|
||||
if not VAULT_ROOT.is_dir():
|
||||
print(f"vault does not exist: {VAULT_ROOT}", file=sys.stderr)
|
||||
return 1
|
||||
if not OBSIDIAN_VAULT:
|
||||
print("vault is not registered in Obsidian", file=sys.stderr)
|
||||
return 1
|
||||
uri = note_uri()
|
||||
if uri is None or not dispatch_uri(uri):
|
||||
return 1
|
||||
target: dict[str, object] | None = None
|
||||
for _ in range(150):
|
||||
result = hyprctl("clients", "-j")
|
||||
if result.returncode == 0:
|
||||
try:
|
||||
clients = json.loads(result.stdout)
|
||||
except json.JSONDecodeError as error:
|
||||
print(f"cannot parse hyprctl clients: {error}", file=sys.stderr)
|
||||
clients = []
|
||||
matches = [
|
||||
client for client in clients if client.get("class") == OBSIDIAN_CLASS
|
||||
]
|
||||
if matches:
|
||||
target = min(
|
||||
matches, key=lambda client: client.get("focusHistoryID", 1_000_000)
|
||||
)
|
||||
break
|
||||
time.sleep(0.1)
|
||||
|
||||
if target is None:
|
||||
return 1
|
||||
|
||||
workspace = target.get("workspace")
|
||||
address = target.get("address")
|
||||
if not isinstance(workspace, dict) or not isinstance(address, str):
|
||||
return 1
|
||||
workspace_id = workspace.get("id")
|
||||
if not isinstance(workspace_id, int):
|
||||
return 1
|
||||
|
||||
workspace_expression = focus_expression("workspace", str(workspace_id))
|
||||
selector = f"address:{address}"
|
||||
window_expression = focus_expression("window", selector)
|
||||
focused_samples = 0
|
||||
for attempt in range(20):
|
||||
if attempt == 0 or attempt % 5 == 0:
|
||||
hyprctl("eval", workspace_expression)
|
||||
hyprctl("eval", window_expression)
|
||||
active = hyprctl("activewindow", "-j")
|
||||
if active.returncode == 0:
|
||||
try:
|
||||
active_window = json.loads(active.stdout)
|
||||
except json.JSONDecodeError as error:
|
||||
print(f"cannot parse active Hyprland window: {error}", file=sys.stderr)
|
||||
active_window = {}
|
||||
focused_samples = (
|
||||
focused_samples + 1 if active_window.get("address") == address else 0
|
||||
)
|
||||
if focused_samples >= 3:
|
||||
# Re-send after focus: Electron occasionally consumes the URI
|
||||
# before its renderer is ready, leaving the prior note visible.
|
||||
if not dispatch_uri(uri):
|
||||
return 1
|
||||
time.sleep(0.35)
|
||||
hyprctl("eval", window_expression)
|
||||
return 0
|
||||
time.sleep(0.1)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user