X-Git-Url: https://git.ucc.asn.au/?a=blobdiff_plain;ds=sidebyside;f=server%2Fsensor.c;h=f2c73c5ddb81af983fd6d135c7178d9b43f9d077;hb=a0c96fa8feb1938db84d95b14c320e6c7511c282;hp=3d5aec7b735afa153fcd2c3297d13f89b2b662a8;hpb=5d9c02bb1ca8a99e08800be88af9a9e02caf1041;p=matches%2FMCTX3420.git

diff --git a/server/sensor.c b/server/sensor.c
index 3d5aec7..f2c73c5 100644
--- a/server/sensor.c
+++ b/server/sensor.c
@@ -35,7 +35,7 @@ DataPoint * GetData(SensorId sensor_id, DataPoint * d)
 	
 	struct timeval t;
 	gettimeofday(&t, NULL);
-	d->time_stamp = (t.tv_sec - g_options.start_time.tv_sec) + 1e-6*(t.tv_usec - g_options.start_time.tv_usec);
+	d->time_stamp = TIMEVAL_DIFF(t, g_options.start_time);
 
 	// Make time relative
 	//d->time_stamp.tv_sec -= g_options.start_time.tv_sec;
@@ -45,6 +45,8 @@ DataPoint * GetData(SensorId sensor_id, DataPoint * d)
 	{
 		case ANALOG_TEST0:
 		{
+			//CheckSensor( sensor_id, *sensor value*); 
+		
 			static int count = 0;
 			d->value = count++;
 			break;
@@ -69,6 +71,33 @@ DataPoint * GetData(SensorId sensor_id, DataPoint * d)
 	return d;
 }
 
+/**
+ * Checks the sensor data for unsafe or unexpected results 
+ * @param sensor_id - The ID of the sensor
+ *
+*
+void CheckSensor( SensorId sensor_id)
+{
+	switch (sensor_id)
+	{
+		case ANALOG_TEST0:
+		{
+			if( *sensor value* > ANALOG_TEST0_SAFETY)
+			{
+				LogEx(LOGERR, GetData, Sensor analog_test0 is above the safe value);
+			//new log function that stops actuators?
+			}
+			//Also include a warning level?
+			else if( *sensor value* > ANALOG_TEST0_WARN)
+			{
+				LogEx(LOGWARN, GetData, Sensor analog_test0);	
+			}
+		}
+	}
+		
+	
+*/	
+
 
 /**
  * Destroy a sensor
@@ -90,6 +119,8 @@ void Init(Sensor * s, int id)
 {
 	s->write_index = 0;
 	s->id = id;
+	s->points_written = 0;
+	s->points_read = 0;
 
 	#define FILENAMESIZE 3
 	char filename[FILENAMESIZE];
@@ -135,7 +166,6 @@ void * Sensor_Main(void * arg)
 				Fatal("Error collecting data");
 			}
 			s->write_index += 1;
-			//TODO: s->points_read not used?
 		}
 
 		//Log(LOGDEBUG, "Filled buffer");
@@ -150,7 +180,7 @@ void * Sensor_Main(void * arg)
 			{
 				Fatal("Wrote %d data points and expected to write %d to \"%s\" - %s", amount_written, SENSOR_DATABUFSIZ, strerror(errno));
 			}
-			s->points_stored += amount_written;
+			s->points_written += amount_written;
 			//Log(LOGDEBUG, "Wrote %d data points for sensor %d", amount_written, s->id);
 		pthread_mutex_unlock(&(s->mutex));
 		// End of critical section
@@ -162,8 +192,135 @@ void * Sensor_Main(void * arg)
 	return NULL; 
 }
 
+/**
+ * Get position in a binary sensor file with a timestamp using a binary search
+ * @param s - Sensor to use
+ * @param time_stamp - Timestamp
+ * @param count - If not NULL, used to provide number of searches required
+ * @param found - If not NULL, set to the closest DataPoint
+ * @returns Integer giving the *closest* index in the file
+ * TODO: Refactor or replace?
+ */
+int FindTime(Sensor * s, double time_stamp, int * count, DataPoint * found)
+{
+	DataPoint d;
+
+	int lower = 0;
+	int upper = s->points_written - 1;
+	int index = 0;
+	if (count != NULL)
+		*count = 0;	
+
+	while (upper - lower > 1)
+	{
+		index = lower + ((upper - lower)/2);
+
+		// Basically anything with fseek is critical; if we don't make it critical the sensor thread may alter data at a random point in the file!
+		// CRITICAL SECTION (May need to rethink how this is done, but I can't see how to do it without fseek :S)
+		// Regarding the suggestion that we have 2 file pointers; one for reading and one for writing:
+		// That seems like it will work... but we will have to be very careful and test it first
+		pthread_mutex_lock(&s->mutex);
+			fseek(s->file, index*sizeof(DataPoint), SEEK_SET);
+			int amount_read = fread(&d, sizeof(DataPoint), 1, s->file);
+		pthread_mutex_unlock(&s->mutex);
+		
+		if (amount_read != 1)
+		{
+			Fatal("Couldn't read single data point from sensor %d", s->id);
+		}
+
+		if (d.time_stamp > time_stamp)
+		{
+			upper = index;
+		}
+		else if (d.time_stamp < time_stamp)
+		{
+			lower = index;
+		}
+		if (count != NULL)
+			*count += 1;
+	}
+
+	if (found != NULL)
+		*found = d;
+
+	return index;
+	
+}
+
+/**
+ * Print sensor data between two indexes in the file, using a given format
+ * @param s - Sensor to use
+ * @param start - Start index
+ * @param end - End index
+ * @param output_type - JSON, CSV or TSV output format
+ */
+void PrintData(Sensor * s, int start, int end, OutputType output_type)
+{
+	DataPoint buffer[SENSOR_QUERYBUFSIZ];
+	int index = start;
+
+	if (output_type == JSON)
+	{
+		FCGI_JSONValue("[");
+	}
+
+
+	while (index < end)
+	{
+		int to_read = end - index;
+		if (to_read > SENSOR_QUERYBUFSIZ)
+		{
+			to_read = SENSOR_QUERYBUFSIZ;
+		}
+
+		int amount_read = 0;
+		// CRITICAL SECTION
+		pthread_mutex_lock(&(s->mutex));
+
+			fseek(s->file, index*sizeof(DataPoint), SEEK_SET);
+			amount_read = fread(buffer, sizeof(DataPoint), to_read, s->file);
+
+		pthread_mutex_unlock(&(s->mutex));
+		// End critical section
+
+		if (amount_read != to_read)
+		{
+			Fatal("Failed to read %d DataPoints from sensor %d; read %d instead", to_read, s->id, amount_read);
+		}
+
+		// Print the data
+		for (int i = 0; i < amount_read; ++i)
+		{
+			//TODO: Reformat?
+			switch (output_type)
+			{
+				case JSON:
+					FCGI_JSONValue("[%f, %f]", buffer[i].time_stamp, buffer[i].value);
+					if (i+1 < amount_read)
+						FCGI_JSONValue(",");
+					break;
+				case CSV:
+					FCGI_PrintRaw("%f,%f\n", buffer[i].time_stamp, buffer[i].value);
+					break;
+				case TSV:
+				default:
+					FCGI_PrintRaw("%f\t%f\n", buffer[i].time_stamp, buffer[i].value);
+					break;
+			}
+		}
+		index += amount_read;
+	}
+
+	if (output_type == JSON)
+	{
+		FCGI_JSONValue("]");
+	}
+}
+
 /**
  * Fill buffer with most recent sensor data
+ * TODO: This may be obselete; remove?
  * @param s - Sensor to use
  * @param buffer - Buffer to fill
  * @param bufsiz - Size of buffer to fill
@@ -205,115 +362,47 @@ Sensor * Sensor_Identify(const char * id_str)
 	return g_sensors+id;
 }
 
-/*
- * Behaviour: 
- * Dump true:
- * - from < 0: From beginning of file, else from that point onwards (0-indexed)
- * - count < 0: All points available, else *at most* that many points
- * Dump false:
- * - from < 0: From the end of file (last available points), else from that point onwards (0-indexed)
- * - count < 0: Default buffer size (SENSOR_QUERYBUFSIZ), else *at most* that many points
- */
-void Sensor_Handler2(FCGIContext *context, char *params)
-{
-	const char *key, *value;
-	int id = -1, from = -1, count = -1;
-	bool dump = false;
-
-	//Lazy checking
-	while ((params = FCGI_KeyPair(params, &key, &value))) {
-		if (!strcmp(key, "id") && *value) {
-			char *end;
-			id = strtol(value, &end, 10);
-			if (*end != '\0')
-				id = -1;
-		} else if (!strcmp(key, "dump")) {
-			dump = !dump;
-		} else if (!strcmp(key, "from") && *value) {
-			from = strtol(value, NULL, 10);
-		} else if (!strcmp(key, "count") && *value) {
-			count = strtol(value, NULL, 10);
-		}
-	}
-
-	if (id < 0 || id >= NUMSENSORS) {
-		FCGI_RejectJSON(context, "Invalid sensor id specified.");
-		return;
-	}
-	
-	Sensor *sensor = &g_sensors[id];
-	DataPoint buffer[SENSOR_QUERYBUFSIZ];
-	int amount_read = 0, total = 0;
-	
-	//Critical section
-	pthread_mutex_lock(&(sensor->mutex));
-	if (from >= sensor->points_stored) {
-		FCGI_RejectJSONEx(context, STATUS_OUTOFRANGE, "Invalid range specified.");
-	} else if (dump) {
-		from =  (from < 0) ? 0 : from;
-		count = (count < 0) ? sensor->points_stored : count;
-
-		FCGI_PrintRaw("Content-type: text/plain\r\n"
-			"Content-disposition: attachment;filename=%d.csv\r\n\r\n", id);
-
-		fseek(sensor->file, sizeof(DataPoint) * from, SEEK_SET);
-		//Force download with content-disposition
-		do {
-			amount_read = fread(buffer, sizeof(DataPoint), SENSOR_QUERYBUFSIZ, sensor->file);
-			for (int i = 0; i < amount_read && total < count; i++, total++) {
-				FCGI_PrintRaw("%f\t%f\n", buffer[i].time_stamp, buffer[i].value);
-			}
-		} while (amount_read > 0 && total < count);
-	} else {
-		count = (count < 0) ? SENSOR_QUERYBUFSIZ : count;
-		if (from < 0) {
-			from = sensor->points_stored - count;
-			if (from < 0)
-				from = 0;
-		}
-		fseek(sensor->file, sizeof(DataPoint) * from, SEEK_SET);
-
-		FCGI_BeginJSON(context, STATUS_OK);	
-		FCGI_JSONLong("id", id); 
-		FCGI_JSONKey("data");
-		FCGI_JSONValue("[");
-		while (total < count) {
-			amount_read = fread(buffer, sizeof(DataPoint), SENSOR_QUERYBUFSIZ, sensor->file);
-			if (amount_read > 0) {
-				FCGI_JSONValue("[%f, %f]", buffer[0].time_stamp, buffer[0].value);
-				total++;
-				for (int i = 1; i < amount_read && total < count; i++, total++)
-					FCGI_JSONValue(", [%f, %f]", buffer[i].time_stamp, buffer[i].value);
-			} else {
-				break;
-			}
-		}
-		FCGI_JSONValue("]");
-
-		FCGI_JSONLong("total_points", sensor->points_stored);
-		FCGI_JSONLong("next_point", from + total);
-		FCGI_EndJSON();	
-	}
-	pthread_mutex_unlock(&(sensor->mutex));
-	//End critical section
-}
-
 /**
  * Handle a request to the sensor module
  * @param context - The context to work in
  * @param params - Parameters passed
+ * TODO: Seriously need to write more helper functions and decrease the size of this function!
  */
 void Sensor_Handler(FCGIContext *context, char * params)
 {
-	DataPoint buffer[SENSOR_QUERYBUFSIZ];
 	StatusCodes status = STATUS_OK;
 
-	enum {DEFAULT, DUMP} operation = DEFAULT;
+	OutputType output_type = JSON;
+	
+
 
 	const char * key; const char * value;
 
 	Sensor * sensor = NULL;
 
+	struct timeval now;
+	gettimeofday(&now, NULL);
+
+	double start_time = -1;
+	double end_time = -1;
+	double current_time = TIMEVAL_DIFF(now, g_options.start_time)
+	bool seek_time = false;
+	bool points_specified = false;
+	int query_size = SENSOR_QUERYBUFSIZ;
+	int start_index = -1;
+	int end_index = -1;
+
+	/* //Possible use case?
+	FCGIValue values[5] = {
+		{"id", &id, FCGI_REQUIRED(FCGI_INT_T)},
+		{"format", &format, FCGI_STRING_T},
+		{"points", &points, FCGI_STRING_T},
+		{"start_time", &start_time, FCGI_DOUBLE_T},
+		{"end_time", &end_time, FCGI_DOUBLE_T}
+	};
+	if (!FCGI_ParseRequest(context, params, values, 5))
+		return;*/
+
 	while ((params = FCGI_KeyPair(params, &key, &value)) != NULL)
 	{
 		Log(LOGDEBUG, "Got key=%s and value=%s", key, value);
@@ -340,15 +429,78 @@ void Sensor_Handler(FCGIContext *context, char * params)
 				break;
 			}
 		}
-		else if (strcmp(key, "dump") == 0)
+		else if (strcmp(key, "format") == 0)
 		{
-			if (operation != DEFAULT)
+			if (strcmp(value, "json") == 0)
+				output_type = JSON;
+			else if (strcmp(value, "csv") == 0)
+				output_type = CSV;
+			else if (strcmp(value, "tsv") == 0)
+				output_type = TSV;			
+		}
+		else if (strcmp(key, "points") == 0)
+		{
+			points_specified = true;
+			if (strcmp(value, "all") == 0)
+			{
+				query_size = sensor->points_written;
+			}
+			else
 			{
-				Log(LOGERR, "Operation already specified!");
+				char * end;
+				query_size = strtol(value, &end, 10);
+				if (*end != '\0')
+				{
+					Log(LOGERR, "Require \"all\" or an integer value: %s = %s", key, value);
+					status = STATUS_ERROR;
+					break;
+				}
+			}
+			
+		}
+		else if (strcmp(key, "start_time") == 0)
+		{
+			seek_time = true;
+			char * end;
+			start_time = strtod(value, &end);
+			if (*end != '\0')
+			{
+				Log(LOGERR, "Require a double: %s = %s", key, value);
 				status = STATUS_ERROR;
 				break;
+			}			
+
+			// Treat negative values as being relative to the current time
+			if (start_time < 0)
+			{
+				start_time = current_time + start_time;
 			}
-			operation = DUMP;
+			start_time = floor(start_time);
+		}
+		else if (strcmp(key, "end_time") == 0)
+		{
+			seek_time = true;
+			char * end;
+			end_time = strtod(value, &end);
+			if (*end != '\0')
+			{
+				Log(LOGERR, "Require a double: %s = %s", key, value);
+				status = STATUS_ERROR;
+				break;
+			}	
+
+			// Treat negative values as being relative to the current time
+			if (end_time < 0)
+			{
+				end_time = current_time + end_time;
+			}		
+			end_time = ceil(end_time);
+		}
+		// For backward compatability:
+		else if (strcmp(key, "dump") == 0)
+		{
+			output_type = TSV;
+			query_size = sensor->points_written+1;
 		}
 		else
 		{
@@ -369,57 +521,67 @@ void Sensor_Handler(FCGIContext *context, char * params)
 		FCGI_RejectJSON(context, "Invalid input parameters");
 		return;
 	}
-	
-	switch (operation)
+
+
+	if (seek_time)
 	{
-		case DUMP:
+		if (end_time < 0 && !points_specified)
+			end_index = sensor->points_written;
+		else
 		{
-			//Force download with content-disposition
-			FCGI_PrintRaw("Content-type: text/plain\r\n"
-				"Content-disposition: attachment;filename=%d.csv\r\n\r\n",
-				sensor->id);
-			//CRITICAL SECTION
-			pthread_mutex_lock(&(sensor->mutex));
-				fseek(sensor->file, 0, SEEK_SET);
-				int amount_read = 0;
-				do
-				{
-					amount_read = fread(buffer, sizeof(DataPoint), SENSOR_QUERYBUFSIZ, sensor->file);
-					for (int i = 0; i < amount_read; ++i)
-					{
-						FCGI_PrintRaw("%f\t%f\n", buffer[i].time_stamp, buffer[i].value);
-					}
-
-				}
-				while (amount_read == SENSOR_QUERYBUFSIZ);
-			pthread_mutex_unlock(&(sensor->mutex));
-			// end critical section
-			break;
+			int count = 0; DataPoint d;
+			end_index = FindTime(sensor, end_time, &count, &d);
+			Log(LOGDEBUG, "FindTime - Looked for %f; found [%f,%f] after %d iterations; sensor %d, position %d", end_time, d.time_stamp, d.value, count, sensor->id, end_index);
 		}
-		default:
+		if (start_time < 0)
+			start_time = 0;
+		else
 		{
-			FCGI_BeginJSON(context, status);	
-			FCGI_JSONPair(key, value); // should spit back sensor ID
-			//Log(LOGDEBUG, "Call Sensor_Query...");
-			int amount_read = Sensor_Query(sensor, buffer, SENSOR_QUERYBUFSIZ);
-			//Log(LOGDEBUG, "Read %d DataPoints", amount_read);
-			//Log(LOGDEBUG, "Produce JSON response");
+			int count = 0; DataPoint d;
+			start_index = FindTime(sensor, start_time, &count, &d);
+			Log(LOGDEBUG, "FindTime - Looked for %f; found [%f,%f] after %d iterations; sensor %d, position %d", start_time, d.time_stamp, d.value, count, sensor->id, start_index);
+		}
+
+		if (points_specified)
+			end_index = start_index + query_size;
+	}
+	else
+	{
+		start_index = sensor->points_written - query_size;
+		
+		end_index = sensor->points_written;
+	}
+	
+	if (start_index < 0)
+	{
+		Log(LOGNOTE, "start_index = %d => Clamped to 0", start_index);
+		start_index = 0;
+	}
+	if (end_index > sensor->points_written)
+	{
+		Log(LOGNOTE, "end_index = %d => Clamped to %d", end_index, sensor->points_written);
+		end_index = sensor->points_written;
+	}
+	
+	switch (output_type)
+	{
+		case JSON:
+			FCGI_BeginJSON(context, status);
+			FCGI_JSONLong("id", sensor->id);
 			FCGI_JSONKey("data");
-			FCGI_JSONValue("[");
-			for (int i = 0; i < amount_read; ++i)
-			{
-				//TODO: Consider; is it better to give both tv_sec and tv_usec to the client seperately, instead of combining here?
-				
-				FCGI_JSONValue("[%f, %f]", buffer[i].time_stamp, buffer[i].value);
-				if (i+1 < amount_read)
-					FCGI_JSONValue(",");
-			}
-			FCGI_JSONValue("]");
-			//Log(LOGDEBUG, "Done producing JSON response");
-			FCGI_EndJSON();	
+			PrintData(sensor, start_index, end_index, output_type);
+			FCGI_EndJSON();
+			break;
+		default:
+			FCGI_PrintRaw("Content-type: text/plain\r\n\r\n");
+			PrintData(sensor, start_index, end_index, output_type);
+			//Force download with content-disposition
+			// Sam: This is cool, but I don't think we should do it
+			//  - letting the user view it in the browser and then save with their own filename is more flexible
+			//"Content-disposition: attachment;filename=%d.csv\r\n\r\n", sensor->id);
 			break;
-		}
 	}
+	
 }
 
 /**