Python to Capture Video Frames with Source Code

by

In order to Capture Video Frames

Execute

python capture_video_frames.py <video_file>Code language: HTML, XML (xml)

In ordet to get live video stream from webcam

Execute

python capture_video_from_webcam.pyCode language: CSS (css)

In order to stream video from webcam to AWS Kinesis follow the below instructions

Amazon Kinesis Video Streams CPP Producer, GStreamer Plugin and JNI

Amazon Kinesis Video Streams | Secure Video Ingestion for Analysis & Storage

Key Features

  1. C++ SDK
  2. GStreamer Plugin (kvssink)
  3. JNI
  • Amazon Kinesis Video Streams Producer SDK for C/C++ makes it easy to build an on-device application that securely connects to a video stream, and reliably publishes video and other media data to Kinesis Video Streams.
  • It takes care of all the underlying tasks required to package the frames and fragments generated by the device’s media pipeline.
  • The SDK also handles stream creation, token rotation for secure and uninterrupted streaming, processing acknowledgements returned by Kinesis Video Streams, and other tasks.

Build:

Download

To download run the following command:

git clone https://github.com/awslabs/amazon-kinesis-video-streams-producer-sdk-cpp.gitCode language: PHP (php)

Note: You will also need to install pkg-config, CMake, m4 and a build environment. If you are building the GStreamer plugin you will also need GStreamer and GStreamer (Development Libraries).

Configure

Prepare a build directory in the newly checked out repository:

mkdir -p amazon-kinesis-video-streams-producer-sdk-cpp/build
cd amazon-kinesis-video-streams-producer-sdk-cpp/build
  • If you are building on Windows you need to generate NMake Makefiles, you should run cmake .. -G "NMake Makefiles"
  • GStreamer and JNI is NOT built by default, if you wish to build both you MUST execute cmake .. -DBUILD_GSTREAMER_PLUGIN=ON -DBUILD_JNI=TRUE
  • By default we download all the libraries from GitHub and build them locally, so should require nothing to be installed ahead of time.
  • If you do wish to link to existing libraries you can do cmake .. -DBUILD_DEPENDENCIES=OFF Libraries needed to build producer are: Curl, Openssl and Log4cplus. If you want to build the gstreamer plugin you will need to have gstreamer in your system. On Mac OS you can get the libraries using homebrew
$ brew install pkg-config openssl cmake gstreamer gst-plugins-base gst-plugins-good gst-plugins-bad gst-plugins-ugly log4cplus gst-libav

On Ubuntu and Raspberry Pi OS you can get the libraries by running

$ sudo apt-get install libssl-dev libcurl4-openssl-dev liblog4cplus-dev libgstreamer1.0-dev libgstreamer-plugins-base1.0-dev gstreamer1.0-plugins-base-apps gstreamer1.0-plugins-bad gstreamer1.0-plugins-good gstreamer1.0-plugins-ugly gstreamer1.0-tools
Code language: JavaScript (javascript)

Cross-Compilation

If you wish to cross-compile CC and CXX are respected when building the library and all its dependencies.

CMake Arguments

You can pass the following options to cmake ...

  • -DBUILD_GSTREAMER_PLUGIN — Build kvssink GStreamer plugin
  • -DBUILD_JNI — Build C++ wrapper for JNI to expose the functionality to Java/Android
  • -DBUILD_DEPENDENCIES — Build depending libraries from source
  • -DBUILD_TEST=TRUE — Build unit/integration tests, may be useful for confirm support for your device. ./tst/producerTest
  • -DCODE_COVERAGE — Enable coverage reporting
  • -DCOMPILER_WARNINGS — Enable all compiler warnings
  • -DADDRESS_SANITIZER — Build with AddressSanitizer
  • -DMEMORY_SANITIZER — Build with MemorySanitizer
  • -DTHREAD_SANITIZER — Build with ThreadSanitizer
  • -DUNDEFINED_BEHAVIOR_SANITIZER Build with UndefinedBehaviorSanitizer
  • -DALIGNED_MEMORY_MODEL Build for aligned memory model only devices. Default is OFF.

To Include JNI

JNI examples are NOT built by default. If you wish to build JNI you MUST add -DBUILD_JNI=TRUE when running cmake:

cmake -DBUILD_JNI=TRUE
Code language: PHP (php)

To Include Building GStreamer Sample Programs

The GStreamer plugin and samples are NOT built by default. If you wish to build them you MUST add -DBUILD_GSTREAMER_PLUGIN=TRUE when running cmake:

cmake -DBUILD_GSTREAMER_PLUGIN=TRUE
Code language: PHP (php)

Compiling

After running cmake, in the same build directior run make:

make

On Windows you should run nmake instead of make

In your build directory you will now have shared objects for all the targets you have selected

Run

