Merge branch 'tg-45-alf-classifier' into 'main'

WIF - ALF Classifier

See merge request feta/wif-group/libwif!36

Author: plnyrich

include/wif/classifiers/alfClassifier.hpp

@@ -0,0 +1,93 @@
+/**
+ * @file
+ * @author Pavel Siska <[email protected]>
+ * @author Richard Plny <[email protected]>
+ * @brief ALF classifier interface
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#pragma once
+
+#include "wif/classifiers/scikitMlClassifier.hpp"
+#include "wif/filesystem/fileModificationChecker.hpp"
+#include "wif/reporters/unirecReporter.hpp"
+#include "wif/utils/timer.hpp"
+
+#include <chrono>
+#include <memory>
+#include <vector>
+
+namespace WIF {
+
+/**
+ * @brief Classifier performing ML classification which is interconnected with ALF
+ *
+ */
+class AlfClassifier : public Classifier {
+private:
+	class AlfCallback : public TimerCallback {
+	public:
+		AlfCallback(const std::string& modelPath, AlfClassifier& classifier);
+
+		void onTick() override;
+
+	private:
+		FileModificationChecker m_fileModificationChecker;
+		AlfClassifier& m_classifier;
+	};
+
+public:
+	/**
+	 * @brief Construct a new Alf Classifier object
+	 *
+	 * @param mlClassifier underlying ML classifier, which has already set source feature IDs
+	 * @param reporter UniRec reporter for ALF interconnection, which already has its interface
+	 * ready
+	 * @param timerIntervalInSeconds number of seconds between each check of file on disk
+	 */
+	AlfClassifier(
+		ScikitMlClassifier& mlClassifier,
+		UnirecReporter& reporter,
+		unsigned timerIntervalInSeconds);
+
+	/**
+	 * @brief Classify single flowFeature object
+	 * See std::vector<ClfResult> classify(const std::vector<FlowFeatures>&) for more details
+	 *
+	 * @param flowFeatures flow features to classify
+	 * @return ClfResult result of the classification
+	 */
+	ClfResult classify(const FlowFeatures& flowFeatures) override;
+
+	/**
+	 * @brief Classify a burst of flow features
+	 * Firstly, ML model is reloaded if needed, then ML classifier is used to obtain results
+	 * Then, source features of ALF classifier are reported, then source features of ML classifier
+	 * are reported, then last model load time (uint64_t) and finally classification results are
+	 * reported
+	 *
+	 * @param burstOfFlowsFeatures the burst of flow features to classify
+	 * @return std::vector<ClfResult> the results of the classification
+	 */
+	std::vector<ClfResult> classify(const std::vector<FlowFeatures>& burstOfFlowsFeatures) override;
+
+	/**
+	 * @brief Method which marks used ML model as obsolete
+	 * This operation will result into ML model reload during the next call of classify() method
+	 */
+	void markModelAsObsolete() { m_modelReloadNeeded = true; }
+
+private:
+	void updateLastModelLoadTime();
+	void handleModelUpdate();
+	void handleSingleReport(const FlowFeatures flowFeatures, const ClfResult& result);
+
+	bool m_modelReloadNeeded = false;
+	uint64_t m_lastModelLoadTime;
+	ScikitMlClassifier& m_mlClassifier;
+	UnirecReporter& m_reporter;
+	Timer m_timer;
+};
+
+} // namespace WIF

include/wif/classifiers/scikitMlClassifier.hpp

