application.h

Go to the documentation of this file.

// Copyright 2026:
//   GobySoft, LLC (2013-)
//   Community contributors (see AUTHORS file)
// File authors:
//   Toby Schneider <toby@gobysoft.org>
//
//
// This file is part of the Goby Underwater Autonomy Project Libraries
// ("The Goby Libraries").
//
// The Goby Libraries are free software: you can redistribute them and/or modify
// them under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or
// (at your option) any later version.
//
// The Goby Libraries are distributed in the hope that they will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with Goby.  If not, see <http://www.gnu.org/licenses/>.
 
#ifndef GOBY_MIDDLEWARE_LANGUAGES_PYTHON_APPLICATION_H
#define GOBY_MIDDLEWARE_LANGUAGES_PYTHON_APPLICATION_H
 
#include <chrono>    // for microseconds
#include <csignal>   // for signal, raise, sig_atomic_t
#include <cstdint>   // for int64_t
#include <cstdlib>   // for exit
#include <exception> // for exception_ptr
#include <iostream>  // for cout
#include <memory>    // for unique_ptr
#include <string>    // for string
#include <vector>    // for vector
 
#include <google/protobuf/text_format.h>
 
#include <pybind11/pybind11.h>
#include <pybind11/stl.h>
 
#include "goby/exception.h"
#include "goby/middleware/application/configurator.h"
#include "goby/middleware/application/detail/simulation_time.h"
#include "goby/middleware/application/interface.h"
#include "goby/middleware/group.h"
#include "goby/middleware/languages/common/interface.h"
#include "goby/middleware/marshalling/interface.h"
#include "goby/middleware/protobuf/coroner.pb.h"
#include "goby/time.h"
#include "goby/time/simulation.h"
#include "goby/util/debug_logger.h"
 
namespace goby
{
namespace middleware
{
namespace python
{
namespace py = pybind11;
 
using goby::middleware::languages::Identifier;
using goby::middleware::languages::PubSubLayer;
 
namespace detail
{
inline void invoke_callback(const py::object& callback, const std::vector<char>& bytes)
{
    py::gil_scoped_acquire gil;
    callback(py::bytes(bytes.data(), bytes.size()));
}
 
struct SignalState
{
    void (*quit_fn)(void*) = nullptr;
    void* quit_ctx = nullptr;
    volatile std::sig_atomic_t requested = 0;
};
 
inline SignalState& signal_state()
{
    static SignalState state;
    return state;
}
 
inline void signal_handler(int signum)
{
    SignalState& state = signal_state();
 
    if (state.quit_fn != nullptr && state.requested == 0)
    {
        state.requested = 1;
        state.quit_fn(state.quit_ctx);
    }
    else
    {
        // a second signal means the application isn't unwinding (e.g. it is blocked waiting on
        // data with no loop frequency set), so fall back to the default disposition
        std::signal(signum, SIG_DFL);
        std::raise(signum);
    }
}
 
class SignalGuard
{
  public:
    SignalGuard(void (*quit_fn)(void*), void* quit_ctx)
    {
        SignalState& state = signal_state();
        state.quit_fn = quit_fn;
        state.quit_ctx = quit_ctx;
        state.requested = 0;
 
        previous_sigint_ = std::signal(SIGINT, &signal_handler);
        previous_sigterm_ = std::signal(SIGTERM, &signal_handler);
    }
 
    ~SignalGuard()
    {
        if (previous_sigint_ != SIG_ERR)
            std::signal(SIGINT, previous_sigint_);
        if (previous_sigterm_ != SIG_ERR)
            std::signal(SIGTERM, previous_sigterm_);
 
        SignalState& state = signal_state();
        state.quit_fn = nullptr;
        state.quit_ctx = nullptr;
    }
 
    SignalGuard(const SignalGuard&) = delete;
    SignalGuard& operator=(const SignalGuard&) = delete;
 
