2.25.3.2. pycsamt.ai.data#

Canonical survey contracts, normalization, realization splits, and dataset manifests for learned inversion workflows.

Canonical data contracts for reproducible AI-assisted EM inversion.

This package owns survey tensors, validity masks, coordinate metadata, normalisation state, realization-level dataset splits, and dataset manifests. It complements pycsamt.ai.training.dataset, which owns framework-facing training dataset wrappers.

The first implementation milestone will add dependency-light canonical data objects here. Importing this package intentionally has no optional machine- learning or forward-solver side effects.

class pycsamt.ai.data.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>)#

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]#

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#

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#

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#

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#

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#

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#

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)#

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)#

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()#

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)#

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)#

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)#

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)#

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)#

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()#

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)#

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)#

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.

class pycsamt.ai.data.SurveyCoverage(overall, by_station, by_frequency, by_component, tipper_overall=None)#

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#

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.ImpedanceConvention(time_dependence='exp(+iwt)', units='V/A', rotation_deg=0.0, coordinate_orientation='x_north_y_east')#

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()#

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)#

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

pycsamt.ai.data.merge_surveys(surveys, *, metadata=None)#

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.

class pycsamt.ai.data.ComplexZScore(mean, scale, frequencies_hz, components, eps=1e-08, count=None, weight_sum=None, weighting='uniform', ddof=0, convention=None, training_survey_count=None, training_station_count=None)#

Bases: object

Immutable per-frequency/component complex z-score state.

Real and imaginary impedance parts are standardized independently. The stored statistic shape is (frequency, component, channel) where the final channels are real and imaginary.

Parameters:

mean (ndarray) – Per-feature location and positive scale arrays.
scale (ndarray) – Per-feature location and positive scale arrays.
frequencies_hz (ndarray) – Exact fitted frequency grid and order.
components (sequence of str) – Exact fitted component order.
eps (float, default=1e-8) – Minimum allowed scale.
count (ndarray or None, optional) – Number of valid training observations supporting each channel.
weight_sum (ndarray or None, optional) – Sum of fitting weights supporting each channel.
weighting ({"uniform", "inverse_variance"}, default="uniform") – Statistic weighting policy.
ddof (int, default=0) – Delta degrees of freedom used for uniform variance.
convention (ImpedanceConvention or None, optional) – Complex convention bound to the fitted state. None is accepted only for legacy schema-1 states and skips convention compatibility checks.
training_survey_count (int, optional) – Audit counts describing the data used during fitting.
training_station_count (int, optional) – Audit counts describing the data used during fitting.

Examples

Fit on training data and reuse the same state for later data:

>>> z = np.array([[[1 + 2j]], [[3 + 4j]]])
>>> training = SurveyData(z, [1], ["A", "B"], ["xy"], [[0, 0], [1, 0]])
>>> normalizer = ComplexZScore.fit(training)
>>> features, mask = normalizer.transform(training)
>>> np.allclose(features[:, 0, 0, 0], [-1, 1])
True
>>> mask.all()
True

mean: ndarray#

scale: ndarray#

frequencies_hz: ndarray#

components: tuple[str, ...]#

eps: float = 1e-08#

count: ndarray | None = None#

weight_sum: ndarray | None = None#

weighting: str = 'uniform'#

ddof: int = 0#

convention: ImpedanceConvention | None = None#

training_survey_count: int | None = None#

training_station_count: int | None = None#

classmethod fit(surveys, *, eps=1e-08, weighting='uniform', ddof=0)#

Fit statistics from explicitly supplied training surveys.

Parameters:

surveys (SurveyData or sequence of SurveyData) – Training surveys sharing frequency grid/order, component order, and impedance convention. Stations are pooled across surveys.
eps (float, default=1e-8) – Positive scale floor for constant features.
weighting ({"uniform", "inverse_variance"}, default="uniform") – Use equal weights or inverse squared impedance errors. The latter requires error arrays on every training survey.
ddof (int, default=0) – Delta degrees of freedom for uniform variance. It must be smaller than every feature’s valid observation count and must be zero for inverse-variance weighting.

Returns:

Immutable fitted state for reuse on validation/test/field data.

Return type:

ComplexZScore

Raises:

