Changeset 1158 for XIOS/dev/dev_olga/src/filter

Timestamp:

06/06/17 17:58:16 (7 years ago)

Author:

oabramkina

Message:

Two server levels: merging with trunk r1137.
There are bugs.

Location:

XIOS/dev/dev_olga/src/filter

Files:

• file_server_writer_filter.cpp (modified) (1 diff)
• file_server_writer_filter.hpp (modified) (1 diff)
• file_writer_filter.cpp (modified) (2 diffs)
• 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) (2 diffs)
• source_filter.hpp (modified) (2 diffs)
• spatial_transform_filter.cpp (modified) (6 diffs)
• spatial_transform_filter.hpp (modified) (1 diff)
• store_filter.cpp (modified) (2 diffs)
• store_filter.hpp (modified) (1 diff)
• temporal_filter.cpp (modified) (3 diffs)
• temporal_filter.hpp (modified) (1 diff)

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

@@ -18 +18 @@
     field->writeUpdateData(data[0]->data);
   }
+
+  bool CFileServerWriterFilter::isDataExpected(const CDate& date) const
+  {
+    return true;
+  }
 } // namespace xios

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

@@ -23 +23 @@
       CFileServerWriterFilter(CGarbageCollector& gc, CField* field);
 
+      bool virtual isDataExpected(const CDate& date) const;
     protected:
       /*!

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

@@ -2 +2 @@
 #include "exception.hpp"
 #include "field.hpp"
+#include "utils.hpp"
 
 namespace xios
@@ -16 +17 @@
   void CFileWriterFilter::onInputReady(std::vector<CDataPacketPtr> data)
   {
+    bool ignoreMissingValue = (!field->detect_missing_value.isEmpty() && 
+                               !field->default_value.isEmpty() && 
+                               field->detect_missing_value == true);
+    if (ignoreMissingValue)
+    {
+      double missingValue = field->default_value;
+      size_t nbData = data[0]->data.numElements();
+      for (size_t idx = 0; idx < nbData; ++idx)
+      {
+        if (NumTraits<double>::isnan(data[0]->data(idx)))
+          data[0]->data(idx) = missingValue;
+      }
+    }    
+
     field->sendUpdateData(data[0]->data);
   }
+
+  bool CFileWriterFilter::isDataExpected(const CDate& date) const
+  {
+    return true;
+  }
 } // namespace xios

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

@@ -23 +23 @@
       CFileWriterFilter(CGarbageCollector& gc, CField* field);
 
+      /*!
+       * Tests whether data is expected for the specified date.
+       *
+       * \param date the date associated to the data
+       */
+      bool virtual isDataExpected(const CDate& date) const;
+
     protected:
       /*!

XIOS/dev/dev_olga/src/filter/filter.cpp

@@ -40 +40 @@
     return (CInputPin::canBeTriggered() || COutputPin::canBeTriggered());
   }
+
+  bool CFilter::isDataExpected(const CDate& date) const
+  {
+    return COutputPin::isDataExpected(date);
+  }
 } // namespace xios

XIOS/dev/dev_olga/src/filter/filter.hpp

@@ -48 +48 @@
       bool virtual canBeTriggered() const;
 
+      /*!
+       * Tests whether data is expected for the specified date.
+       *
+       * \param date the date associated to the data
+       */
+      bool virtual isDataExpected(const CDate& date) const;
+
     protected:
       IFilterEngine* engine; //!< The filter engine, might be the filter itself

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

