Turn anything into a realtime stream, then decide exactly who can reach it: your own machines, your team, or any app or agent through a URL. Nothing leaves your machine until you say so.
Working today. We're letting teams in a few at a time.
Every exposed stream is a URL. Open it in a browser, hand it to an agent, or pipe it into the tools you already use.
# record the live video, no player needed
$ ffmpeg \
-i https://live.tatolab.com/acme/pitch/video.ts \
-c copy pitch.mp4
# or let the CLI look up the URL
$ ffmpeg -i "$(tatolab url pitch/video --as ts)" \
-c copy pitch.mp4
Input #0, mpegts, from 'https://live.tatolab.com/s/7c…': Stream #0:0: Video: h264 (High), 1920x1080, 23.98 fps
# every detection, as it happens
$ curl -N \
"$(tatolab url pitch/detections --as ndjson)" \
| jq -c '{t, players, ball}'
{"t":"14:02:11.402","players":5,"ball":[0.49,0.46]}
{"t":"14:02:11.444","players":5,"ball":[0.48,0.46]}
{"t":"14:02:13.070","players":4,"ball":[0.49,0.42]}
# box just the ball
$ curl -X POST "$(tatolab url pitch/commands)" \
-H 'content-type: application/json' \
-d '{"highlight": "ball"}'
Run it, then watch the video.
One Python file wires the camera, the detector and the encoder together, and names what the stream offers. A second decides who can reach each part. Agents use it like any other tool.
from tatolab.stream import CameraSource, H264Encoder, Stream
from processors.player_detector import PlayerDetector # your own processors
from processors.highlight import Highlight
from processors.ptz import FollowBall
def setup(rt: Stream) -> None:
camera = rt.add(CameraSource)
detector = rt.add(PlayerDetector)
highlight = rt.add(Highlight)
follow = rt.add(FollowBall)
encoder = rt.add(H264Encoder)
rt.connect(camera.output("video"), detector.input("video"))
rt.connect(camera.output("video"), encoder.input("video"))
rt.connect(detector.output("detections"), highlight.input("detections"))
rt.connect(detector.output("detections"), follow.input("detections"))
# Name what this stream offers. Nothing leaves the machine yet.
rt.offer("video", encoder.output("encoded_video"))
rt.offer("detections", highlight.output("detections"))
rt.offer("commands", highlight.input("commands"))
from tatolab import host
pitch = host.load("pitch", "stream.py") # runs on this machine
pitch.expose("video", to="public") # a link you can share
pitch.expose("detections", to="peers") # your machines and team
pitch.expose("commands", to="peers", write=True) # they can send back
for event in pitch.read("detections"): # your own control logic
ball = event["ball"]
near_goal = ball is not None and not 0.15 < ball[0] < 0.85
pitch.send("commands", {"highlight": "ball" if near_goal else "all"})
loaded pitch this machine exposed pitch/video public → live.tatolab.com/acme/pitch/video exposed pitch/detections peers exposed pitch/commands peers, writable
Anything with an API can join a stream: a camera SDK, a robot's driver, a model, or a library in any language with an FFI. Expose it to Python and wire it in.
Use the SDKs and drivers you already have. Code in C, C++, Rust, or anything else with an FFI becomes a processor once Python can call it.
Add processors and connect an output to an input. Every connection is one line you can read.
Each Python processor runs in its own process, so a slow or crashing stage can't stall the others. Add, remove and rewire them while the stream runs, from code or from an agent.
import ctypes
from tatolab.stream import input, processor
# The camera vendor's C SDK. A library in C++, Rust, Go, or any
# language with an FFI wires in the same way.
ptz = ctypes.CDLL("libvendorptz.so")
@processor
class FollowBall:
"""Pans the pitch camera to keep the ball in frame."""
@input(delivery_profile="newest")
def detections(self) -> dict: ...
def setup(self, ctx) -> None:
self.cam = ptz.ptz_connect(b"192.168.1.40")
def process(self, ctx) -> None:
event = ctx.inputs.read("detections")
if event and event["ball"]:
x, y = (ctypes.c_float(v) for v in event["ball"])
ptz.ptz_center(self.cam, x, y)
from tatolab.stream import VideoFrame, input, output, processor
from models import load_detector # your model, any framework
@processor
class PlayerDetector:
"""Finds the players and the ball in every frame."""
@input(delivery_profile="newest")
def video(self) -> VideoFrame: ...
@output()
def detections(self) -> dict: ...
def setup(self, ctx) -> None:
self.model = load_detector("players.onnx")
def process(self, ctx) -> None:
frame = ctx.inputs.read("video", into=VideoFrame)
if frame is None:
return
with frame.cpu() as pixels:
found = self.model(pixels)
ball = [d.center for d in found if d.label == "ball"]
ctx.outputs.write("detections", {
"boxes": [(d.label, d.score, d.box) for d in found],
"players": sum(d.label == "player" for d in found),
"ball": ball[0] if ball else None,
})
Video, audio, tensors, JSON, bytes. Streams carry any payload, with no schema to register.
Nothing leaves your machine until you expose it. You choose who can reach each stream, and who can write back.
A stream stays on the machine that runs it until you expose it.
pitch/video private
pitch/detections private
Expose a stream to your own machines and team. They connect directly, with no port forwarding.
pitch.expose("detections",
to="peers")
Give one stream a URL anyone you share it with can open. Nothing else goes with it.
pitch.expose("video",
to="public")
Let the apps and agents you allow send data back in, like what the feed should highlight.
pitch.expose("commands",
to="peers",
write=True)
Streams are encrypted before they leave your machine. Already on Tailscale? Run over your tailnet too.
$ tailscale up # optional
Run the whole path yourself, or use ours.
nodes your machines
relay yours, or ours
Tunnels put a port on the network. VPNs put a device on it. Tatolab puts the streams you expose on it.
| Tunnelngrok and similar | Device networkTailscale and similar | TatolabApplication-defined | |
|---|---|---|---|
| What's on the network | A port on one machine | Whole devices | Streams you expose |
| Addressed by | A URL per port | A device name or IP | A stream name, or a URL |
| Who can use it | Anyone with the URL | Devices on the network | The people, apps and agents you allow |
| Configured in | A CLI or config file | An admin console | Your code, the CLI, or an agent |
Tell us what you'd stream. We're letting teams in a few at a time, and we'll reach out to get you set up.