Blame for src/nida/data/stats.py

"""S8: dataset statistics (duration, tag/emotion distributions, characters per second).

Pure functions over small typed rows; no I/O. Durations are integer milliseconds.
Nothing is dropped silently: duration values outside the histogram edges land in explicit
underflow/overflow bins, and segments without tags or emotion are counted separately.
"""

from __future__ import annotations

import math
from bisect import bisect_right
from collections import Counter
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import asdict, dataclass
from itertools import pairwise

import numpy as np

__all__ = [
    "DEFAULT_CPS_PERCENTILES",
    "MS_PER_HOUR",
    "SegmentStat",
    "StatsSummary",
    "duration_histogram",
    "render_stats_markdown",
    "summarize",
]

MS_PER_HOUR = 3_600_000
_MS_PER_SECOND = 1000
DEFAULT_CPS_PERCENTILES: tuple[float, ...] = (1.0, 5.0, 50.0, 95.0, 99.0)


def _is_int(value: object) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


@dataclass(frozen=True)
class SegmentStat:
    """The per-segment facts S8 needs.

    Attributes:
        duration_ms: Segment length in milliseconds, > 0.
        num_chars: Characters of the final transcript used for the chars-per-second rate
            (the caller decides whether tags count), >= 0.
        tags: Tags occurring in the segment, one entry per occurrence.
        emotion: Emotion label of the segment, or ``None`` when unlabelled.
    """

    duration_ms: int
    num_chars: int
    tags: tuple[str, ...] = ()
    emotion: str | None = None

    def __post_init__(self) -> None:
        if not _is_int(self.duration_ms) or self.duration_ms <= 0:
            raise ValueError(f"duration_ms must be a positive int, got {self.duration_ms!r}")
        if not _is_int(self.num_chars) or self.num_chars < 0:
            raise ValueError(f"num_chars must be a non-negative int, got {self.num_chars!r}")
        if any(not isinstance(tag, str) or not tag for tag in self.tags):
            raise ValueError(f"tags must be non-empty strings, got {self.tags!r}")
        if self.emotion is not None and (not isinstance(self.emotion, str) or not self.emotion):
            raise ValueError(f"emotion must be None or a non-empty string, got {self.emotion!r}")

    @property
    def chars_per_second(self) -> float:
        return self.num_chars / (self.duration_ms / _MS_PER_SECOND)

    @staticmethod
    def from_mapping(
        row: Mapping[str, object],
        *,
        duration_key: str = "duration_ms",
        chars_key: str = "num_chars",
        tags_key: str = "tags",
        emotion_key: str = "emotion",
    ) -> SegmentStat:
        """Build from a manifest row, validating the types strictly.

        Raises:
            ValueError: if a key is missing, or a value has the wrong type or range.
        """
        for key in (duration_key, chars_key, tags_key, emotion_key):
            if key not in row:
                raise ValueError(f"row has no column {key!r}")
        duration, chars, tags, emotion = (
            row[duration_key],
            row[chars_key],
            row[tags_key],
            row[emotion_key],
        )
        if not isinstance(duration, int) or not isinstance(chars, int):
            raise ValueError(f"{duration_key!r} and {chars_key!r} must be ints")
        if not isinstance(tags, list | tuple) or any(not isinstance(t, str) for t in tags):
            raise ValueError(f"{tags_key!r} must be a list of strings, got {tags!r}")
        if emotion is not None and not isinstance(emotion, str):
            raise ValueError(f"{emotion_key!r} must be a string or None, got {emotion!r}")
        return SegmentStat(duration_ms=duration, num_chars=chars, tags=tuple(tags), emotion=emotion)


def _check_edges(bin_edges_ms: Sequence[int]) -> None:
    if not bin_edges_ms:
        raise ValueError("bin_edges_ms must not be empty")
    if any(not _is_int(edge) or edge < 0 for edge in bin_edges_ms):
        raise ValueError(f"bin edges must be non-negative ints, got {list(bin_edges_ms)!r}")
    if any(b <= a for a, b in pairwise(bin_edges_ms)):
        raise ValueError(f"bin edges must be strictly increasing, got {list(bin_edges_ms)!r}")


