Mirrors/meshcore_py
1
0
Fork 0
mirror of https://github.com/meshcore-dev/meshcore_py.git synced 2026-09-23 14:06:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

let user specify rts state upon connect

Browse Source
This commit is contained in:
Florent
2026-09-14 12:37:28 -04:00
parent 6ffb875c28
commit f757e2b2f6
2 changed files with 5 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/meshcore/meshcore.py
+2 -1
View File
@@ -118,9 +118,10 @@ class MeshCore:
auto_reconnect: bool = False, auto_reconnect: bool = False,
max_reconnect_attempts: int = 3, max_reconnect_attempts: int = 3,
cx_dly: float = 0.1, cx_dly: float = 0.1,
rts = True
) -> "MeshCore": ) -> "MeshCore":
"""Create and connect a MeshCore instance using serial connection""" """Create and connect a MeshCore instance using serial connection"""
connection = SerialConnection(port, baudrate, cx_dly=cx_dly) connection = SerialConnection(port, baudrate, cx_dly=cx_dly, rts=rts)
mc = cls( mc = cls(
connection, connection,
src/meshcore/serial_cx.py
+3 -2
View File
@@ -11,7 +11,7 @@ logger = logging.getLogger("meshcore")
class SerialConnection: class SerialConnection:
def __init__(self, port, baudrate, cx_dly=0.2): def __init__(self, port, baudrate, cx_dly=0.2, rts=True):
self.port = port self.port = port
self.baudrate = baudrate self.baudrate = baudrate
self.transport = None self.transport = None
@@ -21,6 +21,7 @@ class SerialConnection:
self.cx_dly = cx_dly self.cx_dly = cx_dly
self._connected_event = asyncio.Event() self._connected_event = asyncio.Event()
self._background_tasks: set[asyncio.Task] = set() self._background_tasks: set[asyncio.Task] = set()
self.rts = rts
self.frame_expected_size = 0 self.frame_expected_size = 0
self.inframe = b"" self.inframe = b""
@@ -42,7 +43,7 @@ class SerialConnection:
logger.debug('port opened') logger.debug('port opened')
if isinstance(transport, serial_asyncio.SerialTransport) and transport.serial: if isinstance(transport, serial_asyncio.SerialTransport) and transport.serial:
transport.serial.dtr = False # Deassert DTR to avoid serial-open reset/handshake issues transport.serial.dtr = False # Deassert DTR to avoid serial-open reset/handshake issues
transport.serial.rts = False # You can manipulate Serial object via transport transport.serial.rts = self.cx.rts
self.cx._connected_event.set() self.cx._connected_event.set()
def data_received(self, data): def data_received(self, data):