Reference Waveform

Reference waveform management

Reference waveform storage and management for comparison.

ReferenceWaveform

ReferenceWaveform(storage_dir: Optional[str] = None)

Manages reference waveforms for comparison with live data.

Reference waveforms are stored as NPZ files in a user directory, allowing users to capture and compare against known-good signals.

Initialize reference waveform manager.

Parameters:

Name	Type	Description	Default
storage_dir	Optional[str]	Directory to store reference waveforms. Defaults to ~/.siglent/references/	None

Source code in scpi_control/reference_waveform.py

def __init__(self, storage_dir: Optional[str] = None):
    """Initialize reference waveform manager.

    Args:
        storage_dir: Directory to store reference waveforms.
                    Defaults to ~/.siglent/references/
    """
    if storage_dir is None:
        # Use default directory in user's home
        home_dir = Path.home()
        storage_dir = home_dir / ".siglent" / "references"
    else:
        storage_dir = Path(storage_dir)

    self.storage_dir = Path(storage_dir)

    # Create directory if it doesn't exist
    self.storage_dir.mkdir(parents=True, exist_ok=True)

    logger.info(f"Reference waveform storage initialized at: {self.storage_dir}")

save_reference

save_reference(waveform, name: str, metadata: Optional[Dict[str, Any]] = None) -> str

Save a waveform as a reference.

Parameters:

Name	Type	Description	Default
waveform		WaveformData object	required
name	str	Reference name (will be sanitized)	required
metadata	Optional[Dict[str, Any]]	Optional metadata dictionary	None

Returns:

Type	Description
str	Path to saved reference file

Raises:

Type	Description
ValueError	If waveform is None or name is empty
IOError	If save fails

Source code in scpi_control/reference_waveform.py

def save_reference(self, waveform, name: str, metadata: Optional[Dict[str, Any]] = None) -> str:
    """Save a waveform as a reference.

    Args:
        waveform: WaveformData object
        name: Reference name (will be sanitized)
        metadata: Optional metadata dictionary

    Returns:
        Path to saved reference file

    Raises:
        ValueError: If waveform is None or name is empty
        IOError: If save fails
    """
    if waveform is None:
        raise ValueError("Cannot save None waveform as reference")

    if not name or not name.strip():
        raise ValueError("Reference name cannot be empty")

    # Sanitize name (remove special characters)
    safe_name = self._sanitize_name(name)

    # Create filename
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"{safe_name}_{timestamp}.npz"
    filepath = self.storage_dir / filename

    # Prepare metadata
    if metadata is None:
        metadata = {}

    metadata["name"] = name
    metadata["timestamp"] = datetime.now().isoformat()
    metadata["channel"] = getattr(waveform, "channel", "Unknown")

    # Add waveform statistics
    metadata["min_voltage"] = float(np.min(waveform.voltage))
    metadata["max_voltage"] = float(np.max(waveform.voltage))
    metadata["mean_voltage"] = float(np.mean(waveform.voltage))
    metadata["std_voltage"] = float(np.std(waveform.voltage))
    metadata["num_samples"] = len(waveform.voltage)
    metadata["time_span"] = float(waveform.time[-1] - waveform.time[0]) if len(waveform.time) > 1 else 0.0

    try:
        # Save as NPZ with compression
        np.savez_compressed(filepath, time=waveform.time, voltage=waveform.voltage, metadata=metadata)

        logger.info(f"Reference waveform saved: {filepath}")
        return str(filepath)

    except Exception as e:
        logger.error(f"Failed to save reference waveform: {e}")
        raise IOError(f"Failed to save reference: {e}")

load_reference

load_reference(name: str) -> Optional[Dict[str, Any]]

Load a reference waveform by name.

Parameters:

Name	Type	Description	Default
name	str	Reference name or filename	required

Returns:

Type	Description
Optional[Dict[str, Any]]	Dictionary with 'time', 'voltage', and 'metadata' keys, or None if not found

Source code in scpi_control/reference_waveform.py

