Datasets API

Dataset generation is controlled by two user-defined builders. Both receive an ordinary case mapping; one returns the structural system and the other returns the load. Sensors and axes are supplied separately, and the required record argument names the leaf fields retained from each completed simulation.

Pass cases for an exact design. For a random or correlated study, provide n_samples with case_builder(rng). These two case sources are mutually exclusive.

import dynaresp as dr

dataset = dr.generate_dataset(
    cases=cases,
    system_builder=build_system,
    load_builder=build_load,
    sensors=sensor_layout,
    time=time,
    positions=positions,
    record=[
        dr.Record.signal,
        dr.Record.case_parameter,
    ],
    seed=42,
)

Start with the dataset generation guide, then consult the schema reference for stored dimensions and metadata.

Generation

Record dataclass

Record(path)

One typed dataset field, organized through related selector families.

Use the predefined class attributes; unrecognized paths are rejected by :func:generate_dataset.

generate_dataset

generate_dataset(
    *,
    n_samples=None,
    system_builder,
    load_builder,
    sensors,
    cases=None,
    case_builder=None,
    time,
    positions,
    record,
    storage_dtype=float64,
    output=None,
    seed=None,
    workers=1,
    progress=False,
)

Generate structural realizations from user-controlled builder strategies.

The system and load builders receive the same plain, mutable parameter dictionary, in that order. Either builder may augment it with numeric derived values; later builders see those additions, and the final dictionary is stored once as ground truth and exposed through both the parameter and target views.

Supply cases for exact, fully controlled realizations. For streamed random or correlated variability, supply case_builder and n_samples; it receives the per-sample NumPy generator and returns an ordinary numeric mapping. The two inputs are mutually exclusive. With neither, n_samples repeats constant builders with an empty mapping.

Sample seeds are derived only from seed and sample index. Consequently, case generation, stochastic loads, and sensor effects produce the same ordered dataset regardless of workers. If output is provided, records stream to a temporary sibling Zarr store that is validated and renamed into place only after successful completion.

Parameters:

Name	Type	Description	Default
n_samples	int | None	Number of realizations. Required without cases and inferred from them when omitted.	None
system_builder	Callable[[dict[str, CaseValue]], StructuralSystem]	Function receiving the sample case and returning a system.	required
load_builder	Callable[[dict[str, CaseValue]], Load]	Function receiving the same case and returning a load.	required
sensors	Sequence[Sensor]	Fixed sensor layout applied to every realization.	required
cases	Iterable[Mapping[str, ArrayLike]] | None	Ordered exact parameter mappings, one per realization. All mappings must contain the same names and value shapes.	None
case_builder	Callable[[Generator], Mapping[str, ArrayLike]] | None	Optional function that creates one numeric case from the sample generator. It cannot be combined with cases.	None
time	ArrayLike	Explicit simulation time axis.	required
positions	ArrayLike	Explicit spatial axis.	required
record	Sequence[Record]	Leaf fields from :class:Record to retain after each simulation.	required
storage_dtype	DTypeLike	Floating dtype used for selected arrays. Integer case values use int64 storage, while sample seeds remain uint64.	float64
output	str | Path | None	New Zarr path for transactional, incremental storage. Existing destinations are not overwritten.	None
seed	int | None	Root seed for the case builder, load generation, and sensor effects. A random root seed is created when omitted.	None
workers	int	Number of builder and simulation threads. This changes throughput, not sample values or ordering.	1
progress	bool	Display an opt-in tqdm progress bar as samples are stored.	False

Returns:

Type	Description
StructuralDataset	An in-memory or lazily Zarr-backed structural dataset.

Raises:

Type	Description
RuntimeError	If a builder or simulation fails for a sample.
ValueError	If cases, sample shapes, or parameter names are inconsistent.
FileExistsError	If output already exists.

Source code in src/dynaresp/datasets/generation.py

