source: XIOS/dev/dev_olga/src/filter/source_filter.cpp @ 1568

Last change on this file since 1568 was 1568, checked in by oabramkina, 3 years ago

Grid mask is now applied in the source filter of clients: unmasked values are replaced by NaN. It is not reconstructed any more by servers.

This needs to be tested more rigorously before commiting to trunk.

File size: 4.3 KB
Line 
1#include "source_filter.hpp"
2#include "grid.hpp"
3#include "exception.hpp"
4#include "calendar_util.hpp"
5#include <limits>
6
7namespace xios
8{
9  CSourceFilter::CSourceFilter(CGarbageCollector& gc, CGrid* grid,
10                               bool compression /*= true*/, bool mask /*= false*/,
11                               const CDuration offset /*= NoneDu*/, bool manualTrigger /*= false*/,
12                               bool hasMissingValue /*= false*/,
13                               double defaultValue /*= 0.0*/)
14    : COutputPin(gc, manualTrigger)
15    , grid(grid)
16    , compression(compression)
17    , mask(mask)
18    , offset(offset)
19    , hasMissingValue(hasMissingValue), defaultValue(defaultValue)
20  {
21    if (!grid)
22      ERROR("CSourceFilter::CSourceFilter(CGrid* grid)",
23            "Impossible to construct a source filter without providing a grid.");
24  }
25
26  template <int N>
27  void CSourceFilter::streamData(CDate date, const CArray<double, N>& data)
28  {
29    date = date + offset; // this is a temporary solution, it should be part of a proper temporal filter
30
31    CDataPacketPtr packet(new CDataPacket);
32    packet->date = date;
33    packet->timestamp = date;
34    packet->status = CDataPacket::NO_ERROR;
35
36    packet->data.resize(grid->storeIndex_client.numElements());   
37   
38    if (compression)
39    {
40      packet->data = defaultValue;
41      grid->uncompressField(data, packet->data);   
42    }
43    else
44    {
45      if (mask)
46        grid->maskField(data, packet->data);
47      else
48        grid->inputField(data, packet->data);
49    }
50    // Convert missing values to NaN
51    if (hasMissingValue)
52    {
53      const double nanValue = std::numeric_limits<double>::quiet_NaN();
54      const size_t nbData = packet->data.numElements();
55      for (size_t idx = 0; idx < nbData; ++idx)
56      {
57        if (defaultValue == packet->data(idx))
58          packet->data(idx) = nanValue;
59      }
60    }
61
62    onOutputReady(packet);
63  }
64
65  template void CSourceFilter::streamData<1>(CDate date, const CArray<double, 1>& data);
66  template void CSourceFilter::streamData<2>(CDate date, const CArray<double, 2>& data);
67  template void CSourceFilter::streamData<3>(CDate date, const CArray<double, 3>& data);
68  template void CSourceFilter::streamData<4>(CDate date, const CArray<double, 4>& data);
69  template void CSourceFilter::streamData<5>(CDate date, const CArray<double, 5>& data);
70  template void CSourceFilter::streamData<6>(CDate date, const CArray<double, 6>& data);
71  template void CSourceFilter::streamData<7>(CDate date, const CArray<double, 7>& data);
72
73  void CSourceFilter::streamDataFromServer(CDate date, const std::map<int, CArray<double, 1> >& data)
74  {
75    date = date + offset; // this is a temporary solution, it should be part of a proper temporal filter
76
77    CDataPacketPtr packet(new CDataPacket);
78    packet->date = date;
79    packet->timestamp = date;
80    packet->status = CDataPacket::NO_ERROR;
81   
82    if (data.size() != grid->storeIndex_fromSrv.size())
83      ERROR("CSourceFilter::streamDataFromServer(CDate date, const std::map<int, CArray<double, 1> >& data)",
84            << "Incoherent data received from servers,"
85            << " expected " << grid->storeIndex_fromSrv.size() << " chunks but " << data.size() << " were given.");
86
87    packet->data.resize(grid->storeIndex_client.numElements());
88    std::map<int, CArray<double, 1> >::const_iterator it, itEnd = data.end();
89    for (it = data.begin(); it != itEnd; it++)
90    {     
91      CArray<int,1>& index = grid->storeIndex_fromSrv[it->first];
92      for (int n = 0; n < index.numElements(); n++)
93        packet->data(index(n)) = it->second(n);
94    }
95
96    // Convert missing values to NaN
97    if (hasMissingValue)
98    {
99      const double nanValue = std::numeric_limits<double>::quiet_NaN();
100      const size_t nbData = packet->data.numElements();
101      for (size_t idx = 0; idx < nbData; ++idx)
102      {
103        if (defaultValue == packet->data(idx))
104          packet->data(idx) = nanValue;
105      }
106    }
107
108    onOutputReady(packet);
109  }
110
111  void CSourceFilter::signalEndOfStream(CDate date)
112  {
113    date = date + offset; // this is a temporary solution, it should be part of a proper temporal filter
114
115    CDataPacketPtr packet(new CDataPacket);
116    packet->date = date;
117    packet->timestamp = date;
118    packet->status = CDataPacket::END_OF_STREAM;
119    onOutputReady(packet);
120  }
121} // namespace xios
Note: See TracBrowser for help on using the repository browser.