unitelabs.bus.transports
Packages
Attributes
- Name
__all__- Type
- Value
= [ "BaseTransport", "ReadTransport", "WriteTransport", "Transport", "SerialTransport", "ByteSize", "Parity", "StopBits", "UsbTransport" ]
- Description
Classes
BaseTransport
- Bases
- asyncio.BaseTransport, typing.Generic[P_co]
Methods
__init__(self) -> None
is_closing(self) -> bool
Whether the transport is closing or closed.
Response
- Type
- bool
- Description
True if the transport is closing or closed, False otherwise.
open(self) -> None
Open the transport and inform the `Protocol` whether the connection was established successfully.
_open(self) -> None
Underlying action performed when calling `open`. Subclasses override this method to specify behavior when opening the connection.
close(self) -> None
Close the transport and inform the protocol whether the connection was terminated successfully.
_close(self) -> None
Underlying action performed when calling `close`. Subclasses override this method to specify behavior when closing the connection.
abort(self) -> None
Close the transport immediately.
_abort(self, exception : typing.Optional[Exception]) -> None
Underlying action performed when calling `abort`. Subclasses may override this method to specify behavior when aborting the connection.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- typing.Optional[Exception]
- Default
- = None
- Description
The Exception to propagate to the protocol, if connected.
_exception(
- self,
- exception : Exception,
- message : str
Report a fatal error to the event-loop and abort the transport.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- Exception
- Default
- Description
The Exception to pass on the the loop's exception handler.
- Name
message- Type
- str
- Default
- Description
Human-readable text describing the exception's execution state, cause, etc.
Attributes
- Name
_loop- Type
- Value
= asyncio.get_event_loop_policy().get_event_loop()
- Description
- Name
_protocol- Type
- typing.Optional[P_co]
- Value
= None
- Description
- Name
_is_closing- Type
- Value
= True
- Description
ReadTransport
Interface for read-only transports.
- Bases
- asyncio.ReadTransport,
Methods
__init__(self) -> None
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.
Response
- Type
- bool
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- typing.Optional[Exception]
- Default
- = None
- Description
The Exception to propagate to the protocol when aborting, if connected.
_ensure_reader(self) -> None
Set state to enable future read operations.
_remove_reader(self) -> None
Set state to disable further read operations.
_safe_read(self) -> None
Safely and asynchronously read data from the transport.
_read(self) -> typing.Optional[bytes]
Read data from the transport. Called within `_safe_read` for asynchronous reading.
Response
- Type
- typing.Optional[bytes]
- Description
Data read 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.
Response
- Type
- bytes
- Description
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.
is_closing(self) -> bool
Whether the transport is closing or closed.
Response
- Type
- bool
- Description
True if the transport is closing or closed, False otherwise.
_open(self) -> None
Underlying action performed when calling `open`. Subclasses override this method to specify behavior when opening the connection.
_close(self) -> None
Underlying action performed when calling `close`. Subclasses override this method to specify behavior when closing the connection.
abort(self) -> None
Close the transport immediately.
_exception(
- self,
- exception : Exception,
- message : str
Report a fatal error to the event-loop and abort the transport.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- Exception
- Default
- Description
The Exception to pass on the the loop's exception handler.
- Name
message- Type
- str
- Default
- Description
Human-readable text describing the exception's execution state, cause, etc.
Attributes
- Name
_has_reader- Type
- Value
= False
- Description
- Name
_loop- Type
- Value
= asyncio.get_event_loop_policy().get_event_loop()
- Description
- Name
_protocol- Type
- typing.Optional[P_co]
- Value
= None
- Description
- Name
_is_closing- Type
- Value
= True
- Description
ByteSize
- Bases
- enum.IntEnum
Attributes
- Name
FIVEBITS- Type
- Value
= 5
- Description
- Name
SIXBITS- Type
- Value
= 6
- Description
- Name
SEVENBITS- Type
- Value
= 7
- Description
- Name
EIGHTBITS- Type
- Value
= 8
- Description
Parity
- Bases
- str, enum.Enum
Attributes
- Name
NONE- Type
- Value
= 'N'
- Description
- Name
ODD- Type
- Value
= 'O'
- Description
- Name
EVEN- Type
- Value
= 'E'
- Description
- Name
MARK- Type
- Value
= 'M'
- Description
- Name
SPACE- Type
- Value
= 'S'
- Description
SerialTransport
Transport for serial devices.
- MRO
- ⎿⎽⎽⎽Transport
- ⎿⎽⎽⎽⎽SerialTransport
Methods
_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]
Response
- Type
- typing.Optional[bytes]
- Description
_ensure_writer(self) -> None
Adds a writer to the loop if not already added.
Exceptions
- Type
- serial.PortNotOpenError
- Description
- If the serial port is not open.
_poll_write(self) -> None
_remove_writer(self) -> None
Removes a writer from the loop.
_write(self, data : bytes) -> int
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
Response
- Type
- int
- Description
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.
Response
- Type
- bool
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- typing.Optional[Exception]
- Default
- = None
- Description
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.
Response
- Type
- bytes
- Description
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.
Response
- Type
- int
- Description
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.
Response
- Type
- tuple[int, int]
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The maximum allowed number of bytes in the write buffer.
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- typing.Union[bytes, bytearray, memoryview]
- Default
- Description
The bytes to write to the Transport.
can_write_eof(self) -> bool
Whether or not this transport has implemented `write_eof()` method.
Response
- Type
- bool
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
list_of_data- Type
- typing.Iterable[typing.Union[bytes, bytearray, memoryview]]
- Default
- Description
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).
Parameters
- Name
self- Type
- Default
- Description
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
The low-water limit for write flow control.
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The high-water limit for write flow control.
is_closing(self) -> bool
Whether the transport is closing or closed.
Response
- Type
- bool
- Description
True if the transport is closing or closed, False otherwise.
abort(self) -> None
Close the transport immediately.
_exception(
- self,
- exception : Exception,
- message : str
Report a fatal error to the event-loop and abort the transport.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- Exception
- Default
- Description
The Exception to pass on the the loop's exception handler.
- Name
message- Type
- str
- Default
- Description
Human-readable text describing the exception's execution state, cause, etc.
Attributes
- Name
_serial- Type
- Value
= _serial
- Description
- Name
_max_read_size- Type
- Value
= 1024
- Description
- Name
_read_buffer- Type
- Value
= []
- Description
- Name
_has_reader- Type
- Value
= False
- Description
- Name
_has_writer- Type
- Value
= False
- Description
- Name
_is_writing_paused- Type
- Value
= False
- Description
- Name
_write_buffer- Type
- list[typing.Union[bytes, bytearray, memoryview]]
- Value
= []
- Description
- Name
writes_pending- Type
- bool
- Value
= None
- Description
Whether or not there is data in the write buffer waiting to be written.
- Name
_loop- Type
- Value
= asyncio.get_event_loop_policy().get_event_loop()
- Description
- Name
_protocol- Type
- typing.Optional[P_co]
- Value
= None
- Description
- Name
_is_closing- Type
- Value
= True
- Description
StopBits
- Bases
- float, enum.Enum
Attributes
- Name
ONE- Type
- Value
= 1.0
- Description
- Name
ONE_POINT_FIVE- Type
- Value
= 1.5
- Description
- Name
TWO- Type
- Value
= 2.0
- Description
Transport
Interface defining bidirectional communication with a device. Transports read bytes from and write bytes to a device.
- Bases
- asyncio.Transport,
Methods
_open(self) -> None
_close(self) -> None
_ensure_reader(self) -> None
_remove_reader(self) -> None
_read(self) -> typing.Optional[bytes]
Response
- Type
- typing.Optional[bytes]
- Description
_ensure_writer(self) -> None
_remove_writer(self) -> None
_write(self, data : bytes) -> int
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
Response
- Type
- int
- Description
__init__(self) -> None
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.
Response
- Type
- bool
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- typing.Optional[Exception]
- Default
- = None
- Description
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.
Response
- Type
- bytes
- Description
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.
Response
- Type
- int
- Description
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.
Response
- Type
- tuple[int, int]
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The maximum allowed number of bytes in the write buffer.
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- typing.Union[bytes, bytearray, memoryview]
- Default
- Description
The bytes to write to the Transport.
can_write_eof(self) -> bool
Whether or not this transport has implemented `write_eof()` method.
Response
- Type
- bool
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
list_of_data- Type
- typing.Iterable[typing.Union[bytes, bytearray, memoryview]]
- Default
- Description
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).
Parameters
- Name
self- Type
- Default
- Description
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
The low-water limit for write flow control.
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The high-water limit for write flow control.
is_closing(self) -> bool
Whether the transport is closing or closed.
Response
- Type
- bool
- Description
True if the transport is closing or closed, False otherwise.
abort(self) -> None
Close the transport immediately.
_exception(
- self,
- exception : Exception,
- message : str
Report a fatal error to the event-loop and abort the transport.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- Exception
- Default
- Description
The Exception to pass on the the loop's exception handler.
- Name
message- Type
- str
- Default
- Description
Human-readable text describing the exception's execution state, cause, etc.
Attributes
- Name
_has_reader- Type
- Value
= False
- Description
- Name
_has_writer- Type
- Value
= False
- Description
- Name
_is_writing_paused- Type
- Value
= False
- Description
- Name
_write_buffer- Type
- list[typing.Union[bytes, bytearray, memoryview]]
- Value
= []
- Description
- Name
writes_pending- Type
- bool
- Value
= None
- Description
Whether or not there is data in the write buffer waiting to be written.
- Name
_loop- Type
- Value
= asyncio.get_event_loop_policy().get_event_loop()
- Description
- Name
_protocol- Type
- typing.Optional[P_co]
- Value
= None
- Description
- Name
_is_closing- Type
- Value
= True
- Description
UsbTransport
Transport for devices connected via USB. By default, this implementation uses Interface 0 of Configuration 1 on the device.
- MRO
- ⎿⎽⎽⎽Transport
- ⎿⎽⎽⎽⎽UsbTransport
Methods
__init__(
- self,
- vendor : int,
- product : int,
- protocol : asyncio.BaseProtocol,
- interface_index : int
Parameters
- Name
self- Type
- Default
- Description
- Name
vendor- Type
- int
- Default
- Description
The vendor ID of the USB device.
- Name
product- Type
- int
- Default
- Description
The product ID of the USB device.
- Name
protocol- Type
- asyncio.BaseProtocol
- Default
- Description
The protocol instance to use for communication.
- Name
interface_index- Type
- int
- Default
- = DEFAULT_USB_INTERFACE
- Description
The index of the USB Interface to use. Defaults to 0.
_open(self) -> None
_close(self) -> None
_ensure_reader(self) -> None
_remove_reader(self) -> None
__read(self) -> None
_read(self) -> None
Read data from the transport.
_ensure_writer(self) -> None
_remove_writer(self) -> None
_write(self, data) -> None
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- Default
- Description
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.
Response
- Type
- bool
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- typing.Optional[Exception]
- Default
- = None
- Description
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.
Response
- Type
- bytes
- Description
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.
Response
- Type
- int
- Description
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.
Response
- Type
- tuple[int, int]
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The maximum allowed number of bytes in the write buffer.
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- typing.Union[bytes, bytearray, memoryview]
- Default
- Description
The bytes to write to the Transport.
can_write_eof(self) -> bool
Whether or not this transport has implemented `write_eof()` method.
Response
- Type
- bool
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
list_of_data- Type
- typing.Iterable[typing.Union[bytes, bytearray, memoryview]]
- Default
- Description
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).
Parameters
- Name
self- Type
- Default
- Description
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
The low-water limit for write flow control.
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The high-water limit for write flow control.
is_closing(self) -> bool
Whether the transport is closing or closed.
Response
- Type
- bool
- Description
True if the transport is closing or closed, False otherwise.
abort(self) -> None
Close the transport immediately.
_exception(
- self,
- exception : Exception,
- message : str
Report a fatal error to the event-loop and abort the transport.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- Exception
- Default
- Description
The Exception to pass on the the loop's exception handler.
- Name
message- Type
- str
- Default
- Description
Human-readable text describing the exception's execution state, cause, etc.
Attributes
- Name
vendor- Type
- Value
= vendor
- Description
- Name
product- Type
- Value
= product
- Description
- Name
_closing- Type
- Value
= True
- Description
- Name
_protocol- Type
- Value
= protocol
- Description
- Name
_interface_index- Type
- Value
= interface_index
- Description
- Name
_device- Type
- typing.Optional[usb.core.Device]
- Value
= None
- Description
- Name
read_endpoint- Type
- typing.Optional[usb.core.Endpoint]
- Value
= None
- Description
- Name
write_endpoint- Type
- typing.Optional[usb.core.Endpoint]
- Value
= None
- Description
- Name
_reader_event- Type
- Value
= threading.Event()
- Description
- Name
_has_reader- Type
- Value
= False
- Description
- Name
_has_writer- Type
- Value
= False
- Description
- Name
_is_writing_paused- Type
- Value
= False
- Description
- Name
_write_buffer- Type
- list[typing.Union[bytes, bytearray, memoryview]]
- Value
= []
- Description
- Name
writes_pending- Type
- bool
- Value
= None
- Description
Whether or not there is data in the write buffer waiting to be written.
- Name
_loop- Type
- Value
= asyncio.get_event_loop_policy().get_event_loop()
- Description
- Name
_is_closing- Type
- Value
= True
- Description
WriteTransport
Interface for write-only transports.
- Bases
- asyncio.WriteTransport,
Methods
__init__(self) -> None
open(self) -> None
Opens the transport and if write buffer currently contains data, sets state to allow future write operations.
close(self) -> None
Closes the transport. If write buffer is empty, sets state to disallow further write operations.
get_write_buffer_size(self) -> int
Calculate the current size of the write buffer.
Response
- Type
- int
- Description
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.
Response
- Type
- tuple[int, int]
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The maximum allowed number of bytes in the write buffer.
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- typing.Union[bytes, bytearray, memoryview]
- Default
- Description
The bytes to write to the Transport.
can_write_eof(self) -> bool
Whether or not this transport has implemented `write_eof()` method.
Response
- Type
- bool
- Description
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.
Parameters
- Name
self- Type
- Default
- Description
- Name
list_of_data- Type
- typing.Iterable[typing.Union[bytes, bytearray, memoryview]]
- Default
- Description
The list of bytes to concatenate and write to the Transport.
flush(self) -> None
Flush the write buffer and disable further writing.
_ensure_writer(self) -> None
Sets state to allow future write operations, if transport is still open.
_remove_writer(self) -> None
Sets state to disallow future write operations.
_abort(self, exception) -> None
Closes the transport immediately and sets state to disallow further write operations.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- Default
- = None
- Description
The exception to propagate to the transport when closing.
_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.
_write(self, data : bytes) -> int
Write data to the Transport. Called within `_safe_write` for asynchronous writing.
Parameters
- Name
self- Type
- Default
- Description
- Name
data- Type
- bytes
- Default
- Description
The bytes to write to the Transport.
Response
- Type
- int
- Description
the number of bytes written.
_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).
Parameters
- Name
self- Type
- Default
- Description
- Name
low- Type
- typing.Optional[int]
- Default
- = None
- Description
The low-water limit for write flow control.
- Name
high- Type
- typing.Optional[int]
- Default
- = None
- Description
The high-water limit for write flow control.
is_closing(self) -> bool
Whether the transport is closing or closed.
Response
- Type
- bool
- Description
True if the transport is closing or closed, False otherwise.
_open(self) -> None
Underlying action performed when calling `open`. Subclasses override this method to specify behavior when opening the connection.
_close(self) -> None
Underlying action performed when calling `close`. Subclasses override this method to specify behavior when closing the connection.
abort(self) -> None
Close the transport immediately.
_exception(
- self,
- exception : Exception,
- message : str
Report a fatal error to the event-loop and abort the transport.
Parameters
- Name
self- Type
- Default
- Description
- Name
exception- Type
- Exception
- Default
- Description
The Exception to pass on the the loop's exception handler.
- Name
message- Type
- str
- Default
- Description
Human-readable text describing the exception's execution state, cause, etc.
Attributes
- Name
_has_writer- Type
- Value
= False
- Description
- Name
_is_writing_paused- Type
- Value
= False
- Description
- Name
_write_buffer- Type
- list[typing.Union[bytes, bytearray, memoryview]]
- Value
= []
- Description
- Name
writes_pending- Type
- bool
- Value
= None
- Description
Whether or not there is data in the write buffer waiting to be written.
- Name
_loop- Type
- Value
= asyncio.get_event_loop_policy().get_event_loop()
- Description
- Name
_protocol- Type
- typing.Optional[P_co]
- Value
= None
- Description
- Name
_is_closing- Type
- Value
= True
- Description