3 * @brief Implementation of sensor thread
4 * TODO: Finalise implementation
12 /** Array of sensors, initialised by Sensor_Init **/
13 static Sensor g_sensors[NUMSENSORS]; //global to this file
14 const char * g_sensor_names[NUMSENSORS] = {
15 "analog_test0", "analog_test1",
16 "digital_test0", "digital_test1"
20 * Read a data value from a sensor; block until value is read
21 * @param sensor_id - The ID of the sensor
22 * @param d - DataPoint to set
23 * @returns NULL for digital sensors when data is unchanged, otherwise d
25 DataPoint * GetData(SensorId sensor_id, DataPoint * d)
27 // switch based on the sensor_id at the moment for testing;
28 // might be able to just directly access ADC from sensor_id?
29 //TODO: Implement for real sensors
32 //TODO: We should ensure the time is *never* allowed to change on the server if we use gettimeofday
33 // Another way people might think of getting the time is to count CPU cycles with clock()
34 // But this will not work because a) CPU clock speed may change on some devices (RPi?) and b) It counts cycles used by all threads
37 gettimeofday(&t, NULL);
38 d->time_stamp = (t.tv_sec - g_options.start_time.tv_sec) + 1e-6*(t.tv_usec - g_options.start_time.tv_usec);
41 //d->time_stamp.tv_sec -= g_options.start_time.tv_sec;
42 //d->time_stamp.tv_usec -= g_options.start_time.tv_usec;
53 d->value = (double)(rand() % 100) / 100;
56 //TODO: For digital sensors, consider only updating when sensor is actually changed
58 d->value = t.tv_sec % 2;
61 d->value = (t.tv_sec+1)%2;
64 Fatal("Unknown sensor id: %d", sensor_id);
67 usleep(100000); // simulate delay in sensor polling
75 * @param s - Sensor to destroy
77 void Destroy(Sensor * s)
79 // Maybe move the binary file into long term file storage?
87 * @param s - Sensor to initialise
89 void Init(Sensor * s, int id)
94 #define FILENAMESIZE 3
95 char filename[FILENAMESIZE];
96 if (s->id >= pow(10, FILENAMESIZE))
98 Fatal("Too many sensors! FILENAMESIZE is %d; increase it and recompile.", FILENAMESIZE);
101 pthread_mutex_init(&(s->mutex), NULL);
103 sprintf(filename, "%d", s->id);
104 unlink(filename); //TODO: Move old files somewhere
106 s->file = fopen(filename, "a+b"); // open binary file
107 Log(LOGDEBUG, "Initialised sensor %d; binary file is \"%s\"", id, filename);
112 * Run the main sensor polling loop
113 * @param arg - Cast to Sensor* - Sensor that the thread will handle
114 * @returns NULL (void* required to use the function with pthreads)
116 void * Sensor_Main(void * arg)
118 Sensor * s = (Sensor*)(arg);
120 while (Thread_Runstate() == RUNNING) //TODO: Exit condition
122 // The sensor will write data to a buffer until it is full
123 // Then it will open a file and dump the buffer to the end of it.
126 // The reason I've added the buffer is because locks are expensive
127 // But maybe it's better to just write data straight to the file
128 // I'd like to do some tests by changing SENSOR_DATABUFSIZ
130 while (s->write_index < SENSOR_DATABUFSIZ)
132 DataPoint * d = &(s->buffer[s->write_index]);
133 if (GetData(s->id, d) == NULL)
135 Fatal("Error collecting data");
138 //TODO: s->points_read not used?
141 //Log(LOGDEBUG, "Filled buffer");
143 // CRITICAL SECTION (no threads should be able to read/write the file at the same time)
144 pthread_mutex_lock(&(s->mutex));
145 //TODO: Valgrind complains about this fseek: "Syscall param write(buf) points to uninitialised byte(s)"
146 // Not sure why, but we should find out and fix it.
147 fseek(s->file, 0, SEEK_END);
148 int amount_written = fwrite(s->buffer, sizeof(DataPoint), SENSOR_DATABUFSIZ, s->file);
149 if (amount_written != SENSOR_DATABUFSIZ)
151 Fatal("Wrote %d data points and expected to write %d to \"%s\" - %s", amount_written, SENSOR_DATABUFSIZ, strerror(errno));
153 s->points_stored += amount_written;
154 //Log(LOGDEBUG, "Wrote %d data points for sensor %d", amount_written, s->id);
155 pthread_mutex_unlock(&(s->mutex));
156 // End of critical section
158 s->write_index = 0; // reset position in buffer
161 Log(LOGDEBUG, "Thread for sensor %d exits", s->id);
166 * Fill buffer with most recent sensor data
167 * @param s - Sensor to use
168 * @param buffer - Buffer to fill
169 * @param bufsiz - Size of buffer to fill
170 * @returns The number of DataPoints actually read
172 int Sensor_Query(Sensor * s, DataPoint * buffer, int bufsiz)
175 //CRITICAL SECTION (Don't access file while sensor thread is writing to it!)
176 pthread_mutex_lock(&(s->mutex));
178 fseek(s->file, -bufsiz*sizeof(DataPoint), SEEK_END);
179 amount_read = fread(buffer, sizeof(DataPoint), bufsiz, s->file);
180 //Log(LOGDEBUG, "Read %d data points", amount_read);
181 pthread_mutex_unlock(&(s->mutex));
186 * Get a Sensor given an ID string
187 * @param id_str ID string
188 * @returns Sensor* identified by the string; NULL on error
190 Sensor * Sensor_Identify(const char * id_str)
193 // Parse string as integer
194 int id = strtol(id_str, &end, 10);
200 if (id < 0 || id >= NUMSENSORS)
204 Log(LOGDEBUG, "Sensor \"%s\" identified", g_sensor_names[id]);
211 * - from < 0: From beginning of file, else from that point onwards (0-indexed)
212 * - count < 0: All points available, else *at most* that many points
214 * - from < 0: From the end of file (last available points), else from that point onwards (0-indexed)
215 * - count < 0: Default buffer size (SENSOR_QUERYBUFSIZ), else *at most* that many points
217 void Sensor_Handler2(FCGIContext *context, char *params)
219 const char *key, *value;
220 int id = -1, from = -1, count = -1;
224 while ((params = FCGI_KeyPair(params, &key, &value))) {
225 if (!strcmp(key, "id") && *value) {
227 id = strtol(value, &end, 10);
230 } else if (!strcmp(key, "dump")) {
232 } else if (!strcmp(key, "from") && *value) {
233 from = strtol(value, NULL, 10);
234 } else if (!strcmp(key, "count") && *value) {
235 count = strtol(value, NULL, 10);
239 if (id < 0 || id >= NUMSENSORS) {
240 FCGI_RejectJSON(context, "Invalid sensor id specified.");
244 Sensor *sensor = &g_sensors[id];
245 DataPoint buffer[SENSOR_QUERYBUFSIZ];
246 int amount_read = 0, total = 0;
249 pthread_mutex_lock(&(sensor->mutex));
250 if (from >= sensor->points_stored) {
251 FCGI_RejectJSONEx(context, STATUS_OUTOFRANGE, "Invalid range specified.");
253 from = (from < 0) ? 0 : from;
254 count = (count < 0) ? sensor->points_stored : count;
256 FCGI_PrintRaw("Content-type: text/plain\r\n"
257 "Content-disposition: attachment;filename=%d.csv\r\n\r\n", id);
259 fseek(sensor->file, sizeof(DataPoint) * from, SEEK_SET);
260 //Force download with content-disposition
262 amount_read = fread(buffer, sizeof(DataPoint), SENSOR_QUERYBUFSIZ, sensor->file);
263 for (int i = 0; i < amount_read && total < count; i++, total++) {
264 FCGI_PrintRaw("%f\t%f\n", buffer[i].time_stamp, buffer[i].value);
266 } while (amount_read > 0 && total < count);
268 count = (count < 0) ? SENSOR_QUERYBUFSIZ : count;
270 from = sensor->points_stored - count;
274 fseek(sensor->file, sizeof(DataPoint) * from, SEEK_SET);
276 FCGI_BeginJSON(context, STATUS_OK);
277 FCGI_JSONLong("id", id);
278 FCGI_JSONKey("data");
280 while (total < count) {
281 amount_read = fread(buffer, sizeof(DataPoint), SENSOR_QUERYBUFSIZ, sensor->file);
282 if (amount_read > 0) {
283 FCGI_JSONValue("[%f, %f]", buffer[0].time_stamp, buffer[0].value);
285 for (int i = 1; i < amount_read && total < count; i++, total++)
286 FCGI_JSONValue(", [%f, %f]", buffer[i].time_stamp, buffer[i].value);
293 FCGI_JSONLong("total_points", sensor->points_stored);
294 FCGI_JSONLong("next_point", from + total);
297 pthread_mutex_unlock(&(sensor->mutex));
298 //End critical section
302 * Handle a request to the sensor module
303 * @param context - The context to work in
304 * @param params - Parameters passed
306 void Sensor_Handler(FCGIContext *context, char * params)
308 DataPoint buffer[SENSOR_QUERYBUFSIZ];
309 StatusCodes status = STATUS_OK;
311 enum {DEFAULT, DUMP} operation = DEFAULT;
313 const char * key; const char * value;
315 Sensor * sensor = NULL;
317 while ((params = FCGI_KeyPair(params, &key, &value)) != NULL)
319 Log(LOGDEBUG, "Got key=%s and value=%s", key, value);
320 if (strcmp(key, "id") == 0)
324 Log(LOGERR, "Only one sensor id should be specified");
325 status = STATUS_ERROR;
330 Log(LOGERR, "No id specified.");
331 status = STATUS_ERROR;
335 sensor = Sensor_Identify(value);
338 Log(LOGERR, "Invalid sensor id: %s", value);
339 status = STATUS_ERROR;
343 else if (strcmp(key, "dump") == 0)
345 if (operation != DEFAULT)
347 Log(LOGERR, "Operation already specified!");
348 status = STATUS_ERROR;
355 Log(LOGERR, "Unknown key \"%s\" (value = %s)", key, value);
356 status = STATUS_ERROR;
361 if (status != STATUS_ERROR && sensor == NULL)
363 Log(LOGERR, "No valid sensor id given");
364 status = STATUS_ERROR;
367 if (status == STATUS_ERROR)
369 FCGI_RejectJSON(context, "Invalid input parameters");
377 //Force download with content-disposition
378 FCGI_PrintRaw("Content-type: text/plain\r\n"
379 "Content-disposition: attachment;filename=%d.csv\r\n\r\n",
382 pthread_mutex_lock(&(sensor->mutex));
383 fseek(sensor->file, 0, SEEK_SET);
387 amount_read = fread(buffer, sizeof(DataPoint), SENSOR_QUERYBUFSIZ, sensor->file);
388 for (int i = 0; i < amount_read; ++i)
390 FCGI_PrintRaw("%f\t%f\n", buffer[i].time_stamp, buffer[i].value);
394 while (amount_read == SENSOR_QUERYBUFSIZ);
395 pthread_mutex_unlock(&(sensor->mutex));
396 // end critical section
401 FCGI_BeginJSON(context, status);
402 FCGI_JSONPair(key, value); // should spit back sensor ID
403 //Log(LOGDEBUG, "Call Sensor_Query...");
404 int amount_read = Sensor_Query(sensor, buffer, SENSOR_QUERYBUFSIZ);
405 //Log(LOGDEBUG, "Read %d DataPoints", amount_read);
406 //Log(LOGDEBUG, "Produce JSON response");
407 FCGI_JSONKey("data");
409 for (int i = 0; i < amount_read; ++i)
411 //TODO: Consider; is it better to give both tv_sec and tv_usec to the client seperately, instead of combining here?
413 FCGI_JSONValue("[%f, %f]", buffer[i].time_stamp, buffer[i].value);
414 if (i+1 < amount_read)
418 //Log(LOGDEBUG, "Done producing JSON response");
426 * Setup Sensors, start Sensor polling thread(s)
430 // start sensor threads
431 for (int i = 0; i < NUMSENSORS; ++i)
433 Init(g_sensors+i, i);
434 pthread_create(&(g_sensors[i].thread), NULL, Sensor_Main, (void*)(g_sensors+i));
443 if (!Thread_Runstate())
445 Fatal("This function should not be called before Thread_QuitProgram");
447 for (int i = 0; i < NUMSENSORS; ++i)
449 pthread_join(g_sensors[i].thread, NULL);
450 Destroy(g_sensors+i);