Automation

High-level automation and data collection

Automation and programmatic data collection for Siglent oscilloscopes.

This module provides high-level APIs for automated data collection, batch processing, and analysis of oscilloscope traces. It simplifies common workflows for users who want to collect and analyze data programmatically.

Example

Simple waveform capture:

from scpi_control import Oscilloscope from scpi_control.automation import DataCollector

collector = DataCollector('192.168.1.100') collector.connect() data = collector.capture_single([1, 2]) # Capture channels 1 and 2 collector.save_data(data, 'measurement.npz') collector.disconnect()

Batch collection with different timebase settings:

collector = DataCollector('192.168.1.100') with collector: ... results = collector.batch_capture( ... channels=[1], ... timebase_scales=['1us', '10us', '100us'], ... triggers_per_config=10 ... ) ... collector.save_batch(results, 'batch_data')

Time-series collection:

collector = DataCollector('192.168.1.100') with collector: ... collector.start_continuous_capture( ... channels=[1, 2], ... duration=60, # 60 seconds ... interval=1.0, # 1 capture per second ... output_dir='time_series_data' ... )

DataCollector

DataCollector(host: str, port: int = 5025, timeout: float = 5.0, connection: Optional[BaseConnection] = None, dialect: Optional[str] = None)

High-level API for automated oscilloscope data collection.

This class wraps the Oscilloscope class and provides convenient methods for common data collection workflows, batch processing, and automated measurements.

Initialize data collector.

Parameters:

Name	Type	Description	Default
host	str	IP address or hostname of the oscilloscope	required
port	int	TCP port for SCPI communication (default: 5025, the Siglent raw SCPI socket; 5024 is the telnet-style port with prompts and is not recommended)	5025
timeout	float	Command timeout in seconds (default: 5.0)	5.0
connection	Optional[BaseConnection]	Optional connection implementation (e.g., MockConnection for offline tests)	None
dialect	Optional[str]	Optional SCPI dialect override passed to Oscilloscope ("legacy" or "modern"; None auto-detects)	None

Source code in scpi_control/automation.py

def __init__(
    self,
    host: str,
    port: int = 5025,
    timeout: float = 5.0,
    connection: Optional[BaseConnection] = None,
    dialect: Optional[str] = None,
):
    """Initialize data collector.

    Args:
        host: IP address or hostname of the oscilloscope
        port: TCP port for SCPI communication (default: 5025, the Siglent raw SCPI socket; 5024 is the telnet-style port with prompts and is not recommended)
        timeout: Command timeout in seconds (default: 5.0)
        connection: Optional connection implementation (e.g., MockConnection for offline tests)
        dialect: Optional SCPI dialect override passed to Oscilloscope ("legacy" or "modern"; None auto-detects)
    """
    self.scope = Oscilloscope(host, port, timeout, connection=connection, dialect=dialect)
    self._connected = False

from_scope classmethod

from_scope(scope: Oscilloscope) -> DataCollector

Wrap an Oscilloscope the caller already owns and connected.

init builds its own Oscilloscope from a host, which leaves no way to reuse capture_single's arm-and-wait logic against an existing session. The scope's lifetime stays with the caller: this does not connect, and disconnect() on the result would close a session it did not open.

Source code in scpi_control/automation.py

@classmethod
def from_scope(cls, scope: Oscilloscope) -> "DataCollector":
    """Wrap an Oscilloscope the caller already owns and connected.

    __init__ builds its own Oscilloscope from a host, which leaves no way to
    reuse capture_single's arm-and-wait logic against an existing session.
    The scope's lifetime stays with the caller: this does not connect, and
    disconnect() on the result would close a session it did not open.
    """
    collector = cls.__new__(cls)
    collector.scope = scope
    collector._connected = True
    return collector

connect

connect() -> None

Connect to the oscilloscope.

Source code in scpi_control/automation.py

def connect(self) -> None:
    """Connect to the oscilloscope."""
    self.scope.connect()
    self._connected = True
    logger.info(f"Connected to {self.scope.identify()}")

disconnect

disconnect() -> None

Disconnect from the oscilloscope.

Source code in scpi_control/automation.py

def disconnect(self) -> None:
    """Disconnect from the oscilloscope."""
    if self._connected:
        self.scope.disconnect()
        self._connected = False
        logger.info("Disconnected from oscilloscope")

capture_single

capture_single(channels: List[int], auto_setup: bool = False, max_wait: Optional[float] = None) -> Dict[int, WaveformData]

Capture waveforms from specified channels.

If the scope is already in NORM trigger mode, this waits for the next natural trigger instead of arming a single acquisition -- the user's configured mode is left running rather than being overwritten. In any other mode (including the default) it arms a single acquisition as before. Previously this method always forced a single-shot regardless of mode; a NORM-mode caller upgrading past this change will see capture_single wait for a real trigger instead of forcing one.

Parameters:

Name	Type	Description	Default
channels	List[int]	List of channel numbers to capture (e.g., [1, 2, 3])	required
auto_setup	bool	If True, run auto-setup before capture	False
max_wait	Optional[float]	Seconds to wait for the acquisition to complete. Must be positive. None (default) derives a budget from the current timebase.	None

Returns:

Type	Description
Dict[int, WaveformData]	Dictionary mapping channel number to WaveformData object

Raises:

Type	Description
InvalidParameterError	If max_wait is not positive.
SiglentTimeoutError	If the acquisition does not complete in time. Reading anyway would return the PREVIOUS acquisition, which is what this replaced.

Example

data = collector.capture_single([1, 2]) print(f"Channel 1: {len(data[1].voltage)} samples") print(f"Sample rate: {data[1].sample_rate} Hz")

Source code in scpi_control/automation.py

def capture_single(self, channels: List[int], auto_setup: bool = False, max_wait: Optional[float] = None) -> Dict[int, WaveformData]:
    """Capture waveforms from specified channels.

    If the scope is already in NORM trigger mode, this waits for the next
    natural trigger instead of arming a single acquisition -- the user's
    configured mode is left running rather than being overwritten. In any
    other mode (including the default) it arms a single acquisition as
    before. Previously this method always forced a single-shot regardless
    of mode; a NORM-mode caller upgrading past this change will see
    capture_single wait for a real trigger instead of forcing one.

    Args:
        channels: List of channel numbers to capture (e.g., [1, 2, 3])
        auto_setup: If True, run auto-setup before capture
        max_wait: Seconds to wait for the acquisition to complete. Must be
            positive. None (default) derives a budget from the current
            timebase.

    Returns:
        Dictionary mapping channel number to WaveformData object

    Raises:
        InvalidParameterError: If max_wait is not positive.
        SiglentTimeoutError: If the acquisition does not complete in time.
            Reading anyway would return the PREVIOUS acquisition, which is
            what this replaced.

    Example:
        >>> data = collector.capture_single([1, 2])
        >>> print(f"Channel 1: {len(data[1].voltage)} samples")
        >>> print(f"Sample rate: {data[1].sample_rate} Hz")
    """
    if not self._connected:
        raise SiglentError(f"Not connected to oscilloscope at {self.scope.host}:{self.scope.port}")

    # Checked explicitly, because the failure is otherwise misattributed: a
    # zero or negative budget skips the poll loop without a single status
    # read and raises "last status: unknown", which reads like an
    # instrument fault rather than the bad argument it is.
    if max_wait is not None and max_wait <= 0:
        raise InvalidParameterError(f"max_wait must be a positive number of seconds, not {max_wait!r}")

    if auto_setup:
        self.scope.auto_setup()
        # Unlike the trigger wait below, this genuinely is an
        # unknown-duration sleep: auto-setup reports no status the library
        # can poll, so there is nothing to wait ON.
        time.sleep(1)

    if self.scope.trigger.mode != "NORM":
        # Mirrors wait_for_trigger: NORMAL mode is already free-running and
        # re-arms itself, so arming a SINGLE here would stomp the user's
        # NORM setting and _wait_for_acquisition would never see it.
        #
        # Ordering is load-bearing, not cosmetic: trigger_single() writes
        # TRIG_MODE SINGLE to the scope, which overwrites the very mode
        # _wait_for_acquisition needs to read to pick its done-states. The
        # mode must be checked BEFORE this call, not just before the poll
        # -- moving this inside/after trigger_single() silently reproduces
        # the bug this guard exists to prevent.
        self.scope.trigger_single()
    self._wait_for_acquisition(self._acquisition_timeout() if max_wait is None else max_wait)

    # Capture waveforms
    waveforms = {}
    for ch in channels:
        try:
            channel = getattr(self.scope, f"channel{ch}")
            if channel.enabled:
                waveforms[ch] = self.scope.waveform.acquire(ch)
                logger.info(f"Captured {len(waveforms[ch].voltage)} samples from channel {ch}")
            else:
                logger.warning(f"Channel {ch} is not enabled, skipping")
        except Exception as e:
            logger.error(f"Failed to capture channel {ch}: {e}")

    return waveforms

batch_capture

batch_capture(channels: List[int], timebase_scales: Optional[List[str]] = None, voltage_scales: Optional[Dict[int, List[str]]] = None, triggers_per_config: int = 1, progress_callback: Optional[Callable[[int, int, str], None]] = None, max_consecutive_failures: Optional[int] = 3) -> List[Dict[str, Any]]

Capture multiple waveforms with different configurations.

Parameters:

Name	Type	Description	Default
channels	List[int]	List of channel numbers to capture	required
timebase_scales	Optional[List[str]]	List of timebase scale strings (e.g., ['1us', '10us', '100us'])	None
voltage_scales	Optional[Dict[int, List[str]]]	Dict mapping channel number to list of voltage scale strings (e.g., {1: ['1V', '2V'], 2: ['500mV', '1V']})	None
triggers_per_config	int	Number of captures per configuration	1
progress_callback	Optional[Callable[[int, int, str], None]]	Optional callback function(current, total, status)	None
max_consecutive_failures	Optional[int]	Stop the run after this many back-to-back capture timeouts, counted ACROSS configurations and reset by any success. Guards the common unattended failure -- a trigger level set where the signal never crosses it -- which otherwise times out on every capture for the whole run: at a 70 s timeout and 100 triggers per configuration that is hours of waiting to collect nothing. Pass None to disable the breaker entirely; a run with genuinely sparse triggers should raise max_wait rather than rely on repeated timeouts.	3

Returns:

Type	Description
List[Dict[str, Any]]	List of dictionaries containing waveforms and configuration metadata.
List[Dict[str, Any]]	When the breaker trips (or the run is interrupted) everything gathered
List[Dict[str, Any]]	so far is still returned -- failed entries carry an error field --
List[Dict[str, Any]]	rather than being discarded.

Raises:

Type	Description
InvalidParameterError	If a scale cannot be parsed, or max_consecutive_failures is not None and not at least 1.

Example

results = collector.batch_capture( ... channels=[1], ... timebase_scales=['1us', '10us', '100us'], ... triggers_per_config=5 ... ) print(f"Collected {len(results)} captures")

Source code in scpi_control/automation.py

def batch_capture(
    self,
    channels: List[int],
    timebase_scales: Optional[List[str]] = None,
    voltage_scales: Optional[Dict[int, List[str]]] = None,
    triggers_per_config: int = 1,
    progress_callback: Optional[Callable[[int, int, str], None]] = None,
    max_consecutive_failures: Optional[int] = 3,
) -> List[Dict[str, Any]]:
    """Capture multiple waveforms with different configurations.

    Args:
        channels: List of channel numbers to capture
        timebase_scales: List of timebase scale strings (e.g., ['1us', '10us', '100us'])
        voltage_scales: Dict mapping channel number to list of voltage scale strings
                       (e.g., {1: ['1V', '2V'], 2: ['500mV', '1V']})
        triggers_per_config: Number of captures per configuration
        progress_callback: Optional callback function(current, total, status)
        max_consecutive_failures: Stop the run after this many back-to-back
            capture timeouts, counted ACROSS configurations and reset by any
            success. Guards the common unattended failure -- a trigger level
            set where the signal never crosses it -- which otherwise times
            out on every capture for the whole run: at a 70 s timeout and
            100 triggers per configuration that is hours of waiting to
            collect nothing. Pass None to disable the breaker entirely; a
            run with genuinely sparse triggers should raise `max_wait`
            rather than rely on repeated timeouts.

    Returns:
        List of dictionaries containing waveforms and configuration metadata.
        When the breaker trips (or the run is interrupted) everything gathered
        so far is still returned -- failed entries carry an ``error`` field --
        rather than being discarded.

    Raises:
        InvalidParameterError: If a scale cannot be parsed, or
            ``max_consecutive_failures`` is not None and not at least 1.

    Example:
        >>> results = collector.batch_capture(
        ...     channels=[1],
        ...     timebase_scales=['1us', '10us', '100us'],
        ...     triggers_per_config=5
        ... )
        >>> print(f"Collected {len(results)} captures")
    """
    # Arguments are validated before the connection check: a bad argument is a
    # bad argument whether or not an instrument happens to be attached, and
    # reporting it only when connected makes it look like a connection problem.
    if max_consecutive_failures is not None and max_consecutive_failures < 1:
        raise InvalidParameterError(f"max_consecutive_failures must be at least 1, or None to disable the breaker: {max_consecutive_failures}")
    if not self._connected:
        raise SiglentError(f"Not connected to oscilloscope at {self.scope.host}:{self.scope.port}")

    results = []
    consecutive_failures = 0
    stop_reason = None

    # Build configuration list
    configs = []
    if timebase_scales:
        for tb in timebase_scales:
            # Parsed HERE, not at apply time, so an unparseable scale fails
            # before any capture is taken rather than part-way through a run.
            configs.append({"timebase": parse_si_value(tb, "timebase scale")})
    else:
        configs.append({})

    if voltage_scales:
        new_configs = []
        for config in configs:
            for ch, scales in voltage_scales.items():
                for scale in scales:
                    new_config = config.copy()
                    new_config[f"ch{ch}_vdiv"] = parse_si_value(scale, f"channel {ch} voltage scale")
                    new_configs.append(new_config)
        if new_configs:
            configs = new_configs

    total = len(configs) * triggers_per_config
    current = 0

    # Execute batch capture. Enumerated rather than looked up with
    # configs.index(config): parsing makes duplicate configs genuinely
    # possible -- ['1us', '1000ns'] used to yield two distinct raw-string
    # dicts and now both parse to {'timebase': 1e-06} -- and index() would
    # then report "Config 1/2" twice.
    for config_index, config in enumerate(configs):
        # The WHOLE per-config body is guarded, not just the capture. Ctrl-C
        # lands wherever the operator happens to press it, and the gap between
        # configs -- two socket writes plus the settle sleep below -- is exactly
        # where an impatient operator watching a doomed run tends to hit it.
        # Guarding only the capture would discard the whole run for a keypress
        # one statement earlier, which is the loss this exists to prevent.
        try:
            # Apply configuration
            if "timebase" in config:
                if hasattr(self.scope, "set_timebase"):
                    self.scope.set_timebase(config["timebase"])
                else:
                    self.scope.timebase = config["timebase"]
                logger.info(f"Set timebase to {config['timebase']}")

            for ch, scale in [(int(k[2]), v) for k, v in config.items() if k.startswith("ch") and k.endswith("_vdiv")]:
                channel = getattr(self.scope, f"channel{ch}")
                if hasattr(channel, "set_scale"):
                    channel.set_scale(scale)
                else:
                    channel.voltage_scale = scale
                logger.info(f"Set channel {ch} scale to {scale}")

            time.sleep(0.2)  # Allow settings to settle

            # Capture multiple triggers with this configuration
            for trigger_num in range(triggers_per_config):
                current += 1

                if progress_callback:
                    status = f"Config {config_index+1}/{len(configs)}, Trigger {trigger_num+1}/{triggers_per_config}"
                    progress_callback(current, total, status)

                entry = {
                    "timestamp": datetime.now().isoformat(),
                    "config": config.copy(),
                    "waveforms": {},
                    "trigger_num": trigger_num,
                }
                try:
                    entry["waveforms"] = self.capture_single(channels)
                    # Any success proves the setup can still trigger, so the breaker
                    # counts CONSECUTIVE failures rather than total ones -- an
                    # occasional miss in a long run is not the same as a run that
                    # cannot trigger at all.
                    consecutive_failures = 0
                except SiglentError as exc:
                    # SiglentError, not just SiglentTimeoutError: a dropped link is
                    # the other likely unattended failure, and letting it propagate
                    # would discard every capture already taken -- precisely the loss
                    # this whole path exists to prevent. A scope that stopped
                    # answering is also exactly what a breaker is for, so it counts.
                    # Record and carry on; an "error" key appears ONLY on failed
                    # entries, so consumers reading config/waveforms/trigger_num are
                    # unaffected.
                    logger.warning(f"Capture failed for config {config}: {exc}")
                    entry["error"] = str(exc)
                    consecutive_failures += 1
                results.append(entry)

                if max_consecutive_failures is not None and consecutive_failures >= max_consecutive_failures:
                    stop_reason = f"{consecutive_failures} consecutive capture failures"
                    break

        except KeyboardInterrupt:
            # Keep what was collected. Without this an operator aborting a run
            # they can see is doomed loses every capture already taken.
            logger.info("Batch capture interrupted by user")
            stop_reason = "interrupted by user"
        except SiglentError as exc:
            # Applying the configuration failed -- the instrument stopped
            # answering between captures. Stop, but return what was collected.
            logger.error(f"Batch capture stopped: applying configuration {config} failed: {exc}")
            stop_reason = f"instrument error while applying a configuration: {exc}"

        if stop_reason is not None:
            break

    if stop_reason is not None:
        # Not silent: the collected results come back, the failed entries carry
        # their own `error`, and the caller is told the run was cut short. ERROR
        # rather than WARNING because a truncated batch is a result the caller
        # will otherwise mistake for a complete one -- `len(results)` is the only
        # other signal, and save_batch reports it without any hint of a shortfall.
        logger.error(f"Batch capture stopped early ({stop_reason}) after {len(results)} of {total} planned captures. Check the trigger level and max_wait if captures are timing out.")
    logger.info(f"Batch capture complete: {len(results)} captures")
    return results

start_continuous_capture

start_continuous_capture(channels: List[int], duration: float, interval: float = 1.0, output_dir: Optional[Union[str, Path]] = None, file_format: str = 'npz', progress_callback: Optional[Callable[[int, str], None]] = None) -> List[Dict[str, Any]]

Capture waveforms continuously over a time period.

Parameters:

Name	Type	Description	Default
channels	List[int]	List of channel numbers to capture	required
duration	float	Total capture duration in seconds	required
interval	float	Time between captures in seconds	1.0
output_dir	Optional[Union[str, Path]]	Optional directory to save captures (saves to memory if None)	None
file_format	str	Format for saved files ('npz', 'csv', 'mat', 'h5')	'npz'
progress_callback	Optional[Callable[[int, str], None]]	Optional callback function(captures_done, status)	None

Returns:

Type	Description
List[Dict[str, Any]]	List of capture dictionaries. Without output_dir each entry
List[Dict[str, Any]]	carries the captured waveforms; with output_dir the bulky
List[Dict[str, Any]]	arrays are omitted (they are on disk) and each entry instead lists
List[Dict[str, Any]]	the files written, so a caller can still tell how many captures
List[Dict[str, Any]]	happened and where they went.

Raises:

Type

Description

SiglentError

If the very first save fails and no file was written. That means the run is misconfigured -- a rejected file_format, an unwritable output_dir, a missing optional dependency -- and every later attempt would fail identically, so an unattended run is stopped immediately rather than writing nothing for its full duration. Once one file has landed the configuration is proven, and later save failures are counted and logged without aborting the run.

Example

Capture for 60 seconds, save to files

collector.start_continuous_capture( ... channels=[1, 2], ... duration=60, ... interval=2.0, ... output_dir='continuous_data', ... file_format='npz' ... )

Source code in scpi_control/automation.py

def start_continuous_capture(
    self,
    channels: List[int],
    duration: float,
    interval: float = 1.0,
    output_dir: Optional[Union[str, Path]] = None,
    file_format: str = "npz",
    progress_callback: Optional[Callable[[int, str], None]] = None,
) -> List[Dict[str, Any]]:
    """Capture waveforms continuously over a time period.

    Args:
        channels: List of channel numbers to capture
        duration: Total capture duration in seconds
        interval: Time between captures in seconds
        output_dir: Optional directory to save captures (saves to memory if None)
        file_format: Format for saved files ('npz', 'csv', 'mat', 'h5')
        progress_callback: Optional callback function(captures_done, status)

    Returns:
        List of capture dictionaries. Without ``output_dir`` each entry
        carries the captured ``waveforms``; with ``output_dir`` the bulky
        arrays are omitted (they are on disk) and each entry instead lists
        the ``files`` written, so a caller can still tell how many captures
        happened and where they went.

    Raises:
        SiglentError: If the very first save fails and no file was written.
            That means the run is misconfigured -- a rejected ``file_format``,
            an unwritable ``output_dir``, a missing optional dependency --
            and every later attempt would fail identically, so an unattended
            run is stopped immediately rather than writing nothing for its
            full duration. Once one file has landed the configuration is
            proven, and later save failures are counted and logged without
            aborting the run.

    Example:
        >>> # Capture for 60 seconds, save to files
        >>> collector.start_continuous_capture(
        ...     channels=[1, 2],
        ...     duration=60,
        ...     interval=2.0,
        ...     output_dir='continuous_data',
        ...     file_format='npz'
        ... )
    """
    if not self._connected:
        raise SiglentError(f"Not connected to oscilloscope at {self.scope.host}:{self.scope.port}")

    if output_dir:
        output_path = Path(output_dir)
        output_path.mkdir(parents=True, exist_ok=True)
        logger.info(f"Saving captures to {output_path}")

    results = []
    start_time = time.time()
    capture_count = 0
    files_written = 0
    save_failures = 0
    fatal_save_error = None

    # Set to AUTO trigger mode for continuous acquisition
    self.scope.trigger.mode = "AUTO"

    while (time.time() - start_time) < duration:
        try:
            capture_start = time.time()

            # Capture waveforms
            waveforms = {}
            for ch in channels:
                try:
                    channel = getattr(self.scope, f"channel{ch}")
                    if channel.enabled:
                        waveforms[ch] = self.scope.waveform.acquire(ch)
                except Exception as e:
                    logger.error(f"Failed to capture channel {ch}: {e}")

            capture_count += 1
            elapsed = time.time() - start_time

            capture_data = {
                "timestamp": datetime.now().isoformat(),
                "elapsed_time": elapsed,
                "capture_num": capture_count,
                "waveforms": waveforms,
            }

            # Save to file or memory
            if output_dir:
                timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
                saved_paths = []
                for ch, waveform in waveforms.items():
                    filename = output_path / f"ch{ch}_{timestamp_str}.{file_format}"
                    try:
                        self.scope.waveform.save_waveform(waveform, str(filename), format=file_format)
                    except Exception as exc:
                        # Saves get their own handler rather than falling through to
                        # the loop's broad one. That handler logs and CONTINUES, so a
                        # rejected file_format used to fail identically on every
                        # iteration for the whole duration while the function returned
                        # an empty list -- an overnight run producing an empty
                        # directory and no signal at all. That is how the npz default
                        # stayed broken (see the 5.6.0 changelog).
                        save_failures += 1
                        logger.error(f"Failed to save channel {ch} to {filename}: {exc}")
                        if files_written == 0:
                            # Nothing has ever been written, so this is configuration,
                            # not a transient hiccup, and every later attempt fails the
                            # same way. Stop now. Once one file has landed the
                            # configuration is proven and later failures are tolerated.
                            #
                            # This abandons any remaining channels in THIS capture, which
                            # is deliberate: a rejected format or an unwritable directory
                            # is channel-independent, so ch2 would fail exactly as ch1
                            # just did. (The reverse order is tolerated -- if ch1 wrote
                            # and ch2 failed, the configuration is already proven.)
                            fatal_save_error = exc
                            break
                        continue
                    files_written += 1
                    saved_paths.append(str(filename))
                # Metadata without the arrays -- they are on disk. This mode used to
                # append nothing at all, so a caller could not tell how many captures
                # happened, where they went, or whether any had failed.
                saved_record = {key: value for key, value in capture_data.items() if key != "waveforms"}
                saved_record["files"] = saved_paths
                results.append(saved_record)
                if fatal_save_error is not None:
                    break
                logger.debug(f"Saved capture {capture_count}")
            else:
                results.append(capture_data)

            if progress_callback:
                remaining = duration - elapsed
                status = f"Captured {capture_count}, {remaining:.1f}s remaining"
                progress_callback(capture_count, status)

            # Wait for next interval
            capture_duration = time.time() - capture_start
            sleep_time = max(0, interval - capture_duration)
            if sleep_time > 0:
                time.sleep(sleep_time)

        except KeyboardInterrupt:
            logger.info("Continuous capture interrupted by user")
            break
        except Exception as e:
            logger.error(f"Error during continuous capture: {e}")

    if fatal_save_error is not None:
        message = f"Continuous capture aborted after {capture_count} capture(s): no file was written. Check output_dir and file_format={file_format!r}."
        # Preserve the original type when it is already one of ours -- wrapping an
        # InvalidParameterError in a plain SiglentError would stop a caller that
        # catches the specific type from catching it at all.
        if isinstance(fatal_save_error, SiglentError):
            raise type(fatal_save_error)(f"{message} ({fatal_save_error})") from fatal_save_error
        raise SiglentError(message) from fatal_save_error
    if save_failures:
        logger.warning(f"Continuous capture: {save_failures} save(s) failed, {files_written} file(s) written")
    elif output_dir and files_written == 0:
        # No save was even ATTEMPTED -- every capture yielded no waveforms, because
        # the channels are disabled or acquire() failed each time (both handled by
        # the inner handler above, which logs and continues). Without this the run
        # ends silently against an empty directory: the same symptom the save
        # handler fixes, reached by a different route.
        logger.warning(f"Continuous capture wrote no files: {capture_count} capture(s) produced no waveforms to save. Check that the requested channels are enabled.")

    logger.info(f"Continuous capture complete: {capture_count} captures over {duration}s")
    return results

save_data

save_data(waveforms: Dict[int, WaveformData], filename: str, format: Optional[str] = None) -> None

Save captured waveform data to file.

Parameters:

Name	Type	Description	Default
waveforms	Dict[int, WaveformData]	Dictionary mapping channel number to WaveformData	required
filename	str	Output filename	required
format	Optional[str]	File format - one of 'CSV', 'CSV_ENHANCED', 'NPY', 'MAT', 'HDF5'. If None (default), the format is auto-detected from each generated per-channel filename's extension.	None

Example

data = collector.capture_single([1, 2]) collector.save_data(data, 'measurement.npz')

writes measurement_ch1.npz and measurement_ch2.npz -- one file

per channel, suffixed with "_ch{n}" before the extension.

Source code in scpi_control/automation.py

def save_data(self, waveforms: Dict[int, WaveformData], filename: str, format: Optional[str] = None) -> None:
    """Save captured waveform data to file.

    Args:
        waveforms: Dictionary mapping channel number to WaveformData
        filename: Output filename
        format: File format - one of 'CSV', 'CSV_ENHANCED', 'NPY', 'MAT', 'HDF5'.
            If None (default), the format is auto-detected from each generated
            per-channel filename's extension.

    Example:
        >>> data = collector.capture_single([1, 2])
        >>> collector.save_data(data, 'measurement.npz')
        # writes measurement_ch1.npz and measurement_ch2.npz -- one file
        # per channel, suffixed with "_ch{n}" before the extension.
    """
    for ch, waveform in waveforms.items():
        base, ext = filename.rsplit(".", 1) if "." in filename else (filename, format or "npz")
        ch_filename = f"{base}_ch{ch}.{ext}"
        self.scope.waveform.save_waveform(waveform, ch_filename, format=format)
        logger.info(f"Saved channel {ch} to {ch_filename}")

save_batch

save_batch(batch_results: List[Dict[str, Any]], output_dir: str, format: Optional[str] = None) -> None

Save batch capture results to directory.

Parameters:

Name	Type	Description	Default
batch_results	List[Dict[str, Any]]	List of batch capture results	required
output_dir	str	Output directory path	required
format	Optional[str]	File format - one of 'CSV', 'CSV_ENHANCED', 'NPY', 'MAT', 'HDF5'. If None (default), the format is auto-detected from each generated filename's extension.	None

Example

results = collector.batch_capture(...) collector.save_batch(results, 'batch_output')

Source code in scpi_control/automation.py

def save_batch(self, batch_results: List[Dict[str, Any]], output_dir: str, format: Optional[str] = None) -> None:
    """Save batch capture results to directory.

    Args:
        batch_results: List of batch capture results
        output_dir: Output directory path
        format: File format - one of 'CSV', 'CSV_ENHANCED', 'NPY', 'MAT', 'HDF5'.
            If None (default), the format is auto-detected from each generated
            filename's extension.

    Example:
        >>> results = collector.batch_capture(...)
        >>> collector.save_batch(results, 'batch_output')
    """
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    # Save metadata
    metadata = {
        "total_captures": len(batch_results),
        "timestamp": datetime.now().isoformat(),
        "configurations": [r["config"] for r in batch_results],
    }

    metadata_file = output_path / "metadata.txt"
    with open(metadata_file, "w") as f:
        f.write(f"Batch Capture Metadata\n")
        f.write(f"=====================\n\n")
        f.write(f"Total Captures: {metadata['total_captures']}\n")
        f.write(f"Timestamp: {metadata['timestamp']}\n\n")
        f.write(f"Configurations:\n")
        for i, config in enumerate(metadata["configurations"]):
            f.write(f"  {i+1}. {config}\n")

    # Save waveforms
    for i, result in enumerate(batch_results):
        config_str = "_".join([f"{k}={v}" for k, v in result["config"].items()]).replace("/", "-")
        trigger_num = result["trigger_num"]

        ext = format or "npz"
        for ch, waveform in result["waveforms"].items():
            filename = f"capture_{i:04d}_ch{ch}_{config_str}_trig{trigger_num}.{ext}"
            filepath = output_path / filename
            self.scope.waveform.save_waveform(waveform, str(filepath), format=format)

    logger.info(f"Saved {len(batch_results)} captures to {output_path}")

analyze_waveform

analyze_waveform(waveform: WaveformData) -> Dict[str, float]

Analyze a waveform and extract common measurements.

Parameters:

Name	Type	Description	Default
waveform	WaveformData	WaveformData object to analyze	required

Returns:

Type	Description
Dict[str, float]	Dictionary of measurement names and values

Example

data = collector.capture_single([1]) stats = collector.analyze_waveform(data[1]) print(f"Peak-to-peak: {stats['vpp']:.3f}V") print(f"RMS: {stats['rms']:.3f}V")

Source code in scpi_control/automation.py

def analyze_waveform(self, waveform: WaveformData) -> Dict[str, float]:
    """Analyze a waveform and extract common measurements.

    Args:
        waveform: WaveformData object to analyze

    Returns:
        Dictionary of measurement names and values

    Example:
        >>> data = collector.capture_single([1])
        >>> stats = collector.analyze_waveform(data[1])
        >>> print(f"Peak-to-peak: {stats['vpp']:.3f}V")
        >>> print(f"RMS: {stats['rms']:.3f}V")
    """
    voltage = waveform.voltage

    analysis = {
        "vpp": np.max(voltage) - np.min(voltage),
        "amplitude": (np.max(voltage) - np.min(voltage)) / 2,
        "max": np.max(voltage),
        "min": np.min(voltage),
        "mean": np.mean(voltage),
        "rms": np.sqrt(np.mean(voltage**2)),
        "std_dev": np.std(voltage),
        "median": np.median(voltage),
    }

    # Try to detect frequency (simple zero-crossing method)
    frequency = 0.0
    period = 0.0

    try:
        mean_val = analysis["mean"]
        crossings = np.where(np.diff(np.sign(voltage - mean_val)))[0]

        # Estimate sample interval from the time axis, falling back to sample_rate
        dt = float(np.mean(np.diff(waveform.time))) if len(waveform.time) > 1 else None
        if (dt is None or dt <= 0) and getattr(waveform, "sample_rate", None):
            if waveform.sample_rate > 0:
                dt = 1.0 / float(waveform.sample_rate)

        if len(crossings) > 2 and dt and dt > 0:
            # Average time between positive-going zero crossings
            periods = np.diff(crossings[::2]) * dt
            avg_period = float(np.mean(periods))
            if avg_period > 0:
                period = avg_period
                frequency = 1.0 / avg_period
    except Exception:
        # Keep zero defaults on parsing errors
        pass

    analysis["frequency"] = frequency
    analysis["period"] = period

    return analysis

TriggerWaitCollector

TriggerWaitCollector(host: str, port: int = 5025, timeout: float = 5.0, connection: Optional[BaseConnection] = None, dialect: Optional[str] = None)

Specialized collector for waiting on specific trigger conditions.

Useful for capturing events that occur sporadically or based on specific signal conditions.

Initialize trigger wait collector.

Parameters:

Name	Type	Description	Default
host	str	IP address or hostname of the oscilloscope	required
port	int	TCP port for SCPI communication (default: 5025, the Siglent raw SCPI socket; 5024 is the telnet-style port with prompts and is not recommended)	5025
timeout	float	Command timeout in seconds	5.0
connection	Optional[BaseConnection]	Optional connection implementation (e.g., MockConnection for offline tests)	None
dialect	Optional[str]	Optional SCPI dialect override passed to Oscilloscope ("legacy" or "modern"; None auto-detects)	None

Source code in scpi_control/automation.py

def __init__(
    self,
    host: str,
    port: int = 5025,
    timeout: float = 5.0,
    connection: Optional[BaseConnection] = None,
    dialect: Optional[str] = None,
):
    """Initialize trigger wait collector.

    Args:
        host: IP address or hostname of the oscilloscope
        port: TCP port for SCPI communication (default: 5025, the Siglent raw SCPI socket; 5024 is the telnet-style port with prompts and is not recommended)
        timeout: Command timeout in seconds
        connection: Optional connection implementation (e.g., MockConnection for offline tests)
        dialect: Optional SCPI dialect override passed to Oscilloscope ("legacy" or "modern"; None auto-detects)
    """
    self.collector = DataCollector(host, port, timeout, connection=connection, dialect=dialect)

wait_for_trigger

wait_for_trigger(channels: List[int], max_wait: float = 60.0, save_on_trigger: bool = True, output_dir: Optional[str] = None) -> Optional[Dict[int, WaveformData]]

Wait for a trigger event and capture waveform.

If the trigger mode has been set to NORMAL beforehand (e.g. via trigger.set_mode('NORMAL')), that mode is preserved and the wait completes on the first trigger event. In any other mode, the scope is switched to SINGLE and armed for a one-shot acquisition.

Parameters:

Name	Type	Description	Default
channels	List[int]	List of channel numbers to capture	required
max_wait	float	Maximum time to wait for trigger in seconds	60.0
save_on_trigger	bool	If True, save waveform when triggered	True
output_dir	Optional[str]	Directory to save waveforms (required if save_on_trigger=True)	None

Returns:

Type	Description
Optional[Dict[int, WaveformData]]	Captured waveforms or None if timeout

Example

with TriggerWaitCollector('192.168.1.100') as tc: ... # Configure trigger on channel 1, edge = rising, level = 1V ... tc.collector.scope.trigger.set_source(1) ... tc.collector.scope.trigger.set_slope('POS') ... tc.collector.scope.trigger.set_level(1, 1.0) ... ... # Wait for trigger ... data = tc.wait_for_trigger([1, 2], max_wait=30.0) ... if data: ... print("Trigger captured!")

Source code in scpi_control/automation.py

def wait_for_trigger(
    self,
    channels: List[int],
    max_wait: float = 60.0,
    save_on_trigger: bool = True,
    output_dir: Optional[str] = None,
) -> Optional[Dict[int, WaveformData]]:
    """Wait for a trigger event and capture waveform.

    If the trigger mode has been set to NORMAL beforehand (e.g. via
    ``trigger.set_mode('NORMAL')``), that mode is preserved and the wait
    completes on the first trigger event. In any other mode, the scope is
    switched to SINGLE and armed for a one-shot acquisition.

    Args:
        channels: List of channel numbers to capture
        max_wait: Maximum time to wait for trigger in seconds
        save_on_trigger: If True, save waveform when triggered
        output_dir: Directory to save waveforms (required if save_on_trigger=True)

    Returns:
        Captured waveforms or None if timeout

    Example:
        >>> with TriggerWaitCollector('192.168.1.100') as tc:
        ...     # Configure trigger on channel 1, edge = rising, level = 1V
        ...     tc.collector.scope.trigger.set_source(1)
        ...     tc.collector.scope.trigger.set_slope('POS')
        ...     tc.collector.scope.trigger.set_level(1, 1.0)
        ...
        ...     # Wait for trigger
        ...     data = tc.wait_for_trigger([1, 2], max_wait=30.0)
        ...     if data:
        ...         print("Trigger captured!")
    """
    # Honor a user-configured NORMAL trigger mode; otherwise arm a
    # one-shot SINGLE acquisition. trigger.mode is dialect-normalized.
    current_mode = self.collector.scope.trigger.mode

    if current_mode == "NORM":
        # NORMAL mode re-arms after every trigger and never reports
        # STOP, so watch for the trigger event itself.
        done_states = {"TRIGD", "STOP"}
    else:
        self.collector.scope.trigger_single()
        done_states = {"STOP"}

    start_time = time.time()
    while (time.time() - start_time) < max_wait:
        status = self.collector.scope.acquisition_status()

        if status in done_states:
            # Trigger occurred, capture waveform
            logger.info("Trigger detected!")
            waveforms = {}
            for ch in channels:
                try:
                    waveforms[ch] = self.collector.scope.waveform.acquire(ch)
                except Exception as e:
                    logger.error(f"Failed to capture channel {ch}: {e}")

            if save_on_trigger and output_dir:
                timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
                self.collector.save_data(waveforms, f"{output_dir}/trigger_{timestamp_str}")

            return waveforms

        time.sleep(0.1)  # Check every 100ms

    logger.warning(f"Trigger timeout after {max_wait}s")
    return None

See Also

• Oscilloscope - Main oscilloscope control class for SCPI communication
• Waveform - Waveform acquisition and data handling
• Measurement - Automated measurements (frequency, voltage, timing)