クイックスタート

pip install websockets

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

python wifi_pilot.py

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

┌────────────────────────────────────────┐
│  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]                   │
└────────────────────────────────────────┘

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

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