From 28b040953467a4cfa5a09761f7c239f670117529 Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Tue, 8 Jul 2025 20:24:58 -0400 Subject: [PATCH 01/13] new async-friendly Connection API --- meshtastic/connection.py | 58 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) create mode 100644 meshtastic/connection.py diff --git a/meshtastic/connection.py b/meshtastic/connection.py new file mode 100644 index 00000000..242b2801 --- /dev/null +++ b/meshtastic/connection.py @@ -0,0 +1,58 @@ +import asyncio +from abc import ABC, abstractmethod +from typing import * + +from pubsub import pub + +from meshtastic.protobuf.mesh_pb2 import FromRadio, ToRadio + + +class MeshConnection(ABC): + """A client API connection to a meshtastic radio.""" + + def __init__(self): + self._on_disconnect: asyncio.Event = asyncio.Event() + + @abstractmethod + async def _send_bytes(self, msg: buffer): + """Send bytes to the mesh device.""" + pass + + @abstractmethod + async def _recv_bytes(self) -> buffer: + """Recieve bytes from the mesh device.""" + pass + + @abstractmethod + def close(self): + """Close the connection""" + pass + + @staticmethod + @abstractmethod + async def get_available() -> AsyncGenerator[Any]: + """Enumerate any mesh devices that can be connected to. + + Generates values that can be passed to the concrete connection class's + constructor.""" + pass + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, trace): + self.close() + + async def send(self, message: ToRadio): + """Send something to the connected device.""" + msg_str: str = message.SerializeToString() + await self._send_bytes(bytes(msg_str)) + + async def recv(self) -> FromRadio: + """Recieve something from the connected device.""" + msg_bytes: buffer = await self._recv_bytes() + return FromRadio.FromString(str(msg_bytes)) + + async def listen(self) -> AsyncGenerator[FromRadio]: + while True: + yield await self.recv() From 2b1114cdbf208fe9947c82189923cd1dcab72bde Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Wed, 9 Jul 2025 14:48:19 -0400 Subject: [PATCH 02/13] Add seeed xiao ESP32s3 to supported devices --- meshtastic/supported_device.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/meshtastic/supported_device.py b/meshtastic/supported_device.py index 8f491fb5..b67d07cf 100755 --- a/meshtastic/supported_device.py +++ b/meshtastic/supported_device.py @@ -207,6 +207,16 @@ def __init__( usb_product_id_in_hex="55d4", ) +seeed_xiao_s3 = SupportedDevice( + name = "Seeed Xiao ESP32-S3", + version = "", + for_firmware="seeed-xiao-esp32s3", + baseport_on_linux="ttyACM", + baseport_on_mac="cu.usbmodem", + usb_vendor_id_in_hex="2886", + usb_product_id_in_hex="0059", +) + supported_devices = [ tbeam_v0_7, tbeam_v1_1, @@ -226,4 +236,5 @@ def __init__( rak4631_19003, rak11200, nano_g1, + seeed_xiao_s3, ] From a8c271c41944963c8450d65eb5a610f3b2a58538 Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Wed, 9 Jul 2025 15:50:11 -0400 Subject: [PATCH 03/13] Revert "Add seeed xiao ESP32s3 to supported devices" This reverts commit 2b1114cdbf208fe9947c82189923cd1dcab72bde. --- meshtastic/supported_device.py | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/meshtastic/supported_device.py b/meshtastic/supported_device.py index b67d07cf..8f491fb5 100755 --- a/meshtastic/supported_device.py +++ b/meshtastic/supported_device.py @@ -207,16 +207,6 @@ def __init__( usb_product_id_in_hex="55d4", ) -seeed_xiao_s3 = SupportedDevice( - name = "Seeed Xiao ESP32-S3", - version = "", - for_firmware="seeed-xiao-esp32s3", - baseport_on_linux="ttyACM", - baseport_on_mac="cu.usbmodem", - usb_vendor_id_in_hex="2886", - usb_product_id_in_hex="0059", -) - supported_devices = [ tbeam_v0_7, tbeam_v1_1, @@ -236,5 +226,4 @@ def __init__( rak4631_19003, rak11200, nano_g1, - seeed_xiao_s3, ] From 2e3719ee49907acbb4e72e41498160269f8094ab Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Wed, 9 Jul 2025 15:50:38 -0400 Subject: [PATCH 04/13] implement first draft stream and serial connections --- meshtastic/connection.py | 156 ++++++++++++++++++++++++++++++++++----- 1 file changed, 139 insertions(+), 17 deletions(-) diff --git a/meshtastic/connection.py b/meshtastic/connection.py index 242b2801..773ce15a 100644 --- a/meshtastic/connection.py +++ b/meshtastic/connection.py @@ -1,17 +1,45 @@ import asyncio +import io +import logging from abc import ABC, abstractmethod from typing import * -from pubsub import pub +import serial +import serial_asyncio from meshtastic.protobuf.mesh_pb2 import FromRadio, ToRadio +# magic number used in streaming client headers +HEADER_MAGIC: bytes = b"\x94\xc3" + + +class ConnectionError(Exception): + """Base class for MeshConnection-related errors.""" + + +class BadPayloadError(ConnectionError): + def __init__(self, payload, reason: str): + self.payload = payload + super().__init__(reason) + + class MeshConnection(ABC): """A client API connection to a meshtastic radio.""" - def __init__(self): - self._on_disconnect: asyncio.Event = asyncio.Event() + def __init__(self, name: str): + self.name: str = name + self.on_disconnect: asyncio.Event = asyncio.Event() + self._is_ready: bool = False + self._send_lock: asyncio.Lock = asyncio.Lock() + self._recv_lock: asyncio.Lock = asyncio.Lock() + self._init_task: asyncio.Task = asyncio.create_task(self._initialize()) + self._init_task.add_done_callback(self._after_initialize) + + @abstractmethod + async def _initialize(self): + """Perform any connection initialization that must be performed async + (and therefore not from the constructor).""" @abstractmethod async def _send_bytes(self, msg: buffer): @@ -23,11 +51,6 @@ async def _recv_bytes(self) -> buffer: """Recieve bytes from the mesh device.""" pass - @abstractmethod - def close(self): - """Close the connection""" - pass - @staticmethod @abstractmethod async def get_available() -> AsyncGenerator[Any]: @@ -37,22 +60,121 @@ async def get_available() -> AsyncGenerator[Any]: constructor.""" pass - def __enter__(self): - return self + def ready(self): + return self._is_ready - def __exit__(self, exc_type, exc_value, trace): - self.close() + def _after_initialize(self): + self._is_ready = True + del self._init_task async def send(self, message: ToRadio): """Send something to the connected device.""" - msg_str: str = message.SerializeToString() - await self._send_bytes(bytes(msg_str)) + async with self._send_lock: + msg_str: str = message.SerializeToString() + await self._send_bytes(bytes(msg_str)) async def recv(self) -> FromRadio: """Recieve something from the connected device.""" - msg_bytes: buffer = await self._recv_bytes() - return FromRadio.FromString(str(msg_bytes)) + async with self._recv_lock: + msg_bytes: buffer = await self._recv_bytes() + return FromRadio.FromString(str(msg_bytes, errors="ignore")) async def listen(self) -> AsyncGenerator[FromRadio]: - while True: + while not self.on_disconnect.is_set(): yield await self.recv() + + def close(self): + """Close the connection. + Overloaders should remember to call supermethod""" + if not self.ready(): + self._init_task.cancel() + + self.on_disconnect.set() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, trace): + self.close() + + +class StreamConnection(MeshConnection): + """Base class for connections using the aio stream API""" + def __init__(self, name: str): + self._reader: Optional[asyncio.StreamReader] = None + self._writer: Optional[asyncio.StreamWriter] = None + self.stream_debug_out: io.StringIO = io.StringIO() + super().__init__(name) + + def _handle_debug(self, debug_out: bytes): + self.stream_debug_out.write(str(debug_out)) + self.stream_debug_out.flush() + + async def _send_bytes(self, msg: buffer): + length: int = len(msg) + if length > 512: + raise BadPayloadError(msg, "Cannot send client API messages over 512 bytes") + + self._writer.write(HEADER_MAGIC) + self._writer.write(length.to_bytes(2, "big")) + self._writer.write(msg) + await self._writer.drain() + + async def _find_stream_header(self): + """Consumes and logs debug out bytes until a valid header is detected""" + try: + while True: + from_stream: bytes = await self._reader.readuntil((b'\n', HEADER_MAGIC)) + if from_stream.endswith(HEADER_MAGIC): + self._handle_debug(from_stream[:-2]) + return + else: + self._handle_debug(from_stream) + + except asyncio.IncompleteReadError as err: + if len(err.partial) > 0: + self._handle_debug(err.partial) + raise + + async def _recv_bytes(self) -> buffer: + try: + while True: + await self._find_stream_header() + size_bytes: bytes = await self._reader.readexactly(2) + size: int = int.from_bytes(size_bytes, "big") + if 0 < size <= 512: + return await self._reader.readexactly(size) + + self._handle_debug(size_bytes) + + except asyncio.LimitOverrunError as err: + raise ConnectionError( + "Read buffer overrun while reading stream") from err + + except asyncio.IncompleteReadError: + logging.error(f"Connection to {self.name} terminated: stream EOF reached") + self.close() + + def close(self): + super().close() + self._writer.close() + self.stream_debug_out.close() + asyncio.as_completed((self._writer.wait_closed(),)) + + +class SerialConnection(StreamConnection): + def __init__(self, portaddr: str, baudrate: int=115200): + self.port: str = portaddr + self.baudrate: int = baudrate + super().__init__(portaddr) + + async def _initialize(self): + self._reader, self._writer = await serial_asyncio.open_serial_connectio( + port=self._port, baudrate=self._baudrate, + ) + + @staticmethod + async def get_available() -> AsyncGenerator[str]: + for port in serial.tools.list_ports.comports(): + if port.hwid != "n/a": + yield port.device From 743474fdfebe92af0d04e5cb1082e4b609660827 Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Mon, 21 Jul 2025 12:42:49 -0400 Subject: [PATCH 05/13] add connection module docstring --- meshtastic/connection.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/meshtastic/connection.py b/meshtastic/connection.py index 773ce15a..4ca0dfc2 100644 --- a/meshtastic/connection.py +++ b/meshtastic/connection.py @@ -1,3 +1,6 @@ +""" +low-level radio connection API +""" import asyncio import io import logging From 4b8ba4afa2f8b88b3feb46eb1040c36e2c3ff7df Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Mon, 21 Jul 2025 12:45:39 -0400 Subject: [PATCH 06/13] Refactor async connection init/teardown --- meshtastic/connection.py | 95 ++++++++++++++++++++++++---------------- 1 file changed, 57 insertions(+), 38 deletions(-) diff --git a/meshtastic/connection.py b/meshtastic/connection.py index 4ca0dfc2..9bcc550c 100644 --- a/meshtastic/connection.py +++ b/meshtastic/connection.py @@ -13,31 +13,34 @@ from meshtastic.protobuf.mesh_pb2 import FromRadio, ToRadio -# magic number used in streaming client headers -HEADER_MAGIC: bytes = b"\x94\xc3" +STREAM_HEADER_MAGIC: bytes = b"\x94\xc3" # magic number used in streaming client headers +DEFAULT_BAUDRATE: int = 115200 -class ConnectionError(Exception): +class MeshConnectionError(Exception): """Base class for MeshConnection-related errors.""" -class BadPayloadError(ConnectionError): +class BadPayloadError(MeshConnectionError): + """Error indicating invalid payload over connection""" def __init__(self, payload, reason: str): self.payload = payload super().__init__(reason) +class ConnectionTerminatedError(MeshConnectionError): + """Error indicating the connection was terminated.""" + + class MeshConnection(ABC): """A client API connection to a meshtastic radio.""" def __init__(self, name: str): self.name: str = name + self.on_ready: asyncio.Event = asyncio.Event() self.on_disconnect: asyncio.Event = asyncio.Event() - self._is_ready: bool = False self._send_lock: asyncio.Lock = asyncio.Lock() self._recv_lock: asyncio.Lock = asyncio.Lock() - self._init_task: asyncio.Task = asyncio.create_task(self._initialize()) - self._init_task.add_done_callback(self._after_initialize) @abstractmethod async def _initialize(self): @@ -45,14 +48,12 @@ async def _initialize(self): (and therefore not from the constructor).""" @abstractmethod - async def _send_bytes(self, msg: buffer): + async def _send_bytes(self, msg: bytes): """Send bytes to the mesh device.""" - pass @abstractmethod - async def _recv_bytes(self) -> buffer: + async def _recv_bytes(self) -> bytes: """Recieve bytes from the mesh device.""" - pass @staticmethod @abstractmethod @@ -61,44 +62,61 @@ async def get_available() -> AsyncGenerator[Any]: Generates values that can be passed to the concrete connection class's constructor.""" - pass def ready(self): - return self._is_ready + """Returns if the connection is ready for tx/rx""" + return self.on_ready.is_set() - def _after_initialize(self): - self._is_ready = True - del self._init_task + async def open(self): + """Start the connection""" + await self._initialize() + self.on_ready.set() + + def _ensure_ready(self): + """Raise an exception if the connection is not ready for tx/rx""" + if not self.ready(): + raise MeshConnectionError("Connection used before it was ready") async def send(self, message: ToRadio): """Send something to the connected device.""" + self._ensure_ready() async with self._send_lock: msg_str: str = message.SerializeToString() await self._send_bytes(bytes(msg_str)) async def recv(self) -> FromRadio: """Recieve something from the connected device.""" + self._ensure_ready() async with self._recv_lock: - msg_bytes: buffer = await self._recv_bytes() + msg_bytes: bytes = await self._recv_bytes() return FromRadio.FromString(str(msg_bytes, errors="ignore")) async def listen(self) -> AsyncGenerator[FromRadio]: + """Yields new messages from the radio so long as the connection is active.""" + self._ensure_ready() while not self.on_disconnect.is_set(): yield await self.recv() - def close(self): + async def close(self): """Close the connection. Overloaders should remember to call supermethod""" - if not self.ready(): - self._init_task.cancel() - + self.on_ready.unset() self.on_disconnect.set() - def __enter__(self): + async def __aenter__(self): + await self.open() return self - def __exit__(self, exc_type, exc_value, trace): - self.close() + async def __aexit__(self, exc_type, exc_value, trace): + await self.close() + + #def __enter__(self): + # self.open() + # asyncio.run(self._init_task) + # return self + + #def __exit__(self, exc_type, exc_value, trace): + # self.close() class StreamConnection(MeshConnection): @@ -113,12 +131,12 @@ def _handle_debug(self, debug_out: bytes): self.stream_debug_out.write(str(debug_out)) self.stream_debug_out.flush() - async def _send_bytes(self, msg: buffer): + async def _send_bytes(self, msg: bytes): length: int = len(msg) if length > 512: raise BadPayloadError(msg, "Cannot send client API messages over 512 bytes") - self._writer.write(HEADER_MAGIC) + self._writer.write(STREAM_HEADER_MAGIC) self._writer.write(length.to_bytes(2, "big")) self._writer.write(msg) await self._writer.drain() @@ -127,8 +145,8 @@ async def _find_stream_header(self): """Consumes and logs debug out bytes until a valid header is detected""" try: while True: - from_stream: bytes = await self._reader.readuntil((b'\n', HEADER_MAGIC)) - if from_stream.endswith(HEADER_MAGIC): + from_stream: bytes = await self._reader.readuntil((b'\n', STREAM_HEADER_MAGIC)) + if from_stream.endswith(STREAM_HEADER_MAGIC): self._handle_debug(from_stream[:-2]) return else: @@ -139,7 +157,7 @@ async def _find_stream_header(self): self._handle_debug(err.partial) raise - async def _recv_bytes(self) -> buffer: + async def _recv_bytes(self) -> bytes: try: while True: await self._find_stream_header() @@ -151,33 +169,34 @@ async def _recv_bytes(self) -> buffer: self._handle_debug(size_bytes) except asyncio.LimitOverrunError as err: - raise ConnectionError( - "Read buffer overrun while reading stream") from err + raise MeshConnectionError("Read buffer overrun while reading stream") from err except asyncio.IncompleteReadError: logging.error(f"Connection to {self.name} terminated: stream EOF reached") - self.close() + raise ConnectionTerminatedError from None - def close(self): - super().close() + async def close(self): + await super().close() self._writer.close() self.stream_debug_out.close() - asyncio.as_completed((self._writer.wait_closed(),)) + await self._writer.wait_closed() class SerialConnection(StreamConnection): - def __init__(self, portaddr: str, baudrate: int=115200): + """Connection to a mesh radio over serial port""" + def __init__(self, portaddr: str, baudrate: int=DEFAULT_BAUDRATE): self.port: str = portaddr self.baudrate: int = baudrate super().__init__(portaddr) async def _initialize(self): self._reader, self._writer = await serial_asyncio.open_serial_connectio( - port=self._port, baudrate=self._baudrate, - ) + port=self._port, baudrate=self._baudrate) @staticmethod async def get_available() -> AsyncGenerator[str]: for port in serial.tools.list_ports.comports(): + # filtering for hwid gets rid of linux VT serials (e.g, /dev/ttyS0 and friends) + # FIXME: this may not be cross-platform or non-USB serial friendly if port.hwid != "n/a": yield port.device From 05e52e38faeef9e01f2e7da5c58047ccb7da18dc Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Mon, 21 Jul 2025 12:52:18 -0400 Subject: [PATCH 07/13] Rename MeshConnection -> RadioConnection --- meshtastic/connection.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/meshtastic/connection.py b/meshtastic/connection.py index 9bcc550c..1b3ae2c0 100644 --- a/meshtastic/connection.py +++ b/meshtastic/connection.py @@ -17,22 +17,22 @@ DEFAULT_BAUDRATE: int = 115200 -class MeshConnectionError(Exception): - """Base class for MeshConnection-related errors.""" +class RadioConnectionError(Exception): + """Base class for RadioConnection-related errors.""" -class BadPayloadError(MeshConnectionError): +class BadPayloadError(RadioConnectionError): """Error indicating invalid payload over connection""" def __init__(self, payload, reason: str): self.payload = payload super().__init__(reason) -class ConnectionTerminatedError(MeshConnectionError): +class ConnectionTerminatedError(RadioConnectionError): """Error indicating the connection was terminated.""" -class MeshConnection(ABC): +class RadioConnection(ABC): """A client API connection to a meshtastic radio.""" def __init__(self, name: str): @@ -75,7 +75,7 @@ async def open(self): def _ensure_ready(self): """Raise an exception if the connection is not ready for tx/rx""" if not self.ready(): - raise MeshConnectionError("Connection used before it was ready") + raise RadioConnectionError("Connection used before it was ready") async def send(self, message: ToRadio): """Send something to the connected device.""" @@ -119,7 +119,7 @@ async def __aexit__(self, exc_type, exc_value, trace): # self.close() -class StreamConnection(MeshConnection): +class StreamConnection(RadioConnection): """Base class for connections using the aio stream API""" def __init__(self, name: str): self._reader: Optional[asyncio.StreamReader] = None @@ -169,7 +169,7 @@ async def _recv_bytes(self) -> bytes: self._handle_debug(size_bytes) except asyncio.LimitOverrunError as err: - raise MeshConnectionError("Read buffer overrun while reading stream") from err + raise RadioConnectionError("Read buffer overrun while reading stream") from err except asyncio.IncompleteReadError: logging.error(f"Connection to {self.name} terminated: stream EOF reached") From ad37c2191db20a234b0bd9e64717e424003bf10d Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Mon, 21 Jul 2025 13:50:11 -0400 Subject: [PATCH 08/13] fix typo --- meshtastic/connection.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/meshtastic/connection.py b/meshtastic/connection.py index 1b3ae2c0..f79d8b03 100644 --- a/meshtastic/connection.py +++ b/meshtastic/connection.py @@ -190,7 +190,7 @@ def __init__(self, portaddr: str, baudrate: int=DEFAULT_BAUDRATE): super().__init__(portaddr) async def _initialize(self): - self._reader, self._writer = await serial_asyncio.open_serial_connectio( + self._reader, self._writer = await serial_asyncio.open_serial_connection( port=self._port, baudrate=self._baudrate) @staticmethod From 664c19a0aeacf531c7113c66bdf6722a986a17ee Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Mon, 21 Jul 2025 14:02:35 -0400 Subject: [PATCH 09/13] implement lock for connection listen method --- meshtastic/connection.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/meshtastic/connection.py b/meshtastic/connection.py index f79d8b03..cdea081f 100644 --- a/meshtastic/connection.py +++ b/meshtastic/connection.py @@ -41,6 +41,7 @@ def __init__(self, name: str): self.on_disconnect: asyncio.Event = asyncio.Event() self._send_lock: asyncio.Lock = asyncio.Lock() self._recv_lock: asyncio.Lock = asyncio.Lock() + self._listen_lock: asyncio.Lock = asyncio.Lock() @abstractmethod async def _initialize(self): @@ -94,8 +95,9 @@ async def recv(self) -> FromRadio: async def listen(self) -> AsyncGenerator[FromRadio]: """Yields new messages from the radio so long as the connection is active.""" self._ensure_ready() - while not self.on_disconnect.is_set(): - yield await self.recv() + async with self._listen_lock: + while not self.on_disconnect.is_set(): + yield await self.recv() async def close(self): """Close the connection. From 5f48ed6cf5d9a512fa33998bc0446dbb7900fcb2 Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Mon, 21 Jul 2025 16:53:44 -0400 Subject: [PATCH 10/13] Ensure proper EOF handling for stream connections --- meshtastic/connection.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/meshtastic/connection.py b/meshtastic/connection.py index cdea081f..f6ab6b58 100644 --- a/meshtastic/connection.py +++ b/meshtastic/connection.py @@ -174,11 +174,15 @@ async def _recv_bytes(self) -> bytes: raise RadioConnectionError("Read buffer overrun while reading stream") from err except asyncio.IncompleteReadError: + self._reader.feed_eof() logging.error(f"Connection to {self.name} terminated: stream EOF reached") raise ConnectionTerminatedError from None async def close(self): await super().close() + if self._writer.can_write_eof(): + self._writer.write_eof() + self._writer.close() self.stream_debug_out.close() await self._writer.wait_closed() From 0e320cd64baa0928f2ee6dcf90e03328342958e1 Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Mon, 21 Jul 2025 16:54:14 -0400 Subject: [PATCH 11/13] Fix serial connection args --- meshtastic/connection.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/meshtastic/connection.py b/meshtastic/connection.py index f6ab6b58..eb3a085e 100644 --- a/meshtastic/connection.py +++ b/meshtastic/connection.py @@ -72,6 +72,7 @@ async def open(self): """Start the connection""" await self._initialize() self.on_ready.set() + logging.info(f"Connected to mesh radio {self.name}") def _ensure_ready(self): """Raise an exception if the connection is not ready for tx/rx""" @@ -197,7 +198,7 @@ def __init__(self, portaddr: str, baudrate: int=DEFAULT_BAUDRATE): async def _initialize(self): self._reader, self._writer = await serial_asyncio.open_serial_connection( - port=self._port, baudrate=self._baudrate) + url=self.port, baudrate=self.baudrate) @staticmethod async def get_available() -> AsyncGenerator[str]: From 773d1f930bf52b91a8f0aa768f604cf1bdad1f68 Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Mon, 21 Jul 2025 16:55:06 -0400 Subject: [PATCH 12/13] Implement first draft TCP connections --- meshtastic/connection.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/meshtastic/connection.py b/meshtastic/connection.py index eb3a085e..67fc59e2 100644 --- a/meshtastic/connection.py +++ b/meshtastic/connection.py @@ -15,6 +15,7 @@ STREAM_HEADER_MAGIC: bytes = b"\x94\xc3" # magic number used in streaming client headers DEFAULT_BAUDRATE: int = 115200 +DEFAULT_TCP_PORT: int = 4403 class RadioConnectionError(Exception): @@ -207,3 +208,19 @@ async def get_available() -> AsyncGenerator[str]: # FIXME: this may not be cross-platform or non-USB serial friendly if port.hwid != "n/a": yield port.device + + +class TCPConnection(StreamConnection): + """Connection to a mesh radio over TCP""" + + def __init__(self, host: str, port: int=DEFAULT_TCP_PORT): + self.host: str = host + self.port: int = port + super().__init__(f"{host}:{port}") + + async def _initialize(self): + self._reader, self._writer = await asyncio.open_connection(self.host, self.port) + + @staticmethod + async def get_available() -> AsyncGenerator[None]: + yield None # FIXME From b0b8be80df09188dc93f0edbc831344d70a3b16b Mon Sep 17 00:00:00 2001 From: flockofsparrows Date: Mon, 21 Jul 2025 16:55:39 -0400 Subject: [PATCH 13/13] Implement first draft BLE connection --- meshtastic/connection.py | 59 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/meshtastic/connection.py b/meshtastic/connection.py index 67fc59e2..004a18b2 100644 --- a/meshtastic/connection.py +++ b/meshtastic/connection.py @@ -9,6 +9,7 @@ import serial import serial_asyncio +from bleak import BleakClient, BLEDevice, BleakScanner from meshtastic.protobuf.mesh_pb2 import FromRadio, ToRadio @@ -16,6 +17,12 @@ STREAM_HEADER_MAGIC: bytes = b"\x94\xc3" # magic number used in streaming client headers DEFAULT_BAUDRATE: int = 115200 DEFAULT_TCP_PORT: int = 4403 +BLE_SERVICE_UUID: str = "6ba1b218-15a8-461f-9fa8-5dcae273eafd" +BLE_TORADIO_UUID: str = "f75c76d2-129e-4dad-a1dd-7866124401e7" +BLE_FROMRADIO_UUID: str = "2c55e69e-4993-11ed-b878-0242ac120002" +BLE_FROMNUM_UUID: str = "ed9da18c-a800-4f66-a670-aa7547e34453" +BLE_LEGACY_LOGRADIO_UUID: str = "6c6fd238-78fa-436b-aacf-15c5be1ef2e2" +BLE_LOGRADIO_UUID: str = "5a3d6e49-06e6-4423-9944-e9de8cdf9547" class RadioConnectionError(Exception): @@ -224,3 +231,55 @@ async def _initialize(self): @staticmethod async def get_available() -> AsyncGenerator[None]: yield None # FIXME + + +class BLEConnection(RadioConnection): + """Connection to a mesh radio over BLE""" + + def __init__(self, device: Union[str, BLEDevice]): + self._recieved_messages: asyncio.Queue = asyncio.Queue() + self._ble_client = BleakClient(device, disconnected_callback=lambda _: self.close()) + self._ble_client.mtu_size = 512 + + name: str = device + if isinstance(device, BLEDevice): + name = device.name + super().__init__(name) + + async def _initialize(self): + await self._ble_client.connect() + await self._ble_client.start_notify(BLE_FROMNUM_UUID, self._on_recv) + + async def _on_recv(self, _sender: Any, _data: bytearray): + """Callback for handling fromnum endpoint notifs""" + data: bytearray = await self._ble_client.read_gatt_char(BLE_FROMRADIO_UUID) + if len(data) > 512: + raise BadPayloadError(data, "Cannot recieve client API messages over 512 bytes") + + await self._recieved_messages.put(data) + + async def _read_bytes(self) -> bytes: + return bytes(await self._recieved_messages.get()) + + async def _send_bytes(self, msg: bytes): + if len(msg) > 512: + raise BadPayloadError(msg, "Cannot send client API messages over 512 bytes") + + await self._ble_client.write_gatt_char(BLE_TORADIO_UUID, msg, response=True) + + async def close(self): + await super().close() + self._recieved_messages.shutdown(True) + await self._ble_client.stop_notify(BLE_FROMNUM_UUID) + await self._ble_client.disconnect() + + @staticmethod + async def get_available() -> AsyncGenerator[BLEDevice]: + async with BleakScanner(service_uuids=(BLE_SERVICE_UUID,)) as scanner: + try: + async with asyncio.timeout(10): + async for dev, _ad in scanner.advertisement_data(): + yield dev + + except TimeoutError: + pass