Industrial-AIOps Energy

[READ][risk=low][PERSIST] Adopt + persist the canonical alias map for a site. Writes a local owner-only advisory JSON file (NOT an OT-device write — hence risk=low); see the persistence note below. Runs the cross-protocol asset model over ``feeds``, extracts the adopted map ``{canonical_alias: {ref, protocol, asset, name, class}}``, and persists it as the site's baseline (owner-only JSON under the iaiops home). Re-running overwrites the baseline. Advisory — the map is a SUGGESTION, never a server-side rename (OT-dangerous). Args: feeds: Per-protocol tag feeds ``[{protocol, source, asset?, tags:[...]}]``, the SAME shape ``cross_protocol_asset_model`` takes. site: Site label (a safe file leaf: alphanumeric/_/-). Default 'site'. Returns dict: {site, path, tag_count, adopted:{alias: {...}}}. Example: adopt_alias_map(feeds=[{"protocol":"opcua","source":"l1","tags":[...]}], site="plant").

[READ][risk=low] ISA-18.2 alarm-flood analysis over a list of alarm events. Args: events: Alarm/condition events — {source, timestamp (ISO-8601), priority?, state? (ACTIVE/RTN/ACK)}. window_minutes: Analysis window; omitted → inferred from event timestamps. chatter_window_s: A source with >=3 transitions inside this window chatters. standing_s: An alarm active longer than this is 'standing/stale' (default 24h). top_n: How many top offenders to return. Returns dict: {event_count, window_minutes, alarms_per_hour, isa_18_2:{ok_max:6, manageable_max:12, flood_min:30}, flood_verdict ('ok'|'manageable'|'over_target'|'flood'), priority_distribution, pareto_sources_for_80pct, top_offenders:[{source, count, share_pct, chattering, standing}], chattering:[...], standing:[...]}. Example: alarm_bad_actors(events=[{"source":"FIC101","timestamp":"...", "priority":"high"}, ...]).

[READ][risk=low] Collapse an alarm flood into cascades + each cascade's first-out root. Answers "which alarm to look at first" when 100+ alarms hit in minutes: groups annunciations into cascades (a new cascade starts after a quiet gap > window_s) and reports the FIRST-OUT alarm (earliest in the burst) as the likely root, plus downstream members and any chattering sources. First-out is a transparent heuristic cited by timestamp — NOT causal (use downtime_root_cause for causality). Pass 'events' for pure analysis, or an endpoint to collect live via the OPC-UA active-condition scan. Read-only; bounded. Args: endpoint: Endpoint name from config (used only when events is omitted). duration_s: Live collection window in seconds (1..300, default 60). window_s: Quiet gap (seconds) that separates one cascade from the next (default 60). min_cascade: Minimum annunciations for a group to count as a cascade (default 2). events: Injected alarm events — {source, timestamp (ISO-8601), state?}; skips live collect. Returns dict: {cascade_count, total_activations, cascades:[{root:{source, ts}, size, distinct_sources, span_s, members[], chattering[]}], collected?}. Example: alarm_cascade(events=[{"source": "PT101", "timestamp": "2026-06-28T10:00:00Z"}, ...]).

[READ][risk=low] Collapse ten phrasings of one fault into one row. `alarm_bad_actors` ranks by SOURCE, which answers "which instrument is noisiest" and not "which fault is noisiest". A plant that words one condition ten ways — `PT-101 HIGH`, `PT-102 HIGH`, `PT-103 high alarm` — gets ten bad actors and no sign that they are one problem, so a rationalization meeting works the list top-down and fixes the same thing three times. This groups the same events by what they SAY instead of by who said it. Clustering is **exact equality of a normalized string, not similarity**: case, punctuation and embedded numbers are removed, and what remains must match exactly. That is deliberately dumber than it could be, and it is why the result needs no model and can be checked — two messages land together only when they are literally the same sentence with the identifiers taken out. Every cluster carries the distinct wordings and sources it merged, so you can see what was combined. It does **not** claim two differently-worded alarms mean the same thing; a person decides that. Events carrying no message text are counted separately and excluded from the shares, rather than being lumped together as one type. Args: events: [{source?, message|description|text|condition|type, ...}]. top_n: Clusters returned, largest first (default 20, capped at 100). min_count: Only report clusters with at least this many events (default 1 — a one-off is still reported, not tidied away). Returns dict: {events_supplied, events_clustered, events_without_text, cluster_count, collapsed_count, clusters:[{signature, count, share_pct, distinct_wordings, distinct_sources, variants:[{text, count}], sources:[{source, count}]}], note}. Example: alarm_event_clusters(events=[{"source":"PT-101","message":"pressure HIGH"}, {"source":"PT-102","message":"Pressure high!"}]).

