Local Abilities

result = await self.capability_worker.send_devkit_capability_action(
    function_name="your_function_name",
    args=["arg1", "arg2"],
    timeout=10,
)

{
    "success": True,                  # True if the DevKit function completed successfully
    "output": "captured stdout",      # Output from print() calls in devkit_functions.py
    "error": None,                    # Captured stderr or execution error details
    "function_name": "function_name", # Function that was executed
    "args": ["arg1", "arg2"],         # Arguments passed to the function
    "capability_name": "ability_name" # Ability that handled the request
}

import json
import sys
import subprocess
from devkit_utils.devkit_logging import web_logger as log


def _print_payload(payload):
    output = json.dumps(payload)
    log.info("stdout payload: %s", output)
    print(output)


def check_wifi():
    try:
        result = subprocess.run(
            ["iwgetid", "-r"], capture_output=True, text=True, timeout=5
        )
        ssid = result.stdout.strip()
        if ssid:
            _print_payload({
                "success": True,
                "metric": "wifi",
                "spoken_response": f"Wi-Fi is connected to {ssid}.",
                "data": {"connected": True, "ssid": ssid},
                "error": None,
            })
        else:
            _print_payload({
                "success": True,
                "metric": "wifi",
                "spoken_response": "Wi-Fi is not connected.",
                "data": {"connected": False, "ssid": None},
                "error": None,
            })
    except Exception as error:
        log.exception("check_wifi failed")
        _print_payload({
            "success": False,
            "metric": "wifi",
            "spoken_response": "I couldn't read Wi-Fi status.",
            "data": {},
            "error": {
                "code": "wifi_error",
                "message": str(error),
            },
        })


FUNCTION_REGISTRY = {
    "check_wifi": check_wifi,
}


if __name__ == "__main__":
    function_name = sys.argv[1]
    FUNCTION_REGISTRY[function_name](*sys.argv[2:])

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


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

    #{{register capability}}

    async def first_function(self):
        try:
            result = await self.capability_worker.send_devkit_capability_action(
                function_name="check_wifi",
                args=[],
                timeout=5,
            )
            await self.capability_worker.speak(self._spoken_response_from_result(result))
        finally:
            self.capability_worker.resume_normal_flow()

    def _spoken_response_from_result(self, result):
        if not isinstance(result, dict) or not result.get("success"):
            return "I couldn't fetch Wi-Fi status from the DevKit."

        output = (result.get("output") or "").strip()
        if not output:
            return "The DevKit did not return Wi-Fi status."

        try:
            payload = json.loads(output)
        except json.JSONDecodeError:
            return "I couldn't read the DevKit response."

        return payload.get("spoken_response") or "I couldn't read Wi-Fi status."

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

result = await self.capability_worker.send_devkit_capability_action(
    function_name="get_sensor_value",
    args=["temperature"],
    timeout=10,
    capability_name="target_ability_name",
)

from devkit_utils.devkit_logging import web_logger as log

log.info("devkit stats functions loaded")

def check_temperature():
    log.info("check_temperature: entry")
    # Your DevKit-side code runs here
    log.info("check_temperature: completed")

rpi-ws281x        # NeoPixel / WS281x LED strip control
gpiozero          # high-level GPIO pin control
RPi.GPIO          # low-level GPIO access
picamera2         # camera access
adafruit-blinka   # CircuitPython compatibility for sensors
smbus2            # I2C bus communication
pyserial          # serial port communication

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


AVAILABLE_STATS = {
    "get_cpu": "CPU usage",
    "get_memory": "Memory usage",
    "get_temperature": "Device temperature",
    "get_uptime": "Device uptime",
    "get_wifi": "Wi-Fi connection",
    "get_disk": "Disk usage",
    "get_health": "Overall device health",
    "get_all_stats": "Summary of all key metrics",
}

FUNCTIONS_DESCRIPTION = "\n".join(
    f"- {name}: {description}" for name, description in AVAILABLE_STATS.items()
)

