Home Assistant

DEVKIT_AUTH_PATHS = [
    os.environ.get("OPENHOME_HA_AUTH_PATH", ""),
    os.path.expanduser("~/.ha_refresh_token"),
    "/home/openhome/.ha_refresh_token",
    "/root/.ha_refresh_token",
]

_AUTH = {
    "url": None,
    "client_id": None,
    "refresh_token": None,
    "access_token": None,
    "exp": 0,
}


def _ensure_auth():
    if not _AUTH["refresh_token"]:
        path = None
        for p in DEVKIT_AUTH_PATHS:
            if p and os.path.exists(p) and os.access(p, os.R_OK):
                path = p
                break
        if not path:
            log.warning("[HA] no refresh token file found")
            return None, None
        try:
            values = {}
            with open(path, "r") as f:
                for line in f:
                    line = line.strip()
                    if not line or line.startswith("#") or "=" not in line:
                        continue
                    k, v = line.split("=", 1)
                    values[k.strip()] = v.strip().strip('"\'')
        except Exception as exc:
            log.error(f"[HA] read {path}: {exc}")
            return None, None

        url = (values.get("HA_URL") or "").rstrip("/")
        rt = values.get("HA_REFRESH_TOKEN")
        cid = values.get("HA_CLIENT_ID") or (url + "/" if url else None)
        if not url or not rt:
            log.warning("[HA] auth file missing HA_URL or HA_REFRESH_TOKEN")
            return None, None
        _AUTH["url"] = url
        _AUTH["refresh_token"] = rt
        _AUTH["client_id"] = cid

    if _AUTH["access_token"] and time.time() < _AUTH["exp"]:
        return _AUTH["url"], _AUTH["access_token"]

    data = urllib.parse.urlencode({
        "grant_type": "refresh_token",
        "refresh_token": _AUTH["refresh_token"],
        "client_id": _AUTH["client_id"],
    }).encode()
    req = urllib.request.Request(
        f"{_AUTH['url']}/auth/token", data=data,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            body = json.loads(resp.read().decode("utf-8", errors="replace"))
    except urllib.error.HTTPError as e:
        if e.code == 400:
            _AUTH["refresh_token"] = None
        log.error(f"[HA] /auth/token HTTP {e.code}")
        return _AUTH["url"], None
    except Exception as exc:
        log.error(f"[HA] /auth/token failed: {exc}")
        return _AUTH["url"], None

    access_token = body.get("access_token")
    expires_in = int(body.get("expires_in") or 1800)
    if not access_token:
        return _AUTH["url"], None
    _AUTH["access_token"] = access_token
    _AUTH["exp"] = time.time() + expires_in - 60
    return _AUTH["url"], access_token

HA_URL=http://localhost:8123
HA_REFRESH_TOKEN=your_home_assistant_refresh_token
HA_CLIENT_ID=http://localhost:8123/

import json
import re

from src.agent.capability import MatchingCapability
from src.main import AgentWorker
from src.agent.capability_worker import CapabilityWorker

def _tasmota_light_context():
    return (
        "=== TASMOTA LIGHTS (integration: tasmota_light) ===\n"
        "Smart bulbs/strips controlled via Tasmota over MQTT. The user may have "
        "one or more. Set `hint` to a short natural-language descriptor of which "
        "device (e.g. 'kitchen', 'bedroom lamp', 'the bulb'). Leave hint empty if "
        "only one is plausible from context — the DevKit will pick the only one.\n"
        "Actions:\n"
        "  on_off      params={\"state\":\"on\"|\"off\"|\"toggle\"}\n"
        "  brightness  params={\"percent\":0-100}  (\"bright\"=90, \"dim\"=20, \"half\"=50)\n"
        "  color       params={\"rgb\":[R,G,B]}    (0-255 each; convert any color name yourself)\n"
        "  color_temp  params={\"kelvin\":2000-6500}  (cozy=2700, reading=4000, daylight=6000)"
    )


def _tasmota_light_execute(action, params, hint):
    p = params or {}
    h = (hint or "").strip()
    if action == "on_off":
        state = (p.get("state") or "on").lower()
        if state == "on":
            service = "turn_on"
        elif state == "off":
            service = "turn_off"
        else:
            service = "toggle"
        return ("tasmota_light_action", [service, h, "{}"])
    if action == "brightness":
        try:
            pct = float(p.get("percent", 50))
        except Exception:
            return None
        b = max(0, min(255, int(round(pct * 2.55))))
        return ("tasmota_light_action", ["turn_on", h, json.dumps({"brightness": b})])
    if action == "color":
        rgb = p.get("rgb") or [255, 255, 255]
        try:
            rgb = [max(0, min(255, int(c))) for c in list(rgb)[:3]]
        except Exception:
            return None
        if len(rgb) != 3:
            return None
        return ("tasmota_light_action", ["turn_on", h, json.dumps({"rgb_color": rgb})])
    if action == "color_temp":
        try:
            k = max(2000, min(6500, int(p.get("kelvin", 4000))))
        except Exception:
            return None
        return ("tasmota_light_action", ["turn_on", h, json.dumps({"color_temp_kelvin": k})])
    return None


# Integration registry
INTEGRATION_REGISTRY = {
    "tasmota_light": {
        "context": _tasmota_light_context,
        "execute": _tasmota_light_execute,
    },
}


# Prompt builder
def _build_system_prompt():
    blocks = "\n\n".join(
        handler["context"]() for handler in INTEGRATION_REGISTRY.values()
    )
    return f"""You are a voice assistant for a smart home. Your output is read aloud by TTS to a native US English speaker.

Return ONLY valid JSON (no markdown, no code fences):
{{
  "integration": "<one of the loaded integration names, or 'none' for end_session/unrelated>",
  "action": "<an action listed under that integration, or 'end_session', or 'none'>",
  "params": {{ ... action-specific ... }},
  "hint": "<short natural-language descriptor of which device (e.g. 'kitchen', 'lamp'), or empty>",
  "spoken_response": "<short reply, see TTS RULES>"
}}

TTS RULES (apply to spoken_response):
- Plain spoken English. No markdown, asterisks, dashes, bullets, emojis, URLs, code, symbols.
- No abbreviations. Spell them out: "for example" not "e.g.", "by the way" not "FYI".
- Numbers spoken naturally: "thirty" not "30" when natural.
- No brand or technical jargon. Never say "Tasmota", "entity", "service", "RGB", "kelvin".
- Don't restate the user's command. Just confirm and stop.

LENGTH RULES:
- On/off: under 6 words. ("On it." / "Lights out.")
- Brightness/color/temp: under 10 words. ("Going warm." / "Setting it to red.")
- end_session: under 6 words. ("Catch you later." / "Anytime.")

INTENT RULES:
1. Return ONLY valid JSON.
2. integration MUST be one of the loaded names (or "none").
3. Only use actions listed for the chosen integration.
4. "it"/"that"/"the light" in follow-ups = previously hinted device. Reuse the same hint.
5. Goodbye, "I'm done", "that's all", "never mind", "thanks bye", silence-style closers -> action="end_session".
6. We CANNOT read the device's actual current state. NEVER claim it is on or off without a fresh action.
7. Unrelated request -> integration="none", action="none", short polite reply.

AVAILABLE INTEGRATIONS:
{blocks}
"""


REPROMPTS = ["Sorry?", "One more time?", "Didn't catch that.", "Try saying turn it on or change the color."]


class AgentAbilityCapability(MatchingCapability):
    worker: AgentWorker = None
    capability_worker: CapabilityWorker = None

    #{{register capability}}

    def call(self, worker: AgentWorker):
        self.worker = worker
        self.capability_worker = CapabilityWorker(self.worker)
        self.worker.session_tasks.create(self.run())

    async def run(self):
        try:
            try:
                await self.capability_worker.send_devkit_capability_action(
                    function_name="ha_startup", args=[], timeout=15,
                )
            except Exception as exc:
                self.worker.editor_logging_handler.error(
                    f"[HA] ha_startup ping failed: {exc}"
                )

            system_prompt = _build_system_prompt()
            await self.capability_worker.speak("Smart home ready. What can I do?")

            history = []
            last_hint = ""
            miss_count = 0

            while True:
                msg = await self.capability_worker.user_response()
                if not msg or not msg.strip():
                    prompt = REPROMPTS[min(miss_count, len(REPROMPTS) - 1)]
                    await self.capability_worker.speak(prompt)
                    miss_count += 1
                    if miss_count >= 4:
                        await self.capability_worker.speak("Talk later.")
                        break
                    continue
                miss_count = 0

                hint_note = f" (last device hint: {last_hint})" if last_hint else ""
                user_prompt = f'User said: "{msg}"{hint_note}'

                raw = self.capability_worker.text_to_text_response(
                    user_prompt, history, system_prompt=system_prompt,
                )

                intent = self._parse_json(raw)
                if intent is None:
                    await self.capability_worker.speak("Try that again?")
                    continue

                history.append({"role": "user", "content": user_prompt})
                history.append({"role": "assistant", "content": raw or ""})
                if len(history) > 20:
                    history = history[-20:]

                hint_val = (intent.get("hint") or "").strip()
                if hint_val:
                    last_hint = hint_val

                if intent.get("action") == "end_session":
                    await self.capability_worker.speak(
                        intent.get("spoken_response") or "Bye."
                    )
                    break

                await self._dispatch(intent)

        except Exception as exc:
            self.worker.editor_logging_handler.error(f"[HA] run error: {exc}")
            await self.capability_worker.speak("Something didn't work right.")
        finally:
            self.capability_worker.resume_normal_flow()

    async def _dispatch(self, intent):
        integration = intent.get("integration", "")
        action = intent.get("action", "")
        params = intent.get("params") or {}
        hint = (intent.get("hint") or "").strip()
        spoken = intent.get("spoken_response") or "Done."

        if integration in ("none", "") or action in ("none", ""):
            await self.capability_worker.speak(spoken)
            return

        handler = INTEGRATION_REGISTRY.get(integration)
        if not handler:
            await self.capability_worker.speak("I can't do that one.")
            return

        translated = handler["execute"](action, params, hint)
        if not translated:
            await self.capability_worker.speak("Can't do that one.")
            return

        devkit_fn, devkit_args = translated
        try:
            await self.capability_worker.send_devkit_capability_action(
                function_name=devkit_fn, args=devkit_args, timeout=10,
            )
        except Exception as exc:
            self.worker.editor_logging_handler.error(
                f"[HA] devkit call {devkit_fn} failed: {exc}"
            )
        await self.capability_worker.speak(spoken)

    @staticmethod
    def _parse_json(text):
        if not text:
            return None
        try:
            cleaned = re.sub(r"^```[a-zA-Z]*\n|\n```$", "", text.strip())
            return json.loads(cleaned)
        except Exception:
            return None

import os
import sys
import json
import time
import urllib.parse
import urllib.request
import urllib.error

try:
    from devkit_utils.devkit_logging import web_logger as log
except Exception:
    class _StubLogger:
        def info(self, *a, **k): pass
        def warning(self, *a, **k): pass
        def error(self, *a, **k): pass
        def debug(self, *a, **k): pass
    log = _StubLogger()


log.info("[HA devkit] module loaded")

REQUEST_TIMEOUT = 8

DEVKIT_AUTH_PATHS = [
    os.environ.get("OPENHOME_HA_AUTH_PATH", ""),
    os.path.expanduser("~/.ha_refresh_token"),
    "/home/openhome/.ha_refresh_token",
    "/root/.ha_refresh_token",
]

_AUTH = {
    "url": None,
    "client_id": None,
    "refresh_token": None,
    "access_token": None,
    "exp": 0,
}

_NON_TASMOTA_LIGHT_PLATFORMS = ("hue", "lifx", "shelly", "wled", "yeelight", "tplink")


def _ensure_auth():
    if not _AUTH["refresh_token"]:
        path = None
        for p in DEVKIT_AUTH_PATHS:
            if p and os.path.exists(p) and os.access(p, os.R_OK):
                path = p
                break
        if not path:
            log.warning("[HA] no refresh token file found")
            return None, None
        try:
            values = {}
            with open(path, "r") as f:
                for line in f:
                    line = line.strip()
                    if not line or line.startswith("#") or "=" not in line:
                        continue
                    k, v = line.split("=", 1)
                    values[k.strip()] = v.strip().strip('"\'')
        except Exception as exc:
            log.error(f"[HA] read {path}: {exc}")
            return None, None

        url = (values.get("HA_URL") or "").rstrip("/")
        rt = values.get("HA_REFRESH_TOKEN")
        cid = values.get("HA_CLIENT_ID") or (url + "/" if url else None)
        if not url or not rt:
            log.warning("[HA] auth file missing HA_URL or HA_REFRESH_TOKEN")
            return None, None
        _AUTH["url"] = url
        _AUTH["refresh_token"] = rt
        _AUTH["client_id"] = cid

    if _AUTH["access_token"] and time.time() < _AUTH["exp"]:
        return _AUTH["url"], _AUTH["access_token"]

    data = urllib.parse.urlencode({
        "grant_type": "refresh_token",
        "refresh_token": _AUTH["refresh_token"],
        "client_id": _AUTH["client_id"],
    }).encode()
    req = urllib.request.Request(
        f"{_AUTH['url']}/auth/token", data=data,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            body = json.loads(resp.read().decode("utf-8", errors="replace"))
    except urllib.error.HTTPError as e:
        if e.code == 400:
            _AUTH["refresh_token"] = None
        log.error(f"[HA] /auth/token HTTP {e.code}")
        return _AUTH["url"], None
    except Exception as exc:
        log.error(f"[HA] /auth/token failed: {exc}")
        return _AUTH["url"], None

    access_token = body.get("access_token")
    expires_in = int(body.get("expires_in") or 1800)
    if not access_token:
        return _AUTH["url"], None
    _AUTH["access_token"] = access_token
    _AUTH["exp"] = time.time() + expires_in - 60
    return _AUTH["url"], access_token


def _ha_request(method, path, body=None, timeout=REQUEST_TIMEOUT):
    url, token = _ensure_auth()
    if not url or not token:
        return 0, "auth unavailable"

    data = json.dumps(body).encode("utf-8") if body is not None else None

    for attempt in (1, 2):
        req = urllib.request.Request(
            url + path, data=data, method=method,
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {token}",
            },
        )
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                text = resp.read().decode("utf-8", errors="replace")
                return resp.status, text
        except urllib.error.HTTPError as e:
            if e.code == 401 and attempt == 1:
                log.warning("[HA] <- 401, refreshing token")
                _AUTH["access_token"] = None
                _AUTH["exp"] = 0
                _, token = _ensure_auth()
                if not token:
                    return 401, "auth refresh failed"
                continue
            text = ""
            try:
                text = e.read().decode("utf-8", errors="replace")
            except Exception:
                pass
            log.warning(f"[HA] <- HTTP {e.code}")
            return e.code, text
        except Exception as e:
            log.error(f"[HA] {type(e).__name__}: {e}")
            return 0, str(e)

    return 0, "exhausted retries"


def _emit(data):
    payload = json.dumps(data, separators=(",", ":"))
    sys.stdout.write(payload + "\n")
    try:
        sys.stdout.flush()
    except Exception:
        pass
    return data


def _pick_entity(candidates, name_hint):
    if not candidates:
        return None
    needle = (name_hint or "").strip().lower()
    if not needle:
        return candidates[0].get("entity_id")
    for s in candidates:
        attrs = s.get("attributes") or {}
        fn = (attrs.get("friendly_name") or "").lower()
        eid = s.get("entity_id", "").lower()
        if needle in fn or needle in eid:
            return s.get("entity_id")
    for s in candidates:
        attrs = s.get("attributes") or {}
        fn = (attrs.get("friendly_name") or "").lower()
        eid = s.get("entity_id", "").lower()
        for word in needle.split():
            if word and (word in fn or word in eid):
                return s.get("entity_id")
    return candidates[0].get("entity_id")


# DevKit functions
def ha_startup():
    url, token = _ensure_auth()
    if not url or not token:
        return _emit({"ok": False, "err": "auth unavailable"})

    status, _ = _ha_request("GET", "/api/")
    if status != 200:
        return _emit({"ok": False, "err": f"HA ping HTTP {status}"})

    integrations = []
    status, body = _ha_request("GET", "/api/config/config_entries/entry", timeout=10)
    if status == 200:
        try:
            entries = json.loads(body)
            integrations = sorted({e.get("domain") for e in entries if e.get("domain")})
        except Exception as exc:
            log.warning(f"[HA] config_entries parse: {exc}")

    entities = 0
    status, body = _ha_request("GET", "/api/states", timeout=15)
    if status == 200:
        try:
            entities = len(json.loads(body))
        except Exception as exc:
            log.warning(f"[HA] states parse: {exc}")

    return _emit({
        "ok": True,
        "url": url,
        "integrations": integrations,
        "entities": entities,
    })


def tasmota_light_action(service=None, name_hint="", extra_json="{}"):
    if not service:
        return _emit({"ok": False, "err": "service required"})

    status, body = _ha_request("GET", "/api/states", timeout=15)
    if status != 200:
        return _emit({"ok": False, "err": f"states HTTP {status}"})
    try:
        states = json.loads(body)
    except Exception as exc:
        return _emit({"ok": False, "err": f"parse: {exc}"})

    candidates = []
    for s in states:
        if not isinstance(s, dict):
            continue
        eid = s.get("entity_id", "")
        if not eid.startswith("light."):
            continue
        if s.get("state") in ("unavailable", "unknown"):
            continue
        attrs = s.get("attributes") or {}
        if attrs.get("platform", "") in _NON_TASMOTA_LIGHT_PLATFORMS:
            continue
        candidates.append(s)

    entity_id = _pick_entity(candidates, name_hint)
    if not entity_id:
        log.warning(f"[HA] no tasmota lights to match hint={name_hint!r}")
        return _emit({"ok": False, "err": "no tasmota lights", "hint": name_hint})

    try:
        extra = json.loads(extra_json) if extra_json and extra_json != "{}" else {}
    except Exception:
        extra = {}
    payload = {"entity_id": entity_id}
    payload.update(extra)

    status, _ = _ha_request("POST", f"/api/services/light/{service}", body=payload)
    return _emit({"ok": status in (200, 201), "status": status, "entity_id": entity_id})


FUNCTION_REGISTRY = {
    "ha_startup": ha_startup,
    "tasmota_light_action": tasmota_light_action,
}

log.info(f"[HA devkit] registry: {list(FUNCTION_REGISTRY.keys())}")


def list_functions():
    print("=" * 60)
    print("  HA Devkit Functions (Tasmota Lights)")
    print("=" * 60)
    for name, func in FUNCTION_REGISTRY.items():
        doc = (func.__doc__ or "").strip().split("\n")[0]
        print(f"  {name}: {doc}")


FUNCTION_REGISTRY["list_functions"] = list_functions


def main():
    if len(sys.argv) < 2:
        print("Usage: python3 devkit_functions.py <function_name> [args...]")
        sys.exit(1)
    func_name = sys.argv[1]
    func_args = sys.argv[2:]
    if func_name in ("--help", "-h"):
        list_functions()
        sys.exit(0)
    if func_name not in FUNCTION_REGISTRY:
        print(f"[error] unknown function '{func_name}'")
        sys.exit(1)
    try:
        FUNCTION_REGISTRY[func_name](*func_args)
        sys.exit(0)
    except TypeError as e:
        print(f"[error] wrong arguments for '{func_name}': {e}")
        sys.exit(1)
    except Exception as e:
        print(f"[error] {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()