alex
ed063ab1fe
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.
140 lines
4.4 KiB
Python
140 lines
4.4 KiB
Python
#!/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())
|