pycsamt.agents#

AI-assisted workflow agents for loading, quality control, processing, inversion preparation, interpretation, reporting, and orchestration.

AI-assisted MT, AMT, and CSAMT workflow automation.

All agents are lazy-loaded: importing this module costs nothing unless you actually instantiate an agent. This keeps the base pycsamt import fast even when LLM libraries (anthropic, openai, google-generativeai) are not installed.

Agent catalogue#

ContextInputAgent: Translate a natural-language request into a structured workflow config.
MTLoaderAgent: Load EDI / AVG / J files into a Sites object with a per-station QC report.
AgentCoordinator: Chain agents into named workflows with checkpointing and cost tracking.
DataQCAgent: SNR section, dead-band detection, per-station quality scores.
StaticShiftAgent: Static-shift detection and correction (AMA / LOESS / spatial median).
PhaseAnalysisAgent: Phase tensor, strike, dimensionality, Mohr circles, Argand diagrams.
ForwardModelAgent: 1-D, 2-D, and 3-D MT forward modelling.
InversionPrepAgent / Occam2DAgent / ModEmAgent / Mare2DEMAgent: Write Occam2D, ModEM3D, or MARE2DEM inversion input files; optionally run the binary and scan results.
InversionEvaluationAgent: Compute RMS, residual PT section, misfit pseudosection.
InterpretationAgent: Map resistivity ranges to lithology; correlate with borehole logs.
ReportAgent: Assemble all figures and tables into Markdown / HTML / PDF.
CodeGenerationAgent: Emit a standalone reproducible Python script from the workflow config.
WorkflowOrchestratorAgent: NL → classify workflow → build and run the correct agent chain.
AgentMaster: One-line front door over the orchestrator: AgentMaster(provider="anthropic").run("...").
DenoisingAgent: RPCA / Hampel / EMAP / AI-CAE denoising.
AnomalyDetectionAgent: Unsupervised CAE anomaly flagging per (station, frequency).
AIInversionAgent: End-to-end 1-D AI inversion (ResNet / CNN / FCN).
Inv2DAgent: U-Net 2-D profile inversion with lateral continuity.
Inv3DAgent: GCN 3-D spatial inversion using inter-station graph message-passing.
EnsembleAgent: Ensemble 1-D inversion with conformal uncertainty bands.
JointInversionAgent: DRCNN multi-modal joint inversion (MT + TEM / CSAMT / gravity).
ModelZooAgent: Browse, download, and deploy pre-trained EM inverter checkpoints.
PINNInversionAgent: Physics-informed 1-D/2-D/3-D MT inversion via Adam (no labelled training data required).
HybridInversionAgent: Two-stage AI warm-start + physics refinement for 1-D, 2-D, and 3-D MT inversion.
IoTFieldAgent: Monitor an IoT-enabled AMT/MT/CSAMT/CSEM field acquisition from its edge telemetry: monitoring status, sync/power summaries, provenance manifest, and dashboard figures.

Supporting classes#

AgentResult: Standardised return type for every agent.
BaseAgent: Abstract base class; inherit to build custom agents.

Quick start#

Without an LLM key (regex fallback, no cost):

from pycsamt.agents import ContextInputAgent, MTLoaderAgent, AgentCoordinator

ctx    = ContextInputAgent()   # no api_key → pure regex
loader = MTLoaderAgent()

coord  = AgentCoordinator("mt_qc")
coord.add_step("parse", ctx,
               description="Parse request into config")
coord.add_step("load",  loader,
               input_fn=lambda r: {"path": r["parse"]["config"]["data_path"]},
               description="Load EDI files and QC scan")

result = coord.execute(
    {"request": "Load /data/EDIs, QC, period range 1e-4 to 1 s"},
    dry_run=True,
)

With a Claude API key:

ctx = ContextInputAgent(api_key="sk-ant-…", llm_provider="claude")
res = ctx.execute({"request": "Load L22PLT EDIs, run PT analysis …"})
print(res["config"])
print(res.llm_interpretation)

LLM providers supported#

Anthropic Claude (default) — llm_provider="claude"
OpenAI — llm_provider="openai"
Google Gemini — llm_provider="gemini"
DeepSeek — llm_provider="deepseek"
MiniMax — llm_provider="minimax"

Install the relevant client library before use: pip install anthropic / pip install openai / pip install google-generativeai DeepSeek and MiniMax reuse the openai package with a custom base_url; no extra install is needed.

class pycsamt.agents.AgentConfig#

Bases: object

Global LLM configuration singleton for all pycsamt agents.

BaseAgent calls resolve() on every instantiation and get_rate() + _add_spend() on every LLM call, so keys, rates, and the budget cap set here apply automatically to every agent in the session.

:param (none — use configure() after construction):

Variables:

provider (str or None) – Currently active LLM provider.
model (str or None) – Resolved model (explicit override or provider default).
api_key (str or None) – Resolved key for the active provider (stored key → env var → None).
is_configured (bool) – True when a provider and a resolvable key are both present.
spent_usd (float) – Accumulated LLM spend this session.
remaining_usd (float or None) – Remaining budget, or None when no cap is set.

Examples

from pycsamt.api.agents import AGENT_CONFIG

# one-call setup
AGENT_CONFIG.configure(provider="claude", api_key="sk-ant-…")

# pricing — override a rate (USD / 1 M tokens)
AGENT_CONFIG.set_rate("claude", "claude-opus-4-8",
                      input=12.0, output=60.0)

# pricing — add a model not in the built-in table
AGENT_CONFIG.set_rate("openai", "gpt-5-turbo", input=5.0, output=20.0)

# budget cap
AGENT_CONFIG.set_budget(usd=5.0)
print(AGENT_CONFIG.remaining_usd)   # 5.0

# inspect
print(AGENT_CONFIG.info())

# reset
AGENT_CONFIG.reset()

configure(*, provider, api_key, model=None)#

Set the active provider, key, and optional model in one call.

Parameters:

provider ({"claude", "openai", "gemini"}) – LLM provider to activate.
api_key (str) – API key for provider. Stored per-provider so switching away and back does not require re-supplying the key.
model (str or None) – Model identifier override. None uses the provider default.

