analyze_energy_usage

Instructions

Implementation Reference

• src/domoticz_mcp/server.py:1111-1133 (handler)

  The actual tool handler function implementation: fetches all devices, filters for those with 'Usage' or 'CounterToday' fields, and returns a JSON summary of energy usage per device.

  @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:1111-1111 (registration)

  The @mcp.tool() decorator registers 'analyze_energy_usage' as an MCP tool.

  @mcp.tool()

• src/domoticz_mcp/server.py:373-376 (helper)

  Helper function _resolve_device_idx and _get_cached_data used by the handler to fetch and cache device data.

  async def _resolve_device_idx(client: "httpx.AsyncClient", idx: Optional[int] = None, name: Optional[str] = None) -> Optional[int]:
      """Resolve a device to its idx."""
      return await _resolve_idx(client, idx, name, _device_cache, f"{DOMOTICZ_API_URL}?type=command¶m=getdevices&filter=all&used=true")

• src/domoticz_mcp/server.py:387-394 (helper)

  Helper function _simplify_device used to reduce device dictionaries (not directly used by analyze_energy_usage but part of the device processing patterns).

  def _simplify_device(dev: Dict[str, Any]) -> Dict[str, Any]:
      """Reduce device dictionary to essential fields to save context space."""
      keys_to_keep = [
          "idx", "Name", "Type", "SubType", "Data", "Status", 
          "BatteryLevel", "Favorite", "HardwareName", "LastUpdate", 
          "TypeImg", "Usage", "CounterToday", "Temp", "Humidity"
      ]
      return {k: dev[k] for k in keys_to_keep if k in dev}

• tests/test_server.py:245-246 (helper)

  Test for the analyze_energy_usage tool, verifying it filters devices without energy data and returns only those with Usage/CounterToday fields.

  @respx.mock
  async def test_analyze_energy_usage():