2.25.3.5. pycsamt.ai.domain_gap

Survey auditing, corruption simulation, survey-fit estimation, and feature distribution comparisons for synthetic-to-field domain-gap analysis.

Domain-gap and noise simulation for AI-assisted EM inversion (M3).

This package turns a clean SurveyData into realistic training data by injecting heteroscedastic noise, dropout, static shift, galvanic distortion, coordinate perturbation, and outliers (simulator); by fitting plausible parameter ranges from a real survey’s own QC diagnostics, AMT, CSAMT, MT, or otherwise (survey_fit); and by comparing simulated and field feature distributions quantitatively (report).

class pycsamt.ai.domain_gap.StationExclusion(station, reason)

Bases: object

One station dropped before the canonical survey bridge.

Parameters:

• station (str) – Station identifier as reported by the raw site collection.

• reason (str) – Human-readable, non-empty exclusion reason.

Examples

>>> StationExclusion("18-099Z", "missing freq or z array").reason
'missing freq or z array'

station: str

reason: str

to_dict()

Return a JSON-serializable representation.

Returns:

station and reason fields.

Return type:

dict

Examples

>>> StationExclusion("A", "missing freq or z array").to_dict()
{'station': 'A', 'reason': 'missing freq or z array'}

class pycsamt.ai.domain_gap.FrequencyGridReport(matched, reference_station, n_frequencies_by_station, mismatched_stations)

Bases: object

Whether every included station shares one frequency grid.

Parameters:

• matched (bool) – True only when every included station’s frequency grid equals the reference station’s, within tolerance.

• reference_station (str or None) – Station whose grid every other station was compared against.

• n_frequencies_by_station (mapping) – Station name to its own frequency count, for every included station, regardless of whether it matched.

• mismatched_stations (tuple of str) – Included stations whose frequency grid differs from the reference. Empty when matched is True.

Examples

>>> report = FrequencyGridReport(True, "A", {"A": 10, "B": 10}, ())
>>> report.matched
True

matched: bool

reference_station: str | None

n_frequencies_by_station: Mapping[str, int]

mismatched_stations: tuple[str, ...]

to_dict()

Return a JSON-serializable representation.

Returns:

All fields as plain JSON-compatible values.

Return type:

dict

Examples

>>> FrequencyGridReport(True, "A", {"A": 1}, ()).to_dict()["matched"]
True

class pycsamt.ai.domain_gap.DimensionalitySummary(n_samples, frac_1d, frac_2d, frac_3d, strike_consensus_deg, strike_consensus_iqr_deg, stations_recommending_3d_review=())

Bases: object

Survey-wide aggregate of pre2d_inversion_assessment().

Parameters:

• n_samples (int) – Total station-period samples the per-station table was built from.

• frac_1d (float) – Sample-weighted fraction classified 1-D, 2-D, and 3-D. They sum to one when n_samples is positive.

• frac_2d (float) – Sample-weighted fraction classified 1-D, 2-D, and 3-D. They sum to one when n_samples is positive.

• frac_3d (float) – Sample-weighted fraction classified 1-D, 2-D, and 3-D. They sum to one when n_samples is positive.

• strike_consensus_deg (float or None) – Median across stations of each station’s consensus strike angle and its interquartile spread; None when no station produced a finite value.

• strike_consensus_iqr_deg (float or None) – Median across stations of each station’s consensus strike angle and its interquartile spread; None when no station produced a finite value.

• stations_recommending_3d_review (tuple of str) – Stations pre2d_inversion_assessment() flagged "review_3d_effects_before_2d".

Examples

>>> summary = DimensionalitySummary(10, 0.7, 0.2, 0.1, 12.0, 5.0, ())
>>> round(summary.frac_1d + summary.frac_2d + summary.frac_3d, 6)
1.0

n_samples: int

frac_1d: float

frac_2d: float

frac_3d: float

strike_consensus_deg: float | None

strike_consensus_iqr_deg: float | None

stations_recommending_3d_review: tuple[str, ...] = ()

to_dict()

Return a JSON-serializable representation.

Returns:

All fields as plain JSON-compatible values.

Return type:

dict

Examples

>>> DimensionalitySummary(1, 1.0, 0.0, 0.0, None, None).to_dict()[
...     "n_samples"
... ]
1

class pycsamt.ai.domain_gap.SurveyAuditReport(n_stations_input, excluded_stations, frequency_grid, coverage, frequency_range_hz, error_ratio_p05, error_ratio_p50, error_ratio_p95, station_spacing_m, elevation_coverage, crs_declared, dimensionality, static_shift_log10_sigma, distortion_gain_log10_sigma, distortion_twist_deg_sigma, distortion_shear_sigma, distortion_anisotropy_sigma, generated_utc, metadata=<factory>)

Bases: object

Complete M1 accounting of one survey’s data quality.

Parameters:

• n_stations_input (int) – Stations found by ensure_sites() before any exclusion.

• excluded_stations (tuple of StationExclusion) – Every station dropped before the canonical bridge, with a reason.

• frequency_grid (FrequencyGridReport) – Whether every included station shares one frequency grid.

• coverage (SurveyCoverage or None) – Impedance coverage from coverage(), or None when frequency_grid did not match (the canonical bridge cannot run without a shared grid).

• frequency_range_hz ((float, float) or None) – Minimum and maximum frequency, when available.

• error_ratio_p05 (float or None) – 5th/50th/95th percentile of declared impedance_error / |Z| over valid observations, when available.

• error_ratio_p50 (float or None) – 5th/50th/95th percentile of declared impedance_error / |Z| over valid observations, when available.

