core/schemas.py

from __future__ import annotations
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field


class ManifestItem(BaseModel):
    utt_id: str
    audio_uri: str
    ref_text: Optional[str] = None
    meta: Dict[str, Any] = Field(default_factory=dict)


class ASRConfig(BaseModel):
    language: Optional[str] = "zh"
    task: str = "transcribe"
    beam_size: int = 5
    best_of: int = 5
    temperature: float = 0.0
    vad_filter: bool = True
    chunk_length_s: Optional[float] = None
    initial_prompt: Optional[str] = None


class Segment(BaseModel):
    start: float
    end: float
    text: str
    avg_logprob: Optional[float] = None
    no_speech_prob: Optional[float] = None


class ASROutput(BaseModel):
    utt_id: str
    hyp_text: str
    segments: List[Segment] = Field(default_factory=list)
    language: Optional[str] = None
    duration_s: Optional[float] = None
    latency_ms: Optional[float] = None
    confidence: Optional[float] = None
    extras: Dict[str, Any] = Field(default_factory=dict)


class AlignOp(BaseModel):
    op: str
    ref: str
    hyp: str
    i_ref: int
    i_hyp: int


class AlignResult(BaseModel):
    utt_id: str
    ref_text: Optional[str]
    hyp_text: str
    norm_ref: Optional[str] = None
    norm_hyp: Optional[str] = None
    lang_type: str = "other"
    primary_level: str = "char"
    primary_metric_name: Optional[str] = None
    primary_metric_value: Optional[float] = None
    wer: Optional[float] = None
    cer: Optional[float] = None
    ops_word: List[AlignOp] = Field(default_factory=list)
    ops_char: List[AlignOp] = Field(default_factory=list)


class ErrorEvent(BaseModel):
    run_id: str
    utt_id: str
    op_type: str
    ref: str
    hyp: str
    position: int
    level: str
    lang_type: str = "other"
    is_primary_level: bool = False
    error_class: str
    meta: Dict[str, Any] = Field(default_factory=dict)