pycsamt.ai.data.contracts#

Canonical, dependency-light data contracts for MT/AMT surveys.

The classes in this module define the boundary between survey ingestion and all downstream AI or physics code. They deliberately depend only on NumPy: EDI readers, interpolation policies, neural-network frameworks, and Maxwell solver backends belong elsewhere.

The canonical impedance axis order is (station, frequency, component). Keeping this order explicit prevents a common class of scientifically quiet errors in which stations, tensor components, or frequency order are swapped without changing an array’s rank.

Functions

merge_surveys(surveys, *[, metadata])

Concatenate compatible surveys along the station axis.

Classes

`ImpedanceConvention`([time_dependence, ...])	Describe the sign, units, and rotation convention of impedance data.
`SurveyCoverage`(overall, by_station, ...[, ...])	Summarize the usable fraction of a survey along each data axis.
`SurveyData`(impedance, frequencies_hz, ...[, ...])	Validated MT/AMT observations on a common survey grid.

class pycsamt.ai.data.contracts.ImpedanceConvention(time_dependence='exp(+iwt)', units='V/A', rotation_deg=0.0, coordinate_orientation='x_north_y_east')[source]

Bases: object

Describe the sign, units, and rotation convention of impedance data.

Parameters:

time_dependence ({"exp(+iwt)", "exp(-iwt)"}, default="exp(+iwt)") – Fourier time convention used for the stored complex impedance. Solver predictions must use the same convention before residuals are formed.
units ({"V/A"}, default="V/A") – Physical unit of electric field divided by magnetic field. The current canonical contract accepts SI impedance only.
rotation_deg (float, default=0.0) – Clockwise rotation already applied to the horizontal tensor axes, in degrees. Values are normalized to the interval [0, 360).
coordinate_orientation (str, default="x_north_y_east") – Human-readable definition of the horizontal tensor axes.

Examples

Record a tensor rotated clockwise into a geological strike frame:

>>> convention = ImpedanceConvention(rotation_deg=32.0)
>>> convention.rotation_deg
32.0
>>> convention.to_dict()["time_dependence"]
'exp(+iwt)'

Notes

This object records a convention; it does not rotate or conjugate data. Such transformations must be explicit preprocessing operations that create a new SurveyData object with updated provenance.

time_dependence: str = 'exp(+iwt)'

units: str = 'V/A'

rotation_deg: float = 0.0

coordinate_orientation: str = 'x_north_y_east'

to_dict()[source]

Return a JSON-serializable representation.

Returns:: Convention fields with a schema discriminator.
Return type:: dict

Examples

>>> state = ImpedanceConvention().to_dict()
>>> state["schema_version"], state["units"]
(1, 'V/A')

classmethod from_dict(data)[source]

Restore and validate a serialized convention.

Parameters:: data (mapping) – State previously returned by to_dict().
Returns:: Validated immutable convention.
Return type:: ImpedanceConvention
Raises:: ValueError – If the schema version is unsupported or a field is invalid.

Examples

>>> state = ImpedanceConvention(rotation_deg=15).to_dict()
>>> ImpedanceConvention.from_dict(state).rotation_deg
15.0

class pycsamt.ai.data.contracts.SurveyCoverage(overall, by_station, by_frequency, by_component, tipper_overall=None)[source]

Bases: object

Summarize the usable fraction of a survey along each data axis.

Parameters:

overall (float) – Fraction of valid impedance observations across the complete cube.
by_station (ndarray, shape (n_station,)) – Valid fraction for each station.
by_frequency (ndarray, shape (n_frequency,)) – Valid fraction for each frequency.
by_component (ndarray, shape (n_component,)) – Valid fraction for each impedance component.
tipper_overall (float or None, optional) – Fraction of valid tipper observations, or None when absent.

Examples

Coverage is normally obtained from SurveyData.coverage():

>>> coverage = SurveyCoverage(1.0, [1.0], [1.0], [1.0])
>>> coverage.complete
True