def generate_dataset(
    *,
    n_samples: int | None = None,
    system_builder: Callable[[dict[str, CaseValue]], StructuralSystem],
    load_builder: Callable[[dict[str, CaseValue]], Load],
    sensors: Sequence[Sensor],
    cases: Iterable[Mapping[str, ArrayLike]] | None = None,
    case_builder: Callable[[np.random.Generator], Mapping[str, ArrayLike]]
    | None = None,
    time: ArrayLike,
    positions: ArrayLike,
    record: Sequence[Record],
    storage_dtype: DTypeLike = np.float64,
    output: str | Path | None = None,
    seed: int | None = None,
    workers: int = 1,
    progress: bool = False,
) -> StructuralDataset:
    """Generate structural realizations from user-controlled builder strategies.

    The system and load builders receive the same plain, mutable parameter
    dictionary, in that order. Either builder may augment it with numeric
    derived values; later builders see those additions, and the final dictionary
    is stored once as ground truth and exposed through both the parameter and
    target views.

    Supply ``cases`` for exact, fully controlled realizations. For streamed
    random or correlated variability, supply ``case_builder`` and ``n_samples``;
    it receives the per-sample NumPy generator and returns an ordinary numeric
    mapping. The two inputs are mutually exclusive. With neither, ``n_samples``
    repeats constant builders with an empty mapping.

    Sample seeds are derived only from ``seed`` and sample index. Consequently,
    case generation, stochastic loads, and sensor effects produce the same
    ordered dataset regardless of ``workers``. If ``output`` is provided,
    records stream to a temporary sibling Zarr store that is validated and
    renamed into place only after successful completion.

    Args:
        n_samples: Number of realizations. Required without ``cases`` and inferred
            from them when omitted.
        system_builder: Function receiving the sample case and returning a system.
        load_builder: Function receiving the same case and returning a load.
        sensors: Fixed sensor layout applied to every realization.
        cases: Ordered exact parameter mappings, one per realization. All mappings
            must contain the same names and value shapes.
        case_builder: Optional function that creates one numeric case from the
            sample generator. It cannot be combined with ``cases``.
        time: Explicit simulation time axis.
        positions: Explicit spatial axis.
        record: Leaf fields from :class:`Record` to retain after each simulation.
        storage_dtype: Floating dtype used for selected arrays. Integer case
            values use int64 storage, while sample seeds remain uint64.
        output: New Zarr path for transactional, incremental storage. Existing
            destinations are not overwritten.
        seed: Root seed for the case builder, load generation, and sensor effects.
            A random root seed is created when omitted.
        workers: Number of builder and simulation threads. This changes throughput,
            not sample values or ordering.
        progress: Display an opt-in ``tqdm`` progress bar as samples are stored.

    Returns:
        An in-memory or lazily Zarr-backed structural dataset.

    Raises:
        RuntimeError: If a builder or simulation fails for a sample.
        ValueError: If cases, sample shapes, or parameter names are inconsistent.
        FileExistsError: If ``output`` already exists.
    """
    if not callable(system_builder):
        raise TypeError("system_builder must be callable")
    if not callable(load_builder):
        raise TypeError("load_builder must be callable")
    if case_builder is not None and not callable(case_builder):
        raise TypeError("case_builder must be callable")
    if cases is not None and case_builder is not None:
        raise ValueError("cases and case_builder are mutually exclusive")
    selected_paths = _record_paths(record)
    selected_dtype = _floating_dtype(storage_dtype)
    sensor_items = tuple(sensors)
    if any(not isinstance(sensor, Sensor) for sensor in sensor_items):
        raise TypeError("sensors must contain Sensor objects")
    fixed_cases = _cases(cases)
    if n_samples is None:
        if fixed_cases is None:
            raise ValueError("n_samples is required when cases is omitted")
        sample_count = len(fixed_cases)
    else:
        if (
            isinstance(n_samples, bool)
            or not isinstance(n_samples, Integral)
            or n_samples <= 0
        ):
            raise ValueError("n_samples must be a positive integer")
        sample_count = int(n_samples)
        if fixed_cases is not None and sample_count != len(fixed_cases):
            raise ValueError("n_samples must match the number of cases")
    if isinstance(workers, bool) or not isinstance(workers, Integral) or workers < 1:
        raise ValueError("workers must be a positive integer")
    worker_count = int(workers)
    if not isinstance(progress, bool):
        raise TypeError("progress must be a boolean")
    if seed is not None and (
        isinstance(seed, bool) or not isinstance(seed, Integral) or seed < 0
    ):
        raise ValueError("seed must be a non-negative integer")
    base_seed = (
        int(np.random.SeedSequence().generate_state(1, dtype=np.uint64)[0])
        if seed is None
        else int(seed)
    )

    plan = _GenerationPlan(
        system_builder,
        load_builder,
        case_builder,
        sensor_items,
        validate_time(time),
        validate_positions(positions),
        fixed_cases,
        selected_paths,
        selected_dtype,
    )

    if worker_count == 1:
        samples = (
            plan.create_sample(base_seed, index) for index in range(sample_count)
        )
        return _dataset_from_samples(
            plan,
            samples,
            n_samples=sample_count,
            seed=base_seed,
            controlled=fixed_cases is not None,
            output=output,
            progress=progress,
        )

    with ThreadPoolExecutor(max_workers=worker_count) as executor:
        samples = plan.parallel_samples(
            executor,
            base_seed=base_seed,
            n_samples=sample_count,
            workers=worker_count,
        )
        return _dataset_from_samples(
            plan,
            samples,
            n_samples=sample_count,
            seed=base_seed,
            controlled=fixed_cases is not None,
            output=output,
            progress=progress,
        )

