unitelabs.bus.commands
Packages
Attributes
- Name
__all__- Type
- Value
= [ "Command", "InType_co", "OutType_co", "SerialCommand", "ByteCommand", "Request", "Response", "multiline" ]
- Description
- Name
InType_co- Type
- Value
= typing.TypeVar('InType_co', covariant=True)
- Description
- Name
OutType_co- Type
- Value
= typing.TypeVar('OutType_co', covariant=True)
- Description
Classes
ByteCommand
The most basic form of a `Command`, which ingests and returns bytes.
- MRO
- Bases
- Command[bytes, bytes]
Methods
__init__(self, command : bytes, timeout : typing.Optional[float]) -> None
Parameters
- Name
self- Type
- Default
- Description
- Name
command- Type
- bytes
- Default
- Description
- Name
timeout- Type
- typing.Optional[float]
- Default
- = None
- Description
serialize(self, message : typing.Optional[bytes]) -> bytes
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- typing.Optional[bytes]
- Default
- = None
- Description
Response
- Type
- bytes
- Description
deserialize(self, response : typing.Optional[bytes]) -> bytes
Parameters
- Name
self- Type
- Default
- Description
- Name
response- Type
- typing.Optional[bytes]
- Default
- = None
- Description
Response
- Type
- bytes
- Description
result(self) -> typing.Optional[OutType_co]
Deserializes the `Response.payload.result()` bytes.
Response
- Type
- typing.Optional[OutType_co]
- Description
The deserialized `Response.payload`
validate_request(self, message : bytes) -> bool
Validate a serialized message. Called within `Command.request` before generating a `Request` object.
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- bytes
- Default
- Description
The message to set as the `Request.payload`, if valid.
Response
- Type
- bool
- Description
Whether or not the `message` is valid.
_set_response(self) -> None
Set the result of `self.response.payload` to `self._response_buffer` and clears `self._response_buffer`.
validate_response(self, data : bytes) -> None
This method is called by `Protocol.data_received` and is responsible for setting the `Response.payload`. It manages the `_response_buffer` that accumulates the response bytes and calls `_validate_response` to determine whether the accumulated message in the `_response_buffer` is finished or 'valid'. If the response is valid, it sets the `Response.payload.result` to the accumulated bytes from the `_response_buffer`.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes from the `Transport` to add to the response_buffer and evaluate for completeness.
_validate_response(self, data : bytes) -> bool
Validate the data received from the `Transport` and determine if the data is a complete response. Subclasses override this method to specify behavior when validating a response.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes to evaluate for completeness.
Response
- Type
- bool
- Description
match_response(self, data : bytes) -> bool
For devices that allow parallel command processing, first check if `data` belongs to this command and then validate the response.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes to check for match during parallel processing, usually an identifier shared by request and response.
Response
- Type
- bool
- Description
True if the `data` matches to this command, otherwise False.
Attributes
- Name
receiver- Type
- typing.Optional[Protocol]
- Value
= None
- Description
- Name
_response- Type
- typing.Optional[Response]
- Value
= None
- Description
- Name
_request- Type
- typing.Optional[Request]
- Value
= None
- Description
- Name
message- Type
- Value
= message
- Description
- Name
timeout- Type
- Value
= timeout
- Description
- Name
_response_buffer- Type
- Value
= b''
- Description
- Name
is_void- Type
- Value
= is_void
- Description
- Name
request- Type
- Request
- Value
= None
- Description
The `Request` which will be used by the `Protocol` to send bytes to the device. Calls `validate_request` on the `command` before serializing it and creating the `Request` object.
- Name
response- Type
- Response
- Value
= None
- Description
The `Response` used by `Protocol.data_received` to set the `payload` of the command. When using the `Protocol.execute` method, the `Protocol` will call `validate_response` from within it's `Protocol.data_received` method and only set the result if valid.
Command
Generic Command that can be used with `Protocol.execute`. The first type parameter of the `Command` determines the type that the `Command` accepts on init and serialization, and the second type parameter determines the type returned by deserialization, e.g. `Command[str, typing.List[bool]]` would ingest strings and returns a list of booleans.
- Bases
- typing.Generic[InType_co, OutType_co]
Methods
__init__(
- self,
- message : InType_co,
- timeout : typing.Optional[float],
- is_void : typing.Optional[bool]
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- InType_co
- Default
- Description
The contents of the message to be sent to the device, pre-serialization.
- Name
timeout- Type
- typing.Optional[float]
- Default
- = None
- Description
How long is seconds to wait for a response.
- Name
is_void- Type
- typing.Optional[bool]
- Default
- = False
- Description
If true, does not return a response. Void commands ignore all response validations.
result(self) -> typing.Optional[OutType_co]
Deserializes the `Response.payload.result()` bytes.
Response
- Type
- typing.Optional[OutType_co]
- Description
The deserialized `Response.payload`
@abc.abstractmethod
serialize(self, message : typing.Optional[InType_co]) -> bytesSerializes the message into bytes. Should use `self.message` if `message` is None.
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- typing.Optional[InType_co]
- Default
- = None
- Description
A command input, or None to use `self.message`.
Response
- Type
- bytes
- Description
The serialized message to be sent to the device.
@abc.abstractmethod
deserialize(self, response : typing.Optional[bytes]) -> typing.Optional[OutType_co]Deserializes the `response` bytes. Should call `self.result()` if `response` is None.
Parameters
- Name
self- Type
- Default
- Description
- Name
response- Type
- typing.Optional[bytes]
- Default
- = None
- Description
bytes to be deserialized, or None to use `self.result()`.
Response
- Type
- typing.Optional[OutType_co]
- Description
The deserialized `response`.
validate_request(self, message : bytes) -> bool
Validate a serialized message. Called within `Command.request` before generating a `Request` object.
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- bytes
- Default
- Description
The message to set as the `Request.payload`, if valid.
Response
- Type
- bool
- Description
Whether or not the `message` is valid.
_set_response(self) -> None
Set the result of `self.response.payload` to `self._response_buffer` and clears `self._response_buffer`.
validate_response(self, data : bytes) -> None
This method is called by `Protocol.data_received` and is responsible for setting the `Response.payload`. It manages the `_response_buffer` that accumulates the response bytes and calls `_validate_response` to determine whether the accumulated message in the `_response_buffer` is finished or 'valid'. If the response is valid, it sets the `Response.payload.result` to the accumulated bytes from the `_response_buffer`.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes from the `Transport` to add to the response_buffer and evaluate for completeness.
_validate_response(self, data : bytes) -> bool
Validate the data received from the `Transport` and determine if the data is a complete response. Subclasses override this method to specify behavior when validating a response.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes to evaluate for completeness.
Response
- Type
- bool
- Description
match_response(self, data : bytes) -> bool
For devices that allow parallel command processing, first check if `data` belongs to this command and then validate the response.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes to check for match during parallel processing, usually an identifier shared by request and response.
Response
- Type
- bool
- Description
True if the `data` matches to this command, otherwise False.
Attributes
- Name
receiver- Type
- typing.Optional[Protocol]
- Value
= None
- Description
- Name
_response- Type
- typing.Optional[Response]
- Value
= None
- Description
- Name
_request- Type
- typing.Optional[Request]
- Value
= None
- Description
- Name
message- Type
- Value
= message
- Description
- Name
timeout- Type
- Value
= timeout
- Description
- Name
_response_buffer- Type
- Value
= b''
- Description
- Name
is_void- Type
- Value
= is_void
- Description
- Name
request- Type
- Request
- Value
= None
- Description
The `Request` which will be used by the `Protocol` to send bytes to the device. Calls `validate_request` on the `command` before serializing it and creating the `Request` object.
- Name
response- Type
- Response
- Value
= None
- Description
The `Response` used by `Protocol.data_received` to set the `payload` of the command. When using the `Protocol.execute` method, the `Protocol` will call `validate_response` from within it's `Protocol.data_received` method and only set the result if valid.
Request
Protocols use `Request`s to specify data that is sent to a transport.
- Decorators
- dataclasses.dataclass
Methods
__init__(self, payload : bytes, timeout : typing.Optional[float]) -> None
Parameters
- Name
self- Type
- Default
- Description
- Name
payload- Type
- bytes
- Default
- Description
- Name
timeout- Type
- typing.Optional[float]
- Default
- = None
- Description
__post_init__(self) -> None
Attributes
- Name
payload- Type
- bytes
- Value
= None
- Description
- Name
timeout- Type
- typing.Optional[float]
- Value
= None
- Description
Response
Protocols use `Response`s to specify data that is received from a transport.
- Decorators
- dataclasses.dataclass
Methods
__post_init__(self) -> None
__handle_done(self, _payload : asyncio.Future[bytes]) -> None
The callback to be run when the payload `Future` becomes done.
Parameters
- Name
self- Type
- Default
- Description
- Name
_payload- Type
- asyncio.Future[bytes]
- Default
- Description
The `Future` object.
Attributes
- Name
request- Type
- Request
- Value
= None
- Description
SerialCommand
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
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- str
- Default
- Description
the string message to send to the device.
- Name
read_terminator- Type
- bytes
- Default
- = b'\r\n'
- Description
the byte-string expected at the end of messages coming from the device.
- Name
write_terminator- Type
- bytes
- Default
- = b'\r\n'
- Description
the byte-string to append to our string `message` which indicates to the device that a complete message has been received.
- Name
encoding- Type
- str
- Default
- = 'ascii'
- Description
The encoding used to convert between strings and bytes.
- Name
**kwargs- Type
- Default
- = {}
- Description
Additional `Command` kwargs.
serialize(self, message : typing.Optional[str]) -> bytes
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- typing.Optional[str]
- Default
- = None
- Description
Response
- Type
- bytes
- Description
deserialize(self, response : typing.Optional[bytes]) -> str
Parameters
- Name
self- Type
- Default
- Description
- Name
response- Type
- typing.Optional[bytes]
- Default
- Description
Response
- Type
- str
- Description
_validate_response(self, data : bytes) -> bool
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
Response
- Type
- bool
- Description
result(self) -> typing.Optional[OutType_co]
Deserializes the `Response.payload.result()` bytes.
Response
- Type
- typing.Optional[OutType_co]
- Description
The deserialized `Response.payload`
validate_request(self, message : bytes) -> bool
Validate a serialized message. Called within `Command.request` before generating a `Request` object.
Parameters
- Name
self- Type
- Default
- Description
- Name
message- Type
- bytes
- Default
- Description
The message to set as the `Request.payload`, if valid.
Response
- Type
- bool
- Description
Whether or not the `message` is valid.
_set_response(self) -> None
Set the result of `self.response.payload` to `self._response_buffer` and clears `self._response_buffer`.
validate_response(self, data : bytes) -> None
This method is called by `Protocol.data_received` and is responsible for setting the `Response.payload`. It manages the `_response_buffer` that accumulates the response bytes and calls `_validate_response` to determine whether the accumulated message in the `_response_buffer` is finished or 'valid'. If the response is valid, it sets the `Response.payload.result` to the accumulated bytes from the `_response_buffer`.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes from the `Transport` to add to the response_buffer and evaluate for completeness.
match_response(self, data : bytes) -> bool
For devices that allow parallel command processing, first check if `data` belongs to this command and then validate the response.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes to check for match during parallel processing, usually an identifier shared by request and response.
Response
- Type
- bool
- Description
True if the `data` matches to this command, otherwise False.
Attributes
- Name
_read_terminator- Type
- Value
= read_terminator
- Description
- Name
_write_terminator- Type
- Value
= write_terminator
- Description
- Name
_encoding- Type
- Value
= encoding
- Description
- Name
receiver- Type
- typing.Optional[Protocol]
- Value
= None
- Description
- Name
_response- Type
- typing.Optional[Response]
- Value
= None
- Description
- Name
_request- Type
- typing.Optional[Request]
- Value
= None
- Description
- Name
message- Type
- Value
= message
- Description
- Name
timeout- Type
- Value
= timeout
- Description
- Name
_response_buffer- Type
- Value
= b''
- Description
- Name
is_void- Type
- Value
= is_void
- Description
- Name
request- Type
- Request
- Value
= None
- Description
The `Request` which will be used by the `Protocol` to send bytes to the device. Calls `validate_request` on the `command` before serializing it and creating the `Request` object.
- Name
response- Type
- Response
- Value
= None
- Description
The `Response` used by `Protocol.data_received` to set the `payload` of the command. When using the `Protocol.execute` method, the `Protocol` will call `validate_response` from within it's `Protocol.data_received` method and only set the result if valid.