session_controller.py

"""Session controller: drives a real /usr/bin/caffeinate owned process.

Vocabulary (binding, per docs/DOMAIN.md): session, owned process, duration,
mode, active, inactive. This module imports stdlib only; it must never
import rumps or any other UI dependency (see docs/ARCHITECTURE.md).
"""

from __future__ import annotations

import enum
import math
import subprocess
import time
from collections.abc import Callable
from dataclasses import dataclass

CAFFEINATE_EXECUTABLE = "/usr/bin/caffeinate"


class Mode(enum.Enum):
    """A session's plain-language wake mode."""

    KEEP_MAC_AWAKE = "keep_mac_awake"
    KEEP_MAC_AND_DISPLAY_AWAKE = "keep_mac_and_display_awake"

    @property
    def argv(self) -> list[str]:
        if self is Mode.KEEP_MAC_AWAKE:
            return ["-i"]
        if self is Mode.KEEP_MAC_AND_DISPLAY_AWAKE:
            return ["-i", "-d"]
        raise AssertionError(f"unhandled mode: {self}")


class Duration(enum.Enum):
    """A session's requested length."""

    THIRTY_MINUTES = "30_minutes"
    ONE_HOUR = "1_hour"
    TWO_HOURS = "2_hours"
    INDEFINITE = "indefinite"

    @property
    def seconds(self) -> int | None:
        if self is Duration.THIRTY_MINUTES:
            return 1_800
        if self is Duration.ONE_HOUR:
            return 3_600
        if self is Duration.TWO_HOURS:
            return 7_200
        if self is Duration.INDEFINITE:
            return None
        raise AssertionError(f"unhandled duration: {self}")


@dataclass(frozen=True)
class SessionConfiguration:
    """A validated selected configuration for a session."""

    duration: Duration
    mode: Mode


@dataclass(frozen=True)
class SessionObservation:
    """One coherent observation of the owned process and live session."""

    active: bool
    configuration: SessionConfiguration | None
    remaining_seconds: int | None


def argument_vector(configuration: SessionConfiguration) -> list[str]:
    """Build the caffeinate argument vector from a session configuration."""
    argv = [CAFFEINATE_EXECUTABLE, *configuration.mode.argv]
    seconds = configuration.duration.seconds
    if seconds is not None:
        argv.extend(["-t", str(seconds)])
    return argv


class SessionLaunchError(Exception):
    """Raised when the owned process fails to launch."""


ProcessLauncher = Callable[[list[str]], subprocess.Popen[bytes]]
MonotonicClock = Callable[[], float]


def _launch_process(argv: list[str]) -> subprocess.Popen[bytes]:
    return subprocess.Popen(argv, shell=False)


class SessionController:
    """Owns at most one caffeinate process at a time."""

    def __init__(
        self,
        process_launcher: ProcessLauncher = _launch_process,
        monotonic_clock: MonotonicClock = time.monotonic,
    ) -> None:
        self._process_launcher = process_launcher
        self._monotonic_clock = monotonic_clock
        self._process: subprocess.Popen[bytes] | None = None
        self._configuration: SessionConfiguration | None = None
        self._deadline: float | None = None

    def start(self, configuration: SessionConfiguration) -> None:
        if self.observe().active:
            self.stop()

        try:
            process = self._process_launcher(argument_vector(configuration))
        except OSError as error:
            self._clear_session()
            raise SessionLaunchError("Could not start session.") from error

        seconds = configuration.duration.seconds
        deadline = (
            None
            if seconds is None
            else self._monotonic_clock() + seconds
        )
        self._process = process
        self._configuration = configuration
        self._deadline = deadline

    def stop(self) -> None:
        process = self._process
        if process is None:
            self._clear_session()
            return

        if process.poll() is None:
            process.terminate()
            try:
                process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait(timeout=5)

        self._clear_session()

    def observe(self) -> SessionObservation:
        """Observe the retained process once and return a coherent snapshot."""
        process = self._process
        if process is None:
            return SessionObservation(False, None, None)

        if process.poll() is not None:
            self._clear_session()
            return SessionObservation(False, None, None)

        configuration = self._configuration
        assert configuration is not None
        deadline = self._deadline
        remaining_seconds = (
            None
            if deadline is None
            else math.ceil(max(0.0, deadline - self._monotonic_clock()))
        )
        return SessionObservation(True, configuration, remaining_seconds)

    def owned_process_id(self) -> int | None:
        observation = self.observe()
        if not observation.active:
            return None
        assert self._process is not None
        return self._process.pid

    def _clear_session(self) -> None:
        self._process = None
        self._configuration = None
        self._deadline = None