示例 - Qaptly Docs

概述

此頁面提供自訂執行節點的完整、生產級別範例。每個範例展示不同的 UPP 功能和最佳實踐。

範例 1: Raspberry Pi WiFi 控制器

具有實際網路操作的完整 WiFi 控制器執行節點。

功能

✅ 實際 WiFi 掃描和連接
✅ 錯誤處理
✅ 事件通知
✅ 金鑰持久性
✅ 自動重新連接

前置條件

pip install websockets nmcli

完整實現

#!/usr/bin/env python3
"""
Raspberry Pi WiFi Controller Pilot
Requires: nmcli (NetworkManager CLI tool)
"""

import asyncio
import json
import uuid
import os
import subprocess
from datetime import datetime
from pathlib import Path
import websockets

class WiFiController:
    """Handles actual WiFi operations using nmcli"""

    @staticmethod
    def scan():
        """Scan for WiFi networks"""
        try:
            result = subprocess.run(
                ["nmcli", "-t", "-f", "SSID,SIGNAL,SECURITY", "dev", "wifi", "list"],
                capture_output=True,
                text=True,
                timeout=15
            )

            networks = []
            for line in result.stdout.strip().split("\n"):
                if line:
                    ssid, signal, security = line.split(":")
                    if ssid:  # Skip empty SSIDs
                        networks.append({
                            "ssid": ssid,
                            "signal": -100 + int(signal),  # Convert to dBm
                            "security": security if security else "open"
                        })

            return {"success": True, "networks": networks}
        except Exception as e:
            return {"success": False, "error": str(e)}

    @staticmethod
    def connect(ssid, password=None):
        """Connect to WiFi network"""
        try:
            cmd = ["nmcli", "dev", "wifi", "connect", ssid]
            if password:
                cmd.extend(["password", password])

            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)

            if result.returncode == 0:
                # Get IP address
                ip_result = subprocess.run(
                    ["hostname", "-I"],
                    capture_output=True,
                    text=True
                )
                ip = ip_result.stdout.strip().split()[0] if ip_result.stdout else "unknown"

                return {"success": True, "connected": True, "ip": ip}
            else:
                return {
                    "success": False,
                    "error": f"Connection failed: {result.stderr}"
                }

        except subprocess.TimeoutExpired:
            return {"success": False, "error": "Connection timeout"}
        except Exception as e:
            return {"success": False, "error": str(e)}

    @staticmethod
    def status():
        """Get current WiFi status"""
        try:
            # Get active connection
            result = subprocess.run(
                ["nmcli", "-t", "-f", "NAME,TYPE,DEVICE", "connection", "show", "--active"],
                capture_output=True,
                text=True
            )

            for line in result.stdout.strip().split("\n"):
                if "wireless" in line:
                    name, _, device = line.split(":")

                    # Get signal strength
                    signal_result = subprocess.run(
                        ["nmcli", "-t", "-f", "IN-USE,SIGNAL,SSID", "dev", "wifi"],
                        capture_output=True,
                        text=True
                    )

                    for wifi_line in signal_result.stdout.strip().split("\n"):
                        if wifi_line.startswith("*"):
                            _, signal, ssid = wifi_line.split(":")
                            return {
                                "success": True,
                                "connected": True,
                                "ssid": ssid,
                                "signal": -100 + int(signal),
                                "device": device
                            }

            return {"success": True, "connected": False}

        except Exception as e:
            return {"success": False, "error": str(e)}

