Changeset 1681 for XIOS/dev/dev_trunk_omp/src/filter/temporal_filter.cpp

Timestamp:

07/19/19 15:28:33 (5 years ago)

Author:

yushan

Message:

MARK: Dynamic workflow graph developement. Branch up to date with trunk @1676. Bug fixed

File:

• XIOS/dev/dev_trunk_omp/src/filter/temporal_filter.cpp (modified) (6 diffs)

XIOS/dev/dev_trunk_omp/src/filter/temporal_filter.cpp

@@ -33 +33 @@
   }
 
-  CDataPacketPtr CTemporalFilter::apply(std::vector<CDataPacketPtr> data)
-  {
-    bool building_graph=this->tag ? data[0]->timestamp >= this->start_graph && data[0]->timestamp <= this->end_graph : false;
-
-    if(building_graph)
-    {
-      int edgeID = InvalidableObject::edgeIdGenerator++;
-
-      CWorkflowGraph::allocNodeEdge();
-
-      if(CWorkflowGraph::build_begin)
-      {
-        CWorkflowGraph::addEdge(edgeID, -1, data[0]);        
-      }
-      this->filterIDoutputs_pair.push_back(make_pair(edgeID, data[0]->src_filterID)); 
-    }
-    
-    
-    CDataPacketPtr packet;
-
-    if (data[0]->status != CDataPacket::END_OF_STREAM)
-    {
-      bool usePacket, outputResult, copyLess;
-      if (isOnceOperation)
-        usePacket = outputResult = copyLess = isFirstOperation;
-      else
-      {
-        usePacket = (data[0]->date >= nextSamplingDate);
-        outputResult = (data[0]->date  > initDate + nbOperationDates*opFreq - samplingFreq + offsetMonth + offsetAllButMonth);
-        copyLess = (isInstantOperation && usePacket && outputResult);
-      }
-
-      if (usePacket)
-      {
-        nbSamplingDates ++;
-        if (!copyLess)
-        {
-          if (!tmpData.numElements())
-            tmpData.resize(data[0]->data.numElements());
-
-          (*functor)(data[0]->data);
-        }
-
-        nextSamplingDate = ((initDate + offsetMonth) + nbSamplingDates * samplingFreq) + offsetAllButMonth + initDate.getRelCalendar().getTimeStep();
-      }
-
-      if (outputResult)
-      {
-        if(building_graph)
-        {
-          this->filterID = InvalidableObject::filterIdGenerator++;
-          CWorkflowGraph::addNode(this->filterID, "Temporal Filter\\n("+this->temp_op+")", 5, 1, 0, data[0]);   
-          (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].transform_type = this->temp_op;   
-          (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].inputs_complete = true ;
-
-          (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].attributes = this->field->recordXiosAttributes();
-          if(this->field->file) (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].attributes += "</br>file attributes : </br>" +this->field->file->recordXiosAttributes();
-
-
-          for(int i=0; i<this->filterIDoutputs_pair.size(); i++)
-          {
-            (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterIDoutputs_pair[i].second].filter_filled = 0 ;  
-            (*CWorkflowGraph::mapFieldToFilters_ptr_with_info)[this->filterIDoutputs_pair[i].first].to = this->filterID ;  
-          }
-
-          (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].expected_entry_nb = this->filterIDoutputs_pair.size() ;
-        }
+  
 
 
 
-        nbOperationDates ++;
-        if (!copyLess)
-        {
-          functor->final();
-
-          packet = CDataPacketPtr(new CDataPacket);
-          packet->date = data[0]->date;
-          packet->timestamp = data[0]->timestamp;
-          packet->status = data[0]->status;
-          packet->data.resize(tmpData.numElements());
-          packet->data = tmpData;
-        }
-        else
-          packet = data[0];
-
-        isFirstOperation = false;
-        
-        packet->field = this->field;
-        if(building_graph) packet->src_filterID=this->filterID;
-        if(building_graph) this->filterIDoutputs_pair.clear();
-        if(building_graph) CWorkflowGraph::build_begin=true;
-      }
-    }
-
-    return packet;
-  }
-
-
-
-  CDataPacketPtr CTemporalFilter::apply_old(std::vector<CDataPacketPtr> data)
+  bool CTemporalFilter::buildGraph(std::vector<CDataPacketPtr> data)
   {
     bool building_graph=this->tag ? data[0]->timestamp >= this->start_graph && data[0]->timestamp <= this->end_graph : false;
@@ -150 +55 @@
         (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].transform_type = this->temp_op;   
         (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].inputs_complete = false ;
+        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].clusterID = 1 ;
+        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].distance = (data[0]->distance);
+
+
+        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].attributes = this->field->record4graphXiosAttributes();
+        if(this->field->file) (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].attributes += "</br>file attributes : </br>" +this->field->file->record4graphXiosAttributes();
       }
 
@@ -159 +70 @@
         (*CWorkflowGraph::mapFilters_ptr_with_info)[data[0]->src_filterID].filter_filled = 0 ;
         (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].expected_entry_nb += 1 ;
+        (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].distance = max(data[0]->distance+1, (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].distance);
       }
 
@@ -164 +76 @@
       this->filterIDoutputs.push_back(data[0]->src_filterID); 
     }
-    
-    
+
+    return building_graph;
+  }
+
+
+  CDataPacketPtr CTemporalFilter::apply(std::vector<CDataPacketPtr> data)
+  {
+    bool BG = buildGraph(data);
+
     CDataPacketPtr packet;
 
@@ -207 +126 @@
           packet->data.resize(tmpData.numElements());
           packet->data = tmpData;
+          
         }
         else
@@ -214 +134 @@
         
         packet->field = this->field;
-        if(building_graph) packet->src_filterID=this->filterID;
-        if(building_graph) this->filterIDoutputs.clear();
-        if(building_graph) CWorkflowGraph::build_begin=true;
-        if(building_graph) (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].inputs_complete = true ;
+        
+        if(BG)
+        {
+          packet->src_filterID=this->filterID;
+          packet->distance = data[0]->distance+1;
+          this->filterIDoutputs.clear();
+          CWorkflowGraph::build_begin=true;
+          (*CWorkflowGraph::mapFilters_ptr_with_info)[this->filterID].inputs_complete = true ;
+        }
       }
     }