• error_ratio_p95 (float or None) – 5th/50th/95th percentile of declared impedance_error / |Z| over valid observations, when available.

• station_spacing_m (mapping or None) – min, median, and max consecutive station spacing in the survey’s stored order.

• elevation_coverage (float) – Fraction of included stations with a finite elevation.

• crs_declared (bool) – Whether a coordinate reference system was supplied explicitly. Station coordinates for MT/AMT sites are ordinarily projected locally from latitude/longitude, which by itself declares no formal CRS.

• dimensionality (DimensionalitySummary) – Aggregate dimensionality and strike indicators.

• static_shift_log10_sigma (float) – Empirical spreads from fit_distortion_priors_from_sites().

• distortion_gain_log10_sigma (float) – Empirical spreads from fit_distortion_priors_from_sites().

• distortion_twist_deg_sigma (float) – Empirical spreads from fit_distortion_priors_from_sites().

• distortion_shear_sigma (float) – Empirical spreads from fit_distortion_priors_from_sites().

• distortion_anisotropy_sigma (float) – Empirical spreads from fit_distortion_priors_from_sites().

• generated_utc (str) – Timezone-aware ISO-8601 timestamp of when the audit ran.

• metadata (Mapping[str, Any])

Examples

Reports are normally produced by audit_survey(), not built directly:

>>> report = SurveyAuditReport(
...     n_stations_input=1,
...     excluded_stations=(),
...     frequency_grid=FrequencyGridReport(True, "A", {"A": 1}, ()),
...     coverage=None,
...     frequency_range_hz=None,
...     error_ratio_p05=None,
...     error_ratio_p50=None,
...     error_ratio_p95=None,
...     station_spacing_m=None,
...     elevation_coverage=0.0,
...     crs_declared=False,
...     dimensionality=DimensionalitySummary(0, 0.0, 0.0, 0.0, None, None),
...     static_shift_log10_sigma=0.0,
...     distortion_gain_log10_sigma=0.0,
...     distortion_twist_deg_sigma=0.0,
...     distortion_shear_sigma=0.0,
...     distortion_anisotropy_sigma=0.0,
...     generated_utc="2026-01-01T00:00:00Z",
... )
>>> report.n_stations_included
1

n_stations_input: int

excluded_stations: tuple[StationExclusion, ...]

frequency_grid: FrequencyGridReport

coverage: SurveyCoverage | None

frequency_range_hz: tuple[float, float] | None

error_ratio_p05: float | None

error_ratio_p50: float | None

error_ratio_p95: float | None

station_spacing_m: Mapping[str, float] | None

elevation_coverage: float

crs_declared: bool

dimensionality: DimensionalitySummary

static_shift_log10_sigma: float

distortion_gain_log10_sigma: float

distortion_twist_deg_sigma: float

distortion_shear_sigma: float

distortion_anisotropy_sigma: float

generated_utc: str

metadata: Mapping[str, Any]

property n_stations_included: int

Return the number of stations that reached the canonical bridge.

Returns:

n_stations_input minus the number of exclusions.

Return type:

int

Examples

Constructed directly for illustration; see audit_survey() for the normal way to obtain a report.

>>> report = SurveyAuditReport(
...     2,
...     (StationExclusion("B", "missing freq or z array"),),
...     FrequencyGridReport(True, "A", {"A": 1}, ()),
...     None,
...     None,
...     None,
...     None,
...     None,
...     None,
...     0.0,
...     False,
...     DimensionalitySummary(0, 0.0, 0.0, 0.0, None, None),
...     0.0,
...     0.0,
...     0.0,
...     0.0,
...     0.0,
...     "2026-01-01T00:00:00Z",
... )
>>> report.n_stations_included
1

to_dict()

Return a complete JSON-serializable representation.

Returns:

Every field, with nested records converted to plain dicts.

Return type:

dict

Examples

>>> report = SurveyAuditReport(
...     1,
...     (),
...     FrequencyGridReport(True, "A", {"A": 1}, ()),
...     None,
...     None,
...     None,
...     None,
...     None,
...     None,
...     0.0,
...     False,
...     DimensionalitySummary(0, 0.0, 0.0, 0.0, None, None),
...     0.0,
...     0.0,
...     0.0,
...     0.0,
...     0.0,
...     "2026-01-01T00:00:00Z",
... )
>>> report.to_dict()["schema_version"]
1

write_json(path, *, overwrite=True)

Write a deterministic, human-readable audit report 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
>>> report = SurveyAuditReport(
...     1,
...     (),
...     FrequencyGridReport(True, "A", {"A": 1}, ()),
...     None,
...     None,
...     None,
...     None,
...     None,
...     None,
...     0.0,
...     False,
...     DimensionalitySummary(0, 0.0, 0.0, 0.0, None, None),
...     0.0,
...     0.0,
...     0.0,
...     0.0,
...     0.0,
...     "2026-01-01T00:00:00Z",
... )
>>> with TemporaryDirectory() as directory:
...     path = report.write_json(Path(directory) / "audit.json")
...     loaded = SurveyAuditReport.read_json(path)
>>> loaded.n_stations_input
1

classmethod from_dict(data)

Restore a report from its JSON representation.

Parameters:

data (mapping) – State previously returned by to_dict().

Returns:

Validated immutable report.

Return type:

SurveyAuditReport

Raises:

ValueError – If the schema version is unsupported.

Examples

>>> report = SurveyAuditReport(
...     1,
...     (),
...     FrequencyGridReport(True, "A", {"A": 1}, ()),
...     None,
...     None,
...     None,
...     None,
...     None,
...     None,
...     0.0,
...     False,
...     DimensionalitySummary(0, 0.0, 0.0, 0.0, None, None),
...     0.0,
...     0.0,
...     0.0,
...     0.0,
...     0.0,
...     "2026-01-01T00:00:00Z",
... )
>>> SurveyAuditReport.from_dict(report.to_dict()) == report
True

