---
title: "Several sources, one record"
description: "Record video, audio and instrument readings on one machine, on one clock, so a trial's data can be lined up afterwards, and watch it live without touching the recording. A research lab example."
type: "docs-page"
url: "https://tatolab.com/docs/use-cases/research"
site: "Tatolab"
section: "Use cases"
---

# Several sources, one record

> Record video, audio and instrument readings on one machine, on one clock, so a trial's data can be lined up afterwards, and watch it live without touching the recording. A research lab example.

An experiment often has a camera, a microphone and an instrument or two, each with its own software, its own files and its own idea of the time. Lining them up afterwards becomes its own project: which video frame matches this sensor reading, and was anything lost?

With Tatolab, one stream reads every source on the machine, and every frame, audio block and reading is stamped by the same clock. The trial's data lands in files you can analyze with your usual tools, and the stream exposes its live outputs, so a colleague's monitor or an agent can watch without touching the recording.

```txt
CameraSource ─┬─ MotionDetector ─ motion ─────┐
              │                               ├─ TrialLogger ─▶ trial-….csv
UdpListener (sensor) ─ readings ──────────────┘
              └─ H264Encoder ─▶ Mp4Sink ─▶ trial-….mp4
MicrophoneSource ─▶ WavWriter ─▶ trial-….wav
```

## The stream

```python title="stream.py"
from datetime import datetime
from pathlib import Path

from tatolab.stream import CameraSource, H264Encoder, MicrophoneSource, Mp4Sink, StreamBuilder, stream

from nodes.motion_detector import MotionDetector
from nodes.trial_logger import TrialLogger
from nodes.udp_listener import UdpListener
from nodes.wav_writer import WavWriter

FOLDER = Path(__file__).resolve().parent


@stream
def arena(stream_builder: StreamBuilder) -> None:
    """One trial in the arena: video, audio, motion and sensor readings, on one clock."""
    trial = FOLDER / f"trial-{datetime.now():%Y%m%d-%H%M%S}"
    camera = stream_builder.add(CameraSource, name="camera")
    microphone = stream_builder.add(MicrophoneSource)
    motion = stream_builder.add(MotionDetector, name="motion", config={"checks_per_second": 30})
    sensor = stream_builder.add(UdpListener, name="sensor", config={"port": 9000})
    logger = stream_builder.add(TrialLogger, config={"path": f"{trial}.csv"})
    audio = stream_builder.add(WavWriter, config={"path": f"{trial}.wav"})
    encoder = stream_builder.add(H264Encoder)
    recording = stream_builder.add(Mp4Sink, config={"path": f"{trial}.mp4"})

    stream_builder.connect(camera.output("video"), motion.input("video"))
    stream_builder.connect(motion.output("motion"), logger.input("motion"))
    stream_builder.connect(sensor.output("datagrams"), logger.input("readings"))
    stream_builder.connect(microphone.output("audio"), audio.input("audio"))
    stream_builder.connect(camera.output("video"), encoder.input("video"))
    stream_builder.connect(encoder.output("encoded_video"), recording.input("tracks"))

    stream_builder.expose(camera.output("video"))
    stream_builder.expose(motion.output("motion"))
```

