unitelabs.bus.protocols.protocol
Attributes
- Name
InType_co- Type
- Value
= typing.TypeVar('InType_co', covariant=True)
- Description
- Name
OutType_co- Type
- Value
= typing.TypeVar('OutType_co', covariant=True)
- Description
- Name
DEFAULT_RECONNECT_DELAY- Type
- Value
= 20.0
- Description
- Name
DEFAULT_MAX_RECONNECT_ATTEMPTS- Type
- Value
= 3
- Description
- Name
DEFAULT_MAX_PARALLEL_COMMANDS- Type
- Value
= 1
- Description
Classes
Protocol
Base communication Protocol.
- Bases
- asyncio.Protocol
Methods
__init__(
- self,
- transport_factory : TransportFactory,
- reconnect : bool,
- reconnect_delay : float,
- max_reconnect_attempts : int,
- autodetect : bool,
- max_parallel_commands : int,
- **kwargs
Parameters
- Name
self- Type
- Default
- Description
- Name
transport_factory- Type
- TransportFactory
- Default
- Description
A callable used to create a connection to a transport.
- Name
reconnect- Type
- bool
- Default
- = True
- Description
Whether or not to a attempt to reconnect to a device when the connection is lost.
- Name
reconnect_delay- Type
- float
- Default
- = DEFAULT_RECONNECT_DELAY
- Description
How long in seconds to wait between reconnection attempts.
- Name
max_reconnect_attempts- Type
- int
- Default
- = DEFAULT_MAX_RECONNECT_ATTEMPTS
- Description
How many times to attempt to reconnect to a device before connection is considered lost.
- Name
autodetect- Type
- bool
- Default
- = False
- Description
Whether or not to use autodetection for device connectivity.
- Name
max_parallel_commands- Type
- int
- Default
- = DEFAULT_MAX_PARALLEL_COMMANDS
- Description
The maximum number of commands to process in parallel. This should be 1 for serial devices, but can be configured to allow more depending on the processing capacity of the usb device.
- Name
**kwargs- Type
- Default
- = {}
- Description
additional kwargs, including kwargs for use with `TransportFactory`.
_connect_transport(self, **kwargs) -> None
Create a new transport instance.
Parameters
- Name
self- Type
- Default
- Description
- Name
**kwargs- Type
- Default
- = {}
- Description
@abc.abstractmethod
identity(self, **config_kwargs) -> boolMethod for validating the identity of the connected device. This method will call another user-defined method on `Protocol` and compare the device's response (i.e. the method's return value) to values provided in `config_kwargs`.
Parameters
- Name
self- Type
- Default
- Description
- Name
**config_kwargs- Type
- Default
- = {}
- Description
kwargs sent from `validate`
Response
- Type
- bool
- Description
True if the result of the inner call matches the expectation from `config_kwargs` else False.
validate(
- self,
- timeout : float,
- **validation_kwargs
This method will be called by user after `__init__` via `open`; it calls `identity` to determine if the connected device is the one that was expected. If autodetect=True this is called internally by `unitelabs.bus.utils.AutoDetector` to cycle through possile devices until the correct device or no device is found. `validation_kwargs` and `__init__` kwargs are stored on the `Protocol` such that they must only be provided once. Should there be values which overlap, `__init__` values are overwritten by values in `validation_kwargs`. `validation_kwargs` may contain stable information about the device. Check `unitelabs.bus.utils.device_manager` for more information about valid device filter kwargs.
Parameters
- Name
self- Type
- Default
- Description
- Name
timeout- Type
- float
- Default
- Description
How long in seconds to wait for a response from the device.
- Name
**validation_kwargs- Type
- Default
- = {}
- Description
Kwargs to use to run validation of the device, in the case of use with `autodetect=True`, these kwargs will be stored on the first call for all future validations.
Response
- Type
- bool
- Description
Propagated return value from `identity`; True if `identity` returns True else False
open(
- self,
- validation_timeout : float,
- **validation_kwargs
Open underlying `Transport`, establish a connection to a device and validate the device's identity.
Parameters
- Name
self- Type
- Default
- Description
- Name
validation_timeout- Type
- float
- Default
- = 1.0
- Description
How long in seconds to wait for a response to `Protocol.validate`.
- Name
**validation_kwargs- Type
- Default
- = {}
- Description
kwargs to be passed to `Protocol.validate` to test device identity against.
close(self) -> None
Close underlying `Transport`. Explicitly calling `close` will NOT attempt to reconnect to the `Transport`.
connection_made(self, transport : asyncio.Transport) -> None
Invoked by `transport` when connection is made. Logs the connection.
Parameters
- Name
self- Type
- Default
- Description
- Name
transport- Type
- asyncio.Transport
- Default
- Description
connection_lost(self, exc : typing.Optional[Exception]) -> None
Invoked by transport when connection is lost. Attempts to reconnect after `reconnect_delay` seconds. Here `exc` can be None as a result of : - manual abort through direct call of transport's `abort` method - connection closing after `_safe_write` successfully wrote all data in write-buffer
Parameters
- Name
self- Type
- Default
- Description
- Name
exc- Type
- typing.Optional[Exception]
- Default
- = None
- Description
pause_writing(self) -> None
resume_writing(self) -> None
data_received(self, data : bytes) -> None
Invoked by transport when data is received. Logs the data and sets the response if not already set. Further invocations with the same `Response` will only be logged.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The data received.
error_received(self, exc : typing.Union[Exception, type[Exception]]) -> None
Invoked by transport when an error is received. Logs the error and sets the response if not already set. Further invocations with the same `Response` will only be logged.
Parameters
- Name
self- Type
- Default
- Description
- Name
exc- Type
- typing.Union[Exception, type[Exception]]
- Default
- Description
The error received.
execute(self, command : Command[InType_co, OutType_co]) -> typing.Optional[OutType_co]
Executes a `Command` by sending the `Request` within the `Command` to the `Transport`.
Parameters
- Name
self- Type
- Default
- Description
- Name
command- Type
- Command[InType_co, OutType_co]
- Default
- Description
The `Command` to be executed.
Response
- Type
- typing.Optional[OutType_co]
- Description
The deserialized response, created by `command.result()` or None if `Command.is_void` is True.
Exceptions
- Type
- RuntimeError
- Description
- If the transport is not open, does not currently allow writing, or is busy processing and cannot accept more work at the moment.
- Type
- TimeoutError
- Description
- If the response is not received within the specified timeout.
Attributes
- Name
_transport_allows_writing- Type
- Value
= True
- Description
- Name
_transport_factory- Type
- Value
= transport_factory
- Description
- Name
_transport_kwargs- Type
- Value
= kwargs
- Description
- Name
_commands- Type
- list[Command]
- Value
= []
- Description
- Name
max_parallel_commands- Type
- Value
= max_parallel_commands
- Description
- Name
is_open- Type
- Value
= asyncio.Event()
- Description
- Name
autodetect- Type
- Value
= autodetect
- Description
- Name
_autodetector- Type
- Value
= None
- Description
- Name
_validation_kwargs- Type
- Value
= None
- Description
- Name
is_validated- Type
- Value
= asyncio.Event()
- Description
- Name
reconnect- Type
- Value
= reconnect
- Description
- Name
reconnect_delay- Type
- Value
= reconnect_delay
- Description
- Name
remaining_reconnect_attempts- Type
- Value
= max_reconnect_attempts
- Description
- Name
max_reconnect_attempts- Type
- Value
= max_reconnect_attempts
- Description
- Name
logger- Type
- logging.Logger
- Value
= None
- Description
A standard python logger.
- Name
autodetector- Type
- AutoDetector
- Value
= None
- Description