ml-services/evaluation/v2/adapters.py

"""Normalize current and enriched pipeline payloads into SignalBundle."""
from __future__ import annotations

from collections import defaultdict, deque
from typing import Any

from .schemas import (
    Modality,
    ModalityCoverage,
    ReliabilityAssessment,
    ReliabilityStatus,
    SignalBundle,
    SignalRecord,
    SignalScope,
    SourceProvenance,
    Speaker,
    TranscriptSegment,
    Visibility,
)

_AUDIO_FEATURES = {
    "duration_seconds": ("acoustic.duration", "seconds"),
    "pitch_mean_hz": ("acoustic.pitch.mean", "hz"),
    "pitch_min_hz": ("acoustic.pitch.min", "hz"),
    "pitch_max_hz": ("acoustic.pitch.max", "hz"),
    "pitch_std_hz": ("acoustic.pitch.std", "hz"),
    "rms_energy_mean": ("acoustic.energy.rms_mean", None),
    "rms_energy_max": ("acoustic.energy.rms_max", None),
    "volume_db_mean": ("acoustic.volume.mean", "db"),
    "total_pause_duration_seconds": (
        "acoustic.pause.total_duration",
        "seconds",
    ),
    "pause_count": ("acoustic.pause.count", "count"),
    "pause_ratio": ("acoustic.pause.ratio", "ratio"),
    "speech_rate_words_per_minute": (
        "acoustic.speech_rate",
        "words_per_minute",
    ),
}

_SEGMENT_OUTPUTS = {
    "sentiment": ("acoustic.sentiment.label", None),
    "dominant_emotion": ("acoustic.emotion.label", None),
    "escalation_score": ("acoustic.escalation.score", "ratio"),
    "prediction_confidence": ("acoustic.emotion.top_probability", "ratio"),
    "negative_emotion_probability": (
        "acoustic.emotion.negative_probability",
        "ratio",
    ),
}

_TRAJECTORY_FIELDS = {
    "trajectory_direction": ("candidate.trajectory.direction", None),
    "trajectory_delta": ("candidate.trajectory.delta", "ratio"),
    "start_escalation": ("candidate.trajectory.start_escalation", "ratio"),
    "end_escalation": ("candidate.trajectory.end_escalation", "ratio"),
    "peak_escalation": ("candidate.trajectory.peak_escalation", "ratio"),
    "deescalation_detected": (
        "candidate.trajectory.deescalation_detected",
        None,
    ),
    "unresolved_end_risk": (
        "candidate.trajectory.unresolved_end_risk",
        None,
    ),
}


def _speaker(value: Any) -> Speaker:
    normalized = str(value or "").strip().lower()
    if normalized == "agent":
        return Speaker.AGENT
    if normalized == "customer":
        return Speaker.CUSTOMER
    if normalized == "system":
        return Speaker.SYSTEM
    return Speaker.UNKNOWN


def _transcript_segments(
    transcript: dict[str, Any],
    provenance: SourceProvenance,
) -> list[TranscriptSegment]:
    segments = []
    for index, sentence in enumerate(transcript.get("sentences") or []):
        segment_id = str(
            sentence.get("id")
            or sentence.get("segment_key")
            or f"segment-{index + 1:04d}"
        )
        start = float(sentence.get("start", sentence.get("start_time", 0.0)))
        end = float(sentence.get("end", sentence.get("end_time", start)))
        segments.append(
            TranscriptSegment(
                segment_id=segment_id,
                segment_index=index,
                seq_id=sentence.get("seq_id"),
                speaker=_speaker(sentence.get("speaker")),
                start_seconds=start,
                end_seconds=end,
                text=str(sentence.get("text") or ""),
                provenance=provenance,
            )
        )
    return segments


def _align_sentiment(
    transcript_segments: list[TranscriptSegment],
    sentiment_segments: list[dict[str, Any]],
) -> list[tuple[TranscriptSegment | None, dict[str, Any]]]:
    if len(transcript_segments) == len(sentiment_segments):
        return list(zip(transcript_segments, sentiment_segments))

    by_seq: dict[int, deque[TranscriptSegment]] = defaultdict(deque)
    for segment in transcript_segments:
        if segment.seq_id is not None:
            by_seq[segment.seq_id].append(segment)

    aligned = []
    used = set()
    for sentiment in sentiment_segments:
        match = None
        segment_index = sentiment.get("segment_index")
        if isinstance(segment_index, int):
            possible_indexes = (
                (segment_index - 1, segment_index)
                if segment_index > 0
                else (segment_index,)
            )
            for index in possible_indexes:
                if (
                    0 <= index < len(transcript_segments)
                    and transcript_segments[index].segment_id not in used
                ):
                    match = transcript_segments[index]
                    break

        seq_id = sentiment.get("seq_id")
        if match is None and isinstance(seq_id, int):
            while by_seq[seq_id] and by_seq[seq_id][0].segment_id in used:
                by_seq[seq_id].popleft()
            if by_seq[seq_id]:
                match = by_seq[seq_id].popleft()

        if match is not None:
            used.add(match.segment_id)
        aligned.append((match, sentiment))
    return aligned


def _model_reliability(segment: dict[str, Any]) -> ReliabilityAssessment:
    status = str(segment.get("processing_status") or "unknown")
    reasons = []
    reliability = ReliabilityStatus.USABLE

    if status != "success":
        reliability = ReliabilityStatus.UNAVAILABLE
        reasons.append(f"processing_status:{status}")
    else:
        confidence = segment.get(
            "prediction_confidence",
            segment.get("emotion_confidence"),
        )
        if isinstance(confidence, (int, float)) and confidence < 0.5:
            reliability = ReliabilityStatus.LIMITED
            reasons.append("low_model_probability")

    return ReliabilityAssessment(
        status=reliability,
        reasons=reasons,
    )


def _feature_reliability(segment: dict[str, Any]) -> ReliabilityAssessment:
    features = segment.get("audio_features") or {}
    flags = features.get("audio_quality_flags") or {}
    active_flags = [
        str(name)
        for name, active in flags.items()
        if active is True
    ]
    limiting = {
        "very_short_segment",
        "low_energy_segment",
        "missing_pitch",
        "unrealistic_speech_rate",
    }
    status = (
        ReliabilityStatus.LIMITED
        if limiting.intersection(active_flags)
        else ReliabilityStatus.USABLE
    )
    reasons = (
        ["one_or_more_feature_quality_flags"]
        if status == ReliabilityStatus.LIMITED
        else []
    )
    return ReliabilityAssessment(
        status=status,
        reasons=reasons,
        quality_flags=active_flags,
    )


def _append_segment_signal(
    signals: list[SignalRecord],
    *,
    segment: TranscriptSegment,
    name: str,
    value: Any,
    unit: str | None,
    reliability: ReliabilityAssessment,
    provenance: SourceProvenance,
):
    if value is None or isinstance(value, (dict, list)):
        return
    signals.append(
        SignalRecord(
            signal_id=f"{segment.segment_id}:{name}",
            name=name,
            modality=Modality.ACOUSTIC,
            scope=SignalScope.SEGMENT,
            value=value,
            unit=unit,
            segment_id=segment.segment_id,
            seq_id=segment.seq_id,
            speaker=segment.speaker,
            start_seconds=segment.start_seconds,
            end_seconds=segment.end_seconds,
            reliability=reliability,
            provenance=provenance,
            visibility=Visibility.INTERNAL,
        )
    )


def _candidate_call_signals(
    sentiment: dict[str, Any],
    provenance: SourceProvenance,
) -> list[SignalRecord]:
    signals = []
    reliability = ReliabilityAssessment(
        status=ReliabilityStatus.LIMITED,
        reasons=["imported_unvalidated_derived_signal"],
    )
    trajectory = sentiment.get("temporal_emotion_trajectory") or {}

    for field, (name, unit) in _TRAJECTORY_FIELDS.items():
        value = trajectory.get(field)
        if value is None or isinstance(value, (dict, list)):
            continue
        signals.append(
            SignalRecord(
                signal_id=f"call:{name}",
                name=name,
                modality=Modality.MULTIMODAL,
                scope=SignalScope.CALL,
                value=value,
                unit=unit,
                reliability=reliability,
                provenance=provenance,
                visibility=Visibility.INTERNAL,
            )
        )
    return signals


