InputRemapper/inputremapper/focus/backends/hyprland.py

# -*- coding: utf-8 -*-
# input-remapper - GUI for device specific keyboard mappings
# Copyright (C) 2025 sezanzeb <b8x45ygc9@mozmail.com>
#
# This file is part of input-remapper.
#
# input-remapper is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# input-remapper is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with input-remapper.  If not, see <https://www.gnu.org/licenses/>.

"""Focus backend for Hyprland (socket2 event stream + socket1 queries)."""

from __future__ import annotations

import json
import os
import socket
from typing import Optional

import gi

from inputremapper.focus.focus_backend import FocusBackend, FocusEvent, OnFocus
from inputremapper.logging.logger import logger

gi.require_version("GLib", "2.0")
from gi.repository import GLib  # noqa: E402


def _runtime_dirs(signature: str) -> list[str]:
    """Candidate hypr runtime directories, newest layout first."""
    candidates = []
    xdg_runtime = os.environ.get("XDG_RUNTIME_DIR")
    if xdg_runtime:
        candidates.append(os.path.join(xdg_runtime, "hypr", signature))
    # legacy location for older Hyprland versions
    candidates.append(os.path.join("/tmp", "hypr", signature))
    return candidates


class HyprlandFocusBackend(FocusBackend):
    """Reads ``activewindow`` events from the Hyprland socket2."""

    name = "hyprland"

    def __init__(self) -> None:
        self._socket: Optional[socket.socket] = None
        self._watch_id: Optional[int] = None
        self._buffer = b""
        self._on_focus: Optional[OnFocus] = None

    @staticmethod
    def _signature() -> Optional[str]:
        return os.environ.get("HYPRLAND_INSTANCE_SIGNATURE")

    @classmethod
    def _socket_path(cls, name: str) -> Optional[str]:
        signature = cls._signature()
        if not signature:
            return None
        for directory in _runtime_dirs(signature):
            path = os.path.join(directory, name)
            if os.path.exists(path):
                return path
        return None

    @staticmethod
    def is_available() -> bool:
        return HyprlandFocusBackend._socket_path(".socket2.sock") is not None

    def start(self, on_focus: OnFocus) -> None:
        path = self._socket_path(".socket2.sock")
        if not path:
            raise RuntimeError("Hyprland socket2 not found")

        self._on_focus = on_focus
        self._socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self._socket.connect(path)
        self._socket.setblocking(False)

        self._watch_id = GLib.io_add_watch(
            self._socket.fileno(),
            GLib.IO_IN | GLib.IO_HUP | GLib.IO_ERR,
            self._on_readable,
        )
        logger.debug("Hyprland focus backend listening on %s", path)

    def _on_readable(self, _fd: int, condition: int) -> bool:
        if condition & (GLib.IO_HUP | GLib.IO_ERR):
            logger.error("Hyprland socket2 closed")
            return False

        assert self._socket is not None
        try:
            data = self._socket.recv(65536)
        except BlockingIOError:
            return True
        except OSError as error:
            logger.error("Hyprland socket2 read failed: %s", error)
            return False

        if not data:
            return False

        self._buffer += data
        while b"\n" in self._buffer:
            line, self._buffer = self._buffer.split(b"\n", 1)
            self._handle_line(line.decode(errors="replace"))
        return True

    def _handle_line(self, line: str) -> None:
        # Lines look like "activewindow>>CLASS,TITLE"
        name, separator, data = line.partition(">>")
        if not separator or name != "activewindow":
            return

        app_id, _, title = data.partition(",")
        event = FocusEvent(app_id=app_id, title=title, backend=self.name)
        if self._on_focus is not None:
            self._on_focus(event)

    def stop(self) -> None:
        if self._watch_id is not None:
            GLib.source_remove(self._watch_id)
            self._watch_id = None
        if self._socket is not None:
            self._socket.close()
            self._socket = None
        self._buffer = b""

    def get_current(self) -> Optional[FocusEvent]:
        path = self._socket_path(".socket.sock")
        if not path:
            return None

        try:
            with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
                sock.connect(path)
                sock.sendall(b"j/activewindow")
                chunks = []
                while True:
                    chunk = sock.recv(65536)
                    if not chunk:
                        break
                    chunks.append(chunk)
        except OSError as error:
            logger.error("Failed to query Hyprland active window: %s", error)
            return None

        raw = b"".join(chunks).decode(errors="replace").strip()
        if not raw:
            return None

        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            return None

        if not data:
            return None

        return FocusEvent(
            app_id=data.get("class", "") or "",
            title=data.get("title", "") or "",
            backend=self.name,
        )