feat: add NMEAQueue to support flexible queueing (#166)

* feat: add NMEAQueue to support flexible queueing

* chore: renames examples/tee.py to examples/queue.py

* chore: renames examples/tee.py to examples/nmea_queue.py

* chore: adds docs, bumps version, updates changelog

Author: M0r13n

CHANGELOG.txt

@@ -1,6 +1,14 @@
 ====================
  pyais CHANGELOG
 ====================
+-------------------------------------------------------------------------------
+ Version 2.9.0 23 Feb 2025
+-------------------------------------------------------------------------------
+* added NMEAQueue class
+  * supports both single-line and multi-line NMEA sentences
+  * buffers fragments until all parts are available
+  * handles tag blocks
+  * supports gatehouse wrappers
 -------------------------------------------------------------------------------
  Version 2.8.4 26 Jan 2025
 -------------------------------------------------------------------------------

docs/index.rst

@@ -14,3 +14,4 @@ Welcome to Pyais's documentation!
    examples
    messages
    filters
+   queue

docs/queue.rst

@@ -0,0 +1,54 @@
+NMEAQueue: Assembling Complete NMEA Sentences
+=============================================
+
+The ``NMEAQueue`` class provides a robust mechanism for assembling and managing NMEA sentences, including both single-line and multi-line messages. It is designed to handle AIS messages and integrates with ``TagBlockQueue`` for processing tag blocks.
+
+Features
+--------
+- **Single-Line Sentence Handling**: Single-line NMEA sentences are added directly to the queue.
+- **Multi-Line Sentence Assembly**: Multi-line sentences are buffered until all fragments are available, after which they are assembled and added to the queue.
+- **Gatehouse Wrappers**: Supports gatehouse wrappers for AIS messages, associating them with the next AIS message in the sequence.
+- **Graceful Error Handling**: Invalid or malformed sentences are skipped without interrupting the processing flow.
+- **Integration with TagBlockQueue**: If a ``TagBlockQueue`` instance is provided, sentences are added to it for tag block processing.
+
+Constructor
+-----------
+.. code-block:: python
+
+    NMEAQueue(maxsize: int = 0, tbq: typing.Optional[TagBlockQueue] = None)
+
+- ``maxsize``: The maximum size of the queue. Defaults to 0 (unlimited).
+- ``tbq``: An optional ``TagBlockQueue`` instance for handling tag blocks.
+
+Methods
+-------
+- ``put_line(line: bytes, block: bool = True, timeout: typing.Optional[float] = None)``: Adds a line of raw bytes to the queue. This method processes the line, handles multi-line assembly, and integrates with ``TagBlockQueue`` if provided.
+- ``get_or_none() -> typing.Optional[NMEASentence]``: Retrieves the last message from the queue in a non-blocking manner. Returns ``None`` if the queue is empty.
+
+Example Usage
+-------------
+.. code-block:: python
+
+    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)
+
+Notes
+-----
+- **Error Handling**: The ``put_line`` method skips invalid messages (e.g., malformed sentences or those with non-printable characters) without raising exceptions.
+- **Buffering**: Multi-line sentences are buffered using a unique key based on the sequence ID and channel. Once all fragments are received, the sentence is assembled and added to the queue.
+
+This class simplifies the process of handling NMEA sentences, making it easier to work with AIS data streams in real-time applications.

examples/nmea_queue.py