def duration_histogram(durations_ms: Iterable[int], bin_edges_ms: Sequence[int]) -> list[int]:
    """Histogram of durations over ``n`` strictly increasing edges.

    Returns ``n + 1`` counts: index 0 is the underflow bin ``(-inf, e0)``, index ``k`` for
    ``1 <= k < n`` is ``[e(k-1), e(k))``, and index ``n`` is the overflow bin ``[e(n-1), inf)``.
    Every duration is counted exactly once, so ``sum(result) == len(durations_ms)``.

    Raises:
        ValueError: if the edges are empty, negative or not strictly increasing, or a
            duration is not a non-negative int.
    """
    _check_edges(bin_edges_ms)
    counts = [0] * (len(bin_edges_ms) + 1)
    for duration in durations_ms:
        if not _is_int(duration) or duration < 0:
            raise ValueError(f"duration must be a non-negative int in ms, got {duration!r}")
        counts[bisect_right(bin_edges_ms, duration)] += 1
    return counts


def _percentile_name(percentile: float) -> str:
    return f"p{percentile:g}"


@dataclass(frozen=True)
class StatsSummary:
    """Result of :func:`summarize`.

    ``duration_histogram`` has ``len(duration_bin_edges_ms) + 1`` entries (see
    :func:`duration_histogram`). ``cps_percentiles`` maps names such as ``"p5"`` to
    characters per second and is empty when there are no segments. Count mappings are
    ordered most frequent first (ties alphabetically).
    """

    num_segments: int
    total_duration_ms: int
    total_hours: float
    duration_bin_edges_ms: tuple[int, ...]
    duration_histogram: tuple[int, ...]
    tag_counts: dict[str, int]
    tag_segment_counts: dict[str, int]
    num_segments_without_tags: int
    emotion_counts: dict[str, int]
    num_segments_without_emotion: int
    cps_percentiles: dict[str, float]

    def to_dict(self) -> dict[str, object]:
        """JSON-compatible dict (for ``metrics.json``)."""
        return asdict(self)


def _ranked(counter: Mapping[str, int]) -> dict[str, int]:
    return dict(sorted(counter.items(), key=lambda item: (-item[1], item[0])))


def summarize(
    rows: Iterable[SegmentStat],
    *,
    duration_bin_edges_ms: Sequence[int],
    percentiles: Sequence[float] = DEFAULT_CPS_PERCENTILES,
) -> StatsSummary:
    """Aggregate segment statistics.

    Args:
        rows: Segments to describe (consumed once).
        duration_bin_edges_ms: Edges for the duration histogram.
        percentiles: Percentiles in [0, 100] of the chars-per-second distribution
            (linear interpolation, ``numpy.percentile`` default); no duplicates.

    Raises:
        ValueError: on invalid edges or percentiles.
    """
    _check_edges(duration_bin_edges_ms)
    if any(not (math.isfinite(p) and 0.0 <= p <= 100.0) for p in percentiles):
        raise ValueError(f"percentiles must be in [0, 100], got {list(percentiles)!r}")
    if len(set(percentiles)) != len(percentiles):
        raise ValueError(f"duplicate percentiles: {list(percentiles)!r}")

    durations: list[int] = []
    rates: list[float] = []
    tag_counts: Counter[str] = Counter()
    tag_segment_counts: Counter[str] = Counter()
    emotion_counts: Counter[str] = Counter()
    without_tags = 0
    without_emotion = 0
    for row in rows:
        durations.append(row.duration_ms)
        rates.append(row.chars_per_second)
        tag_counts.update(row.tags)
        tag_segment_counts.update(set(row.tags))
        if not row.tags:
            without_tags += 1
        if row.emotion is None:
            without_emotion += 1
        else:
            emotion_counts[row.emotion] += 1

    total_ms = sum(durations)
    cps: dict[str, float] = {}
    if rates:
        values = np.percentile(np.asarray(rates, dtype=np.float64), list(percentiles))
        cps = {_percentile_name(p): float(v) for p, v in zip(percentiles, values, strict=True)}
    return StatsSummary(
        num_segments=len(durations),
        total_duration_ms=total_ms,
        total_hours=total_ms / MS_PER_HOUR,
        duration_bin_edges_ms=tuple(duration_bin_edges_ms),
        duration_histogram=tuple(duration_histogram(durations, duration_bin_edges_ms)),
        tag_counts=_ranked(tag_counts),
        tag_segment_counts=_ranked(tag_segment_counts),
        num_segments_without_tags=without_tags,
        emotion_counts=_ranked(emotion_counts),
        num_segments_without_emotion=without_emotion,
        cps_percentiles=cps,
    )