GStreamer Plugin (kvssink)

Loading Element

The GStreamer plugin is located in your build directory.

To load this plugin set the following environment variables. This should be run from the root of the repo, NOT the build directory.

export GST_PLUGIN_PATH=`pwd`/build
export LD_LIBRARY_PATH=`pwd`/open-source/local/lib
Code language: JavaScript (javascript)

The equivalent for Windows is

set GST_PLUGIN_PATH=%CD%\build
set PATH=%PATH%;%CD%\open-source\local\bin;%CD%\open-source\local\lib
Code language: JavaScript (javascript)

Now if you execute gst-inspect-1.0 kvssink you should get information on the plugin like

Factory Details:
  Rank                     primary + 10 (266)
  Long-name                KVS Sink
  Klass                    Sink/Video/Network
  Description              GStreamer AWS KVS plugin
  Author                   AWS KVS <kinesis-video-support@amazon.com>

Plugin Details:
  Name                     kvssink
  Description              GStreamer AWS KVS plugin
  Filename                 /Users/seaduboi/workspaces/amazon-kinesis-video-streams-producer-sdk-cpp/build/libgstkvssink.so
  Version                  1.0
  License                  Proprietary
  Source module            kvssinkpackage
  Binary package           GStreamer
  Origin URL               http://gstreamer.net
Code language: JavaScript (javascript)

If the build failed, or GST_PLUGIN_PATH is not properly set you will get output like

No such element or plugin 'kvssink'
Code language: JavaScript (javascript)

Using Element

The kvssink element has the following required parameters:

  • stream-name — The name of the destination Kinesis video stream.
  • storage-size — The storage size of the device in megabytes. For information about configuring device storage, see StorageInfo.
  • access-key — The AWS access key that is used to access Kinesis Video Streams. You must provide either this parameter or credential-path.
  • secret-key — The AWS secret key that is used to access Kinesis Video Streams. You must provide either this parameter or credential-path.
  • credential-path — A path to a file containing your credentials for accessing Kinesis Video Streams. For example credential files, see Sample Static Credential and Sample Rotating Credential. For more information on rotating credentials, see Managing Access Keys for IAM Users. You must provide either this parameter or access-key and secret-key.

For examples of common use cases you can look at Example: Kinesis Video Streams Producer SDK GStreamer Plugin

Dockerscripts

DEBUG

  • If you are successfully streaming but run into issue with playback. You can do export KVS_DEBUG_DUMP_DATA_FILE_DIR=/path/to/directory before streaming. Producer will then dump MKV files into that path. The file is exactly what KVS will receive. You can use MKVToolNIX to check that everything looks correct. You can also try to play the MKV file in compatible players.

Source Code:

capture_video_frames.py

import os
import shutil
import sys
import cv2

class FrameCapture:
    '''
        Class definition to capture frames
    '''
    def __init__(self, file_path):
        '''
            initializing directory where the captured frames will be stored.
            Also truncating the directory where captured frames are stored, if exists.
        '''
        self.directory = "captured_frames"
        self.file_path = file_path
        if os.path.exists(self.directory):
            shutil.rmtree(self.directory)
        os.mkdir(self.directory)

    def capture_frames(self):
        '''
            This method captures the frames from the video file provided.
            This program makes use of openCV library
        '''
        cv2_object = cv2.VideoCapture(self.file_path)

        frame_number = 0
        frame_found = 1

        while frame_found:
            frame_found, image = cv2_object.read()
            capture = f'{self.directory}/frame{frame_number}.jpg'
            cv2.imwrite(capture, image)

            frame_number += 1

if __name__ == '__main__':
    file_path = sys.argv[1]
    fc = FrameCapture(file_path)
    fc.capture_frames()

Enhanced_Capture_Video_Frames_Test

import unittest


# enhancement 1
def get_file_and_destination():
    input_file = input("What is the file name?")
    mp4_converter = ".mp4"
    input_file = input_file + mp4_converter

    destination_file = input("Where you want to save it to?")
    return input_file, destination_file


# enhancement 2
def get_user_input():
    user_preference = input("Enter your frame preference in total: ")
    if not user_preference.isnumeric():
        print("Please enter an integer!")
        new_value = get_user_input()
        user_preference = new_value
    return user_preference


# test variables for enhancement 1
first, second = get_file_and_destination()
# test variable for enhancement 2
input_number = get_user_input()


class BasicTests(unittest.TestCase):

    def test_get_string(self):
        file_name = "video.mp4"
        self.assertEqual(first, file_name)

    def test_get_destination(self):
        destination = "testing destination"
        self.assertEqual(second, destination)

    def test_input_number(self):
        output_expected = input_number.isnumeric()
        self.assertTrue(output_expected)


if __name__ == "__main__":
    unittest.main()

Leave a Comment