radangel/radangeld/main.cpp

/*
radangeld -- a linux daemon to readout and distribute data measured by a Kromek 
RadAngel γ-spectrometer
Copyright (C) 2019 Matthias Janke <matthias.janke@physi.uni-heidelberg.de>

This program is free software: you can redistribute it and/or modify it under 
the terms of the GNU Affero General Public License as published by the Free 
Software Foundation, version 3.

This program 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 Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License along 
with this program. If not, see <https://www.gnu.org/licenses/>
*/
// SPDX-License-Identifier: AGPL-3.0-only


#include <iostream>
#include <fstream>
#include <chrono>
#include <array>
#include <limits>
#include <thread>
#include <atomic>
#include <csignal>
#include <cstdio>

#include <wchar.h>

// generate timestamps that are identifiable
#include "date/tz.h"

// program option parser
#include <boost/program_options.hpp>

// HID device access helper library
#include <hidapi.h>

// configuration file library
#include <libconfig.h++>

// thread/ipc/tcp communications
#include <zmqpp/zmqpp.hpp>

// systemd new-style daemon library
#include <systemd/sd-daemon.h>

namespace po = boost::program_options;

namespace
{
  volatile std::sig_atomic_t terminate{0};
}
 
void signal_handler(int signal)
{
    if(signal==SIGTERM)
        terminate = 1;
}

struct measurement_properties
{
	bool eventswrite{true};
	bool reportswrite{false};
	bool demonize{false};

	int eventswritten{0};
	int reportswritten{0};

	std::chrono::seconds maxduration{30};
	unsigned int readreports{0};

	std::string fileprefix{"radangel"};
	std::string serial{""};
	hid_device *handle;

	std::chrono::time_point<std::chrono::system_clock> starttime;
	std::chrono::time_point<std::chrono::system_clock> endtime;

    std::chrono::time_point<std::chrono::steady_clock> starterrortime;
	std::chrono::time_point<std::chrono::steady_clock> startmeastime;
};

std::string ws2mbs (const wchar_t *ws)
{
	std::mbstate_t state = std::mbstate_t();
	const int len = 1 + std::wcsrtombs (nullptr, &ws, 0, &state);
	std::vector<char> mbs (len);
	std::wcsrtombs (&mbs[0], &ws, mbs.size(), &state);
	return std::string (mbs.data(), mbs.size() );
}

