blitztext-app-linux/linux/blitztext/caret.py

"""Best-effort screen anchor for the overlay: *where* to point the bubble's tip.

The overlay wants to sit at "the cursor where the text will land". On X11 there
is no portable way to read the text caret of an arbitrary app, so we degrade
through a chain of decreasing precision:

  1. AT-SPI caret  — the real text insertion point, when the focused app exposes
     it over accessibility (native GTK/Qt apps do; many terminals / Electron /
     web views do not). Tracked passively via the GLib main loop, so reads are
     instant and never block.
  2. Mouse pointer — `xdotool getmouselocation`. Always available on X11; a good
     proxy since the pointer is usually near where you're typing.
  3. Window / screen — top-centre of the target window, else screen bottom-centre.

Everything here is defensive: any failure falls through to the next tier, and
the whole module is optional — if AT-SPI isn't running you simply get the
pointer anchor. Returns a `(x, y, region)` anchor in **root/screen** pixels,
where `region` is the caret/line box so the overlay can avoid covering it.
"""

from __future__ import annotations

import shutil
import subprocess
import time
from dataclasses import dataclass

from .logbuffer import log


@dataclass
class Anchor:
    x: int            # root-relative x to point the tail tip at
    y: int            # root-relative y (top of the caret/line box)
    height: int = 0   # caret/line height, so the bubble can clear the line
    source: str = ""  # "caret" | "pointer" | "window" | "corner" — for logging


# --------------------------------------------------------------------------- #
# Tier 1: AT-SPI caret tracking (best-effort, passive)
# --------------------------------------------------------------------------- #
class _CaretTracker:
    """Passively follow the focused editable text's caret via AT-SPI events.

    We never poll the a11y tree on the hot path (that can be slow and can block
    on unresponsive apps). Instead we subscribe to caret-moved / focus events and
    cache the last caret rectangle; `rect()` just returns the cached value if it
    is fresh enough to still be meaningful.
    """

    STALE_SECONDS = 30.0  # ignore a cached caret older than this

    def __init__(self) -> None:
        self._ok = False
        self._listener = None
        self._rect: tuple[int, int, int, int] | None = None  # x, y, w, h (screen)
        self._stamp = 0.0
        self._Atspi = None

    def start(self) -> bool:
        """Register AT-SPI listeners on the (already running) GLib main loop.

        Safe to call when accessibility is disabled — it just returns False and
        the anchor logic skips this tier from then on.
        """
        try:
            import gi
            gi.require_version("Atspi", "2.0")
            from gi.repository import Atspi  # noqa: N813
        except (ImportError, ValueError) as exc:
            log(f"[overlay] AT-SPI unavailable, caret anchor disabled: {exc}")
            return False
        try:
            # init() is idempotent; returns 0/1. Connects to the a11y registry.
            Atspi.init()
            self._Atspi = Atspi
            self._listener = Atspi.EventListener.new(self._on_event)
            # Caret moves give us the live position; focus changes let us grab the
            # caret of a freshly-focused field even before it moves.
            self._listener.register("object:text-caret-moved")
            self._listener.register("object:state-changed:focused")
            self._ok = True
            log("[overlay] AT-SPI caret tracking active")
            return True
        except Exception as exc:  # noqa: BLE001 - a11y bus may be down/locked
            log(f"[overlay] AT-SPI init failed, caret anchor disabled: {exc}")
            self._ok = False
            return False

    def stop(self) -> None:
        try:
            if self._listener is not None:
                self._listener.deregister("object:text-caret-moved")
                self._listener.deregister("object:state-changed:focused")
        except Exception:  # noqa: BLE001
            pass
        self._listener = None
        self._ok = False

    def _on_event(self, event) -> None:
        # Runs on the GLib main thread (same loop GTK uses). Keep it cheap and
        # never raise — an exception here would bubble into the a11y dispatcher.
        try:
            if event.type.startswith("object:state-changed:focused") and not event.detail1:
                return  # a *de*focus event — nothing to read
            source = event.source
            if source is None:
                return
            rect = self._caret_rect(source)
            if rect is not None:
                self._rect = rect
                self._stamp = time.time()
        except Exception:  # noqa: BLE001
            pass

    def _caret_rect(self, acc) -> tuple[int, int, int, int] | None:
        Atspi = self._Atspi
        text = None
        try:
            text = acc.get_text_iface()
        except Exception:  # noqa: BLE001
            text = None
        if text is None:
            return None
        try:
            offset = text.get_caret_offset()
            # Extents of the character at the caret, in absolute screen coords.
            ext = text.get_character_extents(offset, Atspi.CoordType.SCREEN)
            x, y, w, h = ext.x, ext.y, ext.width, ext.height
            if w == 0 and h == 0:
                # End-of-line / empty field: fall back to the component box so we
                # at least anchor on the right widget.
                comp = acc.get_component_iface()
                if comp is None:
                    return None
                cext = comp.get_extents(Atspi.CoordType.SCREEN)
                return int(cext.x), int(cext.y), 2, int(cext.height) or 18
            if x < 0 or y < 0:
                return None
            return int(x), int(y), int(w) or 2, int(h) or 18
        except Exception:  # noqa: BLE001
            return None

    def rect(self) -> tuple[int, int, int, int] | None:
        if not self._ok or self._rect is None:
            return None
        if time.time() - self._stamp > self.STALE_SECONDS:
            return None
        return self._rect


# --------------------------------------------------------------------------- #
# Tier 2 + 3: pointer and window fallbacks
# --------------------------------------------------------------------------- #
def _pointer() -> tuple[int, int] | None:
    if not shutil.which("xdotool"):
        return None
    try:
        out = subprocess.run(
            ["xdotool", "getmouselocation", "--shell"],
            capture_output=True, text=True, check=True, timeout=0.5,
        ).stdout
    except (OSError, subprocess.SubprocessError):
        return None
    vals: dict[str, str] = {}
    for line in out.splitlines():
        if "=" in line:
            k, _, v = line.partition("=")
            vals[k.strip()] = v.strip()
    try:
        return int(vals["X"]), int(vals["Y"])
    except (KeyError, ValueError):
        return None


def _window_box(window_id: str | None) -> tuple[int, int, int] | None:
    """Top-centre of the target window: (x, y, height_hint)."""
    if not window_id or not shutil.which("xdotool"):
        return None
    try:
        out = subprocess.run(
            ["xdotool", "getwindowgeometry", "--shell", window_id],
            capture_output=True, text=True, check=True, timeout=0.5,
        ).stdout
    except (OSError, subprocess.SubprocessError):
        return None
    vals: dict[str, str] = {}
    for line in out.splitlines():
        if "=" in line:
            k, _, v = line.partition("=")
            vals[k.strip()] = v.strip()
    try:
        x, y = int(vals["X"]), int(vals["Y"])
        w = int(vals["WIDTH"])
        return x + w // 2, y + 40, 0
    except (KeyError, ValueError):
        return None


def _screen_bottom_centre() -> tuple[int, int]:
    try:
        import gi
        gi.require_version("Gdk", "3.0")
        from gi.repository import Gdk
        disp = Gdk.Display.get_default()
        mon = disp.get_primary_monitor() or disp.get_monitor(0)
        geo = mon.get_geometry()
        return geo.x + geo.width // 2, geo.y + geo.height - 140
    except Exception:  # noqa: BLE001
        return 960, 800  # last-ditch constant; better than crashing


_tracker: _CaretTracker | None = None


def start_tracking() -> None:
    """Begin passive caret tracking (call once, from the GTK main thread)."""
    global _tracker
    if _tracker is None:
        _tracker = _CaretTracker()
        _tracker.start()


def stop_tracking() -> None:
    global _tracker
    if _tracker is not None:
        _tracker.stop()
        _tracker = None


def resolve(anchor_mode: str, window_id: str | None) -> Anchor:
    """Resolve the on-screen anchor for the overlay, honouring `anchor_mode`.

    anchor_mode:
      "caret"   -> caret → pointer → window → corner
      "pointer" -> pointer → window → corner
      "corner"  -> window → corner
    Always returns an Anchor (never None) so callers don't special-case failure.
    """
    if anchor_mode == "caret" and _tracker is not None:
        r = _tracker.rect()
        if r is not None:
            x, y, _w, h = r
            return Anchor(x=x, y=y, height=h or 18, source="caret")

    if anchor_mode in ("caret", "pointer"):
        p = _pointer()
        if p is not None:
            return Anchor(x=p[0], y=p[1], height=0, source="pointer")

    wb = _window_box(window_id)
    if wb is not None:
        return Anchor(x=wb[0], y=wb[1], height=0, source="window")

    x, y = _screen_bottom_centre()
    return Anchor(x=x, y=y, height=0, source="corner")