#!/usr/bin/env python3 """ Home Assistant API Script (ha-sofia instance) Control and query Home Assistant entities on ha-sofia.viktorbarzin.me. """ import argparse import json import os import sys from urllib.parse import urljoin try: import requests except ImportError: print("ERROR: Required package not installed. Run:") print(" pip install requests") sys.exit(1) # Configuration from environment variables (ha-sofia specific) HA_URL = os.environ.get("HOME_ASSISTANT_SOFIA_URL", "").rstrip("/") HA_TOKEN = os.environ.get("HOME_ASSISTANT_SOFIA_TOKEN") if not HA_URL or not HA_TOKEN: print("ERROR: HOME_ASSISTANT_SOFIA_URL and HOME_ASSISTANT_SOFIA_TOKEN environment variables must be set.") print("These should be set when activating the Claude venv (~/.venvs/claude)") sys.exit(1) HEADERS = { "Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json", } def api_get(endpoint): """Make GET request to HA API.""" url = f"{HA_URL}/api/{endpoint}" response = requests.get(url, headers=HEADERS, timeout=30) response.raise_for_status() return response.json() def api_post(endpoint, data=None): """Make POST request to HA API.""" url = f"{HA_URL}/api/{endpoint}" response = requests.post(url, headers=HEADERS, json=data or {}, timeout=30) response.raise_for_status() return response.json() if response.text else {} def get_states(): """Get all entity states.""" return api_get("states") def get_state(entity_id): """Get state of a specific entity.""" return api_get(f"states/{entity_id}") def get_services(): """Get all available services.""" return api_get("services") def call_service(domain, service, entity_id=None, data=None): """Call a Home Assistant service.""" payload = data or {} if entity_id: payload["entity_id"] = entity_id return api_post(f"services/{domain}/{service}", payload) def list_entities(domain_filter=None, area_filter=None): """List all entities, optionally filtered by domain or area.""" states = get_states() entities = [] for state in states: entity_id = state["entity_id"] domain = entity_id.split(".")[0] if domain_filter and domain != domain_filter: continue entities.append({ "entity_id": entity_id, "state": state["state"], "friendly_name": state["attributes"].get("friendly_name", entity_id), "domain": domain, }) # Sort by domain, then entity_id entities.sort(key=lambda x: (x["domain"], x["entity_id"])) return entities def turn_on(entity_id): """Turn on an entity.""" domain = entity_id.split(".")[0] return call_service(domain, "turn_on", entity_id) def turn_off(entity_id): """Turn off an entity.""" domain = entity_id.split(".")[0] return call_service(domain, "turn_off", entity_id) def toggle(entity_id): """Toggle an entity.""" domain = entity_id.split(".")[0] return call_service(domain, "toggle", entity_id) def set_value(entity_id, value): """Set value for input entities (input_number, input_text, etc.).""" domain = entity_id.split(".")[0] if domain == "input_number": return call_service(domain, "set_value", entity_id, {"value": float(value)}) elif domain == "input_text": return call_service(domain, "set_value", entity_id, {"value": str(value)}) elif domain == "input_boolean": if value.lower() in ("true", "on", "1", "yes"): return turn_on(entity_id) else: return turn_off(entity_id) elif domain == "input_select": return call_service(domain, "select_option", entity_id, {"option": str(value)}) elif domain == "light": # Assume value is brightness percentage return call_service(domain, "turn_on", entity_id, {"brightness_pct": int(value)}) elif domain == "climate": return call_service(domain, "set_temperature", entity_id, {"temperature": float(value)}) elif domain == "cover": return call_service(domain, "set_cover_position", entity_id, {"position": int(value)}) else: print(f"Warning: set_value not implemented for domain '{domain}'", file=sys.stderr) return {} def run_script(script_id): """Run a script.""" if not script_id.startswith("script."): script_id = f"script.{script_id}" return call_service("script", "turn_on", script_id) def run_scene(scene_id): """Activate a scene.""" if not scene_id.startswith("scene."): scene_id = f"scene.{scene_id}" return call_service("scene", "turn_on", scene_id) def send_notification(message, title=None, target="notify"): """Send a notification.""" data = {"message": message} if title: data["title"] = title return call_service("notify", target, data=data) def format_entities(entities, output_format="text"): """Format entities for display.""" if output_format == "json": return json.dumps(entities, indent=2) if not entities: return "No entities found." lines = [] current_domain = None for entity in entities: if entity["domain"] != current_domain: current_domain = entity["domain"] lines.append(f"\n## {current_domain}") state = entity["state"] name = entity["friendly_name"] eid = entity["entity_id"] # Color-code common states if state in ("on", "home", "open", "playing"): state_display = f"[ON] {state}" elif state in ("off", "away", "closed", "idle", "paused"): state_display = f"[--] {state}" elif state == "unavailable": state_display = "[??] unavailable" else: state_display = state lines.append(f"- {name}: {state_display}") lines.append(f" `{eid}`") return "\n".join(lines) def search_entities(query): """Search entities by name or ID.""" query = query.lower() states = get_states() matches = [] for state in states: entity_id = state["entity_id"] friendly_name = state["attributes"].get("friendly_name", "").lower() if query in entity_id.lower() or query in friendly_name: matches.append({ "entity_id": entity_id, "state": state["state"], "friendly_name": state["attributes"].get("friendly_name", entity_id), "domain": entity_id.split(".")[0], }) matches.sort(key=lambda x: (x["domain"], x["entity_id"])) return matches def main(): parser = argparse.ArgumentParser(description="Control Home Assistant (ha-sofia)") subparsers = parser.add_subparsers(dest="command", help="Command to run") # List command list_parser = subparsers.add_parser("list", help="List entities") list_parser.add_argument("--domain", "-d", help="Filter by domain (light, switch, sensor, etc.)") list_parser.add_argument("--json", action="store_true", help="Output as JSON") # Search command search_parser = subparsers.add_parser("search", help="Search entities") search_parser.add_argument("query", help="Search query") search_parser.add_argument("--json", action="store_true", help="Output as JSON") # State command state_parser = subparsers.add_parser("state", help="Get entity state") state_parser.add_argument("entity_id", help="Entity ID") state_parser.add_argument("--json", action="store_true", help="Output as JSON") # On command on_parser = subparsers.add_parser("on", help="Turn on entity") on_parser.add_argument("entity_id", help="Entity ID") # Off command off_parser = subparsers.add_parser("off", help="Turn off entity") off_parser.add_argument("entity_id", help="Entity ID") # Toggle command toggle_parser = subparsers.add_parser("toggle", help="Toggle entity") toggle_parser.add_argument("entity_id", help="Entity ID") # Set command set_parser = subparsers.add_parser("set", help="Set entity value") set_parser.add_argument("entity_id", help="Entity ID") set_parser.add_argument("value", help="Value to set") # Script command script_parser = subparsers.add_parser("script", help="Run a script") script_parser.add_argument("script_id", help="Script ID (with or without 'script.' prefix)") # Scene command scene_parser = subparsers.add_parser("scene", help="Activate a scene") scene_parser.add_argument("scene_id", help="Scene ID (with or without 'scene.' prefix)") # Service command service_parser = subparsers.add_parser("service", help="Call a service") service_parser.add_argument("domain", help="Service domain") service_parser.add_argument("service", help="Service name") service_parser.add_argument("--entity", "-e", help="Entity ID") service_parser.add_argument("--data", "-d", help="JSON data") # Services list command services_parser = subparsers.add_parser("services", help="List available services") services_parser.add_argument("--domain", "-d", help="Filter by domain") services_parser.add_argument("--json", action="store_true", help="Output as JSON") # Notify command notify_parser = subparsers.add_parser("notify", help="Send notification") notify_parser.add_argument("message", help="Notification message") notify_parser.add_argument("--title", "-t", help="Notification title") notify_parser.add_argument("--target", default="notify", help="Notification target (default: notify)") args = parser.parse_args() if not args.command: parser.print_help() sys.exit(1) try: if args.command == "list": entities = list_entities(domain_filter=args.domain) output_format = "json" if args.json else "text" print(format_entities(entities, output_format)) elif args.command == "search": entities = search_entities(args.query) output_format = "json" if args.json else "text" print(format_entities(entities, output_format)) elif args.command == "state": state = get_state(args.entity_id) if args.json: print(json.dumps(state, indent=2)) else: print(f"Entity: {state['entity_id']}") print(f"State: {state['state']}") print(f"Name: {state['attributes'].get('friendly_name', 'N/A')}") if state['attributes']: print("Attributes:") for key, value in state['attributes'].items(): if key != 'friendly_name': print(f" {key}: {value}") elif args.command == "on": turn_on(args.entity_id) print(f"Turned on: {args.entity_id}") elif args.command == "off": turn_off(args.entity_id) print(f"Turned off: {args.entity_id}") elif args.command == "toggle": toggle(args.entity_id) print(f"Toggled: {args.entity_id}") elif args.command == "set": set_value(args.entity_id, args.value) print(f"Set {args.entity_id} to {args.value}") elif args.command == "script": run_script(args.script_id) print(f"Ran script: {args.script_id}") elif args.command == "scene": run_scene(args.scene_id) print(f"Activated scene: {args.scene_id}") elif args.command == "service": data = json.loads(args.data) if args.data else None call_service(args.domain, args.service, args.entity, data) print(f"Called {args.domain}.{args.service}") elif args.command == "services": services = get_services() if args.domain: services = [s for s in services if s["domain"] == args.domain] if args.json: print(json.dumps(services, indent=2)) else: for svc in services: print(f"\n## {svc['domain']}") for name, info in svc["services"].items(): desc = info.get("description", "") print(f"- {name}: {desc[:60]}...") elif args.command == "notify": send_notification(args.message, args.title, args.target) print(f"Sent notification: {args.message[:50]}...") except requests.exceptions.HTTPError as e: print(f"HTTP Error: {e}", file=sys.stderr) print(f"Response: {e.response.text}", file=sys.stderr) sys.exit(1) except Exception as e: print(f"Error: {e}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()