VILLASnode/lib/api/requests/universal/channel.cpp

/** The Universal Data-exchange API.
 *
 * @author Steffen Vogel <post@steffenvogel.de>
 * @copyright 2014-2022, Institute for Automation of Complex Power Systems, EONERC
 * @license Apache 2.0
 *********************************************************************************/

#include <villas/api/requests/universal.hpp>
#include <villas/api/response.hpp>
#include <villas/timing.hpp>

namespace villas {
namespace node {
namespace api {
namespace universal {

class ChannelRequest : public UniversalRequest {
public:
	using UniversalRequest::UniversalRequest;

	Response * executeGet(const std::string &signalName, PayloadType payload)
	{
		if (body != nullptr)
			throw BadRequest("This endpoint does not accept any body data");

		pthread_mutex_lock(&api_node->write.mutex);

		auto *smp = api_node->write.sample;
		if (!smp) {
			pthread_mutex_unlock(&api_node->write.mutex);
			throw Error(HTTP_STATUS_NOT_FOUND, "No data available");
		}

		auto idx = smp->signals->getIndexByName(signalName);
		if (idx < 0) {
			pthread_mutex_unlock(&api_node->write.mutex);
			throw Error(HTTP_STATUS_NOT_FOUND, "Unknown signal id: {}", signalName);
		}

		auto sig = smp->signals->getByIndex(idx);
		auto ch = api_node->write.channels.at(idx);

		if (payload != ch->payload)
			throw BadRequest("Mismatching payload type");

		auto *json_signal = json_pack("{ s: f, s: o, s: s, s: s, s: s }",
			"timestamp", time_to_double(&smp->ts.origin),
			"value", smp->data[idx].toJson(sig->type),
			"validity", "unknown",
			"source", "unknown",
			"timesource", "unknown"
		);

		if (smp->length <= (unsigned) idx)
			smp->length = idx + 1;

		smp->flags |= (int) SampleFlags::HAS_TS_ORIGIN | (int) SampleFlags::HAS_DATA;

		pthread_mutex_unlock(&api_node->write.mutex);

		return new JsonResponse(session, HTTP_STATUS_OK, json_signal);
	}

	Response * executePut(const std::string &signalName, PayloadType payload)
	{
		int ret;

		pthread_mutex_lock(&api_node->read.mutex);

		auto *smp = api_node->read.sample;
		if (!smp) {
			pthread_mutex_unlock(&api_node->read.mutex);
			throw Error(HTTP_STATUS_INTERNAL_SERVER_ERROR, "Not initialized yet");
		}

		auto idx = smp->signals->getIndexByName(signalName);
		if (idx < 0) {
			pthread_mutex_unlock(&api_node->read.mutex);
			throw BadRequest("Unknown signal id: {}", signalName);
		}

		auto sig = smp->signals->getByIndex(idx);
		auto ch = api_node->read.channels.at(idx);

		if (payload != ch->payload)
			throw BadRequest("Mismatching payload type");

		double timestamp = 0;
		json_t *json_value;
		const char *validity = nullptr;
		const char *source = nullptr;
		const char *timesource = nullptr;

		json_error_t err;
		ret = json_unpack_ex(body, &err, 0, "{ s: F, s: o, s?: s, s?: s, s?: s }",
			"timestamp", &timestamp,
			"value", &json_value,
			"validity", &validity,
			"timesource", &timesource,
			"source", &source
		);
		if (ret) {
			pthread_mutex_unlock(&api_node->read.mutex);
			throw BadRequest("Malformed body: {}", err.text);
		}

		if (validity)
			logger->warn("Attribute 'validity' is not supported by VILLASnode");

		if (source)
			logger->warn("Attribute 'source' is not supported by VILLASnode");

		if (timesource)
			logger->warn("Attribute 'timesource' is not supported by VILLASnode");

		ret = smp->data[idx].parseJson(sig->type, json_value);
		if (ret) {
			pthread_mutex_unlock(&api_node->read.mutex);
			throw BadRequest("Malformed value");
		}

		smp->ts.origin = time_from_double(timestamp);

		pthread_cond_signal(&api_node->read.cv);
		pthread_mutex_unlock(&api_node->read.mutex);

		return new JsonResponse(session, HTTP_STATUS_OK, json_object());
	}

	virtual Response * execute()
	{
		auto const &signalName = matches[2];
		auto const &subResource = matches[3];

		PayloadType payload;
		if (subResource == "event")
			payload = PayloadType::EVENTS;
		else if (subResource == "sample")
			payload = PayloadType::EVENTS;
		else
			throw BadRequest("Unsupported sub-resource: {}", subResource);

		switch (method) {
		case Session::Method::GET:
			return executeGet(signalName, payload);
		case Session::Method::PUT:
			return executePut(signalName, payload);
		default:
			throw InvalidMethod(this);
		}
	}
};

/* Register API requests */
static char n[] = "universal/channel/sample";
static char r[] = "/universal/(" RE_NODE_NAME ")/channel/([a-z0-9_-]+)/(sample|event)";
static char d[] = "retrieve or send samples via universal data-exchange API";
static RequestPlugin<ChannelRequest, n, r, d> p;

} /* namespace universal */
} /* namespace api */
} /* namespace node */
} /* namespace villas */