Skip to content

Data Logger (DAQ)

Data acquisition / switch unit control (Keysight 34970A/DAQ970A style)

Main DataLogger class for controlling SCPI Data Acquisition systems.

Supports generic SCPI-99 compliant DAQ systems and Keysight 34970A/DAQ970A series.

Installation

pip install "SCPI-Instrument-Control"

Features
  • Multiple connection types: Ethernet/LAN, USB, GPIB
  • Multi-channel scanning with configurable sample rates
  • Support for voltage, current, resistance, temperature measurements
  • Alarm/limit checking and scaling (mx+b)
  • Data logging with timestamps
  • Context manager support for automatic connection management

Reading dataclass

Reading(value: float, channel: Optional[int] = None, timestamp: Optional[float] = None, unit: Optional[str] = None, alarm_state: Optional[str] = None)

A single measurement reading from the DAQ.

DataLogger

DataLogger(host: str, port: int = 5025, timeout: float = 10.0, connection: Optional[BaseConnection] = None)

Main class for controlling SCPI Data Acquisition systems.

This class provides a high-level interface for controlling DAQ functions including channel configuration, scanning, and data retrieval.

Supports both generic SCPI-99 DAQs and Keysight 34970A/DAQ970A series with automatic model detection and capability-based feature availability.

Example

daq = DataLogger('192.168.1.100') daq.connect() print(daq.identify())

Configure channels for voltage measurement

daq.configure_voltage_dc([101, 102, 103])

Set up scan list

daq.set_scan_list([101, 102, 103])

Take readings

readings = daq.read() for r in readings: ... print(f"Channel {r.channel}: {r.value} V") daq.disconnect()

Or using context manager:

with DataLogger('192.168.1.100') as daq: ... daq.configure_voltage_dc([101, 102]) ... readings = daq.measure_voltage_dc([101, 102])

Initialize Data Logger connection.

Parameters:

Name Type Description Default
host str

IP address or hostname of the DAQ

required
port int

TCP port for SCPI communication (default: 5025)

5025
timeout float

Command timeout in seconds (default: 10.0)

10.0
connection Optional[BaseConnection]

Optional custom connection object

None
Note

Call connect() to establish connection and initialize capabilities.

Source code in scpi_control/data_logger.py
def __init__(
    self,
    host: str,
    port: int = 5025,
    timeout: float = 10.0,
    connection: Optional[BaseConnection] = None,
):
    """Initialize Data Logger connection.

    Args:
        host: IP address or hostname of the DAQ
        port: TCP port for SCPI communication (default: 5025)
        timeout: Command timeout in seconds (default: 10.0)
        connection: Optional custom connection object

    Note:
        Call connect() to establish connection and initialize capabilities.
    """
    self.host = host
    self.port = port
    self.timeout = timeout

    if connection is not None:
        self._connection = connection
    else:
        self._connection = SocketConnection(host, port, timeout)

    self.model_capability: Optional[DAQCapability] = None
    self._scpi_commands: Optional[DAQSCPICommandSet] = None
    self._device_info: Optional[Dict[str, str]] = None
    self._scan_list: List[int] = []

is_connected property

is_connected: bool

Check if connected to Data Logger.

device_info property

device_info: Optional[Dict[str, str]]

Get parsed device information.

connect

connect() -> None

Establish connection to the Data Logger.

Raises:

Type Description
SiglentConnectionError

If connection fails

SiglentTimeoutError

If connection times out

Source code in scpi_control/data_logger.py
def connect(self) -> None:
    """Establish connection to the Data Logger.

    Raises:
        SiglentConnectionError: If connection fails
        SiglentTimeoutError: If connection times out
    """
    logger.info(f"Connecting to Data Logger at {self.host}:{self.port}")
    self._connection.connect()
    logger.info("Connected successfully")

    try:
        idn_string = self.identify()
        self._device_info = self._parse_idn(idn_string)
        logger.info(f"Connected to: {self._device_info.get('manufacturer', 'Unknown')} " f"{self._device_info.get('model', 'Unknown')}")

        self.model_capability = detect_daq_from_idn(idn_string)
        logger.info(f"Model capability: {self.model_capability}")

        self._scpi_commands = DAQSCPICommandSet(self.model_capability.scpi_variant)
        logger.info(f"Using SCPI variant: {self.model_capability.scpi_variant}")

    except Exception as e:
        logger.error(f"Failed to identify device: {e}")
        self.disconnect()
        raise exceptions.SiglentConnectionError(f"Connected but failed to identify device: {e}")

disconnect

disconnect() -> None

Close connection to the Data Logger.

Source code in scpi_control/data_logger.py
def disconnect(self) -> None:
    """Close connection to the Data Logger."""
    logger.info("Disconnecting from Data Logger")
    self._connection.disconnect()
    self._device_info = None
    self.model_capability = None
    self._scpi_commands = None
    self._scan_list = []

write

write(command: str) -> None

Send a SCPI command to the Data Logger.

Source code in scpi_control/data_logger.py
def write(self, command: str) -> None:
    """Send a SCPI command to the Data Logger."""
    logger.debug(f"Write: {command}")
    self._connection.write(command)

query

query(command: str) -> str

Send a SCPI query and get the response.

Source code in scpi_control/data_logger.py
def query(self, command: str) -> str:
    """Send a SCPI query and get the response."""
    logger.debug(f"Query: {command}")
    response = self._connection.query(command)
    logger.debug(f"Response: {response}")
    return response

identify

identify() -> str

Get device identification string.

Source code in scpi_control/data_logger.py
def identify(self) -> str:
    """Get device identification string."""
    return self.query("*IDN?")

reset

reset() -> None

Reset Data Logger to default settings.

Source code in scpi_control/data_logger.py
def reset(self) -> None:
    """Reset Data Logger to default settings."""
    logger.info("Resetting Data Logger to defaults")
    self.write("*RST")

clear_status

clear_status() -> None

Clear status registers.

Source code in scpi_control/data_logger.py
def clear_status(self) -> None:
    """Clear status registers."""
    self.write("*CLS")

get_error

get_error() -> str

Get the last error from the error queue.

Source code in scpi_control/data_logger.py
def get_error(self) -> str:
    """Get the last error from the error queue."""
    return self.query("SYST:ERR?")

abort

abort() -> None

Abort the current scan operation.

Source code in scpi_control/data_logger.py
def abort(self) -> None:
    """Abort the current scan operation."""
    logger.info("Aborting current scan")
    cmd = self._get_command("abort")
    self.write(cmd)

configure_voltage_dc

configure_voltage_dc(channels: Union[int, List[int]], range: str = 'AUTO', resolution: str = 'AUTO') -> None

Configure channels for DC voltage measurement.

Parameters:

Name Type Description Default
channels Union[int, List[int]]

Channel number(s) to configure

required
range str

Measurement range (AUTO, 0.1, 1, 10, 100, 300)

'AUTO'
resolution str

Resolution (AUTO, MIN, MAX, or specific value)

'AUTO'
Source code in scpi_control/data_logger.py
def configure_voltage_dc(
    self,
    channels: Union[int, List[int]],
    range: str = "AUTO",
    resolution: str = "AUTO",
) -> None:
    """Configure channels for DC voltage measurement.

    Args:
        channels: Channel number(s) to configure
        range: Measurement range (AUTO, 0.1, 1, 10, 100, 300)
        resolution: Resolution (AUTO, MIN, MAX, or specific value)
    """
    ch_str = format_channel_list(channels)
    cmd = self._get_command("configure_voltage_dc", range=range, resolution=resolution, channels=ch_str)
    self.write(cmd)
    logger.info(f"Configured DC voltage on {ch_str}")

configure_voltage_ac

configure_voltage_ac(channels: Union[int, List[int]], range: str = 'AUTO', resolution: str = 'AUTO') -> None

Configure channels for AC voltage measurement.

Source code in scpi_control/data_logger.py
def configure_voltage_ac(
    self,
    channels: Union[int, List[int]],
    range: str = "AUTO",
    resolution: str = "AUTO",
) -> None:
    """Configure channels for AC voltage measurement."""
    ch_str = format_channel_list(channels)
    cmd = self._get_command("configure_voltage_ac", range=range, resolution=resolution, channels=ch_str)
    self.write(cmd)
    logger.info(f"Configured AC voltage on {ch_str}")

configure_current_dc

configure_current_dc(channels: Union[int, List[int]], range: str = 'AUTO', resolution: str = 'AUTO') -> None

