Local Connect

python3 --version

OPENHOME_API_KEY = "your_api_key_here"

python3 local_client.py

User speaks → OpenHome STT → Ability receives transcription
                                        ↓
                          LLM converts to terminal command
                                        ↓
                            exec_local_command(command)
                                        ↓
                            local_client.py (WebSocket)
                                        ↓
                          Executes via subprocess.run
                                        ↓
                 Response ← LLM formats for voice ← Ability

system_prompt = """
You are a terminal command generator. Your ONLY purpose is to convert user
requests into valid shell commands.

Rules:
- Respond ONLY with the terminal command, nothing else
- Use POSIX-compatible commands (bash/zsh)
- Do not include explanations, quotes, or markdown formatting
- Do not use sudo unless absolutely necessary
- Make sure commands are safe and won't harm the system

Examples:
User: "list all files" -> ls -la
User: "show current directory" -> pwd
User: "find python files" -> find . -name "*.py"
User: "check disk space" -> df -h
"""

async def first_function(self):
    user_inquiry = await self.capability_worker.wait_for_complete_transcription()

    # Convert natural language → shell command
    command = self.capability_worker.text_to_text_response(
        user_inquiry, history=[], system=system_prompt,
    ).strip()

    await self.capability_worker.speak(f"Running {command}")
    response = await self.capability_worker.exec_local_command(command)

    # Convert raw output → spoken response
    spoken = self.capability_worker.text_to_text_response(
        f"Format this for voice. Command: {command}\nOutput: {response}",
        history=[],
        system="Convert terminal output to a single spoken sentence under 15 words.",
    )
    await self.capability_worker.speak(spoken)
    self.capability_worker.resume_normal_flow()

async def exec_local_command(
    self,
    command: str | dict,
    target_id: str | None = None,
    timeout: float = 10.0,
)

system_prompt = """
Convert git operations to commands. POSIX shell.

Examples:
"check git status" -> git status
"commit changes" -> git add . && git commit -m "Update"
"push to main" -> git push origin main
"create branch feature-x" -> git checkout -b feature-x
"""

async def first_function(self):
    user_inquiry = await self.capability_worker.wait_for_complete_transcription()

    if "system health" in user_inquiry.lower():
        # Pick commands by OS — detect via Python in the client
        metrics = {
            "CPU":     "top -l 1 | grep 'CPU usage'",     # macOS
            "Memory":  "vm_stat | head -n 10",            # macOS
            "Disk":    "df -h /",
            "Battery": "pmset -g batt",                   # macOS
        }
        report = []
        for name, cmd in metrics.items():
            response = await self.capability_worker.exec_local_command(cmd)
            report.append(f"{name}: {response}")
        await self.capability_worker.speak(". ".join(report))
        self.capability_worker.resume_normal_flow()

async def first_function(self):
    user_inquiry = await self.capability_worker.wait_for_complete_transcription()

    if "start dev environment" in user_inquiry.lower():
        await self.capability_worker.speak("Starting development environment...")
        commands = [
            "cd ~/Projects/my-app",
            "code .",                       # Opens VS Code
            "npm run dev &",                # Background dev server
            "open http://localhost:3000",   # macOS — use `start` on Windows, `xdg-open` on Linux
        ]
        for cmd in commands:
            await self.capability_worker.exec_local_command(cmd, timeout=15.0)
        await self.capability_worker.speak("Development environment is ready!")
        self.capability_worker.resume_normal_flow()

system_prompt = """
File management commands for POSIX systems.

Examples:
"organize downloads by type" ->
  mkdir -p ~/Downloads/Images ~/Downloads/Documents &&
  mv ~/Downloads/*.jpg ~/Downloads/Images/ 2>/dev/null || true &&
  mv ~/Downloads/*.pdf ~/Downloads/Documents/ 2>/dev/null || true

"find large files" -> find ~ -type f -size +100M
"""

# In local_client.py
def execute_command(command):
    if command == "get_battery":
        result = subprocess.run(['pmset', '-g', 'batt'], capture_output=True)
        return parse_battery_output(result.stdout)
    result = subprocess.run(command, shell=True, capture_output=True)
    return result.stdout.decode()

import logging
logging.basicConfig(filename='openhome_commands.log', level=logging.INFO)

def execute_command(command):
    logging.info(f"Executing: {command}")
    result = subprocess.run(command, shell=True, capture_output=True)
    logging.info(f"Result: {result.stdout[:100]}...")
    return result.stdout.decode()

BLOCKED_COMMANDS = ['rm -rf /', 'sudo', 'format', 'dd if=']

def execute_command(command):
    if any(blocked in command for blocked in BLOCKED_COMMANDS):
        return "ERROR: Blocked command for safety"
    result = subprocess.run(command, shell=True, capture_output=True)
    return result.stdout.decode()

ls -la ~/Documents   # test in terminal
                      # verify output matches expectations
                      # then add to the Ability

SAFE_COMMANDS = ['ls', 'pwd', 'cd', 'cat', 'grep', 'find', 'df', 'du']

if not any(safe in terminal_command for safe in SAFE_COMMANDS):
    await self.capability_worker.speak("This command needs confirmation.")
    # ... confirmation logic

# Quick commands: default is fine
response = await self.capability_worker.exec_local_command("pwd")

# Long-running: bump up
response = await self.capability_worker.exec_local_command(
    "find / -name '*.log'",
    timeout=60.0,
)

try:
    response = await self.capability_worker.exec_local_command(terminal_command)
    if "error" in response.lower() or "permission denied" in response.lower():
        await self.capability_worker.speak("That command failed. Try a different approach?")
    else:
        # process success
        ...
except asyncio.TimeoutError:
    await self.capability_worker.speak("That took too long. It might still be running.")
except Exception as e:
    self.worker.editor_logging_handler.error(f"Command failed: {e}")
    await self.capability_worker.speak("Something went wrong.")
finally:
    self.capability_worker.resume_normal_flow()

ALLOWED_COMMANDS = ['ls', 'pwd', 'df', 'du', 'cat', 'grep', 'find']

if not any(cmd in terminal_command for cmd in ALLOWED_COMMANDS):
    await self.capability_worker.speak("That command is not allowed.")
    return

# Check what commands were run
grep "Executing:" /tmp/openhome_client.log

# Alert on dangerous patterns
tail -f /tmp/openhome_client.log | grep -E "rm|sudo|shutdown"

Voice Input → OpenHome Ability → exec_local_command()
                                        ↓
                                  local_client.py
                                  (WebSocket connection)
                                        ↓
                                 Terminal execution
                                  (subprocess.run)
                                        ↓
                        Response ← AI formatting ← Template