@@ -60 +60 @@
 
       /*!
+       * Tests whether data is expected for the specified date.
+       *
+       * \param date the date associated to the data
+       */
+      bool virtual isDataExpected(const CDate& date) const = 0;
+
+      /*!
        * Removes all pending packets which are older than the specified timestamp.
        *

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

@@ -73 +73 @@
   }
 
+  bool COutputPin::isDataExpected(const CDate& date) 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->isDataExpected(date))
+        return true;
+    }
+
+    return false;
+  }
+
   void COutputPin::invalidate(Time timestamp)
   {

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

@@ -46 +46 @@
 
       /*!
+       * Tests whether data is expected for the specified date.
+       *
+       * \param date the date associated to the data
+       */
+      bool virtual isDataExpected(const CDate& date) const;
+
+      /*!
        * Removes all pending packets which are older than the specified timestamp.
        *

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

@@ -3 +3 @@
 #include "exception.hpp"
 #include "calendar_util.hpp"
+#include <limits> 
 
 namespace xios
 {
   CSourceFilter::CSourceFilter(CGarbageCollector& gc, CGrid* grid,
-                               const CDuration offset /*= NoneDu*/, bool manualTrigger /*= false*/)
+                               const CDuration offset /*= NoneDu*/, bool manualTrigger /*= false*/,
+                               bool hasMissingValue /*= false*/,
+                               double defaultValue /*= 0.0*/)
     : COutputPin(gc, manualTrigger)
     , grid(grid)
     , offset(offset)
+    , hasMissingValue(hasMissingValue), defaultValue(defaultValue)
   {
     if (!grid)
@@ -29 +33 @@
     packet->data.resize(grid->storeIndex_client.numElements());
     grid->inputField(data, packet->data);
+
+    // Convert missing values to NaN
+    if (hasMissingValue)
+    {
+      double nanValue = std::numeric_limits<double>::quiet_NaN();
+      size_t nbData = packet->data.numElements();
+      for (size_t idx = 0; idx < nbData; ++idx)
+      {
+        if (defaultValue == packet->data(idx))
+          packet->data(idx) = nanValue;
+      }
+    }
 
     onOutputReady(packet);

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

@@ -23 +23 @@
        * \param offset the offset applied to the timestamp of all packets
        * \param manualTrigger whether the output should be triggered manually
+       * \param hasMissingValue whether data has missing value
+       * \param defaultValue missing value to detect
        */
       CSourceFilter(CGarbageCollector& gc, CGrid* grid,
-                    const CDuration offset = NoneDu, bool manualTrigger = false);
+                    const CDuration offset = NoneDu, bool manualTrigger = false,
+                    bool hasMissingValue = false,
+                    double defaultValue = 0.0);
 
       /*!
@@ -58 +62 @@
       CGrid* grid; //!< The grid attached to the data the filter can accept
       const CDuration offset; //!< The offset applied to the timestamp of all packets
+      bool hasMissingValue;
+      double defaultValue;
   }; // class CSourceFilter
 } // namespace xios

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

@@ -11 +11 @@
 
   std::pair<boost::shared_ptr<CSpatialTransformFilter>, boost::shared_ptr<CSpatialTransformFilter> >
-  CSpatialTransformFilter::buildFilterGraph(CGarbageCollector& gc, CGrid* srcGrid, CGrid* destGrid, double defaultValue)
+  CSpatialTransformFilter::buildFilterGraph(CGarbageCollector& gc, CGrid* srcGrid, CGrid* destGrid, bool hasMissingValue, double missingValue)
   {
     if (!srcGrid || !destGrid)
@@ -26 +26 @@
       const std::vector<StdString>& auxInputs = gridTransformation->getAuxInputs();
       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));
 
@@ -103 +104 @@
       }
       packet->data.resize(gridTransformation->getGridDestination()->storeIndex_client.numElements());
-      packet->data = defaultValue;
+      if (0 != packet->data.numElements())
+        (packet->data)(0) = defaultValue;
       apply(data[0]->data, packet->data);
     }
@@ -115 +117 @@
 
     // Get default value for output data
-    double defaultValue = 0.0;
-    if (0 != dataDest.numElements()) defaultValue = dataDest(0);
+    bool ignoreMissingValue = false; 
+    double defaultValue = std::numeric_limits<double>::quiet_NaN();
+    if (0 != dataDest.numElements()) ignoreMissingValue = NumTraits<double>::isnan(dataDest(0));
 
     const std::list<CGridTransformation::SendingIndexGridSourceMap>& listLocalIndexSend = gridTransformation->getLocalIndexToSendFromGridSource();
@@ -191 +194 @@
         else dataCurrentDest(i) = defaultValue;
 
-      std::vector<bool> localInitFlag(dataCurrentDest.size(), true);
+      std::vector<bool> localInitFlag(dataCurrentDest.numElements(), true);
       currentBuff = 0;
       for (itRecv = itbRecv; itRecv != iteRecv; ++itRecv)
@@ -201 +204 @@
                          dataCurrentDest,
                          localInitFlag,
-                         defaultValue);
+                         ignoreMissingValue);
 
         currentBuff += countSize;

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

@@ -32 +32 @@
        * \param srcGrid the source grid
        * \param destGrid the destination grid
+       * \param hasMissingValue whether field source has missing value
+       * \param defaultValue default value
        * \return the first and the last filters of the filter graph
        */
       static std::pair<boost::shared_ptr<CSpatialTransformFilter>, boost::shared_ptr<CSpatialTransformFilter> >