Configure channels for DC current measurement.

Source code in scpi_control/data_logger.py
def configure_current_dc(
    self,
    channels: Union[int, List[int]],
    range: str = "AUTO",
    resolution: str = "AUTO",
) -> None:
    """Configure channels for DC current measurement."""
    ch_str = format_channel_list(channels)
    cmd = self._get_command("configure_current_dc", range=range, resolution=resolution, channels=ch_str)
    self.write(cmd)
    logger.info(f"Configured DC current on {ch_str}")

configure_current_ac

configure_current_ac(channels: Union[int, List[int]], range: str = 'AUTO', resolution: str = 'AUTO') -> None

Configure channels for AC current measurement.

Source code in scpi_control/data_logger.py
def configure_current_ac(
    self,
    channels: Union[int, List[int]],
    range: str = "AUTO",
    resolution: str = "AUTO",
) -> None:
    """Configure channels for AC current measurement."""
    ch_str = format_channel_list(channels)
    cmd = self._get_command("configure_current_ac", range=range, resolution=resolution, channels=ch_str)
    self.write(cmd)
    logger.info(f"Configured AC current on {ch_str}")

configure_resistance

configure_resistance(channels: Union[int, List[int]], range: str = 'AUTO', resolution: str = 'AUTO', four_wire: bool = False) -> None

Configure channels for resistance measurement.

Parameters:

Name Type Description Default
channels Union[int, List[int]]

Channel number(s) to configure

required
range str

Measurement range (AUTO, 100, 1000, 10000, etc.)

'AUTO'
resolution str

Resolution (AUTO, MIN, MAX)

'AUTO'
four_wire bool

Use 4-wire (Kelvin) measurement for accuracy

False
Source code in scpi_control/data_logger.py
def configure_resistance(
    self,
    channels: Union[int, List[int]],
    range: str = "AUTO",
    resolution: str = "AUTO",
    four_wire: bool = False,
) -> None:
    """Configure channels for resistance measurement.

    Args:
        channels: Channel number(s) to configure
        range: Measurement range (AUTO, 100, 1000, 10000, etc.)
        resolution: Resolution (AUTO, MIN, MAX)
        four_wire: Use 4-wire (Kelvin) measurement for accuracy
    """
    ch_str = format_channel_list(channels)
    cmd_name = "configure_resistance_4w" if four_wire else "configure_resistance_2w"
    cmd = self._get_command(cmd_name, range=range, resolution=resolution, channels=ch_str)
    self.write(cmd)
    logger.info(f"Configured {'4-wire' if four_wire else '2-wire'} resistance on {ch_str}")

configure_temperature

configure_temperature(channels: Union[int, List[int]], sensor_type: str = 'TC', sensor_subtype: str = 'K') -> None

Configure channels for temperature measurement.

Parameters:

Name Type Description Default
channels Union[int, List[int]]

Channel number(s) to configure

required
sensor_type str

Sensor type (TC=thermocouple, RTD, THER=thermistor)

'TC'
sensor_subtype str

Sensor subtype (e.g., K, J, T for TC; PT100 for RTD)

'K'
Source code in scpi_control/data_logger.py
def configure_temperature(
    self,
    channels: Union[int, List[int]],
    sensor_type: str = "TC",
    sensor_subtype: str = "K",
) -> None:
    """Configure channels for temperature measurement.

    Args:
        channels: Channel number(s) to configure
        sensor_type: Sensor type (TC=thermocouple, RTD, THER=thermistor)
        sensor_subtype: Sensor subtype (e.g., K, J, T for TC; PT100 for RTD)
    """
    ch_str = format_channel_list(channels)

    if self._scpi_commands and self._scpi_commands.variant == "keysight_daq":
        if sensor_type.upper() == "TC":
            cmd = self._get_command("configure_temperature_tc", tc_type=sensor_subtype, channels=ch_str)
        elif sensor_type.upper() == "RTD":
            cmd = self._get_command("configure_temperature_rtd", rtd_type=sensor_subtype, channels=ch_str)
        elif sensor_type.upper() in ("THER", "THERM", "THERMISTOR"):
            cmd = self._get_command("configure_temperature_therm", therm_type=sensor_subtype, channels=ch_str)
        else:
            cmd = self._get_command("configure_temperature", sensor_type=sensor_type, channels=ch_str)
    else:
        cmd = self._get_command("configure_temperature", sensor_type=f"{sensor_type},{sensor_subtype}", channels=ch_str)

    self.write(cmd)
    logger.info(f"Configured temperature ({sensor_type} {sensor_subtype}) on {ch_str}")

