From 46a1fb90d4944615eacc2e248c3f37aafed1da0a Mon Sep 17 00:00:00 2001 From: Trygve Date: Sun, 3 Apr 2022 22:00:15 +0200 Subject: [PATCH] Fungerende polling av mtr --- serial_async_test.py | 41 ----------------------------------------- serial_test.py | 19 +++++++++++++++++++ 2 files changed, 19 insertions(+), 41 deletions(-) delete mode 100644 serial_async_test.py create mode 100644 serial_test.py diff --git a/serial_async_test.py b/serial_async_test.py deleted file mode 100644 index 09f858e..0000000 --- a/serial_async_test.py +++ /dev/null @@ -1,41 +0,0 @@ -import asyncio -import serial_asyncio -from rich.console import Console -from rich import inspect -from datetime import datetime - -import mtrreader -import mtrlog - -class OutputProtocol(asyncio.Protocol): - def connection_made(self, transport): - self.transport = transport - print('port opened', transport) - transport.serial.rts = False # You can manipulate Serial object via transport - transport.write(b'Hello, World!\n') # Write serial data via transport - - def data_received(self, data): - print('data received', repr(data)) - msg = mtrreader.MtrDataMessage(data) - inspect(msg) - print(mtrlog.MtrLogFormatter.format(self, msg, datetime.now())) - if b'\n' in data: - self.transport.close() - - def connection_lost(self, exc): - print('port closed') - self.transport.loop.stop() - - def pause_writing(self): - print('pause writing') - print(self.transport.get_write_buffer_size()) - - def resume_writing(self): - print(self.transport.get_write_buffer_size()) - print('resume writing') - -loop = asyncio.get_event_loop() -coro = serial_asyncio.create_serial_connection(loop, OutputProtocol, '/dev/ttyUSB0', baudrate=9600) -transport, protocol = loop.run_until_complete(coro) -loop.run_forever() -loop.close() diff --git a/serial_test.py b/serial_test.py new file mode 100644 index 0000000..449ef91 --- /dev/null +++ b/serial_test.py @@ -0,0 +1,19 @@ +import serial +import mtrreader +from rich import inspect +import mtrlog +from datetime import datetime + +mtr = serial.Serial(port='/dev/ttyUSB0', baudrate=9600, timeout=40) + +def main(): + data = mtr.read_until(expected='FFFFFFFFFF', size=230) + print('START') + msg = mtrreader.MtrDataMessage(data) + f = mtrlog.MtrLogFormatter + print(f.format(f, msg, datetime.now())) + print('END') + main() + +if __name__ == '__main__': + main()