package

unitelabs.bus

unitelabs/bus/__init__.py

Attributes

__all__
= [ "Protocol", "Command", "Request", "Response", "TransportFactory", "create_usb_connection", "create_serial_connection", "create_usb_connection", "SerialTransport", "UsbTransport", "SerialDeviceManager", "testing" ]

Functions

```
create_serial_connection(protocol_factory : typing.Callable[..., P], 
port : str, 
baudrate : int, 
bytesize : ByteSize, 
parity : Parity, 
stopbits : StopBits, 
**kwargs
) -> tuple[Transport, P]
```
Create a serial connection with the specified port.
protocol_factory
typing.Callable[..., P]
A callable that returns an instance of the protocol to be used.
port
str
The port of the serial device.
baudrate
int = 9600
bytesize
ByteSize = EIGHTBITS
parity
Parity = NONE
stopbits
StopBits = ONE
**kwargs
= {}
Additional keyword arguments to be passed to the `SerialTransport` constructor.
tuple[Transport, P]
A tuple containing the `SerialTransport` instance and the `Protocol` instance.
```
create_usb_connection(protocol_factory : typing.Callable[..., P], 
vendor : int, 
product : int, 
**kwargs
) -> tuple[Transport, P]
```
Create a USB connection with a device based on specified `vendor` and `product` IDs.
protocol_factory
typing.Callable[..., P]
A callable that returns an instance of the protocol to be used.
vendor
int
The vendor ID of the USB device.
product
int
The product ID of the USB device.
**kwargs
= {}
Additional keyword arguments to be passed to the `UsbTransport` constructor.
tuple[Transport, P]
A tuple containing the `UsbTransport` instance and the protocol instance.

Classes

Command
class
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]
  ) -> None
  message
  InType_co
  The contents of the message to be sent to the device, pre-serialization.
  timeout
  typing.Optional[float] = None
  How long is seconds to wait for a response.
  is_void
  typing.Optional[bool] = False
  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.
  typing.Optional[OutType_co]
  The deserialized `Response.payload`
- @abc.abstractmethod
  serialize(self, message : typing.Optional[InType_co]) -> bytes
  Serializes the message into bytes. Should use `self.message` if `message` is None.
  message
  typing.Optional[InType_co] = None
  A command input, or None to use `self.message`.
  bytes
  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.
  response
  typing.Optional[bytes] = None
  bytes to be deserialized, or None to use `self.result()`.
  typing.Optional[OutType_co]
  The deserialized `response`.
- validate_request(self, message : bytes) -> bool
  Validate a serialized message. Called within `Command.request` before generating a `Request` object.
  message
  bytes
  The message to set as the `Request.payload`, if valid.
  bool
  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`.
  data
  bytes
  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.
  data
  bytes
  The bytes to evaluate for completeness.
  bool
- 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.
  data
  bytes
  The bytes to check for match during parallel processing, usually an identifier shared by request and response.
  bool
  True if the `data` matches to this command, otherwise False.
Attributes
- receiver
  typing.Optional[Protocol] = None
- _response
  typing.Optional[Response] = None
- _request
  typing.Optional[Request] = None
- message
  = message
- timeout
  = timeout
- _response_buffer
  = b''
- is_void
  = is_void
- request
  Request = None
  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.
- response
  Response = None
  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
class
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
  payload
  bytes
  timeout
  typing.Optional[float] = None
- __post_init__(self) -> None
Attributes
- payload
  bytes = None
- timeout
  typing.Optional[float] = None
Response
class
Protocols use `Response`s to specify data that is received from a transport.
Decorators
dataclasses.dataclass
Methods
- __init__(self, request : Request) -> None
  request
  Request
- __post_init__(self) -> None
- __handle_done(self, _payload : asyncio.Future[bytes]) -> None
  The callback to be run when the payload `Future` becomes done.
  _payload
  asyncio.Future[bytes]
  The `Future` object.
Attributes
- request
  Request = None
TransportFactory
class
Interface representing a factory for creating transports.
Bases
typing.Protocol
Methods
- __call__(
  self,
  protocol_factory : typing.Callable[..., P],
  **kwargs
  ) -> tuple[Transport, P]
  protocol_factory
  typing.Callable[..., P]
  **kwargs
  = {}
  tuple[Transport, P]
Protocol
class
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
  ) -> None
  transport_factory
  TransportFactory
  A callable used to create a connection to a transport.
  reconnect
  bool = True
  Whether or not to a attempt to reconnect to a device when the connection is lost.
  reconnect_delay
  float = DEFAULT_RECONNECT_DELAY
  How long in seconds to wait between reconnection attempts.
  max_reconnect_attempts
  int = DEFAULT_MAX_RECONNECT_ATTEMPTS
  How many times to attempt to reconnect to a device before connection is considered lost.
  autodetect
  bool = False
  Whether or not to use autodetection for device connectivity.
  max_parallel_commands
  int = DEFAULT_MAX_PARALLEL_COMMANDS
  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.
  **kwargs
  = {}
  additional kwargs, including kwargs for use with `TransportFactory`.
- _connect_transport(self, **kwargs) -> None
  Create a new transport instance.
  **kwargs
  = {}
- @abc.abstractmethod
  identity(self, **config_kwargs) -> bool
  Method 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`.
  **config_kwargs
  = {}
  kwargs sent from `validate`
  bool
  True if the result of the inner call matches the expectation from `config_kwargs` else False.
