source: XIOS/dev/dev_trunk_graph/src/filter/temporal_filter.cpp @ 2021

Last change on this file since 2021 was 2021, checked in by yushan, 8 months ago

Graph intermediate commit to a tmp branch.

File size: 4.9 KB
Line 
1#include "temporal_filter.hpp"
2#include "functor_type.hpp"
3#include "calendar_util.hpp"
4#include "workflow_graph.hpp"
5
6namespace xios
7{
8  static func::CFunctor* createFunctor(const std::string& opId, bool ignoreMissingValue, CArray<double, 1>& tmpData);
9
10  CTemporalFilter::CTemporalFilter(CGarbageCollector& gc, const std::string& opId,
11                                   const CDate& initDate, const CDuration samplingFreq, const CDuration samplingOffset, const CDuration opFreq,
12                                   bool ignoreMissingValue /*= false*/)
13    : CFilter(gc, 1, this)
14    , functor(createFunctor(opId, ignoreMissingValue, tmpData))
15    , isOnceOperation(functor->timeType() == func::CFunctor::once)
16    , isInstantOperation(functor->timeType() == func::CFunctor::instant)
17    , samplingFreq(samplingFreq)
18    , samplingOffset(samplingOffset)
19    , opFreq(opFreq)
20    , offsetMonth(0, this->samplingOffset.month, 0, 0, 0, 0, 0)
21    , offsetAllButMonth(this->samplingOffset.year, 0 , this->samplingOffset.day,
22                        this->samplingOffset.hour, this->samplingOffset.minute,
23                        this->samplingOffset.second, this->samplingOffset.timestep)
24    , initDate(initDate)
25    , nextSamplingDate(initDate + offsetMonth + ( offsetAllButMonth + initDate.getRelCalendar().getTimeStep()))
26    , nbOperationDates(1)
27    , nbSamplingDates(0)
28    , isFirstOperation(true)
29  {
30  }
31
32  void CTemporalFilter::buildWorkflowGraph(std::vector<CDataPacketPtr> data)
33  {
34    if(this->graphEnabled)
35    {
36      if(!data[0]->graphPackage)
37      {
38        data[0]->graphPackage = new CGraphDataPackage;
39      }
40     
41      if(this->graphPackage->sourceFilterIds.size()==0)
42      { 
43        this->graphPackage->filterId = CWorkflowGraph::getNodeSize();
44        CWorkflowGraph::addNode("Temporal filter", 3, false, 0, data[0]);
45      }
46     
47      CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, this->graphPackage->filterId, data[0]);
48      data[0]->graphPackage->fromFilter = this->graphPackage->filterId;
49      this->graphPackage->sourceFilterIds.push_back(data[0]->graphPackage->fromFilter); 
50      data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
51      std::rotate(this->graphPackage->inFields.begin(), this->graphPackage->inFields.begin() + 1, this->graphPackage->inFields.end());
52    }
53
54  }
55  CDataPacketPtr CTemporalFilter::apply(std::vector<CDataPacketPtr> data)
56  {
57    buildWorkflowGraph(data);
58 
59    CDataPacketPtr packet;
60
61    if (data[0]->status != CDataPacket::END_OF_STREAM)
62    {
63      bool usePacket, outputResult, copyLess;
64      if (isOnceOperation)
65        usePacket = outputResult = copyLess = isFirstOperation;
66      else
67      {
68        usePacket = (data[0]->date >= nextSamplingDate);
69        outputResult = (data[0]->date  > initDate + nbOperationDates*opFreq - samplingFreq + offsetMonth + offsetAllButMonth);
70        copyLess = (isInstantOperation && usePacket && outputResult);
71      }
72
73      if (usePacket)
74      {
75        nbSamplingDates ++;
76        if (!copyLess)
77        {
78          if (!tmpData.numElements())
79            tmpData.resize(data[0]->data.numElements());
80
81          (*functor)(data[0]->data);
82        }
83
84        nextSamplingDate = ((initDate + offsetMonth) + nbSamplingDates * samplingFreq) + offsetAllButMonth + initDate.getRelCalendar().getTimeStep();
85      }
86
87      if (outputResult)
88      {
89        nbOperationDates ++;
90        if (!copyLess)
91        {
92          functor->final();
93
94          packet = CDataPacketPtr(new CDataPacket);
95          packet->date = data[0]->date;
96          packet->timestamp = data[0]->timestamp;
97          packet->status = data[0]->status;
98          packet->data.resize(tmpData.numElements());
99          packet->data = tmpData;
100          packet->graphPackage = data[0]->graphPackage;
101        }
102        else
103          packet = data[0];
104
105        isFirstOperation = false;
106       
107      }
108    }
109
110    return packet;
111  }
112
113  bool CTemporalFilter::mustAutoTrigger() const
114  {
115    return true;
116  }
117
118  bool CTemporalFilter::isDataExpected(const CDate& date) const
119  {
120    return isOnceOperation ? isFirstOperation : (date >= nextSamplingDate || date > initDate + nbOperationDates*opFreq - samplingFreq + offsetMonth + offsetAllButMonth);
121  }
122
123  static func::CFunctor* createFunctor(const std::string& opId, bool ignoreMissingValue, CArray<double, 1>& tmpData)
124  {
125    func::CFunctor* functor = NULL;
126
127    double defaultValue = std::numeric_limits<double>::quiet_NaN();
128
129#define DECLARE_FUNCTOR(MType, mtype) \
130    if (opId.compare(#mtype) == 0) \
131    { \
132      if (ignoreMissingValue) \
133      { \
134        functor = new func::C##MType(tmpData, defaultValue); \
135      } \
136      else \
137      { \
138        functor = new func::C##MType(tmpData); \
139      } \
140    }
141
142#include "functor_type.conf"
143
144    if (!functor)
145      ERROR("createFunctor(const std::string& opId, ...)",
146            << "\"" << opId << "\" is not a valid operation.");
147
148    return functor;
149  }
150} // namespace xios
Note: See TracBrowser for help on using the repository browser.