LDA.hpp

#ifndef _LDAPLUSPLUS_LDA_HPP_
#define _LDAPLUSPLUS_LDA_HPP_


#include <condition_variable>
#include <list>
#include <memory>
#include <mutex>
#include <vector>
#include <thread>
#include <tuple>

#include <Eigen/Core>

#include "ldaplusplus/events/Events.hpp"
#include "ldaplusplus/em/EStepInterface.hpp"
#include "ldaplusplus/em/MStepInterface.hpp"
#include "ldaplusplus/Parameters.hpp"

namespace ldaplusplus {


template <typename Scalar = double>
class LDA
{
    typedef Eigen::Matrix<Scalar, Eigen::Dynamic, Eigen::Dynamic> MatrixX;
    typedef Eigen::Matrix<Scalar, Eigen::Dynamic, 1> VectorX;

    public:
        LDA(
            std::shared_ptr<parameters::Parameters> model_parameters,
            std::shared_ptr<em::EStepInterface<Scalar> > e_step,
            std::shared_ptr<em::MStepInterface<Scalar> > m_step,
            size_t iterations = 20,
            size_t workers = 1
        );

        LDA(LDA &&lda);

        void fit(const Eigen::MatrixXi &X, const Eigen::VectorXi &y);

        void fit(const Eigen::MatrixXi &X);

        void partial_fit(const Eigen::MatrixXi &X, const Eigen::VectorXi &y);

        void partial_fit(std::shared_ptr<corpus::Corpus> corpus);

        MatrixX transform(const Eigen::MatrixXi &X);

        MatrixX decision_function(const Eigen::MatrixXi &X);

        Eigen::VectorXi predict(const Eigen::MatrixXi &X);

        std::tuple<MatrixX, Eigen::VectorXi> transform_predict(const Eigen::MatrixXi &X);

        std::shared_ptr<events::EventDispatcherInterface> get_event_dispatcher() {
            return event_dispatcher_;
        }

        const std::shared_ptr<parameters::Parameters> model_parameters() {
            return model_parameters_;
        }

        template <typename P>
        const std::shared_ptr<P> model_parameters() {
            return std::static_pointer_cast<P>(model_parameters_);
        }

    protected:
        std::shared_ptr<corpus::Corpus> get_corpus(
            const Eigen::MatrixXi &X,
            const Eigen::VectorXi &y
        );

        std::shared_ptr<corpus::Corpus> get_corpus(const Eigen::MatrixXi &X);

        void create_worker_pool();

        void destroy_worker_pool();

        void process_worker_events() {
            std::static_pointer_cast<events::ThreadSafeEventDispatcher>(
                event_dispatcher_
            )->process_events();
        }

        std::tuple<std::shared_ptr<parameters::Parameters>, size_t> extract_vp_from_queue();

        void doc_e_step_worker();

        MatrixX decision_function(const MatrixX &X);

        Eigen::VectorXi predict(const MatrixX &scores);


    private:
        void set_up_event_dispatcher();

        // The model parameters
        std::shared_ptr<parameters::Parameters> model_parameters_;

        // The LDA implementation
        std::shared_ptr<em::EStepInterface<Scalar> > e_step_;
        std::shared_ptr<em::MStepInterface<Scalar> > m_step_;

        // Member variables that affect the behaviour of fit
        size_t iterations_;

        // The thread related member variables
        std::vector<std::thread> workers_;
        std::mutex queue_in_mutex_;
        std::list<std::tuple<std::shared_ptr<corpus::Corpus>, size_t> > queue_in_;
        std::mutex queue_out_mutex_;
        std::condition_variable queue_out_cv_;
        std::list<std::tuple<std::shared_ptr<parameters::Parameters>, size_t> > queue_out_;

        // An event dispatcher that we will use to communicate with the
        // external components
        std::shared_ptr<events::EventDispatcherInterface> event_dispatcher_;
};


} // namespace ldaplusplus
#endif  // _LDAPLUSPLUS_LDA_HPP_