Local Abilities

A special Ability type that runs on the OpenHome DevKit and can use hardware, connected peripherals, the file system, shell commands, and the device’s Python environment.

Local Abilities are a specialized Ability type for running DevKit-side code from an OpenHome Ability. Unlike other Ability types, which operate only within the standard Ability runtime, Local Abilities can use the DevKit hardware, system resources, and the Python environment installed on the device.

This includes Python imports that are restricted in the standard runtime, file system operations, shell commands, hardware access such as GPIO pins, sensors, LEDs, and connected peripherals, and system-level data such as CPU, memory, temperature, and network state.

Use Local Abilities for IoT projects, custom hardware integrations, voice-controlled physical devices, device telemetry, long-running on-device tasks, and any use case that requires direct interaction with the DevKit or capabilities beyond the standard Ability runtime.

Local Abilities only run on actual OpenHome DevKit hardware. They do not run in the web Live Editor’s simulated environment.

How It Works

A Local Ability is split between the standard Ability runtime and DevKit-side execution. main.py handles the Agent flow, while devkit_functions.py runs hardware, system, and device-level code on the OpenHome DevKit.

File Structure

File	Runtime	Use for
main.py	Standard Ability runtime	Voice interaction, prompts, conversation state, SDK calls, and calls to DevKit-side functions.
devkit_functions.py	OpenHome DevKit	Hardware control, connected peripherals, system operations, shell commands, system telemetry, ambient intelligence workflows, and DevKit-side Python packages.
requirements.txt	OpenHome DevKit	Python dependencies installed for devkit_functions.py.

The DevKit-side file must be named exactly devkit_functions.py. No other filename will be picked up by the platform.

Packages listed in requirements.txt are installed for devkit_functions.py on the OpenHome DevKit. They are not available in the standard Ability runtime where main.py runs.

Calling DevKit Functions

Use send_devkit_capability_action() in main.py to run a registered function from devkit_functions.py.

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

Parameter	Type	Description
function_name	str	Name of the function registered in devkit_functions.py.
args	list[str]	Arguments passed to the DevKit function. Values are passed as strings; cast them inside devkit_functions.py when another type is required.
timeout	int	Maximum number of seconds to wait for the function to complete.
capability_name	str (optional)	Name of another installed Ability whose devkit_functions.py should handle the call. Omit to use the current Ability.

devkit_functions.py Execution Flow

devkit_functions.py runs on the OpenHome DevKit as a Python script. Functions that should be callable from main.py must be registered in FUNCTION_REGISTRY, and the function_name passed from main.py must match one of those registry keys.

devkit_functions.py should include a Python main guard: if __name__ == "__main__". The main guard reads the requested function name and arguments, then runs the matching registered function.

Values in args are passed to the DevKit-side function as strings. Cast them inside devkit_functions.py when the function requires a specific type, such as an integer, boolean, or JSON object.

Use print() for output that should be returned to main.py; standard output is captured in result["output"]. Python return values are not captured by send_devkit_capability_action().

Use web_logger for diagnostics. These logs appear in the DevKit section of the Ability Live Editor and are not returned to main.py.

Response Shape

send_devkit_capability_action() returns an object with the execution status, captured output, and request metadata.

{
    "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
}

output contains the standard output produced during execution. If the function does not print anything, output is None.

error contains the error message when execution fails. Otherwise, it is None.

Logs written with web_logger are separate from the returned object. They appear in the DevKit section of the Ability Live Editor logs and are useful for debugging DevKit-side execution.

Example: Wi-Fi Status

This example reads the DevKit’s current Wi-Fi connection and speaks it back to the user.

devkit_functions.py — runs on the DevKit:

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:])

main.py — runs in the standard Ability runtime:

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())

Calling Functions from Another Local Ability

main.py can also call functions from another installed Local Ability’s devkit_functions.py. This is useful when one Local Ability exposes reusable DevKit-side functions that another Ability needs to use.

capability_name is only needed for cross-Ability calls. When it is omitted, the call uses the current Ability’s devkit_functions.py.

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

To find the name of an installed Ability to use as capability_name, see Installed Abilities later in this document.

Local Abilities in the Live Editor

Select the Local Category

To create a Local Ability, select Local from the Ability categories and choose a template.

If you upload a custom Ability, the project must include devkit_functions.py and requirements.txt.

Advanced DevKit Controls

If your DevKit is online and connected, the Advanced DevKit Controls toggle appears in the Ability Editor. Enable it to expand the Advanced DevKit Controls section.

