SensorBase.h 6.94 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
/*
 * SensorBase.h
 *
 *  Created on: 09.08.2018
 *      Author: Micha Mueller
 */

#ifndef SRC_SENSORBASE_H_
#define SRC_SENSORBASE_H_

11
#include <fstream>
12
#include <memory>
13
#include <string>
14
#include <limits.h>
15
16
17
#include <boost/lockfree/spsc_queue.hpp>

typedef struct {
18
	int64_t  value;
19
20
21
22
23
	uint64_t timestamp;
} reading_t;

class SensorBase {
public:
lu43jih's avatar
lu43jih committed
24
25
	static const size_t QUEUE_MAXLIMIT=1024;

26
	SensorBase(const std::string& name) :
27
28
29
30
31
32
33
34
35
36
37
38
39
40
			_name(name),
			_mqtt(""),
			_sinkPath(""),
			_skipConstVal(false),
			_cacheInterval(900000),
			_subsamplingFactor(1),
			_subsamplingIndex(0),
			_cacheSize(1),
			_cacheIndex(0),
			_cache(nullptr),
			_delta(false),
			_firstReading(true),
			_readingQueue(nullptr),
			_sinkFile(nullptr) {
41
42
43

		_latestValue.timestamp	= 0;
		_latestValue.value		= 0;
44
45
		_lastSentValue.timestamp= 0;
		_lastSentValue.value	= 0;
46
47
		_lastRawValue.timestamp = 0;
		_lastRawValue.value		= 0;
48
49
		_accumulator.timestamp  = 0;
		_accumulator.value		= 0;
50
51
	}

52
	SensorBase(const SensorBase& other) :
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
			_name(other._name),
			_mqtt(other._mqtt),
			_skipConstVal(other._skipConstVal),
			_cacheInterval(other._cacheInterval),
			_subsamplingFactor(other._subsamplingFactor),
			_subsamplingIndex(0),
			_cacheSize(other._cacheSize),
			_cacheIndex(0),
			_cache(nullptr),
			_delta(other._delta),
			_firstReading(true),
			_latestValue(other._latestValue),
			_lastRawValue(other._lastRawValue),
			_lastSentValue(other._lastSentValue),
			_accumulator(other._accumulator),
			_readingQueue(nullptr),
			_sinkFile(nullptr) {}
70
71
72
73
74
75

	virtual ~SensorBase() {}

	SensorBase& operator=(const SensorBase& other) {
		_name = other._name;
		_mqtt = other._mqtt;
76
		_skipConstVal = other._skipConstVal;
77
		_cacheInterval = other._cacheInterval;
78
79
		_subsamplingFactor = other._subsamplingFactor;
		_subsamplingIndex = 0;
80
		_cacheSize = other._cacheSize;
81
		_cacheIndex = 0;
82
		_cache.reset(nullptr);
83
		_delta = other._delta;
84
		_firstReading = true;
85
86
		_latestValue.timestamp	= other._latestValue.timestamp;
		_latestValue.value		= other._latestValue.value;
87
88
		_lastRawValue.timestamp = other._lastRawValue.timestamp;
		_lastRawValue.value		= other._lastRawValue.value;
89
90
91
92
		_lastSentValue.timestamp= other._lastSentValue.timestamp;
		_lastSentValue.value	= other._lastSentValue.value;
		_accumulator.timestamp  = other._accumulator.timestamp;
		_accumulator.value      = other._accumulator.value;
93
		_readingQueue.reset(nullptr);
94
		_sinkFile.reset(nullptr);
95
96

		return *this;
97
98
	}

99
	const bool 				isDelta()			const 	{ return _delta;}
100
101
	const std::string& 		getName() 			const	{ return _name; }
	const std::string&		getMqtt() 			const	{ return _mqtt; }
102
	const std::string&		getSinkPath() 		const	{ return _sinkPath; }
103
	bool					getSkipConstVal()	const	{ return _skipConstVal; }
104
	unsigned				getCacheSize()		const	{ return _cacheSize; }
Alessio Netti's avatar
Alessio Netti committed
105
	unsigned				getCacheInterval()	const	{ return _cacheInterval; }
106
	unsigned 				getSubsampling()	const   { return _subsamplingFactor; }
107
	const reading_t * const	getCache() 			const	{ return _cache.get(); }
Micha Mueller's avatar
Micha Mueller committed
108
	const reading_t&		getLatestValue()	const	{ return _latestValue; }
109

110
	void	setSkipConstVal(bool skipConstVal)				{ _skipConstVal = skipConstVal;	}
111
112
113
	void	setDelta(const bool delta) 						{ _delta = delta; }
	void	setName(const std::string& name, int cpuID=-1)	{ _name = formatName(name, cpuID); }
	void	setMqtt(const std::string& mqtt)				{ _mqtt = mqtt; }
114
	void 	setSinkPath(const std::string& path)			{ _sinkPath = path; }
115
	void 	setCacheInterval(unsigned cacheInterval)		{ _cacheInterval = cacheInterval; }
116
	void	setSubsampling(unsigned factor)					{ _subsamplingFactor = factor; }
117
118
119
120
121

