readalign

12class EnglishSyllableWeighting:
13    def weight(self, word: str) -> float:
14        return float(self.syllable_count(word))
15
16    def syllable_count(self, word: str) -> int:
17        lightest = int(rules().lightest_word)
18        vowels = frozenset(rules().english_vowels)
19        clusters = letter_clusters(word)
20        if not clusters:
21            return lightest
22
23        count = 0
24        previous_was_vowel = False
25        for cluster in clusters:
26            is_vowel = cluster in vowels
27            if is_vowel and not previous_was_vowel:
28                count += 1
29            previous_was_vowel = is_vowel
30        if self.ends_silently(clusters) and count > 1:
31            count -= 1
32        return max(count, lightest)
33
34    def ends_silently(self, clusters: list[str]) -> bool:
35        if len(clusters) < rules().shortest_with_a_silent_ending:
36            return False
37        spelled = "".join(clusters)
38        return spelled.endswith(rules().silent_ending) and not spelled.endswith(rules().silent_ending_except_after)

41class EvenWeighting:
42    def weight(self, _word: str) -> float:
43        return 1.0

 9@dataclass(frozen=True)
10class RecognizedWord:
11    text: str
12    start: float
13    end: float

8class SpeechWeighting(Protocol):
9    def weight(self, word: str) -> float: ...

20class UnevenPiecesError(ValueError):
21    def __init__(self, heard: int, pieces: int) -> None:
22        super().__init__(f"{heard} transcripts for {pieces} pieces")

22@dataclass(frozen=True)
23class WordMatch:
24    expected: range
25    heard: range

16@dataclass(frozen=True)
17class WordSpan:
18    start: float
19    end: float

22def align(
23    expected: list[str],
24    heard: list[RecognizedWord],
25    duration: float,
26    weighting: SpeechWeighting | None = None,
27    equivalent: Equivalent | None = None,
28) -> list[WordSpan]:
29    if not expected or not heard:
30        return []
31    weighting = weighting or EnglishSyllableWeighting()
32    return fill(expected, match(expected, heard, weighting, equivalent), duration, weighting)

57def cuts(samples: Sequence[float], sample_rate: float) -> list[tuple[int, int]]:
58    """Cut the recording into the pieces it is asked in, each overlapping the one before.
59
60    A piece ends at the last pause that leaves it long enough to carry a line and short
61    enough for the runtime to take whole; where no pause falls there, it ends on length
62    alone, because a piece that grows to find a pause is the very window this avoids. The
63    next piece begins one pause earlier than the last ended, so every word is heard whole
64    by at least one of them. Where that would leave no overlap, the next piece begins by
65    `edge_overlap` instead. This is confined to empty seams: a floor that moved every
66    shorter overlap was measured worse.
67    """
68    longest = int(rules().piece_seconds * sample_rate)
69    if len(samples) <= longest or longest <= 0:
70        return [(0, len(samples))]
71    shortest = int(rules().piece_seconds * rules().shortest_piece_share * sample_rate)
72    marks = pauses(samples, sample_rate)
73
74    pieces = []
75    start = 0
76    while len(samples) - start > longest:
77        within = [mark for mark in marks if start + shortest < mark < start + longest]
78        cut = within[-1] if within else start + longest
79        pieces.append((start, cut))
80        # One pause back, but never back past half a piece: the overlap is there to carry
81        # the words at the seam, and a pause near the start of this piece would hand the
82        # next one almost the same range, over and over.
83        back = [mark for mark in marks if start + shortest <= mark < cut]
84        start = back[-1] if back else max(start, cut - int(rules().edge_overlap * sample_rate))
85    pieces.append((start, len(samples)))
86    return pieces

 8def energy_frames(samples: Sequence[float], sample_rate: float) -> list[float]:
 9    size = max(int(rules().frame_seconds * sample_rate), 1)
10    frames = []
11    for start in range(0, len(samples), size):
12        window = samples[start : start + size]
13        frames.append((sum(value * value for value in window) / len(window)) ** 0.5)
14    return frames

128def fold(word: str) -> str:
129    kept = [character for character in unicodedata.normalize("NFD", word) if not rules().lifts(character)]
130    lifted = unicodedata.normalize("NFC", "".join(kept))
131    listed = rules().folded_letters
132    return "".join(listed.get(character, character) for character in lifted)

