pycsamt.ai.data.normalization

Mask-aware, train-fitted normalization for complex MT/AMT surveys.

Normalization state is a scientific artifact. It is fitted only from explicitly supplied surveys, records the frequency/component axes and complex impedance convention, and is then reused unchanged for validation, test, and field data. Invalid observations remain identifiable through explicit masks.

Classes

ComplexZScore(mean, scale, frequencies_hz, ...)	Immutable per-frequency/component complex z-score state.
NormalizedSurvey(values, valid, ...[, ...])	Normalized real/imaginary feature channels and their validity state.

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