ValueError – If axes or conventions differ, a feature has insufficient valid observations, or requested weights are unavailable.
TypeError – If an input is not SurveyData.

Examples

>>> z = np.array([[[1 + 1j]], [[2 + 3j]], [[5 + 7j]]])
>>> survey = SurveyData(
...     z, [1], ["a", "b", "c"], ["xy"], [[0, 0], [1, 0], [2, 0]]
... )
>>> state = ComplexZScore.fit(survey, ddof=1)
>>> state.training_station_count
3
>>> state.count[0, 0, 0]
3

property state_hash: str#

Return a deterministic digest of the complete fitted state.

Returns:: Lowercase SHA-256 digest suitable for artifact provenance.
Return type:: str

Examples

>>> state = ComplexZScore(
...     np.zeros((1, 1, 2)), np.ones((1, 1, 2)), [1], ["xy"]
... )
>>> len(state.state_hash)
64

property feature_names: tuple[str, ...]#

Return flattened feature names in canonical array order.

Returns:: Names ordered by frequency, component, then real/imaginary channel.
Return type:: tuple of str

Examples

>>> state = ComplexZScore(
...     np.zeros((1, 1, 2)), np.ones((1, 1, 2)), [10], ["xy"]
... )
>>> state.feature_names
('10Hz:xy:real', '10Hz:xy:imag')

validate_survey(survey)#

Validate survey axes and complex convention against fitted state.

Parameters:

survey (SurveyData) – Candidate survey for transformation.

Returns:

Successful return means the survey is transform-compatible.

Return type:

None

Raises:

TypeError – If survey is not SurveyData.
ValueError – If frequency values/order, component order, or a recorded complex convention differs.

Examples

>>> survey = SurveyData(
...     np.ones((1, 1, 1), complex), [1], ["S"], ["xy"], [[0, 0]]
... )
>>> state = ComplexZScore.fit(survey)
>>> state.validate_survey(survey) is None
True

transform(survey, *, fill_value=0.0, clip=None)#

Normalize impedance and return features with an explicit mask.

Parameters:

survey (SurveyData) – Compatible survey to transform.
fill_value (float, default=0.0) – Finite value placed in invalid feature channels.
clip (float or None, optional) – Symmetric positive z-score limit. Clipping can improve numerical robustness but makes exact inversion impossible for clipped values.

Returns:

features (ndarray) – Shape (station, frequency, component, 2) with real then imaginary channels.
valid (ndarray of bool) – Same shape, retaining the observation mask independently of fill.

Return type:

tuple[ndarray, ndarray]

Examples

>>> z = np.array([[[1 + 2j]], [[3 + 4j]]])
>>> survey = SurveyData(z, [1], ["A", "B"], ["xy"], [[0, 0], [1, 0]])
>>> features, valid = ComplexZScore.fit(survey).transform(survey)
>>> features.shape, valid.all()
((2, 1, 1, 2), True)

transform_errors(survey, *, fill_value=1.0)#

Propagate absolute impedance errors into normalized channel units.

Parameters:

survey (SurveyData) – Compatible survey containing impedance_error.
fill_value (float, default=1.0) – Positive finite error assigned to invalid channels.

Returns:

errors (ndarray) – Normalized errors shaped (station, frequency, component, 2). The same scalar complex-impedance error is divided by the separate real and imaginary channel scales.
valid (ndarray of bool) – Expanded observation mask.

Raises:

ValueError – If the survey has no impedance errors or fill is invalid.

Return type:

tuple[ndarray, ndarray]

Examples

>>> z = np.array([[[1 + 1j]], [[3 + 3j]]])
>>> s = SurveyData(
...     z,
...     [1],
...     ["A", "B"],
...     ["xy"],
...     [[0, 0], [1, 0]],
...     impedance_error=np.ones_like(z.real),
... )
>>> errors, valid = ComplexZScore.fit(s).transform_errors(s)
>>> errors.shape, valid.all()
((2, 1, 1, 2), True)

transform_survey(survey, *, fill_value=0.0, error_fill_value=1.0, clip=None)#

Create a labeled immutable normalized-survey container.

Parameters:

survey (SurveyData) – Compatible survey.
fill_value (float, optional) – Finite values for invalid features and invalid normalized errors.
error_fill_value (float, optional) – Finite values for invalid features and invalid normalized errors.
clip (float or None, optional) – Symmetric feature clipping threshold.

Returns:

Features, mask, optional propagated errors, axes, and state digest.

Return type:

NormalizedSurvey

Examples

>>> z = np.array([[[1 + 2j]], [[3 + 4j]]])
>>> s = SurveyData(z, [1], ["A", "B"], ["xy"], [[0, 0], [1, 0]])
>>> result = ComplexZScore.fit(s).transform_survey(s)
>>> result.station_names
('A', 'B')

inverse_transform(values, *, valid=None, invalid_value=nan + nanj)#

Convert normalized channels back to complex SI impedance.

Parameters:

values (ndarray) – Shape (station, frequency, component, 2).
valid (ndarray of bool, optional) – Mask with the same shape. An impedance is retained only if both real and imaginary channels are valid.
invalid_value (complex, default=nan+nanj) – Value assigned where either channel is invalid.

Returns:

Shape (station, frequency, component) in V/A.

Return type:

ndarray of complex

Examples

>>> z = np.array([[[1 + 2j]], [[3 + 4j]]])
>>> s = SurveyData(z, [1], ["A", "B"], ["xy"], [[0, 0], [1, 0]])
>>> state = ComplexZScore.fit(s)
>>> features, mask = state.transform(s)
>>> np.allclose(state.inverse_transform(features, valid=mask), z)
True

to_dict()#

Return a complete JSON-serializable schema-2 state.

Returns:: Fitted statistics, axes, counts, fitting policy, convention, and training audit counts.
Return type:: dict

Examples

>>> state = ComplexZScore(
...     np.zeros((1, 1, 2)), np.ones((1, 1, 2)), [1], ["xy"]
... )
>>> state.to_dict()["schema_version"]
2

classmethod from_dict(data)#

Restore a schema-1 or schema-2 fitted state.

Parameters:: data (mapping) – State previously returned by to_dict(), or the earlier schema-1 Cartesian z-score representation.
Returns:: Validated immutable runtime state.
Return type:: ComplexZScore
Raises:: ValueError – If the schema discriminator is unsupported or statistics violate the normalization contract.

Examples

>>> original = ComplexZScore(
...     np.zeros((1, 1, 2)), np.ones((1, 1, 2)), [1], ["xy"]
... )
>>> restored = ComplexZScore.from_dict(original.to_dict())
>>> restored.state_hash == original.state_hash
True

class pycsamt.ai.data.NormalizedSurvey(values, valid, frequencies_hz, station_names, components, errors=None, state_hash=None)#

Bases: object

Normalized real/imaginary feature channels and their validity state.

Parameters:

values (ndarray, shape (n_station, n_frequency, n_component, 2)) – Normalized channels ordered as real then imaginary.
valid (ndarray of bool) – Mask with the same shape as values. Both channels of an impedance observation normally share one validity state.
frequencies_hz (ndarray, shape (n_frequency,)) – Frequency axis used by the fitted normalizer.
station_names (sequence of str) – Axis labels corresponding to values.
components (sequence of str) – Axis labels corresponding to values.
errors (ndarray or None, optional) – Normalized absolute standard errors with the same shape as values.
state_hash (str or None, optional) – Digest of the normalization state that produced the features.

Examples

NormalizedSurvey objects are normally created with ComplexZScore.transform_survey():

>>> values = np.zeros((1, 2, 1, 2))
>>> result = NormalizedSurvey(
...     values, np.ones_like(values, bool), [10, 1], ["S"], ["xy"]
... )
>>> result.shape
(1, 2, 1, 2)
>>> result.n_valid_observations
2

values: ndarray#

valid: ndarray#

frequencies_hz: ndarray#

station_names: tuple[str, ...]#

components: tuple[str, ...]#

errors: ndarray | None = None#

state_hash: str | None = None#

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

Return the normalized feature shape.

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

Examples

>>> x = np.zeros((2, 3, 1, 2))
>>> n = NormalizedSurvey(
...     x, np.ones_like(x, bool), [100, 10, 1], ["A", "B"], ["xy"]
... )
>>> n.shape
(2, 3, 1, 2)

property n_valid_observations: int#

Return the number of valid complex observations.

Returns:: Count on the impedance grid, not the doubled channel count.
Return type:: int

Examples

>>> x = np.zeros((1, 1, 1, 2))
>>> n = NormalizedSurvey(x, np.ones_like(x, bool), [1], ["S"], ["xy"])
>>> n.n_valid_observations
1

flatten()#

Flatten frequency, component, and channel axes for dense models.

Returns:

values (ndarray, shape (n_station, n_feature)) – Read-only station-major feature matrix.
valid (ndarray of bool, shape (n_station, n_feature)) – Read-only feature mask in identical order.

Return type:

tuple[ndarray, ndarray]

Examples

>>> x = np.zeros((2, 3, 2, 2))
>>> n = NormalizedSurvey(
...     x,
...     np.ones_like(x, bool),
...     [100, 10, 1],
...     ["A", "B"],
...     ["xy", "yx"],
... )
>>> n.flatten()[0].shape
(2, 12)

class pycsamt.ai.data.RealizationSplit(train, validation, test, seed=None, lineage=<factory>, strategy='random')#

Bases: object

Immutable train/validation/test assignment of geological realizations.

Parameters:

train (sequence of str) – Unique realization identifiers. The three partitions must be disjoint, and training cannot be empty.
validation (sequence of str) – Unique realization identifiers. The three partitions must be disjoint, and training cannot be empty.
test (sequence of str) – Unique realization identifiers. The three partitions must be disjoint, and training cannot be empty.
seed (int or None, optional) – Random seed used to create the assignment. None records that the seed is unknown or intentionally nondeterministic.
lineage (mapping, optional) – Complete mapping from every realization ID to its parent geological lineage. IDs sharing a lineage must occupy the same partition.
strategy (str, default="random") – Human-readable splitting strategy identifier.

Examples

>>> split = RealizationSplit(("r1", "r2"), ("r3",), ("r4",), seed=7)
>>> split.sizes
{'train': 2, 'validation': 1, 'test': 1}
>>> split.partition("r3")
'validation'

train: tuple[str, ...]#

validation: tuple[str, ...]#

test: tuple[str, ...]#

seed: int | None = None#

lineage: Mapping[str, str]#

strategy: str = 'random'#

property all_ids: tuple[str, ...]#

Return all realization IDs in partition order.

Returns:: Training IDs followed by validation IDs and test IDs.
Return type:: tuple of str

Examples

>>> RealizationSplit(("a",), ("b",), ("c",)).all_ids
('a', 'b', 'c')

property sizes: dict[str, int]#

Return the number of realizations in each partition.

Returns:: train, validation, and test counts.
Return type:: dict

Examples

>>> RealizationSplit(("a", "b"), (), ("c",)).sizes["train"]
2

property fractions: dict[str, float]#

Return realized partition fractions.

Returns:: Counts divided by total realization count. These may differ from requested targets when lineages contain multiple realizations.
Return type:: dict

Examples

>>> RealizationSplit(("a", "b"), ("c",), ("d",)).fractions
{'train': 0.5, 'validation': 0.25, 'test': 0.25}

property split_hash: str#

Return a deterministic SHA-256 digest of the assignment.

Returns:: Digest covering partitions, seed, lineage, and strategy.
Return type:: str

Examples

>>> len(RealizationSplit(("a",), (), ()).split_hash)
64

partition(realization_id)#

Return the partition containing a realization.

Parameters:: realization_id (str) – Exact realization identifier.
Returns:: Partition name.
Return type:: {“train”, “validation”, “test”}
Raises:: KeyError – If the identifier is unknown.

Examples

>>> RealizationSplit(("a",), ("b",), ()).partition("b")
'validation'

ids_for(partition)#

Return IDs assigned to a named partition.

Parameters:: partition ({"train", "validation", "test"}) – Partition to retrieve.
Returns:: Immutable realization IDs.
Return type:: tuple of str
Raises:: ValueError – If partition is unsupported.

Examples

>>> RealizationSplit(("a",), ("b",), ()).ids_for("train")
('a',)

mask(realization_ids, partition, *, unknown='raise')#

Build a Boolean sample mask from realization IDs.

Parameters:

realization_ids (sequence of str) – Per-sample realization IDs; duplicates are allowed because many stations or noise variants may belong to one realization.
partition ({"train", "validation", "test"}) – Partition selected as True.
unknown ({"raise", "false"}, default="raise") – Raise for IDs absent from the split or mark them false.

Returns:

One value per supplied sample ID.

Return type:

ndarray of bool

Examples

>>> split = RealizationSplit(("a",), (), ("b",))
>>> split.mask(["a", "b", "a"], "train").tolist()
[True, False, True]

assert_complete(expected_ids)#

Assert exact coverage of an expected realization collection.

Parameters:: expected_ids (sequence of str) – Unique IDs expected across all partitions.
Returns:: Successful return proves no expected ID is missing or unexpected.
Return type:: None
Raises:: ValueError – If expected IDs are duplicated or coverage differs.

Examples

>>> split = RealizationSplit(("a",), (), ("b",))
>>> split.assert_complete(["b", "a"]) is None
True

assert_no_lineage_leakage(lineage=None)#

Assert that every parent lineage occurs in one partition only.

Parameters:: lineage (mapping or None, optional) – Complete external ID-to-lineage mapping. When omitted, use the mapping persisted on the split. If neither exists, each realization is already an independent unit and the check succeeds.
Returns:: Successful return proves no supplied lineage crosses partitions.
Return type:: None
Raises:: ValueError – If mapping coverage is incomplete or a lineage leaks.

Examples

>>> split = RealizationSplit(("a1", "a2"), (), ("b1",))
>>> split.assert_no_lineage_leakage(
...     {"a1": "a", "a2": "a", "b1": "b"}
... ) is None
True

reassign(realization_ids, partition)#

Return a copy with complete realizations moved to one partition.

Parameters:

realization_ids (sequence of str) – Known IDs to move. When lineage is recorded, all members of each affected lineage must be supplied together.
partition ({"train", "validation", "test"}) – Destination partition.

Returns:

New validated assignment. Existing relative order is retained; moved IDs are appended in the requested order.

Return type:

RealizationSplit

Raises:

KeyError – If an ID is unknown.
ValueError – If a lineage is moved partially or training would become empty.

Examples

>>> split = RealizationSplit(("a", "b"), (), ("c",))
>>> split.reassign(["b"], "validation").validation
('b',)

to_dict()#

Return the complete schema-2 JSON representation.

Returns:: Mutable copy of partitions, seed, lineage, and strategy.
Return type:: dict

Examples

>>> RealizationSplit(("a",), (), ()).to_dict()["schema_version"]
2

classmethod from_dict(data)#

Restore schema-1 or schema-2 split state.

Parameters:: data (mapping) – Serialized split dictionary.
Returns:: Validated immutable runtime split.
Return type:: RealizationSplit
Raises:: ValueError – If the schema is unsupported or assignments leak/overlap.

Examples

>>> split = RealizationSplit(("a",), (), ("b",), seed=1)
>>> RealizationSplit.from_dict(split.to_dict()) == split
True

pycsamt.ai.data.split_realizations(realization_ids, *, validation_fraction=0.1, test_fraction=0.1, seed=0, lineage=None)#

Create a deterministic realization- or lineage-level random split.

Parameters:

realization_ids (sequence of str) – Unique geological realization IDs. Input order does not affect output.
validation_fraction (float, default=0.1) – Target fractions in [0, 1) whose sum is less than one.
test_fraction (float, default=0.1) – Target fractions in [0, 1) whose sum is less than one.
seed (int or None, default=0) – NumPy random-generator seed.
lineage (mapping, optional) – Complete ID-to-parent mapping. Whole lineages are assigned together, so realized fractions can differ from targets.

Returns:

Immutable leakage-checked assignment.

Return type:

RealizationSplit

Raises:

ValueError – If IDs, fractions, or lineage coverage are invalid or no training group can remain.

Examples

>>> ids = [f"r{i}" for i in range(10)]
>>> first = split_realizations(
...     ids, validation_fraction=0.2, test_fraction=0.2, seed=4
... )
>>> second = split_realizations(
...     list(reversed(ids)),
...     validation_fraction=0.2,
...     test_fraction=0.2,
...     seed=4,
... )
>>> first == second
True

Keep multiple noise variants of one parent in a single partition:

