- Add complete homeassistant skill source to skills/ directory - Includes all scripts, references, and automation templates - Matches format of other skills in repository
173 lines
6.1 KiB
Python
Executable File
173 lines
6.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Home Assistant REST API Client
|
|
|
|
This script provides utilities to interact with Home Assistant via the REST API.
|
|
Supports getting states, calling services, and querying configuration.
|
|
|
|
Usage:
|
|
# Get all entity states
|
|
python3 api_client.py --url http://homeassistant.local:8123 --token YOUR_TOKEN get-states
|
|
|
|
# Get specific entity state
|
|
python3 api_client.py --url http://homeassistant.local:8123 --token YOUR_TOKEN get-state light.living_room
|
|
|
|
# Call a service
|
|
python3 api_client.py --url http://homeassistant.local:8123 --token YOUR_TOKEN call-service light turn_on --entity light.living_room --data '{"brightness": 255}'
|
|
|
|
# List all services
|
|
python3 api_client.py --url http://homeassistant.local:8123 --token YOUR_TOKEN list-services
|
|
|
|
Environment variables:
|
|
HA_URL: Home Assistant URL (e.g., http://homeassistant.local:8123)
|
|
HA_TOKEN: Long-lived access token
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Any, Dict, Optional
|
|
import urllib.request
|
|
import urllib.error
|
|
import urllib.parse
|
|
|
|
|
|
class HomeAssistantClient:
|
|
"""Client for interacting with Home Assistant REST API."""
|
|
|
|
def __init__(self, url: str, token: str):
|
|
"""Initialize the client with URL and access token."""
|
|
self.url = url.rstrip('/')
|
|
self.token = token
|
|
self.headers = {
|
|
'Authorization': f'Bearer {token}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
def _request(self, endpoint: str, method: str = 'GET', data: Optional[Dict] = None) -> Any:
|
|
"""Make a request to the Home Assistant API."""
|
|
url = f"{self.url}/api/{endpoint}"
|
|
|
|
req = urllib.request.Request(url, headers=self.headers, method=method)
|
|
|
|
if data is not None:
|
|
req.data = json.dumps(data).encode('utf-8')
|
|
|
|
try:
|
|
with urllib.request.urlopen(req) as response:
|
|
return json.loads(response.read().decode('utf-8'))
|
|
except urllib.error.HTTPError as e:
|
|
error_body = e.read().decode('utf-8')
|
|
print(f"HTTP Error {e.code}: {e.reason}", file=sys.stderr)
|
|
print(f"Response: {error_body}", file=sys.stderr)
|
|
sys.exit(1)
|
|
except urllib.error.URLError as e:
|
|
print(f"URL Error: {e.reason}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
def get_states(self) -> list:
|
|
"""Get all entity states."""
|
|
return self._request('states')
|
|
|
|
def get_state(self, entity_id: str) -> Dict:
|
|
"""Get state of a specific entity."""
|
|
return self._request(f'states/{entity_id}')
|
|
|
|
def get_services(self) -> Dict:
|
|
"""Get all available services."""
|
|
return self._request('services')
|
|
|
|
def get_config(self) -> Dict:
|
|
"""Get Home Assistant configuration."""
|
|
return self._request('config')
|
|
|
|
def call_service(self, domain: str, service: str, entity_id: Optional[str] = None,
|
|
service_data: Optional[Dict] = None) -> list:
|
|
"""Call a service."""
|
|
data = {}
|
|
if entity_id:
|
|
data['entity_id'] = entity_id
|
|
if service_data:
|
|
data.update(service_data)
|
|
|
|
return self._request(f'services/{domain}/{service}', method='POST', data=data if data else None)
|
|
|
|
def get_error_log(self) -> str:
|
|
"""Get the error log."""
|
|
url = f"{self.url}/api/error_log"
|
|
req = urllib.request.Request(url, headers=self.headers)
|
|
with urllib.request.urlopen(req) as response:
|
|
return response.read().decode('utf-8')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='Home Assistant REST API Client')
|
|
parser.add_argument('--url', help='Home Assistant URL',
|
|
default=os.getenv('HA_URL'))
|
|
parser.add_argument('--token', help='Long-lived access token',
|
|
default=os.getenv('HA_TOKEN'))
|
|
|
|
subparsers = parser.add_subparsers(dest='command', help='Command to execute')
|
|
|
|
# get-states command
|
|
subparsers.add_parser('get-states', help='Get all entity states')
|
|
|
|
# get-state command
|
|
state_parser = subparsers.add_parser('get-state', help='Get specific entity state')
|
|
state_parser.add_argument('entity_id', help='Entity ID')
|
|
|
|
# list-services command
|
|
subparsers.add_parser('list-services', help='List all available services')
|
|
|
|
# get-config command
|
|
subparsers.add_parser('get-config', help='Get Home Assistant configuration')
|
|
|
|
# call-service command
|
|
service_parser = subparsers.add_parser('call-service', help='Call a service')
|
|
service_parser.add_argument('domain', help='Service domain (e.g., light, switch)')
|
|
service_parser.add_argument('service', help='Service name (e.g., turn_on, turn_off)')
|
|
service_parser.add_argument('--entity', help='Entity ID to target')
|
|
service_parser.add_argument('--data', help='Service data as JSON string')
|
|
|
|
# get-error-log command
|
|
subparsers.add_parser('get-error-log', help='Get the error log')
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not args.url:
|
|
print("Error: Home Assistant URL is required (use --url or set HA_URL environment variable)", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if not args.token:
|
|
print("Error: Access token is required (use --token or set HA_TOKEN environment variable)", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
client = HomeAssistantClient(args.url, args.token)
|
|
|
|
if args.command == 'get-states':
|
|
result = client.get_states()
|
|
elif args.command == 'get-state':
|
|
result = client.get_state(args.entity_id)
|
|
elif args.command == 'list-services':
|
|
result = client.get_services()
|
|
elif args.command == 'get-config':
|
|
result = client.get_config()
|
|
elif args.command == 'call-service':
|
|
service_data = json.loads(args.data) if args.data else None
|
|
result = client.call_service(args.domain, args.service, args.entity, service_data)
|
|
elif args.command == 'get-error-log':
|
|
result = client.get_error_log()
|
|
print(result)
|
|
return
|
|
|
|
print(json.dumps(result, indent=2))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|