bandsaunter/tests/test_morse.py

import numpy as np
import pytest

from morse_gen import morse_audio
from bandsaunter.morse import decode_morse, encode_morse, find_morse

FS = 16000

MESSAGES = [
    ("CQ CQ DE W1AW K", 18, 25), ("SOS SOS", 12, 20),
    ("TEST DE N0CALL 599 TU", 25, 18), ("HELLO WORLD 73", 35, 15),
    ("PARIS PARIS", 8, 25), ("VVV DE K1ABC", 22, 12),
    ("QRZ? DE VE3XYZ/M", 20, 20), ("R 5NN TU 73 GL", 30, 22),
    ("MAYDAY MAYDAY", 15, 8), ("THE QUICK BROWN FOX 1234567890", 28, 20),
    ("CQ DX DE W1AW", 18, 6),
]


@pytest.mark.parametrize("msg,wpm,snr", MESSAGES)
def test_decodes_exactly(msg, wpm, snr):
    r = decode_morse(morse_audio(msg, wpm, FS, snr), FS)
    assert r.text.strip() == msg
    assert r.is_morse


@pytest.mark.parametrize("wpm", [8, 15, 22, 30, 40])
def test_speed_estimate_is_accurate(wpm):
    r = decode_morse(morse_audio("CQ DE W1AW TEST", wpm, FS, 20), FS)
    assert r.wpm == pytest.approx(wpm, rel=0.10)


def test_tone_frequency_found():
    r = decode_morse(morse_audio("CQ TEST", 20, FS, 20, tone=1100.0), FS)
    assert r.tone_hz == pytest.approx(1100.0, abs=25)


def test_noise_is_not_morse():
    rng = np.random.default_rng(0)
    r = decode_morse(0.1 * rng.standard_normal(FS * 3), FS)
    assert not r.is_morse
    assert r.text == ""


def test_silence_is_rejected():
    assert not decode_morse(np.zeros(FS * 2), FS).is_morse


def test_encode_round_trip():
    assert encode_morse("SOS") == "... --- ..."
    assert encode_morse("A B") == ".- / -..."


# ---------------------------------------------------------------------------
# Short bursts
#
# Most of the CW on the air is not a conversation.  It is a repeater, a
# beacon or an unattended transmitter saying who it is and stopping, which is
# four to six characters and over in a second or two.  Those used to be read
# correctly and then thrown away by gates written for a paragraph of text.
# ---------------------------------------------------------------------------

SHORT = ["W1AW", "K1AA", "KU0W", "N0CALL", "VVV", "DE", "AR", "K", "73", "QRZ"]


@pytest.mark.parametrize("msg", SHORT)
@pytest.mark.parametrize("wpm", [12, 20, 30, 45])
def test_a_short_burst_is_read_and_believed(msg, wpm):
    r = decode_morse(morse_audio(msg, wpm, FS, 18), FS)
    assert r.text.strip() == msg
    assert r.is_morse, f"read {msg} correctly and then refused it: {r.notes}"


def test_a_callsign_on_its_own_is_under_two_seconds_at_thirty_words():
    """The length this is all about, stated so a regression is obvious."""
    from morse_gen import morse_keying
    keyed = np.flatnonzero(morse_keying("W1AW", 30, FS) > 0)
    assert (keyed[-1] - keyed[0]) / FS < 2.0
    assert decode_morse(morse_audio("W1AW", 30, FS, 18), FS).is_morse


@pytest.mark.parametrize("msg", ["E", "T"])
def test_one_keyed_element_is_not_an_identification(msg):
    """A single pulse is an E or a T whether a person sent it or not.

    This is where "no matter how short" stops, and it stops here because
    below it there is nothing left to be right about.
    """
    assert not decode_morse(morse_audio(msg, 20, FS, 20), FS).is_morse


def test_a_repeated_character_with_structure_is_still_text():
    """VVV is the oldest thing anyone sends, and it is not a pulse train.

    A uniform train of identical pulses decodes to EEEE or TTTT, which is
    what the check against one repeated character is for -- but V has dots
    and a dash of its own, which no train of identical pulses can produce.
    """
    r = decode_morse(morse_audio("VVV", 20, FS, 20), FS)
    assert r.text.strip() == "VVV" and r.is_morse