overall: float

by_station: ndarray

by_frequency: ndarray

by_component: ndarray

tipper_overall: float | None = None

property complete: bool[source]

Whether every impedance observation is usable.

Returns:: True only when impedance coverage is exactly one. Tipper coverage is not included because tipper is optional.
Return type:: bool

Examples

>>> SurveyCoverage(0.5, [0.5], [0.5], [0.5]).complete
False

class pycsamt.ai.data.contracts.SurveyData(impedance, frequencies_hz, station_names, components, coordinates_m, impedance_error=None, valid=None, tipper=None, tipper_error=None, tipper_valid=None, crs=None, metadata=<factory>, convention=<factory>)[source]

Bases: object

Validated MT/AMT observations on a common survey grid.

Parameters:

impedance (array-like of complex, shape (n_station, n_frequency, n_component)) – Complex impedance in V/A. Invalid entries may be NaN but must be false in valid after construction.
frequencies_hz (array-like, shape (n_frequency,)) – Positive, finite, unique, strictly monotonic frequencies.
station_names (sequence of str) – Names corresponding exactly to the station and component axes.
components (sequence of str) – Names corresponding exactly to the station and component axes.
coordinates_m (array-like, shape (n_station, 2 or 3)) – Projected x/y coordinates and optional elevation in metres. A missing third column is represented by NaN elevation.
impedance_error (array-like, optional) – Positive absolute standard errors with the same shape as impedance. Entries without usable errors are invalidated.
valid (array-like of bool, optional) – Explicit observation mask. It is combined with finite-value and error checks; invalid data are never silently imputed.
tipper (array-like, optional) – Optional complex magnetic transfer functions shaped (n_station, n_frequency, 2) for Tx and Ty.
tipper_error (array-like, optional) – Optional complex magnetic transfer functions shaped (n_station, n_frequency, 2) for Tx and Ty.
tipper_valid (array-like, optional) – Optional complex magnetic transfer functions shaped (n_station, n_frequency, 2) for Tx and Ty.
crs (str, optional) – Coordinate reference system identifier. Projected coordinates should normally provide an EPSG or WKT identifier.
metadata (mapping, optional) – Finite JSON-serializable provenance only.
convention (ImpedanceConvention, optional) – Explicit complex sign, SI unit, tensor-axis, and rotation convention.

Variables:

impedance (ndarray) – Read-only complex impedance cube.
valid (ndarray of bool) – Read-only authoritative mask for usable impedance observations.

Examples

Construct a two-station survey with descending frequency order. Two- column coordinates are accepted and expanded with unknown elevations:

>>> z = np.ones((2, 3, 2), dtype=complex) * (1 + 2j)
>>> survey = SurveyData(
...     impedance=z,
...     frequencies_hz=[100.0, 10.0, 1.0],
...     station_names=["S01", "S02"],
...     components=["xy", "yx"],
...     coordinates_m=[[0.0, 0.0], [100.0, 0.0]],
...     crs="EPSG:32630",
... )
>>> survey.shape
(2, 3, 2)
>>> survey.frequency_order
'descending'

Notes

Construction copies all numerical inputs and marks them read-only. The dataclass is therefore safe to share between training, validation, and reporting code without accidental in-place mutation.

impedance: ndarray

frequencies_hz: ndarray

station_names: tuple[str, ...]

components: tuple[str, ...]

coordinates_m: ndarray

impedance_error: ndarray | None = None

valid: ndarray | None = None

tipper: ndarray | None = None

tipper_error: ndarray | None = None

tipper_valid: ndarray | None = None

crs: str | None = None

metadata: Mapping[str, Any]

convention: ImpedanceConvention

property shape: tuple[int, int, int][source]

Return the canonical impedance shape.

Returns:: (n_station, n_frequency, n_component).
Return type:: tuple of int

Examples

>>> z = np.ones((1, 2, 1), dtype=complex)
>>> s = SurveyData(z, [10, 1], ["S"], ["xy"], [[0, 0]])
>>> s.shape
(1, 2, 1)

