Task

Hello! I wrote a movie barcode generator in C++ using OpenCV. Self described engineer, space lover, tea drinker Thomas Poulet defines a movie barcode as follows:

A Movie Barcode is the color identity card of the movie. Basically for each frame you take the dominant color and create a stripe out of it. The result is an overview of the overall mood of the movie.

My program uses the following template:

./barcode movieFile samplingRate nBatches nWorkers [-p] [-l]

Here are definitions for the required arguments and flags:

• movieFile: a relative path to a video file (e.g., movie.mp4)
• samplingRate: the rate at which frames are retrieved
• nBatches: the number of batches (a collection of frames) used in the barcode
• nWorkers: the number of workers used to process batches
• p: this flag indicates whether to apply a polar transformation to the barcode
• l: this flag indicates whether to log each step

Approach

Conceptually, my approach uses two key ingredients: a Reader and Worker Class in alignment with the Producer/Consumer design pattern. The reader is responsible for reading in frames and the workers process a batch of frames into a strip.

Code

I wrote this program in a single main.cpp file. The source code is shared below:

/*
 * main.cpp
 * --------
 * This program takes in a video file (e.g., .mp3 or .mp4) and yields a movie barcode
 * with the desired sample rate and batches. The terminal command uses the following template:
 * ./barcode movieFile samplingRate nBatches nWorkers [-p] [-l]
*/

// Include Header Files
#include <opencv2/opencv.hpp>
#include <getopt.h>
#include <iostream>
#include <cassert>
#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>

/*
 * Logger Class
 * ------------
 * This class provides an interface to log messages (e.g., Worker Initiated)
 * By default logging is disabled but may be enabled via the -l flag.
*/
class Logger {
public:
    static bool enableLogging;
    static void log(const std::string &message) {
        if (enableLogging) {
            std::cout << message << std::endl;
        }
    }
};
bool Logger::enableLogging = false;

/*
 * checkArguments()
 * ----------------
 * This helper functions returns TRUE is the arguments supplied appear valid and false otherwise.
 * Errors are reported via assertion statements.
*/
void checkArguments(const std::string &movieFile, int &samplingRate, int &nBatches, int &nWorkers) {
    // Assert that the movie file is readable
    cv::VideoCapture cap(movieFile);
    assert(cap.isOpened());

    // Assert that the arguments are non-negative and do not exceed extreme bounds
    int totalFrames = (int)cap.get(cv::CAP_PROP_FRAME_COUNT);
    int usedFrames = totalFrames / samplingRate;
    assert(samplingRate > 0);
    assert(nBatches > 0);
    assert(nWorkers > 0);
    assert(samplingRate < totalFrames);
    assert(nBatches < usedFrames);
}

/*
 * polarTransform()
 * ----------------
 * This helper function remaps the barcode (passed by reference) from cartesian to 
 * polar space in-place.
*/
void polarTransform(cv::Mat &barcode, int nBatches) {
    // Prepare for transformation
    cv::Mat flipped;
    cv::Mat polarImage;
    cv::rotate(barcode, flipped, cv::ROTATE_90_CLOCKWISE);
    
    // Apply transformation
    cv::Size dsize(nBatches, nBatches);
    cv::Point2f center(nBatches / 2.0f, nBatches / 2.0f);
    int maxRadius = nBatches / 2.0f;
    cv::warpPolar(flipped, polarImage, dsize, center, maxRadius, cv::WARP_INVERSE_MAP);

    // Create a circular mask to make pixels outside the circle transparent
    cv::Mat mask = cv::Mat::zeros(polarImage.size(), CV_8UC1);
    cv::circle(mask, center, maxRadius, cv::Scalar(255), -1);

    // Apply the mask to make pixels outside the circle transparent
    cv::cvtColor(polarImage, polarImage, cv::COLOR_BGR2BGRA);
    polarImage.setTo(cv::Scalar(0, 0, 0, 0), mask == 0);
    barcode = polarImage;
}

/*
 * Batch Struct
 * ------------
 * The Batch Struct maintains a vector of frames and a unique identifier.
*/
struct Batch {
    int id;
    std::vector<cv::Mat> frames;
};

/*
 * Worker Class
 * ------------
 * This class maintains all worker threads. It includes the following functions:
 * processBatch() - reads in batches until the worker queue is empty and reader is finished
 * addBatch() - pushes a batch to the worker queue
 * getResults() - returns `results` vector containing row-reduced batches
 * setState() - sets reader state
*/
class Worker {
public:
    // Static Member Variables
    static std::vector<cv::Mat> results;
    static std::vector<Worker*> workers;
    static std::queue<Batch> queue;
    static std::mutex mtx;
    static std::condition_variable cv;
    static bool doneReading;

    // Non-Static Member Variables
    int workerId;
    std::thread workerThread;

    // Constructor
    Worker(int workerId) : workerId(workerId) {
        workerThread = std::thread(&Worker::processBatch, this);
        workers.push_back(this);
        Logger::log("Worker " + std::to_string(workerId) + " created.");
    }

    void processBatch() {
        while (true) {
            Batch batch;
            {   
                // Check if there are batches to process...
                std::unique_lock<std::mutex> lock(mtx);
                cv.wait(lock, [this] { return !queue.empty() || doneReading; });
                if (queue.empty() && doneReading) break;
                if (queue.empty()) continue;

                // If so, peek and pop off the queue
                batch = queue.front();
                queue.pop();
            }
            Logger::log("Worker " + std::to_string(workerId) + " processing batch " + std::to_string(batch.id) + ".");

            // Process the batch
            cv::Mat concatenation;
            cv::Mat average;
            cv::hconcat(batch.frames, concatenation);
            cv::reduce(concatenation, average, 1, cv::REDUCE_AVG);

            // Store the result
            // thread-safe because we never modify the same memory [https://stackoverflow.com/a/61013298]
            results[batch.id] = average;
            Logger::log("Worker " + std::to_string(workerId) + " finished processing batch " + std::to_string(batch.id) + ".");
        }
        Logger::log("Worker " + std::to_string(workerId) + " exiting.");
    }

