diff --git a/docs/mcp-servers.md b/docs/mcp-servers.md index 5fe34c94..a8ee9652 100644 --- a/docs/mcp-servers.md +++ b/docs/mcp-servers.md @@ -13,43 +13,36 @@ Six FastMCP servers expose the AssetOpsBench domain logic. Each is a standalone ## iot — IoT Asset Registry and Telemetry Records -The IoT server reads from the asset **registry** (`ASSET_DBNAME`, default `asset`, loaded from -`asset_profile_sample.json`) and IoT telemetry records (`IOT_DBNAME`, default `iot`). It exposes -registry discovery tools while the broader IoT tool surface is being rebuilt: `sites()` for site -names, `asset_ids()` for bare `assetnum` values, `asset_detail()` for one asset's registry -details, `assets()` for registry metadata with optional `assettype` filtering, -`find_assets_by_sensors()` to find assets by installed or measured sensors, `installed_sensors()` -for registry sensor inventory, and `measured_sensors()` for telemetry fields observed in records. +Read-only tools for browsing the asset registry and querying IoT telemetry. **Path:** `src/servers/iot/main.py` **Requires:** CouchDB (`COUCHDB_URL`, `COUCHDB_USERNAME`, `COUCHDB_PASSWORD`, `ASSET_DBNAME`, `IOT_DBNAME`) - -**Sample asset profiles shipped in the `asset` database** (loaded by `src/couchdb/init_data.py`): - -| `assetnum` | Asset class | -| ----------- | ---------------- | -| `Chiller 6` | Chiller | -| `mp_1` | Compressor | -| `hyd_1` | Hydraulic pump | - -Source file: `src/couchdb/scenarios_data/shared/iot/asset_profile_sample.json`. - -| Tool | Arguments | Description | -| ----------------- | ------------------------------------------ | ------------------------------------------------------------------------------------------------------------ | -| `sites` | - | List known site names from the asset registry, with a default fallback | -| `asset_ids` | `site_name` | List bare `assetnum` values registered at a site | -| `asset_detail` | `site_name`, `asset_id` | Return one asset's registry details, including `n_installed_sensors` | -| `find_assets_by_sensors` | `site_name`, `sensors`, `match?`, `substring?`, `source?` | Find assets at a site by installed or measured sensor names | -| `installed_sensors` | `site_name`, `asset_id` | List sensor names installed on an asset according to the asset registry | -| `measured_sensors` | `site_name`, `asset_id` | List measured telemetry fields observed for an asset | -| `assets` | `site_name`, `assettype?` | List assets with metadata (assettype, description, vintage, installed sensor count), optionally filtered by assettype | - -`find_assets_by_sensors()` searches only assets registered at the requested site. `match="all"` -requires every query sensor to match, while `match="any"` requires at least one. Set -`substring=true` to perform case-insensitive substring matching instead of exact sensor-name -matching. `source="installed"` searches the asset registry sensor inventory; `source="measured"` -searches telemetry fields observed in IoT records. Duplicate query sensors are deduplicated in -the response while preserving their first occurrence order. +**Sample assets:** `Chiller 6`, `mp_1`, and `hyd_1` from `asset_profile_sample.json` + +### Registry and discovery tools + +| Tool | Arguments | Description | +| ---- | --------- | ----------- | +| `sites` | - | List sorted site identifiers, with `MAIN` as the fallback | +| `asset_ids` | `site_name` | List asset identifiers registered at a site | +| `assets` | `site_name`, `assettype?` | List assets with compact metadata and optional exact type filtering | +| `asset_detail` | `site_name`, `asset_id` | Return registry details and installed-sensor count for one asset | +| `installed_sensors` | `site_name`, `asset_id` | List sensor names assigned in the registry | +| `measured_sensors` | `site_name`, `asset_id` | List measurement fields observed across the telemetry stream | +| `find_assets_by_sensors` | `site_name`, `sensors`, `match?`, `substring?`, `source?` | Find site assets by installed or measured sensor names | + +### Telemetry tools + +| Tool | Arguments | Description | +| ---- | --------- | ----------- | +| `stream_extent` | `site_name`, `asset_id`, `sensor?`, `start?`, `end?` | Count matching records and return their earliest and latest timestamps | +| `latest_reading` | `site_name`, `asset_id`, `sensor?` | Return the newest record, or the newest non-null value for one sensor | +| `history` | `site_name`, `asset_id`, `start?`, `end?`, `sensors?`, `limit?`, `cursor?` | Return chronological, projected observations with cursor paging | +| `sensor_coverage` | `site_name`, `asset_id` | Scan the full stream for per-sensor non-null counts and time coverage | +| `sensor_stats` | `site_name`, `asset_id`, `sensor?`, `start?`, `end?` | Compute per-sensor numeric counts, range, mean, and population standard deviation | + +Telemetry windows are half-open ISO 8601 ranges. `history` supports cursor-based paging with up to +1000 observations per page. ## utilities — Utilities diff --git a/src/servers/iot/main.py b/src/servers/iot/main.py index c734b613..9e62f578 100644 --- a/src/servers/iot/main.py +++ b/src/servers/iot/main.py @@ -1,11 +1,43 @@ import logging import os -from typing import Any, Dict, Iterator, List, Optional, Union +from datetime import datetime +from typing import Any, Dict, List, Optional, Union import couchdb3 from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP -from pydantic import BaseModel + +from servers.iot.models import ( + AssetDetail, + AssetSensorMatch, + AssetSummary, + AssetsResult, + AssetsWithMetadataResult, + ErrorResult, + FindAssetsResult, + HistoryResult, + LatestReadingResult, + SensorCoverageResult, + SensorStatsResult, + SensorsResult, + SitesResult, + StreamExtentResult, +) +from servers.iot.telemetry_helper import ( + _SensorAccumulator, + _SensorCoverageAccumulator, + _TimestampHandlingError, + _coerce_finite_number, + _decode_history_cursor, + _encode_history_cursor, + _history_cursor_context, + _history_observation, + _iter_records, + _iter_records_in_window, + _parse_iso_timestamp, + _timestamp_age_seconds, + _validate_dates, +) load_dotenv() @@ -49,12 +81,9 @@ mcp = FastMCP( "iot", instructions=( - "IoT asset registry and telemetry record tools. Use sites() to discover site names, " - "asset_ids() for bare assetnum values at a site, asset_detail() for one asset's " - "registry details, assets() for registry metadata with optional assettype filtering, " - "find_assets_by_sensors() to find assets by installed or measured sensors, " - "installed_sensors() for registry sensor inventory, and measured_sensors() for " - "observed telemetry fields." + "Read-only IoT sensor data and asset registry tools. Browse sites, assets, and " + "sensors; inspect asset details; compare installed and measured sensors; and query " + "telemetry." ), ) @@ -63,105 +92,10 @@ RESERVED_FIELDS = {"_id", "_rev", "asset_id", "timestamp", "dataset", "type", "doctype"} -class ErrorResult(BaseModel): - error: str - - -class SitesResult(BaseModel): - sites: List[str] - - -class AssetsResult(BaseModel): - site_name: str - total_assets: int - assets: List[str] - message: str - - -class SensorsResult(BaseModel): - site_name: str - asset_id: str - total_sensors: int - sensors: List[str] - message: str - - -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_installed_sensors: int - message: str - - -class AssetSummary(BaseModel): - asset_id: str - description: Optional[str] - assettype: Optional[str] - vintage: Optional[str] - n_sensors: int - - -class AssetsWithMetadataResult(BaseModel): - site_name: str - total_assets: int - assets: List[AssetSummary] - message: str - - -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 - - _registry_sites_cache: Optional[List[str]] = None _sensor_list_cache: Dict[str, List[str]] = {} -def _iter_records( - selector: Dict[str, Any], - fields: Optional[List[str]] = None, - sort: Optional[List[Dict[str, str]]] = None, - page_size: int = PAGE_SIZE, -) -> Iterator[Dict[str, Any]]: - """Yield telemetry records matching a selector across paged database results.""" - if not iot_db: - return - if sort is None: - sort = [{"asset_id": "asc"}, {"timestamp": "asc"}] - bookmark: Optional[str] = None - 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 = iot_db.find(selector, **kwargs) - docs = res.get("docs", []) - if not docs: - break - for doc in docs: - yield doc - bookmark = res.get("bookmark") - if bookmark is None or len(docs) < page_size: - break - - def get_sensor_list(asset_id: str) -> List[str]: """Return sorted telemetry field names observed across all records for an asset.""" if asset_id in _sensor_list_cache: @@ -171,7 +105,7 @@ def get_sensor_list(asset_id: str) -> List[str]: try: found = set() seen = False - for doc in _iter_records({"asset_id": asset_id}): + for doc in _iter_records(iot_db, {"asset_id": asset_id}): seen = True found.update(key for key in doc.keys() if key not in RESERVED_FIELDS) if not seen: @@ -249,28 +183,27 @@ def _installed_sensors(asset_id: str, site_name: Optional[str] = None) -> List[s @mcp.tool(title="List Sites") def sites() -> SitesResult: - """List known site names from the asset registry. + """List sorted site identifiers available in the asset registry. Returns: - SitesResult: `sites` contains sorted distinct `siteid` values from - asset profiles. If the registry database is unavailable or has no sites, - `sites` falls back to `["MAIN"]`. + SitesResult: Distinct site identifiers, or `["MAIN"]` when none are + available. """ return SitesResult(sites=known_sites()) @mcp.tool(title="List Asset IDs") def asset_ids(site_name: str) -> Union[AssetsResult, ErrorResult]: - """List only asset identifiers for one site. + """List asset identifiers registered at one site. + + Use `assets()` when registry metadata is also needed. Args: site_name: Exact site id to query, such as `MAIN`. Use `sites()` to discover valid site ids. Returns: - AssetsResult: Contains `site_name`, `total_assets`, `assets`, and - `message`. The `assets` field is a sorted list of asset registry - `assetnum` values. + AssetsResult: Sorted asset identifiers and their count. """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -298,9 +231,8 @@ def asset_ids(site_name: str) -> Union[AssetsResult, ErrorResult]: def asset_detail(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResult]: """Return registry details for one asset. - This tool reads the asset registry from `asset_db` (`ASSET_DBNAME`, default - `asset`) and returns nameplate fields for one asset record. Use - `installed_sensors()` when you only need the registry sensor inventory. + Includes identity, description, type, status, location, installation date, + vintage, and installed-sensor count. Use `installed_sensors()` for names. Args: site_name: Exact site id to query, such as `MAIN`. Use `sites()` to @@ -308,9 +240,8 @@ def asset_detail(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResul asset_id: Exact asset id from `asset_ids()`, such as `Chiller 6`. Returns: - AssetDetail: Contains `site_name`, `asset_id`, `description`, - `assettype`, `status`, `location`, `installdate`, `vintage`, - `n_installed_sensors`, and `message`. + AssetDetail: The available registry fields; optional values are null + when absent. """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -371,12 +302,10 @@ def asset_detail(site_name: str, asset_id: str) -> Union[AssetDetail, ErrorResul def measured_sensors( site_name: str, asset_id: str ) -> Union[SensorsResult, ErrorResult]: - """List telemetry fields actually measured for one asset. + """List measurement fields observed in an asset's telemetry stream. - This tool reads IoT records from `iot_db` (`IOT_DBNAME`, default `iot`). - It scans records matching `asset_id` and returns the union of observed - measurement fields. Use `installed_sensors()` instead when you need the - asset registry inventory. + Returns the sorted union across the full stream, excluding record metadata. + Use `installed_sensors()` for assigned sensors, which may differ. Args: site_name: Exact site id to query, such as `MAIN`. Use `sites()` to @@ -384,10 +313,7 @@ def measured_sensors( asset_id: Exact asset id from `asset_ids()`, such as `Chiller 6`. Returns: - SensorsResult: Contains `site_name`, `asset_id`, `total_sensors`, - `sensors`, and `message`. The `sensors` field is a sorted list of - observed telemetry fields, excluding record metadata such as `_id`, - `_rev`, `asset_id`, `timestamp`, `dataset`, `type`, and `doctype`. + SensorsResult: Observed measurement field names and their count. """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -414,12 +340,10 @@ def measured_sensors( def installed_sensors( site_name: str, asset_id: str ) -> Union[SensorsResult, ErrorResult]: - """List sensor names installed on one asset according to the registry. + """List sensor names assigned to an asset in the registry. - This tool reads the asset registry from `asset_db` (`ASSET_DBNAME`, default - `asset`) and returns the asset record's `sensors` inventory. Use - `measured_sensors()` instead when you need fields actually observed in IoT - telemetry records. + Use `measured_sensors()` for fields observed in telemetry; the two lists may + differ. Args: site_name: Exact site id to query, such as `MAIN`. Use `sites()` to @@ -427,9 +351,7 @@ def installed_sensors( asset_id: Exact asset id from `asset_ids()`, such as `Chiller 6`. Returns: - SensorsResult: Contains `site_name`, `asset_id`, `total_sensors`, - `sensors`, and `message`. The `sensors` field preserves the registry - sensor inventory order from the asset record. + SensorsResult: Sensor names in registry order and their count. """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -462,19 +384,20 @@ def installed_sensors( def assets( site_name: str, assettype: Optional[str] = None ) -> Union[AssetsWithMetadataResult, ErrorResult]: - """List asset registry records for one site with compact metadata. + """List assets at one site with compact registry metadata. + + Use `asset_ids()` for identifiers only and `asset_detail()` for more fields + about one asset. Args: site_name: Exact site id to query, such as `MAIN`. Use `sites()` to discover valid site ids. - assettype: Optional exact asset type filter, such as `PUMP` or - `COMPRESSOR`. + assettype: Optional exact, case-sensitive asset type filter, such as + `PUMP` or `COMPRESSOR`. Returns: - AssetsWithMetadataResult: Contains `site_name`, `total_assets`, - `assets`, and `message`. Each item in `assets` includes `asset_id` - (the registry `assetnum`), `description`, `assettype`, `vintage`, and - `n_sensors`. + AssetsWithMetadataResult: Assets sorted by identifier, with description, + type, vintage, and installed-sensor count. """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -523,24 +446,24 @@ def find_assets_by_sensors( substring: bool = False, source: str = "measured", ) -> Union[FindAssetsResult, ErrorResult]: - """Return assets at a site that match the requested sensor names. + """Find site assets by installed or measured sensor names. - The search is limited to assets registered at `site_name`. Duplicate query - sensors are ignored while preserving the first occurrence order. + Searches only assets registered at `site_name`. Duplicate queries are + removed while preserving first occurrence order. Args: site_name: Exact site id to query, such as `MAIN`. Use `sites()` to discover valid site ids. - sensors: One or more sensor names or, when `substring` is true, sensor - name fragments to search for. - match: `all` requires every listed sensor; `any` requires at least one. - substring: If true, match sensor names case-insensitively by substring. - source: `measured` checks telemetry fields; `installed` checks the - asset registry inventory. + sensors: Sensor names, or fragments when `substring` is true. + match: `all` requires every query; `any` requires at least one. + substring: Use case-insensitive fragment matching instead of exact, + case-sensitive matching. + source: `measured` searches observed fields; `installed` searches + registry assignments. Defaults to `measured`. Returns: - FindAssetsResult: Contains the deduplicated query sensors, matching - asset ids, and the concrete sensor names that matched each asset. + FindAssetsResult: Query settings and matching assets with the concrete + sensor names found. """ if not _is_known_site(site_name): return ErrorResult(error=f"unknown site {site_name}") @@ -611,6 +534,524 @@ def _hits(sensor_name: 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, + end: Optional[str] = None, +) -> Union[StreamExtentResult, ErrorResult]: + """Inspect the size and timestamp span of an asset's telemetry stream. + + Optionally filter by one measured sensor and a half-open ISO 8601 window: + `start` is inclusive and `end` is exclusive. Sensor-filtered records count + only when the field is present and non-null. Bounds and record timestamps + must consistently include or omit timezone offsets; aware timestamps are + compared by instant even when their offsets differ. + + Args: + site_name: Exact site id to query, such as `MAIN`. Use `sites()` to + discover valid site ids. + asset_id: Exact asset id from `asset_ids()`, such as `Chiller 6`. + sensor: Optional exact field name from `measured_sensors()`. + Reserved metadata fields such as `asset_id` and `timestamp` are not + valid sensor names. + start: Optional inclusive ISO 8601 date or datetime lower bound. + end: Optional exclusive ISO 8601 date or datetime upper bound. + + Returns: + StreamExtentResult: Earliest and latest matching timestamps, record + count, whether the count exceeds one 1000-record page, and average + interval implied by the span and count. + """ + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + validation_error = _validate_dates(start, end) + if validation_error: + return ErrorResult(error=validation_error) + if not iot_db: + return ErrorResult(error="IoT records database not connected") + if sensor is not None and not sensor.strip(): + return ErrorResult(error="sensor must not be empty") + if sensor in RESERVED_FIELDS: + return ErrorResult( + error=f"sensor must be a telemetry field, not reserved metadata field {sensor}" + ) + + start_dt = _parse_iso_timestamp(start) if start is not None else None + end_dt = _parse_iso_timestamp(end) if end is not None else None + selector: Dict[str, Any] = { + "asset_id": asset_id, + "timestamp": {"$exists": True, "$ne": None}, + } + if sensor: + selector[sensor] = {"$exists": True, "$ne": None} + + first_timestamp: Optional[str] = None + last_timestamp: Optional[str] = None + first_datetime: Optional[datetime] = None + last_datetime: Optional[datetime] = None + total_records = 0 + try: + for _, timestamp, timestamp_dt in _iter_records_in_window( + iot_db, selector, start_dt, end_dt, fields=["timestamp"] + ): + if first_datetime is None or timestamp_dt < first_datetime: + first_datetime = timestamp_dt + first_timestamp = timestamp + if last_datetime is None or timestamp_dt > last_datetime: + last_datetime = timestamp_dt + last_timestamp = timestamp + total_records += 1 + + if total_records == 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_datetime is not None + and last_datetime is not None + and total_records > 1 + ): + approx_interval = (last_datetime - first_datetime).total_seconds() / ( + total_records - 1 + ) + + return StreamExtentResult( + site_name=site_name, + asset_id=asset_id, + sensor=sensor, + start_time=first_timestamp, + end_time=last_timestamp, + total_records=total_records, + exceeds_page_limit=total_records > PAGE_SIZE, + approx_interval_seconds=approx_interval, + message=f"{total_records} record(s) for asset_id {asset_id}" + + (f" (sensor {sensor})" if sensor else "") + + f" from {first_timestamp} to {last_timestamp}.", + ) + except _TimestampHandlingError as e: + return ErrorResult(error=str(e)) + except Exception as e: + logger.error(f"stream_extent failed: {e}") + return ErrorResult(error="unable to inspect telemetry stream extent") + + +@mcp.tool(title="Get Sensor History") +def history( + 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]: + """Return one chronological page of telemetry observations for an asset. + + Optional bounds form a half-open ISO 8601 window. Use `sensors` to project + exact measured fields; omit it to return all measured fields present in each + row. Every row includes `asset_id` and `timestamp` and excludes other record + metadata. + + Leave `cursor` unset for the first page. When `has_more` is true, repeat the + same query with `next_cursor`. Cursors are opaque and query-bound. Timestamp + offset rules match `stream_extent()`. + + Args: + site_name: Exact site id to query, such as `MAIN`. Use `sites()` to + discover valid site ids. + asset_id: Exact asset id from `asset_ids()`, such as `Chiller 6`. + start: Optional inclusive ISO 8601 lower bound. + end: Optional exclusive ISO 8601 upper bound. + sensors: Optional non-empty list of exact measured field names. Duplicate + names are removed while preserving order. + limit: Maximum observations in this page, from 1 to 1000. + cursor: Opaque `next_cursor` from the previous page of the same query. + + Returns: + HistoryResult: Projected observations, returned count, echoed bounds, + and cursor fields for the next page. + """ + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + validation_error = _validate_dates(start, end) + if validation_error: + return ErrorResult(error=validation_error) + if not iot_db: + return ErrorResult(error="IoT records database not connected") + if isinstance(limit, bool) or not 1 <= limit <= PAGE_SIZE: + return ErrorResult(error=f"limit must be between 1 and {PAGE_SIZE}") + + selected_sensors: Optional[List[str]] = None + if sensors is not None: + if not sensors: + return ErrorResult(error="sensors must not be empty when provided") + if any(not sensor.strip() for sensor in sensors): + return ErrorResult(error="sensor names must not be empty") + selected_sensors = list(dict.fromkeys(sensors)) + reserved = [sensor for sensor in selected_sensors if sensor in RESERVED_FIELDS] + if reserved: + return ErrorResult( + error=( + "sensors must be telemetry fields, not reserved metadata " + f"fields {reserved}" + ) + ) + available_sensors = get_sensor_list(asset_id) + if not available_sensors: + return ErrorResult(error=f"unknown asset_id {asset_id} or no sensors found") + unknown = [ + sensor for sensor in selected_sensors if sensor not in available_sensors + ] + if unknown: + return ErrorResult( + error=f"unknown sensors {unknown} for asset_id {asset_id}" + ) + + cursor_context = _history_cursor_context( + site_name, asset_id, start, end, selected_sensors + ) + offset = 0 + if cursor is not None: + offset, cursor_error = _decode_history_cursor(cursor, cursor_context) + if cursor_error: + return ErrorResult(error=cursor_error) + assert offset is not None + + start_dt = _parse_iso_timestamp(start) if start is not None else None + end_dt = _parse_iso_timestamp(end) if end is not None else None + selector: Dict[str, Any] = { + "asset_id": asset_id, + "timestamp": {"$exists": True, "$ne": None}, + } + fields = ["timestamp", *selected_sensors] if selected_sensors else None + observations: List[Dict[str, Any]] = [] + matched_records = 0 + has_more = False + previous_datetime: Optional[datetime] = None + try: + for doc, timestamp, timestamp_dt in _iter_records_in_window( + iot_db, selector, start_dt, end_dt, fields=fields + ): + if previous_datetime is not None and timestamp_dt < previous_datetime: + raise _TimestampHandlingError( + "telemetry timestamps cannot be returned in chronological order" + ) + previous_datetime = timestamp_dt + if matched_records < offset: + matched_records += 1 + continue + if len(observations) >= limit: + has_more = True + break + observations.append( + _history_observation( + doc, + asset_id, + timestamp, + selected_sensors, + RESERVED_FIELDS, + ) + ) + matched_records += 1 + except _TimestampHandlingError as e: + return ErrorResult(error=str(e)) + except Exception as e: + logger.error(f"history failed: {e}") + return ErrorResult(error="unable to retrieve telemetry history") + + next_cursor = None + if has_more: + next_cursor = _encode_history_cursor( + offset + len(observations), cursor_context + ) + return HistoryResult( + site_name=site_name, + asset_id=asset_id, + observations=observations, + returned=len(observations), + next_cursor=next_cursor, + has_more=has_more, + start=start, + end=end, + message=( + f"returned {len(observations)} observation(s) for asset_id {asset_id}; " + f"has_more={has_more}." + ), + ) + + +@mcp.tool(title="Latest Reading") +def latest_reading( + site_name: str, + asset_id: str, + sensor: Optional[str] = None, +) -> Union[LatestReadingResult, ErrorResult]: + """Return the newest telemetry observation for an asset. + + Omit `sensor` for all measured fields in the newest record. When a sensor is + provided, returns the newest record where that field is present and non-null. + Records are compared by parsed timestamp using the `stream_extent()` rules. + + `age_seconds` is the difference between the current UTC time and the reading + timestamp. Offset-free timestamps are interpreted as UTC; a negative value + indicates a timestamp in the future. + + Args: + site_name: Exact site id to query, such as `MAIN`. Use `sites()` to + discover valid site ids. + asset_id: Exact asset id from `asset_ids()`, such as `Chiller 6`. + sensor: Optional exact field name from `measured_sensors()`. Reserved + metadata fields are not valid sensor names. + + Returns: + LatestReadingResult: Selected timestamp, values, and age in seconds. + """ + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + if not iot_db: + return ErrorResult(error="IoT records database not connected") + if sensor is not None and not sensor.strip(): + return ErrorResult(error="sensor must not be empty") + if sensor in RESERVED_FIELDS: + return ErrorResult( + error=( + "sensor must be a telemetry field, not reserved metadata " + f"field {sensor}" + ) + ) + if sensor is not None: + available_sensors = get_sensor_list(asset_id) + if not available_sensors: + return ErrorResult(error=f"unknown asset_id {asset_id} or no sensors found") + if sensor not in available_sensors: + return ErrorResult(error=f"unknown sensor {sensor} for asset_id {asset_id}") + + selector: Dict[str, Any] = { + "asset_id": asset_id, + "timestamp": {"$exists": True, "$ne": None}, + } + fields = None + if sensor is not None: + selector[sensor] = {"$exists": True, "$ne": None} + fields = ["timestamp", sensor] + + latest_doc: Optional[Dict[str, Any]] = None + latest_timestamp: Optional[str] = None + latest_datetime: Optional[datetime] = None + try: + for doc, timestamp, timestamp_dt in _iter_records_in_window( + iot_db, selector, None, None, fields=fields + ): + if latest_datetime is None or timestamp_dt > latest_datetime: + latest_doc = doc + latest_timestamp = timestamp + latest_datetime = timestamp_dt + except _TimestampHandlingError as e: + return ErrorResult(error=str(e)) + except Exception as e: + logger.error(f"latest_reading failed: {e}") + return ErrorResult(error="unable to retrieve latest telemetry reading") + + if latest_doc is None or latest_timestamp is None or latest_datetime is None: + return ErrorResult( + error=f"no records for asset_id {asset_id}" + + (f", sensor {sensor}" if sensor else "") + ) + + if sensor is None: + values = { + field: value + for field, value in latest_doc.items() + if field not in RESERVED_FIELDS + } + else: + values = {sensor: latest_doc[sensor]} + return LatestReadingResult( + site_name=site_name, + asset_id=asset_id, + timestamp=latest_timestamp, + values=values, + age_seconds=_timestamp_age_seconds(latest_datetime), + message=f"latest reading for asset_id {asset_id} at {latest_timestamp}.", + ) + + +@mcp.tool(title="Sensor Coverage") +def sensor_coverage( + site_name: str, + asset_id: str, +) -> Union[SensorCoverageResult, ErrorResult]: + """Summarize non-null observation coverage for every measured sensor. + + Scans the full timestamped stream. `non_null_count` includes present, + non-null values of any type. Fields observed only with null values have zero + count and null bounds. Timestamp offset rules match `stream_extent()`. + + Args: + site_name: Exact site id to query, such as `MAIN`. Use `sites()` to + discover valid site ids. + asset_id: Exact asset id from `asset_ids()`, such as `Chiller 6`. + + Returns: + SensorCoverageResult: Timestamped records scanned and per-sensor counts + and chronological bounds, sorted by field name. + """ + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + if not iot_db: + return ErrorResult(error="IoT records database not connected") + + selector: Dict[str, Any] = { + "asset_id": asset_id, + "timestamp": {"$exists": True, "$ne": None}, + } + coverage: Dict[str, _SensorCoverageAccumulator] = {} + docs_scanned = 0 + try: + for doc, timestamp, timestamp_dt in _iter_records_in_window( + iot_db, selector, None, None + ): + docs_scanned += 1 + for field, value in doc.items(): + if field in RESERVED_FIELDS: + continue + accumulator = coverage.setdefault( + field, _SensorCoverageAccumulator() + ) + if value is not None: + accumulator.add_non_null(timestamp, timestamp_dt) + except _TimestampHandlingError as e: + return ErrorResult(error=str(e)) + except Exception as e: + logger.error(f"sensor_coverage failed: {e}") + return ErrorResult(error="unable to calculate sensor coverage") + + if docs_scanned == 0: + return ErrorResult(error=f"unknown asset_id {asset_id} or no records found") + + sensors = [coverage[field].result(field) for field in sorted(coverage)] + message = ( + f"coverage for {len(sensors)} sensor(s) on asset_id {asset_id} " + f"across {docs_scanned} timestamped record(s)." + ) + return SensorCoverageResult( + site_name=site_name, + asset_id=asset_id, + docs_scanned=docs_scanned, + sensors=sensors, + message=message, + ) + + +@mcp.tool(title="Sensor Statistics") +def sensor_stats( + site_name: str, + asset_id: str, + sensor: Optional[str] = None, + start: Optional[str] = None, + end: Optional[str] = None, +) -> Union[SensorStatsResult, ErrorResult]: + """Compute numeric statistics for one or all measured sensors. + + Omit `sensor` to summarize every measured field. Optional bounds form a + half-open ISO 8601 window. Timestamp offset rules match `stream_extent()`. + + Finite numbers and numeric strings contribute to the numeric statistics. + Present null, boolean, non-numeric, and non-finite values contribute to + `null_count`; missing fields contribute to neither count. `stddev` is the + population standard deviation: `0.0` for one numeric value and null for none. + + Args: + site_name: Exact site id to query, such as `MAIN`. Use `sites()` to + discover valid site ids. + asset_id: Exact asset id from `asset_ids()`, such as `Chiller 6`. + sensor: Optional exact field name from `measured_sensors()`. Omit it to + summarize all measured fields. + start: Optional inclusive ISO 8601 lower bound. + end: Optional exclusive ISO 8601 upper bound. + + Returns: + SensorStatsResult: Per-field numeric and null counts, range, mean, + population standard deviation, and earliest/latest numeric timestamps. + """ + if not _is_known_site(site_name): + return ErrorResult(error=f"unknown site {site_name}") + validation_error = _validate_dates(start, end) + if validation_error: + return ErrorResult(error=validation_error) + if not iot_db: + return ErrorResult(error="IoT records database not connected") + if sensor is not None and not sensor.strip(): + return ErrorResult(error="sensor must not be empty") + if sensor in RESERVED_FIELDS: + return ErrorResult( + error=f"sensor must be a telemetry field, not reserved metadata field {sensor}" + ) + + available_sensors = get_sensor_list(asset_id) + if not available_sensors: + return ErrorResult(error=f"unknown asset_id {asset_id} or no sensors found") + if sensor is not None and sensor not in available_sensors: + return ErrorResult(error=f"unknown sensor {sensor} for asset_id {asset_id}") + + targets = [sensor] if sensor is not None else available_sensors + accumulators = {target: _SensorAccumulator() for target in targets} + start_dt = _parse_iso_timestamp(start) if start is not None else None + end_dt = _parse_iso_timestamp(end) if end is not None else None + selector: Dict[str, Any] = { + "asset_id": asset_id, + "timestamp": {"$exists": True, "$ne": None}, + } + if sensor is not None: + selector[sensor] = {"$exists": True} + + records_in_window = 0 + try: + for doc, timestamp, timestamp_dt in _iter_records_in_window( + iot_db, + selector, + start_dt, + end_dt, + fields=["timestamp", *targets], + ): + records_in_window += 1 + for target in targets: + if target not in doc: + continue + numeric_value = _coerce_finite_number(doc.get(target)) + if numeric_value is None: + accumulators[target].add_invalid() + continue + accumulators[target].add_numeric( + numeric_value, timestamp, timestamp_dt + ) + except _TimestampHandlingError as e: + return ErrorResult(error=str(e)) + except Exception as e: + logger.error(f"sensor_stats failed: {e}") + return ErrorResult(error="unable to calculate sensor statistics") + + if records_in_window == 0: + return ErrorResult( + error=f"no records for asset_id {asset_id}" + + (f", sensor {sensor}" if sensor else "") + ) + + stats = [accumulators[target].result(target) for target in targets] + return SensorStatsResult( + site_name=site_name, + asset_id=asset_id, + stats=stats, + message=f"numeric stats for {len(stats)} sensor(s) on asset_id {asset_id}.", + ) + + def main(): mcp.run(transport="stdio") diff --git a/src/servers/iot/models.py b/src/servers/iot/models.py new file mode 100644 index 00000000..c7d2e04e --- /dev/null +++ b/src/servers/iot/models.py @@ -0,0 +1,136 @@ +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel + + +class ErrorResult(BaseModel): + error: str + + +class SitesResult(BaseModel): + sites: List[str] + + +class AssetsResult(BaseModel): + site_name: str + total_assets: int + assets: List[str] + message: str + + +class SensorsResult(BaseModel): + site_name: str + asset_id: str + total_sensors: int + sensors: List[str] + message: str + + +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_installed_sensors: int + message: str + + +class AssetSummary(BaseModel): + asset_id: str + description: Optional[str] + assettype: Optional[str] + vintage: Optional[str] + n_sensors: int + + +class AssetsWithMetadataResult(BaseModel): + site_name: str + total_assets: int + assets: List[AssetSummary] + message: str + + +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 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 HistoryResult(BaseModel): + site_name: str + asset_id: str + observations: List[Dict[str, Any]] + returned: int + next_cursor: Optional[str] + has_more: bool + start: Optional[str] + end: Optional[str] + message: str + + +class LatestReadingResult(BaseModel): + site_name: str + asset_id: str + timestamp: str + values: Dict[str, Any] + age_seconds: float + 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 SensorStat(BaseModel): + sensor: str + count: int + null_count: int + min: Optional[float] + max: Optional[float] + mean: Optional[float] + stddev: Optional[float] + first_timestamp: Optional[str] + last_timestamp: Optional[str] + + +class SensorStatsResult(BaseModel): + site_name: str + asset_id: str + stats: List[SensorStat] + message: str diff --git a/src/servers/iot/telemetry_helper.py b/src/servers/iot/telemetry_helper.py new file mode 100644 index 00000000..51d9f888 --- /dev/null +++ b/src/servers/iot/telemetry_helper.py @@ -0,0 +1,289 @@ +import base64 +import binascii +import json +import math +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Dict, Iterator, List, Optional, Set, Tuple + +from servers.iot.models import SensorCoverage, SensorStat + + +class _TimestampHandlingError(ValueError): + pass + + +@dataclass +class _SensorAccumulator: + count: int = 0 + null_count: int = 0 + min_value: Optional[float] = None + max_value: Optional[float] = None + mean_value: float = 0.0 + squared_deviation_sum: float = 0.0 + first_timestamp: Optional[str] = None + last_timestamp: Optional[str] = None + first_datetime: Optional[datetime] = None + last_datetime: Optional[datetime] = None + + def add_invalid(self) -> None: + self.null_count += 1 + + def add_numeric( + self, value: float, timestamp: str, timestamp_dt: datetime + ) -> None: + self.count += 1 + delta = value - self.mean_value + self.mean_value += delta / self.count + self.squared_deviation_sum += delta * (value - self.mean_value) + self.min_value = ( + value if self.min_value is None else min(self.min_value, value) + ) + self.max_value = ( + value if self.max_value is None else max(self.max_value, value) + ) + if self.first_datetime is None or timestamp_dt < self.first_datetime: + self.first_datetime = timestamp_dt + self.first_timestamp = timestamp + if self.last_datetime is None or timestamp_dt > self.last_datetime: + self.last_datetime = timestamp_dt + self.last_timestamp = timestamp + + def result(self, sensor: str) -> SensorStat: + mean = None + stddev = None + if self.count: + if math.isfinite(self.mean_value): + mean = self.mean_value + variance = self.squared_deviation_sum / self.count + if math.isfinite(variance): + stddev = math.sqrt(max(variance, 0.0)) + return SensorStat( + sensor=sensor, + count=self.count, + null_count=self.null_count, + min=self.min_value, + max=self.max_value, + mean=mean, + stddev=stddev, + first_timestamp=self.first_timestamp, + last_timestamp=self.last_timestamp, + ) + + +@dataclass +class _SensorCoverageAccumulator: + non_null_count: int = 0 + first_timestamp: Optional[str] = None + last_timestamp: Optional[str] = None + first_datetime: Optional[datetime] = None + last_datetime: Optional[datetime] = None + + def add_non_null(self, timestamp: str, timestamp_dt: datetime) -> None: + self.non_null_count += 1 + if self.first_datetime is None or timestamp_dt < self.first_datetime: + self.first_datetime = timestamp_dt + self.first_timestamp = timestamp + if self.last_datetime is None or timestamp_dt > self.last_datetime: + self.last_datetime = timestamp_dt + self.last_timestamp = timestamp + + def result(self, sensor: str) -> SensorCoverage: + return SensorCoverage( + sensor=sensor, + non_null_count=self.non_null_count, + first_timestamp=self.first_timestamp, + last_timestamp=self.last_timestamp, + ) + + +def _iter_records( + database: Any, + selector: Dict[str, Any], + fields: Optional[List[str]] = None, + sort: Optional[List[Dict[str, str]]] = None, + page_size: int = 1000, +) -> Iterator[Dict[str, Any]]: + """Yield records matching a selector across paged query results.""" + if not database: + return + if sort is None: + sort = [{"asset_id": "asc"}, {"timestamp": "asc"}] + bookmark: Optional[str] = None + 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 = database.find(selector, **kwargs) + docs = res.get("docs", []) + if not docs: + break + yield from docs + bookmark = res.get("bookmark") + if bookmark is None or len(docs) < page_size: + break + + +def _parse_iso_timestamp(value: Any) -> Optional[datetime]: + """Parse one ISO 8601 timestamp, returning None for invalid values.""" + if not isinstance(value, str): + return None + try: + return datetime.fromisoformat(value) + except (TypeError, ValueError): + return None + + +def _is_timezone_aware(value: datetime) -> bool: + return value.utcoffset() is not None + + +def _validate_dates(start: Optional[str], end: Optional[str]) -> Optional[str]: + """Return None when the optional ISO 8601 bounds are valid.""" + start_dt = _parse_iso_timestamp(start) if start is not None else None + end_dt = _parse_iso_timestamp(end) if end is not None else None + if start is not None and start_dt is None: + return "Invalid date format for start (expected ISO 8601)" + if end is not None and end_dt is None: + return "Invalid date format for end (expected ISO 8601)" + if start_dt is not None and end_dt is not None: + if _is_timezone_aware(start_dt) != _is_timezone_aware(end_dt): + return "start and end must use matching timezone awareness" + if start_dt >= end_dt: + return "start >= end" + return None + + +def _iter_records_in_window( + database: Any, + selector: Dict[str, Any], + start_dt: Optional[datetime], + end_dt: Optional[datetime], + fields: Optional[List[str]] = None, +) -> Iterator[Tuple[Dict[str, Any], str, datetime]]: + """Yield timestamped records in a parsed half-open time window.""" + stream_is_aware: Optional[bool] = None + for doc in _iter_records(database, selector, fields=fields): + timestamp = doc.get("timestamp") + if timestamp is None: + continue + timestamp_dt = _parse_iso_timestamp(timestamp) + if timestamp_dt is None: + raise _TimestampHandlingError( + "telemetry record has an invalid ISO 8601 timestamp" + ) + + timestamp_is_aware = _is_timezone_aware(timestamp_dt) + if stream_is_aware is None: + stream_is_aware = timestamp_is_aware + for bound in (start_dt, end_dt): + if ( + bound is not None + and _is_timezone_aware(bound) != stream_is_aware + ): + raise _TimestampHandlingError( + "timestamp bounds must use the same timezone awareness " + "as telemetry timestamps" + ) + elif timestamp_is_aware != stream_is_aware: + raise _TimestampHandlingError( + "telemetry timestamps use mixed timezone awareness" + ) + + if start_dt is not None and timestamp_dt < start_dt: + continue + if end_dt is not None and timestamp_dt >= end_dt: + continue + yield doc, timestamp, timestamp_dt + + +def _coerce_finite_number(value: Any) -> Optional[float]: + """Return a finite float for numeric values and numeric strings.""" + if isinstance(value, bool): + return None + try: + number = float(value) + except (TypeError, ValueError): + return None + return number if math.isfinite(number) else None + + +def _history_cursor_context( + site_name: str, + asset_id: str, + start: Optional[str], + end: Optional[str], + sensors: Optional[List[str]], +) -> Dict[str, Any]: + return { + "site_name": site_name, + "asset_id": asset_id, + "start": start, + "end": end, + "sensors": sensors, + } + + +def _encode_history_cursor(offset: int, context: Dict[str, Any]) -> str: + payload = json.dumps( + {"version": 1, "offset": offset, "context": context}, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + return base64.urlsafe_b64encode(payload).decode("ascii").rstrip("=") + + +def _decode_history_cursor( + cursor: str, expected_context: Dict[str, Any] +) -> Tuple[Optional[int], Optional[str]]: + try: + padding = "=" * (-len(cursor) % 4) + payload = json.loads( + base64.urlsafe_b64decode(cursor + padding).decode("utf-8") + ) + except (binascii.Error, UnicodeDecodeError, ValueError, json.JSONDecodeError): + return None, "invalid cursor" + if not isinstance(payload, dict) or payload.get("version") != 1: + return None, "invalid cursor" + offset = payload.get("offset") + if isinstance(offset, bool) or not isinstance(offset, int) or offset < 0: + return None, "invalid cursor" + if payload.get("context") != expected_context: + return None, "cursor does not match history query" + return offset, None + + +def _history_observation( + doc: Dict[str, Any], + asset_id: str, + timestamp: str, + sensors: Optional[List[str]], + reserved_fields: Set[str], +) -> Dict[str, Any]: + observation: Dict[str, Any] = { + "asset_id": asset_id, + "timestamp": timestamp, + } + if sensors is None: + observation.update( + { + field: value + for field, value in doc.items() + if field not in reserved_fields + } + ) + else: + observation.update( + {field: doc[field] for field in sensors if field in doc} + ) + return observation + + +def _timestamp_age_seconds(timestamp_dt: datetime) -> float: + if not _is_timezone_aware(timestamp_dt): + timestamp_dt = timestamp_dt.replace(tzinfo=timezone.utc) + return ( + datetime.now(timezone.utc) - timestamp_dt.astimezone(timezone.utc) + ).total_seconds() diff --git a/src/servers/iot/tests/test_tools.py b/src/servers/iot/tests/test_tools.py index ff8d9c30..8896275e 100644 --- a/src/servers/iot/tests/test_tools.py +++ b/src/servers/iot/tests/test_tools.py @@ -15,11 +15,37 @@ async def test_registry_tools_are_registered(self): "asset_ids", "assets", "find_assets_by_sensors", + "history", "installed_sensors", + "latest_reading", "measured_sensors", + "sensor_coverage", + "sensor_stats", "sites", + "stream_extent", ] + @pytest.mark.anyio + async def test_stream_extent_description_is_storage_neutral(self): + tools = await mcp.list_tools() + descriptions = { + tool.name: tool.description + for tool in tools + if tool.name + in { + "history", + "latest_reading", + "sensor_coverage", + "sensor_stats", + "stream_extent", + } + } + + assert all( + "couchdb" not in description.lower() + for description in descriptions.values() + ) + class TestSites: @pytest.mark.anyio @@ -473,6 +499,1222 @@ def find_records(selector, **kwargs): ] +class TestStreamExtent: + @pytest.mark.anyio + async def test_invalid_site(self): + data = await call_tool( + mcp, + "stream_extent", + {"site_name": "INVALID", "asset_id": "Pump-1"}, + ) + + assert "error" in data + assert "unknown site" in data["error"] + + @pytest.mark.anyio + async def test_invalid_date(self, mock_asset_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "stream_extent", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "start": "not-a-date", + }, + ) + + assert "error" in data + assert "Invalid date format" in data["error"] + + @pytest.mark.anyio + async def test_start_must_precede_end(self, mock_asset_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "stream_extent", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "start": "2024-01-02T00:00:00", + "end": "2024-01-01T00:00:00", + }, + ) + + assert data == {"error": "start >= end"} + + @pytest.mark.anyio + async def test_db_disconnected(self, mock_asset_db, no_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "stream_extent", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert "error" in data + assert "not connected" in data["error"].lower() + + @pytest.mark.anyio + async def test_rejects_reserved_sensor_field(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "stream_extent", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensor": "asset_id", + }, + ) + + assert data == { + "error": ( + "sensor must be a telemetry field, " + "not reserved metadata field asset_id" + ) + } + mock_iot_db.find.assert_not_called() + + @pytest.mark.anyio + async def test_with_mock_iot_db(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {"timestamp": "2024-01-01T00:00:00"}, + {"timestamp": "2024-01-01T00:01:00"}, + {"timestamp": "2024-01-01T00:02:00"}, + ] + } + + data = await call_tool( + mcp, + "stream_extent", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data == { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensor": None, + "start_time": "2024-01-01T00:00:00", + "end_time": "2024-01-01T00:02:00", + "total_records": 3, + "exceeds_page_limit": False, + "approx_interval_seconds": 60.0, + "message": ( + "3 record(s) for asset_id Pump-1 from " + "2024-01-01T00:00:00 to 2024-01-01T00:02:00." + ), + } + mock_iot_db.find.assert_called_once_with( + { + "asset_id": "Pump-1", + "timestamp": {"$exists": True, "$ne": None}, + }, + limit=1000, + sort=[{"asset_id": "asc"}, {"timestamp": "asc"}], + fields=["timestamp"], + ) + + @pytest.mark.anyio + async def test_sensor_and_window_selector(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {"timestamp": "2023-12-31T23:59:00"}, + {"timestamp": "2024-01-01T00:01:00"}, + {"timestamp": "2024-01-01T00:05:00"}, + ] + } + + data = await call_tool( + mcp, + "stream_extent", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensor": "Pressure", + "start": "2024-01-01T00:00:00", + "end": "2024-01-01T00:05:00", + }, + ) + + assert data["sensor"] == "Pressure" + assert data["total_records"] == 1 + assert data["start_time"] == "2024-01-01T00:01:00" + assert data["end_time"] == "2024-01-01T00:01:00" + selector = mock_iot_db.find.call_args.args[0] + assert selector == { + "asset_id": "Pump-1", + "timestamp": {"$exists": True, "$ne": None}, + "Pressure": {"$exists": True, "$ne": None}, + } + + @pytest.mark.anyio + async def test_date_only_window_selector_preserves_input( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {"timestamp": "2023-12-31T23:59:00"}, + {"timestamp": "2024-01-01T12:00:00"}, + {"timestamp": "2024-01-02T00:00:00"}, + ] + } + + data = await call_tool( + mcp, + "stream_extent", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "start": "2024-01-01", + "end": "2024-01-02", + }, + ) + + assert data["total_records"] == 1 + assert data["start_time"] == "2024-01-01T12:00:00" + assert data["end_time"] == "2024-01-01T12:00:00" + selector = mock_iot_db.find.call_args.args[0] + assert selector["timestamp"] == {"$exists": True, "$ne": None} + + @pytest.mark.anyio + async def test_bounds_must_match_stream_timezone_awareness( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [{"timestamp": "2024-01-01T00:00:00"}] + } + + data = await call_tool( + mcp, + "stream_extent", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "start": "2024-01-01T00:00:00+00:00", + }, + ) + + assert data == { + "error": ( + "timestamp bounds must use the same timezone awareness " + "as telemetry timestamps" + ) + } + + @pytest.mark.anyio + async def test_compares_explicit_offsets_chronologically( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {"timestamp": "2023-12-31T23:00:00+00:00"}, + {"timestamp": "2024-01-01T00:30:00+02:00"}, + {"timestamp": "2024-01-01T00:00:00+00:00"}, + ] + } + + data = await call_tool( + mcp, + "stream_extent", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "start": "2023-12-31T22:15:00+00:00", + "end": "2023-12-31T23:30:00+00:00", + }, + ) + + assert data["total_records"] == 2 + assert data["start_time"] == "2024-01-01T00:30:00+02:00" + assert data["end_time"] == "2023-12-31T23:00:00+00:00" + assert data["approx_interval_seconds"] == 1800.0 + + @pytest.mark.anyio + async def test_mixed_stream_timezone_awareness_returns_error( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {"timestamp": "2024-01-01T00:00:00"}, + {"timestamp": "2024-01-01T01:00:00+00:00"}, + ] + } + + data = await call_tool( + mcp, + "stream_extent", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data == {"error": "telemetry timestamps use mixed timezone awareness"} + + @pytest.mark.anyio + async def test_invalid_stream_timestamp_returns_error( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = {"docs": [{"timestamp": "not-a-date"}]} + + data = await call_tool( + mcp, + "stream_extent", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data == { + "error": "telemetry record has an invalid ISO 8601 timestamp" + } + + @pytest.mark.anyio + async def test_records_without_timestamps_are_not_counted( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {}, + {"timestamp": "2024-01-01T00:00:00"}, + {"timestamp": None}, + ] + } + + data = await call_tool( + mcp, + "stream_extent", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data["total_records"] == 1 + assert data["start_time"] == "2024-01-01T00:00:00" + assert data["end_time"] == "2024-01-01T00:00:00" + + @pytest.mark.anyio + async def test_no_records(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = {"docs": []} + + data = await call_tool( + mcp, + "stream_extent", + {"site_name": "MAIN", "asset_id": "Pump-1", "sensor": "Pressure"}, + ) + + assert "error" in data + assert data["error"] == "no records for asset_id Pump-1, sensor Pressure" + + @pytest.mark.anyio + async def test_query_error_does_not_expose_storage_backend( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.side_effect = RuntimeError("CouchDB unavailable") + + data = await call_tool( + mcp, + "stream_extent", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data == {"error": "unable to inspect telemetry stream extent"} + assert "couchdb" not in data["error"].lower() + + @requires_iot_db + @pytest.mark.anyio + async def test_discovery_integration(self): + data = await call_tool( + mcp, + "stream_extent", + {"site_name": "MAIN", "asset_id": "Chiller 6"}, + ) + + assert data["asset_id"] == "Chiller 6" + assert data["total_records"] > 0 + assert data["start_time"] is not None + assert data["end_time"] is not None + + +class TestHistory: + @pytest.mark.anyio + async def test_invalid_site(self): + data = await call_tool( + mcp, + "history", + {"site_name": "INVALID", "asset_id": "Pump-1"}, + ) + + assert "unknown site" in data["error"] + + @pytest.mark.anyio + async def test_invalid_date(self, mock_asset_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "history", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "start": "not-a-date", + }, + ) + + assert "Invalid date format" in data["error"] + + @pytest.mark.anyio + async def test_db_disconnected(self, mock_asset_db, no_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "history", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert "not connected" in data["error"].lower() + + @pytest.mark.anyio + async def test_rejects_limits_outside_page_range( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + for limit in (0, 1001): + data = await call_tool( + mcp, + "history", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "limit": limit, + }, + ) + assert data == {"error": "limit must be between 1 and 1000"} + mock_iot_db.find.assert_not_called() + + @pytest.mark.anyio + async def test_rejects_empty_sensor_list(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "history", + {"site_name": "MAIN", "asset_id": "Pump-1", "sensors": []}, + ) + + assert data == {"error": "sensors must not be empty when provided"} + mock_iot_db.find.assert_not_called() + + @pytest.mark.anyio + async def test_rejects_unknown_sensor(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [{"timestamp": "2024-01-01T00:00:00", "Temp": 1.0}] + } + + data = await call_tool( + mcp, + "history", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensors": ["Pressure"], + }, + ) + + assert data == {"error": "unknown sensors ['Pressure'] for asset_id Pump-1"} + + @pytest.mark.anyio + async def test_exact_cursor_pagination_and_metadata_exclusion( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + { + "_id": "a", + "_rev": "1-a", + "asset_id": "Pump-1", + "timestamp": "2024-01-01T00:00:00", + "Temp": 1.0, + }, + { + "_id": "b", + "asset_id": "Pump-1", + "timestamp": "2024-01-01T00:01:00", + "Temp": 2.0, + }, + { + "_id": "c", + "asset_id": "Pump-1", + "timestamp": "2024-01-01T00:02:00", + "Temp": 3.0, + }, + ] + } + + first = await call_tool( + mcp, + "history", + {"site_name": "MAIN", "asset_id": "Pump-1", "limit": 2}, + ) + + assert first["returned"] == 2 + assert first["has_more"] is True + assert first["next_cursor"] is not None + assert first["observations"] == [ + { + "asset_id": "Pump-1", + "timestamp": "2024-01-01T00:00:00", + "Temp": 1.0, + }, + { + "asset_id": "Pump-1", + "timestamp": "2024-01-01T00:01:00", + "Temp": 2.0, + }, + ] + + second = await call_tool( + mcp, + "history", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "limit": 2, + "cursor": first["next_cursor"], + }, + ) + + assert second["returned"] == 1 + assert second["has_more"] is False + assert second["next_cursor"] is None + assert second["observations"][0]["timestamp"] == "2024-01-01T00:02:00" + + @pytest.mark.anyio + async def test_cursor_is_bound_to_query(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {"timestamp": "2024-01-01T00:00:00", "Temp": 1.0}, + {"timestamp": "2024-01-01T00:01:00", "Temp": 2.0}, + ] + } + first = await call_tool( + mcp, + "history", + {"site_name": "MAIN", "asset_id": "Pump-1", "limit": 1}, + ) + + data = await call_tool( + mcp, + "history", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "start": "2024-01-01T00:00:00", + "limit": 1, + "cursor": first["next_cursor"], + }, + ) + + assert data == {"error": "cursor does not match history query"} + + @pytest.mark.anyio + async def test_invalid_cursor(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "history", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "cursor": "not-a-cursor", + }, + ) + + assert data == {"error": "invalid cursor"} + mock_iot_db.find.assert_not_called() + + @pytest.mark.anyio + async def test_sensor_projection_deduplicates_fields( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + { + "timestamp": "2024-01-01T00:00:00", + "Temp": 1.0, + "Pressure": 5.0, + } + ] + } + + data = await call_tool( + mcp, + "history", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensors": ["Temp", "Temp"], + }, + ) + + assert data["observations"] == [ + { + "asset_id": "Pump-1", + "timestamp": "2024-01-01T00:00:00", + "Temp": 1.0, + } + ] + history_query = mock_iot_db.find.call_args_list[1] + assert history_query.kwargs["fields"] == ["timestamp", "Temp"] + + @pytest.mark.anyio + async def test_half_open_window(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {"timestamp": "2024-01-01T00:00:00", "Temp": 0.0}, + {"timestamp": "2024-01-01T00:01:00", "Temp": 1.0}, + {"timestamp": "2024-01-01T00:02:00", "Temp": 2.0}, + ] + } + + data = await call_tool( + mcp, + "history", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "start": "2024-01-01T00:01:00", + "end": "2024-01-01T00:02:00", + }, + ) + + assert data["returned"] == 1 + assert data["observations"][0]["timestamp"] == "2024-01-01T00:01:00" + + @pytest.mark.anyio + async def test_rejects_non_chronological_timestamp_representations( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {"timestamp": "2023-12-31T23:00:00+00:00", "Temp": 1.0}, + {"timestamp": "2024-01-01T00:30:00+02:00", "Temp": 2.0}, + ] + } + + data = await call_tool( + mcp, + "history", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data == { + "error": "telemetry timestamps cannot be returned in chronological order" + } + + @requires_iot_db + @pytest.mark.anyio + async def test_discovery_integration(self): + first = await call_tool( + mcp, + "history", + {"site_name": "MAIN", "asset_id": "Chiller 6", "limit": 2}, + ) + + assert first["returned"] == 2 + assert first["has_more"] is True + second = await call_tool( + mcp, + "history", + { + "site_name": "MAIN", + "asset_id": "Chiller 6", + "limit": 2, + "cursor": first["next_cursor"], + }, + ) + assert second["returned"] == 2 + assert ( + first["observations"][-1]["timestamp"] + < second["observations"][0]["timestamp"] + ) + + +class TestLatestReading: + @pytest.mark.anyio + async def test_invalid_site(self): + data = await call_tool( + mcp, + "latest_reading", + {"site_name": "INVALID", "asset_id": "Pump-1"}, + ) + + assert "unknown site" in data["error"] + + @pytest.mark.anyio + async def test_db_disconnected(self, mock_asset_db, no_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "latest_reading", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert "not connected" in data["error"].lower() + + @pytest.mark.anyio + async def test_rejects_reserved_sensor(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "latest_reading", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensor": "timestamp", + }, + ) + + assert "reserved metadata field timestamp" in data["error"] + mock_iot_db.find.assert_not_called() + + @pytest.mark.anyio + async def test_rejects_unknown_sensor(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [{"timestamp": "2024-01-01T00:00:00", "Temp": 1.0}] + } + + data = await call_tool( + mcp, + "latest_reading", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensor": "Pressure", + }, + ) + + assert data == {"error": "unknown sensor Pressure for asset_id Pump-1"} + + @pytest.mark.anyio + async def test_selects_latest_parsed_timestamp_and_excludes_metadata( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + { + "_id": "a", + "timestamp": "2023-12-31T23:00:00+00:00", + "Temp": 1.0, + }, + { + "_rev": "1-b", + "timestamp": "2024-01-01T00:30:00+02:00", + "Temp": 2.0, + }, + { + "dataset": "sample", + "timestamp": "2024-01-01T00:00:00+00:00", + "Temp": 3.0, + }, + ] + } + + data = await call_tool( + mcp, + "latest_reading", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data["timestamp"] == "2024-01-01T00:00:00+00:00" + assert data["values"] == {"Temp": 3.0} + assert data["age_seconds"] > 0 + + @pytest.mark.anyio + async def test_sensor_filter_returns_only_requested_value( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.side_effect = [ + { + "docs": [ + { + "timestamp": "2024-01-01T00:00:00", + "Temp": 1.0, + "Pressure": 5.0, + } + ] + }, + { + "docs": [ + {"timestamp": "2024-01-01T00:00:00", "Temp": 1.0}, + {"timestamp": "2024-01-01T00:01:00", "Temp": 2.0}, + ] + }, + ] + + data = await call_tool( + mcp, + "latest_reading", + {"site_name": "MAIN", "asset_id": "Pump-1", "sensor": "Temp"}, + ) + + assert data["timestamp"] == "2024-01-01T00:01:00" + assert data["values"] == {"Temp": 2.0} + + @pytest.mark.anyio + async def test_no_records(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = {"docs": []} + + data = await call_tool( + mcp, + "latest_reading", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data == {"error": "no records for asset_id Pump-1"} + + @pytest.mark.anyio + async def test_timestamp_error_is_returned(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [{"timestamp": "not-a-date", "Temp": 1.0}] + } + + data = await call_tool( + mcp, + "latest_reading", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data == { + "error": "telemetry record has an invalid ISO 8601 timestamp" + } + + @requires_iot_db + @pytest.mark.anyio + async def test_discovery_integration(self): + data = await call_tool( + mcp, + "latest_reading", + { + "site_name": "MAIN", + "asset_id": "Chiller 6", + "sensor": "Chiller 6 Supply Temperature", + }, + ) + + assert data["timestamp"] == "2020-06-30T23:45:00" + assert set(data["values"]) == {"Chiller 6 Supply Temperature"} + assert data["age_seconds"] > 0 + + +class TestSensorCoverage: + @pytest.mark.anyio + async def test_invalid_site(self): + data = await call_tool( + mcp, + "sensor_coverage", + {"site_name": "INVALID", "asset_id": "Pump-1"}, + ) + + assert "unknown site" in data["error"] + + @pytest.mark.anyio + async def test_db_disconnected(self, mock_asset_db, no_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "sensor_coverage", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert "not connected" in data["error"].lower() + + @pytest.mark.anyio + async def test_counts_non_null_values_and_chronological_bounds( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + { + "timestamp": "2024-01-01T00:02:00", + "Temp": 2.0, + "Pressure": None, + }, + { + "timestamp": "2024-01-01T00:00:00", + "Temp": None, + "Pressure": 1.0, + "Mode": None, + }, + { + "timestamp": "2024-01-01T00:01:00", + "Temp": 1.0, + "Pressure": False, + }, + ] + } + + data = await call_tool( + mcp, + "sensor_coverage", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data["docs_scanned"] == 3 + coverage = {item["sensor"]: item for item in data["sensors"]} + assert list(item["sensor"] for item in data["sensors"]) == [ + "Mode", + "Pressure", + "Temp", + ] + assert coverage["Mode"] == { + "sensor": "Mode", + "non_null_count": 0, + "first_timestamp": None, + "last_timestamp": None, + } + assert coverage["Pressure"]["non_null_count"] == 2 + assert coverage["Pressure"]["first_timestamp"] == "2024-01-01T00:00:00" + assert coverage["Pressure"]["last_timestamp"] == "2024-01-01T00:01:00" + assert coverage["Temp"]["non_null_count"] == 2 + assert coverage["Temp"]["first_timestamp"] == "2024-01-01T00:01:00" + assert coverage["Temp"]["last_timestamp"] == "2024-01-01T00:02:00" + + @pytest.mark.anyio + async def test_no_records(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = {"docs": []} + + data = await call_tool( + mcp, + "sensor_coverage", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data == {"error": "unknown asset_id Pump-1 or no records found"} + + @pytest.mark.anyio + async def test_timestamp_error_is_returned(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [{"timestamp": "not-a-date", "Temp": 1.0}] + } + + data = await call_tool( + mcp, + "sensor_coverage", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert data == { + "error": "telemetry record has an invalid ISO 8601 timestamp" + } + + @requires_iot_db + @pytest.mark.anyio + async def test_discovery_integration(self): + data = await call_tool( + mcp, + "sensor_coverage", + { + "site_name": "MAIN", + "asset_id": "Chiller 6", + }, + ) + + assert data["docs_scanned"] > 0 + assert any( + item["sensor"] == "Chiller 6 Supply Temperature" + for item in data["sensors"] + ) + + +class TestSensorStats: + @pytest.mark.anyio + async def test_invalid_site(self): + data = await call_tool( + mcp, + "sensor_stats", + {"site_name": "INVALID", "asset_id": "Pump-1"}, + ) + + assert "unknown site" in data["error"] + + @pytest.mark.anyio + async def test_invalid_date(self, mock_asset_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "sensor_stats", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "start": "not-a-date", + }, + ) + + assert "Invalid date format" in data["error"] + + @pytest.mark.anyio + async def test_db_disconnected(self, mock_asset_db, no_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "sensor_stats", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + assert "not connected" in data["error"].lower() + + @pytest.mark.anyio + async def test_rejects_reserved_sensor_field(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + + data = await call_tool( + mcp, + "sensor_stats", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensor": "timestamp", + }, + ) + + assert data == { + "error": ( + "sensor must be a telemetry field, " + "not reserved metadata field timestamp" + ) + } + mock_iot_db.find.assert_not_called() + + @pytest.mark.anyio + async def test_rejects_unknown_sensor(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [{"timestamp": "2024-01-01T00:00:00", "Pressure": 1.0}] + } + + data = await call_tool( + mcp, + "sensor_stats", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensor": "Temperature", + }, + ) + + assert data == { + "error": "unknown sensor Temperature for asset_id Pump-1" + } + + @pytest.mark.anyio + async def test_single_sensor_numeric_summary(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {"timestamp": "2024-01-01T00:02:00", "Temp": None}, + {"timestamp": "2024-01-01T00:00:00", "Temp": 1.0}, + {"timestamp": "2024-01-01T00:04:00", "Temp": "bad"}, + {"timestamp": "2024-01-01T00:01:00", "Temp": "2.5"}, + {"timestamp": "2024-01-01T00:03:00", "Temp": True}, + {"timestamp": "2024-01-01T00:05:00", "Temp": float("inf")}, + {"timestamp": "2024-01-01T00:06:00"}, + ] + } + + data = await call_tool( + mcp, + "sensor_stats", + {"site_name": "MAIN", "asset_id": "Pump-1", "sensor": "Temp"}, + ) + + assert data == { + "site_name": "MAIN", + "asset_id": "Pump-1", + "stats": [ + { + "sensor": "Temp", + "count": 2, + "null_count": 4, + "min": 1.0, + "max": 2.5, + "mean": 1.75, + "stddev": 0.75, + "first_timestamp": "2024-01-01T00:00:00", + "last_timestamp": "2024-01-01T00:01:00", + } + ], + "message": "numeric stats for 1 sensor(s) on asset_id Pump-1.", + } + stats_query = mock_iot_db.find.call_args_list[1] + assert stats_query.args[0] == { + "asset_id": "Pump-1", + "timestamp": {"$exists": True, "$ne": None}, + "Temp": {"$exists": True}, + } + assert stats_query.kwargs["fields"] == ["timestamp", "Temp"] + + @pytest.mark.anyio + async def test_all_measured_sensors_when_omitted( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + { + "timestamp": "2024-01-01T00:00:00", + "Temp": 10.0, + "Pressure": 1.0, + "Mode": "auto", + }, + { + "timestamp": "2024-01-01T00:01:00", + "Temp": 20.0, + "Pressure": 3.0, + "Mode": "manual", + }, + ] + } + + data = await call_tool( + mcp, + "sensor_stats", + {"site_name": "MAIN", "asset_id": "Pump-1"}, + ) + + stats = {item["sensor"]: item for item in data["stats"]} + assert list(item["sensor"] for item in data["stats"]) == [ + "Mode", + "Pressure", + "Temp", + ] + assert stats["Pressure"]["mean"] == 2.0 + assert stats["Temp"]["stddev"] == 5.0 + assert stats["Mode"]["count"] == 0 + assert stats["Mode"]["null_count"] == 2 + + @pytest.mark.anyio + async def test_half_open_window_uses_chronological_order( + self, mock_asset_db, mock_iot_db + ): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [ + {"timestamp": "2024-01-01T00:05:00", "Temp": 5.0}, + {"timestamp": "2024-01-01T00:02:00", "Temp": 2.0}, + {"timestamp": "2024-01-01T00:00:00", "Temp": 0.0}, + {"timestamp": "2024-01-01T00:04:00", "Temp": 4.0}, + ] + } + + data = await call_tool( + mcp, + "sensor_stats", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensor": "Temp", + "start": "2024-01-01T00:02:00", + "end": "2024-01-01T00:05:00", + }, + ) + + stat = data["stats"][0] + assert stat["count"] == 2 + assert stat["mean"] == 3.0 + assert stat["first_timestamp"] == "2024-01-01T00:02:00" + assert stat["last_timestamp"] == "2024-01-01T00:04:00" + + @pytest.mark.anyio + async def test_no_records_in_window(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [{"timestamp": "2024-01-01T00:00:00", "Temp": 1.0}] + } + + data = await call_tool( + mcp, + "sensor_stats", + { + "site_name": "MAIN", + "asset_id": "Pump-1", + "sensor": "Temp", + "start": "2024-01-02T00:00:00", + }, + ) + + assert data == {"error": "no records for asset_id Pump-1, sensor Temp"} + + @pytest.mark.anyio + async def test_timestamp_error_is_returned(self, mock_asset_db, mock_iot_db): + mock_asset_db.find.return_value = {"docs": [{"siteid": "MAIN"}]} + mock_iot_db.find.return_value = { + "docs": [{"timestamp": "not-a-date", "Temp": 1.0}] + } + + data = await call_tool( + mcp, + "sensor_stats", + {"site_name": "MAIN", "asset_id": "Pump-1", "sensor": "Temp"}, + ) + + assert data == { + "error": "telemetry record has an invalid ISO 8601 timestamp" + } + + @requires_iot_db + @pytest.mark.anyio + async def test_discovery_integration(self): + data = await call_tool( + mcp, + "sensor_stats", + { + "site_name": "MAIN", + "asset_id": "Chiller 6", + "sensor": "Chiller 6 Supply Temperature", + "start": "2020-06-10", + "end": "2020-06-11", + }, + ) + + stat = data["stats"][0] + assert stat["sensor"] == "Chiller 6 Supply Temperature" + assert stat["count"] > 0 + assert stat["min"] is not None + assert stat["max"] is not None + + class TestAssets: @pytest.mark.anyio async def test_invalid_site(self):