parameter_grid

parameter_grid(**dimensions)

Build exact cases from the Cartesian product of named dimensions.

This function performs no random sampling. Keyword order is retained in every case and the rightmost dimension varies fastest. Pass the result to :func:generate_dataset to run one realization per combination.

Example

parameter_grid(temperature=[0, 20], force=[100, 200]) returns four cases in the order (0, 100), (0, 200), (20, 100), and (20, 200).

Source code in src/dynaresp/datasets/generation.py

def parameter_grid(**dimensions: Sequence[ArrayLike]) -> list[dict[str, ArrayLike]]:
    """Build exact cases from the Cartesian product of named dimensions.

    This function performs no random sampling. Keyword order is retained in
    every case and the rightmost dimension varies fastest. Pass the result to
    :func:`generate_dataset` to run one realization per combination.

    Example:
        ``parameter_grid(temperature=[0, 20], force=[100, 200])`` returns four
        cases in the order ``(0, 100)``, ``(0, 200)``, ``(20, 100)``, and
        ``(20, 200)``.
    """
    if not dimensions:
        raise ValueError("parameter_grid requires at least one dimension")
    names = tuple(dimensions)
    values = tuple(tuple(dimension) for dimension in dimensions.values())
    if any(not dimension for dimension in values):
        raise ValueError("parameter_grid dimensions must not be empty")
    cases = [
        dict(zip(names, combination, strict=True)) for combination in product(*values)
    ]
    for case in cases:
        _fixed_parameters(case)
    return cases

Dataset access

StructuralDataset

StructuralDataset(arrays, metadata, *, store=None)

Validated, indexable collection of structural simulations.

Schema version 4 stores only explicitly selected arrays. Shared time and positions coordinates have no sample dimension; every selected sample array begins with it. Signals use (sample, sensor, time), responses use (sample, degrees_of_freedom, time), and modal shapes use (sample, mode, position). metadata["n_samples"] remains authoritative even when no sample arrays were selected.

In-memory arrays are ordinary NumPy arrays. Arrays opened from Zarr remain lazy until a property, sample, or analysis adapter materializes them. Logical parameter and target names are resolved through parameter_keys and target_keys metadata instead of inferred from path prefixes. This keeps parameter and target views independent of their physical paths.

Source code in src/dynaresp/datasets/core.py

def __init__(
    self,
    arrays: Mapping[str, Array],
    metadata: Mapping[str, object],
    *,
    store: object | None = None,
) -> None:
    self._arrays = dict(arrays)
    self.metadata = dict(metadata)
    self._store = store
    self._n_samples = 0
    self._validate()

signal property

signal

The full signal array; remains lazy for a Zarr-backed dataset.

clean_signal property

clean_signal

The full noise-free measurement array.

time property

time

Shared simulation time coordinate.

positions property

positions

Shared physical-position or MDOF-index axis.

__getitem__

__getitem__(index)

Materialize one realization while shared coordinates stay reusable.

Source code in src/dynaresp/datasets/core.py