  private:
    void (*previous_sigint_)(int){SIG_ERR};
    void (*previous_sigterm_)(int){SIG_ERR};
};
 
} // namespace detail
 
template <typename AppBase> class Application : public AppBase
{
  public:
    using AppBase::AppBase;
 
    using Hook = std::function<void()>;
    using HealthHook = std::function<std::string(const std::string&)>;
 
    void set_hooks(Hook loop, Hook initialize, Hook finalize, HealthHook health)
    {
        loop_ = std::move(loop);
        initialize_ = std::move(initialize);
        finalize_ = std::move(finalize);
        health_ = std::move(health);
    }
 
  protected:
    void loop() override
    {
        if (loop_)
            loop_();
    }
 
    void health(goby::middleware::protobuf::ThreadHealth& health) override
    {
        AppBase::health(health);
 
        if (!health_)
            return;
 
        std::string before;
        health.SerializeToString(&before);
 
        std::string after = health_(before);
        if (after.empty())
            return;
 
        goby::middleware::protobuf::ThreadHealth updated;
        if (updated.ParseFromString(after))
            health.Swap(&updated);
        else
            goby::glog.is_warn() && goby::glog << "Python health() returned a ThreadHealth that "
                                                  "could not be parsed; keeping the Goby one"
                                               << std::endl;
    }
 
    // SingleThreadApplication inherits initialize()/finalize() from both Application and Thread.
    // Only one set is ever invoked for a given application type, but guard anyway so that a
    // Python override cannot be called twice.
    void initialize() override
    {
        if (initialize_ && !initialize_called_)
        {
            initialize_called_ = true;
            initialize_();
        }
    }
 
    void finalize() override
    {
        if (finalize_ && !finalize_called_)
        {
            finalize_called_ = true;
            finalize_();
        }
    }
 
  private:
    Hook loop_;
    Hook initialize_;
    Hook finalize_;
    HealthHook health_;
    bool initialize_called_{false};
    bool finalize_called_{false};
};
 
template <typename App> class ApplicationWrapper
{
  public:
    ApplicationWrapper(double loop_frequency_hertz = 0)
    {
        if (!App::app_cfg_)
            throw(goby::Exception(
                "Goby configuration has not been read. Applications must be started with "
                "goby.run(), which parses the command line before constructing the application."));
 
        app_ptr_.reset(new App(loop_frequency_hertz));
        app_ptr_->set_hooks([this]() { this->loop(); }, [this]() { this->initialize(); },
                            [this]() { this->finalize(); },
                            [this](const std::string& serialized) -> std::string
                            {
                                // health() is called from the Goby event loop, which runs with
                                // the GIL released
                                py::gil_scoped_acquire gil;
                                return this->health(py::bytes(serialized));
                            });
    }
 
    virtual ~ApplicationWrapper() = default;
 
    ApplicationWrapper(const ApplicationWrapper&) = delete;
    ApplicationWrapper& operator=(const ApplicationWrapper&) = delete;
 
    virtual void loop()
    {
        // mirrors the diagnostic from Thread::loop()
        throw(goby::Exception("loop() must be overridden when loop_frequency is non-zero"));
    }
 
    virtual void initialize() {}
 
    virtual void finalize() {}
 
    virtual std::string health(py::bytes serialized) { return std::string(serialized); }
 
    static py::bytes configure(const std::vector<std::string>& argv)
    {
        using ConfigType = typename App::ConfigType;
 
        std::vector<char*> argv_c;
        argv_c.reserve(argv.size() + 1);
        for (const auto& arg : argv) argv_c.push_back(const_cast<char*>(arg.c_str()));
        argv_c.push_back(nullptr);
 
        ProtobufConfigurator<ConfigType> protobuf_cfgtor(static_cast<int>(argv.size()),
                                                         argv_c.data());
        // ProtobufConfigurator narrows the access of these overrides, so go through the
        // interface, as goby::run does
        const ConfiguratorInterface<ConfigType>& cfgtor = protobuf_cfgtor;
 
        try
        {
            cfgtor.validate();
        }
        catch (middleware::ConfigException& e)
        {
            cfgtor.handle_config_error(e);
            throw;
        }
 
        // simply print the configuration and exit
        if (cfgtor.app_configuration().debug_cfg())
        {
            std::cout << cfgtor.str() << std::endl;
            std::exit(EXIT_SUCCESS);
        }
 
        App::app_cfg_.reset(new ConfigType(cfgtor.cfg()));
        App::app3_base_configuration_.reset(
            new goby::middleware::protobuf::AppConfig(cfgtor.app_configuration()));
 
        goby::middleware::detail::configure_simulation_time(*App::app3_base_configuration_);
 
        return set_configuration(cfgtor.cfg());
    }
 
    static py::bytes configure_from_text(const std::string& config_text)
    {
        typename App::ConfigType cfg;
 
        google::protobuf::TextFormat::Parser parser;
        goby::util::FlexOStreamErrorCollector error_collector(config_text);
        parser.RecordErrorsTo(&error_collector);
        parser.AllowPartialMessage(false);
 
        if (!parser.ParseFromString(config_text, &cfg))
            throw(middleware::ConfigException(
                "Failed to parse the configuration as Protobuf TextFormat"));
 
        return set_configuration(cfg);
    }
 
    static py::bytes configure_from_serialized(const std::string& config_bytes)
    {
        typename App::ConfigType cfg;
 
        if (!cfg.ParseFromString(config_bytes))
            throw(middleware::ConfigException("Failed to parse the serialized configuration"));
 
        return set_configuration(cfg);
    }
 
    int run()
    {
        detail::SignalGuard signal_guard(&ApplicationWrapper::signal_quit, this);
        py::gil_scoped_release release_gil;
        return app_ptr_->__run();
    }
 
    void quit(int return_value = 0) { app_ptr_->quit(return_value); }
 
    std::string app_name() { return app_ptr_->app_name(); }
 
    py::bytes cfg_serialized()
    {
        std::string serialized;
        app_ptr_->app_cfg().SerializeToString(&serialized);
        return py::bytes(serialized);
    }
 
    void set_loop_frequency_hertz(double loop_frequency_hertz)
    {
        app_ptr_->set_loop_frequency_hertz(loop_frequency_hertz);
    }
 
    void publish(int layer, const std::string& type_name, int scheme, const std::string& group,
                 const std::string& data)
    {
        app_ptr_->publish(Identifier{static_cast<PubSubLayer>(layer), type_name, scheme, group},
                          data);
    }
 
    void subscribe(int layer, const std::string& type_name, int scheme, const std::string& group,
                   py::object callback)
    {
        app_ptr_->subscribe(Identifier{static_cast<PubSubLayer>(layer), type_name, scheme, group},
                            std::move(callback));
    }
 
  private:
    static py::bytes set_configuration(const typename App::ConfigType& cfg)
    {
        ConfigReader::check_required_cfg(cfg, cfg.app().binary());
 
        App::app_cfg_.reset(new typename App::ConfigType(cfg));
        App::app3_base_configuration_.reset(new goby::middleware::protobuf::AppConfig(cfg.app()));
 
        goby::middleware::detail::configure_simulation_time(*App::app3_base_configuration_);
 
        std::string serialized;
        cfg.SerializeToString(&serialized);
        return py::bytes(serialized);
    }
 
    static void signal_quit(void* self)
    {
        static_cast<ApplicationWrapper*>(self)->app_ptr_->quit();
    }
 
  private:
    std::unique_ptr<App> app_ptr_;
};
 
template <typename App> class ApplicationWrapperTrampoline : public ApplicationWrapper<App>
{
  public:
    using ApplicationWrapper<App>::ApplicationWrapper;
 
    void loop() override { PYBIND11_OVERRIDE(void, ApplicationWrapper<App>, loop, ); }
    void initialize() override { PYBIND11_OVERRIDE(void, ApplicationWrapper<App>, initialize, ); }
    void finalize() override { PYBIND11_OVERRIDE(void, ApplicationWrapper<App>, finalize, ); }
    std::string health(py::bytes serialized) override
    {
        PYBIND11_OVERRIDE_NAME(std::string, ApplicationWrapper<App>, "_goby_health", health,
                               serialized);
    }
};
 
namespace detail
{
inline void glog_write(int verbosity, const std::string& group, const std::string& text)
{
    py::gil_scoped_release release_gil;
 
    if (!goby::glog.is(static_cast<goby::util::logger::Verbosity>(verbosity)))
        return;
 
    if (!group.empty())
        goby::glog << ::group(group);
 
    goby::glog << text << std::endl;
}
 
inline bool glog_is(int verbosity)
{
    auto level = static_cast<goby::util::logger::Verbosity>(verbosity);
    if (level == goby::util::logger::DIE)
        return true;
 
    const auto* buf = dynamic_cast<const goby::util::FlexOStreamBuf*>(goby::glog.rdbuf());
    return buf != nullptr && buf->highest_verbosity() >= level;
}
 
} // namespace detail
 
template <typename App>
inline void define_python_module(py::module_& m, const std::string& app_name)
{
    // ConfigException is raised as goby.ConfigError so that all generated modules share one
    // exception type, which goby.run() catches
    py::register_exception_translator(
        [](std::exception_ptr p)
        {
            try
            {
                if (p)
                    std::rethrow_exception(p);
            }
            catch (const middleware::ConfigException& e)
            {
                py::object goby_module = py::module_::import("goby");
                PyErr_SetString(goby_module.attr("ConfigError").ptr(), e.what());
            }
        });
 
    py::class_<ApplicationWrapper<App>, ApplicationWrapperTrampoline<App>>(m, "_ApplicationBase")
        .def(py::init<double>(), py::arg("loop_frequency_hertz") = 0)
        .def("loop", &ApplicationWrapper<App>::loop)
        .def("initialize", &ApplicationWrapper<App>::initialize)
        .def("finalize", &ApplicationWrapper<App>::finalize)
        .def("quit", &ApplicationWrapper<App>::quit, py::arg("return_value") = 0)
        .def("app_name", &ApplicationWrapper<App>::app_name)
        .def("_run", &ApplicationWrapper<App>::run)
        .def("_publish", &ApplicationWrapper<App>::publish, py::arg("layer"), py::arg("type_name"),
             py::arg("scheme"), py::arg("group"), py::arg("data"))
        .def("_subscribe", &ApplicationWrapper<App>::subscribe, py::arg("layer"),
             py::arg("type_name"), py::arg("scheme"), py::arg("group"), py::arg("callback"))
        .def("_cfg_serialized", &ApplicationWrapper<App>::cfg_serialized)
        .def("_set_loop_frequency_hertz", &ApplicationWrapper<App>::set_loop_frequency_hertz)
        .def_static("_configure", &ApplicationWrapper<App>::configure, py::arg("argv"))
        .def_static("_configure_from_text", &ApplicationWrapper<App>::configure_from_text,
                    py::arg("config_text"))
        .def_static("_configure_from_serialized",
                    &ApplicationWrapper<App>::configure_from_serialized, py::arg("config_bytes"))
        .def("_goby_health", &ApplicationWrapper<App>::health, py::arg("serialized"));
 
    // Simulation time lives in process-wide statics rather than on the application, and is set
    // when the configuration is read. goby.time reads it from here so that a Python loop is
    // scheduled on the same warped clock the C++ loop runs on.
    m.def("_sim_time",
          []() {
              return py::make_tuple(
                  goby::time::SimulatorSettings::using_sim_time,
                  goby::time::SimulatorSettings::warp_factor,
                  static_cast<std::int64_t>(
                      goby::time::SimulatorSettings::reference_time.time_since_epoch() /
                      std::chrono::microseconds(1)));
          });
 
    m.def("_glog_write", &detail::glog_write, py::arg("verbosity"), py::arg("group"),
          py::arg("text"));
    m.def("_glog_is", &detail::glog_is, py::arg("verbosity"));
    m.def("_glog_add_group",
          [](const std::string& name, const std::string& description)
          { goby::glog.add_group(name, goby::util::Colors::nocolor, description); });
 
    m.attr("_GOBY_APPLICATION_NAME") = app_name;
}
 
} // namespace python
} // namespace middleware
} // namespace goby
 
