Changeset 2193 for XIOS/dev/dev_ym/XIOS_COUPLING/src/filter

Timestamp:

07/22/21 14:05:58 (3 years ago)

Author:

yushan

Message:

workflow graph : enable unary and binary arithmetic filters

Location:

XIOS/dev/dev_ym/XIOS_COUPLING/src/filter

Files:

• binary_arithmetic_filter.cpp (modified) (7 diffs)
• binary_arithmetic_filter.hpp (modified) (3 diffs)
• temporal_filter.cpp (modified) (3 diffs)
• temporal_filter.hpp (modified) (2 diffs)
• unary_arithmetic_filter.cpp (modified) (3 diffs)
• unary_arithmetic_filter.hpp (modified) (1 diff)

XIOS/dev/dev_ym/XIOS_COUPLING/src/filter/binary_arithmetic_filter.cpp

@@ -1 +1 @@
 #include "binary_arithmetic_filter.hpp"
+#include "workflow_graph.hpp"
 
 namespace xios
@@ -9 +10 @@
   { /* Nothing to do */ };
 
+  std::pair<int, int> CScalarFieldArithmeticFilter::buildGraph(std::vector<CDataPacketPtr> data)
+  {
+    bool building_graph = this->graphEnabled;
+    int unique_filter_id;
+    bool firstround = true;
+    
+    if(building_graph)
+    {
+      if(!data[0]->graphPackage)
+      {
+        data[0]->graphPackage = new CGraphDataPackage;
+        data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
+        data[0]->graphPackage->fromFilter = -1;
+      }
+
+      if(!CWorkflowGraph::mapHashFilterID_) CWorkflowGraph::mapHashFilterID_ = new std::unordered_map <size_t, int>;
+
+      size_t filterhash = std::hash<StdString>{}(this->graphPackage->inFields[0]->content+to_string(data[0]->timestamp)+this->graphPackage->inFields[0]->getId());
+
+      // first round
+      if(CWorkflowGraph::mapHashFilterID_->find(filterhash) == CWorkflowGraph::mapHashFilterID_->end())
+      {
+        this->graphPackage->filterId = CWorkflowGraph::getNodeSize();
+        unique_filter_id = this->graphPackage->filterId;
+        CWorkflowGraph::addNode("Arithmetic filter\\n ("+this->graphPackage->inFields[0]->content+")", 4, false, 0, data[0]);        
+
+        CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, this->graphPackage->filterId, data[0]);
+        data[0]->graphPackage->fromFilter = this->graphPackage->filterId;
+        data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
+        std::rotate(this->graphPackage->inFields.begin(), this->graphPackage->inFields.begin() + 1, this->graphPackage->inFields.end());
+
+       
+        (*CWorkflowGraph::mapHashFilterID_)[filterhash] = unique_filter_id; 
+        
+      }
+      // not first round
+      else 
+      {
+        unique_filter_id = (*CWorkflowGraph::mapHashFilterID_)[filterhash];
+        if(data[0]->graphPackage->fromFilter != unique_filter_id)
+        {
+          CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, unique_filter_id, data[0]);  
+        }
+      }  
+    }
+
+    return std::make_pair(building_graph, unique_filter_id);
+  }
+
   CDataPacketPtr CScalarFieldArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
@@ -15 +65 @@
     packet->timestamp = data[0]->timestamp;
     packet->status = data[0]->status;
+    
+    std::pair<int, int> graph = buildGraph(data);
+
+    if(std::get<0>(graph))
+    {  
+      packet->graphPackage = new CGraphDataPackage;
+      packet->graphPackage->fromFilter = std::get<1>(graph);
+      packet->graphPackage->currentField = this->graphPackage->inFields[0]; 
+    }
 
     if (packet->status == CDataPacket::NO_ERROR)
@@ -28 +87 @@
   { /* Nothing to do */ };
 