classmethod read_json(path)

Read and validate a UTF-8 JSON audit report.

Parameters:

path (str or pathlib.Path) – Existing report file written by write_json().

Returns:

Validated immutable report.

Return type:

SurveyAuditReport

Examples

See write_json() for a complete round trip.

summary()

Return a compact, human-readable multi-line report.

Returns:

Plain-text accounting of inclusion, coverage, geometry, dimensionality, and distortion indicators.

Return type:

str

Examples

>>> report = SurveyAuditReport(
...     1,
...     (),
...     FrequencyGridReport(True, "A", {"A": 1}, ()),
...     None,
...     None,
...     None,
...     None,
...     None,
...     None,
...     0.0,
...     False,
...     DimensionalitySummary(0, 0.0, 0.0, 0.0, None, None),
...     0.0,
...     0.0,
...     0.0,
...     0.0,
...     0.0,
...     "2026-01-01T00:00:00Z",
... )
>>> print(report.summary())
Survey audit (generated 2026-01-01T00:00:00Z)
  Stations: 1 input, 1 included, 0 excluded
  Frequency grid: matched
  CRS declared: False
  Elevation coverage: 0.0%
  Dimensionality: n=0, 1D=0.0%, 2D=0.0%, 3D=0.0%
  Static shift log10 sigma: 0.0000
  Distortion sigma: gain(log10)=0.0000, twist_deg=0.00, shear=0.0000, anisotropy=0.0000

pycsamt.ai.domain_gap.audit_survey(sites, *, recursive=True, on_dup='replace', verbose=0, freq_rtol=1e-06, band=None, skew_th=3.0, ellipt_th=0.2, station_spacing_fallback=500.0, metadata=None)

Audit a raw survey and account for every included/excluded station.

Unlike survey_data_from_sites(), which deliberately raises on a mismatched frequency grid so training can never proceed silently on inconsistent data, this function never raises for that reason: it is meant to be run before deciding whether a survey is fit for that stricter bridge, and reports a mismatch as a finding rather than an exception.

Parameters:

• sites (Any) – Anything accepted by pycsamt.emtools._core.ensure_sites().

• recursive (bool) – Forwarded to ensure_sites and the underlying diagnostics.

• on_dup (str) – Forwarded to ensure_sites and the underlying diagnostics.

• verbose (int) – Forwarded to ensure_sites and the underlying diagnostics.

• freq_rtol (float, default=1e-6) – Relative tolerance used when comparing each station’s frequency grid against the reference station’s.

• band ((float, float), optional) – Period band in seconds forwarded to pre2d_inversion_assessment().

• skew_th (float, optional) – Phase-tensor skew and ellipticity thresholds forwarded to the same dimensionality assessment.

• ellipt_th (float, optional) – Phase-tensor skew and ellipticity thresholds forwarded to the same dimensionality assessment.

• station_spacing_fallback (float, default=500.0) – Uniform-grid spacing in metres used only when no station reports usable latitude/longitude.

• metadata (mapping, optional) – Extra provenance recorded on the returned report.

Returns:

Complete accounting of inclusion, coverage, geometry, dimensionality, and distortion indicators.

Return type:

SurveyAuditReport

Raises:

ValueError – If no station in sites has usable impedance data at all.

Examples

>>> report = audit_survey(
...     "data/AMT/WILLY_data/L18PLT", recursive=True, verbose=0
... )
>>> report.frequency_grid.matched
True

class pycsamt.ai.domain_gap.EmpiricalCorruptionResult(survey, seed, field_station_indices, static_log10_resistivity_factor, relative_error_fraction, noise_realization, missing_mask, measurement_reliability, dimensionality_reliability, observation_reliability)

Bases: object

One corrupted survey and every sampled latent array.

Parameters:

• survey (SurveyData) – Corrupted survey with declared empirical impedance errors.

• seed (int) – Parent random seed.

• field_station_indices (ndarray of int) – Empirical field station selected for each synthetic station.

• static_log10_resistivity_factor (ndarray) – Sampled log10(rho_observed / rho_smooth).

• relative_error_fraction (ndarray) – Sampled impedance-error fractions by component.

• noise_realization (ndarray of complex) – Additive complex noise in impedance units.

• missing_mask (ndarray of bool) – True where an observation was removed.

• measurement_reliability (ndarray) – Separate and combined reliability factors.

• dimensionality_reliability (ndarray) – Separate and combined reliability factors.

• observation_reliability (ndarray) – Separate and combined reliability factors.

survey: SurveyData

seed: int

field_station_indices: ndarray

static_log10_resistivity_factor: ndarray

relative_error_fraction: ndarray

noise_realization: ndarray

missing_mask: ndarray

measurement_reliability: ndarray

dimensionality_reliability: ndarray

observation_reliability: ndarray

property record_hash: str

Return a deterministic hash of seed and sampled summaries.

Returns:

SHA-256 digest of JSON-compatible provenance.

Return type:

str

to_dict()

Return compact JSON-compatible corruption provenance.

Returns:

Seed, selected field stations, and sampled ranges.

Return type:

dict

pycsamt.ai.domain_gap.apply_empirical_corruption(survey, *, field_frequencies_hz, static_log10_resistivity_profiles, measurement_reliability_profiles, dimensionality_reliability_profiles, observation_reliability_profiles, relative_error_quantiles, seed, missing_rate_by_component=None)

Apply jointly sampled field profiles and empirical error quantiles.

