20111957357

Committed 10 Dec 2025 08:13PM UTC coverage: 48.009% (-0.2%) from 48.224%

Build # 20111957357

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #2728 from mavlink/pr-log-streaming-ardupilot

log_streaming: add ArduPilot support

Run Details

0 of 425 new or added lines in 6 files covered. (0.0%)

13 existing lines in 5 files now uncovered.

17655 of 36774 relevant lines covered (48.01%)

462.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/mavsdk/plugins/log_streaming/log_streaming_impl.cpp

#include <future>

#include "log_streaming_impl.h"
#include "log_streaming_backend_px4.h"
#include "log_streaming_backend_ardupilot.h"
#include "plugins/log_streaming/log_streaming.h"
#include "callback_list.tpp"
#include "base64.h"

namespace mavsdk {

template class CallbackList<LogStreaming::LogStreamingRaw>;

LogStreamingImpl::LogStreamingImpl(System& system) : PluginImplBase(system)
{
    _system_impl->register_plugin(this);
}

LogStreamingImpl::LogStreamingImpl(std::shared_ptr<System> system) :
    PluginImplBase(std::move(system))
{
    _system_impl->register_plugin(this);
}

LogStreamingImpl::~LogStreamingImpl()
{
    _system_impl->unregister_plugin(this);
}

void LogStreamingImpl::init()
{
    if (const char* env_p = std::getenv("MAVSDK_LOG_STREAMING_DEBUGGING")) {
        if (std::string(env_p) == "1") {
            LogDebug() << "Log streaming debugging is on.";
            _debugging = true;
        }
    }
}

void LogStreamingImpl::deinit()
{
    std::lock_guard<std::mutex> lock(_mutex);

    // Cancel any pending autopilot type polling
    if (_check_autopilot_cookie) {
        _system_impl->remove_call_every(_check_autopilot_cookie);
        _check_autopilot_cookie = {};
    }
    _start_callback = nullptr;

    if (_backend) {
        _backend->deinit();
        _backend.reset();
    }
}

void LogStreamingImpl::enable() {}

void LogStreamingImpl::disable()
{
    std::lock_guard<std::mutex> lock(_mutex);
    if (_backend) {
        _backend->deinit();
        _backend.reset();
    }
}

bool LogStreamingImpl::maybe_create_backend()
{
    // Already have a backend
    if (_backend) {
        return true;
    }

    auto autopilot = _system_impl->autopilot();

    // Don't create backend yet if autopilot type is unknown
    if (autopilot == Autopilot::Unknown) {
        if (_debugging) {
            LogDebug() << "Autopilot type unknown, cannot create backend yet";
        }
        return false;
    }

    if (autopilot == Autopilot::ArduPilot) {
        if (_debugging) {
            LogDebug() << "Creating ArduPilot log streaming backend";
        }
        _backend = std::make_unique<LogStreamingBackendArdupilot>();
    } else {
        if (_debugging) {
            LogDebug() << "Creating PX4 log streaming backend";
        }
        _backend = std::make_unique<LogStreamingBackendPx4>();
    }

    _backend->set_debugging(_debugging);
    _backend->init(_system_impl.get());
    _backend->set_data_callback([this](const std::vector<uint8_t>& data) { process_data(data); });
    return true;
}

void LogStreamingImpl::process_data(const std::vector<uint8_t>& data)
{
    std::lock_guard<std::mutex> lock(_mutex);

    if (_debugging) {
        LogDebug() << "Processing log data with size " << data.size();
    }

    // Convert to base64
    LogStreaming::LogStreamingRaw part;
    std::vector<uint8_t> data_copy = data;
    part.data_base64 = base64_encode(data_copy);

    // Let's pass it to the user.
    if (!_subscription_callbacks.empty()) {
        _subscription_callbacks.queue(
            part, [this](const auto& func) { _system_impl->call_user_callback(func); });
    }
}

void LogStreamingImpl::start_log_streaming_async(const LogStreaming::ResultCallback& callback)
{
    std::lock_guard<std::mutex> lock(_mutex);

    // Try to create backend if not yet created
    if (maybe_create_backend()) {
        _backend->start_log_streaming_async(callback);
        return;
    }

    // Autopilot type unknown - wait for it with polling.
    // Use timeout_s() for the total wait time, polling every 0.1s.
    const float poll_interval_s = 0.1f;
    const unsigned max_polls = static_cast<unsigned>(_system_impl->timeout_s() / poll_interval_s);

    _start_callback = callback;
    _autopilot_poll_count = 0;

    _check_autopilot_cookie = _system_impl->add_call_every(
        [this, max_polls]() {
            std::lock_guard<std::mutex> lock2(_mutex);

            if (maybe_create_backend()) {
                // Success - stop polling and start streaming
                _system_impl->remove_call_every(_check_autopilot_cookie);
                _backend->start_log_streaming_async(_start_callback);
                _start_callback = nullptr;
                return;
            }

            _autopilot_poll_count++;
            if (_autopilot_poll_count >= max_polls) {
                // Timeout - stop polling and report failure
                _system_impl->remove_call_every(_check_autopilot_cookie);
                if (_start_callback) {
                    auto cb = _start_callback;
                    _start_callback = nullptr;
                    _system_impl->call_user_callback(
                        [cb]() { cb(LogStreaming::Result::NoSystem); });
                }
            }
        },
        poll_interval_s);
}

LogStreaming::Result LogStreamingImpl::start_log_streaming()
{
    auto prom = std::promise<LogStreaming::Result>{};
    auto fut = prom.get_future();

    start_log_streaming_async([&](LogStreaming::Result result) { prom.set_value(result); });

    return fut.get();
}

void LogStreamingImpl::stop_log_streaming_async(const LogStreaming::ResultCallback& callback)
{
    std::lock_guard<std::mutex> lock(_mutex);

    if (!_backend) {
        if (callback) {
            _system_impl->call_user_callback(
                [callback]() { callback(LogStreaming::Result::NoSystem); });
        }
        return;
    }

    _backend->stop_log_streaming_async(callback);
}

LogStreaming::Result LogStreamingImpl::stop_log_streaming()
{
    auto prom = std::promise<LogStreaming::Result>{};
    auto fut = prom.get_future();

    stop_log_streaming_async([&](LogStreaming::Result result) { prom.set_value(result); });

    return fut.get();
}

LogStreaming::LogStreamingRawHandle
LogStreamingImpl::subscribe_log_streaming_raw(const LogStreaming::LogStreamingRawCallback& callback)
{
    std::lock_guard<std::mutex> lock(_mutex);
    auto handle = _subscription_callbacks.subscribe(callback);

    return handle;
}

void LogStreamingImpl::unsubscribe_log_streaming_raw(LogStreaming::LogStreamingRawHandle handle)
{
    std::lock_guard<std::mutex> lock(_mutex);
    _subscription_callbacks.unsubscribe(handle);
}

} // namespace mavsdk

