Sincronización de video multicámara con ISOBMFF

Esta página fue traducida automáticamente. Si encuentra errores o tiene sugerencias, contáctenos.

Al trabajar con segmentos de video—especialmente en sistemas que dependen de cámaras distribuidas, almacenamiento en la nube y múltiples flujos de datos—la sincronización temporal se convierte en uno de los retos más críticos. Sin información de temporización precisa, alinear eventos entre varios flujos (por ejemplo, video, audio y datos de sensores) se vuelve propenso a errores. Rhombus ha implementado una estrategia de embebido de marcas de tiempo personalizadas que ofrece precisión de milisegundos dentro de los segmentos de video .mp4/.m4v, yendo más allá de los campos de temporización gruesos que se encuentran tradicionalmente en los contenedores ISOBMFF (ISO Base Media File Format) estándar.

Esta guía explica qué significa este enfoque, por qué es importante y cómo los desarrolladores pueden recuperar y usar esta marca de tiempo para crear aplicaciones alineadas temporalmente.

Comprender ISOBMFF y el átomo “free”

El estándar ISOBMFF (ISO/IEC 14496-12) es el formato de contenedor que subyace a .mp4, .m4v, .mov y muchos formatos de segmentos de streaming como .fMP4. Su estructura se basa en cajas (átomos)—unidades de datos autocontenidas identificadas por un código de 4 caracteres (por ejemplo, moov, mdat, free).

Las marcas de tiempo estándar en ISOBMFF (por ejemplo, creation_time en la caja mvhd) normalmente tienen resolución de segundos. Esto es suficiente para algunos flujos de trabajo multimedia, pero insuficiente para la sincronización multicámara o la correlación de eventos de alta velocidad.

Rhombus resuelve esto embebiendo una marca de tiempo con precisión de milisegundos dentro de un átomo free. Este es un método no estándar, pero totalmente compatible con ISOBMFF.

La marca de tiempo personalizada de Rhombus

Cuando Rhombus segmenta el video (y si se obtienen segmentos reales, no un flujo de transporte en vivo), se escribe una firma de metadatos personalizada en la caja free:

Tipo de caja: free

En ISOBMFF, free normalmente es una caja de relleno que contiene espacio sin usar. Rhombus la reutiliza para transportar metadatos de marca de tiempo.

Firma: rhom

Los primeros 4 bytes son la cadena ASCII rhom—que la identifica como datos específicos de Rhombus.

Marca de tiempo en milisegundos

Los siguientes 8 bytes son un entero de 64 bits que representa la hora de inicio del contenido de video, medida en milisegundos desde la época Unix (UTC).

Ejemplo de disposición binaria:

[ free box length ][ 'free' ][ 'rhom' ][ 8-byte timestamp (ms since epoch) ]

Por qué esto es importante para los desarrolladores

Esta decisión de diseño desbloquea capacidades de sincronización precisa en múltiples casos de uso:

Alineación multicámara

Alinea las transmisiones de video de diferentes cámaras con una precisión de 1 ms para un monitoreo coordinado

Fusión de sensores

Combina video con datos de sensores IoT (eventos de control de acceso, lecturas ambientales)

Precisión forense

Reconstruye eventos hasta intervalos de subsegundos en investigaciones

Menor deriva

Evita los errores que se acumulan al depender únicamente de los relojes del sistema cliente o de la sincronización NTP

Para los socios del ecosistema y de integración, esto hace que las transmisiones de video de Rhombus sean altamente interoperables con análisis de terceros, canalizaciones de IA/ML y sistemas de monitoreo en tiempo real.

Recuperar la marca de tiempo

Analizar archivos ISOBMFF

Puedes usar bibliotecas de código abierto para leer la caja free de un segmento .mp4/.m4v y comprobar la firma rhom.

Analizador ISOBMFF (C++/Swift)

Biblioteca profesional para analizar el ISO Base Media File Format

Visor interactivo MP4Box.js

Herramienta en línea para inspeccionar visualmente las estructuras de cajas MP4

Ejemplos de implementación

Python
JavaScript
C++

Python Implementation

import struct
import datetime

