chore: auxiliary craft now decode mothership MMSI instead of vessel dimensions (#196) (#197)

Author: M0r13n

CHANGELOG.txt

@@ -1,6 +1,12 @@
 ====================
  pyais CHANGELOG
 ====================
+-------------------------------------------------------------------------------
+ Version 2.14.0 07 Dec 2025
+-------------------------------------------------------------------------------
+* closes: https://github.com/M0r13n/pyais/issues/196
+* add support for AIS Message Type 24 Part B auxiliary craft variant
+* auxiliary craft now decode mothership MMSI instead of vessel dimensions
 -------------------------------------------------------------------------------
  Version 2.13.3 15 Nov 2025
 -------------------------------------------------------------------------------

pyais/__init__.py

@@ -5,7 +5,7 @@
 from pyais.tracker import AISTracker, AISTrack
 
 __license__ = 'MIT'
-__version__ = '2.13.3'
+__version__ = '2.14.0'
 __author__ = 'Leon Morten Richter'
 
 __all__ = (

pyais/messages.py

@@ -15,7 +15,7 @@
     InvalidDataTypeException, MissingPayloadException
 from pyais.util import checksum, decode_into_bit_array, compute_checksum, get_itdma_comm_state, get_sotdma_comm_state, int_to_bin, str_to_bin, \
     encode_ascii_6, from_bytes, from_bytes_signed, decode_bin_as_ascii6, get_int, chk_to_int, coerce_val, \
-    bits2bytes, bytes2bits, b64encode_str
+    bits2bytes, bytes2bits, b64encode_str, is_auxiliary_craft
 
 NMEA_VALUE = typing.Union[str, float, int, bool, bytes]
 
@@ -1701,6 +1701,32 @@ class MessageType24PartB(Payload):
     spare_1 = bit_field(6, bytes, default=b'', is_spare=True)
 
 
+@attr.s(slots=True)
+class MessageType24PartBAuxiliaryCraft(Payload):
+    """
+    Static Data Report - Part B (Auxiliary Craft Variant)
+
+    When the MMSI follows the pattern 98XXXYYYY (auxiliary craft),
+    bits 132-161 contain the mothership MMSI instead of vessel dimensions.
+
+    See ITU-R M.1371-5 for specification details.
+    """
+    msg_type = bit_field(6, int, default=24, signed=False)
+    repeat = bit_field(2, int, default=0, signed=False)
+    mmsi = bit_field(30, int, from_converter=from_mmsi)
+
+    partno = bit_field(2, int, default=0, signed=False)
+    ship_type = bit_field(8, int, default=0, signed=False)
+    vendorid = bit_field(18, str, default='', signed=False)
+    model = bit_field(4, int, default=0, signed=False)
+    serial = bit_field(20, int, default=0, signed=False)
+    callsign = bit_field(42, str, default='')
+
+    mothership_mmsi = bit_field(30, int, from_converter=from_mmsi)
+
+    spare_1 = bit_field(6, bytes, default=b'', is_spare=True)
+
+
 class MessageType24(Payload):
     """
     Static Data Report
@@ -1714,20 +1740,26 @@ class MessageType24(Payload):
 
     @classmethod
     def create(cls, **kwargs: typing.Union[str, float, int, bool, bytes]) -> "ANY_MESSAGE":
+        mmsi: int = int(kwargs.get('mmsi', 0))
         partno: int = int(kwargs.get('partno', 0))
         if partno == 0:
             return MessageType24PartA.create(**kwargs)
         elif partno == 1:
+            if is_auxiliary_craft(mmsi):
+                return MessageType24PartBAuxiliaryCraft.create(**kwargs)
             return MessageType24PartB.create(**kwargs)
         else:
             raise UnknownPartNoException(f"Partno {partno} is not allowed!")
 
     @classmethod
     def from_bitarray(cls, bit_arr: bitarray) -> "ANY_MESSAGE":
+        mmsi: int = get_int(bit_arr, 8, 38)
         partno: int = get_int(bit_arr, 38, 40)
         if partno == 0:
             return MessageType24PartA.from_bitarray(bit_arr)
         elif partno == 1:
+            if is_auxiliary_craft(mmsi):
+                return MessageType24PartBAuxiliaryCraft.from_bitarray(bit_arr)
             return MessageType24PartB.from_bitarray(bit_arr)
         else:
             raise UnknownPartNoException(f"Partno {partno} is not allowed!")

pyais/util.py

@@ -430,3 +430,11 @@ def get_first_three_digits(num: int) -> int:
 
 def get_country(mmsi: int) -> typing.Tuple[str, str]:
     return COUNTRY_MAPPING.get(get_first_three_digits(mmsi), ('NA', 'Unknown'))
+
+
+def is_auxiliary_craft(mmsi: int) -> bool:
+    # An auxiliary craft has a mmsi of the form 98XXXYYYY, where:
+    #   1. it starts with 98
+    #   2. is followed by a MID (XXX)
+    #   3. if followed by any decimal literal YYYY (0000 - 9999)
+    return 98_000_0000 <= mmsi <= 98_999_9999

tests/test_decode.py

@@ -52,7 +52,7 @@
     MessageType26BroadcastUnstructured,
 )
 from pyais.stream import ByteStream, IterMessages
-from pyais.util import b64encode_str, bits2bytes, bytes2bits, decode_into_bit_array
+from pyais.util import b64encode_str, bits2bytes, bytes2bits, decode_into_bit_array, is_auxiliary_craft
 from pyais.exceptions import MissingPayloadException
 
 
@@ -763,6 +763,49 @@ def test_msg_type_24_with_168_bits(self):
         assert msg["mmsi"] == 123456789
         assert msg["spare_1"] == b"\x00"
 
+    def test_msg_type_24_b_regular(self):
+        msg = decode(b"!AIVDO,1,1,,A,H8=;nnT000000000000000Wg8Jb0,0*26").asdict(ignore_spare=False)
+
+        assert msg["msg_type"] == 24
+        assert msg["partno"] == 1
+        assert msg["mmsi"] == 550696666
+        assert msg["to_bow"] == 317
+        assert msg["to_stern"] == 456
+        assert msg["to_port"] == 26
+        assert msg["to_starboard"] == 42
+        assert 'mothership_mmsi' not in msg
+
+    def test_msg_type_24_b_aux_craft(self):
+        msg = decode(b"!AIVDO,1,1,,A,H>W@vFTe6??406t2??21J0Wg8Jb0,0*6F").asdict(ignore_spare=False)
+
+        assert msg["msg_type"] == 24
+        assert msg["partno"] == 1
+        assert msg["mmsi"] == 980696666
+        assert msg["ship_type"] == 45
+        assert msg["vendorid"] == 'FOO'
+        assert msg["model"] == 1
+        assert msg["callsign"] == 'BOOBAZ'
+
+        assert 'to_bow' not in msg
+        assert 'to_stern' not in msg
+        assert 'to_port' not in msg
+        assert 'to_starboard' not in msg
+
+        assert msg["mothership_mmsi"] == 666666666
+
+    def test_is_auxiliary_craft_boundary_cases(self):
+        # Test lower boundary
+        assert is_auxiliary_craft(980000000)
+
+        # Test upper boundary
+        assert is_auxiliary_craft(989999999)
+
+        # Test just below range (should use regular PartB)
+        assert not is_auxiliary_craft(979999999)
+
+        # Test just above range (should use regular PartB)
+        assert not is_auxiliary_craft(990000000)
+
     def test_msg_type_25_a(self):
         msg = decode(b"!AIVDM,1,1,,A,I6SWo?8P00a3PKpEKEVj0?vNP<65,0*73").asdict(ignore_spare=False)