def __getitem__(self, index: int) -> StructuralSample:
    """Materialize one realization while shared coordinates stay reusable."""
    resolved = self._resolve_index(index)
    parameter_keys = self._metadata_paths("parameter_keys")
    parameters = {
        str(name): _as_sample_value(self._arrays[str(path)][resolved])
        for name, path in parameter_keys.items()
    }
    target_keys = self._metadata_paths("target_keys")
    target = {
        str(name): _as_sample_value(self._arrays[str(path)][resolved])
        for name, path in target_keys.items()
    }

    system = self._sample_group("system", resolved)
    modal = self._sample_group("modal", resolved)
    load = self._sample_group("load", resolved)
    sample_metadata: dict[str, object] = {"index": resolved}
    if "metadata/seed" in self._arrays:
        sample_metadata["seed"] = int(self._arrays["metadata/seed"][resolved])
    if parameters:
        sample_metadata["parameters"] = parameters
    for name in ("dynaresp_version", "generation", "sensors", "variability"):
        if name in self.metadata:
            sample_metadata[name] = self.metadata[name]
    return StructuralSample(
        signal=(
            np.asarray(self._arrays["signal"][resolved])
            if "signal" in self._arrays
            else None
        ),
        clean_signal=(
            np.asarray(self._arrays["clean_signal"][resolved])
            if "clean_signal" in self._arrays
            else None
        ),
        time=self.time if "time" in self._arrays else None,
        positions=self.positions if "positions" in self._arrays else None,
        target=target,
        system=system,
        modal=modal,
        load=load,
        metadata=sample_metadata,
    )

sample_field

sample_field(sample, name)

Resolve an adapter field by public alias, target, or parameter name.

Source code in src/dynaresp/datasets/core.py

def sample_field(self, sample: StructuralSample, name: str) -> object:
    """Resolve an adapter field by public alias, target, or parameter name."""
    index = sample.metadata.get("index")
    if not isinstance(index, int):
        raise TypeError("sample metadata does not contain a valid index")
    return self._sample_field_at(index, name)

to_torch

to_torch(*, inputs='signal', targets, extras=())

Return a lazy, pickleable PyTorch-compatible view.

inputs, targets, and extras use the same names accepted by :meth:sample_field. Values are converted to torch.float32 only when an item is requested, so a Zarr-backed source remains lazy.

Source code in src/dynaresp/datasets/core.py

def to_torch(
    self,
    *,
    inputs: str = "signal",
    targets: str,
    extras: Sequence[str] = (),
) -> object:
    """Return a lazy, pickleable PyTorch-compatible view.

    ``inputs``, ``targets``, and ``extras`` use the same names accepted by
    :meth:`sample_field`. Values are converted to ``torch.float32`` only
    when an item is requested, so a Zarr-backed source remains lazy.
    """
    try:
        __import__("torch")
    except ImportError as error:
        raise ImportError(
            "PyTorch is optional; install dynaresp with the 'torch' extra"
        ) from error
    return TorchStructuralDataset(self, inputs, targets, extras)

to_xarray

to_xarray()

Materialize signals, frequencies, and scalar parameters as xarray.

Source code in src/dynaresp/datasets/core.py

def to_xarray(self) -> object:
    """Materialize signals, frequencies, and scalar parameters as xarray."""
    try:
        import xarray as xr
    except ImportError as error:
        raise ImportError(
            "xarray is optional; install dynaresp with the 'analysis' extra"
        ) from error
    data_vars: dict[str, tuple[tuple[str, ...], Array]] = {}
    for name in ("signal", "clean_signal"):
        if name in self._arrays:
            data_vars[name] = (
                ("sample", "sensor", "time"),
                np.asarray(self._arrays[name]),
            )
    if "modal/frequencies_hz" in self._arrays:
        data_vars["frequencies_hz"] = (
            ("sample", "mode"),
            np.asarray(self._arrays["modal/frequencies_hz"]),
        )
    for name, path in self._metadata_paths("parameter_keys").items():
        values = np.asarray(self._arrays[path])
        if values.ndim == 1:
            data_vars[name] = (("sample",), values)
    coords: dict[str, Array] = {"sample": np.arange(len(self))}
    if "time" in self._arrays:
        coords["time"] = self.time
    if "positions" in self._arrays:
        coords["position"] = self.positions
    return xr.Dataset(
        data_vars=data_vars,
        coords=coords,
        attrs={"dynaresp": self.metadata},
    )

