Changeset 1677 for XIOS/dev/dev_trunk_omp/src/filter

Timestamp:

06/25/19 16:14:54 (5 years ago)

Author:

yushan

Message:

MARK: Dynamic workflow graph developement. Branch up to date with trunk @1663.

Location:

XIOS/dev/dev_trunk_omp/src/filter

Files:

• binary_arithmetic_filter.cpp (modified) (5 diffs)
• binary_arithmetic_filter.hpp (modified) (3 diffs)
• data_packet.hpp (modified) (1 diff)
• file_server_writer_filter.cpp (modified) (2 diffs)
• file_server_writer_filter.hpp (modified) (1 diff)
• file_writer_filter.cpp (modified) (4 diffs)
• file_writer_filter.hpp (modified) (3 diffs)
• filter.cpp (modified) (2 diffs)
• filter.hpp (modified) (1 diff)
• garbage_collector.cpp (modified) (1 diff)
• garbage_collector.hpp (modified) (1 diff)
• input_pin.cpp (modified) (3 diffs)
• input_pin.hpp (modified) (2 diffs)
• output_pin.cpp (modified) (4 diffs)
• output_pin.hpp (modified) (4 diffs)
• pass_through_filter.cpp (modified) (2 diffs)
• source_filter.cpp (modified) (4 diffs)
• source_filter.hpp (modified) (3 diffs)
• spatial_transform_filter.cpp (modified) (6 diffs)
• spatial_transform_filter.hpp (modified) (2 diffs)
• store_filter.cpp (modified) (2 diffs)
• store_filter.hpp (modified) (3 diffs)
• temporal_filter.cpp (modified) (4 diffs)
• temporal_filter.hpp (modified) (2 diffs)
• ternary_arithmetic_filter.cpp (modified) (8 diffs)
• unary_arithmetic_filter.cpp (modified) (2 diffs)

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

@@ -1 +1 @@
 #include "binary_arithmetic_filter.hpp"
+#include "workflow_graph.hpp"
 
 namespace xios
@@ -7 +8 @@
     , op(operatorExpr.getOpScalarField(op))
     , value(value)
-  { /* Nothing to do */ };
+  { 
+    StdString input_op_expression=op;
+    if(input_op_expression == "add")
+      op_expression = "+";
+    else if(input_op_expression == "minus")
+      op_expression = "-";
+    else if(input_op_expression == "mult")
+      op_expression = "x";
+    else if(input_op_expression == "div")
+      op_expression = "/";
+    else if(input_op_expression == "eq")
+      op_expression = "=";
+    else if(input_op_expression == "lt")
+      op_expression = "<";
+    else if(input_op_expression == "gt")
+      op_expression = ">";
+    else if(input_op_expression == "le")
+      op_expression = "<=";
+    else if(input_op_expression == "ge")
+      op_expression = ">=";
+    else if(input_op_expression == "ne")
+      op_expression = "!=";
+    else
+      op_expression = " ";
+  };
 
   CDataPacketPtr CScalarFieldArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;    
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "S "+op_expression +" F Filter" ;
+      std::cout<<"CScalarFieldArithmeticFilter::apply filter tag = "<<this->tag<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
     CDataPacketPtr packet(new CDataPacket);
     packet->date = data[0]->date;
     packet->timestamp = data[0]->timestamp;
     packet->status = data[0]->status;
+    if(this->tag) packet->src_filterID = this->filterID;
+    packet->fieldID = this->output_field_id;
 
     if (packet->status == CDataPacket::NO_ERROR)
@@ -26 +68 @@
     , op(operatorExpr.getOpFieldScalar(op))
     , value(value)
-  { /* Nothing to do */ };
+  { 
+    StdString input_op_expression=op;
+    if(input_op_expression == "add")
+      op_expression = "+";
+    else if(input_op_expression == "minus")
+      op_expression = "-";
+    else if(input_op_expression == "mult")
+      op_expression = "x";
+    else if(input_op_expression == "div")
+      op_expression = "/";
+    else if(input_op_expression == "pow")
+      op_expression = "^";
+    else if(input_op_expression == "eq")
+      op_expression = "=";
+    else if(input_op_expression == "lt")
+      op_expression = "<";
+    else if(input_op_expression == "gt")
+      op_expression = ">";
+    else if(input_op_expression == "le")
+      op_expression = "<=";
+    else if(input_op_expression == "ge")
+      op_expression = ">=";
+    else if(input_op_expression == "ne")
+      op_expression = "!=";
+    else
+      op_expression = " ";
+  };
 
   CDataPacketPtr CFieldScalarArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+ 
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "F "+op_expression +" S Filter" ;
+      std::cout<<"CFieldScalarArithmeticFilter::apply filter tag = "<<this->tag<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
     CDataPacketPtr packet(new CDataPacket);
     packet->date = data[0]->date;
     packet->timestamp = data[0]->timestamp;
     packet->status = data[0]->status;
