FIX: Removing output_path paramater from video converter

doc/code/converters/4_video_converters.ipynb

@@ -68,7 +68,7 @@
     "\n",
     "await initialize_pyrit_async(memory_db_type=IN_MEMORY)  # type: ignore\n",
     "\n",
-    "input_video = str(pathlib.Path(\".\") / \"..\" / \"..\" / \"..\" / \"assets\" / \"sample_video.mp4\")\n",
+    "input_video = pathlib.Path(\".\") / \"..\" / \"..\" / \"..\" / \"assets\" / \"sample_video.mp4\"\n",
     "input_image = str(pathlib.Path(\".\") / \"..\" / \"..\" / \"..\" / \"assets\" / \"pyrit_architecture.png\")\n",
     "\n",
     "video = AddImageVideoConverter(video_path=input_video)\n",

doc/code/converters/4_video_converters.py

@@ -36,7 +36,7 @@
 
 await initialize_pyrit_async(memory_db_type=IN_MEMORY)  # type: ignore
 
-input_video = str(pathlib.Path(".") / ".." / ".." / ".." / "assets" / "sample_video.mp4")
+input_video = pathlib.Path(".") / ".." / ".." / ".." / "assets" / "sample_video.mp4"
 input_image = str(pathlib.Path(".") / ".." / ".." / ".." / "assets" / "pyrit_architecture.png")
 
 video = AddImageVideoConverter(video_path=input_video)

pyrit/converter/add_image_to_video_converter.py

@@ -4,6 +4,7 @@
 import asyncio
 import contextlib
 import logging
+import tempfile
 from pathlib import Path
 
 import numpy as np
@@ -38,27 +39,18 @@ class AddImageVideoConverter(Converter):
     def __init__(
         self,
         *,
-        video_path: str,
-        output_path: str | None = None,
+        video_path: Path,
         img_position: tuple[int, int] = (10, 10),
         img_resize_size: tuple[int, int] = (500, 500),
     ) -> None:
         """
         Initialize the converter with the video path and image properties.
 
         Args:
-            video_path (str): File path of video to add image to.
-            output_path (str, Optional): File path of output video. Defaults to None.
+            video_path (Path): File path of video to add image to.
             img_position (tuple): Position to place image in video. Defaults to (10, 10).
             img_resize_size (tuple): Size to resize image to. Defaults to (500, 500).
-
-        Raises:
-            ValueError: If ``video_path`` is empty or invalid.
         """
-        if not video_path:
-            raise ValueError("Please provide valid video path")
-
-        self._output_path = output_path
         self._img_position = img_position
         self._img_resize_size = img_resize_size
         self._video_path = video_path
@@ -78,16 +70,15 @@ def _build_identifier(self) -> ComponentIdentifier:
             }
         )
 
-    async def _add_image_to_video_async(self, image_path: str, output_path: str) -> str:
+    async def _add_image_to_video_async(self, image_path: str) -> bytes:
         """
         Add an image to video.
 
         Args:
             image_path (str): The image path to add to video.
-            output_path (str): The output video path.
 
         Returns:
-            str: The output video path.
+            bytes: The converted video data.
 
         Raises:
             ModuleNotFoundError: If OpenCV is not installed.
@@ -106,121 +97,111 @@ async def _add_image_to_video_async(self, image_path: str, output_path: str) ->
             category="prompt-memory-entries", data_type="image_path", value=image_path
         )
         input_video_data = data_serializer_factory(
-            category="prompt-memory-entries", data_type="video_path", value=self._video_path
+            category="prompt-memory-entries", data_type="video_path", value=str(self._video_path)
         )
 
         # Open the video to ensure it exists
         video_bytes = await input_video_data.read_data_async()
         input_image_bytes = await input_image_data.read_data_async()
 
-        azure_storage_flag = input_video_data._is_azure_storage_url(self._video_path)
-
-        await asyncio.to_thread(
+        return await asyncio.to_thread(
             self._add_image_to_video_sync,
             video_bytes=video_bytes,
             image_bytes=input_image_bytes,
-            output_path=output_path,
-            azure_storage_flag=azure_storage_flag,
         )
 
-        logger.info(f"Video saved as {output_path}")
-
-        return output_path
-
     def _add_image_to_video_sync(
         self,
         *,
         video_bytes: bytes,
         image_bytes: bytes,
-        output_path: str,
-        azure_storage_flag: bool,
-    ) -> None:
+    ) -> bytes:
         """
         Run the blocking cv2 pipeline and temp-file I/O on a worker thread.
 
         Designed to be invoked via ``asyncio.to_thread`` so the event loop is not blocked.
 
+        Returns:
+            bytes: The converted video data.
+
         Raises:
             ValueError: If the input video format is unsupported or the overlay image cannot
                 be decoded.
         """
         import cv2
 
-        video_path = self._video_path
-        local_temp_path: Path | None = None
+        file_extension = self._video_path.suffix.removeprefix(".").lower()
+        if file_extension not in video_encoding_map:
+            raise ValueError(f"Unsupported video format: {file_extension}")
+
+        with tempfile.NamedTemporaryFile(suffix=f".{file_extension}", dir=DB_DATA_PATH, delete=False) as input_file:
+            input_file.write(video_bytes)
+            input_path = Path(input_file.name)
+        with tempfile.NamedTemporaryFile(suffix=f".{file_extension}", dir=DB_DATA_PATH, delete=False) as output_file:
+            output_path = Path(output_file.name)
+
         cap: cv2.VideoCapture | None = None
         output_video: cv2.VideoWriter | None = None
 
         try:
-            if azure_storage_flag:
-                # If the video is in Azure storage, download it first
-
-                # Save the video bytes to a temporary file
-                local_temp_path = Path(DB_DATA_PATH, "temp_video.mp4")
-                with open(local_temp_path, "wb") as f:
-                    f.write(video_bytes)
-                video_path = str(local_temp_path)
-
-            cap = cv2.VideoCapture(video_path)
-
-            # Get video properties
-            fps = int(cap.get(cv2.CAP_PROP_FPS))
-            width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
-            height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
-            file_extension = video_path.split(".")[-1].lower()
-            if file_extension in video_encoding_map:
-                video_char_code = cv2.VideoWriter.fourcc(*video_encoding_map[file_extension])
-                output_video = cv2.VideoWriter(output_path, video_char_code, fps, (width, height))
-            else:
-                raise ValueError(f"Unsupported video format: {file_extension}")
-
-            # Load and resize the overlay image
-
-            image_np_arr = np.frombuffer(image_bytes, np.uint8)
-            decoded = cv2.imdecode(image_np_arr, cv2.IMREAD_UNCHANGED)
-            if decoded is None:
-                raise ValueError("Failed to decode overlay image")
-            overlay = cv2.resize(decoded, self._img_resize_size)
-
-            # Get overlay image dimensions
-            image_height, image_width, _ = overlay.shape
-            x, y = self._img_position  # Position where the overlay will be placed
-
-            while cap.isOpened():
-                ret, frame = cap.read()
-                if not ret:
-                    break
-
-                # Ensure overlay fits within the frame boundaries
-                if x + image_width > width or y + image_height > height:
-                    logger.info("Overlay image is too large for the video frame. Resizing to fit.")
-                    overlay = cv2.resize(overlay, (width - x, height - y))
-                    image_height, image_width, _ = overlay.shape
-
-                # Blend overlay with frame
-                if overlay.shape[2] == 4:  # Check number of channels on image
-                    alpha_overlay = overlay[:, :, 3] / 255.0
-                    for c in range(3):
-                        frame[y : y + image_height, x : x + image_width, c] = (
-                            alpha_overlay * overlay[:, :, c]
-                            + (1 - alpha_overlay) * frame[y : y + image_height, x : x + image_width, c]
-                        )
-                else:
-                    frame[y : y + image_height, x : x + image_width] = overlay
-
-                # Write the modified frame to the output video
-                output_video.write(frame)
+            try:
+                cap = cv2.VideoCapture(str(input_path))
 
+                # Get video properties
+                fps = int(cap.get(cv2.CAP_PROP_FPS))
+                width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+                height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+                video_char_code = cv2.VideoWriter.fourcc(*video_encoding_map[file_extension])
+                output_video = cv2.VideoWriter(str(output_path), video_char_code, fps, (width, height))
+
+                # Load and resize the overlay image
+
+                image_np_arr = np.frombuffer(image_bytes, np.uint8)
+                decoded = cv2.imdecode(image_np_arr, cv2.IMREAD_UNCHANGED)
+                if decoded is None:
+                    raise ValueError("Failed to decode overlay image")
+                overlay = cv2.resize(decoded, self._img_resize_size)
+
+                # Get overlay image dimensions
+                image_height, image_width, _ = overlay.shape
+                x, y = self._img_position  # Position where the overlay will be placed
+
+                while cap.isOpened():
+                    ret, frame = cap.read()
+                    if not ret:
+                        break
+
+                    # Ensure overlay fits within the frame boundaries
+                    if x + image_width > width or y + image_height > height:
+                        logger.info("Overlay image is too large for the video frame. Resizing to fit.")
+                        overlay = cv2.resize(overlay, (width - x, height - y))
+                        image_height, image_width, _ = overlay.shape
+
+                    # Blend overlay with frame
+                    if overlay.shape[2] == 4:  # Check number of channels on image
+                        alpha_overlay = overlay[:, :, 3] / 255.0
+                        for c in range(3):
+                            frame[y : y + image_height, x : x + image_width, c] = (
+                                alpha_overlay * overlay[:, :, c]
+                                + (1 - alpha_overlay) * frame[y : y + image_height, x : x + image_width, c]
+                            )
+                    else:
+                        frame[y : y + image_height, x : x + image_width] = overlay
+
+                    # Write the modified frame to the output video
+                    output_video.write(frame)
+            finally:
+                if cap is not None:
+                    cap.release()
+                if output_video is not None:
+                    output_video.release()
+                with contextlib.suppress(cv2.error):
+                    cv2.destroyAllWindows()  # Not available in headless OpenCV builds
+
+            return output_path.read_bytes()
         finally:
-            # Release everything (guarded — early raises may leave cap/output_video unbound)
-            if cap is not None:
-                cap.release()
-            if output_video is not None:
-                output_video.release()
-            with contextlib.suppress(cv2.error):
-                cv2.destroyAllWindows()  # Not available in headless OpenCV builds
-            if azure_storage_flag and local_temp_path is not None:
-                local_temp_path.unlink()
+            input_path.unlink(missing_ok=True)
+            output_path.unlink(missing_ok=True)
 
     async def convert_async(self, *, prompt: str, input_type: PromptDataType = "image_path") -> ConverterResult:
         """
@@ -239,15 +220,13 @@ async def convert_async(self, *, prompt: str, input_type: PromptDataType = "imag
         if not self.input_supported(input_type):
             raise ValueError("Input type not supported")
 
-        output_video_serializer = data_serializer_factory(category="prompt-memory-entries", data_type="video_path")
-
-        if not self._output_path:
-            output_video_serializer.value = str(await output_video_serializer.get_data_filename_async())
-        else:
-            output_video_serializer.value = self._output_path
-
-        # Add video to the image
-        updated_video = await self._add_image_to_video_async(
-            image_path=prompt, output_path=output_video_serializer.value
+        output_video_serializer = data_serializer_factory(
+            category="prompt-memory-entries",
+            data_type="video_path",
+            extension=self._video_path.suffix.removeprefix("."),
         )
-        return ConverterResult(output_text=str(updated_video), output_type="video_path")
+        updated_video = await self._add_image_to_video_async(image_path=prompt)
+        await output_video_serializer.save_data_async(data=updated_video)
+        logger.info(f"Video saved as {output_video_serializer.value}")
+
+        return ConverterResult(output_text=str(output_video_serializer.value), output_type="video_path")

pyrit/models/parameter.py

@@ -9,14 +9,15 @@
 import types
 from dataclasses import dataclass
 from enum import Enum
+from pathlib import Path
 from typing import Any, Literal, Union, get_args, get_origin
 
 from pydantic import BaseModel, ConfigDict, Field, computed_field, field_serializer, model_validator
 
 from pyrit.common.apply_defaults import REQUIRED_VALUE
 
-_SUPPORTED_SCALAR_TYPES: tuple[type, ...] = (str, int, float, bool)
-_SCALAR_NAME_TO_TYPE: dict[str, type] = {"int": int, "float": float, "bool": bool, "str": str}
+_SUPPORTED_SCALAR_TYPES: tuple[type, ...] = (str, int, float, bool, Path)
+_SCALAR_NAME_TO_TYPE: dict[str, type] = {"int": int, "float": float, "bool": bool, "str": str, "Path": Path}
 
 
 class ComponentType(str, Enum):
@@ -191,7 +192,7 @@ def is_string_coercible(self) -> bool:
         Whether a single string token can be coerced to this parameter's value.
 
         True for a non-reference plain scalar (``str`` / ``int`` / ``float`` /
-        ``bool``), ``Literal[...]``, or ``Enum`` parameter — exactly the forms a
+        ``bool`` / ``Path``), ``Literal[...]``, or ``Enum`` parameter — exactly the forms a
         text field or CLI token can supply. References and structured types (lists
         and arbitrary objects) are False and are surfaced/handled elsewhere.
 
@@ -261,7 +262,7 @@ def validate(self) -> None:  # type: ignore[ty:invalid-method-override]
         Reject a declaration with an unsupported ``param_type``.
 
         Supported forms are a plain scalar, a constrained scalar
-        (``Literal``/``Enum``), a ``list`` of any of those, a registry reference,
+        (``Literal``/``Enum``), ``Path``, a ``list`` of any of those, a registry reference,
         an opaque passthrough, or ``None``. An otherwise-unsupported type is
         tolerated only when the parameter declares a default (the builder simply
         does not supply it, and the value passes through unchanged).
@@ -284,7 +285,7 @@ def validate(self) -> None:  # type: ignore[ty:invalid-method-override]
 
         raise ValueError(
             f"Parameter '{self.name}' has unsupported param_type {param_type!r}. "
-            f"Supported types: str, int, float, bool, Literal[...], Enum, a list of those, "
+            f"Supported types: str, int, float, bool, Path, Literal[...], Enum, a list of those, "
             f"or None (or provide a default)."
         )
 
@@ -314,7 +315,7 @@ def _is_scalar_param_type(annotation: Any) -> bool:
     """
     Return True when ``annotation`` is a coercible scalar form.
 
-    A scalar form is a plain scalar (``str`` / ``int`` / ``float`` / ``bool``) or a
+    A scalar form is a plain scalar (``str`` / ``int`` / ``float`` / ``bool`` / ``Path``) or a
     constrained scalar (``Literal[...]`` or an ``Enum`` subclass) that carries its
     own allowed set.
 
@@ -333,12 +334,12 @@ def _coerce_simple_value(*, param_name: str, annotation: Any, raw_value: Any) ->
     Coerce ``raw_value`` to a scalar ``annotation`` — the shared coercion core.
 
     Handles ``Optional[X]`` unwrap, ``Literal``/``Enum`` membership, and
-    int/float/bool/str. Anything else passes through unchanged. Both the
+    int/float/bool/str/Path. Anything else passes through unchanged. Both the
     ``Parameter`` path (``coerce_value``) and the resolver's annotation path route
     through this function so they cannot diverge on coerced values.
 
     Returns:
-        Any: The coerced value (a ``Literal``/``Enum`` member, an int/float/bool/str, or
+        Any: The coerced value (a ``Literal``/``Enum`` member, an int/float/bool/str/Path, or
             the raw value unchanged for unsupported annotations).
 
     Raises:
@@ -358,6 +359,8 @@ def _coerce_simple_value(*, param_name: str, annotation: Any, raw_value: Any) ->
         return _coerce_scalar(param_name=param_name, scalar_type=float, raw_value=raw_value)
     if annotation is str:
         return str(raw_value)
+    if annotation is Path:
+        return Path(raw_value)
     return raw_value
 
 
@@ -466,7 +469,7 @@ def _coerce_list(*, param_name: str, param_type: Any, raw_value: Any) -> list[An
         ]
     raise ValueError(
         f"Parameter '{param_name}' has unsupported list element type {element_type!r}. "
-        f"Supported list element types: str, int, float, bool, or Literal[...]."
+        f"Supported list element types: str, int, float, bool, Path, or Literal[...]."
     )
 
 

tests/unit/backend/test_converter_service.py

@@ -161,6 +161,17 @@ async def test_catalog_serializes_parameter_type(self) -> None:
         caesar_param = next(p for p in caesar_entry.parameters if p.name == "caesar_offset")
         assert caesar_param.type_name == "int"
 
+    async def test_catalog_exposes_video_input_as_path_without_output_path(self) -> None:
+        """The video converter accepts an uploaded input but no caller-controlled destination."""
+        service = ConverterService()
+
+        result = await service.list_converter_catalog_async()
+
+        video_entry = next(item for item in result.items if item.converter_type == "AddImageVideoConverter")
+        video_path_param = next(parameter for parameter in video_entry.parameters if parameter.name == "video_path")
+        assert video_path_param.type_name == "Path"
+        assert all(parameter.name != "output_path" for parameter in video_entry.parameters)
+
     async def test_catalog_excludes_non_coercible_params(self) -> None:
         """Catalog only surfaces params that can be set from a string (e.g. not the LLM target)."""
         service = ConverterService()