pycsamt.iot.edge_amt#

AMT/CSAMT-specific edge quality-control metrics.

The generic EdgeProcessor checks coverage, RMS, and spikes. Field electromagnetics needs more: powerline-harmonic contamination, channel SNR, saturation/clipping, contact-resistance proxies, resolvable frequency coverage, live spectra, impedance stability, and sensor dropout. Those diagnostics live here.

All spectral routines are implemented with numpy only (a small internal Welch estimator) so the subpackage imports without SciPy. If SciPy is installed it is used automatically for a better PSD.

Functions

`amt_edge_report`(data, sample_rate, *[, ...])	Run the core AMT edge diagnostics on one channel and collate them.
`amt_edge_table`(reports, *[, api])	Flatten one or more `amt_edge_report()` results into a table.
`assess_impedance_stability`(z_windows, *[, ...])	Assess the stability of per-window impedance estimates.
`check_channel_saturation`(data, *[, limit, ...])	Detect ADC clipping / saturation in a channel.
`check_contact_resistance`(ex[, ey, ...])	Proxy assessment of electrode contact quality.
`compute_live_spectra`(data, sample_rate, *[, ...])	Return `{"frequency_hz": ..., "psd": ...}` for live display.
`detect_powerline_harmonics`(data, sample_rate, *)	Detect mains-frequency harmonics in a time series.
`detect_sensor_dropout`(data, *[, ...])	Detect NaN gaps and stuck-value (flatline) runs in a channel.
`estimate_channel_snr`(data[, sample_rate, ...])	Estimate channel SNR in decibels.
`estimate_frequency_coverage`(timeseries, ...)	Estimate the resolvable frequency band of a recording.
`estimate_static_shift`(res_xy, res_yx, *[, ...])	Flag a static shift between the two apparent-resistivity modes.

Classes

`FrequencyCoverage`(sample_rate_hz, ...[, ...])	Result of `estimate_frequency_coverage()`.
`HarmonicPeak`(order, frequency_hz, ...)	Contamination measured at one powerline harmonic.
`ImpedanceStability`(n_windows, cv_magnitude, ...)	Result of `assess_impedance_stability()`.
`PowerlineHarmonics`(mains_hz[, peaks, ...])	Result of `detect_powerline_harmonics()`.
`StaticShift`(shift_factor, split_decades, ...)	Result of `estimate_static_shift()`.

class pycsamt.iot.edge_amt.HarmonicPeak(order, frequency_hz, power_ratio, flagged)[source]#

Bases: PyCSAMTObject

Contamination measured at one powerline harmonic.

Parameters:

order (int)
frequency_hz (float)
power_ratio (float)
flagged (bool)

order: int#

frequency_hz: float#

power_ratio: float#

flagged: bool#

as_dict()[source]#

Return type:: dict[str, Any]

class pycsamt.iot.edge_amt.PowerlineHarmonics(mains_hz, peaks=<factory>, total_ratio=0.0, contaminated=False)[source]#

Bases: PyCSAMTObject

Result of detect_powerline_harmonics().

Parameters:

mains_hz (float)
peaks (list[HarmonicPeak])
total_ratio (float)
contaminated (bool)

mains_hz: float#

peaks: list[HarmonicPeak]#

total_ratio: float = 0.0#

contaminated: bool = False#

property dominant: HarmonicPeak | None[source]#: Return the strongest harmonic, if any were measured.

as_dict()[source]#

Return type:: dict[str, Any]

class pycsamt.iot.edge_amt.FrequencyCoverage(sample_rate_hz, nyquist_hz, f_low_hz, f_high_hz, n_decades, coverage_fraction=nan, missing_bands=<factory>)[source]#

Bases: PyCSAMTObject

Result of estimate_frequency_coverage().

Parameters:

sample_rate_hz (float)
nyquist_hz (float)
f_low_hz (float)
f_high_hz (float)
n_decades (float)
coverage_fraction (float)
missing_bands (list[tuple[float, float]])

sample_rate_hz: float#

nyquist_hz: float#

f_low_hz: float#

f_high_hz: float#

n_decades: float#