property n_valid: int[source]

Return the number of usable impedance observations.

Returns:: Count of True entries in valid.
Return type:: int

Examples

>>> z = np.array([[[1 + 1j], [complex(np.nan, np.nan)]]])
>>> s = SurveyData(z, [10, 1], ["S"], ["xy"], [[0, 0]])
>>> s.n_valid
1

property n_stations: int[source]

Return the number of stations.

Returns:: Length of the station axis.
Return type:: int

Examples

>>> s = SurveyData(
...     np.ones((2, 1, 1), complex),
...     [1],
...     ["a", "b"],
...     ["xy"],
...     [[0, 0], [1, 0]],
... )
>>> s.n_stations
2

property n_frequencies: int[source]

Return the number of frequencies.

Returns:: Length of the frequency axis.
Return type:: int

Examples

>>> s = SurveyData(
...     np.ones((1, 2, 1), complex), [10, 1], ["S"], ["xy"], [[0, 0]]
... )
>>> s.n_frequencies
2

property n_components: int[source]

Return the number of impedance components.

Returns:: Length of the component axis.
Return type:: int

Examples

>>> s = SurveyData(
...     np.ones((1, 1, 2), complex), [1], ["S"], ["xy", "yx"], [[0, 0]]
... )
>>> s.n_components
2

property frequency_order: str[source]

Return the monotonic direction of the frequency axis.

Returns:: Direction in which frequencies are stored.
Return type:: {“ascending”, “descending”}

Examples

>>> s = SurveyData(
...     np.ones((1, 2, 1), complex), [1, 10], ["S"], ["xy"], [[0, 0]]
... )
>>> s.frequency_order
'ascending'

property has_tipper: bool[source]

Whether optional Tx/Ty transfer functions are present.

Returns:: True when tipper is populated.
Return type:: bool

Examples

>>> s = SurveyData(
...     np.ones((1, 1, 1), complex), [1], ["S"], ["xy"], [[0, 0]]
... )
>>> s.has_tipper
False

station_index(name)[source]

Return the integer position of a named station.

Parameters:: name (str) – Exact, case-sensitive station identifier.
Returns:: Position along the station axis.
Return type:: int
Raises:: KeyError – If name does not occur in station_names.

Examples

>>> s = SurveyData(
...     np.ones((2, 1, 1), complex),
...     [1],
...     ["A", "B"],
...     ["xy"],
...     [[0, 0], [1, 0]],
... )
>>> s.station_index("B")
1

component_index(name)[source]

Return the integer position of an impedance component.

Parameters:: name (str) – Exact, case-sensitive component name such as "xy".
Returns:: Position along the component axis.
Return type:: int
Raises:: KeyError – If name is not stored.

Examples

>>> s = SurveyData(
...     np.ones((1, 1, 2), complex), [1], ["S"], ["xy", "yx"], [[0, 0]]
... )
>>> s.component_index("yx")
1

coverage()[source]

Calculate valid-data coverage along every impedance axis.

Returns:: Overall, station, frequency, and component fractions. Optional tipper coverage is included when tipper data exist.
Return type:: SurveyCoverage

Examples

>>> z = np.ones((2, 2, 1), dtype=complex)
>>> mask = np.array([[[True], [False]], [[True], [True]]])
>>> s = SurveyData(
...     z, [10, 1], ["A", "B"], ["xy"], [[0, 0], [1, 0]], valid=mask
... )
>>> s.coverage().overall
0.75
>>> s.coverage().by_station.tolist()
[0.5, 1.0]

component_data(name)[source]

Return values, errors, and validity mask for one component.

Parameters:

name (str) – Exact component name.

Returns:

values (ndarray, shape (n_station, n_frequency)) – Read-only complex impedance view.
errors (ndarray or None) – Read-only absolute standard-error view, when available.
valid (ndarray of bool) – Read-only validity-mask view.

Raises:

KeyError – If the named component is unavailable.

