快速開始

概述

在此快速開始指南中，您將建立一個簡單的WiFi 控制器執行節點，連接到 Qaptly Desktop 並執行 WiFi 命令。此示例在實作中演示了核心 UPP 概念。

前置需求

Qaptly Desktop 0.15.0 版或更新版本
Python 3.8+ 已安裝
Python 和 JSON 的基本知識

步驟 1：安裝相依性

首先，安裝 WebSocket 程式庫：

pip install websockets

步驟 2：建立執行節點

建立名為 wifi_pilot.py 的檔案：

重要: 將 <qaptly_desktop_ip> 替換為執行 Qaptly Desktop 的機器在區域網路上的實際 IP 位址（例如：192.168.1.100）。您可以在網路設定中找到此 IP 位址，或執行 ifconfig（macOS/Linux）或 ipconfig（Windows）命令查看。

import asyncio
import json
import uuid
from datetime import datetime
import websockets

class WiFiPilot:
    def __init__(self):
        self.pilot_id = "wifi-controller-001"
        self.websocket = None

    async def connect(self):
        """Connect to Qaptly Desktop"""
        uri = "ws://<qaptly_desktop_ip>:18765/upp"
        print(f"Connecting to {uri}...")
        self.websocket = await websockets.connect(uri)
        print("Connected!")

    async def send_message(self, msg_type, payload):
        """Send a 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))
        return message

    async def register(self):
        """Register this pilot with Qaptly"""
        payload = {
            "pilot": {
                "id": self.pilot_id,
                "name": "WiFi Controller",
                "version": "1.0.0",
                "vendor": "My Company",
                "description": "Controls WiFi networks for testing"
            },
            "auth": {},  # Empty for first-time connection
            "capabilities": [
                {
                    "action": "wifi.scan",
                    "name": "Scan WiFi Networks",
                    "description": "Scans for available WiFi networks",
                    "params": {
                        "type": "object",
                        "properties": {},
                        "required": []
                    },
                    "returns": {
                        "type": "object",
                        "properties": {
                            "networks": {
                                "type": "array",
                                "items": {
                                    "type": "object",
                                    "properties": {
                                        "ssid": {"type": "string"},
                                        "signal": {"type": "number"}
                                    }
                                }
                            }
                        }
                    },
                    "events": ["wifi.scan_complete"]
                },
                {
                    "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"}
                        }
                    },
                    "events": []
                }
            ]
        }

        await self.send_message("register", payload)
        print("Registration sent, waiting for approval...")

    async def handle_command(self, command_id, action, params):
        """Handle a command from Qaptly"""
        print(f"Executing: {action} with params: {params}")

        if action == "wifi.scan":
            # Simulate WiFi scan
            result = {
                "networks": [
                    {"ssid": "HomeWiFi", "signal": -45},
                    {"ssid": "GuestNetwork", "signal": -72}
                ]
            }
            return {"success": True, "data": result}

        elif action == "wifi.status":
            # Simulate WiFi status
            result = {
                "connected": True,
                "ssid": "HomeWiFi"
            }
            return {"success": True, "data": result}

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

    async def send_heartbeat(self):
        """Send heartbeat every 30 seconds"""
        while True:
            await asyncio.sleep(30)
            await self.send_message("heartbeat", {})
            print("❤️  Heartbeat sent")

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

                print(f"\n📨 Received: {msg_type}")

                if msg_type == "register_response":
                    if payload.get("success"):
                        print("✅ Registration successful!")
                        if "assignedKey" in payload:
                            print(f"🔑 Save this key: {payload['assignedKey']}")
                            print("   (Use it in 'auth.key' for future connections)")
                    elif payload.get("pending"):
                        print("⏳ Waiting for user approval in Qaptly Desktop...")
                    else:
                        error = payload.get("error", {})
                        print(f"❌ Registration failed: {error.get('message')}")

                elif msg_type == "command":
                    command_id = payload.get("commandId")
                    action = payload.get("action")
                    params = payload.get("params", {})

                    # Execute command
                    response_payload = await self.handle_command(command_id, action, params)
                    response_payload["commandId"] = command_id

                    # Send response
                    await self.send_message("response", response_payload)
                    print(f"✅ Response sent for {action}")

                elif msg_type == "subscribe":
                    events = payload.get("subscribe", [])
                    print(f"📡 Qaptly subscribed to events: {events}")

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

    async def run(self):
        """Main run loop"""
        await self.connect()
        await self.register()

        # Start heartbeat task
        heartbeat_task = asyncio.create_task(self.send_heartbeat())

        # Listen for messages
        await self.listen()

        # Cancel heartbeat when done
        heartbeat_task.cancel()

if __name__ == "__main__":
    pilot = WiFiPilot()
    asyncio.run(pilot.run())

步驟 3：啟動 Qaptly Desktop

啟動 Qaptly Desktop 應用程式
UPP 服務會自動在連接埠 18765 上啟動

步驟 4：執行執行節點

執行執行節點指令碼：

python wifi_pilot.py

您應該會看到：

Connecting to ws://<qaptly_desktop_ip>:18765/upp...
Connected!
Registration sent, waiting for approval...

步驟 5：批准連接

Qaptly Desktop 中將出現對話方塊：

┌────────────────────────────────────────┐
│  New Pilot Connection Request         │
├────────────────────────────────────────┤
│  Name: WiFi Controller                 │
│  ID: wifi-controller-001               │
│  Vendor: My Company                    │
│                                        │
│  Capabilities:                         │
│  • wifi.scan - Scan WiFi Networks      │
│  • wifi.status - Get WiFi Status       │
│                                        │
│  [Reject]  [Approve]                   │
└────────────────────────────────────────┘

點擊批准。

步驟 6：測試連接

執行節點應該會收到：

✅ Registration successful!
🔑 Save this key: abc123-generated-key-xyz789
   (Use it in 'auth.key' for future connections)
❤️  Heartbeat sent

步驟 7：執行命令

您現在可以從 Qaptly 執行命令。執行節點將接收命令並自動回應。例如，如果 Qaptly 發送 wifi.scan 命令，您將看到：

📨 Received: command
Executing: wifi.scan with params: {}
✅ Response sent for wifi.scan

下一次連接(使用金鑰)

對於後續連接，請使用指派的金鑰：

async def register(self):
    payload = {
        "pilot": {
            "id": self.pilot_id,
            "name": "WiFi Controller",
            "version": "1.0.0"
        },
        "auth": {
            "key": "abc123-generated-key-xyz789"  # ← 使用首次連接中的金鑰
        },
        "capabilities": [...]
    }
    await self.send_message("register", payload)

使用有效的金鑰，您的執行節點將自動批准，無需使用者互動。

展示的關鍵概念

此示例展示了： ✅ WebSocket 連接 - 連接到 ws://<qaptly_desktop_ip>:18765/upp ✅ 訊息格式 - 所有訊息遵循 UPP 結構 ✅ 註冊 - 宣告執行節點資訊和功能 ✅ 認證 - 首次批准 + 基於金鑰的認證 ✅ 命令處理 - 接收和回應命令 ✅ 心跳 - 每 30 秒保持連接活躍 ✅ JSON Schema - 為每個功能定義參數和返回值

疑難排解

”Connection refused”

確保 Qaptly Desktop 正在運行
檢查防火牆未阻止連接埠 18765
驗證您是否連接到 <qaptly_desktop_ip> (localhost)

“Registration failed: AUTH_FAILED”

您的金鑰可能無效或已過期
嘗試不使用金鑰連接(首次流程)
檢查金鑰字串中是否有拼寫錯誤

心跳逾時

執行節點必須每 30 秒發送一次心跳
90 秒內沒有任何訊息後連接將關閉

下一步

現在您已擁有可運作的執行節點：

新增實際 WiFi 控制 - 將模擬替換為實際網路命令
實現更多功能 - 新增 wifi.connect、wifi.disconnect
發送事件 - 通知 Qaptly 有關 WiFi 狀態變更
閱讀協議 - 了解所有訊息類型和流程

協議參考

完整的訊息規範

功能指南

使用 JSON Schema 定義複雜功能

認證

安全性模型和金鑰管理

範例

更完整的實作

Android Pilot

iOS Pilot

自訂 Pilots (UPP)

概述

前置需求

步驟 1：安裝相依性

步驟 2：建立執行節點

步驟 3：啟動 Qaptly Desktop

步驟 4：執行執行節點

步驟 5：批准連接

步驟 6：測試連接

步驟 7：執行命令

下一次連接(使用金鑰)

展示的關鍵概念

疑難排解

”Connection refused”

“Registration failed: AUTH_FAILED”

心跳逾時

下一步

協議參考

功能指南

認證

範例

​概述

​前置需求

​步驟 1：安裝相依性

​步驟 2：建立執行節點

​步驟 3：啟動 Qaptly Desktop

​步驟 4：執行執行節點

​步驟 5：批准連接

​步驟 6：測試連接

​步驟 7：執行命令

​下一次連接(使用金鑰)

​展示的關鍵概念

​疑難排解

​”Connection refused”

​“Registration failed: AUTH_FAILED”

​心跳逾時

​下一步

協議參考

功能指南

認證

範例

概述

前置需求

步驟 1：安裝相依性

步驟 2：建立執行節點

步驟 3：啟動 Qaptly Desktop

步驟 4：執行執行節點

步驟 5：批准連接

步驟 6：測試連接

步驟 7：執行命令

下一次連接(使用金鑰)

展示的關鍵概念

疑難排解

”Connection refused”

“Registration failed: AUTH_FAILED”

心跳逾時

下一步