p5control.drivers.basedriver

Base Class for a driver, as an example using pyvisa

exception p5control.drivers.basedriver.BaseDriverError

Bases: Exception

Exception related to the Driver

class p5control.drivers.basedriver.BaseDriver(name: str, address: str, refresh_delay: float = 0.5)

Bases: object

Base class for a driver. Implements some common funcionality for a message base driver with a given address.

Inherit this class when implementing your own driver. Overwrite the different functions as detailed below to allow the server to automatically collect data and track the status of the instrument.

Thread-Safety:

pyvisa itself is not thread safe, even if the faq on their docs states this, see https://github.com/pyvisa/pyvisa/issues/726. Thus it is necessary to think about thread safety. If you use a different library, make sure that it is thread safe or implement a driver-wide lock to stop weird behavior.

__init__(name: str, address: str, refresh_delay: float = 0.5)

Init the device and change constant settings, for example termination of requests and replies. After this, call self.open() to open the connection to the device

Inheritance: parameter name is required, InstrumentServer._add creates an instance by calling class_ref(name, *args, **kwargs). This is done such that different instances of the driver can be dinstinguished if you own an instrument multiple times

Parameters:

  • name (str) – name for the device this driver is for

  • address (str) – visa adress of the resource

  • refresh_delay (float, default = 0.5) – time spend asleep between self.get_data() calls in the measuremet thread

open()

Open connection to the device.

Raises:

BaseDriverError – if an connecton has already been established.

close()

Close connection to the device and reset self._inst variable to None

Raises:

BaseDriverError – if no connection exists which can be closed

_measuring_thread(stop_event: Event, entry_barrier: Barrier | None = None, exit_barrier: Barrier | None = None, hdf5_path: str = '', refresh_delay: float | None = None)

Measurement thread which periodically asks the instrument for its data and saves it. Starts by calling setup_measuring and calls stop_measuring after the stop event is set.

Parameters:

  • stop_event (threading.Event) – Event which signals the thread to end

  • entry_barrier (threading.Barrier, optional) – if specified, entry_barrier.wait() is called after setup. can be used to synchronise different measurements on different devices.

  • exit_barrier (threading.Barrier, optional) – if specified, exit_barrier.wait() is called after measurements have been stopped, can be used detect when all threads have stopped measuring

  • hdf5_path (str, optional) – specify a path to store the data under, will be stored at hdf5_path/name

  • refresh_delay (float, optional) – specify the time period after which new data is requested, can be used to overwrite the value from initialization

_save_data(hdf5_path: str, array, dgw: DataGateway)

Save data to hdf5 through a gateway. Overwrite this method if you want to change how or where this driver saves it data when being measured.

Parameters:

  • hdf5_path (str) – base hdf5_path under which you should save the data

  • array – data to save

  • dgw (DataGateway) – gateway to the dataserver

stop_measuring()

In this method, write the commands necessary to restore the original state of the device after a measurement has been completed.

setup_measuring()

Setup the device prior to measurement, for example the rate of acquisition etc…

start_measuring()

Start the measurement

get_data()

Request data from the device and process it. This can for example include the creation of time stamps for the corresponding time points.

get_status()

Query the status of the device, including parameters which you want to be logged during to whole uptime of the instrument server, this can include temperature measurement of just the status of the device.

This method is called periodically but with a significant longer period than get_data during an actual measurment.

If this method is not implemented, nothing will get logged.

This method should return a dictionary of the form: {“ampl”: 1.2, “freq”: 2.3}, note that “time” will be automatically added in the status_measurement_thread.

class p5control.drivers.basedriver.ThreadSafeBaseDriver(name: str, address: str, refresh_delay: float = 0.5)

Bases: BaseDriver

Extends BaseDriver with thread safe read, write and query methods.

open()

Open connection to the device.

Raises:

BaseDriverError – if an connecton has already been established.

write(message: str)

Write a string message to the device, thread safe.

read()

Read a message from the device, thread safe.

query(message: str)

Write a message to the device and read a response.