SYSTEM_PROMPT = f"""You are a request router for a DevKit telemetry Ability. Your sole responsibility is to map user input to exactly one function name. You do not answer questions, explain concepts, or generate conversational responses.

## Device Context
The OpenHome DevKit is the user's locally connected device. Telemetry refers to its live runtime metrics: CPU, memory, temperature, uptime, Wi-Fi, disk, and health. This Ability is limited strictly to the functions listed below.

## Response Format
Always return a single JSON object. No prose, no markdown, no extra keys.
{{"function_name": "<function_name | none | exit>"}}

## Available Functions
{FUNCTIONS_DESCRIPTION}

## Routing Rules
- General status, "all stats", "everything", "snapshot", "system info" -> get_all_stats
- CPU, processor, load, compute, busy, usage -> get_cpu
- Memory, RAM, available memory, used memory -> get_memory
- Temperature, temp, heat, thermal, hot, warm -> get_temperature
- Uptime, boot time, running time, how long running -> get_uptime
- Wi-Fi, wifi, network, SSID, connection -> get_wifi
- Disk, storage, free space, used space -> get_disk
- Health, diagnostics, issues, problems, anything wrong -> get_health

## Exit Routing
Trigger `exit` when the user says: stop, quit, cancel, end, done, all done, that's all, thank you, thanks, goodbye, bye — or any close variation, even with filler words.

## Unsupported Requests
If the request is unrelated to DevKit telemetry, or asks for telemetry not covered by any available function, return:
{{"function_name": "none"}}

## Hard Rules
- Return exactly one function_name per response.
- Never explain, define, or discuss any concept — even if directly asked.
- Route by intent: if the user asks "what is my CPU usage?" that is a CPU telemetry request -> get_cpu.
- Do not include any text outside the JSON object.
"""


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

    #{{register capability}}

    async def first_function(self):
        try:
            is_first_turn = True
            conversation_history = []

            while True:
                if is_first_turn:
                    user_message = await self.capability_worker.wait_for_complete_transcription()
                else:
                    user_message = await self.capability_worker.user_response()

                if not user_message or not user_message.strip():
                    continue

                route = self._route_to_devkit_function(user_message, conversation_history)
                function_name = route.get("function_name", "")

                if is_first_turn and function_name in ("", "none"):
                    function_name = "get_all_stats"

                if function_name == "exit":
                    await self.capability_worker.speak("Exiting DevKit stats.")
                    break

                if function_name not in AVAILABLE_STATS:
                    await self.capability_worker.speak(
                        "I can't fetch that DevKit information. Try asking for CPU, memory, temperature, disk, uptime, Wi-Fi, or health."
                    )
                    is_first_turn = False
                    continue

                result = await self.capability_worker.send_devkit_capability_action(
                    function_name=function_name,
                    args=[],
                    timeout=8,
                )
                spoken_message = self._spoken_response_from_result(result)
                await self.capability_worker.speak(spoken_message)

                conversation_history.append({"role": "user", "content": user_message})
                conversation_history.append({"role": "assistant", "content": spoken_message})
                conversation_history = conversation_history[-12:]

                await self.capability_worker.speak("Want me to check anything else, or say stop to exit.")
                is_first_turn = False

        except Exception as error:
            self.worker.editor_logging_handler.error(f"DevKit stats failed: {error}")
            await self.capability_worker.speak("Something went wrong while checking DevKit stats.")
        finally:
            self.capability_worker.resume_normal_flow()

    def _route_to_devkit_function(self, user_message, conversation_history):
        response = self.capability_worker.text_to_text_response(
            f'User request: "{user_message}"',
            conversation_history,
            system_prompt=SYSTEM_PROMPT,
        )
        cleaned = re.sub(r"^```[a-zA-Z]*\n|\n```$", "", response.strip())
        try:
            return json.loads(cleaned)
        except (json.JSONDecodeError, TypeError, ValueError):
            return {"function_name": ""}

    def _spoken_response_from_result(self, result):
        if not isinstance(result, dict):
            return "I couldn't reach the DevKit."

        if not result.get("success"):
            self.worker.editor_logging_handler.error(
                f"DevKit call failed: {result.get('error')}"
            )
            return "I couldn't fetch that DevKit information. Try asking for another stat."

        output = (result.get("output") or "").strip()
        if not output:
            return "I couldn't fetch that DevKit information. Try asking for another stat."

        try:
            payload = json.loads(output)
        except json.JSONDecodeError:
            self.worker.editor_logging_handler.error(f"Invalid DevKit output: {output}")
            return "I couldn't read the DevKit response."

        if not payload.get("success"):
            error = payload.get("error") or {}
            self.worker.editor_logging_handler.warning(
                f"DevKit stat unavailable: {error.get('code')} {error.get('message')}"
            )

        return payload.get("spoken_response") or "I couldn't read that DevKit stat."

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