set_scan_list

set_scan_list(channels: Union[int, List[int]]) -> None

Set the scan list for automatic scanning.

Parameters:

Name Type Description Default
channels Union[int, List[int]]

Channel number(s) to include in scan

required
Source code in scpi_control/data_logger.py
def set_scan_list(self, channels: Union[int, List[int]]) -> None:
    """Set the scan list for automatic scanning.

    Args:
        channels: Channel number(s) to include in scan
    """
    ch_str = format_channel_list(channels)
    cmd = self._get_command("set_scan_list", channels=ch_str)
    self.write(cmd)

    # Update internal scan list tracking
    if isinstance(channels, int):
        self._scan_list = [channels]
    else:
        self._scan_list = list(channels)

    logger.info(f"Scan list set to {ch_str}")

get_scan_list

get_scan_list() -> str

Get the current scan list.

Source code in scpi_control/data_logger.py
def get_scan_list(self) -> str:
    """Get the current scan list."""
    cmd = self._get_command("get_scan_list")
    return self.query(cmd)

clear_scan_list

clear_scan_list() -> None

Clear the scan list.

Source code in scpi_control/data_logger.py
def clear_scan_list(self) -> None:
    """Clear the scan list."""
    cmd = self._get_command("clear_scan_list")
    self.write(cmd)
    self._scan_list = []
    logger.info("Scan list cleared")

set_trigger_source

set_trigger_source(source: str = 'IMM') -> None

Set the trigger source.

Parameters:

Name Type Description Default
source str

Trigger source (IMM, TIM, BUS, EXT)

'IMM'
Source code in scpi_control/data_logger.py
def set_trigger_source(self, source: str = "IMM") -> None:
    """Set the trigger source.

    Args:
        source: Trigger source (IMM, TIM, BUS, EXT)
    """
    cmd = self._get_command("set_trigger_source", source=source)
    self.write(cmd)
    logger.info(f"Trigger source set to {source}")

set_trigger_count

set_trigger_count(count: Union[int, str] = 1) -> None

Set number of triggers to accept.

Parameters:

Name Type Description Default
count Union[int, str]

Number of triggers (int or "INF" for infinite)

1
Source code in scpi_control/data_logger.py
def set_trigger_count(self, count: Union[int, str] = 1) -> None:
    """Set number of triggers to accept.

    Args:
        count: Number of triggers (int or "INF" for infinite)
    """
    cmd = self._get_command("set_trigger_count", count=count)
    self.write(cmd)
    logger.info(f"Trigger count set to {count}")

set_trigger_timer

set_trigger_timer(interval: float) -> None

Set the trigger timer interval for timed scanning.

Parameters:

Name Type Description Default
interval float

Time between scans in seconds

required
Source code in scpi_control/data_logger.py
def set_trigger_timer(self, interval: float) -> None:
    """Set the trigger timer interval for timed scanning.

    Args:
        interval: Time between scans in seconds
    """
    cmd = self._get_command("set_trigger_timer", interval=interval)
    self.write(cmd)
    logger.info(f"Trigger timer set to {interval}s")

set_sample_count

set_sample_count(count: Union[int, str] = 1) -> None

Set the number of samples per trigger.

Parameters:

Name Type Description Default
count Union[int, str]

Number of samples (int or "INF")

1
Source code in scpi_control/data_logger.py
def set_sample_count(self, count: Union[int, str] = 1) -> None:
    """Set the number of samples per trigger.

    Args:
        count: Number of samples (int or "INF")
    """
    if self._scpi_commands and self._scpi_commands.supports_command("set_sample_count"):
        cmd = self._get_command("set_sample_count", count=count)
        self.write(cmd)
        logger.info(f"Sample count set to {count}")
    else:
        logger.warning("Sample count command not supported on this model")

initiate

initiate() -> None

Initiate a scan sequence.