// Macros for use by the autogenerated code to define the C++ side of the pub/sub setup.
//
// GROUP_KEY is the interface.yml expression for GROUP as a string, which is the only name
// Python has for the group; a group's runtime name need not be its C++ variable name.
#define GOBY_PYTHON_IF_PUBLICATION(SCHEME, LAYER_ENUM, LAYER_FUNCTION, GROUP, GROUP_KEY, TYPE)   \
    if (id == goby::middleware::python::Identifier{                                              \
                  goby::middleware::python::PubSubLayer::LAYER_ENUM, TYPE::descriptor()->name(), \
                  goby::middleware::MarshallingScheme::SCHEME, GROUP_KEY})                       \
    {                                                                                            \
        decltype(data.end()) actual_end;                                                         \
        auto msg = goby::middleware::SerializerParserHelper<                                     \
            TYPE, goby::middleware::MarshallingScheme::SCHEME>::parse(data.begin(), data.end(),  \
                                                                      actual_end);               \
        LAYER_FUNCTION().publish<GROUP>(msg);                                                    \
        return;                                                                                  \
    }
 
#define GOBY_PYTHON_IF_SUBSCRIPTION(SCHEME, LAYER_ENUM, LAYER_FUNCTION, GROUP, GROUP_KEY, TYPE)  \
    if (id == goby::middleware::python::Identifier{                                              \
                  goby::middleware::python::PubSubLayer::LAYER_ENUM, TYPE::descriptor()->name(), \
                  goby::middleware::MarshallingScheme::SCHEME, GROUP_KEY})                       \
    {                                                                                            \
        LAYER_FUNCTION().subscribe<GROUP>(                                                       \
            [callback](const TYPE& pb)                                                           \
            {                                                                                    \
                std::vector<char> bytes = goby::middleware::SerializerParserHelper<              \
                    TYPE, goby::middleware::MarshallingScheme::SCHEME>::serialize(pb);           \
                goby::middleware::python::detail::invoke_callback(callback, bytes);              \
            });                                                                                  \
        return;                                                                                  \
    }
 
#define GOBY_PYTHON_FAIL(PUBLISH_OR_SUBSCRIBE)                                              \
    goby::glog.is_die() &&                                                                  \
        goby::glog << PUBLISH_OR_SUBSCRIBE " not defined for these parameters: [" << id     \
                   << "]. Please include in interface.yml and re-generate to include them." \
                   << std::endl;
 
// used to stringify application name
#define GOBY_PYTHON_QUOTE(name) #name
#define GOBY_PYTHON_DEFINE_MODULE(APPLICATION_NAME, MODULE_NAME)          \
    PYBIND11_MODULE(MODULE_NAME, goby_python_module)                      \
    {                                                                     \
        goby::middleware::python::define_python_module<APPLICATION_NAME>( \
            goby_python_module, GOBY_PYTHON_QUOTE(APPLICATION_NAME));     \
    }
 
#endif