+    if(this->tag) packet->src_filterID = this->filterID;
+    packet->fieldID = this->output_field_id;
+    
 
     if (packet->status == CDataPacket::NO_ERROR)
@@ -44 +130 @@
     : CFilter(gc, 2, this)
     , op(operatorExpr.getOpFieldField(op))
-  { /* Nothing to do */ };
+  { 
+    StdString input_op_expression=op;
+    if(input_op_expression == "add")
+      op_expression = "+";
+    else if(input_op_expression == "minus")
+      op_expression = "-";
+    else if(input_op_expression == "mult")
+      op_expression = "x";
+    else if(input_op_expression == "div")
+      op_expression = "/";
+    else if(input_op_expression == "pow")
+      op_expression = "^";
+    else if(input_op_expression == "eq")
+      op_expression = "=";
+    else if(input_op_expression == "lt")
+      op_expression = "<";
+    else if(input_op_expression == "gt")
+      op_expression = ">";
+    else if(input_op_expression == "le")
+      op_expression = "<=";
+    else if(input_op_expression == "ge")
+      op_expression = ">=";
+    else if(input_op_expression == "ne")
+      op_expression = "!=";
+    else
+      op_expression = " ";
+  };
 
   CDataPacketPtr CFieldFieldArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
+    if(this->tag)
+    {
+
+      this->filterID = InvalidableObject::filterIdGenerator++;
+   
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "F "+op_expression +" F Filter" ;
+      std::cout<<"CFieldFieldArithmeticFilter::apply filter tag = "<<this->tag<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+
+      str = data[1]->fieldID +" ts=" + to_string(data[0]->timestamp);
+
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[1]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
+
     CDataPacketPtr packet(new CDataPacket);
     packet->date = data[0]->date;
     packet->timestamp = data[0]->timestamp;
+    if(this->tag) packet->src_filterID = this->filterID;
+    packet->fieldID = this->output_field_id;
+    
 
     if (data[0]->status != CDataPacket::NO_ERROR)
@@ -64 +201 @@
     return packet;
   }
+
+  StdString CScalarFieldArithmeticFilter::GetName(void)    { return StdString("CScalarFieldArithmeticFilter"); }
+  StdString CFieldScalarArithmeticFilter::GetName(void)    { return StdString("CFieldScalarArithmeticFilter"); }
+  StdString CFieldFieldArithmeticFilter::GetName(void)     { return StdString("CFieldFieldArithmeticFilter"); }
+
+
 } // namespace xios

XIOS/dev/dev_trunk_omp/src/filter/binary_arithmetic_filter.hpp

@@ -23 +23 @@
        */
       CScalarFieldArithmeticFilter(CGarbageCollector& gc, const std::string& op, double value);
+      StdString virtual GetName(void);
 
     protected:
@@ -51 +52 @@
        */
       CFieldScalarArithmeticFilter(CGarbageCollector& gc, const std::string& op, double value);
+      StdString virtual GetName(void);
 
     protected:
@@ -78 +80 @@
        */
       CFieldFieldArithmeticFilter(CGarbageCollector& gc, const std::string& op);
+      StdString virtual GetName(void);
 
     protected:

XIOS/dev/dev_trunk_omp/src/filter/data_packet.hpp

@@ -26 +26 @@
     Time timestamp;         //!< Timestamp of the data
     StatusCode status;      //!< Status of the packet
-
+    int src_filterID;
+    std::vector<int> filterIDoutputs;
+    StdString fieldID;
+    
     /*!
      * Creates a deep copy of the packet.

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

@@ -16 +16 @@
   void CFileServerWriterFilter::onInputReady(std::vector<CDataPacketPtr> data)
   {
+    if(CXios::isClient)
+    std::cout<<"CFileServerWriterFilter::onInputReady"<<std::endl;
     field->writeUpdateData(data[0]->data);
   }
@@ -29 +31 @@
   }
 
-  int CFileServerWriterFilter::getFilterId(void)
-  {
-    return filterId;
-  }
 
 } // namespace xios

XIOS/dev/dev_trunk_omp/src/filter/file_server_writer_filter.hpp

@@ -37 +37 @@
       bool virtual isDataExpected(const CDate& date) const;
 
-      /*!
-       * Returns filter's id needed in case of building workflow graph
-       */
-      int getFilterId();
 
     protected:

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

@@ -4 +4 @@
 #include "utils.hpp"
 #include "workflow_graph.hpp"
+#include "graphviz.hpp"
+
+using namespace ep_lib;
 
 namespace xios
@@ -14 +17 @@
       ERROR("CFileWriterFilter::CFileWriterFilter(CField* field)",
             "The field cannot be null.");
