| |
| """Line guards for the scribe cleanup model: keep the raw line wherever the cleaned one is unsafe. |
| |
| The cleanup model is trained to keep every word and number, but no model is perfect. Run its output through this |
| check and show the raw text for any line where: |
| - the output doesn't have the same lines, timestamps and speakers (then the whole window stays raw) |
| - a number's value was lost, changed or added ("twenty five" and "25" count as equal) |
| - an ambiguous spoken number ("twelve fifty") was turned into digits without context that settles it |
| - a negation (not, no, never, n't, ...) was lost from a word that is still there |
| An explicit self-correction is allowed: "ten no wait four" -> "4" loses neither a number nor a negation. |
| |
| python3 guard.py raw.txt cleaned.txt # both in "[m:ss] Speaker: text" lines; prints the guarded lines |
| python3 guard.py --selftest |
| |
| Generated by build_guard.py in the training repository from the code its evaluation uses. Standard library |
| only. Apache-2.0. |
| """ |
| import difflib, re, sys |
| from collections import Counter |
|
|
| FILLERS = {"um", "uh", "umm", "uhm", "mm", "mmm", "hmm", "hm", "mm-hmm", "mhm", "uh-huh", "er", "erm", "ah", "eh", "mkay", "mm'kay"} |
|
|
| FUNCTION = set("a an the i to and of in on it is we you he she they my your our this but so for with at as or".split()) |
|
|
| ONES = "zero one two three four five six seven eight nine ten eleven twelve thirteen fourteen fifteen sixteen seventeen eighteen nineteen".split() |
|
|
| TENS_WORDS = "_ _ twenty thirty forty fifty sixty seventy eighty ninety".split() |
|
|
| ORD = {1: "first", 2: "second", 3: "third", 4: "fourth", 5: "fifth", 6: "sixth", 7: "seventh", 8: "eighth", 9: "ninth", |
| 10: "tenth", 11: "eleventh", 12: "twelfth", 13: "thirteenth", 14: "fourteenth", 15: "fifteenth", 16: "sixteenth", |
| 17: "seventeenth", 18: "eighteenth", 19: "nineteenth", 20: "twentieth", 30: "thirtieth"} |
|
|
| def say(n): |
| """Spoken English for 0 <= n < 1,000,000 as ASR writes it ("three hundred twenty", no "and").""" |
| if n < 20: |
| return ONES[n] |
| if n < 100: |
| return TENS_WORDS[n // 10] + ("" if n % 10 == 0 else " " + ONES[n % 10]) |
| if n < 1000: |
| return ONES[n // 100] + " hundred" + ("" if n % 100 == 0 else " " + say(n % 100)) |
| return say(n // 1000) + " thousand" + ("" if n % 1000 == 0 else " " + say(n % 1000)) |
|
|
| def say_ord(n): |
| return ORD.get(n) or TENS_WORDS[n // 10] + " " + ORD[n % 10] |
|
|
| def say_minutes(m): |
| return "oh " + ONES[m] if 0 < m < 10 else say(m) |
|
|
| UNITS = {w: i for i, w in enumerate(ONES)} |
|
|
| TENS = {w: 10 * i for i, w in enumerate(TENS_WORDS) if w != "_"} |
|
|
| ORDINALS = {w: n for n, w in ORD.items()} |
|
|
| ORDINALS.update({w + "s": v for w, v in list(ORDINALS.items()) if w not in ("first", "second")}) |
|
|
| SHORT_WORDS = set("a i o am an as at be by do go he if in is it me my no of oh ok on or pm so to up us we".split()) |
|
|
| CUES = ["no wait", "no no", "actually no", "no actually", "or rather", "i mean", "sorry", "wait no"] |
|
|
| CUE_RE = re.compile(r"\b(" + "|".join(c.replace(" ", r"\s+") for c in CUES) + r")\b", re.I) |
|
|
| SCALES = {"hundred": 100, "thousand": 1000, "million": 10**6, "billion": 10**9} |
|
|
| def number_word(w): |
| return w in UNITS or w in TENS or w in ORDINALS or w in SCALES or w == "point" |
|
|
| def number_spans(text, context=False): |
| """Number mentions in order. Each span is a list of groups: a group is one value. "twenty five" -> [25]; |
| "three hundred and twenty" -> [320]; "two point five" -> [2.5]; "fifty million" / "50 million" -> [5e7]; |
| adjacent values that do not combine form an ambiguous span: "twelve fifty" -> [12, 50], "12 50" -> [12, 50]. |
| A digit token with ':' or '/' (12:50, 08/01) is one group holding its parts, as a tuple.""" |
| spans, groups = [], [] |
| state = None |
| total = current = 0 |
| decimals = None |
|
|
| def value(): |
| v = total + current |
| return v + float("0." + decimals) if decimals else v |
|
|
| def close_group(): |
| nonlocal total, current, decimals, state |
| if state is not None: |
| groups.append(value()) |
| total = current = 0 |
| decimals, state = None, None |
|
|
| def close_span(): |
| close_group() |
| if groups: |
| |
| |
| spans.append((list(groups), tokens[sentence_start:span_start], tokens[i:i + 8] if i is not None else [], span_start)) |
| groups.clear() |
|
|
| i, span_start, sentence_start = None, 0, 0 |
| tokens = raw_tokens = re.findall(r"\$?\d+(?:,\d{3})*(?:\.\d+)?(?:[:/]\d+)*%?(?:st|nd|rd|th)?|[a-z]+|[.,;?!]", |
| text.lower().replace("-", " ")) |
| for i, t in enumerate(raw_tokens): |
| if not groups and state is None: |
| span_start = i |
| if t in ".,;?!": |
| close_span() |
| if t in ".?!": |
| sentence_start = i + 1 |
| continue |
| if t in FILLERS or (t == "oh" and state is not None): |
| continue |
| if state is not None and t.isalpha() and len(t) <= 2 and t not in SHORT_WORDS and not number_word(t): |
| continue |
| if t[0].isdigit() or t[0] == "$": |
| d = re.sub(r"[$,%]|(st|nd|rd|th)$", "", t) |
| close_group() |
| if ":" in d or "/" in d: |
| groups.append(tuple(int(x) for x in re.split(r"[:/]", d))) |
| else: |
| total, current, state = 0, float(d), "digit" |
| continue |
| if t == "and" and state in ("hundred", "big"): |
| continue |
| if t == "point" and state in ("unit", "teen", "tens", "digit", "hundred", "big") and decimals is None: |
| decimals, state = "", "point"; continue |
| if state == "point" or (decimals is not None and t in UNITS and UNITS[t] < 10): |
| if t in UNITS and UNITS[t] < 10: |
| decimals += str(UNITS[t]); continue |
| if t in SCALES and state is not None: |
| if SCALES[t] == 100: |
| current = (current or 1) * 100; state = "hundred" |
| else: |
| total += (current or 1) * SCALES[t]; current = 0; state = "big" |
| continue |
| v = UNITS.get(t, TENS.get(t, ORDINALS.get(t))) |
| if v is None: |
| close_span(); continue |
| kind = "tens" if t in TENS or (t in ORDINALS and v >= 20) else ("unit" if v < 10 else "teen") |
| combines = (state == "tens" and kind == "unit") or state in ("hundred", "big") |
| if not combines: |
| close_group() |
| current += v |
| state = kind |
| i = len(raw_tokens) |
| close_span() |
| return spans if context else [g for g, _, _, _ in spans] |
|
|
| CURRENCY = {"dollars": "$", "dollar": "$", "euros": "€", "euro": "€", "pounds": "£", "pound": "£"} |
|
|
| def written(value, unit): |
| """How an unambiguous amount is written: (30000, "dollars") -> "$30,000", (2.5, "percent") -> "2.5%".""" |
| if unit == "percent": |
| return f"{value:g}%" |
| return CURRENCY[unit] + (f"{value:,.2f}" if value != int(value) else f"{int(value):,}") |
|
|
| SPELLING = {"programme": "program", "favourite": "favorite", "colour": "color", "centre": "center", "grey": "gray", |
| "realise": "realize", "organise": "organize", "behaviour": "behavior", "neighbourhoods": "neighborhoods", |
| "reprogramme": "reprogram", "okay": "ok"} |
|
|
| NEG = re.compile(r"\b(not|no|never|nothing|none|nobody|neither|nor|cannot)\b|n't\b", re.I) |
|
|
| LINE = re.compile(r"^\[(\d+:\d\d)\] ([^:\n]+):[ \t]*(.*)$") |
|
|
| def words(s): |
| s = re.sub(r"\b([a-z])\.([a-z])\.", r"\1\2", s.lower()) |
| return [SPELLING.get(w, w) for w in re.findall(r"[a-z0-9']+", s.replace("-", " "))] |
|
|
| DATE_CUES = {"date", "deadline", "due", "dated"} |
|
|
| def justified(before, after): |
| """Context that makes a two-part spoken number safe to normalize (SPEC.md): am/pm right after (a time), or |
| a month name or a date cue just before (a date). A currency does not: "twelve fifty euro" is still 12.50 |
| or 1,250.""" |
| return bool(set(after[:2]) & {"am", "pm"} or set(before) & DATE_CUES |
| or set(before[-1:]) & {m for m in "january february march april may june july august september october november december".split()}) |
|
|
| def readings(group_pair): |
| """Digit forms a two-group ambiguous span could be resolved into: 12 50 -> 12.5, 1250, (12, 50).""" |
| a, b = group_pair |
| if not (float(a).is_integer() and float(b).is_integer()): |
| return set() |
| a, b = int(a), int(b) |
| return {float(f"{a}.{b:02d}") if b < 100 else None, float(f"{a}{b:02d}") if b < 100 else None, float(f"{a}{b}"), |
| (a, b)} - {None} |
|
|
| def corrected(after): |
| """True when a self-correction cue starts within the next three words and another number follows it: |
| "ten [no wait four]", "ten am [no wait four pm]", "twenty five to eight [no wait eight o'clock]".""" |
| w = [t for t in after if t not in ".,;?!"] |
| for k in range(min(3, len(w))): |
| for cue in CUES: |
| c = cue.split() |
| if w[k:k + len(c)] == c and any(number_word(t) or t[0].isdigit() or t in ("quarter", "half") |
| for t in w[k + len(c):k + len(c) + 3]): |
| return True |
| return False |
|
|
| def restarted(tokens, i): |
| """True when token i is in the first copy of an immediately repeated phrase of up to four words, perhaps with a |
| cut-off word between ("one of the f one of the", "I don't I don't"): dropping that copy loses nothing. A phrase of |
| numbers alone ("two two", "one five five") is never a restart: dropping a digit there changes a value.""" |
| w = [t for t in tokens if t not in ".,;?!"] |
| j = sum(1 for t in tokens[:i] if t not in ".,;?!") |
| for n in range(1, 5): |
| for a in range(max(0, j - n + 1), j + 1): |
| for gap in (0, 1): |
| b = a + n + gap |
| if (b + n <= len(w) and w[a:a + n] == w[b:b + n] |
| and not all(number_word(t) or t[:1].isdigit() for t in w[a:a + n]) |
| and (gap == 0 or (len(w[a + n]) <= 3 and not number_word(w[a + n]) and w[a + n] not in ("oh", "o")))): |
| return True |
| return False |
|
|
| def negations(s): |
| """Negations, not counting the "no" inside correction cues ("ten no wait four").""" |
| return len(NEG.findall(CUE_RE.sub(" ", re.sub(r"[^\w\s']", " ", s)))) |
|
|
| def is_neg(w): |
| return bool(NEG.fullmatch(w)) or w.endswith("n't") |
|
|
| def neg_lost(src, hyp): |
| """A negation of src is lost when it has no counterpart in hyp but the word it negated does ("should not |
| exceed" -> "should exceed"), by word alignment, or when hyp has no negation left at all. Dropping a false |
| start that held a repeat of it is not a loss ("you can't make it you can't make a mistake").""" |
| plain = lambda t: words(CUE_RE.sub(" ", " ".join(words(t)))) |
| s, h = plain(src), plain(hyp) |
| to_h = {blk.a + k: blk.b + k for blk in difflib.SequenceMatcher(None, s, h, autojunk=False).get_matching_blocks() |
| for k in range(blk.size)} |
| if any(is_neg(w) and i not in to_h and not restarted(s, i) and to_h.get(i + 1) is not None and not is_neg(s[i + 1]) |
| and not (to_h[i + 1] > 0 and is_neg(h[to_h[i + 1] - 1])) |
| for i, w in enumerate(s)): |
| return True |
| return any(map(is_neg, s)) and not any(map(is_neg, h)) |
|
|
| def num_check(raw, hyp, ref=None): |
| """(lost, invented, resolved) number mentions of hyp against raw. A raw span matches a hyp span with the |
| same groups (form ignored). An ambiguous two-group raw span matched by one digit token built from its |
| parts counts as resolved. Lost and invented mentions are excused when ref agrees with hyp.""" |
| rs = number_spans(raw, context=True) |
| hs = number_spans(hyp) |
| if ref is not None and hs == number_spans(ref): |
| return 0, 0, 0 |
| ref_vals = {g for s in number_spans(ref) for g in s} if ref is not None else set() |
| used, lost, resolved, lost_vals = set(), 0, 0, Counter() |
| raw_tokens = re.findall(r"\$?\d+(?:,\d{3})*(?:\.\d+)?(?:[:/]\d+)*%?(?:st|nd|rd|th)?|[a-z]+|[.,;?!]", raw.lower().replace("-", " ")) |
| optional = Counter() |
| for s0, before, after, start in rs: |
| if corrected(after) or restarted(raw_tokens, start): |
| optional.update(s0) |
| continue |
| |
| |
| s = s0 |
| stutter = len(s) == 2 and s[0] == s[1] |
| hit = next((k for k, h in enumerate(hs) if k not in used and (h == s or (stutter and h == s[:1]) |
| or (len(s) == 1 and after[:1] == ["o"] and h == [(int(s[0]), 0)]))), None) |
| if hit is None and len(s) == 2: |
| hit = next((k for k, h in enumerate(hs) if k not in used and len(h) == 1 and h[0] in readings(s)), None) |
| resolved += hit is not None and not justified(before, after) |
| if hit is None and len(s) > 1: |
| hit = next((k for k, h in enumerate(hs) if k not in used and sorted(map(str, h)) == sorted(map(str, s))), None) |
| if hit is None: |
| if not (ref is not None and not set(s) & ref_vals): |
| lost += 1 |
| lost_vals.update(s) |
| else: |
| used.add(hit) |
| unmatched = [] |
| for k, h in enumerate(hs): |
| if k not in used: |
| if Counter(h) <= optional: |
| optional.subtract(h) |
| else: |
| unmatched.append(h) |
| extra = [h for h in unmatched if not (ref is not None and set(h) <= ref_vals)] |
| if lost and Counter(v for h in unmatched for v in h) == lost_vals: |
| return 0, 0, resolved |
| if Counter(v for sp, *_ in rs for v in sp) == Counter(v for h in hs for v in h): |
| return 0, 0, resolved |
| return lost, len(extra), resolved |
|
|
| def unsafe(raw, hyp): |
| """A number lost, invented or ambiguously rewritten ("twelve fifty" -> 12.50 without context), or a negation |
| lost. Self-corrections are allowed: "ten no wait four" -> "4" loses neither a number nor a negation.""" |
| return bool(sum(num_check(raw, hyp))) or neg_lost(raw, hyp) |
|
|
| def guard_lines(raw_lines, cleaned_lines): |
| """raw_lines, cleaned_lines: lists of "[m:ss] Speaker: text" strings. Returns (lines to show, reverted count).""" |
| raw = [LINE.match(l.strip()) for l in raw_lines if l.strip()] |
| out = [LINE.match(l.strip()) for l in cleaned_lines if l.strip()] |
| if len(raw) != len(out) or not all(raw) or not all(out) or any( |
| (a.group(1), a.group(2).strip()) != (b.group(1), b.group(2).strip()) for a, b in zip(raw, out)): |
| return [m.group(0) for m in raw if m], len(raw) |
| bad = [unsafe(a.group(3), b.group(3)) for a, b in zip(raw, out)] |
| return [a.group(0) if x else b.group(0) for a, b, x in zip(raw, out, bad)], sum(bad) |
|
|
|
|
| def selftest(): |
| ok = lambda raw, out: guard_lines([f"[0:01] A B: {raw}"], [f"[0:01] A B: {out}"])[1] == 0 |
| assert ok("um so the the cost is fine", "So the cost is fine.") |
| assert ok("twenty five thousand dollars", "$25,000") |
| assert ok("let's meet at three thirty pm", "Let's meet at 3:30 PM.") |
| assert ok("the deadline is eight one", "The deadline is 08/01.") |
| assert ok("stay below twelve uh fifty", "Stay below twelve fifty.") |
| assert ok("the review moved to ten no wait four pm", "The review moved to 4 PM.") |
| assert ok("I have one one little question", "I have one little question.") |
| assert not ok("it's on channel six one one one one oh", "It's on channel six one one one oh.") |
| assert ok("so you can't make it you can't make a mistake", "So you can't make a mistake.") |
| assert not ok("less than twelve fifty euro", "less than 125,000 Euro") |
| assert not ok("less than twelve fifty", "less than 12.50") |
| assert not ok("so that's two off the price", "So that's 20 off the price.") |
| assert not ok("it should not exceed it", "It should exceed it.") |
| assert guard_lines(["[0:01] A B: hi"], ["[0:02] A B: Hi."])[1] == 1 |
| print("ok") |
|
|
|
|
| if __name__ == "__main__": |
| if sys.argv[1:2] == ["--selftest"]: |
| selftest() |
| else: |
| shown, reverted = guard_lines(open(sys.argv[1]).read().splitlines(), open(sys.argv[2]).read().splitlines()) |
| print("\n".join(shown)) |
| print(f"{reverted} line(s) kept raw", file=sys.stderr) |
|
|
