Examples

Overview

This page provides complete, production-ready examples of custom pilots. Each example demonstrates different UPP features and best practices.

Example 1: Raspberry Pi WiFi Controller

A complete WiFi controller pilot with real network operations.

Features

✅ Real WiFi scanning and connection
✅ Error handling
✅ Event notifications
✅ Key persistence
✅ Automatic reconnection

Prerequisites

pip install websockets nmcli

Complete Implementation

#!/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...")

Running the Example

# Save as wifi_pilot.py
chmod +x wifi_pilot.py

# Run
./wifi_pilot.py

Example 2: GPIO Controller (Python)

Control Raspberry Pi GPIO pins.

Implementation Highlights

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"]
    }

Example 3: Sensor Monitor (Node.js)

Monitor temperature sensors and send events.

Implementation Highlights

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']
    }
  ];
}

Testing Your Pilot

Manual Testing

# 1. Start Qaptly Desktop

# 2. Run your pilot
python wifi_pilot.py

# 3. Approve in Qaptly

# 4. Test commands (from Qaptly or test script)

Automated Testing

Create a test script:

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

More Examples

GitHub Repository

Complete examples repository

Quick Start

Simple getting started example

Protocol Reference

Full protocol documentation

Capabilities

JSON Schema guide

Community Examples

Share your pilots with the community! Submit your examples to:

GitHub: qaptly-upp-examples
Discord: Qaptly Community

Android Pilot

iOS Pilot

Chrome Extension Pilot

Custom Pilots (UPP)

Overview

Example 1: Raspberry Pi WiFi Controller

Features

Prerequisites

Complete Implementation

Running the Example

Example 2: GPIO Controller (Python)

Implementation Highlights

Example 3: Sensor Monitor (Node.js)

Implementation Highlights

Testing Your Pilot

Manual Testing

Automated Testing

More Examples

GitHub Repository

Quick Start

Protocol Reference

Capabilities

Community Examples

​Overview

​Example 1: Raspberry Pi WiFi Controller

​Features

​Prerequisites

​Complete Implementation

​Running the Example

​Example 2: GPIO Controller (Python)

​Implementation Highlights

​Example 3: Sensor Monitor (Node.js)

​Implementation Highlights

​Testing Your Pilot

​Manual Testing

​Automated Testing

​More Examples

GitHub Repository

Quick Start

Protocol Reference

Capabilities

​Community Examples

Overview

Example 1: Raspberry Pi WiFi Controller

Features

Prerequisites

Complete Implementation

Running the Example

Example 2: GPIO Controller (Python)

Implementation Highlights

Example 3: Sensor Monitor (Node.js)

Implementation Highlights

Testing Your Pilot

Manual Testing

Automated Testing

More Examples

Community Examples