properties

properties()

Materialize scalar parameters and modal frequencies as a dataframe.

Source code in src/dynaresp/datasets/core.py

def properties(self) -> object:
    """Materialize scalar parameters and modal frequencies as a dataframe."""
    try:
        import pandas as pd
    except ImportError as error:
        raise ImportError(
            "pandas is optional; install dynaresp with the 'analysis' extra"
        ) from error
    columns: dict[str, Array] = {}
    for name, path in self._metadata_paths("parameter_keys").items():
        values = np.asarray(self._arrays[path])
        if values.ndim == 1:
            columns[name.replace(".", "_")] = values
    if "modal/frequencies_hz" in self._arrays:
        frequencies = np.asarray(self._arrays["modal/frequencies_hz"])
        for mode in range(frequencies.shape[1]):
            columns[f"f{mode + 1}_hz"] = frequencies[:, mode]
    return pd.DataFrame(columns)

summary

summary()

Return a compact human-readable dataset summary.

Source code in src/dynaresp/datasets/core.py

def summary(self) -> str:
    """Return a compact human-readable dataset summary."""
    variable_names = self.metadata.get("parameter_keys", {})
    variable_text = (
        ", ".join(variable_names) if isinstance(variable_names, Mapping) else ""
    )
    lines = [
        "dynaresp StructuralDataset",
        f"Samples: {len(self):,}",
        f"System: {self.metadata.get('system_type', 'unknown')}",
    ]
    if "time" in self._arrays:
        lines.append(f"Time samples: {self._arrays['time'].shape[0]}")
    if "positions" in self._arrays:
        lines.append(f"Position samples: {self._arrays['positions'].shape[0]}")
    if "signal" in self._arrays:
        signal_shape = self._arrays["signal"].shape
        lines.extend(
            [
                f"Sensors: {signal_shape[1]}",
                f"Signal shape: {signal_shape[1:]}",
            ]
        )
    lines.extend(
        [
            f"Recorded parameters: {variable_text or 'none'}",
            f"Storage: {'Zarr (lazy)' if self._store is not None else 'memory'}",
        ]
    )
    return "\n".join(lines)

from_records classmethod

from_records(
    records,
    *,
    n_samples,
    time=None,
    positions=None,
    metadata,
    output=None,
)

Construct a dataset from identically shaped per-sample records.

With no output, records are stacked into memory. With output, each record is written directly to a temporary Zarr directory beside the destination. The completed store is reopened and validated before one same-filesystem rename publishes it. Failures remove the temporary store, and an existing destination is never overwritten.

Source code in src/dynaresp/datasets/core.py