Source code in scpi_control/data_logger.py
def initiate(self) -> None:
    """Initiate a scan sequence."""
    cmd = self._get_command("initiate")
    self.write(cmd)
    logger.info("Scan initiated")

trigger

trigger() -> None

Send a software trigger.

Source code in scpi_control/data_logger.py
def trigger(self) -> None:
    """Send a software trigger."""
    cmd = self._get_command("trigger")
    self.write(cmd)
    logger.debug("Software trigger sent")

read

read() -> List[Reading]

Initiate scan and read results (blocking).

Returns:

Type Description
List[Reading]

List of Reading objects with measurement data

Source code in scpi_control/data_logger.py
def read(self) -> List[Reading]:
    """Initiate scan and read results (blocking).

    Returns:
        List of Reading objects with measurement data
    """
    cmd = self._get_command("read")
    response = self.query(cmd)
    return self._parse_readings(response)

fetch

fetch() -> List[Reading]

Fetch readings from memory (does not initiate new scan).

Returns:

Type Description
List[Reading]

List of Reading objects

Source code in scpi_control/data_logger.py
def fetch(self) -> List[Reading]:
    """Fetch readings from memory (does not initiate new scan).

    Returns:
        List of Reading objects
    """
    cmd = self._get_command("fetch")
    response = self.query(cmd)
    return self._parse_readings(response)

read_and_remove

read_and_remove(max_readings: int = 100) -> List[Reading]

Read and remove readings from memory.

Parameters:

Name Type Description Default
max_readings int

Maximum number of readings to retrieve

100

Returns:

Type Description
List[Reading]

List of Reading objects

Source code in scpi_control/data_logger.py
def read_and_remove(self, max_readings: int = 100) -> List[Reading]:
    """Read and remove readings from memory.

    Args:
        max_readings: Maximum number of readings to retrieve

    Returns:
        List of Reading objects
    """
    cmd = self._get_command("read_remove", max_readings=max_readings)
    response = self.query(cmd)
    return self._parse_readings(response)

get_data_points

get_data_points() -> int

Get number of readings in memory.

Source code in scpi_control/data_logger.py
def get_data_points(self) -> int:
    """Get number of readings in memory."""
    cmd = self._get_command("get_data_points")
    response = self.query(cmd)
    return int(float(response.strip()))

clear_data

clear_data() -> None

Clear all readings from memory.

Source code in scpi_control/data_logger.py
def clear_data(self) -> None:
    """Clear all readings from memory."""
    cmd = self._get_command("clear_data")
    self.write(cmd)
    logger.info("Data memory cleared")

measure_voltage_dc

measure_voltage_dc(channels: Union[int, List[int]], range: str = 'AUTO', resolution: str = 'AUTO') -> List[Reading]

Measure DC voltage immediately.

Parameters:

Name Type Description Default
channels Union[int, List[int]]

Channel(s) to measure

required
range str

Measurement range

'AUTO'
resolution str

Resolution

'AUTO'

Returns:

Type Description
List[Reading]

List of Reading objects

Source code in scpi_control/data_logger.py
def measure_voltage_dc(
    self,
    channels: Union[int, List[int]],
    range: str = "AUTO",
    resolution: str = "AUTO",
) -> List[Reading]:
    """Measure DC voltage immediately.

    Args:
        channels: Channel(s) to measure
        range: Measurement range
        resolution: Resolution

    Returns:
        List of Reading objects
    """
    ch_str = format_channel_list(channels)
    cmd = self._get_command("measure_voltage_dc", range=range, resolution=resolution, channels=ch_str)
    response = self.query(cmd)
    readings = self._parse_readings(response)

    # Assign channel numbers if not included in response
    if isinstance(channels, int):
        channels = [channels]
    for i, reading in enumerate(readings):
        if reading.channel is None and i < len(channels):
            reading.channel = channels[i]
        if reading.unit is None:
            reading.unit = "V"

    return readings

measure_voltage_ac

measure_voltage_ac(channels: Union[int, List[int]], range: str = 'AUTO', resolution: str = 'AUTO') -> List[Reading]

Measure AC voltage immediately.

