Added async support and test cases.

dalymople · dalymople · commit 0bec30dabfbb · 2026-01-29T05:09:12.000+11:00
diff --git a/README.md b/README.md
@@ -67,4 +67,3 @@ Nothing in this package is very complicated, please have a crack and help me to
 - Add the ability to pack messages into packets for two way communications
 - Add more and better tests
 - Add Field type RU1_3
-- Add async support
diff --git a/ubxtranslator/core.py b/ubxtranslator/core.py
@@ -392,13 +392,6 @@ def register_cls(self, cls: Cls):
         """Register a message  class."""
         self.classes[cls.id_] = cls
 
-    async def async_receive_from(self, stream) -> namedtuple:
-        """Receive a message from a stream and return as a namedtuple.
-        raise IOError in case of errors due to insufficient data.
-        raise ValueError in case of errors due to sufficient but invalid data.
-        """
-        pass
-
     def receive_from(self, stream) -> namedtuple:
         """Receive a message from a stream and return as a namedtuple.
         raise IOError in case of errors due to insufficient data.
@@ -448,6 +441,44 @@ def receive_from(self, stream) -> namedtuple:
 
         return self.classes[msg_cls].parse(msg_id, buff[4:])
 
+    async def receive_from_async(self, stream) -> namedtuple:
+        """Async version of receive_from."""
+        while True:
+            # Search for the prefix
+            buff = await self._read_until_async(stream, terminator=Parser.PREFIX)
+            if buff[-2:] == Parser.PREFIX:
+                break
+
+        # read the first four bytes
+        buff = await stream.readexactly(4)
+
+        if len(buff) != 4:
+            raise IOError("A stream read returned {} bytes, expected 4 bytes".format(len(buff)))
+
+        # convert them into the packet descriptors
+        msg_cls, msg_id, length = struct.unpack("BBH", buff)
+
+        # check the packet validity
+        if msg_cls not in self.classes:
+            raise ValueError("Received unsupported message class of {:x}".format(msg_cls))
+        if msg_id not in self.classes[msg_cls]:
+            raise ValueError("Received unsupported message id of {:x} in class {:x}".format(msg_id, msg_cls))
+
+        # Read the payload
+        payload = await stream.readexactly(length)
+        full_msg_for_checksum = buff + payload
+
+        # Read the checksum
+        checksum_sup = await stream.readexactly(2)
+        checksum_cal = self._generate_fletcher_checksum(full_msg_for_checksum)
+
+        if checksum_cal != checksum_sup:
+            raise ValueError("Checksum mismatch. Calculated {:x} {:x}, received {:x} {:x}".format(
+                checksum_cal[0], checksum_cal[1], checksum_sup[0], checksum_sup[1]
+            ))
+
+        return self.classes[msg_cls].parse(msg_id, payload)
+
     @staticmethod
     def _read_until(stream, terminator: bytes, size=None):
         """Read from the stream until the terminator byte/s are read.
@@ -468,6 +499,24 @@ def _read_until(stream, terminator: bytes, size=None):
 
         return bytes(line)
 
+    @staticmethod
+    async def _read_until_async(stream, terminator: bytes, size=None):
+        """Async version of Parser._read_until."""
+        term_len = len(terminator)
+        line = bytearray()
+        while True:
+            c = await stream.read(1)
+            if c:
+                line += c
+                if line[-term_len:] == terminator:
+                    break
+                if size is not None and len(line) >= size:
+                    break
+            else:
+                break
+
+        return bytes(line)
+
     @staticmethod
     def _generate_fletcher_checksum(payload: bytes) -> bytes:
         """Return the checksum for the provided payload"""
diff --git a/ubxtranslator/examples/async.py b/ubxtranslator/examples/async.py
@@ -0,0 +1,58 @@
+"""An example of a state class with an async parser using pyserial-asyncio.
+
+You will need to change the port name to that of the port you want to connect to. Also make sure that the baud rate is
+correct and that the device has been setup to output UBX messages protocol to your desired port!
+"""
+
+import asyncio
+from asyncio import Event
+
+import serial_asyncio
+
+from ubxtranslator.core import Parser
+from ubxtranslator.predefined import NAV_CLS, ACK_CLS
+
+
+class GnssService:
+    def __init__(self, port, baud_rate):
+        self.parser = Parser([NAV_CLS, ACK_CLS])
+        self.port = port
+        self.baud_rate = baud_rate
+        self.last_message = None
+        self.new_message_event = Event()
+
+    async def read_serial(self):
+        """Read from serial port asynchronously using pyserial-asyncio."""
+        while True:
+            try:
+                reader, _ = await serial_asyncio.open_serial_connection(
+                    url=self.port, baudrate=self.baud_rate
+                )
+                print(f"Starting to listen for UBX packets on {self.port}")
+                while True:
+                    try:
+                        msg = await self.parser.receive_from_async(reader)
+                        if msg:
+                            self.last_message = msg
+                            self.new_message_event.set()
+
+                    except (ValueError, IOError, asyncio.IncompleteReadError) as e:
+                        print(f"Error parsing UBX message: {e}")
+                        continue
+
+            except Exception as e:
+                print(f"Could not open serial port {self.port}: {e}")
+                # Wait before trying to reconnect
+                await asyncio.sleep(5)
+
+    async def print_last_message(self):
+        """An event based approach to consumption."""
+        while True:
+            await self.new_message_event.wait()
+            print(self.last_message)
+            self.new_message_event.clear()
+
+    async def run(self):
+        """Run the service."""
+        await asyncio.gather(self.read_serial(), self.print_last_message())
+
diff --git a/ubxtranslator/tests/__init__.py b/ubxtranslator/tests/__init__.py
@@ -1,6 +1,6 @@
 import unittest
 