import json
import shutil
import subprocess
import sys
import time
from devkit_utils.devkit_logging import web_logger as log


def _emit_success(metric, spoken, data=None):
    payload = {
        "success": True,
        "metric": metric,
        "spoken_response": spoken,
        "data": data or {},
        "error": None,
    }
    serialized_payload = json.dumps(payload)
    log.info("stdout payload: %s", serialized_payload)
    print(serialized_payload)


def _emit_error(metric, code, message, spoken):
    log.error("%s failed [%s]: %s", metric, code, message)
    payload = {
        "success": False,
        "metric": metric,
        "spoken_response": spoken,
        "data": {},
        "error": {
            "code": code,
            "message": message,
        },
    }
    serialized_payload = json.dumps(payload)
    log.info("stdout payload: %s", serialized_payload)
    print(serialized_payload)


def _read_text_file(path):
    try:
        with open(path, "r", encoding="utf-8") as file_handle:
            return file_handle.read().strip()
    except (FileNotFoundError, PermissionError, OSError) as error:
        log.warning("Could not read %s: %s", path, error)
        return ""


def _run_command(command, timeout=5):
    try:
        completed = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        log.warning("Command timed out: %s", command)
        return ""
    except OSError as error:
        log.warning("Command failed: %s: %s", command, error)
        return ""

    if completed.returncode != 0:
        log.warning("Command returned %s: %s", completed.returncode, command)
        return ""

    return completed.stdout.strip()


def _safe_int(value):
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _safe_float(value):
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _read_memory_kb(field_name):
    meminfo = _read_text_file("/proc/meminfo")
    for line in meminfo.splitlines():
        if line.startswith(field_name):
            value = line.split(":", 1)[1].strip().split()[0]
            return _safe_int(value)
    return None


def _read_cpu_sample():
    stat = _read_text_file("/proc/stat")
    for line in stat.splitlines():
        if line.startswith("cpu "):
            values = [_safe_int(value) or 0 for value in line.split()[1:]]
            if len(values) < 4:
                return None
            idle = values[3] + (values[4] if len(values) > 4 else 0)
            return {"idle": idle, "total": sum(values)}
    return None


def _read_cpu_usage_percent(sample_seconds=0.4):
    first = _read_cpu_sample()
    time.sleep(sample_seconds)
    second = _read_cpu_sample()

    if not first or not second:
        return None

    total_delta = second["total"] - first["total"]
    idle_delta = second["idle"] - first["idle"]
    if total_delta <= 0:
        return None

    return round((1 - idle_delta / total_delta) * 100)


def _gb_from_kb(value):
    if value is None:
        return None
    return round(value / 1024 / 1024, 1)


def _temperature_status(celsius):
    if celsius < 50:
        return "running cool"
    if celsius < 65:
        return "comfortable"
    if celsius < 75:
        return "warm"
    if celsius < 85:
        return "hot"
    return "very hot"


def get_cpu():
    metric = "cpu"
    log.info("get_cpu called")
    try:
        used_percent = _read_cpu_usage_percent()
        if used_percent is None:
            _emit_error(metric, "cpu_unavailable", "CPU usage could not be read.", "I couldn't read CPU usage.")
            return

        free_percent = 100 - used_percent
        _emit_success(
            metric,
            f"CPU is {used_percent} percent used and {free_percent} percent free.",
            {"used_percent": used_percent, "free_percent": free_percent},
        )
    except Exception as error:
        log.exception("Unhandled error in get_cpu")
        _emit_error(metric, "cpu_error", str(error), "I couldn't read CPU usage.")


def get_memory():
    metric = "memory"
    log.info("get_memory called")
    try:
        total_gb = _gb_from_kb(_read_memory_kb("MemTotal:"))
        available_gb = _gb_from_kb(_read_memory_kb("MemAvailable:"))
        if total_gb is None or available_gb is None:
            _emit_error(metric, "memory_unavailable", "Memory info could not be read.", "I couldn't read memory usage.")
            return

        used_gb = round(total_gb - available_gb, 1)
        _emit_success(
            metric,
            f"Memory has {used_gb} gigabytes used out of {total_gb}, with {available_gb} gigabytes available.",
            {"total_gb": total_gb, "used_gb": used_gb, "available_gb": available_gb},
        )
    except Exception as error:
        log.exception("Unhandled error in get_memory")
        _emit_error(metric, "memory_error", str(error), "I couldn't read memory usage.")


