Adding home assistant filter than can return time in EST (yeah i'll make that better in next version to use local timezon) and control light entities based on the easy names. Currently can only control one entity at a time, so recommend making light groups.

examples/filters/home_assistant_filter.py

@@ -0,0 +1,116 @@
+"""
+title: HomeAssistant Filter Pipeline
+author: open-webui
+date: 2024-06-15
+version: 1.0
+license: MIT
+description: A pipeline for controlling Home Assistant entities based on their easy names. Only supports lights at the moment.
+requirements: pytz, difflab
+"""
+import requests
+from typing import Literal, Dict, Any
+from datetime import datetime
+import pytz
+from difflib import get_close_matches
+
+from blueprints.function_calling_blueprint import Pipeline as FunctionCallingBlueprint
+
+class Pipeline(FunctionCallingBlueprint):
+    class Valves(FunctionCallingBlueprint.Valves):
+        HOME_ASSISTANT_URL: str = ""
+        HOME_ASSISTANT_TOKEN: str = ""
+
+    class Tools:
+        def __init__(self, pipeline) -> None:
+            self.pipeline = pipeline
+
+        def get_current_time(self) -> str:
+            """
+            Get the current time in EST.
+
+            :return: The current time in EST.
+            """
+            now_est = datetime.now(pytz.timezone('US/Eastern'))  # Get the current time in EST
+            current_time = now_est.strftime("%I:%M %p")  # %I for 12-hour clock, %M for minutes, %p for am/pm
+            return f"ONLY RESPOND 'Current time is {current_time}'"
+
+        def get_all_lights(self) -> Dict[str, Any]:
+            """
+            Lists my lights.
+            Shows me my lights.
+            Get a dictionary of all lights in my home.
+
+            :return: A dictionary of light entity names and their IDs.
+            """
+            if not self.pipeline.valves.HOME_ASSISTANT_URL or not self.pipeline.valves.HOME_ASSISTANT_TOKEN:
+                return {"error": "Home Assistant URL or token not set, ask the user to set it up."}
+            else:
+                url = f"{self.pipeline.valves.HOME_ASSISTANT_URL}/api/states"
+                headers = {
+                    "Authorization": f"Bearer {self.pipeline.valves.HOME_ASSISTANT_TOKEN}",
+                    "Content-Type": "application/json",
+                }
+
+                response = requests.get(url, headers=headers)
+                response.raise_for_status()  # Raises an HTTPError for bad responses
+                data = response.json()
+
+                lights = {entity["attributes"]["friendly_name"]: entity["entity_id"]
+                          for entity in data if entity["entity_id"].startswith("light.")}
+
+                return lights
+
+        def control_light(self, name: str, state: Literal['on', 'off']) -> str:
+            """
+            Turn a light on or off based on its name.
+
+            :param name: The friendly name of the light.
+            :param state: The desired state ('on' or 'off').
+            :return: The result of the operation.
+            """
+            if not self.pipeline.valves.HOME_ASSISTANT_URL or not self.pipeline.valves.HOME_ASSISTANT_TOKEN:
+                return "Home Assistant URL or token not set, ask the user to set it up."
+
+            # Normalize the light name by converting to lowercase and stripping extra spaces
+            normalized_name = " ".join(name.lower().split())
+
+            # Get a dictionary of all lights
+            lights = self.get_all_lights()
+            if "error" in lights:
+                return lights["error"]
+
+            # Find the closest matching light name
+            light_names = list(lights.keys())
+            closest_matches = get_close_matches(normalized_name, light_names, n=1, cutoff=0.6)
+
+            if not closest_matches:
+                return f"Light named '{name}' not found."
+
+            best_match = closest_matches[0]
+            light_id = lights[best_match]
+
+            url = f"{self.pipeline.valves.HOME_ASSISTANT_URL}/api/services/light/turn_{state}"
+            headers = {
+                "Authorization": f"Bearer {self.pipeline.valves.HOME_ASSISTANT_TOKEN}",
+                "Content-Type": "application/json",
+            }
+            payload = {
+                "entity_id": light_id
+            }
+
+            response = requests.post(url, headers=headers, json=payload)
+            if response.status_code == 200:
+                return f"ONLY RESPOND 'Will do' TO THE USER. DO NOT SAY ANYTHING ELSE!"
+            else:
+                return f"ONLY RESPOND 'Couldn't find light' TO THE USER. DO NOT SAY ANYTHING ELSE!"
+
+    def __init__(self):
+        super().__init__()
+        self.name = "My Tools Pipeline"
+        self.valves = self.Valves(
+            **{
+                **self.valves.model_dump(),
+                "pipelines": ["*"],  # Connect to all pipelines
+            },
+        )
+        self.tools = self.Tools(self)