Changeset 1460 for XIOS/dev/branch_openmp/src/filter

Timestamp:

03/22/18 10:43:20 (6 years ago)

Author:

yushan

Message:

branch_openmp merged with XIOS_DEV_CMIP6@1459

Location:

XIOS/dev/branch_openmp/src/filter

Files:

• file_server_writer_filter.cpp (added)
• file_server_writer_filter.hpp (added)
• file_writer_filter.cpp (modified) (1 diff)
• file_writer_filter.hpp (modified) (1 diff)
• filter.cpp (modified) (1 diff)
• filter.hpp (modified) (1 diff)
• input_pin.hpp (modified) (1 diff)
• output_pin.cpp (modified) (1 diff)
• output_pin.hpp (modified) (1 diff)
• source_filter.cpp (modified) (5 diffs)
• source_filter.hpp (modified) (2 diffs)
• spatial_transform_filter.cpp (modified) (5 diffs)
• spatial_transform_filter.hpp (modified) (2 diffs)
• store_filter.cpp (modified) (1 diff)
• store_filter.hpp (modified) (1 diff)
• temporal_filter.cpp (modified) (6 diffs)
• temporal_filter.hpp (modified) (3 diffs)

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

@@ -37 +37 @@
   }
 
+  bool CFileWriterFilter::mustAutoTrigger() const
+  {
+    return true;
+  }
+
   bool CFileWriterFilter::isDataExpected(const CDate& date) const
   {

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

@@ -24 +24 @@
 
       /*!
+       * Tests if the filter must auto-trigger.
+       *
+       * \return true if the filter must auto-trigger
+       */
+      bool virtual mustAutoTrigger() const;
+
+      /*!
        * Tests whether data is expected for the specified date.
        *

XIOS/dev/branch_openmp/src/filter/filter.cpp

@@ -41 +41 @@
   }
 
+  bool CFilter::mustAutoTrigger() const
+  {
+    return COutputPin::mustAutoTrigger();
+  }
+
   bool CFilter::isDataExpected(const CDate& date) const
   {

XIOS/dev/branch_openmp/src/filter/filter.hpp

@@ -49 +49 @@
 
       /*!
+       * Tests if the filter must auto-trigger.
+       *
+       * \return true if the filter must auto-trigger
+       */
+      bool virtual mustAutoTrigger() const;
+
+      /*!
        * Tests whether data is expected for the specified date.
        *

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

@@ -60 +60 @@
 
       /*!
+       * Tests if the pin must auto-trigger.
+       *
+       * \return true if the pin must auto-trigger
+       */
+      bool virtual mustAutoTrigger() const = 0;
+
+      /*!
        * Tests whether data is expected for the specified date.
        *

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

@@ -66 +66 @@
   }
 
+  bool COutputPin::mustAutoTrigger() const
+  {
+    std::vector<std::pair<boost::shared_ptr<CInputPin>, size_t> >::const_iterator it, itEnd;
+    for (it = outputs.begin(), itEnd = outputs.end(); it != itEnd; ++it)
+    {
+      if (it->first->mustAutoTrigger())
+        return true;
+    }
+
+    return false;
+  }
+
   void COutputPin::setOutputTriggers()
   {

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

@@ -44 +44 @@
        */
       bool virtual canBeTriggered() const;
+
+      /*!
+       * Tests if the pin must auto-trigger.
+       *
+       * \return true if the pin must auto-trigger
+       */
+      bool virtual mustAutoTrigger() const;
 
       /*!

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

@@ -7 +7 @@
 namespace xios
 {
-  CSourceFilter::CSourceFilter(CGarbageCollector& gc, CGrid* grid,
+  CSourceFilter::CSourceFilter(CGarbageCollector& gc, CGrid* grid, bool compression, 
                                const CDuration offset /*= NoneDu*/, bool manualTrigger /*= false*/,
                                bool hasMissingValue /*= false*/,
@@ -13 +13 @@
     : COutputPin(gc, manualTrigger)
     , grid(grid)
+    , compression(compression)
     , offset(offset)
     , hasMissingValue(hasMissingValue), defaultValue(defaultValue)
@@ -31 +32 @@
     packet->status = CDataPacket::NO_ERROR;
 
-    packet->data.resize(grid->storeIndex_client.numElements());
-    grid->inputField(data, packet->data);
+    packet->data.resize(grid->storeIndex_client.numElements());    
+    
+    if (compression)
+    {
+      packet->data = defaultValue;
+      grid->uncompressField(data, packet->data);    
+    }
+    else
+      grid->inputField(data, packet->data);
 
+    
+    
+    // if (compression) grid->inputField(data, packet->data) ;
+    // else
+    // {
+    //   // just make a flat copy
+    //   CArray<double, N> data_tmp(data.copy()) ; // supress const attribute
+    //   CArray<double,1> dataTmp2(data_tmp.dataFirst(),shape(data.numElements()),neverDeleteData) ;
+    //   packet->data = dataTmp2 ;
+    // }
     // Convert missing values to NaN
     if (hasMissingValue)
@@ -65 +83 @@
     packet->timestamp = date;
     packet->status = CDataPacket::NO_ERROR;
-
-    // if (data.size() != grid->storeIndex_toSrv.size())
+    
     if (data.size() != grid->storeIndex_fromSrv.size())
       ERROR("CSourceFilter::streamDataFromServer(CDate date, const std::map<int, CArray<double, 1> >& data)",
@@ -75 +92 @@
     std::map<int, CArray<double, 1> >::const_iterator it, itEnd = data.end();
     for (it = data.begin(); it != itEnd; it++)
-    {
-      // CArray<int,1>& index = grid->storeIndex_toSrv[it->first];
+    {      
       CArray<int,1>& index = grid->storeIndex_fromSrv[it->first];
       for (int n = 0; n < index.numElements(); n++)

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

@@ -27 +27 @@
        */
       CSourceFilter(CGarbageCollector& gc, CGrid* grid,
+                    bool compression=true,
                     const CDuration offset = NoneDu, bool manualTrigger = false,
                     bool hasMissingValue = false,
@@ -64 +65 @@
       const bool hasMissingValue;
       const double defaultValue;
+      const bool compression ; //!< indicate if the data need to be compressed : on client size : true, on server side : false
   }; // class CSourceFilter
 } // namespace xios

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

@@ -4 +4 @@
 #include "context.hpp"
 #include "context_client.hpp"
+#include "timer.hpp"
 using namespace ep_lib;
 
@@ -29 +30 @@
       size_t inputCount = 1 + (auxInputs.empty() ? 0 : auxInputs.size());
       double defaultValue  = (hasMissingValue) ? std::numeric_limits<double>::quiet_NaN() : 0.0;
-      boost::shared_ptr<CSpatialTransformFilter> filter(new CSpatialTransformFilter(gc, engine, defaultValue, inputCount));
-
+
+
+      const CGridTransformationSelector::ListAlgoType& algoList = gridTransformation->getAlgoList() ;
+      CGridTransformationSelector::ListAlgoType::const_iterator it  ;
+
+      bool isSpatialTemporal=false ;
+      for (it=algoList.begin();it!=algoList.end();++it)  if (it->second.first == TRANS_TEMPORAL_SPLITTING) isSpatialTemporal=true ;
+
+      boost::shared_ptr<CSpatialTransformFilter> filter ;
+      if( isSpatialTemporal) filter = boost::shared_ptr<CSpatialTransformFilter>(new CSpatialTemporalFilter(gc, engine, gridTransformation, defaultValue, inputCount));
+      else filter = boost::shared_ptr<CSpatialTransformFilter>(new CSpatialTransformFilter(gc, engine, defaultValue, inputCount));
+
+      
       if (!lastFilter)
         lastFilter = filter;
@@ -58 +70 @@
       onOutputReady(outputPacket);
   }
+
+
+
+
+
+  CSpatialTemporalFilter::CSpatialTemporalFilter(CGarbageCollector& gc, CSpatialTransformFilterEngine* engine, CGridTransformation* gridTransformation, double outputValue, size_t inputSlotsCount)
+    : CSpatialTransformFilter(gc, engine, outputValue, inputSlotsCount), record(0)
+  {
+      const CGridTransformationSelector::ListAlgoType& algoList = gridTransformation->getAlgoList() ;
+      CGridTransformationSelector::ListAlgoType::const_iterator it  ;
+
+      int pos ;
+      for (it=algoList.begin();it!=algoList.end();++it) 
+        if (it->second.first == TRANS_TEMPORAL_SPLITTING)
+        {
+          pos=it->first ;
+          if (pos < algoList.size()-1)
+            ERROR("SpatialTemporalFilter::CSpatialTemporalFilter(CGarbageCollector& gc, CSpatialTransformFilterEngine* engine, CGridTransformation* gridTransformation, double outputValue, size_t inputSlotsCount))",
+                  "temporal splitting operation must be the last of whole transformation on same grid") ;
+        }
+          
+      CGrid* grid=gridTransformation->getGridDestination() ;
+
+      CAxis* axis = grid->getAxis(gridTransformation->getElementPositionInGridDst2AxisPosition().find(pos)->second) ;
+
+      nrecords = axis->index.numElements() ;
+  }
+
+
+  void CSpatialTemporalFilter::onInputReady(std::vector<CDataPacketPtr> data)
+  {
+    CSpatialTransformFilterEngine* spaceFilter = static_cast<CSpatialTransformFilterEngine*>(engine);
+    CDataPacketPtr outputPacket = spaceFilter->applyFilter(data, outputDefaultValue);
+
+    if (outputPacket)
+    {
+      size_t nelements=outputPacket->data.numElements() ;
+      if (!tmpData.numElements())
+      {
+        tmpData.resize(nelements);
+        tmpData=outputDefaultValue ;
+      }
+
+      nelements/=nrecords ;
+      size_t offset=nelements*record ;
+      for(size_t i=0;i<nelements;++i)  tmpData(i+offset) = outputPacket->data(i) ;
+    
+      record ++ ;
+      if (record==nrecords)
+      {
+        record=0 ;
+        CDataPacketPtr 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;
+        onOutputReady(packet);
+        tmpData.resize(0) ;
+      }
+    }
+  }
+
 
   CSpatialTransformFilterEngine::CSpatialTransformFilterEngine(CGridTransformation* gridTransformation)
@@ -122 +197 @@
   void CSpatialTransformFilterEngine::apply(const CArray<double, 1>& dataSrc, CArray<double,1>& dataDest)
   {
+    CTimer::get("CSpatialTransformFilterEngine::apply").resume(); 
+    
     CContextClient* client = CContext::getCurrent()->client;
 
@@ -235 +312 @@
 
     dataDest = dataCurrentDest;
+
+    CTimer::get("CSpatialTransformFilterEngine::apply").suspend() ;
   }
 } // namespace xios

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

@@ -3 +3 @@
 
 #include "filter.hpp"
+
 namespace xios
 {
@@ -48 +49 @@
       double outputDefaultValue;
   }; // class CSpatialTransformFilter
+
+
+ /*!
+   * A specific spatial filter for the temporal_splitting transformation scalar -> axis. An incoming flux will be stored in an aditional dimension given by the destination axis.
+   * At each flux received, the storing index (record) is increased. When it reach the size of the axis (nrecords) a new flux is generated and the record is reset to 0
+   */
+
+ class CSpatialTemporalFilter : public CSpatialTransformFilter
+  {
+    public:
+      /*!
+       * Constructs a filter wrapping the specified spatial transformation.
+       *
+       * \param gc the associated garbage collector
+       * \param engine the engine defining the spatial transformation
+       * \param [in] gridTransformation the associated transformations
+       * \param outputValue default value of output pin
+       * \param [in] inputSlotsCount number of input, by default there is only one for field src
+       */
+      CSpatialTemporalFilter(CGarbageCollector& gc, CSpatialTransformFilterEngine* engine, CGridTransformation* gridTransformation, double outputValue, size_t inputSlotsCount = 1);
+
+
+    protected:
+      /*!
+        Overriding this function to process transformations with auxillary inputs
+      */
+      void virtual onInputReady(std::vector<CDataPacketPtr> data);
+      //! Current record in the filter
+      int record ;
+      //! Maximum number of records
+      int nrecords;
+      //! Temporary storage for output flux
+      CArray<double, 1> tmpData; 
+
+
+  }; // class CSpatialTemporalFilter
+
 
   /*!

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

@@ -111 +111 @@
   }
 
+  bool CStoreFilter::mustAutoTrigger() const
+  {
+    return false;
+  }
+
   bool CStoreFilter::isDataExpected(const CDate& date) const
   {

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

@@ -53 +53 @@
 
       /*!
+       * Tests if the filter must auto-trigger.
+       *
+       * \return true if the filter must auto-trigger
+       */
+      bool virtual mustAutoTrigger() const;
+
+      /*!
        * Tests whether data is expected for the specified date.
        *

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

@@ -5 +5 @@
 namespace xios
 {
-  static func::CFunctor* createFunctor(const std::string& opId, bool ignoreMissingValue, double missingValue, CArray<double, 1>& tmpData);
+  static func::CFunctor* createFunctor(const std::string& opId, bool ignoreMissingValue, CArray<double, 1>& tmpData);
 
   CTemporalFilter::CTemporalFilter(CGarbageCollector& gc, const std::string& opId,
                                    const CDate& initDate, const CDuration samplingFreq, const CDuration samplingOffset, const CDuration opFreq,
-                                   bool ignoreMissingValue /*= false*/, double missingValue /*= 0.0*/)
+                                   bool ignoreMissingValue /*= false*/)
     : CFilter(gc, 1, this)
