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");
140 //Log(LOGDEBUG, "Filled buffer");
142 // CRITICAL SECTION (no threads should be able to read/write the file at the same time)
143 pthread_mutex_lock(&(s->mutex));
144 //TODO: Valgrind complains about this fseek: "Syscall param write(buf) points to uninitialised byte(s)"
145 // Not sure why, but we should find out and fix it.
146 fseek(s->file, 0, SEEK_END);
147 int amount_written = fwrite(s->buffer, sizeof(DataPoint), SENSOR_DATABUFSIZ, s->file);
148 if (amount_written != SENSOR_DATABUFSIZ)
150 Fatal("Wrote %d data points and expected to write %d to \"%s\" - %s", amount_written, SENSOR_DATABUFSIZ, strerror(errno));
152 //Log(LOGDEBUG, "Wrote %d data points for sensor %d", amount_written, s->id);
153 pthread_mutex_unlock(&(s->mutex));
154 // End of critical section
156 s->write_index = 0; // reset position in buffer
159 Log(LOGDEBUG, "Thread for sensor %d exits", s->id);
164 * Fill buffer with most recent sensor data
165 * @param s - Sensor to use
166 * @param buffer - Buffer to fill
167 * @param bufsiz - Size of buffer to fill
168 * @returns The number of DataPoints actually read
170 int Sensor_Query(Sensor * s, DataPoint * buffer, int bufsiz)
173 //CRITICAL SECTION (Don't access file while sensor thread is writing to it!)
174 pthread_mutex_lock(&(s->mutex));
176 fseek(s->file, -bufsiz*sizeof(DataPoint), SEEK_END);
177 amount_read = fread(buffer, sizeof(DataPoint), bufsiz, s->file);
178 //Log(LOGDEBUG, "Read %d data points", amount_read);
179 pthread_mutex_unlock(&(s->mutex));
184 * Get a Sensor given an ID string
185 * @param id_str ID string
186 * @returns Sensor* identified by the string; NULL on error
188 Sensor * Sensor_Identify(const char * id_str)
191 // Parse string as integer
192 int id = strtol(id_str, &end, 10);
198 if (id < 0 || id >= NUMSENSORS)
202 Log(LOGDEBUG, "Sensor \"%s\" identified", g_sensor_names[id]);
207 * Handle a request to the sensor module
208 * @param context - The context to work in
209 * @param params - Parameters passed
211 void Sensor_Handler(FCGIContext *context, char * params)
213 DataPoint buffer[SENSOR_QUERYBUFSIZ];
214 StatusCodes status = STATUS_OK;
216 enum {DEFAULT, DUMP} operation = DEFAULT;
218 const char * key; const char * value;
220 Sensor * sensor = NULL;
222 while ((params = FCGI_KeyPair(params, &key, &value)) != NULL)
224 Log(LOGDEBUG, "Got key=%s and value=%s", key, value);
225 if (strcmp(key, "id") == 0)
229 Log(LOGERR, "Only one sensor id should be specified");
230 status = STATUS_ERROR;
235 Log(LOGERR, "No id specified.");
236 status = STATUS_ERROR;
240 sensor = Sensor_Identify(value);
243 Log(LOGERR, "Invalid sensor id: %s", value);
244 status = STATUS_ERROR;
248 else if (strcmp(key, "dump") == 0)
250 if (operation != DEFAULT)
252 Log(LOGERR, "Operation already specified!");
253 status = STATUS_ERROR;
260 Log(LOGERR, "Unknown key \"%s\" (value = %s)", key, value);
261 status = STATUS_ERROR;
266 if (status != STATUS_ERROR && sensor == NULL)
268 Log(LOGERR, "No valid sensor id given");
269 status = STATUS_ERROR;
272 if (status == STATUS_ERROR)
274 FCGI_RejectJSON(context, "Invalid input parameters");
282 //Force download with content-disposition
283 FCGI_PrintRaw("Content-type: text/plain\r\n"
284 "Content-disposition: attachment;filename=%d.csv\r\n\r\n",
287 pthread_mutex_lock(&(sensor->mutex));
288 fseek(sensor->file, 0, SEEK_SET);
292 amount_read = fread(buffer, sizeof(DataPoint), SENSOR_QUERYBUFSIZ, sensor->file);
293 for (int i = 0; i < amount_read; ++i)
295 FCGI_PrintRaw("%f\t%f\n", buffer[i].time_stamp, buffer[i].value);
299 while (amount_read == SENSOR_QUERYBUFSIZ);
300 pthread_mutex_unlock(&(sensor->mutex));
301 // end critical section
306 FCGI_BeginJSON(context, status);
307 FCGI_JSONPair(key, value); // should spit back sensor ID
308 //Log(LOGDEBUG, "Call Sensor_Query...");
309 int amount_read = Sensor_Query(sensor, buffer, SENSOR_QUERYBUFSIZ);
310 //Log(LOGDEBUG, "Read %d DataPoints", amount_read);
311 //Log(LOGDEBUG, "Produce JSON response");
312 FCGI_JSONKey("data");
314 for (int i = 0; i < amount_read; ++i)
316 //TODO: Consider; is it better to give both tv_sec and tv_usec to the client seperately, instead of combining here?
318 FCGI_JSONValue("[%f, %f]", buffer[i].time_stamp, buffer[i].value);
319 if (i+1 < amount_read)
323 //Log(LOGDEBUG, "Done producing JSON response");
331 * Setup Sensors, start Sensor polling thread(s)
335 // start sensor threads
336 for (int i = 0; i < NUMSENSORS; ++i)
338 Init(g_sensors+i, i);
339 pthread_create(&(g_sensors[i].thread), NULL, Sensor_Main, (void*)(g_sensors+i));
348 if (!Thread_Runstate())
350 Fatal("This function should not be called before Thread_QuitProgram");
352 for (int i = 0; i < NUMSENSORS; ++i)
354 pthread_join(g_sensors[i].thread, NULL);
355 Destroy(g_sensors+i);