Parameters:

• survey (SurveyData) – Clean synthetic survey.

• field_frequencies_hz (array-like) – Positive unique frequencies of the empirical field profiles.

• static_log10_resistivity_profiles (array-like) – Field log10(rho_observed / rho_smooth) profiles shaped (n_field_station, n_field_frequency).

• measurement_reliability_profiles (array-like) – Aligned empirical reliability profiles in [0, 1].

• dimensionality_reliability_profiles (array-like) – Aligned empirical reliability profiles in [0, 1].

• observation_reliability_profiles (array-like) – Aligned empirical reliability profiles in [0, 1].

• relative_error_quantiles (mapping) – Component names mapped to {"levels": ..., "values": ...} monotone empirical quantile curves.

• seed (int) – Parent random seed.

• missing_rate_by_component (mapping or None, optional) – Empirical independent missing probabilities. Missing observations are set to complex NaN and marked invalid.

Returns:

Corrupted survey and complete sampled provenance.

Return type:

EmpiricalCorruptionResult

Notes

Static shift is supplied in apparent-resistivity space and therefore applied to impedance as 10**(0.5 * log10_rho_factor).

Examples

>>> clean = SurveyData(
...     np.ones((2, 2, 1), complex),
...     [10.0, 1.0],
...     ["A", "B"],
...     ["zxy"],
...     [[0, 0], [1, 0]],
... )
>>> result = apply_empirical_corruption(
...     clean,
...     field_frequencies_hz=[10.0, 1.0],
...     static_log10_resistivity_profiles=[[0.0, 0.0]],
...     measurement_reliability_profiles=[[0.8, 0.7]],
...     dimensionality_reliability_profiles=[[1.0, 0.5]],
...     observation_reliability_profiles=[[0.8, 0.35]],
...     relative_error_quantiles={
...         "zxy": {"levels": [0, 1], "values": [0.01, 0.05]}
...     },
...     seed=0,
... )
>>> result.survey.shape
(2, 2, 1)

class pycsamt.ai.domain_gap.CorruptionConfig(noise_level_range=(0.0, 0.0), error_floor_fraction=0.0, static_shift_log10_sigma=0.0, distortion_gain_log10_sigma=0.0, distortion_twist_deg_sigma=0.0, distortion_shear_sigma=0.0, distortion_anisotropy_sigma=0.0, station_dropout_rate=0.0, frequency_dropout_rate=0.0, random_dropout_rate=0.0, outlier_rate=0.0, outlier_log10_shift_range=(0.5, 1.5), coordinate_sigma_m=0.0, elevation_sigma_m=0.0)

Bases: object

Parameter ranges for one corruption pass over a SurveyData.

All defaults are zero/no-op so CorruptionConfig() is the clean synthetic control set required by the M3 gate.

Parameters:

• noise_level_range ((float, float), default=(0.0, 0.0)) – Bounds on the relative heteroscedastic noise standard deviation sampled independently per station/frequency observation.

• error_floor_fraction (float, default=0.0) – Minimum declared impedance_error as a fraction of |Z|.

• static_shift_log10_sigma (float, default=0.0) – Std. dev. of the log₁₀ per-station static-shift factor applied identically across all frequencies.

• distortion_gain_log10_sigma (float, default=0.0) – Std. dev. of the per-station Groom-Bailey-style gain, twist, shear, and anisotropy parameters of the injected galvanic distortion.

• distortion_twist_deg_sigma (float, default=0.0) – Std. dev. of the per-station Groom-Bailey-style gain, twist, shear, and anisotropy parameters of the injected galvanic distortion.

• distortion_shear_sigma (float, default=0.0) – Std. dev. of the per-station Groom-Bailey-style gain, twist, shear, and anisotropy parameters of the injected galvanic distortion.

• distortion_anisotropy_sigma (float, default=0.0) – Std. dev. of the per-station Groom-Bailey-style gain, twist, shear, and anisotropy parameters of the injected galvanic distortion.

• station_dropout_rate (float, default=0.0) – Probability that an entire station, an entire frequency (across all stations), or an individual observation is marked missing.

• frequency_dropout_rate (float, default=0.0) – Probability that an entire station, an entire frequency (across all stations), or an individual observation is marked missing.

• random_dropout_rate (float, default=0.0) – Probability that an entire station, an entire frequency (across all stations), or an individual observation is marked missing.

• outlier_rate (float, default=0.0) – Fraction of remaining valid observations perturbed by a large, undetected multiplicative shift.

• outlier_log10_shift_range ((float, float), default=(0.5, 1.5)) – Bounds on the magnitude (in log₁₀ decades) of injected outliers; the sign is randomized.

• coordinate_sigma_m (float, default=0.0) – Std. dev. of Gaussian perturbation applied to station horizontal coordinates and elevation, respectively.

• elevation_sigma_m (float, default=0.0) – Std. dev. of Gaussian perturbation applied to station horizontal coordinates and elevation, respectively.

Examples

>>> config = CorruptionConfig(noise_level_range=(0.01, 0.05))
>>> config.config_hash() == CorruptionConfig(
...     noise_level_range=(0.01, 0.05)
... ).config_hash()
True

noise_level_range: tuple[float, float] = (0.0, 0.0)

error_floor_fraction: float = 0.0

static_shift_log10_sigma: float = 0.0

distortion_gain_log10_sigma: float = 0.0

distortion_twist_deg_sigma: float = 0.0

distortion_shear_sigma: float = 0.0

distortion_anisotropy_sigma: float = 0.0

station_dropout_rate: float = 0.0

frequency_dropout_rate: float = 0.0

random_dropout_rate: float = 0.0

