kuhyx/utils
1
0
Fork 0
mirror of https://github.com/kuhyx/utils.git synced 2026-11-10 19:07:17 +01:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity

gatelock: split _outputs.py 401 -> 137 across four modules

Browse Source 
_output_types.py 62  — Output, OutputRect, OutputScan (data only)
  _randr.py        157 — the optional python-xlib backend and its Protocols
  _xrandr.py       112 — the `xrandr --query` subprocess fallback
  _outputs.py      137 — the enumerator that picks between them

The two backends both needed the value types, which would have made the
dependency circular. The first attempt papered over that with function-local
imports; extracting the types into a leaf module removes the cycle instead, so
every import is top-level again — and, importantly, the tests' existing
`patch("gatelock._outputs.scan_xrandr")` targets keep working, which the
lazy-import version had silently broken.

_RR_CONNECTED moved with the RandR code and _XRANDR_TIMEOUT_S/_OUTPUT_LINE/
_GEOMETRY with the xrandr code, each carrying its comment. An earlier pass
tried to relocate those constants with a regex, which ate part of an unrelated
function body; that attempt was reverted and redone with asserted line
boundaries plus an AST check for unresolved names in both halves.

389 passed at 100% branch coverage; gatelock.__all__ still the same 58 names.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CFSGeWxHPY5rHUGrPqhcmG
This commit is contained in:
Krzysztof Rudnicki 2026-08-22 15:15:55 +02:00
parent 02659e0abf
commit 40f94833c1
9 changed files with 354 additions and 274 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
gatelock/gatelock/__init__.py
View File

@ -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,

62
gatelock/gatelock/_output_types.py Normal file
View File

@ -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)

282
gatelock/gatelock/_outputs.py
View File

@ -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<name>\S+)\s+(?P<state>connected|disconnected)\b(?P<rest>.*)$",
re.MULTILINE,
)
_GEOMETRY = re.compile(r"(?P<w>\d+)x(?P<h>\d+)\+(?P<x>-?\d+)\+(?P<y>-?\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.

157
gatelock/gatelock/_randr.py Normal file
View File

@ -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)

112
gatelock/gatelock/_xrandr.py Normal file
View File

@ -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<name>\S+)\s+(?P<state>connected|disconnected)\b(?P<rest>.*)$",
re.MULTILINE,
)
_GEOMETRY = re.compile(r"(?P<w>\d+)x(?P<h>\d+)\+(?P<x>-?\d+)\+(?P<y>-?\d+)")
_GEOMETRY = re.compile(r"(?P<w>\d+)x(?P<h>\d+)\+(?P<x>-?\d+)\+(?P<y>-?\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

2
gatelock/gatelock/tests/conftest.py
View File

@ -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"),

5
gatelock/gatelock/tests/test_outputs.py
View File

@ -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

2
gatelock/gatelock/tests/test_outputs_enumerator.py
View File

@ -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,
)

2
gatelock/gatelock/tests/test_outputs_randr.py
View File

@ -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