Device

This module provides base classes for device connections.

ATTRIBUTE	DESCRIPTION
READ_FORMAT	default read format for device connections TYPE: str
WRITE_FORMAT	default write format for device connections TYPE: str
Data	default data type for device connections TYPE: NamedTuple

CLASS	DESCRIPTION
Device	Protocol for device connection classes
StreamingDevice	Protocol for streaming device connection classes
TimedDeviceMixin	Mixin class for timed device operations
BaseDevice	Base class for device connections
SerialDevice	Class for serial device connections
SocketDevice	Class for socket device connections

Documentation last updated: 2025-02-22

BaseDevice

BaseDevice(
    *,
    connection_details: dict | None = None,
    init_timeout: int = 1,
    data_type: NamedTuple = Data,
    read_format: str = READ_FORMAT,
    write_format: str = WRITE_FORMAT,
    simulation: bool = False,
    verbose: bool = False,
    **kwargs,
)

BaseDevice provides an interface for handling device connections

ATTRIBUTE	DESCRIPTION
connection	connection object for the device TYPE: Any | None
connection_details	connection details for the device TYPE: dict
flags	flags for the device TYPE: SimpleNamespace[str, bool]
init_timeout	timeout for initialization TYPE: int
data_type	data type for the device TYPE: NamedTuple
read_format	read format for the device TYPE: str
write_format	write format for the device TYPE: str
eol	end of line character for the read format TYPE: str
buffer	buffer for storing streamed data TYPE: deque
data_queue	queue for storing processed data TYPE: Queue
show_event	event for showing streamed data TYPE: Event
stream_event	event for controlling streaming TYPE: Event
threads	dictionary of threads used in streaming TYPE: dict

METHOD	DESCRIPTION
clear	clear the input and output buffers, and reset the data queue and buffer
connect	connect to the device
disconnect	disconnect from the device
checkDeviceConnection	check the connection to the device
checkDeviceBuffer	check the connection buffer
clearDeviceBuffer	clear the device input and output buffers
read	read data from the device
readAll	read all data from the device
write	write data to the device
poll	poll the device (i.e. write and read data)
processInput	process the input data
processOutput	process the output data
query	query the device (i.e. write and read data)
startStream	start the stream
stopStream	stop the stream
stream	toggle the stream
showStream	show the stream

Initialize BaseDevice class

PARAMETER	DESCRIPTION
connection_details	connection details for the device. Defaults to None. TYPE: dict | None DEFAULT: None
init_timeout	timeout for initialization. Defaults to 1. TYPE: int DEFAULT: 1
data_type	data type for the device. Defaults to Data. TYPE: NamedTuple DEFAULT: Data
read_format	read format for the device. Defaults to READ_FORMAT. TYPE: str DEFAULT: READ_FORMAT
write_format	write format for the device. Defaults to WRITE_FORMAT. TYPE: str DEFAULT: WRITE_FORMAT
simulation	whether to simulate the device. Defaults to False. TYPE: bool DEFAULT: False
verbose	verbosity of class. Defaults to False. TYPE: bool DEFAULT: False

is_connected property

is_connected: bool

Whether the device is connected

verbose property writable

verbose: bool

Verbosity of class

__enter__

__enter__()

Context manager enter method

__exit__

__exit__(exc_type, exc_value, traceback)

Context manager exit method

checkDeviceBuffer

checkDeviceBuffer() -> bool

Check the connection buffer

RETURNS	DESCRIPTION
bool	whether there is data in the connection buffer TYPE: bool

checkDeviceConnection

checkDeviceConnection() -> bool

Check the connection to the device

RETURNS	DESCRIPTION
bool	whether the device is connected TYPE: bool

clear

clear()

Clear the input and output buffers, and reset the data queue and buffer

clearDeviceBuffer

clearDeviceBuffer()

Clear the device input and output buffers

connect

connect()

Connect to the device

disconnect

disconnect()

Disconnect from the device

poll

poll(data: str | None = None) -> str

Poll the device

PARAMETER	DESCRIPTION
data	data to write to the device. Defaults to None. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
str	str|None: data read from the device

processInput

processInput(
    data: Any = None, format_in: str | None = None, **kwargs
) -> str | None

Process the input

PARAMETER	DESCRIPTION
data	data to process. Defaults to None. TYPE: Any DEFAULT: None
format_in	format for the data. Defaults to None. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
str | None	str|None: processed input data

processOutput

