It may be useful to read or stream NMEA messages containing any type of metadata
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
# - Always consider that the NMEA message are bytes when parsing
# - The metadata field can be also parsed during the process (any types: string, float, datetime, etc.)
def parse_function(line: bytes) -> Tuple[bytes, Any]:
nmea_message = re.search(b'.* (.*)', line).group(1) # NMEA
metadata_bytes = re.search(b'(.*) .*', line).group(1) # Metadata
timestamp = datetime.strptime(metadata_bytes.decode("utf-8"), "[%Y-%m-%d %X.%f]").timestamp()
return nmea_message, timestamp
# Whatever if enhanced_fake_stream datas are bytes or strings:
for message, infos in IterMessages(enhanced_fake_stream, parse_function):
print(f"Timestamp: {infos} --", message.decode())
The retro-compatibility is done by a inputs/outputs mapper function:
io_mapper_function = (lambda msg, infos: msg) if parse_function is None else (lambda msg, infos: (msg, infos))
It's the biggest weakness of this new feature: we break the consistency of the output data structure.
However, we could imagine for a future version of pyais, with a breaking change (3.y.z), to change the global output to (message, metadata) where metadata can be None.
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?
It may be useful to read or stream NMEA messages containing any type of metadata
And the result would be:
The retro-compatibility is done by a inputs/outputs mapper function:
It's the biggest weakness of this new feature: we break the consistency of the output data structure. However, we could imagine for a future version of pyais, with a breaking change (3.y.z), to change the global output to (message, metadata) where metadata can be None.
As far as I'm concerned, all the tests passed except:
If I change the path ‘tests/ais_test_messages’ to ‘ais_test_messages’ in test_decode_from_file and test_parser then :
Anyways, when I git checkout on master (without any modifications), the same tests don't pass. Is this normal for these tests?