visualdynamics.io.rattlesnake

rattlesnake

Reader for Rattlesnake vibration controller output files (.nc4).

Rattlesnake writes its results to netCDF4 with a channel table that includes each channel's engineering unit (read in whatever case it was typed), so an import is unit-aware with no unit declaration needed. Values are converted to SI as they are read, and each record keeps the unit the file named. The layout, for file_version 3.x:

• root attributes: sample_rate, hardware, file_version and others
• time_data (response_channels, time_samples), in each channel's engineering unit
• a channels group: node_number, node_direction, unit, channel_type and the other per-channel fields
• one group per environment (for example 'Random'), which may hold a specification CPSD matrix (specification_frequency_lines, specification_cpsd_matrix_*) and its limits (specification_warning_matrix and specification_abort_matrix, each (2, lines, channels), lower first)

A run is saved to two files that share no data: streaming writes the time histories, and a separate save writes what the environment computed from them. Both are read here, and so is a saved system-ID package.

load returns a dict: 'channel_table' -> ChannelTable, and whichever of these the file holds: 'time_data' -> TimeHistory (with 'time_data_2' and so on for later streams); '_specification' -> Specification (the auto-spectra with their warning and abort limits, plus the cross terms when the file holds values for them or full_cpsd is set), or a SineSweepSpecification or TransientSpecification for a sine or transient environment; '_frf' -> Frf; '_coherence' -> MultipleCoherence; and '_response_cpsd' and '_drive_cpsd' -> Psd. A system-ID package has no channel table and gives '_frf', '_coherence', '_excitation_psds' and '_noise_psds'.

A modal environment counts every enabled channel as a response and never excludes its references, so the drives appear among the responses and the saved matrix includes each drive against the drives. Those rows are identity and cross-talk by construction (|H| exactly 1 against itself, numerically zero against the others, coherence exactly 1), so they are dropped: they come from the arithmetic, not from a measurement, and would add a force-per-force axis to the plot.

A spectral file has no time data and, after Rattlesnake's own random save, no root attributes either, so the sample rate may be missing. The frequency axis then comes from the specification's lines, or with no specification the lines are numbered by index. The noise cross spectra a random save also writes are not read: they describe the measurement's noise floor and would double the number of objects for something rarely looked at.

The reader does not write .nc4 files. A .nc4 records a controller run, including hardware settings and environments, that Visual Dynamics does not hold, so a file written from here would describe a test that never happened (see docs/export.md). The one exception adds nothing of its own: save_window copies a run with its streams cut to a time window, and every other attribute, dimension and variable as the file has it.

Functions:

Name	Description
machine_memory	The machine's physical memory in bytes, or 0 when it cannot be
last_window	The start time of a window holding the last last seconds of a
window_name	The file name for a copy of a run cut to a window: the run's own
save_window	Copy a run with its time data cut to a window, and nothing else
stream_summary	The size of each of a run's streams and the memory an import of
stream_preview	One channel's envelope over a whole stream, read in slabs.
sniff	Whether path is a Rattlesnake .nc4 or .nc file this module
environment_kinds	{environment name: kind} for the environments the file holds.
run_kind	The kind of test the file holds, as the file states it.
project_type	The Visual Dynamics project type this file is a run of, or None
streamed_sysid_candidate	Whether a streamed save looks like a system ID: exactly two
load	Everything a Rattlesnake .nc4 holds, keyed by the names the

Classes

Functions:

machine_memory

machine_memory() -> int

The machine's physical memory in bytes, or 0 when it cannot be found.

This is the one query of the operating system this package makes, used to tell whether a stream about to be imported would take a large share of memory. POSIX systems are read through sysconf, Windows through GlobalMemoryStatusEx.

Source code in src/visualdynamics/io/rattlesnake.py

def machine_memory() -> int:
    """The machine's physical memory in bytes, or 0 when it cannot be
    found.

    This is the one query of the operating system this package makes,
    used to tell whether a stream about to be imported would take a
    large share of memory. POSIX systems are read through `sysconf`,
    Windows through `GlobalMemoryStatusEx`.
    """
    try:
        if hasattr(os, 'sysconf'):
            return int(os.sysconf('SC_PAGE_SIZE')) * int(os.sysconf('SC_PHYS_PAGES'))
        import ctypes

        class Status(ctypes.Structure):
            _fields_ = [('dwLength', ctypes.c_ulong),
                        ('dwMemoryLoad', ctypes.c_ulong),
                        ('ullTotalPhys', ctypes.c_ulonglong),
                        ('ullAvailPhys', ctypes.c_ulonglong),
                        ('ullTotalPageFile', ctypes.c_ulonglong),
                        ('ullAvailPageFile', ctypes.c_ulonglong),
                        ('ullTotalVirtual', ctypes.c_ulonglong),
                        ('ullAvailVirtual', ctypes.c_ulonglong),
                        ('ullAvailExtendedVirtual', ctypes.c_ulonglong)]

        status = Status()
        status.dwLength = ctypes.sizeof(Status)
        ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(status))  # type: ignore[attr-defined]
        return int(status.ullTotalPhys)
    except Exception:  # noqa: BLE001 - an unanswerable question is 0, not a crash
        return 0

last_window

last_window(seconds: float, last: float) -> float | None

The start time of a window holding the last last seconds of a run.

The import dialog's Last field and a script's last= both use this function. A run seconds long is read from seconds - last to its end, and a run no longer than last is read whole, which the None result means. Asking for the last 100 s of a 28 s run is not an error; it is the whole run. Raises ValueError unless last is positive.

Parameters:

Name	Type	Description	Default
seconds	float	The instant of the run's last sample on its own clock.	required
last	float	How many seconds before the end the window opens; positive.	required

Returns:

Type	Description
float or None	The start to import from, or None for the whole run.

Source code in src/visualdynamics/io/rattlesnake.py

def last_window(seconds: float, last: float) -> float | None:
    """The start time of a window holding the last `last` seconds of a
    run.

    The import dialog's Last field and a script's `last=` both use
    this function. A run `seconds` long is read from `seconds - last`
    to its end, and a run no longer than `last` is read whole, which
    the None result means. Asking for the last 100 s of a 28 s run is
    not an error; it is the whole run. Raises ValueError unless `last`
    is positive.

    Parameters
    ----------
    seconds : float
        The instant of the run's last sample on its own clock.
    last : float
        How many seconds before the end the window opens; positive.

    Returns
    -------
    float or None
        The `start` to import from, or None for the whole run.
    """
    # One rule for the dialog's Last field and a script's last=
    # (Brandon, 2026-09-19).
    if not last > 0:
        raise ValueError(f'last must be a positive number of seconds, not {last!r}')
    if last >= seconds:
        return None
    return float(seconds - last)

window_name

window_name(path: str | PathLike, start: float | None, stop: float | None) -> str

The file name for a copy of a run cut to a window: the run's own name with the window added, for example random_30-60s.nc4.

The copy's time starts at zero again and nothing in the file says where the window began, so the name records it. An open start is written as 0 and an open stop as end (random_30s-end.nc4).

Parameters:

Name	Type	Description	Default
path	str or PathLike	The run.	required
start	float or None	The window, in seconds on the run's time axis; None for the run's start or end.	required
stop	float or None	The window, in seconds on the run's time axis; None for the run's start or end.	required

Returns:

Type	Description
str	A path in the same folder as the run.

Source code in src/visualdynamics/io/rattlesnake.py

def window_name(path: str | os.PathLike, start: float | None,
                stop: float | None) -> str:
    """The file name for a copy of a run cut to a window: the run's own
    name with the window added, for example `random_30-60s.nc4`.

    The copy's time starts at zero again and nothing in the file says
    where the window began, so the name records it. An open start is
    written as 0 and an open stop as `end` (`random_30s-end.nc4`).

    Parameters
    ----------
    path : str or os.PathLike
        The run.
    start, stop : float or None
        The window, in seconds on the run's time axis; None for the
        run's start or end.

    Returns
    -------
    str
        A path in the same folder as the run.
    """
    stem, suffix = os.path.splitext(os.fspath(path))
    first = 0.0 if start is None else float(start)
    span = f'{first:g}s-end' if stop is None else f'{first:g}-{float(stop):g}s'
    return f'{stem}_{span}{suffix or ".nc4"}'

save_window

save_window(path: str | PathLike, target: str | PathLike, start: float | None = None, stop: float | None = None, progress: Callable[[int, int], None] | None = None) -> dict[str, int]

Copy a run with its time data cut to a window, and nothing else changed.

