Local Connect

curl -fsSL https://app.openhome.com/install.sh | sh

openhome login

openhome local start        # start the bridge in the background
openhome local status       # confirm it's running
openhome local logs         # stream requests and replies live (Ctrl-C to stop)
openhome local stop         # stop it

Voice Input → OpenHome Ability (main.py)
                      ↓ exec_local_command()
              OpenHome Local Link (openhome CLI, your PC)
                      ↓
              local-link handler (raw shell executor)
                      ↓
                subprocess.run
                      ↓
        Response ← Ability speaks the reply

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 — branch on platform.system() in the Ability
        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
"""

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

async def first_function(self):
    ...
    if any(blocked in command for blocked in BLOCKED_COMMANDS):
        await self.capability_worker.speak("That command is blocked for safety.")
        self.capability_worker.resume_normal_flow()
        return
    response = await self.capability_worker.exec_local_command(command)

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

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

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

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