server/sensor.c

   1 /**
   2  * @file sensor.c
   3  * @brief Implementation of sensor thread
   4  * TODO: Finalise implementation
   5  */
   6
   7 #include "common.h"
   8 #include "sensor.h"
   9 #include "options.h"
  10 #include <math.h>
  11
  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"
  17 };
  18
  19 /**
  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
  24  */
  25 DataPoint * GetData(SensorId sensor_id, DataPoint * d)
  26 {
  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
  30
  31
  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
  35
  36         struct timeval t;
  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);
  39
  40         // Make time relative
  41         //d->time_stamp.tv_sec -= g_options.start_time.tv_sec;
  42         //d->time_stamp.tv_usec -= g_options.start_time.tv_usec;
  43
  44         switch (sensor_id)
  45         {
  46                 case ANALOG_TEST0:
  47                 {
  48                         //CheckSensor( sensor_id, *sensor value*);
  49
  50                         static int count = 0;
  51                         d->value = count++;
  52                         break;
  53                 }
  54                 case ANALOG_TEST1:
  55                         d->value = (double)(rand() % 100) / 100;
  56                         break;
  57
  58                 //TODO: For digital sensors, consider only updating when sensor is actually changed
  59                 case DIGITAL_TEST0:
  60                         d->value = t.tv_sec % 2;
  61                         break;
  62                 case DIGITAL_TEST1:
  63                         d->value = (t.tv_sec+1)%2;
  64                         break;
  65                 default:
  66                         Fatal("Unknown sensor id: %d", sensor_id);
  67                         break;
  68         }
  69         usleep(100000); // simulate delay in sensor polling
  70
  71         return d;
  72 }
  73
  74 /**
  75  * Checks the sensor data for unsafe or unexpected results
  76  * @param sensor_id - The ID of the sensor
  77  *
  78 *
  79 void CheckSensor( SensorId sensor_id)
  80 {
  81         switch (sensor_id)
  82         {
  83                 case ANALOG_TEST0:
  84                 {
  85                         if( *sensor value* > ANALOG_TEST0_SAFETY)
  86                         {
  87                                 LogEx(LOGERR, GetData, Sensor analog_test0 is above the safe value);
  88                         //new log function that stops actuators?
  89                         }
  90                         //Also include a warning level?
  91                         else if( *sensor value* > ANALOG_TEST0_WARN)
  92                         {
  93                                 LogEx(LOGWARN, GetData, Sensor analog_test0);
  94                         }
  95                 }
  96         }
  97
  98
  99 */
 100
 101
 102 /**
 103  * Destroy a sensor
 104  * @param s - Sensor to destroy
 105  */
 106 void Destroy(Sensor * s)
 107 {
 108         // Maybe move the binary file into long term file storage?
 109         fclose(s->file);
 110 }
 111
 112
 113
 114 /**
 115  * Initialise a sensor
 116  * @param s - Sensor to initialise
 117  */
 118 void Init(Sensor * s, int id)
 119 {
 120         s->write_index = 0;
 121         s->id = id;
 122
 123         #define FILENAMESIZE 3
 124         char filename[FILENAMESIZE];
 125         if (s->id >= pow(10, FILENAMESIZE))
 126         {
 127                 Fatal("Too many sensors! FILENAMESIZE is %d; increase it and recompile.", FILENAMESIZE);
 128         }
 129
 130         pthread_mutex_init(&(s->mutex), NULL);
 131
 132         sprintf(filename, "%d", s->id);
 133         unlink(filename); //TODO: Move old files somewhere
 134
 135         s->file = fopen(filename, "a+b"); // open binary file
 136         Log(LOGDEBUG, "Initialised sensor %d; binary file is \"%s\"", id, filename);
 137 }
 138
 139
 140 /**
 141  * Run the main sensor polling loop
 142  * @param arg - Cast to Sensor* - Sensor that the thread will handle
 143  * @returns NULL (void* required to use the function with pthreads)
 144  */
 145 void * Sensor_Main(void * arg)
 146 {
 147         Sensor * s = (Sensor*)(arg);
 148
 149         while (Thread_Runstate() == RUNNING) //TODO: Exit condition
 150         {
 151                 // The sensor will write data to a buffer until it is full
 152                 // Then it will open a file and dump the buffer to the end of it.
 153                 // Rinse and repeat
 154
 155                 // The reason I've added the buffer is because locks are expensive
 156                 // But maybe it's better to just write data straight to the file
 157                 // I'd like to do some tests by changing SENSOR_DATABUFSIZ
 158
 159                 while (s->write_index < SENSOR_DATABUFSIZ)
 160                 {
 161                         DataPoint * d = &(s->buffer[s->write_index]);
 162                         if (GetData(s->id, d) == NULL)
 163                         {
 164                                 Fatal("Error collecting data");
 165                         }
 166                         s->write_index += 1;
 167                 }
 168
 169                 //Log(LOGDEBUG, "Filled buffer");
 170
 171                 // CRITICAL SECTION (no threads should be able to read/write the file at the same time)
 172                 pthread_mutex_lock(&(s->mutex));
 173                         //TODO: Valgrind complains about this fseek: "Syscall param write(buf) points to uninitialised byte(s)"
 174                         //              Not sure why, but we should find out and fix it.
 175                         fseek(s->file, 0, SEEK_END);
 176                         int amount_written = fwrite(s->buffer, sizeof(DataPoint), SENSOR_DATABUFSIZ, s->file);
 177                         if (amount_written != SENSOR_DATABUFSIZ)
 178                         {
 179                                 Fatal("Wrote %d data points and expected to write %d to \"%s\" - %s", amount_written, SENSOR_DATABUFSIZ, strerror(errno));
 180                         }
 181                         //Log(LOGDEBUG, "Wrote %d data points for sensor %d", amount_written, s->id);
 182                 pthread_mutex_unlock(&(s->mutex));
 183                 // End of critical section
 184
 185                 s->write_index = 0; // reset position in buffer
 186
 187         }
 188         Log(LOGDEBUG, "Thread for sensor %d exits", s->id);
 189         return NULL;
 190 }
 191
 192 /**
 193  * Fill buffer with most recent sensor data
 194  * @param s - Sensor to use
 195  * @param buffer - Buffer to fill
 196  * @param bufsiz - Size of buffer to fill
 197  * @returns The number of DataPoints actually read
 198  */
 199 int Sensor_Query(Sensor * s, DataPoint * buffer, int bufsiz)
 200 {
 201         int amount_read = 0;
 202         //CRITICAL SECTION (Don't access file while sensor thread is writing to it!)
 203         pthread_mutex_lock(&(s->mutex));
 204
 205                 fseek(s->file, -bufsiz*sizeof(DataPoint), SEEK_END);
 206                 amount_read = fread(buffer, sizeof(DataPoint), bufsiz, s->file);
 207                 //Log(LOGDEBUG, "Read %d data points", amount_read);
 208         pthread_mutex_unlock(&(s->mutex));
 209         return amount_read;
 210 }
 211
 212 /**
 213  * Get a Sensor given an ID string
 214  * @param id_str ID string
 215  * @returns Sensor* identified by the string; NULL on error
 216  */
 217 Sensor * Sensor_Identify(const char * id_str)
 218 {
 219         char * end;
 220         // Parse string as integer
 221         int id = strtol(id_str, &end, 10);
 222         if (*end != '\0')
 223         {
 224                 return NULL;
 225         }
 226         // Bounds check
 227         if (id < 0 || id >= NUMSENSORS)
 228                 return NULL;
 229
 230
 231         Log(LOGDEBUG, "Sensor \"%s\" identified", g_sensor_names[id]);
 232         return g_sensors+id;
 233 }
 234
 235 /**
 236  * Handle a request to the sensor module
 237  * @param context - The context to work in
 238  * @param params - Parameters passed
 239  */
 240 void Sensor_Handler(FCGIContext *context, char * params)
 241 {
 242         DataPoint buffer[SENSOR_QUERYBUFSIZ];
 243         StatusCodes status = STATUS_OK;
 244
 245         enum {DEFAULT, DUMP} operation = DEFAULT;
 246
 247         const char * key; const char * value;
 248
 249         Sensor * sensor = NULL;
 250
 251         while ((params = FCGI_KeyPair(params, &key, &value)) != NULL)
 252         {
 253                 Log(LOGDEBUG, "Got key=%s and value=%s", key, value);
 254                 if (strcmp(key, "id") == 0)
 255                 {
 256                         if (sensor != NULL)
 257                         {
 258                                 Log(LOGERR, "Only one sensor id should be specified");
 259                                 status = STATUS_ERROR;
 260                                 break;
 261                         }
 262                         if (*value == '\0')
 263                         {
 264                                 Log(LOGERR, "No id specified.");
 265                                 status = STATUS_ERROR;
 266                                 break;
 267                         }
 268
 269                         sensor = Sensor_Identify(value);
 270                         if (sensor == NULL)
 271                         {
 272                                 Log(LOGERR, "Invalid sensor id: %s", value);
 273                                 status = STATUS_ERROR;
 274                                 break;
 275                         }
 276                 }
 277                 else if (strcmp(key, "dump") == 0)
 278                 {
 279                         if (operation != DEFAULT)
 280                         {
 281                                 Log(LOGERR, "Operation already specified!");
 282                                 status = STATUS_ERROR;
 283                                 break;
 284                         }
 285                         operation = DUMP;
 286                 }
 287                 else
 288                 {
 289                         Log(LOGERR, "Unknown key \"%s\" (value = %s)", key, value);
 290                         status = STATUS_ERROR;
 291                         break;
 292                 }
 293         }
 294
 295         if (status != STATUS_ERROR && sensor == NULL)
 296         {
 297                 Log(LOGERR, "No valid sensor id given");
 298                 status = STATUS_ERROR;
 299         }
 300
 301         if (status == STATUS_ERROR)
 302         {
 303                 FCGI_RejectJSON(context, "Invalid input parameters");
 304                 return;
 305         }
 306
 307         switch (operation)
 308         {
 309                 case DUMP:
 310                 {
 311                         //Force download with content-disposition
 312                         FCGI_PrintRaw("Content-type: text/plain\r\n"
 313                                 "Content-disposition: attachment;filename=%d.csv\r\n\r\n",
 314                                 sensor->id);
 315                         //CRITICAL SECTION
 316                         pthread_mutex_lock(&(sensor->mutex));
 317                                 fseek(sensor->file, 0, SEEK_SET);
 318                                 int amount_read = 0;
 319                                 do
 320                                 {
 321                                         amount_read = fread(buffer, sizeof(DataPoint), SENSOR_QUERYBUFSIZ, sensor->file);
 322                                         for (int i = 0; i < amount_read; ++i)
 323                                         {
 324                                                 FCGI_PrintRaw("%f\t%f\n", buffer[i].time_stamp, buffer[i].value);
 325                                         }
 326
 327                                 }
 328                                 while (amount_read == SENSOR_QUERYBUFSIZ);
 329                         pthread_mutex_unlock(&(sensor->mutex));
 330                         // end critical section
 331                         break;
 332                 }
 333                 default:
 334                 {
 335                         FCGI_BeginJSON(context, status);
 336                         FCGI_JSONPair(key, value); // should spit back sensor ID
 337                         //Log(LOGDEBUG, "Call Sensor_Query...");
 338                         int amount_read = Sensor_Query(sensor, buffer, SENSOR_QUERYBUFSIZ);
 339                         //Log(LOGDEBUG, "Read %d DataPoints", amount_read);
 340                         //Log(LOGDEBUG, "Produce JSON response");
 341                         FCGI_JSONKey("data");
 342                         FCGI_JSONValue("[");
 343                         for (int i = 0; i < amount_read; ++i)
 344                         {
 345                                 //TODO: Consider; is it better to give both tv_sec and tv_usec to the client seperately, instead of combining here?
 346
 347                                 FCGI_JSONValue("[%f, %f]", buffer[i].time_stamp, buffer[i].value);
 348                                 if (i+1 < amount_read)
 349                                         FCGI_JSONValue(",");
 350                         }
 351                         FCGI_JSONValue("]");
 352                         //Log(LOGDEBUG, "Done producing JSON response");
 353                         FCGI_EndJSON();
 354                         break;
 355                 }
 356         }
 357 }
 358
 359 /**
 360  * Setup Sensors, start Sensor polling thread(s)
 361  */
 362 void Sensor_Spawn()
 363 {
 364         // start sensor threads
 365         for (int i = 0; i < NUMSENSORS; ++i)
 366         {
 367                 Init(g_sensors+i, i);
 368                 pthread_create(&(g_sensors[i].thread), NULL, Sensor_Main, (void*)(g_sensors+i));
 369         }
 370 }
 371
 372 /**
 373  * Quit Sensor loops
 374  */
 375 void Sensor_Join()
 376 {
 377         if (!Thread_Runstate())
 378         {
 379                 Fatal("This function should not be called before Thread_QuitProgram");
 380         }
 381         for (int i = 0; i < NUMSENSORS; ++i)
 382         {
 383                 pthread_join(g_sensors[i].thread, NULL);
 384                 Destroy(g_sensors+i);
 385         }
 386 }