coverage_fraction: float = nan#

missing_bands: list[tuple[float, float]]#

as_dict()[source]#

Return type:: dict[str, Any]

class pycsamt.iot.edge_amt.ImpedanceStability(n_windows, cv_magnitude, phase_std_deg, stable)[source]#

Bases: PyCSAMTObject

Result of assess_impedance_stability().

Parameters:

n_windows (int)
cv_magnitude (float)
phase_std_deg (float)
stable (bool)

n_windows: int#

cv_magnitude: float#

phase_std_deg: float#

stable: bool#

as_dict()[source]#

Return type:: dict[str, Any]

class pycsamt.iot.edge_amt.StaticShift(shift_factor, split_decades, consistency_std, phase_diff_deg, static_shift)[source]#

Bases: PyCSAMTObject

Result of estimate_static_shift().

Parameters:

shift_factor (float)
split_decades (float)
consistency_std (float)
phase_diff_deg (float)
static_shift (bool)

shift_factor: float#

split_decades: float#

consistency_std: float#

phase_diff_deg: float#

static_shift: bool#

as_dict()[source]#

Return type:: dict[str, Any]

pycsamt.iot.edge_amt.detect_powerline_harmonics(data, sample_rate, *, mains_hz=50.0, n_harmonics=5, bandwidth_hz=1.0, threshold_ratio=0.05)[source]#

Detect mains-frequency harmonics in a time series.

Parameters:

data (array-like) – Single-channel time series.
sample_rate (float) – Sampling frequency in Hz.
mains_hz (float) – Powerline fundamental (50 or 60 Hz typically).
n_harmonics (int) – Number of harmonics (including the fundamental) to test.
bandwidth_hz (float) – Half-width of the integration band around each harmonic.
threshold_ratio (float) – Per-harmonic band-power fraction above which a harmonic is flagged as contaminating.

Return type:

PowerlineHarmonics

pycsamt.iot.edge_amt.estimate_static_shift(res_xy, res_yx, *, phase_xy=None, phase_yx=None, min_split_decades=0.15, max_log_std=0.15, max_phase_diff_deg=10.0)[source]#

Flag a static shift between the two apparent-resistivity modes.

Static shift is a galvanic distortion that multiplies apparent resistivity by a frequency-independent factor while leaving phase unchanged. It therefore shows up as the xy and yx resistivity curves running parallel on a log scale (a near-constant split) even though their phases coincide – unlike true anisotropy, which splits the phases too.

Parameters:

res_xy (array-like) – Apparent resistivity (\(\Omega\cdot m\)) for the two off-diagonal modes, one value per frequency.
res_yx (array-like) – Apparent resistivity (\(\Omega\cdot m\)) for the two off-diagonal modes, one value per frequency.
phase_xy (array-like, optional) – Corresponding phases in degrees. When given, agreeing phases strengthen a static-shift call (and disagreeing phases veto it).
phase_yx (array-like, optional) – Corresponding phases in degrees. When given, agreeing phases strengthen a static-shift call (and disagreeing phases veto it).
min_split_decades (float) – Minimum |log10(shift_factor)| for a split to matter.
max_log_std (float) – Maximum standard deviation of the per-frequency log split for it to count as frequency-independent.
max_phase_diff_deg (float) – Maximum mean phase difference (when phases are supplied) for the distortion to read as purely galvanic.

Return type:

StaticShift

pycsamt.iot.edge_amt.estimate_channel_snr(data, sample_rate=None, *, signal_band_hz=None)[source]#

Estimate channel SNR in decibels.

Two estimators are provided:

If sample_rate and signal_band_hz are given, SNR is the ratio of in-band to out-of-band spectral power.
Otherwise a time-domain estimate is used: the signal power is the variance of the series and the noise power is derived from the variance of first differences (a white-noise proxy).

Parameters:

data (Any)
sample_rate (float | None)
signal_band_hz (tuple[float, float] | None)

Return type:

float

pycsamt.iot.edge_amt.check_channel_saturation(data, *, limit=None, max_clip_fraction=0.01, tol=1e-09)[source]#

