Add script for media validation and debugging utilities

Author: sirknightj

.github/workflows/ci.yml

@@ -10,6 +10,10 @@ on:
       - develop
       - master
 
+concurrency:
+  group: ${{ github.workflow }}
+  cancel-in-progress: true
+
 jobs:
   load-matrix:
     runs-on: ubuntu-latest
@@ -170,17 +174,17 @@ jobs:
           cache: maven
 
       - name: Set up Python 3.12
-        if: ${{ runner.os == 'macOS' || runner.os == 'linux' }}
         uses: actions/setup-python@v5
         with:
           python-version: "3.12"
 
       - name: Install Python dependencies
-        if: ${{ runner.os == 'macOS' || runner.os == 'linux' }}
-        working-directory: scripts/python
+        working-directory: scripts/python/benchmarking
         run: |
           python -m pip install --upgrade pip
           pip install -r requirements.txt
+          cd ../getmediavalidation
+          pip install .
 
       - name: Download JNI Library
         uses: actions/download-artifact@v4
@@ -239,7 +243,7 @@ jobs:
             com.amazonaws.kinesisvideo.demoapp.${{ matrix.sample }} &
           JAVA_PID=$!
 
-          python ./scripts/python/capture_rss_and_cpu.py $JAVA_PID &
+          python ./scripts/python/benchmarking/capture_rss_and_cpu.py $JAVA_PID &
           MONITOR_PID=$!
 
           # Wait for maximum 30 seconds
@@ -307,12 +311,36 @@ jobs:
         run: |
           DATA_FILE=$(find . -name 'process_*.txt' | head -n 1)
           COMMIT_HASH=$(git rev-parse --short HEAD)
-          python ./scripts/python/plot_rss_and_cpu.py "$DATA_FILE" \
+          python ./scripts/python/benchmarking/plot_rss_and_cpu.py "$DATA_FILE" \
             --title "Memory and CPU Usage (${{ matrix.sample }} @ ${COMMIT_HASH})\n${{ matrix.platform.os }}, Java ${{ matrix.java }}" \
             --output "${{ env.STREAM_NAME }}-mem-cpu.png"
 
         shell: bash
 
+      - name: Check uploaded media (Mac and Linux) (DemoAppMain)
+        if: ${{ (runner.os == 'macOS' || runner.os == 'linux') && matrix.sample == 'DemoAppMain' }}
+        working-directory: scripts/python/getmediavalidation/bin
+        run: |
+          python ./fetch_fragment_info.py --stream-name "$STREAM_NAME" --last 5m
+          python ./validate_media.py --stream-name "$STREAM_NAME" \
+            --keyframe-interval 25 \
+            -fps 25 \
+            --frames-path "${{ github.workspace }}/src/main/resources/data/h264/*.h264" \
+            --last 5m
+
+      - name: Check uploaded media (Windows) (DemoAppMain)
+        if: ${{ runner.os == 'Windows' && matrix.sample == 'DemoAppMain' }}
+        working-directory: scripts/python/getmediavalidation/bin
+        run: |
+          python ./fetch_fragment_info.py --stream-name "%STREAM_NAME%" --last 5m
+          python ./validate_media.py --stream-name "%STREAM_NAME%" ^
+            --keyframe-interval 25 ^
+            -fps 25 ^
+            --frames-path "${{ github.workspace }}/src/main/resources/data/h264/*.h264" ^
+            --last 5m
+
+        shell: cmd
+
       - name: Upload memory and CPU graph (Mac and Linux)
         if: ${{ runner.os == 'macOS' || runner.os == 'linux' }}
         uses: actions/upload-artifact@v4

scripts/python/capture_rss_and_cpu.py renamed to scripts/python/benchmarking/capture_rss_and_cpu.py

