pycsamt.ai.domain_gap.audit#

Survey-level audit reports for the M1 data-contract gate.

The M1 milestone requires that “an audit report accounts for every included/excluded datum” before a survey is trusted as a training or validation source. audit_survey() is that report: it runs a raw sites input (anything ensure_sites() accepts) through the same canonical bridge used everywhere else in this package (survey_data_from_sites()) without letting a single bad station abort the whole audit the way that bridge’s own fail-fast frequency-grid check does. Every station dropped for missing data, and every station whose frequency grid does not match the rest, is named and given a reason; dimensionality, strike, static-shift, and distortion indicators come from the existing, heavier pandas-based diagnostics in pycsamt.emtools.dimensionality, pycsamt.emtools.gb, and pycsamt.emtools.ss rather than being re-implemented here.

This module depends on those pandas-based diagnostics and therefore lives in pycsamt.ai.domain_gap, not in the NumPy-only pycsamt.ai.data, matching the layering pycsamt.ai.data.contracts documents for itself.

Functions

audit_survey(sites, *[, recursive, on_dup, ...])

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

Classes

DimensionalitySummary(n_samples, frac_1d, ...)

Survey-wide aggregate of pre2d_inversion_assessment().

FrequencyGridReport(matched, ...)

Whether every included station shares one frequency grid.

StationExclusion(station, reason)

One station dropped before the canonical survey bridge.

SurveyAuditReport(n_stations_input, ...[, ...])

Complete M1 accounting of one survey's data quality.
class pycsamt.ai.domain_gap.audit.StationExclusion(station, reason)[source]

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

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.audit.FrequencyGridReport(matched, reference_station, n_frequencies_by_station, mismatched_stations)[source]

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

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.audit.DimensionalitySummary(n_samples, frac_1d, frac_2d, frac_3d, strike_consensus_deg, strike_consensus_iqr_deg, stations_recommending_3d_review=())[source]

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

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

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

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

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

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

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

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

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

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