class WiFiPilot:
    """UPP WiFi Pilot Implementation"""

    def __init__(self):
        self.pilot_id = "rpi-wifi-controller-001"
        self.websocket = None
        self.wifi = WiFiController()
        self.config_dir = Path.home() / ".qaptly_pilot"
        self.config_file = self.config_dir / "config.json"
        self.auth_key = None
        self.subscribed_events = set()
        self.load_config()

    def load_config(self):
        """Load authentication key from config"""
        self.config_dir.mkdir(exist_ok=True)

        if self.config_file.exists():
            with open(self.config_file, "r") as f:
                config = json.load(f)
                self.auth_key = config.get("auth_key")
                print(f"✅ Loaded config from {self.config_file}")

    def save_config(self, auth_key):
        """Save authentication key"""
        config = {
            "pilot_id": self.pilot_id,
            "auth_key": auth_key,
            "updated_at": datetime.utcnow().isoformat() + "Z"
        }

        with open(self.config_file, "w") as f:
            json.dump(config, f, indent=2)

        # Secure permissions
        self.config_file.chmod(0o600)
        print(f"💾 Saved config to {self.config_file}")

    async def send_message(self, msg_type, payload):
        """Send UPP message"""
        message = {
            "id": str(uuid.uuid4()),
            "protocolVersion": "1.0.0",
            "type": msg_type,
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "payload": payload
        }
        await self.websocket.send(json.dumps(message))

    async def send_event(self, event_name, data):
        """Send event if subscribed"""
        if event_name in self.subscribed_events or "*" in self.subscribed_events:
            await self.send_message("event", {
                "event": event_name,
                "data": data
            })
            print(f"📡 Sent event: {event_name}")

    async def register(self):
        """Register pilot with Qaptly"""
        payload = {
            "pilot": {
                "id": self.pilot_id,
                "name": "Raspberry Pi WiFi Controller",
                "version": "1.0.0",
                "vendor": "Qaptly",
                "description": "Controls WiFi on Raspberry Pi for network testing"
            },
            "auth": {"key": self.auth_key} if self.auth_key else {},
            "capabilities": [
                {
                    "action": "wifi.scan",
                    "name": "Scan WiFi Networks",
                    "description": "Scans for available WiFi networks and returns SSID, signal strength, and security type",
                    "params": {
                        "type": "object",
                        "properties": {},
                        "required": []
                    },
                    "returns": {
                        "type": "object",
                        "properties": {
                            "networks": {
                                "type": "array",
                                "items": {
                                    "type": "object",
                                    "properties": {
                                        "ssid": {"type": "string"},
                                        "signal": {"type": "number"},
                                        "security": {"type": "string"}
                                    }
                                }
                            }
                        }
                    },
                    "events": ["wifi.scan_complete"]
                },
                {
                    "action": "wifi.connect",
                    "name": "Connect to WiFi",
                    "description": "Connects to a specified WiFi network",
                    "params": {
                        "type": "object",
                        "properties": {
                            "ssid": {
                                "type": "string",
                                "description": "Network name"
                            },
                            "password": {
                                "type": "string",
                                "description": "Network password (optional for open networks)"
                            }
                        },
                        "required": ["ssid"]
                    },
                    "returns": {
                        "type": "object",
                        "properties": {
                            "connected": {"type": "boolean"},
                            "ip": {"type": "string"}
                        }
                    },
                    "events": ["wifi.connected", "wifi.connection_failed"]
                },
                {
                    "action": "wifi.status",
                    "name": "Get WiFi Status",
                    "description": "Returns current WiFi connection status",
                    "params": {
                        "type": "object",
                        "properties": {},
                        "required": []
                    },
                    "returns": {
                        "type": "object",
                        "properties": {
                            "connected": {"type": "boolean"},
                            "ssid": {"type": "string"},
                            "signal": {"type": "number"}
                        }
                    },
                    "events": []
                }
            ]
        }

        await self.send_message("register", payload)
        print("📝 Registration sent")

    async def handle_command(self, command_id, action, params):
        """Execute command and return response"""
        print(f"⚙️  Executing: {action}")

        try:
            if action == "wifi.scan":
                result = self.wifi.scan()
                if result.get("success"):
                    await self.send_event("wifi.scan_complete", {
                        "count": len(result.get("networks", []))
                    })
                return {
                    "commandId": command_id,
                    "success": True,
                    "data": {"networks": result.get("networks", [])}
                }

            elif action == "wifi.connect":
                ssid = params.get("ssid")
                password = params.get("password")
                result = self.wifi.connect(ssid, password)

                if result.get("success"):
                    await self.send_event("wifi.connected", {
                        "ssid": ssid,
                        "ip": result.get("ip")
                    })
                    return {
                        "commandId": command_id,
                        "success": True,
                        "data": result
                    }
                else:
                    await self.send_event("wifi.connection_failed", {
                        "ssid": ssid,
                        "reason": result.get("error")
                    })
                    return {
                        "commandId": command_id,
                        "success": False,
                        "error": {
                            "code": "wifi.CONNECTION_FAILED",
                            "message": result.get("error")
                        }
                    }

            elif action == "wifi.status":
                result = self.wifi.status()
                if result.get("success"):
                    return {
                        "commandId": command_id,
                        "success": True,
                        "data": {
                            "connected": result.get("connected", False),
                            "ssid": result.get("ssid", ""),
                            "signal": result.get("signal", 0)
                        }
                    }
                else:
                    return {
                        "commandId": command_id,
                        "success": False,
                        "error": {
                            "code": "wifi.STATUS_ERROR",
                            "message": result.get("error")
                        }
                    }

            else:
                return {
                    "commandId": command_id,
                    "success": False,
                    "error": {
                        "code": "UNKNOWN_ACTION",
                        "message": f"Unknown action: {action}"
                    }
                }

        except Exception as e:
            return {
                "commandId": command_id,
                "success": False,
                "error": {
                    "code": "EXECUTION_ERROR",
                    "message": str(e)
                }
            }

    async def heartbeat_loop(self):
        """Send heartbeat every 30 seconds"""
        try:
            while True:
                await asyncio.sleep(30)
                await self.send_message("heartbeat", {})
                print("❤️  Heartbeat")
        except asyncio.CancelledError:
            pass

    async def listen(self):
        """Listen for messages from Qaptly"""
        try:
            async for message in self.websocket:
                data = json.loads(message)
                msg_type = data.get("type")
                payload = data.get("payload", {})

                if msg_type == "register_response":
                    if payload.get("success"):
                        print("✅ Registration successful")
                        assigned_key = payload.get("assignedKey")
                        if assigned_key:
                            self.save_config(assigned_key)
                            self.auth_key = assigned_key
                    elif payload.get("pending"):
                        print("⏳ Waiting for user approval...")
                    else:
                        error = payload.get("error", {})
                        print(f"❌ Registration failed: {error.get('message')}")
                        if error.get("code") == "AUTH_FAILED":
                            self.config_file.unlink(missing_ok=True)
                            print("🔄 Deleted invalid config, restart to retry")
                            return

                elif msg_type == "command":
                    response = await self.handle_command(
                        payload.get("commandId"),
                        payload.get("action"),
                        payload.get("params", {})
                    )
                    await self.send_message("response", response)
                    print("✅ Response sent")

                elif msg_type == "subscribe":
                    subscribe = payload.get("subscribe", [])
                    unsubscribe = payload.get("unsubscribe", [])

                    for event in subscribe:
                        self.subscribed_events.add(event)

                    for event in unsubscribe:
                        self.subscribed_events.discard(event)

                    print(f"📡 Subscriptions updated: {self.subscribed_events}")

        except websockets.exceptions.ConnectionClosed:
            print("❌ Connection closed")

    async def run(self):
        """Main run loop"""
        uri = "ws://<qaptly_desktop_ip>:18765/upp"

        while True:
            try:
                print(f"🔌 Connecting to {uri}...")
                async with websockets.connect(uri) as websocket:
                    self.websocket = websocket
                    print("✅ Connected")

                    await self.register()

                    # Start heartbeat
                    heartbeat_task = asyncio.create_task(self.heartbeat_loop())

                    # Listen for messages
                    await self.listen()

                    # Cancel heartbeat
                    heartbeat_task.cancel()
                    try:
                        await heartbeat_task
                    except asyncio.CancelledError:
                        pass

            except Exception as e:
                print(f"❌ Error: {e}")
                print("🔄 Reconnecting in 5 seconds...")
                await asyncio.sleep(5)