def extract_rhombus_timestamp(file_path):
    """
    Extrae la marca de tiempo con precisión de milisegundos de un segmento de video de Rhombus.

    Args:
        file_path: Ruta al archivo .mp4 o .m4v

    Returns:
        tuple: (timestamp_ms, objeto datetime) o (None, None) si no se encuentra
    """
    with open(file_path, "rb") as f:
        data = f.read()

    # Buscar la caja 'free'
    idx = data.find(b'free')
    if idx == -1:
        return None, None

    # Buscar la firma 'rhom' después de 'free'
    rhom_idx = data.find(b'rhom', idx)
    if rhom_idx == -1:
        return None, None

    # Leer los siguientes 8 bytes después de 'rhom' (entero de 64 bits big-endian)
    timestamp_bytes = data[rhom_idx + 4 : rhom_idx + 12]
    timestamp_ms = int.from_bytes(timestamp_bytes, byteorder="big")

    # Convertir a hora UTC legible para humanos
    timestamp_dt = datetime.datetime.fromtimestamp(timestamp_ms / 1000.0, tz=datetime.timezone.utc)

    return timestamp_ms, timestamp_dt

# Ejemplo de uso:
timestamp_ms, timestamp_dt = extract_rhombus_timestamp("video_segment.mp4")

if timestamp_ms:
    print(f"Timestamp (ms since epoch): {timestamp_ms}")
    print(f"UTC Time: {timestamp_dt}")
    print(f"ISO Format: {timestamp_dt.isoformat()}")
else:
    print("Rhombus timestamp not found in file")

# Salida de ejemplo:
# Timestamp (ms since epoch): 1722945678123
# UTC Time: 2024-08-06 15:21:18.123000
# ISO Format: 2024-08-06T15:21:18.123000

Uso avanzado en Python

Multi-Segment Processing

import os
import glob
from datetime import datetime

class RhombusTimestampExtractor:
    """Extrae y administra marcas de tiempo de múltiples segmentos de video."""

    def __init__(self):
        self.timestamps = []

    def process_directory(self, directory_path, pattern="*.mp4"):
        """Procesa todos los archivos de video en un directorio."""
        files = glob.glob(os.path.join(directory_path, pattern))

        for file_path in sorted(files):
            timestamp_ms, timestamp_dt = extract_rhombus_timestamp(file_path)

            if timestamp_ms:
                self.timestamps.append({
                    'file': os.path.basename(file_path),
                    'timestamp_ms': timestamp_ms,
                    'datetime': timestamp_dt
                })

    def get_timeline(self):
        """Obtiene una lista de segmentos ordenada cronológicamente."""
        return sorted(self.timestamps, key=lambda x: x['timestamp_ms'])

    def find_segment_at_time(self, target_datetime):
        """Encuentra el segmento de video que contiene una hora específica."""
        target_ms = int(target_datetime.timestamp() * 1000)

        for segment in self.get_timeline():
            if segment['timestamp_ms'] <= target_ms:
                closest = segment
            else:
                break

        return closest

# Ejemplo de uso
extractor = RhombusTimestampExtractor()
extractor.process_directory("/path/to/video/segments")

# Encontrar el segmento en una hora específica
target = datetime(2024, 8, 6, 15, 21, 0)
segment = extractor.find_segment_at_time(target)
print(f"Segment at {target}: {segment['file']}")

JavaScript Implementation

/**
 * Extrae la marca de tiempo de Rhombus de un ArrayBuffer de segmento de video
 * @param {ArrayBuffer} arrayBuffer - El archivo de video como ArrayBuffer
 * @returns {Object|null} Objeto con timestampMs y utcTime, o null si no se encuentra
 */
