GCP Videos to Images

Extract images from video in Python

https://www.geeksforgeeks.org/extract-images-from-video-in-python/

OpenCV - Getting Started with Videos

https://docs.opencv.org/3.4/dd/d43/tutorial_py_video_display.html

import cv2
import glob
import os
import string
import re
import numpy as np
from sewar.full_ref import mse, msssim
from multiprocessing import Pool

LOGS = False
SAMPLING_RATE = 0.08 # 0.08 seconds; Feature Crosses and TensorFlow Playground 
INITIAL_WINDOW = 0.6 # seconds

MSE_SAME = 0.01 # 0.01 Designing Adaptable ML Systems: Intro
MSE_NEW = 1.9  # 1.9 Designing Adaptable ML Systems: Intro
MSE_VALID = 12.35 # 12.34 (Optional) Lab demo_ AutoML tabular
MSSSIM_SAME = 0.9993 # Prediction using Vertex AI
WHITE_FRAME = 254.7125 # NLP architecture
BLACK_FRAME = 0.0012
IMG_DIM_PROC = (320, 180)
IMG_DIM_OUT = (1280, 720)

video_source = '/mnt/c/Users/Eduardo/Downloads'
videos_path = [f for f in glob.iglob(f'{video_source}/*') if f.endswith('.mp4')]

def save_image(video_name, frame, slide):
    name = f'data_tmp/{video_name}/{slide:03d}.jpg'
    new_frame = cv2.resize(frame, IMG_DIM_OUT, interpolation=cv2.INTER_AREA)
    cv2.imwrite(name, new_frame)

def is_static(val):
    return val < MSE_SAME

def printf(info):
    if LOGS:
        print(info)

def get_lesson_id(resource):
    item = re.sub('[%s ]+' % re.escape(string.punctuation), '_', resource).lower()

    keywords = {
        'introduction': 'intro',
        'computer_vision': 'cv',
        'convolutional_neural_networks': 'cnns',
        'convolutional_neural_network': 'cnn',
        'deep_neural_networks': 'dnns',
        'deep_neural_network': 'dnn',
        'neural_networks': 'nns',
        'neural_network': 'nn',
        'google_cloud': 'gcloud',
        'bigquery': 'bq',
        'tensorflow': 'tf',
        'machine_learning': 'ml',
        'natural_language_processing': 'nlp'    
    }

    for key, value in keywords.items():
        item = item.replace(key, value)

    item = item[:51]

    if item.endswith('_'):
        item = item[:-1]

    return item

def check_bw_frame(frame, num_slide, frame_idx):
    bw_check = np.mean(frame)
    valid_frame = True

    img_info = f'X Image{num_slide}-{frame_idx}'

    if(bw_check >= WHITE_FRAME):
        printf(f'{img_info} -> white by {bw_check}')
        valid_frame = False
    elif(bw_check <= BLACK_FRAME):
        printf(f'{img_info} -> black by {bw_check}')
        valid_frame = False

    return valid_frame, round(bw_check, 4)


def get_resized_grame(frame):
    return cv2.resize(frame, IMG_DIM_PROC, interpolation=cv2.INTER_AREA)

def process_video(video_path):
    video_name = video_path.split('/')[-1].split('.')[0]
    video_name = get_lesson_id(video_name)

    if not os.path.exists(f'data_tmp/{video_name}'):
        os.mkdir(f'data_tmp/{video_name}')
    
    cam = cv2.VideoCapture(video_path)

    frame_idx = 0
    num_slide = 1

    #### Check if video start static
    _, frame = cam.read()
    frame_idx += 1
    last_frame_resized = get_resized_grame(frame)
    last_image_resized = last_frame_resized

    static_found = False

    frame_step = round(SAMPLING_RATE * cam.get(cv2.CAP_PROP_FPS))
    num_frames_chk = round(INITIAL_WINDOW * cam.get(cv2.CAP_PROP_FPS))

    while((static_found == False) and (frame_idx < num_frames_chk)):
        for  _ in range(frame_step):
            ret, frame = cam.read()
            frame_idx += 1

        frame_resized = get_resized_grame(frame)
        similarity = mse(frame_resized, last_frame_resized)

        if(is_static(similarity)):
            valid_image, bw = check_bw_frame(last_image_resized, num_slide, frame_idx)

            if(valid_image):
                printf(f'Image{num_slide}-{frame_idx} -> {bw}')
                save_image(video_name, frame, num_slide)
                num_slide += 1

            static_found = True

        last_frame_resized = frame_resized
        last_image_resized = frame_resized
        
    #### end

    while(True):
        
        for  _ in range(frame_step):
            ret, frame = cam.read()
            frame_idx += 1
    
        if ret:
            frame_resized = get_resized_grame(frame)

            similarity = mse(last_frame_resized, frame_resized)

            if(is_static(similarity)):
                newimg_thresh = mse(frame_resized, last_image_resized)

                if(newimg_thresh >= MSE_NEW):
                    valid_image = True

                    if(newimg_thresh <= MSE_VALID):
                        double_check = msssim(frame_resized, last_image_resized).real
                        image_info  = f'Image{num_slide}-{frame_idx}'
                        msssim_info = f'{newimg_thresh:.2f}, {double_check:5f} msssim'

                        log_info = f'{image_info} -> {msssim_info}'
                        if(double_check < MSSSIM_SAME):
                            printf(f'Ok ' + log_info)
                        else:
                            printf('X ' + log_info)
                            valid_image = False

                    if(valid_image):
                        valid_image, bw = check_bw_frame(frame_resized, num_slide, frame_idx)

                    if(valid_image):
                        image_info = f'Image{num_slide}-{frame_idx}'
                        printf(f'{image_info} -> {newimg_thresh:.2f}, {bw}')
                        save_image(video_name, frame, num_slide)
                        num_slide += 1
                    
                    last_image_resized = frame_resized
                        
            last_frame_resized = frame_resized
        else:
            break

    cam.release()
    print(f'\n{video_name} - Finished\n')

with Pool(8) as p:
    p.map(process_video, videos_path)

lab_intro_vertex_ai_qwik_start - Finished


what_is_vertex_ai_and_why_does_a_unified_platform_m - Finished


how_does_vertex_ai_help_with_the_mlops_workflow_par - Finished


intro_to_mlops_on_vertex_ai - Finished