`visualdynamics.io.native`¶

native ¶

The native .vdyn file format, which is HDF5.

A file with a single object has one group at the root, named for the object's kind ('geometry', 'data', 'shapes', and so on). A file with a whole test has an objects/ group of numbered groups, numbered in the test's order, with each object's name and kind as attributes, so a name may contain anything, including characters h5py would read as a path. The test's name, active geometry, project type, object groups and provenance are attributes of the root. Ragged connectivity is stored as a flat array plus offsets. Every file is stamped with its schema version, and a file from a newer schema is refused with a message to update.

Functions:

Name	Description
`schema_for`	The schema version a file holding `objects` (one object, or a
`save_predictions`	Write a `Predictions` object into `group`: its comment, and one
`save_run_log_extract`	Write a `RunLogExtract` into `group`: the snapshot's JSON Lines
`save_sine_specification`	Write a sine sweep specification into `group`, one subgroup per
`save_sine_levels`	Write a sine level set into `group`, one subgroup per tone, each
`save`	Save one Visual Dynamics object to a .vdyn (HDF5) file.
`save_into`	Write one object into an open, empty HDF5 file.
`save_test`	Save a whole test, every named object, to one .vdyn file.
`save_test_into`	`save_test` into an open, empty HDF5 file (see `save_into`).
`load`	Load a .vdyn file: the object it holds, or a whole test.
`load_from`	`load` from an open HDF5 file (see `save_into`). `path` names the

Classes¶

Functions:¶

schema_for ¶

schema_for(objects: Any) -> int

The schema version a file holding objects (one object, or a mapping of names to objects) is stamped with: 3 when it holds a run log extract, 2 when it holds predictions or a report reading one (and no extract), 1 otherwise.

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

def schema_for(objects: Any) -> int:
    """The schema version a file holding `objects` (one object, or a
    mapping of names to objects) is stamped with: 3 when it holds a run
    log extract, 2 when it holds predictions or a report reading one
    (and no extract), 1 otherwise."""
    from ..core.prediction import Predictions
    from ..core.report import Report
    from ..core.run_log_extract import RunLogExtract

    held = objects.values() if isinstance(objects, Mapping) else [objects]
    stamp = BASE_SCHEMA
    for obj in held:
        if isinstance(obj, RunLogExtract):
            return RUN_LOG_SCHEMA
        if isinstance(obj, Predictions) or (
                isinstance(obj, Report) and any(
                    block.get('kind') == 'prediction'
                    for block in obj.blocks)):
            stamp = PREDICTIONS_SCHEMA
    return stamp

save_predictions ¶

save_predictions(predictions, group) -> None

Write a Predictions object into group: its comment, and one subgroup per prediction holding its values as JSON (Prediction.as_record) and its curve, when it has one.

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

def save_predictions(predictions, group) -> None:
    """Write a `Predictions` object into `group`: its comment, and one
    subgroup per prediction holding its values as JSON
    (`Prediction.as_record`) and its curve, when it has one."""
    import json

    group.attrs['comment'] = predictions.comment
    group.attrs['count'] = len(predictions)
    for k, prediction in enumerate(predictions):
        # index-named subgroups, in the set's order: a channel is user
        # text, which an HDF5 group name cannot always hold
        sub = group.create_group(str(k))
        sub.attrs['record'] = json.dumps(prediction.as_record())
        if prediction.curve is not None:
            sub.create_dataset('curve_frequency', data=prediction.curve[0])
            sub.create_dataset('curve_values', data=prediction.curve[1])

save_run_log_extract ¶

save_run_log_extract(extract: RunLogExtract, group: Group) -> None

Write a RunLogExtract into group: the snapshot's JSON Lines whole, in a dataset (it can pass the 64 KiB an attribute holds), and what it was taken from as attributes. The log's path is stored relative to the project file when the log is in the file's folder or one beneath it (run_log_extract.stored_path), so a project moved with its log still finds it.

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

def save_run_log_extract(extract: RunLogExtract, group: h5py.Group) -> None:
    """Write a `RunLogExtract` into `group`: the snapshot's JSON Lines
    whole, in a dataset (it can pass the 64 KiB an attribute holds), and
    what it was taken from as attributes. The log's path is stored
    relative to the project file when the log is in the file's folder
    or one beneath it (`run_log_extract.stored_path`), so a project
    moved with its log still finds it."""
    import json

    import h5py

    from ..core.run_log_extract import stored_path

    group.attrs['path'] = stored_path(extract.path, _file_path(group))
    group.attrs['test'] = extract.test
    group.attrs['upto'] = extract.upto
    group.attrs['taken'] = extract.taken.isoformat()
    group.attrs['log_size'] = extract.log_size
    group.attrs['after'] = extract.after
    group.attrs['after_in_test'] = extract.after_in_test
    group.attrs['problems'] = json.dumps(list(extract.problems))
    group.create_dataset('log', data=extract.text,
                         dtype=h5py.string_dtype('utf-8'))

save_sine_specification ¶

save_sine_specification(spec, group) -> None

Write a sine sweep specification into group, one subgroup per tone: its breakpoints with the sweep law of each segment, and its bands where it has them.

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

def save_sine_specification(spec, group) -> None:
    """Write a sine sweep specification into `group`, one subgroup per
    tone: its breakpoints with the sweep law of each segment, and its
    bands where it has them."""
    _write_strings(group, 'response_dof', spec.response_dof)
    group.attrs['ordinate_dim'] = spec.ordinate_dim
    group.attrs['ordinate_unit'] = spec.ordinate_unit or ''
    group.attrs['comment'] = spec.comment
    _write_strings(group, 'tone_order', [tone.name for tone in spec.tones])
    tones = group.create_group('tones')
    for k, tone in enumerate(spec.tones):
        # index-named subgroups: tone names are user text and h5py
        # group names cannot hold a '/', so the order array holds the
        # names and the groups hold the numbers
        sub = tones.create_group(str(k))
        sub.attrs['name'] = tone.name
        sub.attrs['start_time'] = tone.start_time
        sub.create_dataset('frequency', data=tone.frequency)
        sub.create_dataset('amplitude', data=tone.amplitude)
        sub.create_dataset('phase', data=tone.phase)
        sub.create_dataset('segment_type', data=tone.segment_type)
        sub.create_dataset('segment_rate', data=tone.segment_rate)
        for name, values in tone.limits.items():
            sub.create_dataset(name, data=values)

save_sine_levels ¶

save_sine_levels(levels, group) -> None

Write a sine level set into group, one subgroup per tone, each in the ordinary data layout, since the set only groups the levels.

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

def save_sine_levels(levels, group) -> None:
    """Write a sine level set into `group`, one subgroup per tone, each
    in the ordinary data layout, since the set only groups the levels."""
    _write_strings(group, 'tone_order',
                   [level.tone for level in levels.levels])
    if levels.cycles is not None:
        group.attrs['sine_cycles'] = float(levels.cycles)
    tones = group.create_group('tones')
    for k, level in enumerate(levels.levels):
        save_data(level, tones.create_group(str(k)))

save ¶

save(obj: Any, path: str | PathLike) -> None

Save one Visual Dynamics object to a .vdyn (HDF5) file.

Raises TypeError for an object of a kind the format does not hold.

Parameters:

Name	Type	Description	Default
`obj`	`object`	The object to save.	required
`path`	`str or PathLike`	Where; `.vdyn` is added when the suffix is not there.	required

Returns:

Type	Description
`None`

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

def save(obj: Any, path: str | os.PathLike) -> None:
    """Save one Visual Dynamics object to a .vdyn (HDF5) file.

    Raises TypeError for an object of a kind the format does not hold.

    Parameters
    ----------
    obj : object
        The object to save.
    path : str or os.PathLike
        Where; `.vdyn` is added when the suffix is not there.

    Returns
    -------
    None
    """
    import h5py

    with h5py.File(_visualdynamics_path(path), 'w') as f:
        save_into(obj, f)

save_into ¶

save_into(obj: Any, f: File) -> None

Write one object into an open, empty HDF5 file.

This is save without opening the file, so that other formats (io.matlab, io.escdf_objects) can write the same layout into memory and store it inside their own files.

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

def save_into(obj: Any, f: h5py.File) -> None:
    """Write one object into an open, empty HDF5 file.

    This is `save` without opening the file, so that other formats
    (`io.matlab`, `io.escdf_objects`) can write the same layout into
    memory and store it inside their own files.
    """
    group_name, saver = _saver_for(obj)
    f.attrs['visualdynamics_schema'] = schema_for(obj)
    saver(obj, f.create_group(group_name))

save_test ¶

save_test(path: str | PathLike, name: str, objects: Mapping[str, Any], active_geometry: str | None = None, project_type: str | None = None, object_groups: Sequence[Mapping[str, Any]] | None = None, provenance: Mapping[str, Any] | None = None, data_index: Sequence[Mapping[str, Any]] | None = None) -> None

Save a whole test, every named object, to one .vdyn file.

Objects go in numbered groups with the name as an attribute, so a name may contain anything h5py would read as a path. objects maps each name to its object, in order. object_groups is the object groups, each a mapping with its members (object names); provenance records how each derived object was computed. Both are stored as JSON. data_index is the project's data index, its entries as plain values (DataIndex.records), stored as JSON in a data_index dataset at the root when there are any. .vdyn is added to path when it is not there.

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

def save_test(path: str | os.PathLike, name: str,
              objects: Mapping[str, Any],
              active_geometry: str | None = None,
              project_type: str | None = None,
              object_groups: Sequence[Mapping[str, Any]] | None = None,
              provenance: Mapping[str, Any] | None = None,
              data_index: Sequence[Mapping[str, Any]] | None = None) -> None:
    """Save a whole test, every named object, to one .vdyn file.

    Objects go in numbered groups with the name as an attribute, so a
    name may contain anything h5py would read as a path. `objects` maps
    each name to its object, in order. `object_groups` is the object
    groups, each a mapping with its `members` (object names);
    `provenance` records how each derived object was computed. Both are
    stored as JSON. `data_index` is the project's data index, its
    entries as plain values (`DataIndex.records`), stored as JSON in a
    `data_index` dataset at the root when there are any. `.vdyn` is
    added to `path` when it is not there.
    """
    import h5py

    with h5py.File(_visualdynamics_path(path), 'w') as f:
        save_test_into(f, name, objects, active_geometry, project_type,
                       object_groups, provenance, data_index)

save_test_into ¶

save_test_into(f: File, name: str, objects: Mapping[str, Any], active_geometry: str | None = None, project_type: str | None = None, object_groups: Sequence[Mapping[str, Any]] | None = None, provenance: Mapping[str, Any] | None = None, data_index: Sequence[Mapping[str, Any]] | None = None) -> None

save_test into an open, empty HDF5 file (see save_into).

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

def save_test_into(f: h5py.File, name: str, objects: Mapping[str, Any],
                   active_geometry: str | None = None,
                   project_type: str | None = None,
                   object_groups: Sequence[Mapping[str, Any]] | None = None,
                   provenance: Mapping[str, Any] | None = None,
                   data_index: Sequence[Mapping[str, Any]] | None = None
                   ) -> None:
    """`save_test` into an open, empty HDF5 file (see `save_into`)."""
    f.attrs['visualdynamics_schema'] = schema_for(objects)
    f.attrs['test_name'] = name
    f.attrs['active_geometry'] = active_geometry or ''
    f.attrs['project_type'] = project_type or ''
    import json
    f.attrs['links'] = json.dumps(object_groups or [])
    # how each derived object was computed, for the staleness
    # badges — settings fingerprints, so they survive the file
    f.attrs['provenance'] = json.dumps(provenance or {})
    if data_index:
        # the files the project rests on, by name, size and SHA-256. A
        # dataset, not an attribute: an attached settings list rides in
        # it, and an HDF5 attribute holds 64 KiB at most. Written only
        # when there is one, so a project with none is stored as it
        # always was; a reader that predates it opens the file and
        # leaves the dataset alone
        import h5py

        f.create_dataset('data_index', data=json.dumps(list(data_index)),
                         dtype=h5py.string_dtype('utf-8'))
    container = f.create_group('objects')
    for i, (obj_name, obj) in enumerate(objects.items()):
        kind, saver = _saver_for(obj)
        group = container.create_group(f'{i:04d}')
        group.attrs['name'] = obj_name
        group.attrs['kind'] = kind
        saver(obj, group)

load ¶

load(path: str | PathLike, progress: Callable[[int, int], None] | None = None) -> Any

Load a .vdyn file: the object it holds, or a whole test.

For a whole test, progress is called as (objects loaded, objects in the file), once at the start with 0 and once per object, because a large project can take minutes to load and only the reader knows how far along it is. For a single-object file it is not called, since loading one object is a single step.

Parameters:

Name	Type	Description	Default
`path`	`str or PathLike`	The `.vdyn` file.	required
`progress`	`callable`	Called as `progress(loaded, total)`, once up front and once per object.	`None`

Returns:

Type	Description
`object`	The object the file holds, or a `Project` for a whole test.

Raises:

Type	Description
`ValueError`	When the file has no schema stamp (it was not written by Visual Dynamics), was written with a newer schema, or holds nothing this reader recognizes.

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

def load(path: str | os.PathLike,
         progress: Callable[[int, int], None] | None = None) -> Any:
    """Load a .vdyn file: the object it holds, or a whole test.

    For a whole test, `progress` is called as (objects loaded, objects
    in the file), once at the start with 0 and once per object, because
    a large project can take minutes to load and only the reader knows
    how far along it is. For a single-object file it is not called,
    since loading one object is a single step.

    Parameters
    ----------
    path : str or os.PathLike
        The `.vdyn` file.
    progress : callable, optional
        Called as ``progress(loaded, total)``, once up front and once per
        object.

    Returns
    -------
    object
        The object the file holds, or a `Project` for a whole test.

    Raises
    ------
    ValueError
        When the file has no schema stamp (it was not written by Visual
        Dynamics), was written with a newer schema, or holds nothing
        this reader recognizes.
    """
    import h5py

    with h5py.File(path, 'r') as f:
        return load_from(f, progress, str(path))

load_from ¶

load_from(f: File, progress: Callable[[int, int], None] | None = None, path: str = '') -> Any

load from an open HDF5 file (see save_into). path names the file in error messages.

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

def load_from(f: h5py.File, progress: Callable[[int, int], None] | None = None,
              path: str = '') -> Any:
    """`load` from an open HDF5 file (see `save_into`). `path` names the
    file in error messages."""
    # A newer stamp means a newer Visual Dynamics wrote fields this
    # reader has no idea exist, and half-loading someone's project
    # quietly is worse than telling them to update. A missing stamp
    # means the file is not ours at all — every writer stamps.
    if 'visualdynamics_schema' not in f.attrs:
        raise ValueError(f'{path} is not a Visual Dynamics file '
                         '(no schema stamp)')
    written = int(f.attrs['visualdynamics_schema'])
    if written > SCHEMA_VERSION:
        raise ValueError(
            f'{path} was written by a newer Visual Dynamics '
            f'(schema {written}; this build reads up to '
            f'{SCHEMA_VERSION}). Update to open it.')
    if 'objects' in f:
        objects = {}
        keys = sorted(f['objects'])
        if progress is not None:
            progress(0, len(keys))
        for done, key in enumerate(keys, start=1):
            group = f['objects'][key]
            objects[group.attrs['name']] = (
                _LOADERS[group.attrs['kind']](group))
            if progress is not None:
                progress(done, len(keys))
        import json
        indexed = []
        if 'data_index' in f:
            raw = f['data_index'][()]
            indexed = json.loads(raw.decode('utf-8') if isinstance(raw, bytes)
                                 else raw)
        return Project(f.attrs['test_name'], objects,
                       f.attrs['active_geometry'] or None,
                       f.attrs['project_type'] or None,
                       json.loads(f.attrs['links']),
                       provenance=json.loads(f.attrs['provenance']),
                       data_index=indexed)
    for group_name, loader in _LOADERS.items():
        if group_name in f:
            return loader(f[group_name])
    raise ValueError(f"No recognized content in {path}")

visualdynamics.io.native¶

native ¶

Classes¶

Functions:¶

schema_for ¶

save_predictions ¶

save_run_log_extract ¶

save_sine_specification ¶

save_sine_levels ¶

save ¶

save_into ¶

save_test ¶

save_test_into ¶

load ¶

load_from ¶

`visualdynamics.io.native`¶