基于opencv实现模块化图像处理管道

import cv2import osimport jsonimport numpy as np
def parse_args():    import argparse
    # Parse command line arguments    ap = argparse.ArgumentParser(description="Image processing pipeline")    ap.add_argument("-i", "--input", required=True,                    help="path to input image files")    ap.add_argument("-o", "--output", default="output",                    help="path to output directory")    ap.add_argument("-os", "--out-summary", default=None,                    help="output JSON summary file name")    ap.add_argument("-c", "--classifier", default="models/haarcascade/haarcascade_frontalface_default.xml",                    help="path to where the face cascade resides")
    return vars(ap.parse_args())
def list_images(path, valid_exts=None):    image_files = []    # Loop over the input directory structure    for (root_dir, dir_names, filenames) in os.walk(path):        for filename in sorted(filenames):            # Determine the file extension of the current file            ext = filename[filename.rfind("."):].lower()            if valid_exts and ext.endswith(valid_exts):                # Construct the path to the file and yield it                file = os.path.join(root_dir, filename)                image_files.append(file)
    return image_files
def main(args):    os.makedirs(args["output"], exist_ok=True)
    # load the face detector    detector = cv2.CascadeClassifier(args["classifier"])
    # list images from input directory    input_image_files = list_images(args["input"], (".jpg", ".png"))
    # Storage for JSON summary    summary = {}
    # Loop over the image paths    for image_file in input_image_files:        # Load the image and convert it to grayscale        image = cv2.imread(image_file)        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
        # Detect faces        face_rects = detector.detectMultiScale(gray, scaleFactor=1.05, minNeighbors=5,                                               minSize=(30, 30), flags=cv2.CASCADE_SCALE_IMAGE)        summary[image_file] = {}        # Loop over all detected faces        for i, (x, y, w, h) in enumerate(face_rects):            face = image[y:y+w, x:x+h]
            # Prepare output directory for faces            output = os.path.join(*(image_file.split(os.path.sep)[1:]))            output = os.path.join(args["output"], output)            os.makedirs(output, exist_ok=True)
            # Save faces            face_file = os.path.join(output, f"{i:05d}.jpg")            cv2.imwrite(face_file, face)
            # Store summary data            summary[image_file][face_file] = np.array([x, y, w, h], dtype=int).tolist()
        # Display summary        print(f"[INFO] {image_file}: face detections {len(face_rects)}")
    # Save summary data    if args["out_summary"]:        summary_file = os.path.join(args["output"], args["out_summary"])        print(f"[INFO] Saving summary to {summary_file}...")        with open(summary_file, 'w') as json_file:            json_file.write(json.dumps(summary))
if __name__ == '__main__':    args = parse_args()    main(args)

#! /usr/bin/env python
class Pipeline(object):    def __init__(self):        self.source = None        def __iter__(self):        return self.generator()        def generator(self):        while True:            value = self.source.next()            if self.filter(value):                yield self.map(value)        def __or__(self, other):        other.source = self.generator()        return other        def filter(self, value):        return True        def map(self, value):        return value

class AllNumbers(Pipeline):    def generator(self):        value = 0        while True:            yield value            value += 1

class Evens(Pipeline):    def filter(self, value):        return value % 2 == 0

class MultipleOf(Pipeline):    def __init__(self, factor=1):        self.factor = factor        super(MultipleOf, self).__init__()            def filter(self, value):        return value % self.factor == 0

class Printer(Pipeline):    def map(self, value):        print value        return value

class First(Pipeline):    def __init__(self, total=10):        self.total = total        self.count = 0        super(First, self).__init__()        def map(self, value):        self.count += 1        if self.count > self.total:            raise StopIteration        return value

def main():    all_numbers = AllNumbers()    evens = MultipleOf(2)    multiple_of_3 = MultipleOf(3)    printer = Printer()    first_10 = First(10)    pipeline = all_numbers | evens | multiple_of_3 | first_10 | printer        for i in pipeline:        pass

if __name__ == '__main__':    main()

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 {}            if self.filter(data):                yield self.map(data)
    def __or__(self, other):        other.source = self.generator()        return other
    def filter(self, data):        return True
    def map(self, data):        return data
    def has_next(self):        return True

import cv2
from pipeline.pipeline import Pipelineimport pipeline.utils as utils
class LoadImages(Pipeline):    def __init__(self, src, valid_exts=(".jpg", ".png")):        self.src = src        self.valid_exts = valid_exts
        super(LoadImages, self).__init__()
    def generator(self):        source = utils.list_images(self.src, self.valid_exts)        while self.has_next():            image_file = next(source)            image = cv2.imread(image_file)
            data = {                "image_file": image_file,                "image": image            }
            if self.filter(data):                yield self.map(data)

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__()
    def map(self, data):        image = data["image"]
        # Detect faces        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)        face_rects = self.detector.detectMultiScale(gray, scaleFactor=1.05, minNeighbors=5,                                                    minSize=(30, 30), flags=cv2.CASCADE_SCALE_IMAGE)        data["face_rects"] = face_rects
        return data

import os
from pipeline.load_images import LoadImagesfrom pipeline.cascade_detect_faces import CascadeDetectFacesfrom pipeline.save_faces import SaveFacesfrom pipeline.save_summary import SaveSummaryfrom pipeline.display_summary import DisplaySummary
def parse_args():    import argparse
    # Parse command line arguments    ap = argparse.ArgumentParser(description="Image processing pipeline")    ap.add_argument("-i", "--input", required=True,                    help="path to input image files")    ap.add_argument("-o", "--output", default="output",                    help="path to output directory")    ap.add_argument("-os", "--out-summary", default=None,                    help="output JSON summary file name")    ap.add_argument("-c", "--classifier", default="models/haarcascade/haarcascade_frontalface_default.xml",                    help="path to where the face cascade resides")
    return vars(ap.parse_args())
def main(args):    # Create pipeline steps    load_images = LoadImages(args["input"])    detect_faces = CascadeDetectFaces(args["classifier"])    save_faces = SaveFaces(args["output"])    if args["out_summary"]:        summary_file = os.path.join(args["output"], args["out_summary"])        save_summary = SaveSummary(summary_file)    display_summary = DisplaySummary()
    # Create image processing pipeline    pipeline = load_images | detect_faces | save_faces    if args["out_summary"]:        pipeline |= save_summary    pipeline |= display_summary
    # Iterate through pipeline    for _ in pipeline:        pass
    if args["out_summary"]:        print(f"[INFO] Saving summary to {summary_file}...")        save_summary.write()
if __name__ == '__main__':    args = parse_args()    main(args)

python process_video_pipeline.py -i assets/videos/faces.mp4 -ov faces.avi -p 12%|██████████████████████████ ██████▋ | 71/577 [00:08<01:12, 6.99it/s]