-from . import test_core, test_fields
+from . import test_core, test_fields, test_async
 
 
 def suite():
@@ -17,6 +17,9 @@ def suite():
     suite.addTest(test_core.UbxClsTester())
     suite.addTest(test_core.UbxParserTester())
 
+    # test async
+    suite.addTest(test_async.UbxAsyncParserTester())
+
     return suite
 
 
diff --git a/ubxtranslator/tests/test_async.py b/ubxtranslator/tests/test_async.py
@@ -0,0 +1,104 @@
+import unittest
+import asyncio
+from ubxtranslator.core import *
+
+
+class MockStreamReader:
+    def __init__(self, data):
+        self.data = data
+        self.pos = 0
+
+    async def read(self, n):
+        chunk = self.data[self.pos:self.pos + n]
+        self.pos += len(chunk)
+        return chunk
+
+    async def readexactly(self, n):
+        chunk = self.data[self.pos:self.pos + n]
+        if len(chunk) < n:
+            raise asyncio.IncompleteReadError(chunk, n)
+        self.pos += n
+        return chunk
+
+
+class UbxAsyncParserTester(unittest.IsolatedAsyncioTestCase):
+    async def test_parser_async(self):
+        cls = Cls(1, 'TEST', [
+            Message(1, 'TEST', [
+                PadByte(),
+                Field('F1', 'U1'),
+                Field('F2', 'I1'),
+                PadByte(),
+                Field('F3', 'U1'),
+                BitField('F4', 'X1', [
+                    Flag('SF1', 0, 4),
+                    Flag('SF2', 4, 8)
+                ])
+            ])
+        ])
+
+        parser = Parser([cls])
+
+        with self.subTest(msg='Test correct async usage'):
+            test_packet = bytes([1, 1, 6, 0, 0, 1, 2, 3, 4, 5])
+            test_packet = parser.PREFIX + test_packet + parser._generate_fletcher_checksum(test_packet)
+
+            test_stream = MockStreamReader(test_packet)
+
+            cls_name, msg_name, msg = await parser.receive_from_async(test_stream)
+
+            self.assertEqual(msg.F1, 1)
+            self.assertEqual(msg.F2, 2)
+            self.assertEqual(msg.F3, 4)
+            self.assertEqual(msg.F4.SF1, 5)
+            self.assertEqual(msg.F4.SF2, 0)
+
+        with self.subTest(msg='Test async bad class id'):
+            test_packet = bytes([2, 1, 6, 0, 0, 1, 2, 3, 4, 5])
+            test_packet = parser.PREFIX + test_packet + parser._generate_fletcher_checksum(test_packet)
+            test_stream = MockStreamReader(test_packet)
+            with self.assertRaises(ValueError):
+                await parser.receive_from_async(test_stream)
+
+        with self.subTest(msg='Test async bad msg'):
+            test_packet = bytes([1, 2, 6, 0, 0, 1, 2, 3, 4, 5])
+            test_packet = parser.PREFIX + test_packet + parser._generate_fletcher_checksum(test_packet)
+            test_stream = MockStreamReader(test_packet)
+            with self.assertRaises(ValueError):
+                await parser.receive_from_async(test_stream)
+
+        with self.subTest(msg='Test async bad checksum'):
+            test_packet = bytes([1, 1, 6, 0, 0, 1, 2, 3, 4, 5])
+            test_packet = parser.PREFIX + test_packet + bytes([0, 1])
+            test_stream = MockStreamReader(test_packet)
+            with self.assertRaises(ValueError):
+                await parser.receive_from_async(test_stream)
+
+        with self.subTest(msg='Test async insufficient data'):
+            test_packet = bytes([1, 1, 6, 0, 0, 1, 2, 3, 5])
+            test_packet = parser.PREFIX + test_packet + parser._generate_fletcher_checksum(test_packet)
+            test_stream = MockStreamReader(test_packet)
+            with self.assertRaises(asyncio.IncompleteReadError):
+                await parser.receive_from_async(test_stream)
+
+    async def test_read_until_async(self):
+        data = b"some junk" + Parser.PREFIX + b"more data"
+        stream = MockStreamReader(data)
+
+        # Test reading until PREFIX
+        result = await Parser._read_until_async(stream, terminator=Parser.PREFIX)
+        self.assertEqual(result, b"some junk" + Parser.PREFIX)
+
+        # Test reading with size limit
+        stream = MockStreamReader(b"1234567890")
+        result = await Parser._read_until_async(stream, terminator=b"XYZ", size=5)
+        self.assertEqual(result, b"12345")
+
+        # Test reading until EOF
+        stream = MockStreamReader(b"eof reach")
+        result = await Parser._read_until_async(stream, terminator=b"NOT_HERE")
+        self.assertEqual(result, b"eof reach")
+
+
+if __name__ == '__main__':
+    unittest.main()
