#!/usr/bin/env python3 """Background Zed activity tracker for Wayland (Arch Linux) and fallback. Logs heartbeats to .zed-hours/heartbeats.jsonl. """ import argparse import json import os import re import shutil import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path ZED_APP_NAMES = {"zed", "zeditor"} PROJECT_ZED_PATTERN = re.compile(r"(?:^|\s)([^/]+?)(?:\s+[-—]\s+Zed(?:\s+Editor)?)$") def now_iso() -> str: return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds") def load_config(project_root: Path) -> dict: config_path = project_root / ".zed-hours" / "config.json" defaults = { "project_root": str(project_root), "log_dir": ".zed-hours", "heartbeat_interval_seconds": 60, "idle_threshold_seconds": 300, "max_inactive_gap_seconds": 600, } if config_path.exists(): try: loaded = json.loads(config_path.read_text(encoding="utf-8")) defaults.update(loaded) except Exception as e: print(f"Warning: could not read config: {e}", file=sys.stderr) return defaults def get_active_window_wayland() -> dict | None: """Try several Wayland compositor methods to detect the active window.""" # 1. Hyprland if shutil.which("hyprctl"): try: out = subprocess.check_output( ["hyprctl", "activewindow", "-j"], text=True, stderr=subprocess.DEVNULL, timeout=2, ) data = json.loads(out) title = (data.get("title") or "").strip() class_ = (data.get("class") or "").strip().lower() return {"title": title, "class": class_, "source": "hyprctl"} except Exception: pass # 2. Sway if shutil.which("swaymsg"): try: out = subprocess.check_output( ["swaymsg", "-t", "get_tree"], text=True, stderr=subprocess.DEVNULL, timeout=2, ) data = json.loads(out) def find_focused(node): if node.get("focused"): return node for child in node.get("nodes", []) + node.get("floating_nodes", []): found = find_focused(child) if found: return found return None focused = find_focused(data) if focused: props = focused.get("window_properties", {}) title = focused.get("name") or props.get("title") or "" class_ = props.get("class", "").lower() return {"title": title, "class": class_, "source": "swaymsg"} except Exception: pass # 3. GNOME / Mutter via gdbus if shutil.which("gdbus"): try: out = subprocess.check_output( [ "gdbus", "call", "--session", "--dest", "org.gnome.Shell", "--object-path", "/org/gnome/Shell", "--method", "org.gnome.Shell.Eval", "global.display.focus_window.get_title()", ], text=True, stderr=subprocess.DEVNULL, timeout=2, ) match = re.search(r"'(.*)'", out) if match: title = match.group(1) return {"title": title, "class": "", "source": "gdbus-gnome"} except Exception: pass # 4. KDE / KWin via qdbus / qdbus6 for qdbus in ("qdbus6", "qdbus"): if shutil.which(qdbus): try: out = subprocess.check_output( [qdbus, "org.kde.KWin", "/KWin", "org.kde.KWin.activeWindow"], text=True, stderr=subprocess.DEVNULL, timeout=2, ).strip() # qdbus returns a window id; we would need more introspection for title. # Fallback: try activeWindowCaption if available caption = subprocess.check_output( [qdbus, "org.kde.KWin", "/KWin", "org.kde.KWin.activeWindowCaption"], text=True, stderr=subprocess.DEVNULL, timeout=2, ).strip() if caption: return {"title": caption, "class": "", "source": f"{qdbus}-kwin"} except Exception: pass return None def is_zed(window: dict | None) -> bool: if not window: return False title = window.get("title", "").lower() class_ = window.get("class", "").lower() return ( class_ in ZED_APP_NAMES or "zed" in class_ or title.endswith(" - zed") or title.endswith(" - zed editor") or "zeditor" in class_ ) def extract_project_from_title(title: str, project_root: Path) -> str | None: """Try to identify which project/folder is open in Zed from window title.""" match = PROJECT_ZED_PATTERN.search(title) if match: return match.group(1) # Fallback: if title ends with the project root name, assume this project if project_root.name.lower() in title.lower(): return project_root.name return None def get_latest_project_file_activity(project_root: Path) -> datetime | None: """Fallback: find the most recent access or modification time under the project.""" latest = None # Common directories to ignore ignored = {".git", ".zed-hours", "__pycache__", "node_modules", ".venv", "venv"} try: for dirpath, dirnames, filenames in os.walk(project_root): # Prune ignored dirs dirnames[:] = [d for d in dirnames if d not in ignored] for name in filenames: # Skip hidden, binary-ish, or generated files if name.startswith("."): continue path = Path(dirpath) / name try: stat = path.stat() mtime = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc) if latest is None or mtime > latest: latest = mtime except Exception: continue except Exception as e: print(f"Warning: filesystem scan failed: {e}", file=sys.stderr) return latest def log_heartbeat(log_dir: Path, project_root: Path, window: dict | None, active: bool, source: str) -> None: title = window.get("title") if window else None project_in_title = None if title: project_in_title = extract_project_from_title(title, project_root) record = { "timestamp": now_iso(), "project": str(project_root), "project_in_title": project_in_title, "zed_active": active, "source": source, "title": title, "class": window.get("class") if window else None, } log_file = log_dir / "heartbeats.jsonl" with log_file.open("a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") def tracker_loop(project_root: Path, config: dict) -> None: log_dir = project_root / config["log_dir"] log_dir.mkdir(parents=True, exist_ok=True) interval = int(config.get("heartbeat_interval_seconds", 60)) idle_threshold = int(config.get("idle_threshold_seconds", 300)) last_window_state: dict | None = None last_activity_time = time.time() filesystem_last = None print(f"Zed tracker started for {project_root}") print(f"Logging to {log_dir}") print(f"Heartbeat interval: {interval}s") print("Press Ctrl+C to stop.\n") try: while True: now = time.time() window = get_active_window_wayland() active = is_zed(window) source = window.get("source") if window else None if active: last_activity_time = now else: # Fallback: check filesystem activity latest = get_latest_project_file_activity(project_root) if latest and (filesystem_last is None or latest > filesystem_last): filesystem_last = latest last_activity_time = now active = True source = "filesystem-fallback" log_heartbeat(log_dir, project_root, window, active, source or "unknown") if active != (last_window_state is not None and is_zed(last_window_state)): status = "ACTIVE" if active else "inactive" print(f"[{now_iso()}] Zed {status} (source: {source})") last_window_state = window time.sleep(interval) except KeyboardInterrupt: print("\nZed tracker stopped.") def main() -> None: parser = argparse.ArgumentParser(description="Track Zed activity and log heartbeats.") parser.add_argument( "--project-root", type=Path, default=None, help="Project root directory (defaults to git root or script location)", ) args = parser.parse_args() project_root = args.project_root if project_root: project_root = project_root.resolve() else: # Try git root try: project_root = Path( subprocess.check_output( ["git", "rev-parse", "--show-toplevel"], text=True, stderr=subprocess.DEVNULL, ).strip() ).resolve() except (subprocess.CalledProcessError, FileNotFoundError): pass # Fallback: script location if project_root is None: script_project = Path(__file__).resolve().parents[2] if (script_project / ".git").exists() or (script_project / ".zed-hours").exists(): project_root = script_project # Final fallback: cwd if project_root is None: project_root = Path.cwd().resolve() config = load_config(project_root) tracker_loop(project_root, config) if __name__ == "__main__": main()