import math
from collections.abc import Iterable, Iterator
from unittest.mock import MagicMock, patch

import numpy as np
import numpy.typing as npt
import pytest
from src.atmos.loudness_check import (
    LOUDNESS_WARN_LKFS,
    TRUE_PEAK_WARN_DBTP,
    LoudnessMeasurement,
    _check_loudness,
    _check_true_peak,
    _measure_loudnorm,
    _parse_loudnorm_output,
    apply_loudness_measurement,
)
from src.atmos.models import (
    AtmosValidationBuilder,
    AtmosValidationFindingCode,
    AtmosValidationMetadata,
)
from src.atmos.render_errors import FfmpegError, FfmpegUnavailable, LoudnormOutputError

_LOUDNESS_KEY = AtmosValidationFindingCode.LOUDNESS_TOO_HIGH.metadata_key
_TRUE_PEAK_KEY = AtmosValidationFindingCode.TRUE_PEAK_TOO_HIGH.metadata_key

_LOUDNORM_STDERR = (
    "[Parsed_loudnorm_0 @ 0x55d0] \n"
    "{\n"
    '\t"input_i" : "-9.05",\n'
    '\t"input_tp" : "-0.92",\n'
    '\t"input_lra" : "0.40",\n'
    '\t"input_thresh" : "-19.13",\n'
    '\t"output_i" : "-24.31",\n'
    '\t"normalization_type" : "dynamic",\n'
    '\t"target_offset" : "-0.69"\n'
    "}\n"
)


def _atmos_metadata() -> AtmosValidationMetadata:
    return AtmosValidationMetadata(
        container="Wave",
        codec="PCM",
        codec_id="1",
        channels=12,
        sample_rate_hz=48000,
        bits_per_sample=24,
        is_truncated=False,
        adm_profile="Dolby Atmos Master",
        duration_ms=180500,
    )


class TestParseLoudnormOutput:
    def test_parses_input_i_and_true_peak(self) -> None:
        measurement = _parse_loudnorm_output(_LOUDNORM_STDERR)

        assert measurement.integrated_loudness_lkfs == -9.05
        assert measurement.true_peak_dbtp == -0.92

    def test_ignores_braces_in_earlier_log_lines(self) -> None:
        # A "{" in an earlier ffmpeg log line must not mis-anchor the slice — rindex takes the
        # loudnorm object (the last brace-block), not the first "{".
        noisy = "[Parsed_aformat_0 @ 0x55d0] config {layout: 5.1}\n" + _LOUDNORM_STDERR
        measurement = _parse_loudnorm_output(noisy)

        assert measurement.integrated_loudness_lkfs == -9.05
        assert measurement.true_peak_dbtp == -0.92

    def test_raises_when_no_json_present(self) -> None:
        with pytest.raises(LoudnormOutputError):
            _parse_loudnorm_output("ffmpeg failed before emitting loudnorm output")

    def test_raises_when_required_keys_missing(self) -> None:
        with pytest.raises(LoudnormOutputError):
            _parse_loudnorm_output('{"input_lra": "0.40"}')


class TestThresholds:
    def test_match_apple_published_limits(self) -> None:
        # Locks the gate values to Apple's verbatim "should not exceed" figures
        # (quoted on the AtmosValidationFindingCode members).
        assert LOUDNESS_WARN_LKFS == -18.0
        assert TRUE_PEAK_WARN_DBTP == -1.0


class TestCheckLoudness:
    def test_exactly_at_apple_limit_no_warning(self) -> None:
        # Apple: "should not exceed -18 LKFS" -> exactly -18 passes (strict >).
        atmos_validation_builder = AtmosValidationBuilder()
        _check_loudness(
            LoudnessMeasurement(integrated_loudness_lkfs=-18.0, true_peak_dbtp=-5.0),
            atmos_validation_builder,
        )

        assert _LOUDNESS_KEY not in atmos_validation_builder.warnings

    def test_above_apple_limit_warns(self) -> None:
        atmos_validation_builder = AtmosValidationBuilder()
        _check_loudness(
            LoudnessMeasurement(integrated_loudness_lkfs=-17.9, true_peak_dbtp=-5.0),
            atmos_validation_builder,
        )

        assert _LOUDNESS_KEY in atmos_validation_builder.warnings


class TestCheckTruePeak:
    def test_exactly_at_apple_limit_no_warning(self) -> None:
        # Apple: "should not exceed -1 dB TP" -> exactly -1 passes (strict >).
        atmos_validation_builder = AtmosValidationBuilder()
        _check_true_peak(
            LoudnessMeasurement(integrated_loudness_lkfs=-20.0, true_peak_dbtp=-1.0),
            atmos_validation_builder,
        )

        assert _TRUE_PEAK_KEY not in atmos_validation_builder.warnings

    def test_above_apple_limit_warns(self) -> None:
        atmos_validation_builder = AtmosValidationBuilder()
        _check_true_peak(
            LoudnessMeasurement(integrated_loudness_lkfs=-20.0, true_peak_dbtp=-0.9),
            atmos_validation_builder,
        )

        assert _TRUE_PEAK_KEY in atmos_validation_builder.warnings