int writeprotocol (const std::string &filename, measurement_properties &mp)
{
	const int MAX_STR = 255;

	int res;

	wchar_t wstr[MAX_STR];

	libconfig::Config cfg;

	libconfig::Setting &root = cfg.getRoot();

	libconfig::Setting &measurement = root.add ("Measurement", libconfig::Setting::TypeGroup);

	measurement.add ("StartTime", libconfig::Setting::TypeString) = date::format ("%F %T ", date::make_zoned (date::current_zone(), mp.starttime) ) + date::current_zone()->name();
	measurement.add ("StartTimeError_ns", libconfig::Setting::TypeInt64) = std::chrono::duration_cast<std::chrono::nanoseconds>(mp.startmeastime-mp.starterrortime).count()/2;
	measurement.add ("EndTime", libconfig::Setting::TypeString) = date::format ("%F %T ", date::make_zoned (date::current_zone(), mp.endtime) ) + date::current_zone()->name();
	measurement.add ("RunAsDaemon", libconfig::Setting::TypeBoolean) = mp.demonize;
	measurement.add ("MaxIntegrationTime_s", libconfig::Setting::TypeInt) = static_cast<int> (mp.maxduration.count() );
	measurement.add ("SavedFilePrefix", libconfig::Setting::TypeString) = mp.fileprefix;
	measurement.add ("CountedReports", libconfig::Setting::TypeInt) = static_cast<int> (mp.readreports);

	libconfig::Setting &detector = measurement.add ("Detector", libconfig::Setting::TypeGroup);
	libconfig::Setting &usb = detector.add ("USB", libconfig::Setting::TypeGroup);
	// Read the Manufacturer String
	wstr[0] = 0x0000;
	res = hid_get_manufacturer_string (mp.handle, wstr, MAX_STR);

	if (res == 0)
		usb.add ("Manufacturer", libconfig::Setting::TypeString) = ws2mbs (wstr);

	// Read the Product String
	wstr[0] = 0x0000;
	res = hid_get_product_string (mp.handle, wstr, MAX_STR);

	if (res == 0)
		usb.add ("Product", libconfig::Setting::TypeString) = ws2mbs (wstr);

	// Read the Serial Number String
	wstr[0] = 0x0000;
	res = hid_get_serial_number_string (mp.handle, wstr, MAX_STR);

	if (res == 0)
		usb.add ("Serial", libconfig::Setting::TypeString) = ws2mbs (wstr);

	libconfig::Setting &manual = detector.add ("Manual", libconfig::Setting::TypeGroup);
	manual.add ("Serial", libconfig::Setting::TypeString) = mp.serial;

	libconfig::Setting &clock = measurement.add ("Clock", libconfig::Setting::TypeGroup);
	clock.add ("Steady", libconfig::Setting::TypeBoolean) = std::chrono::steady_clock::is_steady;
	std::chrono::steady_clock::duration unit (1);
	clock.add ("Granularity_ns", libconfig::Setting::TypeInt64) = std::chrono::nanoseconds (unit).count() ;

	std::array<std::chrono::time_point<std::chrono::steady_clock>, 1000000> vec;

	for (auto &elem : vec)
		elem = std::chrono::steady_clock::now();

	std::vector<std::chrono::nanoseconds> differences (vec.size() - 1);

	for (auto i = 0u; i < differences.size(); i++)
		differences[i] = vec[i + 1] - vec[i];

	auto result = std::min_element (std::begin (differences), std::end (differences) );

	clock.add ("MaximumAccuracy_ns", libconfig::Setting::TypeInt64) = result->count();

	libconfig::Setting &files = measurement.add ("Files", libconfig::Setting::TypeGroup);

	libconfig::Setting &data = files.add ("Events", libconfig::Setting::TypeList);
	data.add (libconfig::Setting::TypeBoolean) = mp.eventswrite;
	data.add (libconfig::Setting::TypeInt) = mp.eventswritten;

	libconfig::Setting &report = files.add ("Reports", libconfig::Setting::TypeList);
	report.add (libconfig::Setting::TypeBoolean) = mp.reportswrite;
	report.add (libconfig::Setting::TypeInt) = mp.reportswritten;

	cfg.writeFile ( (filename + "-protocol.conf").c_str() );

	return 0;
}

void read_hid_reports (measurement_properties &mp, zmqpp::context &zmqc)
{
	int ret = 0 ;
	// buffer to transfer data via HID IN-descriptor
	std::array<uint8_t, 63> reportdata{0};// maximum report length specified by RadAngel HID descriptor report 4

	zmqpp::socket reportsocket (zmqc, mp.reportswrite ? zmqpp::socket_type::publish : zmqpp::socket_type::pair);
	reportsocket.bind ("inproc://rawreports");
	zmqpp::message report;

	if (mp.demonize)
		sd_notifyf(0, "READY=1\nSTATUS=Processing events…\nMAINPID=%lu", (unsigned long) getpid());
                 
    mp.starterrortime = std::chrono::steady_clock::now();
	mp.starttime = std::chrono::system_clock::now();
	mp.startmeastime = std::chrono::steady_clock::now();

	do
	{
		do
		{
			ret = hid_read (mp.handle, reportdata.data(), reportdata.size() );

			if (ret == 0)
				std::this_thread::sleep_for (std::chrono::microseconds (500) );
			else if (ret > 0)
				break;
			else
				std::cerr <<  "Unable to read() HID reports!\n";
		}
		while (ret == 0);

		if (reportdata[0] > 0)
		{
			report << "report"
			       << std::chrono::time_point_cast<std::chrono::microseconds> (std::chrono::steady_clock::now() ).time_since_epoch().count()
			       << ret;

			for (const auto &element : reportdata)
				report << element;

			reportsocket.send (report);

			mp.readreports++;
		}
	}
	while ( (ret >= 0) && (mp.demonize || ( (std::chrono::steady_clock::now() - mp.startmeastime) < mp.maxduration) ) && !terminate );

	mp.endtime = std::chrono::system_clock::now();

	report << "done";
	reportsocket.send (report);
	
	if(mp.demonize)
	    sd_notify(0, "STOPPING=1");
}

