jagin

import os
import cv2

from pipeline.libs.face_detector import FaceDetector
import tests.config as config


class TestFaceDetector:
    def test_face_detector(self):
        prototxt = os.path.join(config.MODELS_FACE_DETECTOR_DIR, "deploy.prototxt.txt")

jagin

from pipeline.pipeline import Pipeline
from pipeline.libs.face_detector import FaceDetector

class DetectFaces(Pipeline):
    def __init__(self, prototxt, model, batch_size=1, confidence=0.5):
        self.detector = FaceDetector(prototxt, model, confidence=confidence)
        self.batch_size = batch_size

        super(DetectFaces, self).__init__()

jagin

import cv2
import numpy as np

class FaceDetector:
    def __init__(self, prototxt, model, confidence=0.5):
        self.confidence = confidence

        self.net = cv2.dnn.readNetFromCaffe(prototxt, model)

    def detect(self, images):

jagin

import cv2
from pipeline.pipeline import Pipeline

class CaptureVideo(Pipeline):
    def __init__(self, src=0):
        self.cap = cv2.VideoCapture(src)
        if not self.cap.isOpened():
            raise IOError(f"Cannot open video {src}")

        self.fps = int(self.cap.get(cv2.CAP_PROP_FPS))

jagin

import os

from pipeline.load_images import LoadImages
from pipeline.cascade_detect_faces import CascadeDetectFaces
from pipeline.save_faces import SaveFaces
from pipeline.save_summary import SaveSummary
from pipeline.display_summary import DisplaySummary

def parse_args():
    import argparse

jagin

import cv2

from pipeline.pipeline import Pipeline

class CascadeDetectFaces(Pipeline):
    def __init__(self, classifier):
        # load the face detector
        self.detector = cv2.CascadeClassifier(classifier)

        super(DetectFaces, self).__init__()

jagin

import cv2

from pipeline.pipeline import Pipeline
import pipeline.utils as utils

class LoadImages(Pipeline):
    def __init__(self, src, valid_exts=(".jpg", ".png")):
        self.src = src
        self.valid_exts = valid_exts

jagin

class Pipeline(object):
    def __init__(self):
        self.source = None

    def __iter__(self):
        return self.generator()

    def generator(self):
        while self.has_next():
            data = next(self.source) if self.source else {}

jagin

import cv2
import os
import json
import numpy as np

def parse_args():
    import argparse

    # Parse command line arguments
    ap = argparse.ArgumentParser(description="Image processing pipeline")