diff --git a/tests/test_dsc.py b/tests/test_dsc.py index 65dc71c..c66dad9 100644 --- a/tests/test_dsc.py +++ b/tests/test_dsc.py @@ -1,805 +1,817 @@ -"""Tests for DSC (Digital Selective Calling) utilities.""" - -import json - -import pytest - - -class TestDSCParser: - """Tests for DSC parser utilities.""" - - def test_get_country_from_mmsi_ship_station(self): - """Test country lookup for standard ship MMSI.""" - from utils.dsc.parser import get_country_from_mmsi - - # UK ships start with 232-235 - assert get_country_from_mmsi("232123456") == "United Kingdom" - assert get_country_from_mmsi("235987654") == "United Kingdom" - - # US ships start with 303, 338, 366-369 - assert get_country_from_mmsi("366123456") == "USA" - assert get_country_from_mmsi("369000001") == "USA" - - # Panama (common flag of convenience) - assert get_country_from_mmsi("351234567") == "Panama" - assert get_country_from_mmsi("370000001") == "Panama" - - # Norway - assert get_country_from_mmsi("257123456") == "Norway" - - # Germany - assert get_country_from_mmsi("211000001") == "Germany" - - def test_get_country_from_mmsi_coast_station(self): - """Test country lookup for coast station MMSI (starts with 00).""" - from utils.dsc.parser import get_country_from_mmsi - - # Coast stations: 00 + MID - assert get_country_from_mmsi("002320001") == "United Kingdom" - assert get_country_from_mmsi("003660001") == "USA" - - def test_get_country_from_mmsi_group_station(self): - """Test country lookup for group station MMSI (starts with 0).""" - from utils.dsc.parser import get_country_from_mmsi - - # Group call: 0 + MID - assert get_country_from_mmsi("023200001") == "United Kingdom" - assert get_country_from_mmsi("036600001") == "USA" - - def test_get_country_from_mmsi_unknown(self): - """Test country lookup returns None for unknown MID.""" - from utils.dsc.parser import get_country_from_mmsi - - assert get_country_from_mmsi("999999999") is None - assert get_country_from_mmsi("100000000") is None - - def test_get_country_from_mmsi_invalid(self): - """Test country lookup handles invalid input.""" - from utils.dsc.parser import get_country_from_mmsi - - assert get_country_from_mmsi("") is None - assert get_country_from_mmsi(None) is None - assert get_country_from_mmsi("12") is None - - def test_get_distress_nature_text(self): - """Test distress nature code to text conversion.""" - from utils.dsc.parser import get_distress_nature_text - - assert get_distress_nature_text(100) == "UNDESIGNATED" - assert get_distress_nature_text(101) == "FIRE" - assert get_distress_nature_text(102) == "FLOODING" - assert get_distress_nature_text(103) == "COLLISION" - assert get_distress_nature_text(106) == "SINKING" - assert get_distress_nature_text(109) == "PIRACY" - assert get_distress_nature_text(110) == "MOB" # Man overboard - - def test_get_distress_nature_text_unknown(self): - """Test distress nature returns formatted unknown for invalid codes.""" - from utils.dsc.parser import get_distress_nature_text - - assert "UNKNOWN" in get_distress_nature_text(999) - assert "999" in get_distress_nature_text(999) - - def test_get_distress_nature_text_string_input(self): - """Test distress nature accepts string input.""" - from utils.dsc.parser import get_distress_nature_text - - assert get_distress_nature_text("101") == "FIRE" - assert get_distress_nature_text("invalid") == "invalid" - - def test_get_format_text(self): - """Test format code to text conversion per ITU-R M.493.""" - from utils.dsc.parser import get_format_text - - assert get_format_text(102) == "ALL_SHIPS" - assert get_format_text(112) == "INDIVIDUAL" - assert get_format_text(114) == "INDIVIDUAL_ACK" - assert get_format_text(116) == "GROUP" - assert get_format_text(120) == "DISTRESS" - assert get_format_text(123) == "ALL_SHIPS_URGENCY_SAFETY" - - def test_get_format_text_unknown(self): - """Test format code returns unknown for invalid codes.""" - from utils.dsc.parser import get_format_text - - result = get_format_text(999) - assert "UNKNOWN" in result - - def test_get_format_text_removed_codes(self): - """Test that non-ITU format codes are no longer recognized.""" - from utils.dsc.parser import get_format_text - - # These were previously defined but are not ITU-R M.493 specifiers - for code in [100, 104, 106, 108, 110, 118]: - result = get_format_text(code) - assert "UNKNOWN" in result - - def test_get_telecommand_text(self): - """Test telecommand code to text conversion.""" - from utils.dsc.parser import get_telecommand_text - - assert get_telecommand_text(100) == "F3E_G3E_ALL" - assert get_telecommand_text(105) == "DATA" - assert get_telecommand_text(107) == "DISTRESS_ACK" - assert get_telecommand_text(111) == "TEST" - - def test_get_category_priority(self): - """Test category priority values.""" - from utils.dsc.parser import get_category_priority - - # Distress has highest priority (0) - assert get_category_priority("DISTRESS") == 0 - assert get_category_priority("distress") == 0 - - # Urgency/safety - assert get_category_priority("ALL_SHIPS_URGENCY_SAFETY") == 2 - - # Routine-level - assert get_category_priority("ALL_SHIPS") == 5 - assert get_category_priority("GROUP") == 5 - assert get_category_priority("INDIVIDUAL") == 5 - - # Unknown gets default high number - assert get_category_priority("UNKNOWN") == 10 - - def test_validate_mmsi_valid(self): - """Test MMSI validation with valid numbers.""" - from utils.dsc.parser import validate_mmsi - - assert validate_mmsi("232123456") is True - assert validate_mmsi("366000001") is True - assert validate_mmsi("002320001") is True # Coast station - assert validate_mmsi("023200001") is True # Group station - - def test_validate_mmsi_invalid(self): - """Test MMSI validation rejects invalid numbers.""" - from utils.dsc.parser import validate_mmsi - - assert validate_mmsi("") is False - assert validate_mmsi(None) is False - assert validate_mmsi("12345678") is False # Too short - assert validate_mmsi("1234567890") is False # Too long - assert validate_mmsi("abcdefghi") is False # Not digits - assert validate_mmsi("000000000") is False # All zeros - - def test_classify_mmsi(self): - """Test MMSI classification.""" - from utils.dsc.parser import classify_mmsi - - # Ship stations (start with 2-7) - assert classify_mmsi("232123456") == "ship" - assert classify_mmsi("366000001") == "ship" - assert classify_mmsi("503000001") == "ship" - - # Coast stations (start with 00) - assert classify_mmsi("002320001") == "coast" - - # Group stations (start with 0, not 00) - assert classify_mmsi("023200001") == "group" - - # SAR aircraft (start with 111) - assert classify_mmsi("111232001") == "sar" - - # Aids to Navigation (start with 99) - assert classify_mmsi("992321001") == "aton" - - # Unknown - assert classify_mmsi("invalid") == "unknown" - assert classify_mmsi("812345678") == "unknown" - - def test_parse_dsc_message_distress(self): - """Test parsing a distress message with ITU format 120.""" - from utils.dsc.parser import parse_dsc_message - - raw = json.dumps( - { - "type": "dsc", - "format": 120, - "source_mmsi": "232123456", - "dest_mmsi": "002320001", - "category": "DISTRESS", - "nature": 101, - "position": {"lat": 51.5, "lon": -0.1}, - "telecommand1": 100, - "timestamp": "2025-01-15T12:00:00Z", - "raw": "120002032123456101100127", - } - ) - - msg = parse_dsc_message(raw) - - assert msg is not None - assert msg["type"] == "dsc_message" - assert msg["source_mmsi"] == "232123456" - assert msg["category"] == "DISTRESS" - assert msg["source_country"] == "United Kingdom" - assert msg["nature_of_distress"] == "FIRE" - assert msg["latitude"] == 51.5 - assert msg["longitude"] == -0.1 - assert msg["is_critical"] is True - assert msg["priority"] == 0 - - def test_parse_dsc_message_group(self): - """Test parsing a group call message.""" - from utils.dsc.parser import parse_dsc_message - - raw = json.dumps( - { - "type": "dsc", - "format": 116, - "source_mmsi": "366000001", - "dest_mmsi": "023200001", - "category": "GROUP", - "timestamp": "2025-01-15T12:00:00Z", - "raw": "116023200001366000001117", - } - ) - - msg = parse_dsc_message(raw) - - assert msg is not None - assert msg["category"] == "GROUP" - assert msg["source_country"] == "USA" - assert msg["is_critical"] is False - assert msg["priority"] == 5 - - def test_parse_dsc_message_individual(self): - """Test parsing an individual call message.""" - from utils.dsc.parser import parse_dsc_message - - raw = json.dumps( - { - "type": "dsc", - "format": 112, - "source_mmsi": "366000001", - "dest_mmsi": "232123456", - "category": "INDIVIDUAL", - "telecommand1": 100, - "timestamp": "2025-01-15T12:00:00Z", - "raw": "112232123456366000001100122", - } - ) - - msg = parse_dsc_message(raw) - - assert msg is not None - assert msg["category"] == "INDIVIDUAL" - assert msg["is_critical"] is False - - def test_parse_dsc_message_invalid_json(self): - """Test parsing rejects invalid JSON.""" - from utils.dsc.parser import parse_dsc_message - - assert parse_dsc_message("not json") is None - assert parse_dsc_message("{invalid}") is None - - def test_parse_dsc_message_missing_type(self): - """Test parsing rejects messages without correct type.""" - from utils.dsc.parser import parse_dsc_message - - raw = json.dumps({"source_mmsi": "232123456"}) - assert parse_dsc_message(raw) is None - - raw = json.dumps({"type": "other", "source_mmsi": "232123456"}) - assert parse_dsc_message(raw) is None - - def test_parse_dsc_message_missing_mmsi(self): - """Test parsing rejects messages without source MMSI.""" - from utils.dsc.parser import parse_dsc_message - - raw = json.dumps({"type": "dsc"}) - assert parse_dsc_message(raw) is None - - def test_parse_dsc_message_empty(self): - """Test parsing handles empty input.""" - from utils.dsc.parser import parse_dsc_message - - assert parse_dsc_message("") is None - assert parse_dsc_message(None) is None - assert parse_dsc_message(" ") is None - - def test_parse_dsc_message_rejects_non_itu_format(self): - """Test parser rejects records with non-ITU format specifier.""" - from utils.dsc.parser import parse_dsc_message - - for bad_format in [100, 104, 106, 108, 110, 118, 999]: - raw = json.dumps( - { - "type": "dsc", - "format": bad_format, - "source_mmsi": "232123456", - "category": "ROUTINE", - "raw": "120232123456100127", - } - ) - assert parse_dsc_message(raw) is None, f"Format {bad_format} should be rejected" - - def test_parse_dsc_message_rejects_telecommand_out_of_range(self): - """Test parser rejects records with telecommand out of 100-127 range.""" - from utils.dsc.parser import parse_dsc_message - - raw = json.dumps( - { - "type": "dsc", - "format": 120, - "source_mmsi": "232123456", - "dest_mmsi": "002320001", - "category": "DISTRESS", - "telecommand1": 200, - "timestamp": "2025-01-15T12:00:00Z", - "raw": "120002032123456200127", - } - ) - assert parse_dsc_message(raw) is None - - def test_parse_dsc_message_accepts_zero_telecommand(self): - """Test parser does not drop telecommand with value 100 (truthiness fix).""" - from utils.dsc.parser import parse_dsc_message - - raw = json.dumps( - { - "type": "dsc", - "format": 112, - "source_mmsi": "232123456", - "dest_mmsi": "366000001", - "category": "INDIVIDUAL", - "telecommand1": 100, - "telecommand2": 100, - "timestamp": "2025-01-15T12:00:00Z", - "raw": "112366000001232123456100100122", - } - ) - - msg = parse_dsc_message(raw) - assert msg is not None - assert msg["telecommand1"] == 100 - assert msg["telecommand2"] == 100 - - def test_parse_dsc_message_validates_raw_field(self): - """Test parser validates raw field structure.""" - from utils.dsc.parser import parse_dsc_message - - # Non-digit raw field - raw = json.dumps( - { - "type": "dsc", - "format": 120, - "source_mmsi": "232123456", - "category": "DISTRESS", - "raw": "12abc", - } - ) - assert parse_dsc_message(raw) is None - - # Raw field length not divisible by 3 - raw = json.dumps( - { - "type": "dsc", - "format": 120, - "source_mmsi": "232123456", - "category": "DISTRESS", - "raw": "1201", - } - ) - assert parse_dsc_message(raw) is None - - # Raw field with non-EOS last token - raw = json.dumps( - { - "type": "dsc", - "format": 120, - "source_mmsi": "232123456", - "category": "DISTRESS", - "raw": "120100", - } - ) - assert parse_dsc_message(raw) is None - - def test_parse_dsc_message_accepts_valid_eos_in_raw(self): - """Test parser accepts all three valid EOS values in raw field.""" - from utils.dsc.parser import parse_dsc_message - - for eos in [117, 122, 127]: - raw = json.dumps( - { - "type": "dsc", - "format": 120, - "source_mmsi": "232123456", - "dest_mmsi": "002320001", - "category": "DISTRESS", - "timestamp": "2025-01-15T12:00:00Z", - "raw": f"120002032123456{eos:03d}", - } - ) - msg = parse_dsc_message(raw) - assert msg is not None, f"EOS {eos} should be accepted" - - def test_parse_dsc_message_rejects_invalid_mmsi(self): - """Test parser rejects invalid MMSI values.""" - from utils.dsc.parser import parse_dsc_message - - # All-zeros MMSI - raw = json.dumps( - { - "type": "dsc", - "format": 120, - "source_mmsi": "000000000", - "category": "DISTRESS", - "raw": "120000000000127", - } - ) - assert parse_dsc_message(raw) is None - - # Short MMSI - raw = json.dumps( - { - "type": "dsc", - "format": 120, - "source_mmsi": "12345", - "category": "DISTRESS", - "raw": "120127", - } - ) - assert parse_dsc_message(raw) is None - - def test_parse_dsc_message_nature_zero_not_dropped(self): - """Test that nature code 0 is not dropped by truthiness check.""" - from utils.dsc.parser import parse_dsc_message - - raw = json.dumps( - { - "type": "dsc", - "format": 120, - "source_mmsi": "232123456", - "dest_mmsi": "002320001", - "category": "DISTRESS", - "nature": 0, - "timestamp": "2025-01-15T12:00:00Z", - "raw": "120002032123456000127", - } - ) - - msg = parse_dsc_message(raw) - assert msg is not None - assert msg["nature_code"] == 0 - - def test_parse_dsc_message_channel_zero_not_dropped(self): - """Test that channel value 0 is not dropped by truthiness check.""" - from utils.dsc.parser import parse_dsc_message - - raw = json.dumps( - { - "type": "dsc", - "format": 112, - "source_mmsi": "232123456", - "dest_mmsi": "366000001", - "category": "INDIVIDUAL", - "channel": 0, - "telecommand1": 100, - "timestamp": "2025-01-15T12:00:00Z", - "raw": "112366000001232123456100122", - } - ) - - msg = parse_dsc_message(raw) - assert msg is not None - assert msg["channel"] == 0 - - def test_format_dsc_for_display(self): - """Test message formatting for display.""" - from utils.dsc.parser import format_dsc_for_display - - msg = { - "category": "DISTRESS", - "source_mmsi": "232123456", - "source_country": "United Kingdom", - "dest_mmsi": "002320001", - "nature_of_distress": "FIRE", - "latitude": 51.5074, - "longitude": -0.1278, - "telecommand1_text": "F3E_G3E_ALL", - "channel": 16, - "timestamp": "2025-01-15T12:00:00Z", - } - - output = format_dsc_for_display(msg) - - assert "DISTRESS" in output - assert "232123456" in output - assert "United Kingdom" in output - assert "FIRE" in output - assert "51.5074" in output - assert "Channel: 16" in output - - -class TestDSCDecoder: - """Tests for DSC decoder utilities.""" - - @pytest.fixture - def decoder(self): - """Create a DSC decoder instance.""" - # Skip if scipy not available - pytest.importorskip("scipy") - pytest.importorskip("numpy") - from utils.dsc.decoder import DSCDecoder - - return DSCDecoder() - - def test_decode_mmsi_valid(self, decoder): - """Test MMSI decoding from symbols.""" - # Each symbol is 2 BCD digits - # To encode MMSI 232123456, we need: - # 02-32-12-34-56 -> symbols [2, 32, 12, 34, 56] - symbols = [2, 32, 12, 34, 56] - result = decoder._decode_mmsi(symbols) - assert result == "232123456" - - def test_decode_mmsi_with_leading_zeros(self, decoder): - """Test MMSI decoding handles leading zeros.""" - # Coast station: 002320001 - # Padded to 10 digits: 0002320001 - # BCD pairs: 00-02-32-00-01 -> [0, 2, 32, 0, 1] - symbols = [0, 2, 32, 0, 1] - result = decoder._decode_mmsi(symbols) - assert result == "002320001" - - def test_decode_mmsi_short_symbols(self, decoder): - """Test MMSI decoding returns None for short symbol list.""" - result = decoder._decode_mmsi([1, 2, 3]) - assert result is None - - def test_decode_mmsi_invalid_symbols(self, decoder): - """Test MMSI decoding returns None for out-of-range symbols.""" - # Symbols > 99 should cause decode to fail - symbols = [100, 32, 12, 34, 56] - result = decoder._decode_mmsi(symbols) - assert result is None - - def test_decode_position_northeast(self, decoder): - """Test position decoding for NE quadrant.""" - # Quadrant 10 = NE (lat+, lon+) - # Position: 51°30'N, 0°10'E - # lon_deg = symbols[3]*100 + symbols[4] = 0, lon_min = symbols[5] = 10 - symbols = [10, 51, 30, 0, 0, 10, 0, 0, 0, 0] - result = decoder._decode_position(symbols) - - assert result is not None - assert result["lat"] == pytest.approx(51.5, rel=0.01) - assert result["lon"] == pytest.approx(0.1667, rel=0.01) - - def test_decode_position_northwest(self, decoder): - """Test position decoding for NW quadrant.""" - # Quadrant 11 = NW (lat+, lon-) - # Position: 40°42'N, 74°00'W (NYC area) - symbols = [11, 40, 42, 0, 74, 0, 0, 0, 0, 0] - result = decoder._decode_position(symbols) - - assert result is not None - assert result["lat"] > 0 # North - assert result["lon"] < 0 # West - - def test_decode_position_southeast(self, decoder): - """Test position decoding for SE quadrant.""" - # Quadrant 0 = SE (lat-, lon+) - symbols = [0, 33, 51, 1, 51, 12, 0, 0, 0, 0] - result = decoder._decode_position(symbols) - - assert result is not None - assert result["lat"] < 0 # South - assert result["lon"] > 0 # East - - def test_decode_position_short_symbols(self, decoder): - """Test position decoding handles short symbol list.""" - result = decoder._decode_position([10, 51, 30]) - assert result is None - - def test_decode_position_invalid_values(self, decoder): - """Test position decoding handles invalid values gracefully.""" - # Latitude > 90 should be treated as 0 - symbols = [10, 95, 30, 0, 10, 0, 0, 0, 0, 0] - result = decoder._decode_position(symbols) - assert result is not None - assert result["lat"] == pytest.approx(0.5, rel=0.01) # 0 deg + 30 min - - def test_bits_to_symbol(self, decoder): - """Test bit to symbol conversion.""" - # Symbol value is first 7 bits (LSB first) - # Value 100 = 0b1100100 -> bits [0,0,1,0,0,1,1] -> 3 ones - # Check bits must make total even -> need 1 more one -> [1,0,0] - bits = [0, 0, 1, 0, 0, 1, 1, 1, 0, 0] - result = decoder._bits_to_symbol(bits) - assert result == 100 - - def test_bits_to_symbol_wrong_length(self, decoder): - """Test bit to symbol returns -1 for wrong length.""" - result = decoder._bits_to_symbol([0, 1, 0, 1, 0]) - assert result == -1 - - def test_detect_dot_pattern(self, decoder): - """Test dot pattern detection with 200+ alternating bits.""" - # Dot pattern requires at least 200 bits / 100 alternations - decoder.bit_buffer = [1, 0] * 110 # 220 alternating bits - assert decoder._detect_dot_pattern() is True - - def test_detect_dot_pattern_insufficient(self, decoder): - """Test dot pattern not detected with insufficient alternations.""" - decoder.bit_buffer = [1, 0] * 40 # Only 80 bits, below 200 threshold - assert decoder._detect_dot_pattern() is False - - def test_detect_dot_pattern_not_alternating(self, decoder): - """Test dot pattern not detected without alternation.""" - decoder.bit_buffer = [1, 1, 1, 1, 0, 0, 0, 0] * 5 - assert decoder._detect_dot_pattern() is False - - def test_bounded_phasing_strip(self, decoder): - """Test that >7 phasing symbols causes decode to return None.""" - # Build message bits: 10 phasing symbols (120) + format + data - # Each symbol is 10 bits. Phasing symbol 120 = 0b1111000 LSB first - # 120 in 7 bits LSB-first: 0,0,0,1,1,1,1 + 3 check bits - # 120 = 0b1111000 -> LSB first: 0,0,0,1,1,1,1 -> ones=4 (even) -> check [0,0,0] - phasing_bits = [0, 0, 0, 1, 1, 1, 1, 0, 0, 0] # symbol 120 - # 10 phasing symbols (>7 max) - decoder.message_bits = phasing_bits * 10 - # Add some non-phasing symbols after (enough for a message) - # Symbol 112 (INDIVIDUAL) = 0b1110000 LSB-first: 0,0,0,0,1,1,1 -> ones=3 (odd) -> need odd check - # For simplicity, just add enough bits for the decoder to attempt - for _ in range(20): - decoder.message_bits.extend([0, 0, 0, 0, 1, 1, 1, 1, 0, 0]) - result = decoder._try_decode_message() - assert result is None - - def test_eos_minimum_length(self, decoder): - """Test that EOS found too early in the symbol stream is skipped.""" - # Build a message where EOS appears at position 5 (< MIN_SYMBOLS_FOR_FORMAT=12) - # This should not be accepted as a valid message end - # Symbol 127 (EOS) = 0b1111111 LSB-first: 1,1,1,1,1,1,1 -> ones=7 (odd) -> check needs 1 one - # Use a simple approach: create symbols directly via _try_decode_message - # Create 5 normal symbols + EOS at position 5 — should be skipped - # Followed by more symbols and a real EOS at position 15 - from utils.dsc.decoder import DSCDecoder - - d = DSCDecoder() - - # Build symbols manually: we need _try_decode_message to find EOS too early - # Symbol 112 = format code. We'll build 10 bits per symbol. - # Since check bit validation is now active, we need valid check bits. - # Symbol value 10 = 0b0001010 LSB-first: 0,1,0,1,0,0,0, ones=2 (even) -> check [0,0,0] - sym_10 = [0, 1, 0, 1, 0, 0, 0, 0, 0, 0] - # Symbol 127 (EOS) = 0b1111111, ones=7 (odd) -> check needs odd total -> [1,0,0] - sym_eos = [1, 1, 1, 1, 1, 1, 1, 1, 0, 0] - - # 5 normal symbols + early EOS (should be skipped) + 8 more normal + real EOS - d.message_bits = sym_10 * 5 + sym_eos + sym_10 * 8 + sym_eos - result = d._try_decode_message() - # The early EOS at index 5 should be skipped; the one at index 14 - # is past MIN_SYMBOLS_FOR_FORMAT so it can be accepted. - # But the message content is garbage, so _decode_symbols will likely - # return None for other reasons. The key test: it doesn't return a - # message truncated at position 5. - # Just verify no crash and either None or a valid longer message - # (not truncated at the early EOS) - assert result is None or len(result.get("raw", "")) > 18 - - def test_bits_to_symbol_check_bit_validation(self, decoder): - """Test that _bits_to_symbol rejects symbols with invalid check bits.""" - # Symbol 100 = 0b1100100 LSB-first: 0,0,1,0,0,1,1 - # ones in data = 3, need total even -> check bits need 1 one - # Valid: [0,0,1,0,0,1,1, 1,0,0] -> total ones = 4 (even) -> valid - valid_bits = [0, 0, 1, 0, 0, 1, 1, 1, 0, 0] - assert decoder._bits_to_symbol(valid_bits) == 100 - - # Invalid: flip one check bit -> total ones = 5 (odd) -> invalid - invalid_bits = [0, 0, 1, 0, 0, 1, 1, 0, 0, 0] - assert decoder._bits_to_symbol(invalid_bits) == -1 - - def test_safety_is_critical(self): - """Test that SAFETY category is marked as critical.""" - import json - - from utils.dsc.parser import parse_dsc_message - - raw = json.dumps( - { - "type": "dsc", - "format": 123, - "source_mmsi": "232123456", - "category": "SAFETY", - "timestamp": "2025-01-15T12:00:00Z", - "raw": "123232123456100122", - } - ) - msg = parse_dsc_message(raw) - assert msg is not None - assert msg["is_critical"] is True - - -class TestDSCConstants: - """Tests for DSC constants.""" - - def test_format_codes_completeness(self): - """Test that all ITU-R M.493 format specifiers are defined.""" - from utils.dsc.constants import FORMAT_CODES - - # ITU-R M.493 format specifiers (and only these) - expected_keys = {102, 112, 114, 116, 120, 123} - assert set(FORMAT_CODES.keys()) == expected_keys - - def test_valid_format_specifiers_set(self): - """Test VALID_FORMAT_SPECIFIERS matches FORMAT_CODES keys.""" - from utils.dsc.constants import FORMAT_CODES, VALID_FORMAT_SPECIFIERS - - assert set(FORMAT_CODES.keys()) == VALID_FORMAT_SPECIFIERS - - def test_valid_eos_symbols(self): - """Test VALID_EOS contains the three ITU-defined EOS symbols.""" - from utils.dsc.constants import VALID_EOS - - assert {117, 122, 127} == VALID_EOS - - def test_distress_nature_codes_completeness(self): - """Test that all distress nature codes are defined.""" - from utils.dsc.constants import DISTRESS_NATURE_CODES - - # ITU-R M.493 distress nature codes - assert 100 in DISTRESS_NATURE_CODES # UNDESIGNATED - assert 101 in DISTRESS_NATURE_CODES # FIRE - assert 102 in DISTRESS_NATURE_CODES # FLOODING - assert 103 in DISTRESS_NATURE_CODES # COLLISION - assert 106 in DISTRESS_NATURE_CODES # SINKING - assert 109 in DISTRESS_NATURE_CODES # PIRACY - assert 110 in DISTRESS_NATURE_CODES # MOB - - def test_mid_country_map_completeness(self): - """Test that common MID codes are defined.""" - from utils.dsc.constants import MID_COUNTRY_MAP - - # Verify some key maritime nations - assert "232" in MID_COUNTRY_MAP # UK - assert "366" in MID_COUNTRY_MAP # USA - assert "351" in MID_COUNTRY_MAP # Panama - assert "257" in MID_COUNTRY_MAP # Norway - assert "211" in MID_COUNTRY_MAP # Germany - assert "503" in MID_COUNTRY_MAP # Australia - assert "431" in MID_COUNTRY_MAP # Japan - - def test_vhf_channel_70_frequency(self): - """Test DSC Channel 70 frequency constant.""" - from utils.dsc.constants import VHF_CHANNELS - - assert VHF_CHANNELS[70] == 156.525 - - def test_dsc_modulation_parameters(self): - """Test DSC modulation constants per ITU-R M.493.""" - from utils.dsc.constants import ( - DSC_BAUD_RATE, - DSC_MARK_FREQ, - DSC_SPACE_FREQ, - ) - - assert DSC_BAUD_RATE == 1200 - assert DSC_MARK_FREQ == 2100 - assert DSC_SPACE_FREQ == 1300 - - def test_telecommand_codes_full(self): - """Test TELECOMMAND_CODES_FULL covers 0-127 range.""" - from utils.dsc.constants import TELECOMMAND_CODES_FULL - - assert len(TELECOMMAND_CODES_FULL) == 128 - # Known codes map correctly - assert TELECOMMAND_CODES_FULL[100] == "F3E_G3E_ALL" - assert TELECOMMAND_CODES_FULL[107] == "DISTRESS_ACK" - # Unknown codes map to "UNKNOWN" - assert TELECOMMAND_CODES_FULL[0] == "UNKNOWN" - assert TELECOMMAND_CODES_FULL[99] == "UNKNOWN" - - def test_telecommand_formats(self): - """Test TELECOMMAND_FORMATS contains correct format codes.""" - from utils.dsc.constants import TELECOMMAND_FORMATS - - assert {112, 114, 116, 120, 123} == TELECOMMAND_FORMATS - - def test_min_symbols_for_format(self): - """Test MIN_SYMBOLS_FOR_FORMAT constant.""" - from utils.dsc.constants import MIN_SYMBOLS_FOR_FORMAT - - assert MIN_SYMBOLS_FOR_FORMAT == 12 +"""Tests for DSC (Digital Selective Calling) utilities.""" + +import json + +import pytest + + +class TestDSCParser: + """Tests for DSC parser utilities.""" + + def test_get_country_from_mmsi_ship_station(self): + """Test country lookup for standard ship MMSI.""" + from utils.dsc.parser import get_country_from_mmsi + + # UK ships start with 232-235 + assert get_country_from_mmsi("232123456") == "United Kingdom" + assert get_country_from_mmsi("235987654") == "United Kingdom" + + # US ships start with 303, 338, 366-369 + assert get_country_from_mmsi("366123456") == "USA" + assert get_country_from_mmsi("369000001") == "USA" + + # Panama (common flag of convenience) + assert get_country_from_mmsi("351234567") == "Panama" + assert get_country_from_mmsi("370000001") == "Panama" + + # Norway + assert get_country_from_mmsi("257123456") == "Norway" + + # Germany + assert get_country_from_mmsi("211000001") == "Germany" + + def test_get_country_from_mmsi_coast_station(self): + """Test country lookup for coast station MMSI (starts with 00).""" + from utils.dsc.parser import get_country_from_mmsi + + # Coast stations: 00 + MID + assert get_country_from_mmsi("002320001") == "United Kingdom" + assert get_country_from_mmsi("003660001") == "USA" + + def test_get_country_from_mmsi_group_station(self): + """Test country lookup for group station MMSI (starts with 0).""" + from utils.dsc.parser import get_country_from_mmsi + + # Group call: 0 + MID + assert get_country_from_mmsi("023200001") == "United Kingdom" + assert get_country_from_mmsi("036600001") == "USA" + + def test_get_country_from_mmsi_unknown(self): + """Test country lookup returns None for unknown MID.""" + from utils.dsc.parser import get_country_from_mmsi + + assert get_country_from_mmsi("999999999") is None + assert get_country_from_mmsi("100000000") is None + + def test_get_country_from_mmsi_invalid(self): + """Test country lookup handles invalid input.""" + from utils.dsc.parser import get_country_from_mmsi + + assert get_country_from_mmsi("") is None + assert get_country_from_mmsi(None) is None + assert get_country_from_mmsi("12") is None + + def test_get_distress_nature_text(self): + """Test distress nature code to text conversion.""" + from utils.dsc.parser import get_distress_nature_text + + assert get_distress_nature_text(100) == "UNDESIGNATED" + assert get_distress_nature_text(101) == "FIRE" + assert get_distress_nature_text(102) == "FLOODING" + assert get_distress_nature_text(103) == "COLLISION" + assert get_distress_nature_text(106) == "SINKING" + assert get_distress_nature_text(109) == "PIRACY" + assert get_distress_nature_text(110) == "MOB" # Man overboard + + def test_get_distress_nature_text_unknown(self): + """Test distress nature returns formatted unknown for invalid codes.""" + from utils.dsc.parser import get_distress_nature_text + + assert "UNKNOWN" in get_distress_nature_text(999) + assert "999" in get_distress_nature_text(999) + + def test_get_distress_nature_text_string_input(self): + """Test distress nature accepts string input.""" + from utils.dsc.parser import get_distress_nature_text + + assert get_distress_nature_text("101") == "FIRE" + assert get_distress_nature_text("invalid") == "invalid" + + def test_get_format_text(self): + """Test format code to text conversion per ITU-R M.493.""" + from utils.dsc.parser import get_format_text + + assert get_format_text(102) == "ALL_SHIPS" + assert get_format_text(112) == "INDIVIDUAL" + assert get_format_text(114) == "INDIVIDUAL_ACK" + assert get_format_text(116) == "GROUP" + assert get_format_text(120) == "DISTRESS" + assert get_format_text(123) == "ALL_SHIPS_URGENCY_SAFETY" + + def test_get_format_text_unknown(self): + """Test format code returns unknown for invalid codes.""" + from utils.dsc.parser import get_format_text + + result = get_format_text(999) + assert "UNKNOWN" in result + + def test_get_format_text_removed_codes(self): + """Test that non-ITU format codes are no longer recognized.""" + from utils.dsc.parser import get_format_text + + # These were previously defined but are not ITU-R M.493 specifiers + for code in [100, 104, 106, 108, 110, 118]: + result = get_format_text(code) + assert "UNKNOWN" in result + + def test_get_telecommand_text(self): + """Test telecommand code to text conversion.""" + from utils.dsc.parser import get_telecommand_text + + assert get_telecommand_text(100) == "F3E_G3E_ALL" + assert get_telecommand_text(105) == "DATA" + assert get_telecommand_text(107) == "DISTRESS_ACK" + assert get_telecommand_text(111) == "TEST" + + def test_get_category_priority(self): + """Test category priority values.""" + from utils.dsc.parser import get_category_priority + + # Distress has highest priority (0) + assert get_category_priority("DISTRESS") == 0 + assert get_category_priority("distress") == 0 + + # Urgency/safety + assert get_category_priority("ALL_SHIPS_URGENCY_SAFETY") == 2 + + # Routine-level + assert get_category_priority("ALL_SHIPS") == 5 + assert get_category_priority("GROUP") == 5 + assert get_category_priority("INDIVIDUAL") == 5 + + # Unknown gets default high number + assert get_category_priority("UNKNOWN") == 10 + + def test_validate_mmsi_valid(self): + """Test MMSI validation with valid numbers.""" + from utils.dsc.parser import validate_mmsi + + assert validate_mmsi("232123456") is True + assert validate_mmsi("366000001") is True + assert validate_mmsi("002320001") is True # Coast station + assert validate_mmsi("023200001") is True # Group station + + def test_validate_mmsi_invalid(self): + """Test MMSI validation rejects invalid numbers.""" + from utils.dsc.parser import validate_mmsi + + assert validate_mmsi("") is False + assert validate_mmsi(None) is False + assert validate_mmsi("12345678") is False # Too short + assert validate_mmsi("1234567890") is False # Too long + assert validate_mmsi("abcdefghi") is False # Not digits + assert validate_mmsi("000000000") is False # All zeros + + def test_classify_mmsi(self): + """Test MMSI classification.""" + from utils.dsc.parser import classify_mmsi + + # Ship stations (start with 2-7) + assert classify_mmsi("232123456") == "ship" + assert classify_mmsi("366000001") == "ship" + assert classify_mmsi("503000001") == "ship" + + # Coast stations (start with 00) + assert classify_mmsi("002320001") == "coast" + + # Group stations (start with 0, not 00) + assert classify_mmsi("023200001") == "group" + + # SAR aircraft (start with 111) + assert classify_mmsi("111232001") == "sar" + + # Aids to Navigation (start with 99) + assert classify_mmsi("992321001") == "aton" + + # Unknown + assert classify_mmsi("invalid") == "unknown" + assert classify_mmsi("812345678") == "unknown" + + def test_parse_dsc_message_distress(self): + """Test parsing a distress message with ITU format 120.""" + from utils.dsc.parser import parse_dsc_message + + raw = json.dumps( + { + "type": "dsc", + "format": 120, + "source_mmsi": "232123456", + "dest_mmsi": "002320001", + "category": "DISTRESS", + "nature": 101, + "position": {"lat": 51.5, "lon": -0.1}, + "telecommand1": 100, + "timestamp": "2025-01-15T12:00:00Z", + "raw": "120002032123456101100127", + } + ) + + msg = parse_dsc_message(raw) + + assert msg is not None + assert msg["type"] == "dsc_message" + assert msg["source_mmsi"] == "232123456" + assert msg["category"] == "DISTRESS" + assert msg["source_country"] == "United Kingdom" + assert msg["nature_of_distress"] == "FIRE" + assert msg["latitude"] == 51.5 + assert msg["longitude"] == -0.1 + assert msg["is_critical"] is True + assert msg["priority"] == 0 + + def test_parse_dsc_message_group(self): + """Test parsing a group call message.""" + from utils.dsc.parser import parse_dsc_message + + raw = json.dumps( + { + "type": "dsc", + "format": 116, + "source_mmsi": "366000001", + "dest_mmsi": "023200001", + "category": "GROUP", + "timestamp": "2025-01-15T12:00:00Z", + "raw": "116023200001366000001117", + } + ) + + msg = parse_dsc_message(raw) + + assert msg is not None + assert msg["category"] == "GROUP" + assert msg["source_country"] == "USA" + assert msg["is_critical"] is False + assert msg["priority"] == 5 + + def test_parse_dsc_message_individual(self): + """Test parsing an individual call message.""" + from utils.dsc.parser import parse_dsc_message + + raw = json.dumps( + { + "type": "dsc", + "format": 112, + "source_mmsi": "366000001", + "dest_mmsi": "232123456", + "category": "INDIVIDUAL", + "telecommand1": 100, + "timestamp": "2025-01-15T12:00:00Z", + "raw": "112232123456366000001100122", + } + ) + + msg = parse_dsc_message(raw) + + assert msg is not None + assert msg["category"] == "INDIVIDUAL" + assert msg["is_critical"] is False + + def test_parse_dsc_message_invalid_json(self): + """Test parsing rejects invalid JSON.""" + from utils.dsc.parser import parse_dsc_message + + assert parse_dsc_message("not json") is None + assert parse_dsc_message("{invalid}") is None + + def test_parse_dsc_message_missing_type(self): + """Test parsing rejects messages without correct type.""" + from utils.dsc.parser import parse_dsc_message + + raw = json.dumps({"source_mmsi": "232123456"}) + assert parse_dsc_message(raw) is None + + raw = json.dumps({"type": "other", "source_mmsi": "232123456"}) + assert parse_dsc_message(raw) is None + + def test_parse_dsc_message_missing_mmsi(self): + """Test parsing rejects messages without source MMSI.""" + from utils.dsc.parser import parse_dsc_message + + raw = json.dumps({"type": "dsc"}) + assert parse_dsc_message(raw) is None + + def test_parse_dsc_message_empty(self): + """Test parsing handles empty input.""" + from utils.dsc.parser import parse_dsc_message + + assert parse_dsc_message("") is None + assert parse_dsc_message(None) is None + assert parse_dsc_message(" ") is None + + def test_parse_dsc_message_rejects_non_itu_format(self): + """Test parser rejects records with non-ITU format specifier.""" + from utils.dsc.parser import parse_dsc_message + + for bad_format in [100, 104, 106, 108, 110, 118, 999]: + raw = json.dumps( + { + "type": "dsc", + "format": bad_format, + "source_mmsi": "232123456", + "category": "ROUTINE", + "raw": "120232123456100127", + } + ) + assert parse_dsc_message(raw) is None, f"Format {bad_format} should be rejected" + + def test_parse_dsc_message_rejects_telecommand_out_of_range(self): + """Test parser rejects records with telecommand out of 100-127 range.""" + from utils.dsc.parser import parse_dsc_message + + raw = json.dumps( + { + "type": "dsc", + "format": 120, + "source_mmsi": "232123456", + "dest_mmsi": "002320001", + "category": "DISTRESS", + "telecommand1": 200, + "timestamp": "2025-01-15T12:00:00Z", + "raw": "120002032123456200127", + } + ) + assert parse_dsc_message(raw) is None + + def test_parse_dsc_message_accepts_zero_telecommand(self): + """Test parser does not drop telecommand with value 100 (truthiness fix).""" + from utils.dsc.parser import parse_dsc_message + + raw = json.dumps( + { + "type": "dsc", + "format": 112, + "source_mmsi": "232123456", + "dest_mmsi": "366000001", + "category": "INDIVIDUAL", + "telecommand1": 100, + "telecommand2": 100, + "timestamp": "2025-01-15T12:00:00Z", + "raw": "112366000001232123456100100122", + } + ) + + msg = parse_dsc_message(raw) + assert msg is not None + assert msg["telecommand1"] == 100 + assert msg["telecommand2"] == 100 + + def test_parse_dsc_message_validates_raw_field(self): + """Test parser validates raw field structure.""" + from utils.dsc.parser import parse_dsc_message + + # Non-digit raw field + raw = json.dumps( + { + "type": "dsc", + "format": 120, + "source_mmsi": "232123456", + "category": "DISTRESS", + "raw": "12abc", + } + ) + assert parse_dsc_message(raw) is None + + # Raw field length not divisible by 3 + raw = json.dumps( + { + "type": "dsc", + "format": 120, + "source_mmsi": "232123456", + "category": "DISTRESS", + "raw": "1201", + } + ) + assert parse_dsc_message(raw) is None + + # Raw field with non-EOS last token + raw = json.dumps( + { + "type": "dsc", + "format": 120, + "source_mmsi": "232123456", + "category": "DISTRESS", + "raw": "120100", + } + ) + assert parse_dsc_message(raw) is None + + def test_parse_dsc_message_accepts_valid_eos_in_raw(self): + """Test parser accepts all three valid EOS values in raw field.""" + from utils.dsc.parser import parse_dsc_message + + for eos in [117, 122, 127]: + raw = json.dumps( + { + "type": "dsc", + "format": 120, + "source_mmsi": "232123456", + "dest_mmsi": "002320001", + "category": "DISTRESS", + "timestamp": "2025-01-15T12:00:00Z", + "raw": f"120002032123456{eos:03d}", + } + ) + msg = parse_dsc_message(raw) + assert msg is not None, f"EOS {eos} should be accepted" + + def test_parse_dsc_message_rejects_invalid_mmsi(self): + """Test parser rejects invalid MMSI values.""" + from utils.dsc.parser import parse_dsc_message + + # All-zeros MMSI + raw = json.dumps( + { + "type": "dsc", + "format": 120, + "source_mmsi": "000000000", + "category": "DISTRESS", + "raw": "120000000000127", + } + ) + assert parse_dsc_message(raw) is None + + # Short MMSI + raw = json.dumps( + { + "type": "dsc", + "format": 120, + "source_mmsi": "12345", + "category": "DISTRESS", + "raw": "120127", + } + ) + assert parse_dsc_message(raw) is None + + def test_parse_dsc_message_nature_zero_not_dropped(self): + """Test that nature code 0 is not dropped by truthiness check.""" + from utils.dsc.parser import parse_dsc_message + + raw = json.dumps( + { + "type": "dsc", + "format": 120, + "source_mmsi": "232123456", + "dest_mmsi": "002320001", + "category": "DISTRESS", + "nature": 0, + "timestamp": "2025-01-15T12:00:00Z", + "raw": "120002032123456000127", + } + ) + + msg = parse_dsc_message(raw) + assert msg is not None + assert msg["nature_code"] == 0 + + def test_parse_dsc_message_channel_zero_not_dropped(self): + """Test that channel value 0 is not dropped by truthiness check.""" + from utils.dsc.parser import parse_dsc_message + + raw = json.dumps( + { + "type": "dsc", + "format": 112, + "source_mmsi": "232123456", + "dest_mmsi": "366000001", + "category": "INDIVIDUAL", + "channel": 0, + "telecommand1": 100, + "timestamp": "2025-01-15T12:00:00Z", + "raw": "112366000001232123456100122", + } + ) + + msg = parse_dsc_message(raw) + assert msg is not None + assert msg["channel"] == 0 + + def test_format_dsc_for_display(self): + """Test message formatting for display.""" + from utils.dsc.parser import format_dsc_for_display + + msg = { + "category": "DISTRESS", + "source_mmsi": "232123456", + "source_country": "United Kingdom", + "dest_mmsi": "002320001", + "nature_of_distress": "FIRE", + "latitude": 51.5074, + "longitude": -0.1278, + "telecommand1_text": "F3E_G3E_ALL", + "channel": 16, + "timestamp": "2025-01-15T12:00:00Z", + } + + output = format_dsc_for_display(msg) + + assert "DISTRESS" in output + assert "232123456" in output + assert "United Kingdom" in output + assert "FIRE" in output + assert "51.5074" in output + assert "Channel: 16" in output + + +class TestDSCDecoder: + """Tests for DSC decoder utilities.""" + + @pytest.fixture + def decoder(self): + """Create a DSC decoder instance.""" + # Skip if scipy not available + pytest.importorskip("scipy") + pytest.importorskip("numpy") + from utils.dsc.decoder import DSCDecoder + + return DSCDecoder() + + def test_decode_mmsi_valid(self, decoder): + """Test MMSI decoding from symbols.""" + # 5 symbols, transmitted C5C4C3C2C1, are 10 digits X1...X10. + # Per ITU-R M.493-16: "digit X10 is always the digit 0" - X10 is + # the LAST digit, so the real 9-digit MMSI is digits 1-9. + # Uses a real MMSI seen throughout this decoder's over-the-air testing. + symbols = [33, 82, 16, 79, 40] + result = decoder._decode_mmsi(symbols) + assert result == "338216794" + def test_decode_mmsi_with_leading_zeros(self, decoder): + """Test MMSI decoding handles leading zeros (coast station).""" + # Coast station MMSI 002275100 (CROSS Gris Nez, a real coast + # station seen in real DSC traffic during this decoder's testing) + symbols = [0, 22, 75, 10, 0] + result = decoder._decode_mmsi(symbols) + assert result == "002275100" + def test_decode_mmsi_short_symbols(self, decoder): + """Test MMSI decoding returns None for short symbol list.""" + result = decoder._decode_mmsi([1, 2, 3]) + assert result is None + + def test_decode_mmsi_invalid_symbols(self, decoder): + """Test MMSI decoding returns None for out-of-range symbols.""" + symbols = [100, 32, 12, 34, 56] + result = decoder._decode_mmsi(symbols) + assert result is None + def test_decode_position_northeast(self, decoder): + """Test position decoding for NE quadrant.""" + # Quadrant 0 = NE per ITU-R M.493 Annex 1 Sec 8.1.2 (single digit, + # not a 2-digit symbol value). Position: 51 deg 30'N, 000 deg 10'E + symbols = [5, 13, 0, 0, 10] + result = decoder._decode_position(symbols) + assert result is not None + assert result["lat"] == pytest.approx(51.5, rel=0.01) + assert result["lon"] == pytest.approx(0.1667, rel=0.01) + def test_decode_position_northwest(self, decoder): + """Test position decoding for NW quadrant.""" + # Quadrant 1 = NW. Position: 40 deg 42'N, 074 deg 00'W (NYC area) + symbols = [14, 4, 20, 74, 0] + result = decoder._decode_position(symbols) + assert result is not None + assert result["lat"] > 0 # North + assert result["lon"] < 0 # West + def test_decode_position_southeast(self, decoder): + """Test position decoding for SE quadrant.""" + # Quadrant 2 = SE. Position: 33 deg 51'S, 151 deg 12'E (Sydney area) + symbols = [23, 35, 11, 51, 12] + result = decoder._decode_position(symbols) + assert result is not None + assert result["lat"] < 0 # South + assert result["lon"] > 0 # East + def test_decode_position_short_symbols(self, decoder): + """Test position decoding handles short symbol list.""" + result = decoder._decode_position([10, 51, 30]) + assert result is None + + def test_decode_position_invalid_values(self, decoder): + """Test position decoding flags (not silently discards or + clamps) physically-impossible values. + + Design: a partial/uncertain position can still be valuable + (e.g. for search and rescue, where an approximate position + beats none at all) - so out-of-range components are NOT + discarded, but the result is explicitly flagged via + position_uncertain/position_issue so downstream consumers know + not to trust it blindly. Constructed so degrees genuinely + decode to 95 (impossible) - see position_issue for exact + derivation. + """ + symbols = [9, 50, 30, 0, 10] + result = decoder._decode_position(symbols) + assert result is not None + assert result["position_uncertain"] is True + assert "latitude degrees" in result["position_issue"] + # the longitude portion is unaffected and still usable + assert result["lon"] == pytest.approx(0.1667, rel=0.01) + def test_bits_to_symbol(self, decoder): + """Test bit to symbol conversion.""" + # Symbol value is first 7 bits (LSB first) + # Value 100 = 0b1100100 -> bits [0,0,1,0,0,1,1] -> 3 ones + # Check bits must make total even -> need 1 more one -> [1,0,0] + bits = [0, 0, 1, 0, 0, 1, 1, 1, 0, 0] + result = decoder._bits_to_symbol(bits) + assert result == 100 + + def test_bits_to_symbol_wrong_length(self, decoder): + """Test bit to symbol returns -1 for wrong length.""" + result = decoder._bits_to_symbol([0, 1, 0, 1, 0]) + assert result == -1 + + def test_detect_dot_pattern(self, decoder): + """Test dot pattern detection with 200+ alternating bits.""" + # Dot pattern requires at least 200 bits / 100 alternations + decoder.bit_buffer = [1, 0] * 110 # 220 alternating bits + assert decoder._detect_dot_pattern() is True + + def test_detect_dot_pattern_insufficient(self, decoder): + """Test dot pattern not detected with insufficient bits buffered. + + VHF DSC uses a 20-bit dot pattern (checked within a 24-bit + window) - not the old HF/MF 200-bit/100-alternation threshold. + 16 bits is below the minimum window needed to even attempt + detection. + """ + decoder.bit_buffer = [1, 0] * 8 # 16 bits, below the 24-bit window + assert decoder._detect_dot_pattern() is False + + def test_detect_dot_pattern_not_alternating(self, decoder): + """Test dot pattern not detected without alternation.""" + decoder.bit_buffer = [1, 1, 1, 1, 0, 0, 0, 0] * 5 + assert decoder._detect_dot_pattern() is False + + def test_no_anchor_returns_none(self, decoder): + """Test that a symbol stream with no valid, repeating format + specifier never produces a decode. + + Replaces the old test_bounded_phasing_strip, which tested the + pre-fix-9 flat-parser's phasing-symbol-counting logic (bounding + a linear strip to <=7 symbols). That mechanism no longer exists + - fix 9 replaced it with anchor detection (the first position + where a real format-specifier value repeats 2 symbol-positions + later). This test exercises the equivalent safety property + (malformed/non-message data never produces a bogus decode) + against the actual current mechanism: symbol 126 is a real, + valid ten-unit-code value but is never a format specifier, so + no anchor can ever be found no matter how much data accumulates. + """ + # Symbol 126 = info bits [0,1,1,1,1,1,1] (LSB first) + check + # bits [0,0,1] (MSB first, encoding 1 B-element) - a genuinely + # valid, check-bit-correct symbol, just never a format specifier. + symbol_126_bits = [0, 1, 1, 1, 1, 1, 1, 0, 0, 1] + decoder.message_bits = symbol_126_bits * 60 # 60 symbols, plenty + result = decoder._try_decode_message() + assert result is None + + def test_eos_minimum_length(self, decoder): + """Test that EOS found too early in the symbol stream is skipped.""" + # Build a message where EOS appears at position 5 (< MIN_SYMBOLS_FOR_FORMAT=12) + # This should not be accepted as a valid message end + # Symbol 127 (EOS) = 0b1111111 LSB-first: 1,1,1,1,1,1,1 -> ones=7 (odd) -> check needs 1 one + # Use a simple approach: create symbols directly via _try_decode_message + # Create 5 normal symbols + EOS at position 5 — should be skipped + # Followed by more symbols and a real EOS at position 15 + from utils.dsc.decoder import DSCDecoder + + d = DSCDecoder() + + # Build symbols manually: we need _try_decode_message to find EOS too early + # Symbol 112 = format code. We'll build 10 bits per symbol. + # Since check bit validation is now active, we need valid check bits. + # Symbol value 10 = 0b0001010 LSB-first: 0,1,0,1,0,0,0, ones=2 (even) -> check [0,0,0] + sym_10 = [0, 1, 0, 1, 0, 0, 0, 0, 0, 0] + # Symbol 127 (EOS) = 0b1111111, ones=7 (odd) -> check needs odd total -> [1,0,0] + sym_eos = [1, 1, 1, 1, 1, 1, 1, 1, 0, 0] + + # 5 normal symbols + early EOS (should be skipped) + 8 more normal + real EOS + d.message_bits = sym_10 * 5 + sym_eos + sym_10 * 8 + sym_eos + result = d._try_decode_message() + # The early EOS at index 5 should be skipped; the one at index 14 + # is past MIN_SYMBOLS_FOR_FORMAT so it can be accepted. + # But the message content is garbage, so _decode_symbols will likely + # return None for other reasons. The key test: it doesn't return a + # message truncated at position 5. + # Just verify no crash and either None or a valid longer message + # (not truncated at the early EOS) + assert result is None or len(result.get("raw", "")) > 18 + + def test_bits_to_symbol_check_bit_validation(self, decoder): + """Test that _bits_to_symbol rejects symbols with invalid check bits.""" + # Symbol 100 = 0b1100100 LSB-first: 0,0,1,0,0,1,1 + # ones in data = 3, need total even -> check bits need 1 one + # Valid: [0,0,1,0,0,1,1, 1,0,0] -> total ones = 4 (even) -> valid + valid_bits = [0, 0, 1, 0, 0, 1, 1, 1, 0, 0] + assert decoder._bits_to_symbol(valid_bits) == 100 + + # Invalid: flip one check bit -> total ones = 5 (odd) -> invalid + invalid_bits = [0, 0, 1, 0, 0, 1, 1, 0, 0, 0] + assert decoder._bits_to_symbol(invalid_bits) == -1 + + def test_safety_is_critical(self): + """Test that SAFETY category is marked as critical.""" + import json + + from utils.dsc.parser import parse_dsc_message + + raw = json.dumps( + { + "type": "dsc", + "format": 123, + "source_mmsi": "232123456", + "category": "SAFETY", + "timestamp": "2025-01-15T12:00:00Z", + "raw": "123232123456100122", + } + ) + msg = parse_dsc_message(raw) + assert msg is not None + assert msg["is_critical"] is True + + +class TestDSCConstants: + """Tests for DSC constants.""" + + def test_format_codes_completeness(self): + """Test that all ITU-R M.493 format specifiers are defined.""" + from utils.dsc.constants import FORMAT_CODES + + # ITU-R M.493 format specifiers (and only these) + expected_keys = {102, 112, 114, 116, 120, 123} + assert set(FORMAT_CODES.keys()) == expected_keys + + def test_valid_format_specifiers_set(self): + """Test VALID_FORMAT_SPECIFIERS matches FORMAT_CODES keys.""" + from utils.dsc.constants import FORMAT_CODES, VALID_FORMAT_SPECIFIERS + + assert set(FORMAT_CODES.keys()) == VALID_FORMAT_SPECIFIERS + + def test_valid_eos_symbols(self): + """Test VALID_EOS contains the three ITU-defined EOS symbols.""" + from utils.dsc.constants import VALID_EOS + + assert {117, 122, 127} == VALID_EOS + + def test_distress_nature_codes_completeness(self): + """Test that all distress nature codes are defined.""" + from utils.dsc.constants import DISTRESS_NATURE_CODES + + # ITU-R M.493 distress nature codes + assert 100 in DISTRESS_NATURE_CODES # UNDESIGNATED + assert 101 in DISTRESS_NATURE_CODES # FIRE + assert 102 in DISTRESS_NATURE_CODES # FLOODING + assert 103 in DISTRESS_NATURE_CODES # COLLISION + assert 106 in DISTRESS_NATURE_CODES # SINKING + assert 109 in DISTRESS_NATURE_CODES # PIRACY + assert 110 in DISTRESS_NATURE_CODES # MOB + + def test_mid_country_map_completeness(self): + """Test that common MID codes are defined.""" + from utils.dsc.constants import MID_COUNTRY_MAP + + # Verify some key maritime nations + assert "232" in MID_COUNTRY_MAP # UK + assert "366" in MID_COUNTRY_MAP # USA + assert "351" in MID_COUNTRY_MAP # Panama + assert "257" in MID_COUNTRY_MAP # Norway + assert "211" in MID_COUNTRY_MAP # Germany + assert "503" in MID_COUNTRY_MAP # Australia + assert "431" in MID_COUNTRY_MAP # Japan + + def test_vhf_channel_70_frequency(self): + """Test DSC Channel 70 frequency constant.""" + from utils.dsc.constants import VHF_CHANNELS + + assert VHF_CHANNELS[70] == 156.525 + + def test_dsc_modulation_parameters(self): + """Test DSC modulation constants per ITU-R M.493.""" + from utils.dsc.constants import ( + DSC_BAUD_RATE, + DSC_MARK_FREQ, + DSC_SPACE_FREQ, + ) + + assert DSC_BAUD_RATE == 1200 + assert DSC_MARK_FREQ == 2100 + assert DSC_SPACE_FREQ == 1300 + + def test_telecommand_codes_full(self): + """Test TELECOMMAND_CODES_FULL covers 0-127 range.""" + from utils.dsc.constants import TELECOMMAND_CODES_FULL + + assert len(TELECOMMAND_CODES_FULL) == 128 + # Known codes map correctly + assert TELECOMMAND_CODES_FULL[100] == "F3E_G3E_ALL" + assert TELECOMMAND_CODES_FULL[107] == "DISTRESS_ACK" + # Unknown codes map to "UNKNOWN" + assert TELECOMMAND_CODES_FULL[0] == "UNKNOWN" + assert TELECOMMAND_CODES_FULL[99] == "UNKNOWN" + + def test_telecommand_formats(self): + """Test TELECOMMAND_FORMATS contains correct format codes.""" + from utils.dsc.constants import TELECOMMAND_FORMATS + + assert {112, 114, 116, 120, 123} == TELECOMMAND_FORMATS + + def test_min_symbols_for_format(self): + """Test MIN_SYMBOLS_FOR_FORMAT constant.""" + from utils.dsc.constants import MIN_SYMBOLS_FOR_FORMAT + + assert MIN_SYMBOLS_FOR_FORMAT == 12 diff --git a/utils/dsc/decoder.py b/utils/dsc/decoder.py index fd91970..9176bbb 100644 --- a/utils/dsc/decoder.py +++ b/utils/dsc/decoder.py @@ -64,8 +64,11 @@ class DSCDecoder: self.samples_per_bit = sample_rate // self.baud_rate # FSK frequencies - self.mark_freq = DSC_MARK_FREQ # 2100 Hz = binary 1 - self.space_freq = DSC_SPACE_FREQ # 1300 Hz = binary 0 + # Per ITU-R M.493 Annex 1 Sec 1.4: higher frequency (2100 Hz) is + # the B-state (binary 0), lower frequency (1300 Hz) is the + # Y-state (binary 1). + self.mark_freq = DSC_MARK_FREQ # 2100 Hz = binary 0 (B state) + self.space_freq = DSC_SPACE_FREQ # 1300 Hz = binary 1 (Y state) # Bandpass filter for DSC band (1100-2300 Hz) nyq = sample_rate / 2 @@ -83,54 +86,87 @@ class DSCDecoder: self.message_bits = [] def _build_correlators(self): - """Build matched filter correlators for mark and space frequencies.""" + """ + Build quadrature (I/Q) matched filter correlators for mark and + space frequencies. + + A single-phase sine correlator is phase-sensitive: if the + incoming tone is ~90 degrees out of phase with the reference, + the correlation magnitude collapses toward zero even when the + correct tone is present, since there's no timing recovery + locking the symbol window to the actual bit transitions. Using + both a sine (Q) and cosine (I) reference and combining them as + sqrt(I^2 + Q^2) makes the magnitude phase-independent, which is + required for a free-running (non-timing-recovered) symbol + window like this one. + """ # Duration for one bit t = np.arange(self.samples_per_bit) / self.sample_rate - # Mark correlator (1800 Hz) - self.mark_ref = np.sin(2 * np.pi * self.mark_freq * t) + # Mark (2100 Hz) quadrature references + self.mark_ref_i = np.cos(2 * np.pi * self.mark_freq * t) + self.mark_ref_q = np.sin(2 * np.pi * self.mark_freq * t) - # Space correlator (1200 Hz) - self.space_ref = np.sin(2 * np.pi * self.space_freq * t) + # Space (1300 Hz) quadrature references + self.space_ref_i = np.cos(2 * np.pi * self.space_freq * t) + self.space_ref_q = np.sin(2 * np.pi * self.space_freq * t) def process_audio(self, audio_data: bytes) -> Generator[dict, None, None]: """ Process audio data and yield decoded DSC messages. + Uses a stateful bandpass filter (carrying scipy's lfilter `zi` + state across calls) instead of re-filtering an overlapping + buffer each call. The previous approach re-filtered a retained + tail of samples on every call and re-yielded the resulting + bits, duplicating already-processed bits into the bit stream + on every call after the first - this corrupted sync/symbol + alignment starting right after the first chunk, and compounded + with every subsequent call. It also introduced small filter + transients at each chunk boundary from resetting filter state + to zero every time. Carrying filter state eliminates both: + confirmed to produce a bit-for-bit identical stream to a + single continuous one-shot filter pass on real captured audio. + + Any leftover samples that don't complete a full bit period are + buffered and prepended to the next call's audio, so this works + correctly regardless of the caller's chunk size. + Args: audio_data: Raw 16-bit signed PCM audio bytes Yields: Decoded DSC message dicts """ - # Convert bytes to numpy array samples = np.frombuffer(audio_data, dtype=np.int16) if len(samples) == 0: return - # Append to buffer - self.buffer = np.concatenate([self.buffer, samples]) + if not hasattr(self, "_filter_zi"): + self._filter_zi = scipy_signal.lfilter_zi(self.bp_b, self.bp_a) * samples[0] + if not hasattr(self, "_leftover_samples"): + self._leftover_samples = np.array([], dtype=np.int16) - # Need at least one bit worth of samples - if len(self.buffer) < self.samples_per_bit: + samples = np.concatenate([self._leftover_samples, samples]) + + usable_len = (len(samples) // self.samples_per_bit) * self.samples_per_bit + if usable_len == 0: + self._leftover_samples = samples return - # Apply bandpass filter + to_process = samples[:usable_len] + self._leftover_samples = samples[usable_len:] + try: - filtered = scipy_signal.lfilter(self.bp_b, self.bp_a, self.buffer) + filtered, self._filter_zi = scipy_signal.lfilter( + self.bp_b, self.bp_a, to_process.astype(np.float64), zi=self._filter_zi + ) except Exception as e: logger.warning(f"Filter error: {e}") return - # Demodulate FSK using correlation bits = self._demodulate_fsk(filtered) - # Keep unprocessed samples (last bit's worth) - keep_samples = self.samples_per_bit * 2 - if len(self.buffer) > keep_samples: - self.buffer = self.buffer[-keep_samples:] - - # Process decoded bits for bit in bits: message = self._process_bit(bit) if message: @@ -138,7 +174,8 @@ class DSCDecoder: def _demodulate_fsk(self, samples: np.ndarray) -> list[int]: """ - Demodulate FSK audio to bits using correlation. + Demodulate FSK audio to bits using phase-independent quadrature + correlation. Args: samples: Filtered audio samples @@ -157,15 +194,22 @@ class DSCDecoder: if len(segment) < self.samples_per_bit: break - # Correlate with mark and space references - mark_corr = np.abs(np.correlate(segment, self.mark_ref, mode="valid")) - space_corr = np.abs(np.correlate(segment, self.space_ref, mode="valid")) + # Quadrature (I/Q) magnitude - phase independent, unlike a + # single-phase correlation against just a sine reference. + mark_i = np.dot(segment, self.mark_ref_i) + mark_q = np.dot(segment, self.mark_ref_q) + mark_mag = np.sqrt(mark_i**2 + mark_q**2) - # Decision: mark (1) if mark correlation > space correlation - if np.max(mark_corr) > np.max(space_corr): - bits.append(1) - else: + space_i = np.dot(segment, self.space_ref_i) + space_q = np.dot(segment, self.space_ref_q) + space_mag = np.sqrt(space_i**2 + space_q**2) + + # Per ITU-R M.493: mark (2100 Hz) = binary 0 (B state), + # space (1300 Hz) = binary 1 (Y state) + if mark_mag > space_mag: bits.append(0) + else: + bits.append(1) return bits @@ -173,6 +217,15 @@ class DSCDecoder: """ Process a decoded bit and detect/decode DSC messages. + Uses a two-stage sync: coarse dot-pattern detection (20+ + alternating bits, per VHF spec) followed by a fine-sync step + that tests bit offsets 0-9 against the check-bit test to find + the transmitter's actual 10-bit symbol grid. The coarse + trigger point is not reliably grid-aligned on its own - the + grid position is arbitrary relative to wherever our sample + buffer happens to start, and being off by even a few bits + breaks every subsequent check-bit test even on a clean signal. + Args: bit: Decoded bit (0 or 1) @@ -180,313 +233,576 @@ class DSCDecoder: Decoded message dict if complete message found, None otherwise """ self.bit_buffer.append(bit) - - # Keep buffer manageable if len(self.bit_buffer) > 2000: self.bit_buffer = self.bit_buffer[-1500:] - # Look for dot pattern (sync) - alternating 1010101... + if not hasattr(self, "sync_pending"): + self.sync_pending = False + self.sync_lookahead = [] + + if self.sync_pending: + self.sync_lookahead.append(bit) + if len(self.sync_lookahead) >= 40: + best_offset, best_valid = 0, -1 + for offset in range(10): + valid = 0 + for k in range(3): + s = offset + k * 10 + if s + 10 <= len(self.sync_lookahead): + symbol_bits = self.sync_lookahead[s:s + 10] + if self._bits_to_symbol(symbol_bits) != -1: + valid += 1 + if valid > best_valid: + best_valid, best_offset = valid, offset + if best_valid >= 2: + self.in_message = True + self.message_bits = self.sync_lookahead[best_offset:] + logger.debug( + f"DSC fine sync: offset={best_offset} valid={best_valid}/3" + ) + else: + logger.debug("DSC fine sync failed, discarding false trigger") + self.sync_pending = False + self.sync_lookahead = [] + return None + if not self.in_message and self._detect_dot_pattern(): - self.in_message = True - self.message_bits = [] - logger.debug("DSC sync detected") + self.sync_pending = True + self.sync_lookahead = [] + logger.debug("DSC sync detected, refining bit alignment") return None # Collect message bits if self.in_message: self.message_bits.append(bit) - # Check for end of message or timeout - if len(self.message_bits) >= 10: # One symbol + if len(self.message_bits) >= 10 and len(self.message_bits) % 10 == 0: # One new complete symbol # Try to decode accumulated symbols message = self._try_decode_message() if message: self.in_message = False self.message_bits = [] return message - # Timeout - too many bits without valid message if len(self.message_bits) > 1800: # ~180 symbols max - logger.debug("DSC message timeout") + diagnosis = self._diagnose_decode_failure() + logger.debug(f"DSC message timeout: {diagnosis}") self.in_message = False self.message_bits = [] - return None def _detect_dot_pattern(self) -> bool: """ Detect DSC dot pattern for synchronization. - The dot pattern is at least 200 alternating bits (1010101...). - We require at least 100 consecutive alternations to avoid - false sync triggers from noise. + Per ITU-R M.493 Annex 1 Sec 3.4, there are two dot pattern + lengths: 200 bits (HF/MF acknowledgements and coast-station + calls) and 20 bits (VHF - all calls, per Sec 3.4.2, confirmed + by ETSI EN 300 338-3: "the equipment shall automatically set + the dot pattern length to 20 bits for all transmitted DSC + messages" on VHF). This decoder targets VHF ch. 70, so it must + use the 20-bit pattern - the previous 100-alternation/200-bit + requirement was the HF/MF number and made VHF sync structurally + unreachable, since real VHF traffic only ever produces roughly + 20-27 alternating bits. + + We check a slightly wider window (24 bits) than the strict + 20-bit minimum to tolerate a possible bit error right at the + start of the pattern. """ - if len(self.bit_buffer) < 200: + if len(self.bit_buffer) < 24: return False - - # Check last 200 bits for alternating pattern - last_bits = self.bit_buffer[-200:] + last_bits = self.bit_buffer[-24:] alternations = 0 - for i in range(1, len(last_bits)): if last_bits[i] != last_bits[i - 1]: alternations += 1 else: alternations = 0 - - if alternations >= 100: + if alternations >= 20: return True - return False def _try_decode_message(self) -> dict | None: """ - Try to decode accumulated message bits as DSC message. + Decode accumulated message bits into a DSC message. + + Per ITU-R M.493 Annex 1 Sec 1.2: apart from phasing, every + character in a DSC call is transmitted TWICE (DX then RX, ~5 + symbol-positions apart) - this is the actual wire format, not + an optional extra. The format specifier is additionally doubled + at the FIELD level (Sec 4.1: "2 identical characters"), giving + 4 raw occurrences for it specifically and providing a reliable + anchor: the first position where a real format-specifier value + repeats exactly 2 symbol-positions later. From that anchor, + every other symbol (stride 2) is the de-interleaved, real + message content - confirmed empirically against real captures + and against Annex 1 Figure 1 (call sequence construction). Returns: - Decoded message dict or None if not yet complete/valid + Decoded message dict, or None if not yet decodable (anchor + not found, fields incomplete, or EOS not found yet). """ - # Need at least a few symbols to start decoding - num_symbols = len(self.message_bits) // 10 + FORMAT_SPECIFIERS = {102, 112, 114, 116, 120, 123} + FORMAT_TEXT = { + 102: "GEOGRAPHIC_AREA", + 112: "DISTRESS", + 114: "GROUP", + 116: "ALL_SHIPS", + 120: "INDIVIDUAL", + 123: "INDIVIDUAL_SEMI_AUTO", + } + CATEGORY_TEXT = { + 112: "DISTRESS", + 110: "URGENCY", + 108: "SAFETY", + 106: "SHIPS_BUSINESS", + 100: "ROUTINE", + } + # Nature of distress (Table 10 / Table A1-3), confirmed against + # spec text directly, consistent across M.493-8 through -15. + NATURE_TEXT = { + 100: "FIRE_EXPLOSION", + 101: "FLOODING", + 102: "COLLISION", + 103: "GROUNDING", + 104: "LISTING_CAPSIZE_DANGER", + 105: "SINKING", + 106: "DISABLED_ADRIFT", + 107: "UNDESIGNATED", + 108: "ABANDONING_SHIP", + 109: "PIRACY_ARMED_ROBBERY", + 110: "MAN_OVERBOARD", + } + # Telecommand values (Table 11/12), confirmed directly against + # spec text. This is not the full table - only values we've + # actually cited are included; anything else reports as + # UNKNOWN- rather than a guess. 126 ("no information") is + # by far the most common value for the second telecommand slot, + # since most calls don't use it for anything. + TELECOMMAND_TEXT = { + 100: "F3E/G3E ALL MODES TP", + 101: "F3E/G3E DUPLEX TP", + 118: "TEST", + 121: "POSITION_OR_LOCATION_UPDATING", + 126: "NO_INFORMATION", + } + # EOS (end of sequence) meaning, per ITU-R M.493 Sec 9, confirmed + # directly against spec text. + EOS_TEXT = { + 117: "ACKNOWLEDGEMENT_REQUESTED", + 122: "ACKNOWLEDGEMENT_REPLY", + 127: "NO_ACKNOWLEDGEMENT_REQUIRED", + } + VALID_EOS = {117, 122, 127} + # Field layouts per format specifier: (field_name, symbol_count), + # in transmission order after the format specifier. Per Table 5 + # (selective calls) and Table 4 (distress). Distress has no + # address/category field (priority is the format specifier + # itself). + FIELD_LAYOUTS = { + 120: [("address", 5), ("category", 1), ("self_id", 5), + ("telecommand", 2), ("frequency_1", 3), ("frequency_2", 3)], + 123: [("address", 5), ("category", 1), ("self_id", 5), + ("telecommand", 2), ("frequency_1", 3)], + 116: [("category", 1), ("self_id", 5), ("telecommand", 2), + ("frequency_1", 3)], + 114: [("address", 5), ("category", 1), ("self_id", 5), + ("telecommand", 2), ("frequency_1", 3), ("frequency_2", 3)], + 102: [("address", 5), ("category", 1), ("self_id", 5), + ("telecommand", 2), ("frequency_1", 3), ("frequency_2", 3)], + 112: [("self_id", 5), ("nature", 1), ("coordinates", 5), + ("time", 2), ("telecommand_msg4", 1)], + } + # Format 116 is overloaded: normally "All Ships" broadcast (the + # FIELD_LAYOUTS[116] entry above), but per ITU-R M.493-16 it is + # ALSO used for distress alert acknowledgement / distress + # self-cancel (Table A1-4.2) - a structurally different message. + # Per spec: "Distress acknowledgments where the transmitting ID + # and ship in distress ID are the same...should be interpreted + # as a self-cancel operation." Distinguished by category: a + # normal All Ships broadcast uses ROUTINE/SAFETY/URGENCY: a + # distress acknowledgement/self-cancel uses category DISTRESS + # (112). Layout below empirically mapped and validated against a + # real captured self-cancel message (own MMSI confirmed present, + # nature/coordinates/time matching the alert being cancelled). + # field_8_unconfirmed's exact semantic meaning is not yet + # confirmed against spec text - flagged honestly rather than + # guessed. + DISTRESS_ACK_LAYOUT_116 = [ + ("ship_in_distress_id", 5), + ("field_8_unconfirmed", 1), + ("ship_in_distress_id_repeat", 5), + ("nature", 1), + ("coordinates", 5), + ("time", 2), + ] + num_symbols = len(self.message_bits) // 10 if num_symbols < 5: return None - # Extract symbols (10 bits each) symbols = [] for i in range(num_symbols): - start = i * 10 - end = start + 10 - if end <= len(self.message_bits): - symbol_bits = self.message_bits[start:end] - symbol_value = self._bits_to_symbol(symbol_bits) - if symbol_value == -1: - logger.debug("DSC symbol check bit failure, aborting decode") - return None - symbols.append(symbol_value) + symbol_bits = self.message_bits[i * 10:i * 10 + 10] + symbol_value = self._bits_to_symbol(symbol_bits) + symbols.append(symbol_value) - # Strip phasing sequence (RX/DX symbols 120-126) from the - # start of the message. Per ITU-R M.493, after the dot pattern - # there are 7 phasing symbols before the format specifier. - # Bound to max 7 — if more are present, this is a bad sync. - msg_start = 0 - for i, sym in enumerate(symbols): - if 120 <= sym <= 126: - msg_start = i + 1 - else: + # Find the message anchor: first position where a real format + # specifier value repeats exactly 2 positions later. + anchor = None + for i in range(len(symbols) - 2): + if symbols[i] in FORMAT_SPECIFIERS and symbols[i] == symbols[i + 2]: + anchor = i break - if msg_start > 7: - logger.debug("DSC bad sync: >7 phasing symbols stripped") - return None - symbols = symbols[msg_start:] - - if len(symbols) < 5: + if anchor is None: return None - # Look for EOS (End of Sequence) - symbols 117, 122, or 127 - # EOS must appear after at least MIN_SYMBOLS_FOR_FORMAT symbols - eos_found = False - eos_index = -1 - for i, sym in enumerate(symbols): - if sym in VALID_EOS: - if i < MIN_SYMBOLS_FOR_FORMAT: - continue # Too early — not a real EOS - eos_found = True - eos_index = i + # Track A is the primary de-interleaved sequence used for field + # boundaries. Track B (offset by one raw symbol position) is the + # redundant time-diversity copy: per ITU-R M.493 Annex 1 Sec 1.2, + # every character apart from phasing is transmitted twice (DX + # then RX). Confirmed empirically: Track A's logical position k + # pairs with Track B's position k+2 (Track B starts 2 positions + # "behind" since it still has 2 trailing phasing remnants before + # it catches up to real content) - validated with zero mismatches + # across a full real capture, and confirmed to correctly recover + # a synthetically-corrupted symbol in testing. When Track A's + # copy of a character fails the check-bit test, fall back to + # Track B's copy before giving up on that character - this is + # the actual error-correction mechanism ITU-R M.493 relies on. + dedup = symbols[anchor::2] + dedup_b = symbols[anchor + 1::2] + if len(dedup) < 2: + return None + + def value_at(k): + """Track A position k, falling back to Track B's diversity + copy (position k+2) if Track A's copy failed check bits. + This is the diversity-combining fallback that recovers + isolated real-world bit errors using ITU-R M.493's built-in + time-diversity redundancy.""" + primary = dedup[k] if k < len(dedup) else -1 + if primary != -1: + return primary + fallback_idx = k + 2 + if 0 <= fallback_idx < len(dedup_b): + return dedup_b[fallback_idx] + return -1 + + format_code = value_at(0) + layout = FIELD_LAYOUTS.get(format_code) + if layout is None: + logger.debug(f"DSC unsupported/unrecognized format specifier: {format_code}") + return None + + # Format 116 needs a peek at its category field to know which of + # the two structurally different layouts applies (see + # DISTRESS_ACK_LAYOUT_116 definition above for why). + is_distress_ack = False + if format_code == 116: + peeked_category = value_at(2) + if peeked_category == 112: # DISTRESS + layout = DISTRESS_ACK_LAYOUT_116 + is_distress_ack = True + + idx = 2 # skip the two format-specifier copies + fields = {} + if is_distress_ack: + fields["category"] = [112] + idx += 1 # already consumed by the peek above + for name, count in layout: + if idx + count > len(dedup): + return None # not enough data yet, keep accumulating + chunk = [value_at(idx + j) for j in range(count)] + fields[name] = chunk + idx += count + + eos = None + for k in range(idx, len(dedup)): + v = value_at(k) + if v in VALID_EOS: + eos = v break + if eos is None: + return None # message not complete yet - if not eos_found: - # Not complete yet + message = { + "type": "dsc", + "format": format_code, + "format_text": FORMAT_TEXT.get(format_code, f"UNKNOWN-{format_code}"), + "eos": eos, + "eos_text": EOS_TEXT.get(eos, f"UNKNOWN-{eos}"), + "timestamp": datetime.utcnow().isoformat() + "Z", + } + if is_distress_ack: + message["format_text"] = "DISTRESS_ACK_OR_SELF_CANCEL" + if "address" in fields: + message["dest_mmsi"] = self._decode_mmsi(fields["address"]) + if "self_id" in fields: + message["source_mmsi"] = self._decode_mmsi(fields["self_id"]) + if "ship_in_distress_id" in fields: + ship_mmsi = self._decode_mmsi(fields["ship_in_distress_id"]) + message["ship_in_distress_mmsi"] = ship_mmsi + if "source_mmsi" not in message: + # for a self-cancel, transmitting station IS the ship in + # distress - report it as source too for consistency with + # other message types + message["source_mmsi"] = ship_mmsi + if "category" in fields: + cat_val = fields["category"][0] + message["category"] = CATEGORY_TEXT.get(cat_val, f"UNKNOWN-{cat_val}") + if format_code == 112 or is_distress_ack: + message["category"] = "DISTRESS" + if "nature" in fields: + nature_val = fields["nature"][0] + message["nature"] = nature_val + message["nature_text"] = NATURE_TEXT.get(nature_val, f"UNKNOWN-{nature_val}") + if "coordinates" in fields: + message["coordinates_raw"] = fields["coordinates"] + message["position"] = self._decode_position(fields["coordinates"]) + if "time" in fields: + message["time_raw"] = fields["time"] + if "telecommand" in fields: + message["telecommand"] = fields["telecommand"] + message["telecommand_text"] = [ + TELECOMMAND_TEXT.get(t, f"UNKNOWN-{t}") for t in fields["telecommand"] + ] + + # Plain-English summary, built from the fields actually present - + # different message types (individual/group/area, all ships, + # distress alert, distress ack/self-cancel) have different fields, + # so this is assembled per-type rather than one fixed template. + eos_phrase = { + 117: "requesting acknowledgement", + 122: "acknowledging a previous call", + 127: "no acknowledgement required", + }.get(eos, "unknown EOS") + tc_phrase = message.get("telecommand_text", [None])[0] + tc_phrase = f", purpose: {tc_phrase}" if tc_phrase and tc_phrase != "NO_INFORMATION" else "" + + if is_distress_ack: + message["summary"] = ( + f"Distress self-cancel/acknowledgement from {message.get('source_mmsi')} " + f"for vessel {message.get('ship_in_distress_mmsi')} " + f"(original distress nature: {message.get('nature_text', 'unknown')})" + ) + elif format_code == 112: + message["summary"] = ( + f"DISTRESS ALERT from {message.get('source_mmsi')} - " + f"nature: {message.get('nature_text', 'unknown')}" + ) + elif "dest_mmsi" in message: + message["summary"] = ( + f"{message.get('category', 'UNKNOWN')} individual/group call from " + f"{message.get('source_mmsi')} to {message.get('dest_mmsi')}, " + f"{eos_phrase}{tc_phrase}" + ) + else: + message["summary"] = ( + f"{message.get('category', 'UNKNOWN')} broadcast from " + f"{message.get('source_mmsi')} to all ships, {eos_phrase}{tc_phrase}" + ) + + return message + + def _diagnose_decode_failure(self) -> str: + """ + Explain exactly why the current message_bits buffer failed to + decode, for logging at timeout. Mirrors _try_decode_message's + logic but reports where it stopped instead of returning None + silently - normal accumulation still returns None quietly (this + is only called once, at the moment of timeout, not on every + partial-accumulation call). + """ + FORMAT_SPECIFIERS = {102, 112, 114, 116, 120, 123} + FIELD_LAYOUTS = { + 120: [("address", 5), ("category", 1), ("self_id", 5), + ("telecommand", 2), ("frequency_1", 3), ("frequency_2", 3)], + 123: [("address", 5), ("category", 1), ("self_id", 5), + ("telecommand", 2), ("frequency_1", 3)], + 116: [("category", 1), ("self_id", 5), ("telecommand", 2), + ("frequency_1", 3)], + 114: [("address", 5), ("category", 1), ("self_id", 5), + ("telecommand", 2), ("frequency_1", 3), ("frequency_2", 3)], + 102: [("address", 5), ("category", 1), ("self_id", 5), + ("telecommand", 2), ("frequency_1", 3), ("frequency_2", 3)], + 112: [("self_id", 5), ("nature", 1), ("coordinates", 5), + ("time", 2), ("telecommand_msg4", 1)], + } + VALID_EOS = {117, 122, 127} + + num_symbols = len(self.message_bits) // 10 + symbols = [self._bits_to_symbol(self.message_bits[i * 10:i * 10 + 10]) for i in range(num_symbols)] + invalid_count = sum(1 for s in symbols if s == -1) + + anchor = None + for i in range(len(symbols) - 2): + if symbols[i] in FORMAT_SPECIFIERS and symbols[i] == symbols[i + 2]: + anchor = i + break + if anchor is None: + return (f"no anchor found ({num_symbols} symbols, {invalid_count} invalid, " + f"first 15 symbols={symbols[:15]})") + + dedup = symbols[anchor::2] + dedup_b = symbols[anchor + 1::2] + + def value_at(k): + primary = dedup[k] if k < len(dedup) else -1 + if primary != -1: + return primary + fb = k + 2 + return dedup_b[fb] if 0 <= fb < len(dedup_b) else -1 + + format_code = value_at(0) + layout = FIELD_LAYOUTS.get(format_code) + if layout is None: + return f"anchor found at {anchor}, but format specifier {format_code} not recognized" + + idx = 2 + for name, count in layout: + if idx + count > len(dedup): + have = len(dedup) - idx + return (f"format={format_code}, stalled waiting for field " + f"'{name}' (need {count} symbols, have {max(have, 0)})") + idx += count + + for k in range(idx, len(dedup)): + if value_at(k) in VALID_EOS: + return "all fields complete, EOS found - should have decoded (unexpected)" + + return (f"format={format_code}, all fields complete, but no valid EOS " + f"found in remaining {len(dedup) - idx} symbols") + + def _decode_mmsi(self, symbols: list[int]) -> str | None: + """ + Decode 5 symbols into a 9-digit MMSI string. + + Per ITU-R M.493-16: identities are transmitted as five + characters C5C4C3C2C1, comprising ten digits X1...X10, "whereas + digit X10 is always the digit 0" (unless M.1080 extended + addressing is in use, not handled here). X10 is the LAST digit + of the sequence, not the first - so the real 9-digit MMSI is + digits 1-9, dropping the trailing padding digit. Confirmed + against real over-the-air captures: this produces MMSIs with + recognizable, valid Maritime Identification Digits (e.g. "338" + = USA) consistently across independent real recordings. + + Args: + symbols: 5 raw symbol values (0-99 each) + + Returns: + 9-digit MMSI string, or None if malformed input. + """ + if len(symbols) != 5 or any(s < 0 or s > 99 for s in symbols): return None + digits = "".join(f"{s:02d}" for s in symbols) + return digits[:9] - # Decode the message from symbols - return self._decode_symbols(symbols[: eos_index + 1]) + def _decode_position(self, symbols: list[int]) -> dict | None: + """ + Decode 5 coordinate symbols into a position dict, per ITU-R + M.493 Annex 1 Sec 8.1.2/8.2.3 (Table A1-2), confirmed + consistent across M.493-8 through -15. 10 digits total: + digit 1 = quadrant (0=NE, 1=NW, 2=SE, 3=SW), digits 2-5 = + latitude (2 degrees + 2 minutes), digits 6-10 = longitude + (3 degrees + 2 minutes). All-9s (9999999999) is the spec- + defined "position not available" sentinel. + + Physically-impossible values (degrees/minutes outside valid + range - e.g. from a residual bit error the diversity-combining + fallback in _process_bit didn't catch) are NOT silently + discarded or clamped: the computed position is still returned + (a partial position can still be useful - e.g. for search and + rescue, where even an approximate position beats none at all), + but flagged via "position_uncertain"/"position_issue" so + downstream consumers can decide how much to trust it rather + than being silently given a wrong-but-plausible-looking value. + + Returns a dict with "lat"/"lon" as signed decimal degrees + (matching the pre-existing message["position"] shape this + decoder's downstream consumers - parser.py/routes/dsc.py - + expect) plus human-readable degree/minute text fields and the + uncertainty flag described above. Returns None only if the + position is unavailable (all-9s sentinel) or the input itself + is malformed (wrong length/out-of-range symbols). + + Args: + symbols: 5 raw symbol values (0-99 each) + """ + if len(symbols) != 5 or any(s < 0 or s > 99 for s in symbols): + return None + digits = "".join(f"{s:02d}" for s in symbols) + if digits == "9999999999": + return None + quadrant = int(digits[0]) + lat_deg = int(digits[1:3]) + lat_min = int(digits[3:5]) + lon_deg = int(digits[5:8]) + lon_min = int(digits[8:10]) + + issues = [] + if lat_deg > 90: + issues.append(f"latitude degrees ({lat_deg}) exceeds valid range (0-90)") + if lon_deg > 180: + issues.append(f"longitude degrees ({lon_deg}) exceeds valid range (0-180)") + if lat_min > 59: + issues.append(f"latitude minutes ({lat_min}) exceeds valid range (0-59)") + if lon_min > 59: + issues.append(f"longitude minutes ({lon_min}) exceeds valid range (0-59)") + + lat_sign = 1 if quadrant in (0, 1) else -1 # NE, NW -> North + lon_sign = 1 if quadrant in (0, 2) else -1 # NE, SE -> East + lat = lat_sign * (lat_deg + lat_min / 60.0) + lon = lon_sign * (lon_deg + lon_min / 60.0) + return { + "lat": round(lat, 6), + "lon": round(lon, 6), + "lat_text": f"{lat_deg:02d}°{lat_min:02d}'{'N' if lat_sign > 0 else 'S'}", + "lon_text": f"{lon_deg:03d}°{lon_min:02d}'{'E' if lon_sign > 0 else 'W'}", + "position_uncertain": bool(issues), + "position_issue": "; ".join(issues) if issues else None, + } def _bits_to_symbol(self, bits: list[int]) -> int: """ Convert 10 bits to symbol value. - DSC uses 10-bit symbols: 7 information bits + 3 error bits. - The 3 check bits provide parity such that the total number of - '1' bits across all 10 bits should be even (even parity). - Returns -1 if the check bits are invalid. + Per ITU-R M.493 Annex 1 Sec 2 / Table 1: each DSC character is + a 10-bit code - 7 information bits (bits 1-7, transmitted LSB + first) plus 3 check bits (bits 8-10, transmitted MSB first). + The check bits encode, as a 3-bit binary number, the count of + "B" elements (binary 0) among the 7 information bits - this is + NOT simple parity. E.g. a BYY check-bit sequence (0,1,1) means + the symbol's info bits contain exactly 3 B-elements (spec's own + worked example). + + Returns -1 if the check bits don't match the actual B-element + count in the info bits (character failed the ten-unit + error-detecting code). """ if len(bits) != 10: return -1 - # First 7 bits are data (LSB first in DSC) + info_bits = bits[:7] + check_bits = bits[7:10] + + # Information bits: LSB first value = 0 for i in range(7): - if bits[i]: + if info_bits[i]: value |= 1 << i - # Validate check bits: total number of 1s should be even - ones = sum(bits) - if ones % 2 != 0: + # Check bits: MSB first, encode count of B (0-value) info bits + check_value = (check_bits[0] << 2) | (check_bits[1] << 1) | check_bits[2] + num_b_elements = 7 - sum(info_bits) + + if check_value != num_b_elements: return -1 return value - def _decode_symbols(self, symbols: list[int]) -> dict | None: - """ - Decode DSC symbol sequence to message. - - Message structure (symbols): - 0: Format specifier - 1-5: Address/MMSI (encoded) - 6-10: Self-ID/MMSI (encoded) - 11+: Variable fields depending on format - Last: EOS (127) - - Args: - symbols: List of decoded symbol values - - Returns: - Decoded message dict or None if invalid - """ - if len(symbols) < 12: - return None - - try: - # Format specifier (first non-phasing symbol) - format_code = symbols[0] - format_text = FORMAT_CODES.get(format_code, f"UNKNOWN-{format_code}") - - # Derive category from format specifier per ITU-R M.493 - if format_code == 120: - category = "DISTRESS" - elif format_code == 123: - category = "ALL_SHIPS_URGENCY_SAFETY" - elif format_code == 102: - category = "ALL_SHIPS" - elif format_code == 116: - category = "GROUP" - elif format_code == 112: - category = "INDIVIDUAL" - elif format_code == 114: - category = "INDIVIDUAL_ACK" - else: - category = FORMAT_CODES.get(format_code, "UNKNOWN") - - # Decode MMSI from symbols 1-5 (destination/address) - dest_mmsi = self._decode_mmsi(symbols[1:6]) - if dest_mmsi is None: - return None - - # Decode self-ID from symbols 6-10 (source) - source_mmsi = self._decode_mmsi(symbols[6:11]) - if source_mmsi is None: - return None - - message = { - "type": "dsc", - "format": format_code, - "format_text": format_text, - "category": category, - "source_mmsi": source_mmsi, - "dest_mmsi": dest_mmsi, - "timestamp": datetime.utcnow().isoformat() + "Z", - } - - # Parse additional fields based on format - remaining = symbols[11:-1] # Exclude EOS - - if category in ("DISTRESS", "DISTRESS_RELAY"): - # Distress messages have nature and position - if len(remaining) >= 1: - message["nature"] = remaining[0] - message["nature_text"] = DISTRESS_NATURE_CODES.get(remaining[0], f"UNKNOWN-{remaining[0]}") - - # Try to decode position - if len(remaining) >= 11: - position = self._decode_position(remaining[1:11]) - if position: - message["position"] = position - - # Telecommand fields (last two before EOS) — only for formats - # that carry telecommand fields per ITU-R M.493 - if format_code in TELECOMMAND_FORMATS and len(remaining) >= 2: - message["telecommand1"] = remaining[-2] - message["telecommand2"] = remaining[-1] - - # Add raw data for debugging - message["raw"] = "".join(f"{s:03d}" for s in symbols) - - logger.info(f"Decoded DSC: {category} from {source_mmsi}") - return message - - except Exception as e: - logger.warning(f"DSC decode error: {e}") - return None - - def _decode_mmsi(self, symbols: list[int]) -> str | None: - """ - Decode MMSI from 5 DSC symbols. - - Each symbol represents 2 BCD digits (00-99). - 5 symbols = 10 digits, but MMSI is 9 digits (first symbol has leading 0). - Returns None if any symbol is out of valid BCD range. - """ - if len(symbols) < 5: - return None - - digits = [] - for sym in symbols: - if sym < 0 or sym > 99: - return None - # Each symbol is 2 BCD digits - digits.append(f"{sym:02d}") - - mmsi = "".join(digits) - # MMSI is 9 digits - trim the leading digit from the 10-digit - # BCD result since the first symbol's high digit is always 0 - if len(mmsi) > 9: - mmsi = mmsi[1:] - - return mmsi.zfill(9) - - def _decode_position(self, symbols: list[int]) -> dict | None: - """ - Decode position from 10 DSC symbols. - - Position encoding (ITU-R M.493): - - Quadrant (10=NE, 11=NW, 00=SE, 01=SW) - - Latitude degrees (2 digits) - - Latitude minutes (2 digits) - - Longitude degrees (3 digits) - - Longitude minutes (2 digits) - """ - if len(symbols) < 10: - return None - - try: - # Quadrant indicator - quadrant = symbols[0] - lat_sign = 1 if quadrant in (10, 11) else -1 - lon_sign = 1 if quadrant in (10, 0) else -1 - - # Latitude degrees and minutes - lat_deg = symbols[1] if symbols[1] <= 90 else 0 - lat_min = symbols[2] if symbols[2] < 60 else 0 - - # Longitude degrees (3 digits from 2 symbols) - lon_deg_high = symbols[3] if symbols[3] < 10 else 0 - lon_deg_low = symbols[4] if symbols[4] < 100 else 0 - lon_deg = lon_deg_high * 100 + lon_deg_low - if lon_deg > 180: - lon_deg = 0 - - lon_min = symbols[5] if symbols[5] < 60 else 0 - - lat = lat_sign * (lat_deg + lat_min / 60.0) - lon = lon_sign * (lon_deg + lon_min / 60.0) - - return {"lat": round(lat, 6), "lon": round(lon, 6)} - - except Exception: - return None - - def read_audio_stdin() -> Generator[bytes, None, None]: """ Read audio from stdin in chunks.