visualdynamics.io.rattlesnake

rattlesnake

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

Rattlesnake streams results to netCDF4 with a channel table that includes engineering units (read in whatever case they were typed), so imports are fully unit-aware — no unit declaration needed. Layout (file_version 3.x):

• root attrs: sample_rate, hardware, file_version, ...
• time_data (response_channels, time_samples), in channel engineering units
• channels group: node_number, node_direction, unit, channel_type, and other per-channel metadata
• one group per environment (e.g. 'Random') that may carry a specification CPSD matrix (specification_frequency_lines, specification_cpsd_matrix_*) and the bands around it (specification_warning_matrix, specification_abort_matrix, each (2, lines, channels), lower first)

A run saves twice and the two files share nothing: streaming writes the time histories, and a separate save writes what the environment computed from them. Both are read here.

Returns a dict: 'channel_table' -> ChannelTable, and whichever of these the file holds — 'time_data' -> TimeHistory, '_specification' -> Specification (diagonal ASDs with their warning and abort limits; full_cpsd=True for the whole matrix as cross-PSD records), '_frf' -> Frf, '_coherence' -> MultipleCoherence, '_response_cpsd' and '_drive_cpsd' -> Psd.

A modal environment counts every enabled channel as a response and never excludes its references, so the drives appear among them and the saved matrix carries each drive against the drives. Those rows are identity and cross-talk by construction — |H| exactly 1 against itself, numerically zero against the other, coherence exactly 1 — so they are dropped: they are arithmetic rather than measurement, and would put a force-per-force axis on the plot beside the real one.

A spectral file carries neither time data nor, once rattlesnake's own random save has been through it, the root attributes — so the sample rate may be missing and the frequency axis has to come from the specification's own lines. The noise cross spectra a random save also writes are deliberately left alone: they describe the measurement's noise floor and would double the object count for something rarely looked at.

Reading only, and deliberately: a .nc4 records a controller run — hardware settings, environments, the lot — that visualdynamics does not hold, so a file written from here would describe a test that never happened. See docs/export.md.

Functions:

Name	Description
machine_memory	The machine's physical memory in bytes, or 0 when it cannot be asked.
last_window	Where a window of the last last seconds of a run opens.
stream_summary	What a run's streams would cost to import, read without reading one.
stream_preview	One channel's envelope over a whole stream, read in slabs.
environment_kinds	{environment name: kind} the file says it holds.
run_kind	What kind of test the file holds, by its own account.
project_type	The visualdynamics project this file is a run of, or None if visualdynamics has no
streamed_sysid_candidate	Whether a streamed save has the shape a system ID leaves:
load	Everything a Rattlesnake .nc4 holds, keyed the way the tree names it.

Classes

Functions:

machine_memory

machine_memory() -> int

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

The one system question this package asks: whether a stream about to be imported would take a large share of it. POSIX answers 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 asked.

    The one system question this package asks: whether a stream about
    to be imported would take a large share of it. POSIX answers
    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

Where a window of the last last seconds of a run opens.

The one rule behind the import dialog's Last field and a script's last= (Brandon, 2026-09-19): a run seconds long is read from seconds - last to its end — and a run no longer than that is taken whole, which is what None says. Asking for the last 100 s of a 28 s run is not a mistake to refuse; it is all of it.

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:
    """Where a window of the last `last` seconds of a run opens.

    The one rule behind the import dialog's *Last* field and a
    script's `last=` (Brandon, 2026-09-19): a run `seconds` long is
    read from `seconds - last` to its end — and a run no longer than
    that is taken whole, which is what `None` says. Asking for the last
    100 s of a 28 s run is not a mistake to refuse; it is all of it.

    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.
    """
    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)

stream_summary

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

What a run's streams would cost to import, read without reading one.

The import dialog's first question — is this stream a large share of the machine? — answered from the file's dimensions alone, so asking it costs nothing 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, what the import holds); 'memory', machine_memory(). A system-ID package has no streams and says so with an empty list.

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

def stream_summary(path: str | os.PathLike) -> dict[str, Any]:
    """What a run's streams would cost to import, read without reading one.

    The import dialog's first question — is this stream a large share
    of the machine? — answered from the file's dimensions alone, so
    asking it costs nothing 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, what the import holds);
        'memory', `machine_memory()`. A system-ID package has no
        streams and says so with an empty list.
    """
    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.

The picture an import window is chosen from: where the run reaches level, where it stops, the false start at the front. Each of points buckets keeps its least and greatest sample — the stage's peak thinning, applied as the channel is read — so the preview of a 22 GB run holds a slab at a time and comes back as a few thousand points. The file's chunks run along one channel at a time, so one channel is a sequential read of its own bytes.

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 clock; 'low' and 'high', each bucket's least and greatest value as the file holds them, NaN where a bucket holds nothing else; 'dof', 'unit', 'dim' of the channel; and 'samples', '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.

    The picture an import window is chosen from: where the run
    reaches level, where it stops, the false start at the front. Each
    of `points` buckets keeps its least and greatest sample — the
    stage's peak thinning, applied as the channel is read — so the
    preview of a 22 GB run holds a slab at a time and comes back as a
    few thousand points. The file's chunks run along one channel at a
    time, so one channel is a sequential read of its own bytes.

    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 clock;
        'low' and 'high', each bucket's least and greatest value as
        the file holds them, NaN where a bucket holds nothing else;
        'dof', 'unit', 'dim' of the channel; and
        'samples', '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}