Returns:

self — allows chaining.

Return type:

AgentConfig

Examples

AGENT_CONFIG.configure(provider="claude", api_key="sk-ant-…")
AGENT_CONFIG.configure(
    provider="openai",
    api_key="sk-…",
    model="gpt-4o-mini",
)

set_key(provider, api_key)#

Store an API key for provider without changing the active provider.

Useful for pre-loading keys for multiple providers so you can switch() between them without re-supplying credentials.

Parameters:

provider (str)
api_key (str)

Returns:

self.

Return type:

AgentConfig

Examples

AGENT_CONFIG.set_key("claude", "sk-ant-…")
AGENT_CONFIG.set_key("openai", "sk-…")
AGENT_CONFIG.set_key("gemini", "AIza…")
AGENT_CONFIG.switch("claude")

switch(provider, *, model=None)#

Switch the active provider.

The key for provider must have been stored via configure() or set_key(), or be available as an environment variable.

Parameters:

provider (str)
model (str or None) – Override the model for this provider. None keeps the current model override (or uses the provider default).

Returns:

self.

Return type:

AgentConfig

Examples

AGENT_CONFIG.switch("openai")
AGENT_CONFIG.switch("gemini", model="gemini-2.0-flash")

reset(*, keys=True)#

Clear the active configuration, custom rates, and budget.

Parameters:: keys (bool) – When True (default) also wipe all stored per-provider keys. Pass False to clear only the active provider / model while keeping stored keys available for future switch() calls.
Returns:: self.
Return type:: AgentConfig

Notes

Custom rates and the budget cap/counter are always reset. Call reset_budget() to zero only the spend counter.

set_rate(provider, model, *, input, output)#

Override or add the cost rate for a specific provider + model.

Custom rates take precedence over the built-in table for every cost estimate made by agents in this session.

Parameters:

provider (str) – "claude" | "openai" | "gemini"
model (str) – Model identifier exactly as passed to the LLM client, e.g. "claude-opus-4-8" or "gpt-5-turbo".
input (float) – Cost per 1 000 000 input tokens in USD.
output (float) – Cost per 1 000 000 output tokens in USD.

Returns:

self.

Return type:

AgentConfig

Examples

# provider lowered their price
AGENT_CONFIG.set_rate("claude", "claude-opus-4-8",
                      input=12.0, output=60.0)

# a new model not yet in the built-in table
AGENT_CONFIG.set_rate("openai", "gpt-5-turbo",
                      input=5.0, output=20.0)

get_rate(provider, model)#

Return the resolved {"input": …, "output": …} rate.

Resolution order: custom override → built-in table (exact match → prefix match) → provider default.

Parameters:

provider (str)
model (str)

Return type:

dict with keys "input" and "output" (USD / 1 M tokens).

Examples

rate = AGENT_CONFIG.get_rate("claude", "claude-sonnet-4-6")
# {"input": 3.0, "output": 15.0}

estimate_cost(provider, model, input_tokens, output_tokens)#

Compute the USD cost for one LLM call using the resolved rate.

Respects any rate overrides set via set_rate().

Parameters:

provider (str)
model (str)
input_tokens (int)
output_tokens (int)

Return type:

float — estimated cost in USD.

Examples

cost = AGENT_CONFIG.estimate_cost(
    "claude", "claude-sonnet-4-6", 500, 200
)

list_rates(provider=None)#

Return the effective rate table (custom overrides merged with built-in).

Parameters:: provider (str or None) – When given, return only that provider’s table. When None, return all providers.
Return type:: dict — same structure as the built-in table.

Examples

import pprint
pprint.pprint(AGENT_CONFIG.list_rates("claude"))

set_budget(*, usd)#

Set a session spend cap.

Once spent_usd reaches usd, any subsequent LLM call will raise BudgetExceededError before the API call is made.

Parameters:: usd (float) – Maximum spend in USD for this session.
Returns:: self.
Return type:: AgentConfig

Examples

AGENT_CONFIG.set_budget(usd=2.0)
# … after several agent calls …
print(f"${AGENT_CONFIG.spent_usd:.4f} used, "
      f"${AGENT_CONFIG.remaining_usd:.4f} left")

reset_budget(*, cap=False)#

Reset the session spend counter.

Parameters:: cap (bool) – When True also remove the budget cap (set_budget must be called again to re-enable it). Default False — zeroes the counter but keeps the existing cap.
Returns:: self.
Return type:: AgentConfig

Examples

AGENT_CONFIG.reset_budget()          # zero counter, keep cap
AGENT_CONFIG.reset_budget(cap=True)  # zero counter and remove cap

property spent_usd: float#: Accumulated LLM spend this session (USD).

property remaining_usd: float | None#: Remaining budget (USD), or None when no cap is set.

property provider: str | None#: Currently active provider, or None if unconfigured.

property api_key: str | None#

Resolved key for the active provider.

Resolution order: stored key → environment variable → None.

property model: str | None#: Active model (explicit override, or provider default).

property is_configured: bool#: True when a provider and a resolvable key are both present.

resolve(provider, api_key, model)#

Resolve the effective (provider, api_key, model) for an agent.

Called automatically by __init__. Users should not call this directly.

Resolution rules#

If api_key is explicitly given:

Use provider, api_key, and model (or provider default) exactly as supplied. The global config is ignored.

If api_key is None:

If provider equals the default "claude" and the global config has an active provider, inherit that provider.
Look up the key: stored key → env var.
Inherit the model from the global config when the effective provider matches the globally active provider.

Parameters:

provider (str)
api_key (str | None)
model (str | None)

Return type:

tuple[str, str | None, str | None]

info()#

Return a summary dict suitable for display or logging.

The API key is masked to its last four characters for safety.

Returns:: Keys: provider, model, has_key, key_source, stored_providers, custom_rate_models, budget_usd, spent_usd, remaining_usd.
Return type:: dict

using(*, provider=None, api_key=None, model=None)#

Temporarily override the global config inside a with block.

All arguments are optional; omitted values are left unchanged. The original config (including custom rates and budget) is restored on exit even if an exception occurs.