>>> lineage = {"a-clean": "a", "a-noisy": "a", "b": "b", "c": "c"}
>>> split = split_realizations(
...     list(lineage),
...     validation_fraction=0.25,
...     test_fraction=0.25,
...     lineage=lineage,
... )
>>> split.partition("a-clean") == split.partition("a-noisy")
True

pycsamt.ai.data.realization_folds(realization_ids, *, n_splits=5, seed=0, lineage=None)#

Build deterministic group-safe cross-validation test folds.

Parameters:

realization_ids (sequence of str) – Unique realization IDs.
n_splits (int, default=5) – Number of folds. It cannot exceed independent lineage count.
seed (int or None, default=0) – Seed controlling shuffled group order.
lineage (mapping, optional) – Complete ID-to-parent mapping. Each lineage appears in exactly one test fold and never crosses train/test within a fold.

Returns:

Splits with empty validation partitions. Across the tuple, every input realization appears in test exactly once.

Return type:

tuple of RealizationSplit

Raises:

ValueError – If fewer than two folds are requested or independent groups are insufficient.

Examples

>>> folds = realization_folds(["a", "b", "c", "d"], n_splits=2, seed=1)
>>> len(folds)
2
>>> sorted(item for fold in folds for item in fold.test)
['a', 'b', 'c', 'd']

class pycsamt.ai.data.DatasetManifest(dataset_id, generator, generator_version, configuration, split, sample_count, created_utc=None, artifacts=<factory>, schema_version=2)#

Bases: object

Identify a generated dataset and its complete reproducibility state.

Parameters:

dataset_id (str) – Portable identifier containing letters, digits, dots, underscores, or hyphens. It must start with a letter or digit.
generator (str) – Fully qualified generator name and its version or source revision.
generator_version (str) – Fully qualified generator name and its version or source revision.
configuration (mapping) – Finite JSON-compatible generator configuration. It is recursively copied and frozen.
split (RealizationSplit) – Disjoint realization-level train/validation/test assignment.
sample_count (int) – Number of samples represented by the dataset.
created_utc (str or None, optional) – Timezone-aware ISO-8601 creation time. It is normalized to UTC.
artifacts (mapping, optional) – Normalized relative paths mapped to ArtifactRecord objects or their serialized dictionaries.
schema_version (int, default=2) – Manifest format version. New manifests use version 2.

Examples

>>> split = RealizationSplit(("r1", "r2"), ("r3",), ("r4",), seed=7)
>>> manifest = DatasetManifest(
...     dataset_id="willy-2d-v1",
...     generator="pycsamt.ai.geology.correlated2d",
...     generator_version="0.1.0",
...     configuration={"seed": 7, "correlation_m": [1000, 100]},
...     split=split,
...     sample_count=4,
... )
>>> len(manifest.configuration_hash)
64

dataset_id: str#

generator: str#

generator_version: str#

configuration: Mapping[str, Any]#

split: RealizationSplit#

sample_count: int#

created_utc: str | None = None#

artifacts: Mapping[str, ArtifactRecord | Mapping[str, Any] | str]#

schema_version: int = 2#

property configuration_hash: str#

Return the canonical configuration digest.

Returns:: SHA-256 digest of generator configuration only.
Return type:: str

Examples

>>> split = RealizationSplit(("r1",), (), ())
>>> m = DatasetManifest("d", "g", "1", {"seed": 0}, split, 1)
>>> m.configuration_hash == canonical_hash({"seed": 0})
True

property manifest_hash: str#

Return a digest of the complete serialized manifest.

Returns:: SHA-256 digest covering configuration, split, timestamps, and all artifact records.
Return type:: str

Examples

>>> split = RealizationSplit(("r1",), (), ())
>>> m = DatasetManifest("d", "g", "1", {}, split, 1)
>>> len(m.manifest_hash)
64

property realization_count: int#

Return the total number of split realizations.

Returns:: Length of the combined train, validation, and test ID sets.
Return type:: int

Examples

>>> split = RealizationSplit(("a", "b"), ("c",), ())
>>> DatasetManifest("d", "g", "1", {}, split, 3).realization_count
3

with_artifact(path, record)#

Return a copy containing or replacing one artifact record.

Parameters:

path (str) – Portable relative artifact path.
record (ArtifactRecord, mapping, or str) – Integrity record, serialized record, or SHA-256 digest.

Returns:

New immutable manifest; the original is unchanged.

Return type:

DatasetManifest

Examples

>>> split = RealizationSplit(("r1",), (), ())
>>> m = DatasetManifest("d", "g", "1", {}, split, 1)
>>> updated = m.with_artifact("data/models.npz", "a" * 64)
>>> list(updated.artifacts)
['data/models.npz']

verify_artifacts(root, *, paths=None, raise_on_error=False)#

Verify recorded artifact sizes and SHA-256 digests on disk.

Parameters:

root (str or pathlib.Path) – Directory against which relative artifact paths are resolved.
paths (sequence of str, optional) – Subset of recorded paths. By default all artifacts are checked.
raise_on_error (bool, default=False) – Raise on the first missing, size-mismatched, or hash-mismatched artifact instead of returning False for it.

Returns:

Normalized artifact paths mapped to verification results.

Return type:

dict

Raises:

KeyError – If a requested path is not recorded.
ValueError – If root is not a directory or verification fails while raise_on_error is true.

Examples

>>> from tempfile import TemporaryDirectory
>>> split = RealizationSplit(("r1",), (), ())
>>> with TemporaryDirectory() as directory:
...     root = Path(directory)
...     file = root / "data.bin"
...     _ = file.write_bytes(b"data")
...     record = ArtifactRecord.from_file(file)
...     manifest = DatasetManifest(
...         "d", "g", "1", {}, split, 1, artifacts={"data.bin": record}
...     )
...     result = manifest.verify_artifacts(root)
>>> result
{'data.bin': True}

to_dict()#

Return the complete schema-2 JSON representation.

Returns:: Mutable JSON-compatible copy including the configuration digest.
Return type:: dict

Examples

>>> split = RealizationSplit(("r1",), (), ())
>>> m = DatasetManifest("d", "g", "1", {}, split, 1)
>>> m.to_dict()["schema_version"]
2

write_json(path, *, overwrite=True)#

Write a deterministic, human-readable manifest file.

Parameters:

path (str or pathlib.Path) – Destination JSON file.
overwrite (bool, default=True) – Permit replacement of an existing file.

Returns:

Destination path.

Return type:

pathlib.Path

Raises:

FileExistsError – If the destination exists and overwrite is false.

Examples

>>> from tempfile import TemporaryDirectory
>>> split = RealizationSplit(("r1",), (), ())
>>> m = DatasetManifest("d", "g", "1", {}, split, 1)
>>> with TemporaryDirectory() as directory:
...     path = m.write_json(Path(directory) / "manifest.json")
...     loaded = DatasetManifest.read_json(path)
>>> loaded.manifest_hash == m.manifest_hash
True

classmethod from_dict(data)#

Restore a manifest and verify its recorded configuration digest.

Parameters:: data (mapping) – Schema-1 or schema-2 serialized manifest.
Returns:: Validated schema-2 runtime object.
Return type:: DatasetManifest
Raises:: ValueError – If the schema is unsupported, required content is invalid, or the recorded configuration hash does not match its configuration.

Examples

>>> split = RealizationSplit(("r1",), (), ())
>>> original = DatasetManifest("d", "g", "1", {"seed": 2}, split, 1)
>>> restored = DatasetManifest.from_dict(original.to_dict())
>>> restored.configuration_hash == original.configuration_hash
True

classmethod read_json(path)#

Read and validate a UTF-8 JSON manifest.

Parameters:

path (str or pathlib.Path) – Existing manifest file.

Returns:

Validated immutable manifest.

Return type:

DatasetManifest

Raises:

OSError – If the file cannot be read.
json.JSONDecodeError – If its content is not valid JSON.
ValueError – If decoded content violates the manifest contract.

Examples

>>> from tempfile import TemporaryDirectory
>>> split = RealizationSplit(("r1",), (), ())
>>> source = DatasetManifest("d", "g", "1", {}, split, 1)
>>> with TemporaryDirectory() as directory:
...     path = source.write_json(Path(directory) / "manifest.json")
...     loaded = DatasetManifest.read_json(path)
>>> loaded.dataset_id
'd'

class pycsamt.ai.data.ArtifactRecord(sha256, size_bytes=None, media_type=None, role=None)#