function extractRhombusTimestamp(arrayBuffer) {
    const data = new Uint8Array(arrayBuffer);

    // Buscar la caja 'free'
    const freePattern = new Uint8Array([0x66, 0x72, 0x65, 0x65]); // 'free'
    let freeIndex = -1;

    for (let i = 0; i <= data.length - 4; i++) {
        if (data.subarray(i, i + 4).every((val, idx) => val === freePattern[idx])) {
            freeIndex = i;
            break;
        }
    }

    if (freeIndex === -1) return null;

    // Buscar la firma 'rhom'
    const rhomPattern = new Uint8Array([0x72, 0x68, 0x6f, 0x6d]); // 'rhom'
    let rhomIndex = -1;

    for (let i = freeIndex; i <= data.length - 4; i++) {
        if (data.subarray(i, i + 4).every((val, idx) => val === rhomPattern[idx])) {
            rhomIndex = i;
            break;
        }
    }

    if (rhomIndex === -1) return null;

    // Extraer la marca de tiempo de 8 bytes (big-endian)
    const timestampBytes = data.subarray(rhomIndex + 4, rhomIndex + 12);
    let timestamp = 0;

    for (let i = 0; i < 8; i++) {
        timestamp = timestamp * 256 + timestampBytes[i];
    }

    return {
        timestampMs: timestamp,
        utcTime: new Date(timestamp),
        isoString: new Date(timestamp).toISOString()
    };
}

// Ejemplo de uso con la File API
async function processVideoFile(file) {
    const arrayBuffer = await file.arrayBuffer();
    const result = extractRhombusTimestamp(arrayBuffer);

    if (result) {
        console.log('Timestamp (ms):', result.timestampMs);
        console.log('UTC Time:', result.utcTime);
        console.log('ISO String:', result.isoString);
    } else {
        console.log('Rhombus timestamp not found');
    }
}

// Uso con un elemento input
document.querySelector('#fileInput').addEventListener('change', async (e) => {
    const file = e.target.files[0];
    if (file) {
        await processVideoFile(file);
    }
});

Uso avanzado en JavaScript

Multi-Camera Synchronization

class RhombusVideoSync {
    constructor() {
        this.cameras = new Map();
    }

    /**
     * Agrega el segmento de video de una cámara con su marca de tiempo
     */
    async addCameraSegment(cameraId, videoFile) {
        const arrayBuffer = await videoFile.arrayBuffer();
        const timestamp = extractRhombusTimestamp(arrayBuffer);

        if (timestamp) {
            if (!this.cameras.has(cameraId)) {
                this.cameras.set(cameraId, []);
            }

            this.cameras.get(cameraId).push({
                timestamp: timestamp.timestampMs,
                datetime: timestamp.utcTime,
                file: videoFile
            });
        }
    }

    /**
     * Encuentra segmentos de todas las cámaras en una hora específica
     */
    findSegmentsAtTime(targetTime) {
        const targetMs = targetTime.getTime();
        const results = {};

        for (const [cameraId, segments] of this.cameras.entries()) {
            // Encontrar el segmento más cercano antes o en la hora objetivo
            const sorted = segments.sort((a, b) => a.timestamp - b.timestamp);
            const segment = sorted.find(s => s.timestamp <= targetMs);

            if (segment) {
                results[cameraId] = {
                    timestamp: segment.timestamp,
                    datetime: segment.datetime,
                    offset: targetMs - segment.timestamp
                };
            }
        }

        return results;
    }

    /**
     * Obtiene el rango de tiempo cubierto por todas las cámaras
     */
    getCommonTimeRange() {
        let earliestEnd = Infinity;
        let latestStart = 0;

        for (const segments of this.cameras.values()) {
            if (segments.length === 0) continue;

            const start = Math.min(...segments.map(s => s.timestamp));
            const end = Math.max(...segments.map(s => s.timestamp));

            latestStart = Math.max(latestStart, start);
            earliestEnd = Math.min(earliestEnd, end);
        }

        return {
            start: new Date(latestStart),
            end: new Date(earliestEnd),
            duration: earliestEnd - latestStart
        };
    }
}

// Uso
const sync = new RhombusVideoSync();
await sync.addCameraSegment('camera1', file1);
await sync.addCameraSegment('camera2', file2);
await sync.addCameraSegment('camera3', file3);

// Encontrar qué estaba sucediendo en todas las cámaras en una hora específica
const target = new Date('2024-08-06T15:21:18.123Z');
const segments = sync.findSegmentsAtTime(target);
console.log('Synchronized segments:', segments);

C++ Implementation

#include <iostream>
#include <fstream>
#include <vector>
#include <cstdint>
#include <chrono>
#include <optional>
#include <algorithm>

struct RhombusTimestamp {
    uint64_t timestampMs;
    std::chrono::system_clock::time_point utcTime;

    // Convertir a cadena ISO 8601
    std::string toISO8601() const {
        auto timeT = std::chrono::system_clock::to_time_t(utcTime);
        auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(
            utcTime.time_since_epoch()
        ).count() % 1000;

        char buffer[32];
        std::strftime(buffer, sizeof(buffer), "%Y-%m-%dT%H:%M:%S",
                     std::gmtime(&timeT));

        return std::string(buffer) + "." +
               std::to_string(ms) + "Z";
    }
};

/**
 * Extrae la marca de tiempo de Rhombus de un archivo de video
 */
std::optional<RhombusTimestamp> extractRhombusTimestamp(
    const std::string& filePath
) {
    std::ifstream file(filePath, std::ios::binary);
    if (!file) {
        return std::nullopt;
    }

    // Leer el archivo completo en memoria
    std::vector<uint8_t> data(
        (std::istreambuf_iterator<char>(file)),
        std::istreambuf_iterator<char>()
    );

    // Buscar la caja 'free'
    const std::vector<uint8_t> freePattern = {0x66, 0x72, 0x65, 0x65};
    auto freeIt = std::search(
        data.begin(), data.end(),
        freePattern.begin(), freePattern.end()
    );

    if (freeIt == data.end()) {
        return std::nullopt;
    }

    // Buscar la firma 'rhom'
    const std::vector<uint8_t> rhomPattern = {0x72, 0x68, 0x6f, 0x6d};
    auto rhomIt = std::search(
        freeIt, data.end(),
        rhomPattern.begin(), rhomPattern.end()
    );

    if (rhomIt == data.end()) {
        return std::nullopt;
    }

    // Extraer la marca de tiempo de 8 bytes (big-endian)
    uint64_t timestamp = 0;
    for (int i = 0; i < 8; i++) {
        timestamp = (timestamp << 8) | *(rhomIt + 4 + i);
    }

    // Convertir a time_point
    auto utcTime = std::chrono::system_clock::from_time_t(
        timestamp / 1000
    );
    utcTime += std::chrono::milliseconds(timestamp % 1000);

    return RhombusTimestamp{timestamp, utcTime};
}

// Ejemplo de uso
int main() {
    auto result = extractRhombusTimestamp("video_segment.mp4");

    if (result) {
        std::cout << "Timestamp (ms): " << result->timestampMs << std::endl;
        std::cout << "ISO 8601: " << result->toISO8601() << std::endl;
    } else {
        std::cout << "Rhombus timestamp not found" << std::endl;
    }

    return 0;
}

Uso avanzado en C++

Multi-Segment Timeline

#include <filesystem>
#include <map>

class RhombusTimeline {
private:
    struct Segment {
        std::filesystem::path filePath;
        RhombusTimestamp timestamp;
    };

    std::vector<Segment> segments;

public:
    /**
     * Carga todos los segmentos de video de un directorio
     */
    void loadDirectory(const std::filesystem::path& directory) {
        for (const auto& entry :
             std::filesystem::directory_iterator(directory)) {

            if (entry.path().extension() == ".mp4" ||
                entry.path().extension() == ".m4v") {

                auto timestamp = extractRhombusTimestamp(
                    entry.path().string()
                );

                if (timestamp) {
                    segments.push_back({entry.path(), *timestamp});
                }
            }
        }

        // Ordenar por marca de tiempo
        std::sort(segments.begin(), segments.end(),
            [](const Segment& a, const Segment& b) {
                return a.timestamp.timestampMs < b.timestamp.timestampMs;
            }
        );
    }

    /**
     * Encuentra el segmento que contiene una hora específica
     */
    std::optional<Segment> findSegmentAtTime(
        const std::chrono::system_clock::time_point& targetTime
    ) const {
        auto targetMs = std::chrono::duration_cast<
            std::chrono::milliseconds
        >(targetTime.time_since_epoch()).count();

        Segment* closest = nullptr;

        for (const auto& segment : segments) {
            if (segment.timestamp.timestampMs <= targetMs) {
                closest = const_cast<Segment*>(&segment);
            } else {
                break;
            }
        }

        return closest ? std::optional<Segment>(*closest) : std::nullopt;
    }

    /**
     * Obtiene la lista cronológica de todos los segmentos
     */
    const std::vector<Segment>& getTimeline() const {
        return segments;
    }
};

Casos de uso del mundo real

Reconstrucción de eventos multicámara

Sincroniza el material de varias cámaras para reconstruir incidentes de seguridad:

Event Timeline Reconstruction

from datetime import datetime, timedelta

class EventReconstructor:
    def __init__(self):
        self.camera_segments = {}  # cameraId -> lista de segmentos

    def add_camera_footage(self, camera_id, segment_files):
        """Agrega segmentos de video de una cámara específica."""
        segments = []

        for file_path in segment_files:
            timestamp_ms, timestamp_dt = extract_rhombus_timestamp(file_path)
            if timestamp_ms:
                segments.append({
                    'file': file_path,
                    'start_time': timestamp_dt,
                    'timestamp_ms': timestamp_ms
                })

        self.camera_segments[camera_id] = sorted(
            segments,
            key=lambda x: x['timestamp_ms']
        )

    def reconstruct_event(self, event_time, window_seconds=30):
        """
        Encuentra todos los segmentos de cámara dentro de la ventana de tiempo de un evento.

        Args:
            event_time: datetime del evento
            window_seconds: segundos antes/después del evento a incluir
        """
        window = timedelta(seconds=window_seconds)
        start_time = event_time - window
        end_time = event_time + window

        relevant_footage = {}

        for camera_id, segments in self.camera_segments.items():
            camera_clips = []

            for segment in segments:
                # Comprobar si el segmento se superpone con la ventana del evento
                if start_time <= segment['start_time'] <= end_time:
                    offset = (segment['start_time'] - event_time).total_seconds()
                    camera_clips.append({
                        'file': segment['file'],
                        'start_time': segment['start_time'],
                        'offset_from_event': offset
                    })

            if camera_clips:
                relevant_footage[camera_id] = camera_clips

        return relevant_footage

# Uso
reconstructor = EventReconstructor()
reconstructor.add_camera_footage('entrance', entrance_files)
reconstructor.add_camera_footage('lobby', lobby_files)
reconstructor.add_camera_footage('parking', parking_files)

# Reconstruir el evento en una hora específica
event_time = datetime(2024, 8, 6, 15, 21, 18)
footage = reconstructor.reconstruct_event(event_time, window_seconds=30)

print(f"Footage for event at {event_time}:")
for camera_id, clips in footage.items():
    print(f"\n{camera_id}:")
    for clip in clips:
        print(f"  - {clip['file']}")
        print(f"    Offset: {clip['offset_from_event']:.2f}s")

Correlación de datos de sensores

Alinea el video con eventos de control de acceso o de sensores ambientales:

Sensor-Video Correlation

class SensorVideoCorrelator {
    constructor() {
        this.videoTimestamps = [];
        this.sensorEvents = [];
    }

    /**
     * Agrega un segmento de video con su marca de tiempo
     */
    addVideoSegment(cameraId, timestamp, videoUrl) {
        this.videoTimestamps.push({
            cameraId,
            timestamp: timestamp.timestampMs,
            datetime: timestamp.utcTime,
            url: videoUrl
        });
    }

    /**
     * Agrega un evento de sensor con su marca de tiempo
     */
    addSensorEvent(sensorId, eventType, timestampMs, data) {
        this.sensorEvents.push({
            sensorId,
            eventType,
            timestamp: timestampMs,
            datetime: new Date(timestampMs),
            data
        });
    }

    /**
     * Encuentra la cobertura de video para un evento de sensor
     */
    findVideoForEvent(eventTimestampMs, cameras = null) {
        const relevantVideos = this.videoTimestamps.filter(video => {
            // El segmento de video comienza antes o en la hora del evento
            const inTimeRange = video.timestamp <= eventTimestampMs;

            // Filtrar por cámara si se especifica
            const inCameraList = !cameras || cameras.includes(video.cameraId);

            return inTimeRange && inCameraList;
        });

        // Obtener el video más reciente antes del evento para cada cámara
        const latestByCamera = {};

        relevantVideos.forEach(video => {
            if (!latestByCamera[video.cameraId] ||
                video.timestamp > latestByCamera[video.cameraId].timestamp) {
                latestByCamera[video.cameraId] = video;
            }
        });

        return Object.values(latestByCamera);
    }

    /**
     * Genera un informe de correlación
     */
    generateCorrelationReport() {
        return this.sensorEvents.map(event => {
            const videos = this.findVideoForEvent(event.timestamp);

            return {
                event: {
                    type: event.eventType,
                    sensor: event.sensorId,
                    time: event.datetime.toISOString(),
                    data: event.data
                },
                associatedVideos: videos.map(v => ({
                    camera: v.cameraId,
                    url: v.url,
                    offset: event.timestamp - v.timestamp
                }))
            };
        });
    }
}

// Ejemplo de uso
const correlator = new SensorVideoCorrelator();

// Agregar un evento de acceso a una puerta
correlator.addSensorEvent(
    'door-entrance-1',
    'ACCESS_GRANTED',
    1722945678123,
    { userId: 'user123', cardId: '12345' }
);

// Agregar el video correspondiente
correlator.addVideoSegment(
    'camera-entrance',
    { timestampMs: 1722945670000, utcTime: new Date(1722945670000) },
    'https://media.rhombussystems.com/segment1.mp4'
);

// Generar informe
const report = correlator.generateCorrelationReport();
console.log(JSON.stringify(report, null, 2));

Mejores prácticas para la integración

Validar la firma

Confirma siempre la etiqueta rhom antes de interpretar los bytes siguientes como una marca de tiempo. Esto evita la mala interpretación de datos no relacionados.

def is_valid_rhombus_timestamp(data, rhom_index):
    # Verificar la firma 'rhom'
    if data[rhom_index:rhom_index + 4] != b'rhom':
        return False

    # Verificar que haya datos suficientes para la marca de tiempo
    if len(data) < rhom_index + 12:
        return False

    # Extraer y validar el rango de la marca de tiempo
    timestamp_bytes = data[rhom_index + 4:rhom_index + 12]
    timestamp_ms = int.from_bytes(timestamp_bytes, byteorder="big")

    # Verificación de coherencia: la marca de tiempo debe ser razonable
    # (entre 2015 y 2050)
    min_timestamp = 1420070400000  # 1 de enero de 2015
    max_timestamp = 2524608000000  # 1 de enero de 2050

    return min_timestamp <= timestamp_ms <= max_timestamp

Manejar las zonas horarias correctamente

La marca de tiempo está basada en UTC. Conviértela apropiadamente si tu aplicación necesita la hora local.

import datetime as dt
import pytz

def convert_to_local_time(timestamp_ms, timezone='America/New_York'):
    # Crear un datetime UTC
    utc_dt = dt.datetime.fromtimestamp(timestamp_ms / 1000.0, tz=dt.timezone.utc)
    utc_dt = pytz.utc.localize(utc_dt)

    # Convertir a la zona horaria local
    local_tz = pytz.timezone(timezone)
    local_dt = utc_dt.astimezone(local_tz)

    return local_dt

# Uso
timestamp_ms = 1722945678123
local_time = convert_to_local_time(timestamp_ms, 'America/Los_Angeles')
print(f"Local time: {local_time}")

Usar como reloj de referencia

Combina la marca de tiempo de inicio del segmento con las marcas de tiempo de los fotogramas para una alineación con precisión de fotograma.

class FrameTimestampCalculator {
    constructor(segmentStartMs, frameRate) {
        this.segmentStartMs = segmentStartMs;
        this.frameRate = frameRate;
        this.frameDurationMs = 1000 / frameRate;
    }

    /**
     * Calcula la marca de tiempo absoluta para un fotograma específico
     */
    getFrameTimestamp(frameNumber) {
        const offsetMs = frameNumber * this.frameDurationMs;
        return this.segmentStartMs + offsetMs;
    }

    /**
     * Encuentra el número de fotograma para una marca de tiempo específica
     */
    getFrameAtTimestamp(targetTimestampMs) {
        const offsetMs = targetTimestampMs - this.segmentStartMs;
        return Math.floor(offsetMs / this.frameDurationMs);
    }
}

// Uso
const segmentTimestamp = 1722945678123;
const calculator = new FrameTimestampCalculator(segmentTimestamp, 30);

// Obtener la marca de tiempo del fotograma 150
const frameTime = calculator.getFrameTimestamp(150);
console.log('Frame 150 timestamp:', new Date(frameTime));

// Encontrar el fotograma en una hora específica
const targetTime = 1722945683123;
const frameNum = calculator.getFrameAtTimestamp(targetTime);
console.log('Frame at target time:', frameNum);

Versiona para prepararte a futuro

Almacena tu lógica de análisis de forma modular por si Rhombus agrega nuevos formatos de metadatos.

class RhombusMetadataParser:
    """Analizador extensible para los formatos de metadatos de Rhombus."""

    VERSION = "1.0"

    def __init__(self):
        self.parsers = {
            b'rhom': self._parse_v1_timestamp
        }

    def parse(self, file_path):
        """Analiza los metadatos de un archivo de video."""
        with open(file_path, "rb") as f:
            data = f.read()

        # Buscar la caja 'free'
        free_idx = data.find(b'free')
        if free_idx == -1:
            return None

        # Comprobar todas las firmas conocidas
        for signature, parser_func in self.parsers.items():
            sig_idx = data.find(signature, free_idx)
            if sig_idx != -1:
                return parser_func(data, sig_idx)

        return None

    def _parse_v1_timestamp(self, data, rhom_idx):
        """Analiza el formato de marca de tiempo v1."""
        timestamp_bytes = data[rhom_idx + 4:rhom_idx + 12]
        timestamp_ms = int.from_bytes(timestamp_bytes, byteorder="big")

        return {
            'version': 1,
            'type': 'timestamp',
            'timestamp_ms': timestamp_ms,
            'datetime': datetime.fromtimestamp(timestamp_ms / 1000.0, tz=datetime.timezone.utc)
        }

    def add_parser(self, signature, parser_func):
        """Agrega un analizador personalizado para nuevos formatos de metadatos."""
        self.parsers[signature] = parser_func

# Uso
parser = RhombusMetadataParser()
metadata = parser.parse("video_segment.mp4")

if metadata:
    print(f"Version: {metadata['version']}")
    print(f"Type: {metadata['type']}")
    print(f"Timestamp: {metadata['datetime']}")

Consideraciones de rendimiento

Lectura eficiente de archivos

Para archivos grandes, lee solo la parte del encabezado en lugar del archivo completo

Estrategia de caché

Almacena en caché las marcas de tiempo extraídas para evitar volver a analizar los mismos archivos

Procesamiento por lotes

Procesa varios archivos en paralelo al construir líneas de tiempo

Administración de memoria

Usa analizadores de streaming para archivos de video muy grandes

Lectura optimizada de archivos

Optimized Parser

def extract_rhombus_timestamp_optimized(file_path, max_search_bytes=100_000):
    """
    Versión optimizada que lee solo el comienzo del archivo.
    La mayoría de los metadatos están en la primera parte de los archivos MP4.
    """
    with open(file_path, "rb") as f:
        # Leer solo la parte del encabezado
        data = f.read(max_search_bytes)

    # Buscar la firma 'rhom'
    rhom_idx = data.find(b'rhom')
    if rhom_idx == -1:
        return None, None

    # Verificar que tenemos datos suficientes para la marca de tiempo
    if len(data) < rhom_idx + 12:
        return None, None

    timestamp_bytes = data[rhom_idx + 4:rhom_idx + 12]
    timestamp_ms = int.from_bytes(timestamp_bytes, byteorder="big")
    timestamp_dt = datetime.fromtimestamp(timestamp_ms / 1000.0, tz=datetime.timezone.utc)

    return timestamp_ms, timestamp_dt

Conclusión

El método de Rhombus de embeber una marca de tiempo UTC con precisión de milisegundos en el átomo free de los segmentos ISOBMFF proporciona a los desarrolladores una herramienta poderosa para la alineación precisa de eventos en entornos de múltiples flujos. Este enfoque preserva la compatibilidad con las herramientas de video existentes a la vez que desbloquea precisión de subsegundos para análisis, IA y monitoreo en tiempo real—algo crítico para integraciones avanzadas en el ecosistema de Rhombus.

Próximos pasos para los desarrolladores:

Experimenta con la biblioteca ISOBMFF de GitHub para analizar segmentos de Rhombus
Usa el visor en línea MP4Box.js para inspeccionar visualmente las estructuras de cajas
Incorpora la extracción de marcas de tiempo en tu canalización de ingesta para conjuntos de datos multiorigen perfectamente sincronizados

Recursos adicionales

Guía del reproductor de video

Aprende a implementar streaming de video en vivo con DashJS

Referencia de la API

Explora los endpoints de la API de cámara y medios, así como otras opciones de video

Especificación ISOBMFF

Lee la especificación oficial del ISO Base Media File Format

Comunidad de desarrolladores

Obtén ayuda y comparte implementaciones en la Comunidad de Desarrolladores

Soporte

¿Necesitas ayuda con la extracción de marcas de tiempo o la sincronización de video?

Correo electrónico: support@rhombus.com
Comunidad: rhombus.community
Documentación: Explora nuestra referencia completa de la API

Esta guía de implementación avanzada se actualiza periódicamente para reflejar las mejores prácticas más recientes para trabajar con segmentos de video de Rhombus.

Comenzar

Guías de implementación

Eventos en tiempo real

Comprender ISOBMFF y el átomo “free”

La marca de tiempo personalizada de Rhombus

Por qué esto es importante para los desarrolladores

Alineación multicámara

Fusión de sensores

Precisión forense

Menor deriva

Recuperar la marca de tiempo

Analizar archivos ISOBMFF

Analizador ISOBMFF (C++/Swift)

Visor interactivo MP4Box.js

Ejemplos de implementación

Uso avanzado en Python

Uso avanzado en JavaScript

Uso avanzado en C++

Casos de uso del mundo real

Reconstrucción de eventos multicámara

Correlación de datos de sensores

Mejores prácticas para la integración

Consideraciones de rendimiento

Lectura eficiente de archivos

Estrategia de caché

Procesamiento por lotes

Administración de memoria

Lectura optimizada de archivos

Conclusión

Recursos adicionales

Guía del reproductor de video

Referencia de la API

Especificación ISOBMFF

Comunidad de desarrolladores

Soporte

​Comprender ISOBMFF y el átomo “free”

​La marca de tiempo personalizada de Rhombus

​Por qué esto es importante para los desarrolladores

Alineación multicámara

Fusión de sensores

Precisión forense

Menor deriva

​Recuperar la marca de tiempo

​Analizar archivos ISOBMFF

Analizador ISOBMFF (C++/Swift)

Visor interactivo MP4Box.js

​Ejemplos de implementación

​Uso avanzado en Python

​Uso avanzado en JavaScript

​Uso avanzado en C++

​Casos de uso del mundo real

​Reconstrucción de eventos multicámara

​Correlación de datos de sensores

​Mejores prácticas para la integración

​Consideraciones de rendimiento

Lectura eficiente de archivos

Estrategia de caché

Procesamiento por lotes

Administración de memoria

​Lectura optimizada de archivos

​Conclusión

​Recursos adicionales

Guía del reproductor de video

Referencia de la API

Especificación ISOBMFF

Comunidad de desarrolladores

​Soporte

Comprender ISOBMFF y el átomo “free”

La marca de tiempo personalizada de Rhombus

Por qué esto es importante para los desarrolladores

Recuperar la marca de tiempo

Analizar archivos ISOBMFF

Ejemplos de implementación

Uso avanzado en Python

Uso avanzado en JavaScript

Uso avanzado en C++

Casos de uso del mundo real

Reconstrucción de eventos multicámara

Correlación de datos de sensores

Mejores prácticas para la integración

Consideraciones de rendimiento

Lectura optimizada de archivos

Conclusión

Recursos adicionales

Soporte