void save_hid_reports (measurement_properties &mp, zmqpp::context &zmqc)
{
	mp.reportswritten = 0;

	if (mp.reportswrite)
	{
		int64_t time = 0;
		int validbytes = 0 ;

		std::string reporttype;

		// buffer to transfer data via HID IN-descriptor
		std::array<uint8_t, 63> reportdata{0};// maximum report length specified by RadAngel HID descriptor report 4

		zmqpp::socket reportsocket (zmqc, zmqpp::socket_type::subscribe);
		reportsocket.connect ("inproc://rawreports");
		reportsocket.subscribe ("");
		zmqpp::message reports;

		std::ofstream reportsfile;

		do
		{
			reportsocket.receive (reports);

			reports >> reporttype;

			if (reporttype == "report")
			{
				if (!reportsfile.is_open())
				{
					reportsfile.open (mp.fileprefix + date::format ("-%F-%T-%Z", date::make_zoned (date::current_zone(), mp.starttime) ) + "-reports.csv", std::ios::app);

					reportsfile << "\"relative time since start [µs]\",\"number of valid report bytes\"";

					for (auto i = 0u; i < 63; i++)
						reportsfile << ",\"report byte " << i << "\"";

					reportsfile << "\n";
				}

				reports >> time
				        >> validbytes;

				for (auto &element : reportdata)
					reports >> element;

				reportsfile << std::chrono::duration_cast<std::chrono::microseconds> (std::chrono::steady_clock::time_point{std::chrono::microseconds{time}} - mp.startmeastime).count() << ","
				            << validbytes;

				for (auto &element : reportdata)
					reportsfile << "," << static_cast<int> (element);

				reportsfile << "\n";

				mp.reportswritten++;
			}
			else if (reporttype == "done")
				break;
			else
			{
				std::cerr << "reporttype \"" << reporttype << "\" unknown" << std::endl;
				break;
			}
		}
		while (true);

		if (reportsfile.is_open())
		{
			reportsfile.flush();
			reportsfile.close();
		}
	}
}

void decode_hid_reports (measurement_properties &mp, zmqpp::context &zmqc)
{
	auto init = true;
	auto reportcounter = 0u;

	uint8_t validevents = 0;
	uint8_t invalidevents = 0;
	int64_t time;

	int validbytes;

	std::string reporttype;

	// buffer to transfer data via HID IN-descriptor
	std::array<uint8_t, 63> reportdata{0};// maximum report length specified by RadAngel HID descriptor report 4

	std::array<std::pair<uint16_t, uint8_t>, 31> currentevents;
	std::array<std::array<std::pair<uint16_t, uint8_t>, 31>, 2> oldevents{{{std::make_pair (0, 0) }}};

	zmqpp::socket reportsocket (zmqc, mp.reportswrite ? zmqpp::socket_type::subscribe : zmqpp::socket_type::pair);
	reportsocket.connect ("inproc://rawreports");

	if (mp.reportswrite)
		reportsocket.subscribe ("");

	zmqpp::message reports;

	zmqpp::socket eventssocket (zmqc, zmqpp::socket_type::publish);
	eventssocket.bind ("inproc://events");
	zmqpp::message events;

	do
	{
		reportsocket.receive (reports);

		reports >> reporttype;

		if (reporttype == "report")
		{
			reports >> time
			        >> validbytes;

			for (auto &element : reportdata)
				reports >> element;

			if (reportdata[0] == 4)
			{
				if (validbytes == 63)
				{
					for (auto i = 0u; i < currentevents.size(); i++)
						currentevents[i] = std::make_pair ( ( (reportdata[2 * i + 1] << 8) | reportdata[2 * i + 2]) >> 4, 0x0f & reportdata[2 * i + 2]);

					validevents = 0;
					invalidevents = 0;

					for (auto i = 0u; i < currentevents.size(); i++)
						if (currentevents[i] != oldevents[reportcounter % 2][i])
						{
							oldevents[reportcounter % 2][i] = currentevents[i];

							if (currentevents[i].second != 0)
								validevents++;
							else
								invalidevents++;
						}
						else
						{
							oldevents[reportcounter % 2][i] = currentevents[i];
							currentevents[i].second = 0;
						}

					if (!init)
					{
						events << "events"
						       << std::chrono::duration_cast<std::chrono::microseconds> (std::chrono::steady_clock::time_point{std::chrono::microseconds{time}} - mp.startmeastime).count()
						       << invalidevents
						       << validevents;

						for (const auto &element : currentevents)
							if (element.second != 0)
								events << element.first;

						eventssocket.send (events);
					}

					reportcounter++;

					if (init && (reportcounter == 2) )
						init = false;
				}
				else
					std::cerr << "invalid reportsize! received: " << validbytes << " expected: 63\n";
			}
		}
		else if (reporttype == "done")
			break;
		else
		{
			std::cerr << "reporttype \"" << reporttype << "\" unknown" << std::endl;
			break;
		}
	}
	while (true);

	events << "done";
	eventssocket.send (events);
}