def load_reference(self, name: str) -> Optional[Dict[str, Any]]:
    """Load a reference waveform by name.

    Args:
        name: Reference name or filename

    Returns:
        Dictionary with 'time', 'voltage', and 'metadata' keys, or None if not found
    """
    # Try to find the file
    filepath = self._find_reference_file(name)

    if filepath is None:
        logger.warning(f"Reference waveform not found: {name}")
        return None

    try:
        # Load NPZ file
        data = np.load(filepath, allow_pickle=True)

        result = {
            "time": data["time"],
            "voltage": data["voltage"],
            "metadata": data["metadata"].item() if "metadata" in data else {},
            "filepath": str(filepath),
        }

        logger.info(f"Reference waveform loaded: {filepath}")
        return result

    except Exception as e:
        logger.error(f"Failed to load reference waveform: {e}")
        return None

list_references

list_references() -> List[Dict[str, Any]]

List all available reference waveforms.

Returns:

Type	Description
List[Dict[str, Any]]	List of dictionaries with reference information

Source code in scpi_control/reference_waveform.py

def list_references(self) -> List[Dict[str, Any]]:
    """List all available reference waveforms.

    Returns:
        List of dictionaries with reference information
    """
    references = []

    try:
        # Find all NPZ files in storage directory
        for filepath in self.storage_dir.glob("*.npz"):
            try:
                # Load metadata without loading full data
                data = np.load(filepath, allow_pickle=True)
                metadata = data["metadata"].item() if "metadata" in data else {}

                ref_info = {
                    "filename": filepath.name,
                    "filepath": str(filepath),
                    "name": metadata.get("name", filepath.stem),
                    "timestamp": metadata.get("timestamp", ""),
                    "channel": metadata.get("channel", "Unknown"),
                    "num_samples": metadata.get("num_samples", 0),
                    "time_span": metadata.get("time_span", 0.0),
                    "min_voltage": metadata.get("min_voltage", 0.0),
                    "max_voltage": metadata.get("max_voltage", 0.0),
                    "file_size": filepath.stat().st_size,
                    "modified_time": datetime.fromtimestamp(filepath.stat().st_mtime).isoformat(),
                }

                references.append(ref_info)

            except Exception as e:
                logger.warning(f"Failed to read reference {filepath.name}: {e}")

        # Sort by timestamp (most recent first)
        references.sort(key=lambda x: x.get("timestamp", ""), reverse=True)

        logger.info(f"Found {len(references)} reference waveform(s)")
        return references

    except Exception as e:
        logger.error(f"Failed to list references: {e}")
        return []

delete_reference

delete_reference(name: str) -> bool

Delete a reference waveform.

Parameters:

Name	Type	Description	Default
name	str	Reference name or filename	required

Returns:

Type	Description
bool	True if deleted successfully, False otherwise

Source code in scpi_control/reference_waveform.py

def delete_reference(self, name: str) -> bool:
    """Delete a reference waveform.

    Args:
        name: Reference name or filename

    Returns:
        True if deleted successfully, False otherwise
    """
    filepath = self._find_reference_file(name)

    if filepath is None:
        logger.warning(f"Reference waveform not found: {name}")
        return False

    try:
        filepath.unlink()
        logger.info(f"Reference waveform deleted: {filepath}")
        return True

    except Exception as e:
        logger.error(f"Failed to delete reference: {e}")
        return False

rename_reference

rename_reference(old_name: str, new_name: str) -> bool

Rename a reference waveform.

Parameters:

Name	Type	Description	Default
old_name	str	Current reference name	required
new_name	str	New reference name	required

Returns:

Type	Description
bool	True if renamed successfully, False otherwise

Source code in scpi_control/reference_waveform.py

def rename_reference(self, old_name: str, new_name: str) -> bool:
    """Rename a reference waveform.

    Args:
        old_name: Current reference name
        new_name: New reference name

    Returns:
        True if renamed successfully, False otherwise
    """
    old_filepath = self._find_reference_file(old_name)

    if old_filepath is None:
        logger.warning(f"Reference waveform not found: {old_name}")
        return False

    try:
        # Load and update metadata
        data = np.load(old_filepath, allow_pickle=True)
        time = data["time"]
        voltage = data["voltage"]
        metadata = data["metadata"].item() if "metadata" in data else {}

        # Update name in metadata
        metadata["name"] = new_name

        # Create new filename
        safe_name = self._sanitize_name(new_name)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        new_filename = f"{safe_name}_{timestamp}.npz"
        new_filepath = self.storage_dir / new_filename

        # Save with new name
        np.savez_compressed(new_filepath, time=time, voltage=voltage, metadata=metadata)

        # Delete old file
        old_filepath.unlink()

        logger.info(f"Reference waveform renamed: {old_name} -> {new_name}")
        return True

    except Exception as e:
        logger.error(f"Failed to rename reference: {e}")
        return False

