diff --git a/INSTRUCTIONS.md b/INSTRUCTIONS.md index 9955691f..76b853ce 100644 --- a/INSTRUCTIONS.md +++ b/INSTRUCTIONS.md @@ -127,7 +127,7 @@ Six FastMCP servers cover IoT data, time-series ML, work orders, vibration diagn | Server | Tools | Categories | Backing service | | ----------- | ----- | ------------------------ | -------------------------------------- | -| `iot` | 4 | read | CouchDB | +| `iot` | 7 | 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 726cb915..cf79790c 100644 --- a/docs/mcp-servers.md +++ b/docs/mcp-servers.md @@ -13,8 +13,17 @@ 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. + **Path:** `src/servers/iot/main.py` -**Requires:** CouchDB (`COUCHDB_URL`, `COUCHDB_USERNAME`, `COUCHDB_PASSWORD`, `IOT_DBNAME`) +**Requires:** CouchDB (`COUCHDB_URL`, `COUCHDB_USERNAME`, `COUCHDB_PASSWORD`, `IOT_DBNAME`, `ASSET_DBNAME`) **Sample assets shipped in the `iot` database** (loaded by `src/couchdb/couchdb_setup.sh`): @@ -26,12 +35,15 @@ Six FastMCP servers expose the AssetOpsBench domain logic. Each is a standalone Synthetic motor vibration data (`asset_id: Motor_01`, from `motor_01.json`) ships in a separate `vibration` database for the vibration MCP server. -| Tool | Category | Arguments | Description | -| --------- | -------- | ------------------------------------------ | ----------------------------------------------------------------------- | -| `sites` | read | — | List all available sites | -| `assets` | read | `site_name` | List all asset IDs for a site | -| `sensors` | read | `site_name`, `asset_id` | List sensor names for an asset | -| `history` | read | `site_name`, `asset_id`, `start`, `final?` | Fetch historical sensor readings for a time range (ISO 8601 timestamps) | +| 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 | ## utilities — Utilities diff --git a/src/couchdb/.allowed_datafiles b/src/couchdb/.allowed_datafiles index e4f4804d..0b0d865c 100644 --- a/src/couchdb/.allowed_datafiles +++ b/src/couchdb/.allowed_datafiles @@ -9,3 +9,4 @@ src/couchdb/scenarios_data/shared/iot/hydraulic_pump_1.json src/couchdb/scenarios_data/shared/iot/metro_pump_1.json src/couchdb/scenarios_data/shared/iot/motor_01.json src/couchdb/scenarios_data/shared/work_order/workorders.csv +src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json diff --git a/src/couchdb/collections.json b/src/couchdb/collections.json index bac2576b..e99f367d 100644 --- a/src/couchdb/collections.json +++ b/src/couchdb/collections.json @@ -37,6 +37,32 @@ ] ] }, + "asset": { + "format": "json", + "primary_key": [ + "assetnum" + ], + "id_prefix": "asset", + "doctype": "asset", + "indexes": [ + [ + "doctype", + "assetnum" + ], + [ + "doctype", + "assettype" + ], + [ + "doctype", + "siteid" + ], + [ + "doctype", + "iot_asset_id" + ] + ] + }, "vibration": { "format": "json", "primary_key": [ diff --git a/src/couchdb/scenarios_data/default/manifest.json b/src/couchdb/scenarios_data/default/manifest.json index 06760289..3f227ba8 100644 --- a/src/couchdb/scenarios_data/default/manifest.json +++ b/src/couchdb/scenarios_data/default/manifest.json @@ -5,6 +5,7 @@ "shared/iot/metro_pump_1.json", "shared/iot/hydraulic_pump_1.json" ], + "asset": "shared/iot/asset_profile_sample.json", "vibration": "shared/iot/motor_01.json", "failurecode": "shared/failure_code/failure_code_sample.csv" -} \ No newline at end of file +} diff --git a/src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json b/src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json new file mode 100644 index 00000000..24acd967 --- /dev/null +++ b/src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json @@ -0,0 +1,184 @@ +[ + { + "_id": "asset:CHILLER6", + "doctype": "asset", + "assetnum": "CHILLER6", + "iot_asset_id": "Chiller 6", + "wo_assetnum": "CHILLER6", + "description": "Chiller 6 (chilled-water plant)", + "siteid": "MAIN", + "orgid": "RIVPLANT", + "status": "OPERATING", + "assettype": "CHILLER", + "serialnum": null, + "manufacturer": null, + "vendor": null, + "priority": null, + "installdate": null, + "changedate": null, + "location": "MAIN-MECH-CH6", + "parent": null, + "vintage": null, + "sensors": [ + "Chiller 6 Condenser Water Return To Tower Temperature", + "Chiller 6 Chiller Efficiency", + "Chiller 6 Tonnage", + "Chiller 6 Supply Temperature", + "Chiller 6 Return Temperature", + "Chiller 6 Condenser Water Flow", + "Chiller 6 Schedule", + "Chiller 6 Power Input", + "Chiller 6 Chiller % Loaded", + "Chiller 6 Liquid Refrigerant Evaporator Temperature", + "Chiller 6 Setpoint Temperature", + "Chiller 6 Oil Pressure" + ] + }, + { + "_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", + "status": "OPERATING", + "assettype": "COMPRESSOR", + "serialnum": null, + "manufacturer": null, + "vendor": null, + "priority": null, + "installdate": "2023-10-05", + "changedate": "2008-12-09", + "location": null, + "parent": null, + "vintage": "mid", + "sensors": [ + "Compressor_Pressure_bar", + "Pneumatic_Panel_Pressure_bar", + "Cyclone_Filter_Drop_Pressure_bar", + "Tower_Discharge_Pressure_Drop_bar", + "Reservoir_Pressure_bar", + "Oil_Temperature_C", + "Motor_Current_A", + "Air_Intake_Valve_Status", + "Compressor_Outlet_Valve_Status", + "Active_Tower_ID", + "Load_Start_Signal", + "Low_Pressure_Switch_Status", + "Tower_Discharge_Switch_Status", + "Low_Oil_Level_Alarm", + "Airflow_Pulse_Counter" + ] + }, + { + "_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", + "status": "OPERATING", + "assettype": "HYDRAULIC_PUMP", + "serialnum": null, + "manufacturer": null, + "vendor": null, + "priority": null, + "installdate": null, + "changedate": null, + "location": null, + "parent": null, + "vintage": null, + "sensors": [ + "PS1_Pressure_bar_100Hz", + "PS2_Pressure_bar_100Hz", + "PS3_Pressure_bar_100Hz", + "PS4_Pressure_bar_100Hz", + "PS5_Pressure_bar_100Hz", + "PS6_Pressure_bar_100Hz", + "FS1_VolumeFlow_l_per_min_10Hz", + "FS2_VolumeFlow_l_per_min_10Hz", + "TS1_Temperature_C_1Hz", + "TS2_Temperature_C_1Hz", + "TS3_Temperature_C_1Hz", + "TS4_Temperature_C_1Hz", + "P1_MotorPower_W_100Hz", + "VS1_Vibration_mm_per_s_1Hz", + "CE_CoolingEfficiency_percent_1Hz", + "CP_CoolingPower_kW_1Hz", + "SE_EfficiencyFactor_percent_1Hz", + "cycle" + ] + }, + { + "_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", + "status": "OPERATING", + "assettype": "MOTOR", + "serialnum": null, + "manufacturer": null, + "vendor": null, + "priority": null, + "installdate": null, + "changedate": null, + "location": null, + "parent": null, + "vintage": null, + "sensors": [ + "Vibration_X" + ] + }, + { + "_id": "asset:PUMP3", + "doctype": "asset", + "assetnum": "PUMP3", + "iot_asset_id": null, + "wo_assetnum": "PUMP3", + "description": "Pump 3", + "siteid": "MAIN", + "orgid": "RIVPLANT", + "status": "OPERATING", + "assettype": "PUMP", + "serialnum": null, + "manufacturer": null, + "vendor": null, + "priority": null, + "installdate": null, + "changedate": null, + "location": "MAIN-PUMPHOUSE", + "parent": null, + "vintage": null, + "sensors": [] + }, + { + "_id": "asset:AHU2", + "doctype": "asset", + "assetnum": "AHU2", + "iot_asset_id": null, + "wo_assetnum": "AHU2", + "description": "Air handling unit 2", + "siteid": "NORTH", + "orgid": "RIVPLANT", + "status": "OPERATING", + "assettype": "AHU", + "serialnum": null, + "manufacturer": null, + "vendor": null, + "priority": null, + "installdate": null, + "changedate": null, + "location": "NORTH-ROOF", + "parent": null, + "vintage": null, + "sensors": [] + } +] diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index c6950379..eb3bc89e 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -37,13 +37,34 @@ logger.error(f"Failed to connect to CouchDB: {e}") db = None +# The asset registry is loaded as its own collection (manifest key "asset"), and the loader makes +# database name == collection key — so it lives in the "asset" database, NOT in IOT_DBNAME. Open a +# second handle for it. Telemetry (assets/sensors/history) keeps using `db` (the iot readings DB). +ASSET_DBNAME = os.environ.get("ASSET_DBNAME", "asset") +try: + asset_db = couchdb3.Database( + ASSET_DBNAME, + url=COUCHDB_URL, + user=COUCHDB_USERNAME, + password=COUCHDB_PASSWORD, + ) + logger.info(f"Connected to CouchDB: {ASSET_DBNAME}") +except Exception as e: + logger.error(f"Failed to connect to asset registry DB: {e}") + asset_db = None + mcp = FastMCP( "iot", - instructions="IoT sensor data: browse sites, assets, sensors, and query historical readings from CouchDB.", + instructions=( + "IoT sensor data + asset registry. Browse sites, assets, and sensors, read the asset " + "nameplate (registry), see which installed sensors are actually measured (streaming), and " + "query historical readings from CouchDB. NOTE: assets()/sensors() reflect TELEMETRY (what " + "streams = measured); get_asset()/asset_sensors()/registry_assets() reflect the REGISTRY " + "(what is installed, by name). Compare the two to find installed-but-not-streaming sensors." + ), ) -# Static site as per original requirement -SITES = ["MAIN"] +DEFAULT_SITES = ["MAIN"] class ErrorResult(BaseModel): @@ -79,6 +100,35 @@ class HistoryResult(BaseModel): message: str +# ── Asset-registry result models (identity / nameplate + installed sensor names) ── +class AssetDetail(BaseModel): + site_name: str + asset_id: str + description: Optional[str] + assettype: Optional[str] + status: Optional[str] + location: Optional[str] + installdate: Optional[str] + vintage: Optional[str] + n_sensors: int + message: str + + +class AssetSensorsResult(BaseModel): + site_name: str + asset_id: str + total_sensors: int + sensors: List[str] + message: str + + +class RegistryAssetsResult(BaseModel): + site_name: str + total_assets: int + assets: List[Dict[str, Any]] + message: str + + _asset_list_cache: Optional[List[str]] = None @@ -109,8 +159,13 @@ def get_asset_list() -> List[str]: def get_sensor_list(asset_id: str) -> List[str]: - """Helper to fetch sensor names for a given asset from CouchDB. - Result is cached per asset_id after the first successful call.""" + """The sensors an asset actually measures = the UNION of measurement keys across ALL of the + asset's reading documents. + + IoT data may be sparse / non-uniform: different sensors are recorded at different timestamps + (a timestamp may carry one sensor, several, or all), so a single document does NOT reveal the + full measured set. We therefore scan every reading doc for the asset and union the non-metadata + keys. Result is cached per asset_id after the first successful call.""" if asset_id in _sensor_list_cache: return _sensor_list_cache[asset_id] @@ -118,15 +173,16 @@ def get_sensor_list(asset_id: str) -> List[str]: return [] try: - # Get one document for the asset to inspect keys - res = db.find({"asset_id": asset_id}, limit=1) - if not res["docs"]: + res = db.find({"asset_id": asset_id}, limit=100000) + docs = res["docs"] + if not docs: return [] - doc = res["docs"][0] - # Exclude metadata and standard fields + # Exclude metadata; union the measurement keys across every reading document. exclude = {"_id", "_rev", "asset_id", "timestamp"} - sensors = sorted(key for key in doc.keys() if key not in exclude) + sensors = sorted( + {key for doc in docs for key in doc.keys() if key not in exclude} + ) _sensor_list_cache[asset_id] = sensors return sensors except Exception as e: @@ -134,31 +190,105 @@ def get_sensor_list(asset_id: str) -> List[str]: return [] +_asset_doc_cache: Dict[str, Dict[str, Any]] = {} + + +def get_asset_doc(asset_id: str) -> Optional[Dict[str, Any]]: + """Fetch one asset-registry document, resolving ANY of the asset's id spaces — the registry + `assetnum` (Maximo id), the telemetry `iot_asset_id`, or the work-order `wo_assetnum`. This lets + the same profile be found whether the caller holds the IoT id (e.g. 'Chiller 6') or the WO id + ('CHILLER6'). Cached per asset_id.""" + if asset_id in _asset_doc_cache: + return _asset_doc_cache[asset_id] + if not asset_db: + return None + try: + for field in ("assetnum", "iot_asset_id", "wo_assetnum"): + res = asset_db.find({"doctype": "asset", field: asset_id}, limit=1) + docs = res["docs"] + if docs: + _asset_doc_cache[asset_id] = docs[0] + return docs[0] + return None + except Exception as e: + logger.error(f"Error fetching asset doc {asset_id}: {e}") + return None + + +_registry_sites_cache: Optional[List[str]] = None + + +def get_registry_sites() -> List[str]: + """Distinct site ids present in the asset registry (from each asset profile's `siteid`). Cached.""" + global _registry_sites_cache + if _registry_sites_cache is not None: + return _registry_sites_cache + if not asset_db: + return [] + try: + res = asset_db.find({"doctype": "asset"}, fields=["siteid"], limit=100000) + found = sorted({d.get("siteid") for d in res["docs"] if d.get("siteid")}) + _registry_sites_cache = found + return found + except Exception as e: + logger.error(f"get_registry_sites failed: {e}") + return [] + + +def known_sites() -> List[str]: + """The server's site list — discovered DYNAMICALLY from the asset registry (each asset profile's + `siteid`). Falls back to DEFAULT_SITES only if the registry is empty / unavailable. + """ + return get_registry_sites() or DEFAULT_SITES + + +def _is_known_site(site_name: str) -> bool: + return site_name in known_sites() + + @mcp.tool(title="List Sites") def sites() -> SitesResult: - """Retrieves a list of sites. Each site is represented by a name.""" - return SitesResult(sites=SITES) + """Retrieves the list of sites, discovered dynamically from the asset registry (the distinct + `siteid` across asset profiles). Falls back to the default only if the registry has no assets. + """ + return SitesResult(sites=known_sites()) @mcp.tool(title="List Assets") def assets(site_name: str) -> Union[AssetsResult, ErrorResult]: - """Returns a list of assets for a given site. Each asset includes an id and a name.""" - if site_name not in SITES: + """Returns the assets registered at a given site, from the asset registry filtered by `siteid`. + Each returned id is the asset's telemetry id (`iot_asset_id`) where it has one, otherwise its + registry `assetnum` — so the id works with sensors()/history() when telemetry exists. + """ + if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") - - asset_list = get_asset_list() - return AssetsResult( - site_name=site_name, - total_assets=len(asset_list), - assets=asset_list, - message=f"found {len(asset_list)} asset ids for site_name {site_name}: {', '.join(asset_list)}.", - ) + if not asset_db: + return ErrorResult(error="CouchDB not connected") + try: + res = asset_db.find( + {"doctype": "asset", "siteid": site_name}, + fields=["assetnum", "iot_asset_id"], + limit=100000, + ) + ids = sorted((d.get("iot_asset_id") or d["assetnum"]) for d in res["docs"]) + return AssetsResult( + site_name=site_name, + total_assets=len(ids), + assets=ids, + message=f"found {len(ids)} assets at site {site_name}: {', '.join(ids)}.", + ) + except Exception as e: + logger.error(f"assets failed: {e}") + return ErrorResult(error=str(e)) @mcp.tool(title="List Sensors") def sensors(site_name: str, asset_id: str) -> Union[SensorsResult, ErrorResult]: - """Lists the sensors available for a specified asset at a given site.""" - if site_name not in SITES: + """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().""" + if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") sensor_list = get_sensor_list(asset_id) @@ -218,6 +348,111 @@ def history( return ErrorResult(error=str(e)) +@mcp.tool(title="Get Asset") +def get_asset(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResult]: + """Return registry/nameplate details for one asset (Maximo MXASSET-aligned: description, + assettype, status, location, installdate, vintage) plus installed sensor count. + This is asset IDENTITY — distinct from the telemetry-derived assets() list.""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + doc = get_asset_doc(asset_id) + if not doc: + return ErrorResult(error=f"unknown asset_id {asset_id} in registry") + n = len(doc.get("sensors", [])) + assettype = doc.get("assettype") + vintage = doc.get("vintage") + location = doc.get("location") + + # Build a human message that omits clauses whose fields are null. + parts = [f"asset {asset_id} is a {assettype or 'asset'}"] + if vintage: + parts.append(f" ({vintage} vintage)") + if location: + parts.append(f" at {location}") + parts.append(f" with {n} installed sensors.") + message = "".join(parts) + + return AssetDetail( + site_name=site_name, + asset_id=doc.get("assetnum", asset_id), + description=doc.get("description"), + assettype=assettype, + status=doc.get("status"), + location=location, + installdate=doc.get("installdate"), + vintage=vintage, + n_sensors=n, + message=message, + ) + + +@mcp.tool(title="List Asset Sensors") +def asset_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.""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + doc = get_asset_doc(asset_id) + if not doc: + return ErrorResult(error=f"unknown asset_id {asset_id} in registry") + names = list(doc.get("sensors", [])) + return AssetSensorsResult( + site_name=site_name, + asset_id=asset_id, + total_sensors=len(names), + sensors=names, + message=f"{len(names)} sensors installed on {asset_id}: {', '.join(names)}.", + ) + + +@mcp.tool(title="List Registry Assets") +def registry_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.""" + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + if not asset_db: + return ErrorResult(error="CouchDB not connected") + try: + selector: Dict[str, Any] = {"doctype": "asset"} + if assettype: + selector["assettype"] = assettype + res = asset_db.find( + selector, + fields=["assetnum", "assettype", "vintage", "sensors"], + limit=100000, + ) + rows = sorted( + ( + { + "asset_id": d["assetnum"], + "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"], + ) + return RegistryAssetsResult( + site_name=site_name, + total_assets=len(rows), + assets=rows, + message=f"found {len(rows)} registry assets" + + (f" of type '{assettype}'" if assettype else "") + + ".", + ) + except Exception as e: + logger.error(f"registry_assets failed: {e}") + return ErrorResult(error=str(e)) + + def main(): # Initialize and run the server mcp.run(transport="stdio")