src/turn_detection/data/schema.py

"""Canonical record schema and validation for turn-detection datasets.

The upstream smart-turn v3.2 schema uses ``endpoint_bool``, ``midfiller`` and
``endfiller``.  In particular, filler values are nullable: ``None`` means that
the example was not annotated for that auxiliary task and must not silently be
converted to ``False``.
"""

from __future__ import annotations

from dataclasses import dataclass, field
import math
from typing import Any, Mapping


_ENDPOINT_FIELDS = ("endpoint_bool", "endpoint", "is_endpoint", "label", "target")
_MID_FILLER_FIELDS = ("midfiller", "mid_filler", "is_midfiller")
_END_FILLER_FIELDS = ("endfiller", "end_filler", "is_endfiller")
_SYNTHETIC_FIELDS = ("synthetic", "is_synthetic")
_ID_FIELDS = ("id", "record_id", "example_id", "utterance_id")
_DATASET_FIELDS = ("dataset", "source_dataset", "data_source")

_TRUE_VALUES = {
    "1",
    "true",
    "t",
    "yes",
    "y",
    "end",
    "ended",
    "endpoint",
    "complete",
    "completed",
    "finished",
    "final",
}
_FALSE_VALUES = {
    "0",
    "false",
    "f",
    "no",
    "n",
    "hold",
    "continue",
    "continuation",
    "incomplete",
    "not_endpoint",
    "non_endpoint",
    "nonendpoint",
}
_NULL_VALUES = {"", "na", "n/a", "nan", "none", "null", "unknown", "unlabeled"}


@dataclass(frozen=True)
class ValidationIssue:
    """A machine-readable validation finding."""

    severity: str
    code: str
    message: str
    field: str | None = None

    def to_dict(self) -> dict[str, str]:
        result = {"severity": self.severity, "code": self.code, "message": self.message}
        if self.field is not None:
            result["field"] = self.field
        return result


class RecordValidationError(ValueError):
    """Raised when strict normalization encounters one or more invalid fields."""

    def __init__(self, issues: list[ValidationIssue]) -> None:
        self.issues = tuple(issues)
        summary = "; ".join(f"{item.field or 'record'}: {item.message}" for item in issues)
        super().__init__(summary)


@dataclass(frozen=True)
class TurnRecord:
    """Normalized representation of one labeled audio example."""

    record_id: str
    audio: Any
    language: str | None
    endpoint: bool | None
    midfiller: bool | None
    endfiller: bool | None
    synthetic: bool | None
    dataset: str | None
    spoken_text: str | None
    source_file: str | None = None
    source_row: int | None = None
    metadata: Mapping[str, Any] = field(default_factory=dict)

    def to_dict(self, *, include_audio: bool = True) -> dict[str, Any]:
        """Convert to a plain mapping, optionally excluding the heavy audio value."""

        result: dict[str, Any] = {
            "record_id": self.record_id,
            "language": self.language,
            "endpoint": self.endpoint,
            "midfiller": self.midfiller,
            "endfiller": self.endfiller,
            "synthetic": self.synthetic,
            "dataset": self.dataset,
            "spoken_text": self.spoken_text,
            "source_file": self.source_file,
            "source_row": self.source_row,
            "metadata": dict(self.metadata),
        }
        if include_audio:
            result["audio"] = self.audio
        return result


@dataclass(frozen=True)
class NormalizationResult:
    """A permissively normalized record together with validation findings."""

    record: TurnRecord
    issues: tuple[ValidationIssue, ...]

    @property
    def errors(self) -> tuple[ValidationIssue, ...]:
        return tuple(item for item in self.issues if item.severity == "error")

    @property
    def warnings(self) -> tuple[ValidationIssue, ...]:
        return tuple(item for item in self.issues if item.severity == "warning")


def normalize_nullable_bool(value: Any, *, field_name: str = "value") -> bool | None:
    """Normalize a nullable boolean without conflating missing and false labels.

    Accepted values include booleans, numeric 0/1 and common textual forms.  A
    null-like value returns ``None``.  Other values raise ``ValueError``.
    """

    if value is None:
        return None
    if isinstance(value, bool):
        return value
    if isinstance(value, float) and math.isnan(value):
        return None
    if isinstance(value, (int, float)) and value in (0, 1):
        return bool(value)
    if isinstance(value, str):
        normalized = value.strip().casefold().replace("-", "_").replace(" ", "_")
        if normalized in _NULL_VALUES:
            return None
        if normalized in _TRUE_VALUES:
            return True
        if normalized in _FALSE_VALUES:
            return False
    raise ValueError(f"{field_name} must be boolean, 0/1, or null; got {value!r}")


def _first_present(record: Mapping[str, Any], names: tuple[str, ...]) -> tuple[str | None, Any]:
    for name in names:
        if name in record:
            return name, record[name]
    return None, None


