#!/usr/bin/env python3 """Score complete transcripts and correctly aligned word delivery, without hiding omissions.""" from __future__ import annotations import argparse from array import array import hashlib import json import math from pathlib import Path import re import sys import unicodedata ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from baseline_probe import normalize def digest(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def timed_hypothesis(deltas, language): """Last constituent character arrival is a word's readable completion time. A later boundary/space is not charged to an already complete word. This is retrospective reference-aligned measurement, not a stable-word API promise. """ raw, normalized, times = [], [], [] previous = -math.inf for delta in deltas: text, at = delta['text'], delta['seconds'] if not isinstance(text, str) or type(at) not in (int, float) or not math.isfinite(at) or at < 0 or at < previous: raise ValueError('invalid or nonmonotonic text delivery') previous = at raw.append(text) for char in text: for converted in unicodedata.normalize('NFKC', char).lower(): normalized.append(' ' if unicodedata.category(converted).startswith(('P', 'S')) else converted) times.append(at) raw_text = ''.join(raw) text = ''.join(normalized) if language == 'zh': words = [{'text': c, 'seconds': t} for c, t in zip(text, times) if not c.isspace()] joined = ''.join(w['text'] for w in words) elif language == 'en': words = [dict(text=m.group(), seconds=max(times[m.start():m.end()])) for m in re.finditer(r'\S+', text)] joined = ' '.join(w['text'] for w in words) else: raise ValueError('unsupported language') if joined != normalize(raw_text, language): raise ValueError('cross-character Unicode normalization needs an explicit timing map') return raw_text, words def normalized_reference_words(case): if not case.get('words'): return None rows = [] for i, word in enumerate(case['words']): start, end = word['start'], word['end'] if not (math.isfinite(start) and math.isfinite(end) and 0 <= start <= end <= case['seconds']): raise ValueError('invalid reference word interval') value = normalize(word['text'], case['language']) units = list(value) if case['language'] == 'zh' else value.split() rows.extend(dict(text=unit, start=start, end=end, source_word_index=i) for unit in units) combined = ('' if case['language'] == 'zh' else ' ').join(w['text'] for w in rows) if combined != normalize(case['reference'], case['language']): raise ValueError('timed reference does not cover the exact scored reference') return rows def optimal_match_alternatives(reference, hypothesis): """Identify equal edges shared or ambiguous across all minimal edit paths. Word timing is not used to pick an alignment. Repeated words may have more than one equally valid textual alignment; those cannot prove strict lag. """ n, m = len(reference), len(hypothesis) if (n+1)*(m+1) > 4_000_000: raise ValueError('alignment ambiguity matrix exceeds bounded scoring scope') def distances(a, b): rows=[array('I', range(len(b)+1))] for i, x in enumerate(a): row=array('I', [i+1]) prev=rows[-1] for j, y in enumerate(b): row.append(min(prev[j+1]+1, row[j]+1, prev[j]+(x != y))) rows.append(row) return rows forward=distances(reference,hypothesis) backward=distances(reference[::-1],hypothesis[::-1]) total=forward[n][m] result=[] for i,x in enumerate(reference): matches=[] may_be_unmatched=False for j in range(m+1): cost=forward[i][j] if cost+1+backward[n-i-1][m-j] == total: may_be_unmatched=True if j < m: suffix=backward[n-i-1][m-j-1] if x == hypothesis[j] and cost+suffix == total: matches.append(j) elif x != hypothesis[j] and cost+1+suffix == total: may_be_unmatched=True result.append(dict(possible_hypothesis_indices=matches, may_be_unmatched_in_optimal_alignment=may_be_unmatched)) return result def score(case, deltas, *, actual_arrival_paced, timing_basis): import jiwer raw, hypothesis = timed_hypothesis(deltas, case['language']) reference = normalize(case['reference'], case['language']) hyp = normalize(raw, case['language']) # Use one common edit-distance operation for words or normalized characters. reference_units = list(reference) if case['language'] == 'zh' else reference.split() hyp_units = [w['text'] for w in hypothesis] aligned = jiwer.process_words(' '.join(reference_units), ' '.join(hyp_units)) counts = dict(hits=aligned.hits, substitutions=aligned.substitutions, deletions=aligned.deletions, insertions=aligned.insertions, reference_units=len(reference_units), hypothesis_units=len(hyp_units), errors=aligned.substitutions + aligned.deletions + aligned.insertions) counts['error_rate'] = counts['errors'] / len(reference_units) if reference_units else None timed = normalized_reference_words(case) alternatives = optimal_match_alternatives(reference_units,hyp_units) if timed else None matched, unmatched = [], [] longest_deletion = 0 for chunk in aligned.alignments[0]: if chunk.type == 'delete': longest_deletion = max(longest_deletion, chunk.ref_end_idx - chunk.ref_start_idx) if chunk.type != 'equal': unmatched.append(dict(type=chunk.type, reference_range=[chunk.ref_start_idx, chunk.ref_end_idx], hypothesis_range=[chunk.hyp_start_idx, chunk.hyp_end_idx])) elif timed is not None: for r, h in zip(range(chunk.ref_start_idx, chunk.ref_end_idx), range(chunk.hyp_start_idx, chunk.hyp_end_idx), strict=True): word, output = timed[r], hypothesis[h] assert word['text'] == output['text'] matched.append(dict(reference_index=r, hypothesis_index=h, text=word['text'], reference_start=word['start'], reference_end=word['end'], emitted_seconds=output['seconds'], lag_from_word_end_seconds=output['seconds'] - word['end'], lag_from_word_start_seconds=output['seconds'] - word['start'], alignment_unambiguous=(alternatives[r]['possible_hypothesis_indices'] == [h] and not alternatives[r]['may_be_unmatched_in_optimal_alignment']), **alternatives[r])) lags = sorted(w['lag_from_word_end_seconds'] for w in matched) unambiguous = sorted(w['lag_from_word_end_seconds'] for w in matched if w['alignment_unambiguous']) ambiguous_count = len(matched)-len(unambiguous) available = bool(lags) and actual_arrival_paced return dict(case_id=case['id'], language=case['language'], input_sha256=case['sha256'], raw_hypothesis=raw, normalized_reference=reference, normalized_hypothesis=hyp, metric='CER' if case['language'] == 'zh' else 'WER', counts=counts, longest_contiguous_deleted_reference_units=longest_deletion, aligned_word_timing=dict(measured=available, actual_arrival_paced=actual_arrival_paced, timing_basis=timing_basis, correctly_matched_units=len(matched), ambiguous_matched_units=ambiguous_count, ambiguity_method='all minimum unit-cost Levenshtein paths, without timing-based selection', unambiguous_matched_units=len(unambiguous), unambiguous_maximum_seconds=unambiguous[-1] if available and unambiguous else None, unmatched_reference_units=counts['deletions'] + counts['substitutions'], matched_reference_fraction=len(matched) / len(reference_units) if timed and reference_units else None, min_seconds=lags[0] if available else None, p50_seconds=lags[math.ceil(.50 * len(lags)) - 1] if available else None, p95_seconds=lags[math.ceil(.95 * len(lags)) - 1] if available else None, maximum_seconds=lags[-1] if available else None, matched_units_above_one_second=sum(v > 1 for v in lags) if available else None, all_matched_units_within_one_second=(lags[-1] <= 1) if available and not ambiguous_count else None, whole_release_pass=False, limitation='Lag summaries use one deterministic text alignment and are descriptive when alternative optimal alignments exist. Only unambiguous matches can support per-word timing claims. Omissions/substitutions remain errors, never zero-latency successes. Published reference timing uncertainty and overlapping-speaker word order are not independently corrected.'), alignment=matched, error_spans=unmatched, first_nonempty_delta_seconds=next((r['seconds'] for r in deltas if r['text']), None), production_accuracy_approved=False) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--manifest', type=Path, required=True) parser.add_argument('--case', required=True) parser.add_argument('--deltas', type=Path, required=True, help='JSON list of {text, seconds}; seconds are measured consumer delivery or declared runtime callback times.') parser.add_argument('--actual-arrival-paced', action='store_true') parser.add_argument('--timing-basis', required=True) parser.add_argument('--output', type=Path, required=True) args = parser.parse_args() cases = json.loads(args.manifest.read_text())['cases'] matches = [c for c in cases if c['id'] == args.case] if len(matches) != 1: raise ValueError('missing/duplicate case') case = matches[0] if digest(case['path']) != case['sha256']: raise ValueError('input bytes changed') deltas = json.loads(args.deltas.read_text()) result = score(case, deltas, actual_arrival_paced=args.actual_arrival_paced, timing_basis=args.timing_basis) result['provenance'] = dict(manifest_sha256=digest(args.manifest), deltas_sha256=digest(args.deltas), scorer_sha256=digest(__file__)) with args.output.open('x') as f: json.dump(result, f, indent=2, ensure_ascii=False) f.write('\n') print(json.dumps(dict(counts=result['counts'], timing=result['aligned_word_timing']))) if __name__ == '__main__': main()