pnpq.apt.connection module

You are reading the documentation for v0.1.0, an older version. The most recent release is v0.2.1.

pnpq.apt.connection module#

class pnpq.apt.connection.AbstractAptConnection#

Bases: ABC

abstractmethod close()#
Return type:

None

abstractmethod is_closed()#
Return type:

bool

abstractmethod open()#
Return type:

None

abstractmethod send_message_expect_reply(message, match_reply)#

Send a message and block until an expected reply is received.

Parameters:

  • message (AptMessage) – The message to send.

  • match_reply (Callable[[AptMessage], bool]) – A function that returns True if a received message should be recognized as a reply to the sent message, and False otherwise.

Returns:

The first message to match match_reply’s test.

Return type:

AptMessage

abstractmethod send_message_no_reply(message)#

Send a message and return immediately, without waiting for any reply.

Parameters:

message (AptMessage) – The message to send.

Return type:

None

abstractmethod send_message_unordered(message)#

Send a message as soon as the connection lock will allow, bypassing the message queue. This allows us to poll for status messages while the main message thread is blocked waiting for a reply.

Parameters:

message (AptMessage) – The message to send.

Return type:

None

abstractmethod subscribe()#

Subscribe to all messages received from the device. This can be used to, for example, implement custom logging.

Yield:

A queue which receives all messages received by the device.

Return type:

Iterator[Queue[AptMessage]]

class pnpq.apt.connection.AptConnection(*, serial_number: str, serial_config: pnpq.apt.connection.SerialConfig = <factory>)#

Bases: AbstractAptConnection

Parameters:
close()#
Return type:

None

is_closed()#
Return type:

bool

log = <BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=())>#
open()#
Return type:

None

send_message_expect_reply(message, match_reply)#

Send a message and block until an expected reply is received.

Parameters:

  • message (AptMessage) – The message to send.

  • match_reply (Callable[[AptMessage], bool]) – A function that returns True if a received message should be recognized as a reply to the sent message, and False otherwise.

Returns:

The first message to match match_reply’s test.

Return type:

AptMessage

send_message_no_reply(message)#

Send a message and return immediately, without waiting for any reply.

Parameters:

message (AptMessage)

Return type:

None

send_message_unordered(message)#

Send a message as soon as the serial connection lock will allow, bypassing the message queue. This allows us to poll for status messages while the main message thread is blocked waiting for a reply.

Parameters:

message (AptMessage)

Return type:

None

serial_config: SerialConfig#
serial_number: str#
subscribe()#

Subscribe to all messages received from the device. This can be used to, for example, implement custom logging.

Return type:

Iterator[Queue[AptMessage]]

class pnpq.apt.connection.SerialConfig(*, baudrate=115200, bytesize=8, exclusive=True, parity='N', rtscts=True, stopbits=1, timeout=1.0, write_timeout=1.0)#

Bases: object

Serial connection configuration parameters, to be passed to serial.Serial. These defaults are used by all known Thorlabs devices that implement the APT protocol and should not need to be changed.

Parameters:

  • baudrate (int)

  • bytesize (int)

  • exclusive (bool)

  • parity (str)

  • rtscts (bool)

  • stopbits (int)

  • timeout (None | float)

  • write_timeout (None | float)

baudrate: int = 115200#
bytesize: int = 8#
exclusive: bool = True#
parity: str = 'N'#
rtscts: bool = True#
stopbits: int = 1#
timeout: None | float = 1.0#
write_timeout: None | float = 1.0#
Contents