screen_brightness.py

#!/usr/bin/env python3
"""Drive display brightness from an incoming MSIG1 signal envelope."""


import argparse
import math
import signal
import sys
import time
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent
LIB_ROOT = REPO_ROOT / "lib"
for _p in reversed((LIB_ROOT, REPO_ROOT)):
    if str(_p) not in sys.path:
        sys.path.insert(0, str(_p))

from lib.bootstrap import maybe_reexec_venv

maybe_reexec_venv(__file__)


class BeatFollower:
    """Beat-friendly envelope follower with baseline subtraction and AGC."""

    def __init__(
        self,
        sample_rate: float,
        *,
        attack_ms: float,
        release_ms: float,
        baseline_ms: float,
        decay_per_s: float,
        gain: float,
    ) -> None:
        self.fs = max(1.0, float(sample_rate))
        self.attack_s = max(1e-4, float(attack_ms) / 1000.0)
        self.release_s = max(1e-4, float(release_ms) / 1000.0)
        self.baseline_s = max(1e-3, float(baseline_ms) / 1000.0)
        self.decay_per_s = max(0.1, float(decay_per_s))
        self.gain = max(0.0, float(gain))

        self.fast_env = 0.0
        self.slow_env = 0.0
        self.agc_peak = 1e-6
        self.out = 0.0

    @staticmethod
    def _lpf(prev: float, x: float, dt: float, tau: float) -> float:
        a = math.exp(-max(0.0, dt) / max(1e-6, tau))
        return a * prev + (1.0 - a) * x

    def update(self, sample: float, dt: float) -> float:
        x = abs(float(sample))
        tau = self.attack_s if x > self.fast_env else self.release_s
        self.fast_env = self._lpf(self.fast_env, x, dt, tau)
        self.slow_env = self._lpf(self.slow_env, self.fast_env, dt, self.baseline_s)

        onset = max(0.0, self.fast_env - self.slow_env)

        if onset > self.agc_peak:
            self.agc_peak += 0.08 * (onset - self.agc_peak)
        else:
            self.agc_peak += 0.002 * (onset - self.agc_peak)

        norm = onset / max(1e-7, self.agc_peak)
        norm = max(0.0, min(1.0, norm * self.gain))

        decay = math.exp(-self.decay_per_s * max(0.0, dt))
        self.out = max(norm, self.out * decay)
        return self.out


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Read MSIG1 mono signal and map beat/envelope to display brightness."
    )
    parser.add_argument(
        "--send-hz",
        type=float,
        default=20.0,
        help="Maximum display brightness update rate in Hz.",
    )
    parser.add_argument(
        "--min-level",
        type=float,
        default=0.08,
        help="Minimum output brightness level (0..1).",
    )
    parser.add_argument(
        "--max-level",
        type=float,
        default=1.0,
        help="Maximum output brightness level (0..1).",
    )
    parser.add_argument(
        "--gain",
        type=float,
        default=1.4,
        help="Post-normalization gain (default: 1.4).",
    )
    parser.add_argument(
        "--attack-ms",
        type=float,
        default=12.0,
        help="Envelope attack time in ms (default: 12).",
    )
    parser.add_argument(
        "--release-ms",
        type=float,
        default=260.0,
        help="Envelope release time in ms (default: 260).",
    )
    parser.add_argument(
        "--baseline-ms",
        type=float,
        default=900.0,
        help="Moving baseline time in ms (default: 900).",
    )
    parser.add_argument(
        "--decay-per-s",
        type=float,
        default=6.0,
        help="Output decay constant per second (default: 6.0).",
    )
    parser.add_argument(
        "--no-restore",
        action="store_true",
        help="Do not restore the initial display brightness on exit.",
    )
    parser.add_argument(
        "--debug",
        action="store_true",
        help="Print live stats to stderr (rate, sends, level/backend).",
    )
    parser.add_argument(
        "--pulse",
        type=int,
        default=None,
        help="Ignore stdin and pulse screen brightness N times.",
    )
    parser.add_argument(
        "--on-time",
        type=float,
        default=1.2,
        help="Seconds to stay on per pulse (default: 1.2).",
    )
    parser.add_argument(
        "--off-time",
        type=float,
        default=5.5,
        help="Seconds to stay off per pulse (default: 5.5).",
    )
    parser.add_argument(
        "--set",
        dest="set_percent",
        type=float,
        default=None,
        help="Set brightness percent (0..100). With --pulse this is pulse max.",
    )
    return parser.parse_args()


def clamp01(value: float) -> float:
    return max(0.0, min(1.0, float(value)))


