pycsamt.agents.denoising

pycsamt.agents.denoising

DenoisingAgent — Multi-method MT data denoising.

Wraps both classical emtools filters and the AI-based denoiser:

Classical (no ML deps)

• RPCA — robust PCA off-diagonal denoising (rpca_offdiag_denoise())

• Hampel — frequency-domain Hampel outlier filter (hampel_filter_freq())

• EMAP — array-based EM array processing filter (apply_emap_filter())

• Pipeline — multi-step combined filter (remove_noise_pipeline())

AI-based (requires PyTorch or TensorFlow)

• CAE — convolutional autoencoder denoiser (EMDenoiser)

Classes

DenoisingAgent(*[, api_key, model, ...])

Denoise MT impedance data using classical or AI-based methods.

class pycsamt.agents.denoising.DenoisingAgent(*, api_key=None, model=None, llm_provider='claude', method='rpca', rank=2, half_window=3)

Bases: BaseAgent

Denoise MT impedance data using classical or AI-based methods.

Parameters:

• api_key (str)

• model (str)

• llm_provider (str)

• method (str, optional — overrides constructor default) – "rpca" (default), "hampel", "emap", "pipeline", or "ai" / "ai_cae" (requires PyTorch/TF).

• rank (int) – RPCA rank for off-diagonal denoising (default 2).

• half_window (int) – Hampel filter half-window (default 3).

• keys (Output data)

• ----------

• path (sites /)

• method

• output_dir (str, optional)

• period_range ([T_min, T_max], optional)

• keys

• ----------------

• impedance (denoised_sites Sites with denoised)

• per-(station (snr_after ndarray —)

• before (freq) SNR proxy)

• per-(station

• after (freq) SNR proxy)

• improvement (snr_gain float — mean SNR)

• threshold (n_recovered int — frequencies recovered above SNR)

• dict (figure_paths)

• dict

SYSTEM_PROMPT: str = 'You are an expert MT noise analysis and denoising specialist.\nGiven a denoising result summary, write 3–4 sentences that:\n1. State which noise sources were addressed (powerline, cultural, source effects).\n2. Quantify the improvement (e.g. SNR gain, number of frequencies recovered).\n3. Identify any remaining problematic frequencies or stations.\n4. Recommend follow-up processing steps.\nReply in plain English.\n'

Override in subclasses to give the LLM its domain expertise.

execute(input_data)

Run this agent on input_data and return an AgentResult.

Subclasses must implement this method. The contract:

• Reset self._last_cost = 0.0 at the top.

• Record wall-clock time with t0 = time.time().

• Return AgentResult(elapsed_seconds=time.time()-t0, cost_estimate_usd=self._last_cost, ...).

Parameters:

input_data (dict[str, Any])

Return type:

AgentResult