hat.drivers.serial

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

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

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

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

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

93def endpoint_info_to_json(info: EndpointInfo) -> json.Data:
94    return {'name': info.name,
95            'port': info.port}

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