hat.drivers.serial

15class ByteSize(enum.Enum):
16    FIVEBITS = 5
17    SIXBITS = 6
18    SEVENBITS = 7
19    EIGHTBITS = 8

22class Parity(enum.Enum):
23    NONE = 'N'
24    EVEN = 'E'
25    ODD = 'O'
26    MARK = 'M'
27    SPACE = 'S'

30class StopBits(enum.Enum):
31    ONE = 1
32    ONE_POINT_FIVE = 1.5
33    TWO = 2

36class EndpointInfo(typing.NamedTuple):
37    name: str | None
38    port: str

41class Endpoint(aio.Resource):
42    """Serial endpoint"""
43
44    @property
45    @abc.abstractmethod
46    def info(self) -> EndpointInfo:
47        """Endpoint info"""
48
49    @abc.abstractmethod
50    async def read(self, size: int) -> util.Bytes:
51        """Read
52
53        Args:
54            size: number of bytes to read
55
56        Raises:
57            ConnectionError
58
59        """
60
61    @abc.abstractmethod
62    async def write(self, data: util.Bytes):
63        """Write
64
65        Raises:
66            ConnectionError
67
68        """
69
70    @abc.abstractmethod
71    async def drain(self):
72        """Drain output buffer
73
74        Raises:
75            ConnectionError
76
77        """
78
79    @abc.abstractmethod
80    async def clear_input_buffer(self) -> int:
81        """Reset input buffer
82
83        Returns number of bytes available in buffer immediately before
84        buffer was cleared.
85
86        Raises:
87            ConnectionError
88
89        """

29async def create(port: str,
30                 *,
31                 name: str | None = None,
32                 baudrate: int = 9600,
33                 bytesize: ByteSize = ByteSize.EIGHTBITS,
34                 parity: Parity = Parity.NONE,
35                 stopbits: StopBits = StopBits.ONE,
36                 xonxoff: bool = False,
37                 rtscts: bool = False,
38                 dsrdtr: bool = False,
39                 silent_interval: float = 0
40                 ) -> Endpoint:
41    """Open serial port
42
43    Args:
44        port: port name dependent of operating system
45            (e.g. `/dev/ttyUSB0`, `COM3`, ...)
46        name: endpoint name
47        baudrate: baud rate
48        bytesize: number of data bits
49        parity: parity checking
50        stopbits: number of stop bits
51        xonxoff: enable software flow control
52        rtscts: enable hardware RTS/CTS flow control
53        dsrdtr: enable hardware DSR/DTR flow control
54        silent_interval: minimum time in seconds between writing two
55            consecutive messages
56
57    """
58    impl = native_serial or py_serial
59    return await impl.create(port=port,
60                             name=name,
61                             baudrate=baudrate,
62                             bytesize=bytesize,
63                             parity=parity,
64                             stopbits=stopbits,
65                             xonxoff=xonxoff,
66                             rtscts=rtscts,
67                             dsrdtr=dsrdtr,
68                             silent_interval=silent_interval)