pycsamt.agents.iot_agent#

IoTFieldAgent — AI-assisted monitoring for IoT-enabled AMT / MT / CSAMT / CSEM field acquisition.

Where DataQCAgent reasons about processed EDI data, this agent reasons about the live field campaign: the telemetry stream coming off edge nodes while the survey is still running. It is a thin, defensive orchestration layer over pycsamt.iot that turns whatever acquisition evidence the caller has — a FieldSession, a bag of telemetry packets, or an existing EDI survey to re-occupy — into:

a single MonitoringStatus (OK / WARN / CRITICAL) with the concrete issues that tripped it,
per-topic telemetry, per-station, and pipeline hand-off tables,
optional clock-sync and power-budget summaries,
an optional reproducible AcquisitionManifest (written and, on request, signed),
dashboard / edge-QC / power / sync figures rendered with the shared pycsamt plotting configuration, and
a plain-English field-ops interpretation from the configured LLM.

Every sub-step is wrapped so that a failure in one diagnostic degrades to a warning rather than sinking the whole assessment — the agent always returns a usable AgentResult.

Wraps pycsamt.iot:

Classes

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

Assess and monitor an IoT-enabled EM field acquisition.

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