- validate(
  self,
  timeout : float,
  **validation_kwargs
  ) -> bool
  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.
  timeout
  float
  How long in seconds to wait for a response from the device.
  **validation_kwargs
  = {}
  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.
  bool
  Propagated return value from `identity`; True if `identity` returns True else False
- open(
  self,
  validation_timeout : float,
  **validation_kwargs
  ) -> None
  Open underlying `Transport`, establish a connection to a device and validate the device's identity.
  validation_timeout
  float = 1.0
  How long in seconds to wait for a response to `Protocol.validate`.
  **validation_kwargs
  = {}
  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.
  transport
  asyncio.Transport
- 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
  exc
  typing.Optional[Exception] = None
- 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.
  data
  bytes
  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.
  exc
  typing.Union[Exception, type[Exception]]
  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`.
  command
  Command[InType_co, OutType_co]
  The `Command` to be executed.
  typing.Optional[OutType_co]
  The deserialized response, created by `command.result()` or None if `Command.is_void` is True.
Attributes
- _transport_allows_writing
  = True
- _transport_factory
  = transport_factory
- _transport_kwargs
  = kwargs
- _commands
  list[Command] = []
- max_parallel_commands
  = max_parallel_commands
- is_open
  = asyncio.Event()
- autodetect
  = autodetect
- _autodetector
  = None
- _validation_kwargs
  = None
- is_validated
  = asyncio.Event()
- reconnect
  = reconnect
- reconnect_delay
  = reconnect_delay
- remaining_reconnect_attempts
  = max_reconnect_attempts
- max_reconnect_attempts
  = max_reconnect_attempts
- logger
  logging.Logger = None
  A standard python logger.
- autodetector
  AutoDetector = None
SerialTransport
class
Transport for serial devices.
MRO
BaseTransport
└ WriteTransport
└─ ReadTransport
└── Transport
└─── SerialTransport
Methods
- __init__(
  self,
  port : str,
  baudrate : int,
  bytesize : ByteSize,
  parity : Parity,
  stopbits : StopBits
  ) -> None
  port
  str
  The port where the serial device is connected.
  baudrate
  int = 9600
  The baud rate.
  bytesize
  ByteSize = EIGHTBITS
  Number of data bits.
  parity
  Parity = NONE
  Parity checking method for error-detection.
  stopbits
  StopBits = ONE
  The number of stopbits.
- _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]
  typing.Optional[bytes]
- _ensure_writer(self) -> None
  Adds a writer to the loop if not already added.
- _poll_write(self) -> None
- _remove_writer(self) -> None
  Removes a writer from the loop.
- _write(self, data : bytes) -> int
  data
  bytes
  int
- 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.
  bool
  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.
  exception
  typing.Optional[Exception] = None
  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.
  bytes
  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.
  int
  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.
  tuple[int, int]
  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.
  high
  typing.Optional[int] = None
  The maximum allowed number of bytes in the write buffer.
  low
  typing.Optional[int] = None
  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.
  data
  typing.Union[bytes, bytearray, memoryview]
  The bytes to write to the Transport.
- can_write_eof(self) -> bool
  Whether or not this transport has implemented `write_eof()` method.
  bool
  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.
  list_of_data
  typing.Iterable[typing.Union[bytes, bytearray, memoryview]]
  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).
  low
  typing.Optional[int] = None
  The low-water limit for write flow control.
  high
  typing.Optional[int] = None
  The high-water limit for write flow control.
- get_protocol(self) -> typing.Optional[P_co]
  Get the current `Protocol` associated with this transport.
  typing.Optional[P_co]
  The current `Protocol` instance.
- set_protocol(self, protocol : P_co) -> None
  Associate a new `Protocol` with this transport.
  protocol
  P_co
  The new `Protocol` instance.
- is_closing(self) -> bool
  Whether the transport is closing or closed.
  bool
  True if the transport is closing or closed, False otherwise.
- abort(self) -> None
  Close the transport immediately.
- _exception(
  self,
  exception : Exception,
  message : str
  ) -> None
  Report a fatal error to the event-loop and abort the transport.
  exception
  Exception
  The Exception to pass on the the loop's exception handler.
  message
  str
  Human-readable text describing the exception's execution state, cause, etc.
Attributes
- _serial
  = _serial
- _max_read_size
  = 1024
- _read_buffer
  = []
- _has_reader
  = False
- _has_writer
  = False
- _is_writing_paused
  = False
- _write_buffer
  list[typing.Union[bytes, bytearray, memoryview]] = []
- writes_pending
  bool = None
  Whether or not there is data in the write buffer waiting to be written.
- _loop
  = asyncio.get_event_loop_policy().get_event_loop()
- _protocol
  typing.Optional[P_co] = None
- _is_closing
  = True
UsbTransport
class
Transport for devices connected via USB. By default, this implementation uses Interface 0 of Configuration 1 on the device.
MRO
BaseTransport
└ WriteTransport
└─ ReadTransport
└── Transport
└─── UsbTransport
Methods
- __init__(
  self,
  vendor : int,
  product : int,
  protocol : asyncio.BaseProtocol,
  interface_index : int
  ) -> None
  vendor
  int
  The vendor ID of the USB device.
  product
  int
  The product ID of the USB device.
  protocol
  asyncio.BaseProtocol
  The protocol instance to use for communication.
  interface_index
  int = DEFAULT_USB_INTERFACE
  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
  data
- 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.
  bool
  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.
  exception
  typing.Optional[Exception] = None
  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.
  bytes
  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.
  int
  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.
  tuple[int, int]
  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.
  high
  typing.Optional[int] = None
  The maximum allowed number of bytes in the write buffer.
  low
  typing.Optional[int] = None
  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.
  data
  typing.Union[bytes, bytearray, memoryview]
  The bytes to write to the Transport.
- can_write_eof(self) -> bool
  Whether or not this transport has implemented `write_eof()` method.
  bool
  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.
  list_of_data
  typing.Iterable[typing.Union[bytes, bytearray, memoryview]]
  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).
  low
  typing.Optional[int] = None
  The low-water limit for write flow control.
  high
  typing.Optional[int] = None
  The high-water limit for write flow control.
- get_protocol(self) -> typing.Optional[P_co]
  Get the current `Protocol` associated with this transport.
  typing.Optional[P_co]
  The current `Protocol` instance.
- set_protocol(self, protocol : P_co) -> None
  Associate a new `Protocol` with this transport.
  protocol
  P_co
  The new `Protocol` instance.
- is_closing(self) -> bool
  Whether the transport is closing or closed.
  bool
  True if the transport is closing or closed, False otherwise.
- abort(self) -> None
  Close the transport immediately.
- _exception(
  self,
  exception : Exception,
  message : str
  ) -> None
  Report a fatal error to the event-loop and abort the transport.
  exception
  Exception
  The Exception to pass on the the loop's exception handler.
  message
  str
  Human-readable text describing the exception's execution state, cause, etc.
Attributes
- vendor
  = vendor
- product
  = product
- _closing
  = True
- _protocol
  = protocol
- _interface_index
  = interface_index
- _device
  typing.Optional[usb.core.Device] = None
- read_endpoint
  typing.Optional[usb.core.Endpoint] = None
- write_endpoint
  typing.Optional[usb.core.Endpoint] = None
- _reader_event
  = threading.Event()
- _has_reader
  = False
- _has_writer
  = False
- _is_writing_paused
  = False
- _write_buffer
  list[typing.Union[bytes, bytearray, memoryview]] = []
- writes_pending
  bool = None
  Whether or not there is data in the write buffer waiting to be written.
- _loop
  = asyncio.get_event_loop_policy().get_event_loop()
- _is_closing
  = True
SerialDeviceManager
class
Detect, filter, and get info for connected serial devices.
MRO
DeviceManager
└ SerialDeviceManager
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`.
  kwargs
  typing.Dict[str, str]
  A dictionary of key-value pairs to filter.
  typing.Dict[str, str]
  The filtered dictionary.
- @classmethod
  get_all(cls) -> list[DeviceInfo]
  Get all connected devices.
  list[DeviceInfo]
  A list of all devices detected.
- @classmethod
  check_device_match(
  cls,
  device_info : DeviceInfo,
  **kwargs
  ) -> bool
  Determine if the `DeviceInfo` instance's attributes match the filter `**kwargs`
  device_info
  DeviceInfo
  The device to check for a match against.
  **kwargs
  = {}
  The key-value pairs, which will be filtered, and then used to evaluate the device for a match.
  bool
  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.
  **kwargs
  = {}
  Search criteria for finding a device.
  list[DeviceInfo]
  A list of devices which match all `**kwargs` provided.
Attributes
- SERIAL_SEARCH_KEYS
  = ['device', 'name', 'description', 'hwid', 'vid', 'pid', 'serial_number', 'location', 'manufacturer', 'product', 'interface']

unitelabs.bus

Packages

Attributes

Functions

Classes

Command

Methods

Attributes

Request

Methods

Attributes

Response

Methods

Attributes

TransportFactory

Methods

Protocol

Methods

Attributes

SerialTransport

Methods

Attributes

UsbTransport

Methods

Attributes

SerialDeviceManager

Methods

Attributes