@@ -0,0 +1,40 @@
+import pathlib
+from pyais.queue import NMEAQueue
+
+
+if __name__ == '__main__':
+
+    # Example 1: use a queue to read NMEA/AIS messages from a file
+    filename = pathlib.Path(__file__).parent.joinpath('../tests/mixed.txt')
+
+    q = NMEAQueue()
+
+    with open(filename, 'rb') as fd:
+        for line in fd.readlines():
+            q.put_line(line)
+
+            if x := q.get_or_none():
+                print(x.decode())
+            else:
+                print(line)
+
+    # Example 2: put lines into the queue manually
+
+    q = NMEAQueue()
+    q.qsize()  # Initially empty
+
+    # Raw text.
+    q.put_line(b'Hello there!')
+    q.qsize()  # Still empty
+
+    # Put a multi-line message into the queue
+    q.put_line(b'!AIVDM,2,1,1,A,55?MbV02;H;s<HtKR20EHE:0@T4@Dn2222222216L961O5Gf0NSQEp6ClRp8,0*1C')
+    q.put_line(b'!AIVDM,2,2,1,A,88888888880,2*25')
+    q.qsize()  # Returns 1
+    q.get_or_none()
+    q.qsize()  # Empty again
+
+    # A multi-line message with tag blocks
+    q.put_line(b'\\g:1-2-73874*A\\!AIVDM,1,1,,A,15MrVH0000KH<:V:NtBLoqFP2H9:,0*2F')
+    q.put_line(b'\\g:2-2-73874,n:157037*A\\!AIVDM,1,1,,B,100h00PP0@PHFV`Mg5gTH?vNPUIp,0*3B')
+    q.get_or_none()

pyais/__init__.py

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

pyais/queue.py

@@ -0,0 +1,86 @@
+
+import queue
+import typing
+
+from pyais.exceptions import InvalidNMEAMessageException, NonPrintableCharacterException, UnknownMessageException
+from pyais.messages import AISSentence, GatehouseSentence, NMEASentence, NMEASentenceFactory
+from pyais.stream import TagBlockQueue
+
+
+class NMEAQueue(queue.Queue[AISSentence]):
+    """Assembles complete NMEA sentences.
+
+    Single-line sentences are added to the queue directly. Multi-line sentences
+    are buffered until all fragments are available, after which they are added to the queue."""
+
+    def __init__(self, maxsize: int = 0, tbq: typing.Optional[TagBlockQueue] = None) -> None:
+        super().__init__(maxsize)
+        self.tbq = tbq
+        self.buffer: typing.Dict[typing.Tuple[int, str], typing.List[typing.Optional[AISSentence]]] = {}
+        self.last_wrapper: typing.Optional[GatehouseSentence] = None
+
+    def __add_to_tbq(self, sentence: NMEASentence) -> None:
+        if not self.tbq:
+            # Tag Block Queue not defined. Do nothing.
+            return
+        self.tbq.put_sentence(sentence)
+
+    def put(self, item: object, block: bool = True, timeout: typing.Optional[float] = None) -> None:
+        """This method only exists to please mypy. Use put_line instead."""
+        raise ValueError('do not call NMEAQueue.put() directly. Use NMEAQueue.put_line() instead!')
+
+    def put_line(self, line: bytes, block: bool = True, timeout: typing.Optional[float] = None) -> None:
+        """Put a line of raw bytes, as part of an NMEA sentence, into the queue."""
+        try:
+            sentence = NMEASentenceFactory.produce(line)
+            self.__add_to_tbq(sentence)
+            if sentence.TYPE == GatehouseSentence.TYPE:
+                # Remember gatehouse wrappers for the next AIS message
+                sentence = typing.cast(GatehouseSentence, sentence)
+                self.last_wrapper = sentence
+                return None
+        except (InvalidNMEAMessageException, NonPrintableCharacterException, UnknownMessageException, IndexError):
+            # Be gentle and just skip invalid messages
+            return None
+
+        if not sentence.TYPE == AISSentence.TYPE:
+            return None
+
+        sentence = typing.cast(AISSentence, sentence)
+
+        if sentence.is_single:
+            if self.last_wrapper:
+                # Check if there was a wrapper message right before this line
+                sentence.wrapper_msg = self.last_wrapper
+                self.last_wrapper = None
+            super().put(sentence, block, timeout)
+        else:
+            # Instead of None use -1 as a seq_id
+            seq_id = sentence.seq_id
+            if seq_id is None:
+                seq_id = -1
+
+            # seq_id and channel make a unique stream
+            slot = (seq_id, sentence.channel)
+
+            if slot not in self.buffer:
+                # Create a new array in the buffer that has enough space for all fragments
+                self.buffer[slot] = [None, ] * max(sentence.fragment_count, 0xff)
+
+            self.buffer[slot][sentence.frag_num - 1] = sentence
+            msg_parts = self.buffer[slot][0:sentence.fragment_count]
+
+            # Check if all fragments are found
+            not_none_parts = [m for m in msg_parts if m is not None]
+            if len(not_none_parts) == sentence.fragment_count:
+                # Assemble the full message and clear the buffer
+                full = AISSentence.assemble_from_iterable(not_none_parts)
+                del self.buffer[slot]
+                super().put(full, block, timeout)
+
+    def get_or_none(self) -> typing.Optional[NMEASentence]:
+        """Non-blocking helper method to retrieve the last message, if one is available"""
+        try:
+            return self.get(block=False)
+        except queue.Empty:
+            return None