@@ -0,0 +1,133 @@
+# GetMediaValidation
+
+Tools for validating, viewing, and downloading Kinesis Video Stream media fragments.
+
+## Installation
+
+1. Clone the repository.
+
+2. Change directories to the project root:
+    ```shell
+    cd getmediavalidation
+    ```
+
+3. Create a virtual environment.
+    ```shell
+    python -m venv .venv
+    source .venv/bin/activate  # On Windows: venv\Scripts\activate
+    ```
+
+4. Install the project's dependencies. The `-e` flag is optional but recommended. Instead of creating a static copy of
+   this package in site-packages, it will create a symlink. This lets modifications made to the source be reflected immediately without
+   needing to reinstall this package.
+    ```shell
+    pip install -e .
+    ```
+
+5. Use the scripts.
+    ```shell
+    python3 ./bin/download_fragment.py --help
+    python3 ./bin/scripts/fetch_fragment_info.py --help
+    python3 ./bin/scripts/validate_media.py --help
+    ```
+
+## Project structure
+
+You can find the scripts in the `bin` folder.
+
+```shell
+getmediavalidation/
+├── bin/              # Scripts
+├── src/
+│   ├── api/          # API interaction functions
+│   ├── mkv/          # MKV processing functions
+│   └── utils/        # Utility functions
+├── requirements.txt  # Project dependencies
+├── setup.py          # Package configuration
+└── README.md         # This file
+```
+
+## About the scripts
+
+### _Fetch Fragment Info_ and _Download Fragment_
+
+These scripts are intended to be used in tandem. First, use the Fetch Fragment Info script, which will display a table
+of the fragments in a stream during the specified time range. One of the columns displayed is the fragment number, which
+is that fragment's ID.
+
+You can use the Download Fragment script with that specified ID to download just that fragment. You can then use
+open-source tools like `mkvtoolnix`, `ffmpeg`, or the KVS Parser Library to interact with the MKV.
+
+Example with `mkvtoolnix` (`mkvinfo`):
+
+```shell
+sudo apt-get update
+sudo apt-get install mkvtoolnix
+mkvinfo -v ./downloaded-fragment.mkv
+```
+
+### _Validate Media_
+
+This script validates frame data integrity between the original input frames and the uploaded MKV content.
+
+The validation process ensures that frame data buffer written via `putFrame()` maintains byte-for-byte equality when
+stored in the MKV container's `Simple Block`s.
+
+In the MKV, the frame data in organized in the following hierarchical structure:
+
+```shell
++ EBML head
+|...
++ Segment: size unknown
+| ...
+|+ Cluster
+| + Cluster timestamp: 00:00:00.000000000
+| + Cluster position: 0
+| + Simple block: key, track number 1, 1 frame(s), timestamp 00:00:00.000000000
+|  + Frame with size 23917  # The frame is written here
+```
+
+Fragmentation Rules
+
+- Media content is segmented into fragments
+- Each fragment begins with a keyframe (boundaries are determined by keyframe occurrence)
+
+If a continuous frame insertion loop is used with a fixed frameset, we expect the media uploaded to follow the pattern.
+
+Given:
+
+- A fixed keyframe interval $k$
+- A finite set of $f$ frames, indexed $1$ through $f$
+
+The first frame $f_i$ in the $i$th fragment in an uploading session $F_i$ can be calculated using:
+
+$$ f_i = \left( \left( \left( F_i - 1 \right) * k \right) \bmod f \right) + 1 $$
+
+Example:
+
+- Using a keyframe interval of 25
+- A set of 45 frames
+
+The expected first frame of the fragments are: 1, 26, 6, 31, ...
+
+| Fragment number | Calculation                  | Result   |
+|-----------------|------------------------------|----------|
+| 1               | ((1-1) × 25 mod 45) + 1 = 1  | Frame 1  |
+| 2               | ((2-1) × 25 mod 45) + 1 = 26 | Frame 26 |
+| 3               | ((3-1) × 25 mod 45) + 1 = 6  | Frame 6  |
+| 4               | ((4-1) × 25 mod 45) + 1 = 31 | Frame 31 |
+| 5               | ((5-1) × 25 mod 45) + 1 = 11 | Frame 11 |
+
+We can then simply compare the bytes of the original file for equality.
+
+If any fragments are uploaded out of order, or fragments are dropped, this script will catch it.
+
+## Uninstalling
+
+The name of the package is specified in the `setup.py`: `getmediautils`
+
+To uninstall globally:
+
+```shell
+pip uninstall getmediautils
+```

scripts/python/plot_rss_and_cpu.py renamed to scripts/python/benchmarking/plot_rss_and_cpu.py