def _optional_text(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _optional_int(value: Any) -> int | None:
    if value is None:
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def normalize_record_with_issues(record: Mapping[str, Any]) -> NormalizationResult:
    """Normalize a raw row and retain all validation errors and warnings."""

    issues: list[ValidationIssue] = []

    endpoint_key, endpoint_raw = _first_present(record, _ENDPOINT_FIELDS)
    endpoint: bool | None
    if endpoint_key is None:
        endpoint = None
        issues.append(
            ValidationIssue("error", "missing_endpoint", "required endpoint label is missing", "endpoint")
        )
    else:
        try:
            endpoint = normalize_nullable_bool(endpoint_raw, field_name=endpoint_key)
        except ValueError as exc:
            endpoint = None
            issues.append(ValidationIssue("error", "invalid_endpoint", str(exc), endpoint_key))
        if endpoint is None:
            issues.append(
                ValidationIssue("error", "null_endpoint", "endpoint label cannot be null", endpoint_key)
            )

    normalized_booleans: dict[str, bool | None] = {}
    for canonical, aliases in (
        ("midfiller", _MID_FILLER_FIELDS),
        ("endfiller", _END_FILLER_FIELDS),
        ("synthetic", _SYNTHETIC_FIELDS),
    ):
        key, raw = _first_present(record, aliases)
        if key is None:
            normalized_booleans[canonical] = None
            continue
        try:
            normalized_booleans[canonical] = normalize_nullable_bool(raw, field_name=key)
        except ValueError as exc:
            normalized_booleans[canonical] = None
            issues.append(ValidationIssue("error", f"invalid_{canonical}", str(exc), key))

    id_key, raw_id = _first_present(record, _ID_FIELDS)
    source_file = _optional_text(record.get("__source_file", record.get("source_file")))
    source_row = _optional_int(record.get("__source_row", record.get("source_row")))
    record_id = _optional_text(raw_id)
    if record_id is None:
        if source_file is not None and source_row is not None:
            record_id = f"{source_file}#{source_row}"
        else:
            record_id = ""
        issues.append(ValidationIssue("warning", "missing_id", "record has no explicit id", id_key or "id"))

    audio = record.get("audio")
    if audio is None:
        issues.append(ValidationIssue("error", "missing_audio", "audio value is missing", "audio"))

    language = _optional_text(record.get("language", record.get("lang")))
    if language is None:
        issues.append(ValidationIssue("warning", "missing_language", "language is missing", "language"))

    dataset_key, raw_dataset = _first_present(record, _DATASET_FIELDS)
    dataset = _optional_text(raw_dataset)
    if dataset is None:
        issues.append(
            ValidationIssue("warning", "missing_dataset", "source dataset is missing", dataset_key or "dataset")
        )

    spoken_text = _optional_text(record.get("spoken_text", record.get("transcript", record.get("text"))))

    consumed = {
        "audio",
        "language",
        "lang",
        "spoken_text",
        "transcript",
        "text",
        "__source_file",
        "source_file",
        "__source_row",
        "source_row",
        *_ENDPOINT_FIELDS,
        *_MID_FILLER_FIELDS,
        *_END_FILLER_FIELDS,
        *_SYNTHETIC_FIELDS,
        *_ID_FIELDS,
        *_DATASET_FIELDS,
    }
    metadata = {key: value for key, value in record.items() if key not in consumed}

    normalized = TurnRecord(
        record_id=record_id,
        audio=audio,
        language=language,
        endpoint=endpoint,
        midfiller=normalized_booleans["midfiller"],
        endfiller=normalized_booleans["endfiller"],
        synthetic=normalized_booleans["synthetic"],
        dataset=dataset,
        spoken_text=spoken_text,
        source_file=source_file,
        source_row=source_row,
        metadata=metadata,
    )
    return NormalizationResult(normalized, tuple(issues))


def normalize_record(record: Mapping[str, Any], *, strict: bool = True) -> TurnRecord:
    """Return a canonical record, raising on invalid fields in strict mode."""

    result = normalize_record_with_issues(record)
    if strict and result.errors:
        raise RecordValidationError(list(result.errors))
    return result.record


def validate_record(record: Mapping[str, Any] | TurnRecord) -> tuple[ValidationIssue, ...]:
    """Validate a raw mapping or an already-normalized :class:`TurnRecord`."""

    if isinstance(record, Mapping):
        return normalize_record_with_issues(record).issues

    issues: list[ValidationIssue] = []
    if not record.record_id:
        issues.append(ValidationIssue("warning", "missing_id", "record has no explicit id", "record_id"))
    if record.audio is None:
        issues.append(ValidationIssue("error", "missing_audio", "audio value is missing", "audio"))
    if not isinstance(record.endpoint, bool):
        issues.append(ValidationIssue("error", "invalid_endpoint", "endpoint must be boolean", "endpoint"))
    for field_name in ("midfiller", "endfiller", "synthetic"):
        value = getattr(record, field_name)
        if value is not None and not isinstance(value, bool):
            issues.append(
                ValidationIssue("error", f"invalid_{field_name}", f"{field_name} must be boolean or null", field_name)
            )
    return tuple(issues)