ElectrometerController

class lsst.ts.electrometer.ElectrometerController(log=None)

Bases: ABC

Provide high-level control for an electrometer.

Parameters:

log (logging.Logger or None, optional) – Logger for controller messages. If None, a logger named for the concrete controller class is created.

commander

The tcpip interface for writing and reading from the device.

Type:

commander.Commander

commands

The interface for providing formatted commands for the commander.

Type:

ElectrometerCommandFactory

mode

The current measurement mode name.

Type:

str or None

range

The current measurement range.

Type:

float

integration_time

Integration time for one measurement, in seconds.

Type:

float

median_filter_active

Whether the median filter is active.

Type:

bool

filter_active

Whether any filter is active.

Type:

bool

avg_filter_active

Whether the average filter is active.

Type:

bool

last_value

The last intensity value read from the electrometer.

Type:

int

read_freq

The frequency that readings are gotten from the device buffer.

Type:

float

configuration_delay

The delay to allow the electrometer to configure.

Type:

float

auto_range

Whether automatic range is active.

Type:

bool

manual_start_time

The start TAI time of a scan [s].

Type:

float

manual_end_time

The end TAI time of a scan [s].

Type:

float

modes

Associate SAL Command number with electrometer UnitMode enum.

Type:

dict

group_id

Optional observing group identifier for the current scan.

Type:

str or None

voltage_source

Whether the voltage source is enabled.

Type:

bool

voltage_range

Voltage source range.

Type:

int

voltage_limit

Voltage source limit.

Type:

int

voltage_level

Voltage source level.

Type:

int

temperature

Temperature returned from the probe, in degrees C.

Type:

float or None

vsource

Voltage source input, in volts.

Type:

float or None

Attributes Summary

connected

Whether the TCP/IP client is connected.

Methods Summary

after_scan()	Run the post-scan workflow and return the scan result.
before_scan(*, group_id, timed_scan)	Run the pre-scan workflow for a new acquisition.
begin_acquisition(timed_scan)	Begin storing data for the current scan mode.
build_scan_result(*, response, trace_elements)	Build a scan result from a raw device payload.
check_error(from_command)	Drain and log the electrometer error queue.
configure(config)	Configure the controller from one electrometer config entry.
configure_scan_storage()	Configure and clear device storage used by scans.
configure_scan_trigger(timed_scan)	Configure the scan trigger source for the current scan mode.
connect()	Connect to, identify, and initialize the electrometer.
disconnect()	Disconnect from the electrometer.
during_scan(*, timed_scan[, scan_duration])	Run acquisition for either manual or timed scans.
end_acquisition()	Stop data acquisition before reading the buffer.
estimate_read_timeout()	Estimate the buffer readout timeout from the scan duration.
get_avg_filter_status()	Get and cache the average filter status.
get_config_schema()	Return the JSON schema for this controller configuration.
get_integration_time()	Get and cache the integration time.
get_intensity()	Get and cache the latest intensity reading.
get_med_filter_status()	Get and cache the median filter status.
get_mode()	Get and cache the measurement mode.
get_range()	Get and cache the measurement range.
get_settings()	Return the cached controller settings needed by the CSC.
get_timer()	Get and cache the NPLC timer value.
get_voltage_level()	Get and cache the voltage source level.
get_voltage_limit()	Get and cache the voltage source limit.
get_voltage_range()	Get and cache the voltage source range.
get_voltage_source_status()	Get and cache the voltage source status.
make_primary_header(scan_result, fits_data)	Make the primary FITS header.
normalize_mode(mode)	Return a normalized electrometer mode name.
normalize_trace_elements(trace_format)	Normalize trace elements reported by the device.
parse_buffer(response[, num_categories])	Parse buffer values into per-column sample lists.
perform_zero_calibration([mode, auto, ...])	Perform zero calibration for the current measurement setup.
prepare_scan()	Prepare the electrometer trace format for scanning.
read_scan_payload()	Read the device buffer and return the payload and trace elements.
send_command(command[, has_reply, timeout])	Send a command to the electrometer.
set_digital_filter(activate_filter, ...)	Set the digital filter state.
set_integration_time(int_time)	Set the integration time.
set_mode(mode)	Set the mode/unit.
set_range(set_range)	Set the range.
set_timer(nplc)	Set the NPLC timer value.
set_voltage_level(level)	Set the voltage source level.
set_voltage_limit(limit)	Set the voltage source limit.
set_voltage_range(range)	Set the voltage source range.
setup_scan()	Set up the electrometer for scan preparation.
start_scan([group_id])	Start a manual scan.
start_scan_dt(scan_duration[, group_id])	Start a timed scan.
stop_scan()	Stop the active scan and read back the scan buffer.
supports_median_filter()	Return whether the active device supports median filtering.
toggle_voltage_source(toggle)	Enable or disable the voltage source.
wait_for_scan_completion(scan_duration)	Wait for device-driven acquisition to complete.
write_fits_file(scan_result, data_format, ...)	Write scan data to an in-memory FITS file.

