unitelabs.bus
Packages
Attributes
- Name
__all__- Type
- Value
= [ "Protocol", "Command", "Request", "Response", "TransportFactory", "create_usb_connection", "create_serial_connection", "create_usb_connection", "SerialTransport", "UsbTransport", "SerialDeviceManager", "testing" ]
- Description
Functions
create_serial_connection(
- protocol_factory : typing.Callable[..., P],
- port : str,
- baudrate : int,
- bytesize : ByteSize,
- parity : Parity,
- stopbits : StopBits,
- **kwargs
Create a serial connection with the specified port.
Parameters
- Name
protocol_factory- Type
- typing.Callable[..., P]
- Default
- Description
A callable that returns an instance of the protocol to be used.
- Name
port- Type
- str
- Default
- Description
The port of the serial device.
- Name
baudrate- Type
- int
- Default
- = 9600
- Description
- Name
**kwargs- Type
- Default
- = {}
- Description
Additional keyword arguments to be passed to the `SerialTransport` constructor.
Response
create_usb_connection(
- protocol_factory : typing.Callable[..., P],
- vendor : int,
- product : int,
- **kwargs
Create a USB connection with a device based on specified `vendor` and `product` IDs.
Parameters
- Name
protocol_factory- Type
- typing.Callable[..., P]
- Default
- Description
A callable that returns an instance of the protocol to be used.
- Name
vendor- Type
- int
- Default
- Description
The vendor ID of the USB device.
- Name
product- Type
- int
- Default
- Description
The product ID of the USB device.
- Name
**kwargs- Type
- Default
- = {}
- Description
Additional keyword arguments to be passed to the `UsbTransport` constructor.
Response
Classes
Command
Generic Command that can be used with `Protocol.execute`. The first type parameter of the `Command` determines the type that the `Command` accepts on init and serialization, and the second type parameter determines the type returned by deserialization, e.g. `Command[str, typing.List[bool]]` would ingest strings and returns a list of booleans.
- Bases
- typing.Generic[InType_co, OutType_co]
Methods
__init__(
- self,
- message : InType_co,
- timeout : typing.Optional[float],
- is_void : typing.Optional[bool]
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- InType_co
- Default
- Description
The contents of the message to be sent to the device, pre-serialization.
- Name
timeout- Type
- typing.Optional[float]
- Default
- = None
- Description
How long is seconds to wait for a response.
- Name
is_void- Type
- typing.Optional[bool]
- Default
- = False
- Description
If true, does not return a response. Void commands ignore all response validations.
result(self) -> typing.Optional[OutType_co]
Deserializes the `Response.payload.result()` bytes.
Response
- Type
- typing.Optional[OutType_co]
- Description
The deserialized `Response.payload`
@abc.abstractmethod
serialize(self, message : typing.Optional[InType_co]) -> bytesSerializes the message into bytes. Should use `self.message` if `message` is None.
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- typing.Optional[InType_co]
- Default
- = None
- Description
A command input, or None to use `self.message`.
Response
- Type
- bytes
- Description
The serialized message to be sent to the device.
@abc.abstractmethod
deserialize(self, response : typing.Optional[bytes]) -> typing.Optional[OutType_co]Deserializes the `response` bytes. Should call `self.result()` if `response` is None.
Parameters
- Name
self- Type
- Default
- Description
- Name
response- Type
- typing.Optional[bytes]
- Default
- = None
- Description
bytes to be deserialized, or None to use `self.result()`.
Response
- Type
- typing.Optional[OutType_co]
- Description
The deserialized `response`.
validate_request(self, message : bytes) -> bool
Validate a serialized message. Called within `Command.request` before generating a `Request` object.
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- bytes
- Default
- Description
The message to set as the `Request.payload`, if valid.
Response
- Type
- bool
- Description
Whether or not the `message` is valid.
_set_response(self) -> None
Set the result of `self.response.payload` to `self._response_buffer` and clears `self._response_buffer`.
validate_response(self, data : bytes) -> None
This method is called by `Protocol.data_received` and is responsible for setting the `Response.payload`. It manages the `_response_buffer` that accumulates the response bytes and calls `_validate_response` to determine whether the accumulated message in the `_response_buffer` is finished or 'valid'. If the response is valid, it sets the `Response.payload.result` to the accumulated bytes from the `_response_buffer`.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes from the `Transport` to add to the response_buffer and evaluate for completeness.
_validate_response(self, data : bytes) -> bool
Validate the data received from the `Transport` and determine if the data is a complete response. Subclasses override this method to specify behavior when validating a response.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes to evaluate for completeness.
Response
- Type
- bool
- Description
match_response(self, data : bytes) -> bool
For devices that allow parallel command processing, first check if `data` belongs to this command and then validate the response.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes to check for match during parallel processing, usually an identifier shared by request and response.
Response
- Type
- bool
- Description
True if the `data` matches to this command, otherwise False.
Attributes
- Name
receiver- Type
- typing.Optional[Protocol]
- Value
= None
- Description
- Name
_response- Type
- typing.Optional[Response]
- Value
= None
- Description
- Name
_request- Type
- typing.Optional[Request]
- Value
= None
- Description
- Name
message- Type
- Value
= message
- Description
- Name
timeout- Type
- Value
= timeout
- Description
- Name
_response_buffer- Type
- Value
= b''
- Description
- Name
is_void- Type
- Value
= is_void
- Description
- Name
request- Type
- Request
- Value
= None
- Description
The `Request` which will be used by the `Protocol` to send bytes to the device. Calls `validate_request` on the `command` before serializing it and creating the `Request` object.
- Name
response- Type
- Response
- Value
= None
- Description
The `Response` used by `Protocol.data_received` to set the `payload` of the command. When using the `Protocol.execute` method, the `Protocol` will call `validate_response` from within it's `Protocol.data_received` method and only set the result if valid.
Request
Protocols use `Request`s to specify data that is sent to a transport.
- Decorators
- dataclasses.dataclass
Methods
__init__(self, payload : bytes, timeout : typing.Optional[float]) -> None
Parameters
- Name
self- Type
- Default
- Description
- Name
payload- Type
- bytes
- Default
- Description
- Name
timeout- Type
- typing.Optional[float]
- Default
- = None
- Description
__post_init__(self) -> None
Attributes
- Name
payload- Type
- bytes
- Value
= None
- Description
- Name
timeout- Type
- typing.Optional[float]
- Value
= None
- Description
Response
Protocols use `Response`s to specify data that is received from a transport.
- Decorators
- dataclasses.dataclass
Methods
__post_init__(self) -> None
__handle_done(self, _payload : asyncio.Future[bytes]) -> None
The callback to be run when the payload `Future` becomes done.
Parameters
- Name
self- Type
- Default
- Description
- Name
_payload- Type
- asyncio.Future[bytes]
- Default
- Description
The `Future` object.
Attributes
- Name
request- Type
- Request
- Value
= None
- Description
TransportFactory
Interface representing a factory for creating transports.
- Bases
- typing.Protocol
Methods
Protocol
Base communication Protocol.
- Bases
- asyncio.Protocol
Methods
__init__(
- self,
- transport_factory : TransportFactory,
- reconnect : bool,
- reconnect_delay : float,
- max_reconnect_attempts : int,
- autodetect : bool,
- max_parallel_commands : int,
- **kwargs
Parameters
- Name
self- Type
- Default
- Description
- Name
transport_factory- Type
- TransportFactory
- Default
- Description
A callable used to create a connection to a transport.
- Name
reconnect- Type
- bool
- Default
- = True
- Description
Whether or not to a attempt to reconnect to a device when the connection is lost.
- Name
reconnect_delay- Type
- float
- Default
- = DEFAULT_RECONNECT_DELAY
- Description
How long in seconds to wait between reconnection attempts.
- Name
max_reconnect_attempts- Type
- int
- Default
- = DEFAULT_MAX_RECONNECT_ATTEMPTS
- Description
How many times to attempt to reconnect to a device before connection is considered lost.
- Name
autodetect- Type
- bool
- Default
- = False
- Description
Whether or not to use autodetection for device connectivity.
- Name
max_parallel_commands- Type
- int
- Default
- = DEFAULT_MAX_PARALLEL_COMMANDS
- Description
The maximum number of commands to process in parallel. This should be 1 for serial devices, but can be configured to allow more depending on the processing capacity of the usb device.
- Name
**kwargs- Type
- Default
- = {}
- Description
additional kwargs, including kwargs for use with `TransportFactory`.
_connect_transport(self, **kwargs) -> None
Create a new transport instance.
Parameters
- Name
self- Type
- Default
- Description
- Name
**kwargs- Type
- Default
- = {}
- Description
@abc.abstractmethod
identity(self, **config_kwargs) -> boolMethod for validating the identity of the connected device. This method will call another user-defined method on `Protocol` and compare the device's response (i.e. the method's return value) to values provided in `config_kwargs`.
Parameters
- Name
self- Type
- Default
- Description
- Name
**config_kwargs- Type
- Default
- = {}
- Description
kwargs sent from `validate`
Response
- Type
- bool
- Description
True if the result of the inner call matches the expectation from `config_kwargs` else False.
validate(
- self,
- timeout : float,
- **validation_kwargs
This method will be called by user after `__init__` via `open`; it calls `identity` to determine if the connected device is the one that was expected. If autodetect=True this is called internally by `unitelabs.bus.utils.AutoDetector` to cycle through possile devices until the correct device or no device is found. `validation_kwargs` and `__init__` kwargs are stored on the `Protocol` such that they must only be provided once. Should there be values which overlap, `__init__` values are overwritten by values in `validation_kwargs`. `validation_kwargs` may contain stable information about the device. Check `unitelabs.bus.utils.device_manager` for more information about valid device filter kwargs.
Parameters
- Name
self- Type
- Default
- Description
- Name
timeout- Type
- float
- Default
- Description
How long in seconds to wait for a response from the device.
- Name
**validation_kwargs- Type
- Default
- = {}
- Description
Kwargs to use to run validation of the device, in the case of use with `autodetect=True`, these kwargs will be stored on the first call for all future validations.
Response
- Type
- bool
- Description
Propagated return value from `identity`; True if `identity` returns True else False
open(
- self,
- validation_timeout : float,
- **validation_kwargs
Open underlying `Transport`, establish a connection to a device and validate the device's identity.
Parameters
- Name
self- Type
- Default
- Description
- Name
validation_timeout- Type
- float
- Default
- = 1.0
- Description
How long in seconds to wait for a response to `Protocol.validate`.
- Name
**validation_kwargs- Type
- Default
- = {}
- Description
kwargs to be passed to `Protocol.validate` to test device identity against.
close(self) -> None
Close underlying `Transport`. Explicitly calling `close` will NOT attempt to reconnect to the `Transport`.
connection_made(self, transport : asyncio.Transport) -> None
Invoked by `transport` when connection is made. Logs the connection.
Parameters
- Name
self- Type
- Default
- Description
- Name
transport- Type
- asyncio.Transport
- Default
- Description
connection_lost(self, exc : typing.Optional[Exception]) -> None
Invoked by transport when connection is lost. Attempts to reconnect after `reconnect_delay` seconds. Here `exc` can be None as a result of : - manual abort through direct call of transport's `abort` method - connection closing after `_safe_write` successfully wrote all data in write-buffer
Parameters
- Name
self- Type
- Default
- Description
- Name
exc- Type
- typing.Optional[Exception]
- Default
- = None
- Description
pause_writing(self) -> None
resume_writing(self) -> None
data_received(self, data : bytes) -> None
Invoked by transport when data is received. Logs the data and sets the response if not already set. Further invocations with the same `Response` will only be logged.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The data received.
error_received(self, exc : typing.Union[Exception, type[Exception]]) -> None
Invoked by transport when an error is received. Logs the error and sets the response if not already set. Further invocations with the same `Response` will only be logged.
Parameters
- Name
self- Type
- Default
- Description
- Name
exc- Type
- typing.Union[Exception, type[Exception]]
- Default
- Description
The error received.
execute(self, command : Command[InType_co, OutType_co]) -> typing.Optional[OutType_co]
Executes a `Command` by sending the `Request` within the `Command` to the `Transport`.
Parameters
- Name
self- Type
- Default
- Description
- Name
command- Type
- Command[InType_co, OutType_co]
- Default
- Description
The `Command` to be executed.
Response
- Type
- typing.Optional[OutType_co]
- Description
The deserialized response, created by `command.result()` or None if `Command.is_void` is True.
Exceptions
- Type
- RuntimeError
- Description
- If the transport is not open, does not currently allow writing, or is busy processing and cannot accept more work at the moment.
- Type
- TimeoutError
- Description
- If the response is not received within the specified timeout.
Attributes
- Name
_transport_allows_writing- Type
- Value
= True
- Description
- Name
_transport_factory- Type
- Value
= transport_factory
- Description
- Name
_transport_kwargs- Type
- Value
= kwargs
- Description
- Name
_commands- Type
- list[Command]
- Value
= []
- Description
- Name
max_parallel_commands- Type
- Value
= max_parallel_commands
- Description
- Name
is_open- Type
- Value
= asyncio.Event()
- Description
- Name
autodetect- Type
- Value
= autodetect
- Description
- Name
_autodetector- Type
- Value
= None
- Description
- Name
_validation_kwargs- Type
- Value
= None
- Description
- Name
is_validated- Type
- Value
= asyncio.Event()
- Description
- Name
reconnect- Type
- Value
= reconnect
- Description
- Name
reconnect_delay- Type
- Value
= reconnect_delay
- Description
- Name
remaining_reconnect_attempts- Type
- Value
= max_reconnect_attempts
- Description
- Name
max_reconnect_attempts- Type
- Value
= max_reconnect_attempts
- Description
- Name
logger- Type
- logging.Logger
- Value
= None
- Description
A standard python logger.
- Name
autodetector- Type
- AutoDetector
- Value
= None
- Description
SerialTransport
Transport for serial devices.
- MRO
- ⎿⎽⎽⎽Transport
- ⎿⎽⎽⎽⎽SerialTransport
Methods
_open(self) -> None
Opens underlying serial port, if not already open.
_close(self) -> None
Closes underlying serial port, if open.
_ensure_reader(self) -> None
_poll_read(self) -> None
_remove_reader(self) -> None
_read(self) -> typing.Optional[bytes]
Response
- Type
- typing.Optional[bytes]
- Description
_ensure_writer(self) -> None
Adds a writer to the loop if not already added.
Exceptions
- Type
- serial.PortNotOpenError
- Description
- If the serial port is not open.
_poll_write(self) -> None
_remove_writer(self) -> None
Removes a writer from the loop.
_write(self, data : bytes) -> int
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
Response
- Type
- int
- Description
open(self) -> None
Opens the transport and sets state to allow future read operations.
close(self) -> None
Closes the transport and sets state to disallow further read operations.
is_reading(self) -> bool
Whether or not reading operations are currently being performed.
Response
- Type
- bool
- Description
True if the transport is receiving, otherwise False.
pause_reading(self) -> None
Pause the receiving end. No data will be passed to the protocol's `data_received()` method until `resume_reading()` is called.
resume_reading(self) -> None
Resume the receiving end. Data received will once again be passed to the protocol's `data_received()` method.
_abort(self, exception : typing.Optional[Exception]) -> None
Closes the transport immediately and updates state to disable further read operations.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- typing.Optional[Exception]
- Default
- = None
- Description
The Exception to propagate to the protocol when aborting, if connected.
_safe_read(self) -> None
Safely and asynchronously read data from the transport.
read_all(self) -> bytes
Read all available data from the transport. Repeatedly calls `_read` and aggregates the results until no further data is available.
Response
- Type
- bytes
- Description
All available data from the transport.
clear_read_buffer(self) -> None
Clears out all available read data without notifying the protocol. Calls `read_all` and throws away the result.
get_write_buffer_size(self) -> int
Calculate the current size of the write buffer.
Response
- Type
- int
- Description
The number of bytes in the write buffer.
get_write_buffer_limits(self) -> tuple[int, int]
Get the high and low watermarks for write flow control.
Response
- Type
- tuple[int, int]
- Description
a tuple (low, high) where low and high are positive number of bytes.
set_write_buffer_limits(self, high : typing.Optional[int], low : typing.Optional[int]) -> None
Set the high- and low-water limits for write flow control. These two values control when to call the protocol's `pause_writing()` and `resume_writing()` methods. If specified, the low-water limit must be less than or equal to the high-water limit. Neither value can be negative. The defaults are implementation-specific. If only the high-water limit is given, the low-water limit defaults to an implementation-specific value less than or equal to the high-water limit. Setting high to zero forces low to zero as well, and causes `pause_writing()` to be called whenever the buffer becomes non-empty. Setting low to zero causes `resume_writing()` to be called only once the buffer is empty. Use of zero for either limit is generally sub-optimal as it reduces opportunities for doing I/O and computation concurrently.
Parameters
- Name
self- Type
- Default
- Description
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The maximum allowed number of bytes in the write buffer.
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
The minimum allowed number of bytes in the write buffer.
write(self, data : typing.Union[bytes, bytearray, memoryview]) -> None
Write some data bytes to the transport. This does not block; it buffers the data and arranges for it to be sent out asynchronously.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- typing.Union[bytes, bytearray, memoryview]
- Default
- Description
The bytes to write to the Transport.
can_write_eof(self) -> bool
Whether or not this transport has implemented `write_eof()` method.
Response
- Type
- bool
- Description
True if this transport supports `write_eof()`, False if not.
write_eof(self) -> None
Close the write with end-of-file after flushing buffered data. (This is like typing ^D into a UNIX program reading from stdin.) Data may still be received.
writelines(self, list_of_data : typing.Iterable[typing.Union[bytes, bytearray, memoryview]]) -> None
Write a list (or any iterable) of data bytes to the transport. The default implementation concatenates the arguments and calls `write()` on the result.
Parameters
- Name
self- Type
- Default
- Description
- Name
list_of_data- Type
- typing.Iterable[typing.Union[bytes, bytearray, memoryview]]
- Default
- Description
The list of bytes to concatenate and write to the Transport.
flush(self) -> None
Flush the write buffer and disable further writing.
_safe_write(self) -> None
Asynchronously write buffered data. This method is called back asynchronously as a writer registered with the asyncio event-loop against the underlying file descriptor for the serial port. If this method is invoked while the transport is closing, and the write-buffer is then emptied by this method, the protocol's `connection_lost()` method will be called with None as its argument.
_maybe_pause_writing(self) -> None
To be called whenever the write-buffer size increases. Tests the current write-buffer size against the high water mark configured for this transport. If the high water mark is exceeded, the `Protocol` is instructed to `pause_writing()`.
_maybe_resume_protocol(self) -> None
To be called whenever the write-buffer size decreases. Tests the current write-buffer size against the low water mark configured for this transport. If writing is currently paused and the write-buffer size is below the low water mark, the `Protocol` is instructed to `resume_writing()`.
_set_write_buffer_limits(self, low : typing.Optional[int], high : typing.Optional[int]) -> None
Set the high- and low-water limits for write flow control. By default, the high-water limit is 4 times the high-water limit and if neither is specified, (16384, 65536).
Parameters
- Name
self- Type
- Default
- Description
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
The low-water limit for write flow control.
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The high-water limit for write flow control.
is_closing(self) -> bool
Whether the transport is closing or closed.
Response
- Type
- bool
- Description
True if the transport is closing or closed, False otherwise.
abort(self) -> None
Close the transport immediately.
_exception(
- self,
- exception : Exception,
- message : str
Report a fatal error to the event-loop and abort the transport.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- Exception
- Default
- Description
The Exception to pass on the the loop's exception handler.
- Name
message- Type
- str
- Default
- Description
Human-readable text describing the exception's execution state, cause, etc.
Attributes
- Name
_serial- Type
- Value
= _serial
- Description
- Name
_max_read_size- Type
- Value
= 1024
- Description
- Name
_read_buffer- Type
- Value
= []
- Description
- Name
_has_reader- Type
- Value
= False
- Description
- Name
_has_writer- Type
- Value
= False
- Description
- Name
_is_writing_paused- Type
- Value
= False
- Description
- Name
_write_buffer- Type
- list[typing.Union[bytes, bytearray, memoryview]]
- Value
= []
- Description
- Name
writes_pending- Type
- bool
- Value
= None
- Description
Whether or not there is data in the write buffer waiting to be written.
- Name
_loop- Type
- Value
= asyncio.get_event_loop_policy().get_event_loop()
- Description
- Name
_protocol- Type
- typing.Optional[P_co]
- Value
= None
- Description
- Name
_is_closing- Type
- Value
= True
- Description
UsbTransport
Transport for devices connected via USB. By default, this implementation uses Interface 0 of Configuration 1 on the device.
- MRO
- ⎿⎽⎽⎽Transport
- ⎿⎽⎽⎽⎽UsbTransport
Methods
__init__(
- self,
- vendor : int,
- product : int,
- protocol : asyncio.BaseProtocol,
- interface_index : int
Parameters
- Name
self- Type
- Default
- Description
- Name
vendor- Type
- int
- Default
- Description
The vendor ID of the USB device.
- Name
product- Type
- int
- Default
- Description
The product ID of the USB device.
- Name
protocol- Type
- asyncio.BaseProtocol
- Default
- Description
The protocol instance to use for communication.
- Name
interface_index- Type
- int
- Default
- = DEFAULT_USB_INTERFACE
- Description
The index of the USB Interface to use. Defaults to 0.
_open(self) -> None
_close(self) -> None
_ensure_reader(self) -> None
_remove_reader(self) -> None
__read(self) -> None
_read(self) -> None
Read data from the transport.
_ensure_writer(self) -> None
_remove_writer(self) -> None
_write(self, data) -> None
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- Default
- Description
open(self) -> None
Opens the transport and sets state to allow future read operations.
close(self) -> None
Closes the transport and sets state to disallow further read operations.
is_reading(self) -> bool
Whether or not reading operations are currently being performed.
Response
- Type
- bool
- Description
True if the transport is receiving, otherwise False.
pause_reading(self) -> None
Pause the receiving end. No data will be passed to the protocol's `data_received()` method until `resume_reading()` is called.
resume_reading(self) -> None
Resume the receiving end. Data received will once again be passed to the protocol's `data_received()` method.
_abort(self, exception : typing.Optional[Exception]) -> None
Closes the transport immediately and updates state to disable further read operations.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- typing.Optional[Exception]
- Default
- = None
- Description
The Exception to propagate to the protocol when aborting, if connected.
_safe_read(self) -> None
Safely and asynchronously read data from the transport.
read_all(self) -> bytes
Read all available data from the transport. Repeatedly calls `_read` and aggregates the results until no further data is available.
Response
- Type
- bytes
- Description
All available data from the transport.
clear_read_buffer(self) -> None
Clears out all available read data without notifying the protocol. Calls `read_all` and throws away the result.
get_write_buffer_size(self) -> int
Calculate the current size of the write buffer.
Response
- Type
- int
- Description
The number of bytes in the write buffer.
get_write_buffer_limits(self) -> tuple[int, int]
Get the high and low watermarks for write flow control.
Response
- Type
- tuple[int, int]
- Description
a tuple (low, high) where low and high are positive number of bytes.
set_write_buffer_limits(self, high : typing.Optional[int], low : typing.Optional[int]) -> None
Set the high- and low-water limits for write flow control. These two values control when to call the protocol's `pause_writing()` and `resume_writing()` methods. If specified, the low-water limit must be less than or equal to the high-water limit. Neither value can be negative. The defaults are implementation-specific. If only the high-water limit is given, the low-water limit defaults to an implementation-specific value less than or equal to the high-water limit. Setting high to zero forces low to zero as well, and causes `pause_writing()` to be called whenever the buffer becomes non-empty. Setting low to zero causes `resume_writing()` to be called only once the buffer is empty. Use of zero for either limit is generally sub-optimal as it reduces opportunities for doing I/O and computation concurrently.
Parameters
- Name
self- Type
- Default
- Description
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The maximum allowed number of bytes in the write buffer.
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
The minimum allowed number of bytes in the write buffer.
write(self, data : typing.Union[bytes, bytearray, memoryview]) -> None
Write some data bytes to the transport. This does not block; it buffers the data and arranges for it to be sent out asynchronously.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- typing.Union[bytes, bytearray, memoryview]
- Default
- Description
The bytes to write to the Transport.
can_write_eof(self) -> bool
Whether or not this transport has implemented `write_eof()` method.
Response
- Type
- bool
- Description
True if this transport supports `write_eof()`, False if not.
write_eof(self) -> None
Close the write with end-of-file after flushing buffered data. (This is like typing ^D into a UNIX program reading from stdin.) Data may still be received.
writelines(self, list_of_data : typing.Iterable[typing.Union[bytes, bytearray, memoryview]]) -> None
Write a list (or any iterable) of data bytes to the transport. The default implementation concatenates the arguments and calls `write()` on the result.
Parameters
- Name
self- Type
- Default
- Description
- Name
list_of_data- Type
- typing.Iterable[typing.Union[bytes, bytearray, memoryview]]
- Default
- Description
The list of bytes to concatenate and write to the Transport.
flush(self) -> None
Flush the write buffer and disable further writing.
_safe_write(self) -> None
Asynchronously write buffered data. This method is called back asynchronously as a writer registered with the asyncio event-loop against the underlying file descriptor for the serial port. If this method is invoked while the transport is closing, and the write-buffer is then emptied by this method, the protocol's `connection_lost()` method will be called with None as its argument.
_maybe_pause_writing(self) -> None
To be called whenever the write-buffer size increases. Tests the current write-buffer size against the high water mark configured for this transport. If the high water mark is exceeded, the `Protocol` is instructed to `pause_writing()`.
_maybe_resume_protocol(self) -> None
To be called whenever the write-buffer size decreases. Tests the current write-buffer size against the low water mark configured for this transport. If writing is currently paused and the write-buffer size is below the low water mark, the `Protocol` is instructed to `resume_writing()`.
_set_write_buffer_limits(self, low : typing.Optional[int], high : typing.Optional[int]) -> None
Set the high- and low-water limits for write flow control. By default, the high-water limit is 4 times the high-water limit and if neither is specified, (16384, 65536).
Parameters
- Name
self- Type
- Default
- Description
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
The low-water limit for write flow control.
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The high-water limit for write flow control.
is_closing(self) -> bool
Whether the transport is closing or closed.
Response
- Type
- bool
- Description
True if the transport is closing or closed, False otherwise.
abort(self) -> None
Close the transport immediately.
_exception(
- self,
- exception : Exception,
- message : str
Report a fatal error to the event-loop and abort the transport.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- Exception
- Default
- Description
The Exception to pass on the the loop's exception handler.
- Name
message- Type
- str
- Default
- Description
Human-readable text describing the exception's execution state, cause, etc.
Attributes
- Name
vendor- Type
- Value
= vendor
- Description
- Name
product- Type
- Value
= product
- Description
- Name
_closing- Type
- Value
= True
- Description
- Name
_protocol- Type
- Value
= protocol
- Description
- Name
_interface_index- Type
- Value
= interface_index
- Description
- Name
_device- Type
- typing.Optional[usb.core.Device]
- Value
= None
- Description
- Name
read_endpoint- Type
- typing.Optional[usb.core.Endpoint]
- Value
= None
- Description
- Name
write_endpoint- Type
- typing.Optional[usb.core.Endpoint]
- Value
= None
- Description
- Name
_reader_event- Type
- Value
= threading.Event()
- Description
- Name
_has_reader- Type
- Value
= False
- Description
- Name
_has_writer- Type
- Value
= False
- Description
- Name
_is_writing_paused- Type
- Value
= False
- Description
- Name
_write_buffer- Type
- list[typing.Union[bytes, bytearray, memoryview]]
- Value
= []
- Description
- Name
writes_pending- Type
- bool
- Value
= None
- Description
Whether or not there is data in the write buffer waiting to be written.
- Name
_loop- Type
- Value
= asyncio.get_event_loop_policy().get_event_loop()
- Description
- Name
_is_closing- Type
- Value
= True
- Description
SerialDeviceManager
Detect, filter, and get info for connected serial devices.
Methods
@classmethod
filter_kwargs(cls, kwargs : typing.Dict[str, str]) -> typing.Dict[str, str]Filter kwargs to those which are returned from `serial.tools.list_ports.comports`, i.e. the attributes of `DeviceInfo`. Supports the use of `port` as alternative name for `device`.
Parameters
- Name
cls- Type
- Default
- Description
- Name
kwargs- Type
- typing.Dict[str, str]
- Default
- Description
A dictionary of key-value pairs to filter.
Response
- Type
- typing.Dict[str, str]
- Description
The filtered dictionary.
@classmethod
get_all(cls) -> list[DeviceInfo]Get all connected devices.
Parameters
- Name
cls- Type
- Default
- Description
Response
- Type
- list[DeviceInfo]
- Description
A list of all devices detected.
@classmethod
check_device_match(- cls,
- device_info : DeviceInfo,
- **kwargs
Determine if the `DeviceInfo` instance's attributes match the filter `**kwargs`
Parameters
- Name
cls- Type
- Default
- Description
- Name
device_info- Type
- DeviceInfo
- Default
- Description
The device to check for a match against.
- Name
**kwargs- Type
- Default
- = {}
- Description
The key-value pairs, which will be filtered, and then used to evaluate the device for a match.
Response
- Type
- bool
- Description
True if the `DeviceInfo` matches (or no kwargs provided, or all kwargs have been filtered), else False.
@classmethod
filter(cls, **kwargs) -> list[DeviceInfo]Search through all detectable devices. `**kwargs` are first filtered by `filter_kwargs` based on attrs of `DeviceInfo`. This allows the number of devices detected to be pared down based on known, stable information about the device being searched for.
Parameters
- Name
cls- Type
- Default
- Description
- Name
**kwargs- Type
- Default
- = {}
- Description
Search criteria for finding a device.
Response
- Type
- list[DeviceInfo]
- Description
A list of devices which match all `**kwargs` provided.
Attributes
- Name
SERIAL_SEARCH_KEYS- Type
- Value
= ['device', 'name', 'description', 'hwid', 'vid', 'pid', 'serial_number', 'location', 'manufacturer', 'product', 'interface']
- Description