outlier_rate: float = 0.0

outlier_log10_shift_range: tuple[float, float] = (0.5, 1.5)

coordinate_sigma_m: float = 0.0

elevation_sigma_m: float = 0.0

to_dict()

Return a JSON-serializable, order-stable representation.

Returns:

Field values with a schema discriminator.

Return type:

dict

Examples

>>> CorruptionConfig().to_dict()["schema_version"]
1

classmethod from_dict(data)

Restore and validate a serialized configuration.

Parameters:

data (mapping) – State previously returned by to_dict().

Returns:

Validated immutable configuration.

Return type:

CorruptionConfig

Examples

>>> state = CorruptionConfig(error_floor_fraction=0.02).to_dict()
>>> CorruptionConfig.from_dict(state).error_floor_fraction
0.02

config_hash()

Return the SHA-256 digest of this configuration’s canonical JSON.

Returns:

Lowercase 64-character hexadecimal digest.

Return type:

str

Examples

>>> len(CorruptionConfig().config_hash())
64

class pycsamt.ai.domain_gap.CorruptionRecord(config, seed, sampled=<factory>, severity=None)

Bases: object

Provenance of one applied apply_corruption_suite() call.

Parameters:

• config (CorruptionConfig) – Configuration the sampled parameters were drawn from.

• seed (int) – Parent seed used to spawn every corruption step’s generator.

• severity (str or None) – Name of the severity preset used, when applicable.

• sampled (mapping) – Concrete, JSON-serializable per-step summary statistics (e.g. the number of dropped stations, or the mean sampled distortion gain).

config: CorruptionConfig

seed: int

sampled: Mapping[str, Any]

severity: str | None = None

to_dict()

Return a JSON-serializable provenance record.

Returns:

Schema-versioned config, seed, severity, and sampled summary.

Return type:

dict

Examples

>>> record = CorruptionRecord(CorruptionConfig(), seed=0)
>>> record.to_dict()["schema_version"]
1

pycsamt.ai.domain_gap.add_heteroscedastic_noise(survey, *, level_range=(0.02, 0.05), rng)

Add complex, per-observation heteroscedastic Gaussian noise.

Parameters:

• survey (SurveyData) – Clean or already-corrupted survey.

• level_range ((float, float), default=(0.02, 0.05)) – Bounds on the relative noise standard deviation sampled independently for every (station, frequency) pair and shared across components at that pair, mimicking correlated instrument noise.

• rng (numpy.random.Generator) – Source of randomness; callers control reproducibility.

Returns:

New survey with perturbed impedance and an impedance_error that combines any pre-existing error with the injected noise in quadrature.

Return type:

SurveyData

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.full((1, 2, 1), 100 + 0j)
>>> survey = SurveyData(z, [10.0, 1.0], ["S"], ["xy"], [[0.0, 0.0]])
>>> noisy = add_heteroscedastic_noise(
...     survey, level_range=(0.05, 0.05), rng=np.random.default_rng(0)
... )
>>> noisy.impedance_error is not None
True

pycsamt.ai.domain_gap.apply_corruption_suite(survey, config=None, *, severity=None, seed)

Apply the full, ordered M3 corruption pipeline from a single seed.

Steps run in this fixed order: static shift, galvanic distortion, heteroscedastic noise, error floor, dropout, outliers, coordinate perturbation. Systematic distortions are applied to the clean signal before random noise and missingness, matching how these effects compose physically.

Parameters:

• survey (SurveyData) – Clean survey to corrupt.

• config (CorruptionConfig, optional) – Explicit configuration. Mutually exclusive with severity.

• severity (str, optional) – Name of an entry in SEVERITY_PRESETS to use as config.

• seed (int) – Parent seed. Each step draws from an independently spawned child generator so adding a new step does not change earlier steps’ draws.

Returns:

• survey (SurveyData) – Corrupted survey.

• record (CorruptionRecord) – Provenance of the applied configuration and sampled parameters.

Raises:

ValueError – If both or neither of config/severity are given, or severity is unknown.

Return type:

tuple[SurveyData, CorruptionRecord]

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.full((3, 5, 2), 100 + 50j)
>>> survey = SurveyData(
...     z,
...     np.linspace(100, 1, 5),
...     ["A", "B", "C"],
...     ["xy", "yx"],
...     np.zeros((3, 2)),
... )
>>> corrupted, record = apply_corruption_suite(
...     survey, severity="in_distribution", seed=0
... )
>>> record.severity
'in_distribution'

pycsamt.ai.domain_gap.apply_dropout(survey, *, station_rate=0.0, frequency_rate=0.0, random_rate=0.0, rng, return_info=False)

Mark stations, frequencies, or individual observations as missing.

Dropped observations are invalidated by setting the impedance to NaN; SurveyData construction then recomputes valid from finiteness, so masks stay authoritative automatically.

Parameters:

• survey (SurveyData) – Survey to thin out.

• station_rate (float, default=0.0) – Independent probabilities that a whole station (all frequencies and components), a whole frequency (all stations and components), or an individual observation is dropped. Effects are combined (a station or frequency dropout wins over a random one at the same cell).

• frequency_rate (float, default=0.0) – Independent probabilities that a whole station (all frequencies and components), a whole frequency (all stations and components), or an individual observation is dropped. Effects are combined (a station or frequency dropout wins over a random one at the same cell).

• random_rate (float, default=0.0) – Independent probabilities that a whole station (all frequencies and components), a whole frequency (all stations and components), or an individual observation is dropped. Effects are combined (a station or frequency dropout wins over a random one at the same cell).

• rng (numpy.random.Generator) – Source of randomness.

• return_info (bool)

Returns:

New survey with additional invalid observations.

Return type:

SurveyData

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.ones((4, 4, 1), dtype=complex)
>>> survey = SurveyData(
...     z,
...     [4.0, 3.0, 2.0, 1.0],
...     ["A", "B", "C", "D"],
...     ["xy"],
...     np.zeros((4, 2)),
... )
>>> thinned = apply_dropout(
...     survey, station_rate=1.0, rng=np.random.default_rng(0)
... )
>>> thinned.n_valid
0

pycsamt.ai.domain_gap.apply_error_floor(survey, *, floor_fraction)

Clamp the declared error to a minimum fraction of |Z|.

Parameters:

• survey (SurveyData) – Survey whose error floor should be enforced.

• floor_fraction (float) – Minimum impedance_error as a fraction of |Z|. Zero is a no-op.

Returns:

New survey with an error array at least as large as floor_fraction * |Z| on valid observations.

Return type:

SurveyData

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.full((1, 1, 1), 100 + 0j)
>>> survey = SurveyData(
...     z,
...     [1.0],
...     ["S"],
...     ["xy"],
...     [[0.0, 0.0]],
...     impedance_error=np.ones((1, 1, 1)),
... )
>>> floored = apply_error_floor(survey, floor_fraction=0.5)
>>> floored.impedance_error[0, 0, 0]
50.0

pycsamt.ai.domain_gap.apply_galvanic_distortion(survey, *, gain_log10_sigma=0.0, twist_deg_sigma=0.0, shear_sigma=0.0, anisotropy_sigma=0.0, rng, return_info=False)

Inject a per-station real Groom & Bailey-style distortion matrix.

Parameters:

• survey (SurveyData) – Survey to distort. Must expose at least one impedance component; any of xx, xy, yx, yy absent from SurveyData.components is treated as zero for the purpose of building the dense 2x2 impedance used internally, which is an approximation for surveys that only store off-diagonal components.

• gain_log10_sigma (float) – Std. dev. of the per-station gain (log₁₀), twist (degrees), shear, and anisotropy parameters. Zero for every parameter is a no-op.

• twist_deg_sigma (float) – Std. dev. of the per-station gain (log₁₀), twist (degrees), shear, and anisotropy parameters. Zero for every parameter is a no-op.

• shear_sigma (float) – Std. dev. of the per-station gain (log₁₀), twist (degrees), shear, and anisotropy parameters. Zero for every parameter is a no-op.

• anisotropy_sigma (float) – Std. dev. of the per-station gain (log₁₀), twist (degrees), shear, and anisotropy parameters. Zero for every parameter is a no-op.

• rng (numpy.random.Generator) – Source of randomness.

• return_info (bool)

Returns:

New survey with distorted impedance; declared error, if any, is scaled by the sampled gain as a first-order approximation.

Return type:

SurveyData

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.ones((1, 1, 2), dtype=complex) * (1 + 1j)
>>> survey = SurveyData(z, [1.0], ["S"], ["xy", "yx"], [[0, 0]])
>>> distorted = apply_galvanic_distortion(
...     survey, twist_deg_sigma=10.0, rng=np.random.default_rng(0)
... )
>>> distorted.shape == survey.shape
True

pycsamt.ai.domain_gap.apply_static_shift(survey, *, log10_sigma, rng, return_info=False)

Multiply every present component by a per-station real factor.

Static shift is modelled as a frequency-independent real scalar c_s = 10 ** N(0, log10_sigma) per station, applied identically to every impedance component and to the declared error, consistent with the linear scaling of a real multiplicative distortion.

Parameters:

• survey (SurveyData) – Survey to distort.

• log10_sigma (float) – Std. dev. of the log₁₀ static-shift factor. Zero is a no-op.

• rng (numpy.random.Generator) – Source of randomness.

• return_info (bool)

Returns:

New survey with static shift applied.

Return type:

SurveyData

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.full((2, 1, 1), 100 + 0j)
>>> survey = SurveyData(z, [1.0], ["A", "B"], ["xy"], [[0, 0], [1, 0]])
>>> shifted = apply_static_shift(
...     survey, log10_sigma=0.1, rng=np.random.default_rng(0)
... )
>>> shifted.shape == survey.shape
True

pycsamt.ai.domain_gap.inject_outliers(survey, *, rate=0.0, log10_shift_range=(0.5, 1.5), rng, return_info=False)

Perturb a random fraction of valid observations by a large factor.

Outliers remain marked valid and keep their existing declared error, simulating a bad reading that quality control failed to flag — the case a robust inverter must tolerate.

Parameters:

• survey (SurveyData) – Survey to perturb.

• rate (float, default=0.0) – Fraction of currently valid observations perturbed. Zero is a no-op.

• log10_shift_range ((float, float), default=(0.5, 1.5)) – Bounds on the outlier magnitude in log₁₀ decades; the sign is randomized per outlier.

• rng (numpy.random.Generator) – Source of randomness.

• return_info (bool)

Returns:

New survey with a subset of valid impedance values shifted by 10 ** (+/- shift).

Return type:

SurveyData

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.full((1, 10, 1), 100 + 0j)
>>> survey = SurveyData(
...     z, np.arange(10.0, 0.0, -1.0), ["S"], ["xy"], [[0, 0]]
... )
>>> corrupted = inject_outliers(
...     survey, rate=0.5, rng=np.random.default_rng(0)
... )
>>> corrupted.shape == survey.shape
True

pycsamt.ai.domain_gap.perturb_coordinates(survey, *, coordinate_sigma_m=0.0, elevation_sigma_m=0.0, rng)

Add Gaussian noise to station coordinates and elevation.

Parameters:

• survey (SurveyData) – Survey whose station geometry should be perturbed.

• coordinate_sigma_m (float, default=0.0) – Std. dev. of Gaussian noise added to the horizontal (x, y) coordinates and to elevation, respectively. Elevation entries that are NaN (unknown) stay NaN.

• elevation_sigma_m (float, default=0.0) – Std. dev. of Gaussian noise added to the horizontal (x, y) coordinates and to elevation, respectively. Elevation entries that are NaN (unknown) stay NaN.

• rng (numpy.random.Generator) – Source of randomness.

Returns:

New survey with perturbed SurveyData.coordinates_m.

Return type:

SurveyData

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.ones((1, 1, 1), dtype=complex)
>>> survey = SurveyData(z, [1.0], ["S"], ["xy"], [[0.0, 0.0, 100.0]])
>>> moved = perturb_coordinates(
...     survey, coordinate_sigma_m=5.0, rng=np.random.default_rng(0)
... )
>>> moved.coordinates_m.shape
(1, 3)

pycsamt.ai.domain_gap.fit_corruption_config(survey, *, severity_scale=1.0)

Derive plausible noise/dropout ranges from a real survey’s QC.

Only quantities already present in the canonical SurveyData contract are used: the impedance_error-to-|Z| ratio for heteroscedastic noise and error floor, and coverage() for dropout rates.

Parameters:

• survey (SurveyData) – Real (or realistically corrupted) survey to profile.

• severity_scale (float, default=1.0) – Multiplier applied to every fitted range/rate, letting a caller derive a milder or harsher preset from the same empirical fit.

Returns:

Configuration whose noise range spans the interquartile range of the observed relative error, whose error floor is the fifth percentile of that ratio, and whose dropout rates equal the observed missing fractions. Distortion and outlier parameters are left at zero; see fit_distortion_priors_from_sites() for those.

Return type:

CorruptionConfig

Raises:

ValueError – If survey has no declared impedance_error to profile.

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.full((4, 6, 2), 100 + 50j)
>>> err = np.full((4, 6, 2), 3.0)
>>> survey = SurveyData(
...     z,
...     np.linspace(1000, 1, 6),
...     ["A", "B", "C", "D"],
...     ["xy", "yx"],
...     np.zeros((4, 2)),
...     impedance_error=err,
... )
>>> config = fit_corruption_config(survey)
>>> config.noise_level_range[0] >= 0.0
True

pycsamt.ai.domain_gap.fit_distortion_priors_from_sites(sites, *, recursive=True, on_dup='replace', verbose=0, **kwargs)

Estimate empirical static-shift and distortion spreads from real EDI.

This is the one M3 entry point that genuinely depends on the heavier, pandas-based EM diagnostics in pycsamt.emtools.gb and pycsamt.emtools.ss, run directly on real sites (e.g. a WILLY line) rather than on the numpy-only SurveyData contract.

Parameters:

• sites (Any) – Anything accepted by pycsamt.emtools._core.ensure_sites().

• recursive (bool) – Forwarded to the underlying diagnostics.

• on_dup (str) – Forwarded to the underlying diagnostics.

• verbose (int) – Forwarded to the underlying diagnostics.

• **kwargs (Any) – Forwarded to pycsamt.emtools.gb.groom_bailey_table().

Returns:

static_shift_log10_sigma, distortion_gain_log10_sigma, distortion_twist_deg_sigma, distortion_shear_sigma, and distortion_anisotropy_sigma, each the population standard deviation of the corresponding per-station fitted parameter across stations with a successful fit. A parameter is 0.0 when fewer than two stations produced a usable fit.

Return type:

dict

Examples

>>> priors = fit_distortion_priors_from_sites(
...     "data/AMT/WILLY_DATA/L18PLT", recursive=False, verbose=0
... )
>>> sorted(priors)
['distortion_anisotropy_sigma', 'distortion_gain_log10_sigma', 'distortion_shear_sigma', 'distortion_twist_deg_sigma', 'static_shift_log10_sigma']

pycsamt.ai.domain_gap.survey_data_from_sites(sites, *, crs=None, freq_rtol=1e-06, station_spacing=500.0, recursive=True, on_dup='replace', verbose=0, metadata=None)

Bridge EDI/Sites/APISurvey input to canonical SurveyData.

Parameters:

• sites (Any) – Anything accepted by pycsamt.emtools._core.ensure_sites(): a filesystem path/glob/directory, EDIFile/EDICollection, Site/Sites, APISurvey, or an iterable of these.

• crs (str, optional) – Coordinate reference system identifier to record. Station positions are always projected with a local equirectangular approximation (see pycsamt.ai.inversion._sites_bridge.sites_to_coords_3d()); pass a CRS string only if it genuinely describes that projection.

• freq_rtol (float, default=1e-6) – Relative tolerance used when checking that every station shares the same frequency grid.

• station_spacing (float, default=500.0) – Forwarded to the coordinate bridge as a uniform-grid fallback spacing, used only when no station reports finite coordinates.

• recursive (bool) – Forwarded to ensure_sites.

• on_dup (str) – Forwarded to ensure_sites.

• verbose (int) – Forwarded to ensure_sites.

• metadata (dict, optional) – Extra provenance recorded on the returned survey.

Returns:

Canonical survey with full xx, xy, yx, yy components, with impedance and its declared error converted from Site’s EDI-native [mV/km]/[nT] convention to SI (V/A), matching SurveyData’s default ImpedanceConvention.

Return type:

SurveyData

Raises:

ValueError – If no station has usable impedance data, or stations do not share a common frequency grid within freq_rtol.

Notes

