Implement the gst_transform_resize scenario (New)

contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_resources_generator.py

@@ -22,7 +22,6 @@
 import json
 import logging
 from itertools import product
-from typing import Dict, List
 from checkbox_support.scripts.image_checker import has_desktop_environment
 from checkbox_support.snap_utils.system import on_ubuntucore
 
@@ -94,7 +93,7 @@ def _v4l2_video_decoder_md5_checksum_comparison_helper(
         height: str,
         color_space: str,
         source_format: str,
-    ) -> Dict:
+    ) -> dict:
         """
         Generate a resource item dictionary for
         gst_v4l2_video_decoder_md5_checksum_comparison scenario
@@ -128,7 +127,7 @@ def _v4l2_video_decoder_md5_checksum_comparison_helper(
         return returned_dict
 
     def gst_v4l2_video_decoder_md5_checksum_comparison(
-        self, scenario_data: List[Dict]
+        self, scenario_data: list[dict]
     ) -> None:
         for item in scenario_data:
             self._resource_items.extend(
@@ -145,7 +144,7 @@ def gst_v4l2_video_decoder_md5_checksum_comparison(
                 ]
             )
 
-    def gst_encoder_psnr(self, scenario_data: List[Dict]) -> None:
+    def gst_encoder_psnr(self, scenario_data: list[dict]) -> None:
         # Iterate through each encoder plugin configuration
         for item in scenario_data:
             encoder_plugin = item.get("encoder_plugin")
@@ -172,7 +171,7 @@ def gst_encoder_psnr(self, scenario_data: List[Dict]) -> None:
                 self._resource_items.append(config)
 
     def gst_v4l2_audio_video_synchronization(
-        self, scenario_data: Dict
+        self, scenario_data: dict
     ) -> None:
         video_sink = ""
         if on_ubuntucore():
@@ -202,7 +201,7 @@ def gst_v4l2_audio_video_synchronization(
                 )
 
     def gst_v4l2_video_decoder_performance_fakesink(
-        self, scenario_data: List[Dict]
+        self, scenario_data: list[dict]
     ) -> None:
         for item in scenario_data:
             self._resource_items.append(
@@ -247,6 +246,28 @@ def gst_transform_rotate_and_flip(self, scenario_data: list[dict]) -> None:
                 }
                 self._resource_items.append(config)
 
+    def gst_transform_resize(self, scenario_data: list[dict]) -> None:
+        # Iterate through each encoder plugin configuration
+        for item in scenario_data:
+            encoder_plugin = item.get("encoder_plugin")
+            if not encoder_plugin:
+                continue
+            resolutions = item.get("resolutions", [{}])
+
+            # Use itertools.product to create combinations of all parameters
+            for resolution in resolutions:
+                config = {
+                    "platform": self._conf_name,
+                    "scenario": self._current_scenario_name,
+                    "encoder_plugin": encoder_plugin,
+                    "width_from": resolution.get("width_from"),
+                    "height_from": resolution.get("height_from"),
+                    "width_to": resolution.get("width_to"),
+                    "height_to": resolution.get("height_to"),
+                    "framerate": resolution.get("fps"),
+                }
+                self._resource_items.append(config)
+
     def main(self):
         for scenario in self._scenarios:
             self._current_scenario_name = scenario

contrib/checkbox-ce-oem/checkbox-provider-ce-oem/bin/gst_transform_resize.py

@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+# This file is part of Checkbox.
+#
+# Copyright 2024 Canonical Ltd.
+# Written by:
+#   Patrick Chang <patrick.chang@canonical.com>
+#
+# Checkbox is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 3,
+# as published by the Free Software Foundation.
+#
+# Checkbox is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.
+
+import argparse
+import logging
+import os
+
+from typing import Any
+
+from gst_utils import (
+    GST_LAUNCH_BIN,
+    VIDEO_CODEC_TESTING_DATA,
+    SAMPLE_2_FOLDER,
+    PipelineInterface,
+    GStreamerEncodePlugins,
+    MetadataValidator,
+    get_big_bug_bunny_golden_sample,
+    generate_artifact_name,
+    compare_psnr,
+    delete_file,
+    execute_command,
+)
+
+logging.basicConfig(level=logging.INFO)
+
+
+def register_arguments():
+    parser = argparse.ArgumentParser(
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        description=(
+            "Script helps verify the gst_encoder_psnr scenario of specific"
+            " encoder."
+        ),
+    )
+
+    parser.add_argument(
+        "-p",
+        "--platform",
+        required=True,
+        type=str,
+        help="Json file name which is also the platform e.g. genio-1200",
+    )
+
+    parser.add_argument(
+        "-ep",
+        "--encoder_plugin",
+        required=True,
+        type=str,
+        help="Encoder plugin be used in gstreamer pipeline e.g. v4l2h264enc",
+    )
+
+    parser.add_argument(
+        "-wf",
+        "--width_from",
+        type=str,
+        required=True,
+        help="Value of width of the origianl resolution",
+    )
+
+    parser.add_argument(
+        "-hf",
+        "--height_from",
+        type=str,
+        required=True,
+        help="Value of height of the origianl resolution",
+    )
+
+    parser.add_argument(
+        "-wt",
+        "--width_to",
+        type=str,
+        required=True,
+        help="Value of width of the target resolution",
+    )
+
+    parser.add_argument(
+        "-ht",
+        "--height_to",
+        type=str,
+        required=True,
+        help="Value of height of the target resolution",
+    )
+
+    parser.add_argument(
+        "-f",
+        "--framerate",
+        type=str,
+        default="",
+        help="Value of framerate. e.g. 60, 30",
+    )
+
+    args = parser.parse_args()
+    return args
+
+
+def project_factory(args: argparse.Namespace) -> Any:
+    """
+    Factory function to create a project instance based on the platform
+    specified in the argparse arguments.
+    Args:
+        args (argparse.Namespace): A parsed argument object that contains the
+            project parameters.
+    Returns:
+        Any: An instance of the project class (e.g., `GenioProject`) created
+            with the specified parameters.
+    Raises:
+        SystemExit: If the platform is not recognized or supported.
+    """
+    if "genio" in args.platform:
+        return GenioProject(
+            platform=args.platform,
+            codec=args.encoder_plugin,
+            width_from=args.width_from,
+            height_from=args.height_from,
+            width_to=args.width_to,
+            height_to=args.height_to,
+            framerate=args.framerate,
+        )
+    else:
+        raise SystemExit(
+            "Error: Cannot get the implementation for '{}'".format(
+                args.platform
+            )
+        )
+
+
+class GenioProject(PipelineInterface):
+    """
+    Genio project manages platforms and codecs, and handles
+    building.
+    Spec: https://download.mediatek.com/aiot/download/release-note/v24.0/v24.0_IoT_Yocto_Feature_Table_v1.0.pdf     # noqa: E501
+    """
+
+    def __init__(
+        self,
+        platform: str,
+        codec: str,
+        width_from: int,
+        height_from: int,
+        width_to: int,
+        height_to: int,
+        framerate: int,
+    ):
+        self._platform = platform
+        self._codec = codec
+        self._width_from = width_from
+        self._height_from = height_from
+        self._width_to = width_to
+        self._height_to = height_to
+        self._framerate = framerate
+        self._codec_parser_map = {
+            GStreamerEncodePlugins.V4L2H264ENC.value: "h264parse"
+        }
+        # This sample video file will be consumed by any gstreamer piple as
+        # input video.
+        self._golden_sample = get_big_bug_bunny_golden_sample(
+            width=self._width_from,
+            height=self._height_from,
+            framerate=self._framerate,
+        )
+        self._artifact_file = ""
+
+    @property
+    def artifact_file(self) -> str:
+        if not self._artifact_file:
+            self._artifact_file = generate_artifact_name(extension="mp4")
+        return self._artifact_file
+
+    @property
+    def psnr_reference_file(self) -> str:
+        """
+        A golden reference which has been transformed in advance. It's used to
+        be the compared reference file for PSNR.
+        """
+        golden_reference = get_big_bug_bunny_golden_sample(
+            self._width_to, self._height_to, self._framerate
+        )
+
+        full_path = os.path.join(
+            VIDEO_CODEC_TESTING_DATA, SAMPLE_2_FOLDER, golden_reference
+        )
+        if not os.path.exists(full_path):
+            raise SystemExit(
+                "Error: Golden PSNR reference '{}' doesn't exist".format(
+                    full_path
+                )
+            )
+
+        return full_path
+
+    def build_pipeline(self) -> str:
+        """
+        Build the GStreamer commands based on the platform and codec.
+        Returns:
+            str: A GStreamer command based on the platform and
+            codec.
+        """
+        pipeline = (
+            "{} filesrc location={} ! decodebin ! v4l2convert ! "
+            "video/x-raw,width={},height={} ! {} ! {} ! mp4mux ! filesink"
+            " location={}"
+        ).format(
+            GST_LAUNCH_BIN,
+            self._golden_sample,
+            self._width_to,
+            self._height_to,
+            self._codec,
+            self._codec_parser_map.get(self._codec),
+            self.artifact_file,
+        )
+        return pipeline
+
+
+def main() -> None:
+    args = register_arguments()
+    p = project_factory(args)
+    logging.info("Step 1: Generating artifact...")
+    cmd = p.build_pipeline()
+    # execute command
+    execute_command(cmd=cmd)
+    logging.info("\nStep 2: Checking metadata...")
+    mv = MetadataValidator(file_path=p.artifact_file)
+    mv.validate("width", args.width_to).validate(
+        "height", args.height_to
+    ).validate("frame_rate", args.framerate).validate(
+        "codec", args.encoder_plugin
+    ).is_valid()
+    logging.info("\nStep 3: Comparing PSNR...")
+    compare_psnr(
+        golden_reference_file=p.psnr_reference_file,
+        artifact_file=p.artifact_file,
+    )
+    delete_file(file_path=p.artifact_file)
+
+
+if __name__ == "__main__":
+    main()