UniteLabs
package

unitelabs.bus

unitelabs/bus/__init__.py

Packages

Attributes

  • __all__
    = [ "Protocol", "Command", "CommandBuilder", "Request", "Response", "ByteCommand", "SerialCommand", "HTTPCommand", "TransportFactory", "create_serial_connection", "create_usb_connection", "create_udp_connection", "create_tcp_connection", "SerialTransport", "USBTransport", "SerialDeviceManager", "InputValidationError", "CommandValidationError", "CommandExecutionError", "NoSuchDeviceFound", "testing" ]

Functions

  • create_serial_connection(
      protocol_factory : typing.Callable[..., P], 
      port : str, 
      baudrate : int, 
      **kwargs
    ) -> tuple[SerialTransport, 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
    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.
    **kwargs
    = {}
    Additional keyword arguments to be passed to the `SerialTransport` constructor.
    tuple[SerialTransport, P]
    A tuple containing the `SerialTransport` instance and the `Protocol` instance.
  • create_tcp_connection(
      protocol_factory : typing.Callable[..., P], 
      host : str, 
      port : int, 
      **kwargs
    ) -> tuple[asyncio.Transport, P]

    Create and open a streaming TCP connection to a given address specified by host and port.

    protocol_factory
    typing.Callable[..., P]
    A callable that returns an instance if the `Protocol` to be used.
    host
    str = 'localhost'
    The host of the remote server.
    port
    int = 80
    The port number of the remote server.
    **kwargs
    = {}
    Additional keyword arguments passed to `loop.create_connection`.
    tuple[asyncio.Transport, P]
    A tuple containing the transport and protocol.
  • create_usb_connection(
      protocol_factory : typing.Callable[..., P], 
      vendor : int, 
      product : int, 
      **kwargs
    ) -> tuple[USBTransport, 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[USBTransport, P]
    A tuple containing the `UsbTransport` instance and the protocol instance.
  • create_udp_connection(
      protocol_factory : typing.Callable[..., P], 
      server_port : int, 
      server_host : str, 
      client_port : int, 
      client_host : str, 
      **kwargs
    ) -> tuple[asyncio.Transport, P]

    Create a UDP connection with the specified server (sender) and client (receiver) addresses.

    protocol_factory
    typing.Callable[..., P]
    A callable that returns an instance of the `Protocol` to be used.
    server_port
    int = 0
    The port of the UDP server.
    server_host
    str = ''
    The IP address of the UDP server.
    client_port
    int = 0
    The desired port of the client.
    client_host
    str = ''
    The desired IP address of the client.
    **kwargs
    = {}
    Additional keyword arguments to be passed to the `UDPTransport` constructor.
    tuple[asyncio.Transport, P]
    A tuple containing the `UDPTransport` instance and the `Protocol` instance.

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 : typing.Optional[float], 
        is_void : typing.Optional[bool], 
        **kwargs
      ) -> None

      message
      bytes
      timeout
      typing.Optional[float] = None
      is_void
      typing.Optional[bool] = False
      **kwargs
      = {}
    • _serialize(self, message : typing.Optional[bytes]) -> bytes

      message
      typing.Optional[bytes] = None
      bytes
    • _deserialize(self, response : bytes) -> bytes

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

      *args
      = ()
      is_void
      bool = False
      **kwargs
      = {}
      Command[InType, OutType, ResType][]
    • 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 : typing.Optional[InType]) -> bytes

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

      message
      typing.Optional[InType] = None
      A command input, or None to use `self.message`.
      bytes
      The serialized message to be sent to the device.
    • deserialize(self, response : typing.Optional[bytes]) -> OutType

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

      response
      typing.Optional[bytes] = None
      bytes to be deserialized, or None to use `Response` payload.
      OutType
      None if `self.is_void`, else 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 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

      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

      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.
    • 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 : ResType) -> ResType

      data
      ResType
      ResType

    Attributes

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

  • 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, typing.List[str], typing.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][]

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

      message
      InType
      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
      bool = False
      If true, does not return a response. Void commands ignore all response validations.
    • 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 : typing.Optional[InType]) -> bytes

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

      message
      typing.Optional[InType] = 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 : typing.Optional[bytes]) -> OutType

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

      response
      typing.Optional[bytes] = 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

      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

      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.
    • 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 : ResType) -> ResType

      data
      ResType
      ResType

    Attributes

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

  • CommandBuilder

    class

    Bases
    typing.Generic[InType, OutType, R]

    Methods

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

      Class that coordinates the chainable building of a `Command` instance. Args: command: The `Command` class to build on top of. cls_name: The name of the `Command` subclass returned by `build`. Examples: Call functions individually, replacing the builder instance >>> builder = CommandBuilder(ByteCommand) >>> builder = builder.with_serializer(lambda x: x + b" ") >>> builder = builder.with_deserializer(lambda x: x.strip(b" ")) >>> cmd = builder.build("message", timeout=0.5, is_void=True) CommandBuilder `with_` methods do not operate on themselves, but rather create and operate on a new instance of the CommandBuilder. Chain functions together >>> cmd = ( >>> CommandBuilder(ByteCommand) >>> .with_serializer(lambda x: x + b" ") >>> .with_deserializer(lambda x: x.strip(b" ")) >>> .build("message", timeout=0.5, is_void=True) >>> ) Create multiple commands from the same builder, or builder intermediate >>> crlf_builder = ( >>> CommandBuilder(ByteCommand) >>> .with_serializer(lambda x: x + b" ") >>> .with_deserializer(lambda x: x.strip(b" ")) >>> ) >>> cmd = crlf_builder.build("message", timeout=0.5) >>> cmd2 = crlf_builder.build("message2", timeout=0.5) >>> cmd3 = crlf_builder.with_serializer(lambda x: x).build(b"message3", timeout=0.5) Here `cmd` and `cmd2` both use the same serializer and deserializer methods, but `cmd3` uses a different serializer.

      command
      type[Command[InType, OutType]] = ByteCommand
      cls_name
      str = 'BuiltCommand'
    • __copy__(self) -> CommandBuilder[InType, OutType, R]

      CommandBuilder[InType, OutType, R]
    • __deepcopy__(self, memo) -> CommandBuilder[InType, OutType, R]

      memo
      = None
      CommandBuilder[InType, OutType, R]
    • @staticmethod
      create_command_method(func : typing.Callable[P, T], **kwargs) -> typing.Union[typing.Callable[typing_extensions.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.Union[typing.Callable[typing_extensions.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
      typing.Union[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
      typing.Union[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`, which will be applied, in order, to the deserialized data when calling `Command.result()`. Parsers are chained together consequetively 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.Optional[typing.Union[typing.Callable[[], bool], typing.Callable[[], bool]]]
      ) -> CommandBuilder[InType, OutType, R]

      Handle multi-line responses. Examples: Use the base class's `_validate_response` method. >>> builder = CommandBuilder().with_multiline(0.01) >>> cmd = builder.build(b"message") Provide a custom validation function. >>> builder = builder.with_multiline(0.01, validate_response=lambda x: x.endswith(b" ")) >>> cmd = builder.build(b"message") Args: timeout: The time in seconds to wait for more data before calling the `_validate_response` method. validate_response: The function to replace the `Command._validate_response` method. Raises: CommandValidationError: If the timeout is lower than 0.

      timeout
      float
      validate_response
      typing.Optional[typing.Union[typing.Callable[[], bool], typing.Callable[[], bool]]] = None
      CommandBuilder[InType, OutType, R]
    • build(
        self, 
        *args, 
        timeout : typing.Optional[float], 
        is_void : typing.Optional[bool], 
        **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
      typing.Optional[float] = None
      The timeout to apply to the command.
      is_void
      typing.Optional[bool] = 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 : typing.Optional[bytes], 
        host : typing.Optional[str], 
        path : str, 
        headers : typing.Optional[dict[str, str]], 
        protocol_version : int, 
        timeout : typing.Optional[float], 
        is_void : bool
      ) -> None

      message
      typing.Optional[bytes] = None
      The `body` of the request.
      host
      typing.Optional[str] = None
      The host of the remote server.
      path
      str = '/'
      The path of the HTTP request.
      method
      _Method = GET
      The type of HTTP request.
      headers
      typing.Optional[dict[str, str]] = None
      The headers of the HTTP request.
      protocol_version
      int = 11
      The protocol version to use. Either 10 or 11.
      timeout
      typing.Optional[float] = 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
    • __new__(
        cls, 
        *args, 
        is_void : bool, 
        **kwargs
      ) -> Command[InType, OutType, ResType][]

      *args
      = ()
      is_void
      bool = False
      **kwargs
      = {}
      Command[InType, OutType, ResType][]
    • 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 : typing.Optional[InType]) -> bytes

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

      message
      typing.Optional[InType] = None
      A command input, or None to use `self.message`.
      bytes
      The serialized message to be sent to the device.
    • deserialize(self, response : typing.Optional[bytes]) -> OutType

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

      response
      typing.Optional[bytes] = None
      bytes to be deserialized, or None to use `Response` payload.
      OutType
      None if `self.is_void`, else 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 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

      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.
    • 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.
    • 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 : ResType) -> ResType

      data
      ResType
      ResType

    Attributes

    • Method
      type[_Method] = _Method
    • host
      = host
    • path
      = path
    • method
      = method
    • headers
      = headers or {}
    • protocol_version
      = protocol_version
    • http_response
      typing.Optional[HTTPResponse] = None
    • receiver
      typing.Optional[Protocol] = None
    • _parsers
      typing.Optional[list[Parser]] = 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.
    • parsers
      list[Parser] = None
      Parsers are functions that are applied to the deserialized data when calling `result()`.

  • Request

    class

    `Protocol`s 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
      The payload of the request.
      timeout
      typing.Optional[float] = 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
      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

  • 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 : typing.Optional[str]) -> bytes

      message
      typing.Optional[str] = None
      bytes
    • _deserialize(self, response : typing.Optional[bytes]) -> str

      response
      typing.Optional[bytes]
      str
    • _validate_response(self, data : bytes) -> bool

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

      *args
      = ()
      is_void
      bool = False
      **kwargs
      = {}
      Command[InType, OutType, ResType][]
    • 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 : typing.Optional[InType]) -> bytes

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

      message
      typing.Optional[InType] = None
      A command input, or None to use `self.message`.
      bytes
      The serialized message to be sent to the device.
    • deserialize(self, response : typing.Optional[bytes]) -> OutType

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

      response
      typing.Optional[bytes] = None
      bytes to be deserialized, or None to use `Response` payload.
      OutType
      None if `self.is_void`, else 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 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

      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.
    • 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.
    • 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 : ResType) -> ResType

      data
      ResType
      ResType

    Attributes

    • _read_terminator
      = read_terminator
    • _write_terminator
      = write_terminator
    • _encoding
      = encoding
    • receiver
      typing.Optional[Protocol] = None
    • _parsers
      typing.Optional[list[Parser]] = 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.
    • parsers
      list[Parser] = None
      Parsers are functions that are applied to the deserialized data when calling `result()`.

  • CommandExecutionError

    class

    Bases
    Exception

  • CommandValidationError

    class

  • InputValidationError

    class

    Bases
    ValueError

  • NoSuchDeviceFound

    class

    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
      ) -> typing.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
      typing.Optional[TransportFactory] = 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`.
      typing.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

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

      Invoked by transport when a datagram is received.

      data
      bytes
      The datagram data received.
      addr
      tuple[str, int]
      The address of the sender.
    • 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, OutType, ResType]) -> ResType

      Executes 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
      typing.Optional[asyncio.Transport] = 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

  • 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

      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]

      Get the current `Protocol` associated with this transport.

      typing.Optional[P]
      The current `Protocol` instance.
    • set_protocol(self, protocol : P) -> None

      Associate a new `Protocol` with this transport.

      protocol
      P
      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] = 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.

    Methods

    • __init__(
        self, 
        vendor : int, 
        product : int, 
        interface_index : int
      ) -> 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.
    • _open(self) -> None

    • _close(self) -> None

    • _ensure_reader(self) -> None

    • _remove_reader(self) -> None

    • __read(self) -> None

    • _read(self) -> typing.Optional[bytes]

      Read data from the transport.

      typing.Optional[bytes]
    • _ensure_writer(self) -> None

    • _remove_writer(self) -> None

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

      data
      bytes
    • 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]

      Get the current `Protocol` associated with this transport.

      typing.Optional[P]
      The current `Protocol` instance.
    • set_protocol(self, protocol : P) -> None

      Associate a new `Protocol` with this transport.

      protocol
      P
      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
    • _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()
    • _protocol
      typing.Optional[P] = None
    • _is_closing
      = True

  • SerialDeviceManager

    class

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

    Methods

    • @classmethod
      filter_kwargs(cls, kwargs : dict[str, str]) -> 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
      dict[str, str]
      A dictionary of key-value pairs to filter.
      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, 
        **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']