From 9b6a45bbd346b59ca280d3f752fc2096669fce54 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Wed, 8 Jul 2026 22:48:24 -0400 Subject: [PATCH 01/14] revised code Signed-off-by: Dhaval Patel --- src/servers/iot/main.py | 592 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 569 insertions(+), 23 deletions(-) diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index 7d2ba944..6df94f89 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -1,8 +1,9 @@ import os import logging -from datetime import datetime +import statistics +from datetime import datetime, timezone from functools import lru_cache -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, Iterator, List, Optional, Tuple, Union from mcp.server.fastmcp import FastMCP from pydantic import BaseModel import couchdb3 @@ -66,6 +67,12 @@ 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 @@ -129,6 +136,83 @@ class RegistryAssetsResult(BaseModel): message: str +# ── New tool result models ── +class AssetSensorMatch(BaseModel): + asset_id: str + matched_sensors: List[str] + + +class FindAssetsResult(BaseModel): + site_name: str + query_sensors: List[str] + match: str + source: str + total_assets: int + assets: List[AssetSensorMatch] + message: str + + +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 PagedHistoryResult(BaseModel): + site_name: str + asset_id: str + total_in_page: int + observations: List[Dict[str, Any]] + next_bookmark: Optional[str] + has_more: bool + message: str + + +class SensorStatsResult(BaseModel): + site_name: str + asset_id: str + 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] + message: str + + +class LatestReadingResult(BaseModel): + site_name: str + asset_id: str + timestamp: Optional[str] + values: Dict[str, Any] + age_seconds: Optional[float] + message: str + + _asset_list_cache: Optional[List[str]] = None @@ -246,6 +330,143 @@ 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], final: 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 final is not None: + datetime.fromisoformat(final) + except ValueError as e: + return f"Invalid date format: {e}" + if start is not None and final is not None and start >= final: + return "start >= final" + return None + + +def _time_selector( + asset_id: str, start: Optional[str], final: 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 final is not None: + ts["$lt"] = datetime.fromisoformat(final).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 (iot_asset_id where present, else assetnum) — mirrors assets().""" + if not asset_db: + return [] + try: + res = asset_db.find( + {"doctype": "asset", "siteid": site_name}, + fields=["assetnum", "iot_asset_id"], + limit=100000, + ) + return sorted((d.get("iot_asset_id") or d.get("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) -> List[str]: + doc = get_asset_doc(asset_id) + 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 +475,12 @@ 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`. +@mcp.tool(title="List Asset IDs") +def asset_ids(site_name: str) -> Union[AssetsResult, ErrorResult]: + """Returns the asset IDs 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. + registry `assetnum` — so the id works with measured_sensors()/history() when telemetry exists. + For assets with metadata (type, vintage, sensor count), use assets(). """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -282,12 +504,11 @@ def assets(site_name: str) -> Union[AssetsResult, ErrorResult]: 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. For the full + INSTALLED inventory (including sensors fitted but not streaming), use installed_sensors().""" if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -348,11 +569,11 @@ def history( return ErrorResult(error=str(e)) -@mcp.tool(title="Get Asset") -def get_asset(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResult]: +@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 (Maximo MXASSET-aligned: description, assettype, status, location, installdate, vintage) plus installed sensor count. - This is asset IDENTITY — distinct from the telemetry-derived assets() list.""" + This is asset IDENTITY — distinct from the telemetry-derived asset_ids() list.""" if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") doc = get_asset_doc(asset_id) @@ -386,13 +607,13 @@ 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) @@ -408,13 +629,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.""" + filtered by assettype (e.g. 'PUMP', 'COMPRESSOR'). Complements asset_ids(), which returns bare + ids derived from telemetry.""" if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -455,6 +676,331 @@ def registry_assets( 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) + ) + + 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, + final: Optional[str] = None, +) -> Union[StreamExtentResult, ErrorResult]: + """Time bounds + record count for an asset's stream (optionally one sensor and/or a window), + so callers can size a history() request and know if it exceeds the 1000-row page limit.""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + err = _validate_dates(start, final) + if err: + return ErrorResult(error=err) + if not db: + return ErrorResult(error="CouchDB not connected") + + selector = _time_selector(asset_id, start, final) + 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="Get Sensor History (Paged)") +def history_paged( + site_name: str, + asset_id: str, + start: str, + final: Optional[str] = None, + sensors: Optional[List[str]] = None, + page_size: int = PAGE_SIZE, + bookmark: Optional[str] = None, +) -> Union[PagedHistoryResult, ErrorResult]: + """Like history() but paginated past the 1000-row limit via a CouchDB bookmark cursor. + Pass `sensors` to project only those columns. Call repeatedly, feeding back next_bookmark, + until has_more is false.""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + err = _validate_dates(start, final) + if err: + return ErrorResult(error=err) + if not db: + return ErrorResult(error="CouchDB not connected") + page_size = max(1, min(page_size, PAGE_SIZE)) + + selector = _time_selector(asset_id, start, final) + fields: Optional[List[str]] = None + if sensors: + fields = ["_id", "asset_id", "timestamp"] + list(sensors) + + kwargs: Dict[str, Any] = { + "limit": page_size, + "sort": [{"asset_id": "asc"}, {"timestamp": "asc"}], + } + if fields is not None: + kwargs["fields"] = fields + if bookmark is not None: + kwargs["bookmark"] = bookmark + + try: + 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_bookmark = res.get("bookmark") + has_more = len(docs) == page_size + return PagedHistoryResult( + site_name=site_name, + asset_id=asset_id, + total_in_page=len(docs), + observations=docs, + next_bookmark=next_bookmark if has_more else None, + has_more=has_more, + message=f"page of {len(docs)} observation(s) for asset_id {asset_id}; has_more={has_more}.", + ) + + +@mcp.tool(title="Sensor Coverage") +def sensor_coverage( + site_name: str, + asset_id: str, + sample_limit: 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. + sample_limit=0 scans all docs.""" + 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=sample_limit if sample_limit > 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: str, + start: Optional[str] = None, + final: Optional[str] = None, +) -> Union[SensorStatsResult, ErrorResult]: + """Numeric summary (count/min/max/mean/stddev) for one sensor over an optional time window, + without returning the raw rows.""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + err = _validate_dates(start, final) + if err: + return ErrorResult(error=err) + if not db: + return ErrorResult(error="CouchDB not connected") + + selector = _time_selector(asset_id, start, final) + selector[sensor] = {"$exists": True} + + values: List[float] = [] + null_count = 0 + first_ts: Optional[str] = None + last_ts: Optional[str] = None + + for doc in _iter_docs(selector, fields=["timestamp", sensor]): + ts = doc.get("timestamp") + if ts is not None: + if first_ts is None: + first_ts = ts + last_ts = ts + v = doc.get(sensor) + if v is None: + null_count += 1 + continue + try: + values.append(float(v)) + except (TypeError, ValueError): + null_count += 1 + + if not values and null_count == 0: + return ErrorResult(error=f"no records for asset_id {asset_id} sensor {sensor}") + + stddev = statistics.pstdev(values) if len(values) > 1 else 0.0 + return SensorStatsResult( + site_name=site_name, + asset_id=asset_id, + sensor=sensor, + count=len(values), + null_count=null_count, + min=min(values) if values else None, + max=max(values) if values else None, + mean=statistics.fmean(values) if values else None, + stddev=stddev if values else None, + first_timestamp=first_ts, + last_timestamp=last_ts, + message=f"{len(values)} value(s) for sensor {sensor} on asset_id {asset_id} " + f"({null_count} null/non-numeric).", + ) + + +@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") From 907342f440442c684fd08af44c8a196e7e8f8ee0 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Wed, 8 Jul 2026 23:08:24 -0400 Subject: [PATCH 02/14] revised code Signed-off-by: Dhaval Patel --- src/servers/iot/main.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index 6df94f89..3410a804 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -59,8 +59,8 @@ 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 " + "query historical readings from CouchDB. NOTE: asset_ids()/measured_sensors() reflect TELEMETRY " + "(what streams = measured); get_asset_detail()/installed_sensors()/assets() reflect the REGISTRY " "(what is installed, by name). Compare the two to find installed-but-not-streaming sensors." ), ) From 5676cfbee3ac7e0c37741b7691da03689ce63e51 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Wed, 8 Jul 2026 23:13:36 -0400 Subject: [PATCH 03/14] revised Signed-off-by: Dhaval Patel --- src/servers/iot/main.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index 3410a804..6d647d6a 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -444,7 +444,7 @@ def _age_seconds(ts_iso: str) -> Optional[float]: def _site_asset_ids(site_name: str) -> List[str]: - """Asset ids registered at a site (iot_asset_id where present, else assetnum) — mirrors assets().""" + """Asset ids registered at a site (iot_asset_id where present, else assetnum) — mirrors asset_ids().""" if not asset_db: return [] try: @@ -500,7 +500,7 @@ def asset_ids(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)) @@ -672,7 +672,7 @@ def assets( + ".", ) except Exception as e: - logger.error(f"registry_assets failed: {e}") + logger.error(f"assets failed: {e}") return ErrorResult(error=str(e)) From fbe66c72c1fe2fbbbf15a10b9668d62f74344ad7 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 00:29:06 -0400 Subject: [PATCH 04/14] revised docs Signed-off-by: Dhaval Patel --- INSTRUCTIONS.md | 8 ++++---- README.md | 2 +- docs/mcp-servers.md | 41 ++++++++++++++++++++++++++--------------- 3 files changed, 31 insertions(+), 20 deletions(-) diff --git a/INSTRUCTIONS.md b/INSTRUCTIONS.md index 2f7b6769..1c37f0ed 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` | 13 | 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 d58342ea..41779d9c 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_sensors`, `get_failure_modes`, `get_failure_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 cf79790c..3e041623 100644 --- a/docs/mcp-servers.md +++ b/docs/mcp-servers.md @@ -13,14 +13,19 @@ 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. 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. +Additional telemetry tools help plan and page history queries: `stream_extent` and `sensor_coverage` +report a stream's time bounds and per-sensor record counts, `history_paged` walks results past the +1000-row page cap, and `sensor_stats`, `latest_reading`, and `find_assets_by_sensors` provide quick +summaries and reverse lookup. **Path:** `src/servers/iot/main.py` **Requires:** CouchDB (`COUCHDB_URL`, `COUCHDB_USERNAME`, `COUCHDB_PASSWORD`, `IOT_DBNAME`, `ASSET_DBNAME`) @@ -35,15 +40,21 @@ 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 asset ids registered at a site (telemetry id where present, else `assetnum`) | +| `measured_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); capped at 1000 rows | +| `get_asset_detail` | `site_name`, `asset_id` | Registry/nameplate detail for one asset (description, assettype, status, location, vintage, installed count) | +| `installed_sensors` | `site_name`, `asset_id` | List the **installed** sensors for an asset, by name (registry inventory) | +| `assets` | `site_name`, `assettype?` | List registry assets with metadata (assettype, vintage, sensor count), optionally filtered by assettype | +| `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` | +| `stream_extent` | `site_name`, `asset_id`, `sensor?`, `start?`, `final?` | Time bounds + record count for a stream; flags if it exceeds the 1000-row page cap | +| `history_paged` | `site_name`, `asset_id`, `start`, `final?`, `sensors?`, `page_size?`, `bookmark?` | Paginated `history` past the 1000-row cap via a CouchDB bookmark cursor; optional column projection | +| `sensor_coverage` | `site_name`, `asset_id`, `sample_limit?` | Per-measured-sensor record counts and first/last timestamps | +| `sensor_stats` | `site_name`, `asset_id`, `sensor`, `start?`, `final?` | Numeric summary (count/min/max/mean/stddev) for one sensor over an optional window | +| `latest_reading` | `site_name`, `asset_id`, `sensor?` | Most recent observation for an asset (+ age in seconds for staleness) | ## utilities — Utilities From 745907355e34f029f101ee835c5594b44535cb18 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 13:08:30 -0400 Subject: [PATCH 05/14] review the tool Signed-off-by: Dhaval Patel --- src/servers/iot/main.py | 352 +++++++++++++++++++--------------------- 1 file changed, 165 insertions(+), 187 deletions(-) diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index 6d647d6a..21eba20d 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -100,10 +100,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 @@ -129,10 +131,18 @@ 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 @@ -179,19 +189,7 @@ class StreamExtentResult(BaseModel): message: str -class PagedHistoryResult(BaseModel): - site_name: str - asset_id: str - total_in_page: int - observations: List[Dict[str, Any]] - next_bookmark: Optional[str] - has_more: bool - message: str - - -class SensorStatsResult(BaseModel): - site_name: str - asset_id: str +class SensorStat(BaseModel): sensor: str count: int null_count: int @@ -201,6 +199,12 @@ class SensorStatsResult(BaseModel): 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 @@ -278,10 +282,9 @@ def get_sensor_list(asset_id: str) -> List[str]: 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.""" + """Fetch one asset-registry document by asset_id. Cached per asset_id.""" + # impl: resolves across the internal id spaces (assetnum / iot_asset_id / wo_assetnum) so the + # same profile is found whether the caller holds the telemetry id or the work-order id. if asset_id in _asset_doc_cache: return _asset_doc_cache[asset_id] if not asset_db: @@ -333,29 +336,29 @@ def _is_known_site(site_name: str) -> bool: # --------------------------------------------------------------------------- # Shared helpers for the new tools # --------------------------------------------------------------------------- -def _validate_dates(start: Optional[str], final: Optional[str]) -> Optional[str]: +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 final is not None: - datetime.fromisoformat(final) + if end is not None: + datetime.fromisoformat(end) except ValueError as e: - return f"Invalid date format: {e}" - if start is not None and final is not None and start >= final: - return "start >= final" + 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], final: Optional[str] + 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 final is not None: - ts["$lt"] = datetime.fromisoformat(final).isoformat() + if end is not None: + ts["$lt"] = datetime.fromisoformat(end).isoformat() if ts: selector["timestamp"] = ts return selector @@ -444,7 +447,7 @@ def _age_seconds(ts_iso: str) -> Optional[float]: def _site_asset_ids(site_name: str) -> List[str]: - """Asset ids registered at a site (iot_asset_id where present, else assetnum) — mirrors asset_ids().""" + """Asset ids registered at a site — mirrors asset_ids().""" if not asset_db: return [] try: @@ -477,10 +480,9 @@ def sites() -> SitesResult: @mcp.tool(title="List Asset IDs") def asset_ids(site_name: str) -> Union[AssetsResult, ErrorResult]: - """Returns the asset IDs 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 measured_sensors()/history() when telemetry exists. - For assets with metadata (type, vintage, sensor count), use assets(). + """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, ...). For assets with metadata + (description, type, vintage, installed sensor count), use `assets`. """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -527,47 +529,70 @@ def measured_sensors(site_name: str, asset_id: str) -> Union[SensorsResult, Erro @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 = _time_selector(asset_id, start, end) + fields: Optional[List[str]] = None + if sensors: + fields = ["_id", "asset_id", "timestamp"] + list(sensors) - selector = { - "asset_id": asset_id, - "timestamp": {"$gte": start_iso}, + 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 Detail") def get_asset_detail(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResult]: @@ -632,11 +657,10 @@ def installed_sensors( @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 asset_ids(), 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: @@ -652,22 +676,22 @@ def 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 "") + ".", ) @@ -734,25 +758,26 @@ def _hits(w: str) -> List[str]: ) -@mcp.tool(title="Stream Extent") def stream_extent( site_name: str, asset_id: str, sensor: Optional[str] = None, start: Optional[str] = None, - final: 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), - so callers can size a history() request and know if it exceeds the 1000-row page limit.""" + """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, final) + 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, final) + selector = _time_selector(asset_id, start, end) if sensor: selector[sensor] = {"$exists": True, "$ne": None} @@ -794,71 +819,16 @@ def stream_extent( ) -@mcp.tool(title="Get Sensor History (Paged)") -def history_paged( - site_name: str, - asset_id: str, - start: str, - final: Optional[str] = None, - sensors: Optional[List[str]] = None, - page_size: int = PAGE_SIZE, - bookmark: Optional[str] = None, -) -> Union[PagedHistoryResult, ErrorResult]: - """Like history() but paginated past the 1000-row limit via a CouchDB bookmark cursor. - Pass `sensors` to project only those columns. Call repeatedly, feeding back next_bookmark, - until has_more is false.""" - if not _is_known_site(site_name): - return ErrorResult(error=f"unknown site {site_name}") - err = _validate_dates(start, final) - if err: - return ErrorResult(error=err) - if not db: - return ErrorResult(error="CouchDB not connected") - page_size = max(1, min(page_size, PAGE_SIZE)) - - selector = _time_selector(asset_id, start, final) - fields: Optional[List[str]] = None - if sensors: - fields = ["_id", "asset_id", "timestamp"] + list(sensors) - - kwargs: Dict[str, Any] = { - "limit": page_size, - "sort": [{"asset_id": "asc"}, {"timestamp": "asc"}], - } - if fields is not None: - kwargs["fields"] = fields - if bookmark is not None: - kwargs["bookmark"] = bookmark - - try: - 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_bookmark = res.get("bookmark") - has_more = len(docs) == page_size - return PagedHistoryResult( - site_name=site_name, - asset_id=asset_id, - total_in_page=len(docs), - observations=docs, - next_bookmark=next_bookmark if has_more else None, - has_more=has_more, - message=f"page of {len(docs)} observation(s) for asset_id {asset_id}; has_more={has_more}.", - ) - - @mcp.tool(title="Sensor Coverage") def sensor_coverage( site_name: str, asset_id: str, - sample_limit: int = 5000, + 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. - sample_limit=0 scans all docs.""" + 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: @@ -872,7 +842,7 @@ def sensor_coverage( for doc in _iter_docs( {"asset_id": asset_id}, sort=[{"asset_id": "asc"}, {"timestamp": "asc"}], - max_docs=sample_limit if sample_limit > 0 else None, + max_docs=max_scan_docs if max_scan_docs > 0 else None, ): scanned += 1 ts = doc.get("timestamp") @@ -910,61 +880,69 @@ def sensor_coverage( def sensor_stats( site_name: str, asset_id: str, - sensor: str, + sensor: Optional[str] = None, start: Optional[str] = None, - final: Optional[str] = None, + end: Optional[str] = None, ) -> Union[SensorStatsResult, ErrorResult]: - """Numeric summary (count/min/max/mean/stddev) for one sensor over an optional time window, - without returning the raw rows.""" + """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, final) + 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, final) - selector[sensor] = {"$exists": True} - - values: List[float] = [] - null_count = 0 - first_ts: Optional[str] = None - last_ts: Optional[str] = None + 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") - for doc in _iter_docs(selector, fields=["timestamp", sensor]): + 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") - if ts is not None: - if first_ts is None: - first_ts = ts - last_ts = ts - v = doc.get(sensor) - if v is None: - null_count += 1 - continue - try: - values.append(float(v)) - except (TypeError, ValueError): - null_count += 1 - - if not values and null_count == 0: - return ErrorResult(error=f"no records for asset_id {asset_id} sensor {sensor}") - - stddev = statistics.pstdev(values) if len(values) > 1 else 0.0 + 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, - sensor=sensor, - count=len(values), - null_count=null_count, - min=min(values) if values else None, - max=max(values) if values else None, - mean=statistics.fmean(values) if values else None, - stddev=stddev if values else None, - first_timestamp=first_ts, - last_timestamp=last_ts, - message=f"{len(values)} value(s) for sensor {sensor} on asset_id {asset_id} " - f"({null_count} null/non-numeric).", + stats=stats, + message=f"stats for {len(stats)} sensor(s) on asset_id {asset_id}.", ) From 420398dbee86308343c1a9a9256d0c8f5d5f0d28 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 13:17:06 -0400 Subject: [PATCH 06/14] adding decorateor tool Signed-off-by: Dhaval Patel --- src/servers/iot/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index 21eba20d..ef489a69 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -757,7 +757,7 @@ def _hits(w: str) -> List[str]: f"(match={match}, substring={substring}, source={source}).", ) - +@mcp.tool(title="Stream Extent") def stream_extent( site_name: str, asset_id: str, From dee1984f3fc49d4a170cabb511bcecf88d55b5b4 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 13:22:02 -0400 Subject: [PATCH 07/14] revised docs Signed-off-by: Dhaval Patel --- INSTRUCTIONS.md | 2 +- docs/mcp-servers.md | 38 +++++++++++++++++++------------------- 2 files changed, 20 insertions(+), 20 deletions(-) diff --git a/INSTRUCTIONS.md b/INSTRUCTIONS.md index 1c37f0ed..b9fff039 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` | 13 | 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 | diff --git a/docs/mcp-servers.md b/docs/mcp-servers.md index 3e041623..c322f4b2 100644 --- a/docs/mcp-servers.md +++ b/docs/mcp-servers.md @@ -22,10 +22,11 @@ REGISTRY — the asset nameplate and the **installed** sensor inventory (by name `installed_sensors()` against `measured_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. -Additional telemetry tools help plan and page history queries: `stream_extent` and `sensor_coverage` -report a stream's time bounds and per-sensor record counts, `history_paged` walks results past the -1000-row page cap, and `sensor_stats`, `latest_reading`, and `find_assets_by_sensors` provide quick -summaries and reverse lookup. +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`) @@ -40,21 +41,20 @@ summaries and reverse lookup. 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`) | -| `asset_ids` | `site_name` | List asset ids registered at a site (telemetry id where present, else `assetnum`) | -| `measured_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); capped at 1000 rows | -| `get_asset_detail` | `site_name`, `asset_id` | Registry/nameplate detail for one asset (description, assettype, status, location, vintage, installed count) | -| `installed_sensors` | `site_name`, `asset_id` | List the **installed** sensors for an asset, by name (registry inventory) | -| `assets` | `site_name`, `assettype?` | List registry assets with metadata (assettype, vintage, sensor count), optionally filtered by assettype | -| `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` | -| `stream_extent` | `site_name`, `asset_id`, `sensor?`, `start?`, `final?` | Time bounds + record count for a stream; flags if it exceeds the 1000-row page cap | -| `history_paged` | `site_name`, `asset_id`, `start`, `final?`, `sensors?`, `page_size?`, `bookmark?` | Paginated `history` past the 1000-row cap via a CouchDB bookmark cursor; optional column projection | -| `sensor_coverage` | `site_name`, `asset_id`, `sample_limit?` | Per-measured-sensor record counts and first/last timestamps | -| `sensor_stats` | `site_name`, `asset_id`, `sensor`, `start?`, `final?` | Numeric summary (count/min/max/mean/stddev) for one sensor over an optional window | -| `latest_reading` | `site_name`, `asset_id`, `sensor?` | Most recent observation for an asset (+ age in seconds for staleness) | +| 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 From d7e6d75f45719960fbab6dcf9e8223ae55e8bdc8 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 15:21:42 -0400 Subject: [PATCH 08/14] revised scenarios Signed-off-by: Dhaval Patel --- src/couchdb/collections.json | 16 ------------ .../shared/iot/asset_profile_sample.json | 26 +------------------ 2 files changed, 1 insertion(+), 41 deletions(-) diff --git a/src/couchdb/collections.json b/src/couchdb/collections.json index db67d67f..8303bc85 100644 --- a/src/couchdb/collections.json +++ b/src/couchdb/collections.json @@ -78,22 +78,6 @@ "id_prefix": "asset", "doctype": "asset", "indexes": [ - [ - "doctype", - "assetnum" - ], - [ - "doctype", - "assettype" - ], - [ - "doctype", - "siteid" - ], - [ - "doctype", - "iot_asset_id" - ] ] }, "vibration": { 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 24acd967..db9f1498 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", From e37d86352a4ec21f826a78bf8c8221943e6a77bb Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 16:06:10 -0400 Subject: [PATCH 09/14] revised code Signed-off-by: Dhaval Patel --- src/servers/iot/main.py | 28 ++++++++++++---------------- 1 file changed, 12 insertions(+), 16 deletions(-) diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index ef489a69..03291322 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -280,28 +280,25 @@ def get_sensor_list(asset_id: str) -> List[str]: _asset_doc_cache: Dict[str, Dict[str, Any]] = {} - def get_asset_doc(asset_id: str) -> Optional[Dict[str, Any]]: """Fetch one asset-registry document by asset_id. Cached per asset_id.""" - # impl: resolves across the internal id spaces (assetnum / iot_asset_id / wo_assetnum) so the - # same profile is found whether the caller holds the telemetry id or the work-order id. + # impl: registry docs are keyed by `assetnum`, which holds the same value as the telemetry + # `asset_id` (e.g. "CHILLER 6"), so one equality query resolves the profile. if asset_id in _asset_doc_cache: return _asset_doc_cache[asset_id] 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] + res = asset_db.find({"doctype": "asset", "assetnum": asset_id}, limit=1) + docs = res["docs"] + if docs: + _asset_doc_cache[asset_id] = docs[0] + return docs[0] return None except Exception as e: logger.error(f"Error fetching asset doc {asset_id}: {e}") return None - - + _registry_sites_cache: Optional[List[str]] = None @@ -453,15 +450,14 @@ def _site_asset_ids(site_name: str) -> List[str]: try: res = asset_db.find( {"doctype": "asset", "siteid": site_name}, - fields=["assetnum", "iot_asset_id"], + fields=["assetnum"], limit=100000, ) - return sorted((d.get("iot_asset_id") or d.get("assetnum")) for d in res["docs"]) + 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) -> List[str]: doc = get_asset_doc(asset_id) return list(doc.get("sensors", [])) if doc else [] @@ -491,10 +487,10 @@ def asset_ids(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), From a31429474b65220499fca0debf22a6950101c530 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 16:11:12 -0400 Subject: [PATCH 10/14] revised main Signed-off-by: Dhaval Patel --- src/servers/iot/main.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index 03291322..bb5b12fd 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -59,9 +59,7 @@ 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: asset_ids()/measured_sensors() reflect TELEMETRY " - "(what streams = measured); get_asset_detail()/installed_sensors()/assets() reflect the REGISTRY " - "(what is installed, by name). Compare the two to find installed-but-not-streaming sensors." + "query historical readings." ), ) From 5e3b61e0a7e2d0d73eba92da0b770595d1fcd3b9 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 17:43:16 -0400 Subject: [PATCH 11/14] revised code Signed-off-by: Dhaval Patel --- src/servers/iot/main.py | 118 +++++++++++++++++----------------------- 1 file changed, 51 insertions(+), 67 deletions(-) diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index bb5b12fd..bdf65bcc 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -134,7 +134,7 @@ class AssetSummary(BaseModel): description: Optional[str] assettype: Optional[str] vintage: Optional[str] - n_sensors: int # installed sensor count + n_sensors: int # installed sensor count class AssetsWithMetadataResult(BaseModel): @@ -202,7 +202,7 @@ class SensorStat(BaseModel): class SensorStatsResult(BaseModel): site_name: str asset_id: str - stats: List[SensorStat] # one entry if `sensor` given, else every measured sensor + stats: List[SensorStat] # one entry if `sensor` given, else every measured sensor message: str @@ -215,32 +215,6 @@ class LatestReadingResult(BaseModel): message: str -_asset_list_cache: Optional[List[str]] = None - - -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 - - 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 [] - - _sensor_list_cache: Dict[str, List[str]] = {} @@ -259,16 +233,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: @@ -278,25 +254,29 @@ def get_sensor_list(asset_id: str) -> List[str]: _asset_doc_cache: Dict[str, Dict[str, Any]] = {} -def get_asset_doc(asset_id: str) -> Optional[Dict[str, Any]]: - """Fetch one asset-registry document by asset_id. Cached per asset_id.""" - # impl: registry docs are keyed by `assetnum`, which holds the same value as the telemetry - # `asset_id` (e.g. "CHILLER 6"), so one equality query resolves the profile. - 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: - res = asset_db.find({"doctype": "asset", "assetnum": asset_id}, limit=1) + 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"] - if docs: - _asset_doc_cache[asset_id] = docs[0] - return docs[0] - return None + 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 @@ -308,7 +288,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 @@ -456,11 +436,10 @@ def _site_asset_ids(site_name: str) -> List[str]: logger.error(f"_site_asset_ids failed: {e}") return [] -def _installed_sensors(asset_id: str) -> List[str]: - doc = get_asset_doc(asset_id) +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) # --------------------------------------------------------------------------- @@ -475,8 +454,7 @@ def sites() -> SitesResult: @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, ...). For assets with metadata - (description, type, vintage, installed sensor count), use `assets`. + other tools (measured_sensors, history, stream_extent, ...). """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -501,10 +479,12 @@ def asset_ids(site_name: str) -> Union[AssetsResult, ErrorResult]: @mcp.tool(title="List Measured Sensors") -def measured_sensors(site_name: str, asset_id: str) -> Union[SensorsResult, ErrorResult]: +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. For the full - INSTALLED inventory (including sensors fitted but not streaming), use installed_sensors().""" + 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}") @@ -588,16 +568,15 @@ def history( message=f"{len(docs)} observation(s) for asset_id {asset_id}; has_more={has_more}.", ) + @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 (Maximo MXASSET-aligned: description, - assettype, status, location, installdate, vintage) plus installed sensor count. - This is asset IDENTITY — distinct from the telemetry-derived asset_ids() list.""" + """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") @@ -635,9 +614,9 @@ def installed_sensors( 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, @@ -724,7 +703,7 @@ def find_assets_by_sensors( available = ( get_sensor_list(asset_id) if source == "measured" - else _installed_sensors(asset_id) + else _installed_sensors(asset_id, site_name) ) def _hits(w: str) -> List[str]: @@ -751,6 +730,7 @@ def _hits(w: str) -> List[str]: f"(match={match}, substring={substring}, source={source}).", ) + @mcp.tool(title="Stream Extent") def stream_extent( site_name: str, @@ -927,7 +907,11 @@ def sensor_stats( 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)), + stddev=( + statistics.pstdev(vals) + if len(vals) > 1 + else (0.0 if vals else None) + ), first_timestamp=a["first"], last_timestamp=a["last"], ) From 74cae8233d048722287962f599f1e92b590161cc Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 18:00:19 -0400 Subject: [PATCH 12/14] revised code Signed-off-by: Dhaval Patel --- src/couchdb/collections.json | 6 +++++- src/servers/iot/main.py | 5 ++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/src/couchdb/collections.json b/src/couchdb/collections.json index 8303bc85..3edd36d0 100644 --- a/src/couchdb/collections.json +++ b/src/couchdb/collections.json @@ -78,6 +78,10 @@ "id_prefix": "asset", "doctype": "asset", "indexes": [ + [ + "assetnum", + "siteid" + ] ] }, "vibration": { @@ -128,4 +132,4 @@ "id_prefix": "fc", "indexes": [] } -} +} \ No newline at end of file diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index bdf65bcc..1f06903a 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -2,8 +2,7 @@ import logging import statistics from datetime import datetime, timezone -from functools import lru_cache -from typing import Any, Dict, Iterator, List, Optional, Tuple, Union +from typing import Any, Dict, Iterator, List, Optional, Union, Tuple from mcp.server.fastmcp import FastMCP from pydantic import BaseModel import couchdb3 @@ -252,7 +251,7 @@ 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, site_name: Optional[str] = None) -> Optional[Dict[str, Any]]: From 4d66ecd6d06394d0569d6ee2ad949dd1bf194084 Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 18:19:39 -0400 Subject: [PATCH 13/14] revised test dataset Signed-off-by: Dhaval Patel --- src/servers/iot/tests/conftest.py | 41 ++- src/servers/iot/tests/test_tools.py | 452 ++++++++++++++++++++-------- 2 files changed, 363 insertions(+), 130 deletions(-) diff --git a/src/servers/iot/tests/conftest.py b/src/servers/iot/tests/conftest.py index b99bcd0c..a369381b 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 20b24b59..2cdabd90 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 From be9ac234aecb2fb1e550f03e5000f26837e60aac Mon Sep 17 00:00:00 2001 From: Dhaval Patel Date: Thu, 9 Jul 2026 18:47:21 -0400 Subject: [PATCH 14/14] revised code Signed-off-by: Dhaval Patel --- docs/mcp-servers.md | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/docs/mcp-servers.md b/docs/mcp-servers.md index c322f4b2..f7dd8cfa 100644 --- a/docs/mcp-servers.md +++ b/docs/mcp-servers.md @@ -20,9 +20,7 @@ two answer different questions: `asset_ids()`/`measured_sensors()` reflect TELEM 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. 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. -Beyond browsing: `stream_extent` and `sensor_coverage` report a stream's time bounds and +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