	const std::size_t	getSizeOfReadingQueue() const { return _readingQueue->read_available(); }
	std::size_t 		popReadingQueue(reading_t *reads, std::size_t max) const	{ return _readingQueue->pop(reads, max); }
	void				pushReadingQueue(reading_t *reads, std::size_t count) const	{ _readingQueue->push(reads, count); }

122
123
	void initSensor(unsigned interval) {
		_cacheSize = _cacheInterval / interval + 1;
124
		if(!_cache) {
125
126
			_cache.reset(new reading_t[_cacheSize]);
			for(unsigned i = 0; i < _cacheSize; i++) {
127
128
129
130
				_cache[i] = _latestValue;	//_latestValue should equal (0,0) at this point
			}
		}
		if(!_readingQueue) {
lu43jih's avatar
lu43jih committed
131
			_readingQueue.reset(new boost::lockfree::spsc_queue<reading_t>(QUEUE_MAXLIMIT));
132
		}
133
134
135
136
137
		if(!_sinkFile && _sinkPath != "") {
			_sinkFile.reset(new std::ofstream(_sinkPath));
			if(!_sinkFile->is_open())
				_sinkFile.reset(nullptr);
		}
138
139
	}

Michael Ott's avatar
Michael Ott committed
140
141
	void storeReading(reading_t rawReading, double factor=1.0, unsigned long long maxValue=ULLONG_MAX) {
		reading_t reading = rawReading;
142
		if( _delta ) {
143
144
145
146
147
148
149
150
151
152
			if (!_firstReading) {
				if (rawReading.value < _lastRawValue.value)
					reading.value = (rawReading.value + (maxValue - _lastRawValue.value)) * factor;
				else
					reading.value = (rawReading.value - _lastRawValue.value) * factor;
			} else {
				_firstReading = false;
				_lastRawValue = rawReading;
				return;
			}
Michael Ott's avatar
Michael Ott committed
153
			_lastRawValue = rawReading;
154
155
		}
		else
Michael Ott's avatar
Michael Ott committed
156
			reading.value = rawReading.value * factor;
157

Alessio Netti's avatar
Alessio Netti committed
158
159
		if( _delta )
			// If in delta mode, _accumulator acts as a buffer, summing all deltas for the subsampling period
160
			_accumulator.value += reading.value;
Alessio Netti's avatar
Alessio Netti committed
161
		else
162
			_accumulator.value = reading.value;
Alessio Netti's avatar
Alessio Netti committed
163
164

		if (_subsamplingIndex++ % _subsamplingFactor == 0) {
165
166
167
168
169
170
171
			_accumulator.timestamp = reading.timestamp;
			//TODO: if sensor starts with values of 0, these won't be pushed. This should be fixed
			if( !(_skipConstVal && (_accumulator.value == _lastSentValue.value)) ) {
				_readingQueue->push(_accumulator);
				_lastSentValue = _accumulator;
			}
			// We reset the accumulator's value for the correct accumulation of deltas
Alessio Netti's avatar
Alessio Netti committed
172
			_accumulator.value = 0;
173
174
		}
		if (_sinkFile) {
175
176
177
178
			try {
				_sinkFile->seekp(0, std::ios::beg);
				*_sinkFile << reading.value << std::endl;
			} catch(const std::exception &e) { _sinkFile->close(); _sinkFile.reset(nullptr); }
179
		}
180

Michael Ott's avatar
Michael Ott committed
181
		_cache[_cacheIndex] = reading;
182
		_cacheIndex = (_cacheIndex + 1) % _cacheSize;
Michael Ott's avatar
Michael Ott committed
183
		_latestValue = reading;
184
185
	}

186
187
188
189
	int64_t getCacheOffset(int64_t t) {
		if( t < 0)
			return -1;
		// Converting from milliseconds to nanoseconds
Alessio Netti's avatar
Alessio Netti committed
190
		int64_t offset = ( ( (int64_t)_cacheSize * t ) / ( (int64_t)_cacheInterval * 1000000 ) ) + 1;
191
192
193
194
195
		if(offset > _cacheSize)
			return -1;
		return ( _cacheSize + _cacheIndex - offset ) % _cacheSize;
	}

Alessio Netti's avatar
Alessio Netti committed
196
197
	static std::string formatName(const std::string& name, int cpuID=-1) {return cpuID<0 ? name : "cpu" + std::to_string(cpuID) + "." + name;}

198
199
200
201
protected:

	std::string _name;
	std::string _mqtt;
202
	std::string _sinkPath;
203
	bool _skipConstVal;
204
	unsigned int _cacheInterval;
205
206
	unsigned int _subsamplingFactor;
	unsigned int _subsamplingIndex;
207
	unsigned int _cacheSize;
208
	unsigned int _cacheIndex;
209
	std::unique_ptr<reading_t[]> _cache;
210
	bool _delta;
211
	bool _firstReading;
212
	reading_t _latestValue;
213
	reading_t _lastRawValue;
214
	reading_t _lastSentValue;
Alessio Netti's avatar
Alessio Netti committed
215
	reading_t _accumulator;
216
	std::unique_ptr<boost::lockfree::spsc_queue<reading_t>> _readingQueue;
217
	std::unique_ptr<std::ofstream> _sinkFile;
218
219
};

220
221
222
//for better readability
using SBasePtr = std::shared_ptr<SensorBase>;

223
#endif /* SRC_SENSORBASE_H_ */