Skip to content

Instantly share code, notes, and snippets.

@agners
Created April 2, 2022 10:36
Show Gist options
  • Save agners/dae71532bbdf13ca131fe0c724586876 to your computer and use it in GitHub Desktop.
Save agners/dae71532bbdf13ca131fe0c724586876 to your computer and use it in GitHub Desktop.
Test serial_asyncio with socat
import asyncio
import serial_asyncio
import serial
class Output(asyncio.Protocol):
def connection_made(self, transport):
self.transport = transport
print('port opened', transport)
# This causes an exception
#transport.serial.rts = False
transport.write(b'hello world\n')
def data_received(self, data):
print('data received', repr(data))
self.transport.close()
def connection_lost(self, exc):
print('port closed')
asyncio.get_event_loop().stop()
loop = asyncio.get_event_loop()
coro = serial_asyncio.create_serial_connection(loop, Output,
url='/dev/virtualcom0', baudrate=57600,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
xonxoff=True,
rtscts=False)
loop.run_until_complete(coro)
loop.run_forever()
loop.close()
# Try connecting to a virtual serial port created using:
# socat pty,link=/dev/virtualcom0,raw TCP-LISTEN:11313,reuseaddr,fork
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment