Zuletzt aktualisiert: 11. September

FFmpeg mit Scala: Video in Echtzeit

Kevin van Zonneveld

Mitgründer · Amsterdam, Niederlande ·

Die Videoverarbeitung in Echtzeit ist für Anwendungen wie die Übertragung von Live-Events, die Videoüberwachung und interaktives Streaming unverzichtbar. Scala bietet in Kombination mit FFmpeg eine leistungsstarke und effiziente Lösung für diese anspruchsvollen Aufgaben.

Herausforderungen bei der Videoverarbeitung in Echtzeit

Bei der Videoverarbeitung in Echtzeit werden kontinuierliche Videodatenströme mit minimaler Latenz verarbeitet. Zu den zentralen Herausforderungen gehören:

Geringe Latenz aufrechterhalten
Hohen Durchsatz sicherstellen
Verschiedene Eingabequellen (RTSP, RTMP) verarbeiten
Ressourcenintensive Vorgänge effizient steuern

Scala und FFmpeg einrichten

Kompatibilitätsmatrix

Stellen Sie vor dem Start sicher, dass Sie kompatible Versionen verwenden:

Komponente	Mindestversion	Empfohlene Version
Scala	3.3.0	3.6.3
Scala CLI	1.0.0	1.6.2
FFmpeg	v4.4	v6.0

Scala CLI installieren

Scala CLI bietet eine moderne und komfortable Möglichkeit, mit Scala zu arbeiten:

# 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

FFmpeg installieren

Für die Videoverarbeitung wird FFmpeg benötigt:

# Ubuntu/Debian
sudo apt update
sudo apt install ffmpeg

# macOS (homebrew)
brew install ffmpeg

# Verify installation
ffmpeg -version

FFmpeg-Grundlagen

FFmpeg ist ein vielseitiges Multimedia-Framework, das Videos transkodieren, in der Größe ändern, mit Wasserzeichen versehen und streamen kann. Hier einige grundlegende Befehle:

Transkodieren:

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

Größenänderung:

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

Wasserzeichen hinzufügen:

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

FFmpeg in Scala integrieren

Scala kann FFmpeg-Befehle mithilfe der Process-API ausführen. Hier eine robuste Implementierung mit sauberer Fehlerbehandlung:

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")
}

Verarbeitung von Videostreams in Echtzeit

Beispiel 1: Lesen aus rtsp/rtmp

So lesen Sie mit sauberer Fehlerbehandlung aus einer Netzwerkquelle:

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")

Beispiel 2: Zeitstempel-Overlay hinzufügen

Legen Sie vor der Verwendung dieses Beispiels eine installierte TrueType-Schriftart unter font.ttf im Arbeitsverzeichnis ab. Die explizit angegebene Schriftdatei unterstützt FFmpeg-Builds ohne fontconfig; der Build benötigt weiterhin 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")
  }
}

Beispiel 3: Ausgabe an HLS oder MPEG-DASH

Streaming-Ausgabe mit korrekter Konfiguration:

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")
  }
}

Fehlerbehandlung und Ressourcenverwaltung

Bei der Arbeit mit externen Prozessen ist eine saubere Ressourcenverwaltung entscheidend:

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()
    }
  }
}

FFmpeg-Integrationen testen

Verwenden Sie den Helper runFFmpeg im selben Package wie der Test. Stellen Sie eine echte Fixture test_resources/sample.mp4 und ein leeres Verzeichnis test_output bereit:

//> 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())
  }
}

Performance optimieren

Nutzen Sie die Nebenläufigkeitsfunktionen von Scala, um die Performance zu verbessern:

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")
}

Häufige Probleme beheben

Bei der Arbeit mit FFmpeg und Scala können Ihnen diese häufigen Probleme begegnen:

FFmpeg nicht im PATH gefunden: Stellen Sie sicher, dass FFmpeg korrekt installiert und im PATH Ihres Systems verfügbar ist.

Fehlgeschlagene Stream-Verbindungen: Netzwerkprobleme können dazu führen, dass RTSP/RTMP-Verbindungen fehlschlagen. Implementieren Sie eine Retry-Logik:

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
}

Ressourcenerschöpfung: Die Videoverarbeitung ist ressourcenintensiv. Überwachen Sie die Systemressourcen und implementieren Sie bei Bedarf eine Drosselung.

Fortgeschrittene Techniken

FFmpeg unterstützt benutzerdefinierte Filter und Effekte. So wenden Sie etwa einen Weichzeichnerfilter mit sauberer Fehlerbehandlung an:

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)

Mit den robusten Funktionen von Scala und den leistungsstarken Möglichkeiten von FFmpeg können Entwicklerinnen und Entwickler effiziente und zuverlässige Anwendungen für die Videoverarbeitung in Echtzeit erstellen. Transloadit setzt FFmpeg in mehreren seiner Robots ein, etwa in 🤖/video/encode und /video/adaptive.