概要
このページは、カスタム実行ノードの完全なプロダクション対応の例を提供します。各例は異なる 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 コミュニティ