pycsamt.ai.domain_gap.simulator#

Reproducible corruption of canonical surveys for domain-gap training.

Every function in this module maps a validated SurveyData to a new, independently validated SurveyData. Corruptions are composed from a single integer seed so a training run can be reproduced exactly from its CorruptionConfig and seed alone.

Two families of corruption are distinguished:

Systematic (bias-like): apply_static_shift() and apply_galvanic_distortion() model near-surface, frequency-independent effects that multiply the true impedance by a real matrix. They change the signal, not its declared uncertainty, beyond the linear scaling of that uncertainty.
Random (noise-like): add_heteroscedastic_noise(), apply_error_floor(), apply_dropout(), and inject_outliers() model acquisition noise, missing data, and undetected bad readings. These update impedance_error and valid so that masks and errors stay consistent with the injected corruption, per the M3 acceptance gate.

apply_corruption_suite() composes all of the above in a fixed, documented order using seeds spawned from a single parent seed, and returns a CorruptionRecord describing exactly what was sampled.

Module Attributes

SEVERITY_PRESETS

Named corruption suites for M3's clean/in-distribution/severe/OOD split.

Functions

`add_heteroscedastic_noise`(survey, *[, ...])	Add complex, per-observation heteroscedastic Gaussian noise.
`apply_corruption_suite`(survey[, config, ...])	Apply the full, ordered M3 corruption pipeline from a single seed.
`apply_dropout`(survey, *[, station_rate, ...])	Mark stations, frequencies, or individual observations as missing.
`apply_error_floor`(survey, *, floor_fraction)	Clamp the declared error to a minimum fraction of `\|Z\|`.
`apply_galvanic_distortion`(survey, *[, ...])	Inject a per-station real Groom & Bailey-style distortion matrix.
`apply_static_shift`(survey, *, log10_sigma, rng)	Multiply every present component by a per-station real factor.
`inject_outliers`(survey, *[, rate, ...])	Perturb a random fraction of valid observations by a large factor.
`perturb_coordinates`(survey, *[, ...])	Add Gaussian noise to station coordinates and elevation.

Classes

`CorruptionConfig`([noise_level_range, ...])	Parameter ranges for one corruption pass over a `SurveyData`.
`CorruptionRecord`(config, seed[, sampled, ...])	Provenance of one applied `apply_corruption_suite()` call.

class pycsamt.ai.domain_gap.simulator.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)[source]

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()[source]

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)[source]

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()[source]

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.simulator.CorruptionRecord(config, seed, sampled=<factory>, severity=None)[source]

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()[source]

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.simulator.SEVERITY_PRESETS: Mapping[str, CorruptionConfig] = {'clean': 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), 'held_out_corruption': CorruptionConfig(noise_level_range=(0.1, 0.25), error_floor_fraction=0.05, static_shift_log10_sigma=0.3, distortion_gain_log10_sigma=0.1, distortion_twist_deg_sigma=25.0, distortion_shear_sigma=0.4, distortion_anisotropy_sigma=0.3, station_dropout_rate=0.15, frequency_dropout_rate=0.15, random_dropout_rate=0.1, outlier_rate=0.08, outlier_log10_shift_range=(1.0, 2.5), coordinate_sigma_m=15.0, elevation_sigma_m=5.0), 'in_distribution': CorruptionConfig(noise_level_range=(0.01, 0.03), error_floor_fraction=0.02, static_shift_log10_sigma=0.05, distortion_gain_log10_sigma=0.02, distortion_twist_deg_sigma=3.0, distortion_shear_sigma=0.05, distortion_anisotropy_sigma=0.0, station_dropout_rate=0.0, frequency_dropout_rate=0.0, random_dropout_rate=0.02, outlier_rate=0.0, outlier_log10_shift_range=(0.5, 1.5), coordinate_sigma_m=1.0, elevation_sigma_m=0.0), 'severe': CorruptionConfig(noise_level_range=(0.05, 0.15), error_floor_fraction=0.05, static_shift_log10_sigma=0.15, distortion_gain_log10_sigma=0.05, distortion_twist_deg_sigma=10.0, distortion_shear_sigma=0.15, distortion_anisotropy_sigma=0.0, station_dropout_rate=0.05, frequency_dropout_rate=0.05, random_dropout_rate=0.05, outlier_rate=0.02, outlier_log10_shift_range=(0.5, 1.5), coordinate_sigma_m=5.0, elevation_sigma_m=2.0)}: Named corruption suites for M3’s clean/in-distribution/severe/OOD split.

pycsamt.ai.domain_gap.simulator.add_heteroscedastic_noise(survey, *, level_range=(0.02, 0.05), rng)[source]

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.simulator.apply_error_floor(survey, *, floor_fraction)[source]

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.simulator.apply_static_shift(survey, *, log10_sigma, rng, return_info=False)[source]

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.simulator.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)[source]

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.simulator.apply_dropout(survey, *, station_rate=0.0, frequency_rate=0.0, random_rate=0.0, rng, return_info=False)[source]

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.simulator.inject_outliers(survey, *, rate=0.0, log10_shift_range=(0.5, 1.5), rng, return_info=False)[source]

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.simulator.perturb_coordinates(survey, *, coordinate_sigma_m=0.0, elevation_sigma_m=0.0, rng)[source]

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.simulator.apply_corruption_suite(survey, config=None, *, severity=None, seed)[source]

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'