Coverage for src/data_api/mcp/tools/energy.py: 100%

1"""Energy usage MCP tool.""" 

2 

3from data_api.mcp.errors import live_tool_errors 

4from data_api.mcp.server import mcp 

5from data_api.mcp.tools.result_types import EnergyResult 

6from data_api.runtime import get_runtime 

7 

8 

9async def _get_energy_usage() -> EnergyResult: 

10 """ 

11 Get real-time electricity usage for campus. 

12 

13 Returns: 

14 Dictionary with current electricity usage by zone. 

15 """ 

16 with live_tool_errors(): 

17 data = await get_runtime().energy.get_realtime_electricity_usage() 

18 zones = [] 

19 for item in data: 

20 capacity = item.get("capacity") or 1 # Avoid division by zero 

21 usage = item.get("data", 0) 

22 zones.append( 

23 { 

24 "name": item.get("name"), 

25 "usage_kw": usage, 

26 "capacity_kw": item.get("capacity"), 

27 "usage_percent": round(usage / capacity * 100, 1) if capacity > 0 else 0, 

28 "last_updated": item.get("last_updated"), 

29 } 

30 ) 

31 return {"zones": zones} 

32 

33 

34@mcp.tool( 

35 name="get_energy_usage", 

36 title="查詢校園即時用電", 

37 description="Get real-time campus electricity usage. " 

38 "Use this to check current power consumption on campus.", 

39) 

40async def get_energy_usage() -> dict: 

41 """Get real-time electricity usage for campus.""" 

42 return await _get_energy_usage()