UniteLabs
package

unitelabs.bus

unitelabs/bus/__init__.py

Packages

Attributes

  • __all__
    = [ "unitelabs.bus.commands.byte_command.ByteCommand", "unitelabs.bus.commands.command.Command", "unitelabs.bus.commands.builder.CommandBuilder", "unitelabs.bus.exceptions.CommandExecutionError", "unitelabs.bus.exceptions.CommandValidationError", "unitelabs.bus.transports.hid_transport.HIDTransport", "unitelabs.bus.commands.http_command.HTTPCommand", "unitelabs.bus.exceptions.InputValidationError", "unitelabs.bus.exceptions.NoSuchDeviceFound", "unitelabs.bus.protocols.protocol.Protocol", "unitelabs.bus.commands.request.Request", "unitelabs.bus.commands.response.Response", "unitelabs.bus.commands.serial_command.SerialCommand", "unitelabs.bus.utils.device_manager.SerialDeviceManager", "unitelabs.bus.transports.serial_transport.SerialTransport", "unitelabs.bus.factories.transport_factory.TransportFactory", "unitelabs.bus.transports.usb_transport.USBTransport", "unitelabs.bus.factories.hid_factory.create_hid_connection", "unitelabs.bus.factories.serial_factory.create_serial_connection", "unitelabs.bus.factories.tcp_factory.create_tcp_connection", "unitelabs.bus.factories.udp_factory.create_udp_connection", "unitelabs.bus.factories.usb_factory.create_usb_connection" ]