calculate_difference

calculate_difference(waveform, reference_data: Dict[str, Any]) -> Optional[np.ndarray]

Calculate difference between a waveform and reference.

Parameters:

Name	Type	Description	Default
waveform		WaveformData object	required
reference_data	Dict[str, Any]	Reference data dictionary from load_reference()	required

Returns:

Type	Description
Optional[ndarray]	Difference array (waveform - reference) or None if incompatible

Source code in scpi_control/reference_waveform.py

def calculate_difference(self, waveform, reference_data: Dict[str, Any]) -> Optional[np.ndarray]:
    """Calculate difference between a waveform and reference.

    Args:
        waveform: WaveformData object
        reference_data: Reference data dictionary from load_reference()

    Returns:
        Difference array (waveform - reference) or None if incompatible
    """
    if waveform is None or reference_data is None:
        return None

    try:
        ref_voltage = reference_data["voltage"]

        # Check if lengths match
        if len(waveform.voltage) != len(ref_voltage):
            logger.warning("Waveform and reference have different lengths, interpolating...")
            # Interpolate reference to match waveform time base
            ref_time = reference_data["time"]
            ref_voltage_interp = np.interp(waveform.time, ref_time, ref_voltage)
            return waveform.voltage - ref_voltage_interp
        else:
            return waveform.voltage - ref_voltage

    except Exception as e:
        logger.error(f"Failed to calculate difference: {e}")
        return None

calculate_correlation

calculate_correlation(waveform, reference_data: Dict[str, Any]) -> Optional[float]

Calculate correlation coefficient between waveform and reference.

Parameters:

Name	Type	Description	Default
waveform		WaveformData object	required
reference_data	Dict[str, Any]	Reference data dictionary from load_reference()	required

Returns:

Type	Description
Optional[float]	Correlation coefficient (0.0 to 1.0) or None if incompatible

Source code in scpi_control/reference_waveform.py

def calculate_correlation(self, waveform, reference_data: Dict[str, Any]) -> Optional[float]:
    """Calculate correlation coefficient between waveform and reference.

    Args:
        waveform: WaveformData object
        reference_data: Reference data dictionary from load_reference()

    Returns:
        Correlation coefficient (0.0 to 1.0) or None if incompatible
    """
    if waveform is None or reference_data is None:
        return None

    try:
        ref_voltage = reference_data["voltage"]

        # Interpolate if needed
        if len(waveform.voltage) != len(ref_voltage):
            ref_time = reference_data["time"]
            ref_voltage = np.interp(waveform.time, ref_time, ref_voltage)

        # Calculate correlation coefficient
        correlation = np.corrcoef(waveform.voltage, ref_voltage)[0, 1]

        return float(correlation)

    except Exception as e:
        logger.error(f"Failed to calculate correlation: {e}")
        return None

get_storage_size

get_storage_size() -> int

Get total size of reference storage in bytes.

Returns:

Type	Description
int	Total size in bytes

Source code in scpi_control/reference_waveform.py

def get_storage_size(self) -> int:
    """Get total size of reference storage in bytes.

    Returns:
        Total size in bytes
    """
    total_size = 0
    try:
        for filepath in self.storage_dir.glob("*.npz"):
            total_size += filepath.stat().st_size
    except Exception as e:
        logger.error(f"Failed to calculate storage size: {e}")

    return total_size

clear_all_references

clear_all_references() -> int

Delete all reference waveforms.

Returns:

Type	Description
int	Number of references deleted

Source code in scpi_control/reference_waveform.py

def clear_all_references(self) -> int:
    """Delete all reference waveforms.

    Returns:
        Number of references deleted
    """
    count = 0
    try:
        for filepath in self.storage_dir.glob("*.npz"):
            try:
                filepath.unlink()
                count += 1
            except Exception as e:
                logger.warning(f"Failed to delete {filepath.name}: {e}")

        logger.info(f"Cleared {count} reference waveform(s)")
        return count

    except Exception as e:
        logger.error(f"Failed to clear references: {e}")
        return count

See Also

• Oscilloscope - Main oscilloscope control class for SCPI communication
• Waveform - Waveform acquisition and data handling