245 lines
8.2 KiB
Python
245 lines
8.2 KiB
Python
|
|
""" View and control devices of your HomeAssistant.io instance.
|
|
|
|
Synopsis: <trigger> <entity filter>"""
|
|
|
|
import os
|
|
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
|
|
configurationFileName = "homeassistant_config.json"
|
|
configuration_directory = os.path.join(configLocation())
|
|
configuration_file = os.path.join(configuration_directory, configurationFileName)
|
|
config = {}
|
|
#config["sort_order"] = [["light", "switch"], ["scene", "group"], ["automation"]]
|
|
|
|
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"] # types where toggle doesn't work
|
|
#action_words = ["on", "off", "open", "close"]
|
|
|
|
|
|
def initialize():
|
|
global config
|
|
|
|
# load HASS config
|
|
if os.path.exists(configuration_file):
|
|
with open(configuration_file) as json_config:
|
|
config = 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)
|
|
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"])
|
|
|
|
# Trim hass URL
|
|
config["hass_url"] = config["hass_url"].strip("/")
|
|
|
|
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 a query to control your Home Assistant",
|
|
actions=[UrlAction("Open in Browser", config["hass_url"])])
|
|
|
|
action_word = query.string.split()[0].lower().strip()
|
|
#is_action_word = action_word in action_words
|
|
entity_query_list = query.string.split()#[1:] if is_action_word else query.string.split()
|
|
|
|
if not entity_query_list:
|
|
return Item(id=__title__,
|
|
icon=os.path.dirname(__file__) + "/" + icon_files["logo"],
|
|
text="Enter the name of the entity which you want to control",
|
|
subtext="Open Home Assistant in Browser",
|
|
actions=[UrlAction("Open in Browser", config["hass_url"])])
|
|
|
|
|
|
# Set up HASS state query
|
|
state_query = config["hass_url"] + "/api/states"
|
|
headers = {
|
|
"Authorization": "Bearer " + config["hass_key"],
|
|
"content-type": "application/json",
|
|
}
|
|
|
|
try:
|
|
response = requests.get(state_query, headers=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
|
|
#entities = sorted(response.json(), key=lambda s: s.lower())
|
|
entities = response.json()
|
|
|
|
# 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"]},
|
|
"hass_key": config["hass_key"],
|
|
"headers": headers,
|
|
}
|
|
state = entity["state"]
|
|
state_colored = "<font color=\"{}\">{}</font>".format(
|
|
"Green" if entity["state"] == "on" else
|
|
"Red" if entity["state"] == "off" else "",
|
|
entity["state"].capitalize()
|
|
)
|
|
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())
|
|
)
|
|
|
|
if entity_class in toggle_types:
|
|
item.addAction(FuncAction("Toggle", lambda d=data: sendCommand(d, "homeassistant/toggle")))
|
|
|
|
on_or_off = "on" if state != "on" else "off"
|
|
if entity_class in on_off_types:
|
|
item.addAction(FuncAction("Turn %s" % (on_or_off), lambda d=data: sendCommand(d, "homeassistant/turn_%s" % (on_or_off))))
|
|
|
|
#if entity_class in open_close_types:
|
|
#item.addAction(FuncAction("Open Cover", lambda d=data: sendCommand(d, "cover/open_cover")))
|
|
|
|
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 specify 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=data["headers"],
|
|
)
|
|
response.raise_for_status()
|
|
except requests.exceptions.RequestException as error:
|
|
warning("Error while sending command to Home Assistant:\n%s" % (str(error)))
|
|
|