Source code in scpi_control/data_logger.py
def measure_voltage_ac(
    self,
    channels: Union[int, List[int]],
    range: str = "AUTO",
    resolution: str = "AUTO",
) -> List[Reading]:
    """Measure AC voltage immediately."""
    ch_str = format_channel_list(channels)
    cmd = self._get_command("measure_voltage_ac", range=range, resolution=resolution, channels=ch_str)
    response = self.query(cmd)
    readings = self._parse_readings(response)

    if isinstance(channels, int):
        channels = [channels]
    for i, reading in enumerate(readings):
        if reading.channel is None and i < len(channels):
            reading.channel = channels[i]
        if reading.unit is None:
            reading.unit = "V"

    return readings

measure_current_dc

measure_current_dc(channels: Union[int, List[int]], range: str = 'AUTO', resolution: str = 'AUTO') -> List[Reading]

Measure DC current immediately.

Source code in scpi_control/data_logger.py
def measure_current_dc(
    self,
    channels: Union[int, List[int]],
    range: str = "AUTO",
    resolution: str = "AUTO",
) -> List[Reading]:
    """Measure DC current immediately."""
    ch_str = format_channel_list(channels)
    cmd = self._get_command("measure_current_dc", range=range, resolution=resolution, channels=ch_str)
    response = self.query(cmd)
    readings = self._parse_readings(response)

    if isinstance(channels, int):
        channels = [channels]
    for i, reading in enumerate(readings):
        if reading.channel is None and i < len(channels):
            reading.channel = channels[i]
        if reading.unit is None:
            reading.unit = "A"

    return readings

measure_resistance

measure_resistance(channels: Union[int, List[int]], range: str = 'AUTO', resolution: str = 'AUTO', four_wire: bool = False) -> List[Reading]

Measure resistance immediately.

Source code in scpi_control/data_logger.py
def measure_resistance(
    self,
    channels: Union[int, List[int]],
    range: str = "AUTO",
    resolution: str = "AUTO",
    four_wire: bool = False,
) -> List[Reading]:
    """Measure resistance immediately."""
    ch_str = format_channel_list(channels)
    cmd_name = "measure_resistance_4w" if four_wire else "measure_resistance_2w"
    cmd = self._get_command(cmd_name, range=range, resolution=resolution, channels=ch_str)
    response = self.query(cmd)
    readings = self._parse_readings(response)

    if isinstance(channels, int):
        channels = [channels]
    for i, reading in enumerate(readings):
        if reading.channel is None and i < len(channels):
            reading.channel = channels[i]
        if reading.unit is None:
            reading.unit = "Ω"

    return readings

measure_temperature

measure_temperature(channels: Union[int, List[int]], sensor_type: str = 'TC,K') -> List[Reading]

Measure temperature immediately.

Source code in scpi_control/data_logger.py
def measure_temperature(
    self,
    channels: Union[int, List[int]],
    sensor_type: str = "TC,K",
) -> List[Reading]:
    """Measure temperature immediately."""
    ch_str = format_channel_list(channels)
    cmd = self._get_command("measure_temperature", sensor_type=sensor_type, channels=ch_str)
    response = self.query(cmd)
    readings = self._parse_readings(response)

    if isinstance(channels, int):
        channels = [channels]
    for i, reading in enumerate(readings):
        if reading.channel is None and i < len(channels):
            reading.channel = channels[i]
        if reading.unit is None:
            reading.unit = "°C"

    return readings

set_alarm_limits

set_alarm_limits(channels: Union[int, List[int]], high: Optional[float] = None, low: Optional[float] = None) -> None

Set alarm limits for channels.

Parameters:

Name Type Description Default
channels Union[int, List[int]]

Channel(s) to configure

required
high Optional[float]

Upper limit (None to disable)

None
low Optional[float]

Lower limit (None to disable)

None
Source code in scpi_control/data_logger.py
def set_alarm_limits(
    self,
    channels: Union[int, List[int]],
    high: Optional[float] = None,
    low: Optional[float] = None,
) -> None:
    """Set alarm limits for channels.

    Args:
        channels: Channel(s) to configure
        high: Upper limit (None to disable)
        low: Lower limit (None to disable)
    """
    if not self.model_capability or not self.model_capability.has_alarm:
        logger.warning("Alarm limits not supported on this model")
        return

    ch_str = format_channel_list(channels)

    if high is not None:
        cmd = self._get_command("set_alarm_high", limit=high, channels=ch_str)
        self.write(cmd)

    if low is not None:
        cmd = self._get_command("set_alarm_low", limit=low, channels=ch_str)
        self.write(cmd)

    logger.info(f"Alarm limits set on {ch_str}: low={low}, high={high}")

