get_unusual_power_consumption
Identify unusual power consumption patterns in Tesla vehicles by analyzing data from the TeslaMate database to optimize energy usage and detect anomalies.
Instructions
Get the unusual power consumption for each car.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/tools.py:102-106 (registration)Registers the 'get_unusual_power_consumption' tool by defining its name, description, and the SQL file containing the query logic.ToolDefinition( name="get_unusual_power_consumption", description="Get the unusual power consumption for each car. Identifies anomalies in power usage that might indicate issues.", sql_file="unusual_power_consumption.sql", ),
- main.py:22-28 (handler)Generic handler factory that creates the execution function for the tool by calling DatabaseManager.execute_query_sync with the tool's SQL file. Used for STDIO transport server.def create_tool_handler(sql_file: str): """Factory function to create tool handlers""" def handler() -> List[Dict[str, Any]]: return db_manager.execute_query_sync(sql_file) return handler
- main.py:32-38 (registration)Dynamically creates and registers the handler function for 'get_unusual_power_consumption' (and all tools) with the FastMCP server using the tool decorator.for tool_def in TOOL_DEFINITIONS: tool_func = create_tool_handler(tool_def.sql_file) tool_func.__doc__ = tool_def.description tool_func.__name__ = tool_def.name # Register the tool with the MCP server mcp.tool()(tool_func)
- main_remote.py:118-127 (handler)Handler function for executing predefined tools including 'get_unusual_power_consumption' asynchronously in the HTTP transport server.async def execute_predefined_tool(tool_name: str) -> List[Dict[str, Any]]: """Execute a predefined tool by name""" if not app_context: raise RuntimeError("Application context not initialized") tool = get_tool_by_name(tool_name) return await app_context.db_manager.execute_query_async( tool.sql_file, app_context.db_pool )
- src/database.py:27-34 (helper)Core helper method that reads the SQL file and executes the query on the TeslaMate database, returning results as list of dicts.def execute_query_sync(self, sql_file_path: str) -> List[Dict[str, Any]]: """Execute SQL query synchronously""" sql_query = self.read_sql_file(sql_file_path) with psycopg.connect(self.connection_string, row_factory=dict_row) as conn: with conn.cursor() as cur: cur.execute(sql_query) return cur.fetchall()