Every group, attribute, dimension and variable is written as the file has it (the hardware settings, the channel fields, each environment's control settings and specification) except the streams and their time_samples dimensions. Each stream is cut to the samples the window holds, the same samples an import of that window reads. A stream the window misses is written with no samples, as a spectral file's is. Nothing is added: the file has no attribute for where the window began, so the copy's time starts at zero again (window_name puts the window in the file name instead). All channels are kept, because leaving one out would mean rewriting the channel table, each environment's active channels and its control indices.

Raises ValueError when target is the run itself, when the run has no time data, when the window runs backward, or when it holds fewer than two samples of every stream.

Parameters:

Name	Type	Description	Default
path	str or PathLike	The run, a Rattlesnake .nc4 with time data.	required
target	str or PathLike	Where the copy goes; not the run itself.	required
start	float	The window, in seconds on the run's time axis, both ends included; the run's start and end when not given.	None
stop	float	The window, in seconds on the run's time axis, both ends included; the run's start and end when not given.	None
progress	callable	Called with (done, total) samples as the streams are copied. An exception it raises stops the copy, and the partly written copy is deleted.	None

Returns:

Type	Description
dict	{stream variable: samples kept}.

Source code in src/visualdynamics/io/rattlesnake.py

def save_window(path: str | os.PathLike, target: str | os.PathLike,
                start: float | None = None, stop: float | None = None,
                progress: Callable[[int, int], None] | None = None) -> dict[str, int]:
    """Copy a run with its time data cut to a window, and nothing else
    changed.

    Every group, attribute, dimension and variable is written as the
    file has it (the hardware settings, the channel fields, each
    environment's control settings and specification) except the
    streams and their `time_samples` dimensions. Each stream is cut to
    the samples the window holds, the same samples an import of that
    window reads. A stream the window misses is written with no
    samples, as a spectral file's is. Nothing is added: the file has no
    attribute for where the window began, so the copy's time starts at
    zero again (`window_name` puts the window in the file name
    instead). All channels are kept, because leaving one out would
    mean rewriting the channel table, each environment's active
    channels and its control indices.

    Raises ValueError when `target` is the run itself, when the run has
    no time data, when the window runs backward, or when it holds fewer
    than two samples of every stream.

    Parameters
    ----------
    path : str or os.PathLike
        The run, a Rattlesnake `.nc4` with time data.
    target : str or os.PathLike
        Where the copy goes; not the run itself.
    start, stop : float, optional
        The window, in seconds on the run's time axis, both ends
        included; the run's start and end when not given.
    progress : callable, optional
        Called with `(done, total)` samples as the streams are copied.
        An exception it raises stops the copy, and the partly written
        copy is deleted.

    Returns
    -------
    dict
        {stream variable: samples kept}.
    """
    import netCDF4

    if os.path.exists(target) and os.path.samefile(path, target):
        raise ValueError('the copy cannot be written over the run it is cut from')
    with netCDF4.Dataset(path) as ds:
        streams = {variable: dimension for variable, dimension, _key in _streams(ds)}
        if not streams:
            raise ValueError(f'{os.path.basename(path)} holds no time data to cut')
        rate = float(ds.sample_rate)
        windows = {}
        for dimension in streams.values():
            found = _sample_window(start, stop, len(ds.dimensions[dimension]), rate)
            windows[dimension] = found if found is not None else (0, 0)
        if all(b - a < 2 for a, b in windows.values()):
            raise ValueError('the window holds fewer than two samples of any stream')
        from ..core.progress import Ticker

        ticker = Ticker(progress)
        ticker.add(sum(b - a for a, b in windows.values()))
        try:
            with netCDF4.Dataset(target, 'w', format=ds.data_model) as out:
                _copy_group(ds, out, windows, ticker)
        except BaseException:
            # a copy stopped part way is not a shorter run
            if os.path.exists(target):
                os.remove(target)
            raise
    return {variable: windows[dimension][1] - windows[dimension][0]
            for variable, dimension in streams.items()}

stream_summary

stream_summary(path: str | PathLike) -> dict[str, Any]

The size of each of a run's streams and the memory an import of it would take, without reading the stream data.

The import dialog uses this to ask whether a stream is a large share of the machine's memory. It is read from the file's dimensions alone, so it is fast even on a 22 GB run.

Parameters:

Name	Type	Description	Default
path	str or PathLike	A Rattlesnake .nc4.	required

Returns:

Type

Description

dict

'sample_rate'; 'channels', the table's coordinates in row order; 'units', their units as the file spells them; 'streams', one dict per stream with 'key' (the name the import gives it), 'variable', 'channels', 'samples', 'seconds' and 'bytes' (as float64, the size the import holds); 'memory', machine_memory(). A file with no channel table (a system-ID package) gives a 'sample_rate' of None and empty lists.

Source code in src/visualdynamics/io/rattlesnake.py

def stream_summary(path: str | os.PathLike) -> dict[str, Any]:
    """The size of each of a run's streams and the memory an import of
    it would take, without reading the stream data.

    The import dialog uses this to ask whether a stream is a large
    share of the machine's memory. It is read from the file's
    dimensions alone, so it is fast even on a 22 GB run.

    Parameters
    ----------
    path : str or os.PathLike
        A Rattlesnake `.nc4`.

    Returns
    -------
    dict
        'sample_rate'; 'channels', the table's coordinates in row
        order; 'units', their units as the file spells them;
        'streams', one dict per stream with 'key' (the name the
        import gives it), 'variable', 'channels', 'samples',
        'seconds' and 'bytes' (as float64, the size the import holds);
        'memory', `machine_memory()`. A file with no channel table (a
        system-ID package) gives a 'sample_rate' of None and empty
        lists.
    """
    import netCDF4

    with netCDF4.Dataset(path) as ds:
        if 'channels' not in ds.groups:
            return {'sample_rate': None, 'channels': [], 'units': [],
                    'streams': [], 'memory': machine_memory()}
        _table, dofs, units, _columns = _channel_table(ds)
        rate = float(ds.sample_rate)
        streams = []
        for variable, _dimension, key in _streams(ds):
            channels, samples = ds.variables[variable].shape
            streams.append({'key': key, 'variable': variable,
                            'channels': int(channels), 'samples': int(samples),
                            'seconds': (samples - 1) / rate,
                            'bytes': int(channels) * int(samples) * 8})
    return {'sample_rate': rate, 'channels': dofs, 'units': units,
            'streams': streams, 'memory': machine_memory()}

stream_preview

stream_preview(path: str | PathLike, channel: int | str = 0, stream: str = 'time_data', points: int = PREVIEW_POINTS) -> dict[str, Any]

One channel's envelope over a whole stream, read in slabs.

This is the plot an import window is chosen from: it shows where the run reaches level, where it stops, and any false start at the front. Each of points buckets keeps its smallest and largest sample (the same peak thinning the stage applies, done as the channel is read), so the preview of a 22 GB run holds one slab at a time in memory and returns a few thousand points. The file's chunks run along one channel at a time, so reading one channel is a sequential read.

Parameters:

Name	Type	Description	Default
path	str or PathLike	A Rattlesnake .nc4.	required
channel	int or str	The channel's row in the table, or its coordinate ('101Z+').	0
stream	str	Which stream variable ('time_data', 'time_data_1', …).	'time_data'
points	int	How many buckets to thin to.	PREVIEW_POINTS

Returns:

Type	Description
dict	'times', the bucket centers in seconds on the run's time axis; 'low' and 'high', each bucket's smallest and largest value in the file's own unit, not converted, and NaN where a bucket holds only NaN; 'dof', 'unit' and 'dim' of the channel; and 'samples' and 'sample_rate' of the stream.

Source code in src/visualdynamics/io/rattlesnake.py

def stream_preview(path: str | os.PathLike, channel: int | str = 0,
                   stream: str = 'time_data',
                   points: int = PREVIEW_POINTS) -> dict[str, Any]:
    """One channel's envelope over a whole stream, read in slabs.

    This is the plot an import window is chosen from: it shows where
    the run reaches level, where it stops, and any false start at the
    front. Each of `points` buckets keeps its smallest and largest
    sample (the same peak thinning the stage applies, done as the
    channel is read), so the preview of a 22 GB run holds one slab at
    a time in memory and returns a few thousand points. The file's
    chunks run along one channel at a time, so reading one channel is
    a sequential read.

    Parameters
    ----------
    path : str or os.PathLike
        A Rattlesnake `.nc4`.
    channel : int or str
        The channel's row in the table, or its coordinate ('101Z+').
    stream : str
        Which stream variable ('time_data', 'time_data_1', …).
    points : int
        How many buckets to thin to.

    Returns
    -------
    dict
        'times', the bucket centers in seconds on the run's time axis;
        'low' and 'high', each bucket's smallest and largest value in
        the file's own unit, not converted, and NaN where a bucket
        holds only NaN; 'dof', 'unit' and 'dim' of the channel; and
        'samples' and 'sample_rate' of the stream.
    """
    import netCDF4

    with netCDF4.Dataset(path) as ds:
        _table, dofs, units, _columns = _channel_table(ds)
        row = _channel_rows([channel], dofs)[0]
        # the file's own values in the file's own unit: the preview is
        # read against what the run was set up in, not converted
        _scale, dim, unit = _unit_scale(units[row])
        variable = ds.variables[stream]
        variable.set_auto_maskandscale(False)
        rate = float(ds.sample_rate)
        samples = int(variable.shape[1])
        bucket = max(1, -(-samples // max(1, int(points))))
        buckets = -(-samples // bucket)
        low = np.full(buckets, np.inf)
        high = np.full(buckets, -np.inf)
        # slabs that are whole buckets, so a bucket never straddles two
        slab = bucket * max(1, READ_SAMPLES // bucket)
        for start in range(0, samples, slab):
            stop = min(start + slab, samples)
            chunk = np.asarray(variable[row, start:stop], dtype=np.float64)
            if len(chunk) % bucket:
                chunk = np.concatenate(
                    [chunk, np.full(bucket - len(chunk) % bucket, np.nan)])
            grid = chunk.reshape(-1, bucket)
            first = start // bucket
            # a bucket of nothing but NaN — a stream's lead-in before its
            # first sample arrived writes NaN, and the stress stream has
            # a hundred seconds of it — is a gap in the envelope, NaN,
            # said without numpy's warning about it
            with warnings.catch_warnings():
                warnings.simplefilter('ignore', RuntimeWarning)
                low[first:first + len(grid)] = np.nanmin(grid, axis=1)
                high[first:first + len(grid)] = np.nanmax(grid, axis=1)
    centers = (np.arange(buckets) * bucket + (bucket - 1) / 2.0) / rate
    centers[-1] = min(centers[-1], (samples - 1) / rate)
    return {'times': centers, 'low': low, 'high': high,
            'dof': dofs[row], 'unit': unit, 'dim': dim,
            'samples': samples, 'sample_rate': rate}

sniff

sniff(path: str | PathLike) -> bool

Whether path is a Rattlesnake .nc4 or .nc file this module can read: a streamed run, a spectral save or a system-ID package. Never raises; a file that cannot be opened is not one.

Source code in src/visualdynamics/io/rattlesnake.py

def sniff(path: str | os.PathLike) -> bool:
    """Whether `path` is a Rattlesnake `.nc4` or `.nc` file this module
    can read: a streamed run, a spectral save or a system-ID package.
    Never raises; a file that cannot be opened is not one."""
    if not str(path).lower().endswith(('.nc4', '.nc')):
        return False
    try:
        import netCDF4
        with netCDF4.Dataset(path) as ds:
            if 'channels' not in ds.groups:
                # a saved system-ID package: no channel table, one
                # group per environment holding the measured plant
                return any('frf_data_real' in group.variables
                           for group in ds.groups.values())
            # a run saves two files: the streamed time data, and what the
            # environment computed from it. Neither has the other's contents.
            if 'time_data' in ds.variables:
                return True
            return any(name in group.variables
                       for group in ds.groups.values()
                       for name in _SPECTRAL_MARKERS)
    except Exception:  # noqa: BLE001 - sniffers must not raise on foreign files
        return False

environment_kinds

environment_kinds(path: str | PathLike) -> dict[str, str]

{environment name: kind} for the environments the file holds.

The kinds are 'random', 'transient', 'sine', 'time' and 'modal', read from the file's environment type codes, or from what each environment group holds when the file has no codes. The result is empty when the file identifies none, for example an nc4 put together by hand.

Source code in src/visualdynamics/io/rattlesnake.py

def environment_kinds(path: str | os.PathLike) -> dict[str, str]:
    """{environment name: kind} for the environments the file holds.

    The kinds are 'random', 'transient', 'sine', 'time' and 'modal',
    read from the file's environment type codes, or from what each
    environment group holds when the file has no codes. The result is
    empty when the file identifies none, for example an nc4 put
    together by hand.
    """
    import netCDF4

    with netCDF4.Dataset(str(path)) as ds:
        return _environment_kinds(ds)

run_kind

run_kind(path: str | PathLike) -> str | None

The kind of test the file holds, as the file states it.

'modal', 'random', 'transient', 'sine' or 'time' for a run of one kind; 'mixed' when more than one environment drove the article (a random run with a sine sweep, for example); None when the file does not say. An environment that only records ('time') does not make a run mixed.

Source code in src/visualdynamics/io/rattlesnake.py

def run_kind(path: str | os.PathLike) -> str | None:
    """The kind of test the file holds, as the file states it.

    'modal', 'random', 'transient', 'sine' or 'time' for a run of one
    kind; 'mixed' when more than one environment drove the article (a
    random run with a sine sweep, for example); None when the file
    does not say. An environment that only records ('time') does not
    make a run mixed.
    """
    return _run_kind(environment_kinds(path).values())

project_type

project_type(path: str | PathLike) -> str | None

The Visual Dynamics project type this file is a run of, or None when there is no project type for that kind of run or the file does not say what kind it is.

A saved system-ID package is a 'System ID' project. A random run with a sine sweep is a 'Random and Sine' project (MIXED_PROJECTS), whose report covers both. Any other mixed run takes the type of its first kind in MIXED_PRECEDENCE. Every environment's specification is imported either way, so the type only decides which workflow the tree shows first.

Source code in src/visualdynamics/io/rattlesnake.py

def project_type(path: str | os.PathLike) -> str | None:
    """The Visual Dynamics project type this file is a run of, or None
    when there is no project type for that kind of run or the file does
    not say what kind it is.

    A saved system-ID package is a 'System ID' project. A random run
    with a sine sweep is a 'Random and Sine' project
    (`MIXED_PROJECTS`), whose report covers both. Any other mixed run
    takes the type of its first kind in `MIXED_PRECEDENCE`. Every
    environment's specification is imported either way, so the type
    only decides which workflow the tree shows first.
    """
    import netCDF4

    with netCDF4.Dataset(str(path)) as ds:
        kinds = set(_environment_kinds(ds).values())
        # a saved system-ID package carries no environment_types — it
        # is the measured plant alone, and that is its own kind of
        # test. A *streamed* sysid save is indistinguishable from a
        # run of its environment and keeps that type; the person can
        # switch it to System ID.
        if not kinds and any('frf_data_real' in group.variables
                             for group in ds.groups.values()):
            return 'System ID'
    kind = _run_kind(kinds)
    if kind == 'mixed':
        driving = frozenset(kinds - PASSIVE_KINDS)
        if driving in MIXED_PROJECTS:
            return MIXED_PROJECTS[driving]
        kind = next((k for k in MIXED_PRECEDENCE if k in kinds), None)
    return PROJECT_FOR_KIND.get(kind)

streamed_sysid_candidate

streamed_sysid_candidate(path: str | PathLike) -> bool

Whether a streamed save looks like a system ID: exactly two streams, a quiet one and then a loud one (the ambient measurement and the driven excitation).

The file cannot answer this by itself, because a run whose stream was stopped and restarted once has the same layout. So a True result is a reason to ask the person, not to decide. The level check keeps the question from being asked about every two-stream file: a restarted run's two streams are at the same level, while a system ID's differ by the whole test. The driven stream must be more than 10 dB above the quiet one (standard deviation of the first SNIFF_SAMPLES samples of each), which is well under any real ambient-to-driven difference and well over a level change within one run. Never raises; False for a file that cannot be read.

Source code in src/visualdynamics/io/rattlesnake.py

def streamed_sysid_candidate(path: str | os.PathLike) -> bool:
    """Whether a streamed save looks like a system ID: exactly two
    streams, a quiet one and then a loud one (the ambient measurement
    and the driven excitation).

    The file cannot answer this by itself, because a run whose stream
    was stopped and restarted once has the same layout. So a True
    result is a reason to ask the person, not to decide. The level
    check keeps the question from being asked about every two-stream
    file: a restarted run's two streams are at the same level, while a
    system ID's differ by the whole test. The driven stream must be
    more than 10 dB above the quiet one (standard deviation of the
    first `SNIFF_SAMPLES` samples of each), which is well under any
    real ambient-to-driven difference and well over a level change
    within one run. Never raises; False for a file that cannot be read.
    """
    if not str(path).lower().endswith(('.nc4', '.nc')):
        return False
    try:
        import netCDF4
        import numpy as np

        with netCDF4.Dataset(str(path)) as ds:
            if ('time_data' not in ds.variables
                    or 'time_data_1' not in ds.variables
                    or 'time_data_2' in ds.variables):
                return False
            # a slice, not the stream: the level of a run is settled in
            # its first seconds, and reading a 22 GB recording twice to
            # take a standard deviation was most of what an import
            # cost before it began (2026-09-18)
            quiet = float(np.asarray(
                ds.variables['time_data'][:, :SNIFF_SAMPLES]).std())
            driven = float(np.asarray(
                ds.variables['time_data_1'][:, :SNIFF_SAMPLES]).std())
        # a simulated ambient can be exact silence; that is the
        # extreme of the same shape, not a different case
        return driven > 10 ** (10 / 20) * quiet and driven > 0.0
    except Exception:  # noqa: BLE001 - sniffers must not raise on foreign files
        return False

load

load(path: str | PathLike, full_cpsd: bool = False, start: float | None = None, stop: float | None = None, channels: Iterable[int | str] | None = None, progress: Callable[[int, int], None] | None = None) -> dict[str, Any]

Everything a Rattlesnake .nc4 holds, keyed by the names the tree shows.

The streamed time data, the channel table, and each environment's specification, FRF, coherence and spectra; or a saved system-ID package on its own. start, stop and channels import part of the streams, for a long run that would not fit in memory whole (the import dialog asks for them; a script passes them). The window includes both ends, on the run's time axis, and the records keep those times. A stream the window misses entirely is left out, and each windowed record's comment says which part of which file it is. A part of a stream is imported as one recording, so it is never split into frames the way a spectral save's stack of captures is.

Parameters:

Name	Type	Description	Default
path	str or PathLike	The file.	required
full_cpsd	bool	Import the cross terms of a random specification's CPSD matrix even when its off-diagonal holds only zeros. Cross terms that hold values are imported either way, and a pair the file never wrote (all NaN) never is.	False
start	float	The window in seconds; either end open when omitted.	None
stop	float	The window in seconds; either end open when omitted.	None
channels	iterable of int or str	Which channels of the streams to import, by table row or by coordinate ('101Z+'). All of them when omitted. An environment's virtual responses are included only when every control channel they are computed from is selected.	None
progress	callable	Called with (done, total) samples as the streams are read.	None

Returns:

Type	Description
dict	Objects by key: 'time_data' (and 'time_data_2', …), 'channel_table', and '_specification', '_frf', '_coherence', '_response_cpsd' and the rest as the file has them.

Source code in src/visualdynamics/io/rattlesnake.py

def load(path: str | os.PathLike, full_cpsd: bool = False,
         start: float | None = None, stop: float | None = None,
         channels: Iterable[int | str] | None = None,
         progress: Callable[[int, int], None] | None = None) -> dict[str, Any]:
    """Everything a Rattlesnake `.nc4` holds, keyed by the names the
    tree shows.

    The streamed time data, the channel table, and each environment's
    specification, FRF, coherence and spectra; or a saved system-ID
    package on its own. `start`, `stop` and `channels` import part of
    the streams, for a long run that would not fit in memory whole
    (the import dialog asks for them; a script passes them). The
    window includes both ends, on the run's time axis, and the records
    keep those times. A stream the window misses entirely is left out,
    and each windowed record's comment says which part of which file
    it is. A part of a stream is imported as one recording, so it is
    never split into frames the way a spectral save's stack of
    captures is.

    Parameters
    ----------
    path : str or os.PathLike
        The file.
    full_cpsd : bool
        Import the cross terms of a random specification's CPSD matrix
        even when its off-diagonal holds only zeros. Cross terms that
        hold values are imported either way, and a pair the file never
        wrote (all NaN) never is.
    start, stop : float, optional
        The window in seconds; either end open when omitted.
    channels : iterable of int or str, optional
        Which channels of the streams to import, by table row or by
        coordinate ('101Z+'). All of them when omitted. An
        environment's virtual responses are included only when every
        control channel they are computed from is selected.
    progress : callable, optional
        Called with `(done, total)` samples as the streams are read.

    Returns
    -------
    dict
        Objects by key: 'time_data' (and 'time_data_2', …),
        'channel_table', and '<environment>_specification', '_frf',
        '_coherence', '_response_cpsd' and the rest as the file has them.
    """
    import netCDF4

    out = {}
    with netCDF4.Dataset(path) as ds:
        if 'channels' not in ds.groups:
            return _load_sysid_package(ds, path)
        table, dofs, units, columns = _channel_table(ds)
        num_channels = len(dofs)
        out['channel_table'] = table

        scales_dims = [_unit_scale(u) for u in units]
        chosen = None if channels is None else _channel_rows(channels, dofs)
        base = os.path.basename(str(path))
        # a channel is a drive exactly when it has a feedback device, which
        # is how rattlesnake itself decides (`Channel.is_output_channel`)
        feedback = columns.get('feedback_device', [''] * num_channels)
        drive_channels = [i for i, value in enumerate(feedback)
                          if str(value).strip()]
        _record_environments(ds, table, drive_channels, scales_dims)

        # the streams the window keeps, sized before any is read, so
        # the progress is one count over all of them
        from ..core.progress import Ticker

        ticker = Ticker(progress)
        kept = []
        for variable, _dimension, key in _streams(ds):
            sample_rate = float(ds.sample_rate)
            samples = int(ds.variables[variable].shape[1])
            span = _sample_window(start, stop, samples, sample_rate)
            if span is None:
                continue
            kept.append((variable, key, sample_rate, samples, span))
        ticker.add(sum(last - first for _v, _k, _r, _s, (first, last) in kept))
        for variable, key, sample_rate, samples, (first, last) in kept:
            rows = list(range(num_channels)) if chosen is None else chosen
            # a part of the run is a recording of that part: the window
            # is read as a slice of the file, never the whole and cut
            partial = (first, last) != (0, samples) or chosen is not None
            time_data = _read_stream(ds.variables[variable], rows, first, last,
                                     ticker)
            scales = np.array([scales_dims[i][0] for i in rows])
            rows_dof = [dofs[i] for i in rows]
            rows_dim = [scales_dims[i][1] for i in rows]
            rows_unit = [scales_dims[i][2] for i in rows]
            # a windowed record says so on every row: the clock is the
            # run's, and a record starting at 120 s with nothing to say
            # why would read as a recording that began late
            note = ('' if (first, last) == (0, samples) else
                    f'{first / sample_rate:g} to {(last - 1) / sample_rate:g} s '
                    f'of {base}')
            rows_comment = [note] * len(rows)
            # the virtual responses an environment controlled, as more
            # records of the same history: the controller's own matrix
            # over the raw channels, so they share the frames, the
            # averaging and every act the raw channels get — processed
            # once, beside them, not twice (Brandon, 2026-09-18: "all the
            # time data can be processed identically"). Taken from the
            # raw values, before the scaling below rewrites them
            virtual = []
            for env_name, group in ds.groups.items():
                if 'control_channel_indices' not in group.variables:
                    continue
                indices = np.asarray(
                    group.variables['control_channel_indices'][()], dtype=int)
                found = _response_transformation(group, indices, scales_dims)
                if found is None or any(int(i) not in rows for i in indices):
                    # a subset missing a control channel cannot compute
                    # the virtual row; it is left out, not made up
                    continue
                matrix, labels, (scale, dim, unit) = found
                positions = [rows.index(int(i)) for i in indices]
                virtual.append((matrix @ time_data[positions]) * scale)
                rows_dof += labels
                rows_dim += [dim] * len(labels)
                rows_unit += [unit] * len(labels)
                rows_comment += [
                    (f'{note}; ' if note else '')
                    + f'row {label} of the {env_name} response transformation '
                    f'over {len(indices)} control channels' for label in labels]
            # scaled in place: the array is the file's read and nobody
            # else holds it, and a scaled copy beside it was the second
            # of the two whole copies an import cost (2026-09-18)
            time_data *= scales[:, np.newaxis]
            values = (np.vstack([time_data, *virtual]) if virtual
                      else time_data)
            frames = 0 if partial else _frame_count(ds, time_data.shape[1])
            if frames:
                # (channels, frames * samples) -> one record per capture.
                # reshape only, so the samples are not copied.
                samples = time_data.shape[1] // frames
                values = values.reshape(-1, frames, samples).reshape(-1, samples)
                block = [f'avg {i + 1}' for _dof in rows_dof
                         for i in range(frames)]
                response_dof = [dof for dof in rows_dof for _ in range(frames)]
                repeat = frames
            else:
                block, response_dof, repeat = None, rows_dof, 1
            history = TimeHistory(
                # the window's own instants; a stack split into frames
                # has first = 0 and a frame's worth of columns
                abscissa=np.arange(first, first + values.shape[1],
                                   dtype=np.float64) / sample_rate,
                ordinate=values,
                response_dof=response_dof, block=block,
                ordinate_dim=[d for d in rows_dim for _ in range(repeat)],
                ordinate_unit=[u for u in rows_unit for _ in range(repeat)],
                comment=[c for c in rows_comment for _ in range(repeat)],
            )
            kind = _run_kind(_environment_kinds(ds).values())
            # a pure sine run has no averaging to describe: the sweep
            # is read sample by sample through a tracking filter, and
            # the only frame-shaped attributes in its file are the
            # sysid_* ones — the plant-measurement phase, not the
            # sweep. Shading its frames on the recording claimed an
            # analysis that never happens (Brandon, 2026-08-22).
            averaging = (None if kind == 'sine'
                         else _averaging(ds, values.shape[1],
                                         sample_rate))
            if averaging is not None:
                averaging = _started(history, averaging, kind)
            history.averaging = averaging
            # each channel's role, from the table read beside it — the
            # file's own statement (a feedback device makes a
            # reference), so FRFs start from the channels the
            # controller drove; the virtual rows are the controlled
            # responses they are
            table_roles = table.roles()
            roles = {(dof, dim): (table_roles[i] or 'response')
                     for i, dof, dim in zip(rows, rows_dof, rows_dim)}
            for dof, dim in zip(rows_dof[len(rows):], rows_dim[len(rows):]):
                roles[(dof, dim)] = 'response'
            history.roles = roles
            out[key] = history

        # rattlesnake's own spectral save keeps only the environment group;
        # the root attributes, the channel table and the time data all go.
        # So the sample rate may simply not be there.
        rate = (float(ds.sample_rate) if 'sample_rate' in ds.ncattrs()
                else None)
        for env_name, group in ds.groups.items():
            out.update(_spectra(env_name, group, dofs, scales_dims, rate,
                                drive_channels))

        for env_name, group in ds.groups.items():
            if 'specification_frequency_lines' not in group.variables:
                continue
            freq = np.asarray(group.variables['specification_frequency_lines'][()])
            cpsd = (np.asarray(group.variables['specification_cpsd_matrix_real'][()])
                    + 1j * np.asarray(
                        group.variables['specification_cpsd_matrix_imag'][()]))
            indices = np.asarray(group.variables['control_channel_indices'][()],
                                 dtype=int)
            channels = _control_channels(group, indices, dofs, scales_dims)
            if cpsd.shape[-1] != len(channels):
                raise ValueError(
                    f'{env_name}: the specification is over '
                    f'{cpsd.shape[-1]} channels but the environment '
                    f'names {len(channels)}')
            # (2, lines, channels), lower first: rattlesnake's own plotting
            # reads [1] as upper and [0] as lower
            bands = {}
            for kind, variable in (('warning', 'specification_warning_matrix'),
                                   ('abort', 'specification_abort_matrix')):
                if variable in group.variables:
                    matrix = np.abs(np.asarray(group.variables[variable][()]))
                    bands[f'{kind}_lower'] = matrix[0]
                    bands[f'{kind}_upper'] = matrix[1]

            # The controller's target is a full matrix, and its cross
            # terms are read when they are real numbers: a file whose
            # off-diagonal is all NaN, or all zero, wrote placeholders
            # (Brandon, 2026-09-04: "probably usually just NaNs or 0"),
            # and a specification with only autos says so honestly —
            # the virtual point transform refuses it rather than
            # inventing the phase between the control channels.
            off_diagonal = cpsd[:, ~np.eye(len(channels), dtype=bool)]
            meaningful = bool(np.any(np.isfinite(off_diagonal)
                                     & (off_diagonal != 0)))
            records, response, reference, dims = [], [], [], []
            units, ref_units = [], []
            limits = {name: [] for name in bands}
            for i, (dof_i, (si, di, ui)) in enumerate(channels):
                for j, (dof_j, (sj, dj, uj)) in enumerate(channels):
                    if i != j and not (full_cpsd or meaningful):
                        continue
                    if i != j and not np.isfinite(cpsd[:, i, j]).any():
                        continue        # this one pair was never written
                    for name, matrix in bands.items():
                        # limits are per control channel, so a cross-spectral
                        # record has none of its own
                        limits[name].append(matrix[:, i] * si * sj if i == j
                                            else np.full(len(freq), np.nan))
                    records.append(cpsd[:, i, j] * si * sj)
                    response.append(dof_i)
                    reference.append(dof_j)
                    known = UNKNOWN not in (di, dj)
                    dims.append(
                        (f'{di}**2/frequency' if di == dj
                         else f'{di}*{dj}/frequency') if known else UNKNOWN)
                    units.append(ui if known else None)
                    ref_units.append(uj if known and ui != uj else None)
            spec = Specification(
                abscissa=freq,
                ordinate=np.array(records),
                response_dof=response,
                reference_dof=reference,
                ordinate_dim=dims,
                ordinate_unit=units,
                reference_unit=ref_units,
                **{name: np.array(values) for name, values in limits.items()},
            )
            # the controller's target on its FFT lines is a density per
            # line and draws as steps; a breakpoint curve is the law
            # between its points (`Specification.reading_of`)
            spec.interpolation = Specification.reading_of(freq)
            out[f'{env_name}_specification'] = spec

        # A sine environment's target is its tone set: a specifications
        # subgroup with one named group per tone, each a breakpoint
        # table with the per-segment sweep law. The stored table pads
        # its segment arrays to breakpoint length (the trailing entry is
        # dead — the leading-rate convention, pinned against the
        # controller's own trajectory record), and the band arrays are
        # (breakpoint, lower/upper, left/right, channel): the left and
        # right sides of a breakpoint may differ in principle, but no
        # file here has shown one that does, so a difference refuses by
        # name rather than being averaged into a band nobody wrote.
        for env_name, group in ds.groups.items():
            if 'specifications' not in group.groups:
                continue
            indices = np.asarray(
                group.variables['control_channel_indices'][()], dtype=int)
            # over the control channels, or over the rows of a response
            # transformation when the sweep controlled a virtual point —
            # the same answer the random specification reads (a sine
            # with a transformation beside an untransformed random
            # failed to load here, 2026-09-24)
            controlled = _control_channels(group, indices, dofs, scales_dims)
            scales = [kind[0] for _dof, kind in controlled]
            control_dims = {kind[1] for _dof, kind in controlled}
            control_units = {kind[2] for _dof, kind in controlled}
            if len(control_dims) > 1:
                raise ValueError(
                    f'{env_name}: control channels mix quantities '
                    f'{sorted(control_dims)}; one sine specification '
                    'holds one')
            tones = []
            for tone_name, sub in group.groups['specifications'].groups.items():
                n = len(sub.variables['spec_frequency'][()])
                bands = {}
                for kind, variable in (('warning', 'spec_warning'),
                                       ('abort', 'spec_abort')):
                    if variable not in sub.variables:
                        continue
                    matrix = np.asarray(sub.variables[variable][()],
                                        dtype=float)
                    left, right = matrix[:, :, 0, :], matrix[:, :, 1, :]
                    if not np.allclose(left, right, equal_nan=True):
                        raise ValueError(
                            f'{env_name} tone {tone_name}: the {kind} '
                            'band differs between the left and right '
                            'sides of a breakpoint — never seen in a '
                            'real file, refused rather than averaged')
                    for side, curve in enumerate(('lower', 'upper')):
                        values = left[:, side, :] * np.asarray(scales)
                        if np.isfinite(values).any():
                            bands[f'{kind}_{curve}'] = values
                tones.append(SineTone(
                    name=tone_name,
                    start_time=float(sub.getncattr('start_time')),
                    frequency=sub.variables['spec_frequency'][()],
                    amplitude=(np.asarray(sub.variables['spec_amplitude'][()])
                               * np.asarray(scales)),
                    phase=sub.variables['spec_phase'][()],
                    segment_type=sub.variables['spec_sweep_type'][()][:n - 1],
                    segment_rate=sub.variables['spec_sweep_rate'][()][:n - 1],
                    **bands))
            dim = control_dims.pop()
            unit = (control_units.pop()
                    if len(control_units) == 1 else None)
            out[f'{env_name}_specification'] = SineSweepSpecification(
                tones=tones, response_dof=[dof for dof, _kind in controlled],
                ordinate_dim=dim, ordinate_unit=unit,
                comment=f'sine tones from the {env_name} environment')

        # A transient environment's target is a waveform, not a spectrum,
        # and rides the file as `control_signal` — (control channels,
        # samples) with no abscissa of its own, because the controller
        # plays it at the hardware's rate.
        for env_name, group in ds.groups.items():
            if 'control_signal' not in group.variables:
                continue
            signal = np.asarray(group.variables['control_signal'][()])
            indices = np.asarray(
                group.variables['control_channel_indices'][()], dtype=int)
            if signal.ndim != 2 or not len(indices):
                continue
            rate = _environment_rate(group, ds)
            if rate is None:
                continue
            records, response, dims, units = [], [], [], []
            for i, ci in enumerate(indices[:signal.shape[0]]):
                scale, dim, unit = scales_dims[ci]
                records.append(signal[i] * scale)
                response.append(dofs[ci])
                dims.append(dim)
                units.append(unit)
            out[f'{env_name}_specification'] = TransientSpecification(
                abscissa=np.arange(signal.shape[-1]) / rate,
                ordinate=np.array(records),
                response_dof=response,
                ordinate_dim=dims,
                ordinate_unit=units,
                comment='transient control signal')

    return out