-    if (buildWorkflowGraph)
-    {
-      filterId = InvalidableObject::count;
-      InvalidableObject::count++;
-    }
   }
 
   void CFileWriterFilter::onInputReady(std::vector<CDataPacketPtr> data)
   {
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "File Writer Filter";
+
+      if(CXios::isClient) std::cout<<"CFileWriterFilter::apply filter tag = "<<this->tag<<std::endl;
+
+      if(CXios::isClient) 
+      {
+        StdString str = this->output_field_id +" ts=" + to_string(data[0]->timestamp);
+        // StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+        (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+        (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);  
+      }
+    }
+    
     const bool detectMissingValue = ( !field->default_value.isEmpty() &&
                                ( (!field->detect_missing_value.isEmpty() || field->detect_missing_value == true)
@@ -41 +59 @@
 
     field->sendUpdateData(dataArray);
+    
+    
+    
   }
 
@@ -53 +74 @@
   }
 
-  int CFileWriterFilter::getFilterId(void)
-  {
-    return filterId;
-  }
+
 
 } // namespace xios

XIOS/dev/dev_trunk_omp/src/filter/file_writer_filter.hpp

@@ -15 +15 @@
   {
     public:
+      int tag;
+      StdString output_field_id;
       /*!
        * Constructs the filter (with one input slot) associated to the specified field
@@ -43 +45 @@
       bool virtual isDataExpected(const CDate& date) const;
 
-      /*!
-       * Returns filter's id needed in case of building workflow graph
-       */
-      int getFilterId();
+
+      int filterID;
+
 
     protected:
@@ -59 +60 @@
       CField* field; //<! The associated field
       std::map<Time, CDataPacketPtr> packets; //<! The stored packets
-      int filterId;                           //<! Filter's id needed in case of building a workflow
 
   }; // class CFileWriterFilter

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

@@ -14 +14 @@
   void CFilter::onInputReady(std::vector<CDataPacketPtr> data)
   {
+    // std::cout<<"CFilter::onInputReady"<<std::endl;
     CDataPacketPtr outputPacket = engine->apply(data);
     if (outputPacket)
+    {
+      // std::cout<<"Filter onOutputReady"<<std::endl;
       onOutputReady(outputPacket);
+    } 
   }
 
@@ -52 +56 @@
     return COutputPin::isDataExpected(date);
   }
+
+  
 } // namespace xios

XIOS/dev/dev_trunk_omp/src/filter/filter.hpp

@@ -65 +65 @@
       bool virtual isDataExpected(const CDate& date) const;
 
+      
+      
+
+      int filterID;
+      StdString op_expression;
+
     protected:
       IFilterEngine* engine; //!< The filter engine, might be the filter itself

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

@@ -3 +3 @@
 namespace xios
 {
-
-  int InvalidableObject::count = 0;
-  #pragma omp threadprivate(InvalidableObject::count)
+  int InvalidableObject::filterIdGenerator = 0;
+  #pragma omp threadprivate(InvalidableObject::filterIdGenerator)
   
   void CGarbageCollector::registerObject(InvalidableObject* Object, Time timestamp)

XIOS/dev/dev_trunk_omp/src/filter/garbage_collector.hpp

@@ -21 +21 @@
     void virtual invalidate(Time timestamp) = 0;
   
-    static int count; //!< Counter used to identify a filter in case building workflow graph
-    #pragma omp threadprivate(count)
+    
+
+    static int filterIdGenerator;
+    #pragma omp threadprivate(filterIdGenerator)
+
+    
     
   }; // struct InvalidableObject

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

@@ -12 +12 @@
     , triggers(slotsCount)
     , hasTriggers(false)
-  {   }
+  {
+    // parent_filters = new std::list< std::shared_ptr<COutputPin> >;
+  }
 
   StdString CInputPin::GetName(void)
@@ -42 +44 @@
       gc.unregisterObject(this, packet->timestamp);
       onInputReady(it->second.packets);
+      //if(CXios::isClient) std::cout<<"setInput timestamp = "<<packet->timestamp<<std::endl;
       inputs.erase(it);
     }
@@ -84 +87 @@
   }
 
-  int CInputPin::getFilterId(void)
-  {
-    return -1;
-  }
+
+  
+
 
 } // namespace xios

XIOS/dev/dev_trunk_omp/src/filter/input_pin.hpp

@@ -18 +18 @@
   {
     public:
+
       /*!
        * Constructs an input pin with the specified number of slots
@@ -82 +83 @@
       void virtual invalidate(Time timestamp);
       
-      /*!
-       * Returns filter's id needed in case of building workflow graph
-       * This function should never be called from this class, instead functions defined in derived classes or in class COutputPin should be used
-       */
-      int virtual getFilterId();
 
     protected:

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

@@ -5 +5 @@
 namespace xios
 {
+
+
   COutputPin::COutputPin(CGarbageCollector& gc, bool manualTrigger /*= false*/, bool buildWorkflowGraph /* =false */)
     : gc(gc)
@@ -10 +12 @@
     , buildWorkflowGraph(buildWorkflowGraph)
   {
-    if (buildWorkflowGraph)
-    {
-      filterId = InvalidableObject::count;
-      InvalidableObject::count++;
-    }
+    // parent_filter = new std::vector< void >(0);
   }
 
@@ -40 +38 @@
             "The packet cannot be null.");
 
-    if (buildWorkflowGraph)
-    {
-        if(CWorkflowGraph::mapFilterTimestamps_ptr==0) CWorkflowGraph::mapFilterTimestamps_ptr = new std::unordered_map <int, vector<Time> >;
-      if(CWorkflowGraph::timestamps_ptr==0) CWorkflowGraph::timestamps_ptr = new set<Time>;
-      (*CWorkflowGraph::mapFilterTimestamps_ptr)[this->getFilterId()].push_back(packet->timestamp);
-        CWorkflowGraph::timestamps_ptr->insert(packet->timestamp);
-    }
+    //if(CXios::isClient) std::cout<<"onOutputReady"<<std::endl; 
+  
 
     if (manualTrigger) // Don't use canBeTriggered here, this function is virtual and can be overriden
@@ -123 +116 @@
   }
 
-  int COutputPin::getFilterId(void)
+  void COutputPin::setParentFiltersTag()
   {
-    return filterId;
+    for(int i=0; i<parent_filters.size(); i++)
+    {
+      parent_filters[i]->tag += tag;
+      parent_filters[i]->setParentFiltersTag();
+    }
   }
 
+
 } // namespace xios

XIOS/dev/dev_trunk_omp/src/filter/output_pin.hpp

@@ -7 +7 @@
 namespace xios
 {
+  class CInputPin;
+  class CFilter;
   /*!
    * An output pin handles the connections with downstream filters.
@@ -13 +15 @@
   {
     public:
+      StdString output_field_id;
+      int tag;
+
+      std::vector< std::shared_ptr<COutputPin> > parent_filters;
+
       /*!
        * Constructs an ouput pin with manual or automatic trigger
@@ -69 +76 @@
       void virtual invalidate(Time timestamp);
 
-      /*!
-       * Returns filter's id needed in case of building workflow graph
-       */
-      int getFilterId();
+      void virtual setParentFiltersTag();
+
 
     protected:
@@ -109 +114 @@
       bool buildWorkflowGraph;
 
-      //! Filter's id needed in case of building a workflow graph
-      int filterId;
-
 
   }; // class COutputPin

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

@@ -1 +1 @@
 #include "pass_through_filter.hpp"
+#include "workflow_graph.hpp"
 
 namespace xios
@@ -5 +6 @@
   CPassThroughFilter::CPassThroughFilter(CGarbageCollector& gc, bool buildWorkflowGraph /*= false*/)
     : CFilter(gc, 1, this, buildWorkflowGraph)
-  { /* Nothing to do */ }
+  { 
+  }
 
   CDataPacketPtr CPassThroughFilter::apply(std::vector<CDataPacketPtr> data)
   {
+        if(this->tag)
+    {
+      std::cout<<"CPassThroughFilter::apply tag = "<<this->tag<<std::endl;
+      this->filterID = InvalidableObject::filterIdGenerator++;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "Pass Through Filter";
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      
+      StdString str = data[0]->fieldID + " ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+      data[0]->src_filterID=this->filterID;
+
+    }
+    
+    data[0]->fieldID = this->output_field_id;
+    
+
     return data[0];
   }
+
 } // namespace xios

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

@@ -26 +26 @@
   }
 
+  
   template <int N>
-  void CSourceFilter::streamData(CDate date, const CArray<double, N>& data)
+  void CSourceFilter::streamData(CDate date, const CArray<double, N>& data, const StdString field_id)
   {
     date = date + offset; // this is a temporary solution, it should be part of a proper temporal filter
@@ -62 +63 @@
     }
 
+    packet->fieldID = field_id;
+    this->output_field_id = field_id;
+
+
+    if(this->tag)
+    {
+      this->filterID.first = InvalidableObject::filterIdGenerator++;  
+      packet->src_filterID=this->filterID.first;
+      
+      
+      if(CXios::isClient) std::cout<<"source filter tag = "<<this->tag<<std::endl;
+    
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID.first] = "Source Filter";
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+
+
+    }
+
+
+    onOutputReady(packet);
+  }
+
+  template void CSourceFilter::streamData<1>(CDate date, const CArray<double, 1>& data, const StdString field_id);
+  template void CSourceFilter::streamData<2>(CDate date, const CArray<double, 2>& data, const StdString field_id);
+  template void CSourceFilter::streamData<3>(CDate date, const CArray<double, 3>& data, const StdString field_id);
+  template void CSourceFilter::streamData<4>(CDate date, const CArray<double, 4>& data, const StdString field_id);
+  template void CSourceFilter::streamData<5>(CDate date, const CArray<double, 5>& data, const StdString field_id);
+  template void CSourceFilter::streamData<6>(CDate date, const CArray<double, 6>& data, const StdString field_id);
+  template void CSourceFilter::streamData<7>(CDate date, const CArray<double, 7>& data, const StdString field_id);
+
+
+  template <int N>
+  void CSourceFilter::streamData(CDate date, const CArray<double, N>& data)
+  {
+    date = date + offset; // this is a temporary solution, it should be part of a proper temporal filter
+
+    CDataPacketPtr packet(new CDataPacket);
+    packet->date = date;
+    packet->timestamp = date;
+    packet->status = CDataPacket::NO_ERROR;
+
+    packet->data.resize(grid->storeIndex_client.numElements());    
+    
+    if (compression)
+    {
+      packet->data = defaultValue;
+      grid->uncompressField(data, packet->data);    
+    }
+    else
+    {
+      if (mask)
+        grid->maskField(data, packet->data);
+      else
+        grid->inputField(data, packet->data);
+    }
+    // Convert missing values to NaN
+    if (hasMissingValue)
+    {
+      const double nanValue = std::numeric_limits<double>::quiet_NaN();
+      const size_t nbData = packet->data.numElements();
+      for (size_t idx = 0; idx < nbData; ++idx)
+      {
+        if (defaultValue == packet->data(idx))
+          packet->data(idx) = nanValue;
+      }
+    }
+    this->filterID.first = InvalidableObject::filterIdGenerator++;
+    packet->src_filterID=this->filterID.first;
+
+    if(CXios::isClient) std::cout<<"source filter filter tag = "<<this->tag<<std::endl;
+    
+    if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+    (*CWorkflowGraph::mapFilters_ptr)[this->filterID.first] = "Source Filter";
+
+
+
     onOutputReady(packet);
   }
@@ -73 +154 @@
   template void CSourceFilter::streamData<7>(CDate date, const CArray<double, 7>& data);
 
+
+
   void CSourceFilter::streamDataFromServer(CDate date, const std::map<int, CArray<double, 1> >& data)
   {
@@ -121 +204 @@
     onOutputReady(packet);
   }
+
+  
 } // namespace xios

XIOS/dev/dev_trunk_omp/src/filter/source_filter.hpp

@@ -50 +50 @@
       void streamData(CDate date, const CArray<double, N>& data);
 
+      template <int N>
+      void streamData(CDate date, const CArray<double, N>& data, const StdString field_id);
+
       /*!
        * Transforms the data received from the server into a packet and send it
@@ -66 +69 @@
        */
       void signalEndOfStream(CDate date);
+      std::pair<int, int> filterID;
+      std::vector<int> filterIDoutputs;
 
     private:
@@ -74 +79 @@
       const bool compression ; //!< indicates if data need to be compressed : on client side : true, on server side : false
       const bool mask ;        //!< indicates whether grid mask should be applied (true for clients, false for servers)
+
   }; // class CSourceFilter
 } // namespace xios

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

@@ -54 +54 @@
       {
         filter->connectOutput(firstFilter, 0);
-        if (buildWorkflowGraph)
-        {
-          if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
-          if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
-          int filterOut = (std::static_pointer_cast<COutputPin>(filter))->getFilterId();
-          int filterIn = (std::static_pointer_cast<COutputPin>(firstFilter))->getFilterId();
-          // PASS field's id here
-          (*CWorkflowGraph::mapFieldToFilters_ptr)["XXX"].push_back(filterOut);
-          (*CWorkflowGraph::mapFieldToFilters_ptr)["XXX"].push_back(filterIn);
-          (*CWorkflowGraph::mapFilters_ptr)[filterOut] = "Spatial transform filter";
-          (*CWorkflowGraph::mapFilters_ptr)[filterIn] = "Spatial transform filter";
-          std::cout<<"CSpatialTransformFilter::CSpatialTransformFilter CWorkflowGraph::mapFieldToFilters_ptr->size = "<<CWorkflowGraph::mapFieldToFilters_ptr->size()<<std::endl;
-        }
       }
 
@@ -86 +73 @@
   void CSpatialTransformFilter::onInputReady(std::vector<CDataPacketPtr> data)
   {
+    // if(CXios::isClient) std::cout<<"CSpatialTransformFilter onInputReady"<<std::endl;
+
     CSpatialTransformFilterEngine* spaceFilter = static_cast<CSpatialTransformFilterEngine*>(engine);
-    CDataPacketPtr outputPacket = spaceFilter->applyFilter(data, outputDefaultValue);
+    CDataPacketPtr outputPacket = spaceFilter->applyFilter(data, outputDefaultValue, this->tag, this->output_field_id);
     if (outputPacket)
+    {
+      // std::cout<<"Spatial Transform Filter onOutputReady"<<std::endl;
       onOutputReady(outputPacket);
+    } 
   }
 
@@ -120 +112 @@
   void CSpatialTemporalFilter::onInputReady(std::vector<CDataPacketPtr> data)
   {
+    if(CXios::isClient) std::cout<<"CSpatialTemporalFilter onInputReady"<<std::endl;
+
     CSpatialTransformFilterEngine* spaceFilter = static_cast<CSpatialTransformFilterEngine*>(engine);
-    CDataPacketPtr outputPacket = spaceFilter->applyFilter(data, outputDefaultValue);
+    CDataPacketPtr outputPacket = spaceFilter->applyFilter(data, outputDefaultValue, this->tag, this->output_field_id);
 
     if (outputPacket)
@@ -146 +140 @@
         packet->data.resize(tmpData.numElements());
         packet->data = tmpData;
+        std::cout<<"Spatial temporal filter onOutputReady"<<std::endl;
         onOutputReady(packet);
         tmpData.resize(0) ;
@@ -187 +182 @@
   }
 
-  CDataPacketPtr CSpatialTransformFilterEngine::applyFilter(std::vector<CDataPacketPtr> data, double defaultValue)
-  {
+  CDataPacketPtr CSpatialTransformFilterEngine::applyFilter(std::vector<CDataPacketPtr> data, double defaultValue, int tag, StdString fieldID)
+  {
+    if(tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+    
+      std::cout<<"CSpatialTransformFilter::apply filter tag = "<<tag<<std::endl;
+    
+
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "Spatial Transform Filter";
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+
+      StdString str = data[0]->fieldID + " ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
+
     CDataPacketPtr packet(new CDataPacket);
     packet->date = data[0]->date;
@@ -207 +221 @@
       apply(data[0]->data, packet->data);
     }
+
+    if(tag) packet->src_filterID=this->filterID;
+    packet->fieldID=fieldID;
 
     return packet;

XIOS/dev/dev_trunk_omp/src/filter/spatial_transform_filter.hpp

@@ -101 +101 @@
   {
     public:
+
+      int filterID;
+      int tag;
       /*!
        * Returns the engine wrapping the specified grid transformation.
@@ -117 +120 @@
        * \return the result of the grid transformation
        */
-      CDataPacketPtr applyFilter(std::vector<CDataPacketPtr> data, double defaultValue = 0);
+      CDataPacketPtr applyFilter(std::vector<CDataPacketPtr> data, double defaultValue = 0, int tag=0, StdString fieldID=0);
 
        /*!

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

@@ -90 +90 @@
   void CStoreFilter::onInputReady(std::vector<CDataPacketPtr> data)
   {
+    if(CXios::isClient) std::cout<<"CStoreFilter onInputReady"<<std::endl;
+
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "Store Filter";
+
+      if(CXios::isClient) std::cout<<"CStoreFilter::apply filter tag = "<<this->tag<<std::endl;
+
+      if(CXios::isClient) 
+      {
+        StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+        (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+        (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);  
+      }
+    }
+
+
 
     CDataPacketPtr packet;
@@ -139 +161 @@
   }
 
-  int CStoreFilter::getFilterId(void)
-  {
-    return filterId;
-  }
 
 } // namespace xios

XIOS/dev/dev_trunk_omp/src/filter/store_filter.hpp

@@ -3 +3 @@
 
 #include "input_pin.hpp"
+#include "workflow_graph.hpp"
+
 
 namespace xios
@@ -73 +75 @@
       void virtual invalidate(Time timestamp);
 
-      /*!
-       * Returns filter's id needed in case of building workflow graph
-       */
-      int getFilterId();
+      int filterID;
+      int tag;
+
 
     protected:
@@ -93 +94 @@
       const double missingValue; //!< The value to use to replace missing values
       std::map<Time, CDataPacketPtr> packets; //<! The stored packets
-      int filterId;                     //!< Filter's id needed in case of building a workflow
+
 
   }; // class CStoreFilter

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

@@ -2 +2 @@
 #include "functor_type.hpp"
 #include "calendar_util.hpp"
+#include "workflow_graph.hpp"
 
 namespace xios
@@ -32 +33 @@
   CDataPacketPtr CTemporalFilter::apply(std::vector<CDataPacketPtr> data)
   {
+   
+    if(this->tag)
+    {
+      if(this->filterIDoutputs.size()==0) this->filterID = InvalidableObject::filterIdGenerator++;
+      
+
+      std::cout<<"CTemporalFilter::apply filter tag = "<<this->tag<<std::endl;
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+       
+    
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      if(this->filterIDoutputs.size()==0) (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "Temporal Filter";  
+
+      this->filterIDoutputs.push_back(data[0]->src_filterID); 
+    }
+    
+    
     CDataPacketPtr packet;
 
@@ -42 +66 @@
       {
         usePacket = (data[0]->date >= nextSamplingDate);
-//        outputResult = (data[0]->date + samplingFreq > nextOperationDate);
         outputResult = (data[0]->date  > initDate + nbOperationDates*opFreq - samplingFreq + offsetMonth + offsetAllButMonth);
         copyLess = (isInstantOperation && usePacket && outputResult);
@@ -79 +102 @@
 
         isFirstOperation = false;
-//        nextOperationDate = initDate + samplingFreq + nbOperationDates*opFreq - samplingFreq + offsetMonth + offsetAllButMonth;
+        if(this->tag) packet->src_filterID=this->filterID;
+        packet->fieldID = this->output_field_id;
+        if(this->tag) this->filterIDoutputs.clear();
       }
     }

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

@@ -54 +54 @@
        */
       bool virtual isDataExpected(const CDate& date) const;
+      std::vector<int> filterIDoutputs;
 
     private:
@@ -72 +73 @@
 //      CDate nextOperationDate; //!< The date of the next operation
       bool isFirstOperation; //!< True before the first operation was been computed
+
   }; // class CTemporalFilter
 } // namespace xios

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

@@ -1 +1 @@
 #include "ternary_arithmetic_filter.hpp"
+#include "workflow_graph.hpp"
 
 namespace xios
@@ -12 +13 @@
   CDataPacketPtr CScalarScalarFieldArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
-    CDataPacketPtr packet(new CDataPacket);
-    packet->date = data[0]->date;
-    packet->timestamp = data[0]->timestamp;
-    packet->status = data[0]->status;
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+    
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "SSF Filter" ;
+      std::cout<<"CScalarScalarFieldArithmeticFilter::apply connection = "<<data[0]->src_filterID<<" <-> "<<this->filterID<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
+    CDataPacketPtr packet(new CDataPacket);
+    packet->date = data[0]->date;
+    packet->timestamp = data[0]->timestamp;
+    packet->status = data[0]->status;
+    if(this->tag) packet->src_filterID = this->filterID;
+    packet->fieldID = this->output_field_id;
 
     if (packet->status == CDataPacket::NO_ERROR)
@@ -32 +51 @@
   CDataPacketPtr CScalarFieldScalarArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
-    CDataPacketPtr packet(new CDataPacket);
-    packet->date = data[0]->date;
-    packet->timestamp = data[0]->timestamp;
-    packet->status = data[0]->status;
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+    
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "SFS Filter" ;
+      std::cout<<"CScalarFieldScalarArithmeticFilter::apply connection = "<<data[0]->src_filterID<<" <-> "<<this->filterID<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
+    CDataPacketPtr packet(new CDataPacket);
+    packet->date = data[0]->date;
+    packet->timestamp = data[0]->timestamp;
+    packet->status = data[0]->status;
+    if(this->tag) packet->src_filterID = this->filterID;
+    packet->fieldID = this->output_field_id;
 
     if (packet->status == CDataPacket::NO_ERROR)
@@ -51 +88 @@
   CDataPacketPtr CScalarFieldFieldArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
-    CDataPacketPtr packet(new CDataPacket);
-    packet->date = data[0]->date;
-    packet->timestamp = data[0]->timestamp;
-    packet->status = data[0]->status;
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+    
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "SFF Filter" ;
+      std::cout<<"CScalarFieldFieldArithmeticFilter::apply connection = "<<data[0]->src_filterID<<" <-> "<<this->filterID<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+
+      str = data[1]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[1]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
+    CDataPacketPtr packet(new CDataPacket);
+    packet->date = data[0]->date;
+    packet->timestamp = data[0]->timestamp;
+    packet->status = data[0]->status;
+    if(this->tag) packet->src_filterID = this->filterID;
+    packet->fieldID = this->output_field_id;
 
     if (data[0]->status != CDataPacket::NO_ERROR)
@@ -79 +138 @@
   CDataPacketPtr CFieldScalarScalarArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
-    CDataPacketPtr packet(new CDataPacket);
-    packet->date = data[0]->date;
-    packet->timestamp = data[0]->timestamp;
-    packet->status = data[0]->status;
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+    
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "FSS Filter" ;
+      std::cout<<"CFieldScalarScalarArithmeticFilter::apply connection = "<<data[0]->src_filterID<<" <-> "<<this->filterID<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
+    CDataPacketPtr packet(new CDataPacket);
+    packet->date = data[0]->date;
+    packet->timestamp = data[0]->timestamp;
+    packet->status = data[0]->status;
+    if(this->tag) packet->src_filterID = this->filterID;
+    packet->fieldID = this->output_field_id;
 
     if (packet->status == CDataPacket::NO_ERROR)
@@ -99 +176 @@
   CDataPacketPtr CFieldScalarFieldArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
-    CDataPacketPtr packet(new CDataPacket);
-    packet->date = data[0]->date;
-    packet->timestamp = data[0]->timestamp;
-    packet->status = data[0]->status;
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+    
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "FSF Filter" ;
+      std::cout<<"CFieldScalarFieldArithmeticFilter::apply connection = "<<data[0]->src_filterID<<" <-> "<<this->filterID<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+
+      str = data[1]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[1]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
+    CDataPacketPtr packet(new CDataPacket);
+    packet->date = data[0]->date;
+    packet->timestamp = data[0]->timestamp;
+    packet->status = data[0]->status;
+    if(this->tag) packet->src_filterID = this->filterID;
+    packet->fieldID = this->output_field_id;
 
     if (data[0]->status != CDataPacket::NO_ERROR)
@@ -124 +223 @@
   CDataPacketPtr CFieldFieldScalarArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
-    CDataPacketPtr packet(new CDataPacket);
-    packet->date = data[0]->date;
-    packet->timestamp = data[0]->timestamp;
-    packet->status = data[0]->status;
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+    
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "FFS Filter" ;
+      std::cout<<"CFieldFieldScalarArithmeticFilter::apply connection = "<<data[0]->src_filterID<<" <-> "<<this->filterID<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+
+      str = data[1]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[1]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
+    CDataPacketPtr packet(new CDataPacket);
+    packet->date = data[0]->date;
+    packet->timestamp = data[0]->timestamp;
+    packet->status = data[0]->status;
+    if(this->tag) packet->src_filterID = this->filterID;
+    packet->fieldID = this->output_field_id;
 
     if (data[0]->status != CDataPacket::NO_ERROR)
@@ -149 +270 @@
   CDataPacketPtr CFieldFieldFieldArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
-    CDataPacketPtr packet(new CDataPacket);
-    packet->date = data[0]->date;
-    packet->timestamp = data[0]->timestamp;
-    packet->status = data[0]->status;
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+    
+
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = "FFF Filter" ;
+      std::cout<<"CFieldFieldFieldArithmeticFilter::apply connection = "<<data[0]->src_filterID<<" <-> "<<this->filterID<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+
+      str = data[1]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[1]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+
+      str = data[2]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[2]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);
+    }
+
+
+    CDataPacketPtr packet(new CDataPacket);
+    packet->date = data[0]->date;
+    packet->timestamp = data[0]->timestamp;
+    packet->status = data[0]->status;
+    if(this->tag) packet->src_filterID = this->filterID;
+    packet->fieldID = this->output_field_id;
 
     if (data[0]->status != CDataPacket::NO_ERROR)

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

