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 5b780df..9176bbb 100644 --- a/utils/dsc/decoder.py +++ b/utils/dsc/decoder.py @@ -527,49 +527,6 @@ class DSCDecoder: if eos is None: return None # message not complete yet - def decode_mmsi(syms): - if len(syms) != 5 or any(s < 0 or s > 99 for s in syms): - return None - digits = "".join(f"{s:02d}" for s in syms) - return digits[:9] - - def decode_position(syms): - """ - 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. - - 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. - Returns None if the position is unavailable or malformed. - """ - if len(syms) != 5 or any(s < 0 or s > 99 for s in syms): - return None - digits = "".join(f"{s:02d}" for s in syms) - 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]) - 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'}", - } - message = { "type": "dsc", "format": format_code, @@ -581,11 +538,11 @@ class DSCDecoder: if is_distress_ack: message["format_text"] = "DISTRESS_ACK_OR_SELF_CANCEL" if "address" in fields: - message["dest_mmsi"] = decode_mmsi(fields["address"]) + message["dest_mmsi"] = self._decode_mmsi(fields["address"]) if "self_id" in fields: - message["source_mmsi"] = decode_mmsi(fields["self_id"]) + message["source_mmsi"] = self._decode_mmsi(fields["self_id"]) if "ship_in_distress_id" in fields: - ship_mmsi = decode_mmsi(fields["ship_in_distress_id"]) + 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 @@ -603,7 +560,7 @@ class DSCDecoder: message["nature_text"] = NATURE_TEXT.get(nature_val, f"UNKNOWN-{nature_val}") if "coordinates" in fields: message["coordinates_raw"] = fields["coordinates"] - message["position"] = decode_position(fields["coordinates"]) + message["position"] = self._decode_position(fields["coordinates"]) if "time" in fields: message["time_raw"] = fields["time"] if "telecommand" in fields: @@ -718,143 +675,95 @@ class DSCDecoder: return (f"format={format_code}, all fields complete, but no valid EOS " f"found in remaining {len(dedup) - idx} symbols") - def _diagnose_decode_failure(self) -> str: + def _decode_mmsi(self, symbols: list[int]) -> str | None: """ - 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). + 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. """ - 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)], + 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] + + 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, } - 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 _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 _bits_to_symbol(self, bits: list[int]) -> int: """