Add: gstreamer st30p Tx/Rx tests, media create for audio files

tests/validation/tests/Engine/GstreamerApp.py

@@ -30,7 +30,7 @@ def create_connection_params(
         })
     return params
 
-def setup_gstreamer_tx_pipeline(
+def setup_gstreamer_st20p_tx_pipeline(
     build: str,
     nic_port_list: str,
     input_path: str,
@@ -72,7 +72,7 @@ def setup_gstreamer_tx_pipeline(
 
     return pipeline_command
 
-def setup_gstreamer_rx_pipeline(
+def setup_gstreamer_st20p_rx_pipeline(
     build: str,
     nic_port_list: str,
     output_path: str,
@@ -111,27 +111,105 @@ def setup_gstreamer_rx_pipeline(
 
     return pipeline_command
 
+def setup_gstreamer_st30_tx_pipeline(
+    build: str,
+    nic_port_list: str,
+    input_path: str,
+    tx_payload_type: int,
+    tx_queues: int,
+    audio_format: str,
+    channels: int,
+    sampling: int
+): 
+    connection_params = create_connection_params(dev_port=nic_port_list, payload_type=tx_payload_type, udp_port=30000, is_tx=True)
+
+    # st30 tx GStreamer command line
+    pipeline_command = [
+        "gst-launch-1.0", 
+        "filesrc", 
+        f"location={input_path}", 
+        "blocksize=70000",
+        "!", 
+        "rawaudioparse",
+        f"format=pcm",
+        f"sample-rate={sampling}",
+        f"pcm-format={audio_format}",
+        f"num-channels={channels}",
+        "!", 
+        "mtl_st30p_tx",
+        f"tx-queues={tx_queues}"
+    ]
+
+    for key, value in connection_params.items():
+        pipeline_command.append(f"{key}={value}")
+
+    pipeline_command.append(f"--gst-plugin-path={build}/ecosystem/gstreamer_plugin/builddir/")
+
+    return pipeline_command
+
+def setup_gstreamer_st30_rx_pipeline(
+    build: str,
+    nic_port_list: str,
+    output_path: str,
+    rx_payload_type: int,
+    rx_queues: int,
+    rx_audio_format: str,
+    rx_channels: int,
+    rx_sampling: int
+): 
+    connection_params = create_connection_params(dev_port=nic_port_list, payload_type=rx_payload_type, udp_port=30000, is_tx=False)
+
+    # st30 rx GStreamer command line
+    pipeline_command = [
+        "gst-launch-1.0", 
+        "-v", 
+        "mtl_st30p_rx"
+    ]
+
+    for key, value in connection_params.items():
+        pipeline_command.append(f"{key}={value}")
+
+    for x in [
+        f"rx-queues={rx_queues}",
+        f"rx-audio-format={rx_audio_format}",
+        f"rx-channel={rx_channels}",
+        f"rx-sampling={rx_sampling}",
+        "!", 
+        "filesink", 
+        f"location={output_path}"
+        ]:
+        pipeline_command.append(x)
+
+    pipeline_command.append(f"--gst-plugin-path={build}/ecosystem/gstreamer_plugin/builddir/")
+
+    return pipeline_command
+
 def execute_test(
         build: str,
         tx_command: list,
         rx_command: list,
-        fps: int,
         input_file: str,
         output_file: str,
+        type: str,
+        fps: int = None
         ):
 
     tx_process = call(' '.join(tx_command), cwd=build, timeout=120)
     rx_process = call(' '.join(rx_command), cwd=build, timeout=120)
 
     tx_output = wait(tx_process)
     rx_output = wait(rx_process)
-
-    tx_result = check_tx_output(fps=fps, output=tx_output.splitlines())
-    #rx_result = check_rx_output(fps=fps, output=rx_output.splitlines())
+    if type == "st20":
+        tx_result = check_tx_output(fps=fps, output=tx_output.splitlines())
+        #rx_result = check_rx_output(fps=fps, output=rx_output.splitlines())
+        if tx_result is False:
+            return False
 
     file_compare = compare_files(input_file, output_file)
 
-    if tx_result and file_compare:
+    log_info(f"File comparison: {file_compare}")
+
+    if file_compare:
         return True
 
     return False
@@ -171,6 +249,7 @@ def compare_files(input_file, output_file):
         log_info(f"Input file size: {input_file_size}")
         log_info(f"Output file size: {output_file_size}")
         if input_file_size != output_file_size:
+            log_fail("File size is different")
             return False
 
         with open(input_file, 'rb') as i_file, open(output_file, 'rb') as o_file:
@@ -184,8 +263,24 @@ def compare_files(input_file, output_file):
     log_fail("Comparison of files failed")
     return False
 
-def format_change(file_format):
+def video_format_change(file_format):
     if file_format in ["YUV422PLANAR10LE", "YUV_422_10bit"]:
         return "I422_10LE"
     else:
-        return file_format
+        return file_format
+
+def audio_format_change(file_format, rx_side: bool = False):
+    if rx_side == True:
+        if file_format == "s8":
+            return "PCM8"
+        elif file_format == "s16le":
+            return "PCM16"
+        else:
+            return "PCM24"
+    else:      
+        if file_format == "s8":
+            return 8
+        elif file_format == "s16le":
+            return 16
+        else:
+            return 24

tests/validation/tests/Engine/media_creator.py

@@ -1,8 +1,10 @@
 import os
 import subprocess
-import logging
 import time
 
+
+from tests.Engine.execute import log_fail, log_info
+
 def create_video_file(
     width: int,
     height: int,
@@ -30,24 +32,58 @@ def create_video_file(
         f"location={file_path}"
     ]
 
-    logging.info(f"Creating video file with command: {' '.join(command)}")
+    log_info(f"Creating video file with command: {' '.join(command)}")
 
     process = subprocess.Popen(command)
 
     try:
         time.sleep(duration)
         process.terminate()
         process.wait()
-        logging.info(f"Video file created at {file_path}")
+        log_info(f"Video file created at {file_path}")
     except subprocess.SubprocessError as e:
-        logging.error(f"Failed to create video file: {e}")
+        log_fail(f"Failed to create video file: {e}")
         raise
 
     return file_path
 
+def create_audio_file_sox(
+    sample_rate: int,
+    channels: int,
+    bit_depth: int,
+    frequency: int = 440,
+    output_path: str = "test.pcm",
+    duration: int = 10,
+):
+    """
+    Create an audio file with the provided arguments using sox.
+    """
+    command = [
+        "sox",
+        "-n",
+        "-r", str(sample_rate),
+        "-c", str(channels),
+        "-b", str(bit_depth),
+        "-t", "raw",
+        output_path,
+        "synth",
+        str(duration),
+        "sine",
+        str(frequency)
+    ]
+
+    log_info(f"Creating audio file with command: {' '.join(command)}")
+
+    try:
+        subprocess.run(command, check=True)
+        log_info(f"Audio file created at {output_path}")
+    except subprocess.CalledProcessError as e:
+        log_fail(f"Failed to create audio file: {e}")
+        raise
+
 def remove_file(file_path: str):
     if os.path.exists(file_path):
         os.remove(file_path)
-        logging.info(f"Removed file: {file_path}")
+        log_info(f"Removed file: {file_path}")
     else:
-        logging.warning(f"File not found: {file_path}")
+        log_info(f"File not found: {file_path}")

tests/validation/tests/single/gstreamer/audio_format/test_audio_format.py

@@ -0,0 +1,62 @@
+import os
+import pytest
+
+import tests.Engine.GstreamerApp as gstreamerapp
+import tests.Engine.media_creator as media_create
+
+@pytest.mark.parametrize("audio_format", ["s8", "s16le", "s24le"])
+@pytest.mark.parametrize("audio_channel", [1, 2])
+@pytest.mark.parametrize("audio_rate", [44100, 48000, 96000])
+def test_audio_format(
+    build,
+    media,
+    nic_port_list,
+    audio_format,
+    audio_channel,
+    audio_rate
+):
+    input_file_path = os.path.join(media, "test.pcm")
+
+    media_create.create_audio_file_sox(
+        sample_rate=audio_rate,
+        channels=audio_channel,
+        bit_depth=gstreamerapp.audio_format_change(audio_format),
+        duration=10,
+        frequency=440,
+        output_path=input_file_path
+    )
+
+    tx_config = gstreamerapp.setup_gstreamer_st30_tx_pipeline(
+        build=build,
+        nic_port_list=nic_port_list[0],
+        input_path=input_file_path,
+        tx_payload_type=111,
+        tx_queues=4,
+        audio_format=audio_format,
+        channels=audio_channel,
+        sampling=audio_rate
+    )
+
+    rx_config = gstreamerapp.setup_gstreamer_st30_rx_pipeline(
+        build=build,
+        nic_port_list=nic_port_list[1],
+        output_path=os.path.join(media, "output.pcm"),
+        rx_payload_type=111,
+        rx_queues=4,
+        rx_audio_format=gstreamerapp.audio_format_change(audio_format, rx_side=True),
+        rx_channels=audio_channel,
+        rx_sampling=audio_rate
+    )
+
+    try:
+        gstreamerapp.execute_test(
+            build=build,
+            tx_command=tx_config,
+            rx_command=rx_config,
+            input_file=input_file_path,
+            output_file=os.path.join(media, "output.pcm"),
+            type="st30"
+        )
+    finally:
+        media_create.remove_file(input_file_path)
+        media_create.remove_file(os.path.join(media, "output.pcm"))

tests/validation/tests/single/gstreamer/video_format/test_video_format.py

@@ -18,31 +18,31 @@ def test_video_format(
         width=video_file["width"],
         height=video_file["height"],
         framerate=video_file["fps"],
-        format=gstreamerapp.format_change(video_file["format"]),
+        format=gstreamerapp.video_format_change(video_file["format"]),
         media_path=media
     )
 
-    tx_config = gstreamerapp.setup_gstreamer_tx_pipeline(
+    tx_config = gstreamerapp.setup_gstreamer_st20p_tx_pipeline(
         build=build,
         nic_port_list=nic_port_list[0],
         input_path=input_file_path,
         width=video_file["width"],
         height=video_file["height"],
         framerate=video_file["fps"],
-        format=gstreamerapp.format_change(video_file["format"]),
+        format=gstreamerapp.video_format_change(video_file["format"]),
         tx_payload_type=112,
         tx_queues=4,
         tx_fps=video_file["fps"]
     )
 
-    rx_config = gstreamerapp.setup_gstreamer_rx_pipeline(
+    rx_config = gstreamerapp.setup_gstreamer_st20p_rx_pipeline(
         build=build,
         nic_port_list=nic_port_list[1],
         output_path=os.path.join(media, "output.yuv"),
         width=video_file["width"],
         height=video_file["height"],
         framerate=video_file["fps"],
-        format=gstreamerapp.format_change(video_file["format"]),
+        format=gstreamerapp.video_format_change(video_file["format"]),
         rx_payload_type=112,
         rx_queues=4,
         rx_fps=video_file["fps"]
@@ -55,7 +55,8 @@ def test_video_format(
             rx_command=rx_config,
             fps=video_file["fps"],
             input_file=input_file_path,
-            output_file=os.path.join(media, "output.yuv")
+            output_file=os.path.join(media, "output.yuv"),
+            type="st20"
         )
     finally:
         # Remove the video file after the test