Detect ADC clipping / saturation in a channel.

When limit is provided, samples with abs(x) >= limit count as saturated. Otherwise, samples equal to the observed min/max (within tol) are treated as clipped, which catches rail-to-rail saturation without a known full-scale value.

Parameters:

data (Any)
limit (float | None)
max_clip_fraction (float)
tol (float)

Return type:

dict[str, Any]

pycsamt.iot.edge_amt.check_contact_resistance(ex, ey=None, *, sample_rate=None, noise_rms_threshold=None)[source]#

Proxy assessment of electrode contact quality.

True contact resistance requires a current injection measurement. On passive AMT electric channels, poor contact manifests as elevated low-frequency noise, large DC offset, and drift. This routine reports those proxies per channel (ex and optional ey) and flags a channel when its high-pass noise RMS exceeds noise_rms_threshold (when provided) or when Ex/Ey noise is strongly imbalanced.

Parameters:

ex (Any)
ey (Any)
sample_rate (float | None)
noise_rms_threshold (float | None)

Return type:

dict[str, Any]

pycsamt.iot.edge_amt.estimate_frequency_coverage(timeseries, sample_rate, *, target_bands=None, snr_floor_db=6.0)[source]#

Estimate the resolvable frequency band of a recording.

The PSD noise floor is taken as its median. Frequencies whose power exceeds the floor by snr_floor_db are considered resolved; the lowest and highest such frequencies define the covered band. When target_bands are supplied, the fraction that falls inside the covered band is reported along with any missing bands.

Parameters:

timeseries (Any)
sample_rate (float)
target_bands (Iterable[tuple[float, float]] | None)
snr_floor_db (float)

Return type:

FrequencyCoverage

pycsamt.iot.edge_amt.compute_live_spectra(data, sample_rate, *, nperseg=None)[source]#

Return {"frequency_hz": ..., "psd": ...} for live display.

Parameters:

data (Any)
sample_rate (float)
nperseg (int | None)

Return type:

dict[str, ndarray]

pycsamt.iot.edge_amt.assess_impedance_stability(z_windows, *, max_cv=0.15, max_phase_std_deg=10.0)[source]#

Assess the stability of per-window impedance estimates.

Parameters:

z_windows (array-like of complex) – Impedance estimates, shape (n_windows,) or (n_windows, n_freq). Real inputs are treated as magnitudes with zero phase.
max_cv (float) – Maximum coefficient of variation of |Z| for a stable result.
max_phase_std_deg (float) – Maximum phase standard deviation (degrees) for a stable result.

Return type:

ImpedanceStability

pycsamt.iot.edge_amt.detect_sensor_dropout(data, *, min_flat_run=8, flat_tol=1e-12)[source]#

Detect NaN gaps and stuck-value (flatline) runs in a channel.

Returns counts for NaN samples and the longest run of (near-)constant consecutive samples, which typically indicates a disconnected or stuck sensor.

Parameters:

data (Any)
min_flat_run (int)
flat_tol (float)

Return type:

dict[str, Any]

pycsamt.iot.edge_amt.amt_edge_report(data, sample_rate, *, method=None, mains_hz=50.0, signal_band_hz=None)[source]#

Run the core AMT edge diagnostics on one channel and collate them.

When method is given ("amt", "mt", "csamt", …), the diagnostics become method-aware: powerline-harmonic detection is only run for powerline-sensitive methods (it is skipped for, e.g., TDEM), and frequency coverage is scored against the method’s target bands. Passing no method preserves the original behaviour.

Parameters:

data (Any)
sample_rate (float)
method (Any)
mains_hz (float)
signal_band_hz (tuple[float, float] | None)

Return type:

dict[str, Any]

pycsamt.iot.edge_amt.amt_edge_table(reports, *, api=None)[source]#

Flatten one or more amt_edge_report() results into a table.

Accepts a {channel: report} mapping (or (channel, report) pairs) and returns one row per channel with the headline metrics.

Parameters:

reports (dict[str, dict[str, Any]] | Iterable[tuple[str, dict[str, Any]]])
api (bool | None)

Return type:

Any