diff --git a/INSTRUCTIONS.md b/INSTRUCTIONS.md index 2f7b6769b..b9fff0394 100644 --- a/INSTRUCTIONS.md +++ b/INSTRUCTIONS.md @@ -140,7 +140,7 @@ Six FastMCP servers cover IoT data, time-series ML, work orders, vibration diagn | Server | Tools | Categories | Backing service | | ----------- | ----- | ------------------------ | -------------------------------------- | -| `iot` | 7 | read | CouchDB (telemetry + asset registry) | +| `iot` | 12 | read | CouchDB (telemetry + asset registry) | | `utilities` | 3 | read | none | | `fmsr` | 2 | read, LLM-use | LiteLLM + `failure_modes.yaml` | | `wo` | 14 | read, write | CouchDB | @@ -164,17 +164,17 @@ from mcphub import ToolUniverse tu = ToolUniverse() # 1. init tu.load_tools() # 2. connect + discover result = tu.run({ # 3. run - "name": "iot.sensors", + "name": "iot.measured_sensors", "arguments": {"site_name": "MAIN", "asset_id": "Chiller 6"}, }) tu.close() ``` -- Tools are namespaced `.` (e.g. `iot.sensors`, +- Tools are namespaced `.` (e.g. `iot.measured_sensors`, `wo.generate_work_order`); a bare name works when unambiguous. - `load_tools(servers=["iot", "fmsr"])` limits the set; no args loads all six. - Discovery: `tu.find_tools("failure mode")`, `tu.list_tools()`, - `tu.tool_specification("iot.sensors")`. + `tu.tool_specification("iot.measured_sensors")`. - Servers are spawned via `uv run -mcp-server` and inherit your `.env`. Run inside the project (`uv run python ...`) or an activated venv. Override the launch table with `ToolUniverse(servers={...})`. diff --git a/README.md b/README.md index 866c417f1..40d36c272 100644 --- a/README.md +++ b/README.md @@ -84,7 +84,7 @@ AssetOpsBench is a **unified framework for developing, orchestrating, and evalua | MCP Servers | Important tools | |---|---| -| **IoT** | `get_sites`, `get_history`, `get_assets`, `get_sensors` | +| **IoT** | `sites`, `asset_ids`, `measured_sensors`, `history`, `installed_sensors`, `assets`, `stream_extent`, ... | | **FMSR** | `get_failure_modes`, `generate_failure_modes`, `add_failure_modes`, `generate_failure_mode_sensor_mapping` | | **TSFM** | `forecasting`, `timeseries_anomaly_detection` | | **WO** | `get_work_order_distribution`, `predict_next_work_order`, ... | diff --git a/docs/mcp-servers.md b/docs/mcp-servers.md index e1ffa6b9b..c56b831d1 100644 --- a/docs/mcp-servers.md +++ b/docs/mcp-servers.md @@ -13,14 +13,18 @@ Six FastMCP servers expose the AssetOpsBench domain logic. Each is a standalone ## iot — IoT Sensor Data + The IoT server reads from **two** databases: telemetry readings (`IOT_DBNAME`, default `iot`) and an asset **registry** (`ASSET_DBNAME`, default `asset`, loaded from `asset_profile_sample.json`). The -two answer different questions: `assets()`/`sensors()` reflect TELEMETRY — what actually streams (the -**measured** set); `get_asset()`/`asset_sensors()`/`registry_assets()` reflect the REGISTRY — the -asset nameplate and the **installed** sensor inventory (by name). Comparing `asset_sensors()` against -`sensors()` surfaces sensors that are installed but not streaming. The registry also reconciles ids -across systems (Maximo `assetnum`, telemetry `iot_asset_id`, work-order `wo_assetnum`), so an asset -can be looked up by any of its ids. +two answer different questions: `asset_ids()`/`measured_sensors()` reflect TELEMETRY — what actually +streams (the **measured** set); `get_asset_detail()`/`installed_sensors()`/`assets()` reflect the +REGISTRY — the asset nameplate and the **installed** sensor inventory (by name). Comparing +`installed_sensors()` against `measured_sensors()` surfaces sensors that are installed but not +streaming. Beyond browsing: `stream_extent` and `sensor_coverage` report a stream's time bounds and +per-sensor record counts so you can size a query; `history` returns readings a page at a time +(`limit` rows per page, follow `cursor`); and `sensor_stats`, `latest_reading`, and +`find_assets_by_sensors` provide quick summaries and reverse lookup. Timestamps are ISO 8601 and +time ranges are half-open `[start, end)`. **Path:** `src/servers/iot/main.py` **Requires:** CouchDB (`COUCHDB_URL`, `COUCHDB_USERNAME`, `COUCHDB_PASSWORD`, `IOT_DBNAME`, `ASSET_DBNAME`) @@ -35,15 +39,20 @@ can be looked up by any of its ids. Synthetic motor vibration data (`asset_id: Motor_01`, from `motor_01.json`) ships in a separate `vibration` database for the vibration MCP server. -| Tool | Arguments | Description | -| ----------------- | ------------------------------------------ | ------------------------------------------------------------------------------------------------------------ | -| `sites` | — | List all sites, discovered dynamically from the asset registry (`siteid`) | -| `assets` | `site_name` | List asset ids registered at a site (telemetry id where present, else `assetnum`) | -| `sensors` | `site_name`, `asset_id` | List **measured** sensor names for an asset (union of keys across its telemetry docs) | -| `history` | `site_name`, `asset_id`, `start`, `final?` | Fetch historical sensor readings for a time range (ISO 8601 timestamps) | -| `get_asset` | `site_name`, `asset_id` | Registry/nameplate detail for one asset (description, assettype, status, location, vintage, installed count) | -| `asset_sensors` | `site_name`, `asset_id` | List the **installed** sensors for an asset, by name (registry inventory) | -| `registry_assets` | `site_name`, `assettype?` | List registry assets with metadata (assettype, vintage, sensor count), optionally filtered by assettype | +| Tool | Arguments | Description | +| ------------------------ | -------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ | +| `sites` | — | List all sites, discovered dynamically from the asset registry (`siteid`) | +| `asset_ids` | `site_name` | List the bare asset IDs at a site — use one as `asset_id` in the other tools | +| `assets` | `site_name`, `assettype?` | Assets with metadata (description, assettype, vintage, installed sensor count); optionally filter by assettype | +| `measured_sensors` | `site_name`, `asset_id` | **Measured** sensor names for an asset (what actually streams; union across telemetry docs) | +| `installed_sensors` | `site_name`, `asset_id` | **Installed** sensor names for an asset (registry nameplate inventory) | +| `get_asset_detail` | `site_name`, `asset_id` | Registry/nameplate detail for one asset (description, assettype, status, location, vintage, installed count) | +| `find_assets_by_sensors` | `site_name`, `sensors`, `match?`, `substring?`, `source?` | Reverse lookup: assets that have the given sensor(s); `source`=`measured`\|`installed`, `match`=`all`\|`any` | +| `history` | `site_name`, `asset_id`, `start?`, `end?`, `sensors?`, `limit?`, `cursor?` | Historical readings, one page at a time. ISO 8601, half-open `[start,end)`; omit range for all. Page with `limit` (≤1000) + `cursor`; response gives `next_cursor`/`has_more`. `sensors` projects columns. | +| `stream_extent` | `site_name`, `asset_id`, `sensor?`, `start?`, `end?` | Time bounds + record count for a stream (pre-flight for `history`); flags if it exceeds the page limit | +| `sensor_coverage` | `site_name`, `asset_id`, `max_scan_docs?` | Per-measured-sensor record counts and first/last timestamps (`max_scan_docs`=0 scans all) | +| `sensor_stats` | `site_name`, `asset_id`, `sensor?`, `start?`, `end?` | Numeric summary (count/min/max/mean/stddev) per sensor; omit `sensor` to summarize every measured sensor | +| `latest_reading` | `site_name`, `asset_id`, `sensor?` | Most recent observation for an asset (+ `age_seconds` for staleness) | ## utilities — Utilities diff --git a/src/couchdb/collections.json b/src/couchdb/collections.json index db67d67f3..3edd36d0c 100644 --- a/src/couchdb/collections.json +++ b/src/couchdb/collections.json @@ -79,20 +79,8 @@ "doctype": "asset", "indexes": [ [ - "doctype", - "assetnum" - ], - [ - "doctype", - "assettype" - ], - [ - "doctype", + "assetnum", "siteid" - ], - [ - "doctype", - "iot_asset_id" ] ] }, @@ -144,4 +132,4 @@ "id_prefix": "fc", "indexes": [] } -} +} \ No newline at end of file diff --git a/src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json b/src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json index 24acd9670..db9f1498a 100644 --- a/src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json +++ b/src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json @@ -1,10 +1,6 @@ [ { - "_id": "asset:CHILLER6", - "doctype": "asset", - "assetnum": "CHILLER6", - "iot_asset_id": "Chiller 6", - "wo_assetnum": "CHILLER6", + "assetnum": "CHILLER 6", "description": "Chiller 6 (chilled-water plant)", "siteid": "MAIN", "orgid": "RIVPLANT", @@ -35,11 +31,7 @@ ] }, { - "_id": "asset:mp_1", - "doctype": "asset", "assetnum": "mp_1", - "iot_asset_id": "mp_1", - "wo_assetnum": null, "description": "Compressor / pneumatic system", "siteid": "MAIN", "orgid": "RIVPLANT", @@ -73,11 +65,7 @@ ] }, { - "_id": "asset:hyd_1", - "doctype": "asset", "assetnum": "hyd_1", - "iot_asset_id": "hyd_1", - "wo_assetnum": null, "description": "Hydraulic pump condition-monitoring rig", "siteid": "MAIN", "orgid": "RIVPLANT", @@ -114,11 +102,7 @@ ] }, { - "_id": "asset:Motor_01", - "doctype": "asset", "assetnum": "Motor_01", - "iot_asset_id": "Motor_01", - "wo_assetnum": null, "description": "Induction motor (vibration-monitored)", "siteid": "MAIN", "orgid": "RIVPLANT", @@ -138,11 +122,7 @@ ] }, { - "_id": "asset:PUMP3", - "doctype": "asset", "assetnum": "PUMP3", - "iot_asset_id": null, - "wo_assetnum": "PUMP3", "description": "Pump 3", "siteid": "MAIN", "orgid": "RIVPLANT", @@ -160,11 +140,7 @@ "sensors": [] }, { - "_id": "asset:AHU2", - "doctype": "asset", "assetnum": "AHU2", - "iot_asset_id": null, - "wo_assetnum": "AHU2", "description": "Air handling unit 2", "siteid": "NORTH", "orgid": "RIVPLANT", diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index 7d2ba9449..1f06903a3 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -1,8 +1,8 @@ import os import logging -from datetime import datetime -from functools import lru_cache -from typing import Any, Dict, List, Optional, Union +import statistics +from datetime import datetime, timezone +from typing import Any, Dict, Iterator, List, Optional, Union, Tuple from mcp.server.fastmcp import FastMCP from pydantic import BaseModel import couchdb3 @@ -58,14 +58,18 @@ instructions=( "IoT sensor data + asset registry. Browse sites, assets, and sensors, read the asset " "nameplate (registry), see which installed sensors are actually measured (streaming), and " - "query historical readings from CouchDB. NOTE: assets()/sensors() reflect TELEMETRY (what " - "streams = measured); get_asset()/asset_sensors()/registry_assets() reflect the REGISTRY " - "(what is installed, by name). Compare the two to find installed-but-not-streaming sensors." + "query historical readings." ), ) DEFAULT_SITES = ["MAIN"] +# Fields on a reading document that are structural, not sensor channels. +RESERVED_FIELDS = {"_id", "_rev", "asset_id", "timestamp"} + +# CouchDB per-request page cap used by history(); new paging tools respect it too. +PAGE_SIZE = 1000 + class ErrorResult(BaseModel): error: str @@ -93,10 +97,12 @@ class SensorsResult(BaseModel): class HistoryResult(BaseModel): site_name: str asset_id: str - total_observations: int - start: str - final: Optional[str] observations: List[Dict[str, Any]] + returned: int + next_cursor: Optional[str] + has_more: bool + start: Optional[str] + end: Optional[str] message: str @@ -122,37 +128,90 @@ class AssetSensorsResult(BaseModel): message: str -class RegistryAssetsResult(BaseModel): +class AssetSummary(BaseModel): + asset_id: str + description: Optional[str] + assettype: Optional[str] + vintage: Optional[str] + n_sensors: int # installed sensor count + + +class AssetsWithMetadataResult(BaseModel): site_name: str total_assets: int - assets: List[Dict[str, Any]] + assets: List[AssetSummary] message: str -_asset_list_cache: Optional[List[str]] = None +# ── New tool result models ── +class AssetSensorMatch(BaseModel): + asset_id: str + matched_sensors: List[str] -def get_asset_list() -> List[str]: - """Helper to fetch unique asset IDs from CouchDB. Result is cached after - the first successful call to avoid repeated full-table scans.""" - global _asset_list_cache - if _asset_list_cache is not None: - return _asset_list_cache +class FindAssetsResult(BaseModel): + site_name: str + query_sensors: List[str] + match: str + source: str + total_assets: int + assets: List[AssetSensorMatch] + message: str - if not db: - return [] - try: - # We limit the fields to just asset_id to minimize data transfer - res = db.find( - {"asset_id": {"$exists": True}}, fields=["asset_id"], limit=100000 - ) - assets = {doc["asset_id"] for doc in res["docs"] if "asset_id" in doc} - _asset_list_cache = sorted(list(assets)) - return _asset_list_cache - except Exception as e: - logger.error(f"Error fetching assets: {e}") - return [] +class SensorCoverage(BaseModel): + sensor: str + non_null_count: int + first_timestamp: Optional[str] + last_timestamp: Optional[str] + + +class SensorCoverageResult(BaseModel): + site_name: str + asset_id: str + docs_scanned: int + sensors: List[SensorCoverage] + message: str + + +class StreamExtentResult(BaseModel): + site_name: str + asset_id: str + sensor: Optional[str] + start_time: Optional[str] + end_time: Optional[str] + total_records: int + exceeds_page_limit: bool + approx_interval_seconds: Optional[float] + message: str + + +class SensorStat(BaseModel): + sensor: str + count: int + null_count: int + min: Optional[float] + max: Optional[float] + mean: Optional[float] + stddev: Optional[float] + first_timestamp: Optional[str] + last_timestamp: Optional[str] + + +class SensorStatsResult(BaseModel): + site_name: str + asset_id: str + stats: List[SensorStat] # one entry if `sensor` given, else every measured sensor + message: str + + +class LatestReadingResult(BaseModel): + site_name: str + asset_id: str + timestamp: Optional[str] + values: Dict[str, Any] + age_seconds: Optional[float] + message: str _sensor_list_cache: Dict[str, List[str]] = {} @@ -173,16 +232,18 @@ def get_sensor_list(asset_id: str) -> List[str]: return [] try: - res = db.find({"asset_id": asset_id}, limit=100000) - docs = res["docs"] - if not docs: + found: set = set() + seen = False + for doc in _iter_docs( + {"asset_id": asset_id} + ): # walks all pages via Mango bookmark + seen = True + for key in doc.keys(): + if key not in RESERVED_FIELDS: + found.add(key) + if not seen: return [] - - # Exclude metadata; union the measurement keys across every reading document. - exclude = {"_id", "_rev", "asset_id", "timestamp"} - sensors = sorted( - {key for doc in docs for key in doc.keys() if key not in exclude} - ) + sensors = sorted(found) _sensor_list_cache[asset_id] = sensors return sensors except Exception as e: @@ -190,31 +251,31 @@ def get_sensor_list(asset_id: str) -> List[str]: return [] -_asset_doc_cache: Dict[str, Dict[str, Any]] = {} +_asset_doc_cache: Dict[Tuple[str, Optional[str]], Optional[Dict[str, Any]]] = {} -def get_asset_doc(asset_id: str) -> Optional[Dict[str, Any]]: - """Fetch one asset-registry document, resolving ANY of the asset's id spaces — the registry - `assetnum` (Maximo id), the telemetry `iot_asset_id`, or the work-order `wo_assetnum`. This lets - the same profile be found whether the caller holds the IoT id (e.g. 'Chiller 6') or the WO id - ('CHILLER6'). Cached per asset_id.""" - if asset_id in _asset_doc_cache: - return _asset_doc_cache[asset_id] +def get_asset_doc(asset_id: str, site_name: Optional[str] = None) -> Optional[Dict[str, Any]]: + """Fetch one asset-registry document by assetnum, optionally constrained to a site so both + conditions must hold. Cached per (asset_id, site_name).""" + # impl: registry docs are keyed by `assetnum`, whose value equals the telemetry `asset_id`. + cache_key = (asset_id, site_name) + if cache_key in _asset_doc_cache: + return _asset_doc_cache[cache_key] if not asset_db: return None try: - for field in ("assetnum", "iot_asset_id", "wo_assetnum"): - res = asset_db.find({"doctype": "asset", field: asset_id}, limit=1) - docs = res["docs"] - if docs: - _asset_doc_cache[asset_id] = docs[0] - return docs[0] - return None + selector: Dict[str, Any] = {"doctype": "asset", "assetnum": asset_id} + if site_name is not None: + selector["siteid"] = site_name + res = asset_db.find(selector, limit=1) + docs = res["docs"] + doc = docs[0] if docs else None + _asset_doc_cache[cache_key] = doc # cache hit AND miss (asset is static per run) + return doc except Exception as e: logger.error(f"Error fetching asset doc {asset_id}: {e}") return None - _registry_sites_cache: Optional[List[str]] = None @@ -226,7 +287,7 @@ def get_registry_sites() -> List[str]: if not asset_db: return [] try: - res = asset_db.find({"doctype": "asset"}, fields=["siteid"], limit=100000) + res = asset_db.find({"doctype": "asset"}, fields=["siteid"], limit=10**12) found = sorted({d.get("siteid") for d in res["docs"] if d.get("siteid")}) _registry_sites_cache = found return found @@ -246,6 +307,141 @@ def _is_known_site(site_name: str) -> bool: return site_name in known_sites() +# --------------------------------------------------------------------------- +# Shared helpers for the new tools +# --------------------------------------------------------------------------- +def _validate_dates(start: Optional[str], end: Optional[str]) -> Optional[str]: + """Return None if ok, else an error message. None inputs are allowed.""" + try: + if start is not None: + datetime.fromisoformat(start) + if end is not None: + datetime.fromisoformat(end) + except ValueError as e: + return f"Invalid date format (expected ISO 8601, e.g. 2024-01-15T00:00:00): {e}" + if start is not None and end is not None and start >= end: + return "start >= end" + return None + + +def _time_selector( + asset_id: str, start: Optional[str], end: Optional[str] +) -> Dict[str, Any]: + selector: Dict[str, Any] = {"asset_id": asset_id} + ts: Dict[str, Any] = {} + if start is not None: + ts["$gte"] = datetime.fromisoformat(start).isoformat() + if end is not None: + ts["$lt"] = datetime.fromisoformat(end).isoformat() + if ts: + selector["timestamp"] = ts + return selector + + +def _iter_docs( + selector: Dict[str, Any], + fields: Optional[List[str]] = None, + sort: Optional[List[Dict[str, str]]] = None, + page_size: int = PAGE_SIZE, + max_docs: Optional[int] = None, +) -> Iterator[Dict[str, Any]]: + """Yield all docs matching `selector`, transparently walking Mango bookmarks so callers are not + limited to a single 1000-row page. Used by the new count/extent/stats tools.""" + if not db: + return + if sort is None: + sort = [{"asset_id": "asc"}, {"timestamp": "asc"}] + bookmark: Optional[str] = None + yielded = 0 + while True: + kwargs: Dict[str, Any] = {"limit": page_size, "sort": sort} + if fields is not None: + kwargs["fields"] = fields + if bookmark is not None: + kwargs["bookmark"] = bookmark + res = db.find(selector, **kwargs) + docs = res.get("docs", []) + if not docs: + break + for doc in docs: + yield doc + yielded += 1 + if max_docs is not None and yielded >= max_docs: + return + bookmark = res.get("bookmark") + if bookmark is None or len(docs) < page_size: + break + + +def _latest_doc( + asset_id: str, sensor: Optional[str] = None +) -> Optional[Dict[str, Any]]: + """Return the newest reading doc for an asset. Fast path uses a descending sort (needs a + descending index); if that index does not exist, falls back to an ascending scan keeping the + last doc — correct either way.""" + if not db: + return None + selector: Dict[str, Any] = {"asset_id": asset_id} + if sensor: + selector[sensor] = {"$exists": True, "$ne": None} + try: + res = db.find( + selector, limit=1, sort=[{"asset_id": "desc"}, {"timestamp": "desc"}] + ) + docs = res.get("docs", []) + if docs: + return docs[0] + # empty result with a working sort => genuinely no matching docs + return None + except Exception as e: + logger.info(f"descending sort unavailable, scanning ascending: {e}") + last: Optional[Dict[str, Any]] = None + for doc in _iter_docs(selector): + last = doc + return last + + +def _span_seconds(a_iso: str, b_iso: str) -> Optional[float]: + try: + return ( + datetime.fromisoformat(b_iso) - datetime.fromisoformat(a_iso) + ).total_seconds() + except ValueError: + return None + + +def _age_seconds(ts_iso: str) -> Optional[float]: + try: + ts = datetime.fromisoformat(ts_iso) + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + return (datetime.now(timezone.utc) - ts).total_seconds() + except ValueError: + return None + + +def _site_asset_ids(site_name: str) -> List[str]: + """Asset ids registered at a site — mirrors asset_ids().""" + if not asset_db: + return [] + try: + res = asset_db.find( + {"doctype": "asset", "siteid": site_name}, + fields=["assetnum"], + limit=100000, + ) + return sorted(d["assetnum"] for d in res["docs"]) + except Exception as e: + logger.error(f"_site_asset_ids failed: {e}") + return [] + +def _installed_sensors(asset_id: str, site_name: Optional[str] = None) -> List[str]: + doc = get_asset_doc(asset_id, site_name) + return list(doc.get("sensors", [])) if doc else [] + +# --------------------------------------------------------------------------- +# Original tools (unchanged) +# --------------------------------------------------------------------------- @mcp.tool(title="List Sites") def sites() -> SitesResult: """Retrieves the list of sites, discovered dynamically from the asset registry (the distinct @@ -254,11 +450,10 @@ def sites() -> SitesResult: return SitesResult(sites=known_sites()) -@mcp.tool(title="List Assets") -def assets(site_name: str) -> Union[AssetsResult, ErrorResult]: - """Returns the assets registered at a given site, from the asset registry filtered by `siteid`. - Each returned id is the asset's telemetry id (`iot_asset_id`) where it has one, otherwise its - registry `assetnum` — so the id works with sensors()/history() when telemetry exists. +@mcp.tool(title="List Asset IDs") +def asset_ids(site_name: str) -> Union[AssetsResult, ErrorResult]: + """List the asset IDs at a site. Use an ID from this list as the `asset_id` argument in the + other tools (measured_sensors, history, stream_extent, ...). """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -267,10 +462,10 @@ def assets(site_name: str) -> Union[AssetsResult, ErrorResult]: try: res = asset_db.find( {"doctype": "asset", "siteid": site_name}, - fields=["assetnum", "iot_asset_id"], + fields=["assetnum"], limit=100000, ) - ids = sorted((d.get("iot_asset_id") or d["assetnum"]) for d in res["docs"]) + ids = sorted(d["assetnum"] for d in res["docs"]) return AssetsResult( site_name=site_name, total_assets=len(ids), @@ -278,16 +473,17 @@ def assets(site_name: str) -> Union[AssetsResult, ErrorResult]: message=f"found {len(ids)} assets at site {site_name}: {', '.join(ids)}.", ) except Exception as e: - logger.error(f"assets failed: {e}") + logger.error(f"asset_ids failed: {e}") return ErrorResult(error=str(e)) -@mcp.tool(title="List Sensors") -def sensors(site_name: str, asset_id: str) -> Union[SensorsResult, ErrorResult]: - """Lists the sensors available for a specified asset at a given site. - These are the MEASURED sensors — names discovered from the asset's telemetry documents, - i.e. points that actually stream to the historian. For the full INSTALLED inventory - (including sensors fitted but not streaming), use asset_sensors().""" +@mcp.tool(title="List Measured Sensors") +def measured_sensors( + site_name: str, asset_id: str +) -> Union[SensorsResult, ErrorResult]: + """Lists the MEASURED sensors for a specified asset at a given site — names discovered from the + asset's telemetry documents, i.e. points that actually stream to the historian. + """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -306,58 +502,80 @@ def sensors(site_name: str, asset_id: str) -> Union[SensorsResult, ErrorResult]: @mcp.tool(title="Get Sensor History") def history( - site_name: str, asset_id: str, start: str, final: Optional[str] = None + site_name: str, + asset_id: str, + start: Optional[str] = None, + end: Optional[str] = None, + sensors: Optional[List[str]] = None, + limit: int = PAGE_SIZE, + cursor: Optional[str] = None, ) -> Union[HistoryResult, ErrorResult]: - """Returns a list of historical sensor values for the specified asset(s) at a site within a given time range (start to final).""" - try: - start_iso = datetime.fromisoformat(start).isoformat() - if final: - datetime.fromisoformat(final) - if start >= final: - return ErrorResult(error="start >= final") - except ValueError as e: - return ErrorResult(error=f"Invalid date format: {e}") + """Return historical sensor readings for an asset, one page at a time. + + Timestamps are ISO 8601 strings, e.g. "2024-01-15T00:00:00" (offset allowed, + "2024-01-15T00:00:00+00:00"). The range is half-open [start, end). Omit `start`/`end` to + use the full available range; call stream_extent(site_name, asset_id) first to learn the + exact bounds and total record count. + + Paging: `limit` = rows per page (<=1000). Leave `cursor` empty on the first call; each + response returns `next_cursor` and `has_more` — repeat with cursor=next_cursor until + `has_more` is false. Pass `sensors` to return only those columns. + On a malformed timestamp or start >= end, returns {"error": ...}. + """ + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + err = _validate_dates(start, end) + if err: + return ErrorResult(error=err) if not db: return ErrorResult(error="CouchDB not connected") + limit = max(1, min(limit, PAGE_SIZE)) - selector = { - "asset_id": asset_id, - "timestamp": {"$gte": start_iso}, + selector = _time_selector(asset_id, start, end) + fields: Optional[List[str]] = None + if sensors: + fields = ["_id", "asset_id", "timestamp"] + list(sensors) + + kwargs: Dict[str, Any] = { + "limit": limit, + "sort": [{"asset_id": "asc"}, {"timestamp": "asc"}], } - if final: - selector["timestamp"]["$lt"] = datetime.fromisoformat(final).isoformat() + if fields is not None: + kwargs["fields"] = fields + if cursor is not None: + kwargs["bookmark"] = cursor - logger.info(f"Querying CouchDB with selector: {selector}") try: - res = db.find( - selector, limit=1000, sort=[{"asset_id": "asc"}, {"timestamp": "asc"}] - ) - docs = res["docs"] - return HistoryResult( - site_name=site_name, - asset_id=asset_id, - total_observations=len(docs), - start=start, - final=final, - observations=docs, - message=f"found {len(docs)} observations for asset_id {asset_id} from {start} to {final or 'now'}.", - ) + res = db.find(selector, **kwargs) except Exception as e: logger.error(f"CouchDB query failed: {e}") return ErrorResult(error=str(e)) + docs = res.get("docs", []) + next_cursor = res.get("bookmark") + has_more = len(docs) == limit + return HistoryResult( + site_name=site_name, + asset_id=asset_id, + observations=docs, + returned=len(docs), + next_cursor=next_cursor if has_more else None, + has_more=has_more, + start=start, + end=end, + message=f"{len(docs)} observation(s) for asset_id {asset_id}; has_more={has_more}.", + ) + -@mcp.tool(title="Get Asset") -def get_asset(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResult]: - """Return registry/nameplate details for one asset (Maximo MXASSET-aligned: description, - assettype, status, location, installdate, vintage) plus installed sensor count. - This is asset IDENTITY — distinct from the telemetry-derived assets() list.""" +@mcp.tool(title="Get Asset Detail") +def get_asset_detail(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResult]: + """Return registry/nameplate details for one asset.""" if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") - doc = get_asset_doc(asset_id) + doc = get_asset_doc(asset_id, site_name) if not doc: - return ErrorResult(error=f"unknown asset_id {asset_id} in registry") + return ErrorResult(error=f"unknown asset_id {asset_id} at site {site_name}") n = len(doc.get("sensors", [])) assettype = doc.get("assettype") vintage = doc.get("vintage") @@ -386,18 +604,18 @@ def get_asset(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResult]: ) -@mcp.tool(title="List Asset Sensors") -def asset_sensors( +@mcp.tool(title="List Installed Sensors") +def installed_sensors( site_name: str, asset_id: str ) -> Union[AssetSensorsResult, ErrorResult]: """List the INSTALLED sensors for an asset, by name (installed is assumed). This is the registry - inventory — distinct from sensors(), which lists only what actually streams (the MEASURED set). - Compare the two to find installed-but-not-streaming sensors.""" + inventory — distinct from measured_sensors(), which lists only what actually streams (the MEASURED + set). Compare the two to find installed-but-not-streaming sensors.""" if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") - doc = get_asset_doc(asset_id) + doc = get_asset_doc(asset_id, site_name) if not doc: - return ErrorResult(error=f"unknown asset_id {asset_id} in registry") + return ErrorResult(error=f"unknown asset_id {asset_id} at site {site_name}") names = list(doc.get("sensors", [])) return AssetSensorsResult( site_name=site_name, @@ -408,14 +626,13 @@ def asset_sensors( ) -@mcp.tool(title="List Registry Assets") -def registry_assets( +@mcp.tool(title="List Assets") +def assets( site_name: str, assettype: Optional[str] = None -) -> Union[RegistryAssetsResult, ErrorResult]: - """List assets from the registry with metadata (assettype, vintage, sensor count), optionally - filtered by assettype (e.g. 'PUMP', 'COMPRESSOR'). Complements assets(), which returns bare ids derived from - telemetry.""" - +) -> Union[AssetsWithMetadataResult, ErrorResult]: + """List assets at a site with metadata: description, assettype, vintage, and installed sensor + count (one AssetSummary per asset). Optionally filter by `assettype` (e.g. 'PUMP', + 'COMPRESSOR'). For just the bare IDs to pass to other tools, use `asset_ids`.""" if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") if not asset_db: @@ -431,30 +648,314 @@ def registry_assets( ) rows = sorted( ( - { - "asset_id": d["assetnum"], - "assettype": d.get("assettype"), - "description": d.get("description"), - "vintage": d.get("vintage"), - "n_sensors": len(d.get("sensors", [])), - } + AssetSummary( + asset_id=d["assetnum"], + description=d.get("description"), + assettype=d.get("assettype"), + vintage=d.get("vintage"), + n_sensors=len(d.get("sensors", [])), + ) for d in res["docs"] ), - key=lambda r: r["asset_id"], + key=lambda r: r.asset_id, ) - return RegistryAssetsResult( + return AssetsWithMetadataResult( site_name=site_name, total_assets=len(rows), assets=rows, - message=f"found {len(rows)} registry assets" + message=f"found {len(rows)} assets" + (f" of type '{assettype}'" if assettype else "") + ".", ) except Exception as e: - logger.error(f"registry_assets failed: {e}") + logger.error(f"assets failed: {e}") return ErrorResult(error=str(e)) +# =========================================================================== +# NEW TOOLS +# =========================================================================== +@mcp.tool(title="Find Assets By Sensors") +def find_assets_by_sensors( + site_name: str, + sensors: List[str], + match: str = "all", + substring: bool = False, + source: str = "measured", +) -> Union[FindAssetsResult, ErrorResult]: + """Return only the assets at a site that have the given sensor(s) — not all assets. + match='all' requires every listed sensor; match='any' requires at least one. + substring=True matches sensor names case-insensitively by substring. + source='measured' checks telemetry (what streams, via measured_sensors()); source='installed' + checks the registry nameplate (via installed_sensors()).""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + if match not in ("all", "any"): + return ErrorResult(error="match must be 'all' or 'any'") + if source not in ("measured", "installed"): + return ErrorResult(error="source must be 'measured' or 'installed'") + if not sensors: + return ErrorResult(error="provide at least one sensor name") + + matches: List[AssetSensorMatch] = [] + for asset_id in _site_asset_ids(site_name): + available = ( + get_sensor_list(asset_id) + if source == "measured" + else _installed_sensors(asset_id, site_name) + ) + + def _hits(w: str) -> List[str]: + if substring: + wl = w.lower() + return [s for s in available if wl in s.lower()] + return [s for s in available if s == w] + + per_query = {w: _hits(w) for w in sensors} + satisfied = [w for w, hits in per_query.items() if hits] + ok = len(satisfied) == len(sensors) if match == "all" else len(satisfied) > 0 + if ok: + matched = sorted({s for hits in per_query.values() for s in hits}) + matches.append(AssetSensorMatch(asset_id=asset_id, matched_sensors=matched)) + + return FindAssetsResult( + site_name=site_name, + query_sensors=sensors, + match=match, + source=source, + total_assets=len(matches), + assets=matches, + message=f"{len(matches)} asset(s) at {site_name} match {sensors} " + f"(match={match}, substring={substring}, source={source}).", + ) + + +@mcp.tool(title="Stream Extent") +def stream_extent( + site_name: str, + asset_id: str, + sensor: Optional[str] = None, + start: Optional[str] = None, + end: Optional[str] = None, +) -> Union[StreamExtentResult, ErrorResult]: + """Time bounds + record count for an asset's stream (optionally one sensor and/or a window). + Use this before history() to learn start_time / end_time / total_records and whether the + result will page (exceeds_page_limit). Timestamps are ISO 8601, half-open [start, end); + omit both for the full range.""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + err = _validate_dates(start, end) + if err: + return ErrorResult(error=err) + if not db: + return ErrorResult(error="CouchDB not connected") + + selector = _time_selector(asset_id, start, end) + if sensor: + selector[sensor] = {"$exists": True, "$ne": None} + + first_ts: Optional[str] = None + last_ts: Optional[str] = None + total = 0 + for doc in _iter_docs(selector, fields=["timestamp"]): + ts = doc.get("timestamp") + if ts is not None: + if first_ts is None: + first_ts = ts + last_ts = ts + total += 1 + + if total == 0: + return ErrorResult( + error=f"no records for asset_id {asset_id}" + + (f", sensor {sensor}" if sensor else "") + ) + + approx_interval: Optional[float] = None + if first_ts and last_ts and total > 1: + span = _span_seconds(first_ts, last_ts) + if span is not None: + approx_interval = span / (total - 1) + + return StreamExtentResult( + site_name=site_name, + asset_id=asset_id, + sensor=sensor, + start_time=first_ts, + end_time=last_ts, + total_records=total, + exceeds_page_limit=total > PAGE_SIZE, + approx_interval_seconds=approx_interval, + message=f"{total} record(s) for asset_id {asset_id}" + + (f" (sensor {sensor})" if sensor else "") + + f" from {first_ts} to {last_ts}.", + ) + + +@mcp.tool(title="Sensor Coverage") +def sensor_coverage( + site_name: str, + asset_id: str, + max_scan_docs: int = 5000, +) -> Union[SensorCoverageResult, ErrorResult]: + """Per-measured-sensor record counts and time coverage (non-null count, first/last timestamp) + for an asset. Complements measured_sensors(), which lists names but not how much data each + channel has. `max_scan_docs` = maximum reading documents to scan (0 = scan all; larger = + more complete, slower).""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + if not db: + return ErrorResult(error="CouchDB not connected") + + counts: Dict[str, int] = {} + first_ts: Dict[str, str] = {} + last_ts: Dict[str, str] = {} + scanned = 0 + + for doc in _iter_docs( + {"asset_id": asset_id}, + sort=[{"asset_id": "asc"}, {"timestamp": "asc"}], + max_docs=max_scan_docs if max_scan_docs > 0 else None, + ): + scanned += 1 + ts = doc.get("timestamp") + for k, v in doc.items(): + if k in RESERVED_FIELDS or v is None: + continue + counts[k] = counts.get(k, 0) + 1 + if k not in first_ts and ts is not None: + first_ts[k] = ts + if ts is not None: + last_ts[k] = ts + + if scanned == 0: + return ErrorResult(error=f"unknown asset_id {asset_id} or no records found") + + cov = [ + SensorCoverage( + sensor=s, + non_null_count=counts[s], + first_timestamp=first_ts.get(s), + last_timestamp=last_ts.get(s), + ) + for s in sorted(counts) + ] + return SensorCoverageResult( + site_name=site_name, + asset_id=asset_id, + docs_scanned=scanned, + sensors=cov, + message=f"{len(cov)} sensor(s) with data for asset_id {asset_id} across {scanned} scanned docs.", + ) + + +@mcp.tool(title="Sensor Statistics") +def sensor_stats( + site_name: str, + asset_id: str, + sensor: Optional[str] = None, + start: Optional[str] = None, + end: Optional[str] = None, +) -> Union[SensorStatsResult, ErrorResult]: + """Numeric summary (count/min/max/mean/stddev) per sensor over an optional time window, + without returning raw rows. Omit `sensor` to summarize EVERY measured sensor; give one to + summarize just that channel. Timestamps are ISO 8601, half-open [start, end); omit both for + the full range.""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + err = _validate_dates(start, end) + if err: + return ErrorResult(error=err) + if not db: + return ErrorResult(error="CouchDB not connected") + + targets = [sensor] if sensor else get_sensor_list(asset_id) + if not targets: + return ErrorResult(error=f"unknown asset_id {asset_id} or no sensors found") + + acc = {s: {"vals": [], "nulls": 0, "first": None, "last": None} for s in targets} + tset = set(targets) + for doc in _iter_docs(_time_selector(asset_id, start, end)): + ts = doc.get("timestamp") + for s in tset: + if s not in doc: + continue + a = acc[s] + if ts is not None: + if a["first"] is None: + a["first"] = ts + a["last"] = ts + v = doc.get(s) + if v is None: + a["nulls"] += 1 + continue + try: + a["vals"].append(float(v)) + except (TypeError, ValueError): + a["nulls"] += 1 + + stats: List[SensorStat] = [] + for s in targets: + a = acc[s] + vals = a["vals"] + stats.append( + SensorStat( + sensor=s, + count=len(vals), + null_count=a["nulls"], + min=min(vals) if vals else None, + max=max(vals) if vals else None, + mean=statistics.fmean(vals) if vals else None, + stddev=( + statistics.pstdev(vals) + if len(vals) > 1 + else (0.0 if vals else None) + ), + first_timestamp=a["first"], + last_timestamp=a["last"], + ) + ) + return SensorStatsResult( + site_name=site_name, + asset_id=asset_id, + stats=stats, + message=f"stats for {len(stats)} sensor(s) on asset_id {asset_id}.", + ) + + +@mcp.tool(title="Latest Reading") +def latest_reading( + site_name: str, + asset_id: str, + sensor: Optional[str] = None, +) -> Union[LatestReadingResult, ErrorResult]: + """Most recent observation for an asset (all measured sensors, or just `sensor`), plus its age + in seconds for staleness / offline checks.""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + if not db: + return ErrorResult(error="CouchDB not connected") + + doc = _latest_doc(asset_id, sensor=sensor) + if not doc: + return ErrorResult(error=f"unknown asset_id {asset_id} or no records found") + + ts = doc.get("timestamp") + if sensor: + values = {sensor: doc.get(sensor)} + else: + values = {k: v for k, v in doc.items() if k not in RESERVED_FIELDS} + + return LatestReadingResult( + site_name=site_name, + asset_id=asset_id, + timestamp=ts, + values=values, + age_seconds=_age_seconds(ts) if ts else None, + message=f"latest reading for asset_id {asset_id} at {ts}.", + ) + + def main(): # Initialize and run the server mcp.run(transport="stdio") diff --git a/src/servers/iot/tests/conftest.py b/src/servers/iot/tests/conftest.py index b99bcd0c2..a369381b0 100644 --- a/src/servers/iot/tests/conftest.py +++ b/src/servers/iot/tests/conftest.py @@ -32,21 +32,54 @@ def _couchdb_reachable() -> bool: # --- Fixtures --- +@pytest.fixture(autouse=True) +def _reset_caches(monkeypatch): + """The IoT server memoizes registry sites, sensor lists, and asset docs in module globals. + Reset them before every test so a mock in one test can't leak into the next.""" + monkeypatch.setattr("servers.iot.main._registry_sites_cache", None, raising=False) + monkeypatch.setattr("servers.iot.main._sensor_list_cache", {}, raising=False) + monkeypatch.setattr("servers.iot.main._asset_doc_cache", {}, raising=False) + yield + + @pytest.fixture def mock_db(): - """Patch the module-level `db` object in main with a mock.""" + """Patch the telemetry reading DB (`db`) with a mock.""" with patch("servers.iot.main.db") as mock: yield mock @pytest.fixture def no_db(): - """Patch the module-level `db` to None (simulate disconnected CouchDB).""" + """Patch the telemetry DB to None (disconnected).""" with patch("servers.iot.main.db", None): yield +@pytest.fixture +def mock_asset_db(): + """Patch the asset-registry DB (`asset_db`) with a mock.""" + with patch("servers.iot.main.asset_db") as mock: + yield mock + + +@pytest.fixture +def no_asset_db(): + """Patch the asset-registry DB to None (disconnected).""" + with patch("servers.iot.main.asset_db", None): + yield + + +@pytest.fixture +def known_main(monkeypatch): + """Make site 'MAIN' a known site deterministically, WITHOUT hitting asset_db — seed the + registry-sites cache. Use in unit tests so `_is_known_site('MAIN')` passes and site + resolution never touches the network.""" + monkeypatch.setattr("servers.iot.main._registry_sites_cache", ["MAIN"]) + yield + + async def call_tool(mcp_instance, tool_name: str, args: dict) -> dict: - """Helper: call an MCP tool and return parsed JSON response.""" + """Helper: call an MCP tool and return the parsed JSON response.""" contents, _ = await mcp_instance.call_tool(tool_name, args) - return json.loads(contents[0].text) + return json.loads(contents[0].text) \ No newline at end of file diff --git a/src/servers/iot/tests/test_tools.py b/src/servers/iot/tests/test_tools.py index 20b24b593..2cdabd909 100644 --- a/src/servers/iot/tests/test_tools.py +++ b/src/servers/iot/tests/test_tools.py @@ -1,7 +1,13 @@ -"""Tests for IoT MCP server tools. +"""Tests for the IoT MCP server tools (12-tool redesign). -Unit tests use a mocked CouchDB; integration tests require a live instance -(skipped unless COUCHDB_URL is set). +Two databases: `db` = telemetry readings, `asset_db` = asset registry (keyed by `assetnum`, +scoped by `siteid`). Unit tests mock the relevant DB(s); integration tests need a live CouchDB +(skipped unless COUCHDB_URL is set). `known_main` seeds the registry-sites cache so site checks +pass offline. + +`get_asset_doc` now filters on BOTH `assetnum` and `siteid`, so registry mocks that must resolve +a specific (asset, site) pair use `_registry_find`, a fake that honours those selector keys like +CouchDB Mango would. """ import pytest @@ -9,209 +15,403 @@ from .conftest import requires_couchdb, call_tool +def _registry_find(docs): + """A fake `asset_db.find` that filters on the selector's assetnum / siteid / assettype + (like Mango), so site-scoped lookups (get_asset_doc) behave realistically in unit tests.""" + def _find(selector, limit=100000, **kwargs): + out = [ + d for d in docs + if all( + selector.get(k) is None or d.get(k) == selector.get(k) + for k in ("assetnum", "siteid", "assettype") + ) + ] + return {"docs": out[:limit] if limit else out} + + return _find + + # --------------------------------------------------------------------------- # sites # --------------------------------------------------------------------------- - - class TestSites: @pytest.mark.anyio - async def test_returns_main(self): + async def test_fallback_to_default_when_registry_empty(self, no_asset_db): data = await call_tool(mcp, "sites", {}) assert data["sites"] == ["MAIN"] + @pytest.mark.anyio + async def test_discovered_from_registry(self, mock_asset_db): + mock_asset_db.find.return_value = { + "docs": [{"siteid": "MAIN"}, {"siteid": "NORTH"}, {"siteid": "MAIN"}] + } + data = await call_tool(mcp, "sites", {}) + assert data["sites"] == ["MAIN", "NORTH"] + # --------------------------------------------------------------------------- -# assets +# asset_ids (bare IDs from the registry, keyed by assetnum) # --------------------------------------------------------------------------- - - -class TestAssets: +class TestAssetIds: @pytest.mark.anyio - async def test_invalid_site(self): - data = await call_tool(mcp, "assets", {"site_name": "INVALID"}) - assert "error" in data - assert "unknown site" in data["error"] + async def test_invalid_site(self, known_main): + data = await call_tool(mcp, "asset_ids", {"site_name": "INVALID"}) + assert "error" in data and "unknown site" in data["error"] @pytest.mark.anyio - async def test_with_mock_db(self, mock_db): - mock_db.find.return_value = { - "docs": [{"asset_id": "Chiller 1"}, {"asset_id": "Chiller 2"}] + async def test_returns_assetnums(self, known_main, mock_asset_db): + mock_asset_db.find.return_value = { + "docs": [{"assetnum": "PUMP3"}, {"assetnum": "CHILLER 6"}] } - data = await call_tool(mcp, "assets", {"site_name": "MAIN"}) - + data = await call_tool(mcp, "asset_ids", {"site_name": "MAIN"}) assert data["total_assets"] == 2 - assert "Chiller 1" in data["assets"] - assert "Chiller 2" in data["assets"] - mock_db.find.assert_called_once() - - @pytest.mark.anyio - async def test_db_disconnected(self, no_db): - data = await call_tool(mcp, "assets", {"site_name": "MAIN"}) - # Should still return valid JSON (empty assets), not crash - assert "assets" in data or "error" in data + assert data["assets"] == ["CHILLER 6", "PUMP3"] # sorted, List[str] @requires_couchdb @pytest.mark.anyio - async def test_discovery_integration(self): - data = await call_tool(mcp, "assets", {"site_name": "MAIN"}) - assert "assets" in data - assert "Chiller 6" in data["assets"] - assert data["total_assets"] > 0 + async def test_integration(self): + data = await call_tool(mcp, "asset_ids", {"site_name": "MAIN"}) + assert "assets" in data and data["total_assets"] > 0 + assert "CHILLER 6" in data["assets"] # --------------------------------------------------------------------------- -# sensors +# assets (typed metadata rows — distinct from asset_ids) # --------------------------------------------------------------------------- +class TestAssets: + @pytest.mark.anyio + async def test_typed_metadata_rows(self, known_main, mock_asset_db): + mock_asset_db.find.return_value = { + "docs": [ + { + "assetnum": "CHILLER 6", + "siteid": "MAIN", + "assettype": "CHILLER", + "description": "Chiller 6", + "vintage": "2015", + "sensors": ["a", "b", "c"], + } + ] + } + data = await call_tool(mcp, "assets", {"site_name": "MAIN"}) + assert data["total_assets"] == 1 + row = data["assets"][0] + assert row["asset_id"] == "CHILLER 6" + assert row["assettype"] == "CHILLER" + assert row["vintage"] == "2015" + assert row["n_sensors"] == 3 # metadata IS present (the reviewer's R2 fix) - -class TestSensors: @pytest.mark.anyio - async def test_invalid_site(self): - data = await call_tool( - mcp, "sensors", {"site_name": "INVALID", "asset_id": "Chiller 6"} - ) - assert "error" in data + async def test_assettype_filter_passthrough(self, known_main, mock_asset_db): + mock_asset_db.find.return_value = {"docs": []} + await call_tool(mcp, "assets", {"site_name": "MAIN", "assettype": "PUMP"}) + selector = mock_asset_db.find.call_args.args[0] + assert selector.get("assettype") == "PUMP" + assert selector.get("siteid") == "MAIN" + +# --------------------------------------------------------------------------- +# measured_sensors (telemetry — union of columns) +# --------------------------------------------------------------------------- +class TestMeasuredSensors: @pytest.mark.anyio - async def test_unknown_asset(self): + async def test_unknown_asset(self, known_main, mock_db): + mock_db.find.return_value = {"docs": []} data = await call_tool( - mcp, "sensors", {"site_name": "MAIN", "asset_id": "INVALID"} + mcp, "measured_sensors", {"site_name": "MAIN", "asset_id": "NOPE"} ) - assert "error" in data - assert "no sensors found" in data["error"] + assert "error" in data and "no sensors found" in data["error"] @pytest.mark.anyio - async def test_with_mock_db(self, mock_db): + async def test_excludes_reserved_fields(self, known_main, mock_db): mock_db.find.return_value = { "docs": [ - { - "asset_id": "Chiller 1", - "timestamp": "2024-01-01T00:00:00", - "Temp": 25.5, - "Pressure": 10.2, - "_id": "doc1", - "_rev": "rev1", - } + {"asset_id": "A", "timestamp": "t0", "Temp": 1.0, "_id": "x", "_rev": "y"}, + {"asset_id": "A", "timestamp": "t1", "Pressure": 2.0}, ] } data = await call_tool( - mcp, "sensors", {"site_name": "MAIN", "asset_id": "Chiller 1"} + mcp, "measured_sensors", {"site_name": "MAIN", "asset_id": "A"} ) + assert data["total_sensors"] == 2 + assert set(data["sensors"]) == {"Temp", "Pressure"} + assert "_id" not in data["sensors"] and "timestamp" not in data["sensors"] + +# --------------------------------------------------------------------------- +# installed_sensors (registry nameplate list — site-scoped) +# --------------------------------------------------------------------------- +class TestInstalledSensors: + _DOC = {"assetnum": "CHILLER 6", "siteid": "MAIN", "sensors": ["s1", "s2"]} + + @pytest.mark.anyio + async def test_from_registry(self, known_main, mock_asset_db): + mock_asset_db.find.side_effect = _registry_find([self._DOC]) + data = await call_tool( + mcp, "installed_sensors", {"site_name": "MAIN", "asset_id": "CHILLER 6"} + ) assert data["total_sensors"] == 2 - assert "Temp" in data["sensors"] - assert "Pressure" in data["sensors"] - # Internal fields must be excluded - assert "_id" not in data["sensors"] - assert "_rev" not in data["sensors"] + assert data["sensors"] == ["s1", "s2"] - @requires_couchdb @pytest.mark.anyio - async def test_success_integration(self): + async def test_asset_not_at_site(self, monkeypatch, mock_asset_db): + # asset exists at MAIN; querying it via NORTH (a known site) must not resolve it + monkeypatch.setattr("servers.iot.main._registry_sites_cache", ["MAIN", "NORTH"]) + mock_asset_db.find.side_effect = _registry_find([self._DOC]) data = await call_tool( - mcp, "sensors", {"site_name": "MAIN", "asset_id": "Chiller 6"} + mcp, "installed_sensors", {"site_name": "NORTH", "asset_id": "CHILLER 6"} ) - assert "sensors" in data - assert len(data["sensors"]) > 0 - assert any("Power" in s or "Efficiency" in s for s in data["sensors"]) + assert "error" in data and "at site NORTH" in data["error"] # --------------------------------------------------------------------------- -# history +# get_asset_detail (nameplate — site-scoped) # --------------------------------------------------------------------------- +class TestGetAssetDetail: + _DOC = { + "assetnum": "CHILLER 6", + "siteid": "MAIN", + "assettype": "CHILLER", + "status": "OPERATING", + "location": "MAIN-MECH-CH6", + "description": "Chiller 6", + "sensors": ["a", "b"], + } + @pytest.mark.anyio + async def test_unknown_asset(self, known_main, mock_asset_db): + mock_asset_db.find.side_effect = _registry_find([]) + data = await call_tool( + mcp, "get_asset_detail", {"site_name": "MAIN", "asset_id": "NOPE"} + ) + assert "error" in data and "at site MAIN" in data["error"] -class TestHistory: @pytest.mark.anyio - async def test_invalid_date_range(self): + async def test_nameplate_fields(self, known_main, mock_asset_db): + mock_asset_db.find.side_effect = _registry_find([self._DOC]) data = await call_tool( - mcp, - "history", - { - "site_name": "MAIN", - "asset_id": "Chiller 6", - "start": "2020-06-01T00:00:00", - "final": "2020-05-01T00:00:00", - }, + mcp, "get_asset_detail", {"site_name": "MAIN", "asset_id": "CHILLER 6"} ) - assert "error" in data - assert "start >= final" in data["error"] + assert data["asset_id"] == "CHILLER 6" + assert data["assettype"] == "CHILLER" + assert data["status"] == "OPERATING" + assert data["n_sensors"] == 2 @pytest.mark.anyio - async def test_malformed_date(self): + async def test_asset_not_at_site(self, monkeypatch, mock_asset_db): + monkeypatch.setattr("servers.iot.main._registry_sites_cache", ["MAIN", "NORTH"]) + mock_asset_db.find.side_effect = _registry_find([self._DOC]) data = await call_tool( - mcp, - "history", - {"site_name": "MAIN", "asset_id": "Chiller 6", "start": "not-a-date"}, + mcp, "get_asset_detail", {"site_name": "NORTH", "asset_id": "CHILLER 6"} ) - assert "error" in data + assert "error" in data and "at site NORTH" in data["error"] + +# --------------------------------------------------------------------------- +# find_assets_by_sensors +# --------------------------------------------------------------------------- +class TestFindAssetsBySensors: @pytest.mark.anyio - async def test_db_disconnected(self, no_db): + async def test_bad_match(self, known_main): data = await call_tool( mcp, - "history", - { - "site_name": "MAIN", - "asset_id": "Chiller 6", - "start": "2020-06-01T00:00:00", - }, + "find_assets_by_sensors", + {"site_name": "MAIN", "sensors": ["Temp"], "match": "bogus"}, ) - assert "error" in data - assert "not connected" in data["error"].lower() + assert "error" in data and "match must be" in data["error"] @pytest.mark.anyio - async def test_with_mock_db(self, mock_db): + async def test_measured_source_substring(self, known_main, mock_db, mock_asset_db): + # _site_asset_ids -> asset_db returns one asset; get_sensor_list -> db returns its columns + mock_asset_db.find.return_value = {"docs": [{"assetnum": "CHILLER 6"}]} mock_db.find.return_value = { - "docs": [ - {"timestamp": "2024-01-01T00:00:00", "Temp": 20.0}, - {"timestamp": "2024-01-01T00:15:00", "Temp": 21.0}, - ] + "docs": [{"asset_id": "CHILLER 6", "timestamp": "t0", + "Chiller 6 Chiller Efficiency": 0.8}] } data = await call_tool( mcp, - "history", - { - "site_name": "MAIN", - "asset_id": "Chiller 1", - "start": "2024-01-01T00:00:00", - }, + "find_assets_by_sensors", + {"site_name": "MAIN", "sensors": ["Efficiency"], + "substring": True, "source": "measured"}, ) + assert data["total_assets"] == 1 + assert data["assets"][0]["asset_id"] == "CHILLER 6" - assert data["total_observations"] == 2 - assert len(data["observations"]) == 2 - assert data["observations"][0]["Temp"] == 20.0 - @requires_couchdb +# --------------------------------------------------------------------------- +# history (merged + paged; start optional; end/limit/cursor) +# --------------------------------------------------------------------------- +class TestHistory: @pytest.mark.anyio - async def test_open_ended_integration(self): + async def test_malformed_date(self, known_main): data = await call_tool( - mcp, - "history", - { - "site_name": "MAIN", - "asset_id": "Chiller 6", - "start": "2020-06-01T00:00:00", - }, + mcp, "history", + {"site_name": "MAIN", "asset_id": "A", "start": "not-a-date"}, ) - assert "observations" in data - assert "total_observations" in data + assert "error" in data and "Invalid date" in data["error"] + + @pytest.mark.anyio + async def test_start_after_end(self, known_main): + data = await call_tool( + mcp, "history", + {"site_name": "MAIN", "asset_id": "A", + "start": "2020-06-02T00:00:00", "end": "2020-06-01T00:00:00"}, + ) + assert "error" in data and "start >= end" in data["error"] + + @pytest.mark.anyio + async def test_db_disconnected(self, known_main, no_db): + data = await call_tool( + mcp, "history", {"site_name": "MAIN", "asset_id": "A"} + ) + assert "error" in data and "not connected" in data["error"].lower() + + @pytest.mark.anyio + async def test_start_optional(self, known_main, mock_db): + # start omitted → full range, no error + mock_db.find.return_value = {"docs": [{"timestamp": "t0", "Temp": 1.0}]} + data = await call_tool(mcp, "history", {"site_name": "MAIN", "asset_id": "A"}) + assert data["returned"] == 1 + assert data["has_more"] is False + assert data["next_cursor"] is None + + @pytest.mark.anyio + async def test_paging_has_more(self, known_main, mock_db): + mock_db.find.return_value = { + "docs": [{"timestamp": "t0"}, {"timestamp": "t1"}], + "bookmark": "cursor-2", + } + data = await call_tool( + mcp, "history", {"site_name": "MAIN", "asset_id": "A", "limit": 2} + ) + assert data["returned"] == 2 + assert data["has_more"] is True # full page => more may follow + assert data["next_cursor"] == "cursor-2" + + @pytest.mark.anyio + async def test_sensors_projection(self, known_main, mock_db): + mock_db.find.return_value = {"docs": [{"timestamp": "t0", "Temp": 1.0}]} + await call_tool( + mcp, "history", + {"site_name": "MAIN", "asset_id": "A", "sensors": ["Temp"]}, + ) + kwargs = mock_db.find.call_args.kwargs + assert "Temp" in kwargs.get("fields", []) + assert "timestamp" in kwargs.get("fields", []) @requires_couchdb @pytest.mark.anyio async def test_bounded_range_integration(self): data = await call_tool( - mcp, - "history", - { - "site_name": "MAIN", - "asset_id": "Chiller 6", - "start": "2020-06-01T00:00:00", - "final": "2020-06-01T01:00:00", - }, + mcp, "history", + {"site_name": "MAIN", "asset_id": "CHILLER 6", + "start": "2020-06-01T00:00:00", "end": "2020-06-01T01:00:00"}, ) assert "observations" in data for obs in data["observations"]: assert obs["timestamp"] >= "2020-06-01T00:00:00" assert obs["timestamp"] < "2020-06-01T01:00:00" + + +# --------------------------------------------------------------------------- +# stream_extent (bounds + count pre-flight) +# --------------------------------------------------------------------------- +class TestStreamExtent: + @pytest.mark.anyio + async def test_bounds_and_count(self, known_main, mock_db): + mock_db.find.return_value = { + "docs": [{"timestamp": "t0"}, {"timestamp": "t1"}, {"timestamp": "t2"}] + } + data = await call_tool( + mcp, "stream_extent", {"site_name": "MAIN", "asset_id": "A"} + ) + assert data["total_records"] == 3 + assert data["start_time"] == "t0" + assert data["end_time"] == "t2" + assert data["exceeds_page_limit"] is False + + @pytest.mark.anyio + async def test_no_records(self, known_main, mock_db): + mock_db.find.return_value = {"docs": []} + data = await call_tool( + mcp, "stream_extent", {"site_name": "MAIN", "asset_id": "A"} + ) + assert "error" in data and "no records" in data["error"] + + +# --------------------------------------------------------------------------- +# sensor_coverage (per-sensor counts) +# --------------------------------------------------------------------------- +class TestSensorCoverage: + @pytest.mark.anyio + async def test_counts_per_sensor(self, known_main, mock_db): + mock_db.find.return_value = { + "docs": [ + {"timestamp": "t0", "Temp": 1.0}, + {"timestamp": "t1", "Temp": 2.0, "Pressure": 5.0}, + ] + } + data = await call_tool( + mcp, "sensor_coverage", {"site_name": "MAIN", "asset_id": "A"} + ) + cov = {c["sensor"]: c for c in data["sensors"]} + assert cov["Temp"]["non_null_count"] == 2 + assert cov["Pressure"]["non_null_count"] == 1 + assert cov["Temp"]["first_timestamp"] == "t0" + assert cov["Temp"]["last_timestamp"] == "t1" + + +# --------------------------------------------------------------------------- +# sensor_stats (sensor optional → all sensors) +# --------------------------------------------------------------------------- +class TestSensorStats: + @pytest.mark.anyio + async def test_single_sensor(self, known_main, mock_db): + mock_db.find.return_value = { + "docs": [ + {"timestamp": "t0", "Temp": 10.0}, + {"timestamp": "t1", "Temp": 20.0}, + ] + } + data = await call_tool( + mcp, "sensor_stats", + {"site_name": "MAIN", "asset_id": "A", "sensor": "Temp"}, + ) + stats = {s["sensor"]: s for s in data["stats"]} + assert stats["Temp"]["count"] == 2 + assert stats["Temp"]["min"] == 10.0 + assert stats["Temp"]["max"] == 20.0 + assert stats["Temp"]["mean"] == 15.0 + + @pytest.mark.anyio + async def test_all_sensors_when_omitted(self, known_main, mock_db): + mock_db.find.return_value = { + "docs": [ + {"timestamp": "t0", "Temp": 10.0, "Pressure": 1.0}, + {"timestamp": "t1", "Temp": 20.0, "Pressure": 3.0}, + ] + } + data = await call_tool( + mcp, "sensor_stats", {"site_name": "MAIN", "asset_id": "A"} + ) + names = {s["sensor"] for s in data["stats"]} + assert names == {"Temp", "Pressure"} + + +# --------------------------------------------------------------------------- +# latest_reading +# --------------------------------------------------------------------------- +class TestLatestReading: + @pytest.mark.anyio + async def test_newest_values(self, known_main, mock_db): + mock_db.find.return_value = { + "docs": [{"asset_id": "A", "timestamp": "2024-01-01T00:00:00", + "Temp": 42.0, "_id": "x", "_rev": "y"}] + } + data = await call_tool( + mcp, "latest_reading", {"site_name": "MAIN", "asset_id": "A"} + ) + assert data["timestamp"] == "2024-01-01T00:00:00" + assert data["values"]["Temp"] == 42.0 + assert "_id" not in data["values"] # reserved fields excluded + assert "age_seconds" in data \ No newline at end of file