-      buildFilterGraph(CGarbageCollector& gc, CGrid* srcGrid, CGrid* destGrid, double defaultValue);
+      buildFilterGraph(CGarbageCollector& gc, CGrid* srcGrid, CGrid* destGrid, bool hasMissingValue, double defaultValue);
 
     protected:

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

@@ -24 +24 @@
     CTimer timer("CStoreFilter::getPacket");
     CConstDataPacketPtr packet;
-    const double timeout = 10 ; // 10 seconds timeout
+    const double timeout = CXios::recvFieldTimeout;
 
     do
@@ -81 +81 @@
   }
 
+  bool CStoreFilter::isDataExpected(const CDate& date) const
+  {
+    return true;
+  }
+
   void CStoreFilter::invalidate(Time timestamp)
   {

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

@@ -50 +50 @@
 
       /*!
+       * Tests whether data is expected for the specified date.
+       *
+       * \param date the date associated to the data
+       */
+      bool virtual isDataExpected(const CDate& date) const;
+
+      /*!
        * Removes all pending packets which are older than the specified timestamp.
        *

XIOS/dev/dev_olga/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);
+
   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*/)
     : CFilter(gc, 1, this)
-    , samplingFreq(samplingFreq)
+    , functor(createFunctor(opId, ignoreMissingValue, missingValue, tmpData))
+    , isOnceOperation(functor->timeType() == func::CFunctor::once)
+    , isInstantOperation(functor->timeType() == func::CFunctor::instant)
+    // If we can optimize the sampling when dealing with an instant functor we do it
+    , samplingFreq((isInstantOperation && samplingFreq == TimeStep && samplingOffset == NoneDu) ? opFreq : samplingFreq)
+    , samplingOffset((isInstantOperation && samplingFreq == TimeStep && samplingOffset == NoneDu) ? opFreq - initDate.getRelCalendar().getTimeStep() : samplingOffset)
     , opFreq(opFreq)
