Roy
92 lines
2.3 KiB
Python
92 lines
2.3 KiB
Python
from time import sleep
|
|
from typing import Optional
|
|
|
|
import serial
|
|
|
|
from . import LowLevelHardwareInterface
|
|
|
|
SerialTimeoutException = serial.SerialTimeoutException
|
|
|
|
|
|
class UARTInterface(LowLevelHardwareInterface):
|
|
"""USB serial interface.
|
|
|
|
if device not found, raise SerialException.
|
|
"""
|
|
|
|
__slots__ = ('_port', '_baudrate', '_serial')
|
|
|
|
def __init__(self, port: str, baudrate: int = 115200):
|
|
self._port = port
|
|
self._baudrate = baudrate
|
|
self._serial: serial.Serial = None
|
|
|
|
def reset(self):
|
|
if self._serial is not None and not self._serial.closed:
|
|
self._serial.close()
|
|
|
|
self._serial = serial.Serial(self._port,
|
|
baudrate=self._baudrate,
|
|
timeout=0,
|
|
writeTimeout=1)
|
|
# deprecate function name which change at version 3.0
|
|
self._serial.flushInput()
|
|
self._serial.flushOutput()
|
|
|
|
def close(self):
|
|
if self._serial is not None and not self._serial.closed:
|
|
self._serial.close()
|
|
|
|
self._serial = None
|
|
|
|
def flush(self):
|
|
self._serial.flush()
|
|
|
|
def flush_input(self):
|
|
self._serial.flushInput()
|
|
def flush_output(self):
|
|
self._serial.flushOutput()
|
|
|
|
def send_byte(self, data: bytes):
|
|
"""send raw byte instruction to the hardware
|
|
|
|
:param data: raw byte instruction
|
|
:raises SerialTimeoutException:
|
|
"""
|
|
data_hex = ''.join(format(i, '02X') for i in data)
|
|
# print('send_byte: 0x{0}'.format(data_hex.upper()))
|
|
self._serial.write(data)
|
|
|
|
def recv_byte(self, size: int) -> Optional[bytes]:
|
|
"""receive raw byte data from hardware.
|
|
|
|
:param size: number of bytes want to receive
|
|
:return: raw byte data, None if fail.
|
|
"""
|
|
return self._serial.read(size)
|
|
|
|
|
|
def _main():
|
|
uart = UARTInterface('/dev/ttyAMA0')
|
|
uart.reset()
|
|
|
|
try:
|
|
while True:
|
|
try:
|
|
uart.send_byte("1234".encode())
|
|
except SerialTimeoutException:
|
|
pass
|
|
|
|
result = uart.recv_byte(4)
|
|
if result is not None:
|
|
print(result.decode())
|
|
|
|
sleep(1)
|
|
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
|
|
if __name__ == '__main__':
|
|
_main()
|