Última actualización: 11 de septiembre

FFmpeg con Scala: video en tiempo real

Kevin van Zonneveld

Cofundador · Ámsterdam, Países Bajos ·

El procesamiento de video en tiempo real es esencial para aplicaciones como la retransmisión de eventos en vivo, la videovigilancia y el streaming interactivo. Scala, combinado con FFmpeg, ofrece una solución potente y eficiente para gestionar estas tareas exigentes.

Retos del procesamiento de video en tiempo real

El procesamiento de video en tiempo real implica gestionar streams continuos de datos de video con una latencia mínima. Entre los principales retos están:

Mantener una latencia baja
Garantizar un alto rendimiento de procesamiento
Gestionar diversas fuentes de entrada (RTSP, RTMP)
Administrar de forma eficiente las operaciones que consumen muchos recursos

Configurar Scala y FFmpeg

Matriz de compatibilidad

Antes de empezar, asegúrate de usar versiones compatibles:

Componente	Versión mínima	Versión recomendada
Scala	3.3.0	3.6.3
Scala CLI	1.0.0	1.6.2
FFmpeg	v4.4	v6.0

Instalar Scala CLI

Scala CLI ofrece una forma moderna y cómoda de trabajar con Scala:

# For Ubuntu/Debian
curl -sSLf https://scala-cli.virtuslab.org/get | sh
source ~/.profile  # Required after installation

# For macOS
brew install Virtuslab/scala-cli/scala-cli

# For Windows
winget install virtuslab.scalacli

Instalar FFmpeg

FFmpeg es necesario para el procesamiento de video:

# Ubuntu/Debian
sudo apt update
sudo apt install ffmpeg

# macOS (homebrew)
brew install ffmpeg

# Verify installation
ffmpeg -version

Conceptos básicos de FFmpeg

FFmpeg es un framework multimedia versátil capaz de transcodificar, redimensionar, aplicar marcas de agua y transmitir video. Estos son algunos comandos básicos:

Transcodificar:

ffmpeg -i input.mp4 -c:v libx264 output.mp4

Redimensionar:

ffmpeg -i input.mp4 -vf scale=1280:720 output.mp4

Añadir una marca de agua:

ffmpeg -i input.mp4 -i watermark.png -filter_complex "overlay=10:10" output.mp4

Integrar FFmpeg con Scala

Scala puede ejecutar comandos de FFmpeg mediante la Process API. Esta es una implementación robusta con un manejo de errores adecuado:

import scala.sys.process._
import scala.util.{Try, Success, Failure}

def runFFmpeg(args: Seq[String]): Either[String, Int] = {
  Try(Process("ffmpeg" +: args).!) match {
    case Success(0) => Right(0)
    case Success(code) => Left(s"FFmpeg failed with exit code $code")
    case Failure(e) => Left(s"FFmpeg execution failed: ${e.getMessage}")
  }
}

// Example usage
val result = runFFmpeg(Seq("-i", "input.mp4", "-c:v", "libx264", "output.mp4"))
result match {
  case Right(_) => println("Processing successful")
  case Left(error) => println(s"Error: $error")
}

Procesamiento de streams de video en tiempo real

Ejemplo 1: lectura desde rtsp/rtmp

Para leer desde una fuente de red con un manejo de errores adecuado:

def processStream(streamUrl: String, outputPath: String): Either[String, Unit] = {
  val args = Seq(
    "-i", streamUrl,
    "-c:v", "copy",
    "-c:a", "copy",
    outputPath
  )

  runFFmpeg(args) match {
    case Right(_) => Right(())
    case Left(error) => Left(s"Stream processing failed: $error")
  }
}

// Example usage
val streamUrl = "rtsp://example.com/live"
processStream(streamUrl, "output.mp4")

Ejemplo 2: añadir una superposición con marca de tiempo

Coloca una fuente TrueType instalada en font.ttf dentro del directorio de trabajo antes de usar este ejemplo. El archivo de fuente explícito da soporte a compilaciones de FFmpeg sin fontconfig; la compilación sigue necesitando drawtext.

def addTimestamp(streamUrl: String, outputPath: String): Either[String, Unit] = {
  val args = Seq(
    "-i", streamUrl,
    "-vf", "drawtext=fontfile=font.ttf:text='%{localtime}':fontcolor=white:fontsize=24:x=10:y=10",
    "-c:v", "libx264",
    "-c:a", "copy",
    outputPath
  )

  runFFmpeg(args) match {
    case Right(_) => Right(())
    case Left(error) => Left(s"Timestamp overlay failed: $error")
  }
}

Ejemplo 3: salida a HLS o MPEG-DASH

Salida de streaming con la configuración adecuada:

def createHLSStream(streamUrl: String, outputPath: String): Either[String, Unit] = {
  val args = Seq(
    "-i", streamUrl,
    "-c:v", "libx264",
    "-c:a", "aac",
    "-hls_time", "4",
    "-hls_playlist_type", "event",
    "-hls_segment_filename", s"${outputPath}_%03d.ts",
    s"$outputPath.m3u8"
  )

  runFFmpeg(args) match {
    case Right(_) => Right(())
    case Left(error) => Left(s"HLS streaming failed: $error")
  }
}

Manejo de errores y gestión de recursos

Al trabajar con procesos externos, una gestión adecuada de los recursos es crucial:

import scala.sys.process._
import scala.util.{Try, Success, Failure}
import java.io.{File, IOException}
import java.nio.file.Files

def processVideoWithCleanup(input: String, output: String): Either[String, Unit] = {
  val tempDir = Files.createTempDirectory("video-processing-").toFile

  try {
    val args = Seq(
      "-i", input,
      "-c:v", "libx264",
      s"${tempDir.getPath}/temp_output.mp4"
    )

    runFFmpeg(args) match {
      case Right(_) =>
        // Move from temp to final location
        val tempFile = new File(s"${tempDir.getPath}/temp_output.mp4")
        val outputFile = new File(output)
        Files.move(tempFile.toPath, outputFile.toPath) // Fails if the destination already exists
        Right(())

      case Left(error) => Left(error)
    }
  } catch {
    case e: IOException => Left(s"I/O error: ${e.getMessage}")
    case e: Exception => Left(s"Unexpected error: ${e.getMessage}")
  } finally {
    // Clean up temp files
    if (tempDir.exists()) {
      tempDir.listFiles().foreach(_.delete())
      tempDir.delete()
    }
  }
}

Probar integraciones de FFmpeg

Usa el helper runFFmpeg en el mismo paquete que la prueba. Proporciona un fixture test_resources/sample.mp4 real y un directorio test_output vacío:

//> using test.dep org.scalatest::scalatest:3.2.19

import org.scalatest.funsuite.AnyFunSuite
import java.io.File

class FFmpegIntegrationTest extends AnyFunSuite {
  test("ffmpeg is installed and accessible") {
    val result = runFFmpeg(Seq("-version"))
    assert(result.isRight)
  }

  test("can process a test video file") {
    val testFile = "test_resources/sample.mp4"
    val outputFile = "test_output/result.mp4"
    val result = runFFmpeg(Seq(
      "-i", testFile,
      "-t", "5", // Process only first 5 seconds
      "-c:v", "libx264",
      outputFile
    ))
    assert(result.isRight)
    assert(new File(outputFile).exists())
  }
}

Optimizar el rendimiento

Usa las funciones de concurrencia de Scala para mejorar el rendimiento:

import scala.concurrent._
import scala.concurrent.duration._
import ExecutionContext.Implicits.global

def processMultipleStreams(streams: List[(String, String)]): Future[List[Either[String, Unit]]] = {
  val futures = streams.map { case (input, output) =>
    Future {
      val args = Seq(
        "-i", input,
        "-c:v", "libx264",
        output
      )
      runFFmpeg(args) match {
        case Right(_) => Right(())
        case Left(error) => Left(s"Processing $input failed: $error")
      }
    }
  }

  Future.sequence(futures)
}

// Example usage
val streams = List(
  ("rtsp://camera1.example.com/live", "camera1_output.mp4"),
  ("rtsp://camera2.example.com/live", "camera2_output.mp4")
)

val results = Await.result(processMultipleStreams(streams), 10.minutes)
results.foreach {
  case Right(_) => println("Stream processed successfully")
  case Left(error) => println(s"Error: $error")
}

Solución de problemas comunes

Al trabajar con FFmpeg y Scala, podrías encontrarte con estos problemas comunes:

No se encuentra FFmpeg en el PATH: asegúrate de que FFmpeg esté correctamente instalado y accesible en el PATH de tu sistema.

Fallos de conexión del stream: los problemas de red pueden hacer que fallen las conexiones RTSP/RTMP. Implementa una lógica de reintentos:

def processStreamWithRetry(streamUrl: String, output: String, maxRetries: Int = 3): Either[String, Unit] = {
  var attempts = 0
  var result: Either[String, Unit] = Left("Not attempted yet")

  while (attempts < maxRetries && result.isLeft) {
    attempts += 1
    result = processStream(streamUrl, output)
    if (result.isLeft && attempts < maxRetries) {
      println(s"Attempt $attempts failed, retrying...")
      Thread.sleep(2000) // Wait before retry
    }
  }

  result
}

Agotamiento de recursos: el procesamiento de video consume muchos recursos. Supervisa los recursos del sistema e implementa throttling si es necesario.

Técnicas avanzadas

FFmpeg admite filtros y efectos personalizados. Por ejemplo, aplicar un filtro de desenfoque con un manejo de errores adecuado:

def applyBlurFilter(input: String, output: String, blurAmount: Int = 10): Either[String, Unit] = {
  val args = Seq(
    "-i", input,
    "-vf", s"boxblur=$blurAmount",
    "-c:v", "libx264",
    output
  )

  runFFmpeg(args) match {
    case Right(_) => Right(())
    case Left(error) => Left(s"Blur filter failed: $error")
  }
}

// Example usage
applyBlurFilter("input.mp4", "output_blurred.mp4", 15)

Al aprovechar las robustas funciones de Scala y las potentes capacidades de FFmpeg, los desarrolladores pueden crear aplicaciones de procesamiento de video en tiempo real eficientes y fiables. Transloadit aprovecha FFmpeg en varios de sus Robots, como 🤖/video/encode y /video/adaptive.