Bases: object

Integrity metadata for one external dataset artifact.

Parameters:

sha256 (str) – Lowercase 64-character SHA-256 digest of the complete file.
size_bytes (int or None, optional) – Exact file size. When present it is checked before hashing.
media_type (str or None, optional) – MIME type such as "application/x-npz".
role (str or None, optional) – Human-readable role such as "responses" or "models".

Examples

>>> record = ArtifactRecord("0" * 64, size_bytes=1024, role="models")
>>> record.algorithm
'sha256'

sha256: str#

size_bytes: int | None = None#

media_type: str | None = None#

role: str | None = None#

property algorithm: str#

Return the checksum algorithm identifier.

Returns:: Always "sha256" for the current artifact schema.
Return type:: str

Examples

>>> ArtifactRecord("a" * 64).algorithm
'sha256'

classmethod from_file(path, *, media_type=None, role=None, chunk_size=1048576)#

Create an integrity record from an existing file.

Parameters:

path (str or pathlib.Path) – Existing regular file.
media_type (str, optional) – Optional descriptive metadata.
role (str, optional) – Optional descriptive metadata.
chunk_size (int, default=1048576) – Bytes read per hashing iteration.

Returns:

Digest and exact file size captured from disk.

Return type:

ArtifactRecord

Examples

>>> from tempfile import TemporaryDirectory
>>> with TemporaryDirectory() as directory:
...     path = Path(directory) / "models.npz"
...     _ = path.write_bytes(b"model-data")
...     record = ArtifactRecord.from_file(path, role="models")
>>> record.size_bytes
10

to_dict()#

Return a JSON-serializable artifact record.

Returns:: Versioned checksum, size, media type, and role fields.
Return type:: dict

Examples

>>> ArtifactRecord("f" * 64, size_bytes=2).to_dict()["size_bytes"]
2

classmethod from_dict(data)#

Restore a validated artifact record.

Parameters:: data (mapping) – Versioned state returned by to_dict().
Returns:: Immutable integrity record.
Return type:: ArtifactRecord
Raises:: ValueError – If the schema or checksum algorithm is unsupported.

Examples

>>> state = ArtifactRecord("1" * 64).to_dict()
>>> ArtifactRecord.from_dict(state).sha256 == "1" * 64
True

pycsamt.ai.data.canonical_hash(value)#

Return the SHA-256 digest of deterministic canonical JSON.

Parameters:: value (Any) – Finite JSON-serializable value. Mapping keys are converted to strings, mappings are sorted recursively, and insignificant whitespace is removed before hashing.
Returns:: Lowercase 64-character hexadecimal SHA-256 digest.
Return type:: str
Raises:: ValueError – If value contains NaN, infinity, bytes, arrays, or another object that cannot be represented safely as JSON.

Examples

Mapping insertion order does not affect the digest:

>>> canonical_hash({"a": 1, "b": 2}) == canonical_hash({"b": 2, "a": 1})
True
>>> len(canonical_hash({"frequencies_hz": [100.0, 10.0]}))
64

pycsamt.ai.data.sha256_file(path, *, chunk_size=1048576)#

Hash a file without loading the complete artifact into memory.

Parameters:

path (str or pathlib.Path) – Existing regular file to hash.
chunk_size (int, default=1048576) – Positive number of bytes read per iteration.

Returns:

Lowercase hexadecimal SHA-256 digest.

Return type:

str

Raises:

ValueError – If chunk_size is not a positive integer.
OSError – If the file cannot be opened or read.

Examples

>>> from tempfile import TemporaryDirectory
>>> with TemporaryDirectory() as directory:
...     path = Path(directory) / "artifact.bin"
...     _ = path.write_bytes(b"pycsamt")
...     digest = sha256_file(path)
>>> len(digest)
64

`pycsamt.ai.data.contracts`	Canonical, dependency-light data contracts for MT/AMT surveys.
`pycsamt.ai.data.normalization`	Mask-aware, train-fitted normalization for complex MT/AMT surveys.
`pycsamt.ai.data.splits`	Deterministic, lineage-aware geological-realization dataset splits.
`pycsamt.ai.data.manifest`	Immutable, versioned provenance manifests for generated EM datasets.