def test_a_uniform_pulse_train_is_still_refused():
    dot = 1.2 / 20.0
    env, fs = [], FS
    for _ in range(14):
        env += [1.0] * int(dot * fs) + [0.0] * int(3 * dot * fs)
    env = np.array(env)
    t = np.arange(env.size) / fs
    audio = env * np.sin(2 * np.pi * 700 * t)
    assert not decode_morse(audio, fs).is_morse


# -- what the capture window cut off ----------------------------------------

def _clipped(msg, wpm, head, tail):
    audio = morse_audio(msg, wpm, FS, 20)
    return audio[int(head * FS):audio.size - int(tail * FS)]


def test_a_character_sliced_by_the_window_is_dropped_not_guessed():
    """A K with its first dash missing is an A, not a worse K."""
    r = decode_morse(_clipped("K1AA", 20, 0.55, 0.55), FS)
    assert "A1AA" not in r.text, "half a character was reported as a whole one"
    assert any("cut off" in note for note in r.notes)


def test_what_is_left_of_a_cut_word_is_not_reported_as_a_whole_one():
    """K1AA caught halfway through reads as K1A, which is somebody else."""
    r = decode_morse(_clipped("K1AA", 20, 0.05, 0.9), FS)
    assert r.tail_cut
    assert "K1A" not in r.complete_text.split()


def test_the_words_that_survived_are_still_reported():
    r = decode_morse(_clipped("VVV DE W1AW", 20, 0.6, 0.1), FS)
    assert r.head_cut and not r.tail_cut
    assert "W1AW" in r.complete_text.split()


def test_a_complete_transmission_keeps_every_word():
    r = decode_morse(morse_audio("VVV DE W1AW", 20, FS, 20), FS)
    assert not r.head_cut and not r.tail_cut
    assert r.complete_text == r.text == "VVV DE W1AW"


@pytest.mark.parametrize("head,tail", [(0.05, 0.4), (0.4, 0.05), (0.7, 0.7),
                                       (1.1, 0.3), (0.3, 1.1)])
def test_a_truncated_capture_never_invents_a_callsign(head, tail):
    """The point of all of the above: this text is looked at for callsigns.

    A station heard through half its ident is better reported as half an
    ident than as a different station, because the different station gets
    looked up and pinned to a map.
    """
    from bandsaunter.callsign import find_callsigns
    r = decode_morse(_clipped("VVV DE W1AW/B FN31", 20, head, tail), FS)
    if not r.is_morse:
        return
    found = find_callsigns(r.complete_text, join_words=False)
    assert set(found) <= {"W1AW"}, f"invented {found} from {r.text!r}"


# -- and nothing that was not sent ------------------------------------------

@pytest.mark.parametrize("seed", range(10))
def test_a_blip_of_noise_is_not_a_short_transmission(seed):
    """The gates that let a two-character ident in must not let this in.

    With three or four elements the dot length is fitted to those very
    elements, so they land on the grid whatever produced them: a third of a
    second of noise decodes as a perfectly timed V.  What separates them is
    that keying is a tone and noise is not.
    """
    rng = np.random.default_rng(seed)
    for size in (1200, int(FS * 0.3), FS * 2):
        assert not decode_morse(rng.standard_normal(size), FS).is_morse


@pytest.mark.parametrize("seed", range(6))
def test_a_squelch_click_is_not_a_transmission(seed):
    rng = np.random.default_rng(seed)
    t = np.arange(int(FS * 0.5)) / FS
    for edges in ([(0.20, 0.25)], [(0.10, 0.14), (0.30, 0.34)]):
        env = np.zeros_like(t)
        for a, b in edges:
            env[int(a * FS):int(b * FS)] = 1.0
        audio = env * np.sin(2 * np.pi * 700 * t) + 0.02 * rng.standard_normal(t.size)
        assert not decode_morse(audio, FS).is_morse


def test_a_tone_that_never_keys_is_not_morse():
    t = np.arange(FS * 2) / FS
    assert not decode_morse(np.sin(2 * np.pi * 700 * t), FS).is_morse


@pytest.mark.parametrize("seed", range(4))
def test_speech_is_not_morse(seed):
    from speech import synth_speech
    assert not decode_morse(synth_speech(2.5, FS, seed=seed), FS).is_morse