[READ][risk=low] ISA-18.2 deep alarm-flood analysis: episodes + chattering + stale + advice. Deepens alarm_bad_actors: detects flood *episodes* (start/end/count/peak rate/ top contributors + each episode's first-out annunciation, per ISA-18.2's >=10 alarms per 10 min per operator), alarms chattering ACTIVE↔CLEARED, standing/ stale alarms, and percent-time-in-flood vs the ISA-18.2 targets (~1-2 alarms/ 10 min steady state, <1% time in flood). Also returns an ISA-18.2 'load_profile' (per-bucket rate band + peak period + trend) and per-source 'suppression_advice' (deadband/on-off-delay for chatter, time-limited shelve for standing alarms). The suppression advice is ADVISORY ONLY — starting values for a human to review and approve via your ISA-18.2 / management-of-change process; this tool never applies suppression, shelving, deadband, or delay changes. Pass 'events' for pure analysis, or an endpoint to collect live via the same OPC-UA active- condition scan the RCA copilot uses (polled over duration_s; other protocols contribute no alarms). Output is bounded; 'truncated' flags say when caps bit. Args: endpoint: Endpoint name from config (used only when events is omitted). duration_s: Live collection window in seconds (1..300, default 60). window_s: Flood analysis window in seconds (ISA-18.2 default 600). threshold: Annunciations per window that start a flood (default 10). events: Injected alarm events — {source, timestamp (ISO-8601), state? (ACTIVE/RTN/CLEARED)}; skips live collection entirely. stale_after_s: Continuously-active age that marks a standing alarm (default 24h). max_episodes: Cap on returned flood episodes (default 20). max_rows: Cap on chattering / stale / suppression-advice / worksheet rows (default 50). load_bucket_s: Load-profile bucket width in seconds (ISA-18.2 default 600 = 10 min). Returns dict: {event_count, summary:{insufficient_data, percent_time_in_flood, avg_alarms_per_10min, peak_alarms_per_…

[READ][risk=low] ISA-18.2 alarm-rationalization worksheet (CSV or inline rows). One row per alarm source, count-descending: count, % of total annunciations, chattering?, flood contributor?, and a recommendation stub — the starting document for an ISA-18.2 rationalization review. Pass 'events' for pure analysis, or an endpoint to collect live via the same OPC-UA active-condition scan the RCA copilot uses. With out_path the full worksheet is written as CSV and the path returned; otherwise bounded inline rows (truncation noted). Args: endpoint: Endpoint name from config (used only when events is omitted). duration_s: Live collection window in seconds (1..300, default 60). events: Injected alarm events — {source, timestamp (ISO-8601), state?}. window_s: Flood analysis window in seconds (ISA-18.2 default 600). threshold: Annunciations per window that start a flood (default 10). out_path: Optional CSV destination; parent directory must exist. Returns dict: {row_count, columns:[alarm_id, count, pct_of_total, chattering, in_flood, recommendation], csv_path? , rows?:[...], truncated (bool)}. Example: alarm_rationalization_worksheet(events=[...], out_path="worksheet.csv").

[DEPRECATED → opcua_anomaly_scan][READ][risk=low] Statistical outlier scan. Samples a node over a bounded window and flags statistical outliers. Computes mean/stddev/min/max and flags samples outside mean ± sigma*stddev. Simple statistics only — no ML, no persisted model. Args: node_id: The OPC-UA node id to scan. endpoint: Endpoint name from config. samples: Max samples (capped server-side). interval_ms: Delay between samples in milliseconds. sigma: Outlier band width in standard deviations.

[READ][risk=low] Actively fingerprint endpoints into an asset register. Connects to each target with our own protocol client and reads its identity call (S7 CPU info, EtherNet/IP controller info, OPC-UA server build info, Modbus device identification FC43, Mitsubishi CPU type, MTConnect device model), aggregating vendor/model/firmware/serial per device. Honest scope: ACTIVE fingerprinting (we connect to each device), NOT passive SPAN/tap discovery. Only finds devices we are configured to reach. Args: endpoints: Endpoint names to fingerprint; omit to fingerprint ALL configured endpoints. Returns dict: {asset_count, reachable_count, unreachable_count, method: 'active_fingerprint', assets:[{endpoint, protocol, address, vendor, model, firmware, serial, reachable, last_seen, error}]}. Example: asset_inventory(endpoints=["press1","cell5"]).

[READ][risk=low] Check recent local samples against the learned baseline. Reads the last window_s seconds from ~/.iaiops/data.db (no device I/O) and judges them against the stored band. Conservative by design: a violation is reported ONLY when values are beyond p1/p99 by more than 3×MAD AND sustained for >=3 consecutive samples — a single spike is never flagged. Every violation cites the baseline window (from/to ts, n samples), the band values, and the offending samples' timestamps/values. No stored baseline → an explicit no_baseline answer (never a guess). Bounded output (<=10 violations, <=20 cited samples each). Args: tag: Tag name to check, e.g. 'line1.temp'. endpoint: Only samples from this endpoint label. window_s: Recent window to check, seconds (60..604800; default 3600). Returns dict: {status: 'ok'|'violation'|'no_baseline', tag, checked_samples, thresholds, baseline_citation, violations:[{direction, from_ts, to_ts, consecutive_samples, samples:[{ts,value}], baseline}], note}. Example: baseline_check(tag="line1.temp", window_s=7200).

[READ][risk=low] Check readings against the band for ONE declared context. A reading whose context was never learned comes back `unknown_context` and stops there. It is **not** compared against the global band or the nearest one, and that refusal is the entire point of the tool: borrowing a band turns "we have never seen this regime" into "this regime is normal" — a silent pass, in the direction nobody reports. The response lists the contexts that do have bands so the gap is actionable. Otherwise the usual conservative rules apply: a violation needs values beyond the band by more than `margin_mad` × MAD AND sustained over `sustain_n` consecutive samples, and every flag cites the baseline it was judged against. Args: samples: [{ts, value, ...}] readings to check. contextual: A `baseline_learn_contextual` result. context: Which declared context these readings belong to. margin_mad: MAD margin beyond the band before flagging (default 3.0). sustain_n: Consecutive samples required (default 3 — no single-spike flags). Returns dict (known context): the `baseline_check` shape plus {context, context_key}. (unknown): {status:"unknown_context", tag, context, known_contexts, checked_samples, reason, note}. Example: baseline_check_in_context(samples=[...], contextual={...}, context="recipe-B").

[READ][risk=low] Learn a conservative per-tag normal band from local history. Source is ~/.iaiops/data.db — the local store written by historian_push(sink="sqlite") — NOT a live device read. Learns robust percentiles (p1/p99 + median/MAD, no ML) from the tag's own samples, segmented at the latest change recorded via baseline_record_change (the band reflects only the post-change regime). REFUSES with an explicit insufficient_data verdict (listing exactly what is missing) below 100 usable samples or under 24h of span — it never invents a band from thin data. On success the band is persisted to ~/.iaiops/baselines.json (owner-only local metadata, not an OT write). Args: tag: Tag name to learn, e.g. 'line1.temp'. endpoint: Only samples from this endpoint label. since: Only samples at/after this ISO-8601 time. Returns dict: {status: 'ok'|'insufficient_data', tag, band:{p1,p99,median,mad}, n_samples, window:{from_ts,to_ts,span_s}, segment, missing?:[...], note}. Example: baseline_learn(tag="line1.temp", since="2026-06-01T00:00:00").

[READ][risk=low] Learn one conservative band per declared context, not one per tag. One band per tag is wrong the moment a tag has more than one normal. A dryer running recipe A at 180 °C and recipe B at 240 °C gets a band spanning both, after which neither regime can go wrong — the band is too wide to catch a real excursion and too mixed to mean anything. OT normal ranges move with shift, product/recipe, and start-up versus steady state. **The context is declared, never inferred** (D16). Each sample carries a label under `context_key`; nothing here guesses which shift a timestamp falls in or clusters values into regimes it then treats as real. Each context is handed to the same learner as a global baseline, so it refuses on the same terms — a thin context is left without a band rather than borrowing another context's samples. Samples with no label are counted and named, not pooled into a default bucket, because a default bucket is that same fallback. Pass samples in (as with `spc_check` / `tag_health`). The local store's `samples` table has no context column, so there is deliberately no `iaiops baseline learn --context` yet; wiring one is a schema change and is not done. Args: samples: [{ts, value, quality?, tag?, <context_key>}] rows. tag: The tag being learned. context_key: Field that declares the context (default "context"). min_samples: Per-context minimum before a band is learned (default 100). min_span_s: Per-context minimum history span in seconds (default 86400). Returns dict: {tag, context_key, contexts:{label: learn_baseline result}, learned_contexts, refused_contexts, uncontexted_samples, note}. Example: baseline_learn_contextual(samples=[{"ts":"...","value":181.0, "context":"recipe-A"}], tag="dryer.temp").

[READ][risk=low] Record an operator change-log entry for a tag (local only). Writes ONLY local metadata (~/.iaiops/baselines.json, owner-only) — never an OT device write, hence risk=low. A recorded change (setpoint moved, valve replaced, probe swapped) marks a regime boundary: the next baseline_learn uses only samples AFTER the latest change, so the band never mixes pre-change and post-change behavior. This operator change log — not a black-box score — is what makes the baseline trustworthy. Args: tag: Tag whose process changed, e.g. 'line1.temp'. note: What changed (required), e.g. 'setpoint 60→70C'. Returns dict: {tag, change:{ts, note}, changes_recorded}. Example: baseline_record_change(tag="line1.temp", note="setpoint 60→70C").

[READ][risk=low] Baseline status for one tag, or a bounded listing of all. Read from the local store only (no history scan, no device I/O) and never guesses: 'no_baseline' (nothing learned, no refused attempt), 'learning' (last learn refused — still accumulating history), 'ok' (band learned, last check clean), 'violation' (last check flagged a sustained excursion). With no tag, lists every tracked tag (bounded to 100 entries). Args: tag: Optional tag name; omit to list all tracked tags. Returns dict: {tag, status, band?, baseline_window?, changes_recorded?, ...} for one tag, or {tracked_tags, listed, truncated, tags:[...]} for all. Example: baseline_status(tag="line1.temp").

[READ][risk=low] 等保 2.0 二级 vs 三级 per-pillar deltas + honest iaiops posture. 等保 2.0 (GB/T 22239) is graded — the same control tightens as the level rises. Per governance pillar this shows the 二级 baseline, what 三级 additionally requires, and how far iaiops moves you toward it (with the honest per-control status/gap). An onboarding/self-assessment aid, NOT a certification. Args: level: Focus on one level — 'l2'/'l3', '二级'/'三级', or '2'/'3'. Omit for both. Returns dict: {framework, levels:[{id,name,note}], selected_level, pillar_count, deltas:[{pillar, l2_requires?, l3_adds?, iaiops, iaiops_status, gap}], note}. Example: compliance_dengbao_levels(level="三级").

[READ][risk=low] Export the audit-evidence bundle (zip) for an auditor. Packages the governance evidence trail into one deterministic zip: audit_rows.jsonl (secrets already redacted upstream), chain_verification.json (SHA-256 hash-chain walk result), rules.yaml (if present), doctor_summary.json (non-probing config/secret-store facts), and manifest.json. Path is validated (no '..' traversal; parent created 0700). Args: out_path: Destination zip path (must end in .zip). since: Optional ISO-8601 floor on the audit row timestamp (inclusive). until: Optional ISO-8601 ceiling on the audit row timestamp (inclusive). Returns dict: {path, row_count, chain{ok, checked, unhashed, ...}, files[], since, until}. Example: compliance_evidence_bundle(out_path="/tmp/evidence.zip", since="2026-06-01T00:00:00+00:00").

[READ][risk=low] 跨框架对照: 防护指南 ↔ 等保 2.0 (GB/T 22239) ↔ IEC 62443. One row per governance pillar, showing the matching 《工控系统网络安全防护指南》 requirement, 等保 2.0 control class, IEC 62443 foundational requirement, and the current iaiops status. Companion to compliance_mapping (which carries the honest per-control gap); use this to answer "which 等保 / 62443 clause does this satisfy". Returns dict: {frameworks:[{id,name,region,kind}], framework_count, pillar_count, crosswalk:[{pillar, gjzn, dengbao, iec62443, iaiops_status}], note}. Example: compliance_frameworks().

[READ][risk=low] 《工控系统网络安全防护指南》 ↔ iaiops governance mapping. An honest onboarding/sales self-assessment across the pillars 分区隔离 / 可审计 / 双向认证 / 最小权限 / 数据保护 / 自主可控. Each control names how iaiops addresses it, an honest status (addressed / partial / 待核实), and the remaining gap. Returns dict: {framework, frameworks[], pillars[], control_count, status_summary {addressed, partial, 待核实}, controls:[{pillar, requirement, iaiops, status, gap, crosswalk{dengbao, iec62443}}]}. See compliance_frameworks for the full cross-framework 对照. Example: compliance_mapping().

[READ][risk=low] Render the 等保 2.0 / IEC 62443 compliance report (Markdown). Turns the compliance crosswalk into a deliverable document a CISO can read: title-page metadata (site / date / iaiops version), per-pillar 等保 L2/L3 status table, IEC 62443 FR1–6 crosswalk, honest gap list, and a governance-controls appendix (audit hash chain / approval tokens / dry-run+undo / mTLS). An onboarding/self-assessment aid, NOT a certification. Args: level: 等保 2.0 target level — 'l2'/'l3', '二级'/'三级', '2'/'3'. Omit for both. site: Site / plant name stamped on the title page. out_path: Optional file to write the markdown to (.md). Required when the report exceeds the inline bound (~400 lines): without it the inline markdown is truncated with a note. Returns dict: {format, level, line_count, path?} plus either the full inline {markdown} (when within bounds and no out_path) or {markdown (truncated), truncated: true} with a hint to pass out_path. Example: compliance_report(level="三级", site="示例水厂", out_path="/tmp/compliance-report.md").

[READ][risk=low] Fuse per-protocol tag feeds into ONE unified asset model. Unifies the two per-protocol tag models (OPC-UA address-space discovery + Modbus register templates) into one cross-protocol asset/tag/alias model. Tags are re-classified with the SAME semantic classifier the OPC-UA layer uses, grouped into assets ACROSS protocols (a ``Line1`` OPC-UA folder + a ``Line1`` Modbus block become one asset), and each gets a canonical alias ``<site>.<asset>.<class_or_name>``. Advisory only — aliases are SUGGESTIONS, never a server-side rename (OT-dangerous). Args: feeds: List of per-protocol feeds, each ``{protocol, source, asset?, tags:[...]}``. ``tags`` may be OPC-UA discovery descriptors (from opcua_discover_tags), Modbus template tags (from modbus_apply_template), or already-normalized tags. A feed-level ``asset`` is applied to its tags that don't carry their own. site: Site prefix for canonical aliases (default 'site'). Returns dict: {site, protocols, tag_count, asset_count, assets:[{asset, protocols, tag_count, classes, tags:[{protocol, source, name, ref, asset, unit, klass, canonical_alias, suggested_alias}]}], naming_quality: {alias_collisions, cross_protocol_overlaps, cryptic_names, verdict}}. Example: cross_protocol_asset_model(feeds=[ {"protocol":"opcua","source":"line1","tags":[...]}, {"protocol":"modbus","source":"meter1","asset":"Line1","tags":[...]}], site="plant").

[READ][risk=low] Cross-endpoint fleet rollup of data-TRUST: worst tags + bad quality. Builds on data_quality_scorecard to give a fleet-wide view: endpoints ranked by their single worst tag, bad-quality tag counts aggregated across every endpoint, and a first-class liveness rollup (dead-heartbeat / flatline). Staleness and gap budgets are configurable per tag (staleness_s / gap_threshold_s) and per feed, so a slow daily counter is not judged like a 1Hz sensor. Pure analysis. Args: feeds: Per-endpoint feeds — {endpoint, staleness_s?, tags:[{ref, label?, samples:[scalars or {value, good|quality, timestamp?}], expected_update_s?, staleness_s?, gap_threshold_s?, flatline_after_s?, heartbeat?}]}. default_staleness_s: Fallback max sample-age (seconds) before 'stale' when a tag/feed sets no staleness_s/expected_update_s (default 300). now: ISO-8601 reference time for staleness (deterministic); omit for now-UTC. top_n: How many endpoints / bad-quality rows to return (default 10). Returns dict: {evaluated_endpoints, evaluated_tags, fleet_score (0-100), fleet_status, endpoints_ranked_by_worst_tag:[...], bad_quality_rollup: {total_bad_quality_tags, endpoints_affected, by_endpoint:[{endpoint, bad_quality_tags, fully_bad, partial_bad}]}, liveness_rollup: {dead_heartbeat_count, flatline_count, dead_heartbeats[], flatlines[]}, issue_breakdown{}}. Example: data_quality_fleet_rollup(feeds=[{"endpoint":"line1","tags":[{"ref":"t", "samples":[{"value":None,"good":false}]}]}]).

[READ][risk=low] Fleet data-TRUST scorecard across endpoints' tag feeds. Scores each tag 0-100 on whether its data can be BELIEVED — staleness, dead heartbeat, bad-quality, flatline, gaps, anomaly — then rolls up per endpoint and across the fleet. NOT process health (it does not score whether a value is alarming, only whether it is trustworthy). Pure analysis over provided feeds. Args: feeds: Per-endpoint feeds — {endpoint, tags:[{ref, label?, samples:[scalars or {value, good|quality, timestamp?}], expected_update_s?, heartbeat?}]}. default_staleness_s: Max sample-age before 'stale' when a tag sets no expected_update_s (default 300). now: ISO-8601 reference time for staleness (deterministic); omit for now-UTC. Returns dict: {evaluated_endpoints, evaluated_tags, fleet_score (0-100), fleet_status, issue_breakdown{}, worst_endpoints[], worst_tags[], endpoints:[{endpoint, score, status, status_counts, worst_tag}]}. Example: data_quality_scorecard(feeds=[{"endpoint":"line1","tags":[{"ref":"hb", "heartbeat":true,"samples":[5,5,5,5]}]}]).

[READ][risk=low] Which scanned devices fall inside a mounted advisory's stated range. `iaiops scan` already reads vendor / model / firmware / serial and then does nothing with them. This closes that loop — and deliberately stops short of where a vulnerability scanner would go. **It reports that a device falls inside an advisory's stated range. Nothing more.** Not "vulnerable", not "exploitable", no severity score. Whether a published issue is reachable on a particular machine depends on configuration, network position and compensating controls that a read-only scan cannot see — and in OT most advisories against a protocol stack are not findings at all, because the stack is not reachable from anywhere that matters. A report full of red text that ignores that gets switched off by the site, and then the real one is missed too. **No database ships with this.** A bundled CVE feed is a maintenance commitment this repo has not made, and a stale one that looks current is worse than none — so the library is a file the site controls, which also makes it work air-gapped. Every entry must carry a source, and one bad entry refuses the whole file rather than half-mounting it. Four verdicts, and the middle two are the point: `in_affected_range`, `version_unknown` (model matches, no firmware read — neither a hit nor a pass), `version_unparsed` (a firmware string it will not invent an ordering for), `not_affected`. A device no advisory mentions is **absent** from the findings, not reported clean: "nothing known" is not "nothing there". Args: devices: [{ip?, vendor, model, firmware?}] — e.g. the `hosts` of a scan. library_path: Path to the advisory file (YAML or JSON) this site mounted; entries are {id, vendor, model, source, affected_below|affected_from| affected_versions, title?}. Returns dict: {devices_checked, advisories_mounted, devices_with_findings, summary:{in_affected_range, version_unparsed, version_unknown, not_affected}, findings:[{i…

[READ][risk=low] Localize a 'no data' break across an endpoint's reachable hops. Probes connect → read(ref) → freshness → variance and returns a verdict with per-hop detail and a recommended action. The #1 OT triage: distinguishes "cannot connect" (network/PLC down) from "comms OK but value stale" (upstream/field/source) from "good status but flatline" (sensor stuck). Args: endpoint: Endpoint name from config (any protocol). ref: Tag/node/address/device to read (OPC-UA node id, Modbus address, S7 address string, MELSEC device). Omit to test connectivity only. freshness_threshold_s: Max value-age (seconds) before 'stale' (default 60). series: Optional injected samples (scalars or {value,timestamp}) for flatline/variance reasoning when a live historian is out of reach. flatline_eps: Spread at/below which a series counts as flatline. Returns dict: {verdict ('cannot_connect'|'comms_ok_value_unreadable'| 'comms_ok_bad_quality'|'comms_ok_value_stale'|'comms_ok_flatline'| 'healthy'), diagnosis, recommended_action, hops:[{hop, ok, detail}]}. Example: diagnose_dataflow(endpoint="line1", ref="ns=2;i=5", freshness_threshold_s=30).

[READ][risk=low] Diff a fresh discovery run against the adopted baseline. Loads the site's previously adopted alias map, re-runs the cross-protocol asset model over ``feeds``, and reports how the address space moved: tags added / removed / renamed (same ref, new alias) / reclassified (same ref+alias, new semantic class), plus a stable|changed verdict. Adopt a baseline first with ``adopt_alias_map``. Args: feeds: Fresh per-protocol tag feeds (same shape as adopt_alias_map). site: Site label whose baseline to diff against. Default 'site'. Returns dict: {site, verdict, counts:{added,removed,renamed,reclassified}, added:[...], removed:[...], renamed:[...], reclassified:[...]}. Example: diff_alias_map(feeds=[{"protocol":"opcua","source":"l1","tags":[...]}], site="plant").

[READ][risk=low] Class 0/1/2/3 integrity poll → the outstation's database. Returns all static points grouped by measurement type (binary_input, analog_input, counter, …). Args: endpoint: Endpoint name from config (protocol 'dnp3'). Returns dict: {endpoint, outstation_address, point_count, by_type{}, points:[{type, group, index, value, quality, timestamp}]}. Example: dnp3_integrity_poll(endpoint="rtu2").

[READ][risk=low] Bring the DNP3 master online and report link/outstation status. Args: endpoint: Endpoint name from config (protocol 'dnp3'); omit for default. Returns dict: {endpoint, host, port, outstation_address, master_address, online}. Example: dnp3_link_status(endpoint="rtu2").

[READ][risk=low] Which stoppage started it, and which ones were downstream of it. RCA weights evidence by TIME alone — a signal before onset counts, one after counts less. That is the honest half of the axis. Run `downtime_root_cause` per asset after one upstream stop and every downstream machine comes back with its own confident local root cause: each internally consistent, each citing real signals, and all but one about a machine that stopped because it was starved. The distinguishing fact is not in the evidence — it is the line's topology. Two rules decide an attribution and both must hold: the candidate must be **declared upstream** of the asset, and it must have **stopped first**. An upstream asset that stopped later cannot have caused an earlier stop, however upstream it is. Topology is declared, never inferred (D25). Co-occurrence on a production line is guaranteed — everything stops together — so mining it for edges would manufacture the causality this exists to remove. With no relations declared every row comes back `not_evaluable` and the reason names the command that fixes it. Assets the topology does not connect are left `unattributed` rather than folded into the origin's column. This ranks the stoppages; it does not diagnose the origin. Run `downtime_root_cause` on the origin asset for that. Args: stoppages: [{asset, start, end?}] for one incident window (ISO-8601). site: Which declared line topology to use (default "default"). max_lead_s: A downstream stop is attributed only if it began within this many seconds of the upstream one (default 900). Returns dict: {site, stoppages_evaluated, relations_declared, max_lead_s, verdict ('origin'|'multiple_origins'|'unattributed'|'not_evaluable'), origins, consequence_count, attributions:[{asset, start, status, origin_asset?, hops_upstream?, lead_s?, explains?, detail}], advisory}. Example: downtime_attribution(stoppages=[{"asset":"filler","start":"2026-01-05T06:00:00Z"},…

[READ][risk=low] Detect running→stopped transitions and categorize stoppages. Args: series: Timestamped samples — {timestamp (ISO-8601), state} where state is a string (RUNNING/IDLE/FAULT…), a bool, or a number. category_map: Optional {state_label: category} override (else keyword heuristics map to changeover/material/mechanical/quality/break/unknown). min_duration_s: Ignore stoppages shorter than this (seconds). Returns dict: {samples, event_count, total_downtime_s, by_category:{cat: {count, downtime_s}}, events:[{start, end, duration_s, state, category}]}. Example: downtime_events(series=[{"timestamp":"2026-06-28T08:00:00Z","state":"RUNNING"}, {"timestamp":"2026-06-28T08:05:00Z","state":"FAULT"}, ...]).

[READ][risk=low] AI downtime root-cause copilot — cited verdict, ADVISORY only. Correlates whatever evidence you supply around a downtime/incident window — alarm events, tag samples, a diagnose_dataflow verdict, a machine-state series — ranks candidate root causes, and cites the REAL signals behind each. Read-first: it proposes a human-approved, undoable (MOC-gated) action but executes nothing. Anti-hallucination: only signals present in the input are cited; thin evidence downgrades to 'insufficient_evidence' with a 'recommended_next_data' list rather than a confident guess. Confidence combines independent, time-correlated evidence (signals BEFORE onset outweigh signals during it). Args: window: {start (ISO-8601), end? (ISO-8601), asset?, category?}. If 'end' is omitted but state_series is given, the first running→stopped span bounds it. alarms: Alarm/condition events — {source, timestamp, message?, priority?, state?}. tags: Per-tag samples — {ref, samples:[scalars or {value, good|quality}], warn_high?, alarm_high?, ...} (scored via tag_health). dataflow: A diagnose_dataflow result dict (its 'verdict' localizes comms vs field). state_series: {timestamp, state} samples to bound the window if 'end' is absent. lead_window_s: How far before onset a signal may sit and still count as a cause (default 300s); signals after onset are treated as consequences. cause_weights: Optional per-site {cause: multiplier} override (e.g. from learn_cause_weights) — scales each cause's evidence (1.0 = neutral default) before the noisy-OR. Unknown causes / non-numeric weights are rejected; values are clamped. Omit for the shipped default weighting. include_graph: When true, also return a 'graph' block — the SAME verdict re-projected as a causal graph {nodes, edges, mermaid, meta} (signal → cause → downtime) for a frontend/Grafana. Pure re-shape: signal→cause edge weights are the evide…

[READ][risk=low] AI downtime RCA copilot that GATHERS its own live evidence. Same advisory, read-only, evidence-cited contract as downtime_root_cause — but instead of hand-injecting evidence you give an endpoint + incident window and it pulls the evidence itself: a cross-protocol diagnose_dataflow probe, a short sampled series per ref (so flatline/bad-quality/anomaly surface via tag_health), and active OPC-UA conditions. Light read load; non-destructive; nothing executed. The gathered bundle is echoed under 'collected_evidence' (no hidden inputs). Args: endpoint: Endpoint name from config (any protocol). Omit for the default. window: {start (ISO-8601), end?, asset?, category?, freshness_threshold_s?}. refs: Tags/nodes/addresses to sample for this incident (first is also the diagnose_dataflow target). Capped at 20. sample_count: Reads per ref to build its series (1..60, default 8). interval_ms: Delay between reads (>=50ms, default 200). include_alarms: Surface active OPC-UA conditions as alarm evidence (OPC-UA only). lead_window_s: Causal lead window before onset (default 300s). include_graph: When true, also return the 'graph' block (same {nodes, edges, mermaid, meta} causal-graph re-projection as downtime_root_cause). Pure re-shape of the verdict; no new reasoning. Omit for the flat verdict. Returns dict: same shape as downtime_root_cause plus 'collected_evidence' {endpoint, protocol, refs_sampled, alarms_found, dataflow_verdict}. Example: downtime_root_cause_live(endpoint="line1", window={"start":"2026-06-28T10:00:00Z","asset":"line1"}, refs=["ns=2;i=5","ns=2;i=6"]).

[READ][risk=low] One-call downtime triage: first-look alarm + RCA cause + precursors. Answers the operator's three simultaneous questions on a stopped line — which alarm to look at first, the likely cause, and whether anything warned us — then cross-checks whether the first-out alarm agrees with the RCA verdict. Composes alarm_cascade + downtime_root_cause + pdm_forecast over ONE incident; every field traces to a sub-report echoed under 'cascade'/'rca'/ 'precursor_forecasts'. Read-first and advisory: it proposes but executes nothing. Thin evidence downgrades honestly rather than guessing. Args: window: {start (ISO-8601), end?, asset?, category?}. If 'end' is omitted but state_series is given, the first running→stopped span bounds it. alarms: Alarm/condition events — {source, timestamp, message?, priority?, state?}. Feeds BOTH the first-out cascade and the RCA. tags: Per-tag samples — {ref, samples:[...], warn_high?, ...} (via tag_health). dataflow: A diagnose_dataflow result dict (localizes comms vs field). state_series: {timestamp, state} samples to bound the window if 'end' is absent. precursors: Signals to check for a pre-incident trend — [{signal, series: [scalars or {value, timestamp}], warn_high?, alarm_high?, warn_low?, alarm_low?}]; each is run through pdm_forecast and kept only when it was degrading/imminent before the trip. cascade_window_s: Quiet gap (s) separating alarm cascades (default 60). lead_window_s: Causal lead window before onset (default 300s). cause_weights: Optional per-site {cause: multiplier} RCA override. imminent_within_s: ETA horizon that marks a precursor 'imminent' (default 24h). include_graph: When true, the echoed 'rca' sub-report also carries a 'graph' block — the SAME verdict re-projected as a causal graph {nodes, edges, mermaid, meta} (signal → cause → downtime) for a frontend. Pure re-shape; no new reasoning. Omit to kee…

[READ][risk=low] Export collected samples from the LOCAL SQLite sink to a file. Source is ~/.iaiops/data.db — the local queryable store written by historian_push(sink="sqlite") — NOT a live device read. Writes csv (Excel), sqlite (SQL browser / Power BI) or parquet (pandas/Spark; needs pip install 'iaiops[export]'), and returns the file path + row count with a bounded inline preview (first 200 rows max) so the response never floods. Args: fmt: 'csv' | 'sqlite' | 'parquet'. since/until: Optional ISO-8601 time bounds (inclusive). endpoint: Only samples from this endpoint label. tag: Only samples for this tag. limit: Max rows exported (1..100000; default 10000). out_path: Output file; default ~/.iaiops/exports/iaiops-export-<ts>.<ext>. Returns dict: {format, path, rows, preview_rows:[{ts, endpoint, protocol, tag, value, quality, unit}] (≤200), preview_truncated}. Example: export_data(fmt="csv", tag="line1.temp", since="2026-07-01T00:00:00").

[READ][risk=low] Roll up active RCA incidents across sites → fleet-wide top causes. Aggregates the incidents each site reports into a fleet picture: how many incidents, which sites are affected, and the most common root causes across the whole fleet. Read-only; no device I/O. Args: sites: Per-site reports carrying incidents: [{site, incidents:[{cause|primary_cause, confidence?}]}]. Returns dict: {total_incidents, sites_with_incidents, affected_sites[], top_causes[]}. Example: fleet_incidents(sites=[{"site":"plant-sh","incidents":[{"cause":"network"}]}]).

[READ][risk=low] Roll up per-site status reports into one fleet health view. The tier above data_quality_fleet_rollup (per-endpoint within one site): this aggregates across many edge SITES for central management. A site is 'offline' if its last_seen is older than stale_after_s; fleet_status is the worst site status present. Read-only, pure; no device I/O. Args: sites: Per-site reports, each [{site, location?, profile?, status?, score?, issues?, last_seen?}]; status ∈ ok|degraded|critical|offline (else derived from score); score 0..1. stale_after_s: A site with no report newer than this is 'offline' (default 300). now: Optional ISO-8601 'now' for deterministic staleness (default: current UTC). Returns dict: {site_count, fleet_status, fleet_score, by_status, worst_sites[], sites[]}. Example: fleet_status(sites=[{"site":"sh","score":0.9},{"site":"bj","status":"critical"}]).

[DEPRECATED → opcua_health_summary][READ][risk=low] Classify OPC-UA tags. Classifies tag node-ids against warn/alarm thresholds. Returns ok/warn/alarm/unknown counts plus the offending tags. Thresholds come from config tags, or per-ref overrides in ``thresholds``. Args: endpoint: Endpoint name from config. node_ids: Tag node ids to evaluate; omit to use configured tags. thresholds: Optional {ref: {warn_high, alarm_high, warn_low, alarm_low}}.

[READ][risk=low] Is a heartbeat/watchdog tag still alive? (liveness check). A heartbeat must keep CHANGING; a flatlined one means the upstream is dead even when comms/quality look fine. With timestamped samples + max_interval_s, also flags the longest stall. Args: series: Heartbeat samples — scalars or {value, timestamp?} (a counter/toggle). max_interval_s: Max allowed gap between changes; exceeding it = not alive. Returns dict: {alive (bool), samples, distinct_transitions, spread, longest_stall_s, reason}. Example: heartbeat_health(series=[1,2,3,4,5], max_interval_s=10).

[READ][risk=low] Per-tag history coverage — what history do we actually have. Answers the question every RCA starts with: which tags have stored history, how many rows, and over what time span — per tag {rows, first_ts, last_ts} from the same store historian_push writes. Read-only, bounded (tag list is capped with a truncation flag); no device I/O. Args: reader: 'sqlite' | 'tdengine' | 'iotdb'. Omit to use the per-site 'historian:' block in ~/.iaiops/config.yaml, else the local sqlite store. TSDB readers need their extra: pip install iaiops[tdengine|iotdb]. limit: Max tags returned (1..2000; default 500). Returns dict: {reader, source, tag_count, tags:[{tag, rows, first_ts, last_ts}], truncated} plus the standard return envelope (items_returned, items_total, items_total_is_exact, is_truncated, truncation_note). Example: historian_coverage().

[READ][risk=low] Bad-tag / flatline / gap detection over a provided series. Pure analysis over an injected sample series — no live historian needed. Args: series: Samples — scalars or {value, timestamp (ISO-8601), quality|good}. gap_threshold_s: Time gap (seconds) between consecutive samples that counts as a data gap (default 60). flatline_eps: Spread at/below which the series counts as flatline. Returns dict: {samples, numeric_samples, bad_quality_count, flatline (bool), gap_count, gaps:[{after, gap_seconds}], stdev, verdict ('ok'|'degraded'|'gappy'|'flatline'|'bad_tag')}. Example: historian_health(series=[{"value":10,"timestamp":"2026-06-28T10:00:00Z"}, ...]).

[WRITE][risk=low][→historian] Push collected telemetry to a national TSDB. Writes already-collected points to a domestic historian (信创) — TDengine or IoTDB — instead of binding InfluxDB. Data egress to the operator's OWN database, NOT a control-system write. Non-numeric points are skipped (numeric value column). Args: points: Collected points — {ref|metric, value|present_value, timestamp?, ...} (e.g. the output of interrogate / integrity_poll / read_points / monitor). sink: 'tdengine' or 'iotdb'. host/port/user/password: TSDB connection params (sensible defaults per sink when blank/0). database: Target database (TDengine db / IoTDB storage group, e.g. 'root.iaiops'). Returns dict: {sink, received, written, skipped_non_numeric, database}. Example: historian_push(points=[{"ref":"line1.temp","value":21.5}], sink="tdengine", host="10.0.0.20", database="iaiops").

[READ][risk=low] Query a tag's historical samples from a historian. Reads history back OUT of the store the sinks write — the local SQLite store (~/.iaiops/data.db), TDengine, or IoTDB — so the RCA copilot / an agent can see real pre-incident windows instead of only short live samples. Read-only over the operator's OWN historian; no device I/O. Bounded: rows are capped and a truncation flag is set when more history exists. Args: tag: Tag/metric name as stored by historian_push (e.g. 'line1.temp'). since/until: Optional ISO-8601 time bounds (inclusive). endpoint: Only samples from this endpoint label (sqlite reader only — the TSDB layout stores no endpoint label). reader: 'sqlite' | 'tdengine' | 'iotdb'. Omit to use the per-site 'historian:' block in ~/.iaiops/config.yaml, else the local sqlite store. TSDB readers need their extra: pip install iaiops[tdengine|iotdb]. limit: Max rows returned (1..10000; default 1000). Returns dict: {reader, source, tag, since, until, rows, samples:[{ts, endpoint, protocol, tag, value, quality, unit}], truncated} plus the standard return envelope (items_returned, items_total, items_total_is_exact, is_truncated, truncation_note). Trust `is_truncated`: an empty `samples` with is_truncated=false means the history really is empty, NOT that the result was cut short. Example: historian_query(tag="line1.temp", since="2026-07-02T06:00:00Z", until="2026-07-02T08:00:00Z").

[READ][risk=low] Connect and report IEC-104 link status + discovered stations. Args: endpoint: Endpoint name from config (protocol 'iec104'); omit for default. Returns dict: {endpoint, host, port, connected, configured_common_address, station_count, common_addresses[]}. Example: iec104_connection_info(endpoint="rtu1").

[READ][risk=low] General interrogation: all monitored points of a station (ASDU CA). Args: common_address: ASDU common address; omit for the configured/first station. endpoint: Endpoint name from config (protocol 'iec104'). Returns dict: {endpoint, common_address, point_count, points:[{io_address, type, value, quality, recorded_at}]}. Example: iec104_interrogate(common_address=1, endpoint="rtu1").

[READ][risk=low] Read one monitored point by information-object address (IOA). Args: io_address: The point's information-object address (IOA). common_address: ASDU common address; omit for the configured/first station. endpoint: Endpoint name from config (protocol 'iec104'). Returns dict: {endpoint, common_address, found, io_address, type, value, quality, recorded_at}. Example: iec104_read_point(io_address=1001, common_address=1, endpoint="rtu1").

[READ][risk=low] Browse immediate model children under a reference (LD/LN/DO). Args: reference: Model reference, e.g. 'IED1LD0' or 'IED1LD0/LLN0'. endpoint: Endpoint name from config (protocol 'iec61850'). Returns dict: {endpoint, reference, child_count, children[]}. Example: iec61850_browse(reference="IED1LD0/MMXU1", endpoint="ied1").

[READ][risk=low] List the IED's logical devices (optionally their children). Args: include_children: Also browse each logical device's immediate model children. endpoint: Endpoint name from config (protocol 'iec61850'); omit for default. Returns dict: {endpoint, logical_device_count, logical_devices:[{logical_device, children[]?, child_count?}]}. Example: iec61850_device_directory(include_children=True, endpoint="ied1").

[READ][risk=low] Read one data attribute by object-reference + functional constraint. Args: reference: Data-attribute object reference, e.g. 'IED1MMXU1.TotW.mag.f'. fc: Functional constraint — MX (measurands), ST (status), CF (config), … endpoint: Endpoint name from config (protocol 'iec61850'). Returns dict: {endpoint, reference, fc, value, error}. Example: iec61850_read(reference="IED1MMXU1.TotW.mag.f", fc="MX", endpoint="ied1").

[READ][risk=low] List saved investigations, newest first.

[READ][risk=low] Open an investigation over one past window and walk what can be walked. Contacts no device — the window is already past, and its evidence is whatever was collected at the time. Each of the eight steps records its own outcome: `done` (it ran, here is what it found), `refused` (it could not run HERE — no samples, no alarm source; a site fact) or `not_possible` (this product cannot do it at all). The investigation is persisted, so it can be re-read and advanced later.

[READ][risk=low] How far into an investigation this site could get, and what each gap needs. `readiness` answers "which scenarios can this site run"; this answers the next question down — if something stopped tomorrow, how many of the eight evidence steps could actually be walked, and for each one that could not, what is missing. Contacts nothing: no device, no network, no historian. It is derived from the config and the local store, which is what makes it usable on a site nobody has been authorised to probe yet. Each gap says whether it is *unmet* (you have not supplied it — the fix names the command) or *not yet expressible* (this product offers no way to supply it at all). Those two send a person to very different places.