environment_kinds

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

{environment name: kind} the file says it holds.

Empty for a file that does not say — an nc4 assembled by hand, or an older save from before the controller wrote its types down.

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

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

    Empty for a file that does not say — an nc4 assembled by hand, or an
    older save from before the controller wrote its types down.
    """
    import netCDF4

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

run_kind

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

What kind of test the file holds, by its own account.

'modal', 'random', 'transient', 'sine' or 'time' for a run of one kind; 'mixed' when more than one environment drove the article at once, which is what a random-plus-sine-sweep run is; None when the file does not say.

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

def run_kind(path: str | os.PathLike) -> str | None:
    """What kind of test the file holds, by its own account.

    'modal', 'random', 'transient', 'sine' or 'time' for a run of one
    kind; 'mixed' when more than one environment drove the article at
    once, which is what a random-plus-sine-sweep run is; None when the
    file does not say.
    """
    return _run_kind(environment_kinds(path).values())

project_type

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

The visualdynamics project this file is a run of, or None if visualdynamics has no project of that kind — or the file never said what kind it was.

A mixed run answers with its leading half (MIXED_PRECEDENCE): every environment's specification imports regardless, so the type only chooses which workflow the tree leads with.

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

def project_type(path: str | os.PathLike) -> str | None:
    """The visualdynamics project this file is a run of, or None if visualdynamics has no
    project of that kind — or the file never said what kind it was.

    A mixed run answers with its leading half (`MIXED_PRECEDENCE`):
    every environment's specification imports regardless, so the type
    only chooses which workflow the tree leads with.
    """
    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':
        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 has the shape a system ID leaves: exactly two streams, a quiet one then a loud one — the ambient measurement and the driven excitation.

The file itself cannot settle the question (a run of the environment that stopped and restarted its stream once looks the same on paper), so this is grounds to ask the person, never to decide. The quiet-then-loud check is what keeps the question from being asked about every two-stream file: a restarted run's halves play at one level, a system ID's differ by the whole test. Ten decibels is well under any real ambient-to-driven gap and well over a level change within one run.

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

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

    The file itself cannot settle the question (a run of the
    environment that stopped and restarted its stream once looks the
    same on paper), so this is grounds to *ask the person*, never to
    decide. The quiet-then-loud check is what keeps the question from
    being asked about every two-stream file: a restarted run's halves
    play at one level, a system ID's differ by the whole test. Ten
    decibels is well under any real ambient-to-driven gap and well
    over a level change within one run.
    """
    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) -> dict[str, Any]

Everything a Rattlesnake .nc4 holds, keyed the way the tree names it.

The streamed time data, the channel table, and each environment's specification, FRF, coherence and spectra; a saved system-ID package on its own. start, stop and channels import a part of the streams — the window a long run is read through when the whole would not fit the machine (the import dialog asks; a script says). The window is inclusive at both instants on the run's own clock, which the records keep, and a stream the window misses entirely is left out. A part is a recording, so a windowed stream is not split into a spectral save's frames.

Parameters:

Name	Type	Description	Default
path	str or PathLike	The file.	required
full_cpsd	bool	Import every cross term of the environment's CPSDs, not only the autos.	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 come along only when every control channel they are computed from is among them.	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) -> dict[str, Any]:
    """Everything a Rattlesnake `.nc4` holds, keyed the way the tree names it.

    The streamed time data, the channel table, and each environment's
    specification, FRF, coherence and spectra; a saved system-ID
    package on its own. `start`, `stop` and `channels` import a part
    of the streams — the window a long run is read through when the
    whole would not fit the machine (the import dialog asks; a script
    says). The window is inclusive at both instants on the run's own
    clock, which the records keep, and a stream the window misses
    entirely is left out. A part is a recording, so a windowed stream
    is not split into a spectral save's frames.

    Parameters
    ----------
    path : str or os.PathLike
        The file.
    full_cpsd : bool
        Import every cross term of the environment's CPSDs, not only
        the autos.
    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 come along only when every
        control channel they are computed from is among them.

    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()]

        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
            first, last = span
            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)
            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
            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)
            scales = [scales_dims[ci][0] for ci in indices]
            control_dims = {scales_dims[ci][1] for ci in indices}
            control_units = {scales_dims[ci][2] for ci in indices}
            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=[dofs[ci] for ci in indices],
                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')

    # a run also states its *control* channels, through its
    # specification: the requirement is written per control channel, so
    # a response channel whose DOF the specification names is a control
    # channel. The file's own statement, mapped rather than guessed —
    # and role-guarded, so a drive sharing a controlled DOF stays a
    # reference.
    named = {dof for obj in out.values()
             if isinstance(obj, (Specification, TransientSpecification,
                                 SineSweepSpecification))
             for dof in obj.response_dof}
    if named:
        roles = table.roles()
        for row, dof in enumerate(table.dof_strings()):
            if dof in named and roles[row] == 'response':
                table.set_cell('control', row, 'True')
    return out