check_qps.py

from main import FaceFakePipelineImage, FaceFakePipelineVideo
from utils import readb64, img2base64, binary2video, extract_frames

import GPUtil, psutil
from threading import Thread
import time
import cv2


class Monitor(Thread):
    def __init__(self, delay):
        super(Monitor, self).__init__()
        self.stopped = False
        self.delay = delay # Time between calls to GPUtil
        self.start()

    def run(self):
        self.cpus = []
        while not self.stopped:
            GPUtil.showUtilization()
            cpu = psutil.cpu_percent()
            self.cpus.append(cpu)
            print(cpu)
            time.sleep(self.delay)

    def stop(self):
        self.stopped = True
        print('cpu avg:', sum(self.cpus)/len(self.cpus))

def image_test():

    test_files = [
        'sample_files/face_fake.png',
        'sample_files/face_real.png',
        'sample_files/fem_fake_img.jpg',
        'sample_files/fem_real_img.jpg',
        'sample_files\sapi.jpg',
        'sample_files/face_fake.png',
        'sample_files/face_real.png',
        'sample_files/fem_fake_img.jpg',
        'sample_files/fem_real_img.jpg',
        'sample_files/face_fake.png',
    ]

    images = [img2base64(cv2.imread(test_file)) for test_file in test_files]
    payloads = [{'images': [image]} for image in images]


    # init pipeline
    pipeline = FaceFakePipelineImage()
    config_payload = {
        'face_detection_threshold': 0.997,
        'deformation_detection_threshold': 0.6,
        'deepfake_detection_threshold': 0.65,
    }
    
    monitor = Monitor(0.5)
    t0 = time.time()
    for payload in payloads:
        pred = pipeline(payload, config_payload)
    t = time.time() - t0
    monitor.stop()

    print('QPS:', len(payloads)/t)




    # show results
    # from pprint import pprint
    # pprint(pred)


def video_test():
    payloads = [
        'sample_files/jokowi deepfake - cupid.mp4',
        'sample_files/messi_deepfake.mp4',
        'sample_files/obama_real.mp4',
    ]
    
    video_pipeline = FaceFakePipelineVideo()
    config_payload = {
        'face_detection_threshold': 0.997,
        'deepfake_detection_threshold': 0.65,
        'frame_sampling_interval': 60,
        'frame_sampling_max': 50,
    }

    t0 = time.time()
    for payload in payloads:
        print('name:', payload)
        t0_ = time.time()
        pred = video_pipeline(payload, config_payload)
        t_ = time.time() - t0_
        print('latency:', t_)
    t = time.time() - t0

    print('QPS:', len(payloads)/t)

    from pprint import pprint
    pprint(pred)

if __name__ == "__main__":
    # image_test()
    video_test()