tests/mixed.txt

@@ -0,0 +1,35 @@
+# taken from https://www.aishub.net/ais-dispatcher
+!AIVDM,1,1,,A,13HOI:0P0000VOHLCnHQKwvL05Ip,0*23
+!AIVDM,1,1,,A,133sVfPP00PD>hRMDH@jNOvN20S8,0*7F
+!AIVDM,1,1,,B,100h00PP0@PHFV`Mg5gTH?vNPUIp,0*3B
+!AIVDM,1,1,,B,13eaJF0P00Qd388Eew6aagvH85Ip,0*45
+
+$GPGGA,184353.07,1929.045,S,02410.506,E,1,04,2.6,100.00,M,-33.9,M,,0000*6D
+$GPRTE,2,1,c,0,PBRCPK,PBRTO,PTELGR,PPLAND,PYAMBU,PPFAIR,PWARRN,PMORTL,PLISMR*73
+
+!AIVDM,1,1,,A,14eGrSPP00ncMJTO5C6aBwvP2D0?,0*7A
+!AIVDM,1,1,,A,15MrVH0000KH<:V:NtBLoqFP2H9:,0*2F
+
+$GPR00,A,B,C*29
+foobar
+!AIVDM,2,1,1,A,55?MbV02;H;s<HtKR20EHE:0@T4@Dn2222222216L961O5Gf0NSQEp6ClRp8,0*1C
+!AIVDM,2,2,1,A,88888888880,2*25
+
+$IIMWV,271.0,R,000.2,N,A*3B
+# Lines with a leading `#` are ignored
+# Also invalid lines or invalid messages are ignored
+IamNotAnAisMessage1111
+# Tag blocks are also supported
+\g:1-2-73874,n:157036,s:r003669945,c:12415440354*A\!AIVDM,1,1,,B,15N4cJ005Jrek0H@9nDW5608EP,013
+\g:1-2-27300,n:636994,s:b003669710,c:1428621738*5F\!SAVDM,2,1,2,B,55Mw@A7J1adAL@?;7WPl58F0U<h4pB222222220t1PN5553fN4g?`4iSp5Rc,0*26
+\g:2-2-27300,n:636995*15\!SAVDM,2,2,2,B,iP`88888880,2*5E
+\g:1-2-73874*A\!AIVDM,1,1,,A,15MrVH0000KH<:V:NtBLoqFP2H9:,0*2F
+\g:2-2-73874,n:157037*A\!AIVDM,1,1,,B,100h00PP0@PHFV`Mg5gTH?vNPUIp,0*3B
+
+$PGHP,1,2008,5,9,0,0,0,10,338,2,,1,09*17
+!AIVDM,1,1,,B,15NBj>PP1gG>1PVKTDTUJOv00<0M,0*09
+
+# More comments
+
+!AIVDM,1,1,,A,16:=?;0P00`SstvFnFbeGH6L088h,0*44
+$PGHP,1,2009,5,9,0,0,0,10,338,2,,1,09*17