+  std::pair<int, int> CFieldScalarArithmeticFilter::buildGraph(std::vector<CDataPacketPtr> data)
+  {
+    bool building_graph = this->graphEnabled;
+    int unique_filter_id;
+    bool firstround = true;
+
+    if(building_graph)
+    {
+      if(!data[0]->graphPackage)
+      {
+        data[0]->graphPackage = new CGraphDataPackage;
+        data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
+        data[0]->graphPackage->fromFilter = -1;
+      }
+
+      if(!CWorkflowGraph::mapHashFilterID_) CWorkflowGraph::mapHashFilterID_ = new std::unordered_map <size_t, int>;
+
+      size_t filterhash = std::hash<StdString>{}(this->graphPackage->inFields[0]->content+to_string(data[0]->timestamp)+this->graphPackage->inFields[0]->getId());
+
+      // first round
+      if(CWorkflowGraph::mapHashFilterID_->find(filterhash) == CWorkflowGraph::mapHashFilterID_->end())
+      {
+        this->graphPackage->filterId = CWorkflowGraph::getNodeSize();
+        unique_filter_id = this->graphPackage->filterId;
+        CWorkflowGraph::addNode("Arithmetic filter\\n ("+this->graphPackage->inFields[0]->content+")", 4, false, 0, data[0]);        
+
+        CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, this->graphPackage->filterId, data[0]);
+        data[0]->graphPackage->fromFilter = this->graphPackage->filterId;
+        data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
+        std::rotate(this->graphPackage->inFields.begin(), this->graphPackage->inFields.begin() + 1, this->graphPackage->inFields.end());
+
+       
+        (*CWorkflowGraph::mapHashFilterID_)[filterhash] = unique_filter_id; 
+        
+      }
+      // not first round
+      else 
+      {
+        unique_filter_id = (*CWorkflowGraph::mapHashFilterID_)[filterhash];
+        if(data[0]->graphPackage->fromFilter != unique_filter_id)
+        {
+          CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, unique_filter_id, data[0]);  
+        }
+      }  
+    }
+
+    return std::make_pair(building_graph, unique_filter_id);
+  }
+
   CDataPacketPtr CFieldScalarArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
@@ -35 +143 @@
     packet->status = data[0]->status;
 
+    std::pair<int, int> graph = buildGraph(data);
+
+    if(std::get<0>(graph))
+    {  
+      packet->graphPackage = new CGraphDataPackage;
+      packet->graphPackage->fromFilter = std::get<1>(graph);
+      packet->graphPackage->currentField = this->graphPackage->inFields[0]; 
+    }
+
     if (packet->status == CDataPacket::NO_ERROR)
       packet->data.reference(op(data[0]->data, value));
@@ -46 +163 @@
   { /* Nothing to do */ };
 
+  std::pair<int, int> CFieldFieldArithmeticFilter::buildGraph(std::vector<CDataPacketPtr> data)
+  {
+    bool building_graph = this->graphEnabled;
+    int unique_filter_id;
+    bool firstround = true;
+
+    if(building_graph)
+    {
+      if(!data[0]->graphPackage)
+      {
+        data[0]->graphPackage = new CGraphDataPackage;
+        data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
+        data[0]->graphPackage->fromFilter = -1;
+      }
+      if(!data[1]->graphPackage)
+      {
+        data[1]->graphPackage = new CGraphDataPackage;
+        data[1]->graphPackage->currentField = this->graphPackage->inFields[0];
+        data[1]->graphPackage->fromFilter = -1;
+      }
+
+      if(!CWorkflowGraph::mapHashFilterID_) CWorkflowGraph::mapHashFilterID_ = new std::unordered_map <size_t, int>;
+
+      size_t filterhash = std::hash<StdString>{}(this->graphPackage->inFields[0]->content+to_string(data[0]->timestamp)+this->graphPackage->inFields[0]->getId());
+
+      // first round
+      if(CWorkflowGraph::mapHashFilterID_->find(filterhash) == CWorkflowGraph::mapHashFilterID_->end())
+      {
+        this->graphPackage->filterId = CWorkflowGraph::getNodeSize();
+        unique_filter_id = this->graphPackage->filterId;
+        CWorkflowGraph::addNode("Arithmetic filter\\n ("+this->graphPackage->inFields[0]->content+")", 4, false, 0, data[0]);        
+
+        CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, this->graphPackage->filterId, data[0]);
+        data[0]->graphPackage->fromFilter = this->graphPackage->filterId;
+        data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
+
+        CWorkflowGraph::addEdge(data[1]->graphPackage->fromFilter, this->graphPackage->filterId, data[1]);
+        data[1]->graphPackage->fromFilter = this->graphPackage->filterId;
+        data[1]->graphPackage->currentField = this->graphPackage->inFields[0];
+
+        std::rotate(this->graphPackage->inFields.begin(), this->graphPackage->inFields.begin() + 1, this->graphPackage->inFields.end());
+
+        (*CWorkflowGraph::mapHashFilterID_)[filterhash] = unique_filter_id; 
+        
+      }
+      // not first round
+      else 
+      {
+        unique_filter_id = (*CWorkflowGraph::mapHashFilterID_)[filterhash];
+        if(data[0]->graphPackage->fromFilter != unique_filter_id)
+        {
+          CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, unique_filter_id, data[0]);  
+        }
+        if(data[1]->graphPackage->fromFilter != unique_filter_id)
+        {
+          CWorkflowGraph::addEdge(data[1]->graphPackage->fromFilter, unique_filter_id, data[1]);  
+        }
+      }  
+    }
+
+    return std::make_pair(building_graph, unique_filter_id);
+  }
+
+
   CDataPacketPtr CFieldFieldArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