-    , nextSamplingDate(initDate + samplingOffset + initDate.getRelCalendar().getTimeStep())
-    , nextOperationDate(initDate + opFreq)
+    , nextSamplingDate(initDate + this->samplingOffset + initDate.getRelCalendar().getTimeStep())
+    , nextOperationDate(initDate + this->samplingOffset + opFreq)
     , isFirstOperation(true)
   {
-#define DECLARE_FUNCTOR(MType, mtype) \
-    if (opId.compare(#mtype) == 0) \
-    { \
-      if (ignoreMissingValue) \
-      { \
-        functor.reset(new func::C##MType(tmpData, missingValue)); \
-      } \
-      else \
-      { \
-        functor.reset(new func::C##MType(tmpData)); \
-      } \
-    }
-
-#include "functor_type.conf"
-
-    if (!functor)
-      ERROR("CTemporalFilter::CTemporalFilter(CGarbageCollector& gc, const std::string& opId, ...)",
-            << "\"" << opId << "\" is not a valid operation.");
-
-    isOnceOperation = (functor->timeType() == func::CFunctor::once);
   }
 
@@ -43 +30 @@
     if (data[0]->status != CDataPacket::END_OF_STREAM)
     {
-      const bool usePacket = isOnceOperation ? isFirstOperation : (data[0]->date >= nextSamplingDate);
+      bool usePacket, outputResult, copyLess;
+      if (isOnceOperation)
+        usePacket = outputResult = copyLess = isFirstOperation;
+      else
+      {
+        usePacket = (data[0]->date >= nextSamplingDate);
+        outputResult = (data[0]->date + samplingFreq > nextOperationDate);
+        copyLess = (isInstantOperation && usePacket && outputResult);
+      }
+
       if (usePacket)
       {
-        if (!tmpData.numElements())
-          tmpData.resize(data[0]->data.numElements());
+        if (!copyLess)
+        {
+          if (!tmpData.numElements())
+            tmpData.resize(data[0]->data.numElements());
 
-        (*functor)(data[0]->data);
+          (*functor)(data[0]->data);
+        }
 
         nextSamplingDate = nextSamplingDate + samplingFreq;
       }
 
-      const bool outputResult = isOnceOperation ? isFirstOperation : (data[0]->date + samplingFreq > nextOperationDate);
       if (outputResult)
       {
-        functor->final();
+        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;
+          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;
@@ -73 +76 @@
     return packet;
   }
+
+  bool CTemporalFilter::isDataExpected(const CDate& date) const
+  {
+    return isOnceOperation ? isFirstOperation : (date >= nextSamplingDate || date + samplingFreq > nextOperationDate);
+  }
+
+  static func::CFunctor* createFunctor(const std::string& opId, bool ignoreMissingValue, double missingValue, CArray<double, 1>& tmpData)
+  {
+    func::CFunctor* functor = NULL;
+
+    double defaultValue = ignoreMissingValue ? std::numeric_limits<double>::quiet_NaN() : missingValue;
+
+#define DECLARE_FUNCTOR(MType, mtype) \
+    if (opId.compare(#mtype) == 0) \
+    { \
+      if (ignoreMissingValue) \
+      { \
+        functor = new func::C##MType(tmpData, defaultValue); \
+      } \
+      else \
+      { \
+        functor = new func::C##MType(tmpData); \
+      } \
+    }
+
+#include "functor_type.conf"
+
+    if (!functor)
+      ERROR("createFunctor(const std::string& opId, ...)",
+            << "\"" << opId << "\" is not a valid operation.");
+
+    return functor;
+  }
 } // namespace xios

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

@@ -40 +40 @@
       CDataPacketPtr virtual apply(std::vector<CDataPacketPtr> data);
 
+      /*!
+       * Tests whether data is expected for the specified date.
+       *
+       * \param date the date associated to the data
+       */
+      bool virtual isDataExpected(const CDate& date) const;
+
     private:
-      boost::scoped_ptr<func::CFunctor> functor; //!< The functor corresponding to the temporal operation
+      const boost::scoped_ptr<func::CFunctor> functor; //!< The functor corresponding to the temporal operation
       CArray<double, 1> tmpData; //!< The array of data used for temporary storage
       const CDuration samplingFreq; //!< The sampling frequency, i.e. the frequency at which the input data will be used
+      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
       CDate nextSamplingDate; //!< The date of the next sampling
       CDate nextOperationDate; //!< The date of the next operation
       bool isFirstOperation; //!< True before the first operation was been computed
-      bool isOnceOperation; //!< True if the operation should be computed just once
+      const bool isOnceOperation; //!< True if the operation should be computed just once
+      const bool isInstantOperation; //!< True if the operation is instant
   }; // class CTemporalFilter
 } // namespace xios