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