@@ -51 +232 @@
     packet->date = data[0]->date;
     packet->timestamp = data[0]->timestamp;
+
+    std::pair<int, int> graph = buildGraph(data);
+
+    if(std::get<0>(graph))
+    {  
+      packet->graphPackage = new CGraphDataPackage;
+      packet->graphPackage->fromFilter = std::get<1>(graph);
+      packet->graphPackage->currentField = this->graphPackage->inFields[0]; 
+    }
 
     if (data[0]->status != CDataPacket::NO_ERROR)

XIOS/dev/dev_ym/XIOS_COUPLING/src/filter/binary_arithmetic_filter.hpp

@@ -35 +35 @@
        */
       CDataPacketPtr virtual apply(std::vector<CDataPacketPtr> data);
+      std::pair<int, int> virtual buildGraph(std::vector<CDataPacketPtr> data);
   }; // class CScalarFieldArithmeticFilter
 
@@ -63 +64 @@
        */
       CDataPacketPtr virtual apply(std::vector<CDataPacketPtr> data);
+      std::pair<int, int> virtual buildGraph(std::vector<CDataPacketPtr> data);
   }; // class CFieldScalarArithmeticFilter
 
@@ -89 +91 @@
        */
       CDataPacketPtr virtual apply(std::vector<CDataPacketPtr> data);
+      std::pair<int, int> virtual buildGraph(std::vector<CDataPacketPtr> data);
+      
   }; // class CFieldFieldArithmeticFilter
 } // namespace xios

XIOS/dev/dev_ym/XIOS_COUPLING/src/filter/temporal_filter.cpp

@@ -30 +30 @@
     , isFirstOperation(true)
     , graphCycleCompleted(true)
+    , temporalOperation(opId)
   {
+  }
+
+  std::string CTemporalFilter::getTemporalOperation()
+  {
+    return this->temporalOperation;
   }
 
@@ -45 +51 @@
       {  
         this->graphPackage->filterId = CWorkflowGraph::getNodeSize();
-        CWorkflowGraph::addNode("Temporal filter", 3, false, 0, data[0]);
+        CWorkflowGraph::addNode("Temporal filter \\n("+getTemporalOperation()+")", 3, false, 0, data[0]);
         graphCycleCompleted = false;
       }
@@ -54 +60 @@
       CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, this->graphPackage->filterId, data[0]);
       data[0]->graphPackage->fromFilter = this->graphPackage->filterId;
-      this->graphPackage->sourceFilterIds.push_back(data[0]->graphPackage->fromFilter); 
+      // this->graphPackage->sourceFilterIds.push_back(data[0]->graphPackage->fromFilter); 
       data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
       std::rotate(this->graphPackage->inFields.begin(), this->graphPackage->inFields.begin() + 1, this->graphPackage->inFields.end());

