finish DBStream

include/Algorithm/Algorithm.hpp

@@ -14,7 +14,7 @@
 using namespace std;
 namespace SESAME {
 
-enum algoType { BirchType, StreamKMeansType, CluStreamType, DenStreamType };
+enum algoType { BirchType, StreamKMeansType, CluStreamType, DenStreamType, DBStreamType};
 
 class Algorithm;
 typedef std::shared_ptr<Algorithm> AlgorithmPtr;

include/Algorithm/CluStream.hpp

@@ -44,7 +44,7 @@ class CluStream : public Algorithm {
   int pointsForgot;
   int pointsMerged;
   clock_t startTime;
-  clock_t lastTime;
+  clock_t lastUpdateTime;
   CluStream(param_t &cmd_params);
   ~CluStream();
 

include/Algorithm/DBStream.hpp

@@ -0,0 +1,61 @@
+//
+// Created by 1124a on 2021/8/30.
+//
+
+#ifndef SESAME_INCLUDE_ALGORITHM_DBSTREAM_HPP_
+#define SESAME_INCLUDE_ALGORITHM_DBSTREAM_HPP_
+#include <Algorithm/Algorithm.hpp>
+#include <Algorithm/DataStructure/WeightedAdjacencyList.hpp>
+#include <Utils/BenchmarkUtils.hpp>
+
+namespace SESAME {
+typedef std::vector<std::vector<MicroClusterPtr>> Clusters;
+class DBStreamParams : public AlgorithmParameters {
+ public:
+  double radius;
+  double lambda;
+  int cleanUpInterval;//Time gap
+  double weightMin;//minimum weight
+  double alpha;//α, intersection factor
+  double base;//base of decay function
+};
+
+
+class DBStream : public Algorithm
+    {
+     public:
+      DBStreamParams dbStreamParams;
+      DampedWindowPtr dampedWindow;
+      std::vector<MicroClusterPtr> microClusters;
+      SESAME::WeightedAdjacencyList weightedAdjacencyList;
+      std::vector<MicroClusterPtr> microClusterNN;//micro clusters found in function findFixedRadiusNN
+      int weakEntry;//W_weak, weak entries
+      double aWeakEntry;
+      clock_t startTime;
+      clock_t pointArrivingTime;
+      clock_t lastCleanTime;
+      int microClusterIndex;
+      //Final output of clusters
+      Clusters finalClusters;
+     //Connectivity graph
+     unordered_map<int,std::vector<int>> connecvtivityGraphId;
+
+      //TODO Need to implement weighted a weighted adjacency list S
+      DBStream(param_t &cmd_params);
+      ~DBStream();
+      void Initilize() override;
+      void runOnlineClustering(PointPtr input) override;
+      void runOfflineClustering(DataSinkPtr sinkPtr) override;
+     private:
+      bool isInitial  = false;
+      void update(PointPtr dataPoint);
+      bool checkMove( std::vector<MicroClusterPtr> microClusters) const;
+      std::vector<MicroClusterPtr> findFixedRadiusNN(PointPtr dataPoint);
+      void cleanUp(clock_t nowTime);
+      void reCluster(double threshold);
+      void insertIntoGraph(int microClusterId,int OtherId);
+      void insertIntoGraph(int microClusterId);
+      void findConnectedComponents();
+    };
+}
+#endif //SESAME_INCLUDE_ALGORITHM_DBSTREAM_HPP_

include/Algorithm/DataStructure/DataStructureFactory.hpp

@@ -13,7 +13,7 @@
 #include <Algorithm/DataStructure/Snapshot.hpp>
 #include <Algorithm/DataStructure/CFTree.hpp>
 #include <Algorithm/DataStructure/FeatureVector.hpp>
-
+#include <Algorithm/DataStructure/WeightedAdjacencyList.hpp>
 namespace SESAME {
 class DataStructureFactory {
 
@@ -28,12 +28,16 @@ class DataStructureFactory {
   static CoresetTreePtr createCoresetTree();
   static void clearCoresetTree(CoresetTreePtr tree);
   static MicroClusterPtr createMicroCluster(int dimension, int id);
+  static MicroClusterPtr createMicroCluster(int dimension, int id,PointPtr dataPoint,double radius);
   static void clearMicroCluster(MicroClusterPtr microCluster);
   static SnapshotPtr createSnapshot(MicroClusters & otherMicroClusters,int elapsedTime);
   static void clearSnapshot(SnapshotPtr snapshot);
   static CFTreePtr createCFTree();
   static NodePtr createNode();
-
+  static MicroClusterPairPtr createMicroClusterPair(MicroClusterPtr microCluster1,MicroClusterPtr microCluster2);
+  static void clearMicroClusterPair(MicroClusterPairPtr microClusterPair);
+  static AdjustedWeightPtr createAdjustedWeight(double weight, clock_t pointTime);
+  static void clearAdjustedWeight(AdjustedWeightPtr adjustedWeight);
 };
 }
 #endif //SESAME_INCLUDE_ALGORITHM_DATASTRUCTURE_DATASTRUCTUREFACTORY_HPP_

include/Algorithm/DataStructure/MicroCluster.hpp

@@ -32,23 +32,26 @@ class MicroCluster {
   int SST;//the sum of the squares of the time stamps  Til... Tin
   double weight; //number of data point in the clusters
   int dimension;
-
-  //TODO Need to subtract Base class of CF vector when all cf-vector based-algorithms have been implemented
+  double radius;//Used in DBStream
   //the parameters below is unique for DenStream
   clock_t createTime;
   clock_t lastUpdateTime;
   bool visited;
 
-  //TODO this may need to modify in the future (All algorithms used this, e.g.DenStream,CluStream,DenStream,DBStream,SWEM =.=)
+  //TODO 1. Need to subtract Base class of CF vector when all cf-vector based-algorithms have been implemented
+  // 2.this may need to modify in the future (All algorithms used this, e.g.DenStream,CluStream,DenStream,DBStream,SWEM =.=)
+
+
   MicroCluster(int dimension, int id);
+  MicroCluster(int dimension, int id,PointPtr dataPoint,double radius);//DBStream
+
   ~MicroCluster();
   void init(PointPtr datapoint, int timestamp);
-
-  void insert(PointPtr datapoint, int timestamp);
-  bool insert(PointPtr datapoint,double decayFactor,double epsilon);//Used in DenStream
-
+  void insert(PointPtr datapoint, int timestamp);//Used in CluStream
+  bool insert(PointPtr datapoint,double decayFactor,double epsilon);// DenStream
+  void insert(PointPtr datapoint, double decayFactor);//DBStream
   void merge(MicroClusterPtr other);
-  void substractClusterVector(MicroClusterPtr other);
+  void subtractClusterVector(MicroClusterPtr other);
   void updateId(MicroClusterPtr other);
 
   void resetID(int index);   //Used in DenStream
@@ -65,10 +68,25 @@ class MicroCluster {
   dataPoint getVarianceVector();
   double calCentroidDistance(PointPtr datapoint);
   bool judgeMerge(MicroClusterPtr other);
+  double getDistance(PointPtr datapoint);//DBStream
+  double getDistance(MicroClusterPtr other);//DBStream
+  void move();//DBStream
+  void decayWeight(double decayFactor);
   SESAME::MicroClusterPtr copy();
 
  private:
+  double distance;
   static double inverseError(double x);
 };
+typedef struct finderMicroCluster
+{
+  finderMicroCluster(int n) : id(n) { }
+  bool operator()(const MicroClusterPtr MC) const
+  {
+    return (id == MC->id.front());
+  }
+  int id;
+}finderMicroCluster;
+
 }
 #endif //SESAME_INCLUDE_ALGORITHM_DATASTRUCTURE_MICROCLUSTER_HPP_

include/Algorithm/DataStructure/WeightedAdjacencyList.hpp

@@ -0,0 +1,64 @@
+//
+// Created by 1124a on 2021/8/30.
+//
+
+#ifndef SESAME_INCLUDE_ALGORITHM_DATASTRUCTURE_WEIGHTEDADJACENCYLIST_HPP_
+#define SESAME_INCLUDE_ALGORITHM_DATASTRUCTURE_WEIGHTEDADJACENCYLIST_HPP_
+#include <algorithm>
+#include <functional>
+#include <unordered_map>
+#include<Algorithm/DataStructure/Point.hpp>
+#include<Algorithm/DataStructure/MicroCluster.hpp>
+#include<Algorithm/WindowModel/DampedWindow.hpp>
+
+namespace SESAME {
+struct MicroClusterPair;
+typedef std::shared_ptr<MicroClusterPair> MicroClusterPairPtr;
+struct MicroClusterPair{
+  MicroClusterPtr microCluster1;
+  MicroClusterPtr microCluster2;
+  MicroClusterPair( MicroClusterPtr microCluster1,MicroClusterPtr microCluster2){
+    this->microCluster1=microCluster1->copy();
+    this->microCluster2=microCluster2->copy();
+  }
+ //bool operator==(const MicroClusterPair &other) const;
+
+};
+
+struct KeyHasher{
+ std::size_t operator()(const MicroClusterPair &microClusterPair) const
+  {
+   return (std::hash<int>()(microClusterPair.microCluster1->id.front())) ^ (std::hash<int>()(microClusterPair.microCluster2->id.front()));
+  }
+};
+
+struct EqualKey
+    {
+  bool operator() (const MicroClusterPair &MCPair1, const MicroClusterPair &MCPair2) const
+  {
+     bool equal=false;
+    if( MCPair1.microCluster1->id.front()==MCPair2.microCluster1->id.front() &&MCPair1.microCluster2->id.front()==MCPair2.microCluster2->id.front() )
+      equal=true;
+    if(MCPair1.microCluster1->id.front() ==MCPair2.microCluster2->id.front() &&MCPair1.microCluster2->id.front()==MCPair2.microCluster1->id.front() )
+      equal=true;
+
+    return equal;
+  }
+    };
+
+
+class AdjustedWeight;
+typedef std::shared_ptr<AdjustedWeight> AdjustedWeightPtr;
+class AdjustedWeight{
+ public:
+  double weight;
+  clock_t updateTime;
+  AdjustedWeight(double weight, clock_t pointTime);
+  void add(clock_t startTime,double decayValue);
+  double getCurrentWeight(double decayFactor);
+};
+
+typedef std::unordered_map<MicroClusterPair,AdjustedWeightPtr,KeyHasher,EqualKey> WeightedAdjacencyList;
+typedef std::pair<MicroClusterPair, AdjustedWeightPtr>  DensityGraph;
+}
+#endif //SESAME_INCLUDE_ALGORITHM_DATASTRUCTURE_WEIGHTEDADJACENCYLIST_HPP_

include/Utils/BenchmarkUtils.hpp

@@ -44,6 +44,11 @@ struct param_t {
   double lambda;
   double mu;
   double beta;
+  //used in DBStream
+  double radius;
+  int cleanUpInterval;
+  double weightMin;
+  double alpha;
   std::string inputPath;
   std::string outputPath;
   SESAME::algoType algoType;

src/Algorithm/AlgorithmFactory.cpp

@@ -7,6 +7,7 @@
 #include <Algorithm/StreamKM.hpp>
 #include <Algorithm/CluStream.hpp>
 #include <Algorithm/DenStream.hpp>
+#include <Algorithm/DBStream.hpp>
 #include <Algorithm/Birch.hpp>
 #include <Algorithm/AlgorithmFactory.hpp>
 
@@ -27,5 +28,9 @@ SESAME::AlgorithmPtr SESAME::AlgorithmFactory::create(param_t &cmd_params) {
     shared_ptr<DenStream> denStream = std::make_shared<DenStream>(cmd_params);
     return (SESAME::AlgorithmPtr) denStream;
   }
+  if (cmd_params.algoType == DBStreamType) {
+    shared_ptr<DBStream> dbStream = std::make_shared<DBStream>(cmd_params);
+    return (SESAME::AlgorithmPtr) dbStream;
+  }
   throw std::invalid_argument("Unsupported");
 }

src/Algorithm/CMakeLists.txt

@@ -3,6 +3,7 @@ add_source_sesame(
         CluStream.cpp
         Birch.cpp
         DenStream.cpp
+        DBStream.cpp
         Algorithm.cpp
         AlgorithmFactory.cpp
 )

src/Algorithm/CluStream.cpp

@@ -9,8 +9,8 @@
 #include <iterator>
 /**
  * @Description: "offline" init micro clusters using KMeans
- * @param size: The size of initial data obects,
- *  initialData:input intial data
+ * @param size: The size of initial data objects,
+ *  initialData:input initial data
  *@Return: void
  */
 SESAME::CluStream::CluStream(param_t &cmd_params) {
@@ -98,7 +98,7 @@ double SESAME::CluStream::calRadius(MicroClusterPtr closestCluster) {
     radius = doubleMax;
     dataPoint centroid = closestCluster->getCentroid();
     for (int i = 0; i < this->CluStreamParam.clusterNumber; i++) {
-      // SESAME_INFO("i is for wegf"<<i);
+
       if (microClusters[i]->id == closestCluster->id) {
         continue;
       }
@@ -118,7 +118,7 @@ void SESAME::CluStream::insertIntoCluster(PointPtr data, MicroClusterPtr operate
   operateCluster->insert(data, timestamp);
 }
 
-//Delete oldest cluster and create new one case
+//Delete the oldest cluster and create new one case
 void SESAME::CluStream::deleteCreateCluster(PointPtr data) {
   // 3.1 Try to forget old micro clusters
 
@@ -172,7 +172,7 @@ void SESAME::CluStream::microClusterToPoint(std::vector<MicroClusterPtr> &microC
   for (int i = 0; i < this->CluStreamParam.clusterNumber; i++) {
     PointPtr
         point = DataStructureFactory::createPoint(i, microClusters[i]->weight, microClusters[i]->centroid.size(), 0);
-    for (int j = 0; j < microClusters[i]->centroid.size(); j++)
+    for (SESAME::dataPoint::size_type j = 0; j < microClusters[i]->centroid.size(); j++)
       point->setFeatureItem(microClusters[i]->centroid[j], j);
     //points;
     points.push_back(point);
@@ -200,7 +200,7 @@ void SESAME::CluStream::Initilize() {
   this->window = WindowFactory::createLandmarkWindow();
   this->window->pyramidalWindow.timeInterval = this->CluStreamParam.timeInterval;
   this->startTime = clock();
-  this->lastTime= this->startTime;
+  this->lastUpdateTime=this->startTime;
   window->initPyramidalWindow(this->window->pyramidalWindow.timeInterval);
 
 }
@@ -235,11 +235,11 @@ void SESAME::CluStream::runOnlineClustering(SESAME::PointPtr input) {
   } else {
     int interval;
     clock_t now = clock();
-    interval = (int) ((now - this->lastTime) / CLOCKS_PER_SEC);
-    if (interval >= 1)//
+    interval = (int) ((now - lastUpdateTime) / CLOCKS_PER_SEC);
+    if (interval >= 1)
     {
       window->pyramidalWindowProcess(startTime, microClusters);
-      lastTime = now;
+      lastUpdateTime = now;
     }
     incrementalCluster(input);
 
@@ -255,17 +255,17 @@ void SESAME::CluStream::runOfflineClustering(SESAME::DataSinkPtr sinkPtr) {
     landmarkTime = 0;
   SESAME_INFO("Start offline...");
   SESAME::SnapshotPtr landmarkSnapshot;
-  SESAME::SnapshotPtr substractMiroCluster;
+  SESAME::SnapshotPtr subtractMiroCluster;
   //If offlineTimeWindow ==0, Only Observe the end results of micro clusters
-  substractMiroCluster =
+  subtractMiroCluster =
       DataStructureFactory::createSnapshot(microClusters, (int) ((now - startTime) / CLOCKS_PER_SEC));
   SESAME_INFO("Now Miro Cluster is...");
   for (int i = 0; i < CluStreamParam.clusterNumber; i++) {
     std::stringstream result, re2;
-    std::copy(substractMiroCluster->microClusters[i]->id.begin(),
-              substractMiroCluster->microClusters[i]->id.end(),
+    std::copy(subtractMiroCluster->microClusters[i]->id.begin(),
+              subtractMiroCluster->microClusters[i]->id.end(),
               std::ostream_iterator<int>(re2, " "));
-    SESAME_INFO("The ID is " << re2.str() << "weight is " << substractMiroCluster->microClusters[i]->weight);
+    SESAME_INFO("The ID is " << re2.str() << "weight is " << subtractMiroCluster->microClusters[i]->weight);
   }
 
   //The offline is to observe a process of data stream clustering
@@ -282,21 +282,21 @@ void SESAME::CluStream::runOfflineClustering(SESAME::DataSinkPtr sinkPtr) {
       SESAME_INFO("The ID is " << re2.str() << "weight is " << landmarkSnapshot->microClusters[i]->weight);
     }
     if (landmarkSnapshot->elapsedTime == -1)
-      landmarkSnapshot = substractMiroCluster;
+      landmarkSnapshot = subtractMiroCluster;
 
-    substractMiroCluster = SESAME::Snapshot::substractSnapshot(substractMiroCluster, landmarkSnapshot,
+    subtractMiroCluster = SESAME::Snapshot::substractSnapshot(subtractMiroCluster, landmarkSnapshot,
                                                                this->CluStreamParam.clusterNumber);
   }
-  SESAME_INFO("substract Miro Cluster is...");
+  SESAME_INFO("subtract Miro Cluster is...");
   for (int i = 0; i < CluStreamParam.clusterNumber; i++) {
-    std::stringstream result, re2;
-    std::copy(substractMiroCluster->microClusters[i]->id.begin(),
-              substractMiroCluster->microClusters[i]->id.end(),
+    std::stringstream  re2;
+    std::copy(subtractMiroCluster->microClusters[i]->id.begin(),
+              subtractMiroCluster->microClusters[i]->id.end(),
               std::ostream_iterator<int>(re2, " "));
-    SESAME_INFO("The ID is " << re2.str() << "weight is " << substractMiroCluster->microClusters[i]->weight);
+    SESAME_INFO("The ID is " << re2.str() << "weight is " << subtractMiroCluster->microClusters[i]->weight);
   }
   vector <PointPtr> TransformedSnapshot;
-  microClusterToPoint(substractMiroCluster->microClusters, TransformedSnapshot);
+  microClusterToPoint(subtractMiroCluster->microClusters, TransformedSnapshot);
 
   SESAME_INFO("offline Cluster Number " << this->CluStreamParam.offlineClusterNumber << "Total number of p: "
                   << TransformedSnapshot.size());