205def heard(
206    piece: Sequence[float],
207    sample_rate: float,
208    asking: Callable[[Sequence[float]], Sequence[RecognizedWord]],
209    covered_prefix: float = 0.0,
210) -> list[RecognizedWord]:
211    """Ask for one piece, recovering an empty or prematurely stopped answer.
212
213    Parakeet answers some pieces of ordinary speech with no words at all, and whether it
214    does turns on where the piece starts and how long it is together: the mel statistics
215    are taken over the piece, so its length moves them, and past some edge the decoder
216    predicts blank at every frame. Handing over a little less of the tail moves the piece
217    off that edge. Nothing here tells speech from silence, so a piece that is genuinely
218    silent pays for the whole list before answering nothing, which is why a piece shorter
219    than `shortest_worth_asking_again` is not asked again at all.
220
221    A non-empty answer can also stop before speech resumes later in its audio. That tail
222    is asked again with already recognised context and accepted only when the two answers
223    share enough words to join without a duplicate.
224
225    An empty answer won by trimming is missing whatever was said in the tail that was cut
226    off. Each piece overlaps the next, and that overlap is what covers it.
227    """
228    words = list(asking(piece))
229    if words:
230        with_head = _recovered_head(words, piece, sample_rate, covered_prefix, asking)
231        return _recovered_tail(with_head, piece, sample_rate, asking)
232    if len(piece) / sample_rate < rules().shortest_worth_asking_again:
233        return words
234    for trim in rules().ask_again_trims:
235        shorter = len(piece) - int(trim * sample_rate)
236        if shorter <= 0:
237            break
238        again = list(asking(piece[:shorter]))
239        if again:
240            with_head = _recovered_head(again, piece, sample_rate, covered_prefix, asking)
241            return _recovered_tail(with_head, piece, sample_rate, asking)
242    return words

32def held(
33    spans: list[WordSpan],
34    samples: Sequence[float],
35    sample_rate: float,
36    limit: float | None = None,
37) -> list[WordSpan]:
38    frames = energy_frames(samples, sample_rate)
39    if not frames:
40        return spans
41
42    limit = rules().hold_limit if limit is None else limit
43    frame_seconds = rules().frame_seconds
44    threshold = speech_threshold(frames)
45    duration = len(samples) / sample_rate
46
47    outcome = []
48    for index, span in enumerate(spans):
49        following = spans[index + 1].start if index + 1 < len(spans) else duration
50        ceiling = min(following, span.end + limit)
51        end = span.end
52        frame = int(span.end / frame_seconds)
53        went_quiet = False
54        while (frame + 1) * frame_seconds <= ceiling and frame < len(frames):
55            if frames[frame] < threshold:
56                went_quiet = True
57            elif went_quiet:
58                break
59            end = (frame + 1) * frame_seconds
60            frame += 1
61        outcome.append(WordSpan(span.start, min(max(span.end, end), max(following, span.start))))
62    return outcome

 89def joined(
 90    heard: Sequence[Sequence[RecognizedWord]],
 91    pieces: Sequence[tuple[int, int]],
 92    sample_rate: float,
 93) -> list[RecognizedWord]:
 94    """One reading out of what each piece came back with.
 95
 96    The pieces overlap, so the words at a seam arrive twice, and the second copy is
 97    dropped by the text: the longest run the two pieces say alike inside the overlap is
 98    found, everything the coming piece says up to the end of it comes off, and so does
 99    whatever the piece before said past it. Placed where it falls in the whole recording,
100    so a caller hands over what it was given piece by piece and gets the reading back.
101
102    By the text and not by the clock, because the clock is the one thing two builds of one
103    model do not share: the same word decoded in two pieces comes back a fifth of a second
104    apart on one runtime and differently again on the next, so a rule that asks how close
105    two marks are decides differently on each of them. The words agree where the marks do
106    not.
107
108    A piece the recogniser had nothing to say about is an answer, not a failure: a
109    stretch of silence is transcribed as no words at all. A count of transcripts that
110    does not match the count of pieces is a failure, and is refused rather than quietly
111    paired off until the shorter of the two runs out.
112    """
113    if len(heard) != len(pieces):
114        raise UnevenPiecesError(len(heard), len(pieces))
115    reading: list[RecognizedWord] = []
116    covered_to = 0.0
117    for words, (start, end) in zip(heard, pieces, strict=True):
118        offset = start / sample_rate
119        placed = [RecognizedWord(text=word.text, start=word.start + offset, end=word.end + offset) for word in words]
120        seam = _agreement(reading, placed, offset, covered_to)
121        if seam.insertion_at is not None and seam.gap_word is not None:
122            reading.insert(seam.insertion_at, seam.gap_word)
123        if seam.kept_after_it:
124            del reading[len(reading) - seam.kept_after_it :]
125        reading += placed[seam.coming_up_to_it :]
126        covered_to = end / sample_rate
127    return reading