@@ -1 +1 @@
 #include "unary_arithmetic_filter.hpp"
-
+#include "workflow_graph.hpp"
 namespace xios
 {
@@ -6 +6 @@
     : CFilter(gc, 1, this)
     , op(operatorExpr.getOpField(op))
-  { /* Nothing to do */ };
+  { 
+    StdString input_op_expression=op;
+    if(input_op_expression == "neg")
+      op_expression = "-";
+    else if(input_op_expression == "sin")
+      op_expression = "sin";
+    else if(input_op_expression == "cos")
+      op_expression = "cos";
+    else if(input_op_expression == "tan")
+      op_expression = "tan";
+    else if(input_op_expression == "exp")
+      op_expression = "exp";
+    else if(input_op_expression == "log")
+      op_expression = "log";
+    else if(input_op_expression == "log10")
+      op_expression = "log10";
+    else if(input_op_expression == "sqrt")
+      op_expression = "sqrt";
+    else
+      op_expression = " ";
+  };
 
   CDataPacketPtr CUnaryArithmeticFilter::apply(std::vector<CDataPacketPtr> data)
   {
+    if(this->tag)
+    {
+      this->filterID = InvalidableObject::filterIdGenerator++;
+      
+      if(CWorkflowGraph::mapFieldToFilters_ptr==0) CWorkflowGraph::mapFieldToFilters_ptr = new std::unordered_map <StdString, vector <int> >;
+      if(CWorkflowGraph::mapFilters_ptr==0) CWorkflowGraph::mapFilters_ptr = new std::unordered_map <int, StdString>;
+
+
+      (*CWorkflowGraph::mapFilters_ptr)[this->filterID] = op_expression +" F Filter" ;
+      std::cout<<"CunaryArithmeticFilter::apply filter tag = "<<this->tag<<std::endl;
+
+      StdString str = data[0]->fieldID +" ts=" + to_string(data[0]->timestamp);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(data[0]->src_filterID);
+      (*CWorkflowGraph::mapFieldToFilters_ptr)[str].push_back(this->filterID);   
+  
+    }
+
     CDataPacketPtr packet(new CDataPacket);
     packet->date = data[0]->date;
     packet->timestamp = data[0]->timestamp;
     packet->status = data[0]->status;
+    packet->fieldID = this->output_field_id;
 
     if (packet->status == CDataPacket::NO_ERROR)
       packet->data.reference(op(data[0]->data));
 
+    if(this->tag) packet->src_filterID = this->filterID;
+
     return packet;
   }