Resolves #343 GUI and Frame Refactor

Major rewrite of the GUI into a module and view form, also using a new Feed class. The frame and frameset data structures have also been rewritten to support memory pooling and response frames.

applications/vision/src/main.cpp

@@ -17,6 +17,8 @@
 
 #include <opencv2/opencv.hpp>
 #include <ftl/rgbd.hpp>
+#include <ftl/data/framepool.hpp>
+#include <ftl/streams/builder.hpp>
 //#include <ftl/middlebury.hpp>
 #include <ftl/net/universe.hpp>
 #include <ftl/master.hpp>
@@ -26,6 +28,7 @@
 
 #include <ftl/streams/netstream.hpp>
 #include <ftl/streams/sender.hpp>
+#include <ftl/streams/receiver.hpp>
 
 #include <ftl/audio/source.hpp>
 
@@ -55,6 +58,8 @@ using std::chrono::milliseconds;
 using cv::Mat;
 using json = nlohmann::json;
 
+static bool quiet = false;
+
 static void run(ftl::Configurable *root) {
 	Universe *net = ftl::create<Universe>(root, "net");
 
@@ -86,7 +91,7 @@ static void run(ftl::Configurable *root) {
 	ftl::ctrl::Master ctrl(root, net);
 
 	// Sync clocks!
-	ftl::timer::add(ftl::timer::kTimerMain, [&time_peer,&sync_counter,net](int64_t ts) {
+	auto timer = ftl::timer::add(ftl::timer::kTimerMain, [&time_peer,&sync_counter,net](int64_t ts) {
 		if (sync_counter-- <= 0 && time_peer != ftl::UUID(0) ) {
 			sync_counter = 20;
 			auto start = std::chrono::high_resolution_clock::now();
@@ -114,10 +119,8 @@ static void run(ftl::Configurable *root) {
 
 	auto paths = root->get<vector<string>>("paths");
 	string file = "";
-	//if (paths && (*paths).size() > 0) file = (*paths)[(*paths).size()-1];
 
 	for (auto &x : *paths) {
-		//LOG(INFO) << "PATH - " << x;
 		if (x != "") {
 			ftl::URI uri(x);
 			if (uri.isValid()) {
@@ -132,56 +135,133 @@ static void run(ftl::Configurable *root) {
 		}
 	}
 
-	Source *source = nullptr;
-	source = ftl::create<Source>(root, "source", net);
 	if (file != "") {
-		//source->set("uri", file);
 		ftl::URI uri(file);
-		uri.to_json(source->getConfig());
-		source->set("uri", uri.getBaseURI());
+		uri.to_json(root->getConfig()["source"]);
 	}
+	Source *source = nullptr;
+	source = ftl::create<Source>(root, "source");
 	
 	ftl::stream::Sender *sender = ftl::create<ftl::stream::Sender>(root, "sender");
 	ftl::stream::Net *outstream = ftl::create<ftl::stream::Net>(root, "stream", net);
-	outstream->set("uri", outstream->getID());
+	outstream->set("uri", root->value("uri", outstream->getID()));
 	outstream->begin();
 	sender->setStream(outstream);
 
-	auto *grp = new ftl::rgbd::Group();
-	source->setChannel(Channel::Depth);
-	grp->addSource(source);
-
-	int stats_count = 0;
-
-	grp->onFrameSet([sender,&stats_count](ftl::rgbd::FrameSet &fs) {
-		fs.id = 0;
-		sender->post(fs);
+	ftl::audio::Source *audioSrc = ftl::create<ftl::audio::Source>(root, "audio_test");
 
-		if (--stats_count <= 0) {
-			auto [fps,latency] = ftl::rgbd::Builder::getStatistics();
-			LOG(INFO) << "Frame rate: " << fps << ", Latency: " << latency;
-			stats_count = 20;
+	ftl::data::Pool pool(root->value("mempool_min", 2),root->value("mempool_max", 5));
+	auto *creator = new ftl::streams::IntervalSourceBuilder(&pool, 0, {source, audioSrc});
+	std::shared_ptr<ftl::streams::BaseBuilder> creatorptr(creator);
+
+	ftl::stream::Receiver *receiver = ftl::create<ftl::stream::Receiver>(root, "receiver", &pool);
+	receiver->setStream(outstream);
+	receiver->registerBuilder(creatorptr);
+
+	// Which channels should be encoded
+	std::unordered_set<Channel> encodable;
+
+	// Send channels on flush
+	auto flushhandle = pool.onFlushSet([sender,&encodable](ftl::data::FrameSet &fs, ftl::codecs::Channel c) {
+		if ((int)c >= 32) sender->post(fs, c);
+		else {
+			if (encodable.count(c)) {
+				sender->post(fs, c);
+			} else {
+				//switch (c) {
+				//case Channel::Colour		:
+				//case Channel::Colour2		:
+				//case Channel::Depth			: 
+				sender->post(fs, c, true); //break;
+				//default						: sender->fakePost(fs, c);
+				//}
+			}
 		}
 		return true;
 	});
 
-	// TODO: TEMPORARY
-	ftl::audio::Source *audioSrc = ftl::create<ftl::audio::Source>(root, "audio_test");
-	audioSrc->onFrameSet([sender](ftl::audio::FrameSet &fs) {
-		sender->post(fs);
-		return true;
-	});
-	
-	auto pipeline = ftl::config::create<ftl::operators::Graph>(root, "pipeline");
+	int stats_count = 0;
+	int frames = 0;
+	float latency = 0.0f;
+	int64_t stats_time = 0;
+
+	root->on("quiet", quiet, false);
+
+	auto *pipeline = ftl::config::create<ftl::operators::Graph>(root, "pipeline");
 	pipeline->append<ftl::operators::DetectAndTrack>("facedetection")->value("enabled", false);
 	pipeline->append<ftl::operators::ArUco>("aruco")->value("enabled", false);
 	pipeline->append<ftl::operators::DepthChannel>("depth");  // Ensure there is a depth channel
-	grp->addPipeline(pipeline);
+
+	bool busy = false;
+
+	auto h = creator->onFrameSet([sender,outstream,&stats_count,&latency,&frames,&stats_time,pipeline,&busy,&encodable](const ftl::data::FrameSetPtr &fs) {
+		if (busy) {
+			LOG(WARNING) << "Depth pipeline drop: " << fs->timestamp();
+			fs->firstFrame().message(ftl::data::Message::Warning_PIPELINE_DROP, "Depth pipeline drop");
+			return true;
+		}
+		busy = true;
+
+		encodable.clear();
+		// Decide what to encode here.
+		const auto sel = outstream->selectedNoExcept(fs->frameset());
+		std::vector<Channel> sortedsel(sel.begin(), sel.end());
+		std::sort(sortedsel.begin(),sortedsel.end());
+
+		if (sortedsel.size() > 0) encodable.emplace(sortedsel[0]);
+		if (sortedsel.size() > 1) encodable.emplace(sortedsel[1]);
+
+		// Only allow the two encoders to exist
+		// This ensures we cleanup other encoders
+		sender->setActiveEncoders(fs->frameset(), encodable);
+
+		// Do all processing in another thread... only if encoding of depth
+		//if (encodable.find(Channel::Depth) != encodable.end()) {
+			ftl::pool.push([sender,&stats_count,&latency,&frames,&stats_time,pipeline,&busy,fs](int id) mutable {
+				// Do pipeline here...
+				pipeline->apply(*fs, *fs);
+
+				++frames;
+				latency += float(ftl::timer::get_time() - fs->timestamp());
+
+				// Destruct frameset as soon as possible to send the data...
+				const_cast<ftl::data::FrameSetPtr&>(fs).reset();
+
+				if (!quiet && --stats_count <= 0) {
+					latency /= float(frames);
+					int64_t nowtime = ftl::timer::get_time();
+					stats_time = nowtime - stats_time;
+					float fps = float(frames) / (float(stats_time) / 1000.0f);
+					LOG(INFO) << "Frame rate: " << fps << ", Latency: " << latency;
+					stats_count = 20;
+					frames = 0;
+					latency = 0.0f;
+					stats_time = nowtime;
+				}
+
+				busy = false;
+			});
+		//} else {
+			//LOG(INFO) << "NOT DOING DEPTH";
+		//	sender->forceAvailable(*fs, Channel::Depth);
+		//	busy = false;
+		//}
+
+		// Lock colour right now to encode in parallel
+		fs->flush(ftl::codecs::Channel::Colour);
+		fs->flush(ftl::codecs::Channel::AudioStereo);
+
+		return true;
+	});
+
+	// Start the timed generation of frames
+	creator->start();
 	
+	// Only now start listening for connections
 	net->start();
 
 	LOG(INFO) << "Running...";
-	ftl::timer::start(true);
+	ftl::timer::start(true);  // Blocks
 	LOG(INFO) << "Stopping...";
 	ctrl.stop();
 	
@@ -189,27 +269,16 @@ static void run(ftl::Configurable *root) {
 
 	ftl::pool.stop();
 
-	delete grp;
+	delete source;
+	delete receiver;
 	delete sender;
+	delete pipeline;
+	delete audioSrc;
 	delete outstream;
 
-	//delete source;  // TODO(Nick) Add ftl::destroy
 	delete net;
 }
 
-static void threadSetCUDADevice() {
-	// Ensure all threads have correct cuda device
-	std::atomic<int> ijobs = 0;
-	for (int i=0; i<ftl::pool.size(); ++i) {
-		ftl::pool.push([&ijobs](int id) {
-			ftl::cuda::setDevice();
-			++ijobs;
-			while (ijobs < ftl::pool.size()) std::this_thread::sleep_for(std::chrono::milliseconds(10));
-		});
-	}
-	while (ijobs < ftl::pool.size()) std::this_thread::sleep_for(std::chrono::milliseconds(10));
-}
-
 int main(int argc, char **argv) {
 #ifdef HAVE_PYLON
 	Pylon::PylonAutoInitTerm autoInitTerm;
@@ -219,7 +288,24 @@ int main(int argc, char **argv) {
 	SetPriorityClass(GetCurrentProcess(), HIGH_PRIORITY_CLASS);
 #endif
 	std::cout << "FTL Vision Node " << FTL_VERSION_LONG << std::endl;
-	auto root = ftl::configure(argc, argv, "vision_default");
+	auto root = ftl::configure(argc, argv, "vision_default", {
+		"uri",
+		"fps",
+		"time_master",
+		"time_peer",
+		"quiet"
+	});
+
+	root->value("restart", 0);
+
+	// Allow config controlled restart
+	root->on("restart", [root]() {
+		auto val = root->get<int>("restart");
+		if (val) {
+			ftl::exit_code = *val;
+			ftl::running = false;
+		}
+	});
 
 	// Use other GPU if available.
 	//ftl::cuda::setDevice(ftl::cuda::deviceCount()-1);
@@ -228,6 +314,9 @@ int main(int argc, char **argv) {
 	run(root);
 
 	delete root;
+
+	ftl::config::cleanup();
+
 	LOG(INFO) << "Terminating with code " << ftl::exit_code;
 	LOG(INFO) << "Branch: " << ftl::branch_name;
 	return ftl::exit_code;