VaM 1.x Using the Coyote 3 with Toy Controller plugins

[LittleBird]

I vibe coded a python script that takes a UDP connection from the toy controller scripts (I have been using the Yoooi ToySerialController plugin) and bridges it to a websocket, which then the CoyoteSocket uses to control the Coyote 3.

The script requires asyncio, socket, and websockets.

pip install asyncio socket websockets

// Update the ports to your liking based on your configuration.
Default UDP port is 8000
Default WS port is 12346

### Usage ###

Requires python 3.10+

Run the script before connecting the plugin.

python.exe udp2ws.py

#####

# udp2ws.py

import asyncio
import socket
import websockets

UDP_IP = "0.0.0.0"
UDP_PORT = 8000

WS_URL = "ws://localhost:12346"   # WebSocket server to connect to


async def udp_listener(queue):
    """Listen for UDP packets and push them into an asyncio queue."""
    loop = asyncio.get_running_loop()

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind((UDP_IP, UDP_PORT))
    sock.setblocking(False)

    print(f"UDP server listening on {UDP_IP}:{UDP_PORT}")

    while True:
        data = await loop.sock_recv(sock, 4096)
        message = data.decode(errors="ignore").strip()
        print(f"UDP received: {message}")
        await queue.put(message)


async def websocket_sender(queue):
    """Connect to WebSocket server and forward UDP messages."""
    while True:
        try:
            print(f"Connecting to WebSocket server at {WS_URL}...")
            async with websockets.connect(WS_URL) as ws:
                print("Connected to WebSocket server")

                while True:
                    msg = await queue.get()
                    await ws.send(msg)
                    print(f"Sent to WS: {msg}")

        except Exception as e:
            print(f"WebSocket error: {e}")
            print("Retrying in 3 seconds...")
            await asyncio.sleep(3)


async def main():
    queue = asyncio.Queue()

    await asyncio.gather(
        udp_listener(queue),
        websocket_sender(queue)
    )


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

I am sure this could get improved on, but it works.