@@ -0,0 +1,62 @@
+import argparse
+import logging
+import os
+
+from src.api import fetch_fragments
+
+logger = logging.getLogger(__name__)
+
+
+def setup_argparse():
+    parser = argparse.ArgumentParser(description='KVS Fragment Info Fetcher')
+
+    # Required arguments
+    parser.add_argument('--stream-name', type=str, required=True,
+                        help='Name of the stream (e.g., demo-stream)')
+
+    parser.add_argument('--fragment-number', type=str, required=True,
+                        help='Fragment number to download')
+
+    # Optional arguments
+    parser.add_argument('--log-level',
+                        choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
+                        default='INFO',
+                        help='Set the logging level (default: INFO)')
+
+    parser.add_argument('--output-directory', type=str, required=False,
+                        default=os.getcwd(),
+                        help='Directory to output the downloaded fragment. Default: This directory.')
+
+    return parser
+
+
+def main():
+    parser = setup_argparse()
+    args = parser.parse_args()
+
+    logging.basicConfig(
+        level=getattr(logging, args.log_level.upper()),
+        format='%(asctime)s [%(filename)s:%(lineno)s/%(funcName)-s()] [%(levelname)s] %(message)s',
+        handlers=[
+            logging.StreamHandler()  # Outputs logs to the console
+        ]
+    )
+
+    fragment_number: str = args.fragment_number
+
+    # Result should only be one fragment
+    fetched_fragments_map: dict[str, bytes] = fetch_fragments(stream_name=args.stream_name,
+                                                              fragment_numbers=[args.fragment_number])
+
+    fragment_bytes: bytes = fetched_fragments_map[fragment_number]
+
+    file_path = f'{args.output_directory}/{args.stream_name}-{fragment_number}.mkv'
+
+    with open(file=file_path, mode='wb') as file:
+        file.write(fragment_bytes)
+
+    logger.info(f'Saved {fragment_number} from {args.stream_name} at {file_path} ({len(fragment_bytes)} bytes)')
+
+
+if __name__ == '__main__':
+    main()

scripts/python/requirements.txt renamed to scripts/python/benchmarking/requirements.txt

@@ -0,0 +1,80 @@
+import argparse
+import logging
+from datetime import datetime, timedelta
+from typing import List
+
+from dateutil.tz import tzlocal
+from mypy_boto3_kinesis_video_archived_media.type_defs import FragmentTypeDef
+
+from src.api import api_calls
+from src.mkv import print_fragments
+from src.utils import parse_datetime, parse_duration
+
+logger = logging.getLogger(__name__)
+
+
+def setup_argparse():
+    parser = argparse.ArgumentParser(description='KVS Fragment Info Display')
+
+    # Required arguments
+    parser.add_argument('--stream-name', type=str, required=True,
+                        help='Name of the stream (e.g., demo-stream)')
+
+    # Optional arguments
+    parser.add_argument('--log-level',
+                        choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
+                        default='INFO',
+                        help='Set the logging level (default: INFO)')
+
+    # Mutually exclusive group for either timestamp range, or last d duration
+    group = parser.add_argument_group('clip selection')
+    selection = group.add_mutually_exclusive_group(required=False)
+
+    selection.add_argument('--time-range', nargs=2, type=parse_datetime,
+                           metavar=('START_TIME', 'END_TIME'),
+                           help='Start and end timestamps (ISO format: 2025-05-27T21:30:00-07:00)')
+    selection.add_argument('--last', type=parse_duration,
+                           help='Duration to look back from now (e.g., "1h" for 1 hour, "30m" for 30 minutes), '
+                                'default 1 hour')
+
+    return parser
+
+
+def main():
+    parser = setup_argparse()
+    args = parser.parse_args()
+
+    logging.basicConfig(
+        level=getattr(logging, args.log_level.upper()),
+        format='%(asctime)s [%(filename)s:%(lineno)s/%(funcName)-s()] [%(levelname)s] %(message)s',
+        handlers=[
+            logging.StreamHandler()  # Outputs logs to the console
+        ]
+    )
+
+    if args.time_range is not None:
+        # Use explicit time range
+        start_time = args.time_range[0]
+        end_time = args.time_range[1]
+    else:
+        # Default duration: 1 hour
+        if args.last is None:
+            args.last = timedelta(hours=1)
+
+        # Calculate time range based on duration
+        end_time = datetime.now(tz=tzlocal())
+        start_time = end_time - args.last
+
+    logger.info(f"Checking time range: {start_time} to {end_time}")
+
+    fragment_list: List[FragmentTypeDef] = api_calls.list_fragments(
+        args.stream_name,
+        start_timestamp=start_time,
+        end_timestamp=end_time
+    )
+
+    print_fragments(fragment_list)
+
+
+if __name__ == '__main__':
+    main()