keyboard_brightness.py

#!/usr/bin/env python3
"""Drive KBPulse keyboard backlight intensity from an MSIG1 signal."""


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

COMMAND_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = (
    COMMAND_DIR.parent.parent
    if COMMAND_DIR.name == "bin" and COMMAND_DIR.parent.name == ".venv"
    else COMMAND_DIR
)
LIB_ROOT = PROJECT_ROOT / "lib"
for _p in reversed((LIB_ROOT, PROJECT_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__)


HOLDER_PIDFILE = Path("/tmp/apple-hardware-keyboard-brightness-set.pid")


def _read_holder_pid() -> int | None:
    try:
        txt = HOLDER_PIDFILE.read_text(encoding="utf-8").strip()
        if not txt:
            return None
        return int(txt)
    except Exception:
        return None


def _holder_running(pid: int) -> bool:
    if pid <= 0:
        return False
    try:
        os.kill(pid, 0)
        return True
    except OSError:
        return False


def _stop_holder() -> None:
    pid = _read_holder_pid()
    if pid is None:
        try:
            HOLDER_PIDFILE.unlink()
        except Exception:
            pass
        return
    try:
        os.kill(pid, signal.SIGTERM)
    except OSError:
        pass
    for _ in range(20):
        if not _holder_running(pid):
            break
        time.sleep(0.05)
    if _holder_running(pid):
        try:
            os.kill(pid, signal.SIGKILL)
        except OSError:
            pass
    try:
        HOLDER_PIDFILE.unlink()
    except Exception:
        pass


class BeatFollower:
    """Envelope follower with onset path + deterministic low-frequency pulse path."""

    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_onset = 0.0

        # Low-frequency path for smooth periodic signals (0.3-6 Hz).
        self.lf_min_hz = 0.3
        self.lf_max_hz = 6.0
        self.onset_lf = 0.0
        self.onset_lf_tau = 1.0 / (2.0 * math.pi * self.lf_max_hz)
        self.bp_hp_alpha = self.fs / (self.fs + 2.0 * math.pi * self.lf_min_hz)
        self.bp_lp_alpha = (2.0 * math.pi * self.lf_max_hz) / (
            (2.0 * math.pi * self.lf_max_hz) + self.fs
        )
        self.bp_hp_prev_in = 0.0
        self.bp_hp_prev_out = 0.0
        self.bp_lp_prev = 0.0
        self.lf_peak = 1e-6
        self.onset_energy = 0.0
        self.lf_energy = 0.0
        self.lf_mode = 0.0
        self.energy_tau = 1.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:
        s = float(sample)

        # Slow low-pass used to remove sub-6 Hz carrier from onset path.
        self.onset_lf = self._lpf(self.onset_lf, s, dt, self.onset_lf_tau)

        # Onset path ignores very-slow content so low-frequency carriers
        # don't produce double-rate pulses from full-wave rectification.
        x = abs(s - self.onset_lf)
        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 emphasis: transients above moving baseline drive flashes.
        onset = max(0.0, self.fast_env - self.slow_env)

        # Slow AGC so quiet/loud tracks both produce visible pulses.
        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_onset = max(norm, self.out_onset * decay)
        self.onset_energy = self._lpf(self.onset_energy, onset, dt, self.energy_tau)

        # 0.3-6 Hz first-order bandpass (high-pass then low-pass).
        hp = self.bp_hp_alpha * (self.bp_hp_prev_out + s - self.bp_hp_prev_in)
        self.bp_hp_prev_in = s
        self.bp_hp_prev_out = hp
        lf_band = self.bp_lp_alpha * hp + (1.0 - self.bp_lp_alpha) * self.bp_lp_prev
        self.bp_lp_prev = lf_band

        lf_mag = abs(lf_band)
        if lf_mag > self.lf_peak:
            self.lf_peak += 0.05 * (lf_mag - self.lf_peak)
        else:
            self.lf_peak += 0.001 * (lf_mag - self.lf_peak)
        lf_norm = max(-1.0, min(1.0, lf_band / max(1e-7, self.lf_peak)))
        # One pulse per LF cycle: square-gate the positive half-cycle.
        lf_level = 1.0 if lf_norm > 0.0 else 0.0

        self.lf_energy = self._lpf(self.lf_energy, lf_mag, dt, self.energy_tau)
        lf_dom = self.lf_energy > max(1e-6, self.onset_energy * 1.2)
        target_mode = 1.0 if lf_dom else 0.0
        self.lf_mode = self._lpf(self.lf_mode, target_mode, dt, 0.06)

        if self.lf_mode >= 0.5:
            return lf_level
        return self.out_onset


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Read MSIG1 mono signal and map its envelope to keyboard brightness."
    )
    parser.add_argument(
        "--send-hz",
        type=float,
        default=30.0,
        help="Maximum KBPulse update rate in Hz.",
    )
    parser.add_argument(
        "--fade-ms",
        type=int,
        default=20,
        help="KBPulse fade duration in milliseconds.",
    )
    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=220.0,
        help="Envelope release time in ms (default: 220).",
    )
    parser.add_argument(
        "--baseline-ms",
        type=float,
        default=800.0,
        help="Moving baseline time in ms (default: 800).",
    )
    parser.add_argument(
        "--decay-per-s",
        type=float,
        default=8.0,
        help="Output decay constant per second (default: 8.0).",
    )
    parser.add_argument(
        "--debug",
        action="store_true",
        help="Print live stats to stderr (rate, sends, level).",
    )
    parser.add_argument(
        "--as-root",
        action="store_true",
        help="Run KBPulse as root when this process is root (default: drop to sudo user).",
    )
    parser.add_argument(
        "--pulse",
        type=int,
        default=None,
        help="Ignore stdin and pulse keyboard lights 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 keyboard brightness percent (0..100). With --pulse this is pulse max.",
    )
    parser.add_argument(
        "--_hold-level",
        type=float,
        default=None,
        help=argparse.SUPPRESS,
    )
    return parser.parse_args()


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 launch_kbpulse_stdin, send_kbpulse_level, stop_kbpulse
    from lib.signal_stream import FloatSignalReader, StreamFormatError, is_tty_stdin

    if args._hold_level is not None:
        level = max(0.0, min(1.0, float(args._hold_level)))
        proc, err = launch_kbpulse_stdin(
            fade_ms=max(0, int(args.fade_ms)),
            run_as_user=not args.as_root,
            start_dir=str(PROJECT_ROOT),
        )
        if proc is None:
            raise SystemExit(f"failed to start KBPulse: {err}")
        running = True

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

        signal.signal(signal.SIGINT, _stop)
        signal.signal(signal.SIGTERM, _stop)
        try:
            while running:
                if proc.poll() is not None:
                    raise SystemExit("KBPulse holder exited unexpectedly")
                if not send_kbpulse_level(proc, level):
                    raise SystemExit("failed writing to KBPulse stdin")
                time.sleep(0.7)
            return 0
        finally:
            stop_kbpulse(proc, fade_ms=max(0, int(args.fade_ms)), reset=False)

    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 = max(0.0, min(1.0, float(target_pct) / 100.0))

        if args.pulse is None:
            # Persistent set mode: run a detached holder process so brightness
            # remains at target after this command exits.
            _stop_holder()
            if target_level <= 0.0:
                proc, err = launch_kbpulse_stdin(
                    fade_ms=max(0, int(args.fade_ms)),
                    run_as_user=not args.as_root,
                    start_dir=str(PROJECT_ROOT),
                )
                if proc is not None:
                    try:
                        send_kbpulse_level(proc, 0.0)
                        time.sleep(max(0.08, float(args.fade_ms) / 1000.0 + 0.05))
                    finally:
                        stop_kbpulse(proc, fade_ms=max(0, int(args.fade_ms)), reset=False)
                return 0

            cmd = [
                sys.executable,
                str(Path(__file__).resolve()),
                "--_hold-level",
                f"{target_level:.4f}",
                "--fade-ms",
                str(max(0, int(args.fade_ms))),
            ]
            if args.as_root:
                cmd.append("--as-root")
            try:
                child = subprocess.Popen(
                    cmd,
                    stdin=subprocess.DEVNULL,
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    start_new_session=True,
                    close_fds=True,
                )
                HOLDER_PIDFILE.write_text(str(child.pid), encoding="utf-8")
            except Exception as exc:
                raise SystemExit(f"failed to start keyboard brightness holder: {exc}") from exc
            return 0

        # Pulse mode is transient; stop any persistent holder first.
        _stop_holder()
        proc, err = launch_kbpulse_stdin(
            fade_ms=max(0, int(args.fade_ms)),
            run_as_user=not args.as_root,
            start_dir=str(PROJECT_ROOT),
        )
        if proc is None:
            raise SystemExit(f"failed to start KBPulse: {err}")

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

        try:
            count = int(args.pulse)
            for i in range(count):
                if not running:
                    break
                if not send_kbpulse_level(proc, target_level):
                    raise SystemExit("failed writing to KBPulse stdin")
                _sleep_while_running(float(args.on_time))
                if not send_kbpulse_level(proc, 0.0):
                    raise SystemExit("failed writing to KBPulse stdin")
                if i < (count - 1):
                    _sleep_while_running(float(args.off_time))
            return 0
        finally:
            stop_kbpulse(proc, fade_ms=max(0, int(args.fade_ms)), reset=False)

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

    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),
    )
    proc, err = launch_kbpulse_stdin(
        fade_ms=max(0, int(args.fade_ms)),
        run_as_user=not args.as_root,
        start_dir=str(PROJECT_ROOT),
    )
    if proc is None:
        raise SystemExit(f"failed to start KBPulse: {err}")

    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_level = -1.0
    min_delta = 0.015
    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:
        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
                if proc.poll() is not None:
                    raise RuntimeError("KBPulse exited unexpectedly")
                for sample in chunk:
                    sample_count += 1
                    level = follower.update(float(sample), dt)
                    last_level = level
                    clock += dt
                    if clock >= next_send:
                        if (
                            abs(level - last_sent_level) >= min_delta
                            or level <= 0.04
                            or last_sent_level <= 0.04
                        ):
                            if not send_kbpulse_level(proc, level):
                                raise RuntimeError("failed writing to KBPulse stdin")
                            sends += 1
                            last_sent_level = level
                        next_send = clock + send_dt
                if args.debug:
                    now = time.monotonic()
                    span = now - last_debug
                    if span >= 1.0:
                        est_hz = sample_count / max(1e-6, span)
                        send_hz = sends / max(1e-6, span)
                        print(
                            f"[kb] in_hz~{est_hz:.0f} send_hz~{send_hz:.1f} lvl={last_level:.3f} proc_alive={proc.poll() is None}",
                            file=sys.stderr,
                            flush=True,
                        )
                        sample_count = 0
                        sends = 0
                        last_debug = now
        except RuntimeError as exc:
            print(f"keyboard-brightness: {exc}", file=sys.stderr)
            return 1
    finally:
        stop_kbpulse(proc, fade_ms=max(0, int(args.fade_ms)))

    return 0


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