| |
| """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']) |
| |
| 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() |
|
|