1	#include <future>
2
3	#include "log_streaming_impl.h"
4	#include "log_streaming_backend_px4.h"
5	#include "log_streaming_backend_ardupilot.h"
6	#include "plugins/log_streaming/log_streaming.h"
7	#include "callback_list.tpp"
8	#include "base64.h"
9
10	namespace mavsdk {
11
12	template class CallbackList<LogStreaming::LogStreamingRaw>;
13
14	LogStreamingImpl::LogStreamingImpl(System& system) : PluginImplBase(system)	×
15	{
16	_system_impl->register_plugin(this);	×
17	}	×
18
19	LogStreamingImpl::LogStreamingImpl(std::shared_ptr<System> system) :	×
20	PluginImplBase(std::move(system))	×
21	{
22	_system_impl->register_plugin(this);	×
23	}	×
24
25	LogStreamingImpl::~LogStreamingImpl()	×
26	{
27	_system_impl->unregister_plugin(this);	×
28	}	×
29
30	void LogStreamingImpl::init()	×
31	{
32	if (const char* env_p = std::getenv("MAVSDK_LOG_STREAMING_DEBUGGING")) {	×
33	if (std::string(env_p) == "1") {	×
34	LogDebug() << "Log streaming debugging is on.";	×
35	_debugging = true;	×
36	}
37	}
UNCOV 38	}	×
39
40	void LogStreamingImpl::deinit()	×
41	{
NEW 42	std::lock_guard<std::mutex> lock(_mutex);	×
43
44	// Cancel any pending autopilot type polling
NEW 45	if (_check_autopilot_cookie) {	×
NEW 46	_system_impl->remove_call_every(_check_autopilot_cookie);	×
NEW 47	_check_autopilot_cookie = {};	×
48	}
NEW 49	_start_callback = nullptr;	×
50
NEW 51	if (_backend) {	×
NEW 52	_backend->deinit();	×
NEW 53	_backend.reset();	×
54	}
55	}	×
56
57	void LogStreamingImpl::enable() {}	×
58
NEW 59	void LogStreamingImpl::disable()	×
60	{
NEW 61	std::lock_guard<std::mutex> lock(_mutex);	×
NEW 62	if (_backend) {	×
NEW 63	_backend->deinit();	×
NEW 64	_backend.reset();	×
65	}
UNCOV 66	}	×
67
NEW 68	bool LogStreamingImpl::maybe_create_backend()	×
69	{
70	// Already have a backend
NEW 71	if (_backend) {	×
NEW 72	return true;	×
73	}
74
NEW 75	auto autopilot = _system_impl->autopilot();	×
76
77	// Don't create backend yet if autopilot type is unknown
NEW 78	if (autopilot == Autopilot::Unknown) {	×
NEW 79	if (_debugging) {	×
NEW 80	LogDebug() << "Autopilot type unknown, cannot create backend yet";	×
81	}
NEW 82	return false;	×
83	}
84
NEW 85	if (autopilot == Autopilot::ArduPilot) {	×
NEW 86	if (_debugging) {	×
NEW 87	LogDebug() << "Creating ArduPilot log streaming backend";	×
88	}
NEW 89	_backend = std::make_unique<LogStreamingBackendArdupilot>();	×
90	} else {
NEW 91	if (_debugging) {	×
NEW 92	LogDebug() << "Creating PX4 log streaming backend";	×
93	}
NEW 94	_backend = std::make_unique<LogStreamingBackendPx4>();	×
95	}
96
NEW 97	_backend->set_debugging(_debugging);	×
NEW 98	_backend->init(_system_impl.get());	×
NEW 99	_backend->set_data_callback([this](const std::vector<uint8_t>& data) { process_data(data); });	×
NEW 100	return true;	×
101	}
102
NEW 103	void LogStreamingImpl::process_data(const std::vector<uint8_t>& data)	×
104	{
105	std::lock_guard<std::mutex> lock(_mutex);	×
106
NEW 107	if (_debugging) {	×
NEW 108	LogDebug() << "Processing log data with size " << data.size();	×
109	}
110
111	// Convert to base64
NEW 112	LogStreaming::LogStreamingRaw part;	×
NEW 113	std::vector<uint8_t> data_copy = data;	×
NEW 114	part.data_base64 = base64_encode(data_copy);	×
115
116	// Let's pass it to the user.
NEW 117	if (!_subscription_callbacks.empty()) {	×
NEW 118	_subscription_callbacks.queue(	×
NEW 119	part, [this](const auto& func) { _system_impl->call_user_callback(func); });	×
120	}
UNCOV 121	}	×
122
NEW 123	void LogStreamingImpl::start_log_streaming_async(const LogStreaming::ResultCallback& callback)	×
124	{
NEW 125	std::lock_guard<std::mutex> lock(_mutex);	×
126
127	// Try to create backend if not yet created
NEW 128	if (maybe_create_backend()) {	×
NEW 129	_backend->start_log_streaming_async(callback);	×
UNCOV 130	return;	×
131	}
132
133	// Autopilot type unknown - wait for it with polling.
134	// Use timeout_s() for the total wait time, polling every 0.1s.
NEW 135	const float poll_interval_s = 0.1f;	×
NEW 136	const unsigned max_polls = static_cast<unsigned>(_system_impl->timeout_s() / poll_interval_s);	×
137
NEW 138	_start_callback = callback;	×
NEW 139	_autopilot_poll_count = 0;	×
140
NEW 141	_check_autopilot_cookie = _system_impl->add_call_every(	×
NEW 142	[this, max_polls]() {	×
NEW 143	std::lock_guard<std::mutex> lock2(_mutex);	×
144
NEW 145	if (maybe_create_backend()) {	×
146	// Success - stop polling and start streaming
NEW 147	_system_impl->remove_call_every(_check_autopilot_cookie);	×
NEW 148	_backend->start_log_streaming_async(_start_callback);	×
NEW 149	_start_callback = nullptr;	×
NEW 150	return;	×
151	}
152
NEW 153	_autopilot_poll_count++;	×
NEW 154	if (_autopilot_poll_count >= max_polls) {	×
155	// Timeout - stop polling and report failure
NEW 156	_system_impl->remove_call_every(_check_autopilot_cookie);	×
NEW 157	if (_start_callback) {	×
NEW 158	auto cb = _start_callback;	×
NEW 159	_start_callback = nullptr;	×
NEW 160	_system_impl->call_user_callback(	×
161	[cb]() { cb(LogStreaming::Result::NoSystem); });
NEW 162	}	×
163	}
NEW 164	},	×
165	poll_interval_s);
UNCOV 166	}	×
167
NEW 168	LogStreaming::Result LogStreamingImpl::start_log_streaming()	×
169	{
NEW 170	auto prom = std::promise<LogStreaming::Result>{};	×
NEW 171	auto fut = prom.get_future();	×
172
NEW 173	start_log_streaming_async([&](LogStreaming::Result result) { prom.set_value(result); });	×
174
NEW 175	return fut.get();	×
NEW 176	}	×
177
NEW 178	void LogStreamingImpl::stop_log_streaming_async(const LogStreaming::ResultCallback& callback)	×
179	{
180	std::lock_guard<std::mutex> lock(_mutex);	×
181
NEW 182	if (!_backend) {	×
NEW 183	if (callback) {	×
NEW 184	_system_impl->call_user_callback(	×
185	[callback]() { callback(LogStreaming::Result::NoSystem); });
186	}
UNCOV 187	return;	×
188	}
189
NEW 190	_backend->stop_log_streaming_async(callback);	×
UNCOV 191	}	×
192
NEW 193	LogStreaming::Result LogStreamingImpl::stop_log_streaming()	×
194	{
NEW 195	auto prom = std::promise<LogStreaming::Result>{};	×
NEW 196	auto fut = prom.get_future();	×
197
NEW 198	stop_log_streaming_async([&](LogStreaming::Result result) { prom.set_value(result); });	×
199
NEW 200	return fut.get();	×
201	}	×
202
203	LogStreaming::LogStreamingRawHandle
NEW 204	LogStreamingImpl::subscribe_log_streaming_raw(const LogStreaming::LogStreamingRawCallback& callback)	×
205	{
NEW 206	std::lock_guard<std::mutex> lock(_mutex);	×
NEW 207	auto handle = _subscription_callbacks.subscribe(callback);	×
208
NEW 209	return handle;	×
210	}	×
211
NEW 212	void LogStreamingImpl::unsubscribe_log_streaming_raw(LogStreaming::LogStreamingRawHandle handle)	×
213	{
NEW 214	std::lock_guard<std::mutex> lock(_mutex);	×
NEW 215	_subscription_callbacks.unsubscribe(handle);	×
UNCOV 216	}	×
217
218	} // namespace mavsdk

mavlink / MAVSDK / 20111957357

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous