From 005f27fb324f04254b85ebb8a33ba00c60cc201b Mon Sep 17 00:00:00 2001 From: Aang23 Date: Mon, 4 Jul 2022 16:33:09 +0200 Subject: [PATCH] Add client mode --- pipelines/GOES.json | 6 + src-cli/live.cpp | 242 +++++++++++------- src-core/core/live_pipeline.cpp | 39 +++ src-core/core/module.cpp | 2 + src-core/core/pipeline.cpp | 2 +- .../modules/network/module_network_client.cpp | 100 ++++++++ .../modules/network/module_network_client.h | 34 +++ 7 files changed, 330 insertions(+), 95 deletions(-) create mode 100644 src-core/modules/network/module_network_client.cpp create mode 100644 src-core/modules/network/module_network_client.h diff --git a/pipelines/GOES.json b/pipelines/GOES.json index d352e6e4b..97d9d045f 100644 --- a/pipelines/GOES.json +++ b/pipelines/GOES.json @@ -176,6 +176,12 @@ 2, 0 ] + ], + "client": [ + [ + 3, + 0 + ] ] }, "frequencies": [], diff --git a/src-cli/live.cpp b/src-cli/live.cpp index dfe1ff520..2da60a39c 100644 --- a/src-cli/live.cpp +++ b/src-cli/live.cpp @@ -37,119 +37,173 @@ int main_live(int argc, char *argv[]) // Parse flags nlohmann::json parameters = parse_common_flags(argc - 4, &argv[4]); - uint64_t samplerate; - uint64_t frequency; - uint64_t timeout; - std::string handler_id; - - try + if (parameters.contains("client")) { - samplerate = parameters["samplerate"].get(); - frequency = parameters["frequency"].get(); - timeout = parameters.contains("timeout") ? parameters["timeout"].get() : 0; - handler_id = parameters["source"].get(); - } - catch (std::exception &e) - { - logger->error("Error parsing arguments! {:s}", e.what()); - return 1; - } + logger->info("Starting in client mode!"); - // Create output dir - if (!std::filesystem::exists(output_file)) - std::filesystem::create_directories(output_file); + // Create output dir + if (!std::filesystem::exists(output_file)) + std::filesystem::create_directories(output_file); - // Get all sources - dsp::registerAllSources(); - std::vector source_tr = dsp::getAllAvailableSources(); - dsp::SourceDescriptor selected_src; + // Get pipeline + std::optional pipeline = satdump::getPipelineFromName(downlink_pipeline); - for (dsp::SourceDescriptor src : source_tr) - logger->debug("Device " + src.name); - - // Try to find it and check it's usable - bool src_found = false; - for (dsp::SourceDescriptor src : source_tr) - { - if (handler_id == src.source_type) + if (!pipeline.has_value()) { - selected_src = src; - src_found = true; + logger->critical("Pipeline " + downlink_pipeline + " does not exist!"); + return 1; } - } - if (!src_found) - { - logger->error("Could not find a handler for source type : {:s}!", handler_id.c_str()); - return 1; - } + // Init pipeline + std::unique_ptr live_pipeline = std::make_unique(pipeline.value(), parameters, output_file); - // Init source - std::shared_ptr source_ptr = getSourceFromDescriptor(selected_src); - source_ptr->open(); - source_ptr->set_frequency(frequency); - source_ptr->set_samplerate(samplerate); - source_ptr->set_settings(parameters); + ctpl::thread_pool live_thread_pool(8); - // Get pipeline - std::optional pipeline = satdump::getPipelineFromName(downlink_pipeline); - - if (!pipeline.has_value()) - { - logger->critical("Pipeline " + downlink_pipeline + " does not exist!"); - return 1; - } - - // Init pipeline - parameters["baseband_format"] = "f32"; - parameters["buffer_size"] = STREAM_BUFFER_SIZE; // This is required, as we WILL go over the (usually) default 8192 size - std::unique_ptr live_pipeline = std::make_unique(pipeline.value(), parameters, output_file); - - ctpl::thread_pool live_thread_pool(8); - - bool server_mode = parameters.contains("server_address") || parameters.contains("server_port"); - - // Attempt to start the source and pipeline - try - { - source_ptr->start(); - live_pipeline->start(source_ptr->output_stream, live_thread_pool, server_mode); - } - catch (std::exception &e) - { - logger->error("Fatal error running pipeline/device : " + std::string(e.what())); - return 1; - } - - // Attach signal - signal(SIGINT, sig_handler_live); - - // Now, we wait - uint64_t start_time = time(0); - while (1) - { - if (timeout > 0) + // Attempt to start the source and pipeline + try { - uint64_t elapsed_time = time(0) - start_time; - if (elapsed_time >= timeout) + live_pipeline->start_client(live_thread_pool); + } + catch (std::exception &e) + { + logger->error("Fatal error running pipeline/device : " + std::string(e.what())); + return 1; + } + + // Attach signal + signal(SIGINT, sig_handler_live); + + // Now, we wait + while (1) + { + if (live_should_exit) { - logger->warn("Timeout is over! ({:d}s >= {:d}s) Stopping.", elapsed_time, timeout); + logger->warn("SIGINT Received. Stopping."); break; } + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + + // Stop cleanly + live_pipeline->stop(); + } + else + { + uint64_t samplerate; + uint64_t frequency; + uint64_t timeout; + std::string handler_id; + + try + { + samplerate = parameters["samplerate"].get(); + frequency = parameters["frequency"].get(); + timeout = parameters.contains("timeout") ? parameters["timeout"].get() : 0; + handler_id = parameters["source"].get(); + } + catch (std::exception &e) + { + logger->error("Error parsing arguments! {:s}", e.what()); + return 1; + } + + // Create output dir + if (!std::filesystem::exists(output_file)) + std::filesystem::create_directories(output_file); + + // Get all sources + dsp::registerAllSources(); + std::vector source_tr = dsp::getAllAvailableSources(); + dsp::SourceDescriptor selected_src; + + for (dsp::SourceDescriptor src : source_tr) + logger->debug("Device " + src.name); + + // Try to find it and check it's usable + bool src_found = false; + for (dsp::SourceDescriptor src : source_tr) + { + if (handler_id == src.source_type) + { + selected_src = src; + src_found = true; + } } - if (live_should_exit) + if (!src_found) { - logger->warn("SIGINT Received. Stopping."); - break; + logger->error("Could not find a handler for source type : {:s}!", handler_id.c_str()); + return 1; } - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - } + // Init source + std::shared_ptr source_ptr = getSourceFromDescriptor(selected_src); + source_ptr->open(); + source_ptr->set_frequency(frequency); + source_ptr->set_samplerate(samplerate); + source_ptr->set_settings(parameters); - // Stop cleanly - source_ptr->stop(); - live_pipeline->stop(); + // Get pipeline + std::optional pipeline = satdump::getPipelineFromName(downlink_pipeline); + + if (!pipeline.has_value()) + { + logger->critical("Pipeline " + downlink_pipeline + " does not exist!"); + return 1; + } + + // Init pipeline + parameters["baseband_format"] = "f32"; + parameters["buffer_size"] = STREAM_BUFFER_SIZE; // This is required, as we WILL go over the (usually) default 8192 size + std::unique_ptr live_pipeline = std::make_unique(pipeline.value(), parameters, output_file); + + ctpl::thread_pool live_thread_pool(8); + + bool server_mode = parameters.contains("server_address") || parameters.contains("server_port"); + + // Attempt to start the source and pipeline + try + { + source_ptr->start(); + live_pipeline->start(source_ptr->output_stream, live_thread_pool, server_mode); + } + catch (std::exception &e) + { + logger->error("Fatal error running pipeline/device : " + std::string(e.what())); + return 1; + } + + // Attach signal + signal(SIGINT, sig_handler_live); + + // Now, we wait + uint64_t start_time = time(0); + while (1) + { + if (timeout > 0) + { + uint64_t elapsed_time = time(0) - start_time; + if (elapsed_time >= timeout) + { + logger->warn("Timeout is over! ({:d}s >= {:d}s) Stopping.", elapsed_time, timeout); + break; + } + } + + if (live_should_exit) + { + logger->warn("SIGINT Received. Stopping."); + break; + } + + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + + // Stop cleanly + source_ptr->stop(); + live_pipeline->stop(); + } return 0; } \ No newline at end of file diff --git a/src-core/core/live_pipeline.cpp b/src-core/core/live_pipeline.cpp index 7fea3d6c4..6b2eae722 100644 --- a/src-core/core/live_pipeline.cpp +++ b/src-core/core/live_pipeline.cpp @@ -109,6 +109,45 @@ namespace satdump void LivePipeline::start_client(ctpl::thread_pool &tp) { + // Init modules + { + if (d_pipeline.live_cfg.client_live.size() == 0) + throw std::runtime_error("Pipeline does not support client mode!"); + + d_parameters["pkt_size"] = d_pipeline.live_cfg.pkt_size; + prepare_module("network_client"); + prepare_modules(d_pipeline.live_cfg.client_live); + } + + // Init the first and whatever's in the middle + for (int i = 0; i < (int)modules.size() - 1; i++) + { + modules[i]->input_fifo = modules[i - 1]->output_fifo; + modules[i]->output_fifo = std::make_shared>(1000000); + modules[i]->setInputType(DATA_STREAM); + modules[i]->setOutputType(DATA_STREAM); + modules[i]->init(); + modules[i]->input_active = true; + module_futs.push_back(tp.push([=](int) + { + logger->info("Start processing..."); + modules[i]->process(); })); + } + + // Init the last module + if (modules.size() > 1) + { + int num = modules.size() - 1; + modules[num]->input_fifo = modules[num - 1]->output_fifo; + modules[num]->setInputType(DATA_STREAM); + modules[num]->setOutputType(DATA_FILE); + modules[num]->init(); + modules[num]->input_active = true; + module_futs.push_back(tp.push([=](int) + { + logger->info("Start processing..."); + modules[num]->process(); })); + } } void LivePipeline::stop() diff --git a/src-core/core/module.cpp b/src-core/core/module.cpp index ceaf29417..db05aa117 100644 --- a/src-core/core/module.cpp +++ b/src-core/core/module.cpp @@ -63,6 +63,7 @@ SATDUMP_DLL std::map>>(); if (pipelineConfig.value()["live_cfg"].contains("client")) - newPipeline.live_cfg.server_live = pipelineConfig.value()["live_cfg"]["client"].get>>(); + newPipeline.live_cfg.client_live = pipelineConfig.value()["live_cfg"]["client"].get>>(); if (pipelineConfig.value()["live_cfg"].contains("pkt_size")) newPipeline.live_cfg.pkt_size = pipelineConfig.value()["live_cfg"]["pkt_size"].get(); } diff --git a/src-core/modules/network/module_network_client.cpp b/src-core/modules/network/module_network_client.cpp new file mode 100644 index 000000000..f72f6bd0d --- /dev/null +++ b/src-core/modules/network/module_network_client.cpp @@ -0,0 +1,100 @@ +#include "module_network_client.h" +#include "logger.h" +#include "imgui/imgui.h" +#include "common/utils.h" +#include +#include + +namespace network +{ + NetworkClientModule::NetworkClientModule(std::string input_file, std::string output_file_hint, nlohmann::json parameters) + : ProcessingModule(input_file, output_file_hint, parameters) + { + if (parameters.count("pkt_size") > 0) + pkt_size = parameters["pkt_size"].get(); + else + throw std::runtime_error("pkt_size parameter must be present!"); + + if (parameters.count("server_address") > 0) + address = parameters["server_address"].get(); + else + throw std::runtime_error("server_address parameter must be present!"); + + if (parameters.count("server_port") > 0) + port = parameters["server_port"].get(); + else + throw std::runtime_error("server_port parameter must be present!"); + + buffer = new uint8_t[pkt_size * 10]; + } + + std::vector NetworkClientModule::getInputTypes() + { + return {DATA_FILE, DATA_STREAM}; + } + + std::vector NetworkClientModule::getOutputTypes() + { + return {DATA_FILE}; + } + + NetworkClientModule::~NetworkClientModule() + { + delete[] buffer; + } + + void NetworkClientModule::process() + { + nng_socket sock; + nng_dialer dialer; + + logger->info("Opening TCP socket on " + std::string("tcp://" + address + ":" + std::to_string(port))); + + nng_sub0_open_raw(&sock); + nng_dialer_create(&dialer, sock, std::string("tcp://" + address + ":" + std::to_string(port)).c_str()); + nng_dialer_start(dialer, NULL); + + while (input_active.load()) + { + size_t lpkt_size; + nng_recv(sock, buffer, &lpkt_size, NULL); + + if (pkt_size != lpkt_size) + continue; + + output_fifo->write(buffer, pkt_size); + } + + nng_dialer_close(dialer); + } + + void NetworkClientModule::drawUI(bool window) + { + ImGui::Begin("Network Client", NULL, window ? NULL : NOWINDOW_FLAGS); + + ImGui::Text("Server Address : "); + ImGui::SameLine(); + ImGui::TextColored(IMCOLOR_SYNCED, "%s", address.c_str()); + + ImGui::Text("Server Port : "); + ImGui::SameLine(); + ImGui::TextColored(IMCOLOR_SYNCED, UITO_C_STR(port)); + + ImGui::End(); + } + + std::string NetworkClientModule::getID() + { + return "network_client"; + } + + std::vector NetworkClientModule::getParameters() + { + return {"server_address", "server_port", "pkt_size"}; + } + + std::shared_ptr NetworkClientModule::getInstance(std::string input_file, std::string output_file_hint, nlohmann::json parameters) + { + return std::make_shared(input_file, output_file_hint, parameters); + } +} \ No newline at end of file diff --git a/src-core/modules/network/module_network_client.h b/src-core/modules/network/module_network_client.h new file mode 100644 index 000000000..ddce88765 --- /dev/null +++ b/src-core/modules/network/module_network_client.h @@ -0,0 +1,34 @@ +#pragma once + +#include "core/module.h" +#include +#include + +namespace network +{ + class NetworkClientModule : public ProcessingModule + { + protected: + uint8_t *buffer; + + std::ofstream data_out; + + int pkt_size; + std::string address; + int port; + + public: + NetworkClientModule(std::string input_file, std::string output_file_hint, nlohmann::json parameters); + ~NetworkClientModule(); + void process(); + void drawUI(bool window); + std::vector getInputTypes(); + std::vector getOutputTypes(); + + public: + static std::string getID(); + virtual std::string getIDM() { return getID(); }; + static std::vector getParameters(); + static std::shared_ptr getInstance(std::string input_file, std::string output_file_hint, nlohmann::json parameters); + }; +} \ No newline at end of file