Return type:

tuple[ndarray, ndarray | None, ndarray]

Examples

>>> z = np.ones((1, 2, 2), dtype=complex)
>>> s = SurveyData(z, [10, 1], ["S"], ["xy", "yx"], [[0, 0]])
>>> values, errors, valid = s.component_data("xy")
>>> values.shape, errors, valid.all()
((1, 2), None, True)

select(*, stations=None, frequencies=None, components=None)[source]

Select survey axes by integer position.

Parameters:

stations (sequence of int, optional) – Positions to retain. Omitted axes are retained completely. The requested order is preserved, but duplicates are rejected by the canonical unique-name/frequency validation.
frequencies (sequence of int, optional) – Positions to retain. Omitted axes are retained completely. The requested order is preserved, but duplicates are rejected by the canonical unique-name/frequency validation.
components (sequence of int, optional) – Positions to retain. Omitted axes are retained completely. The requested order is preserved, but duplicates are rejected by the canonical unique-name/frequency validation.

Returns:

New immutable survey containing the requested subset.

Return type:

SurveyData

Raises:

ValueError – If an index collection is not one-dimensional or creates a non-monotonic frequency axis.
IndexError – If an index lies outside its axis.

Examples

>>> z = np.ones((2, 3, 2), dtype=complex)
>>> s = SurveyData(
...     z, [100, 10, 1], ["A", "B"], ["xy", "yx"], [[0, 0], [1, 0]]
... )
>>> subset = s.select(stations=[1], frequencies=[1, 2], components=[0])
>>> subset.station_names, subset.frequencies_hz.tolist()
(('B',), [10.0, 1.0])

select_names(*, stations=None, components=None, frequency_min_hz=None, frequency_max_hz=None)[source]

Select stations/components by name and frequencies by interval.

Parameters:

stations (sequence of str, optional) – Exact names in the desired output order.
components (sequence of str, optional) – Exact names in the desired output order.
frequency_min_hz (float, optional) – Inclusive physical bounds. Their numerical order is independent of the stored frequency direction.
frequency_max_hz (float, optional) – Inclusive physical bounds. Their numerical order is independent of the stored frequency direction.

Returns:

New validated subset.

Return type:

SurveyData

Raises:

KeyError – If a station or component name is unknown.
ValueError – If bounds are invalid or select no frequencies.

Examples

>>> z = np.ones((2, 3, 2), dtype=complex)
>>> s = SurveyData(
...     z, [100, 10, 1], ["A", "B"], ["xy", "yx"], [[0, 0], [1, 0]]
... )
>>> subset = s.select_names(
...     stations=["B"], components=["yx"], frequency_min_hz=5
... )
>>> subset.shape
(1, 2, 1)

assert_compatible(other, *, require_crs=True, require_convention=True)[source]

Assert that two surveys share model-facing axes and conventions.

Parameters:

other (SurveyData) – Survey to compare.
require_crs (bool, default=True) – Require identical CRS identifiers.
require_convention (bool, default=True) – Require identical impedance conventions.

Returns:

Successful return means the surveys can share a fitted normalizer or be concatenated along their station axes.

Return type:

None

Raises:

TypeError – If other is not SurveyData.
ValueError – If frequency values/order, components/order, CRS, or convention differ under the requested policy.

Examples

>>> a = SurveyData(
...     np.ones((1, 1, 1), complex),
...     [1],
...     ["A"],
...     ["xy"],
...     [[0, 0]],
...     crs="EPSG:32630",
... )
>>> b = SurveyData(
...     np.ones((1, 1, 1), complex),
...     [1],
...     ["B"],
...     ["xy"],
...     [[1, 0]],
...     crs="EPSG:32630",
... )
>>> a.assert_compatible(b) is None
True

with_metadata(metadata, *, merge=True)[source]

Return a copy with validated provenance metadata.

Parameters:

metadata (mapping) – Finite JSON-serializable values to add or use as replacement.
merge (bool, default=True) – Merge with existing metadata when true. New keys replace existing keys. When false, replace the complete metadata mapping.

Returns:

New survey sharing no mutable numerical or metadata state.

Return type:

SurveyData

Examples

>>> s = SurveyData(
...     np.ones((1, 1, 1), complex), [1], ["S"], ["xy"], [[0, 0]]
... )
>>> tagged = s.with_metadata({"line": "L18", "rotation_deg": 0.0})
>>> tagged.metadata["line"]
'L18'

summary()[source]

Return a compact JSON-serializable survey summary.

Returns:: Axis sizes, bounds, coverage, CRS, component names, convention, and optional-data flags. Raw observations are not included.
Return type:: dict

Examples

>>> s = SurveyData(
...     np.ones((1, 2, 1), complex), [10, 1], ["S"], ["xy"], [[0, 0]]
... )
>>> s.summary()["frequency_range_hz"]
[1.0, 10.0]

to_npz(path)[source]

Write a lossless, pickle-free compressed archive.

Parameters:: path (str or pathlib.Path) – Destination .npz path. NumPy appends .npz when the supplied filename has no such suffix.
Returns:: Requested destination path.
Return type:: pathlib.Path

Notes

The archive contains only numerical arrays and JSON/Unicode scalars. from_npz() therefore loads it with allow_pickle=False.

Examples

>>> from tempfile import TemporaryDirectory
>>> s = SurveyData(
...     np.ones((1, 1, 1), complex), [1], ["S"], ["xy"], [[0, 0]]
... )
>>> with TemporaryDirectory() as directory:
...     path = s.to_npz(Path(directory) / "survey.npz")
...     restored = SurveyData.from_npz(path)
>>> restored.station_names
('S',)

classmethod from_npz(path)[source]

Load and validate a survey archive without enabling pickle.

Parameters:

path (str or pathlib.Path) – Archive previously written by to_npz().

Returns:

Newly validated immutable survey.

Return type:

SurveyData

Raises:

ValueError – If the schema is unsupported or restored values violate the data contract.
OSError – If the path cannot be read as a NumPy archive.

Examples

>>> from tempfile import TemporaryDirectory
>>> original = SurveyData(
...     np.ones((1, 1, 1), complex), [1], ["S"], ["xy"], [[0, 0]]
... )
>>> with TemporaryDirectory() as directory:
...     path = original.to_npz(Path(directory) / "survey.npz")
...     loaded = SurveyData.from_npz(path)
>>> np.array_equal(loaded.impedance, original.impedance)
True

Notes

Schema version 1 remains readable and receives the default ImpedanceConvention. New writes use schema version 2.

pycsamt.ai.data.contracts.merge_surveys(surveys, *, metadata=None)[source]

Concatenate compatible surveys along the station axis.

Parameters:

surveys (sequence of SurveyData) – Non-empty surveys with identical frequencies, component order, CRS, impedance convention, and optional-data availability. Station names must be globally unique.
metadata (mapping, optional) – Metadata for the merged object. By default a minimal provenance record containing source_survey_count is used; input metadata are not combined implicitly because equal keys may have different meanings.

Returns:

Canonical survey with concatenated station-axis arrays.

Return type:

SurveyData

Raises:

ValueError – If no surveys are supplied, surveys are incompatible, optional error or tipper availability differs, or station names overlap.
TypeError – If an item is not SurveyData.

Examples

>>> a = SurveyData(
...     np.ones((1, 2, 1), complex), [10, 1], ["A"], ["xy"], [[0, 0]]
... )
>>> b = SurveyData(
...     np.ones((1, 2, 1), complex), [10, 1], ["B"], ["xy"], [[1, 0]]
... )
>>> merged = merge_surveys([a, b])
>>> merged.station_names, merged.shape
(('A', 'B'), (2, 2, 1))

Notes

This function performs no frequency interpolation, coordinate projection, tensor rotation, or unit conversion. Those operations must be explicit and completed before merging.