@classmethod
def from_records(
    cls,
    records: Iterable[Mapping[str, np.ndarray]],
    *,
    n_samples: int,
    time: Array | None = None,
    positions: Array | None = None,
    metadata: Mapping[str, object],
    output: str | Path | None = None,
) -> StructuralDataset:
    """Construct a dataset from identically shaped per-sample records.

    With no ``output``, records are stacked into memory. With ``output``,
    each record is written directly to a temporary Zarr directory beside
    the destination. The completed store is reopened and validated before
    one same-filesystem rename publishes it. Failures remove the temporary
    store, and an existing destination is never overwritten.
    """
    if (
        isinstance(n_samples, bool)
        or not isinstance(n_samples, Integral)
        or n_samples < 1
    ):
        raise ValueError("n_samples must be a positive integer")
    sample_count = int(n_samples)
    iterator = iter(records)
    try:
        first = dict(next(iterator))
    except StopIteration as error:
        raise ValueError("records must contain at least one sample") from error
    coordinate_arrays = {
        name: np.asarray(value)
        for name, value in (("time", time), ("positions", positions))
        if value is not None
    }
    overlap = first.keys() & coordinate_arrays.keys()
    if overlap:
        raise ValueError(
            f"sample records contain coordinate arrays: {sorted(overlap)}"
        )
    if output is None:
        collected = {name: [np.asarray(value)] for name, value in first.items()}
        written = 1
        for index, record in enumerate(iterator, start=1):
            if index >= sample_count:
                raise ValueError("record count exceeds n_samples")
            if set(record) != set(first):
                raise ValueError("dataset records changed fields between samples")
            for name, value in record.items():
                sample = np.asarray(value)
                if sample.shape != collected[name][0].shape:
                    raise ValueError(f"{name} changed shape between samples")
                if sample.dtype != collected[name][0].dtype:
                    raise ValueError(f"{name} changed dtype between samples")
                collected[name].append(sample)
            written += 1
        if written != sample_count:
            raise ValueError("record count does not match n_samples")
        arrays: dict[str, Array] = {
            **coordinate_arrays,
            **{name: np.stack(values) for name, values in collected.items()},
        }
        stored_metadata = {
            **metadata,
            "schema_version": cls.schema_version,
            "n_samples": sample_count,
            "array_keys": list(arrays),
        }
        return cls(arrays, stored_metadata)

    destination = Path(output)
    if os.path.lexists(destination):
        raise FileExistsError(f"dataset destination already exists: {destination}")
    destination.parent.mkdir(parents=True, exist_ok=True)
    temporary = Path(
        tempfile.mkdtemp(
            prefix=f".{destination.name}.",
            suffix=".tmp",
            dir=destination.parent,
        )
    )
    committed = False
    try:
        zarr = _import_zarr()
        group = zarr.open_group(str(temporary), mode="w")
        for name, value in coordinate_arrays.items():
            _zarr_create(group, name, value, None)
        array_keys = list(first)
        for name, value in first.items():
            _zarr_create(group, name, value, sample_count)
            _zarr_get(group, name)[0] = value
        written = 1
        for index, record in enumerate(iterator, start=1):
            if index >= sample_count:
                raise ValueError("record count exceeds n_samples")
            if set(record) != set(first):
                raise ValueError("dataset records changed fields between samples")
            for name, value in record.items():
                sample = np.asarray(value)
                target = _zarr_get(group, name)
                if target.shape[1:] != sample.shape:
                    raise ValueError(f"{name} changed shape between samples")
                if np.dtype(target.dtype) != sample.dtype:
                    raise ValueError(f"{name} changed dtype between samples")
                target[index] = sample
            written += 1
        if written != sample_count:
            raise ValueError("record count does not match n_samples")
        stored_metadata = {
            **metadata,
            "schema_version": cls.schema_version,
            "n_samples": sample_count,
            "array_keys": [*coordinate_arrays, *array_keys],
        }
        group.attrs["metadata_json"] = json.dumps(stored_metadata, sort_keys=True)
        cls.open(temporary)
        if os.path.lexists(destination):
            raise FileExistsError(
                f"dataset destination already exists: {destination}"
            )
        temporary.rename(destination)
        committed = True
    finally:
        if not committed:
            shutil.rmtree(temporary, ignore_errors=True)
    return cls.open(destination)

open classmethod

open(path)

Open and validate a schema-v4 Zarr store without loading samples.

Selected array locations come from stored metadata rather than naming assumptions, so opening a compact store does not materialize its samples.

Source code in src/dynaresp/datasets/core.py

@classmethod
def open(cls, path: str | Path) -> Self:
    """Open and validate a schema-v4 Zarr store without loading samples.

    Selected array locations come from stored metadata rather than naming
    assumptions, so opening a compact store does not materialize its samples.
    """
    zarr = _import_zarr()
    group = zarr.open_group(str(Path(path)), mode="r")
    try:
        raw_metadata = group.attrs["metadata_json"]
    except KeyError as error:
        raise ValueError("not a dynaresp Zarr dataset") from error
    metadata = json.loads(str(raw_metadata))
    if not isinstance(metadata, dict):
        raise TypeError("dataset metadata must be a JSON object")
    keys = metadata.get("array_keys", [])
    if not isinstance(keys, list):
        raise TypeError("dataset array_keys metadata is invalid")
    if metadata.get("schema_version") != cls.schema_version:
        raise ValueError(
            f"unsupported dataset schema_version; expected {cls.schema_version}"
        )
    if any(not isinstance(name, str) for name in keys) or len(set(keys)) != len(
        keys
    ):
        raise TypeError("dataset array_keys metadata is invalid")
    arrays: dict[str, Array] = {}
    for name in keys:
        try:
            arrays[name] = _zarr_get(group, name)
        except KeyError as error:
            raise ValueError(
                f"dataset array_keys references missing array {name!r}"
            ) from error
    return cls(arrays, metadata, store=group)

