Stream NMEA message with metadata

[MyrnsWork]

It may be useful to read or stream NMEA messages containing any kind of metadata like this:

# We can also have any king of metadata for each message:
enhanced_fake_stream = [
    b"[2024-07-19 08:45:27.141] !AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23",
    b"[2024-07-19 08:45:30.074] !AIVDM,1,1,,A,133sVfPP00PD>hRMDH@jNOvN20S8,0*7F",
    b"[2024-07-19 08:45:35.007] !AIVDM,1,1,,B,100h00PP0@PHFV`Mg5gTH?vNPUIp,0*3B",
    b"[2024-07-19 08:45:35.301] !AIVDM,1,1,,B,13eaJF0P00Qd388Eew6aagvH85Ip,0*45",
    b"[2024-07-19 08:45:40.021] !AIVDM,1,1,,A,14eGrSPP00ncMJTO5C6aBwvP2D0?,0*7A",
    b"[2024-07-19 08:45:40.074] !AIVDM,1,1,,A,15MrVH0000KH<:V:NtBLoqFP2H9:,0*2F",
]

# Create a custom parsing function:
# - NMEA message must be always in the first position in bytes
# - For flexibility purposes, consider cases where the data can be bytes or string
def parse_function(msg: Union[str, bytes], encoding: str = 'utf-8') -> Tuple[bytes, str]:
    if isinstance(msg, bytes):
        msg = msg.decode(encoding)
    nmea_message = re.search(".* (.*)", msg).group(1)  # NMEA
    metadata = re.search("(.*) .*", msg).group(1)  # Metadata

    return bytes(nmea_message, encoding), metadata

for message, infos in IterMessages(enhanced_fake_stream, parse_function):
    print(infos, message.decode())

And the result would be:

[2024-07-19 08:45:27.141] MessageType1(msg_type=1, repeat=0, mmsi=227006760, status=<NavigationStatus.UnderWayUsingEngine: 0>, turn=<TurnRate.NO_TI_DEFAULT: -128.0>, speed=0.0, accuracy=False, lon=0.13138, lat=49.475577, course=36.7, heading=511, second=14, maneuver=<ManeuverIndicator.NotAvailable: 0>, spare_1=b'\x00', raim=False, radio=22136)
[2024-07-19 08:45:30.074] MessageType1(msg_type=1, repeat=0, mmsi=205448890, status=<NavigationStatus.UnderWayUsingEngine: 0>, turn=<TurnRate.NO_TI_DEFAULT: -128.0>, speed=0.0, accuracy=True, lon=4.419442, lat=51.237658, course=63.3, heading=511, second=15, maneuver=<ManeuverIndicator.NotAvailable: 0>, spare_1=b'\x00', raim=True, radio=2248)
[2024-07-19 08:45:35.007] MessageType1(msg_type=1, repeat=0, mmsi=786434, status=<NavigationStatus.UnderWayUsingEngine: 0>, turn=<TurnRate.NO_TI_DEFAULT: -128.0>, speed=1.6, accuracy=True, lon=5.320033, lat=51.967037, course=112.0, heading=511, second=15, maneuver=<ManeuverIndicator.NoSpecialManeuver: 1>, spare_1=b'\x00', raim=False, radio=153208)
[2024-07-19 08:45:35.301] MessageType1(msg_type=1, repeat=0, mmsi=249191000, status=<NavigationStatus.UnderWayUsingEngine: 0>, turn=<TurnRate.NO_TI_DEFAULT: -128.0>, speed=0.0, accuracy=True, lon=23.603633, lat=37.955883, course=247.0, heading=511, second=12, maneuver=<ManeuverIndicator.NotAvailable: 0>, spare_1=b'@', raim=False, radio=22136)
[2024-07-19 08:45:40.021] MessageType1(msg_type=1, repeat=0, mmsi=316013198, status=<NavigationStatus.UnderWayUsingEngine: 0>, turn=<TurnRate.NO_TI_DEFAULT: -128.0>, speed=0.0, accuracy=True, lon=-130.316237, lat=54.32111, course=237.9, heading=511, second=16, maneuver=<ManeuverIndicator.NotAvailable: 0>, spare_1=b'\x00', raim=True, radio=81935)
[2024-07-19 08:45:40.074] MessageType1(msg_type=1, repeat=0, mmsi=366913120, status=<NavigationStatus.UnderWayUsingEngine: 0>, turn=0.0, speed=0.0, accuracy=False, lon=-64.620662, lat=18.321188, course=329.5, heading=299, second=16, maneuver=<ManeuverIndicator.NotAvailable: 0>, spare_1=b'\x00', raim=True, radio=98890)

I've already coded the functionality (by keeping the current behaviour of IterMessages (or any class that has inherited from AssembleMessages) with 1 argument) on a local branch if you're interested (I'm not sure of the quality of my code though).

[M0r13n]

@MyrnsWork If I understand your proposal correctly, you are suggesting an extension of the streaming API of pyais to allow a custom pre-process function for handling different kinds of formats of metadata.

That sounds like a great idea. However, special care needs to be taken to avoid breaking backward compatibility. I will think about this when time allows. Feel free to open a PR with your changes so that I can take a look and possibly draw some inspiration.

Have a great day!

[MyrnsWork]

If I understand your proposal correctly, you are suggesting an extension of the streaming API of pyais to allow a custom pre-process function for handling different kinds of formats of metadata

@M0r13n You're right !

Before opening a PR, a word about the tests. As far as I'm concerned, all the tests passed except:

• test_decode_from_file
• test_parser
• test_invalid_endpoint: because a TCP connection is required?
• test_run_every_file (logic here)

If I change the path ‘tests/ais_test_messages’ to ‘ais_test_messages’ in test_decode_from_file and test_parser then :

• test_decode_from_file is now ok
• test_parser: I get an error ‘[Errno 2] No such file or directory : “/tmp/foo.bar”’ during parsing.

Anyways, when I git checkout on master (without any modifications), the same tests don't pass. Is this normal for these tests?

[M0r13n]

Hey @MyrnsWork,

Thank you for the PR; I appreciate your effort. I took some inspiration from you but ultimately settled on a slightly different approach that is more robust and flexible. All instances implementing the streaming API of pyais now optionally accept the keyword preprocessor. This must be an instance of a class implementing the PreprocessorProtocol protocol, which essentially requires the implementation of a process method that accepts some bytes and returns some bytes in NMEA 0183 format. The user is free to add additional logic in this class. I added an example that provides access to the metadata.

Please note that IterMessages() does not accept the preprocessor keyword, as this class requires the messages to be already in memory, meaning it should be trivial to parse the custom format.

Feel free to look at my changes and provide your feedback. You can expect the changes to be available in the coming days.

Have a great day!

[MyrnsWork]

Indeed, you're implementation is better !

[M0r13n]

@MyrnsWork Thank you. So I am closing this now. :)