void save_events (measurement_properties &mp, zmqpp::context &zmqc)
{
	mp.eventswritten = 0;

	if (mp.eventswrite)
	{
		uint8_t validevents;
		uint8_t invalidevents;

		int64_t timestamp;

		std::string eventtype;

		std::array<uint16_t, 31> currentevents{0};

		zmqpp::socket eventssocket (zmqc, zmqpp::socket_type::subscribe);
		eventssocket.connect ("inproc://events");
		eventssocket.subscribe ("");
		zmqpp::message events;

		std::ofstream eventfile;

		do
		{
			eventssocket.receive (events);

			events >> eventtype;

			if (eventtype == "events")
			{
				if (!eventfile.is_open())
				{
					eventfile.open (mp.fileprefix + date::format ("-%F-%T-%Z", date::make_zoned (date::current_zone(), mp.starttime) ) + "-events.csv", std::ios::app);

					eventfile << "\"relative time since start [µs]\",\"number of invalid events\",\"number of valid events\"";

					for (auto i = 0u; i < 31; i++)
						eventfile << ",\"event " << i << "\"";

					eventfile << "\n";
				}

				events >> timestamp
				       >> invalidevents
				       >> validevents;

				for (auto i = 0u; i < validevents; i++)
					events >> currentevents[i];

				eventfile << timestamp << ","
				          << static_cast<int> (invalidevents) << ","
				          << static_cast<int> (validevents);

				for (auto i = 0u; i < validevents; i++)
					eventfile << "," << static_cast<int> (currentevents[i]);

				eventfile << "\n";

				mp.eventswritten++;
			}
			else if (eventtype == "done")
				break;
			else
			{
				std::cerr << "save_events: eventtype \"" << eventtype << "\" unknown" << std::endl;
				break;
			}
		}
		while (true);

		if (eventfile.is_open() )
		{
			eventfile.flush();
			eventfile.close();
		}
	}
}

void publish_events (zmqpp::context &zmqc)
{
	uint8_t validevents;
	uint8_t invalidevents;

	uint16_t currentevent;

	int64_t timestamp;

	std::string eventtype;

	zmqpp::socket eventssocket (zmqc, zmqpp::socket_type::subscribe);
	eventssocket.connect ("inproc://events");
	eventssocket.subscribe ("");
	zmqpp::message events;

	zmqpp::socket eventpublisher (zmqc, zmqpp::socket_type::publish);
	eventpublisher.bind ("ipc://radangel.events");
	zmqpp::message published_events;

	do
	{
		eventssocket.receive (events);

		events >> eventtype;

		if (eventtype == "events")
		{
			events >> timestamp
			       >> invalidevents
			       >> validevents;

			for (auto i = 0u; i < validevents; i++)
			{
				events >> currentevent;

				published_events << currentevent
				                 << timestamp;

				eventpublisher.send (published_events);
			}
		}
		else if (eventtype == "done")
			break;
		else
		{
			std::cerr << "eventtype \"" << eventtype << "\" unknown" << std::endl;
			break;
		}
	}
	while (true);

	eventpublisher.close();
	std::remove("radangel.events");
}

