From c01cc551454376c31040bacbd8dbedda90586d6a Mon Sep 17 00:00:00 2001 From: Philipp Jungkamp Date: Mon, 9 Feb 2026 16:38:44 +0100 Subject: [PATCH 1/4] utils: Add glob helper function Signed-off-by: Philipp Jungkamp --- common/include/villas/utils.hpp | 4 ++ common/lib/utils.cpp | 66 +++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/common/include/villas/utils.hpp b/common/include/villas/utils.hpp index 318aeceac..0919013bf 100644 --- a/common/include/villas/utils.hpp +++ b/common/include/villas/utils.hpp @@ -131,6 +131,10 @@ template struct overloaded : Ts... { // Explicit deduction guide (not needed as of C++20) template overloaded(Ts...) -> overloaded; +// glob-style filesystem pattern matching +std::vector glob(fs::path const &pattern, + std::span searchDirectories); + void write_to_file(std::string data, const fs::path file); namespace base64 { diff --git a/common/lib/utils.cpp b/common/lib/utils.cpp index 2bbeda096..618f45a7e 100644 --- a/common/lib/utils.cpp +++ b/common/lib/utils.cpp @@ -12,12 +12,14 @@ #include #include #include +#include #include #include #include #include #include +#include #include #include #include @@ -352,6 +354,70 @@ bool isPrivileged() { return true; } +// internal glob implementation details +namespace { +bool isGlobPattern(fs::path const &path) { + static const auto specialCharacters = fs::path("?*[").native(); + auto const &string = path.native(); + return std::ranges::find_first_of(string, specialCharacters) != string.end(); +} + +bool isGlobMatch(fs::path const &pattern, fs::path const &path) { + return ::fnmatch(pattern.c_str(), path.c_str(), FNM_PATHNAME) == 0; +} + +void globImpl(std::vector &result, fs::path &&path, + std::ranges::subrange pattern) { + [[maybe_unused]] auto discardErrorCode = std::error_code{}; + + if (pattern.empty()) { + // we've reached the end of our pattern + if (fs::exists(path, discardErrorCode)) + result.push_back(path); + return; + } + + if (not fs::is_directory(path, discardErrorCode)) + return; + + if (not isGlobPattern(pattern.front())) { + path /= pattern.front(); + return globImpl(result, std::move(path), std::move(pattern).next()); + } else { + auto nextPattern = pattern.next(); + for (auto entry : fs::directory_iterator(path)) { + if (not isGlobMatch(pattern.front(), entry.path().filename())) + continue; + + globImpl(result, fs::path(entry.path()), nextPattern); + } + } +} +} // namespace + +std::vector glob(fs::path const &pattern, + std::span searchDirectories) { + auto logger = Log::get("glob"); + std::vector result; + if (pattern.is_absolute()) { + logger->debug("Matching absolute pattern {:?}", pattern.string()); + globImpl(result, pattern.root_path(), pattern); + } else { + for (auto path : searchDirectories) { + logger->debug("Matching relative pattern {:?} in {:?}", pattern.string(), + path.string()); + globImpl(result, std::move(path), pattern); + } + } + + if (result.empty()) { + throw std::runtime_error( + fmt::format("Could not find any file matching {:?}", pattern.string())); + } + + return result; +} + void write_to_file(std::string data, const fs::path file) { villas::Log::get("Filewriter")->debug("{} > {}", data, file.string()); std::ofstream outputFile(file.string()); From 0407b1d713ab67f1b88cf9adc94dfb818a2f714a Mon Sep 17 00:00:00 2001 From: Philipp Jungkamp Date: Mon, 26 Jan 2026 16:06:37 +0100 Subject: [PATCH 2/4] Parse configuration files using nlohmann::json Signed-off-by: Philipp Jungkamp --- CMakeLists.txt | 1 + common/include/villas/json.hpp | 32 +++ common/lib/CMakeLists.txt | 2 + common/lib/json.cpp | 370 ++++++++++++++++++++++++++++++++ include/villas/config_class.hpp | 96 --------- include/villas/super_node.hpp | 14 +- lib/CMakeLists.txt | 1 - lib/config.cpp | 368 ------------------------------- lib/nodes/fpga.cpp | 3 +- lib/super_node.cpp | 12 +- packaging/nix/villas.nix | 26 +-- tests/unit/config.cpp | 131 ++++++----- 12 files changed, 511 insertions(+), 545 deletions(-) create mode 100644 common/include/villas/json.hpp create mode 100644 common/lib/json.cpp delete mode 100644 include/villas/config_class.hpp delete mode 100644 lib/config.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 69f9e307c..79c0fb2ef 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -78,6 +78,7 @@ find_package(OpenSSL 1.0.0 REQUIRED) find_package(PkgConfig REQUIRED) find_package(spdlog REQUIRED) find_package(Threads REQUIRED) +find_package(nlohmann_json REQUIRED) find_package(OpenMP) find_package(IBVerbs) find_package(RDMACM) diff --git a/common/include/villas/json.hpp b/common/include/villas/json.hpp new file mode 100644 index 000000000..673ad83b9 --- /dev/null +++ b/common/include/villas/json.hpp @@ -0,0 +1,32 @@ +#pragma once + +#include +#include + +#include + +namespace villas { + +using Json = nlohmann::json; +using JsonPointer = nlohmann::json_pointer; + +// forward declaration for villas/jansson.hpp compatibility header +class JanssonPtr; +void to_json(Json &json, JanssonPtr const &jansson); +void from_json(Json const &json, JanssonPtr &jansson); + +// load a configuration file +Json load_config_deprecated(fs::path const &path, bool resolve_inc, + bool resolve_env); + +}; // namespace villas + +// forward declaration for libjansson's json_t +struct json_t; +void to_json(villas::Json &json, json_t const *jansson); + +template <> // format config_json using operator<< +struct fmt::formatter : ostream_formatter {}; + +template <> // format config_json_pointer using operator<< +struct fmt::formatter : ostream_formatter {}; diff --git a/common/lib/CMakeLists.txt b/common/lib/CMakeLists.txt index 497fe9994..816df3a42 100644 --- a/common/lib/CMakeLists.txt +++ b/common/lib/CMakeLists.txt @@ -15,6 +15,7 @@ add_library(villas-common SHARED cpuset.cpp dsp/pid.cpp hist.cpp + json.cpp kernel/kernel.cpp kernel/rt.cpp list.cpp @@ -55,6 +56,7 @@ target_include_directories(villas-common PUBLIC ) target_link_libraries(villas-common PUBLIC + nlohmann_json::nlohmann_json PkgConfig::JANSSON PkgConfig::UUID ${OPENSSL_LIBRARIES} diff --git a/common/lib/json.cpp b/common/lib/json.cpp new file mode 100644 index 000000000..126fbde6d --- /dev/null +++ b/common/lib/json.cpp @@ -0,0 +1,370 @@ +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include +#include +#include +#include + +using namespace std::string_view_literals; + +void to_json(villas::Json &json, ::json_t const *jansson) { + switch (json_typeof(jansson)) { + using enum json_type; + + case JSON_ARRAY: { + std::size_t index; + ::json_t const *value; + + json = villas::Json::array(); + json_array_foreach (jansson, index, value) + json.push_back(value); + } break; + + case JSON_OBJECT: { + char const *key; + std::size_t keylen; + ::json_t const *value; + + json = villas::Json::object(); + // The const_cast below is safe as long as we don't replace the contained value + // by accessing the underlying iterator with json_object_key_to_iter(key) and + // json_object_iter_set or json_object_iter_set_new. + // + // There is no API in jansson value to iterate or even enumerate keys of a const object. + // + // See https://github.com/akheron/jansson/issues/578 + json_object_keylen_foreach (const_cast<::json_t *>(jansson), key, keylen, + value) + json.emplace(std::string{key, keylen}, value); + } break; + + case JSON_STRING: { + json = std::string{json_string_value(jansson), json_string_length(jansson)}; + } break; + + case JSON_INTEGER: { + json = json_integer_value(jansson); + } break; + + case JSON_REAL: { + json = json_real_value(jansson); + } break; + + case JSON_TRUE: { + json = true; + } break; + + case JSON_FALSE: { + json = false; + } break; + + case JSON_NULL: + default: { + json = nullptr; + } break; + } +} + +namespace villas { + +void from_json(Json const &json, JanssonPtr &jansson) { + switch (json.type()) { + using value_t = Json::value_t; + + case value_t::array: { + jansson.reset(json_array()); + for (auto const &item : json) + json_array_append_new(jansson.get(), item.get().release()); + } break; + + case value_t::object: { + jansson.reset(json_object()); + for (auto const &[key, value] : json.items()) + json_object_setn_new_nocheck(jansson.get(), key.data(), key.size(), + value.get().release()); + } break; + + case value_t::string: { + auto const string = json.get(); + jansson.reset(json_stringn_nocheck(string->data(), string->size())); + } break; + + case value_t::number_integer: { + auto const integer = json.get(); + jansson.reset(json_integer(*integer)); + } break; + + case value_t::number_unsigned: { + auto const integer = json.get(); + jansson.reset(json_integer(*integer)); + } break; + + case value_t::number_float: { + auto const real = json.get(); + jansson.reset(json_real(*real)); + } break; + + case value_t::boolean: { + auto const boolean = json.get(); + jansson.reset(json_boolean(*boolean)); + } break; + + case value_t::null: { + jansson.reset(json_null()); + } break; + + case value_t::binary: + throw std::runtime_error{"cannot convert binary value to jansson"}; + + case value_t::discarded: + throw std::runtime_error{"cannot convert discarded value to jansson"}; + + default: + __builtin_unreachable(); + } +} + +void to_json(Json &json, JanssonPtr const &jansson) { + to_json(json, jansson.get()); +} + +namespace { + +// implementation of deprecated variable substitutions +void expand_substitutions_deprecated(Json &value, bool resolve_env, + fs::path const *include_dir) { + if (not value.is_string()) + return; + + if (not resolve_env and not include_dir) + return; + + constexpr static auto DEPRECATED_INCLUDE_KEYWORD = "@include "sv; + auto logger = Log::get("config"); + auto string = value.get_ref(); + auto do_include = false; + auto expanded = std::size_t{0}; + + // check for legacy @include keyword + if (include_dir and string.starts_with(DEPRECATED_INCLUDE_KEYWORD)) { + do_include = true; + expanded = DEPRECATED_INCLUDE_KEYWORD.length(); + } + + // legacy environment variable substitution syntax + static auto const env_regex = std::regex(R"--(\$\{([^}]+)\})--"); + enum : std::size_t { + CAPTURE_ALL = 0, + CAPTURE_NAME, + }; + + // expand deprecated environment substition syntax + std::smatch match; + while (resolve_env and std::regex_search(string.cbegin() + expanded, + string.cend(), match, env_regex)) { + auto name = std::string(match[CAPTURE_NAME]); + auto env_ptr = std::getenv(name.c_str()); + if (not env_ptr) + throw std::runtime_error( + fmt::format("Could substitute environment variable {:?}", name)); + + auto env_value = std::string_view(env_ptr); + auto [begin, end] = std::pair(match[CAPTURE_ALL]); + string.replace(begin, end, env_value.begin(), env_value.end()); + expanded += match.position() + env_value.length(); + } + + // expand deprecated @include directive + if (do_include) { + auto pattern = + std::string_view(string).substr(DEPRECATED_INCLUDE_KEYWORD.length()); + auto result = Json(nullptr); + for (auto path : utils::glob(pattern, std::span(include_dir, 1))) { + auto partial_result = + load_config_deprecated(path, include_dir != nullptr, resolve_env); + if (result.is_null()) + result = partial_result; + else if (partial_result.is_object() and result.is_object()) + result.update(partial_result, true); + else if (partial_result.is_array() and result.is_array()) + result.insert(result.end(), partial_result.begin(), + partial_result.end()); + } + + logger->warn("Found deprecated @include directive: {}", value); + value = std::move(result); + } else if (expanded) { + logger->warn("Found deprecated environment substitution: {}", value); + value = std::move(string); + } +} + +Json parse_libconfig_setting(::config_setting_t const *setting, + bool resolve_env, fs::path const *include_dir) { + auto logger = Log::get("config"); + + switch (config_setting_type(setting)) { + case CONFIG_TYPE_ARRAY: + case CONFIG_TYPE_LIST: { + auto array = Json::array(); + for (auto const idx : std::views::iota(0, config_setting_length(setting))) { + auto const elem = config_setting_get_elem(setting, idx); + array.push_back(parse_libconfig_setting(elem, resolve_env, include_dir)); + } + + return array; + } + + case CONFIG_TYPE_GROUP: { + auto object = Json::object(); + for (auto const idx : std::views::iota(0, config_setting_length(setting))) { + auto const elem = config_setting_get_elem(setting, idx); + auto name = std::string(config_setting_name(elem)); + object.emplace(std::move(name), + parse_libconfig_setting(elem, resolve_env, include_dir)); + } + + return object; + } + + case CONFIG_TYPE_STRING: { + auto json = Json(std::string(config_setting_get_string(setting))); + expand_substitutions_deprecated(json, resolve_env, include_dir); + return json; + } + + case CONFIG_TYPE_INT: { + return config_setting_get_int(setting); + } + + case CONFIG_TYPE_INT64: { + return std::int64_t{config_setting_get_int64(setting)}; + } + + case CONFIG_TYPE_FLOAT: { + return config_setting_get_float(setting); + } + + case CONFIG_TYPE_BOOL: { + return static_cast(config_setting_get_bool(setting)); + } + + case CONFIG_TYPE_NONE: + default: { + return nullptr; + } + } +} + +struct LibconfigHook { + bool resolve_env; + fs::path const *include_dir; +}; + +extern "C" char const **libconfig_include_func(::config_t *config, char const *, + char const *pattern, + char const **error) noexcept { + auto hook = static_cast(config_get_hook(config)); + auto paths = std::vector{}; + + if (not hook->include_dir) { + *error = "include directives are disabled"; + return nullptr; + } + + auto pattern_json = Json(pattern); + expand_substitutions_deprecated(pattern_json, hook->resolve_env, nullptr); + auto const &pattern_expanded = pattern_json.get_ref(); + + try { + paths = utils::glob(pattern_expanded, std::span(hook->include_dir, 1)); + std::erase_if(paths, [](auto const &path) { + auto ec = std::error_code{}; + return not fs::is_regular_file(path, ec); + }); + } catch (...) { + } + + if (paths.empty()) { + *error = "include directive did not match any file"; + return nullptr; + } + + auto ret = + static_cast(std::calloc(paths.size() + 1, sizeof(char *))); + auto index = std::size_t{0}; + for (auto &path : paths) + ret[index++] = strdup(path.c_str()); + + return ret; +} + +Json load_libconfig_deprecated(std::FILE *file, bool resolve_env, + fs::path const *include_dir) { + using ConfigDeleter = decltype([](::config_t *c) { ::config_destroy(c); }); + using ConfigPtr = std::unique_ptr<::config_t, ConfigDeleter>; + + auto hook = LibconfigHook{ + .resolve_env = resolve_env, + .include_dir = include_dir, + }; + + ::config_t config; + ::config_init(&config); + auto guard = ConfigPtr(&config); + + ::config_set_hook(&config, &hook); + ::config_set_include_func(&config, &libconfig_include_func); + if (auto ret = ::config_read(&config, file); ret != CONFIG_TRUE) { + throw std::runtime_error(fmt::format("Failed to load libconfig file: {}", + config_error_text(&config))); + } + + return parse_libconfig_setting(config_root_setting(&config), resolve_env, + include_dir); +} + +} // namespace + +Json load_config_deprecated(fs::path const &path, bool resolve_inc, + bool resolve_env) { + using FileDeleter = decltype([](std::FILE *c) { std::fclose(c); }); + using FilePtr = std::unique_ptr; + + auto file = FilePtr(std::fopen(path.c_str(), "r")); + if (not file) { + throw std::runtime_error( + fmt::format("Failed to open config file {}", path.string())); + } + + auto include_dir = path.parent_path(); + if (auto ext = path.extension(); ext == ".json") { + auto parser_callback = [&](int depth, Json::parse_event_t event, + Json &value) { + if (event == Json::parse_event_t::value) + expand_substitutions_deprecated(value, resolve_env, + resolve_inc ? &include_dir : nullptr); + + return true; + }; + + return Json::parse(file.get(), parser_callback); + } else if (ext == ".conf") { + return load_libconfig_deprecated(file.get(), resolve_env, + resolve_inc ? &include_dir : nullptr); + } else { + throw std::runtime_error(fmt::format( + "Failed to load config file with unknown extension {}", ext.string())); + } +} + +} // namespace villas diff --git a/include/villas/config_class.hpp b/include/villas/config_class.hpp deleted file mode 100644 index e6b372b9c..000000000 --- a/include/villas/config_class.hpp +++ /dev/null @@ -1,96 +0,0 @@ -/* Configuration file parsing. - * - * Author: Steffen Vogel - * SPDX-FileCopyrightText: 2014-2023 Institute for Automation of Complex Power Systems, RWTH Aachen University - * SPDX-License-Identifier: Apache-2.0 - */ - -#pragma once - -#include -#include - -#include -#include - -#include -#include -#include - -#ifdef WITH_CONFIG -#include -#endif - -namespace villas { -namespace node { - -class Config { - -protected: - using str_walk_fcn_t = std::function; - - Logger logger; - - std::list includeDirectories; - std::string configPath; - - // Check if file exists on local system. - static bool isLocalFile(const std::string &uri) { - return access(uri.c_str(), F_OK) != -1; - } - - // Decode configuration file. - json_t *decode(FILE *f); - -#ifdef WITH_CONFIG - // Convert libconfig .conf file to libjansson .json file. - json_t *libconfigDecode(FILE *f); - - static const char **includeFuncStub(config_t *cfg, const char *include_dir, - const char *path, const char **error); - - const char **includeFunc(config_t *cfg, const char *include_dir, - const char *path, const char **error); -#endif // WITH_CONFIG - - // Load configuration from standard input (stdim). - FILE *loadFromStdio(); - - // Load configuration from local file. - FILE *loadFromLocalFile(const std::string &u); - - std::list resolveIncludes(const std::string &name); - - void resolveEnvVars(std::string &text); - - // Resolve custom include directives. - json_t *expandIncludes(json_t *in); - - // To shell-like subsitution of environment variables in strings. - json_t *expandEnvVars(json_t *in); - - // Run a callback function for each string in the config - json_t *walkStrings(json_t *in, str_walk_fcn_t cb); - - // Get the include dirs - std::list getIncludeDirectories(FILE *f) const; - -public: - json_t *root; - - Config(); - Config(const std::string &u); - - ~Config(); - - json_t *load(std::FILE *f, bool resolveIncludes = true, - bool resolveEnvVars = true); - - json_t *load(const std::string &u, bool resolveIncludes = true, - bool resolveEnvVars = true); - - std::string const &getConfigPath() const { return configPath; } -}; - -} // namespace node -} // namespace villas diff --git a/include/villas/super_node.hpp b/include/villas/super_node.hpp index f4836fdfd..7f197ebfb 100644 --- a/include/villas/super_node.hpp +++ b/include/villas/super_node.hpp @@ -15,11 +15,10 @@ extern "C" { } #endif -#include - #include #include -#include +#include +#include #include #include #include @@ -68,7 +67,8 @@ class SuperNode { struct timespec started; // The time at which the instance has been started. - Config config; // The configuration file. + fs::path configPath; + JanssonPtr configRoot; // The configuration file. public: // Inititalize configuration object before parsing the configuration. @@ -77,7 +77,7 @@ class SuperNode { int init(); // Wrapper for parse() which loads the config first. - void parse(const std::string &name); + void parse(fs::path const &path); /* Parse super-node configuration. * @@ -138,9 +138,9 @@ class SuperNode { Web *getWeb() { return &web; } #endif - json_t *getConfig() { return config.root; } + json_t *getConfig() { return configRoot.get(); } - const std::string &getConfigPath() const { return config.getConfigPath(); } + fs::path const &getConfigPath() const { return configPath; } int getAffinity() const { return affinity; } diff --git a/lib/CMakeLists.txt b/lib/CMakeLists.txt index e95215e5a..e0e904018 100644 --- a/lib/CMakeLists.txt +++ b/lib/CMakeLists.txt @@ -23,7 +23,6 @@ set(LIBRARIES set(LIB_SRC capabilities.cpp config_helper.cpp - config.cpp dumper.cpp format.cpp mapping.cpp diff --git a/lib/config.cpp b/lib/config.cpp deleted file mode 100644 index 0bab007ef..000000000 --- a/lib/config.cpp +++ /dev/null @@ -1,368 +0,0 @@ -/* Configuration file parsing. - * - * Author: Steffen Vogel - * SPDX-FileCopyrightText: 2014-2023 Institute for Automation of Complex Power Systems, RWTH Aachen University - * SPDX-License-Identifier: Apache-2.0 - */ - -#include -#include - -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include -#include - -#ifdef WITH_CONFIG -#include -#endif - -using namespace villas; -using namespace villas::node; - -Config::Config() : logger(Log::get("config")), root(nullptr) {} - -Config::Config(const std::string &u) : Config() { root = load(u); } - -Config::~Config() { json_decref(root); } - -json_t *Config::load(std::FILE *f, bool resolveInc, bool resolveEnvVars) { - json_t *root = decode(f); - - if (resolveInc) { - json_t *root_old = root; - root = expandIncludes(root); - json_decref(root_old); - } - - if (resolveEnvVars) { - json_t *root_old = root; - root = expandEnvVars(root); - json_decref(root_old); - } - - return root; -} - -json_t *Config::load(const std::string &u, bool resolveInc, - bool resolveEnvVars) { - FILE *f; - - if (u == "-") - f = loadFromStdio(); - else - f = loadFromLocalFile(u); - - json_t *root = load(f, resolveInc, resolveEnvVars); - - fclose(f); - - return root; -} - -FILE *Config::loadFromStdio() { - logger->info("Reading configuration from standard input"); - - auto *cwd = new char[PATH_MAX]; - - configPath = getcwd(cwd, PATH_MAX); - - delete[] cwd; - - return stdin; -} - -FILE *Config::loadFromLocalFile(const std::string &u) { - logger->info("Reading configuration from local file: {}", u); - - configPath = u; - FILE *f = fopen(u.c_str(), "r"); - if (!f) - throw RuntimeError("Failed to open configuration from: {}", u); - - return f; -} - -json_t *Config::decode(FILE *f) { - json_error_t err; - - // Update list of include directories - auto incDirs = getIncludeDirectories(f); - includeDirectories.insert(includeDirectories.end(), incDirs.begin(), - incDirs.end()); - - json_t *root = json_loadf(f, 0, &err); - if (root == nullptr) { -#ifdef WITH_CONFIG - // We try again to parse the config in the legacy format - root = libconfigDecode(f); -#else - throw JanssonParseError(err); -#endif // WITH_CONFIG - } - - return root; -} - -std::list Config::getIncludeDirectories(FILE *f) const { - int ret, fd; - char buf[PATH_MAX]; - char *dir; - - std::list dirs; - - // Adding directory of base configuration file - fd = fileno(f); - if (fd < 0) - throw SystemError("Failed to get file descriptor"); - - auto path = fmt::format("/proc/self/fd/{}", fd); - - ret = readlink(path.c_str(), buf, sizeof(buf)); - if (ret > 0) { - buf[ret] = 0; - if (isLocalFile(buf)) { - dir = dirname(buf); - dirs.push_back(dir); - } - } - - // Adding current working directory - dir = getcwd(buf, sizeof(buf)); - if (dir != nullptr) - dirs.push_back(dir); - - return dirs; -} - -std::list Config::resolveIncludes(const std::string &n) { - glob_t gb; - int ret, flags = 0; - - memset(&gb, 0, sizeof(gb)); - - auto name = n; - resolveEnvVars(name); - - if (name.size() >= 1 && name[0] == '/') { // absolute path - ret = glob(name.c_str(), flags, nullptr, &gb); - if (ret && ret != GLOB_NOMATCH) - gb.gl_pathc = 0; - } else { // relative path - for (auto &dir : includeDirectories) { - auto pattern = fmt::format("{}/{}", dir, name.c_str()); - - ret = glob(pattern.c_str(), flags, nullptr, &gb); - if (ret && ret != GLOB_NOMATCH) { - gb.gl_pathc = 0; - - goto out; - } - - flags |= GLOB_APPEND; - } - } - -out: - std::list files; - for (unsigned i = 0; i < gb.gl_pathc; i++) - files.push_back(gb.gl_pathv[i]); - - globfree(&gb); - - return files; -} - -void Config::resolveEnvVars(std::string &text) { - static const std::regex env_re{R"--(\$\{([^}]+)\})--"}; - - std::smatch match; - while (std::regex_search(text, match, env_re)) { - auto const from = match[0]; - auto const var_name = match[1].str(); - char *var_value = std::getenv(var_name.c_str()); - if (!var_value) - throw RuntimeError("Unresolved environment variable: {}", var_name); - - text.replace(from.first - text.begin(), from.second - from.first, - var_value); - - logger->debug("Replace env var {} in \"{}\" with value \"{}\"", var_name, - text, var_value); - } -} - -#ifdef WITH_CONFIG -#if (LIBCONFIG_VER_MAJOR > 1) || \ - ((LIBCONFIG_VER_MAJOR == 1) && (LIBCONFIG_VER_MINOR >= 7)) -const char **Config::includeFuncStub(config_t *cfg, const char *include_dir, - const char *path, const char **error) { - void *ctx = config_get_hook(cfg); - - return reinterpret_cast(ctx)->includeFunc(cfg, include_dir, path, - error); -} - -const char **Config::includeFunc(config_t *cfg, const char *include_dir, - const char *path, const char **error) { - auto paths = resolveIncludes(path); - - unsigned i = 0; - auto files = (const char **)malloc(sizeof(char **) * (paths.size() + 1)); - - for (auto &path : paths) - files[i++] = strdup(path.c_str()); - - files[i] = NULL; - - return files; -} -#endif - -json_t *Config::libconfigDecode(FILE *f) { - int ret; - - config_t cfg; - config_setting_t *cfg_root; - config_init(&cfg); - config_set_auto_convert(&cfg, 1); - - // Setup libconfig include path -#if (LIBCONFIG_VER_MAJOR > 1) || \ - ((LIBCONFIG_VER_MAJOR == 1) && (LIBCONFIG_VER_MINOR >= 7)) - config_set_hook(&cfg, this); - - config_set_include_func(&cfg, includeFuncStub); -#else - if (includeDirectories.size() > 0) { - logger->info("Setting include dir to: {}", includeDirectories.front()); - - config_set_include_dir(&cfg, includeDirectories.front().c_str()); - - if (includeDirectories.size() > 1) { - logger->warn( - "Ignoring all but the first include directories for libconfig"); - logger->warn( - " libconfig does not support more than a single include dir!"); - } - } -#endif - - // Rewind before re-reading - rewind(f); - - ret = config_read(&cfg, f); - if (ret != CONFIG_TRUE) - throw LibconfigParseError(&cfg); - - cfg_root = config_root_setting(&cfg); - - json_t *root = config_to_json(cfg_root); - if (!root) - throw RuntimeError("Failed to convert JSON to configuration file"); - - config_destroy(&cfg); - - return root; -} -#endif // WITH_CONFIG - -json_t *Config::walkStrings(json_t *root, str_walk_fcn_t cb) { - const char *key; - size_t index; - json_t *val, *new_val, *new_root; - - switch (json_typeof(root)) { - case JSON_STRING: - return cb(root); - - case JSON_OBJECT: - new_root = json_object(); - - json_object_foreach (root, key, val) { - new_val = walkStrings(val, cb); - - json_object_set_new(new_root, key, new_val); - } - - return new_root; - - case JSON_ARRAY: - new_root = json_array(); - - json_array_foreach (root, index, val) { - new_val = walkStrings(val, cb); - - json_array_append_new(new_root, new_val); - } - - return new_root; - - default: - return json_incref(root); - }; -} - -json_t *Config::expandEnvVars(json_t *in) { - return walkStrings(in, [this](json_t *str) -> json_t * { - std::string text = json_string_value(str); - - resolveEnvVars(text); - - return json_string(text.c_str()); - }); -} - -json_t *Config::expandIncludes(json_t *in) { - return walkStrings(in, [this](json_t *str) -> json_t * { - int ret; - std::string text = json_string_value(str); - static const std::string kw = "@include "; - - auto res = std::mismatch(kw.begin(), kw.end(), text.begin()); - if (res.first != kw.end()) - return json_incref(str); - else { - std::string pattern = text.substr(kw.size()); - - resolveEnvVars(pattern); - - json_t *incl = nullptr; - - for (auto &path : resolveIncludes(pattern)) { - json_t *other = load(path); - if (!other) - throw ConfigError(str, "include", - "Failed to include config file from {}", path); - - if (!incl) - incl = other; - else if (json_is_object(incl) && json_is_object(other)) { - ret = json_object_update_recursive(incl, other); - if (ret) - throw ConfigError( - str, "include", - "Can not mix object and array-typed include files"); - } else if (json_is_array(incl) && json_is_array(other)) { - ret = json_array_extend(incl, other); - if (ret) - throw ConfigError( - str, "include", - "Can not mix object and array-typed include files"); - } - - logger->debug("Included config from: {}", path); - } - - return incl; - } - }); -} diff --git a/lib/nodes/fpga.cpp b/lib/nodes/fpga.cpp index 4d8ccc3f3..e4af36d3b 100644 --- a/lib/nodes/fpga.cpp +++ b/lib/nodes/fpga.cpp @@ -381,8 +381,7 @@ int FpgaNodeFactory::start(SuperNode *sn) { } if (cards.empty()) { - auto searchPath = - sn->getConfigPath().substr(0, sn->getConfigPath().rfind("/")); + auto searchPath = sn->getConfigPath(); createCards(sn->getConfig(), cards, searchPath, vfioContainer); } diff --git a/lib/super_node.cpp b/lib/super_node.cpp index 110d194c3..4ba712753 100644 --- a/lib/super_node.cpp +++ b/lib/super_node.cpp @@ -7,7 +7,8 @@ #include #include -#include + +#include #include #include @@ -22,6 +23,8 @@ #include #include +#include "villas/json.hpp" + #ifdef WITH_NETEM #include #endif @@ -60,10 +63,11 @@ SuperNode::SuperNode() logger = Log::get("super_node"); } -void SuperNode::parse(const std::string &u) { - config.root = config.load(u); +void SuperNode::parse(fs::path const &path) { + configPath = path; + load_config_deprecated(path, true, true).get_to(configRoot); - parse(config.root); + parse(configRoot.get()); } void SuperNode::parse(json_t *root) { diff --git a/packaging/nix/villas.nix b/packaging/nix/villas.nix index 246150a67..51d3e982b 100644 --- a/packaging/nix/villas.nix +++ b/packaging/nix/villas.nix @@ -1,9 +1,13 @@ # SPDX-FileCopyrightText: 2023 OPAL-RT Germany GmbH # SPDX-License-Identifier: Apache-2.0 { + lib, + stdenv, + makeWrapper, # General configuration src, version, + system, withGpl ? true, withAllExtras ? false, withAllFormats ? false, @@ -41,23 +45,24 @@ bash, cmake, coreutils, + curl, + gnugrep, graphviz, + jansson, jq, - lib, - makeWrapper, + libuuid, + libwebsockets, + nlohmann_json, + openssl, pkg-config, - stdenv, - system, + spdlog, # Optional dependencies boxfort, comedilib, criterion, - curl, czmq, cyrus_sasl, ethercat, - gnugrep, - jansson, lib60870, libconfig, libdatachannel, @@ -70,14 +75,11 @@ libsodium, libuldaq, libusb1, - libuuid, - libwebsockets, libxml2, lua, mosquitto, nanomsg, opendssc, - openssl, orchestra, pcre2, pkgsBuildBuild, @@ -89,7 +91,6 @@ rdkafka, rdma-core, redis-plus-plus, - spdlog, linuxHeaders, }: stdenv.mkDerivation { @@ -190,8 +191,9 @@ stdenv.mkDerivation { ]; propagatedBuildInputs = [ - libuuid jansson + libuuid + nlohmann_json ] ++ lib.optionals withFormatProtobuf [ protobuf diff --git a/tests/unit/config.cpp b/tests/unit/config.cpp index 0ecd3a2fe..3e6774a07 100644 --- a/tests/unit/config.cpp +++ b/tests/unit/config.cpp @@ -9,68 +9,89 @@ #include #include +#include +#include -#include +#include +#include +#include #include -using namespace villas::node; +using namespace std::string_view_literals; +using FileDeleter = decltype([](std::FILE *f) { std::fclose(f); }); +using FilePtr = std::unique_ptr; + +constexpr auto fileNameTemplate = "villas.unit-test.XXXXXX.conf"sv; +constexpr auto fileNameSuffix = ".conf"sv; // cppcheck-suppress syntaxError Test(config, env) { - const char *cfg_f = "test = \"${MY_ENV_VAR}\"\n"; - - std::FILE *f = std::tmpfile(); - std::fputs(cfg_f, f); - std::rewind(f); - - auto c = Config(); - - char env[] = "MY_ENV_VAR=mobydick"; - putenv(env); - - auto *r = c.load(f); - cr_assert_not_null(r); - - auto *j = json_object_get(r, "test"); - cr_assert_not_null(j); - - cr_assert(json_is_string(j)); - cr_assert_str_eq("mobydick", json_string_value(j)); + auto config_string = R"libconfig( + test = "${MY_ENV_VAR}" + )libconfig"; + + auto config_path_template = + std::string(fs::temp_directory_path() / fileNameTemplate); + auto config_fd = + ::mkstemps(config_path_template.data(), fileNameSuffix.length()); + auto config_file = FilePtr(::fdopen(config_fd, "w")); + auto config_path = fs::path(config_path_template); + std::fputs(config_string, config_file.get()); + config_file.reset(); + + auto config = villas::load_config_deprecated(config_path, true, true) + .get(); + ::setenv("MY_ENV_VAR", "mobydick", true); + + auto *root = config.get(); + cr_assert_not_null(root); + + auto *string = json_object_get(root, "test"); + cr_assert_not_null(string); + cr_assert(json_is_string(string)); + cr_assert_str_eq("mobydick", json_string_value(string)); } Test(config, include) { - const char *cfg_f2 = "magic = 1234\n"; - - char f2_fn_tpl[] = "/tmp/villas.unit-test.XXXXXX"; - int f2_fd = mkstemp(f2_fn_tpl); - - std::FILE *f2 = fdopen(f2_fd, "w"); - std::fputs(cfg_f2, f2); - std::rewind(f2); - - auto cfg_f1 = fmt::format("subval = \"@include {}\"\n", f2_fn_tpl); - - std::FILE *f1 = std::tmpfile(); - std::fputs(cfg_f1.c_str(), f1); - std::rewind(f1); - - auto env = fmt::format("{}", f2_fn_tpl); - setenv("INCLUDE_FILE", env.c_str(), true); - - auto c = Config(); - - auto *r = c.load(f1); - cr_assert_not_null(r); - - auto *j = json_object_get(r, "subval"); - cr_assert_not_null(j); - - auto *j2 = json_object_get(j, "magic"); - cr_assert_not_null(j2); - - cr_assert(json_is_integer(j2)); - cr_assert_eq(json_number_value(j2), 1234); - - std::fclose(f2); - std::remove(f2_fn_tpl); + auto incString = R"libconfig( + magic = 1234 + )libconfig"; + + auto inc_path_template = + std::string(fs::temp_directory_path() / fileNameTemplate); + auto inc_fd = ::mkstemps(inc_path_template.data(), fileNameSuffix.length()); + cr_assert(inc_fd >= 0); + auto inc_file = FilePtr(::fdopen(inc_fd, "w")); + cr_assert_not_null(inc_file); + auto inc_path = fs::path(inc_path_template); + std::fputs(incString, inc_file.get()); + inc_file.reset(); + + auto config_string = fmt::format(R"libconfig( + subval = {{ @include "{}" }} + )libconfig", + inc_path.string()); + auto config_path_template = + std::string(fs::temp_directory_path() / fileNameTemplate); + auto config_fd = + ::mkstemps(config_path_template.data(), fileNameSuffix.length()); + cr_assert(config_fd >= 0); + auto config_file = FilePtr(::fdopen(config_fd, "w")); + cr_assert_not_null(config_file); + auto config_path = fs::path(config_path_template); + std::fputs(config_string.c_str(), config_file.get()); + config_file.reset(); + + auto config = villas::load_config_deprecated(config_path, true, true) + .get(); + auto *root = config.get(); + cr_assert_not_null(root); + + auto *subval = json_object_get(root, "subval"); + cr_assert_not_null(subval); + + auto *magic = json_object_get(subval, "magic"); + cr_assert_not_null(magic); + cr_assert(json_is_integer(magic)); + cr_assert_eq(json_number_value(magic), 1234); } From 18929ce970003b50f5213f422d83b6b93ca176cb Mon Sep 17 00:00:00 2001 From: Philipp Jungkamp Date: Wed, 25 Feb 2026 16:02:47 +0100 Subject: [PATCH 3/4] tests: Use /usr/bin/env bash instead of /bin/bash Signed-off-by: Philipp Jungkamp --- tests/integration/CMakeLists.txt | 2 +- tests/unit/CMakeLists.txt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/integration/CMakeLists.txt b/tests/integration/CMakeLists.txt index 777ea498d..0bf5002db 100644 --- a/tests/integration/CMakeLists.txt +++ b/tests/integration/CMakeLists.txt @@ -6,7 +6,7 @@ add_custom_target(run-integration-tests COMMAND - /bin/bash -o pipefail -c \" + /usr/bin/env bash -o pipefail -c \" SRCDIR=${PROJECT_SOURCE_DIR} BUILDDIR=${PROJECT_BINARY_DIR} ${PROJECT_SOURCE_DIR}/tools/integration-tests.sh 2>&1 | c++filt\" diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt index f9098e53e..7f9ddd71c 100644 --- a/tests/unit/CMakeLists.txt +++ b/tests/unit/CMakeLists.txt @@ -28,7 +28,7 @@ target_link_libraries(unit-tests PUBLIC add_custom_target(run-unit-tests COMMAND - /bin/bash -o pipefail -c \" + /usr/bin/env bash -o pipefail -c \" $ 2>&1 | c++filt\" DEPENDS unit-tests From fa1367c1b73eb1a7394e242718076d494805860a Mon Sep 17 00:00:00 2001 From: Philipp Jungkamp Date: Tue, 30 Jun 2026 20:48:24 +0200 Subject: [PATCH 4/4] nodes(c37.118): Introduce new node type. Signed-off-by: Philipp Jungkamp --- .../components/schemas/config/node_obj.yaml | 3 +- .../schemas/config/nodes/_c37_118.yaml | 7 + .../schemas/config/nodes/c37_118.yaml | 321 +++ etc/examples/nodes/c37_118.conf | 27 + include/villas/nodes/c37_118.hpp | 338 +++ lib/nodes/CMakeLists.txt | 1 + lib/nodes/c37_118.cpp | 2206 +++++++++++++++++ tests/unit/CMakeLists.txt | 2 + tests/unit/c37_118.cpp | 187 ++ 9 files changed, 3091 insertions(+), 1 deletion(-) create mode 100644 doc/openapi/components/schemas/config/nodes/_c37_118.yaml create mode 100644 doc/openapi/components/schemas/config/nodes/c37_118.yaml create mode 100644 etc/examples/nodes/c37_118.conf create mode 100644 include/villas/nodes/c37_118.hpp create mode 100644 lib/nodes/c37_118.cpp create mode 100644 tests/unit/c37_118.cpp diff --git a/doc/openapi/components/schemas/config/node_obj.yaml b/doc/openapi/components/schemas/config/node_obj.yaml index c3e796171..0bfd18bdd 100644 --- a/doc/openapi/components/schemas/config/node_obj.yaml +++ b/doc/openapi/components/schemas/config/node_obj.yaml @@ -3,7 +3,7 @@ # SPDX-License-Identifier: Apache-2.0 --- type: object -title: Format Object +title: Node Object required: - type properties: @@ -21,6 +21,7 @@ discriminator: mapping: amqp: nodes/_amqp.yaml api: nodes/_api.yaml + c37.118: nodes/_c37_118.yaml can: nodes/_can.yaml comedi: nodes/_comedi.yaml ethercat: nodes/_ethercat.yaml diff --git a/doc/openapi/components/schemas/config/nodes/_c37_118.yaml b/doc/openapi/components/schemas/config/nodes/_c37_118.yaml new file mode 100644 index 000000000..588a9630d --- /dev/null +++ b/doc/openapi/components/schemas/config/nodes/_c37_118.yaml @@ -0,0 +1,7 @@ +# yaml-language-server: $schema=http://json-schema.org/draft-07/schema +# SPDX-FileCopyrightText: 2024-2025 Institute for Automation of Complex Power Systems, RWTH Aachen University +# SPDX-License-Identifier: Apache-2.0 +--- +allOf: +- $ref: ../node_obj.yaml +- $ref: c37_118.yaml diff --git a/doc/openapi/components/schemas/config/nodes/c37_118.yaml b/doc/openapi/components/schemas/config/nodes/c37_118.yaml new file mode 100644 index 000000000..a51daa525 --- /dev/null +++ b/doc/openapi/components/schemas/config/nodes/c37_118.yaml @@ -0,0 +1,321 @@ +# yaml-language-server: $schema=http://json-schema.org/draft-07/schema +# SPDX-FileCopyrightText: 2024-2025 Institute for Automation of Complex Power Systems, RWTH Aachen University +# SPDX-License-Identifier: Apache-2.0 +--- +allOf: +- type: object + properties: + + in: + type: object + description: | + Client (PDC) side. When an address is given, the node connects to a + remote PMU / PDC, requests its configuration and reads data frames. + properties: + + address: + type: string + description: | + Hostname or IP address of the remote PMU / PDC in the format + `host[:port]`. The port defaults to the C37.118 port 4712 when + omitted. + + idcode: + type: integer + default: 1 + minimum: 0 + maximum: 65535 + description: | + IDCODE placed in the command frames sent to the remote device. + + out: + type: object + description: | + Server (PMU / PDC) side. When an address is given, the node listens for + a connecting PDC, answers configuration and command frames and streams + data frames built from the samples written to the node. + required: + - address + - data_rate + - pmus + properties: + + address: + type: string + description: | + Local address to bind to in the format `host[:port]`. An empty host + binds to all interfaces. The port defaults to the C37.118 port 4712 + when omitted. + + idcode: + type: integer + default: 1 + minimum: 0 + maximum: 65535 + description: | + IDCODE reported in the frames served to the connecting PDC. + + time_base: + type: integer + default: 1000000 + minimum: 1 + maximum: 16777215 + description: | + Resolution of the fractional second (FRACSEC) timestamp, i.e. the + number of sub-second units per second. The C37.118 TIME_BASE field + is 24 bits wide, so the value must not exceed 16777215. + + data_rate: + type: number + minimum: 3.05175e-5 + maximum: 32767 + description: | + Reporting rate in frames per second. Rates below one frame per + second are supported (e.g. 0.5 for one frame every two seconds) and + are encoded using the C37.118 seconds-per-frame representation. + + pmus: + type: array + minItems: 1 + items: + $ref: '#/definitions/node-c37.118-pmu' + description: | + The list of PMU configurations served by this node. + +- $ref: ../node.yaml + +definitions: + node-c37.118-pmu: + type: object + description: | + Configuration of a single PMU. + required: + - name + - frequency + - rocof + properties: + + name: + type: string + maxLength: 255 + description: | + Station name of the PMU. Clients using revision 2005 of the protocol + might see the name truncated to 16 characters. + + idcode: + type: integer + default: 1 + minimum: 0 + maximum: 65535 + description: | + Data stream ID code of the PMU. + + guid: + type: string + format: uuid + description: | + Global PMU identifier. Defaults to the node's UUID. + + nominal_frequency: + type: number + enum: + - 50 + - 60 + default: 50 + description: | + Nominal line frequency in Hz. + + latitude: + type: number + description: | + Latitude of the PMU in degrees (CONFIG3 only). Defaults to unknown. + + longitude: + type: number + description: | + Longitude of the PMU in degrees (CONFIG3 only). Defaults to unknown. + + elevation: + type: number + description: | + Elevation of the PMU in meters (CONFIG3 only). Defaults to unknown. + + service_class: + type: string + default: measurement + enum: + - measurement + - protection + description: | + Performance/service class of the PMU (CONFIG3 only). `measurement` + maps to service class M, `protection` to service class P. + + window: + type: integer + default: 0 + minimum: 0 + description: | + Phasor measurement window length in microseconds (CONFIG3 only). + + group_delay: + type: integer + default: 0 + minimum: 0 + description: | + Phasor measurement group delay in microseconds (CONFIG3 only). + + frequency: + type: string + description: | + Name of the input signal providing the measured frequency (server + direction only). + + rocof: + type: string + description: | + Name of the input signal providing the rate of change of frequency + (server direction only). + + phasor: + type: array + items: + $ref: '#/definitions/node-c37.118-phasor' + description: | + The phasor channels of the PMU. + + analog: + type: array + items: + $ref: '#/definitions/node-c37.118-analog' + description: | + The analog channels of the PMU. + + digital: + type: array + items: + $ref: '#/definitions/node-c37.118-digital' + description: | + The digital status bits of the PMU. Every 16 bits form one digital + status word. + + node-c37.118-phasor: + type: object + description: | + Configuration of a single phasor channel. + required: + - signal + - unit + properties: + + signal: + type: string + description: | + Name of the input signal mapped to this phasor (server direction). + + name: + type: string + maxLength: 255 + description: | + Name of the phasor channel. Defaults to the signal name. Clients using + revision 2005 of the protocol might see the name truncated to 16 + characters. + + unit: + type: string + enum: + - volt + - ampere + description: | + Physical unit of the phasor. + + component: + type: string + default: positive_sequence + enum: + - zero_sequence + - positive_sequence + - negative_sequence + - phase_a + - phase_b + - phase_c + description: | + The phasor component (sequence or phase) that this phasor represents. + + modifications: + type: array + uniqueItems: true + description: | + Measurement modifications applied to the phasor (CONFIG3 only). + items: + type: string + enum: + - upsampled_with_interpolation + - upsampled_with_extrapolation + - downsampled_with_reselection + - downsampled_with_fir_filter + - downsampled_with_non_fir_filter + - filtered_without_resampling + - magnitude_adjusted_for_calibration + - phase_adjusted_for_calibration + - phase_adjusted_for_rotation + - pseudo_phasor_value + - other + + node-c37.118-analog: + type: object + description: | + Configuration of a single analog channel. + required: + - signal + properties: + + signal: + type: string + description: | + Name of the input signal mapped to this analog channel (server + direction). + + name: + type: string + maxLength: 255 + description: | + Name of the analog channel. Defaults to the signal name. Clients using + revision 2005 of the protocol might see the name truncated to 16 + characters. + + unit: + type: string + default: point_on_wave + enum: + - point_on_wave + - rms + - peak + description: | + Type of the analog value. + + node-c37.118-digital: + type: object + description: | + Configuration of a single digital status bit. Every 16 bits form one + digital status word. + required: + - signal + properties: + + signal: + type: string + description: | + Name of the input signal mapped to this bit (server direction). + + name: + type: string + maxLength: 255 + description: | + Name of the bit. Defaults to the signal name. Clients using revision + 2005 of the protocol might see the name truncated to 16 characters. + + normal: + type: boolean + default: false + description: | + Normal state of the digital status bit. diff --git a/etc/examples/nodes/c37_118.conf b/etc/examples/nodes/c37_118.conf new file mode 100644 index 000000000..591e58662 --- /dev/null +++ b/etc/examples/nodes/c37_118.conf @@ -0,0 +1,27 @@ +nodes = { + c37 = { + type = "c37.118" + + out = { + address = "localhost" + idcode = 0xBEEF + data_rate = 10 + pmus = ({ + name = "VILLASnode" + idcode = 0xBEEF + frequency = "frequency_signal0" + rocof = "rocof_signal0" + phasor = ( + { signal = "phasor_signal0", unit = "volt" }, + { signal = "phasor_signal1", unit = "volt" }, + { signal = "phasor_signal2", unit = "volt" }, + ) + }) + } + + in = { + address = "localhost" + idcode = 0xBEEF + } + } +} diff --git a/include/villas/nodes/c37_118.hpp b/include/villas/nodes/c37_118.hpp new file mode 100644 index 000000000..745c5fcd3 --- /dev/null +++ b/include/villas/nodes/c37_118.hpp @@ -0,0 +1,338 @@ +/* Node type: C37-118. + * + * Author: Philipp Jungkamp + * SPDX-FileCopyrightText: 2024-2025 Institute for Automation of Complex Power Systems, RWTH Aachen University + * SPDX-License-Identifier: Apache-2.0 + */ + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace villas::node { + +enum class C37Version : uint8_t { + R1_2004, + R2_2011, +}; + +/** format type enum for C37.118 protocol values */ +enum class C37Format { + FIXED, + FLOATING_POINT, +}; + +/** unit type enum for C37.118 protocol phasors */ +enum class C37PhasorUnit : uint32_t { + VOLT, + AMPERE, +}; + +/** phasor component represented by a C37.118 phasor */ +enum class C37PhasorComponent : uint32_t { + ZERO_SEQUENCE = 0b000, + POSITIVE_SEQUENCE = 0b001, + NEGATIVE_SEQUENCE = 0b010, + PHASE_A = 0b100, + PHASE_B = 0b101, + PHASE_C = 0b110, +}; + +/** unit type enum for C37.118 protocol phasors */ +enum class C37AnalogUnit : uint32_t { + POINT_ON_WAVE, + RMS, + PEAK, +}; + +/** service class of a C37.118 PMU (measurement or protection). + * The values match the ASCII characters used in the SVC_CLASS field. */ +enum class C37ServiceClass : uint8_t { + M = 'M', + P = 'P', +}; + +enum class C37DataTriggerReason : uint16_t { + MANUAL = 0x0, + LOW_MAGNITUDE, + HIGH_MAGNITUDE, + PHASE_SHIFT, + HIGH_FREQUENCY_DEVIATION, + HIGH_ROCOF, + DIGITAL = 0x7, +}; + +enum class C37DataUnlockedTime : uint16_t { + LOCKED_OR_UNLOCKED_FOR_LESS_THAN_10_SECONDS, + UNLOCKED_FOR_10S_TO_100_SECONDS, + UNLOCKED_FOR_100S_TO_1000_SECONDS, + UNLOCKED_FOR_MORE_THAN_1000_SECONDS, +}; + +enum class C37DataTimeError : uint16_t { + NOT_USED, + ESTIMATED_LESS_THAN_100_NANOSECONDS, + ESTIMATED_LESS_THAN_1_MICROSECOND, + ESTIMATED_LESS_THAN_10_MICROSECONDS, + ESTIMATED_LESS_THAN_100_MICROSECONDS, + ESTIMATED_LESS_THAN_1_MILLISECOND, + ESTIMATED_LESS_THAN_10_MILLISECONDS, + UNKNOWN_OR_ESTIMATED_MORE_THAN_100_MILLISECONDS, +}; + +/** data packet in C37.118 protocol frames */ +struct C37Data { + struct { + C37DataTriggerReason trigger_reason : 4; + C37DataUnlockedTime unlocked_time : 2; + C37DataTimeError time_error : 3; + bool configuration_change : 1; + bool trigger_detected : 1; + bool sorted_by_arrival : 1; + bool out_of_sync : 1; + bool data_error : 1; + }; + + std::vector> phasor; + float freq; + float dfreq; + std::vector analog; + std::vector digital; +}; + +/** metadata for C37.118 protocol phasors */ +struct C37PhasorInfo { + std::string name; + C37PhasorUnit unit; + C37PhasorComponent component; + float amplitude_scale; + float phase_shift; + + // Phasor measurement modifications (CONFIG3 PHSCALE flags), one bit each. + struct { + bool upsampled_with_interpolation : 1; // bit 0 + bool upsampled_with_extrapolation : 1; // bit 1 + bool downsampled_with_reselection : 1; // bit 2 + bool downsampled_with_fir_filter : 1; // bit 3 + bool downsampled_with_non_fir_filter : 1; // bit 4 + bool filtered_without_resampling : 1; // bit 5 + bool magnitude_adjusted_for_calibration : 1; // bit 6 + bool phase_adjusted_for_calibration : 1; // bit 7 + bool phase_adjusted_for_rotation : 1; // bit 8 + bool pseudo_phasor_value : 1; // bit 9 + bool other_modification : 1; // modification-flags MSB + }; +}; + +/** metadata for C37.118 protocol analog values */ +struct C37AnalogInfo { + std::string name; + C37AnalogUnit unit : 8; + float scale; + float offset; +}; + +/** metadata for C37.118 protocol digital values */ +struct C37DigitalInfo { + std::array name; + uint16_t normal; + uint16_t mask; +}; + +/** configuration for one C37.118 PMU */ +struct C37Pmu { + std::string name; + uint16_t idcode; + uuid_t guid; + + struct { + bool phasor_is_polar : 1; + bool phasor_is_float : 1; + bool analog_is_float : 1; + bool freq_is_float : 1; + }; + + std::vector phasor; + std::vector analog; + std::vector digital; + + float latitude; + float longitude; + float elevation; + C37ServiceClass service_class; + uint32_t window; + uint32_t group_delay; + + float nominal_frequency; + + uint16_t cfgcnt; +}; + +/** configuration for a C37.118 PMU or PDC containing one or more PMU configurations */ +struct C37Config { + uint32_t time_base; + std::vector pmu; + uint16_t data_rate; +}; + +enum class C37ConfigType { + R1_CONFIG1, + R1_CONFIG2, + R2_CONFIG3, +}; + +enum class C37CommandType : uint16_t { + R1_DATA_STOP = 1, + R1_DATA_START, + R1_GET_HEADER, + R1_GET_CONFIG1, + R1_GET_CONFIG2, + R2_GET_CONFIG3, +}; + +struct C37Command { + C37CommandType cmd; + std::span ext; +}; + +enum class C37Result : uint16_t { + OK, + SEGMENTED, + INVALID_FRAMESIZE, + INVALID_CHECKSUM, + _requested_size_minimum = 14, +}; + +constexpr std::size_t c37_result_requested_size(C37Result result) { + using underlying_type = std::underlying_type_t; + + constexpr auto minimum = + static_cast(C37Result::_requested_size_minimum); + + if (auto value = static_cast(result); value >= minimum) + return value; + + return 0; +} + +enum class C37Sync : uint16_t { + R1_DATA = 0xAA01, + R1_HEADER = 0xAA11, + R1_CONFIG1 = 0xAA21, + R1_CONFIG2 = 0xAA31, + R1_COMMAND = 0xAA41, + R2_CONFIG3 = 0xAA52, +}; + +enum class C37LeapSecondDirection : uint32_t { + ADD, + DELETE, +}; + +/** time quality indicator for a C37.118 message */ +enum class C37MessageTimeQuality : uint32_t { + LOCKED_TO_UTC, + WITHIN_1_NANOS_OF_UTC, + WITHIN_10_NANOS_OF_UTC, + WITHIN_100_NANOS_OF_UTC, + WITHIN_1_MICROS_OF_UTC, + WITHIN_10_MICROS_OF_UTC, + WITHIN_100_MICROS_OF_UTC, + WITHIN_1_MILLIS_OF_UTC, + WITHIN_10_MILLIS_OF_UTC, + WITHIN_100_MILLIS_OF_UTC, + WITHIN_1_SECS_OF_UTC, + WITHIN_10_SECS_OF_UTC, + TIME_NOT_RELIABLE, +}; + +struct C37FrameMetadata { + uint16_t idcode; + timespec soc; + C37LeapSecondDirection leap_second_direction = {}; + bool leap_second_occurred = false; + bool leap_second_pending = false; + C37MessageTimeQuality message_time_quality = + C37MessageTimeQuality::LOCKED_TO_UTC; +}; + +class C37Frame { + C37Sync sync_; + uint16_t size_; + uint16_t idcode_; + uint32_t soc_; + + struct { + uint32_t fracsec_; + C37LeapSecondDirection leap_second_direction_ : 1; + uint32_t leap_second_occurred_ : 1; + uint32_t leap_second_pending_ : 1; + C37MessageTimeQuality message_time_quality_ : 4; + }; + + std::span message_; + +public: + C37Frame() = default; + uint16_t size() const; + C37Version version() const; + C37Sync sync() const; + C37FrameMetadata metadata(C37Config const &config) const; + + /** load a buffer into this frame */ + C37Result load(std::span buffer); + + /** deserialize a data frame */ + C37Result deserialize_data(C37Config const &config, + std::vector &data) const; + + /** deserialize a header frame */ + C37Result deserialize_header(std::string_view &header) const; + + /** deserialize a configuration frame + * + * A CONFIG3 frame may be fragmented across several frames. The + * segmentation_buffer accumulates the segments; the result is SEGMENTED until + * the final segment completes the configuration. */ + C37Result + deserialize_config(C37Config &config, + std::vector &segmentation_buffer) const; + + /** deserialize a command frame */ + C37Result deserialize_command(C37Command &command) const; + + /** serialize a data frame */ + static C37Result serialize_data(std::span data, + C37FrameMetadata metadata, + C37Config const &config, + std::vector &buffer); + + /** serialize a header frame */ + static C37Result serialize_header(std::string_view header, + C37FrameMetadata metadata, + C37Config const &config, + std::vector &buffer); + + /** serialize a configuration frame */ + static C37Result serialize_config(C37ConfigType type, + C37FrameMetadata metadata, + C37Config const &config, + std::vector &buffer); + + /** serialize a command frame */ + static C37Result serialize_command(C37Command const &command, + C37FrameMetadata metadata, + C37Config const &config, + std::vector &buffer); +}; + +} // namespace villas::node diff --git a/lib/nodes/CMakeLists.txt b/lib/nodes/CMakeLists.txt index 068ad9fba..d3fe92174 100644 --- a/lib/nodes/CMakeLists.txt +++ b/lib/nodes/CMakeLists.txt @@ -6,6 +6,7 @@ set(NODE_SRC loopback_internal.cpp + c37_118.cpp ) if(WITH_WEB) diff --git a/lib/nodes/c37_118.cpp b/lib/nodes/c37_118.cpp new file mode 100644 index 000000000..adfb81410 --- /dev/null +++ b/lib/nodes/c37_118.cpp @@ -0,0 +1,2206 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace villas::node { + +void from_json(Json const &json, C37PhasorUnit &unit) { + auto const &str = json.get_ref(); + if (str == "volt") + unit = C37PhasorUnit::VOLT; + else if (str == "ampere") + unit = C37PhasorUnit::AMPERE; + else + throw RuntimeError("Invalid phasor unit '{}'", str); +} + +void from_json(Json const &json, C37AnalogUnit &unit) { + auto const &str = json.get_ref(); + if (str == "point_on_wave") + unit = C37AnalogUnit::POINT_ON_WAVE; + else if (str == "rms") + unit = C37AnalogUnit::RMS; + else if (str == "peak") + unit = C37AnalogUnit::PEAK; + else + throw RuntimeError("Invalid analog unit '{}'", str); +} + +void from_json(Json const &json, C37PhasorComponent &component) { + auto const &str = json.get_ref(); + if (str == "zero_sequence") + component = C37PhasorComponent::ZERO_SEQUENCE; + else if (str == "positive_sequence") + component = C37PhasorComponent::POSITIVE_SEQUENCE; + else if (str == "negative_sequence") + component = C37PhasorComponent::NEGATIVE_SEQUENCE; + else if (str == "phase_a") + component = C37PhasorComponent::PHASE_A; + else if (str == "phase_b") + component = C37PhasorComponent::PHASE_B; + else if (str == "phase_c") + component = C37PhasorComponent::PHASE_C; + else + throw RuntimeError("Invalid phasor component '{}'", str); +} + +void from_json(Json const &json, C37ServiceClass &service_class) { + auto const &str = json.get_ref(); + if (str == "measurement") + service_class = C37ServiceClass::M; + else if (str == "protection") + service_class = C37ServiceClass::P; + else + throw RuntimeError("Invalid service class '{}'", str); +} + +namespace { + +// Sample CRC routine taken from IEEE Std C37.118.2-2011 Annex B +uint16_t calculate_crc(std::span buffer) { + uint16_t crc = 0xFFFF; + uint16_t temp; + uint16_t quick; + + for (auto byte : buffer) { + temp = (crc >> 8) ^ static_cast(byte); + crc <<= 8; + quick = temp ^ (temp >> 4); + crc ^= quick; + quick <<= 5; + crc ^= quick; + quick <<= 7; + crc ^= quick; + } + + return crc; +} + +template +class bit_field_proxy { + T &value; + +public: + constexpr bit_field_proxy(T &value) : value(value) {} + + constexpr T get() const { return (value >> shift) & mask; } + + constexpr operator T() const { return get(); } + + constexpr void set(T other) { + value &= ~(mask << shift); + value |= (other & mask) << shift; + } + + constexpr bit_field_proxy &operator=(T other) { + set(other); + return *this; + } + + template constexpr To as() const { + return std::bit_cast(get()); + } +}; + +template + requires(std::is_integral_v and std::is_unsigned_v and + high < sizeof(T) * CHAR_BIT and high >= low) +constexpr bit_field_proxy +bit_field(T &value) { + return value; +} + +constexpr C37Result c37_result_request(std::size_t size) { + using underlying_type = std::underlying_type_t; + + constexpr auto minimum = + static_cast(C37Result::_requested_size_minimum); + + if (size < minimum or not std::in_range(size)) { + return C37Result::INVALID_FRAMESIZE; + } + + return static_cast(static_cast(size)); +} + +template +constexpr T from_be_bytes_unchecked(std::byte const *bytes, + std::uint16_t &offset) { + if constexpr (std::is_integral_v && std::is_unsigned_v) { + T temp = 0; + + for (auto const i : std::views::iota(size_t(0), sizeof(T))) + temp = temp << 8 | static_cast(bytes[offset + i]); + + offset += sizeof(T); + + return temp; + } else if constexpr (std::is_integral_v && std::is_signed_v) { + return std::bit_cast( + from_be_bytes_unchecked>(bytes, offset)); + } else if constexpr (std::is_same_v) { + static_assert(sizeof(float) == sizeof(uint32_t)); + return std::bit_cast( + from_be_bytes_unchecked(bytes, offset)); + } else { + return static_cast( + from_be_bytes_unchecked>(bytes, offset)); + } +} + +template + requires(... and (std::is_integral_v or std::is_same_v or + std::is_enum_v)) +C37Result deserialize(std::span buffer, std::uint16_t &offset, + T &...value) { + + if (auto required_size = offset + (... + sizeof(T)); + required_size > buffer.size()) + return c37_result_request(required_size); + + ((void)(value = from_be_bytes_unchecked(buffer.data(), offset)), ...); + + return C37Result::OK; +} + +C37Result deserialize_fixed_string(std::span message, + std::uint16_t &offset, + std::string &fixed_string) { + if (message.size() < size_t(offset + 16)) + return c37_result_request(16); + + auto view = std::string_view( + reinterpret_cast(message.data() + offset), 16); + auto begin = view.find_first_not_of(" "); + auto end = view.find_last_not_of(" "); + + if (begin == std::string_view::npos) + fixed_string.clear(); + else + fixed_string = view.substr(begin, end - begin + 1); + + offset += 16; + + return C37Result::OK; +} + +C37Result deserialize_dynamic_string(std::span message, + std::uint16_t &offset, + std::string &dynamic_string) { + uint8_t length; + if (auto result = deserialize(message, offset, length); + result != C37Result::OK) + return result; + + if (message.size() < size_t(offset + length)) + return c37_result_request(offset + length); + + dynamic_string.assign(reinterpret_cast(message.data() + offset), + length); + offset += length; + + return C37Result::OK; +} + +C37Result deserialize_phasor(std::span message, + C37Pmu const &pmu, C37PhasorInfo const &info, + std::uint16_t &offset, + std::complex &phasor) { + if (pmu.phasor_is_float) { + float first, second; + if (auto result = deserialize(message, offset, first, second); + result != C37Result::OK) + return result; + + if (pmu.phasor_is_polar) { + phasor = std::polar(first, second); + } else { + phasor = std::complex(first, second); + } + } else { + int16_t first, second; + if (auto result = deserialize(message, offset, first, second); + result != C37Result::OK) + return result; + + if (pmu.phasor_is_polar) { + auto abs = static_cast(std::bit_cast(first)) * + info.amplitude_scale; + auto arg = (static_cast(second) / 10'000) - info.phase_shift; + phasor = std::polar(abs, arg); + } else { + auto real = first * info.amplitude_scale; + auto imag = second * info.amplitude_scale; + phasor = std::complex(real, imag); + + if (info.phase_shift != 0) + phasor *= std::polar(1, -info.phase_shift); + } + } + + return C37Result::OK; +} + +C37Result deserialize_freq(std::span message, + C37Pmu const &pmu, std::uint16_t &offset, + float &freq) { + if (pmu.freq_is_float) { + if (auto result = deserialize(message, offset, freq); + result != C37Result::OK) + return result; + } else { + int16_t value; + if (auto result = deserialize(message, offset, value); + result != C37Result::OK) + return result; + + freq = static_cast(value) / 1'000 + pmu.nominal_frequency; + } + + return C37Result::OK; +} + +C37Result deserialize_dfreq(std::span message, + C37Pmu const &pmu, std::uint16_t &offset, + float &dfreq) { + if (pmu.freq_is_float) { + if (auto result = deserialize(message, offset, dfreq); + result != C37Result::OK) + return result; + } else { + int16_t value; + if (auto result = deserialize(message, offset, value); + result != C37Result::OK) + return result; + + dfreq = static_cast(value) / 100; + } + + return C37Result::OK; +} + +C37Result deserialize_analog(std::span message, + C37Pmu const &pmu, C37AnalogInfo const &info, + std::uint16_t &offset, float &analog) { + if (pmu.analog_is_float) { + if (auto result = deserialize(message, offset, analog); + result != C37Result::OK) + return result; + } else { + int16_t value; + if (auto result = deserialize(message, offset, value); + result != C37Result::OK) + return result; + + analog = static_cast(value) * info.scale; + } + + return C37Result::OK; +} + +C37Result deserialize_digital(std::span message, + C37Pmu const &pmu, C37DigitalInfo const &info, + std::uint16_t &offset, uint16_t &digital) { + if (auto result = deserialize(message, offset, digital); + result != C37Result::OK) + return result; + + digital &= info.mask; + + return C37Result::OK; +} + +C37Result deserialize_pmu_data(std::span message, + C37Pmu const &pmu, std::uint16_t &offset, + C37Data &data) { + uint16_t stat; + if (auto result = deserialize(message, offset, stat); result != C37Result::OK) + return result; + + data.data_error = bit_field<13>(stat); + data.out_of_sync = bit_field<12>(stat); + data.sorted_by_arrival = bit_field<11>(stat); + data.trigger_detected = bit_field<10>(stat); + data.configuration_change = bit_field<9>(stat); + data.time_error = bit_field<8, 6>(stat).as(); + data.unlocked_time = bit_field<5, 4>(stat).as(); + data.trigger_reason = bit_field<3, 0>(stat).as(); + + data.phasor.resize(pmu.phasor.size()); + data.analog.resize(pmu.analog.size()); + data.digital.resize(pmu.digital.size()); + + for (auto const i : std::views::iota(size_t(0), pmu.phasor.size())) { + if (auto result = deserialize_phasor(message, pmu, pmu.phasor[i], offset, + data.phasor[i]); + result != C37Result::OK) + return result; + } + + if (auto result = deserialize_freq(message, pmu, offset, data.freq); + result != C37Result::OK) + return result; + + if (auto result = deserialize_dfreq(message, pmu, offset, data.dfreq); + result != C37Result::OK) + return result; + + for (auto const i : std::views::iota(size_t(0), pmu.analog.size())) { + if (auto result = deserialize_analog(message, pmu, pmu.analog[i], offset, + data.analog[i]); + result != C37Result::OK) + return result; + } + + for (auto const i : std::views::iota(size_t(0), pmu.digital.size())) { + if (auto result = deserialize_digital(message, pmu, pmu.digital[i], offset, + data.digital[i]); + result != C37Result::OK) + return result; + } + + return C37Result::OK; +} + +C37Result deserialize_r1_pmu_config(std::span message, + std::uint16_t &offset, C37Pmu &pmu) { + uint16_t phasor_count, analog_count, digital_count; + uint16_t format; + + uuid_clear(pmu.guid); + pmu.latitude = std::numeric_limits::quiet_NaN(); + pmu.longitude = std::numeric_limits::quiet_NaN(); + pmu.elevation = std::numeric_limits::quiet_NaN(); + pmu.service_class = C37ServiceClass::M; + pmu.window = 0; + pmu.group_delay = 0; + + if (auto result = deserialize_fixed_string(message, offset, pmu.name); + result != C37Result::OK) + return result; + + if (auto result = deserialize(message, offset, pmu.idcode, format, + phasor_count, analog_count, digital_count); + result != C37Result::OK) + return result; + + pmu.freq_is_float = bit_field<3>(format); + pmu.analog_is_float = bit_field<2>(format); + pmu.phasor_is_float = bit_field<1>(format); + pmu.phasor_is_polar = bit_field<0>(format); + pmu.phasor.resize(phasor_count); + pmu.analog.resize(analog_count); + pmu.digital.resize(digital_count); + + for (auto &phasor : pmu.phasor) { + if (auto result = deserialize_fixed_string(message, offset, phasor.name); + result != C37Result::OK) + return result; + } + + for (auto &analog : pmu.analog) { + if (auto result = deserialize_fixed_string(message, offset, analog.name); + result != C37Result::OK) + return result; + } + + for (auto &digital : pmu.digital) { + for (auto &name : digital.name) { + if (auto result = deserialize_fixed_string(message, offset, name); + result != C37Result::OK) + return result; + } + } + + for (auto &phasor : pmu.phasor) { + uint32_t phunit; + if (auto result = deserialize(message, offset, phunit); + result != C37Result::OK) + return result; + + phasor.unit = bit_field<31, 24>(phunit).as(); + phasor.component = C37PhasorComponent::POSITIVE_SEQUENCE; + phasor.amplitude_scale = bit_field<23, 0>(phunit); + phasor.phase_shift = 0; + phasor.upsampled_with_interpolation = false; + phasor.upsampled_with_extrapolation = false; + phasor.downsampled_with_reselection = false; + phasor.downsampled_with_fir_filter = false; + phasor.downsampled_with_non_fir_filter = false; + phasor.filtered_without_resampling = false; + phasor.magnitude_adjusted_for_calibration = false; + phasor.phase_adjusted_for_calibration = false; + phasor.phase_adjusted_for_rotation = false; + phasor.pseudo_phasor_value = false; + phasor.other_modification = false; + } + + for (auto &analog : pmu.analog) { + uint32_t anunit; + if (auto result = deserialize(message, offset, anunit); + result != C37Result::OK) + return result; + + analog.unit = bit_field<31, 24>(anunit).as(); + analog.scale = bit_field<23, 0>(anunit); + analog.offset = 0; + } + + for (auto &digital : pmu.digital) { + uint32_t dgunit; + if (auto result = deserialize(message, offset, dgunit); + result != C37Result::OK) + return result; + + digital.normal = bit_field<31, 16>(dgunit); + digital.mask = bit_field<15, 0>(dgunit); + } + + uint16_t nominal_frequency_flags; + if (auto result = + deserialize(message, offset, nominal_frequency_flags, pmu.cfgcnt); + result != C37Result::OK) + return result; + + pmu.nominal_frequency = bit_field<0>(nominal_frequency_flags) ? 50.0f : 60.0f; + + return C37Result::OK; +} + +C37Result deserialize_uuid(std::span message, + std::uint16_t &offset, uuid_t &uuid) { + if (message.size() < size_t(offset + sizeof(uuid_t))) + return c37_result_request(offset + sizeof(uuid_t)); + + std::memcpy(uuid, message.data() + offset, sizeof(uuid_t)); + offset += sizeof(uuid_t); + + return C37Result::OK; +} + +C37Result deserialize_r2_phasor_scale(std::span message, + std::uint16_t &offset, + C37PhasorInfo &phasor) { + uint32_t flags; + if (auto result = deserialize(message, offset, flags, phasor.amplitude_scale, + phasor.phase_shift); + result != C37Result::OK) + return result; + + phasor.other_modification = bit_field<31>(flags); + phasor.pseudo_phasor_value = bit_field<26>(flags); + phasor.phase_adjusted_for_rotation = bit_field<25>(flags); + phasor.phase_adjusted_for_calibration = bit_field<24>(flags); + phasor.magnitude_adjusted_for_calibration = bit_field<23>(flags); + phasor.filtered_without_resampling = bit_field<22>(flags); + phasor.downsampled_with_non_fir_filter = bit_field<21>(flags); + phasor.downsampled_with_fir_filter = bit_field<20>(flags); + phasor.downsampled_with_reselection = bit_field<19>(flags); + phasor.upsampled_with_extrapolation = bit_field<18>(flags); + phasor.upsampled_with_interpolation = bit_field<17>(flags); + phasor.unit = bit_field<11>(flags).as(); + phasor.component = bit_field<10, 8>(flags).as(); + + return C37Result::OK; +} + +C37Result deserialize_r2_analog_scale(std::span message, + std::uint16_t &offset, + C37AnalogInfo &analog) { + if (auto result = deserialize(message, offset, analog.scale, analog.offset); + result != C37Result::OK) + return result; + + analog.unit = C37AnalogUnit::POINT_ON_WAVE; + + return C37Result::OK; +} + +C37Result deserialize_r2_pmu_config(std::span message, + std::uint16_t &offset, C37Pmu &pmu) { + uint16_t format; + + if (auto result = deserialize_dynamic_string(message, offset, pmu.name); + result != C37Result::OK) + return result; + + if (auto result = deserialize(message, offset, pmu.idcode); + result != C37Result::OK) + return result; + + if (auto result = deserialize_uuid(message, offset, pmu.guid); + result != C37Result::OK) + return result; + + uint16_t phasor_count; + uint16_t analog_count; + uint16_t digital_count; + if (auto result = deserialize(message, offset, format, phasor_count, + analog_count, digital_count); + result != C37Result::OK) + return result; + + pmu.freq_is_float = bit_field<3>(format); + pmu.analog_is_float = bit_field<2>(format); + pmu.phasor_is_float = bit_field<1>(format); + pmu.phasor_is_polar = bit_field<0>(format); + pmu.phasor.resize(phasor_count); + pmu.analog.resize(analog_count); + pmu.digital.resize(digital_count); + + for (auto &phasor : pmu.phasor) { + if (auto result = deserialize_dynamic_string(message, offset, phasor.name); + result != C37Result::OK) + return result; + } + + for (auto &analog : pmu.analog) { + if (auto result = deserialize_dynamic_string(message, offset, analog.name); + result != C37Result::OK) + return result; + } + + for (auto &digital : pmu.digital) { + for (auto &name : digital.name) { + if (auto result = deserialize_dynamic_string(message, offset, name); + result != C37Result::OK) + return result; + } + } + + for (auto &phasor : pmu.phasor) { + if (auto result = deserialize_r2_phasor_scale(message, offset, phasor); + result != C37Result::OK) + return result; + } + + for (auto &analog : pmu.analog) { + if (auto result = deserialize_r2_analog_scale(message, offset, analog); + result != C37Result::OK) + return result; + } + + for (auto &digital : pmu.digital) { + uint32_t dgunit; + if (auto result = deserialize(message, offset, dgunit); + result != C37Result::OK) + return result; + + digital.normal = bit_field<31, 16>(dgunit); + digital.mask = bit_field<15, 0>(dgunit); + } + + if (auto result = deserialize(message, offset, pmu.latitude, pmu.longitude, + pmu.elevation, pmu.service_class, pmu.window, + pmu.group_delay); + result != C37Result::OK) + return result; + + uint16_t nominal_frequency_flags; + if (auto result = + deserialize(message, offset, nominal_frequency_flags, pmu.cfgcnt); + result != C37Result::OK) + return result; + + pmu.nominal_frequency = bit_field<0>(nominal_frequency_flags) ? 50.0f : 60.0f; + + return C37Result::OK; +} + +C37Result deserialize_r1_config(std::span message, + C37Config &config) { + uint16_t offset = 0; + uint16_t pmu_count; + + if (auto result = deserialize(message, offset, config.time_base, pmu_count); + result != C37Result::OK) + return c37_result_requested_size(result) ? C37Result::INVALID_FRAMESIZE + : result; + + config.pmu.resize(pmu_count); + for (auto &pmu : config.pmu) { + if (auto result = deserialize_r1_pmu_config(message, offset, pmu); + result != C37Result::OK) + return c37_result_requested_size(result) ? C37Result::INVALID_FRAMESIZE + : result; + } + + if (auto result = deserialize(message, offset, config.data_rate); + result != C37Result::OK) + return c37_result_requested_size(result) ? C37Result::INVALID_FRAMESIZE + : result; + + return C37Result::OK; +} + +C37Result deserialize_r2_config(std::span message, + C37Config &config) { + uint16_t offset = 0; + uint16_t pmu_count; + + if (auto result = deserialize(message, offset, config.time_base, pmu_count); + result != C37Result::OK) + return result; + + config.pmu.resize(pmu_count); + for (auto &pmu : config.pmu) { + if (auto result = deserialize_r2_pmu_config(message, offset, pmu); + result != C37Result::OK) + return result; + } + + if (auto result = deserialize(message, offset, config.data_rate); + result != C37Result::OK) + return result; + + return C37Result::OK; +} + +template void to_be_bytes_unchecked(std::byte *buffer, T value) { + if constexpr (std::is_integral_v) { + using U = std::make_unsigned_t; + auto u = static_cast(value); + for (auto i : std::views::iota(size_t(0), sizeof(T)) | std::views::reverse) + *buffer++ = std::byte((u >> (i * 8)) & 0xFF); + } else if constexpr (std::is_same_v) { + static_assert(sizeof(float) == sizeof(uint32_t)); + to_be_bytes_unchecked(buffer, std::bit_cast(value)); + } else { + to_be_bytes_unchecked(buffer, + static_cast>(value)); + } +} + +template + requires(... and (std::is_integral_v or std::is_same_v or + std::is_enum_v)) +void serialize(std::vector &buffer, T... value) { + auto offset = buffer.size(); + buffer.resize(offset + (... + sizeof(T))); + ((to_be_bytes_unchecked(buffer.data() + offset, value), offset += sizeof(T)), + ...); +} + +void serialize_fixed_string(std::vector &buffer, + std::string_view fixed_string) { + auto bytes = reinterpret_cast(fixed_string.data()); + auto count = std::min({fixed_string.size(), size_t(16)}); + auto iter = buffer.insert(buffer.end(), 16, std::byte(' ')); + std::copy_n(bytes, count, iter); +} + +void serialize_dynamic_string(std::vector &buffer, + std::string_view dynamic_string) { + auto bytes = reinterpret_cast(dynamic_string.data()); + auto count = std::min({dynamic_string.size(), size_t(255)}); + serialize(buffer, uint8_t(count)); + buffer.insert(buffer.end(), bytes, bytes + count); +} + +void serialize_phasor(std::vector &buffer, C37Pmu const &pmu, + C37PhasorInfo const &info, std::complex phasor) { + if (pmu.phasor_is_float) { + if (pmu.phasor_is_polar) { + serialize(buffer, std::abs(phasor), std::arg(phasor)); + } else { + serialize(buffer, phasor.real(), phasor.imag()); + } + } else { + if (pmu.phasor_is_polar) { + auto mag = std::bit_cast( + uint16_t(std::abs(phasor) / info.amplitude_scale)); + auto ang = int16_t((std::arg(phasor) + info.phase_shift) * 10'000); + serialize(buffer, mag, ang); + } else { + if (info.phase_shift != 0) + phasor *= std::polar(1, info.phase_shift); + + auto real = int16_t(phasor.real() / info.amplitude_scale); + auto imag = int16_t(phasor.imag() / info.amplitude_scale); + serialize(buffer, real, imag); + } + } +} + +void serialize_freq(std::vector &buffer, C37Pmu const &pmu, + float freq) { + if (pmu.freq_is_float) { + serialize(buffer, freq); + } else { + serialize(buffer, uint16_t((freq - pmu.nominal_frequency) * 1'000)); + } +} + +void serialize_dfreq(std::vector &buffer, C37Pmu const &pmu, + float dfreq) { + if (pmu.freq_is_float) { + serialize(buffer, dfreq); + } else { + serialize(buffer, uint16_t(dfreq * 100)); + } +} + +void serialize_analog(std::vector &buffer, C37Pmu const &pmu, + C37AnalogInfo const &info, float analog) { + if (pmu.analog_is_float) { + serialize(buffer, analog); + } else { + serialize(buffer, uint16_t(info.scale != 0 ? analog / info.scale : 0.0f)); + } +} + +void serialize_digital(std::vector &buffer, uint16_t digital) { + serialize(buffer, digital); +} + +void serialize_pmu_data(std::vector &buffer, C37Pmu const &pmu, + C37Data const &data) { + uint16_t stat = 0; + bit_field<13>(stat) = uint16_t(data.data_error); + bit_field<12>(stat) = uint16_t(data.out_of_sync); + bit_field<11>(stat) = uint16_t(data.sorted_by_arrival); + bit_field<10>(stat) = uint16_t(data.trigger_detected); + bit_field<9>(stat) = uint16_t(data.configuration_change); + bit_field<8, 6>(stat) = uint16_t(data.time_error); + bit_field<5, 4>(stat) = uint16_t(data.unlocked_time); + bit_field<3, 0>(stat) = uint16_t(data.trigger_reason); + serialize(buffer, stat); + + for (auto const i : std::views::iota(size_t(0), pmu.phasor.size())) + serialize_phasor(buffer, pmu, pmu.phasor[i], data.phasor[i]); + + serialize_freq(buffer, pmu, data.freq); + serialize_dfreq(buffer, pmu, data.dfreq); + + for (auto const i : std::views::iota(size_t(0), pmu.analog.size())) + serialize_analog(buffer, pmu, pmu.analog[i], data.analog[i]); + + for (auto const i : std::views::iota(size_t(0), pmu.digital.size())) + serialize_digital(buffer, data.digital[i]); +} + +void serialize_r1_pmu_config(std::vector &buffer, + C37Pmu const &pmu) { + serialize_fixed_string(buffer, pmu.name); + + uint16_t format = 0; + bit_field<3>(format) = uint16_t(pmu.freq_is_float); + bit_field<2>(format) = uint16_t(pmu.analog_is_float); + bit_field<1>(format) = uint16_t(pmu.phasor_is_float); + bit_field<0>(format) = uint16_t(pmu.phasor_is_polar); + + serialize(buffer, pmu.idcode, format, uint16_t(pmu.phasor.size()), + uint16_t(pmu.analog.size()), uint16_t(pmu.digital.size())); + + for (auto const &phasor : pmu.phasor) + serialize_fixed_string(buffer, phasor.name); + for (auto const &analog : pmu.analog) + serialize_fixed_string(buffer, analog.name); + for (auto const &digital : pmu.digital) + for (auto const &name : digital.name) + serialize_fixed_string(buffer, name); + + for (auto const &phasor : pmu.phasor) { + uint32_t phunit = 0; + bit_field<31, 24>(phunit) = uint32_t(phasor.unit); + bit_field<23, 0>(phunit) = uint32_t(phasor.amplitude_scale); + serialize(buffer, phunit); + } + + for (auto const &analog : pmu.analog) { + uint32_t anunit = 0; + bit_field<31, 24>(anunit) = uint32_t(analog.unit); + bit_field<23, 0>(anunit) = uint32_t(analog.scale); + serialize(buffer, anunit); + } + + for (auto const &digital : pmu.digital) { + uint32_t dgunit = 0; + bit_field<31, 16>(dgunit) = uint32_t(digital.normal); + bit_field<15, 0>(dgunit) = uint32_t(digital.mask); + serialize(buffer, dgunit); + } + + uint16_t fnom = pmu.nominal_frequency == 50.0f ? 1 : 0; + serialize(buffer, fnom, pmu.cfgcnt); +} + +void serialize_uuid(std::vector &buffer, uuid_t const &uuid) { + auto bytes = reinterpret_cast(uuid); + buffer.insert(buffer.end(), bytes, bytes + sizeof(uuid_t)); +} + +void serialize_r2_phasor_scale(std::vector &buffer, + C37PhasorInfo const &phasor) { + uint32_t flags = 0; + bit_field<31>(flags) = uint32_t(phasor.other_modification); + bit_field<26>(flags) = uint32_t(phasor.pseudo_phasor_value); + bit_field<25>(flags) = uint32_t(phasor.phase_adjusted_for_rotation); + bit_field<24>(flags) = uint32_t(phasor.phase_adjusted_for_calibration); + bit_field<23>(flags) = uint32_t(phasor.magnitude_adjusted_for_calibration); + bit_field<22>(flags) = uint32_t(phasor.filtered_without_resampling); + bit_field<21>(flags) = uint32_t(phasor.downsampled_with_non_fir_filter); + bit_field<20>(flags) = uint32_t(phasor.downsampled_with_fir_filter); + bit_field<19>(flags) = uint32_t(phasor.downsampled_with_reselection); + bit_field<18>(flags) = uint32_t(phasor.upsampled_with_extrapolation); + bit_field<17>(flags) = uint32_t(phasor.upsampled_with_interpolation); + bit_field<11>(flags) = uint32_t(phasor.unit); + bit_field<10, 8>(flags) = uint32_t(phasor.component); + + serialize(buffer, flags, phasor.amplitude_scale, phasor.phase_shift); +} + +void serialize_r2_analog_scale(std::vector &buffer, + C37AnalogInfo const &analog) { + serialize(buffer, analog.scale, analog.offset); +} + +void serialize_r2_pmu_metadata(std::vector &buffer, + C37Pmu const &pmu) { + serialize(buffer, pmu.latitude, pmu.longitude, pmu.elevation, + pmu.service_class, pmu.window, pmu.group_delay); +} + +void serialize_r2_pmu_config(std::vector &buffer, + C37Pmu const &pmu) { + serialize_dynamic_string(buffer, pmu.name); + serialize(buffer, pmu.idcode); + serialize_uuid(buffer, pmu.guid); + + uint16_t format = 0; + bit_field<3>(format) = uint16_t(pmu.freq_is_float); + bit_field<2>(format) = uint16_t(pmu.analog_is_float); + bit_field<1>(format) = uint16_t(pmu.phasor_is_float); + bit_field<0>(format) = uint16_t(pmu.phasor_is_polar); + + serialize(buffer, format, uint16_t(pmu.phasor.size()), + uint16_t(pmu.analog.size()), uint16_t(pmu.digital.size())); + + for (auto const &phasor : pmu.phasor) + serialize_dynamic_string(buffer, phasor.name); + for (auto const &analog : pmu.analog) + serialize_dynamic_string(buffer, analog.name); + for (auto const &digital : pmu.digital) + for (auto const &name : digital.name) + serialize_dynamic_string(buffer, name); + + for (auto const &phasor : pmu.phasor) + serialize_r2_phasor_scale(buffer, phasor); + for (auto const &analog : pmu.analog) + serialize_r2_analog_scale(buffer, analog); + + for (auto const &digital : pmu.digital) { + uint32_t dgunit = 0; + bit_field<31, 16>(dgunit) = uint32_t(digital.normal); + bit_field<15, 0>(dgunit) = uint32_t(digital.mask); + serialize(buffer, dgunit); + } + + serialize_r2_pmu_metadata(buffer, pmu); + + uint16_t fnom = pmu.nominal_frequency == 50.0f ? 1 : 0; + serialize(buffer, fnom, pmu.cfgcnt); +} + +std::size_t serialize_frame_header(std::vector &buffer, C37Sync sync, + C37FrameMetadata const &metadata, + C37Config const &config) { + auto frame_start = buffer.size(); + + uint32_t fracsec = + config.time_base != 0 + ? uint32_t((uint64_t(metadata.soc.tv_nsec) * config.time_base + + 500'000'000) / + 1'000'000'000) + : 0; + + uint32_t fracsec_bits = 0; + bit_field<30>(fracsec_bits) = + static_cast(metadata.leap_second_direction); + bit_field<29>(fracsec_bits) = + static_cast(metadata.leap_second_occurred); + bit_field<28>(fracsec_bits) = + static_cast(metadata.leap_second_pending); + bit_field<27, 24>(fracsec_bits) = + static_cast(metadata.message_time_quality); + bit_field<23, 0>(fracsec_bits) = fracsec; + + serialize(buffer, sync, + /* placeholder framesize */ uint16_t(0), metadata.idcode, + uint32_t(metadata.soc.tv_sec), fracsec_bits); + + return frame_start; +} + +void finalize_frame(std::vector &buffer, std::size_t frame_start) { + auto total_size = uint16_t(buffer.size() - frame_start + sizeof(uint16_t)); + + to_be_bytes_unchecked(buffer.data() + frame_start + 2, total_size); + + auto frame_data = std::span(buffer).subspan(frame_start); + serialize(buffer, calculate_crc(frame_data)); +} + +C37Result serialize_r1_config(C37Sync sync, C37FrameMetadata const &metadata, + C37Config const &config, + std::vector &buffer) { + buffer.clear(); + + auto frame_start = serialize_frame_header(buffer, sync, metadata, config); + serialize(buffer, config.time_base, uint16_t(config.pmu.size())); + + for (auto const &pmu : config.pmu) + serialize_r1_pmu_config(buffer, pmu); + + serialize(buffer, config.data_rate); + finalize_frame(buffer, frame_start); + + return C37Result::OK; +} + +void serialize_r2_config(C37Config const &config, + std::vector &buffer) { + serialize(buffer, config.time_base, uint16_t(config.pmu.size())); + + for (auto const &pmu : config.pmu) + serialize_r2_pmu_config(buffer, pmu); + + serialize(buffer, config.data_rate); +} + +} // namespace + +uint16_t C37Frame::size() const { return size_; } + +C37Sync C37Frame::sync() const { return sync_; } + +C37FrameMetadata C37Frame::metadata(C37Config const &config) const { + auto soc_nanos = config.time_base != 0 + ? uint64_t(fracsec_) * 1'000'000'000 / config.time_base + : 0; + + return {.idcode = idcode_, + .soc = {.tv_sec = soc_, .tv_nsec = uint32_t(soc_nanos)}, + .leap_second_direction = leap_second_direction_, + .leap_second_occurred = bool(leap_second_occurred_), + .leap_second_pending = bool(leap_second_pending_), + .message_time_quality = message_time_quality_}; +} + +C37Result C37Frame::load(std::span buffer) { + std::uint16_t offset = 0; + + uint32_t fracsec_bits; + uint16_t crc; + + if (auto result = deserialize(buffer, offset, sync_, size_, idcode_, soc_, + fracsec_bits); + result != C37Result::OK) + return result; + + // framesize cannot be smaller than the mandatory header + crc + if (size_ < offset + sizeof(crc)) + return C37Result::INVALID_FRAMESIZE; + + leap_second_direction_ = + bit_field<30>(fracsec_bits).as(); + leap_second_occurred_ = bit_field<29>(fracsec_bits); + leap_second_pending_ = bit_field<28>(fracsec_bits); + message_time_quality_ = + bit_field<27, 24>(fracsec_bits).as(); + fracsec_ = bit_field<23, 0>(fracsec_bits); + + auto message_offset = offset; + auto message_size = size_ - offset - sizeof(crc); + + offset += message_size; + if (auto result = deserialize(buffer, offset, crc); result != C37Result::OK) + return result; + + if (crc != calculate_crc(buffer.subspan(0, size_ - sizeof(crc)))) + return C37Result::INVALID_CHECKSUM; + + message_ = buffer.subspan(message_offset, message_size); + + return C37Result::OK; +} + +C37Result C37Frame::deserialize_data(C37Config const &config, + std::vector &data) const { + assert(sync_ == C37Sync::R1_DATA); + + std::uint16_t offset = 0; + + data.resize(config.pmu.size()); + for (auto const i : std::views::iota(size_t(0), config.pmu.size())) { + if (auto result = + deserialize_pmu_data(message_, config.pmu[i], offset, data[i]); + result != C37Result::OK) { + return c37_result_requested_size(result) ? C37Result::INVALID_FRAMESIZE + : result; + } + } + + return C37Result::OK; +} + +C37Result C37Frame::deserialize_header(std::string_view &header) const { + assert(sync_ == C37Sync::R1_HEADER); + + auto data = reinterpret_cast(message_.data()); + header = std::string_view(data, message_.size()); + + return C37Result::OK; +} + +C37Result C37Frame::deserialize_config( + C37Config &config, std::vector &segmentation_buffer) const { + switch (sync_) { + case C37Sync::R1_CONFIG1: + case C37Sync::R1_CONFIG2: { + auto result = deserialize_r1_config(message_, config); + return c37_result_requested_size(result) ? C37Result::INVALID_FRAMESIZE + : result; + } + + case C37Sync::R2_CONFIG3: { + uint16_t offset = 0; + uint16_t segment; + + if (auto result = deserialize(message_, offset, segment); + result != C37Result::OK) + return c37_result_requested_size(result) ? C37Result::INVALID_FRAMESIZE + : result; + + auto message = message_.subspan(offset); + if (segment == 0) { + auto result = deserialize_r2_config(message, config); + return c37_result_requested_size(result) ? C37Result::INVALID_FRAMESIZE + : result; + } + + if (segment == 1) + segmentation_buffer.clear(); + + segmentation_buffer.insert(segmentation_buffer.end(), message.begin(), + message.end()); + + if (segment != 0xFFFF) + return C37Result::SEGMENTED; + + auto result = deserialize_r2_config(segmentation_buffer, config); + return c37_result_requested_size(result) ? C37Result::INVALID_FRAMESIZE + : result; + } + + default: + throw std::runtime_error("unreachable"); + } +} + +C37Result C37Frame::deserialize_command(C37Command &command) const { + assert(type_ == C37Sync::R1_COMMAND); + + uint16_t offset = 0; + + if (auto result = deserialize(message_, offset, command.cmd); + result != C37Result::OK) + return c37_result_requested_size(result) ? C37Result::INVALID_FRAMESIZE + : result; + + command.ext = message_.subspan(offset); + + return C37Result::OK; +} + +C37Result C37Frame::serialize_data(std::span data, + C37FrameMetadata metadata, + C37Config const &config, + std::vector &buffer) { + buffer.clear(); + auto frame_start = + serialize_frame_header(buffer, C37Sync::R1_DATA, metadata, config); + + for (auto const i : std::views::iota(size_t(0), config.pmu.size())) + serialize_pmu_data(buffer, config.pmu[i], data[i]); + + finalize_frame(buffer, frame_start); + return C37Result::OK; +} + +C37Result C37Frame::serialize_header(std::string_view header, + C37FrameMetadata metadata, + C37Config const &config, + std::vector &buffer) { + buffer.clear(); + auto frame_start = + serialize_frame_header(buffer, C37Sync::R1_HEADER, metadata, config); + + auto const *data = reinterpret_cast(header.data()); + buffer.insert(buffer.end(), data, data + header.size()); + + finalize_frame(buffer, frame_start); + return C37Result::OK; +} + +C37Result C37Frame::serialize_config(C37ConfigType type, + C37FrameMetadata metadata, + C37Config const &config, + std::vector &buffer) { + buffer.clear(); + + switch (type) { + case C37ConfigType::R1_CONFIG1: + return serialize_r1_config(C37Sync::R1_CONFIG1, metadata, config, buffer); + case C37ConfigType::R1_CONFIG2: + return serialize_r1_config(C37Sync::R1_CONFIG2, metadata, config, buffer); + + case C37ConfigType::R2_CONFIG3: { + std::vector content; + serialize_r2_config(config, content); + + auto content_iter = content.begin(); + for (uint16_t segment = 1; content_iter != content.end(); segment += 1) { + auto frame_offset = buffer.size(); + serialize_frame_header(buffer, C37Sync::R2_CONFIG3, metadata, config); + + auto segment_size = + std::min({size_t(content.end() - content_iter), size_t(0xFFED)}); + auto segment_end = content_iter + segment_size; + + if (segment_end == content.end()) { + segment = (segment == 1) ? 0 : 0xFFFF; + } + + serialize(buffer, segment); + buffer.insert(buffer.end(), content_iter, segment_end); + finalize_frame(buffer, frame_offset); + + content_iter = segment_end; + } + + return C37Result::OK; + } + + default: + throw std::runtime_error("unreachable"); + } +} + +C37Result C37Frame::serialize_command(C37Command const &command, + C37FrameMetadata metadata, + C37Config const &config, + std::vector &buffer) { + buffer.clear(); + auto frame_start = + serialize_frame_header(buffer, C37Sync::R1_COMMAND, metadata, config); + + serialize(buffer, command.cmd); + buffer.insert(buffer.end(), command.ext.begin(), command.ext.end()); + + finalize_frame(buffer, frame_start); + return C37Result::OK; +} + +namespace { + +// A frame buffer sized to the maximum possible frame. The C37.118 frame size +// is a 16-bit field, so a single frame never exceeds 64 KiB. +using C37Buffer = std::array; + +constexpr std::tuple +parse_address(std::string const &address, std::string default_service) { + auto colon = address.rfind(':'); + if (colon == std::string::npos) + return {address, std::move(default_service)}; + + auto host = address.substr(0, colon); + auto service = address.substr(colon + 1); + return {std::move(host), std::move(service)}; +} + +class C37Node : public Node { + struct { + std::string host; + std::string service; + uint16_t idcode = 1; + int sd = -1; + C37Config config = {}; + + std::vector tx = {}; + C37Buffer rx = {}; + } client; + + struct { + std::string host; + std::string service; + uint16_t idcode = 1; + C37Config config = {}; + + int listen_sd = -1; + int client_sd = -1; + bool streaming = false; + + CQueueSignalled queue = {}; + std::jthread thread; + + std::vector names = {}; + std::vector indices = {}; + std::vector data = {}; + std::vector tx = {}; + C37Buffer rx = {}; + } server; + + std::string details = {}; + + static void sendAll(int sd, std::span data); + static bool recvAll(int sd, std::span buf); + static C37Result recvFrame(int sd, C37Buffer &buffer, C37Frame &frame); + + C37PhasorInfo parsePhasor(Json const &json, + std::vector &names) const; + C37AnalogInfo parseAnalog(Json const &json, + std::vector &names) const; + void parseDigital(Json const &json, std::vector &digital, + std::vector &names) const; + C37Pmu parsePmu(Json const &json, std::vector &names) const; + C37Config parseConfig(Json const &json, + std::vector &names) const; + + void clientConnect(); + void clientDisconnect(); + void clientSendCommand(C37CommandType cmd); + void clientBuildSignals(); + + void serverListen(); + void serverShutdown(); + void serverLoop(std::stop_token stop); + void serverAcceptClient(); + void serverCloseClient(); + void serverHandleCommand(); + void serverDrainQueue(); + void serverBuildIndices(SignalList &signals); + void serverSendSample(struct Sample const *smp); + +protected: + int _read(struct Sample *smps[], unsigned cnt) override; + int _write(struct Sample *smps[], unsigned cnt) override; + +public: + using Node::Node; + + ~C37Node() override { + clientDisconnect(); + serverShutdown(); + } + + int parse(json_t *json) override; + int prepare() override; + int start() override; + int stop() override; + int reverse() override; + const std::string &getDetails() override; + + std::vector getPollFDs() override { + return client.sd >= 0 ? std::vector{client.sd} : std::vector{}; + } +}; + +void C37Node::clientConnect() { + addrinfo hints = {}; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + + auto addr = std::invoke([&]() { + addrinfo *result = nullptr; + + if (auto ret = getaddrinfo(client.host.c_str(), client.service.c_str(), + &hints, &result); + ret != 0) + throw RuntimeError("Failed to resolve '{}:{}': {}", client.host, + client.service, gai_strerror(ret)); + + return std::unique_ptr(result); + }); + + int fd = -1; + for (auto *rp = addr.get(); rp != nullptr; rp = rp->ai_next) { + fd = ::socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol); + if (fd < 0) + continue; + + if (::connect(fd, rp->ai_addr, rp->ai_addrlen) == 0) + break; + + ::close(fd); + fd = -1; + } + + if (fd < 0) + throw SystemError("Failed to connect to '{}:{}'", client.host, + client.service); + + client.sd = fd; +} + +void C37Node::clientDisconnect() { + if (client.sd < 0) + return; + + ::close(client.sd); + client.sd = -1; +} + +void C37Node::sendAll(int sd, std::span data) { + std::size_t offset = 0; + while (offset < data.size()) { + auto sent = + ::send(sd, data.data() + offset, data.size() - offset, MSG_NOSIGNAL); + if (sent < 0) + throw SystemError("Failed to send C37.118 frame"); + + offset += sent; + } +} + +bool C37Node::recvAll(int sd, std::span buf) { + std::size_t offset = 0; + while (offset < buf.size()) { + auto received = ::recv(sd, buf.data() + offset, buf.size() - offset, 0); + if (received < 0) + throw SystemError("Failed to receive from C37.118 connection"); + + if (received == 0) + return false; + + offset += received; + } + + return true; +} + +C37Result C37Node::recvFrame(int sd, C37Buffer &buffer, C37Frame &frame) { + for (std::size_t count = 0;;) { + auto result = frame.load(std::span(buffer).first(count)); + + auto required = c37_result_requested_size(result); + if (required == 0) + return result; + + if (!recvAll(sd, std::span(buffer).subspan(count, required - count))) + throw RuntimeError("Connection closed by peer"); + + count = required; + } +} + +void C37Node::clientSendCommand(C37CommandType cmd) { + C37Command command = {.cmd = cmd, .ext = {}}; + + C37FrameMetadata metadata = { + .idcode = client.idcode, + .soc = time_now(), + }; + + C37Frame::serialize_command(command, metadata, client.config, client.tx); + + sendAll(client.sd, client.tx); +} + +void C37Node::clientBuildSignals() { + auto signals = std::make_shared(); + + for (auto const i : std::views::iota(size_t(0), client.config.pmu.size())) { + auto const &pmu = client.config.pmu[i]; + + // Prefix every signal with the PMU index so the names are unique across + // PMUs and usable in path signal-mapping expressions. + for (auto const &phasor : pmu.phasor) + signals->push_back(std::make_shared( + fmt::format("pmu{}_{}", i, phasor.name), "", SignalType::COMPLEX)); + + signals->push_back(std::make_shared( + fmt::format("pmu{}_frequency", i), "Hz", SignalType::FLOAT)); + signals->push_back(std::make_shared(fmt::format("pmu{}_rocof", i), + "Hz/s", SignalType::FLOAT)); + + for (auto const &analog : pmu.analog) + signals->push_back(std::make_shared( + fmt::format("pmu{}_{}", i, analog.name), "", SignalType::FLOAT)); + + for (auto const &digital : pmu.digital) + for (auto const b : std::views::iota(size_t(0), digital.name.size())) + if ((digital.mask >> b) & 1) + signals->push_back(std::make_shared( + fmt::format("pmu{}_{}", i, digital.name[b]), "", + SignalType::BOOLEAN)); + } + + in.signals = signals; +} + +C37PhasorInfo C37Node::parsePhasor(Json const &json, + std::vector &names) const { + C37PhasorInfo phasor; + auto const &signal = names.emplace_back(json.at("signal")); + phasor.name = json.value("name", signal); + + phasor.unit = json.at("unit").get(); + phasor.component = + json.value("component", C37PhasorComponent::POSITIVE_SEQUENCE); + phasor.amplitude_scale = 1.0f; + phasor.phase_shift = 0.0f; + + phasor.upsampled_with_interpolation = false; + phasor.upsampled_with_extrapolation = false; + phasor.downsampled_with_reselection = false; + phasor.downsampled_with_fir_filter = false; + phasor.downsampled_with_non_fir_filter = false; + phasor.filtered_without_resampling = false; + phasor.magnitude_adjusted_for_calibration = false; + phasor.phase_adjusted_for_calibration = false; + phasor.phase_adjusted_for_rotation = false; + phasor.pseudo_phasor_value = false; + phasor.other_modification = false; + + if (json.contains("modifications")) { + for (auto const &json_mod : json.at("modifications")) { + auto const &mod = json_mod.get_ref(); + + if (mod == "upsampled_with_interpolation") + phasor.upsampled_with_interpolation = true; + else if (mod == "upsampled_with_extrapolation") + phasor.upsampled_with_extrapolation = true; + else if (mod == "downsampled_with_reselection") + phasor.downsampled_with_reselection = true; + else if (mod == "downsampled_with_fir_filter") + phasor.downsampled_with_fir_filter = true; + else if (mod == "downsampled_with_non_fir_filter") + phasor.downsampled_with_non_fir_filter = true; + else if (mod == "filtered_without_resampling") + phasor.filtered_without_resampling = true; + else if (mod == "magnitude_adjusted_for_calibration") + phasor.magnitude_adjusted_for_calibration = true; + else if (mod == "phase_adjusted_for_calibration") + phasor.phase_adjusted_for_calibration = true; + else if (mod == "phase_adjusted_for_rotation") + phasor.phase_adjusted_for_rotation = true; + else if (mod == "pseudo_phasor_value") + phasor.pseudo_phasor_value = true; + else if (mod == "other") + phasor.other_modification = true; + else + throw RuntimeError("Invalid phasor modification '{}'", mod); + } + } + + return phasor; +} + +C37AnalogInfo C37Node::parseAnalog(Json const &json, + std::vector &names) const { + C37AnalogInfo analog; + auto const &signal = names.emplace_back(json.at("signal")); + analog.name = json.value("name", signal); + + analog.unit = json.value("unit", C37AnalogUnit::POINT_ON_WAVE); + analog.scale = 1.0f; + analog.offset = 0.0f; + + return analog; +} + +void C37Node::parseDigital(Json const &json, + std::vector &digital, + std::vector &names) const { + // Start a new status word when there is none yet or the last one is full. + if (digital.empty() || digital.back().mask == 0xFFFF) + digital.emplace_back(); + + auto &word = digital.back(); + auto const b = std::countr_one(word.mask); + + auto const &signal = names.emplace_back(json.at("signal")); + word.name[b] = json.value("name", signal); + word.mask |= uint16_t(1 << b); + if (json.value("normal", false)) + word.normal |= uint16_t(1 << b); +} + +C37Pmu C37Node::parsePmu(Json const &json, + std::vector &names) const { + C37Pmu pmu; + pmu.name = json.at("name"); + pmu.idcode = json.value("idcode", 1); + + if (json.contains("guid")) { + if (uuid_parse(json.at("guid").get_ref().c_str(), + pmu.guid) != 0) + throw RuntimeError("Invalid PMU guid '{}'", json.at("guid")); + } else { + uuid_copy(pmu.guid, uuid); + } + + pmu.phasor_is_polar = false; + pmu.phasor_is_float = true; + pmu.analog_is_float = true; + pmu.freq_is_float = true; + + pmu.phasor.clear(); + if (json.contains("phasor")) + for (auto const &phasor : json.at("phasor")) + pmu.phasor.push_back(parsePhasor(phasor, names)); + + names.emplace_back(json.at("frequency")); + names.emplace_back(json.at("rocof")); + + pmu.analog.clear(); + if (json.contains("analog")) + for (auto const &analog : json.at("analog")) + pmu.analog.push_back(parseAnalog(analog, names)); + + // Each entry of the digital array is a single bit, batched into 16-bit words. + pmu.digital.clear(); + if (json.contains("digital")) + for (auto const &digital : json.at("digital")) + parseDigital(digital, pmu.digital, names); + + pmu.latitude = + json.value("latitude", std::numeric_limits::quiet_NaN()); + pmu.longitude = + json.value("longitude", std::numeric_limits::quiet_NaN()); + pmu.elevation = + json.value("elevation", std::numeric_limits::quiet_NaN()); + pmu.service_class = json.value("service_class", C37ServiceClass::M); + pmu.window = json.value("window", 0); + pmu.group_delay = json.value("group_delay", 0); + + pmu.nominal_frequency = json.value("nominal_frequency", 50.0f); + if (pmu.nominal_frequency != 50.0f && pmu.nominal_frequency != 60.0f) + throw RuntimeError("Invalid nominal frequency {} Hz; must be 50 or 60 Hz", + pmu.nominal_frequency); + + pmu.cfgcnt = 0; + + return pmu; +} + +C37Config C37Node::parseConfig(Json const &json, + std::vector &names) const { + C37Config config; + config.time_base = json.value("time_base", 1'000'000); + + config.pmu.clear(); + if (json.contains("pmus")) + for (auto const &pmu : json.at("pmus")) + config.pmu.push_back(parsePmu(pmu, names)); + + // The reporting rate is configured in frames per second. C37.118 encodes it + // as a signed integer: a positive value is frames per second, a negative + // value is seconds per frame. + auto data_rate = json.at("data_rate").get(); + if (data_rate <= 0.0) + throw RuntimeError("Invalid data rate {}; must be positive", data_rate); + + config.data_rate = data_rate >= 1.0 ? int16_t(std::lround(data_rate)) + : int16_t(-std::lround(1.0 / data_rate)); + + return config; +} + +int C37Node::parse(json_t *json) { + if (auto ret = Node::parse(json); ret) + return ret; + + Json config = json; + + if (auto in = config.find("in"); + in != config.end() && in->contains("address")) { + auto address = in->at("address").get(); + client.idcode = in->value("idcode", client.idcode); + std::tie(client.host, client.service) = parse_address(address, "4712"); + } + + if (auto out = config.find("out"); + out != config.end() && out->contains("address")) { + auto address = out->at("address").get(); + server.idcode = out->value("idcode", server.idcode); + + // The configuration properties are inlined into the "out" object. + server.names.clear(); + server.config = parseConfig(*out, server.names); + + std::tie(server.host, server.service) = parse_address(address, "4712"); + } + + return 0; +} + +int C37Node::prepare() { + if (out.enabled) { + serverListen(); + + if (auto ret = queue_signalled_init(&server.queue); ret) + throw RuntimeError("Failed to initialize server queue"); + + server.thread = std::jthread( + [this](std::stop_token stop) { serverLoop(std::move(stop)); }); + } + + if (in.enabled) { + clientConnect(); + + clientSendCommand(C37CommandType::R1_GET_CONFIG2); + + C37Frame frame; + std::vector segmentation_buffer; + for (;;) { + if (auto result = recvFrame(client.sd, client.rx, frame); + result != C37Result::OK) + throw RuntimeError("Failed to load frame ({})", int(result)); + + auto sync = frame.sync(); + if (sync != C37Sync::R1_CONFIG2 && sync != C37Sync::R2_CONFIG3) { + logger->debug("Ignoring frame type {:#x}", int(sync)); + continue; + } + + auto result = + frame.deserialize_config(client.config, segmentation_buffer); + if (result == C37Result::OK) + break; + + if (result == C37Result::SEGMENTED) + continue; + + throw RuntimeError("Failed to deserialize configuration frame ({})", + int(result)); + } + + clientBuildSignals(); + + logger->info("Received configuration with {} PMU(s) and {} signal(s)", + client.config.pmu.size(), in.signals->size()); + } + + return Node::prepare(); +} + +int C37Node::start() { + if (auto ret = Node::start(); ret) + return ret; + + if (in.enabled) + clientSendCommand(C37CommandType::R1_DATA_START); + + if (out.enabled) + serverBuildIndices(*getOutputSignals()); + + return 0; +} + +int C37Node::stop() { + if (in.enabled) { + clientSendCommand(C37CommandType::R1_DATA_STOP); + clientDisconnect(); + } + + if (out.enabled) + serverShutdown(); + + return Node::stop(); +} + +int C37Node::reverse() { + // We can turn a server into a client connecting to its own address, but there + // is no meaningful way to turn a client into a server. + if (in.enabled) + throw RuntimeError("Cannot reverse a C37.118 client into a server"); + + client.host = server.host; + client.service = server.service; + client.idcode = server.idcode; + + std::swap(in.enabled, out.enabled); + + return 0; +} + +const std::string &C37Node::getDetails() { + details.clear(); + + auto append = [&](std::string const &entry) { + if (!details.empty()) + details += ", "; + details += entry; + }; + + if (in.enabled) { + append(fmt::format("in.address={}:{}", client.host, client.service)); + for (auto const i : std::views::iota(size_t(0), client.config.pmu.size())) + append(fmt::format("in.pmu[{}].name={}", i, client.config.pmu[i].name)); + } + + if (out.enabled) { + append(fmt::format("out.address={}:{}", server.host, server.service)); + for (auto const i : std::views::iota(size_t(0), server.config.pmu.size())) + append(fmt::format("out.pmu[{}].name={}", i, server.config.pmu[i].name)); + } + + return details; +} + +int C37Node::_read(struct Sample *smps[], unsigned cnt) { + C37Frame frame; + + if (auto result = recvFrame(client.sd, client.rx, frame); + result != C37Result::OK) { + logger->warn("Dropping invalid frame ({})", int(result)); + return 0; + } + + if (frame.sync() != C37Sync::R1_DATA) { + logger->debug("Ignoring non-data frame of type {:#x}", int(frame.sync())); + return 0; + } + + std::vector data; + if (auto result = frame.deserialize_data(client.config, data); + result != C37Result::OK) { + logger->warn("Failed to deserialize data frame ({})", int(result)); + return 0; + } + + auto metadata = frame.metadata(client.config); + + auto *smp = smps[0]; + smp->length = 0; + smp->signals = getInputSignals(); + smp->flags = (int)SampleFlags::HAS_TS_ORIGIN | (int)SampleFlags::HAS_DATA; + smp->ts.origin = metadata.soc; + + auto push = [&](SignalData value) { + if (smp->length < smp->capacity) + smp->data[smp->length++] = value; + }; + + for (auto const i : std::views::iota(size_t(0), data.size())) { + auto const &d = data[i]; + auto const &pmu = client.config.pmu[i]; + + for (auto const &phasor : d.phasor) + push(SignalData(phasor)); + + push(SignalData(d.freq)); + push(SignalData(d.dfreq)); + + for (auto const &analog : d.analog) + push(SignalData(analog)); + + for (auto const j : std::views::iota(size_t(0), d.digital.size())) { + auto const word = d.digital[j]; + auto const &info = pmu.digital[j]; + + for (auto const b : std::views::iota(size_t(0), info.name.size())) + if ((info.mask >> b) & 1) + push(SignalData(bool((word >> b) & 1))); + } + } + + return 1; +} + +void C37Node::serverListen() { + addrinfo hints = {}; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + hints.ai_flags = AI_PASSIVE; + + auto addr = std::invoke([&]() { + addrinfo *result = nullptr; + + // An empty host binds to all local interfaces (AI_PASSIVE). + auto host = server.host.empty() ? nullptr : server.host.c_str(); + + if (auto ret = getaddrinfo(host, server.service.c_str(), &hints, &result); + ret != 0) + throw RuntimeError("Failed to resolve local address '{}:{}': {}", + server.host, server.service, gai_strerror(ret)); + + return std::unique_ptr(result); + }); + + int fd = -1; + for (auto *rp = addr.get(); rp != nullptr; rp = rp->ai_next) { + fd = ::socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol); + if (fd < 0) + continue; + + int on = 1; + ::setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)); + + if (::bind(fd, rp->ai_addr, rp->ai_addrlen) == 0) + break; + + ::close(fd); + fd = -1; + } + + if (fd < 0) + throw SystemError("Failed to bind to local address '{}:{}'", server.host, + server.service); + + if (::listen(fd, 1) < 0) { + ::close(fd); + throw SystemError("Failed to listen on local address '{}:{}'", server.host, + server.service); + } + + server.listen_sd = fd; +} + +void C37Node::serverShutdown() { + // A non-joinable thread means the server was never started (or was already + // shut down), so there is nothing to tear down. + if (!server.thread.joinable()) + return; + + server.thread.request_stop(); + server.thread.join(); + + serverCloseClient(); + + if (server.listen_sd >= 0) { + ::close(server.listen_sd); + server.listen_sd = -1; + } + + // Release references to any samples still queued (e.g. if the loop exited on + // an error before draining). + void *ptr; + while (queue_pull(&server.queue.queue, &ptr) == 1) + sample_decref(static_cast(ptr)); + + std::ignore = queue_signalled_destroy(&server.queue); +} + +void C37Node::serverLoop(std::stop_token stop) { + while (!stop.stop_requested() || + queue_signalled_available(&server.queue) > 0) { + auto queue_fd = queue_signalled_fd(&server.queue); + + std::array fds = {{ + {.fd = server.listen_sd, .events = POLLIN, .revents = 0}, + {.fd = queue_fd, .events = POLLIN, .revents = 0}, + {.fd = server.client_sd, .events = POLLIN, .revents = 0}, + }}; + + auto ret = ::poll(fds.data(), server.client_sd != -1 ? 3 : 2, 100); + if (ret < 0) { + if (errno == EINTR) + continue; + + logger->warn("Server poll failed: {}", strerror(errno)); + break; + } + + if (fds[0].revents & POLLIN) + serverAcceptClient(); + + if (fds[1].revents & POLLIN) + serverDrainQueue(); + + if (fds[2].revents & (POLLIN | POLLHUP | POLLERR)) { + try { + serverHandleCommand(); + } catch (std::exception &e) { + logger->warn("Client connection error: {}", e.what()); + serverCloseClient(); + } + } + } +} + +void C37Node::serverAcceptClient() { + int fd = ::accept(server.listen_sd, nullptr, nullptr); + if (fd < 0) { + logger->warn("Failed to accept connection: {}", strerror(errno)); + return; + } + + if (server.client_sd >= 0) { + logger->info("Replacing existing client connection"); + serverCloseClient(); + } + + server.client_sd = fd; + server.streaming = false; + logger->info("Client connected"); +} + +void C37Node::serverCloseClient() { + if (server.client_sd < 0) + return; + + ::close(server.client_sd); + server.client_sd = -1; + server.streaming = false; +} + +void C37Node::serverHandleCommand() { + C37Frame frame; + if (auto result = recvFrame(server.client_sd, server.rx, frame); + result != C37Result::OK) { + logger->warn("Dropping invalid frame ({})", int(result)); + return; + } + + if (frame.sync() != C37Sync::R1_COMMAND) { + logger->debug("Ignoring non-command frame with sync {:#x}", + int(frame.sync())); + return; + } + + C37Command command; + if (auto result = frame.deserialize_command(command); + result != C37Result::OK) { + logger->warn("Failed to deserialize command ({})", int(result)); + return; + } + + C37FrameMetadata metadata = { + .idcode = server.idcode, + .soc = time_now(), + }; + + switch (command.cmd) { + case C37CommandType::R1_GET_CONFIG1: + C37Frame::serialize_config(C37ConfigType::R1_CONFIG1, metadata, + server.config, server.tx); + sendAll(server.client_sd, server.tx); + break; + + case C37CommandType::R1_GET_CONFIG2: + C37Frame::serialize_config(C37ConfigType::R1_CONFIG2, metadata, + server.config, server.tx); + sendAll(server.client_sd, server.tx); + break; + + case C37CommandType::R2_GET_CONFIG3: + C37Frame::serialize_config(C37ConfigType::R2_CONFIG3, metadata, + server.config, server.tx); + sendAll(server.client_sd, server.tx); + break; + + case C37CommandType::R1_DATA_START: + server.streaming = true; + logger->info("Client started data transmission"); + break; + + case C37CommandType::R1_DATA_STOP: + server.streaming = false; + logger->info("Client stopped data transmission"); + break; + + default: + logger->warn("Ignoring unsupported command {:#x}", int(command.cmd)); + } +} + +void C37Node::serverDrainQueue() { + // Reset the queue's event descriptor so poll stops signalling it. + uint64_t counter; + auto fd = queue_signalled_fd(&server.queue); + while (::read(fd, &counter, sizeof(counter)) < 0 && errno == EINTR) + ; + + void *ptr; + while (queue_pull(&server.queue.queue, &ptr) == 1) { + auto *smp = static_cast(ptr); + + if (server.client_sd >= 0 && server.streaming) { + try { + serverSendSample(smp); + } catch (std::exception &e) { + logger->warn("Failed to send data frame: {}", e.what()); + serverCloseClient(); + } + } + + sample_decref(smp); + } +} + +void C37Node::serverBuildIndices(SignalList &signals) { + server.indices.clear(); + server.indices.reserve(server.names.size()); + + auto push = [&, name = server.names.begin()](SignalType expected) mutable { + auto const &signal_name = *name++; + + auto index = signals.getIndexByName(signal_name); + if (index < 0) + throw RuntimeError("Configured signal '{}' not found in the input", + signal_name); + + auto signal = signals.getByIndex(index); + if (signal->type != expected) + throw RuntimeError( + "Configured signal '{}' has type {} but {} is required", signal_name, + signalTypeToString(signal->type), signalTypeToString(expected)); + + server.indices.push_back(index); + }; + + for (auto const &pmu : server.config.pmu) { + for (auto const &phasor : pmu.phasor) { + std::ignore = phasor; + push(SignalType::COMPLEX); + } + + push(SignalType::FLOAT); // freq + push(SignalType::FLOAT); // rocof + + for (auto const &analog : pmu.analog) { + std::ignore = analog; + push(SignalType::FLOAT); + } + + for (auto const &word : pmu.digital) + for (auto const b : std::views::iota(size_t(0), word.name.size())) + if ((word.mask >> b) & 1) + push(SignalType::BOOLEAN); + } +} + +void C37Node::serverSendSample(struct Sample const *smp) { + auto const &config = server.config; + auto &data = server.data; + + // Reuse the C37Data (and its per-PMU vectors) across samples; after the first + // sample these resizes are no-ops and perform no allocation. + data.resize(config.pmu.size()); + + auto value = [&](int index) -> SignalData { + return index >= 0 && unsigned(index) < smp->length ? smp->data[index] + : SignalData{}; + }; + + auto index = server.indices.begin(); + auto next = [&]() -> SignalData { return value(*index++); }; + + for (auto const i : std::views::iota(size_t(0), config.pmu.size())) { + auto const &pmu = config.pmu[i]; + auto &d = data[i]; + + d.trigger_reason = C37DataTriggerReason::MANUAL; + d.unlocked_time = + C37DataUnlockedTime::LOCKED_OR_UNLOCKED_FOR_LESS_THAN_10_SECONDS; + d.time_error = C37DataTimeError::ESTIMATED_LESS_THAN_100_NANOSECONDS; + d.configuration_change = false; + d.trigger_detected = false; + d.sorted_by_arrival = false; + d.out_of_sync = false; + d.data_error = false; + + d.phasor.resize(pmu.phasor.size()); + d.analog.resize(pmu.analog.size()); + d.digital.resize(pmu.digital.size()); + + for (auto &phasor : d.phasor) + phasor = next().z; + + d.freq = next().f; + d.dfreq = next().f; + + for (auto &analog : d.analog) + analog = next().f; + + for (auto const j : std::views::iota(size_t(0), pmu.digital.size())) { + auto const &info = pmu.digital[j]; + + uint16_t word = 0; + for (auto const b : std::views::iota(size_t(0), info.name.size())) + if ((info.mask >> b) & 1) + if (next().b) + word |= uint16_t(1 << b); + + d.digital[j] = word; + } + } + + C37FrameMetadata metadata = { + .idcode = server.idcode, + .soc = (smp->flags & (int)SampleFlags::HAS_TS_ORIGIN) ? smp->ts.origin + : time_now(), + }; + + C37Frame::serialize_data(data, metadata, config, server.tx); + sendAll(server.client_sd, server.tx); +} + +int C37Node::_write(struct Sample *smps[], unsigned cnt) { + if (!server.thread.joinable()) + return -1; + + for (auto const i : std::views::iota(unsigned(0), cnt)) { + sample_incref(smps[i]); + + if (queue_signalled_push(&server.queue, smps[i]) != 1) { + sample_decref(smps[i]); + logger->warn("Server queue full; dropping sample"); + } + } + + return cnt; +} + +} // namespace + +static char name[] = "c37.118"; +static char description[] = "IEEE C37.118 Synchrophasor protocol"; +static NodePlugin + p; + +} // namespace villas::node diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt index 7f9ddd71c..3037b2621 100644 --- a/tests/unit/CMakeLists.txt +++ b/tests/unit/CMakeLists.txt @@ -5,6 +5,7 @@ # SPDX-License-Identifier: Apache-2.0 set(TEST_SRC + c37_118.cpp config_json.cpp config.cpp format.cpp @@ -23,6 +24,7 @@ add_executable(unit-tests ${TEST_SRC}) target_link_libraries(unit-tests PUBLIC PkgConfig::CRITERION Threads::Threads + nodes villas ) diff --git a/tests/unit/c37_118.cpp b/tests/unit/c37_118.cpp new file mode 100644 index 000000000..b35497e08 --- /dev/null +++ b/tests/unit/c37_118.cpp @@ -0,0 +1,187 @@ +/* Unit tests for C37.118 parser. + * + * Author: Philipp Jungkamp + * SPDX-FileCopyrightText: 2014-2023 Institute for Automation of Complex Power Systems, RWTH Aachen University + * SPDX-License-Identifier: Apache-2.0 + */ + +#include +#include +#include + +#include +#include + +#include + +using namespace villas::node; + +// cppcheck-suppress syntaxError +ParameterizedTestParameters(c37_118, parser) { + static criterion::parameters> params = + {}; + + params.push_back( // Config2 + {0xaa, 0x31, 0x00, 0x86, 0x00, 0xf1, 0x48, 0x93, 0x34, 0x4a, 0x00, 0x19, + 0x99, 0x9a, 0x00, 0xff, 0xff, 0xff, 0x00, 0x01, 0x42, 0x6c, 0x75, 0x65, + 0x20, 0x50, 0x4d, 0x55, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, + 0x00, 0xf1, 0x00, 0x06, 0x00, 0x04, 0x00, 0x00, 0x00, 0x00, 0x56, 0x31, + 0x4c, 0x50, 0x4d, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, + 0x20, 0x20, 0x56, 0x41, 0x4c, 0x50, 0x4d, 0x20, 0x20, 0x20, 0x20, 0x20, + 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x56, 0x42, 0x4c, 0x50, 0x4d, 0x20, + 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x56, 0x43, + 0x4c, 0x50, 0x4d, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, 0x20, + 0x20, 0x20, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, + 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, 0x00, 0x01, 0x00, 0x59, 0x00, 0x32, + 0xc1, 0xe2}); + + params.push_back( // Data + {0xaa, 0x01, 0x00, 0x36, 0x00, 0xf1, 0x48, 0x93, 0x34, 0x4a, 0x00, + 0x1e, 0xb8, 0x52, 0x08, 0x00, 0x42, 0xf6, 0x8f, 0x24, 0xc7, 0xc3, + 0x66, 0x23, 0x43, 0x01, 0x88, 0xcb, 0xc7, 0xc3, 0x63, 0x32, 0xc7, + 0xa9, 0x56, 0x76, 0x47, 0x42, 0xfe, 0x4b, 0x47, 0xa9, 0x1c, 0xdd, + 0x47, 0x43, 0xd1, 0x44, 0x00, 0x00, 0x00, 0x00, 0x47, 0xef}); + + params.push_back( // Command + {0xaa, 0x41, 0x00, 0x12, 0x00, 0xf1, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x02, 0xa7, 0x37}); + + return params; +} + +ParameterizedTest(criterion::parameters *param, c37_118, + parser) { + static C37Config const config = { + .time_base = 0x00FFFFFFu, + .pmu = {C37Pmu{ + .name = {"Blue PMU"}, + .idcode = 241, + .phasor_is_polar = false, + .phasor_is_float = true, + .analog_is_float = true, + .freq_is_float = false, + .phasor = + { + C37PhasorInfo{.name = {"V1LPM"}, + .unit = C37PhasorUnit::VOLT, + .amplitude_scale = 1}, + C37PhasorInfo{.name = {"VALPM"}, + .unit = C37PhasorUnit::VOLT, + .amplitude_scale = 1}, + C37PhasorInfo{.name = {"VBLPM"}, + .unit = C37PhasorUnit::VOLT, + .amplitude_scale = 1}, + C37PhasorInfo{.name = {"VCLPM"}, + .unit = C37PhasorUnit::VOLT, + .amplitude_scale = 1}, + }, + .analog = {}, + .digital = {}, + .nominal_frequency = 50.0f, + .cfgcnt = 0x59, + }}, + .data_rate = 50, + }; + + C37Frame frame; + cr_assert_eq( + frame.load(std::span(reinterpret_cast(param->data()), + param->size())), + C37Result::OK); + + switch (frame.sync()) { + case C37Sync::R1_CONFIG2: { + C37Config parsed; + std::vector segmentation_buffer; + cr_assert_eq(frame.deserialize_config(parsed, segmentation_buffer), + C37Result::OK); + cr_assert_eq(parsed.time_base, config.time_base); + cr_assert_eq(parsed.data_rate, config.data_rate); + cr_assert_eq(parsed.pmu.size(), config.pmu.size()); + + auto const &pmu = parsed.pmu[0]; + auto const &ref = config.pmu[0]; + cr_assert_str_eq(pmu.name.c_str(), ref.name.c_str()); + cr_assert_eq(pmu.idcode, ref.idcode); + cr_assert_eq(pmu.phasor_is_float, ref.phasor_is_float); + cr_assert_eq(pmu.phasor_is_polar, ref.phasor_is_polar); + cr_assert_eq(pmu.analog_is_float, ref.analog_is_float); + cr_assert_eq(pmu.freq_is_float, ref.freq_is_float); + cr_assert_eq(pmu.nominal_frequency, ref.nominal_frequency); + cr_assert_eq(pmu.cfgcnt, ref.cfgcnt); + cr_assert_eq(pmu.phasor.size(), ref.phasor.size()); + cr_assert_eq(pmu.analog.size(), ref.analog.size()); + cr_assert_eq(pmu.digital.size(), ref.digital.size()); + for (auto const i : std::views::iota(size_t(0), ref.phasor.size())) { + cr_assert_str_eq(pmu.phasor[i].name.c_str(), ref.phasor[i].name.c_str()); + cr_assert_eq(pmu.phasor[i].unit, ref.phasor[i].unit); + cr_assert_eq(pmu.phasor[i].amplitude_scale, + ref.phasor[i].amplitude_scale); + } + + std::vector out; + cr_assert_eq(C37Frame::serialize_config(C37ConfigType::R1_CONFIG2, + frame.metadata(parsed), parsed, + out), + C37Result::OK); + C37Frame frame2; + cr_assert_eq(frame2.load(out), C37Result::OK); + C37Config config2; + cr_assert_eq(frame2.deserialize_config(config2, segmentation_buffer), + C37Result::OK); + cr_assert_eq(config2.time_base, parsed.time_base); + cr_assert_eq(config2.data_rate, parsed.data_rate); + cr_assert_eq(config2.pmu[0].nominal_frequency, pmu.nominal_frequency); + cr_assert_eq(config2.pmu[0].cfgcnt, pmu.cfgcnt); + } break; + + case C37Sync::R1_DATA: { + std::vector data; + cr_assert_eq(frame.deserialize_data(config, data), C37Result::OK); + cr_assert_eq(data.size(), 1u); + + auto const &d = data[0]; + cr_assert_eq(d.sorted_by_arrival, 1); + cr_assert_eq(d.trigger_reason, C37DataTriggerReason::MANUAL); + cr_assert_eq(d.data_error, false); + cr_assert_eq(d.phasor.size(), 4u); + cr_assert_eq(d.phasor[0].real(), std::bit_cast(0x42f68f24u)); + cr_assert_eq(d.phasor[0].imag(), std::bit_cast(0xc7c36623u)); + cr_assert_eq(d.phasor[1].real(), std::bit_cast(0x430188cbu)); + cr_assert_eq(d.phasor[1].imag(), std::bit_cast(0xc7c36332u)); + cr_assert_eq(d.phasor[2].real(), std::bit_cast(0xc7a95676u)); + cr_assert_eq(d.phasor[2].imag(), std::bit_cast(0x4742fe4bu)); + cr_assert_eq(d.phasor[3].real(), std::bit_cast(0x47a91cddu)); + cr_assert_eq(d.phasor[3].imag(), std::bit_cast(0x4743d144u)); + cr_assert_float_eq(d.freq, 50.0f, 1e-3f); + cr_assert_float_eq(d.dfreq, 0.0f, 1e-6f); + + std::vector out; + cr_assert_eq( + C37Frame::serialize_data(data, frame.metadata(config), config, out), + C37Result::OK); + cr_assert_eq(out.size(), param->size()); + for (auto const index : std::views::iota(size_t(0), out.size())) + cr_assert_eq(out[index], std::byte(param->at(index)), + "mismatch at index %lu", index); + } break; + + case C37Sync::R1_COMMAND: { + C37Command command; + cr_assert_eq(frame.deserialize_command(command), C37Result::OK); + cr_assert_eq(command.cmd, C37CommandType::R1_DATA_START); + cr_assert_eq(command.ext.size(), 0u); + + std::vector out; + cr_assert_eq(C37Frame::serialize_command( + {.cmd = C37CommandType::R1_DATA_START, .ext = {}}, + frame.metadata(config), config, out), + C37Result::OK); + cr_assert_eq(out.size(), param->size()); + cr_assert_eq(std::memcmp(out.data(), param->data(), param->size()), 0); + } break; + + default: + break; + } +}