analyze_energy_usage

Instructions

Implementation Reference

• src/domoticz_mcp/server.py:1166-1188 (handler)

  The MCP tool handler for analyze_energy_usage. Fetches all devices from Domoticz, filters for those with 'Usage' or 'CounterToday' fields (energy-reporting devices), and returns a JSON summary of current usage and today's total.

  @mcp.tool()
  async def analyze_energy_usage() -> str:
      """Analyze all energy-reporting devices and summarize their 'Today' usage."""
      async with create_client() as client:
          devices = await _get_cached_data(client, _device_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&filter=all&used=true")
          results = []
          
          # Look for devices with 'Usage' or 'Counter' related fields
          for dev in devices:
              usage = dev.get("Usage")
              counter_today = dev.get("CounterToday")
              
              if usage is not None or counter_today is not None:
                  results.append({
                      "idx": dev.get("idx"),
                      "Name": dev.get("Name"),
                      "CurrentUsage": usage,
                      "TodayTotal": counter_today,
                      "Type": dev.get("Type"),
                      "SubType": dev.get("SubType")
                  })
          
          return json.dumps({"status": "OK", "result": results})

• src/domoticz_mcp/server.py:1166-1167 (registration)

  The tool is registered with the MCP framework via the @mcp.tool() decorator on the analyze_energy_usage function.

  @mcp.tool()
  async def analyze_energy_usage() -> str:

• src/domoticz_mcp/server.py:1169-1188 (helper)

  Uses the _get_cached_data helper to fetch devices from the Domoticz API with caching support.

  async with create_client() as client:
      devices = await _get_cached_data(client, _device_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&filter=all&used=true")
      results = []
      
      # Look for devices with 'Usage' or 'Counter' related fields
      for dev in devices:
          usage = dev.get("Usage")
          counter_today = dev.get("CounterToday")
          
          if usage is not None or counter_today is not None:
              results.append({
                  "idx": dev.get("idx"),
                  "Name": dev.get("Name"),
                  "CurrentUsage": usage,
                  "TodayTotal": counter_today,
                  "Type": dev.get("Type"),
                  "SubType": dev.get("SubType")
              })
      
      return json.dumps({"status": "OK", "result": results})

• src/domoticz_mcp/server.py:329-336 (helper)

  The create_client helper used by analyze_energy_usage to get an authenticated HTTP client.

  def create_client(own_client: bool = False) -> DomoticzClient:
      """Create a DomoticzClient instance.
  
      Args:
          own_client: If True, creates a dedicated client that will be closed on exit.
                      If False (default), uses a shared client for connection pooling.
      """
      return DomoticzClient(own_client=own_client)

• src/domoticz_mcp/server.py:346-352 (helper)

  The _get_cached_data helper that handles caching and API requests for device data.

  async def _get_cached_data(client: "httpx.AsyncClient", cache_obj: Dict[str, Any], api_url: str, key_path: str = "result") -> List[Dict[str, Any]]:
      now = time.time()
      if cache_obj["data"] is None or (now - cache_obj["timestamp"]) > CACHE_TTL:
          response = await _do_request(client, "GET", api_url)
          cache_obj["data"] = response.json().get(key_path, [])
          cache_obj["timestamp"] = now
      return cache_obj["data"]