deepcw_lang.py
Rama main · réplica de lectura del repositorio original.
"""
deepcw_lang.py — CW decoder language layer
Improves the model's output using knowledge of CW QSO STRUCTURE. The
model reads character by character with no context, so it confuses
similar letters (KEITH -> KEHTHA). The language layer knows that:
- there's a finite set of typical phrases (UR, TU, TKS, QSO, RST...),
- reports follow a fixed format (599, 5NN),
- a name follows NAME/OP,
- callsigns have a prefix+digit+suffix structure.
Operates PURELY on text, zero audio cost. Only corrects where the match
is certain — when in doubt it leaves the original (a raw reading is
better than a wrong "correction").
"""
from __future__ import annotations
# ── Common CW phrases (prosigns, Q-codes, working words) ─────────────────────
# A closed set — the model often returns garbled versions of these.
CW_WORDS = {
# Calls and sign-offs
"CQ", "DE", "K", "KN", "AR", "SK", "BK", "R", "RR", "RRR", "AS",
# Q-codes
"QRZ", "QTH", "QSL", "QRM", "QRN", "QSB", "QRP", "QRO", "QSY", "QRT",
"QSO", "QRX", "QRL", "QRG", "QRQ", "QRS",
# Q-codes — added (genuinely used in QSOs/contests, not just QRT/QSY)
"QRV", "QSK", "QSP", "QRU", "QST", "QTC", "QSA", "QRA", "QRK",
# BT prosign (a break/separator between message parts) — the model has
# no "=" in its character set (see DEFAULT_META['chars'] in
# deepcw_engine.py), so the prosign comes out as the letters "BT".
"BT",
# Common closing contractions/greetings in CW
"CUAGN", "HNY", "MERRY", "XMAS", "ELMER",
# Working words
"UR", "URS", "TU", "TKS", "TNX", "THX", "FB", "HR", "HW", "PSE", "PWR",
"RST", "RIG", "ANT", "WX", "TEMP", "NAME", "OP", "QTH", "AGN", "CFM",
"GM", "GA", "GE", "GN", "GD", "DR", "OM", "YL", "XYL", "ES", "NW",
"BTU", "WID", "ABT", "VY", "GUD", "GL", "HPE", "CUL", "CU", "SRI",
"WKD", "WKG", "RCVR", "RX", "TX", "TRX", "WATT", "WATTS", "MTR", "MTRS",
"DPL", "VERT", "BEAM", "LOOP", "GND", "COAX", "KEY", "BUG", "PADDLE",
"SIG", "SIGS", "RPT", "RPRT", "SED", "SND", "SENT", "RCVD", "CPY", "COPY",
"SOLID", "PART", "OK", "NO", "YES", "TEST", "CONTEST", "DX", "SASE",
"CARD", "BURO", "DIRECT", "VIA", "MNI", "TMW", "TDY", "MORN", "EVE",
"NITE", "DAY", "SUN", "RAIN", "SNOW", "CLDY", "HOT", "CLD", "WARM",
# Conversational/weather phrases (common in QSOs, the model glues them together)
"FER", "FINE", "BIZ", "FB", "SUNNY", "CLOUDY", "OVERCAST", "COOL",
"MILD", "WINDY", "FOGGY", "CLEAR", "WET", "DRY", "DEG", "DEGS", "DEGREES",
"TEMP", "CELSIUS", "FAHR", "PSE", "CPI", "CPY", "AGN", "AGE", "YRS",
"OLD", "YOUNG", "WIFE", "SON", "DTR", "FAMILY", "JOB", "WORK", "RETIRED",
"STUDENT", "HAM", "LIC", "LICENSE", "FIRST", "SECOND", "YEAR", "MONTH",
"WEEK", "HOUR", "MIN", "SEC", "TIME", "LOCAL", "UTC", "ANT", "TOWER",
"HEIGHT", "FT", "MTRS", "ELEMENTS", "WATTS", "OUTPUT", "INPUT", "FINALS",
"TUBE", "SOLID", "STATE", "HOMEBREW", "COMMERCIAL", "MADE",
# Numeric greetings/sign-offs — 73/88/72/99 are standard, "44" is a
# greeting/sign-off used in park activations (POTA/SOTA).
"73", "88", "72", "99", "44",
}
# ── Common names (after NAME / OP) ───────────────────────────────────────────
CW_NAMES = {
"JOHN", "JIM", "TOM", "BOB", "BILL", "DAVE", "MIKE", "STEVE", "PAUL",
"PETE", "DAN", "KEN", "KEITH", "CARL", "FRED", "GARY", "JACK", "JOE",
"LARRY", "MARK", "RALPH", "RAY", "RICK", "RON", "ROY", "SAM", "TED",
"WALT", "GENE", "AL", "ED", "HANK", "LOU", "MAX", "PHIL", "STAN",
"HANS", "KURT", "FRITZ", "HANS", "PETER", "KLAUS", "WERNER", "DIETER",
"ANDRE", "PIERRE", "JEAN", "LUC", "MARCO", "LUIGI", "ANTON", "PAVEL",
"IVAN", "YURI", "SERGE", "OLEG", "NICK", "ALEX", "VIC", "WALLY",
"ANDY", "CHRIS", "DENNIS", "DON", "DOUG", "FRANK", "GEORGE", "GREG",
"HARRY", "JERRY", "LEO", "LES", "NORM", "RUSS", "SCOTT", "WAYNE",
# Added — common operator names outside the original (mostly
# Anglo/German) list, since the station works DX across many continents.
"JOSE", "JUAN", "CARLOS", "LUIS", "MIGUEL", "PABLO", "PEDRO", "RAUL",
"MARIO", "GIANNI", "GIORGIO", "FRANCO", "ROBERTO", "RENATO",
"PIOTR", "MAREK", "JUREK", "JANEK", "TOMEK", "KRZYSZTOF", "ANDRZEJ",
"WOJCIECH", "GRZEGORZ", "PAWEL", "MIROSLAW", "ZBIGNIEW", "STANISLAW",
"DMITRY", "VLADIMIR", "IGOR", "BORIS", "VIKTOR", "ANATOLY", "MIKHAIL",
"TAKASHI", "HIROSHI", "KENJI", "AKIRA", "YOSHI",
"AHMED", "MOHAMMED", "ALI", "HASSAN", "OMAR",
"ERIK", "LARS", "SVEN", "OLE", "NIELS", "ANDERS", "BJORN",
"WILLEM", "HENK", "PIET", "KEES", "JAN", "GERRIT",
}
# ── Cut numbers ───────────────────────────────────────────────────────────────
CUT = {"T": "0", "N": "9", "E": "5", "A": "1", "U": "2", "V": "3",
"4": "4", "G": "7", "D": "8", "B": "6"}
# ── Morse code — for weighting substitution cost in _edit_dist ───────────────
# The CTC model confuses letters/digits with a SIMILAR dot-dash pattern (E "."
# vs T "-", S "..." vs O "---" is NOT close, but E vs T is — 1 symbol
# difference). A flat Levenshtein distance (every substitution=1) treated
# every mistake the same, so at max_dist=1 it also "corrected" substitutions
# that are impossible in real CW (e.g. T->O, letters with a completely
# different pattern) - these false "corrections" broke an otherwise correct
# raw reading. The substitution cost is now the Levenshtein distance BETWEEN
# THE MORSE CODES THEMSELVES (length 1-5 characters, at most ~25 comparisons
# to compute) — cheap, computed once per token during text correction, zero
# audio/inference cost.
MORSE_CODE = {
"A": ".-", "B": "-...", "C": "-.-.", "D": "-..", "E": ".",
"F": "..-.", "G": "--.", "H": "....", "I": "..", "J": ".---",
"K": "-.-", "L": ".-..", "M": "--", "N": "-.", "O": "---",
"P": ".--.", "Q": "--.-", "R": ".-.", "S": "...", "T": "-",
"U": "..-", "V": "...-", "W": ".--", "X": "-..-", "Y": "-.--",
"Z": "--..",
"0": "-----", "1": ".----", "2": "..---", "3": "...--", "4": "....-",
"5": ".....", "6": "-....", "7": "--...", "8": "---..", "9": "----.",
}
def _morse_edit(a: str, b: str) -> int:
"""Plain Levenshtein distance between two dot/dash strings (Morse
codes, length 1-5) — a helper for _sub_cost, NOT for text."""
if a == b:
return 0
la, lb = len(a), len(b)
prev = list(range(lb + 1))
for i in range(1, la + 1):
cur = [i] + [0] * lb
for j in range(1, lb + 1):
cost = 0 if a[i-1] == b[j-1] else 1
cur[j] = min(prev[j] + 1, cur[j-1] + 1, prev[j-1] + cost)
prev = cur
return prev[lb]
def _build_sub_cost_table() -> dict:
"""Precompute the substitution cost for every character pair once at
import time (36x36 pairs) — in _edit_dist this then becomes a simple
O(1) dict lookup, faster than computing _morse_edit for every DP cell."""
chars = list(MORSE_CODE.keys())
table = {}
for x in chars:
for y in chars:
table[(x, y)] = max(1, _morse_edit(MORSE_CODE[x], MORSE_CODE[y]))
return table
_SUB_COST = _build_sub_cost_table()
def _sub_cost(a: str, b: str) -> int:
"""Cost of substituting character a->b. Morse-close characters (e.g.
E<->T) cost 1 (like the old flat distance), Morse-far ones cost more —
so at the same max_dist there are fewer false "corrections" between
dissimilar characters, while real model mistakes (one Morse symbol of
difference) still go through."""
if a == b:
return 0
return _SUB_COST.get((a, b), 1) # fallback for spaces/characters outside A-Z0-9
def _edit_dist(a: str, b: str) -> int:
"""Edit distance weighted by Morse cost for substitutions (small
strings, simple DP implementation). Insertion/deletion of a character
still costs 1 as before — that's not "a mistake between two
characters", just a whole character too many/too few."""
if a == b:
return 0
la, lb = len(a), len(b)
if abs(la - lb) > 2: # optimization: length difference too large
return 99
prev = list(range(lb + 1))
for i in range(1, la + 1):
cur = [i] + [0] * lb
for j in range(1, lb + 1):
cost = _sub_cost(a[i-1], b[j-1])
cur[j] = min(prev[j] + 1, cur[j-1] + 1, prev[j-1] + cost)
prev = cur
return prev[lb]
def _best_match(token: str, vocab: set, max_dist: int = 1) -> str | None:
"""Find the word in the vocabulary closest to the token (distance <= max_dist).
Returns a match only when it's UNAMBIGUOUS — on a tie, returns None.
"""
t = token.upper()
if t in vocab:
return t
best, best_d, ties = None, max_dist + 1, 0
for w in vocab:
# Quick rejection: at max_dist=1, words differing in length by >1,
# or whose first/last letter doesn't match, are guaranteed to have
# a distance > max_dist. Skips the expensive DP computation for
# obviously different words.
if abs(len(w) - len(t)) > max_dist:
continue
if max_dist == 1 and w[0] != t[0] and w[-1] != t[-1]:
continue
d = _edit_dist(t, w)
if d < best_d:
best, best_d, ties = w, d, 1
elif d == best_d:
ties += 1
if best and best_d <= max_dist and ties == 1:
return best
return None
def _is_report(tok: str) -> str | None:
"""Recognize and normalize an RST report. Returns the corrected form or None."""
t = tok.upper()
# Expand cut numbers to digit form
digits = "".join(CUT.get(c, c) for c in t)
if not digits.isdigit():
return None
# Typical reports: 599, 579, 559, 339... (3 digits, odd middle R-S-T)
if len(digits) == 3 and digits[0] in "12345" and digits[2] in "9":
return tok # already a sensible report, leave it as sent (may be 5NN)
return None
def _longest_piece_at(t: str, i: int, known_calls: set) -> str | None:
"""The longest recognized piece starting at t[i:] — a CW word, a known
callsign, or an RST report. None if nothing matches."""
n = len(t)
best = None
# a) a word from CW_WORDS — the longest match wins (e.g. so "FB"
# doesn't cut off a longer "FBQSO" if both were words).
for w in CW_WORDS:
lw = len(w)
if lw >= 2 and t[i:i+lw] == w and (best is None or lw > len(best)):
best = w
# b) a known callsign glued to an adjacent word, e.g.
# "DESQ3MZM" -> "DE SQ3MZM". Callsign length varies (3-10), so we try
# successive lengths from longest to shortest instead of iterating the
# (potentially large) known_calls set — this is usually O(10)
# membership checks (O(1) each), not O(|known_calls|).
if known_calls:
for lw in range(min(10, n - i), 2, -1):
cand = t[i:i+lw]
if cand in known_calls:
if best is None or lw > len(best):
best = cand
break
# c) an RST report glued on with no space, e.g. "5NNTU" -> "5NN TU" —
# a very common pattern (the operator sends the report and immediately
# "TU" with no gap).
if best is None and i + 3 <= n and _is_report(t[i:i+3]):
best = t[i:i+3]
return best
def _segment(token: str, known_calls: set | None = None) -> str | None:
"""Split a glued-together string into known CW words, known callsigns,
and RST reports.
At fast CW speeds the model doesn't insert spaces (it sees a run of
characters with no clear gaps), so it returns e.g. 'TKSFERFB' instead
of 'TKS FER FB'. We walk the token from the start, taking the LONGEST
recognized piece at each position (a word, a known callsign, a
report); whatever doesn't match is left as a raw "island" between
recognized pieces instead of ruining the whole result.
Previously we required coverage of the WHOLE token, otherwise None —
but one garbage tail (e.g. the signal cutting off at the end of a
transmission) broke the WHOLE match and showed the entire, longer
block with no spaces at all ('TNX FER QSO 73' glued during
transmission into 'TNXFERQSO73TFE'). So we now return a split even
when coverage is PARTIAL — but only when the MAJORITY of characters
(>=50%) were recognized, to avoid inserting spaces into pure noise
(a random 2-letter match inside a long garbage string).
"""
t = token.upper()
if len(t) < 4 or t in CW_WORDS:
return None
known_calls = known_calls or set()
n = len(t)
pieces: list[str] = []
matched_chars = 0
raw = ""
i = 0
while i < n:
piece = _longest_piece_at(t, i, known_calls)
if piece:
if raw:
pieces.append(raw)
raw = ""
pieces.append(piece)
matched_chars += len(piece)
i += len(piece)
else:
raw += t[i]
i += 1
if raw:
pieces.append(raw)
if matched_chars == 0 or len(pieces) < 2:
return None
if matched_chars / n < 0.5:
return None
return " ".join(pieces)
def _reglue_split_call(tokens: list[str], known_calls: set) -> list[str]:
"""Glue a run of short tokens broken up by false gaps back into one
callsign, e.g. ['H','B','9','T','WX'] -> ['HB9TWX'].
The inverse of the problem _segment fixes: there, the model failed to
insert a gap between words (glued together with no space). Here it
inserts a gap WHERE THERE ISN'T ONE — it confuses a short gap between
the letters of a callsign with a gap between words (observed with
clear, unhurried sending). The callsign comes out broken into single/
two-three-letter pieces.
We only merge a RUN of consecutive "candidates" (1-3 characters, NOT
belonging to CW_WORDS — so we don't swallow real short words like
TU/DE/K/BK/WX) and only when the merged result either matches
known_calls directly, or has an unambiguous callsign shape (letters+
digit, length 3-10). If a real word (e.g. "WX") shows up within the
run, the run ends before it — this can cut the merge short too early
(a callsign that happens to end in "WX"), but that's better than
swallowing real words. When in doubt (the shape doesn't check out),
the tokens are left as they were — we don't guess.
"""
out: list[str] = []
i, n = 0, len(tokens)
while i < n:
j = i
while j < n:
tk = tokens[j].upper()
if 1 <= len(tk) <= 3 and tk.isalnum() and tk not in CW_WORDS:
j += 1
else:
break
merged, advance = None, i
if known_calls and j > i:
# First, solid evidence: the whole glued run (run + up to 2
# more tokens, EVEN a real word like "WX") matches known_calls
# directly. We check from the longest variant down — a solid
# match takes priority over shape alone (below), otherwise a
# shorter, "plausible-looking" prefix (e.g. "HB9T") would win
# before we get a chance to check whether the longer variant
# is a known callsign.
for m in range(min(j + 2, n), i, -1):
cand = "".join(t.upper() for t in tokens[i:m])
if cand in known_calls:
merged, advance = cand, m
break
if merged is None and j - i >= 2:
cand = "".join(t.upper() for t in tokens[i:j])
if len(cand) <= 10 and _looks_like_call(cand):
merged, advance = cand, j
if merged is not None:
out.append(merged)
i = advance
continue
out.append(tokens[i])
i += 1
return out
def correct(text: str, known_calls: set | None = None) -> str:
"""Correct decoder text using CW QSO structure.
known_calls: a pool of real callsigns (from FT8/log/cluster) for validation.
"""
if not text:
return text
known_calls = known_calls or set()
tokens = [t for t in text.split(" ") if t]
tokens = _reglue_split_call(tokens, known_calls)
out = []
prev_upper = ""
for tok in tokens:
if not tok:
continue
T = tok.upper()
# 1. RST report — normalize the format
rep = _is_report(T)
if rep is not None:
out.append(rep)
prev_upper = rep
continue
# 2. After NAME / OP — try to match a name
# max_dist=3 (was 2 under the flat distance) — the budget of 2 was
# calibrated when EVERY substitution cost exactly 1. Now a
# substitution between distant characters costs more (see
# _sub_cost), so the same example as in the file header (KEITH ->
# KEHTHA: I->H is 2 dots of difference plus an inserted A) needs a
# budget of 3, otherwise even the author's own example wouldn't be
# caught. _best_match still rejects ties either way, so a bigger
# budget doesn't mean more false matches, just more CHANCES to find
# an unambiguous one.
if prev_upper in ("NAME", "OP", "OM") and len(T) >= 3:
m = _best_match(T, CW_NAMES, max_dist=3)
if m:
out.append(m)
prev_upper = m
continue
# 3. Callsign — validate against the known-callsign pool (if given)
if known_calls and _looks_like_call(T):
if T in known_calls:
out.append(T)
prev_upper = T
continue
# Only match against the pool for tokens with a SENSIBLE
# structure — otherwise every piece of garbage (R3BDR, OKOK5I9)
# would trigger an expensive distance computation against the
# whole pool, bogging down the CPU on longer lines.
m = _best_match(T, known_calls, max_dist=1)
if m:
out.append(m)
prev_upper = m
continue
# 4. Common CW phrase — match against the vocabulary
if len(T) >= 2:
m = _best_match(T, CW_WORDS, max_dist=1)
if m:
out.append(m)
prev_upper = m
continue
# 5. Glued words — try to split using the vocabulary (TKSFER -> TKS FER)
seg = _segment(T, known_calls)
if seg:
out.append(seg)
prev_upper = seg.split()[-1]
continue
# 6. Nothing matches — leave the original
out.append(tok)
prev_upper = T
return " ".join(out)
def _looks_like_call(t: str) -> bool:
"""Whether a token looks like a callsign (letter + digit + letters)."""
if not (3 <= len(t) <= 10):
return False
return any(c.isdigit() for c in t) and any(c.isalpha() for c in t)