Parameters:

provider (str or None)
api_key (str or None)
model (str or None)

Yields:

AgentConfig – self with the override applied.

Return type:

Generator[AgentConfig, None, None]

Examples

with AGENT_CONFIG.using(provider="gemini", api_key="AIza…"):
    r = DataQCAgent().execute(data)
# original provider / key / model restored here

offline()#

Context manager: force truly offline mode in the current thread.

While active, _resolve_key() will not inspect environment variables, so agents created inside the block receive api_key=None even when ANTHROPIC_API_KEY (or similar) is set in the OS environment. Uses threading.local so concurrent requests in other threads are unaffected.

Examples

with AGENT_CONFIG.offline():
    agent = DataQCAgent()  # api_key will be None

Return type:: Generator[AgentConfig, None, None]

exception pycsamt.agents.BudgetExceededError(spent, budget)#

Bases: RuntimeError

Raised when an LLM call would be made after the session budget is used.

Variables:

spent_usd (float) – Accumulated spend so far this session.
budget_usd (float) – The cap that was set.

Parameters:

spent (float)
budget (float)

Return type:

None

pycsamt.agents.configure_agents(*, provider, api_key, model=None)#

Configure AGENT_CONFIG in one call and return it.

Equivalent to AGENT_CONFIG.configure(...).

Parameters:

provider ({"claude", "openai", "gemini"})
api_key (str)
model (str or None)

Return type:

AgentConfig

Examples

from pycsamt.agents import configure_agents
configure_agents(provider="claude", api_key="sk-ant-…")

pycsamt.agents.reset_agents(*, keys=True)#

Reset AGENT_CONFIG to its unconfigured state.

Equivalent to AGENT_CONFIG.reset(...).

Parameters:: keys (bool) – When True (default) also wipe stored per-provider keys.
Return type:: AgentConfig

class pycsamt.agents.AgentResult(status, summary, data=<factory>, warnings=<factory>, llm_interpretation=None, elapsed_seconds=0.0, cost_estimate_usd=0.0, error=None, error_fix_hint=None)#

Bases: object

Standardised output returned by every pycsamt agent.

Variables:

status (str) – "success" | "failed" | "needs_review"
summary (str) – One-sentence human-readable description of what happened.
data (dict) – Agent-specific outputs (arrays, paths, dataframes, figures …).
warnings (list of str) – Non-fatal issues encountered during execution.
llm_interpretation (str or None) – Free-text interpretation written by the LLM, when available.
elapsed_seconds (float)
cost_estimate_usd (float) – Estimated LLM API cost for this execution.
error (str or None) – Exception message if status == "failed".
error_fix_hint (str or None) – Suggested remediation for the error.

Parameters:

status (str)
summary (str)
data (dict[str, Any])
warnings (list[str])
llm_interpretation (str | None)
elapsed_seconds (float)
cost_estimate_usd (float)
error (str | None)
error_fix_hint (str | None)

Examples

>>> result = some_agent.execute({"path": "/data/EDIs"})
>>> result.status
'success'
>>> result["sites"]          # dict-like access to data
<Sites 25 stations>
>>> result.get("n_stations", 0)
25

status: str#

summary: str#

data: dict[str, Any]#

warnings: list[str]#

llm_interpretation: str | None = None#

elapsed_seconds: float = 0.0#

cost_estimate_usd: float = 0.0#

error: str | None = None#

error_fix_hint: str | None = None#

get(key, default=None)#

Parameters:

key (str)
default (Any)

Return type:

Any

classmethod failed(error, *, hint=None, elapsed=0.0)#

Convenience constructor for failure results.

Parameters:

error (str)
hint (str | None)
elapsed (float)

Return type:

AgentResult

class pycsamt.agents.BaseAgent(name, *, api_key=None, model=None, llm_provider='claude', section_preset='pseudosection')#

Bases: ABC

Abstract base class for all pycsamt agents.

Parameters:

name (str) – Human-readable agent name used in logs and reports.
api_key (str or None) – LLM API key. When None the agent runs without LLM support and every llm_interpretation field will be None.
model (str or None) – LLM model identifier. Defaults to the provider’s recommended model.
llm_provider ({"claude", "openai", "gemini", "deepseek", "minimax"}) – Which provider to use. Default "claude".
section_preset (str) – Which PYCSAMT_SECTION preset governs figures produced by this agent. Default "pseudosection".

Examples

Subclass and implement execute():

class MyAgent(BaseAgent):
    SYSTEM_PROMPT = "You are an expert MT data analyst."

    def execute(self, input_data):
        t0 = time.time()
        # ... do work ...
        interp = self.query_llm("Interpret these results: ...")
        return AgentResult(
            status="success",
            summary="Analysis complete.",
            data={"result": 42},
            llm_interpretation=interp,
            elapsed_seconds=time.time() - t0,
            cost_estimate_usd=self._last_cost,
        )

SYSTEM_PROMPT: str = 'You are a geophysics expert specialising in magnetotelluric (MT/AMT/CSAMT) data processing and interpretation.'#: Override in subclasses to give the LLM its domain expertise.

abstractmethod 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

query_llm(prompt, system_message=None, *, temperature=0.2, max_tokens=1024)#

Send prompt to the configured LLM and return the response text.

Returns None when no API key is configured or all retries fail. Accumulates token cost into self._last_cost.

Parameters:

prompt (str)
system_message (str or None) – Overrides SYSTEM_PROMPT for this call.
temperature (float)
max_tokens (int)

Return type:

str or None

extract_json(text)#

Extract the first JSON object / array from text.

Returns None when no valid JSON is found.

Parameters:: text (str)
Return type:: dict | list | None

require_keys(input_data, *keys, agent_name=None)#

Return a list of missing required keys.

Parameters:

input_data (dict)
keys (str)
agent_name (str | None)

Return type:

list[str]

class pycsamt.agents.AgentCoordinator(workflow_name='pycsamt_workflow', *, checkpoint_dir=None, verbose=True)#

Bases: object

Orchestrate a sequence of pycsamt agents as a named workflow.

Parameters:

workflow_name (str) – Name used for log messages and checkpoint directory.
checkpoint_dir (str or Path or None) – Where to save step checkpoints. Defaults to ./pycsamt_agent_checkpoints/<workflow_name>/.
verbose (bool) – Print step-level progress to stdout.

Examples

Build and run a QC workflow:

from pycsamt.agents import AgentCoordinator, MTLoaderAgent, DataQCAgent

coord = AgentCoordinator("mt_qc")
coord.add_step("load",  MTLoaderAgent(...))
coord.add_step("qc",    DataQCAgent(...),
               input_fn=lambda r: {"sites": r["load"]["sites"]})

result = coord.execute({"path": "/data/EDIs"})
print(result.summary)

Dry-run preview:

result = coord.preview({"path": "/data/EDIs"})
print(result["plan"])

add_step(name, agent, *, input_fn=None, description='', required=True)#

Append a workflow step. Returns self for chaining.

Parameters:

name (str)
agent (BaseAgent)
input_fn (callable(prev_results_dict) → dict, optional) – Maps accumulated results to this step’s input dict. prev_results_dict keys are earlier step names; each value is the step’s AgentResult.
description (str)
required (bool)

Returns:

self for method chaining.

Return type:

AgentCoordinator

preview(config)#

Return an execution plan and estimated cost without running anything.

Parameters:: config (dict) – The workflow configuration that would be passed to execute().
Returns:: data["plan"] contains the formatted plan string. data["steps"] is a list of step metadata dicts.
Return type:: AgentResult

execute(config, *, dry_run=False, resume=False)#

Run all workflow steps sequentially.

Parameters:

config (dict) – Top-level workflow configuration forwarded to every step’s input_fn (or directly to agent.execute() when no input_fn is set).
dry_run (bool) – When True return a preview without executing any agents.
resume (bool) – When True skip steps whose checkpoint already exists on disk.

Returns:

data contains one key per step name with its AgentResult.

Return type:

AgentResult

reset_checkpoints()#

Delete all saved checkpoints for this workflow.

Return type:: None

class pycsamt.agents.WorkflowStep(name, agent, *, input_fn=None, description='', required=True)#

Bases: object

One step in an AgentCoordinator workflow.

Parameters:

name (str) – Unique identifier used for checkpointing and logging.
agent (BaseAgent) – The agent instance that runs this step.
input_fn (callable or None) – input_fn(prev_results) → dict fed to agent.execute(). When None the coordinator passes the raw workflow config.
description (str) – Human-readable description shown in the dry-run preview.
required (bool) – When False a failure skips the step rather than aborting.

pycsamt.agents.estimate_cost(provider, model, input_tokens, output_tokens)#

Compute the USD cost for one LLM call.

Delegates to estimate_cost(), so any per-model overrides set via AGENT_CONFIG.set_rate(...) are used.

Parameters:

provider (str)
model (str)
input_tokens (int)
output_tokens (int)

Returns:

Estimated cost in USD.

Return type:

float

Examples

>>> estimate_cost("claude", "claude-sonnet-4-6", 500, 200)
0.00451...

pycsamt.agents.format_cost(usd)#

Return a human-readable cost string.

Examples

>>> format_cost(0.00012)
'$0.000120'
>>> format_cost(1.5)
'$1.5000'

Parameters:: usd (float)
Return type:: str

pycsamt.agents.get_rate(provider, model)#

Return the resolved {"input": …, "output": …} rate.

Delegates to get_rate(), so any per-model overrides set via AGENT_CONFIG.set_rate(...) take effect.

Parameters:

provider (str) – One of "claude", "openai", "gemini", "deepseek", "minimax".
model (str) – Model identifier, e.g. "claude-sonnet-4-6".

Return type:

dict with keys "input" and "output" (USD / 1 M tokens).

class pycsamt.agents.WorkflowPlan(request, workflow_type, data_path='', output_dir='', parameters=<factory>, risk_flags=<factory>, requires_human_review=False, expected_outputs=<factory>, provider='offline')#

Bases: object

Validated intermediate representation of a workflow request.

Produced by ContextInputAgent and consumed by WorkflowOrchestratorAgent.

Parameters:

request (str) – Original natural-language request.
workflow_type (str) – One of the recognised workflow identifiers (see VALID_WORKFLOWS).
data_path (str) – Path to EDI files or survey directory.
output_dir (str) – Destination for outputs.
parameters (dict) – Parsed parameters (period_range, component, etc.).
risk_flags (list of str) – Warnings or risks the user should review.
requires_human_review (bool) – True when ambiguous parameters were inferred.
expected_outputs (list of str) – Files / artefacts the workflow should produce.
provider (str) – LLM provider used for parsing, or 'offline'.

request: str#

workflow_type: str#

data_path: str = ''#

output_dir: str = ''#

parameters: dict[str, Any]#

risk_flags: list[str]#

requires_human_review: bool = False#

expected_outputs: list[str]#

provider: str = 'offline'#

is_valid()#

Return True when the plan passes basic validation.

Return type:: bool

validation_errors()#

Return a list of validation error messages.

Return type:: list[str]

to_dict()#

Return a JSON-serialisable dict.

Return type:: dict[str, Any]

to_json(indent=2)#

Serialise to a JSON string.

Parameters:: indent (int)
Return type:: str

save(path)#

Write the plan to path as JSON.

Parameters:: path (str | Path)
Return type:: None

classmethod from_config(config, request='', provider='offline')#

Build a WorkflowPlan from a config dict as returned by ContextInputAgent.

Parameters:

config (dict[str, Any])
request (str)
provider (str)

Return type:

WorkflowPlan

pycsamt.agents.validate_workflow_plan(plan, *, raise_on_error=False)#

Validate plan and optionally raise on errors.

Parameters:

plan (WorkflowPlan)
raise_on_error (bool) – When True, raises ValueError if errors exist.

Returns:

errors – Empty when the plan is valid.

Return type:

list of str

class pycsamt.agents.MTLoaderAgent(*, api_key=None, model=None, llm_provider='claude', recursive=True, on_dup='replace')#

Bases: BaseAgent

