h3n3/nspanel-lovelace-ui
1
0
Fork 0
mirror of https://github.com/joBr99/nspanel-lovelace-ui.git synced 2026-03-01 02:20:12 +01:00
Code Issues Packages Projects Releases Wiki Activity

.

Browse Source
This commit is contained in:
joBr99
2023-11-26 13:02:18 +01:00
parent b38189d63e
commit ad891115a9
2 changed files with 264 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
nspanel-lovelace-ui/config.yaml
View File

@@ -1,6 +1,6 @@
# https://developers.home-assistant.io/docs/add-ons/configuration#add-on-config # https://developers.home-assistant.io/docs/add-ons/configuration#add-on-config
name: NSPanel Lovelace UI Addon name: NSPanel Lovelace UI Addon
version: "4.7.44" version: "4.7.46"
slug: nspanel-lovelace-ui slug: nspanel-lovelace-ui
description: NSPanel Lovelace UI Addon description: NSPanel Lovelace UI Addon
services: services:
@@ -17,15 +17,8 @@ init: false
map: map:
- addon_config:rw - addon_config:rw
options: options:
loglevel: loglevel: DEBUG
- DEBUG logtofile: true
- INFO
- WARNING
- ERROR
- CRITICAL
logtofile:
- true
- false
schema: schema:
loglevel: list(DEBUG|INFO|WARNING|ERROR|CRITICAL) loglevel: list(DEBUG|INFO|WARNING|ERROR|CRITICAL)
logtofile: bool logtofile: bool

522
nspanel-lovelace-ui/rootfs/usr/bin/mqtt-manager/libs/home_assistant.py
View File