Classes

  • ByteCommand

    class

    The most basic form of a `Command`, which ingests and returns bytes.

    MRO
    Bases
    Command[bytes, bytes]

    Methods

    • __init__(
        self, 
        message : bytes, 
        timeout : float | None, 
        is_void : bool | None, 
        **kwargs
      ) -> None

      message
      bytes
      timeout
      float | None = None
      is_void
      bool | None = False
      **kwargs
      = {}
    • _serialize(self, message : bytes | None) -> bytes

      message
      bytes | None = None
      bytes
    • _deserialize(self, response : bytes) -> bytes

      response
      bytes
      bytes

  • 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, the second type parameter determines the type returned by deserialization, and the third type parameter determines the type returned by the `result` method, e.g. `Command[str, list[str], list[bool]]` would ingest strings from the user, be converted to a list of strings for the parser and then return a list of booleans when used with `Protocol.execute`. Subclasses which change the constructor signature should be careful to preserve the `timeout` and `is_void` parameters. Removal of these arguments may result in compatability issues with the `CommandBuilder`.

    Bases
    typing.Generic[InType, OutType, ResType]

    Methods

    • __new__(
        cls, 
        *args, 
        is_void : bool, 
        **kwargs
      ) -> Command[InType, OutType, ResType] | Command[InType, None, None]

      *args
      = ()
      is_void
      bool = False
      **kwargs
      = {}
      Command[InType, OutType, ResType] | Command[InType, None, None]
    • __init__(
        self, 
        timeout : float | None, 
        is_void : bool
      ) -> None

      message
      InType
      The contents of the message to be sent to the device, pre-serialization.
      timeout
      float | None = None
      How long is seconds to wait for a response.
      is_void
      bool = False
      If true, does not return a response. Void commands ignore all response validations.
    • __repr__(self) -> str

      str
    • result(self) -> ResType

      Deserializes the `Response.payload.result()` bytes and consecutively applies `self.parsers`.

      ResType
      The deserialized `Response.payload` or None if `Command.is_void` is True.
    • serialize(self, message : InType | None) -> bytes

      Serialize the message into bytes. Uses `self.message` if `message` is None. NB: Calling with `message` DOES NOT SET the `self.message'` attribute; the `Request` sent to the device will use the `self.message` attribute.

      message
      InType | None = None
      A command input, or None to use `self.message`.
      bytes
      The serialized message to be sent to the device.
    • @abc.abstractmethod
      _serialize(self, message : InType) -> bytes

      User-configured serialization method.

      message
      InType
      bytes
    • deserialize(self, response : bytes | None) -> OutType

      Deserializes the `response` bytes. Calls `_deserialize` with `self.response` payload if `response` is None.

      response
      bytes | None = None
      bytes to be deserialized, or None to use `Response` payload.
      OutType
      None if `self.is_void`, else the deserialized `response`.
    • @abc.abstractmethod
      _deserialize(self, response : bytes) -> OutType

      User-configured deserialization method.

      response
      bytes
      OutType
    • validate_request(self, message : bytes) -> bool

      Validate a serialized message. Called within `Command.request` before generating a `Request` object.

      message
      bytes
      The serialized 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

      Set `Response.payload` if it belongs to this Command. 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 / '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

      User-configured response validation method. Validate the data received from the `Transport` and determine if the data is a complete response. Subclasses override this method to specify behavior for validating data before setting it as the response.

      data
      bytes
      The bytes to evaluate for completeness.
      bool
      True if `data` is a complete, valid message from the device, otherwise False.
    • match_response(self, data : bytes) -> bool

      Check if `data` belongs to this command and then validate the response. For devices that allow parallel command processing.

      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.
    • add_parser(self, parser : Parser[P, T]) -> Command[InType, OutType, T]

      Add a parser to the list of parsers applied when calling `result()`. Parsers are chained together in the order that they are added and are consecutively applied to the deserialized data based on the return value of the previous parser, meaning that any data filtered by earlier parsers will no longer be accessible to later parsers.

      parser
      Parser[P, T]
      A function that operates on the deserialized `result` (the first `Parser`), or the result of another `Parser` (all parsers after the first are chained together).
      Command[InType, OutType, T]
    • parse(self, data : OutType) -> ResType

      Apply `parsers` iteratively to the `data`.

      data
      OutType
      The data source which is fed into the parsers for processing.
      ResType

    Attributes

    • receiver
      typing.Optional[Protocol] = None
    • _parsers
      list[Parser] | None = None
    • _response
      Response | None = None
    • _request
      Request | None = 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.
    • parsers
      list[Parser] = None
      Parsers are functions that are applied to the deserialized data when calling `result()`.

  • CommandBuilder

    class

    Class that coordinates the chainable building of a `Command` instance.

    Bases
    typing.Generic[InType, OutType, R]

    Methods

    • __init__(self, command : type[Command[InType, OutType]], cls_name : str) -> None

      command
      type[Command[InType, OutType]] = ByteCommand
      The `Command` class to build on top of.
      cls_name
      str = 'BuiltCommand'
      The name of the `Command` subclass returned by `build`.
    • __copy__(self) -> CommandBuilder[InType, OutType, R]

      CommandBuilder[InType, OutType, R]
    • __deepcopy__(self, memo : dict | None) -> CommandBuilder[InType, OutType, R]

      memo
      dict | None = None
      CommandBuilder[InType, OutType, R]
    • @staticmethod
      create_command_method(func : typing.Callable[P, T], **kwargs) -> typing.Callable[typing.Concatenate[Command, P], T] | typing.Callable[P, T]

      Wrap a function into a command method.

      func
      typing.Callable[P, T]
      The function to format into a command method. Must be a callable with a final signature of (input) or (self, input) when partially loaded with `kwargs`.
      **kwargs
      = {}
      Keyword arguments to pass to the command method.
      typing.Callable[typing.Concatenate[Command, P], T] | typing.Callable[P, T]
      A wrapped, partially-loaded function with the final signature of (input) or (self, input).
    • with_serializer(
        self, 
        **kwargs
      ) -> CommandBuilder[S_InType, OutType, R]

      Set a serializer to apply to the `Command.message` before sending it to the device. The serializer will be called with the message to send to the device as the first and only argument. Additional arguments can be preloaded into the serializer with `**kwargs`.

      serializer
      Serializer[S_InType] | SelfSerializer[S_InType]
      The serializer function to apply.
      **kwargs
      = {}
      Keyword arguments to pass to the serializer.
      CommandBuilder[S_InType, OutType, R]
    • with_deserializer(
        self, 
        **kwargs
      ) -> CommandBuilder[InType, D_OutType, R]

      Set a deserializer function to apply to the `Command.response` recieved from the device. The deserializer will be called with the response bytes as the first and only argument. Additional arguments can be preloaded into the deserializer with `**kwargs`. This function will wrap the deserializer in a check for null values to ensure that calls without `response' use the `Response.payload.result()`.

      deserializer
      Deserializer[D_OutType] | SelfDeserializer[D_OutType]
      The deserializer function to apply.
      **kwargs
      = {}
      Keyword arguments to pass to the deserializer.
      CommandBuilder[InType, D_OutType, R]
    • with_parser(self, parser : Parser[P, T], **kwargs) -> CommandBuilder[InType, OutType, T]

      Add a partial parser function to the `Command`. Parsers are chained together consequetively from `Command._parsers` when calling `Command.result()`. The first parser receives the deserialized data and must pass data through to the next parser.

      parser
      Parser[P, T]
      The parser function to apply.
      **kwargs
      = {}
      Keyword arguments to pass to the parser.
      CommandBuilder[InType, OutType, T]
    • with_timeout(self, timeout : float) -> CommandBuilder[InType, OutType, R]

      Set the timeout to apply to the command. Alternative to setting `timeout` in `build`, will be overriden by `timeout` in `build`, if provided.

      timeout
      float
      The timeout to apply to the command.
      CommandBuilder[InType, OutType, R]
    • without_response(self) -> CommandBuilder[InType, None, None]

      Set the command to be a void command, i.e. return None as a response. Alternative to setting `is_void` in `build`, will be overriden by `is_void` in `build`, if provided.

      CommandBuilder[InType, None, None]
    • with_multiline(
        self, 
        timeout : float, 
        validate_response : typing.Callable[[], bool] | typing.Callable[[], bool] | None
      ) -> CommandBuilder[InType, OutType, R]

      Handle multi-line responses.

      timeout
      float
      The time in seconds to wait for more data before calling the `_validate_response` method.
      validate_response
      typing.Callable[[], bool] | typing.Callable[[], bool] | None = None
      The function to replace the `Command._validate_response` method.
      CommandBuilder[InType, OutType, R]
    • build(
        self, 
        *args, 
        timeout : float | None, 
        is_void : bool | None, 
        **kwargs
      ) -> Command[InType, OutType, R]

      Build the `Command` instance with the previously set serializer, deserializer and parsers.

      message
      InType
      The message to send to the device.
      *args
      = ()
      Additional positional arguments to pass to the command.
      timeout
      float | None = None
      The timeout to apply to the command.
      is_void
      bool | None = None
      Whether the command is a void command.
      **kwargs
      = {}
      Additional keyword arguments to pass to the command.
      Command[InType, OutType, R]
      A custom `Command` instance.

    Attributes

    • _command
      = command
    • _cmd_cls_name
      = cls_name
    • _serializer
      = None
    • _deserializer
      = None
    • _response_validator
      = None
    • _response_validator_timeout
      = None
    • _timeout
      = None
    • _is_void
      = False
    • _parsers
      list[Parser] = []

  • HTTPCommand

    class

    An HTTP Request.

    Methods

    • __init__(
        self, 
        message : bytes | None, 
        host : str | None, 
        path : str, 
        headers : dict[str, str] | None, 
        protocol_version : int, 
        timeout : float | None, 
        is_void : bool
      ) -> None

      message
      bytes | None = None
      The `body` of the request.
      host
      str | None = None
      The host of the remote server.
      path
      str = '/'
      The path of the HTTP request.
      method
      _Method = GET
      The type of HTTP request.
      headers
      dict[str, str] | None = None
      The headers of the HTTP request.
      protocol_version
      int = 11
      The protocol version to use. Either 10 or 11.
      timeout
      float | None = None
      How long in seconds to wait for a response.
      is_void
      bool = False
      Whether or not we should wait for a response at all.
    • _serialize(self, message : bytes) -> bytes

      message
      bytes
      bytes
    • _validate_response(self, data : bytes) -> bool

      data
      bytes
      bool
    • _deserialize(self, response : bytes) -> HTTPResponse

      response
      bytes
      HTTPResponse
    • _parse_http_response(self, data : bytes) -> HTTPResponse

      Read and parse the given binary data into an `HTTPResponse`.

      data
      bytes
      The binary data that contains an HTTP response message.
      HTTPResponse
      An `HTTPResponse` instance with its status code, headers and

    Attributes

    • Method
      type[_Method] = _Method
    • host
      = host
    • path
      = path
    • method
      = method
    • headers
      = headers or {}
    • protocol_version
      = protocol_version
    • http_response
      HTTPResponse | None = None

  • Request

    class

    Data that is sent to a `Transport`.

    Decorators
    dataclasses.dataclass

    Methods

    • __init__(self, payload : bytes, timeout : float | None) -> None

      payload
      bytes
      The payload of the request.
      timeout
      float | None = None
      The duration in seconds to wait for a response before raising a timeout exception. Set to 0.0 to wait indefinitely.
    • __post_init__(self) -> None

    Attributes

    • payload
      bytes = None
    • timeout
      float | None = 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

      Set a callback to be run when the payload `Future` becomes done.

      _payload
      asyncio.Future[bytes]
      The `Future` object.

    Attributes

  • SerialCommand

    class

    Command for use with serial communication device.

    MRO
    Bases
    Command[str, str]

    Methods

    • __init__(
        self, 
        message : str, 
        read_terminator : bytes, 
        write_terminator : bytes, 
        encoding : str, 
        **kwargs
      ) -> None

      message
      str
      the string message to send to the device.
      read_terminator
      bytes = b'\r\n'
      the byte-string expected at the end of messages coming from the device.
      write_terminator
      bytes = b'\r\n'
      the byte-string to append to our string `message` which indicates to the device that a complete message has been received.
      encoding
      str = 'ascii'
      The encoding used to convert between strings and bytes.
      **kwargs
      = {}
      Additional `Command` kwargs.
    • _serialize(self, message : str | None) -> bytes

      message
      str | None = None
      bytes
    • _deserialize(self, response : bytes | None) -> str

      response
      bytes | None
      str
    • _validate_response(self, data : bytes) -> bool

      data
      bytes
      bool

    Attributes

    • _read_terminator
      = read_terminator
    • _write_terminator
      = write_terminator
    • _encoding
      = encoding

  • CommandExecutionError

    class

    Command cannot be executed in current state.

    Bases
    Exception

  • CommandValidationError

    class

    Command input is invalid.

  • InputValidationError

    class

    The inputs given are invalid for the interface that received them.

    Bases
    ValueError

  • NoSuchDeviceFound

    class

    No matching device found.

    Bases
    Exception

  • TransportFactory

    class

    Interface representing a factory for creating transports.

    Bases
    typing.Protocol

    Methods

    • __call__(
        self, 
        protocol_factory : typing.Callable[..., P], 
        **kwargs
      ) -> tuple[asyncio.Transport, P]

      protocol_factory
      typing.Callable[..., P]
      **kwargs
      = {}
      tuple[asyncio.Transport, P]

  • Protocol

    class

    Base communication Protocol.

    Bases
    asyncio.Protocol

    Methods

    • __init__(
        self, 
        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 transports, but can be configured to allow more depending on the processing capacity of alternative transports.
      **kwargs
      = {}
      additional kwargs, including kwargs for use with `TransportFactory`.
    • connect(
        self, 
        **transport_kwargs
      ) -> tuple[TransportFactory, dict[str, typing.Any]]

      Open a connection to a new `Transport`. Closes the old connection, if it exists, before opening the new connection.

      transport_factory
      TransportFactory | None = None
      A callable used to create a connection to a transport, defaults to `self._transport_factory` set from the constructor.
      **transport_kwargs
      = {}
      Kwargs to pass into the `transport_factory`.
      tuple[TransportFactory, dict[str, typing.Any]]
      A tuple containing the `TransportFactory` and kwargs for the previously connected `Transport`.
    • _connect_transport(self, **kwargs) -> None

      Create a new transport instance.

      **kwargs
      = {}
    • identity(self, **config_kwargs) -> bool

      Validate 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

      Validate the opened connection. 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. If 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

      Respond to a connection made event invoked by the transport.

      transport
      asyncio.Transport
    • connection_lost(self, exc : Exception | None) -> None

      Respond to a lost connection event invoked by the transport. 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
      Exception | None = None
    • pause_writing(self) -> None

      Set state to disallow writing.

    • resume_writing(self) -> None

      Set state to allow writing.

    • data_received(self, data : bytes) -> None

      Receive data from the `Transport`. 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.
    • datagram_received(self, data : bytes, addr : tuple[str, int]) -> None

      Receive a datagram from the `Transport`.

      data
      bytes
      The datagram data received.
      addr
      tuple[str, int]
      The address of the sender.
    • error_received(self, exc : Exception | type[Exception]) -> None

      Receive an error from the `Transport`. Logs the error and sets the response if not already set. Further invocations with the same `Response` will only be logged.

      exc
      Exception | type[Exception]
      The error received.
    • execute(self, command : Command[InType, OutType, ResType]) -> ResType

      Execute a `Command` by sending the `Request` within the `Command` to the `Transport`.

      command
      Command[InType, OutType, ResType]
      The `Command` to be executed.
      ResType
      The deserialized response, created by `command.result()` or None if `Command.is_void` is True.

    Attributes

    • transport
      asyncio.Transport | None = None
    • _transport_allows_writing
      = True
    • _transport_factory
      = transport_factory
    • _transport_kwargs
      = kwargs
    • _commands
      list[Command] = []
    • lock
      = asyncio.BoundedSemaphore(value=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
      An `Autodetector` for connecting/re-connecting to certian devices.

  • HIDTransport

    class

    Transport for devices connected to an HID USB interface.

    Methods

    • __init__(
        self, 
        vendor : int, 
        product : int, 
        interface_index : int, 
        **kwargs
      ) -> None

      vendor
      int
      The vendor ID of the USB device.
      product
      int
      The product ID of the USB device.
      interface_index
      int = DEFAULT_USB_INTERFACE
      The index of the USB Interface to use. Defaults to 0.
      **kwargs
      = {}
      Additional configuration options: - serial_number: Specific device serial number to connect to - usage_page: HID usage page filter - usage: HID usage filter - write_report_id: HID report ID to use - read_report_id: HID read report ID to use
    • _open(self) -> None

    • _close(self) -> None

    • _ensure_reader(self) -> None

    • _remove_reader(self) -> None

    • __read(self) -> None

    • _read(self) -> bytes | None

      bytes | None
    • _ensure_writer(self) -> None

    • _remove_writer(self) -> None

    • _write(self, data : bytes) -> int

      data
      bytes
      int

    Attributes

    • read_size
      int = 64
    • vendor
      = vendor
    • product
      = product
    • interface_number
      = interface_index
    • serial_number
      = kwargs.get('serial_number')
    • usage_page
      = kwargs.get('usage_page')
    • usage
      = kwargs.get('usage')
    • write_report_id
      bytes = kwargs.get('write_report_id', b'')
    • read_report_id
      bytes = kwargs.get('read_report_id', b'')
    • _device
      hid.device | None = None
    • _was_opened
      = False
    • _reader_event
      = threading.Event()
    • logger
      logging.Logger = None
      A standard python logger.

  • SerialTransport

    class

    Transport for serial devices.

    Methods

    • __init__(
        self, 
        port : str, 
        baudrate : int, 
      ) -> 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

      Open underlying serial port, if not already open.

    • _close(self) -> None

      Close underlying serial port, if open.

    • _ensure_reader(self) -> None

    • _poll_read(self) -> None

    • _remove_reader(self) -> None

    • _read(self) -> bytes | None

      bytes | None
    • _ensure_writer(self) -> None

      Add a writer to the loop if not already added.

    • _poll_write(self) -> None

    • _remove_writer(self) -> None

      Remove a writer from the loop.

    • _write(self, data : bytes) -> int

      data
      bytes
      int

    Attributes

    • _serial
      = serial.serial_for_url(port, baudrate=baudrate, bytesize=bytesize, parity=parity, stopbits=stopbits, timeout=0, write_timeout=0, do_not_open=True)
    • _max_read_size
      = 1024
    • _read_buffer
      = []

  • USBTransport

    class

    Transport for devices connected via USB. By default, uses the `libusb` backend and Interface 0 of Configuration 1 on the device.

    Methods

    • __init__(
        self, 
        vendor : int, 
        product : int, 
        interface_index : int, 
        **kwargs
      ) -> None

      vendor
      int
      The vendor ID of the USB device.
      product
      int
      The product ID of the USB device.
      interface_index
      int = DEFAULT_USB_INTERFACE
      The index of the USB Interface to use. Defaults to 0.
      **kwargs
      = {}
      Additional backend-specific configuration options.
    • _open(self) -> None

      Open the USB device using the selected backend.

    • _close(self) -> None

    • _ensure_reader(self) -> None

    • _remove_reader(self) -> None

    • __read(self) -> None

    • _read(self) -> bytes | None

      Read data from the transport.

      bytes | None
      Data read from the device, if any.
    • _ensure_writer(self) -> None

    • _remove_writer(self) -> None

    • _write(self, data : bytes) -> int

      Write data to USB device. Write errors are caught by `WriteTransport._safe_write`, which wraps this method.

      data
      bytes
      Bytes to write.
      int
      Number of bytes written or 0 if the device connection is not currently available.

    Attributes

    • read_size
      int = None
    • vendor
      = vendor
    • product
      = product
    • interface_number
      = interface_index
    • _backend
      = None
    • _device
      usb.core.Device | None = None
    • read_endpoint
      usb.core.Endpoint | None = None
    • write_endpoint
      usb.core.Endpoint | None = None
    • _was_opened
      = False
    • _reader_event
      = threading.Event()
    • logger
      logging.Logger = None
      A standard python logger.

  • SerialDeviceManager

    class

    Detect, filter, and get info for connected serial devices.

    Methods

    • @typing.override
      @classmethod
      filter_kwargs(cls, kwargs : dict[str, str]) -> dict[str, str]

      kwargs
      dict[str, str]
      dict[str, str]
    • @typing.override
      @classmethod
      get_all(cls) -> list[DeviceInfo]

      list[DeviceInfo]
    • @typing.override
      @classmethod
      check_device_match(
        cls, 
        **kwargs
      ) -> bool

      device_info
      DeviceInfo
      **kwargs
      = {}
      bool
    • @typing.override
      @classmethod
      filter(cls, **kwargs) -> list[DeviceInfo]

      **kwargs
      = {}
      list[DeviceInfo]

    Attributes

    • SERIAL_SEARCH_KEYS
      typing_extensions.ClassVar = ['device', 'name', 'description', 'hwid', 'vid', 'pid', 'serial_number', 'location', 'manufacturer', 'product', 'interface']