pyais: AIS Message Decoding

The AIS filtering system uses a modular approach where individual filter classes evaluate messages based on specific criteria. These filters are orchestrated by a FilterChain object, which processes an AIS message stream sequentially.

The Workflow:

1. AIS Stream: A stream of AIS messages is provided to the FilterChain.
2. Sequential Application: The FilterChain passes messages through each filter in the order they were provided during initialization.
3. Passing Criteria: Each filter evaluates the message; if the message meets the filter's criteria, it is passed to the next filter in the chain. If it fails, it is dropped from the stream.

This allows you to build complex logic by combining simple, specialized filters.

from pyais import FilterChain
from pyais.filters import AttributeFilter, MessageTypeFilter

# Create a chain that first checks a custom attribute, then checks message type
chain = FilterChain([
    AttributeFilter(lambda x: hasattr(x, 'turn')),
    MessageTypeFilter(1, 2, 3)
])

# Use the chain to filter a stream
filtered_data = chain.filter(stream)

The NMEAQueue class is used to manage and assemble NMEA sentences from raw byte streams. It handles both single-line sentences and multi-line messages by buffering fragments until a complete message is formed. It can also integrate with a TagBlockQueue to facilitate tag block processing.

Key behaviors:

• Multi-line Assembly: Buffers fragments based on sequence ID and channel, assembling them once all parts are present.
• Error Resilience: Malformed or invalid sentences are automatically skipped without raising exceptions.
• Tag Block Integration: If a TagBlockQueue is provided during initialization, sentences are passed to it for processing.

from pyais.stream import TagBlockQueue
from my_module import NMEAQueue

# Initialize a TagBlockQueue
tbq = TagBlockQueue()

# Create an NMEAQueue instance
queue = NMEAQueue(tbq=tbq)

# Add a line of raw bytes to the queue
queue.put_line(b"!AIVDM,1,1,,A,15N:;R0P00PD;88MD5NS8v2P00,0*3C")

# Retrieve the assembled sentence
sentence = queue.get_or_none()
if sentence:
    print(sentence)

Tag blocks are optional key-value pairs wrapped between \s prefixes in NMEA sentences, used to hold extra information. In pyais, tag blocks are lazily decoded to save resources. To access the data, you must check if msg.tag_block is not None and explicitly call msg.tag_block.init() before accessing its contents via asdict().

Supported tag block fields include:

• c: Receiver timestamp in Unix epoch
• d: Destination station
• n: Line count
• r: Relative time
• s: Source station
• t: Text

Unknown fields are omitted.

from pyais.stream import IterMessages

text = """
\s:2573135,c:1671620143*0B\!AIVDM,1,1,,A,16:=?;0P00`SstvFnFbeGH6L088h,0*44
"""

messages = [line.encode() for line in text.split() if line]

with IterMessages(messages) as s:
    for msg in s:
        if msg.tag_block is not None:
            # Must call .init() because tag blocks are lazily parsed
            msg.tag_block.init()
            print(msg.tag_block.asdict())
        print(msg.decode())

The filtering system allows you to chain multiple criteria to process an AIS stream. You create individual filter objects and pass them as a list to a FilterChain.

Available filters:

• AttributeFilter(func): Keeps messages where func(msg) is True.
• NoneFilter(*attr_names): Filters out messages where the specified attributes are None.
• MessageTypeFilter(*types): Keeps only specified message types.
• DistanceFilter(ref_point, distance_km): Keeps messages within a certain distance (km) from a (lat, lon) tuple.
• GridFilter(lat_min, lon_min, lat_max, lon_max): Keeps messages within a geographical bounding box.

Use chain.filter(stream) to apply the filters to an AIS stream.

from pyais import TCPConnection
from pyais.filters import AttributeFilter, NoneFilter, MessageTypeFilter, DistanceFilter, GridFilter, FilterChain

# Define filters
attribute_filter = AttributeFilter(lambda x: not hasattr(x, 'turn') or x.turn == -128.0)
none_filter = NoneFilter('lon', 'lat', 'mmsi2')
message_type_filter = MessageTypeFilter(1, 2, 3)
distance_filter = DistanceFilter((51.900, 5.320), distance_km=1000)
grid_filter = GridFilter(lat_min=50, lon_min=0, lat_max=52, lon_max=5)

# Create and apply the chain
chain = FilterChain([
    attribute_filter,
    none_filter,
    message_type_filter,
    distance_filter,
    grid_filter,
])

stream = TCPConnection(host='...', port=...)
filtered_data = list(chain.filter(stream))

for msg in filtered_data:
    print(msg.lat, msg.lon)