ham-ref/satdump
1
0
Fork 0
mirror of https://github.com/SatDump/SatDump synced 2026-08-13 17:47:30 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add client mode

Browse source
This commit is contained in:
Aang23 2022-07-04 16:33:09 +02:00
parent 0218a101ba
commit 005f27fb32
7 changed files with 330 additions and 95 deletions
Download patch file Download diff file Expand all files Collapse all files

6
pipelines/GOES.json
View file

@ -176,6 +176,12 @@
2,
0
]
],
"client": [
[
3,
0
]
]
},
"frequencies": [],

242
src-cli/live.cpp
View file

@ -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<uint64_t>();
frequency = parameters["frequency"].get<uint64_t>();
timeout = parameters.contains("timeout") ? parameters["timeout"].get<uint64_t>() : 0;
handler_id = parameters["source"].get<std::string>();
}
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<dsp::SourceDescriptor> source_tr = dsp::getAllAvailableSources();
dsp::SourceDescriptor selected_src;
// Get pipeline
std::optional<satdump::Pipeline> 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<satdump::LivePipeline> live_pipeline = std::make_unique<satdump::LivePipeline>(pipeline.value(), parameters, output_file);
// Init source
std::shared_ptr<dsp::DSPSampleSource> 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<satdump::Pipeline> 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<satdump::LivePipeline> live_pipeline = std::make_unique<satdump::LivePipeline>(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<uint64_t>();
frequency = parameters["frequency"].get<uint64_t>();
timeout = parameters.contains("timeout") ? parameters["timeout"].get<uint64_t>() : 0;
handler_id = parameters["source"].get<std::string>();
}
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<dsp::SourceDescriptor> 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<dsp::DSPSampleSource> 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<satdump::Pipeline> 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<satdump::LivePipeline> live_pipeline = std::make_unique<satdump::LivePipeline>(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;
}

39
src-core/core/live_pipeline.cpp
View file

@ -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<dsp::RingBuffer<uint8_t>>(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()

2
src-core/core/module.cpp
View file

@ -63,6 +63,7 @@ SATDUMP_DLL std::map<std::string, std::function<std::shared_ptr<ProcessingModule
#include "modules/demod/module_dvbs2_demod.h"
#include "modules/network/module_network_server.h"
#include "modules/network/module_network_client.h"
#include "modules/xrit/module_goesrecv_publisher.h"
#include "modules/xrit/module_s2udp_xrit_cadu_extractor.h"
@ -86,6 +87,7 @@ void registerModules()
// Network
REGISTER_MODULE(network::NetworkServerModule);
REGISTER_MODULE(network::NetworkClientModule);
// xRIT
REGISTER_MODULE(xrit::GOESRecvPublisherModule);

2
src-core/core/pipeline.cpp
View file

@ -326,7 +326,7 @@ namespace satdump
if (pipelineConfig.value()["live_cfg"].contains("server"))
newPipeline.live_cfg.server_live = pipelineConfig.value()["live_cfg"]["server"].get<std::vector<std::pair<int, int>>>();
if (pipelineConfig.value()["live_cfg"].contains("client"))
newPipeline.live_cfg.server_live = pipelineConfig.value()["live_cfg"]["client"].get<std::vector<std::pair<int, int>>>();
newPipeline.live_cfg.client_live = pipelineConfig.value()["live_cfg"]["client"].get<std::vector<std::pair<int, int>>>();
if (pipelineConfig.value()["live_cfg"].contains("pkt_size"))
newPipeline.live_cfg.pkt_size = pipelineConfig.value()["live_cfg"]["pkt_size"].get<int>();
}

100
src-core/modules/network/module_network_client.cpp Normal file
View file

@ -0,0 +1,100 @@
#include "module_network_client.h"
#include "logger.h"
#include "imgui/imgui.h"
#include "common/utils.h"
#include <nng/nng.h>
#include <nng/protocol/pubsub0/sub.h>
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<int>();
else
throw std::runtime_error("pkt_size parameter must be present!");
if (parameters.count("server_address") > 0)
address = parameters["server_address"].get<std::string>();
else
throw std::runtime_error("server_address parameter must be present!");
if (parameters.count("server_port") > 0)
port = parameters["server_port"].get<int>();
else
throw std::runtime_error("server_port parameter must be present!");
buffer = new uint8_t[pkt_size * 10];
}
std::vector<ModuleDataType> NetworkClientModule::getInputTypes()
{
return {DATA_FILE, DATA_STREAM};
}
std::vector<ModuleDataType> 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<std::string> NetworkClientModule::getParameters()
{
return {"server_address", "server_port", "pkt_size"};
}
std::shared_ptr<ProcessingModule> NetworkClientModule::getInstance(std::string input_file, std::string output_file_hint, nlohmann::json parameters)
{
return std::make_shared<NetworkClientModule>(input_file, output_file_hint, parameters);
}
}

34
src-core/modules/network/module_network_client.h Normal file
View file

@ -0,0 +1,34 @@
#pragma once
#include "core/module.h"
#include <complex>
#include <fstream>
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<ModuleDataType> getInputTypes();
std::vector<ModuleDataType> getOutputTypes();
public:
static std::string getID();
virtual std::string getIDM() { return getID(); };
static std::vector<std::string> getParameters();
static std::shared_ptr<ProcessingModule> getInstance(std::string input_file, std::string output_file_hint, nlohmann::json parameters);
};
}