    static std::vector<cv::Mat> getResults() {
        return results;
    }

    static std::vector<Worker*>& getWorkers() {
        return workers;
    }

    static void allocateBatch(const Batch &batch) {
        {
            std::unique_lock<std::mutex> lock(mtx);
            queue.push(batch);
        }
        cv.notify_one();
    }

    static void setState(bool state) {
        doneReading = state;
        cv.notify_all();
    }
};

// Define Static Member Variables
std::vector<cv::Mat> Worker::results;
std::vector<Worker*> Worker::workers;
std::queue<Batch> Worker::queue;
std::mutex Worker::mtx;
std::condition_variable Worker::cv;
bool Worker::doneReading = false;

/*
 * Reader Class
 * ------------
 * This (singleton) class maintains a reader instance which reads frames from a movie.  
 * It includes the following functions:
 * readFrames() - grabs, retrieves, and allocates batches to workers
 * allocateBatch() - allocates batch to workers in round-robin fashion
*/
class Reader {
public:
    // Instance Member Variables
    int samplingRate;
    int nBatches;
    int nWorkers;
    int framesPerBatch;
    int batchSplit;
    int currentWorker;
    std::vector<Worker*>& workers;
    cv::VideoCapture cap;
    
    Reader(const std::string &movieFile, int samplingRate, int nBatches, int nWorkers)
        : samplingRate(samplingRate), nBatches(nBatches), nWorkers(nWorkers),
          currentWorker(0), workers(Worker::getWorkers()) {
            // Open the video file
            cap.open(movieFile);

            // Compute frames per batch
            int totalFrames = (int)cap.get(cv::CAP_PROP_FRAME_COUNT);
            int usedFrames = totalFrames / samplingRate;
            framesPerBatch = usedFrames / nBatches;
            Logger::log("Reader created for file: " + movieFile);
    }

    void readFrames() {
        // Setup Counters and Current Batch
        int frameCounter = 0;
        int batchCounter = 0;
        Batch currentBatch;
        currentBatch.id = batchCounter;

        while (batchCounter < nBatches) {
            // Grab frames from the capture
            cap.grab();

            // Retrieve frames based on the sampling rate
            if (frameCounter % samplingRate == 0) {
                cv::Mat frame;
                cap.retrieve(frame);
                currentBatch.frames.push_back(frame);

                // Add frame and allocate batch (if size is met) 
                if (currentBatch.frames.size() == framesPerBatch) {
                    Worker::allocateBatch(currentBatch);
                    batchCounter++;
                    currentBatch = Batch{batchCounter};
                }
            }
            frameCounter++;
        }
        // Handle remaining frames in the last batch
        if (!currentBatch.frames.empty()) {
            Worker::allocateBatch(currentBatch);
        }
        Worker::setState(true);
        Logger::log("Finished reading frames.");
    }
};

int main(int argc, char** argv) {
    // Initialize variables
    std::string movieFile;
    int samplingRate = 0;
    int nBatches = 0;
    int nWorkers = 0;
    bool applyPolar = false;

    // Parse Arguments using getopt
    int opt;
    while ((opt = getopt(argc, argv, "pl")) != -1) {
        switch (opt) {
            case 'p':
                applyPolar = true;
                break;
            case 'l':
                Logger::enableLogging = true;
                break;
            default:
                std::cerr << "Usage: " << argv[0] << " movieFile samplingRate nBatches nWorkers [-p] [-l]" << std::endl;
                return EXIT_FAILURE;
        }
    }
    std::cerr << applyPolar << std::endl;

    // Parse positional arguments
    if (optind + 4 > argc) {
        std::cerr << "Usage: " << argv[0] << " movieFile samplingRate nBatches nWorkers [-p] [-l]" << std::endl;
        return EXIT_FAILURE;
    }
    movieFile = argv[optind];
    samplingRate = std::stoi(argv[optind + 1]);
    nBatches = std::stoi(argv[optind + 2]);
    nWorkers = std::stoi(argv[optind + 3]);

    // Check Arguments
    checkArguments(movieFile, samplingRate, nBatches, nWorkers);

    // Setup workers
    Worker::results.resize(nBatches);
    for (int id = 0; id < nWorkers; ++id) {
        new Worker(id);
    }

    // Construct and initiate reader
    Reader reader(movieFile, samplingRate, nBatches, nWorkers);
    reader.readFrames();

    // Wait for all workers to finish up their batches
    auto &workers = Worker::getWorkers();
    for (auto &worker : workers) {
        if (worker->workerThread.joinable()) {
            worker->workerThread.join();
        }
    }

    // Package the vector into a single image
    std::vector<cv::Mat> input = Worker::getResults();
    cv::Mat barcode;
    cv::hconcat(input, barcode);
    
    // Apply polar transform is requested...
    if (applyPolar) {
        polarTransform(barcode, nBatches);
    }
    
    // Save the barcode image
    cv::imwrite("../barcode.png", barcode);
    Logger::log("Barcode image saved as barcode.png");

    // Exit with Status 0
    return 0;
}

Concerns

Anything is fair game, but I wanted to draw attention to the following points:

1. Does the code have good style in terms of commenting and modularity?
2. Can the approach be improved to leverage multithreading more effectively?