# -- an ident sent over a carrier --------------------------------------------
#
# The commonest way Morse arrives on the land-mobile bands, and the way that
# was being missed entirely: the carrier stays up and the ident is an audio
# tone keyed inside it, so a CW detector looking for a keyed carrier sees a
# carrier that never stops.

def _ident_over_a_tone(text="KSQ330", wpm=22.0, rate=16000, tone=795.0,
                       before=11.0, after=11.0, seed=3):
    """A keyed ident with a steady tone either side of it, as heard over FM."""
    rng = np.random.default_rng(seed)
    keyed = morse_audio(text, wpm, rate, snr_db=30.0, tone=tone)
    steady = np.sin(2 * np.pi * tone * np.arange(int(before * rate)) / rate)
    tail = np.sin(2 * np.pi * tone
                  * np.arange(int(after * rate)) / rate) * 0.9
    audio = np.concatenate([steady * 0.9, keyed, tail]).astype(np.float64)
    return audio + rng.standard_normal(audio.size) * 0.02


def test_an_ident_is_found_inside_a_capture_that_is_mostly_something_else():
    """decode_morse reads a clip that is Morse; find_morse looks for one
    inside a clip that is not."""
    audio = _ident_over_a_tone()
    assert decode_morse(audio, 16000).complete_text != "KSQ330", \
        "if the whole clip decodes there is nothing for find_morse to fix"
    found = find_morse(audio, 16000)
    assert found is not None and found.is_morse
    assert found.complete_text == "KSQ330"
    assert 19 <= found.wpm <= 25


def test_a_steady_tone_at_the_edge_is_not_a_sliced_character():
    """A mark far longer than any dash is not a truncated element -- it is
    the transmission the ident was sent over.  Read as a cut, it took the
    first and last letter of every ident, and with them the whole callsign:
    one word with no gap in it to survive the drop."""
    found = find_morse(_ident_over_a_tone("W1AW"), 16000)
    assert found is not None
    assert found.text == "W1AW"
    assert found.complete_text == "W1AW", "the ident was thrown away as cut"


def test_a_capture_with_no_morse_in_it_yields_none():
    rng = np.random.default_rng(9)
    rate = 16000
    noise = rng.standard_normal(20 * rate) * 0.2
    speechy = noise + 0.4 * np.sin(
        2 * np.pi * 300 * np.arange(20 * rate) / rate
        * (1 + 0.3 * np.sin(2 * np.pi * 3 * np.arange(20 * rate) / rate)))
    for clip in (noise, speechy, np.zeros(20 * rate)):
        assert find_morse(clip, rate) is None


def test_the_search_is_no_looser_than_the_decoder():
    """Every window is judged by is_morse exactly as a whole capture would
    be; the search widens where the decoder looks, not what it accepts."""
    rng = np.random.default_rng(11)
    clip = rng.standard_normal(30 * 16000) * 0.3
    assert find_morse(clip, 16000) is None


def test_a_reading_of_nothing_but_dots_and_dashes_is_refused():
    """E and T are the one-element characters, so a decode made only of them
    can hardly be wrong -- there is nothing in it to get wrong.  A capture of
    noise on 445.5 MHz came back as "T T T E E E E E E E E" with the element
    mix and the timing fit both inside their bands."""
    from bandsaunter.morse import MorseResult
    flat = MorseResult(text="T T T E E E E", wpm=20.0, confidence=0.9,
                       n_elements=7, n_characters=7, timing_fit=1.0,
                       snr_db=40.0, undecoded=0)
    assert not flat.is_morse
    real = MorseResult(text="W1AW", wpm=20.0, confidence=0.9,
                       n_elements=11, n_characters=4, timing_fit=1.0,
                       snr_db=40.0, undecoded=0)
    assert real.is_morse


def test_the_whole_capture_is_one_of_the_windows():
    """When the capture *is* Morse -- a beacon keying through all of it -- the
    longest window is the best one, because a word is only certain when a gap
    bounds it at both ends and a short window may not contain one."""
    audio = morse_audio("CQ CQ DE W1AW W1AW K", 18, FS, 20)
    whole = decode_morse(audio, FS)
    found = find_morse(audio, FS)
    assert found is not None
    assert len(found.complete_text) >= len(whole.complete_text)