enable_alarm

enable_alarm(channels: Union[int, List[int]], enable: bool = True) -> None

Enable or disable alarm checking on channels.

Source code in scpi_control/data_logger.py
def enable_alarm(self, channels: Union[int, List[int]], enable: bool = True) -> None:
    """Enable or disable alarm checking on channels."""
    if not self.model_capability or not self.model_capability.has_alarm:
        logger.warning("Alarm limits not supported on this model")
        return

    ch_str = format_channel_list(channels)
    state = "ON" if enable else "OFF"
    cmd = self._get_command("set_alarm_enable", state=state, channels=ch_str)
    self.write(cmd)
    logger.info(f"Alarm {'enabled' if enable else 'disabled'} on {ch_str}")

set_scaling

set_scaling(channels: Union[int, List[int]], gain: float = 1.0, offset: float = 0.0, enable: bool = True) -> None

Set mx+b scaling on channels.

Parameters:

Name Type Description Default
channels Union[int, List[int]]

Channel(s) to configure

required
gain float

Gain multiplier (m)

1.0
offset float

Offset value (b)

0.0
enable bool

Enable scaling

True
Source code in scpi_control/data_logger.py
def set_scaling(
    self,
    channels: Union[int, List[int]],
    gain: float = 1.0,
    offset: float = 0.0,
    enable: bool = True,
) -> None:
    """Set mx+b scaling on channels.

    Args:
        channels: Channel(s) to configure
        gain: Gain multiplier (m)
        offset: Offset value (b)
        enable: Enable scaling
    """
    if not self.model_capability or not self.model_capability.has_math:
        logger.warning("Scaling not supported on this model")
        return

    ch_str = format_channel_list(channels)

    cmd = self._get_command("set_scaling_gain", gain=gain, channels=ch_str)
    self.write(cmd)

    cmd = self._get_command("set_scaling_offset", offset=offset, channels=ch_str)
    self.write(cmd)

    state = "ON" if enable else "OFF"
    cmd = self._get_command("set_scaling_enable", state=state, channels=ch_str)
    self.write(cmd)

    logger.info(f"Scaling set on {ch_str}: {gain}x + {offset}")

start_logging

start_logging(channels: Union[int, List[int]], interval: float = 1.0, duration: Optional[float] = None, callback: Optional[callable] = None) -> List[List[Reading]]

Start continuous data logging.

Parameters:

Name Type Description Default
channels Union[int, List[int]]

Channels to log

required
interval float

Time between scans in seconds

1.0
duration Optional[float]

Total logging duration (None for manual stop)

None
callback Optional[callable]

Optional callback function(readings) called after each scan

None

Returns:

Type Description
List[List[Reading]]

List of reading lists (one per scan)

Source code in scpi_control/data_logger.py
def start_logging(
    self,
    channels: Union[int, List[int]],
    interval: float = 1.0,
    duration: Optional[float] = None,
    callback: Optional[callable] = None,
) -> List[List[Reading]]:
    """Start continuous data logging.

    Args:
        channels: Channels to log
        interval: Time between scans in seconds
        duration: Total logging duration (None for manual stop)
        callback: Optional callback function(readings) called after each scan

    Returns:
        List of reading lists (one per scan)
    """
    # Configure scan
    self.set_scan_list(channels)
    self.set_trigger_source("TIM")
    self.set_trigger_timer(interval)

    if duration:
        num_scans = int(duration / interval)
        self.set_trigger_count(num_scans)
    else:
        self.set_trigger_count("INF")

    all_readings = []
    self.initiate()

    try:
        start_time = time.time()
        while True:
            if duration and (time.time() - start_time) >= duration:
                break

            # Wait for data
            time.sleep(interval * 0.9)

            # Read available data
            try:
                points = self.get_data_points()
                if points > 0:
                    readings = self.read_and_remove(points)
                    all_readings.append(readings)

                    if callback:
                        callback(readings)
            except Exception as e:
                logger.warning(f"Error reading data: {e}")

    except KeyboardInterrupt:
        logger.info("Logging interrupted by user")
    finally:
        self.abort()

    return all_readings