52 lines
1.7 KiB
Python
52 lines
1.7 KiB
Python
import json
|
|
from typing import Any
|
|
|
|
import requests
|
|
from aiohttp import ClientSession
|
|
from hass_client import HomeAssistantClient
|
|
|
|
|
|
class HomeAssistant:
|
|
|
|
def __init__(self, config: dict) -> None:
|
|
self.config = config
|
|
self.ha_config = self.call_api("config")
|
|
|
|
def call_api(self, endpoint, payload=None):
|
|
"""
|
|
Call the REST API
|
|
"""
|
|
base_url = self.config["homeassistant"]["url"]
|
|
headers = {
|
|
"Authorization": f"Bearer {self.config['homeassistant']['token']}",
|
|
"content-type": "application/json",
|
|
}
|
|
if payload is None:
|
|
response = requests.get(f"{base_url}/api/{endpoint}", headers=headers)
|
|
else:
|
|
response = requests.post(f"{base_url}/api/{endpoint}", headers=headers, json=payload)
|
|
if response.status_code == 200:
|
|
try:
|
|
return response.json()
|
|
except:
|
|
pass
|
|
try:
|
|
return json.loads(response.text.replace("'", '"').replace("None", "null"))
|
|
except:
|
|
return response.text
|
|
else:
|
|
return {"status": "error", "message": response.text}
|
|
|
|
async def send_command(self, command: str, **kwargs: dict[str, Any]):
|
|
"""
|
|
Send command using the WebSocket API
|
|
"""
|
|
async with ClientSession() as session:
|
|
async with HomeAssistantClient(
|
|
self.config["homeassistant"]["url"] + "/api/websocket",
|
|
self.config["homeassistant"]["token"],
|
|
session,
|
|
) as client:
|
|
response = await client.send_command(command, **kwargs)
|
|
return response
|