get_statistics_range

Instructions

Implementation Reference

• app/server.py:1392-1431 (handler)

  MCP tool handler for 'get_statistics_range'. Decorated with @mcp.tool() and @async_handler. Calls get_entity_statistics_range() from app.hass with the entity_id, start_time, end_time, and period. Catches ValueError for invalid inputs.

  @mcp.tool()
  @async_handler("get_statistics_range")
  async def get_statistics_range(
      entity_id: str,
      start_time: str,
      end_time: Optional[str] = None,
      period: str = "hour",
  ) -> Dict[str, Any]:
      """
      Get long-term aggregated statistics for an entity over a date/time range.
  
      Same data source as `get_statistics`, but with an explicit window —
      useful for "what was my power usage from Jan 1 to Jan 31?" type
      questions. Aggregated bucket data survives the short-term retention
      window, so this works for data months/years old.
  
      Args:
          entity_id: The entity (must be statistics-tracked).
          start_time: ISO-8601 start (`2026-01-01` or
                      `2026-01-01T00:00:00Z`). UTC if no offset.
          end_time: ISO-8601 end. Defaults to now.
          period: `5minute`, `hour`, `day`, `week`, or `month`.
  
      Returns:
          `entity_id`, `period`, `start_time`, `end_time`, `statistics`.
  
      Examples:
          get_statistics_range("sensor.energy", "2026-01-01", "2026-02-01", period="day")
          get_statistics_range("sensor.temperature", "2026-05-01", period="hour")
      """
      logger.info(
          f"Getting statistics range for {entity_id}: "
          f"{start_time} -> {end_time or 'now'}, period={period}"
      )
      try:
          return await get_entity_statistics_range(
              entity_id, start_time, end_time, period=period
          )
      except ValueError as e:
          return {"entity_id": entity_id, "error": str(e), "statistics": []}

• app/hass.py:717-771 (helper)

  Core implementation of get_entity_statistics_range(). Validates period against _STATISTICS_PERIODS, parses ISO-8601 datetimes via _parse_iso_dt(), calls HA's WebSocket API (recorder/statistics_during_period) via app.ws.call_ws(), and returns the result dict with entity_id, period, start_time, end_time, and statistics list.

  @handle_api_errors
  async def get_entity_statistics_range(
      entity_id: str,
      start_time: Union[str, datetime],
      end_time: Optional[Union[str, datetime]] = None,
      period: str = "hour",
  ) -> Dict[str, Any]:
      """Get long-term statistics for an entity over a date/time range.
  
      Hits HA's `recorder/statistics_during_period` over the WebSocket API.
      Statistics survive the recorder's short-term retention window
      (default 10 days), so this is the right call for anything older than
      that or anything you want aggregated (mean / min / max per period).
  
      Args:
          entity_id: The entity (must have a `state_class` that HA records
                     as statistics — e.g. `measurement`, `total_increasing`).
          start_time: ISO-8601 string or datetime, UTC if naive.
          end_time: ISO-8601 string or datetime, defaults to now.
          period: Aggregation bucket — one of `5minute`, `hour`, `day`,
                  `week`, `month`. Defaults to `hour`.
  
      Returns:
          ``{"entity_id", "period", "start_time", "end_time", "statistics"}``.
          ``statistics`` is the list HA returned (each entry has `start`,
          `end`, `mean`, `min`, `max`, optionally `sum`/`state`).
      """
      if period not in _STATISTICS_PERIODS:
          raise ValueError(
              f"period must be one of {sorted(_STATISTICS_PERIODS)}, got {period!r}"
          )
      start_dt = _parse_iso_dt(start_time)
      end_dt = _parse_iso_dt(end_time) if end_time is not None else datetime.now(timezone.utc)
      if start_dt >= end_dt:
          raise ValueError("start_time must be before end_time")
  
      from app.ws import call_ws
      start_iso = start_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
      end_iso = end_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
      result = await call_ws(
          "recorder/statistics_during_period",
          start_time=start_iso,
          end_time=end_iso,
          statistic_ids=[entity_id],
          period=period,
      )
      return {
          "entity_id": entity_id,
          "period": period,
          "start_time": start_iso,
          "end_time": end_iso,
          # HA returns `{entity_id: [points...]}`; flatten to the list when
          # we only asked for one entity.
          "statistics": (result or {}).get(entity_id, []) if isinstance(result, dict) else result,
      }

• app/hass.py:656-672 (helper)

  Helper function _parse_iso_dt() that coerces user-supplied ISO-8601 strings or datetimes to tz-aware UTC datetimes. Used by get_entity_statistics_range() to validate and normalize start_time/end_time.

  def _parse_iso_dt(value: Union[str, datetime]) -> datetime:
      """Coerce a user-supplied datetime to a tz-aware UTC datetime.
  
      Accepts a `datetime` (assumed UTC if naive) or an ISO-8601 string
      (`2026-01-15`, `2026-01-15T12:00:00`, `2026-01-15T12:00:00Z`, or with
      explicit offset). Raises ValueError on anything else.
      """
      if isinstance(value, datetime):
          return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
      if not isinstance(value, str):
          raise ValueError(f"datetime must be str or datetime, got {type(value).__name__}")
      s = value.strip()
      # `fromisoformat` in 3.11+ accepts `Z`, but be explicit for clarity.
      if s.endswith("Z"):
          s = s[:-1] + "+00:00"
      dt = datetime.fromisoformat(s)
      return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)

• app/hass.py:714-714 (schema)

  Allowed period values (_STATISTICS_PERIODS constant): {'5minute', 'hour', 'day', 'week', 'month'}. This is the input validation constraint for the 'period' parameter.

  _STATISTICS_PERIODS = {"5minute", "hour", "day", "week", "month"}

• app/ws.py:38-79 (helper)

  WebSocket client call_ws() used by get_entity_statistics_range() to send the 'recorder/statistics_during_period' message to Home Assistant. Opens a fresh WS connection, authenticates, sends the request, and returns the result.

  async def call_ws(message_type: str, **payload: Any) -> Any:
      """Send a single request over the HA WebSocket API and return its result.
  
      Args:
          message_type: HA WS message type, e.g. ``"recorder/statistics_during_period"``.
          **payload: Additional fields merged into the request body.
  
      Returns:
          The ``result`` field of HA's success response — shape depends on
          the message type (dict, list, etc.).
  
      Raises:
          HassWebSocketError: HA replied with ``success=False`` or auth failed.
      """
      url = _ws_url()
      ssl_ctx = _build_ssl_context() if url.startswith("wss://") else None
  
      async with websockets.connect(url, ssl=ssl_ctx) as ws:
          # 1. Server sends auth_required first.
          auth_required = json.loads(await ws.recv())
          if auth_required.get("type") != "auth_required":
              raise HassWebSocketError(
                  f"Unexpected initial WS message: {auth_required}"
              )
  
          # 2. Authenticate.
          await ws.send(json.dumps({"type": "auth", "access_token": HA_TOKEN}))
          auth_result = json.loads(await ws.recv())
          if auth_result.get("type") != "auth_ok":
              raise HassWebSocketError(f"WS authentication failed: {auth_result}")
  
          # 3. Send the actual request. HA requires monotonically increasing
          #    `id` per connection; since we open a fresh connection per call,
          #    `1` is always valid.
          await ws.send(json.dumps({"id": 1, "type": message_type, **payload}))
          response = json.loads(await ws.recv())
  
          if not response.get("success", False):
              raise HassWebSocketError(
                  f"WS request {message_type!r} failed: {response.get('error', response)}"
              )
          return response.get("result")