"""Live peak/amplitude metering, independent of the volume *setting* — the
whole point is to make "the app is playing but nothing is reaching the mix"
visually obvious even when volume sliders read 100%, instead of only ever
showing what the volume knob is set to.

PipeWire/pipewire-pulse has no `pactl` subcommand for this (pactl only reports
configured volume, never signal level), so this shells out to `parec`
(pipewire-pulse's PulseAudio-protocol recorder) in raw-PCM mode and computes
peak amplitude client-side from the byte stream — the same technique
pavucontrol's meters use under the hood. `parec --monitor-stream=<id>` taps a
single application's own stream (Applications tab); `parec --device=<name>`
taps a physical device directly (`<sink>.monitor` for outputs, the source
name itself for inputs).

One `parec` process per distinct target, shared across every card watching
it, and only running while that card is on the currently-visible tab of a
currently-visible panel (see ui/card_base.py's start_peak/stop_peak, driven by
ui/scroll_row.py's activate/deactivate) — a mixer panel comfortably shows a
dozen live meters, but there's no reason to keep any of them decoding audio
while the panel is hidden.
"""

from __future__ import annotations

import array
from typing import Callable

import gi

gi.require_version("Gtk", "4.0")
from gi.repository import Gio, GLib  # noqa: E402

CHUNK_BYTES = 512     # ~256 s16le mono samples per read, small enough to feel live
RATE_HZ = 8000        # plenty of resolution for a peak/VU bar, cheap to decode
LATENCY_MSEC = 60


def _peak_from_pcm16(data: bytes) -> float:
    usable = len(data) - (len(data) % 2)
    if usable <= 0:
        return 0.0
    samples = array.array("h")
    samples.frombytes(data[:usable])
    peak = max((abs(s) for s in samples), default=0)
    return min(1.0, peak / 32768.0)


class PeakMonitorManager:
    def __init__(self) -> None:
        self._monitors: dict[str, dict] = {}

    def watch(self, key: str, target_args: list[str],
              on_level: Callable[[float], None]) -> Callable[[], None]:
        """target_args is the parec argv tail identifying what to tap, e.g.
        ["--device=alsa_output.foo.monitor"] or ["--monitor-stream=42"].
        Returns an unwatch() callback; call it to stop receiving levels."""
        entry = self._monitors.get(key)
        if entry is None:
            entry = self._start(key, target_args)
        entry["callbacks"].add(on_level)

        def unwatch() -> None:
            live = self._monitors.get(key)
            if live is None:
                return
            live["callbacks"].discard(on_level)
            if not live["callbacks"]:
                self._stop(key)

        return unwatch

    def _start(self, key: str, target_args: list[str]) -> dict:
        argv = ["parec", "--format=s16le", f"--rate={RATE_HZ}", "--channels=1",
                f"--latency-msec={LATENCY_MSEC}", *target_args]
        entry = {"proc": None, "stream": None, "callbacks": set()}
        self._monitors[key] = entry
        try:
            proc = Gio.Subprocess.new(
                argv, Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_SILENCE)
        except GLib.Error:
            return entry
        entry["proc"] = proc
        entry["stream"] = proc.get_stdout_pipe()
        self._read_next(key)
        return entry

    def _read_next(self, key: str) -> None:
        entry = self._monitors.get(key)
        if entry is None or entry["stream"] is None:
            return
        entry["stream"].read_bytes_async(CHUNK_BYTES, GLib.PRIORITY_DEFAULT, None,
                                         self._on_bytes, key)

    def _on_bytes(self, stream, result, key: str) -> None:
        entry = self._monitors.get(key)
        if entry is None:
            return
        try:
            data = stream.read_bytes_finish(result).get_data() or b""
        except GLib.Error:
            data = b""
        if not data:
            # target went away (app closed, device unplugged) — the owning
            # card's next refresh will drop it entirely; just stop quietly.
            self._stop(key)
            return
        level = _peak_from_pcm16(data)
        for cb in list(entry["callbacks"]):
            cb(level)
        self._read_next(key)

    def _stop(self, key: str) -> None:
        entry = self._monitors.pop(key, None)
        if entry is not None and entry.get("proc") is not None:
            entry["proc"].force_exit()

    def stop_all(self) -> None:
        for key in list(self._monitors.keys()):
            self._stop(key)