def get_temperature():
    metric = "temperature"
    log.info("get_temperature called")
    try:
        raw_value = _read_text_file("/sys/class/thermal/thermal_zone0/temp")
        millicelsius = _safe_int(raw_value)
        if millicelsius is None:
            _emit_error(metric, "temperature_unavailable", "Temperature value could not be read.", "I couldn't read the DevKit temperature.")
            return

        celsius = round(millicelsius / 1000, 1)
        status = _temperature_status(celsius)
        _emit_success(
            metric,
            f"DevKit temperature is {celsius} degrees Celsius and {status}.",
            {"celsius": celsius, "status": status},
        )
    except Exception as error:
        log.exception("Unhandled error in get_temperature")
        _emit_error(metric, "temperature_error", str(error), "I couldn't read the DevKit temperature.")


def get_uptime():
    metric = "uptime"
    log.info("get_uptime called")
    try:
        uptime_text = _read_text_file("/proc/uptime")
        uptime_seconds = _safe_float(uptime_text.split()[0]) if uptime_text else None
        if uptime_seconds is None:
            _emit_error(metric, "uptime_unavailable", "Uptime could not be read.", "I couldn't read DevKit uptime.")
            return

        days = int(uptime_seconds // 86400)
        hours = int((uptime_seconds % 86400) // 3600)
        minutes = int((uptime_seconds % 3600) // 60)

        if days:
            spoken_duration = f"{days} days and {hours} hours"
        elif hours:
            spoken_duration = f"{hours} hours and {minutes} minutes"
        else:
            spoken_duration = f"{minutes} minutes"

        _emit_success(
            metric,
            f"The DevKit has been running for {spoken_duration}.",
            {"seconds": round(uptime_seconds), "days": days, "hours": hours, "minutes": minutes},
        )
    except Exception as error:
        log.exception("Unhandled error in get_uptime")
        _emit_error(metric, "uptime_error", str(error), "I couldn't read DevKit uptime.")


def get_wifi():
    metric = "wifi"
    log.info("get_wifi called")
    try:
        ssid = _run_command("iwgetid -r 2>/dev/null")
        if not ssid:
            _emit_success(metric, "Wi-Fi is not connected.", {"connected": False, "ssid": None})
            return

        _emit_success(metric, f"Wi-Fi is connected to {ssid}.", {"connected": True, "ssid": ssid})
    except Exception as error:
        log.exception("Unhandled error in get_wifi")
        _emit_error(metric, "wifi_error", str(error), "I couldn't read Wi-Fi status.")


def get_disk():
    metric = "disk"
    log.info("get_disk called")
    try:
        total_bytes, used_bytes, free_bytes = shutil.disk_usage("/")
        total_gb = round(total_bytes / 1_000_000_000, 1)
        used_gb = round(used_bytes / 1_000_000_000, 1)
        free_gb = round(free_bytes / 1_000_000_000, 1)
        used_percent = round((used_bytes / total_bytes) * 100)

        _emit_success(
            metric,
            f"Disk is {used_percent} percent used, with {free_gb} gigabytes free.",
            {
                "total_gb": total_gb,
                "used_gb": used_gb,
                "free_gb": free_gb,
                "used_percent": used_percent,
            },
        )
    except Exception as error:
        log.exception("Unhandled error in get_disk")
        _emit_error(metric, "disk_error", str(error), "I couldn't read disk usage.")


def get_health():
    metric = "health"
    log.info("get_health called")
    try:
        issues = []
        data = {}

        raw_temperature = _safe_int(_read_text_file("/sys/class/thermal/thermal_zone0/temp"))
        if raw_temperature is not None:
            celsius = round(raw_temperature / 1000, 1)
            data["temperature_celsius"] = celsius
            if celsius >= 75:
                issues.append(f"temperature is high at {celsius} degrees Celsius")

        available_kb = _read_memory_kb("MemAvailable:")
        if available_kb is not None:
            available_mb = round(available_kb / 1024)
            data["memory_available_mb"] = available_mb
            if available_mb < 200:
                issues.append(f"memory is low with {available_mb} megabytes available")

        disk_total, disk_used, _ = shutil.disk_usage("/")
        disk_used_percent = round((disk_used / disk_total) * 100)
        data["disk_used_percent"] = disk_used_percent
        if disk_used_percent >= 90:
            issues.append(f"disk usage is high at {disk_used_percent} percent")

        data["issues"] = issues

        if not issues:
            _emit_success(metric, "The DevKit looks healthy.", data)
        elif len(issues) == 1:
            _emit_success(metric, f"I found one issue: {issues[0]}.", data)
        else:
            _emit_success(metric, f"I found {len(issues)} issues: {', '.join(issues[:2])}.", data)
    except Exception as error:
        log.exception("Unhandled error in get_health")
        _emit_error(metric, "health_error", str(error), "I couldn't run the DevKit health check.")


def get_all_stats():
    metric = "all_stats"
    log.info("get_all_stats called")
    try:
        cpu_percent = _read_cpu_usage_percent()
        raw_temperature = _safe_int(_read_text_file("/sys/class/thermal/thermal_zone0/temp"))
        temperature_celsius = round(raw_temperature / 1000, 1) if raw_temperature is not None else None
        total_memory_gb = _gb_from_kb(_read_memory_kb("MemTotal:"))
        available_memory_gb = _gb_from_kb(_read_memory_kb("MemAvailable:"))
        ssid = _run_command("iwgetid -r 2>/dev/null")
        total_bytes, used_bytes, free_bytes = shutil.disk_usage("/")
        free_disk_gb = round(free_bytes / 1_000_000_000, 1)
        disk_used_percent = round((used_bytes / total_bytes) * 100)

        data = {
            "cpu_used_percent": cpu_percent,
            "temperature_celsius": temperature_celsius,
            "memory_total_gb": total_memory_gb,
            "memory_available_gb": available_memory_gb,
            "wifi_connected": bool(ssid),
            "wifi_ssid": ssid or None,
            "disk_free_gb": free_disk_gb,
            "disk_used_percent": disk_used_percent,
        }

        spoken_parts = []
        if temperature_celsius is not None:
            spoken_parts.append(f"temperature is {temperature_celsius} degrees Celsius")
        if cpu_percent is not None:
            spoken_parts.append(f"CPU is {cpu_percent} percent used")
        if available_memory_gb is not None and total_memory_gb is not None:
            spoken_parts.append(f"memory has {available_memory_gb} gigabytes available")
        spoken_parts.append(f"disk is {disk_used_percent} percent used")
        spoken_parts.append(f"Wi-Fi is connected to {ssid}" if ssid else "Wi-Fi is not connected")

        _emit_success(metric, "DevKit snapshot: " + ", ".join(spoken_parts) + ".", data)
    except Exception as error:
        log.exception("Unhandled error in get_all_stats")
        _emit_error(metric, "all_stats_error", str(error), "I couldn't gather the DevKit snapshot.")


FUNCTION_REGISTRY = {
    "get_cpu": get_cpu,
    "get_memory": get_memory,
    "get_temperature": get_temperature,
    "get_uptime": get_uptime,
    "get_wifi": get_wifi,
    "get_disk": get_disk,
    "get_health": get_health,
    "get_all_stats": get_all_stats,
}


def main():
    if len(sys.argv) < 2:
        _emit_error("dispatch", "missing_function", "No function name was provided.", "No DevKit function was provided.")
        sys.exit(1)

    function_name = sys.argv[1]
    function_args = sys.argv[2:]
    function = FUNCTION_REGISTRY.get(function_name)

    if function is None:
        _emit_error(
            "dispatch",
            "unknown_function",
            f"Unknown function: {function_name}",
            "The requested DevKit function is not available.",
        )
        sys.exit(1)

    try:
        function(*function_args)
    except TypeError as error:
        log.exception("Invalid arguments for %s", function_name)
        _emit_error(
            function_name,
            "invalid_arguments",
            str(error),
            "The DevKit function received invalid arguments.",
        )
        sys.exit(1)
    except Exception as error:
        log.exception("Unhandled error while running %s", function_name)
        _emit_error(
            function_name,
            "unhandled_error",
            str(error),
            "The DevKit function failed unexpectedly.",
        )
        sys.exit(1)


if __name__ == "__main__":
    main()