logging: improve ffmpeg error message quality

src/aind_behavior_video_transformation/etl.py

@@ -1,9 +1,11 @@
 """Module that defines the ETL class for behavior video transformations."""
 
 import logging
+import shlex
 import sys
 from concurrent.futures import ProcessPoolExecutor, as_completed
 from pathlib import Path
+from subprocess import CalledProcessError
 from time import time
 from typing import List, Optional, Tuple, Union
 
@@ -24,6 +26,25 @@
     convert_video,
 )
 
+logger = logging.getLogger(__name__)
+
+
+def _format_ffmpeg_error(video_path: Path, exc: CalledProcessError) -> str:
+    """Format an ffmpeg ``CalledProcessError`` as a single log record body.
+
+    The format keeps each piece on its own line so ffmpeg stderr is
+    clearly visible to readers and to log aggregators.
+    """
+    stderr = (exc.stderr or "(no stderr captured)").rstrip("\n")
+    return (
+        f"FFmpeg conversion failed for {video_path}\n"
+        f"Command: {shlex.join(exc.cmd)}\n"
+        f"Return code: {exc.returncode}\n"
+        f"--- ffmpeg stderr ---\n"
+        f"{stderr}\n"
+        f"--- end stderr ---"
+    )
+
 
 class BehaviorVideoJobSettings(BasicJobSettings):
     """
@@ -77,52 +98,72 @@ class BehaviorVideoJob(GenericEtl[BehaviorVideoJobSettings]):
     run_job() -> JobResponse
     """
 
+    def _run_parallel(
+        self,
+        convert_video_args: list[tuple[Path, Path, tuple[str, str] | None]],
+    ) -> list[tuple[Path, CalledProcessError]]:
+        """Run conversions in a ProcessPoolExecutor, collecting failures."""
+        errors: list[tuple[Path, CalledProcessError]] = []
+        if not convert_video_args:
+            return errors
+        thread_cnt = self.job_settings.ffmpeg_thread_cnt
+        with ProcessPoolExecutor(max_workers=len(convert_video_args)) as ex:
+            futures = {
+                ex.submit(convert_video, *params, thread_cnt): params
+                for params in convert_video_args
+            }
+            for future in as_completed(futures):
+                video_path = futures[future][0]
+                try:
+                    result = future.result()
+                except CalledProcessError as exc:
+                    errors.append((video_path, exc))
+                else:
+                    logger.info("FFmpeg job completed: %s", result)
+        return errors
+
+    def _run_serial(
+        self,
+        convert_video_args: list[tuple[Path, Path, tuple[str, str] | None]],
+    ) -> list[tuple[Path, CalledProcessError]]:
+        """Run conversions one at a time, collecting failures."""
+        errors: list[tuple[Path, CalledProcessError]] = []
+        thread_cnt = self.job_settings.ffmpeg_thread_cnt
+        for params in convert_video_args:
+            video_path = params[0]
+            try:
+                result = convert_video(*params, thread_cnt)
+            except CalledProcessError as exc:
+                errors.append((video_path, exc))
+            else:
+                logger.info("FFmpeg job completed: %s", result)
+        return errors
+
     def _run_compression(
         self,
         convert_video_args: list[tuple[Path, Path, tuple[str, str] | None]],
     ) -> None:
         """
         Runs CompressionRequests at the specified paths.
         """
-        error_traces = []
         if self.job_settings.parallel_compression:
-            # Execute in-parallel
-            if len(convert_video_args) == 0:
-                return
-
-            num_jobs = len(convert_video_args)
-            with ProcessPoolExecutor(max_workers=num_jobs) as executor:
-                jobs = [
-                    executor.submit(
-                        convert_video,
-                        *params,
-                        self.job_settings.ffmpeg_thread_cnt,
-                    )
-                    for params in convert_video_args
-                ]
-                for job in as_completed(jobs):
-                    result = job.result()
-                    if isinstance(result, tuple):
-                        error_traces.append(result[1])
-                    else:
-                        logging.info(f"FFmpeg job completed: {result}")
+            errors = self._run_parallel(convert_video_args)
         else:
-            # Execute serially
-            for params in convert_video_args:
-                result = convert_video(
-                    *params, self.job_settings.ffmpeg_thread_cnt
-                )
-                if isinstance(result, tuple):
-                    error_traces.append(result[1])
-                else:
-                    logging.info(f"FFmpeg job completed: {result}")
-
-        if error_traces:
-            for e in error_traces:
-                logging.error(e)
-            raise RuntimeError(
-                "One or more Ffmpeg jobs failed. See error logs."
-            )
+            errors = self._run_serial(convert_video_args)
+
+        if not errors:
+            return
+
+        formatted = [
+            _format_ffmpeg_error(video_path, exc)
+            for video_path, exc in errors
+        ]
+        for block in formatted:
+            logger.error(block)
+        raise RuntimeError(
+            f"{len(errors)} ffmpeg job(s) failed:\n\n"
+            + "\n\n".join(formatted)
+        )
 
     def run_job(self) -> JobResponse:
         """

src/aind_behavior_video_transformation/transform_videos.py

@@ -13,16 +13,18 @@
 FfmpegInputArgs / FfmpegOutputArgs can be prexisitng or newly-defined in (1)
 """
 
+import logging
 import shlex
 import subprocess
 from enum import Enum
 from os import symlink
 from pathlib import Path
-from subprocess import CalledProcessError
-from typing import Optional, Tuple, Union
+from typing import Optional, Tuple
 
 from pydantic import BaseModel, Field
 
+logger = logging.getLogger(__name__)
+
 
 class CompressionEnum(Enum):
     """
@@ -190,7 +192,7 @@ def convert_video(
     output_dir: Path,
     arg_set: Optional[Tuple[str, str]],
     ffmpeg_thread_cnt: int = 0,
-) -> Union[str, Tuple[str, str]]:
+) -> str:
     """
     Converts a video to a specified format using ffmpeg.
 
@@ -207,51 +209,40 @@ def convert_video(
 
     Returns
     -------
-    Path
+    str
         The path to the converted video file.
 
+    Raises
+    ------
+    subprocess.CalledProcessError
+        If ffmpeg exits with a non-zero return code.  The exception's
+        ``cmd``, ``returncode``, and ``stderr`` attributes carry the
+        full failure context for the caller to log.
+
     Notes
     -----
-    - The function uses ffmpeg for video conversion.
     - If `arg_set` is None, the function creates a symlink to the original
         video file.
     """
 
-    out_path = output_dir / f"{video_path.stem}.mp4"  # noqa: E501
+    out_path = output_dir / f"{video_path.stem}.mp4"
 
-    # Trivial Case, do nothing
     if arg_set is None:
         symlink(video_path, out_path)
-        return out_path
-
-    input_args = arg_set[0]
-    output_args = arg_set[1]
+        return str(out_path)
 
+    input_args, output_args = arg_set
     ffmpeg_command = ["ffmpeg", "-y", "-v", "warning", "-hide_banner"]
-
-    # Set thread count
     if ffmpeg_thread_cnt > 0:
         ffmpeg_command.extend(["-threads", str(ffmpeg_thread_cnt)])
-
     if input_args:
         ffmpeg_command.extend(shlex.split(input_args))
     ffmpeg_command.extend(["-i", str(video_path)])
     if output_args:
         ffmpeg_command.extend(shlex.split(output_args))
     ffmpeg_command.append(str(out_path))
 
-    # Capture and return error message if it exists
-    try:
-        subprocess.run(
-            ffmpeg_command, check=True, capture_output=True, text=True
-        )
-        return str(out_path)
-
-    except CalledProcessError as e:
-        error_msg = (
-            f"FFmpeg conversion failed for {video_path}\n"
-            f"Command: {' '.join(ffmpeg_command)}\n"
-            f"Return code: {e.returncode}\n"
-            f"Error output:\n{e.stderr}\n"
-        )
-        return (str(out_path), error_msg)
+    subprocess.run(
+        ffmpeg_command, check=True, capture_output=True, text=True
+    )
+    return str(out_path)