"""Streaming measurement and FFmpeg execution. No shell interpolation."""
from collections import deque
import json
import math
import os
from pathlib import Path
import selectors
import signal
import subprocess
import time

import numpy as np
import soundfile as sf

from .setup import terminate_group


def ffmpeg_run(ffmpeg: str, args: list[str], stage: str, duration: float,
               progress) -> str:
    command = [ffmpeg, "-hide_banner", "-nostdin", "-y", "-nostats",
               "-progress", "pipe:1"] + args
    progress(stage, 0, duration or None)
    diagnostics = deque(maxlen=128)
    pending = b""
    completed = 0.0
    last_advance = time.monotonic()
    with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                          start_new_session=True) as process:
        try:
            with selectors.DefaultSelector() as selector:
                for stream in (process.stdout, process.stderr):
                    os.set_blocking(stream.fileno(), False)
                    selector.register(stream, selectors.EVENT_READ)
                while selector.get_map() or process.poll() is None:
                    for key, _ in selector.select(timeout=0.2):
                        data = os.read(key.fd, 8192)
                        if not data:
                            selector.unregister(key.fileobj)
                            continue
                        if key.fileobj is process.stderr:
                            diagnostics.append(data)
                        else:
                            pending += data
                            while b"\n" in pending:
                                line, pending = pending.split(b"\n", 1)
                                if line.startswith(b"out_time_us="):
                                    try:
                                        value = max(0.0, int(line.split(b"=", 1)[1]) / 1e6)
                                    except ValueError:
                                        continue
                                    if value > completed:
                                        last_advance = time.monotonic()
                                    completed = value
                    label = stage
                    if time.monotonic() - last_advance > 60:
                        label += " (no new audio progress for 60s; still waiting)"
                    progress(label, min(completed, duration), duration or None)
                code = process.wait()
            output = b"".join(diagnostics).decode(errors="replace")
            if code:
                raise RuntimeError(f"{stage} failed (exit {code}):\n{output[-8000:]}")
            progress(stage, duration, duration or None)
            return output
        except BaseException:
            terminate_group(process)
            raise


def measure(path: Path, progress, stage="Analyzing audio") -> dict:
    """100ms RMS histogram is bounded in memory, including multi-hour input."""
    with sf.SoundFile(path) as source:
        if source.channels not in (1, 2):
            raise ValueError("Only mono or stereo voice recordings are supported")
        if not len(source):
            raise ValueError("The recording is empty")
        peaks = np.zeros(source.channels)
        sums = np.zeros(source.channels)
        squares = np.zeros(source.channels)
        clipped = 0
        histogram = np.zeros(121, dtype=np.int64)
        frames = 0
        progress(stage, 0, len(source))
        for audio in source.blocks(blocksize=max(1, source.samplerate // 10),
                                   dtype="float64", always_2d=True):
            if not np.isfinite(audio).all():
                raise ValueError("Recording contains non-finite samples")
            peaks = np.maximum(peaks, np.max(np.abs(audio), axis=0))
            sums += audio.sum(axis=0)
            squares += (audio * audio).sum(axis=0)
            clipped += int(np.count_nonzero(np.abs(audio) >= 0.9999))
            rms = float(np.sqrt(np.mean(audio * audio)))
            db = 20 * math.log10(max(rms, 1e-6))
            histogram[int(np.clip(round(db) + 120, 0, 120))] += 1
            frames += len(audio)
            progress(stage, frames, len(source))
        cumulative = histogram.cumsum()
        def percentile(p):
            return int(np.searchsorted(cumulative, max(1, math.ceil(cumulative[-1] * p)))) - 120
        quiet, speech = percentile(0.1), percentile(0.8)
        if speech - quiet < 15:
            # Pause-dominated recordings: the 80th percentile of all blocks
            # sits in room tone, so estimate the level from the loud tail
            # instead of mistaking sparse active speech for silence.
            speech = max(speech, percentile(0.98))
        noise_available = speech - quiet >= 15 and quiet > -110
        return {"frames": frames, "duration_seconds": frames / source.samplerate,
                "sample_rate": source.samplerate, "channels": source.channels,
                "subtype": source.subtype,
                "sample_peak_dbfs": float(20 * np.log10(max(float(peaks.max()), 1e-12))),
                "channel_rms_dbfs": (20 * np.log10(np.maximum(np.sqrt(squares / frames), 1e-12))).tolist(),
                "dc_offset": (sums / frames).tolist(), "near_full_scale_samples": clipped,
                "speech_level_estimate_dbfs": speech,
                "quiet_blocks_dbfs": quiet,
                "noise_floor_estimate_dbfs": quiet if noise_available else None,
                "noise_floor_confidence": "low (quiet-block heuristic, not speech recognition)" if noise_available else "insufficient room tone"}


def loudness(ffmpeg, source, settings, duration, progress, stage="Measuring loudness"):
    output = ffmpeg_run(ffmpeg, ["-i", str(source), "-map", "0:a:0", "-af",
        f"loudnorm=I={settings.target_lufs}:TP={settings.true_peak_db}:LRA={settings.loudness_range}:print_format=json",
        "-f", "null", "-"], stage, duration, progress)
    start, end = output.rfind("{"), output.rfind("}")
    if start < 0 or end < start:
        raise RuntimeError("FFmpeg did not return loudness measurements")
    raw = json.loads(output[start:end + 1])
    return {key: (float(value) if math.isfinite(float(value)) else None)
            for key, value in raw.items() if key != "normalization_type"} | {
                "normalization_type": raw.get("normalization_type")}