| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | screen_rotate.py — Hyprland 0.55+ IPC-only Screen Rotation Utility |
| 4 | |
| 5 | Rotates the focused monitor 90° clockwise (+90) or counter-clockwise (-90). |
| 6 | |
| 7 | Usage: |
| 8 | screen_rotate.py +90 |
| 9 | screen_rotate.py -90 |
| 10 | |
| 11 | Design principles: |
| 12 | • Zero config-file access — reads/writes nothing on disk. |
| 13 | • Temporary by design — changes reset on `hyprctl reload`. |
| 14 | • Mode-string fidelity — resolves active mode natively without resolution corruption. |
| 15 | • Flip-transform aware — rotates within the current flip state (0-3 / 4-7). |
| 16 | • Keybind safe — pipes critical errors to notify-send and debounces rapid inputs. |
| 17 | """ |
| 18 | |
| 19 | from __future__ import annotations |
| 20 | |
| 21 | import fcntl |
| 22 | import json |
| 23 | import os |
| 24 | import subprocess |
| 25 | import sys |
| 26 | import tempfile |
| 27 | import time |
| 28 | from typing import Any |
| 29 | |
| 30 | |
| 31 | # ── Concurrency Lock ─────────────────────────────────────────────────────────── |
| 32 | |
| 33 | def acquire_lock() -> None: |
| 34 | """ |
| 35 | Acquire an exclusive lock. If another instance is running, exit silently. |
| 36 | This prevents race conditions when the keybind is spammed. |
| 37 | """ |
| 38 | lock_file = os.path.join(tempfile.gettempdir(), "hypr_screen_rotate.lock") |
| 39 | fd = os.open(lock_file, os.O_CREAT | os.O_RDWR) |
| 40 | try: |
| 41 | fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) |
| 42 | except BlockingIOError: |
| 43 | # Another instance is running. Silently drop this execution. |
| 44 | sys.exit(0) |
| 45 | # The file descriptor remains open and locked until the process exits. |
| 46 | # The OS automatically releases the lock when the script finishes. |
| 47 | |
| 48 | |
| 49 | # ── ANSI colour helpers ──────────────────────────────────────────────────────── |
| 50 | |
| 51 | _COLOURS = sys.stderr.isatty() or sys.stdout.isatty() |
| 52 | |
| 53 | def _c(code: str, text: str) -> str: |
| 54 | return f"\033[{code}m{text}\033[0m" if _COLOURS else text |
| 55 | |
| 56 | def _red(t: str) -> str: return _c("31", t) |
| 57 | def _green(t: str) -> str: return _c("32", t) |
| 58 | def _yellow(t: str) -> str: return _c("33", t) |
| 59 | def _blue(t: str) -> str: return _c("34", t) |
| 60 | def _bold(t: str) -> str: return _c("1", t) |
| 61 | |
| 62 | |
| 63 | # ── Logging ──────────────────────────────────────────────────────────────────── |
| 64 | |
| 65 | def log_info(msg: str) -> None: print(f"{_blue('[INFO]')} {msg}") |
| 66 | def log_ok(msg: str) -> None: print(f"{_green('[OK]')} {msg}") |
| 67 | def log_warn(msg: str) -> None: print(f"{_yellow('[WARN]')} {msg}", file=sys.stderr) |
| 68 | def log_payload(msg: str) -> None: print(f"{_yellow('[PAYLOAD]')} {msg}") |
| 69 | |
| 70 | def die(msg: str, code: int = 1) -> None: |
| 71 | """Print to stderr and send a desktop notification so keybinds don't fail silently.""" |
| 72 | print(f"{_red('[ERROR]')} {msg}", file=sys.stderr) |
| 73 | try: |
| 74 | subprocess.run( |
| 75 | [ |
| 76 | "notify-send", |
| 77 | "--app-name=screen_rotate.py", |
| 78 | "--urgency=critical", |
| 79 | "Screen Rotation Error", |
| 80 | msg |
| 81 | ], |
| 82 | capture_output=True, |
| 83 | timeout=2 |
| 84 | ) |
| 85 | except Exception: |
| 86 | pass |
| 87 | sys.exit(code) |
| 88 | |
| 89 | |
| 90 | # ── Dependency / environment checks ─────────────────────────────────────────── |
| 91 | |
| 92 | def check_environment() -> None: |
| 93 | if os.geteuid() == 0: |
| 94 | die("Do not run as root — hyprctl requires user-space socket access.") |
| 95 | |
| 96 | if not os.environ.get("HYPRLAND_INSTANCE_SIGNATURE"): |
| 97 | die( |
| 98 | "HYPRLAND_INSTANCE_SIGNATURE is not set.\n" |
| 99 | "Is Hyprland running, and are you in the correct session?" |
| 100 | ) |
| 101 | |
| 102 | result = subprocess.run( |
| 103 | ["hyprctl", "--version"], |
| 104 | capture_output=True, |
| 105 | text=True, |
| 106 | ) |
| 107 | if result.returncode != 0 and "not found" in (result.stderr or "").lower(): |
| 108 | die("'hyprctl' not found. Install Hyprland.") |
| 109 | |
| 110 | |
| 111 | # ── Argument parsing ─────────────────────────────────────────────────────────── |
| 112 | |
| 113 | def parse_direction() -> int: |
| 114 | prog = os.path.basename(sys.argv[0]) |
| 115 | |
| 116 | # Graceful help handler |
| 117 | if len(sys.argv) == 2 and sys.argv[1] in ("-h", "--help"): |
| 118 | print(f"Usage: {prog} [+90|-90]") |
| 119 | print("Rotates the focused monitor 90° clockwise (+90) or counter-clockwise (-90).") |
| 120 | sys.exit(0) |
| 121 | |
| 122 | # Strict arg checking for actual usage |
| 123 | if len(sys.argv) != 2 or sys.argv[1] not in ("+90", "-90"): |
| 124 | die(f"Invalid argument.\nUsage: {prog} [+90|-90]") |
| 125 | |
| 126 | return 1 if sys.argv[1] == "+90" else -1 |
| 127 | |
| 128 | |
| 129 | # ── hyprctl IPC wrappers ─────────────────────────────────────────────────────── |
| 130 | |
| 131 | def hyprctl_json(args: list[str]) -> Any: |
| 132 | try: |
| 133 | proc = subprocess.run( |
| 134 | ["hyprctl", "-j"] + args, |
| 135 | capture_output=True, |
| 136 | text=True, |
| 137 | timeout=5, |
| 138 | ) |
| 139 | except FileNotFoundError: |
| 140 | die("'hyprctl' not found.") |
| 141 | except subprocess.TimeoutExpired: |
| 142 | die("hyprctl timed out — is Hyprland responsive?") |
| 143 | |
| 144 | if proc.returncode != 0: |
| 145 | die(f"hyprctl exited {proc.returncode}: {proc.stderr.strip() or '(no stderr)'}") |
| 146 | |
| 147 | raw = proc.stdout.strip() |
| 148 | |
| 149 | if raw and not raw[0] in ("[", "{"): |
| 150 | for i, line in enumerate(raw.splitlines()): |
| 151 | line = line.strip() |
| 152 | if line.startswith(("[", "{")): |
| 153 | raw = "\n".join(raw.splitlines()[i:]) |
| 154 | break |
| 155 | |
| 156 | if not raw: |
| 157 | die("hyprctl returned empty output.") |
| 158 | |
| 159 | try: |
| 160 | return json.loads(raw) |
| 161 | except json.JSONDecodeError as exc: |
| 162 | die(f"Failed to parse hyprctl JSON: {exc}\nRaw output: {raw[:300]!r}") |
| 163 | |
| 164 | |
| 165 | def hyprctl_eval(lua_str: str) -> bool: |
| 166 | try: |
| 167 | proc = subprocess.run( |
| 168 | ["hyprctl", "eval", lua_str], |
| 169 | capture_output=True, |
| 170 | text=True, |
| 171 | timeout=5, |
| 172 | ) |
| 173 | except subprocess.TimeoutExpired: |
| 174 | die("hyprctl eval timed out — compositor may be unresponsive.") |
| 175 | |
| 176 | return proc.returncode == 0 |
| 177 | |
| 178 | |
| 179 | # ── Mode resolution ──────────────────────────────────────────────────────────── |
| 180 | |
| 181 | def _parse_hz(mode_str: str) -> float | None: |
| 182 | try: |
| 183 | _, hz_part = mode_str.split("@", 1) |
| 184 | return float(hz_part.rstrip("HhZz")) |
| 185 | except (ValueError, IndexError): |
| 186 | return None |
| 187 | |
| 188 | |
| 189 | def resolve_active_mode(monitor: dict[str, Any]) -> str: |
| 190 | """ |
| 191 | Return the best-matching physical mode string for `hl.monitor()` |
| 192 | (WITHOUT the Hz suffix, e.g. "1920x1080@144.00"). |
| 193 | """ |
| 194 | # Hyprland IPC `width` and `height` ALWAYS represent the physical |
| 195 | # native resolution. They do not swap when rotated. |
| 196 | native_w: int = int(monitor["width"]) |
| 197 | native_h: int = int(monitor["height"]) |
| 198 | refresh: float = float(monitor["refreshRate"]) |
| 199 | available: list = monitor.get("availableModes", []) |
| 200 | |
| 201 | target_prefix = f"{native_w}x{native_h}@" |
| 202 | |
| 203 | candidates: list[tuple[float, str]] = [] |
| 204 | for mode in available: |
| 205 | if not mode.lower().startswith(target_prefix.lower()): |
| 206 | continue |
| 207 | hz = _parse_hz(mode) |
| 208 | if hz is None: |
| 209 | continue |
| 210 | clean = mode.rstrip("HhZz") |
| 211 | candidates.append((hz, clean)) |
| 212 | |
| 213 | if not candidates: |
| 214 | log_warn( |
| 215 | f"No availableModes entry for {native_w}x{native_h} " |
| 216 | f"(IPC refreshRate={refresh:.5f}). " |
| 217 | "Reconstructing mode string from IPC data." |
| 218 | ) |
| 219 | return f"{native_w}x{native_h}@{refresh:.2f}" |
| 220 | |
| 221 | best_hz, best_clean = min(candidates, key=lambda c: abs(c[0] - refresh)) |
| 222 | return best_clean |
| 223 | |
| 224 | |
| 225 | # ── Transform arithmetic ─────────────────────────────────────────────────────── |
| 226 | |
| 227 | def compute_new_transform(current: int, direction: int) -> int: |
| 228 | flip_bit = current & 4 |
| 229 | rotation_bits = current & 3 |
| 230 | new_rotation = (rotation_bits + direction + 4) % 4 |
| 231 | return flip_bit | new_rotation |
| 232 | |
| 233 | |
| 234 | # ── Scale formatting ─────────────────────────────────────────────────────────── |
| 235 | |
| 236 | def format_scale(scale: float) -> str: |
| 237 | rounded = f"{scale:.6f}".rstrip("0").rstrip(".") |
| 238 | return rounded |
| 239 | |
| 240 | |
| 241 | # ── Main ─────────────────────────────────────────────────────────────────────── |
| 242 | |
| 243 | def main() -> None: |
| 244 | # 1. Establish concurrency lock immediately to debounce keybind spam |
| 245 | acquire_lock() |
| 246 | |
| 247 | check_environment() |
| 248 | direction = parse_direction() |
| 249 | |
| 250 | monitors: list[dict] = hyprctl_json(["monitors"]) |
| 251 | |
| 252 | if not isinstance(monitors, list) or not monitors: |
| 253 | die("hyprctl returned no monitor data.") |
| 254 | |
| 255 | focused: dict | None = next( |
| 256 | (m for m in monitors if m.get("focused") is True), None |
| 257 | ) |
| 258 | if focused is None: |
| 259 | log_warn("No focused monitor found; using first monitor.") |
| 260 | focused = monitors[0] |
| 261 | |
| 262 | name: str = str(focused.get("name", "")) |
| 263 | current_transform = int(focused.get("transform", 0)) |
| 264 | x: int = int(focused.get("x", 0)) |
| 265 | y: int = int(focused.get("y", 0)) |
| 266 | scale: float = float(focused.get("scale", 1.0)) |
| 267 | disabled: bool = bool(focused.get("disabled", False)) |
| 268 | |
| 269 | if not name or name == "null": |
| 270 | die("Could not read monitor name from IPC.") |
| 271 | |
| 272 | if disabled: |
| 273 | die(f"Monitor '{name}' is currently disabled; cannot rotate.") |
| 274 | |
| 275 | if current_transform not in range(8): |
| 276 | die(f"Unexpected transform value '{current_transform}' from IPC. Expected 0-7.") |
| 277 | |
| 278 | new_transform = compute_new_transform(current_transform, direction) |
| 279 | active_mode = resolve_active_mode(focused) |
| 280 | |
| 281 | pos_str = f"{x}x{y}" |
| 282 | scale_str = format_scale(scale) |
| 283 | lua_name = name.replace("\\", "\\\\").replace('"', '\\"') |
| 284 | |
| 285 | lua_call = ( |
| 286 | f'hl.monitor({{ output = "{lua_name}", mode = "{active_mode}", ' |
| 287 | f'position = "{pos_str}", scale = {scale_str}, transform = {new_transform} }})' |
| 288 | ) |
| 289 | |
| 290 | print() |
| 291 | log_info(f"Monitor : {_bold(name)}") |
| 292 | log_info(f"Mode : {active_mode}") |
| 293 | log_info(f"Position : {pos_str} Scale: {scale_str}") |
| 294 | log_info(f"Transform : {current_transform} → {new_transform} ({direction:+d} × 90°)") |
| 295 | log_payload(lua_call) |
| 296 | print() |
| 297 | |
| 298 | eval_ok = hyprctl_eval(lua_call) |
| 299 | if not eval_ok: |
| 300 | die( |
| 301 | "hyprctl eval returned a non-zero exit code.\n" |
| 302 | "Is Hyprland 0.55+ running and the socket accessible?\n" |
| 303 | f"Lua: {lua_call}" |
| 304 | ) |
| 305 | |
| 306 | actual_transform = current_transform |
| 307 | for _ in range(25): |
| 308 | time.sleep(0.1) |
| 309 | try: |
| 310 | polled = hyprctl_json(["monitors"]) |
| 311 | for m in polled: |
| 312 | if m.get("name") == name: |
| 313 | actual_transform = int(m.get("transform", current_transform)) |
| 314 | break |
| 315 | except SystemExit: |
| 316 | break |
| 317 | if actual_transform != current_transform: |
| 318 | break |
| 319 | |
| 320 | if actual_transform != new_transform: |
| 321 | die( |
| 322 | f"Transform did not change after eval (expected {new_transform}, IPC reports {actual_transform}).\n" |
| 323 | "This may indicate the compositor overrode the value.\n" |
| 324 | f"Lua sent: {lua_call}" |
| 325 | ) |
| 326 | |
| 327 | log_ok(f"Rotation applied — transform {current_transform} → {new_transform}") |
| 328 | _notify(name, new_transform, active_mode) |
| 329 | |
| 330 | |
| 331 | def _notify(monitor: str, transform: int, mode: str) -> None: |
| 332 | _TRANSFORM_NAMES = { |
| 333 | 0: "0° (normal)", 1: "90°", 2: "180°", 3: "270°", |
| 334 | 4: "0° (flipped)", 5: "90°+flip", 6: "180°+flip", 7: "270°+flip", |
| 335 | } |
| 336 | label = _TRANSFORM_NAMES.get(transform, str(transform)) |
| 337 | body = f"{monitor} • {label}" |
| 338 | |
| 339 | try: |
| 340 | subprocess.run( |
| 341 | [ |
| 342 | "notify-send", |
| 343 | "--app-name=hypr-rotate", |
| 344 | "--icon=object-rotate-right", |
| 345 | "-t", "1500", |
| 346 | "Display Rotated", |
| 347 | body, |
| 348 | "--hint=string:x-canonical-private-synchronous:display-rotate", |
| 349 | ], |
| 350 | capture_output=True, |
| 351 | timeout=3, |
| 352 | ) |
| 353 | except Exception: |
| 354 | pass |
| 355 | |
| 356 | |
| 357 | if __name__ == "__main__": |
| 358 | main() |