diff --git a/gatelock/gatelock/__init__.py b/gatelock/gatelock/__init__.py index beb9006..384f167 100644 --- a/gatelock/gatelock/__init__.py +++ b/gatelock/gatelock/__init__.py @@ -32,10 +32,10 @@ from gatelock._outputs import ( OutputEnumerator, OutputRect, OutputScan, - RandrBackend, enumerate_outputs, - parse_xrandr_query, ) +from gatelock._randr import RandrBackend +from gatelock._xrandr import parse_xrandr_query from gatelock._queue import ( QUEUE_DEADLINE_SECONDS, QUEUE_POLL_SECONDS, diff --git a/gatelock/gatelock/_output_types.py b/gatelock/gatelock/_output_types.py new file mode 100644 index 0000000..40c13d0 --- /dev/null +++ b/gatelock/gatelock/_output_types.py @@ -0,0 +1,62 @@ +"""The value types describing a display output and a scan of them. + +Split from :mod:`gatelock._outputs` so the two backends +(:mod:`gatelock._randr`, :mod:`gatelock._xrandr`) can import the types without +importing the enumerator that uses them -- which is what would otherwise make +the dependency circular. + +Re-exported from :mod:`gatelock._outputs`, so existing imports keep working. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class OutputRect: + """A rectangle in root-window coordinates.""" + + x: int + y: int + width: int + height: int + + def geometry(self) -> str: + """Return the Tk geometry string for this rectangle.""" + return f"{self.width}x{self.height}+{self.x}+{self.y}" + + +@dataclass(frozen=True) +class Output: + """A single X output and whether anything can actually be seen on it.""" + + name: str + connected: bool + rect: OutputRect | None + primary: bool = False + + @property + def live(self) -> bool: + """Whether this output is connected *and* has a mode.""" + return self.connected and self.rect is not None + + +@dataclass(frozen=True) +class OutputScan: + """The result of one enumeration attempt. + + ``ok`` records whether enumeration itself succeeded, which is a different + question from whether any output is live. Callers must not conflate them: + a failed scan means "no information, change nothing", while a successful + scan with zero live outputs means "the screens really are dark". + """ + + outputs: tuple[Output, ...] + source: OutputSource + ok: bool + + @property + def live(self) -> tuple[Output, ...]: + """The subset of outputs that are connected and have a mode.""" + return tuple(output for output in self.outputs if output.live) diff --git a/gatelock/gatelock/_outputs.py b/gatelock/gatelock/_outputs.py index a617217..d04b45b 100644 --- a/gatelock/gatelock/_outputs.py +++ b/gatelock/gatelock/_outputs.py @@ -25,7 +25,6 @@ precisely the degradation path that must stay tested. from __future__ import annotations -from dataclasses import dataclass import logging import re import shutil @@ -36,56 +35,27 @@ if TYPE_CHECKING: from collections.abc import Sequence import tkinter as tk - # python-xlib ships no type stubs. Rather than typing its objects as ``Any`` - # (or reaching for a suppression comment, which this repo bans), these - # protocols spell out the exact surface gatelock depends on. If a future - # python-xlib renames one of these, the type checker says so. - class _ScreenResources(Protocol): - config_timestamp: int - outputs: Sequence[int] - - class _OutputInfo(Protocol): - name: object - connection: int - crtc: int - - class _CrtcInfo(Protocol): - mode: int - x: int - y: int - width: int - height: int - - class _PrimaryReply(Protocol): - output: int - - class _XRoot(Protocol): - def xrandr_get_screen_resources(self) -> _ScreenResources: ... - - def xrandr_get_output_primary(self) -> _PrimaryReply: ... - - class _XScreen(Protocol): - root: _XRoot - - class _XDisplay(Protocol): - def screen(self) -> _XScreen: ... - - def xrandr_get_output_info( - self, output: int, timestamp: int - ) -> _OutputInfo: ... - - def xrandr_get_crtc_info(self, crtc: int, timestamp: int) -> _CrtcInfo: ... - - def close(self) -> None: ... +from gatelock._output_types import Output, OutputRect, OutputScan +from gatelock._randr import RandrBackend +from gatelock._xrandr import parse_xrandr_query, scan_xrandr +# Re-exported after the 250-line split, so `from gatelock._outputs import +# Output` -- which the tests and sibling modules do -- keeps resolving. +__all__ = [ + "Output", + "OutputEnumerator", + "OutputRect", + "OutputScan", + "RandrBackend", + "enumerate_outputs", + "parse_xrandr_query", + "scan_xrandr", + "tk_fallback_outputs", +] _logger = logging.getLogger(__name__) -_XRANDR_TIMEOUT_S = 5.0 -# RandR's `connection` field: 0 is RR_Connected. Hard-coded rather than -# imported so the constant is available without python-xlib installed. -_RR_CONNECTED = 0 # Output lines start at column 0; mode lines are indented four spaces. That # anchor is the whole robustness story -- without it, the mode line @@ -95,231 +65,11 @@ _OUTPUT_LINE = re.compile( r"^(?P\S+)\s+(?Pconnected|disconnected)\b(?P.*)$", re.MULTILINE, ) -_GEOMETRY = re.compile(r"(?P\d+)x(?P\d+)\+(?P-?\d+)\+(?P-?\d+)") OutputSource = Literal["randr", "xrandr", "tk", "none"] """Which backend produced a scan. Reported so logs can explain a degradation.""" -@dataclass(frozen=True) -class OutputRect: - """A rectangle in root-window coordinates.""" - - x: int - y: int - width: int - height: int - - def geometry(self) -> str: - """Return the Tk geometry string for this rectangle.""" - return f"{self.width}x{self.height}+{self.x}+{self.y}" - - -@dataclass(frozen=True) -class Output: - """A single X output and whether anything can actually be seen on it.""" - - name: str - connected: bool - rect: OutputRect | None - primary: bool = False - - @property - def live(self) -> bool: - """Whether this output is connected *and* has a mode.""" - return self.connected and self.rect is not None - - -@dataclass(frozen=True) -class OutputScan: - """The result of one enumeration attempt. - - ``ok`` records whether enumeration itself succeeded, which is a different - question from whether any output is live. Callers must not conflate them: - a failed scan means "no information, change nothing", while a successful - scan with zero live outputs means "the screens really are dark". - """ - - outputs: tuple[Output, ...] - source: OutputSource - ok: bool - - @property - def live(self) -> tuple[Output, ...]: - """The subset of outputs that are connected and have a mode.""" - return tuple(output for output in self.outputs if output.live) - - -def _parse_rect(head: str) -> OutputRect | None: - """Parse a geometry token out of an xrandr output line's head segment.""" - match = _GEOMETRY.search(head) - if match is None: - return None - width = int(match.group("w")) - height = int(match.group("h")) - if width <= 0 or height <= 0: - return None - return OutputRect( - x=int(match.group("x")), - y=int(match.group("y")), - width=width, - height=height, - ) - - -def parse_xrandr_query(text: str) -> tuple[Output, ...]: - """Parse ``xrandr --query`` output into outputs. - - Args: - text: Raw stdout from ``xrandr --query``. - - Returns: - One :class:`Output` per output line, in the order xrandr listed them. - """ - outputs: list[Output] = [] - for match in _OUTPUT_LINE.finditer(text): - # Everything after the first "(" is rotation/reflection vocabulary and - # panning noise -- never geometry. Slicing it off keeps the search - # honest. - head = match.group("rest").split("(", 1)[0] - outputs.append( - Output( - name=match.group("name"), - connected=match.group("state") == "connected", - rect=_parse_rect(head), - primary="primary" in head.split(), - ) - ) - return tuple(outputs) - - -def _decode_name(raw: object) -> str: - """Normalise a RandR output name, which may arrive as bytes.""" - if isinstance(raw, bytes): - return raw.decode("utf-8", errors="replace") - return str(raw) - - -class RandrBackend: - """A persistent X connection used to enumerate outputs via RandR. - - Held open across scans -- the recovery loop scans every second, and opening - a fresh X connection each time would be pure waste. - """ - - def __init__(self, display: _XDisplay) -> None: - """Wrap an already-opened ``Xlib.display.Display``.""" - self._display = display - - @classmethod - def create(cls) -> RandrBackend | None: - """Open an X connection for RandR queries, or None if unavailable.""" - try: - from Xlib import display as xdisplay - except ImportError: - _logger.debug( - "python-xlib is not installed; RandR events and RandR " - "enumeration are unavailable, falling back to xrandr" - ) - return None - try: - return cls(xdisplay.Display()) - except (OSError, ValueError) as exc: - _logger.warning("could not open an X display for RandR: %s", exc) - return None - - def close(self) -> None: - """Close the underlying X connection, ignoring teardown errors.""" - try: - self._display.close() - except (OSError, ValueError, AttributeError) as exc: - _logger.debug("ignoring error while closing RandR display: %s", exc) - - def scan(self) -> tuple[Output, ...] | None: - """Enumerate outputs, or None if the RandR query failed.""" - try: - return self._scan() - except (OSError, ValueError, KeyError, AttributeError, TypeError) as exc: - _logger.warning("RandR enumeration failed (%s); falling back", exc) - return None - - def _scan(self) -> tuple[Output, ...]: - """Enumerate outputs via RandR. Raises on any protocol problem.""" - root = self._display.screen().root - resources = root.xrandr_get_screen_resources() - timestamp = resources.config_timestamp - primary_id = self._primary_output_id(root) - - outputs: list[Output] = [] - for output_id in resources.outputs: - info = self._display.xrandr_get_output_info(output_id, timestamp) - outputs.append( - Output( - name=_decode_name(info.name), - connected=info.connection == _RR_CONNECTED, - rect=self._crtc_rect(info.crtc, timestamp), - primary=output_id == primary_id, - ) - ) - return tuple(outputs) - - def _primary_output_id(self, root: _XRoot) -> int | None: - """Return the primary output's id, or None when RandR won't say.""" - try: - return root.xrandr_get_output_primary().output - except (OSError, ValueError, KeyError, AttributeError, TypeError): - return None - - def _crtc_rect(self, crtc: int, timestamp: int) -> OutputRect | None: - """Return the CRTC's rectangle, or None when there is no mode. - - ``crtc == 0`` means no CRTC is assigned and ``mode == 0`` means a CRTC - exists without a mode. Either one is the dark-monitor case. - """ - if not crtc: - return None - info = self._display.xrandr_get_crtc_info(crtc, timestamp) - if not info.mode or info.width <= 0 or info.height <= 0: - return None - return OutputRect(x=info.x, y=info.y, width=info.width, height=info.height) - - -def scan_xrandr() -> tuple[Output, ...] | None: - """Enumerate outputs by shelling out to ``xrandr --query``. - - Returns: - The parsed outputs, or None if xrandr is missing, failed, timed out or - produced nothing parseable. - """ - binary = shutil.which("xrandr") - if binary is None: - _logger.warning("xrandr is not on PATH; cannot enumerate outputs") - return None - try: - completed = subprocess.run( - [binary, "--query"], - capture_output=True, - text=True, - timeout=_XRANDR_TIMEOUT_S, - check=False, - ) - except (subprocess.TimeoutExpired, OSError) as exc: - _logger.warning("xrandr --query failed to run: %s", exc) - return None - if completed.returncode != 0: - _logger.warning( - "xrandr --query exited %d: %s", - completed.returncode, - completed.stderr.strip(), - ) - return None - outputs = parse_xrandr_query(completed.stdout) - if not outputs: - _logger.warning("xrandr --query produced no parseable output lines") - return None - return outputs - - def tk_fallback_outputs(root: tk.Misc) -> tuple[Output, ...]: """Synthesise one output covering the whole X screen. diff --git a/gatelock/gatelock/_randr.py b/gatelock/gatelock/_randr.py new file mode 100644 index 0000000..5796a00 --- /dev/null +++ b/gatelock/gatelock/_randr.py @@ -0,0 +1,157 @@ +"""The python-xlib RandR backend: instant output-change events. + +Split from :mod:`gatelock._outputs`, which keeps the value types and the +xrandr-subprocess path. python-xlib is an optional dependency -- without it +the lock still detects output changes, just by polling -- so everything that +touches it lives here, behind :meth:`RandrBackend.create`. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Protocol + +from gatelock._output_types import Output, OutputRect + +if TYPE_CHECKING: + from collections.abc import Sequence + + # python-xlib ships no type stubs. Rather than typing its objects as ``Any`` + # (or reaching for a suppression comment, which this repo bans), these + # protocols spell out the exact surface gatelock depends on. If a future + # python-xlib renames one of these, the type checker says so. + class _ScreenResources(Protocol): + config_timestamp: int + outputs: Sequence[int] + + class _OutputInfo(Protocol): + name: object + connection: int + crtc: int + + class _CrtcInfo(Protocol): + mode: int + x: int + y: int + width: int + height: int + + class _PrimaryReply(Protocol): + output: int + + class _XRoot(Protocol): + def xrandr_get_screen_resources(self) -> _ScreenResources: ... + + def xrandr_get_output_primary(self) -> _PrimaryReply: ... + + class _XScreen(Protocol): + root: _XRoot + + class _XDisplay(Protocol): + def screen(self) -> _XScreen: ... + + def xrandr_get_output_info( + self, output: int, timestamp: int + ) -> _OutputInfo: ... + + def xrandr_get_crtc_info(self, crtc: int, timestamp: int) -> _CrtcInfo: ... + + def close(self) -> None: ... + +_logger = logging.getLogger(__name__) + +# RandR's `connection` field: 0 is RR_Connected. Hard-coded rather than +# imported so the constant is available without python-xlib installed. +_RR_CONNECTED = 0 + +def _decode_name(raw: object) -> str: + """Normalise a RandR output name, which may arrive as bytes.""" + if isinstance(raw, bytes): + return raw.decode("utf-8", errors="replace") + return str(raw) + + +class RandrBackend: + """A persistent X connection used to enumerate outputs via RandR. + + Held open across scans -- the recovery loop scans every second, and opening + a fresh X connection each time would be pure waste. + """ + + def __init__(self, display: _XDisplay) -> None: + """Wrap an already-opened ``Xlib.display.Display``.""" + self._display = display + + @classmethod + def create(cls) -> RandrBackend | None: + """Open an X connection for RandR queries, or None if unavailable.""" + try: + from Xlib import display as xdisplay + except ImportError: + _logger.debug( + "python-xlib is not installed; RandR events and RandR " + "enumeration are unavailable, falling back to xrandr" + ) + return None + try: + return cls(xdisplay.Display()) + except (OSError, ValueError) as exc: + _logger.warning("could not open an X display for RandR: %s", exc) + return None + + def close(self) -> None: + """Close the underlying X connection, ignoring teardown errors.""" + try: + self._display.close() + except (OSError, ValueError, AttributeError) as exc: + _logger.debug("ignoring error while closing RandR display: %s", exc) + + def scan(self) -> tuple[Output, ...] | None: + """Enumerate outputs, or None if the RandR query failed.""" + try: + return self._scan() + except (OSError, ValueError, KeyError, AttributeError, TypeError) as exc: + _logger.warning("RandR enumeration failed (%s); falling back", exc) + return None + + def _scan(self) -> tuple[Output, ...]: + """Enumerate outputs via RandR. Raises on any protocol problem.""" + root = self._display.screen().root + resources = root.xrandr_get_screen_resources() + timestamp = resources.config_timestamp + primary_id = self._primary_output_id(root) + + outputs: list[Output] = [] + for output_id in resources.outputs: + info = self._display.xrandr_get_output_info(output_id, timestamp) + outputs.append( + Output( + name=_decode_name(info.name), + connected=info.connection == _RR_CONNECTED, + rect=self._crtc_rect(info.crtc, timestamp), + primary=output_id == primary_id, + ) + ) + return tuple(outputs) + + def _primary_output_id(self, root: _XRoot) -> int | None: + """Return the primary output's id, or None when RandR won't say.""" + try: + return root.xrandr_get_output_primary().output + except (OSError, ValueError, KeyError, AttributeError, TypeError): + return None + + def _crtc_rect(self, crtc: int, timestamp: int) -> OutputRect | None: + """Return the CRTC's rectangle, or None when there is no mode. + + ``crtc == 0`` means no CRTC is assigned and ``mode == 0`` means a CRTC + exists without a mode. Either one is the dark-monitor case. + """ + if not crtc: + return None + info = self._display.xrandr_get_crtc_info(crtc, timestamp) + if not info.mode or info.width <= 0 or info.height <= 0: + return None + return OutputRect(x=info.x, y=info.y, width=info.width, height=info.height) + + diff --git a/gatelock/gatelock/_xrandr.py b/gatelock/gatelock/_xrandr.py new file mode 100644 index 0000000..953036a --- /dev/null +++ b/gatelock/gatelock/_xrandr.py @@ -0,0 +1,112 @@ +"""Reading outputs by shelling out to `xrandr --query`. + +Split from :mod:`gatelock._outputs`, which keeps the value types and the +enumerator that picks a source. This is the fallback path: slower than the +RandR backend in :mod:`gatelock._randr` and it cannot deliver change events, +but it needs no optional dependency, so it is what runs on a bare install. +""" + +from __future__ import annotations + +import logging +import re +import shutil +import subprocess + +from gatelock._output_types import Output, OutputRect + +_logger = logging.getLogger(__name__) + +_XRANDR_TIMEOUT_S = 5.0 + +# Output lines start at column 0; mode lines are indented four spaces. That +# anchor is the whole robustness story -- without it, the mode line +# " 2560x1440 59.95*+" parses as a geometry and every modeless output +# silently looks live. +_OUTPUT_LINE = re.compile( + r"^(?P\S+)\s+(?Pconnected|disconnected)\b(?P.*)$", + re.MULTILINE, +) +_GEOMETRY = re.compile(r"(?P\d+)x(?P\d+)\+(?P-?\d+)\+(?P-?\d+)") + +_GEOMETRY = re.compile(r"(?P\d+)x(?P\d+)\+(?P-?\d+)\+(?P-?\d+)") + +def _parse_rect(head: str) -> OutputRect | None: + """Parse a geometry token out of an xrandr output line's head segment.""" + match = _GEOMETRY.search(head) + if match is None: + return None + width = int(match.group("w")) + height = int(match.group("h")) + if width <= 0 or height <= 0: + return None + return OutputRect( + x=int(match.group("x")), + y=int(match.group("y")), + width=width, + height=height, + ) + + +def parse_xrandr_query(text: str) -> tuple[Output, ...]: + """Parse ``xrandr --query`` output into outputs. + + Args: + text: Raw stdout from ``xrandr --query``. + + Returns: + One :class:`Output` per output line, in the order xrandr listed them. + """ + outputs: list[Output] = [] + for match in _OUTPUT_LINE.finditer(text): + # Everything after the first "(" is rotation/reflection vocabulary and + # panning noise -- never geometry. Slicing it off keeps the search + # honest. + head = match.group("rest").split("(", 1)[0] + outputs.append( + Output( + name=match.group("name"), + connected=match.group("state") == "connected", + rect=_parse_rect(head), + primary="primary" in head.split(), + ) + ) + return tuple(outputs) + + +def scan_xrandr() -> tuple[Output, ...] | None: + """Enumerate outputs by shelling out to ``xrandr --query``. + + Returns: + The parsed outputs, or None if xrandr is missing, failed, timed out or + produced nothing parseable. + """ + binary = shutil.which("xrandr") + if binary is None: + _logger.warning("xrandr is not on PATH; cannot enumerate outputs") + return None + try: + completed = subprocess.run( + [binary, "--query"], + capture_output=True, + text=True, + timeout=_XRANDR_TIMEOUT_S, + check=False, + ) + except (subprocess.TimeoutExpired, OSError) as exc: + _logger.warning("xrandr --query failed to run: %s", exc) + return None + if completed.returncode != 0: + _logger.warning( + "xrandr --query exited %d: %s", + completed.returncode, + completed.stderr.strip(), + ) + return None + outputs = parse_xrandr_query(completed.stdout) + if not outputs: + _logger.warning("xrandr --query produced no parseable output lines") + return None + return outputs + + diff --git a/gatelock/gatelock/tests/conftest.py b/gatelock/gatelock/tests/conftest.py index c7d5015..dcea351 100644 --- a/gatelock/gatelock/tests/conftest.py +++ b/gatelock/gatelock/tests/conftest.py @@ -73,7 +73,7 @@ def _hermetic(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Generator[None monkeypatch.setenv("GATELOCK_RUNTIME_DIR", str(tmp_path / "runtime")) with ( patch("gatelock._surfaces.tk.Toplevel", side_effect=make_toplevel), - patch("gatelock._outputs.RandrBackend.create", return_value=None), + patch("gatelock._randr.RandrBackend.create", return_value=None), patch("gatelock._outputs.scan_xrandr", return_value=DEFAULT_OUTPUTS), patch("gatelock._detect._RandrEventSource.start", return_value=False), patch("gatelock._vt.shutil.which", return_value="/usr/bin/setxkbmap"), diff --git a/gatelock/gatelock/tests/test_outputs.py b/gatelock/gatelock/tests/test_outputs.py index 45aba58..25b4706 100644 --- a/gatelock/gatelock/tests/test_outputs.py +++ b/gatelock/gatelock/tests/test_outputs.py @@ -12,12 +12,11 @@ from unittest.mock import MagicMock, patch import pytest +from gatelock._randr import RandrBackend, _decode_name +from gatelock._xrandr import parse_xrandr_query from gatelock._outputs import ( Output, OutputRect, - RandrBackend, - _decode_name, - parse_xrandr_query, ) from gatelock._outputs import scan_xrandr as real_scan_xrandr diff --git a/gatelock/gatelock/tests/test_outputs_enumerator.py b/gatelock/gatelock/tests/test_outputs_enumerator.py index 4c582ab..6c9a9b4 100644 --- a/gatelock/gatelock/tests/test_outputs_enumerator.py +++ b/gatelock/gatelock/tests/test_outputs_enumerator.py @@ -9,12 +9,12 @@ from __future__ import annotations from unittest.mock import MagicMock, patch +from gatelock._randr import RandrBackend from gatelock._outputs import ( Output, OutputEnumerator, OutputRect, OutputScan, - RandrBackend, enumerate_outputs, ) diff --git a/gatelock/gatelock/tests/test_outputs_randr.py b/gatelock/gatelock/tests/test_outputs_randr.py index 66351d2..894e5a7 100644 --- a/gatelock/gatelock/tests/test_outputs_randr.py +++ b/gatelock/gatelock/tests/test_outputs_randr.py @@ -8,9 +8,9 @@ from __future__ import annotations from unittest.mock import MagicMock, patch +from gatelock._randr import RandrBackend from gatelock._outputs import ( OutputRect, - RandrBackend, ) # Bound at import, so the autouse hermetic patch (which replaces these