69def normalize(word: str) -> str:
70    return "".join(letters(word))

10def pair(
11    expected: list[str],
12    heard: list[str],
13    threshold: float,
14    equivalent: Equivalent | None = None,
15) -> list[WordMatch]:
16    if not expected or not heard:
17        return []
18    alignment = Alignment(expected, heard, threshold, equivalent)
19    return alignment.matches(alignment.scores())

25def pauses(samples: Sequence[float], sample_rate: float) -> list[int]:
26    """Where the recording is quiet long enough that a cut there takes no word in half.
27
28    The middle of each stretch of quiet, in samples. A stop inside a word is quiet too,
29    which is why a pause has a length to reach before it counts as one.
30    """
31    frames = energy_frames(samples, sample_rate)
32    if not frames:
33        return []
34    threshold = speech_threshold(frames)
35    # Quiet is only quiet between speech. Where nothing stands above the threshold there
36    # is no voice to pause, and the whole recording would otherwise read as one long
37    # pause and offer its own middle as a place to cut.
38    if not any(frame >= threshold for frame in frames):
39        return []
40
41    each_frame = rules().frame_seconds
42    quiet_enough = max(int(rules().pause_seconds / each_frame), 1)
43    found = []
44    quiet = 0
45    for index, energy in enumerate(frames):
46        if energy < threshold:
47            quiet += 1
48            continue
49        if quiet >= quiet_enough:
50            found.append(int((index - quiet // 2) * each_frame * sample_rate))
51        quiet = 0
52    if quiet >= quiet_enough:
53        found.append(int((len(frames) - quiet // 2) * each_frame * sample_rate))
54    return found

73def printed_parts(word: str) -> int:
74    return max(len([part for part in word.split("-") if part]), 1)

51@lru_cache(maxsize=1)
52def rules() -> Rules:
53    text = (Path(__file__).parent / "rules.yaml").read_text(encoding="utf-8")
54    return read(yaml.safe_load(text))

24def speech_level(frames: Sequence[float]) -> float:
25    if not frames:
26        return rules().quietest_speech
27    quieter = int(len(frames) * (1 - rules().speech_from_loudest_share))
28    louder = sorted(frames)[quieter:]
29    return max(sum(louder) / len(louder), rules().quietest_speech)

17def speech_threshold(frames: Sequence[float]) -> float:
18    if not frames:
19        return 0.0
20    room = sorted(frames)[int(len(frames) * rules().room_quantile)]
21    return max(room * rules().speech_above_room, rules().quietest_room)

 84def spoken(tokens: Sequence[RecognizedWord]) -> list[RecognizedWord]:
 85    """Gather a recogniser's tokens into the words they spell.
 86
 87    A recogniser answers in tokens, not in words: `be`, `aut`, `y's`. Where one word ends
 88    and the next begins is the recogniser's convention, not the caller's, and every side
 89    that reads the same model has to read that convention the same way -- otherwise one
 90    returns `beauty's` where another returns `beautys` and a third splits the word in two,
 91    and the three cannot be held against each other however alike they heard the sound.
 92
 93    Each word is timed from the token it opens with to the one it closes with. Marks at
 94    either end of a word are left off: a word the model wrote `increase,` is the word
 95    `increase`, and the comma is the model's punctuation rather than anything it heard.
 96    Marks inside a word stay, because `beauty's` and `self-substantial` are words and
 97    `beautys` is not.
 98    """
 99    words: list[RecognizedWord] = []
100    text = ""
101    start = 0.0
102    end = 0.0
103    for token in tokens:
104        for character in token.text.replace(UNHEARD, ""):
105            if character in OPENS_A_WORD:
106                if text:
107                    words.append(RecognizedWord(text=text, start=start, end=end))
108                    text = ""
109                continue
110            if not text:
111                start = token.start
112            text += character
113            end = token.end
114    if text:
115        words.append(RecognizedWord(text=text, start=start, end=end))
116    return [trimmed for trimmed in (_trimmed(word) for word in words) if trimmed is not None]