106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
import collections
|
|
from html import entities
|
|
from homeassistant_api import Client, HomeassistantAPIError
|
|
|
|
URL = "https://home.kvanewijk.nl:42728/api"
|
|
TOKEN = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiI0YmQ4ZDEzY2JkYTI0NTdhYWI5YWIzMWU0MTE2Y2FhMCIsImlhdCI6MTY0NzgwMjgxMCwiZXhwIjoxOTYzMTYyODEwfQ.bwXhX7au5MKtHxl-Trj7tLijjATvRaA45iNhTmL0NyM"
|
|
|
|
class HomeAssistant(object):
|
|
|
|
instance = None
|
|
|
|
def __init__(self, url, token) -> None:
|
|
self.url = url
|
|
self.token = token
|
|
self.Client = Client(url, token)
|
|
|
|
@classmethod
|
|
def initialize(cls, url, token):
|
|
cls.instance = HomeAssistant(url, token)
|
|
|
|
@classmethod
|
|
def getInstance(cls, url=None, token=None):
|
|
if cls.instance is None:
|
|
cls.instance = HomeAssistant(url, token)
|
|
|
|
return cls.instance
|
|
|
|
@classmethod
|
|
def testConnection(cls, url:str, token:str) -> str:
|
|
response = None
|
|
|
|
try:
|
|
response = Client(url, token).check_api_running()
|
|
except (HomeassistantAPIError, Exception) as e:
|
|
print("Error while connecting: {}".format(str(e)))
|
|
|
|
return response
|
|
|
|
def get_domains(self):
|
|
return self.Client.get_domains()
|
|
|
|
def get_domain(self, domain):
|
|
return self.Client.get_domain(domain)
|
|
|
|
def get_services(self, domain):
|
|
domain = self.Client.get_domain(domain)
|
|
od = collections.OrderedDict(sorted(domain.services.items()))
|
|
services = list(od.values())
|
|
return services
|
|
|
|
def get_service(self, domain, service):
|
|
services = self.get_domain(domain).services
|
|
if service in services:
|
|
return services[service]
|
|
|
|
raise KeyError("Service `{}` does not exist in domain `{}`".format(service, domain))
|
|
|
|
def get_entities(self, domain):
|
|
groups = self.Client.get_entities()
|
|
entities = []
|
|
|
|
if domain in groups:
|
|
entities = [groups[domain].group_id + "." + e.slug for e in list(groups[domain].entities.values())]
|
|
else:
|
|
for group in groups.values():
|
|
entities.extend([group.group_id + "." + e.slug for e in list(group.entities.values())])
|
|
|
|
entities.sort()
|
|
|
|
return entities
|
|
|
|
def get_entity(self, entity):
|
|
return self.Client.get_entity(entity_id=entity)
|
|
|
|
def domain_has_entities(self, domain):
|
|
groups = self.Client.get_entities()
|
|
|
|
if domain in groups:
|
|
if len(groups[domain].entities) > 0:
|
|
return True
|
|
|
|
return False
|
|
|
|
def get_defaults(self):
|
|
domains = self.get_domains()
|
|
domain = list(domains.values())[0].domain_id
|
|
services = self.get_services(domain)
|
|
service = services[0].service_id
|
|
entities = self.get_entities(domain)
|
|
entity = entities[0]
|
|
|
|
return domain, service, entity
|
|
|
|
def trigger_service(self, domain, service, entity, data):
|
|
service_data = {}
|
|
|
|
if entity:
|
|
service_data['entity_id'] = entity
|
|
|
|
if data:
|
|
for line in data.split("\n"):
|
|
keypair = line.split(":")
|
|
service_data[keypair[0].strip()] = keypair[1].strip()
|
|
|
|
states = self.Client.trigger_service(domain, service, **service_data)
|