tests/test_examples.py

@@ -33,4 +33,4 @@ def test_run_every_file(self):
         if csv_file.exists():
             csv_file.unlink()
 
-        assert i == 23
+        assert i == 24

tests/test_queue.py

@@ -0,0 +1,81 @@
+import pathlib
+import unittest
+
+from pyais.queue import NMEAQueue
+from pyais.stream import FileReaderStream
+
+
+class QueueTestCase(unittest.TestCase):
+
+    def test_against_file_reader_stream(self):
+        # HAVING a file with all sorts of NMEA messages as well as other kinds of messages
+        filename = pathlib.Path(__file__).parent.joinpath('mixed.txt')
+
+        # WHEN streaming NMEA/AIS messages from this files using FileReaderStream
+        expected = []
+        with FileReaderStream(filename) as stream:
+            for msg in stream:
+                expected.append(msg)
+
+        # WHEN reading the same file putting the lines into a NMEAQueue
+        q = NMEAQueue()
+        actual = []
+        with open(filename, 'rb') as fd:
+            for line in fd.readlines():
+                q.put_line(line)
+                if x := q.get_or_none():
+                    actual.append(x)
+
+        # THEN both methods of reading a file return the exact same result
+        self.assertEqual(len(expected), len(actual), 'expected list differs from actual list')
+        self.assertEqual(expected, actual, 'expected list differs from actual list')
+
+        for a, b, in zip(expected, actual):
+            self.assertEqual(a, b)
+            self.assertEqual(a.wrapper_msg, b.wrapper_msg)
+            if a.tag_block or b.tag_block:
+                a.tag_block.init()
+                b.tag_block.init()
+                self.assertEqual(a.tag_block.text, b.tag_block.text)
+
+    def test_that_put_raises_value_error(self):
+        q = NMEAQueue()
+
+        with self.assertRaises(ValueError):
+            q.put(b"line")
+
+    def test_manually(self):
+
+        q = NMEAQueue()
+        assert q.qsize() == 0  # Initially empty
+
+        # Raw text.
+        q.put_line(b'Hello there!')
+        assert q.qsize() == 0  # Still empty
+        assert q.get_or_none() is None
+
+        # Put a multi-line message into the queue
+        q.put_line(b'!AIVDM,2,1,1,A,55?MbV02;H;s<HtKR20EHE:0@T4@Dn2222222216L961O5Gf0NSQEp6ClRp8,0*1C')
+        assert q.qsize() == 0  # Still empty
+        q.put_line(b'!AIVDM,2,2,1,A,88888888880,2*25')
+        assert q.qsize() == 1  # Returns 1
+        assert q.get_or_none() is not None
+        assert q.qsize() == 0  # Empty again
+
+        # A multi-line message with tag blocks
+        q.put_line(b'\\g:1-2-73874*A\\!AIVDM,1,1,,A,15MrVH0000KH<:V:NtBLoqFP2H9:,0*2F')
+        q.put_line(b'\\g:2-2-73874,n:157037*A\\!AIVDM,1,1,,B,100h00PP0@PHFV`Mg5gTH?vNPUIp,0*3B')
+        assert q.qsize() == 2
+        assert q.get_or_none() is not None
+        assert q.qsize() == 1
+        assert q.get_or_none() is not None
+        assert q.qsize() == 0
+
+        q.put_line(b'!AIVDM,1,1,,A,169a:nP01g`hm4pB7:E0;@0L088i,0*5E')
+        q.put_line(b'!AIVDM,1,1,,A,169a:nP01g`hm4pB7:E0;@0L088i,0*5E')
+        q.put_line(b'!AIVDM,1,1,,A,169a:nP01g`hm4pB7:E0;@0L088i,0*5E')
+        assert q.qsize() == 3
+
+
+if __name__ == '__main__':
+    unittest.main()