def build_signal_bundle(
    transcript: dict[str, Any],
    sentiment: dict[str, Any] | None = None,
    *,
    transcript_source: str | None = None,
    sentiment_source: str | None = None,
) -> SignalBundle:
    """Map current or Clara-enriched payloads without applying v2 decisions."""
    sentiment = sentiment or {}
    transcript_call_id = str(transcript.get("call_id") or "").strip()
    sentiment_call_id = str(sentiment.get("call_id") or "").strip()
    if sentiment_call_id and sentiment_call_id != transcript_call_id:
        raise ValueError("transcript and sentiment call_id values do not match")

    transcript_provenance = SourceProvenance(
        producer="sentence_segmentation",
        producer_version=str(transcript.get("model") or "unknown"),
        source_artifact=transcript_source,
        method="per_channel_sentence_segmentation",
    )
    sentiment_provenance = SourceProvenance(
        producer=(
            "enriched_acoustic_pipeline"
            if sentiment.get("has_audio_features")
            else "acoustic_sentiment"
        ),
        producer_version=str(sentiment.get("model_version") or "unknown"),
        source_artifact=sentiment_source,
        method="sentence_aligned_audio_inference",
    )
    segments = _transcript_segments(transcript, transcript_provenance)
    sentiment_segments = [
        value
        for value in (sentiment.get("segments") or [])
        if isinstance(value, dict)
    ]
    aligned = _align_sentiment(segments, sentiment_segments)

    signals = []
    usable_model_segments = 0
    feature_segments = 0
    usable_feature_segments = 0
    for segment, sentiment_segment in aligned:
        if segment is None:
            continue

        model_reliability = _model_reliability(sentiment_segment)
        if model_reliability.status == ReliabilityStatus.USABLE:
            usable_model_segments += 1
        for field, (name, unit) in _SEGMENT_OUTPUTS.items():
            _append_segment_signal(
                signals,
                segment=segment,
                name=name,
                value=sentiment_segment.get(field),
                unit=unit,
                reliability=model_reliability,
                provenance=sentiment_provenance,
            )

        features = sentiment_segment.get("audio_features")
        if not isinstance(features, dict):
            continue
        feature_segments += 1
        feature_reliability = _feature_reliability(sentiment_segment)
        if feature_reliability.status == ReliabilityStatus.USABLE:
            usable_feature_segments += 1
        for field, (name, unit) in _AUDIO_FEATURES.items():
            _append_segment_signal(
                signals,
                segment=segment,
                name=name,
                value=features.get(field),
                unit=unit,
                reliability=feature_reliability,
                provenance=sentiment_provenance,
            )

    signals.extend(_candidate_call_signals(sentiment, sentiment_provenance))

    expected = len(segments)
    coverage = [
        ModalityCoverage(
            modality=Modality.TRANSCRIPT,
            source="sentence_segments",
            expected_units=expected,
            usable_units=expected,
            coverage_ratio=1.0 if expected else 0.0,
        )
    ]
    if sentiment:
        coverage.append(
            ModalityCoverage(
                modality=Modality.ACOUSTIC,
                source="emotion_model",
                expected_units=expected,
                usable_units=usable_model_segments,
                coverage_ratio=(
                    round(usable_model_segments / expected, 6)
                    if expected
                    else 0.0
                ),
                limitations=(
                    ["segments_skipped_or_unreliable"]
                    if usable_model_segments < expected
                    else []
                ),
            )
        )
        if feature_segments:
            coverage.append(
                ModalityCoverage(
                    modality=Modality.ACOUSTIC,
                    source="audio_features",
                    expected_units=expected,
                    usable_units=usable_feature_segments,
                    coverage_ratio=(
                        round(usable_feature_segments / expected, 6)
                        if expected
                        else 0.0
                    ),
                    limitations=(
                        ["feature_quality_flags_present"]
                        if usable_feature_segments < feature_segments
                        else []
                    ),
                )
            )

    duration = max((segment.end_seconds for segment in segments), default=0.0)
    sources = [transcript_provenance]
    if sentiment:
        sources.append(sentiment_provenance)
    return SignalBundle(
        call_id=transcript_call_id,
        domain=str(transcript.get("domain") or sentiment.get("domain") or "unknown"),
        accent=transcript.get("accent"),
        duration_seconds=duration,
        transcript_model=transcript.get("model"),
        segments=segments,
        signals=signals,
        coverage=coverage,
        sources=sources,
    )