StructuralSample dataclass

StructuralSample(
    signal,
    clean_signal,
    time,
    positions,
    target,
    system,
    modal,
    load,
    metadata,
)

One realization containing only the fields selected for storage.

signal and clean_signal have shape (sensors, time). System matrices use (degrees_of_freedom, degrees_of_freedom); modal shapes use (modes, positions); and the generalized load["force"] uses (degrees_of_freedom, time). The diagnostic load["values"] uses (positions, time). target contains the logical targets declared by dataset metadata, so callers need not know their storage paths.

The dataclass is frozen to keep its fields bound to one realization, but contained NumPy arrays and dictionaries are ordinary mutable objects.

Predefined datasets

BeamDamage

BeamDamage(
    *,
    root,
    split="train",
    version="1.0",
    generate=False,
    n_samples=None,
    workers=1,
    progress=False,
)

Open or reproducibly generate the finite-element beam-damage benchmark.

Version 1.0 varies damaged element, stiffness reduction, and Young's modulus under a fixed white-noise load configuration. Train and test splits use distinct fixed root seeds. An existing split is opened lazily; otherwise, generate=True is required before the transactional Zarr writer creates it. n_samples overrides the standard 10,000/2,000 split sizes. Set progress=True to display completed samples while generating.

Source code in src/dynaresp/datasets/predefined.py

def BeamDamage(
    *,
    root: str | Path,
    split: str = "train",
    version: str = "1.0",
    generate: bool = False,
    n_samples: int | None = None,
    workers: int = 1,
    progress: bool = False,
) -> StructuralDataset:
    """Open or reproducibly generate the finite-element beam-damage benchmark.

    Version ``1.0`` varies damaged element, stiffness reduction, and Young's
    modulus under a fixed white-noise load configuration. Train and test splits
    use distinct fixed root seeds. An existing split is opened lazily;
    otherwise, ``generate=True`` is required before the transactional Zarr
    writer creates it. ``n_samples`` overrides the standard 10,000/2,000 split
    sizes.
    Set ``progress=True`` to display completed samples while generating.
    """
    if version != "1.0":
        raise ValueError("BeamDamage currently provides version '1.0'")
    if split not in {"train", "test"}:
        raise ValueError("split must be 'train' or 'test'")
    output = Path(root) / f"beam-damage-{version}-{split}.zarr"
    if output.exists():
        return StructuralDataset.open(output)
    if not generate:
        raise FileNotFoundError(
            f"{output} does not exist; pass generate=True to create it"
        )

    sample_count = (
        (10_000 if split == "train" else 2_000) if n_samples is None else n_samples
    )
    seed = 104_729 if split == "train" else 104_759

    def build_case(rng: np.random.Generator) -> dict[str, int | float]:
        return {
            "damaged_element": int(rng.integers(2, 18)),
            "stiffness_reduction": float(rng.uniform(0.0, 0.30)),
            "young_modulus": float(rng.uniform(190.0e9, 220.0e9)),
        }

    def build_beam(case: Mapping[str, CaseValue]) -> Beam:
        damaged_element = int(case["damaged_element"])
        reduction = float(case["stiffness_reduction"])
        return Beam(
            length=5.0,
            material="steel",
            section=("rectangular", 0.10, 0.02),
            young_modulus=float(case["young_modulus"]),
            damping_ratio=0.01,
            n_elements=20,
            n_modes=6,
            stiffness_reductions={damaged_element: reduction},
        )

    def build_load(_case: Mapping[str, CaseValue]) -> WhiteNoise:
        return WhiteNoise(position=4.5, rms=100.0)

    return generate_dataset(
        n_samples=sample_count,
        case_builder=build_case,
        system_builder=build_beam,
        load_builder=build_load,
        sensors=[
            Accelerometer(
                position=position,
                noise=GaussianNoise(snr_db=30.0),
            )
            for position in (1.0, 2.0, 3.0, 4.0)
        ],
        time=np.arange(1000, dtype=np.float64) / 200.0,
        positions=np.linspace(0.0, 5.0, 101),
        record=_COMPLETE_RECORD,
        output=output,
        seed=seed,
        workers=workers,
        progress=progress,
    )