if __name__ == "__main__":
    pilot = WiFiPilot()
    try:
        asyncio.run(pilot.run())
    except KeyboardInterrupt:
        print("\n👋 Shutting down...")

執行範例

# 儲存為 wifi_pilot.py
chmod +x wifi_pilot.py

# 執行
./wifi_pilot.py

範例 2: GPIO 控制器 (Python)

控制 Raspberry Pi GPIO 腳位。

實現重點

import RPi.GPIO as GPIO

class GPIOPilot:
    def __init__(self):
        GPIO.setmode(GPIO.BCM)
        GPIO.setwarnings(False)

    def gpio_write(self, pin, state):
        """Set GPIO pin state"""
        GPIO.setup(pin, GPIO.OUT)
        GPIO.output(pin, GPIO.HIGH if state == "HIGH" else GPIO.LOW)

        # Send event
        await self.send_event("gpio.state_changed", {
            "pin": pin,
            "state": state
        })

        return {"success": True, "pin": pin, "state": state}

    # Register capability
    {
        "action": "gpio.write",
        "name": "Write GPIO Pin",
        "params": {
            "type": "object",
            "properties": {
                "pin": {
                    "type": "number",
                    "minimum": 0,
                    "maximum": 27
                },
                "state": {
                    "type": "string",
                    "enum": ["HIGH", "LOW"]
                }
            },
            "required": ["pin", "state"]
        },
        "events": ["gpio.state_changed"]
    }