@@ -1,261 +1,261 @@
import websocket import websocket
import ssl import ssl
import logging import logging
import json import json
from threading import Thread from threading import Thread
import time import time
import os import os
home_assistant_url = "" home_assistant_url = ""
home_assistant_token = "" home_assistant_token = ""
settings = {} settings = {}
auth_ok = False auth_ok = False
next_id = 0 next_id = 0
request_all_states_id = 0 request_all_states_id = 0
ws_connected = False ws_connected = False
home_assistant_entity_state_cache = {} home_assistant_entity_state_cache = {}
template_cache = {} template_cache = {}
response_buffer = {} response_buffer = {}
ON_CONNECT_HANDLER = None ON_CONNECT_HANDLER = None
ON_DISCONNECT_HANDLER = None ON_DISCONNECT_HANDLER = None
def init(settings_from_manager, on_ha_update_from_manager): def init(settings_from_manager, on_ha_update_from_manager):
global home_assistant_url, home_assistant_token, settings, on_ha_update global home_assistant_url, home_assistant_token, settings, on_ha_update
settings = settings_from_manager settings = settings_from_manager
on_ha_update = on_ha_update_from_manager on_ha_update = on_ha_update_from_manager
home_assistant_url = settings["home_assistant_address"] home_assistant_url = settings["home_assistant_address"]
home_assistant_token = settings["home_assistant_token"] home_assistant_token = settings["home_assistant_token"]
# Disable logging from underlying "websocket" # Disable logging from underlying "websocket"
logging.getLogger("websocket").propagate = False logging.getLogger("websocket").propagate = False
# Disable logging from underlying "websocket" # Disable logging from underlying "websocket"
logging.getLogger("websockets").propagate = False logging.getLogger("websockets").propagate = False
def register_on_connect_handler(handler): def register_on_connect_handler(handler):
global ON_CONNECT_HANDLER global ON_CONNECT_HANDLER
ON_CONNECT_HANDLER = handler ON_CONNECT_HANDLER = handler
def register_on_disconnect_handler(handler): def register_on_disconnect_handler(handler):
global ON_DISCONNECT_HANDLER global ON_DISCONNECT_HANDLER
ON_DISCONNECT_HANDLER = handler ON_DISCONNECT_HANDLER = handler
def on_message(ws, message): def on_message(ws, message):
global auth_ok, request_all_states_id, home_assistant_entity_state_cache, response_buffer, template_cache global auth_ok, request_all_states_id, home_assistant_entity_state_cache, response_buffer, template_cache
json_msg = json.loads(message) json_msg = json.loads(message)
if json_msg["type"] == "auth_required": if json_msg["type"] == "auth_required":
authenticate_client() authenticate_client()
elif json_msg["type"] == "auth_ok": elif json_msg["type"] == "auth_ok":
auth_ok = True auth_ok = True
logging.info("Home Assistant auth OK. Requesting existing states.") logging.info("Home Assistant auth OK. Requesting existing states.")
subscribe_to_events() subscribe_to_events()
_get_all_states() _get_all_states()
if ON_CONNECT_HANDLER is not None: if ON_CONNECT_HANDLER is not None:
ON_CONNECT_HANDLER() ON_CONNECT_HANDLER()
# for templates # for templates
elif json_msg["type"] == "event" and json_msg["id"] in response_buffer: elif json_msg["type"] == "event" and json_msg["id"] in response_buffer:
template_cache[response_buffer[json_msg["id"]]] = { template_cache[response_buffer[json_msg["id"]]] = {
"result": json_msg["event"]["result"], "result": json_msg["event"]["result"],
"listener-entities": json_msg["event"]["listeners"]["entities"] "listener-entities": json_msg["event"]["listeners"]["entities"]
} }
elif json_msg["type"] == "event" and json_msg["event"]["event_type"] == "state_changed": elif json_msg["type"] == "event" and json_msg["event"]["event_type"] == "state_changed":
entity_id = json_msg["event"]["data"]["entity_id"] entity_id = json_msg["event"]["data"]["entity_id"]
home_assistant_entity_state_cache[entity_id] = json_msg["event"]["data"]["new_state"] home_assistant_entity_state_cache[entity_id] = json_msg["event"]["data"]["new_state"]
send_entity_update(entity_id) send_entity_update(entity_id)
# rerender template # rerender template
for template, template_cache_entry in template_cache.items(): for template, template_cache_entry in template_cache.items():
if entity_id in template_cache_entry.get("listener-entities", []): if entity_id in template_cache_entry.get("listener-entities", []):
cache_template(template) cache_template(template)
elif json_msg["type"] == "result" and not json_msg["success"]: elif json_msg["type"] == "result" and not json_msg["success"]:
logging.error("Failed result: ") logging.error("Failed result: ")
logging.error(json_msg) logging.error(json_msg)
elif json_msg["type"] == "result" and json_msg["success"]: elif json_msg["type"] == "result" and json_msg["success"]:
if json_msg["id"] == request_all_states_id: if json_msg["id"] == request_all_states_id:
for entity in json_msg["result"]: for entity in json_msg["result"]:
home_assistant_entity_state_cache[entity["entity_id"]] = entity home_assistant_entity_state_cache[entity["entity_id"]] = entity
else: else:
if json_msg["id"] in response_buffer and json_msg.get("result"): if json_msg["id"] in response_buffer and json_msg.get("result"):
response_buffer[json_msg["id"]] = json_msg["result"] response_buffer[json_msg["id"]] = json_msg["result"]
return None # Ignore success result messages return None # Ignore success result messages
else: else:
logging.debug(message) logging.debug(message)
def _ws_connection_open(ws): def _ws_connection_open(ws):
global ws_connected global ws_connected
ws_connected = True ws_connected = True
logging.info("WebSocket connection to Home Assistant opened.") logging.info("WebSocket connection to Home Assistant opened.")
if ON_CONNECT_HANDLER is not None: if ON_CONNECT_HANDLER is not None:
ON_CONNECT_HANDLER() ON_CONNECT_HANDLER()
def _ws_connection_close(ws, close_status_code, close_msg): def _ws_connection_close(ws, close_status_code, close_msg):
global ws_connected global ws_connected
ws_connected = False ws_connected = False
logging.error("WebSocket connection closed!") logging.error("WebSocket connection closed!")
if ON_DISCONNECT_HANDLER is not None: if ON_DISCONNECT_HANDLER is not None:
ON_DISCONNECT_HANDLER() ON_DISCONNECT_HANDLER()
def connect(): def connect():
Thread(target=_do_connection, daemon=True).start() Thread(target=_do_connection, daemon=True).start()
def _do_connection(): def _do_connection():
global home_assistant_url, ws, settings global home_assistant_url, ws, settings
ws_url = home_assistant_url.replace( ws_url = home_assistant_url.replace(
"https://", "wss://").replace("http://", "ws://") "https://", "wss://").replace("http://", "ws://")
if settings["is_addon"]: if settings["is_addon"]:
ws_url += "/core/websocket" ws_url += "/core/websocket"
else: else:
ws_url += "/api/websocket" ws_url += "/api/websocket"
ws = websocket.WebSocketApp(F"{ws_url}", on_message=on_message, ws = websocket.WebSocketApp(F"{ws_url}", on_message=on_message,
on_open=_ws_connection_open, on_close=_ws_connection_close) on_open=_ws_connection_open, on_close=_ws_connection_close)
while True: while True:
logging.info(F"Connecting to Home Assistant at {ws_url}") logging.info(F"Connecting to Home Assistant at {ws_url}")
ws.close() ws.close()
time.sleep(1) time.sleep(1)
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
time.sleep(10) time.sleep(10)
def authenticate_client(): def authenticate_client():
global home_assistant_token global home_assistant_token
logging.info("Sending auth to Home Assistant") logging.info("Sending auth to Home Assistant")
msg = { msg = {
"type": "auth", "type": "auth",
"access_token": home_assistant_token "access_token": home_assistant_token
} }
send_message(json.dumps(msg)) send_message(json.dumps(msg))
def subscribe_to_events(): def subscribe_to_events():
global next_id global next_id
msg = { msg = {
"id": next_id, "id": next_id,
"type": "subscribe_events", "type": "subscribe_events",
"event_type": "state_changed" "event_type": "state_changed"
} }
send_message(json.dumps(msg)) send_message(json.dumps(msg))
def _get_all_states(): def _get_all_states():
global next_id, request_all_states_id global next_id, request_all_states_id
msg = { msg = {
"id": next_id, "id": next_id,
"type": "get_states", "type": "get_states",
} }
request_all_states_id = next_id request_all_states_id = next_id
send_message(json.dumps(msg)) send_message(json.dumps(msg))
# Got new value from Home Assistant, send update to callback method # Got new value from Home Assistant, send update to callback method
def send_entity_update(entity_id): def send_entity_update(entity_id):
global on_ha_update global on_ha_update
on_ha_update(entity_id) on_ha_update(entity_id)
def call_service(entity_name: str, domain: str, service: str, service_data: dict) -> bool: def call_service(entity_name: str, domain: str, service: str, service_data: dict) -> bool:
global next_id global next_id
try: try:
msg = { msg = {
"id": next_id, "id": next_id,
"type": "call_service", "type": "call_service",
"domain": domain, "domain": domain,
"service": service, "service": service,
"service_data": service_data, "service_data": service_data,
"target": { "target": {
"entity_id": entity_name "entity_id": entity_name
}, },
} }
send_message(json.dumps(msg)) send_message(json.dumps(msg))
return True return True
except Exception as e: except Exception as e:
logging.exception("Failed to call Home Assisatant service.") logging.exception("Failed to call Home Assisatant service.")
return False return False
def execute_script(entity_name: str, domain: str, service: str, service_data: dict) -> str: def execute_script(entity_name: str, domain: str, service: str, service_data: dict) -> str:
global next_id, response_buffer global next_id, response_buffer
try: try:
call_id = next_id call_id = next_id
# request answer for this call # request answer for this call
response_buffer[call_id] = True response_buffer[call_id] = True
msg = { msg = {
"id": call_id, "id": call_id,
"type": "execute_script", "type": "execute_script",
"sequence": [ "sequence": [
{ {
"service": f"{domain}.{service}", "service": f"{domain}.{service}",
"data": service_data, "data": service_data,
"target": { "target": {
"entity_id": [entity_name] "entity_id": [entity_name]
}, },
"response_variable": "service_result" "response_variable": "service_result"
}, },
{ {
"stop": "done", "stop": "done",
"response_variable": "service_result" "response_variable": "service_result"
} }
] ]
} }
send_message(json.dumps(msg)) send_message(json.dumps(msg))
# busy waiting for response with a timeout of 0.2 seconds - maybe there's a better way of doing this # busy waiting for response with a timeout of 0.2 seconds - maybe there's a better way of doing this
mustend = time.time() + 0.4 mustend = time.time() + 0.4
while time.time() < mustend: while time.time() < mustend:
if response_buffer[call_id] == True: if response_buffer[call_id] == True:
#print(f'loooooooooop {time.time()}') #print(f'loooooooooop {time.time()}')
time.sleep(0.0001) time.sleep(0.0001)
else: else:
return response_buffer[call_id]["response"] return response_buffer[call_id]["response"]
raise TimeoutError("Did not recive respose in time to HA script call") raise TimeoutError("Did not recive respose in time to HA script call")
except Exception as e: except Exception as e:
logging.exception("Failed to call Home Assisatant script.") logging.exception("Failed to call Home Assisatant script.")
return False return {}
def cache_template(template): def cache_template(template):
global next_id, response_buffer global next_id, response_buffer
try: try:
call_id = next_id call_id = next_id
response_buffer[call_id] = template response_buffer[call_id] = template
msg = { msg = {
"id": call_id, "id": call_id,
"type": "render_template", "type": "render_template",
"template": template "template": template
} }
send_message(json.dumps(msg)) send_message(json.dumps(msg))
return True return True
except Exception as e: except Exception as e:
logging.exception("Failed to render template.") logging.exception("Failed to render template.")
return False return False
def get_template(template): def get_template(template):
global template_cache global template_cache
if template in template_cache: if template in template_cache:
return template_cache[template].get("result") return template_cache[template].get("result")
else: else:
mustend = time.time() + 0.5 mustend = time.time() + 0.5
while time.time() < mustend: while time.time() < mustend:
if template not in template_cache: if template not in template_cache:
time.sleep(0.0001) time.sleep(0.0001)
else: else:
return template_cache.get(template, []).get("result", "404") return template_cache.get(template, []).get("result", "404")
def get_entity_data(entity_id: str): def get_entity_data(entity_id: str):
if entity_id in home_assistant_entity_state_cache: if entity_id in home_assistant_entity_state_cache:
return home_assistant_entity_state_cache[entity_id] return home_assistant_entity_state_cache[entity_id]
else: else:
return None return None
def is_existent(entity_id: str): def is_existent(entity_id: str):
if entity_id in home_assistant_entity_state_cache: if entity_id in home_assistant_entity_state_cache:
return True return True
else: else:
return False return False
def send_message(message): def send_message(message):
global ws, next_id global ws, next_id
next_id += 1 next_id += 1
ws.send(message) ws.send(message)