

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of PSYCLONE version 2.0.0.170915_Beta4 (15 September 2017)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2017
//_/_/  
//_/_/  in collaboration with 
//_/_/  
//_/_/  Icelandic Institute for Intelligent Machines ses
//_/_/    Menntavegur 1
//_/_/    101 Reykjavik
//_/_/    ICELAND
//_/_/    http://www.iiim.is/
//_/_/
//_/_/  PSYCLONE 2.0 is released under the GNU Lesser General Public License 
//_/_/  (LGPL) version 3.0, and is subject to the following
//_/_/  
//_/_/  - CADIA Clause: The license granted in and to the software under this 
//_/_/    agreement is a limited-use license. The software may not be used in 
//_/_/    furtherance of: (i) intentionally causing bodily injury or severe 
//_/_/    emotional harm to any person; (ii) invading the personal privacy or 
//_/_/    violating the human rights of any person; or (iii) committing or 
//_/_/    preparing for any act of war. 
//_/_/    
//_/_/  A commercial license for PSYCLONE 2.0 is also available to 
//_/_/  any interested party from the copyright holder.
//_/_/  
//_/_/  You can redistribute this copy of PSYCLONE 2.0 and/or modify it under 
//_/_/  the terms of the GNU General Public License as published by the Free 
//_/_/  Software Foundation, either version 3 of the License, or (at your 
//_/_/  option) any later version.
//_/_/  
//_/_/  You should have received a copy of the GNU General Public License
//_/_/  and the GNU Lesser General Public License along with PSYCLONE 2.0.  
//_/_/  If not, see <http://www.gnu.org/licenses/>.
//_/_/  
//_/_/  PSYCLONE 2.0 is distributed in the hope that it will be useful,
//_/_/  but WITHOUT ANY WARRANTY; without even the implied warranty of
//_/_/  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//_/_/  GNU General Public License for more details.
//_/_/
//_/_/  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
//_/_/  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
//_/_/  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A 
//_/_/  PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER 
//_/_/  OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
//_/_/  EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
//_/_/  PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
//_/_/  PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
//_/_/  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
//_/_/  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
//_/_/  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#include "TimeSeriesCatalog.h"