Attributes Documentation

connected

Whether the TCP/IP client is connected.

Methods Documentation

async after_scan()

Run the post-scan workflow and return the scan result.

Return type:

ScanResult

async before_scan(*, group_id, timed_scan)

Run the pre-scan workflow for a new acquisition.

Parameters:

• group_id (str | None)

• timed_scan (bool)

Return type:

None

async begin_acquisition(timed_scan)

Begin storing data for the current scan mode.

Parameters:

timed_scan (bool)

Return type:

None

build_scan_result(*, response, trace_elements)

Build a scan result from a raw device payload.

Parameters:

• response (str)

• trace_elements (list[str])

Return type:

ScanResult

async check_error(from_command)

Drain and log the electrometer error queue.

Parameters:

from_command (str | None) – Command name that triggered the error check.

abstract configure(config)

Configure the controller from one electrometer config entry.

Parameters:

config (types.SimpleNamespace) – Configuration for one electrometer instance.

async configure_scan_storage()

Configure and clear device storage used by scans.

Return type:

None

async configure_scan_trigger(timed_scan)

Configure the scan trigger source for the current scan mode.

Parameters:

timed_scan (bool)

Return type:

None

async connect()

Connect to, identify, and initialize the electrometer.

async disconnect()

Disconnect from the electrometer.

async during_scan(*, timed_scan, scan_duration=None)

Run acquisition for either manual or timed scans.

Parameters:

• timed_scan (bool)

• scan_duration (float | None, default: None)

Return type:

None

async end_acquisition()

Stop data acquisition before reading the buffer.

Return type:

None

estimate_read_timeout()

Estimate the buffer readout timeout from the scan duration.

Return type:

float

async get_avg_filter_status()

Get and cache the average filter status.

abstract classmethod get_config_schema()

Return the JSON schema for this controller configuration.

async get_integration_time()

Get and cache the integration time.

Returns:

integration_time – Current integration time, in seconds.

Return type:

float

async get_intensity()

Get and cache the latest intensity reading.

async get_med_filter_status()

Get and cache the median filter status.

async get_mode()

Get and cache the measurement mode.

async get_range()

Get and cache the measurement range.

Returns:

range – Current measurement range.

Return type:

float

get_settings()

Return the cached controller settings needed by the CSC.

Return type:

ControllerSettings

async get_timer()

Get and cache the NPLC timer value.

async get_voltage_level()

Get and cache the voltage source level.

Returns:

voltage_level – Voltage source level.

Return type:

int

async get_voltage_limit()

Get and cache the voltage source limit.

Returns:

voltage_limit – Voltage source limit.

Return type:

int

async get_voltage_range()

Get and cache the voltage source range.

Returns:

voltage_range – Voltage source range.

Return type:

int

async get_voltage_source_status()

Get and cache the voltage source status.

Returns:

voltage_source – Whether the voltage source is enabled.

Return type:

bool

make_primary_header(scan_result, fits_data)

Make the primary FITS header.

Parameters:

• scan_result (ScanResult) – Scan data and timing metadata.

• fits_data (object) – Object with index, name, and obs_ids attributes.

Returns:

primary_hdu – Primary HDU containing Rubin Observatory metadata.

Return type:

astropy.io.fits.PrimaryHDU

normalize_mode(mode)

Return a normalized electrometer mode name.

Parameters:

mode (int | str)

Return type:

str

normalize_trace_elements(trace_format)

Normalize trace elements reported by the device.

Parameters:

trace_format (str)

Return type:

list[str]

parse_buffer(response, num_categories=2)

Parse buffer values into per-column sample lists.

Parameters:

• response (str) – The response from the read buffer command.

• num_categories (int, optional) – Number of values in each sample.

Returns:

categorized_lists – Parsed values grouped by category.

Return type:

list [list [float]]

async perform_zero_calibration(mode=None, auto=None, set_range=None, integration_time=None)

Perform zero calibration for the current measurement setup.

Parameters:

• mode (str | None) – Mode of measurement, CURR, CHAR, VOLT, RES

• auto (bool | None) – Whether or not in auto range

• set_range (float | None) – The measurement range

• integration_time (float | None) – Integration time value accepted for API compatibility. This value is currently unused and will be removed in a future XML update.

async prepare_scan()

Prepare the electrometer trace format for scanning.

async read_scan_payload()

Read the device buffer and return the payload and trace elements.

Return type:

tuple[str, list[str]]

async send_command(command, has_reply=False, timeout=None)

Send a command to the electrometer.

Parameters:

• command (str) – Command string to send.

• has_reply (bool, optional) – Whether a reply is expected.

• timeout (float or None, optional) – Timeout for the command reply, in seconds.

Returns:

reply – Reply from the electrometer if has_reply is true; otherwise None.

Return type:

str or None

async set_digital_filter(activate_filter, activate_avg_filter, activate_med_filter)

Set the digital filter state.

Parameters:

• activate_filter (bool) – Whether any filter should be activated.

• activate_avg_filter (bool) – Whether the average filter should be activated.

• activate_med_filter (bool) – Whether the median filter should be activated.

async set_integration_time(int_time)

Set the integration time.

Parameters:

int_time (float) – The integration time.

async set_mode(mode)

Set the mode/unit.

Parameters:

mode (int) – The mode of the electrometer.

async set_range(set_range)

Set the range.

Parameters:

set_range (float) – The new range value.

async set_timer(nplc)

Set the NPLC timer value.

Parameters:

nplc (float) – Number of power line cycles.

async set_voltage_level(level)

Set the voltage source level.

Parameters:

level (int) – Voltage source level.

async set_voltage_limit(limit)

Set the voltage source limit.

Parameters:

limit (int) – Voltage source limit.

async set_voltage_range(range)

Set the voltage source range.

Parameters:

range (int) – Voltage source range.

async setup_scan()

Set up the electrometer for scan preparation.

async start_scan(group_id=None)

Start a manual scan.

Parameters:

group_id (str | None) – Optional observing group identifier to store with the scan.

async start_scan_dt(scan_duration, group_id=None)

Start a timed scan.

Parameters:

• scan_duration (float) – The amount of time to store values for.

• group_id (str | None) – Optional observing group identifier to store with the scan.

async stop_scan()

Stop the active scan and read back the scan buffer.

Returns:

scan_result – Parsed scan data and scan metadata.

Return type:

ScanResult

supports_median_filter()

Return whether the active device supports median filtering.

Return type:

bool

async toggle_voltage_source(toggle)

Enable or disable the voltage source.

Parameters:

toggle (bool) – Whether to enable the voltage source.

async wait_for_scan_completion(scan_duration)

Wait for device-driven acquisition to complete.

Parameters:

scan_duration (float | None)

Return type:

None

async write_fits_file(scan_result, data_format, fits_data)

Write scan data to an in-memory FITS file.

Parameters:

• scan_result (ScanResult) – Scan data and timing metadata.

• data_format (list [str]) – Column names for the parsed scan data.

• fits_data (object) – Object with index, name, and obs_ids attributes.

Returns:

file_upload – FITS file content positioned at the beginning of the stream.

Return type:

io.BytesIO