processOutput(
    data: str,
    format_out: str | None = None,
    data_type: NamedTuple | None = None,
    timestamp: datetime | None = None,
) -> tuple[Any, datetime | None]

Process the output

PARAMETER	DESCRIPTION
data	data to process TYPE: str
format_out	format for the data. Defaults to None. TYPE: str | None DEFAULT: None
data_type	data type for the data. Defaults to None. TYPE: NamedTuple | None DEFAULT: None
timestamp	timestamp for the data. Defaults to None. TYPE: datetime | None DEFAULT: None

RETURNS	DESCRIPTION
tuple[Any, datetime | None]	tuple[Any, datetime|None]: processed output data and timestamp

query

query(
    data: Any,
    multi_out: bool = True,
    *,
    timeout: int | float = 1,
    format_in: str | None = None,
    format_out: str | None = None,
    data_type: NamedTuple | None = None,
    timestamp: bool = False,
    **kwargs,
) -> Any | None

Query the device

PARAMETER	DESCRIPTION
data	data to query TYPE: Any
multi_out	whether to return multiple outputs. Defaults to True. TYPE: bool DEFAULT: True
timeout	timeout for the query. Defaults to 1. TYPE: int | float DEFAULT: 1
format_in	format for the input data. Defaults to None. TYPE: str | None DEFAULT: None
format_out	format for the output data. Defaults to None. TYPE: str | None DEFAULT: None
data_type	data type for the data. Defaults to None. TYPE: NamedTuple | None DEFAULT: None
timestamp	whether to return the timestamp. Defaults to False. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
Any | None	Any|None: queried data

read

read() -> str

Read data from the device

RETURNS	DESCRIPTION
str	str|None: data read from the device

readAll

readAll() -> list[str]

Read all data from the device

RETURNS	DESCRIPTION
list[str]	list[str]|None: data read from the device

showStream

showStream(on: bool)

Show the stream

PARAMETER	DESCRIPTION
on	whether to show the stream TYPE: bool

startStream

startStream(
    data: str | None = None,
    buffer: deque | None = None,
    *,
    format_out: str | None = None,
    data_type: NamedTuple | None = None,
    show: bool = False,
    sync_start: Barrier | None = None,
    split_stream: bool = True,
    callback: Callable[[str], Any] | None = None,
)

Start the stream

PARAMETER	DESCRIPTION
data	data to stream. Defaults to None. TYPE: str | None DEFAULT: None
buffer	buffer to store the streamed data. Defaults to None. TYPE: deque | None DEFAULT: None
format_out	format for the data. Defaults to None. TYPE: str | None DEFAULT: None
data_type	data type for the data. Defaults to None. TYPE: NamedTuple | None DEFAULT: None
show	whether to show the stream. Defaults to False. TYPE: bool DEFAULT: False
sync_start	synchronization barrier. Defaults to None. TYPE: Barrier | None DEFAULT: None
split_stream	whether to split the stream and data processing threads. Defaults to True. TYPE: bool DEFAULT: True
callback	callback function to call with the streamed data. Defaults to None. TYPE: Callable[[str], Any] | None DEFAULT: None

stopStream

stopStream()

Stop the stream

stream

stream(
    on: bool,
    data: str | None = None,
    buffer: deque | None = None,
    *,
    sync_start: Barrier | None = None,
    split_stream: bool = True,
    callback: Callable[[str], Any] | None = None,
    **kwargs,
)

Toggle the stream

PARAMETER	DESCRIPTION
on	whether to start or stop the stream TYPE: bool
data	data to stream. Defaults to None. TYPE: str | None DEFAULT: None
buffer	buffer to store the streamed data. Defaults to None. TYPE: deque | None DEFAULT: None
sync_start	synchronization barrier. Defaults to None. TYPE: Barrier | None DEFAULT: None
split_stream	whether to split the stream and data processing threads. Defaults to True. TYPE: bool DEFAULT: True
callback	callback function to call with the streamed data. Defaults to None. TYPE: Callable[[str], Any] | None DEFAULT: None

write

write(data: str) -> bool

Write data to the device

PARAMETER	DESCRIPTION
data	data to write to the device TYPE: str

RETURNS	DESCRIPTION
bool	whether the data was written successfully TYPE: bool

Device

Bases: Protocol

Protocol for device connection classes

clear

clear()

Clear the input and output buffers

connect

connect()

Connect to the device

disconnect

disconnect()

Disconnect from the device

processInput

processInput(
    data: Any, format_in: str, **kwargs
) -> str | None

Process the input

processOutput

