geopulse.io

Data I/O: HDF5, GeoJSON, IAGA-2002.

class geopulse.io.MatpowerGMDCase(source_path, n_dc_nodes, dc_nodes=<factory>, dc_branches=<factory>, bus_latlon=<factory>, ac_bus_kv=<factory>, ac_branches=<factory>)[source]

Bases: object

Structured contents of a MATPOWER-GMD .m file.

Parameters:

source_path (Path)
n_dc_nodes (int)
dc_nodes (list[dict])
dc_branches (list[dict])
bus_latlon (dict[int, tuple[float, float]])
ac_bus_kv (dict[int, float])
ac_branches (list[dict])

source_path

Path the case was loaded from.

Type:: pathlib.Path

n_dc_nodes

Number of DC (grounding) nodes.

Type:: int

dc_nodes

One entry per DC node, ordered by 1-based row position within mpc.gmd_bus. Fields:

row 1-based row index (DC node id used by branches)
parent_ac_bus AC bus number this DC node grounds
status 1 = in service, 0 = out
g_gnd_S grounding conductance in Siemens
name human name (e.g. "dc_sub1")

Type:: list of dict

dc_branches

DC branches. Fields:

from_row, to_row 1-based DC node row indices
parent_ac_branch 1-based AC branch index (or 0 for none)
status 1 in service, 0 out
resistance_Ohm DC resistance
length_km geographic length (km); 0 for xfmr windings
name human name

Type:: list of dict

bus_latlon

ac_bus_id -> (latitude_deg, longitude_deg).

Type:: dict of int -> (float, float)

ac_bus_kv

ac_bus_id -> baseKV.

Type:: dict of int -> float

ac_branches

AC-side branches, retained for case-configuration helpers (each has f, t, rateA, ratio, status).

Type:: list of dict

source_path: Path

n_dc_nodes: int

dc_nodes: list[dict]

dc_branches: list[dict]

bus_latlon: dict[int, tuple[float, float]]

ac_bus_kv: dict[int, float]

ac_branches: list[dict]

geopulse.io.read_matpower_gmd(path)[source]

Parse a MATPOWER-GMD .m file.

Parameters:: path (str | Path) – Path to the .m file.
Return type:: MatpowerGMDCase
Returns:: MatpowerGMDCase – Structured case data.
Raises:: DataError – If the file is missing required blocks (gmd_bus, gmd_branch, bus, bus_gmd) or has malformed rows.

geopulse.io.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.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)

Modules

`geojson`	GeoJSON reader for network topologies — STUB (WP2).
`hdf5`	Schema-versioned HDF5 reader/writer.
`iaga2002`	IAGA-2002 magnetometer format parser.
`matpower`	MATPOWER-GMD (`.m`) parser for power-grid GIC test cases.