@@ -29,13 +29,25 @@ class ScikitMlClassifier : public Classifier {
 	 */
 	ScikitMlClassifier(const std::string& bridgePath, const std::string& mlModelPath);
 
+	/**
+	 * @brief Get the path to the used ML model
+	 *
+	 * @return const std::string& the path to the used ML model
+	 */
+	const std::string& getMlModelPath() const noexcept;
+
 	/**
 	 * @brief Set feature IDs which will be used for classification
 	 *
 	 * @param sourceFeatureIDs
 	 */
 	void setFeatureSourceIDs(const std::vector<FeatureID>& sourceFeatureIDs) override;
 
+	/**
+	 * @brief Reload used ML model from disk
+	 */
+	void reloadModelFromDisk();
+
 	/**
 	 * @brief Classify single flowFeature object
 	 *

include/wif/ml/scikitMlWrapper.hpp

@@ -57,10 +57,10 @@ class ScikitMlWrapper {
 	void setFeatureSourceIDs(const std::vector<FeatureID>& sourceFeatureIDs);
 
 	/**
-	 * @brief Getter for used ML model path
+	 * @brief Getter for path of the used ML model
 	 * @return const std::string&
 	 */
-	const std::string& mlModelPath() const noexcept { return m_mlModelPath; }
+	const std::string& getMlModelPath() const noexcept { return m_mlModelPath; }
 
 	/**
 	 * @brief Reload model from disk

include/wif/storage/flowFeatures.hpp

@@ -45,6 +45,14 @@ class FlowFeatures {
 		return std::get<T>(m_features.at(featureID));
 	}
 
+	/**
+	 * @brief Get raw feature
+	 *
+	 * @param featureID the feature identifier
+	 * @return const DataVariant& raw feature
+	 */
+	const DataVariant& getRaw(FeatureID featureID) const { return m_features.at(featureID); }
+
 	/**
 	 * @brief Set feature
 	 *

src/wif/CMakeLists.txt

@@ -21,6 +21,7 @@ set(LIBWIF_LIBS
 
 if(BUILD_WITH_UNIREC)
 	list(APPEND LIBWIF_SOURCES
+		classifiers/alfClassifier.cpp
 		reporters/unirecReporter.cpp
 	)
 	list(APPEND LIBWIF_LIBS

src/wif/classifiers/alfClassifier.cpp

@@ -0,0 +1,100 @@
+/**
+ * @file
+ * @author Pavel Siska <[email protected]>
+ * @author Richard Plny <[email protected]>
+ * @brief ALF classifier implementation
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#include "wif/classifiers/alfClassifier.hpp"
+
+namespace WIF {
+
+AlfClassifier::AlfCallback::AlfCallback(const std::string& modelPath, AlfClassifier& classifier)
+	: m_fileModificationChecker(modelPath)
+	, m_classifier(classifier)
+{
+}
+
+void AlfClassifier::AlfCallback::onTick()
+{
+	if (m_fileModificationChecker.isChangeDetected()) {
+		m_classifier.markModelAsObsolete();
+	}
+}
+
+AlfClassifier::AlfClassifier(
+	ScikitMlClassifier& mlClassifier,
+	UnirecReporter& reporter,
+	unsigned timerIntervalInSeconds)
+	: m_mlClassifier(mlClassifier)
+	, m_reporter(reporter)
+	, m_timer(
+		  timerIntervalInSeconds,
+		  std::make_unique<AlfCallback>(mlClassifier.getMlModelPath(), *this))
+{
+	updateLastModelLoadTime();
+	m_timer.start();
+}
+
+ClfResult AlfClassifier::classify(const FlowFeatures& flowFeatures)
+{
+	return classify(std::vector<FlowFeatures> {flowFeatures})[0];
+}
+
+std::vector<ClfResult>
+AlfClassifier::classify(const std::vector<FlowFeatures>& burstOfFlowsFeatures)
+{
+	if (m_modelReloadNeeded) {
+		handleModelUpdate();
+	}
+
+	auto burstOfResults = m_mlClassifier.classify(burstOfFlowsFeatures);
+	for (unsigned flowIdx = 0; flowIdx < burstOfResults.size(); ++flowIdx) {
+		auto& flowFeatures = burstOfFlowsFeatures[flowIdx];
+		auto& results = burstOfResults[flowIdx];
+		handleSingleReport(flowFeatures, results);
+	}
+
+	return burstOfResults;
+}
+
+void AlfClassifier::updateLastModelLoadTime()
+{
+	const auto timestamp = std::chrono::system_clock::now();
+	m_lastModelLoadTime
+		= std::chrono::duration_cast<std::chrono::seconds>(timestamp.time_since_epoch()).count();
+}
+
+void AlfClassifier::handleModelUpdate()
+{
+	m_mlClassifier.reloadModelFromDisk();
+	m_modelReloadNeeded = false;
+	updateLastModelLoadTime();
+}
+
+void AlfClassifier::handleSingleReport(const FlowFeatures flowFeatures, const ClfResult& result)
+{
+	m_reporter.onRecordStart();
+
+	// Identification fields are reported first
+	for (const auto id : sourceFeatureIDs()) {
+		m_reporter.report(flowFeatures.getRaw(id));
+	}
+
+	// ML features are reported as second
+	for (const auto id : m_mlClassifier.sourceFeatureIDs()) {
+		m_reporter.report(flowFeatures.getRaw(id));
+	}
+
+	// Report last time ML model was reloaded
+	m_reporter.report({m_lastModelLoadTime});
+
+	// ML proba array is reported as the last one
+	m_reporter.report(result.get<std::vector<double>>());
+
+	m_reporter.onRecordEnd();
+}
+
+} // namespace WIF

src/wif/classifiers/scikitMlClassifier.cpp

@@ -17,12 +17,22 @@ ScikitMlClassifier::ScikitMlClassifier(
 	m_scikitMlWrapper = std::make_unique<ScikitMlWrapper>(bridgePath, mlModelPath);
 }
 
+const std::string& ScikitMlClassifier::getMlModelPath() const noexcept
+{
+	return m_scikitMlWrapper->getMlModelPath();
+}
+
 void ScikitMlClassifier::setFeatureSourceIDs(const std::vector<FeatureID>& sourceFeatureIDs)
 {
 	Classifier::setFeatureSourceIDs(sourceFeatureIDs);
 	m_scikitMlWrapper->setFeatureSourceIDs(sourceFeatureIDs);
 }
 
+void ScikitMlClassifier::reloadModelFromDisk()
+{
+	m_scikitMlWrapper->reloadModel();
+}
+
 ClfResult ScikitMlClassifier::classify(const FlowFeatures& flowFeatures)
 {
 	return m_scikitMlWrapper->classify({flowFeatures})[0];