class TestApplyLoudnessMeasurement:
    def test_stores_measurement_in_metadata_and_emits_warnings(self) -> None:
        atmos_validation_builder = AtmosValidationBuilder()
        atmos_validation_builder.metadata = _atmos_metadata()

        apply_loudness_measurement(
            LoudnessMeasurement(integrated_loudness_lkfs=-9.0, true_peak_dbtp=0.5),
            atmos_validation_builder,
        )

        assert atmos_validation_builder.metadata is not None
        assert atmos_validation_builder.metadata.integrated_loudness_lkfs == -9.0
        assert atmos_validation_builder.metadata.true_peak_dbtp == 0.5
        assert _LOUDNESS_KEY in atmos_validation_builder.warnings
        assert _TRUE_PEAK_KEY in atmos_validation_builder.warnings

    def test_in_spec_measurement_emits_no_warnings(self) -> None:
        atmos_validation_builder = AtmosValidationBuilder()
        atmos_validation_builder.metadata = _atmos_metadata()

        apply_loudness_measurement(
            LoudnessMeasurement(integrated_loudness_lkfs=-18.0, true_peak_dbtp=-2.0),
            atmos_validation_builder,
        )

        assert atmos_validation_builder.warnings == {}


_POPEN = "src.atmos.loudness_check.subprocess.Popen"

_SILENT_BLOCK = np.zeros((1024, 6), dtype=np.float64)


def _fake_ffmpeg_process(*, stderr: bytes = b"", returncode: int = 0) -> MagicMock:
    process = MagicMock()
    process.stderr.read.return_value = stderr
    process.wait.return_value = returncode
    process.poll.return_value = returncode
    return process


def _exploding_blocks() -> Iterator[npt.NDArray[np.float64]]:
    # A block iterable that raises during iteration (in the feeder thread), not when first consumed —
    # models a render failure surfacing mid-stream.
    yield from ()
    raise RuntimeError("render exploded")


def _measure(blocks: Iterable[npt.NDArray[np.float64]]) -> LoudnessMeasurement:
    return _measure_loudnorm(blocks, sample_rate_hz=48000, channel_count=6, ffmpeg_channel_layout="5.1")


class TestMeasureErrors:
    def test_popen_oserror_raises_ffmpeg_unavailable(self) -> None:
        with (
            patch(_POPEN, side_effect=OSError("ffmpeg not found")),
            pytest.raises(FfmpegUnavailable),
        ):
            _measure([_SILENT_BLOCK])

    def test_nonzero_exit_raises_ffmpeg_error(self) -> None:
        with (
            patch(_POPEN, return_value=_fake_ffmpeg_process(stderr=b"boom", returncode=1)),
            pytest.raises(FfmpegError, match="exited with code 1"),
        ):
            _measure([_SILENT_BLOCK])

    def test_render_failure_during_feed_propagates_as_itself(self) -> None:
        # A render/data failure raised by the block generator is not an ffmpeg streaming failure — it
        # surfaces with its own type so it maps to the right failure category, not relabeled Ffmpeg*.
        with (
            patch(_POPEN, return_value=_fake_ffmpeg_process()),
            pytest.raises(RuntimeError, match="render exploded"),
        ):
            _measure(_exploding_blocks())

    def test_broken_pipe_during_feed_raises_ffmpeg_error(self) -> None:
        # A broken stdin pipe (ffmpeg died mid-stream) IS an ffmpeg streaming failure.
        process = _fake_ffmpeg_process()
        process.stdin.write.side_effect = BrokenPipeError("ffmpeg closed stdin")
        with (
            patch(_POPEN, return_value=process),
            pytest.raises(FfmpegError, match="streaming"),
        ):
            _measure([_SILENT_BLOCK])

    def test_kills_ffmpeg_if_error_escapes_before_completion(self) -> None:
        process = _fake_ffmpeg_process()
        process.stderr.read.side_effect = OSError("stderr pipe broke")
        process.poll.return_value = None  # ffmpeg still running when the error escapes
        with (
            patch(_POPEN, return_value=process),
            pytest.raises(OSError, match="stderr pipe broke"),
        ):
            _measure([_SILENT_BLOCK])

        process.kill.assert_called_once()


class TestMeasureLoudnorm:
    def test_measures_synthetic_blocks_independent_of_render(self) -> None:
        # The loudnorm consumer is render-agnostic: feed it raw f32 blocks (not from AtmosRender)
        # and it still produces a measurement — the decoupling that lets any render stream into it.
        sample_rate_hz = 48000
        t = np.arange(sample_rate_hz) / sample_rate_hz
        tone = 0.5 * np.sin(2 * np.pi * 1000 * t)
        block = np.zeros((sample_rate_hz, 6), dtype=np.float64)
        block[:, 0] = tone
        block[:, 1] = tone

        measurement = _measure_loudnorm(
            [block], sample_rate_hz=sample_rate_hz, channel_count=6, ffmpeg_channel_layout="5.1"
        )

        assert math.isfinite(measurement.true_peak_dbtp)
        assert measurement.true_peak_dbtp < 12.0
        assert not math.isnan(measurement.integrated_loudness_lkfs)