hobostreamer_control_example.py
bash · 17 views · 5 unique · 2026-04-08 01:49:20 · raw
#!/usr/bin/env python3
"""
HoboStreamer — Control Buttons & Streaming Example
====================================================
Demonstrates:
1. Dashboard setup overview (printed instructions)
2. REST API — create, list, and delete control buttons
3. Hardware WebSocket — connect as a hardware client and receive commands
4. FFmpeg streaming examples — jsmpeg (TCP pipe) and WebRTC-ready (WHIP)
Setup
-----
pip install websocket-client requests
Your credentials come from the HoboStreamer dashboard:
- JWT token : hobostreamer.com → Dashboard → copy from browser devtools
(Network tab → any /api/ request → Authorization header)
- Stream Key : hobostreamer.com → Dashboard → Stream Key section
Usage
-----
# Just print setup info and walk through examples
python hobostreamer_example.py --token YOUR_JWT --key YOUR_STREAM_KEY --stream 42
# Connect as hardware client and listen for button presses
python hobostreamer_example.py --token YOUR_JWT --key YOUR_STREAM_KEY --stream 42 --listen
# Stream a webcam using jsmpeg (requires ffmpeg)
python hobostreamer_example.py --stream-jsmpeg --key YOUR_STREAM_KEY
# Stream a webcam using the WebRTC WHIP ingest
python hobostreamer_example.py --stream-whip --token YOUR_JWT --stream 42
"""
import argparse
import json
import subprocess
import sys
import signal
import time
import threading
try:
import requests
except ImportError:
print("[ERROR] requests not installed. Run: pip install requests")
sys.exit(1)
try:
import websocket
except ImportError:
print("[ERROR] websocket-client not installed. Run: pip install websocket-client")
sys.exit(1)
DEFAULT_HOST = "hobostreamer.com"
DEFAULT_PORT = 443
DEFAULT_HTTP = "https"
DEFAULT_WS = "wss"
# ═══════════════════════════════════════════════════════════════
# Dashboard Setup Guide
# ═══════════════════════════════════════════════════════════════
def print_setup_guide():
print("""
╔══════════════════════════════════════════════════════════════╗
║ HoboStreamer — Dashboard Setup Guide ║
╠══════════════════════════════════════════════════════════════╣
║ ║
║ 1. CREATE AN ACCOUNT ║
║ Go to https://hobo.tools and sign up. ║
║ HoboStreamer uses the shared Hobo Network identity. ║
║ ║
║ 2. BECOME A STREAMER ║
║ Visit https://hobostreamer.com and log in. ║
║ Request streamer access from the dashboard if needed. ║
║ ║
║ 3. GET YOUR STREAM KEY ║
║ Dashboard → Stream Key section → copy the key. ║
║ Keep this private — it authenticates your hardware. ║
║ ║
║ 4. GET YOUR JWT TOKEN (for REST API calls) ║
║ Log in to hobostreamer.com in your browser. ║
║ Open DevTools (F12) → Network tab → click any ║
║ /api/ request → Headers → Authorization: Bearer <TOKEN> ║
║ Copy that token value. ║
║ ║
║ 5. ADD CONTROL BUTTONS ║
║ Dashboard → Controls section → Add buttons, OR ║
║ use this script's REST API examples below. ║
║ ║
║ 6. GO LIVE ║
║ Dashboard → fill in title/category → click Go Live. ║
║ Note the Stream ID shown under "Active Streams". ║
║ ║
║ 7. CONNECT YOUR HARDWARE ║
║ Run this script (or controller.py) with --listen. ║
║ Viewers clicking buttons triggers commands to your code. ║
║ ║
║ STREAMING OPTIONS: ║
║ jsmpeg : OBS or ffmpeg → TCP pipe → HoboStreamer relay ║
║ WebRTC : Browser-based (in dashboard) or WHIP ingest ║
║ RTMP : OBS → rtmp://hobostreamer.com/live/<KEY> ║
║ ║
╚══════════════════════════════════════════════════════════════╝
""")
# ═══════════════════════════════════════════════════════════════
# REST API Helpers
# ═══════════════════════════════════════════════════════════════
class HoboStreamerAPI:
"""Simple wrapper for HoboStreamer REST API calls."""
def __init__(self, token, host=DEFAULT_HOST, port=DEFAULT_PORT, https=True):
scheme = "https" if https else "http"
port_str = f":{port}" if (https and port != 443) or (not https and port != 80) else ""
self.base = f"{scheme}://{host}{port_str}/api"
self.headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
def _get(self, path):
r = requests.get(f"{self.base}{path}", headers=self.headers, timeout=10)
r.raise_for_status()
return r.json()
def _post(self, path, body=None):
r = requests.post(f"{self.base}{path}", headers=self.headers,
json=body or {}, timeout=10)
r.raise_for_status()
return r.json()
def _put(self, path, body=None):
r = requests.put(f"{self.base}{path}", headers=self.headers,
json=body or {}, timeout=10)
r.raise_for_status()
return r.json()
def _delete(self, path):
r = requests.delete(f"{self.base}{path}", headers=self.headers, timeout=10)
r.raise_for_status()
return r.json()
# ── Stream management ───────────────────────────────────────
def go_live(self, title, description="", protocol="rtmp", category="", nsfw=False):
"""Start a stream session (go live)."""
return self._post("/streams", {
"title": title,
"description": description,
"protocol": protocol, # 'rtmp', 'jsmpeg', or 'webrtc'
"category": category,
"nsfw": nsfw,
})
def end_stream(self, stream_id):
"""End an active stream."""
return self._delete(f"/streams/{stream_id}")
def get_my_streams(self):
"""Get your active streams."""
return self._get("/streams/mine")
def get_endpoint(self, stream_id):
"""Get the streaming endpoint info for a stream."""
return self._get(f"/streams/{stream_id}/endpoint")
def get_stream_key(self):
"""Get your stream key."""
return self._get("/auth/stream-key")
# ── Control buttons ─────────────────────────────────────────
def get_controls(self, stream_id):
"""Get all control buttons for a stream."""
return self._get(f"/controls/{stream_id}")
def add_control(self, stream_id, label, command, icon="fa-gamepad",
control_type="button", key_binding=None, cooldown_ms=500):
"""
Add a control button to a stream.
Parameters
----------
stream_id : int — your active stream ID
label : str — button text shown to viewers (max 50 chars)
command : str — command string sent to hardware when clicked (max 100 chars)
icon : str — FontAwesome icon class, e.g. 'fa-arrow-up'
control_type : str — 'button' (more types may be added)
key_binding : str — optional keyboard shortcut, e.g. 'ArrowUp'
cooldown_ms : int — milliseconds between allowed presses (default 500)
"""
return self._post(f"/controls/{stream_id}", {
"label": label,
"command": command,
"icon": icon,
"control_type": control_type,
"key_binding": key_binding,
"cooldown_ms": cooldown_ms,
})
def update_control(self, stream_id, control_id, **kwargs):
"""Update a control button. Pass any fields to change as kwargs."""
return self._put(f"/controls/{stream_id}/{control_id}", kwargs)
def delete_control(self, stream_id, control_id):
"""Remove a control button."""
return self._delete(f"/controls/{stream_id}/{control_id}")
def generate_api_key(self, label="My Script"):
"""
Generate a persistent API key (shown once, save it!).
Returns: { api_key: '...', label: '...' }
"""
return self._post("/controls/api-key", {"label": label})
def list_api_keys(self):
"""List your generated API keys (hashes only — raw key not retrievable)."""
return self._get("/controls/api-keys")
# ═══════════════════════════════════════════════════════════════
# Example: Create a standard movement button set
# ═══════════════════════════════════════════════════════════════
MOVEMENT_BUTTONS = [
("Forward", "forward", "fa-arrow-up", "ArrowUp"),
("Backward", "backward", "fa-arrow-down", "ArrowDown"),
("Left", "left", "fa-arrow-left", "ArrowLeft"),
("Right", "right", "fa-arrow-right", "ArrowRight"),
("Stop", "stop", "fa-stop", "Space"),
("Horn", "horn", "fa-bullhorn", None),
]
def demo_controls(api, stream_id):
"""Add a movement control set, list them, then clean up."""
print(f"\n[API] Adding movement control buttons to stream {stream_id}...")
added = []
for label, command, icon, key in MOVEMENT_BUTTONS:
result = api.add_control(stream_id, label, command, icon=icon, key_binding=key)
controls = result.get("controls", [])
# The last control in the list is the one just added
if controls:
cid = controls[-1]["id"]
added.append(cid)
print(f" + [{cid}] {label} → '{command}' ({icon})")
print(f"\n[API] Listing controls for stream {stream_id}:")
result = api.get_controls(stream_id)
for ctrl in result.get("controls", []):
status = "enabled" if ctrl.get("is_enabled", True) else "disabled"
print(f" [{ctrl['id']}] {ctrl['label']:<12} cmd={ctrl['command']:<10} {status}")
print(f"\n[API] Cleaning up — deleting the {len(added)} buttons just added...")
for cid in added:
api.delete_control(stream_id, cid)
print(f" - Deleted control {cid}")
print("[API] Done.")
# ═══════════════════════════════════════════════════════════════
# Hardware WebSocket Listener
# ═══════════════════════════════════════════════════════════════
class CommandHandler:
"""
Receives commands from HoboStreamer viewers and dispatches them.
Subclass or edit handle_command() to hook in your own hardware.
"""
def handle_command(self, command, data, from_user):
"""
Called whenever a viewer presses a control button.
Parameters
----------
command : str — the command string defined on the button
data : dict — optional extra data (may be empty)
from_user : str — viewer's username (or 'anonymous')
"""
print(f" [CMD] {command!r} from {from_user}")
# --- Add your hardware logic here ---
if command == "forward":
print(" → drive forward")
elif command == "backward":
print(" → drive backward")
elif command == "left":
print(" → turn left")
elif command == "right":
print(" → turn right")
elif command == "stop":
print(" → stop all motors")
elif command == "horn":
print(" → beep!")
else:
print(f" → unhandled command: {command}")
def listen_for_commands(stream_key, handler, host=DEFAULT_HOST, port=DEFAULT_PORT, ssl=True):
"""
Connect to HoboStreamer control WebSocket as a hardware client.
Your stream_key acts as the authentication credential.
The server will route commands from viewers to this connection.
"""
scheme = "wss" if ssl else "ws"
port_str = f":{port}" if (ssl and port != 443) or (not ssl and port != 80) else ""
url = f"{scheme}://{host}{port_str}/ws/control?mode=hardware&stream_key={stream_key}"
reconnect_delay = 5
running = True
def on_open(ws):
print(f"[WS] Connected as hardware client")
print(f"[WS] Waiting for viewer commands... (Ctrl+C to stop)\n")
def on_message(ws, raw):
try:
msg = json.loads(raw)
msg_type = msg.get("type")
if msg_type == "connected":
print(f"[WS] Server acknowledged hardware connection")
elif msg_type == "command":
handler.handle_command(
command=msg.get("command", ""),
data=msg.get("data") or {},
from_user=msg.get("from_user", "anonymous"),
)
elif msg_type == "error":
print(f"[WS ERROR] {msg.get('message')}")
elif msg_type == "ping":
ws.send(json.dumps({"type": "pong"}))
except json.JSONDecodeError:
print(f"[WS] Bad JSON: {raw}")
def on_error(ws, error):
print(f"[WS ERROR] {error}")
def on_close(ws, code, reason):
print(f"[WS] Disconnected (code={code}, reason={reason})")
if running:
print(f"[WS] Reconnecting in {reconnect_delay}s...")
time.sleep(reconnect_delay)
connect()
def connect():
ws = websocket.WebSocketApp(
url,
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close,
)
ws.run_forever()
def stop(sig, frame):
nonlocal running
running = False
print("\n[WS] Shutting down.")
sys.exit(0)
signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)
connect()
# ═══════════════════════════════════════════════════════════════
# FFmpeg Streaming Examples
# ═══════════════════════════════════════════════════════════════
def ffmpeg_jsmpeg(stream_key, host=DEFAULT_HOST, port=DEFAULT_PORT, ssl=True,
device="/dev/video0", width=1280, height=720, fps=30, bitrate="2500k"):
"""
Stream a camera to HoboStreamer using jsmpeg (TCP socket relay).
jsmpeg uses a custom TCP socket protocol — ffmpeg encodes to MPEG-1 and
pipes it directly into the server's TCP relay endpoint.
How it works:
Camera → ffmpeg → MPEG-1 video → TCP → HoboStreamer jsmpeg relay
→ Server re-muxes and distributes to viewers via WebSocket
Requirements: ffmpeg in PATH
Parameters
----------
stream_key : str — your HoboStreamer stream key
device : str — video device (Linux: /dev/video0, macOS: 0, Windows: 'video="Your Cam"')
width/height/fps : video dimensions
bitrate : str — video bitrate (e.g. '2500k', '4000k')
"""
scheme_tcp = "tcp"
port_str = f":{port}" if port not in (80, 443) else (":80" if not ssl else ":443")
tcp_url = f"{scheme_tcp}://{host}{port_str}/jsmpeg/{stream_key}"
# Platform input flag
import platform
if platform.system() == "Linux":
input_flags = ["-f", "v4l2", "-i", device]
elif platform.system() == "Darwin":
input_flags = ["-f", "avfoundation", "-i", f"{device}:none"]
else:
# Windows — use DirectShow
input_flags = ["-f", "dshow", "-i", f"video={device}"]
cmd = [
"ffmpeg",
*input_flags,
# No audio for jsmpeg (MPEG-1 video only)
"-an",
# Video encoding — jsmpeg needs MPEG-1
"-vcodec", "mpeg1video",
"-b:v", bitrate,
"-r", str(fps),
"-s", f"{width}x{height}",
# Output to TCP socket
"-f", "mpegts",
tcp_url,
]
print(f"[STREAM] jsmpeg → {tcp_url}")
print(f"[STREAM] Command: {' '.join(cmd)}\n")
subprocess.run(cmd)
def ffmpeg_webrtc_whip(stream_key, host=DEFAULT_HOST, port=DEFAULT_PORT, ssl=True,
device="/dev/video0", width=1280, height=720, fps=30,
video_bitrate="2500k", audio_bitrate="128k"):
"""
Stream a camera to HoboStreamer using WebRTC WHIP ingest.
WHIP (WebRTC-HTTP Ingest Protocol) lets ffmpeg push a stream that
viewers watch with near-real-time latency via WebRTC.
Requirements: ffmpeg 6.1+ with libx264 and libopus support.
On Linux: apt install ffmpeg
On macOS: brew install ffmpeg
Parameters
----------
stream_key : str — your HoboStreamer stream key
video_bitrate : str — H.264 video bitrate (e.g. '2500k')
audio_bitrate : str — Opus audio bitrate (e.g. '128k')
"""
scheme = "https" if ssl else "http"
port_str = f":{port}" if (ssl and port != 443) or (not ssl and port != 80) else ""
whip_url = f"{scheme}://{host}{port_str}/api/streams/whip?key={stream_key}"
import platform
if platform.system() == "Linux":
input_flags = ["-f", "v4l2", "-i", device]
elif platform.system() == "Darwin":
input_flags = ["-f", "avfoundation", "-i", f"{device}:default"]
else:
input_flags = ["-f", "dshow", "-i", f"video={device}:audio=Microphone"]
cmd = [
"ffmpeg",
*input_flags,
# Video: H.264 baseline (best WebRTC compatibility)
"-vcodec", "libx264",
"-preset", "veryfast",
"-tune", "zerolatency",
"-b:v", video_bitrate,
"-r", str(fps),
"-s", f"{width}x{height}",
"-g", str(fps * 2), # keyframe every 2 seconds
# Audio: Opus
"-acodec", "libopus",
"-b:a", audio_bitrate,
"-ar", "48000",
"-ac", "2",
# WHIP output
"-f", "whip",
whip_url,
]
print(f"[STREAM] WebRTC WHIP → {whip_url}")
print(f"[STREAM] Command: {' '.join(cmd)}\n")
subprocess.run(cmd)
def print_rtmp_example(stream_key, host=DEFAULT_HOST):
"""Print OBS / ffmpeg RTMP config for reference."""
rtmp_url = f"rtmp://{host}/live"
print(f"""
[STREAM] RTMP ingest (use with OBS or ffmpeg):
Server : {rtmp_url}
Key : {stream_key}
ffmpeg example:
ffmpeg -f v4l2 -i /dev/video0 \\
-vcodec libx264 -preset veryfast -b:v 3000k \\
-acodec aac -b:a 128k \\
-f flv {rtmp_url}/{stream_key}
In OBS:
Settings → Stream → Service: Custom
Server: {rtmp_url}
Stream Key: {stream_key}
""")
# ═══════════════════════════════════════════════════════════════
# Main
# ═══════════════════════════════════════════════════════════════
def main():
parser = argparse.ArgumentParser(
description="HoboStreamer — Control Buttons & Streaming Example",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__,
)
parser.add_argument("--token", help="JWT bearer token from hobostreamer.com")
parser.add_argument("--key", help="Your stream key")
parser.add_argument("--stream", type=int, help="Stream ID (shown in dashboard)")
parser.add_argument("--host", default=DEFAULT_HOST, help=f"Server host (default: {DEFAULT_HOST})")
parser.add_argument("--port", type=int, default=DEFAULT_PORT)
parser.add_argument("--no-ssl", action="store_true", help="Disable SSL (for local dev)")
parser.add_argument("--setup", action="store_true", help="Print setup guide and exit")
parser.add_argument("--demo-api", action="store_true", help="Run REST API button demo (needs --token --stream)")
parser.add_argument("--listen", action="store_true", help="Listen for hardware commands (needs --key)")
parser.add_argument("--stream-jsmpeg", action="store_true", help="Start jsmpeg camera stream (needs --key)")
parser.add_argument("--stream-whip", action="store_true", help="Start WebRTC WHIP camera stream (needs --key)")
parser.add_argument("--stream-rtmp", action="store_true", help="Print RTMP config and exit (needs --key)")
parser.add_argument("--device", default="/dev/video0", help="Camera device (default: /dev/video0)")
parser.add_argument("--width", type=int, default=1280)
parser.add_argument("--height", type=int, default=720)
parser.add_argument("--fps", type=int, default=30)
parser.add_argument("--bitrate", default="2500k")
args = parser.parse_args()
ssl = not args.no_ssl
if args.setup or len(sys.argv) == 1:
print_setup_guide()
if len(sys.argv) == 1:
parser.print_help()
return
if args.demo_api:
if not args.token or not args.stream:
print("[ERROR] --demo-api requires --token and --stream")
sys.exit(1)
api = HoboStreamerAPI(args.token, args.host, args.port, https=ssl)
demo_controls(api, args.stream)
return
if args.listen:
if not args.key:
print("[ERROR] --listen requires --key")
sys.exit(1)
handler = CommandHandler()
listen_for_commands(args.key, handler, args.host, args.port, ssl)
return
if args.stream_jsmpeg:
if not args.key:
print("[ERROR] --stream-jsmpeg requires --key")
sys.exit(1)
ffmpeg_jsmpeg(args.key, args.host, args.port, ssl,
args.device, args.width, args.height, args.fps, args.bitrate)
return
if args.stream_whip:
if not args.key:
print("[ERROR] --stream-whip requires --key")
sys.exit(1)
ffmpeg_webrtc_whip(args.key, args.host, args.port, ssl,
args.device, args.width, args.height, args.fps, args.bitrate)
return
if args.stream_rtmp:
if not args.key:
print("[ERROR] --stream-rtmp requires --key")
sys.exit(1)
print_rtmp_example(args.key, args.host)
return
parser.print_help()
if __name__ == "__main__":
main()