pi_servo_robot.py

# examples/pi_servo_robot.py
# Raspberry Pi servo robot using the Python SDK
# pip install spraay-rtp[pi]

import asyncio
import os
from dotenv import load_dotenv
load_dotenv()

from rtp import RTPDevice, RTPDeviceConfig, ConnectionConfig, ConnectionType, TaskResult

# ─── Servo Control ────────────────────────────────────────────────
try:
    import pigpio
    pi = pigpio.pi()
    SERVO_PIN = 17
    POSITIONS = {
        "home":  1500,
        "pick":  500,
        "place": 2500,
        "scan":  1000,
    }

    async def move_servo(pulse_width: int, hold_ms: int = 1000):
        pi.set_servo_pulsewidth(SERVO_PIN, pulse_width)
        await asyncio.sleep(hold_ms / 1000)

except ImportError:
    # Not on a Pi — simulate servo movements
    print("⚠️  pigpio not found — simulating servo movements")
    POSITIONS = {"home": 1500, "pick": 500, "place": 2500, "scan": 1000}

    async def move_servo(pulse_width: int, hold_ms: int = 1000):
        print(f"  🔄 Servo → {pulse_width}μs")
        await asyncio.sleep(hold_ms / 1000)

# ─── Device Setup ─────────────────────────────────────────────────
config = RTPDeviceConfig(
    name="ServoBot-Pi-01",
    description="Raspberry Pi servo arm — Python SDK",
    capabilities=["pick", "place", "scan", "move"],
    price_per_task="0.01",
    payment_address=os.getenv("WALLET_ADDRESS", ""),
    api_key=os.getenv("SPRAAY_API_KEY", ""),
    connection=ConnectionConfig(
        type=ConnectionType.WEBHOOK,
        webhook_url=f"http://{os.getenv('PUBLIC_IP', 'localhost')}:3100/rtp/task",
        port=3100,
        secret=os.getenv("WEBHOOK_SECRET")
    ),
    tags=["raspberry-pi", "servo", "python", "demo"]
)

device = RTPDevice(config)

# ─── Task Handlers ────────────────────────────────────────────────

@device.on_task("pick")
async def handle_pick(params, task):
    print(f"🦾 PICK: {params.get('item', 'item')}")
    await task.progress()
    await move_servo(POSITIONS["pick"])
    await move_servo(POSITIONS["home"])
    await task.complete(TaskResult(
        success=True,
        output=f"Picked {params.get('item', 'item')}"
    ))


@device.on_task("place")
async def handle_place(params, task):
    print(f"📦 PLACE: {params.get('to_location', 'destination')}")
    await task.progress()
    await move_servo(POSITIONS["place"])
    await move_servo(POSITIONS["home"])
    await task.complete(TaskResult(
        success=True,
        output=f"Placed at {params.get('to_location', 'destination')}"
    ))


@device.on_task("scan")
async def handle_scan(params, task):
    print(f"📷 SCAN: {params.get('target', 'target')}")
    await task.progress()
    await move_servo(POSITIONS["scan"])
    await asyncio.sleep(2)  # hold for scan
    await move_servo(POSITIONS["home"])
    await task.complete(TaskResult(
        success=True,
        output=f"Scanned {params.get('target')}",
        data={
            "value": f"ITEM-{__import__('random').randint(1000, 9999)}",
            "confidence": 0.97
        }
    ))


@device.on_task("move")
async def handle_move(params, task):
    angle = min(180, max(0, params.get("angle", 90)))
    pulse_width = int(500 + (angle / 180) * 2000)
    print(f"🔄 MOVE: {angle}°")
    await task.progress()
    await move_servo(pulse_width)
    await task.complete(TaskResult(
        success=True,
        output=f"Moved to {angle}°",
        data={"angle": angle, "pulse_width": pulse_width}
    ))


# ─── Main ─────────────────────────────────────────────────────────

async def main():
    print("🤖 ServoBot-Pi-01 starting (Python SDK)...")

    # Home the servo
    await move_servo(POSITIONS["home"])
    print("✅ Servo homed")

    # Register and start
    result = await device.register()
    print(f"📡 Registered: {result['robot_id']}")
    print(f"💧 Endpoint:   {result['endpoint']}")

    await device.listen(3100)


if __name__ == "__main__":
    asyncio.run(main())