This function performs no frequency interpolation: a survey whose stations were sampled on different frequency grids must be resolved by an explicit, survey-matched frequency selector (an M1 concern) before reaching this bridge.

Examples

>>> survey = survey_data_from_sites(
...     "data/AMT/WILLY_DATA/L18PLT", recursive=False, verbose=0
... )
>>> survey.components
('xx', 'xy', 'yx', 'yy')

class pycsamt.ai.domain_gap.DistributionComparisonReport(comparisons)

Bases: object

Collection of FeatureComparison results across features.

Parameters:

comparisons (mapping) – Feature name to FeatureComparison.

comparisons: Mapping[str, FeatureComparison]

worst_feature()

Return the feature name with the largest KS statistic.

Returns:

Feature name whose simulated/field distributions differ most, ignoring features with a NaN statistic (empty samples).

Return type:

str

Raises:

ValueError – If every feature has a NaN KS statistic.

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.full((2, 4, 1), 100 + 50j)
>>> survey = SurveyData(
...     z, np.linspace(100, 1, 4), ["A", "B"], ["xy"], np.zeros((2, 2))
... )
>>> report = compare_survey_distributions(survey, survey)
>>> report.worst_feature() in report.comparisons
True

to_dict()

Return a JSON-serializable representation.

Returns:

Mapping of feature name to its comparison dict.

Return type:

dict

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.full((2, 4, 1), 100 + 50j)
>>> survey = SurveyData(
...     z, np.linspace(100, 1, 4), ["A", "B"], ["xy"], np.zeros((2, 2))
... )
>>> report = compare_survey_distributions(survey, survey)
>>> sorted(report.to_dict())
['error_to_magnitude_ratio', 'log_impedance_magnitude', 'phase_deg']

class pycsamt.ai.domain_gap.FeatureComparison(feature, simulated_stats, field_stats, ks_statistic, ks_pvalue, mean_difference, std_ratio)

Bases: object

Quantitative comparison of one feature between two distributions.

Parameters:

• feature (str) – Name of the compared feature.

• simulated_stats (mapping) – count, mean, std, median of each sample.

• field_stats (mapping) – count, mean, std, median of each sample.

• ks_statistic (float) – Two-sample Kolmogorov-Smirnov statistic and p-value; NaN when either sample is empty.

• ks_pvalue (float) – Two-sample Kolmogorov-Smirnov statistic and p-value; NaN when either sample is empty.

• mean_difference (float) – simulated mean - field mean.

• std_ratio (float) – simulated std / field std; NaN when the field std is zero.

feature: str

simulated_stats: Mapping[str, float]

field_stats: Mapping[str, float]

ks_statistic: float

ks_pvalue: float

mean_difference: float

std_ratio: float

to_dict()

Return a JSON-serializable representation.

Returns:

All fields, with statistics mappings converted to plain dicts.

Return type:

dict

Examples

>>> import numpy as np
>>> comparison = compare_feature_distributions(
...     np.array([1.0, 2.0, 3.0]),
...     np.array([1.0, 2.0, 3.0]),
...     feature="custom",
... )
>>> comparison.to_dict()["feature"]
'custom'

pycsamt.ai.domain_gap.compare_feature_distributions(simulated, field, *, feature)

Compare two 1-D samples of the same feature quantitatively.

Parameters:

• simulated (array-like) – Feature values already extracted from each survey (see compare_survey_distributions() to extract them from SurveyData directly).

• field (array-like) – Feature values already extracted from each survey (see compare_survey_distributions() to extract them from SurveyData directly).

• feature (str) – Label recorded on the returned FeatureComparison.

Returns:

Summary statistics, mean difference, std ratio, and two-sample KS test between the samples.

Return type:

FeatureComparison

Examples

>>> import numpy as np
>>> rng = np.random.default_rng(0)
>>> comparison = compare_feature_distributions(
...     rng.normal(0, 1, 200), rng.normal(0, 1, 200), feature="demo"
... )
>>> comparison.ks_pvalue > 0.01
True

pycsamt.ai.domain_gap.compare_survey_distributions(simulated, field, *, features=('log_impedance_magnitude', 'phase_deg', 'error_to_magnitude_ratio'))

Compare simulated and field surveys across several canonical features.

Parameters:

• simulated (SurveyData) – Simulated (e.g. corrupted synthetic) survey.

• field (SurveyData) – Real field survey, ideally sharing the simulated survey’s frequency band and component set for a meaningful comparison.

• features (sequence of str, default features) – Any of "log_impedance_magnitude", "phase_deg", or "error_to_magnitude_ratio". The error-ratio feature is skipped (empty sample) for a survey without declared errors.

Returns:

One FeatureComparison per requested feature.

Return type:

DistributionComparisonReport

Examples

>>> import numpy as np
>>> from pycsamt.ai.data.contracts import SurveyData
>>> z = np.full((2, 4, 1), 100 + 50j)
>>> survey = SurveyData(
...     z, np.linspace(100, 1, 4), ["A", "B"], ["xy"], np.zeros((2, 2))
... )
>>> report = compare_survey_distributions(survey, survey)
>>> report.comparisons["phase_deg"].mean_difference
0.0

pycsamt.ai.domain_gap.audit	Survey-level audit reports for the M1 data-contract gate.
pycsamt.ai.domain_gap.simulator	Reproducible corruption of canonical surveys for domain-gap training.
pycsamt.ai.domain_gap.survey_fit	Fit plausible CorruptionConfig ranges from real data.
pycsamt.ai.domain_gap.report	Quantitative comparison between simulated and field feature distributions.
pycsamt.ai.domain_gap.empirical	Empirical field-calibrated corruption for synthetic EM surveys.