namespace	cmlabs {

static std::map<std::string, std::list<TimeSeriesCatalogEntry*> > *TimeSeriesCatalogMap = NULL;
static std::map<std::string, std::map<std::string, TimeSeriesCatalogDataset*> > *TimeSeriesCatalogDatasetMap = NULL;

int8 TimeSeriesCatalog(PsyAPI* api) {
	std::string myName = api->getModuleName();
	api->logPrint(2, "Started running...");

	std::list<TimeSeriesCatalogEntry*>* entryList = NULL;
	if (!TimeSeriesCatalogMap)
		TimeSeriesCatalogMap = new std::map<std::string, std::list<TimeSeriesCatalogEntry*> >;
	entryList = &((*TimeSeriesCatalogMap)[myName]);

	std::map<std::string, TimeSeriesCatalogDataset*>* datasetMap = NULL;
	if (!TimeSeriesCatalogDatasetMap)
		TimeSeriesCatalogDatasetMap = new std::map<std::string, std::map<std::string, TimeSeriesCatalogDataset*> >;
	datasetMap = &((*TimeSeriesCatalogDatasetMap)[myName]);

	XMLResults xmlResults;
	std::string xml = api->getParameterString("componentsetup");
	XMLNode node, mainNode = XMLNode::parseString(xml.c_str(), "setup", &xmlResults);

	// display error message (if any)
	if (xmlResults.error != eXMLErrorNone)
	{
		if (xmlResults.error == eXMLErrorFirstTagNotFound) {
			api->logPrint(0, "Catalog setup error, main tag '<setup>' not found");
			// return XMLNode::emptyXMLNode;
		}
		else {
			api->logPrint(0, "Catalog setup parsing error at line %i, column %i:\n   %s\n",
				xmlResults.nLine, xmlResults.nColumn, XMLNode::getError(xmlResults.error));
			// return XMLNode::emptyXMLNode;
		}
	}

	TimeSeriesCatalogDataset* dataset;
	std::map<std::string, TimeSeriesCatalogDataset*>::iterator di, de;
	TimeSeriesCatalogEntry* entry;
	std::list<TimeSeriesCatalogEntry*>::iterator i, e;
	int n = 0;
	const char* cdata, *tname;
	while (!(node = mainNode.getChildNode("dataset", n++)).isEmpty()) {
		if ((cdata = node.getAttribute("name")) && (tname = node.getAttribute("datatype"))) {
			if (datasetMap->find(cdata) != datasetMap->end()) {
				api->logPrint(0, "Cannot define dataset '%s' as it already exists", cdata);
				continue;
			}
			dataset = new TimeSeriesCatalogDataset();
			dataset->mutex.enter();
			dataset->reset();
			dataset->name = cdata;
			if (cdata = node.getAttribute("color"))
				dataset->color = Color(cdata);
			else
				dataset->color = Color();
			if (cdata = node.getAttribute("shoulders"))
				dataset->shoulders = cdata;
			if (cdata = node.getAttribute("shading"))
				dataset->shading = utils::Ascii2Uint32(cdata);
			dataset->datatype = GetDataTypeID(tname);
			dataset->maxKeep = utils::Ascii2Int32(node.getAttribute("maxkeep"));
			if (!dataset->maxKeep)
				dataset->maxKeep = 1000;
			if (node.isAttributeSet("maxfrequency")) {
				float64 f = utils::Ascii2Float64(node.getAttribute("maxfrequency"));
				if (f > 0)
					dataset->minMSApart = (uint32)((1.0 / f) * 1000);
			}
			else if (node.isAttributeSet("minmsapart")) {
				uint32 u = utils::Ascii2Int32(node.getAttribute("minmsapart"));
				if (u > 0)
					dataset->minMSApart = u;
			}
			(*datasetMap)[dataset->name] = dataset;
			dataset->mutex.leave();
		}
	}

	n = 0;
	while (!(node = mainNode.getChildNode("store", n++)).isEmpty()) {
		if ((cdata = node.getAttribute("trigger")) && (tname = node.getAttribute("dataset")) && (datasetMap->find(tname) != datasetMap->end())) {
			dataset = (*datasetMap)[tname];
			if (!dataset) {
				api->logPrint(0, "Cannot define entry for unknown dataset '%s'", tname);
				continue;
			}
			entry = new TimeSeriesCatalogEntry();
			entry->mutex.enter();
			entryList->push_back(entry);
			entry->reset();
			entry->trigger = cdata;
			entry->dataset = tname;
			if (cdata = node.getAttribute("datatype"))
				entry->datatype = GetDataTypeID(cdata);
			else
				entry->datatype = dataset->datatype;
			if (cdata = node.getAttribute("key"))
				entry->key = cdata;
			if (cdata = node.getAttribute("timekey"))
				entry->timeKey = cdata;
			if (cdata = node.getAttribute("durationkey"))
				entry->durationKey = cdata;
			if (cdata = node.getAttribute("value"))
				entry->value = cdata;
			entry->mutex.leave();
		}
	}

	std::string json;
	DataMessage* inMsg, *replyMsg;
	const char* triggerName;
	const char* req;
	//jstd::map<std::string, MessageDataCatalogEntry>::iterator it, et;

	api->addPsyProbeCustomView("Time Series", "elements/timeseriescatalog.html");

	int32 ms = 0;
	while (api->shouldContinue()) {
		if (inMsg = api->waitForNewMessage(200, triggerName)) {
			if (triggerName && stricmp(triggerName, "DefaultComponentTrigger") == 0)
				api->logPrint(2, "Got wakeup message...");
			else if (inMsg->getType() == PsyAPI::CTRL_QUERY) {
				if ((req = inMsg->getString("Query")) && (datasetMap->find(req) != datasetMap->end())) {
					dataset = (*datasetMap)[req];
					if (!dataset) {
						api->logPrint(0, "Cannot query unknown dataset '%s'", req);
						api->queryReply((uint32)inMsg->getReference(), QUERY_NAME_UNKNOWN);
					}
					else {
						replyMsg = dataset->retrieveData((uint64)inMsg->getAsInt("sinceid"), (uint64)inMsg->getAsInt("sincetime"), (uint32)inMsg->getAsInt("maxcount"), (uint32)inMsg->getAsInt("maxage"), (uint32)inMsg->getAsInt("minage"));
						if (replyMsg)
							api->queryReply((uint32)inMsg->getReference(), QUERY_SUCCESS, replyMsg);
						else
							api->queryReply((uint32)inMsg->getReference(), QUERY_FAILED);
					}
				}
				else if (req = inMsg->getString("List")) {
					std::string listJSON;
					for (di = datasetMap->begin(), de = datasetMap->end(); di != de; di++) {
						if (listJSON.length())
							listJSON += utils::StringFormat(",\"%s\": { \"datatype\": \"%s\", \"maxkeep\": %u, \"count\": %u, \"color\": \"%s\", \"shading\": 0.%u, \"shoulders\": \"%s\"}\n",
							di->first.c_str(), GetDataTypeName(di->second->datatype).c_str(), di->second->maxKeep, di->second->dataSeries.size(),
								di->second->color.toRGBString().c_str(), di->second->shading, di->second->shoulders.c_str());
						else
							listJSON += utils::StringFormat("\"%s\": { \"datatype\": \"%s\", \"maxkeep\": %u, \"count\": %u, \"color\": \"%s\", \"shading\": 0.%u, \"shoulders\": \"%s\" }\n",
								di->first.c_str(), GetDataTypeName(di->second->datatype).c_str(), di->second->maxKeep, di->second->dataSeries.size(),
								di->second->color.toRGBString().c_str(), di->second->shading, di->second->shoulders.c_str());
					}
					replyMsg = new DataMessage();
					replyMsg->setString("JSON", utils::StringFormat("{ %s }\n", listJSON.c_str()).c_str());
					api->queryReply((uint32)inMsg->getReference(), QUERY_SUCCESS, replyMsg);
				}
				else
					api->queryReply((uint32)inMsg->getReference(), QUERY_FAILED);
			}
			else {
				// Received another message, storing...
				entry = NULL;
				for (i = entryList->begin(), e = entryList->end(); i != e; i++) {
					if (stricmp((*i)->trigger.c_str(), triggerName) == 0) {
						dataset = (*datasetMap)[(*i)->dataset];
						if (dataset) {
							(*i)->addData(inMsg, dataset);
						}
						else {
							api->logPrint(0, "Cannot add data to unknown dataset '%s'", tname);
							continue;
						}
					}
				}
				// if no matches, we don't know about this, don't store, just ignore...
			}
		}
	}

	return 0;
}



uint32 TimeSeriesCatalogEntry::addData(DataMessage* msg, TimeSeriesCatalogDataset* dataset) {

	if (!mutex.enter(1000))
		return 0;

	uint64 t = 0;
	if (timeKey.length())
		t = msg->getTime(timeKey.c_str());
	if (!t)
		t = msg->getCreatedTime();


	if (value.length()) {
		if (stricmp(dataset->shoulders.c_str(), "0") == 0)
			dataset->addData(t, (int64)0);
		else if (stricmp(dataset->shoulders.c_str(), "last") == 0)
			dataset->repeatLastValue(t);
		// Just add value, don't look for data in msg
		dataset->addData(t, value.c_str());
		return 1;
	}

	uint64 interval = 0;
	uint64 duration = 0;

	DataMessage::KeyType keyType = msg->getKeyType(key.c_str());
	if ((keyType == DataMessage::ARRAY) || (keyType == DataMessage::MAP)) {
		if (durationKey.length())
			duration = (uint32)msg->getInt(durationKey.c_str());
		if (!duration)
			duration = (uint32)msg->getInt("Duration");
		if (!duration && lastTimestamp)
			duration = (uint64)(t - lastTimestamp);
		if (!duration)
			duration = 1000000;
		lastTimestamp = t;
		t -= duration;
	}

	std::map<int64, int64> intArr;
	std::map<int64, int64>::iterator intArrI, intArrE;
	std::map<std::string, int64> intMap;
	std::map<std::string, int64>::iterator intMapI, intMapE;
	std::map<int64, double> dblArr;
	std::map<int64, double>::iterator dblArrI, dblArrE;
	std::map<std::string, double> dblMap;
	std::map<std::string, double>::iterator dblMapI, dblMapE;
	std::map<int64, std::string> stringArr;
	std::map<int64, std::string>::iterator stringArrI, stringArrE;
	std::map<std::string, std::string> stringMap;
	std::map<std::string, std::string>::iterator stringMapI, stringMapE;
	std::map<int64, uint64> timeArr;
	std::map<int64, uint64>::iterator timeArrI, timeArrE;
	std::map<std::string, uint64> timeMap;
	std::map<std::string, uint64>::iterator timeMapI, timeMapE;

	const char* data;
	int addedCount = 0;

	if (stricmp(dataset->shoulders.c_str(), "0") == 0)
		dataset->addData(t, (int64)0);
	else if (stricmp(dataset->shoulders.c_str(), "last") == 0)
		dataset->repeatLastValue(t);

	switch (datatype) {
	case CONSTCHARID:
		switch (keyType) {
		case DataMessage::SINGLE:
			data = msg->getString(key.c_str());
			if (data) {
				dataset->addData(t, data);
				addedCount++;
			}
			break;
		case DataMessage::ARRAY:
			stringArr = msg->getStringArray(key.c_str());
			if (stringArr.size()) {
				interval = duration / stringArr.size();
				stringArrI = stringArr.begin(), stringArrE = stringArr.end();
				// in std::maps keys are always sorted smallest to largest
				while (stringArrI != stringArrE) {
					data = msg->getString(stringArrI->first, key.c_str());
					dataset->addData(t, data);
					t += interval;
					addedCount++;
					stringArrI++;
				}
			}
		case DataMessage::MAP:
			stringMap = msg->getStringMap(key.c_str());
			if (stringMap.size()) {
				interval = duration / stringMap.size();
				stringMapI = stringMap.begin(), stringMapE = stringMap.end();
				// in std::maps keys are always sorted smallest to largest
				while (stringMapI != stringMapE) {
					data = msg->getString(stringMapI->first.c_str(), key.c_str());
					dataset->addData(t, data);
					t += interval;
					addedCount++;
					stringMapI++;
				}
			}
		default:
			break;
		}
	case TIMEID:
		switch (keyType) {
		case DataMessage::SINGLE:
			dataset->addData(t, msg->getTime(key.c_str()));
			addedCount++;
			break;
		case DataMessage::ARRAY:
			timeArr = msg->getTimeArray(key.c_str());
			if (timeArr.size()) {
				interval = duration / timeArr.size();
				timeArrI = timeArr.begin(), timeArrE = timeArr.end();
				// in std::maps keys are always sorted smallest to largest
				while (timeArrI != timeArrE) {
					dataset->addData(t, msg->getTime(timeArrI->first, key.c_str()));
					addedCount++;
					t += interval;
					timeArrI++;
				}
			}
		case DataMessage::MAP:
			timeMap = msg->getTimeMap(key.c_str());
			if (timeMap.size()) {
				interval = duration / timeMap.size();
				timeMapI = timeMap.begin(), timeMapE = timeMap.end();
				// in std::maps keys are always sorted smallest to largest
				while (timeMapI != timeMapE) {
					dataset->addData(t, msg->getTime(timeMapI->first.c_str(), key.c_str()));
					addedCount++;
					t += interval;
					timeMapI++;
				}
			}
		default:
			break;
		}
	case INTID:
		switch (keyType) {
		case DataMessage::SINGLE:
			dataset->addData(t, msg->getInt(key.c_str()));
			addedCount++;
			break;
		case DataMessage::ARRAY:
			intArr = msg->getIntArray(key.c_str());
			if (intArr.size()) {
				interval = duration / intArr.size();
				intArrI = intArr.begin(), intArrE = intArr.end();
				// in std::maps keys are always sorted smallest to largest
				while (intArrI != intArrE) {
					dataset->addData(t, msg->getInt(intArrI->first, key.c_str()));
					addedCount++;
					t += interval;
					intArrI++;
				}
			}
		case DataMessage::MAP:
			intMap = msg->getIntMap(key.c_str());
			if (intMap.size()) {
				interval = duration / intMap.size();
				intMapI = intMap.begin(), intMapE = intMap.end();
				// in std::maps keys are always sorted smallest to largest
				while (intMapI != intMapE) {
					dataset->addData(t, msg->getInt(intMapI->first.c_str(), key.c_str()));
					addedCount++;
					t += interval;
					intMapI++;
				}
			}
		default:
			break;
		}
	case DOUBLEID:
		switch (keyType) {
		case DataMessage::SINGLE:
			dataset->addData(t, msg->getFloat(key.c_str()));
			addedCount++;
			break;
		case DataMessage::ARRAY:
			dblArr = msg->getFloatArray(key.c_str());
			if (dblArr.size()) {
				interval = duration / dblArr.size();
				dblArrI = dblArr.begin(), dblArrE = dblArr.end();
				// in std::maps keys are always sorted smallest to largest
				while (dblArrI != dblArrE) {
					dataset->addData(t, msg->getFloat(dblArrI->first, key.c_str()));
					addedCount++;
					t += interval;
					dblArrI++;
				}
			}
		case DataMessage::MAP:
			dblMap = msg->getFloatMap(key.c_str());
			if (dblMap.size()) {
				interval = duration / dblMap.size();
				dblMapI = dblMap.begin(), dblMapE = dblMap.end();
				// in std::maps keys are always sorted smallest to largest
				while (dblMapI != dblMapE) {
					dataset->addData(t, msg->getFloat(dblMapI->first.c_str(), key.c_str()));
					addedCount++;
					t += interval;
					dblMapI++;
				}
			}
		default:
			break;
		}
	case CHARDATAID:
	case DATAMESSAGEID:
	case CONSTCHARINFOID:
	case CHARDATAINFOID:
	case DATAMESSAGEINFOID:
	default:
		break;
	}

	if (stricmp(dataset->shoulders.c_str(), "0") == 0) {
		dataset->addData(t, (int64)0);
	}

	mutex.leave();
	return addedCount;

	// Data of type entryType always stored in userdata key
	// If key is a single value, add to timeseries with timestamp timeKey or use msg created
	// If key is an array spread timestamps evenly from timeKey/created backwards for durationKey ms
	// durationKey defaults to 'duration'
	// If duration not provided use duration = timeKey/created - last(timeKey/created)

	// Always added in the order they come in = in ID order, even if time is then out of order
}

bool TimeSeriesCatalogDataset::addRawData(uint64 t, const char* data, uint32 size) {
	if (!mutex.enter(1000))
		return false;
	TimeSeriesCatalogData rData;
	rData.timestamp = t;
	rData.size = size;
	rData.data = new char[rData.size];
	memcpy(rData.data, data, rData.size);
	rData.id = ++lastID;
	dataSeries.push_back(rData);
	while (dataSeries.size() > maxKeep) {
		delete (dataSeries.front().data);
		dataSeries.pop_front();
	}
	lastStored = GetTimeNow();
	mutex.leave();
	return true;
}


bool TimeSeriesCatalogDataset::addData(uint64 t, const char* value) {
	int64 i64;
	uint64 u64;
	float64 f64;
	switch (datatype) {
	case CONSTCHARID:
		return addRawData(t, value, strlen(value) + 1);
	case TIMEID:
		// convert text to uint64
		u64 = utils::Ascii2Uint64(value);
		return addRawData(t, (const char*)&u64, sizeof(uint64));
	case INTID:
		// convert text to int64
		i64 = utils::Ascii2Int64(value);
		return addRawData(t, (const char*)&i64, sizeof(int64));
	case DOUBLEID:
		// convert text to float64
		f64 = utils::Ascii2Float64(value);
		return addRawData(t, (const char*)&f64, sizeof(float64));
	case CHARDATAID:
	case DATAMESSAGEID:
	case CONSTCHARINFOID:
	case CHARDATAINFOID:
	case DATAMESSAGEINFOID:
	default:
		return false;
	}
}

bool TimeSeriesCatalogDataset::addData(uint64 t, int64 value) {
	uint64 u64;
	float64 f64;
	std::string str;
	switch (datatype) {
	case CONSTCHARID:
		str = utils::StringFormat("%lld", value);
		return addRawData(t, str.c_str(), str.length() + 1);
	case TIMEID:
		// convert text to uint64
		u64 = (uint64)value;
		return addRawData(t, (const char*)&u64, sizeof(uint64));
	case INTID:
		// convert text to int64
		return addRawData(t, (const char*)&value, sizeof(int64));
	case DOUBLEID:
		// convert text to float64
		f64 = (float64)value;
		return addRawData(t, (const char*)&f64, sizeof(float64));
	case CHARDATAID:
	case DATAMESSAGEID:
	case CONSTCHARINFOID:
	case CHARDATAINFOID:
	case DATAMESSAGEINFOID:
	default:
		return false;
	}
}

bool TimeSeriesCatalogDataset::addData(uint64 t, float64 value) {
	int64 i64;
	uint64 u64;
	std::string str;
	switch (datatype) {
	case CONSTCHARID:
		str = utils::StringFormat("%f", (double)value);
		return addRawData(t, str.c_str(), str.length() + 1);
	case TIMEID:
		// convert text to uint64
		u64 = (uint64)value;
		return addRawData(t, (const char*)&u64, sizeof(uint64));
	case INTID:
		// convert text to int64
		i64 = (int64)value;
		return addRawData(t, (const char*)&i64, sizeof(int64));
	case DOUBLEID:
		return addRawData(t, (const char*)&value, sizeof(float64));
	case CHARDATAID:
	case DATAMESSAGEID:
	case CONSTCHARINFOID:
	case CHARDATAINFOID:
	case DATAMESSAGEINFOID:
	default:
		return false;
	}
}

bool TimeSeriesCatalogDataset::addData(uint64 t, uint64 value) {
	int64 i64;
	float64 f64;
	std::string str;
	switch (datatype) {
	case CONSTCHARID:
		str = utils::StringFormat("%llu", value);
		return addRawData(t, str.c_str(), str.length() + 1);
	case TIMEID:
		// convert text to uint64
		return addRawData(t, (const char*)&value, sizeof(uint64));
	case INTID:
		// convert text to int64
		i64 = (int64)value;
		return addRawData(t, (const char*)&i64, sizeof(int64));
	case DOUBLEID:
		// convert text to float64
		f64 = (float64)value;
		return addRawData(t, (const char*)&f64, sizeof(float64));
	case CHARDATAID:
	case DATAMESSAGEID:
	case CONSTCHARINFOID:
	case CHARDATAINFOID:
	case DATAMESSAGEINFOID:
	default:
		return false;
	}
}

bool TimeSeriesCatalogDataset::repeatLastValue(uint64 t) {
	if (!dataSeries.size())
		return true;
	TimeSeriesCatalogData data = dataSeries.back();
	//printf("Adding shoulder of %llu ms...\n", (uint64)((t - data.timestamp)));
	return addRawData(t, data.data, data.size);
}


DataMessage* TimeSeriesCatalogDataset::retrieveData(uint64 sinceID, uint64 sinceTime, uint32 maxCount, uint32 maxAgeMS, uint32 minAgeMS) {

	if (!mutex.enter(1000))
		return NULL;

	std::string json;
	uint64 now = GetTimeNow();

	uint64 fromTime = 0;
	if (sinceTime)
		fromTime = sinceTime + 1;
	else if (maxAgeMS)
		fromTime = now - (maxAgeMS * 1000);

	uint64 toTime = 0;
	if (minAgeMS)
		toTime = now - (minAgeMS * 1000);

	uint32 count = 0;
	uint32 skipped = 0;
	std::list<TimeSeriesCatalogData>::reverse_iterator i = dataSeries.rbegin(), e = dataSeries.rend();
	while (i != e) {
		if (sinceID && ((*i).id <= sinceID))
			break;
		else if (fromTime && ((*i).timestamp <= fromTime)) {
			if (++skipped > 1000)
				break;
			continue;
		}
		else if (toTime && ((*i).timestamp >= toTime))
			continue;
		else if (maxCount && (count >= maxCount))
			break;

		if (json.length())
			json = (*i).toJSON(datatype) + "," + json;
		else
			json = (*i).toJSON(datatype);
		count++;
		i++;
	}

	json = utils::StringFormat("{ \"count\": %u, \"data\": [%s] }\n", count, json.c_str());
	DataMessage* msg = new DataMessage();
	msg->setString("JSON", json.c_str());
	mutex.leave();
	return msg;

}

std::string TimeSeriesCatalogData::toJSON(uint32 datatype) {

	switch (datatype) {
	case 0:
		return utils::StringFormat("{ \"id\": %llu, \"type\": \"none\", \"time\": %llu, \"timetext\": \"%s\" }\n",
			id, timestamp, PrintTimeSortableMicrosecString(timestamp).c_str());
	case CHARDATAID:
		return utils::StringFormat("{ \"id\": %llu, \"type\": \"binary\", \"time\": %llu, \"timetext\": \"%s\", \"size\": %u }\n",
			id, timestamp, PrintTimeSortableMicrosecString(timestamp).c_str(), size);
	case TIMEID:
		return utils::StringFormat("{ \"id\": %llu, \"type\": \"time\", \"time\": %llu, \"timetext\": \"%s\", \"value\": %llu, \"valuetext\": \"%s\" }\n",
			id, timestamp, PrintTimeSortableMicrosecString(timestamp).c_str(), *(uint64*)data, PrintTimeSortableMicrosecString(*(uint64*)data).c_str());
	case INTID:
		return utils::StringFormat("{ \"id\": %llu, \"type\": \"integer\", \"time\": %llu, \"timetext\": \"%s\", \"value\": %lld }\n",
			id, timestamp, PrintTimeSortableMicrosecString(timestamp).c_str(), *(int64*)data);
	case DOUBLEID:
		return utils::StringFormat("{ \"id\": %llu, \"type\": \"float\", \"time\": %llu, \"timetext\": \"%s\", \"value\": %f }\n",
			id, timestamp, PrintTimeSortableMicrosecString(timestamp).c_str(), (double)(*(float64*)data));
	case CONSTCHARID:
		return utils::StringFormat("{ \"id\": %llu, \"type\": \"string\", \"time\": %llu, \"timetext\": \"%s\", \"value\": \"%s\" }\n",
			id, timestamp, PrintTimeSortableMicrosecString(timestamp).c_str(), data);
	default:
		return utils::StringFormat("{ \"id\": %llu, \"type\": \"unknown\", \"time\": %llu, \"timetext\": \"%s\" }\n",
			id, timestamp, PrintTimeSortableMicrosecString(timestamp).c_str());
	}
}



} // namespace cmlabs