-    , functor(createFunctor(opId, ignoreMissingValue, missingValue, tmpData))
+    , functor(createFunctor(opId, ignoreMissingValue, tmpData))
     , isOnceOperation(functor->timeType() == func::CFunctor::once)
     , isInstantOperation(functor->timeType() == func::CFunctor::instant)
@@ -17 +17 @@
     , samplingOffset(samplingOffset)
     , opFreq(opFreq)
-    , nextSamplingDate(initDate + this->samplingOffset + initDate.getRelCalendar().getTimeStep())
-    , nextOperationDate(initDate + this->samplingOffset + opFreq)
+    , offsetMonth({0, this->samplingOffset.month, 0, 0, 0, 0, 0})
+    , offsetAllButMonth({this->samplingOffset.year, 0 , this->samplingOffset.day,
+                       this->samplingOffset.hour, this->samplingOffset.minute,
+                       this->samplingOffset.second, this->samplingOffset.timestep})
+    , initDate(initDate)
+    , nextSamplingDate(initDate + (this->samplingOffset + initDate.getRelCalendar().getTimeStep()))
+    , nbOperationDates(1)
+//    , nextOperationDate(initDate + opFreq + this->samplingOffset)
     , isFirstOperation(true)
   {
@@ -35 +41 @@
       {
         usePacket = (data[0]->date >= nextSamplingDate);
-        outputResult = (data[0]->date + samplingFreq > nextOperationDate);
+//        outputResult = (data[0]->date + samplingFreq > nextOperationDate);
+        outputResult = (data[0]->date  > initDate + nbOperationDates*opFreq - samplingFreq + offsetMonth + offsetAllButMonth);
         copyLess = (isInstantOperation && usePacket && outputResult);
       }
@@ -54 +61 @@
       if (outputResult)
       {
+        nbOperationDates ++;
         if (!copyLess)
         {
@@ -69 +77 @@
 
         isFirstOperation = false;
-        nextOperationDate = nextOperationDate + samplingFreq + opFreq - samplingFreq;
+//        nextOperationDate = initDate + samplingFreq + nbOperationDates*opFreq - samplingFreq + offsetMonth + offsetAllButMonth;
       }
     }
@@ -76 +84 @@
   }
 