void publish_cpm (zmqpp::context &zmqc)
{
	uint8_t validevents;
	uint8_t invalidevents;

	int64_t timestamp;
	int64_t oldtimestamp = 0;

	std::string eventtype;

	zmqpp::socket eventssocket (zmqc, zmqpp::socket_type::subscribe);
	eventssocket.connect ("inproc://events");
	eventssocket.subscribe ("");
	zmqpp::message events;

	zmqpp::socket cpmsocket (zmqc, zmqpp::socket_type::publish);
	cpmsocket.bind ("ipc://radangel.cpm");
	zmqpp::message cpm;

	do
	{
		eventssocket.receive (events);

		events >> eventtype;

		if (eventtype == "events")
		{
			events  >> timestamp
			        >> invalidevents
			        >> validevents;

			if (validevents > 0)
			{
				cpm << timestamp
				    << static_cast<double> (validevents / ((timestamp - oldtimestamp) / 1000.0 / 1000.0 ) * 60.0);

				cpmsocket.send (cpm);

				oldtimestamp = timestamp;
			}
		}
		else if (eventtype == "done")
			break;
		else
		{
			std::cerr << "eventtype \"" << eventtype << "\" unknown" << std::endl;
			break;
		}
	}
	while (true);

	cpmsocket.close();
	std::remove("radangel.cpm");
}

int main (int argc, char *argv[])
{
	zmqpp::context zmqc;

	measurement_properties mp;

	unsigned int integration_time;

	po::options_description desc ("Available options:");
	desc.add_options()
	("help,?", "Prints this list of available options.")
	("events,e", po::value<bool> (&mp.eventswrite)->implicit_value (true)->default_value (true), "Save events data to disk. File postfix is -data.csv.")
	("reports,r", po::value<bool> (&mp.reportswrite)->implicit_value (true)->default_value (false), "Save received raw usb hid reports to disk. File postfix is -reports.csv.")
	("integrate,i", po::value<unsigned int> (&integration_time)->default_value (30), "Maximum duration of a measurement in seconds.")
	("daemon,d", po::value<bool> (&mp.demonize)->implicit_value (true)->default_value (false), "Run endless as daemon.")
	("prefix,p", po::value<std::string> (&mp.fileprefix)->default_value ("radangel"), "Prefix for the files generated.")
	("serial,s", po::value<std::string> (&mp.serial), "Serial number of detector.")
	;

	po::positional_options_description p;
	p.add ("prefix", -1);

	po::variables_map vm;
	po::store (po::command_line_parser (argc, argv).options (desc).positional (p).run(), vm);
	po::notify (vm);

	if (vm.count ("help") )
	{
		std::cout << "Usage: " << argv[0] << " [options] [prefix]\n";
		std::cout << desc;
		return 0;
	}

	mp.maxduration = std::chrono::seconds{integration_time};

	if (vm.count ("daemon") && !vm["daemon"].defaulted() && vm["integrate"].defaulted() )
		mp.maxduration = std::numeric_limits<std::chrono::seconds>::max();

	if (hid_init() )
		return 1;

	mp.handle = hid_open (0x4d8, 0x100, NULL);

	if (!mp.handle)
	{
		std::cerr << "unable to open device\n";
		return 4;
	}

	// Set the hid_read() function to be non-blocking.
	hid_set_nonblocking (mp.handle, 1);

	std::signal (SIGTERM, signal_handler);

	if(!terminate)
	{
		std::thread publishCPM (publish_cpm, std::ref (zmqc) );
		std::thread publishEvents (publish_events, std::ref (zmqc) );
		std::thread saveEvents (save_events, std::ref (mp), std::ref (zmqc) );
		std::thread decodeReports (decode_hid_reports, std::ref (mp), std::ref (zmqc) );
		std::thread saveHIDReports (save_hid_reports, std::ref (mp), std::ref (zmqc) );
		std::thread readHIDReports (read_hid_reports, std::ref (mp), std::ref (zmqc) );

		publishCPM.join();
		publishEvents.join();
		saveEvents.join();
		decodeReports.join();
		saveHIDReports.join();
		readHIDReports.join();

		// write measurement protocol
		writeprotocol (mp.fileprefix + date::format ("-%F-%T-%Z", date::make_zoned (date::current_zone(), mp.starttime) ), mp);
	}

	hid_close (mp.handle);

	// Free static HIDAPI objects.
	hid_exit();

	return 0;
}