def main() -> int:
    args = parse_args()
    if args.send_hz <= 0:
        raise SystemExit("--send-hz must be > 0")
    if args.attack_ms <= 0 or args.release_ms <= 0 or args.baseline_ms <= 0:
        raise SystemExit("--attack-ms, --release-ms, and --baseline-ms must be > 0")
    if args.pulse is not None and args.pulse < 0:
        raise SystemExit("--pulse must be >= 0")
    if args.on_time < 0 or args.off_time < 0:
        raise SystemExit("--on-time and --off-time must be >= 0")
    if args.set_percent is not None and (args.set_percent < 0 or args.set_percent > 100):
        raise SystemExit("--set must be between 0 and 100")

    from lib.hardware import DisplayBrightnessController
    from lib.signal_stream import FloatSignalReader, StreamFormatError, is_tty_stdin

    lo = clamp01(args.min_level)
    hi = clamp01(args.max_level)
    if hi < lo:
        lo, hi = hi, lo

    controller = DisplayBrightnessController()
    if not controller.available:
        raise SystemExit("display brightness control is unavailable on this system")

    control_mode = (args.pulse is not None) or (args.set_percent is not None)
    if control_mode:
        running = True

        def _stop(_sig: int, _frame: object) -> None:
            nonlocal running
            running = False

        signal.signal(signal.SIGINT, _stop)
        signal.signal(signal.SIGTERM, _stop)

        target_pct = args.set_percent if args.set_percent is not None else 100.0
        target_level = clamp01(float(target_pct) / 100.0)

        def _sleep_while_running(seconds: float) -> None:
            end_t = time.monotonic() + max(0.0, float(seconds))
            while running:
                now = time.monotonic()
                if now >= end_t:
                    break
                time.sleep(min(0.05, end_t - now))

        if args.pulse is None:
            if not controller.set(target_level):
                return 1
            return 0

        initial_level = controller.get()
        count = int(args.pulse)
        try:
            for i in range(count):
                if not running:
                    break
                _ = controller.set(target_level)
                _sleep_while_running(float(args.on_time))
                _ = controller.set(0.0)
                if i < (count - 1):
                    _sleep_while_running(float(args.off_time))
            return 0
        finally:
            if not args.no_restore and initial_level is not None:
                _ = controller.set(initial_level)

    if is_tty_stdin():
        raise SystemExit("screen-brightness expects an MSIG1 stream on stdin")

    initial_level = controller.get()
    try:
        reader = FloatSignalReader.from_stdin()
    except (EOFError, StreamFormatError):
        return 0
    follower = BeatFollower(
        sample_rate=float(reader.sample_rate),
        attack_ms=float(args.attack_ms),
        release_ms=float(args.release_ms),
        baseline_ms=float(args.baseline_ms),
        decay_per_s=float(args.decay_per_s),
        gain=float(args.gain),
    )

    running = True
    send_dt = 1.0 / float(args.send_hz)
    dt = 1.0 / float(reader.sample_rate)
    clock = time.monotonic()
    next_send = clock
    last_sent = -1.0
    min_delta = 0.02
    sends = 0
    sample_count = 0
    last_debug = time.monotonic()
    last_level = 0.0

    def _stop(_sig: int, _frame: object) -> None:
        nonlocal running
        running = False

    signal.signal(signal.SIGINT, _stop)
    signal.signal(signal.SIGTERM, _stop)

    try:
        chunk_bytes = max(16, int(reader.sample_rate / max(1.0, args.send_hz)) * 4)
        for chunk in reader.iter_chunks(chunk_bytes=chunk_bytes):
            if not running:
                break
            for sample in chunk:
                sample_count += 1
                env = follower.update(float(sample), dt)
                target = lo + env * (hi - lo)
                last_level = target
                clock += dt
                if clock >= next_send:
                    if (
                        abs(target - last_sent) >= min_delta
                        or target <= lo + 0.02
                        or last_sent <= lo + 0.02
                    ):
                        if controller.set(target):
                            last_sent = target
                            sends += 1
                    next_send = clock + send_dt
            if args.debug:
                now = time.monotonic()
                span = now - last_debug
                if span >= 1.0:
                    in_hz = sample_count / max(1e-6, span)
                    send_hz = sends / max(1e-6, span)
                    print(
                        f"[screen] backend={controller.backend} in_hz~{in_hz:.0f} send_hz~{send_hz:.1f} lvl={last_level:.3f}",
                        file=sys.stderr,
                        flush=True,
                    )
                    sample_count = 0
                    sends = 0
                    last_debug = now
    finally:
        if not args.no_restore and initial_level is not None:
            _ = controller.set(initial_level)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())