+  bool CTemporalFilter::mustAutoTrigger() const
+  {
+    return true;
+  }
+
   bool CTemporalFilter::isDataExpected(const CDate& date) const
   {
-    return isOnceOperation ? isFirstOperation : (date >= nextSamplingDate || date + samplingFreq > nextOperationDate);
+//    return isOnceOperation ? isFirstOperation : (date >= nextSamplingDate || date + samplingFreq > nextOperationDate);
+    return isOnceOperation ? isFirstOperation : (date >= nextSamplingDate || date > initDate + nbOperationDates*opFreq - samplingFreq + offsetMonth + offsetAllButMonth);
   }
 
-  static func::CFunctor* createFunctor(const std::string& opId, bool ignoreMissingValue, double missingValue, CArray<double, 1>& tmpData)
+  static func::CFunctor* createFunctor(const std::string& opId, bool ignoreMissingValue, CArray<double, 1>& tmpData)
   {
     func::CFunctor* functor = NULL;
 
-    double defaultValue = ignoreMissingValue ? std::numeric_limits<double>::quiet_NaN() : missingValue;
+    double defaultValue = std::numeric_limits<double>::quiet_NaN();
 
 #define DECLARE_FUNCTOR(MType, mtype) \

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

@@ -26 +26 @@
        * \param ignoreMissingValue true if and only if the missing value must be ignored
                                    when doing the operation
-       * \param missingValue the missing value
        */
       CTemporalFilter(CGarbageCollector& gc, const std::string& opId,
                       const CDate& initDate, const CDuration samplingFreq, const CDuration samplingOffset, const CDuration opFreq,
-                      bool ignoreMissingValue = false, double missingValue = 0.0);
+                      bool ignoreMissingValue = false);
 
       /*!
@@ -39 +38 @@
        */
       CDataPacketPtr virtual apply(std::vector<CDataPacketPtr> data);
+
+      /*!
+       * Tests if the filter must auto-trigger.
+       *
+       * \return true if the filter must auto-trigger
+       */
+      bool virtual mustAutoTrigger() const;
 
       /*!
@@ -56 +62 @@
       const CDuration samplingOffset; //!< The sampling offset, i.e. the offset after which the input data will be used
       const CDuration opFreq; //!< The operation frequency, i.e. the frequency at which the output data will be computed
+      const CDuration offsetMonth; //!< The month duration of samplingOffset
+      CDuration offsetAllButMonth; //!< All but the month duration of samplingOffset
+      const CDate initDate;
       CDate nextSamplingDate; //!< The date of the next sampling
-      CDate nextOperationDate; //!< The date of the next operation
+      int nbOperationDates; //!< The number of times an operation is performed
+//      CDate nextOperationDate; //!< The date of the next operation
       bool isFirstOperation; //!< True before the first operation was been computed
   }; // class CTemporalFilter