processOutput(
    data: str,
    format_out: str,
    data_type: NamedTuple,
    timestamp: datetime | None,
    **kwargs,
) -> tuple[Any, datetime]

Process the output

query

query(
    data: Any, multi_out: bool = True, **kwargs
) -> Any | None

Query the device

read

read() -> str | None

Read data from the device

write

write(data: str) -> bool

Write data to the device

SerialDevice

SerialDevice(
    port: str | None = None,
    baudrate: int = 9600,
    timeout: int = 1,
    *,
    init_timeout: int = 1,
    data_type: NamedTuple = Data,
    read_format: str = READ_FORMAT,
    write_format: str = WRITE_FORMAT,
    simulation: bool = False,
    verbose: bool = False,
    **kwargs,
)

Bases: BaseDevice

SerialDevice provides an interface for handling serial devices

ATTRIBUTE	DESCRIPTION
port	device serial port TYPE: str
baudrate	device baudrate TYPE: int
timeout	device timeout TYPE: int
connection_details	connection details for the device TYPE: dict
serial	serial object for the device TYPE: Serial
init_timeout	timeout for initialization TYPE: int
message_end	message end character TYPE: str
flags	flags for the device TYPE: SimpleNamespace[str, bool]
is_connected	whether the device is connected TYPE: bool
verbose	verbosity of class TYPE: bool

METHOD	DESCRIPTION
clear	clear the input and output buffers, and reset the data queue and buffer
connect	connect to the device
disconnect	disconnect from the device
checkDeviceConnection	check the connection to the device
checkDeviceBuffer	check the connection buffer
clearDeviceBuffer	clear the device input and output buffers
read	read data from the device
readAll	read all data from the device
write	write data to the device
poll	poll the device (i.e. write and read data)
processInput	process the input data
processOutput	process the output data
query	query the device (i.e. write and read data)
startStream	start the stream
stopStream	stop the stream
stream	toggle the stream
showStream	show the stream

Initialize SerialDevice class

PARAMETER	DESCRIPTION
port	serial port for the device. Defaults to None. TYPE: str | None DEFAULT: None
baudrate	baudrate for the device. Defaults to 9600. TYPE: int DEFAULT: 9600
timeout	timeout for the device. Defaults to 1. TYPE: int DEFAULT: 1
init_timeout	timeout for initialization. Defaults to 2. TYPE: int DEFAULT: 1
data_type	data type for the device. Defaults to Data. TYPE: NamedTuple DEFAULT: Data
read_format	read format for the device. Defaults to READ_FORMAT. TYPE: str DEFAULT: READ_FORMAT
write_format	write format for the device. Defaults to WRITE_FORMAT. TYPE: str DEFAULT: WRITE_FORMAT
simulation	whether to simulate the device. Defaults to False. TYPE: bool DEFAULT: False
verbose	verbosity of class. Defaults to False. TYPE: bool DEFAULT: False

baudrate property writable

baudrate: int

Device baudrate

port property writable

port: str

Device serial port

serial property writable

serial: Serial

Serial object for the device

timeout property writable

timeout: int

Device timeout

checkDeviceBuffer

checkDeviceBuffer() -> bool

Check the connection buffer

checkDeviceConnection

checkDeviceConnection()

Check the connection to the device

clearDeviceBuffer

clearDeviceBuffer()

Clear the device input and output buffers

connect

connect()

Connect to the device

disconnect

disconnect()

Disconnect from the device

read

read() -> str

Read data from the device

readAll

readAll() -> list[str]

Read all data from the device

write

write(data: str) -> bool

Write data to the device

SocketDevice

SocketDevice(
    host: str,
    port: int,
    timeout: int = 0,
    *,
    byte_size: int = 1024,
    simulation: bool = False,
    verbose: bool = False,
    **kwargs,
)

Bases: BaseDevice

SocketDevice provides an interface for handling socket devices

ATTRIBUTE	DESCRIPTION
host	device host TYPE: str
port	device port TYPE: int
timeout	device timeout TYPE: int
byte_size	size of the byte buffer TYPE: int
connection_details	connection details for the device TYPE: dict
socket	socket object for the device TYPE: socket
flags	flags for the device TYPE: SimpleNamespace[str, bool]
is_connected	whether the device is connected TYPE: bool
verbose	verbosity of class TYPE: bool

