""" View and control devices of your HomeAssistant.io instance. Synopsis: """ import os import re import json import requests from albert import Item, ClipAction, FuncAction, UrlAction, configLocation from albert import critical, warning, debug __title__ = "Home Assistant" __version__ = "0.0.1" __triggers__ = "ha " __authors__ = "hoellen" __py_deps__ =["requests"] # global variables config = {} configurationFileName = "homeassistant_config.json" configuration_directory = os.path.join(configLocation()) configuration_file = os.path.join(configuration_directory, configurationFileName) config["sort_order"] = {"light": 1, "switch": 1, "scene": 2, "group": 2, "automation": 3} icon_files = { "logo": "icons/icon.png", "automation": "icons/automation.png", "cover": "icons/cover.png", "group": "icons/group.png", "light": "icons/light.png", "scene": "icons/scene.png", "switch": "icons/switch.png", } toggle_types = ["light", "switch", "automation", "group", "input_boolean", "climate", "camera"] on_off_types = ["scene", "media_player"] # use turn_{on,off} instead of toggle def initialize(): global config # load HASS config if os.path.exists(configuration_file): with open(configuration_file) as json_config: config.update(json.load(json_config)) else: config["hass_url"] = "http://192.168.1.5:8123" config["hass_key"] = "" try: os.makedirs(configuration_directory, exist_ok=True) try: with open(configuration_file, "w") as output_file: json.dump(config, output_file, indent=4) except OSError: critical("There was an error opening the file: %s" % configuration_file) except OSError: critical("There was an error making the directory: %s" % configuration_directory) # Set up HASS state query config["hass_url"] = config["hass_url"].strip("/") config["state_query"] = config["hass_url"] + "/api/states" config["headers"]= { "Authorization": "Bearer " + config["hass_key"], "content-type": "application/json", } # Set up sort_order config["sort_order"]["."] = 999 # Wildcard for sorting config["sort_order"] = RegexDict(config["sort_order"]) debug("Loaded config: " + str(config)) # check config if not config["hass_url"]: critical("Invalid Home Assistant URL") # Trim hass URL config["hass_url"] = config["hass_url"].strip("/") if not config["hass_key"]: critical("Empty Home Assistant API Key") def handleQuery(query): global config if query.isTriggered: query.disableSort() if not query.isValid: return if "hass_url" not in config or not config["hass_key"]: return Item(id=__title__, icon=os.path.dirname(__file__) + "/" + icon_files["logo"], text="Invalid Home Assistant URL", subtext=config["hass_url"]) if "hass_key" not in config or not config["hass_key"]: return Item(id=__title__, icon=os.path.dirname(__file__) + "/" + icon_files["logo"], text="Empty Home Assistant API Key", subtext="Please add you Home Assistant API Key to your config." ) return showEntities(query) def showEntities(query): results = [] # empty query if not query.string.strip(): return Item(id=__title__, icon=os.path.dirname(__file__) + "/" + icon_files["logo"], text=__title__, subtext="Enter the name of the entity which you want to control", actions=[UrlAction("Open in Browser", config["hass_url"])]) entity_query_list = query.string.split() # query entities from HASS try: response = requests.get(config["state_query"], headers=config["headers"]) response.raise_for_status() except requests.exceptions.RequestException as error: return Item(id=__title__, icon=os.path.dirname(__file__) + "/" + icon_files["logo"], text="Error while getting entity states from Home Assistant", subtext=str(error)) # Sort entries by class entities = sorted(response.json(), key=lambda e: config["sort_order"].get_matching(e["entity_id"].split(".")[0])) # Sort entries by query matching entities = sorted(entities, key=lambda e: RegexDict({query.string.lower() + ".": 0, ".": 1}) .get_matching(e["attributes"]["friendly_name"].lower() if "friendly_name" in e["attributes"] else e["entity_id"])) # Parse all entities and states for entity in entities: # number of items if len(results) > 10: break # Not likely, but worth checking if "entity_id" not in entity or "attributes" not in entity: continue # Use entity_id if friendly_name is not given if "friendly_name" not in entity["attributes"]: entity["attributes"]["friendly_name"] = entity["entity_id"] entity_class = entity["entity_id"].split(".")[0] # Don't add this item if the query doesn't appear in either friendly_name or id entity_appears_in_search = True for entity_query_item in entity_query_list: if ( entity_query_item.lower() not in entity["attributes"]["friendly_name"].lower() and entity_query_item.lower() not in entity["entity_id"] ): entity_appears_in_search = False if not entity_appears_in_search: debug("Exclude from search %s" % entity["entity_id"]) continue entity_icon = os.path.dirname(__file__) + "/" + ( icon_files[entity_class] if entity_class in icon_files else icon_files["logo"] ) data = { "endpoint": "{}/api/services/".format(config["hass_url"]), "service_data": {"entity_id": entity["entity_id"]}, } state = entity["state"] state_colored = "{}".format( "Green" if entity["state"] == "on" else "Red" if entity["state"] == "off" else "", entity["state"].capitalize() ) # build item for list item = Item(id=__title__, icon=entity_icon, text=entity["attributes"]["friendly_name"], completion="%s%s" % (__triggers__, entity["attributes"]["friendly_name"]), subtext="%s  |  %s" % (state_colored, entity_class.capitalize()) ) # add actions depending on class if entity_class in toggle_types: item.addAction(FuncAction("Toggle", lambda d=data: sendCommand(d, "homeassistant/toggle"))) if entity_class in on_off_types: on_or_off = "on" if state != "on" else "off" item.addAction(FuncAction("Turn %s" % (on_or_off), lambda d=data: sendCommand(d, "homeassistant/turn_%s" % (on_or_off)))) if entity_class == "cover": open_or_close = "open" if state != "open" else "close" item.addAction(FuncAction("%s Cover" % (open_or_close.capitalize()), lambda d=data: sendCommand(d, "homeassistant/cover/%s_cover") % (open_or_close))) item.addAction(ClipAction("Copy ID", entity["entity_id"])) results.append(item) # no entity found if len(results) == 0: results.append( Item(id=__title__, icon=os.path.dirname(__file__) + "/" + icon_files["logo"], text="Entity not found!", subtext="Please search for another entity." ) ) return results def sendCommand(data, service): data["endpoint"] += service debug(__title__ + ": Sending command \"" + service + "\" to " + data["service_data"]["entity_id"]) # Make POST request to HA service try: response = requests.post( data["endpoint"], data=json.dumps(data["service_data"]), headers=config["headers"], ) response.raise_for_status() except requests.exceptions.RequestException as error: warning("Error while sending command to Home Assistant:\n%s" % (str(error))) # New dictionary class to match keys with regex class RegexDict(dict): def get_matching(self, event): return next(self[key] for key in self if re.match(key, event))