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

Last change on this file since 2028 was 2028, checked in by yushan, 9 months ago

Graph intermediate commit to a tmp branch.

File size: 5.2 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    , graphCycleCompleted(true)
30  {
31  }
32
33  void CTemporalFilter::buildWorkflowGraph(std::vector<CDataPacketPtr> data)
34  {
35    if(this->graphEnabled )
36    {
37      if(!data[0]->graphPackage)
38      {
39        data[0]->graphPackage = new CGraphDataPackage;
40      }
41     
42      //if(this->graphPackage->sourceFilterIds.size()==0)
43      if(graphCycleCompleted)
44      { 
45        this->graphPackage->filterId = CWorkflowGraph::getNodeSize();
46        CWorkflowGraph::addNode("Temporal filter", 3, false, 0, data[0]);
47        graphCycleCompleted = false;
48      }
49     
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      CWorkflowGraph::addEdge(data[0]->graphPackage->fromFilter, this->graphPackage->filterId, data[0]);
54      data[0]->graphPackage->fromFilter = this->graphPackage->filterId;
55      this->graphPackage->sourceFilterIds.push_back(data[0]->graphPackage->fromFilter); 
56      data[0]->graphPackage->currentField = this->graphPackage->inFields[0];
57      std::rotate(this->graphPackage->inFields.begin(), this->graphPackage->inFields.begin() + 1, this->graphPackage->inFields.end());
58    }
59
60  }
61  CDataPacketPtr CTemporalFilter::apply(std::vector<CDataPacketPtr> data)
62  {
63    buildWorkflowGraph(data);
64 
65    CDataPacketPtr packet;
66
67    if (data[0]->status != CDataPacket::END_OF_STREAM)
68    {
69      bool usePacket, outputResult, copyLess;
70      if (isOnceOperation)
71        usePacket = outputResult = copyLess = isFirstOperation;
72      else
73      {
74        usePacket = (data[0]->date >= nextSamplingDate);
75        outputResult = (data[0]->date  > initDate + nbOperationDates*opFreq - samplingFreq + offsetMonth + offsetAllButMonth);
76        copyLess = (isInstantOperation && usePacket && outputResult);
77      }
78
79      if (usePacket)
80      {
81        nbSamplingDates ++;
82        if (!copyLess)
83        {
84          if (!tmpData.numElements())
85            tmpData.resize(data[0]->data.numElements());
86
87          (*functor)(data[0]->data);
88        }
89
90        nextSamplingDate = ((initDate + offsetMonth) + nbSamplingDates * samplingFreq) + offsetAllButMonth + initDate.getRelCalendar().getTimeStep();
91      }
92
93      if (outputResult)
94      {
95        nbOperationDates ++;
96        if (!copyLess)
97        {
98          functor->final();
99
100          packet = CDataPacketPtr(new CDataPacket);
101          packet->date = data[0]->date;
102          packet->timestamp = data[0]->timestamp;
103          packet->status = data[0]->status;
104          packet->data.resize(tmpData.numElements());
105          packet->data = tmpData;
106          packet->graphPackage = data[0]->graphPackage;
107        }
108        else
109          packet = data[0];
110
111        isFirstOperation = false;
112       
113        graphCycleCompleted = true;
114      }
115    }
116
117    return packet;
118  }
119
120  bool CTemporalFilter::mustAutoTrigger() const
121  {
122    return true;
123  }
124
125  bool CTemporalFilter::isDataExpected(const CDate& date) const
126  {
127    return isOnceOperation ? isFirstOperation : (date >= nextSamplingDate || date > initDate + nbOperationDates*opFreq - samplingFreq + offsetMonth + offsetAllButMonth);
128  }
129
130  static func::CFunctor* createFunctor(const std::string& opId, bool ignoreMissingValue, CArray<double, 1>& tmpData)
131  {
132    func::CFunctor* functor = NULL;
133
134    double defaultValue = std::numeric_limits<double>::quiet_NaN();
135
136#define DECLARE_FUNCTOR(MType, mtype) \
137    if (opId.compare(#mtype) == 0) \
138    { \
139      if (ignoreMissingValue) \
140      { \
141        functor = new func::C##MType(tmpData, defaultValue); \
142      } \
143      else \
144      { \
145        functor = new func::C##MType(tmpData); \
146      } \
147    }
148
149#include "functor_type.conf"
150
151    if (!functor)
152      ERROR("createFunctor(const std::string& opId, ...)",
153            << "\"" << opId << "\" is not a valid operation.");
154
155    return functor;
156  }
157} // namespace xios
Note: See TracBrowser for help on using the repository browser.