範例 3: 感應器監視器 (Node.js)

監視溫度感應器並發送事件。

實現重點

const WebSocket = require('ws');
const { v4: uuidv4 } = require('uuid');

class SensorPilot {
  constructor() {
    this.ws = null;
    this.subscriptions = new Set();
    this.lastTemperature = null;
  }

  async startMonitoring() {
    // Monitor temperature every 5 seconds
    setInterval(async () => {
      const temp = await this.readTemperature();

      // Send event if changed significantly
      if (Math.abs(temp - this.lastTemperature) > 0.5) {
        await this.sendEvent('sensor.temperature_changed', {
          temperature: temp,
          unit: 'celsius',
          timestamp: new Date().toISOString()
        });

        this.lastTemperature = temp;
      }
    }, 5000);
  }

  async readTemperature() {
    // Read from sensor (DHT22, DS18B20, etc.)
    // Implementation depends on your hardware
    return 23.5;
  }

  // Capabilities
  capabilities = [
    {
      action: 'sensor.read_temperature',
      name: 'Read Temperature',
      params: {
        type: 'object',
        properties: {
          unit: {
            type: 'string',
            enum: ['celsius', 'fahrenheit'],
            default: 'celsius'
          }
        }
      },
      returns: {
        type: 'object',
        properties: {
          temperature: { type: 'number' },
          unit: { type: 'string' },
          timestamp: { type: 'string' }
        }
      },
      events: ['sensor.temperature_changed']
    }
  ];
}

測試您的執行節點

手動測試

# 1. 啟動 Qaptly Desktop

# 2. 執行您的執行節點
python wifi_pilot.py

# 3. 在 Qaptly 中核准

# 4. 測試命令 (從 Qaptly 或測試指令碼)

自動化測試

建立測試指令碼:

import asyncio
import websockets
import json

async def test_pilot():
    uri = "ws://<qaptly_desktop_ip>:18765/upp"

    async with websockets.connect(uri) as ws:
        # Wait for pilot to register
        await asyncio.sleep(2)

        # Send test command
        command = {
            "id": "test-001",
            "protocolVersion": "1.0.0",
            "type": "command",
            "timestamp": "2026-01-21T10:00:00Z",
            "payload": {
                "commandId": "cmd-001",
                "action": "wifi.status",
                "params": {}
            }
        }

        await ws.send(json.dumps(command))

        # Wait for response
        response = await ws.recv()
        print(f"Response: {response}")

asyncio.run(test_pilot())

GitHub 儲存庫

完整的範例儲存庫

快速開始

簡單的入門範例

協議參考

完整的協議文件

功能

JSON Schema 指南

社群範例

與社群分享您的執行節點! 將您的範例提交至:

GitHub: qaptly-upp-examples
Discord: Qaptly 社群

Android Pilot

iOS Pilot

自訂 Pilots (UPP)

示例

概述

範例 1: Raspberry Pi WiFi 控制器

功能

前置條件

完整實現

執行範例

範例 2: GPIO 控制器 (Python)

實現重點

範例 3: 感應器監視器 (Node.js)

實現重點

測試您的執行節點

手動測試

自動化測試

更多範例

GitHub 儲存庫

快速開始

協議參考

功能

社群範例

​概述

​範例 1: Raspberry Pi WiFi 控制器

​功能

​前置條件

​完整實現

​執行範例

​範例 2: GPIO 控制器 (Python)

​實現重點

​範例 3: 感應器監視器 (Node.js)

​實現重點

​測試您的執行節點

​手動測試

​自動化測試

​更多範例

GitHub 儲存庫

快速開始

協議參考

功能

​社群範例

概述

範例 1: Raspberry Pi WiFi 控制器

功能

前置條件

完整實現

執行範例

範例 2: GPIO 控制器 (Python)

實現重點

範例 3: 感應器監視器 (Node.js)

實現重點

測試您的執行節點

手動測試

自動化測試

更多範例

社群範例