geopulse.io.hdf5

Schema-versioned HDF5 reader/writer.

Every file written by GeoPulse carries a schema_version attribute at the root. The reader migrates older schemas forward automatically so that files written by previous versions remain readable. Writers always emit the current schema.

Layout (schema v1):

/
  attrs:
    schema_version   int
    geopulse_version str
    created_utc      str  (ISO-8601)
    description      str
  source/            (optional)
    attrs: {station_id, latitude_deg, longitude_deg, sampling_rate_Hz}
    time_s, bx_T, by_T, bz_T   datasets
  earth/             (optional)
    attrs: {model_name, tier}
    impedance/       group with impedance-specific layout
  results/           (optional)
    efield/          Ex_Vm, Ey_Vm
    gic/             node_ids, currents_A

Module Attributes

CURRENT_SCHEMA_VERSION

Integer schema version — bump on any breaking change to the layout.

Functions

`read_results`(path)	Read an HDF5 file, migrating older schemas forward.
`write_results`(path[, description])	Write results to HDF5 with the current schema version.

geopulse.io.hdf5.CURRENT_SCHEMA_VERSION: int = 1: Integer schema version — bump on any breaking change to the layout.

geopulse.io.hdf5.read_results(path)[source]

Read an HDF5 file, migrating older schemas forward.

Parameters:: path (str | Path) – Path to read.
Return type:: dict[str, Any]
Returns:: dict – Nested dict of the file contents, including root attributes under keys schema_version, geopulse_version, created_utc, and description.
Raises:: DataError – If the file is missing the schema_version attribute.

geopulse.io.hdf5.write_results(path, description='', **groups)[source]

Write results to HDF5 with the current schema version.

Parameters:

path (str | Path) – Destination path. Any existing file is overwritten.
description (str) – Free-text description recorded at the root.
**groups (Any) – Named groups to write. Each value should be a nested dict of arrays, scalars, or sub-dicts.

Return type:

None

Examples

>>> import numpy as np, tempfile, os
>>> tmp = tempfile.NamedTemporaryFile(suffix='.h5', delete=False).name
>>> write_results(tmp, description="demo",
...               source={"time_s": np.arange(10.0)})
>>> data = read_results(tmp)
>>> data["source"]["time_s"].shape
(10,)
>>> os.unlink(tmp)