METHOD	DESCRIPTION
clear	clear the input and output buffers, and reset the data queue and buffer
connect	connect to the device
disconnect	disconnect from the device
checkDeviceConnection	check the connection to the device
checkDeviceBuffer	check the connection buffer
clearDeviceBuffer	clear the device input and output buffers
read	read data from the device
readAll	read all data from the device
write	write data to the device
poll	poll the device (i.e. write and read data)
processInput	process the input data
processOutput	process the output data
query	query the device (i.e. write and read data)
startStream	start the stream
stopStream	stop the stream
stream	toggle the stream
showStream	show the stream

Initialize SocketDevice class

PARAMETER	DESCRIPTION
host	host for the device TYPE: str
port	port for the device TYPE: int
timeout	timeout for the device. Defaults to 1. TYPE: int DEFAULT: 0
byte_size	size of the byte buffer. Defaults to 1024. TYPE: int DEFAULT: 1024
simulation	whether to simulate the device. Defaults to False. TYPE: bool DEFAULT: False
verbose	verbosity of class. Defaults to False. TYPE: bool DEFAULT: False

address property

address: tuple[str, int]

Device address

host property writable

host: str

Device socket host

port property writable

port: str

Device socket port

socket property writable

socket: socket

Socket object for the device

timeout property writable

timeout: int

Device timeout

checkDeviceBuffer

checkDeviceBuffer() -> bool

Check the connection buffer

checkDeviceConnection

checkDeviceConnection()

Check the connection to the device

clearDeviceBuffer

clearDeviceBuffer()

Clear the device input and output buffers

connect

connect()

Connect to the device

disconnect

disconnect()

Disconnect from the device

read

read() -> str

Read data from the device

readAll

readAll() -> list[str]

Read all data from the device

write

write(data: str) -> bool

Write data to the device

StreamingDevice

Bases: Protocol

Protocol for device connection classes

clear

clear()

Clear the input and output buffers, and reset the data queue and buffer

clearDeviceBuffer

clearDeviceBuffer()

Clear the device input and output buffers

connect

connect()

Connect to the device

disconnect

disconnect()

Disconnect from the device

processInput

processInput(
    data: Any, format_in: str | None = None, **kwargs
) -> str | None

Process the input

processOutput

processOutput(
    data: str,
    format_out: str | None = None,
    data_type: NamedTuple | None = None,
    timestamp: datetime | None = None,
    **kwargs,
) -> tuple[Any, datetime]

Process the output

query

query(
    data: Any, multi_out: bool = True, **kwargs
) -> Any | None

Query the device

read

read() -> str | None

Read data from the device

showStream

showStream(on: bool)

Show the stream

startStream

startStream(
    data: str | None = None,
    buffer: deque | None = None,
    **kwargs,
)

Start the stream

stopStream

stopStream()

Stop the stream

stream

stream(
    on: bool,
    data: str | None = None,
    buffer: deque | None = None,
    **kwargs,
)

Toggle the stream

write

write(data: str) -> bool

Write data to the device

TimedDeviceMixin

TimedDeviceMixin(*args, **kwargs)

Mixin class for timed device operations

METHOD	DESCRIPTION
stopTimer	stop a timer
setValue	set a value
setValueDelayed	set a value after a delay

setValue

setValue(
    value: Any, event: Event | None = None, **kwargs
) -> bool

Set a value

PARAMETER	DESCRIPTION
value	value to set TYPE: Any
event	event to set or clear. Defaults to None. TYPE: Event | None DEFAULT: None

RETURNS	DESCRIPTION
bool	whether the value was set TYPE: bool

setValueDelayed

setValueDelayed(
    duration: int | float,
    initial: Any | None = None,
    final: Any | None = None,
    blocking: bool = True,
    *,
    event: Event | None = None,
    **kwargs,
) -> Timer | None

Set a value after a delay

PARAMETER	DESCRIPTION
duration	duration of the delay TYPE: int | float
initial	initial value. Defaults to None. TYPE: Any | None DEFAULT: None
final	final value. Defaults to None. TYPE: Any | None DEFAULT: None
blocking	whether to block the main thread. Defaults to True. TYPE: bool DEFAULT: True
event	event to set or clear. Defaults to None. TYPE: Event | None DEFAULT: None

RETURNS	DESCRIPTION
Timer | None	threading.Timer|None: timer object if blocking is False

stopTimer

stopTimer(
    timer: Timer | None = None, event: Event | None = None
)

Stop a timer

PARAMETER	DESCRIPTION
timer	timer to stop. Defaults to None. TYPE: Timer | None DEFAULT: None
event	event to clear. Defaults to None TYPE: Event | None DEFAULT: None