Load MT data from any pycsamt-supported format and assess quality.

Parameters:

api_key (str or None)
model (str)
llm_provider (str)
recursive (bool) – When loading a directory, recurse into sub-directories.
on_dup (str) – Duplicate-station handling: "replace" (default) or "skip".

Examples

>>> agent = MTLoaderAgent()
>>> result = agent.execute({"path": "/data/AMT/WILLY_DATA/L22PLT"})
>>> result.status
'success'
>>> result["n_stations"]
25
>>> result["quality_table"].head()
     station  has_z  ...  qc_score
0  22-22BF    True  ...        88

SYSTEM_PROMPT: str = 'You are an expert MT/AMT/CSAMT data quality analyst.\nGiven a per-station data quality summary, write 2–3 concise sentences that:\n1. State the overall data quality.\n2. Flag any stations or frequency ranges that need attention.\n3. Recommend the next processing step.\nReply in plain English — no bullet points, no markdown.\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

class pycsamt.agents.DataQCAgent(*, api_key=None, model=None, llm_provider='claude', method='composite', min_frac_ok=0.6, min_snr_med=2.0, max_skew_med=6.0)#

Bases: BaseAgent

Run data quality control on a MT/AMT dataset.

Parameters:

api_key (str) – LLM configuration (optional).
model (str) – LLM configuration (optional).
llm_provider (str) – LLM configuration (optional).
method (str) – Confidence scoring method: "composite" (default), "presence", "snr", or "spatial".
min_frac_ok (float) – Minimum fraction of OK frequencies for a station to pass (0–1).
min_snr_med (float) – Minimum median SNR for a station to pass.
max_skew_med (float) – Maximum median |β| skewness for a station to pass.
keys (Output data)
----------
sites (Sites or path : str) – EDI data to assess.
output_dir (str, optional) – Where to save QC figures.
period_range ([T_min, T_max], optional) – Restrict QC to this period window.
keys
----------------
metrics (qc_table pandas DataFrame — per-station)
station (flags pandas DataFrame — pass / fail per)
scores (confidence_table pandas DataFrame — per-station confidence)
confidence (freq_conf_table pandas DataFrame — per-frequency)
int (n_flagged)
list[str] (flagged_stations)
objects (figures dict — matplotlib Figure)
set) (figure_paths dict — saved file paths (when output_dir)

Examples

>>> agent  = DataQCAgent()
>>> result = agent.execute({"path": "/data/L22PLT",
...                         "output_dir": "/out/qc"})
>>> result["n_flagged"]
2
>>> result["figures"]["confidence_section"]
<Figure …>

SYSTEM_PROMPT: str = 'You are an expert MT/AMT/CSAMT data quality analyst for pycsamt v2.\nGiven a survey QC summary, write 3–4 sentences that:\n1. State the overall data quality (good / moderate / poor).\n2. Identify specific stations or frequency bands that need attention.\n3. Explain the likely cause (instrument noise, EM interference, near-field).\n4. Recommend the most important next processing step.\nReply in plain English. No bullet points, no markdown headings.\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

class pycsamt.agents.PhaseAnalysisAgent(*, api_key=None, model=None, llm_provider='claude', skew_th=5.0, ellipt_th=0.1, band=None)#

Bases: BaseAgent

Run a full phase tensor, strike, and dimensionality survey analysis.

Parameters:

api_key (str)
model (str)
llm_provider (str)
skew_th (float) – Skewness |β| threshold for 3-D classification (°).
ellipt_th (float) – Ellipticity λ threshold for 2-D classification.
band ((T_min, T_max) or None) – Period band for strike estimation.
keys (Output data)
----------
path (sites /)
period_range ([T_min, T_max], optional)
output_dir (str, optional)
run_mohr (bool, optional — also produce Mohr circles (default False))
run_fingerprint (bool, optional — produce fingerprint grid (default True))
keys
----------------
(station (pt_table pandas DataFrame — full PT metrics per)
period)
(°) (strike_consensus float — consensus strike angle)
stations (strike_iqr float — IQR of strike across all)
per-(station (dim_table pandas DataFrame —)
classification (period))
n_1d
n_2d
class (n_3d int — count of observations per)
objects (figures dict — matplotlib Figure)
paths (figure_paths dict — saved file)

Examples

>>> agent  = PhaseAnalysisAgent()
>>> result = agent.execute({"path": "/data/L22PLT",
...                         "output_dir": "/out/pt"})
>>> result["strike_consensus"]
42.5

SYSTEM_PROMPT: str = 'You are an expert in MT phase tensor analysis and geological interpretation.\nGiven a survey phase tensor summary, write 4–5 sentences that:\n1. State the dominant dimensionality (1-D, 2-D, or 3-D) with evidence.\n2. Report the consensus geoelectric strike direction and its reliability.\n3. Identify periods / depth ranges where 3-D structure becomes significant.\n4. Note any anomalous stations (high skew, low ellipticity).\n5. Recommend whether to rotate data to strike before inversion.\nReply in plain English. No bullet points or markdown.\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

class pycsamt.agents.ForwardModelAgent(*, api_key=None, model=None, llm_provider='claude', dim=1, freqs=None)#

Bases: BaseAgent

Run a 1-D, 2-D, or 3-D MT forward model.

Parameters:

api_key (str)
model (dict or LayeredModel or None)
llm_provider (str)
dim (int, optional — overrides constructor dim for this call) – Forward solver dimensionality.
freqs (array-like, optional — overrides constructor default) – Frequencies (Hz). Defaults to 40 log-spaced points 10⁻⁴–10³ Hz.
keys (Output data)
----------
model –
1-D / 2-D from 1-D layers: {"resistivities": [...], "thicknesses": [...]}.

2-D grid type override: add "type": "halfspace" | "anomaly" and grid parameters such as "bg_rho", "anomaly_rho", "anomaly_bounds".

3-D grid type: "type": "halfspace" | "block_anomaly" with grid parameters.
dim
nx (int / float, optional (2-D grid))
nz (int / float, optional (2-D grid))
x_max (int / float, optional (2-D grid))
z_max (int / float, optional (2-D grid))
ny (int / float (3-D))
y_max (int / float (3-D))
nx_stations (int / float (3-D))
ny_stations (int / float (3-D))
n_stations (int, optional — number of surface receivers (2-D))
method (str, optional — "quasi3d" (default) for 3-D solver)
path (sites /)
freqs
output_dir (str, optional)
component ("xy" (default) or "yx" (1-D component selection))
keys
----------------
int (dim)
1-D) (layered_model LayeredModel (1-D / 2-D from)
3-D) (grid Grid2D or Grid3D (2-D /)
ForwardResponse3D (response ForwardResponse / ForwardResponse2D /)
ρa (rho_a ndarray — 1-D)
(°) (phase ndarray — 1-D phase)
(n_freqs (rho_a_xy ndarray)
TE (n_stations) — 2-D)
phase (phase_yx ndarray — 3-D YX)
TM (rho_a_tm ndarray — 2-D)
phase
(n_freqs
XY (n_stations) — 3-D)
phase
YX (rho_a_yx ndarray — 3-D)
phase
ndarray (freqs)
None (rms float or)
dict (figure_paths)
dict

SYSTEM_PROMPT: str = 'You are an expert in MT forward modelling and resistivity earth models.\nGiven a forward model result, write 3-4 sentences that:\n1. Describe the model geometry (dimensionality, layers / grid, resistivity range).\n2. Comment on the synthetic ρa and phase response (frequency range, lateral variation for 2D/3D).\n3. If observed data are provided, interpret the data-model misfit (1-D only).\n4. Suggest which model parameters to adjust to better fit the data or geology.\nReply in plain English. No bullet points or markdown.\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

class pycsamt.agents.ReportAgent(*, api_key=None, model=None, llm_provider='claude', report_title='MT/AMT Survey Report', formats=None)#

Bases: BaseAgent

Generate a structured MT survey report from agent results.

Parameters:

api_key (str)
model (str)
llm_provider (str)
report_title (str) – Title for the report.
formats (list of {"md", "html", "pdf"}) – Output formats. Default ["md", "html"].
keys (Output data)
----------
results (dict) – Keyed by agent step name → AgentResult. Expected keys: "load", "qc", "static_shift", "phase_analysis", "forward" (all optional).
output_dir (str)
title (str, optional — overrides constructor default)
keys
----------------
text (report_md str — full markdown)
None (report_path_html str or)
file (report_path_md str — path to .md)
None
name (sections dict — section text keyed by)

Examples

>>> agent  = ReportAgent(api_key="sk-ant-…")
>>> result = agent.execute({
...     "results":    {"load": load_result, "qc": qc_result},
...     "output_dir": "/out/report",
...     "title":      "WILLY_DATA AMT Survey — L22PLT",
... })
>>> print(result["report_path_md"])
/out/report/survey_report.md

SYSTEM_PROMPT: str = 'You are a geophysics technical writer specialising in MT surveys.\nWrite clear, concise report sections in formal scientific English.\nUse complete sentences. No markdown headings inside your response.\nKeep each section to 3–5 sentences.\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

class pycsamt.agents.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

class pycsamt.agents.AIInversionAgent(*, api_key=None, model=None, llm_provider='claude', arch='resnet', n_layers=5, n_train_samples=2000, epochs=30, freqs=None, pretrained=None)#

Bases: BaseAgent

Train an AI inverter on synthetic data then predict on observed sites.

Parameters:

api_key (str)
model (str)
llm_provider (str)
arch (optional overrides) – Neural network architecture.
n_layers (int) – Number of model layers the inverter will predict (default 5).
n_train_samples (optional overrides) – Number of synthetic training samples (default 2 000).
epochs (optional overrides) – Training epochs (default 30). Increase for better models.
freqs (array-like or None) – Frequencies used for both training synthesis and observed data interpolation. Default: 40 log-spaced 10⁻⁴–10³ Hz.
pretrained (str or None) – Path to a pre-trained model checkpoint. When set, skips training.
keys (Output data)
----------
path (sites /)
output_dir (str, optional)
arch
epochs
n_train_samples
keys
----------------

:param inverter EMInverter1D: :param predictions dict {station: :type predictions dict {station: ndarray of log₁₀ ρ values} :param best_model dict with “resistivity” and “thickness” for first station: :param rms_per_station dict {station: :type rms_per_station dict {station: float} :param rms_global float: :param train_history dict (loss curves): :param figures dict: :param figure_paths dict:

Examples

>>> agent  = AIInversionAgent(arch="resnet", n_layers=5, epochs=30)
>>> result = agent.execute({
...     "path":       "/data/L22PLT",
...     "output_dir": "/out/ai_inv",
... })
>>> result["rms_global"]
0.24

SYSTEM_PROMPT: str = 'You are an expert in AI-based MT inversion and deep learning for geophysics.\nGiven an AI inversion result, write 4–5 sentences that:\n1. Describe the neural network architecture used and training convergence.\n2. State the prediction quality (RMS, layer count, depth range).\n3. Identify stations where the AI prediction is most / least reliable.\n4. Compare AI results with classical Bostick depth estimates if available.\n5. Recommend next steps (fine-tuning, ensemble, switch to 2-D inversion).\nReply in plain scientific English.\n'#: Override in subclasses to give the LLM its domain expertise.

classmethod from_pretrained(model_name, *, api_key=None, model=None, llm_provider='claude', cache_dir=None, force_download=False)#

Return an AIInversionAgent pre-loaded with a zoo checkpoint.

Parameters:

model_name (str) – Registry name — see list_pretrained().
cache_dir (str or None) – Override default cache ~/.pycsamt/model_zoo/.
force_download (bool) – Re-download even if cached.
api_key (str | None)
model (str | None)
llm_provider (str)

Return type:

AIInversionAgent

Examples

>>> agent = AIInversionAgent.from_pretrained("mt1d-resnet-5layer-v1")
>>> result = agent.execute({"path": "/data/L22PLT"})

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

class pycsamt.agents.Inv2DAgent(*, api_key=None, model=None, llm_provider='claude', n_depth=40, n_freqs=32, n_components=4, arch='unet', n_train_profiles=200, n_stations_per_profile=20, epochs=30)#

Bases: BaseAgent

2-D MT profile inversion using a U-Net convolutional architecture.

Parameters:

api_key (str)
model (str)
llm_provider (str)
n_depth (int) – Number of depth cells in the output section (default 40).
n_freqs (int) – Number of input frequencies (default 32).
n_components (int) – Number of impedance components in input (default 4: Re/Im × xy/yx).
arch (str) – U-Net variant (default "unet").
n_train_profiles (int) – Number of synthetic 2-D profiles for training (default 200).
n_stations_per_profile (int) – Stations per synthetic profile (default 20).
epochs (int) – Training epochs (default 30).
keys (Output data)
----------
path (sites /)
output_dir (str, optional)
period_range ([T_min, T_max], optional)
keys
----------------
ρ (pred_section ndarray (n_depth × n_stations) — log₁₀)
(km) (depths_km ndarray — depth axis)
list[str] (station_names)
float (rms_global)
EMInverter2D (inverter)
dict (figure_paths)
dict

SYSTEM_PROMPT: str = 'You are an expert in 2-D MT inversion using deep learning (U-Net architecture).\nGiven a 2-D AI inversion result, write 4-5 sentences that:\n1. Describe the input pseudosection geometry (stations × frequencies).\n2. Interpret the dominant structural features in the resistivity section.\n3. Assess lateral continuity and compare to classical smoothness-constrained results.\n4. Identify artefacts or stations with poor convergence.\n5. Recommend follow-up (regularisation, 3-D verification, drilling targets).\nReply in plain scientific 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

class pycsamt.agents.Inv3DAgent(*, api_key=None, model=None, llm_provider='claude', n_layers=5, n_freqs=32, n_train_profiles=150, epochs=30, radius=5000.0, hidden=(256, 128, 64), dropout=0.1, n_mc=20)#

Bases: BaseAgent

3-D MT profile inversion using a graph-convolutional network (GCN).

Parameters:

api_key (str)
model (str)
llm_provider (str)
n_layers (int) – Number of depth layers per station (default 5).
n_freqs (int) – Number of frequencies used for feature extraction (default 32).
n_train_profiles (int) – Number of synthetic 3-D training profiles (default 150).
epochs (int) – Training epochs (default 30).
radius (float) – Maximum inter-station edge distance in metres for the adjacency graph (default 5 000 m). Stations farther apart than radius are disconnected in the graph.
hidden (tuple of int) – GCN hidden-layer sizes (default (256, 128, 64)).
dropout (float) – Dropout probability (default 0.1); also used for MC uncertainty.
n_mc (int) – Number of Monte-Carlo dropout passes for uncertainty estimation. Set to 0 to skip uncertainty (faster, default 20).
keys (Output data)
----------
path (sites /)
coords (ndarray (n_stations, 2), optional — station (x, y) in metres.) – Auto-extracted from EDI lat/lon when absent.
adjacency (ndarray (n_stations, n_stations), optional — pre-computed) – normalised adjacency; overrides radius when supplied.
output_dir (str, optional)
period_range ([T_min, T_max], optional)
keys
----------------
(n_sta (adjacency ndarray)
log₁₀ρ (n_layers) —)
(n_sta
(metres) (n_layers-1) — log₁₀h)
(n_sta
std (n_layers) or None — MC-dropout)
(km) (depths_km ndarray — depth axis at station midpoints)
list[str] (station_names)
(n_sta
metres (2) —)
(n_sta
n_sta)
float (rms_global)
GCNInverter3D (inverter)
dict (figure_paths)
dict

Examples

>>> agent = Inv3DAgent(n_layers=5, epochs=20, n_mc=10)
>>> result = agent.execute({
...     "path":       "/data/WILLY_EDIs",
...     "output_dir": "/out/inv3d",
... })
>>> result["rms_global"]
0.28

SYSTEM_PROMPT: str = 'You are an expert in 3-D MT inversion using graph-convolutional deep learning.\nGiven a GCN-based 3-D inversion result, write 4-5 sentences that:\n1. Describe the survey geometry (station count, spatial extent, adjacency radius).\n2. Interpret the dominant 3-D resistivity structures and their spatial continuity.\n3. Assess prediction quality (RMS, depth range) relative to station spacing.\n4. Compare the GCN spatial result to independent 1-D predictions where possible.\n5. Recommend geological follow-up and areas with highest uncertainty.\nReply in plain scientific 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

Agent Modules#

`pycsamt.agents._base`	pycsamt.agents._base
`pycsamt.agents.context`	pycsamt.agents.context
`pycsamt.agents.loader`	pycsamt.agents.loader
`pycsamt.agents.coordinator`	pycsamt.agents.coordinator
`pycsamt.agents.orchestrator`	pycsamt.agents.orchestrator
`pycsamt.agents.qc`	pycsamt.agents.qc
`pycsamt.agents.static_shift`	pycsamt.agents.static_shift
`pycsamt.agents.phase_analysis`	pycsamt.agents.phase_analysis
`pycsamt.agents.tensor_rotation`	pycsamt.agents.tensor_rotation
`pycsamt.agents.tipper_analysis`	pycsamt.agents.tipper_analysis
`pycsamt.agents.freq_decimation`	pycsamt.agents.freq_decimation
`pycsamt.agents.denoising`	pycsamt.agents.denoising
`pycsamt.agents.forward`	pycsamt.agents.forward
`pycsamt.agents.inversion_prep`	pycsamt.agents.inversion_prep
`pycsamt.agents.inversion_eval`	pycsamt.agents.inversion_eval
`pycsamt.agents.inversion_backend`	pycsamt.agents.inversion_backend
`pycsamt.agents.inversion_comparison`	pycsamt.agents.inversion_comparison
`pycsamt.agents.occam2d_agent`	pycsamt.agents.occam2d_agent
`pycsamt.agents.modem_agent`	pycsamt.agents.modem_agent
`pycsamt.agents.inv2d_agent`	pycsamt.agents.inv2d_agent
`pycsamt.agents.inv3d_agent`	pycsamt.agents.inv3d_agent
`pycsamt.agents.ai_inversion`	pycsamt.agents.ai_inversion
`pycsamt.agents.ensemble_agent`	pycsamt.agents.ensemble_agent
`pycsamt.agents.joint_agent`	pycsamt.agents.joint_agent
`pycsamt.agents.model_zoo_agent`	pycsamt.agents.model_zoo_agent
`pycsamt.agents.anomaly_agent`	pycsamt.agents.anomaly_agent
`pycsamt.agents.interpretation`	pycsamt.agents.interpretation
`pycsamt.agents.iot_agent`	pycsamt.agents.iot_agent
`pycsamt.agents.resistivity_map`	pycsamt.agents.resistivity_map
`pycsamt.agents.sensitivity`	pycsamt.agents.sensitivity
`pycsamt.agents.edi_export`	pycsamt.agents.edi_export
`pycsamt.agents.report`	pycsamt.agents.report
`pycsamt.agents.code_gen`	pycsamt.agents.code_gen
`pycsamt.agents.pipeline_agent`	pycsamt.agents.pipeline_agent
`pycsamt.agents.batch_survey`	pycsamt.agents.batch_survey
`pycsamt.agents.hybrid_agent`	pycsamt.agents.hybrid_agent
`pycsamt.agents.mare2dem_agent`	pycsamt.agents.mare2dem_agent
`pycsamt.agents.master`	One-line front door to the pyCSAMT agent stack.
`pycsamt.agents.metrics`	pycsamt.agents.metrics
`pycsamt.agents.package_qa`	pycsamt.agents.package_qa PackageQAAgent — answer free-form questions about the pycsamt v2 package.
`pycsamt.agents.pinn_agent`	pycsamt.agents.pinn_agent
`pycsamt.agents.plotting`	pycsamt.agents.plotting
`pycsamt.agents.router`	pycsamt.agents.router
`pycsamt.agents.tooling`	pycsamt.agents.tooling
`pycsamt.agents.web`	pycsamt.agents.web

IoT Field Agent#

class pycsamt.agents.iot_agent.IoTFieldAgent(*, api_key=None, model=None, llm_provider='claude', method=None, write_manifest=False)[source]#

Bases: BaseAgent

Assess and monitor an IoT-enabled EM field acquisition.

Parameters:

api_key (str, optional) – LLM configuration. When api_key is None the agent runs fully offline and llm_interpretation is None.
model (str, optional) – LLM configuration. When api_key is None the agent runs fully offline and llm_interpretation is None.
llm_provider (str, optional) – LLM configuration. When api_key is None the agent runs fully offline and llm_interpretation is None.
method (str, optional) – EM method hint ("amt", "mt", "csamt", "csem", "tdem", "tem"). Used to seed sessions built from packets and to tag the manifest. When omitted it is inferred from the telemetry.
write_manifest (bool, default False) – When True an AcquisitionManifest is built for every run (and written when manifest_path / output_dir is available).
keys (Output data)
----------
order) (One acquisition source is required (tried in this)
session (FieldSession or mapping) – A live session, or a mapping from to_dict().
packets (iterable of TelemetryPacket or mapping) – Raw telemetry to fold into a fresh session.
sites (path / edis /) – An existing survey to seed a re-occupation session via field_session_from_edis() (no live packets).
deployment (DeploymentConfig, optional) – Declared device capabilities; tabulated via deployment_report().
keys
survey_id (str) – Identifier for sessions built from packets/EDIs (default "iot_survey").
now (float) – Reference epoch seconds for live latency / gap calculations.
output_dir (str) – Directory for saved figures and (when requested) the manifest.
manifest (bool) – Force building the acquisition manifest for this run.
manifest_path (str) – Explicit path to write the manifest JSON.
sign_key (str or bytes) – When given with a manifest, the written manifest is HMAC-signed.
sync_references (mapping) – Per-device reference clocks for batch_assess_sync().
energy_configs (iterable of EnergyConfig) – Device energy budgets for estimate_deployment_energy().
figures (bool, default True) – Whether to render dashboard/edge/power/sync figures.
api (bool) – Passed through to pycsamt table builders (API-object vs raw frame).
keys
----------------

:param session the resolved FieldSession: :param status MonitoringStatus: :param status_table one-row monitoring-status table: :param telemetry_summary packet counts by device and topic: :param packet_table full telemetry packet table: :param station_table registered stations: :param pipeline_input acquisition hand-off dict for the processing flow: :param deployment_table device-capability table (when deployment given): :param sync_table clock-sync table (when sync_references given): :param power_table energy-budget table (when energy_configs given): :param manifest AcquisitionManifest (optional): :param manifest_path written manifest path (optional): :param signature manifest HMAC signature dict (when sign_key given): :param level monitoring level string (ok / warn / critical): :param issues list of issue strings: :param n_packets / n_stations / n_devices: :type n_packets / n_stations / n_devices: int :param figures dict of matplotlib Figure objects: :param figure_paths dict of saved figure paths (when output_dir set):

Examples

>>> agent = IoTFieldAgent()
>>> res = agent.execute({"packets": packets, "output_dir": "/out/iot"})
>>> res["level"]
'warn'
>>> res["status"].issues
['battery_min_v below 11.5 V on 2 device(s)']
>>> res["figures"]["dashboard"]
<Figure ...>

SYSTEM_PROMPT: str = 'You are an expert IoT field-operations analyst for pycsamt v2, supervising a\nlive AMT/MT/CSAMT/CSEM acquisition through its edge telemetry stream.\nGiven a survey monitoring summary, write 3-5 sentences that:\n1. State the overall health of the acquisition (healthy / degraded / critical).\n2. Call out the specific stations, channels, or devices that need field\n attention, and the dominant failure mode (packet loss, high latency,\n low battery, clock drift, edge rejections, sensor dropout).\n3. Explain the most likely field cause (power budget, GPS/clock, comms link,\n contact resistance, powerline or near-field interference).\n4. Recommend the single most important field action to take right now.\nReply in plain English. No bullet points, no markdown headings.\n'#: Override in subclasses to give the LLM its domain expertise.

execute(input_data)[source]#

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