def _backtick_runs(text: str) -> list[int]:
    """Lengths of the maximal runs of backticks in ``text``."""
    runs: list[int] = []
    current = 0
    for char in text:
        if char == "`":
            current += 1
        elif current:
            runs.append(current)
            current = 0
    if current:
        runs.append(current)
    return runs


def _code(text: str) -> str:
    """Inline-code span that survives pipes, angle brackets and backticks in ``text``.

    The fence is one backtick longer than the longest backtick run inside the text.
    """
    runs = _backtick_runs(text)
    fence = "`" * ((max(runs) + 1) if runs else 1)
    return f"{fence} {text} {fence}".replace("|", r"\|")


def _seconds(edge_ms: int) -> str:
    return f"{edge_ms / _MS_PER_SECOND:g}"


def _bin_labels(edges_ms: Sequence[int]) -> list[str]:
    labels = [f"< {_seconds(edges_ms[0])} s"]
    labels += [f"[{_seconds(lo)}, {_seconds(hi)}) s" for lo, hi in pairwise(edges_ms)]
    labels.append(f">= {_seconds(edges_ms[-1])} s")
    return labels


def _share(count: int, total: int) -> str:
    return f"{100.0 * count / total:.1f}%" if total else "-"


def render_stats_markdown(summary: StatsSummary, *, title: str = "Veri istatistikleri") -> str:
    """Render a summary as a Markdown report section (Turkish headings)."""
    n = summary.num_segments
    lines: list[str] = [f"## {title}", ""]
    lines += [
        f"- Segment sayısı: {n}",
        f"- Toplam süre: {summary.total_hours:.2f} saat ({summary.total_duration_ms} ms)",
        f"- Etiketsiz segment: {summary.num_segments_without_tags}",
        f"- Duygu etiketsiz segment: {summary.num_segments_without_emotion}",
        "",
        "### Süre dağılımı",
        "",
        "| Aralık | Sayı | Oran |",
        "|---|---:|---:|",
    ]
    labels = _bin_labels(summary.duration_bin_edges_ms)
    for label, count in zip(labels, summary.duration_histogram, strict=True):
        lines.append(f"| {label} | {count} | {_share(count, n)} |")

    lines += ["", "### Etiketler", ""]
    if summary.tag_counts:
        lines += ["| Etiket | Geçiş | Segment |", "|---|---:|---:|"]
        for tag, count in summary.tag_counts.items():
            lines.append(f"| {_code(tag)} | {count} | {summary.tag_segment_counts[tag]} |")
    else:
        lines.append("Etiket yok.")

    lines += ["", "### Duygular", ""]
    if summary.emotion_counts:
        lines += ["| Duygu | Segment | Oran |", "|---|---:|---:|"]
        for emotion, count in summary.emotion_counts.items():
            lines.append(f"| {_code(emotion)} | {count} | {_share(count, n)} |")
    else:
        lines.append("Duygu etiketi yok.")

    lines += ["", "### Harf/saniye", ""]
    if summary.cps_percentiles:
        lines += ["| Yüzdelik | Harf/sn |", "|---|---:|"]
        for name, value in summary.cps_percentiles.items():
            lines.append(f"| {name} | {value:.2f} |")
    else:
        lines.append("Veri yok.")
    return "\n".join(lines) + "\n"