Once Advanced DevKit Controls are enabled, scroll down and you will see the Advanced DevKit Controls section. From here you can sync your Ability to the DevKit, restart the Agent, and view the DevKit connection status.

Sync Local Abilities with the DevKit

When the DevKit is online and connected, changes saved in the Live Editor are synced to the DevKit automatically.

On save:

• devkit_functions.py or requirements.txt changes are pushed to the DevKit without restarting the Agent. If requirements.txt changed, new dependencies are installed on the DevKit.
• main.py changes are saved to the OpenHome platform, synced with the DevKit sandbox, and the Agent restarts on the DevKit so the latest Ability code is used.

When editing main.py, save after completing the intended change. Each save can restart the Agent on the DevKit while the DevKit is connected.

If the DevKit was offline while you updated a Local Ability:

• main.py changes sync when the DevKit reconnects.
• devkit_functions.py or requirements.txt changes should be synced before testing. After the DevKit reconnects, click Sync Abilities from Advanced DevKit Controls to apply the latest changes.

You can also sync from Advanced DevKit Controls in the Live Editor, or from the OpenHome - Voice AI Devkit App dashboard using the Sync Abilities button .

Logging on the DevKit

Use the DevKit logger inside devkit_functions.py to debug on-device behavior. Messages written with this logger appear in the DevKit section of the Ability Editor logs.

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")

To view the logs, open the DevKit section inside the Ability Editor logs after triggering the Ability on the DevKit.

Installed Abilities

To use functions from another Ability’s devkit_functions.py, you need that Ability’s name to pass in the capability_name parameter. To find it, click the Quick Reference Installed Abilities button in the top-left corner of the Ability Editor.

This opens the installed Local Abilities list. Copy the name of the Local Ability that contains the target devkit_functions.py file and pass it in the capability_name parameter.

Example: DevKit Stats

This is a voice-controlled DevKit telemetry reporter. Users say something like “check cpu” or “how hot is the devkit” and the DevKit reads its system stats and speaks them back.

Trigger words

This example can be triggered with phrases like:

• devkit info
• system info
• how long has my devkit been running

requirements.txt

No third-party packages are required for this example — all stat checks use Python’s standard library and standard Linux interfaces (/proc, /sys, and shell commands like iwgetid, df).

For other Local Abilities that need hardware libraries, list them here. Some common examples:

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

Only the packages you actually import in devkit_functions.py need to go here — they get installed on the DevKit side when you sync.

main.py — standard Ability runtime

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())

devkit_functions.py — DevKit-side telemetry

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()

Interaction Flow

1. Trigger

   The user starts the Ability with a trigger phrase such as “devkit stats” or “check cpu”.

2. Route

   main.py keeps the voice flow in the standard Ability runtime and uses the LLM as a strict router from natural language to a registered DevKit telemetry function.

3. Execute

   main.py calls send_devkit_capability_action() with the selected function name, arguments, and timeout. The matching function runs on the OpenHome DevKit from devkit_functions.py.

4. Return output

   devkit_functions.py reads the requested device data, logs diagnostics with web_logger, and prints a structured JSON payload. The printed payload is captured in result["output"].

5. Respond

   main.py parses result["output"], reads spoken_response, and speaks the result. The structured data field remains available for richer logic.

6. Continue or exit

   The Ability prompts for another stat or exits cleanly. On exit, main.py calls resume_normal_flow() so the Agent returns to its normal flow.

Best practices

Put voice logic in main.py, hardware logic in devkit_functions.py

Clean separation makes both sides easier to debug. Keep devkit_functions.py focused on the hardware work.

Set sensible timeouts

Hardware calls can block. A 5–10 second timeout is typical for lightweight actions; bump to 30 or more for long-running effects or captures.

Use DevKit logs to debug on-device code

Use the DevKit logger web_logger for debugging and inspect messages in the DevKit logs section inside the Ability Editor.

requirements.txt is for the DevKit side only

Packages listed there are installed for devkit_functions.py. They are not available in the sandboxed runtime where main.py runs.

Handle missing hardware gracefully

Not every DevKit has every peripheral. Wrap hardware initialization in try/except and log an informative error instead of crashing — your Ability can still speak a helpful message to the user.

See also

• Ability Types — when Local is the right choice vs. Skill, Brain, or Background Daemon
• Background Abilities — for always-on monitoring that doesn’t need hardware access
• SDK Reference — full method catalog
• Voice-First Best Practices — the UX rules that apply to any Ability, including Local