`MotionDetector` is the [tutorial's](https://tatolab.com/docs/tutorial/first-node) node, here checking 30 times a second, and `UdpListener` is the threaded source from [Write a node](https://tatolab.com/docs/build/nodes#a-source-with-its-own-thread). The logger writes both inputs to one CSV:

```python title="nodes/trial_logger.py"
"""Writes motion measurements and sensor readings to one CSV, each with its timestamp."""

import csv
from dataclasses import dataclass

from tatolab.stream import RuntimeContextFullAccess, RuntimeContextLimitedAccess, node


@dataclass
class TrialLoggerConfig:
    path: str = "trial.csv"


@node
class TrialLogger:
    """Writes motion measurements and sensor readings to one CSV, stamped on the machine's clock."""

    @node.input(delivery_profile="ordered")
    def motion(self) -> dict: ...

    @node.input(delivery_profile="ordered")
    def readings(self) -> dict: ...

    def __init__(self, config: TrialLoggerConfig) -> None:
        self.path = config.path

    def setup(self, ctx: RuntimeContextFullAccess) -> None:
        self.file = open(self.path, "w", newline="")
        self.writer = csv.writer(self.file)
        self.writer.writerow(["timestamp_ns", "source", "value"])

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        while (measurement := ctx.inputs.read("motion")) is not None:
            self.writer.writerow([measurement["timestamp_ns"], "motion", measurement["changed"]])
        while (reading := ctx.inputs.read("readings")) is not None:
            self.writer.writerow([reading["timestamp_ns"], "sensor", reading["data"].decode(errors="replace").strip()])
        self.file.flush()

    def teardown(self, ctx: RuntimeContextFullAccess) -> None:
        self.file.close()
```

The MP4's audio would be Opus, which is lossy and resampled to 48 kHz, so audio for analysis goes to a WAV file instead:

```python title="nodes/wav_writer.py"
"""Writes audio to an uncompressed 16-bit WAV file."""

import wave
from dataclasses import dataclass

import numpy
from tatolab.stream import AudioBlock, RuntimeContextFullAccess, RuntimeContextLimitedAccess, node


@dataclass
class WavWriterConfig:
    path: str = "audio.wav"


@node
class WavWriter:
    """Writes the audio it receives to an uncompressed 16-bit WAV file, at the microphone's own rate."""

    @node.input(delivery_profile="ordered")
    def audio(self) -> AudioBlock: ...

    def __init__(self, config: WavWriterConfig) -> None:
        self.path = config.path
        self.wav: wave.Wave_write | None = None

    def process(self, ctx: RuntimeContextLimitedAccess) -> None:
        while (block := ctx.inputs.read("audio", into=AudioBlock)) is not None:
            if self.wav is None:
                self.wav = wave.open(self.path, "wb")
                self.wav.setnchannels(block.channels)
                self.wav.setsampwidth(2)
                self.wav.setframerate(block.sample_rate)
            samples = block.samples
            if block.dtype == "f32":
                samples = (numpy.clip(samples, -1.0, 1.0) * 32767).astype("<i2")
            self.wav.writeframes(samples.tobytes())

    def teardown(self, ctx: RuntimeContextFullAccess) -> None:
        if self.wav is not None:
            self.wav.close()
```

## Run a trial

Run each trial attached, from the project folder on the rig:

```sh
tatolab run      # starts the trial; Ctrl-C ends it
```

Each run writes a new set of files, named by when it started. Running it attached, rather than keeping it with `run -d`, is deliberate: a kept stream would load again after any restart with the same file names and start them over, erasing that trial. If the machine restarts mid-trial, the trial simply ends, and its files keep what was written. Copy each trial's files off when it ends.

While a trial runs, anyone on the machine can watch it from a stream of their own, reading `arena/camera/video` or `arena/motion/motion`, or with an agent, without touching the recording.

## Lining sources up

A shared clock makes lining up a matter of measuring offsets, not reconciling clocks. Each stamp is taken where the data reaches the machine: a USB camera's frame when the driver receives it, after exposure and transfer; a UDP reading when it arrives. So each device has its own delay:

- **Measure each device's delay once per rig,** with an event every device sees: an LED flash with a click, say, or a pulse your instrument also records. There's no hardware sync input.
- **The MP4's timeline starts at zero** at its first frame. To place CSV rows in the video, find the sync event in both, or log each frame's `timestamp_ns` with a node reading the camera.
- **The timestamps count from when the machine booted,** not from midnight. If you need wall-clock time, log it once at the start of a trial next to a `ctx.time` reading, and convert.

Rows from the two inputs arrive in whatever order the nodes produce them, so sort by timestamp first. With pandas:

```python title="analyze.py"
import sys

import pandas as pd

trial = pd.read_csv(sys.argv[1]).sort_values("timestamp_ns")
trial["seconds"] = (trial["timestamp_ns"] - trial["timestamp_ns"].iloc[0]) / 1e9
motion = trial[trial["source"] == "motion"]
print(motion.describe())
```

## Checking a record is complete

`tatolab graph` counts the bags each input dropped because it fell behind, but zero dropped bags doesn't prove a record is complete. It doesn't count frames a `newest` input passed over (as `MotionDetector` does, by design), frames the camera or USB bus lost, or datagrams lost on the network. To show a record is complete:

- check the recorded timestamps for gaps longer than one frame or one audio block;
- read anything you analyze per frame with an `ordered` input;
- put a sequence number in your instrument's datagrams, and check it in the CSV.

## Today's limits

- **One machine's clock.** Timestamps from different machines aren't on one clock.
- **Cameras the operating system sees.** On Linux, `CameraSource` reads uncompressed video only, and has no exposure, gain or frame-rate settings. Scientific cameras that need a vendor SDK need a source node around the SDK's Python binding.
- **The microphone opens at the device's default rate.** High-rate or ultrasonic work needs a source node that opens the device itself.
- **Not hard real-time.** Tatolab doesn't guarantee deadlines for closed-loop control.

## Related guides

- [Write a node](https://tatolab.com/docs/build/nodes): sources for your instruments.
- [Audio](https://tatolab.com/docs/build/audio): audio formats and windows.
- [Inspect a running stream](https://tatolab.com/docs/run/inspect): what `tatolab graph` shows during a run.