XIOS/dev/dev_ym/XIOS_COUPLING/src/filter/temporal_filter.hpp

@@ -55 +55 @@
       bool virtual isDataExpected(const CDate& date) const;
       bool graphCycleCompleted;
+      std::string getTemporalOperation();
 
     private:
@@ -73 +74 @@
 //      CDate nextOperationDate; //!< The date of the next operation
       bool isFirstOperation; //!< True before the first operation was been computed
+      const std::string temporalOperation;
   }; // class CTemporalFilter
 } // namespace xios

XIOS/dev/dev_ym/XIOS_COUPLING/src/filter/unary_arithmetic_filter.cpp

@@ -1 +1 @@
 #include "unary_arithmetic_filter.hpp"
+#include "workflow_graph.hpp"
+
 
 namespace xios
@@ -8 +10 @@
   { /* Nothing to do */ };
 
+  std::pair<int, int> CUnaryArithmeticFilter::buildGraph(std::vector<CDataPacketPtr> data)
+  {
+    bool building_graph = this->graphEnabled;
+    int unique_filter_id;
+    bool firstround = true;
+    
+    if(building_graph)
+    {
+      if(!data[0]->graphPackage)
+      {
+        data[0]->graphPackage = new CGraphDataPackage;
+        data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
+        data[0]->graphPackage->fromFilter = -1;
+      }
+
+      if(!CWorkflowGraph::mapHashFilterID_) CWorkflowGraph::mapHashFilterID_ = new std::unordered_map <size_t, int>;
+
+      size_t filterhash = std::hash<StdString>{}(this->graphPackage->inFields[0]->content+to_string(data[0]->timestamp)+this->graphPackage->inFields[0]->getId());
+
+      // first round
+      if(CWorkflowGraph::mapHashFilterID_->find(filterhash) == CWorkflowGraph::mapHashFilterID_->end())
+      {
+        this->graphPackage->filterId = CWorkflowGraph::getNodeSize();
+        unique_filter_id = this->graphPackage->filterId;
+        CWorkflowGraph::addNode("Arithmetic filter\\n ("+this->graphPackage->inFields[0]->content+")", 4, false, 0, data[0]);        
+
+        CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, this->graphPackage->filterId, data[0]);
+        data[0]->graphPackage->fromFilter = this->graphPackage->filterId;
+        data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
+        std::rotate(this->graphPackage->inFields.begin(), this->graphPackage->inFields.begin() + 1, this->graphPackage->inFields.end());
+
+       
+        (*CWorkflowGraph::mapHashFilterID_)[filterhash] = unique_filter_id; 
+        
+      }
+      // not first round
+      else 
+      {
+        unique_filter_id = (*CWorkflowGraph::mapHashFilterID_)[filterhash];
+        if(data[0]->graphPackage->fromFilter != unique_filter_id)
+        {
+          CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, unique_filter_id, data[0]);  
+        }
+      }  
+    }
+
+    return std::make_pair(building_graph, unique_filter_id);
+  }
+
+
   CDataPacketPtr CUnaryArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
@@ -15 +67 @@
     packet->status = data[0]->status;
 
+    std::pair<int, int> graph = buildGraph(data);
+
+    if(std::get<0>(graph))
+    {  
+      packet->graphPackage = new CGraphDataPackage;
+      packet->graphPackage->fromFilter = std::get<1>(graph);
+      packet->graphPackage->currentField = this->graphPackage->inFields[0]; 
+    }
+
     if (packet->status == CDataPacket::NO_ERROR)
       packet->data.reference(op(data[0]->data));

XIOS/dev/dev_ym/XIOS_COUPLING/src/filter/unary_arithmetic_filter.hpp

@@ -34 +34 @@
        */
       CDataPacketPtr virtual apply(std::vector<CDataPacketPtr> data);
+      std::pair<int, int> virtual buildGraph(std::vector<CDataPacketPtr> data);
   }; // class CUnaryArithmeticFilter
 } // namespace xios