diff --git a/Makefile b/Makefile index a5a83cc..c267e8a 100644 --- a/Makefile +++ b/Makefile @@ -21,7 +21,8 @@ INCLUDES := \ -Idata_acq_and_processing/preprocessing/reference_master/include \ -Idata_acq_and_processing/preprocessing/data_preprocessor/include \ -Idata_acq_and_processing/processing/processors/include \ - -Idata_acq_and_processing/processing/data_processor/include + -Idata_acq_and_processing/processing/data_processor/include \ + -Idata_acq_and_processing/processing/locator/include COMMON_SOURCES := \ data_acq_and_processing/common_cpp/ipc/src/shm_ring.cpp \ @@ -46,13 +47,18 @@ PREPROC_SOURCES := \ data_acq_and_processing/preprocessing/data_preprocessor/src/data_preprocessor.cpp \ data_acq_and_processing/preprocessing/data_preprocessor/src/main.cpp +LOCATOR_SOURCES := \ + data_acq_and_processing/processing/locator/src/payload_builder.cpp \ + data_acq_and_processing/processing/locator/src/tcp_server.cpp + PROCESSOR_SOURCES := \ data_acq_and_processing/processing/processors/src/bscan_processor.cpp \ data_acq_and_processing/processing/processors/src/gpr_processor.cpp \ data_acq_and_processing/processing/processors/src/passthrough_processor.cpp \ data_acq_and_processing/processing/data_processor/src/processing_live_config.cpp \ data_acq_and_processing/processing/data_processor/src/data_processor.cpp \ - data_acq_and_processing/processing/data_processor/src/main.cpp + data_acq_and_processing/processing/data_processor/src/main.cpp \ + $(LOCATOR_SOURCES) SWEEP_ORCH_OBJS := $(addprefix $(BUILD_DIR)/,$(COMMON_SOURCES:.cpp=.o) $(ORCH_SOURCES:.cpp=.o)) PREPROCESSOR_OBJS := $(addprefix $(BUILD_DIR)/,$(COMMON_SOURCES:.cpp=.o) $(PREPROC_SOURCES:.cpp=.o)) diff --git a/data_acq_and_processing/common_cpp/config/include/run_config.hpp b/data_acq_and_processing/common_cpp/config/include/run_config.hpp index 4326d29..0f4d43a 100644 --- a/data_acq_and_processing/common_cpp/config/include/run_config.hpp +++ b/data_acq_and_processing/common_cpp/config/include/run_config.hpp @@ -4,6 +4,7 @@ #include #include +#include "locator/locator_config.hpp" #include "shared_types.hpp" namespace radar::config { @@ -148,6 +149,7 @@ struct RunConfig { RuntimeConfig runtime{}; PreprocessConfig preprocess{}; GprConfig gpr{}; + radar::locator::LocatorServerConfig locator_server{}; std::vector run_combos{}; }; diff --git a/data_acq_and_processing/common_cpp/config/src/run_config.cpp b/data_acq_and_processing/common_cpp/config/src/run_config.cpp index 44a3488..4ce65c2 100644 --- a/data_acq_and_processing/common_cpp/config/src/run_config.cpp +++ b/data_acq_and_processing/common_cpp/config/src/run_config.cpp @@ -194,6 +194,33 @@ using Json = nlohmann::json; return notch; } +[[nodiscard]] auto parse_locator_server(const Json& object) -> radar::locator::LocatorServerConfig { + radar::locator::LocatorServerConfig locator{}; + locator.enabled = optional_bool(object, "enabled", locator.enabled); + locator.host = optional_string(object, "host", locator.host); + locator.device_id = optional_u32(object, "device_id", locator.device_id); + locator.protocol_version = optional_u32(object, "protocol_version", locator.protocol_version); + locator.max_payload_bytes = optional_u32(object, "max_payload_bytes", locator.max_payload_bytes); + locator.client_queue_size = optional_u32(object, "client_queue_size", locator.client_queue_size); + + if (const auto* port_value = optional_field(object, "port"); port_value != nullptr) { + const auto port_u32 = number_to_u32(as_number(*port_value, "run.locator_server.port"), + "run.locator_server.port"); + if (port_u32 == 0U || port_u32 > 0xFFFFU) { + throw std::runtime_error("run.locator_server.port must be in [1, 65535]"); + } + locator.port = static_cast(port_u32); + } + + if (locator.client_queue_size == 0U) { + throw std::runtime_error("run.locator_server.client_queue_size must be > 0"); + } + if (locator.max_payload_bytes == 0U) { + throw std::runtime_error("run.locator_server.max_payload_bytes must be > 0"); + } + return locator; +} + [[nodiscard]] auto parse_driver_mode(const std::string& value) -> DriverMode { if (value == "mock") { return DriverMode::Mock; @@ -479,6 +506,12 @@ auto load_run_config(const std::string& path) -> RunConfig { "python_app/runtime/processing_live.json" ); + if (const auto* locator_value = optional_field(*run_obj, "locator_server"); + locator_value != nullptr) { + config.locator_server = + parse_locator_server(*as_object(*locator_value, "run.locator_server")); + } + const auto* combos = as_array(required_field(*run_obj, "combos"), "run.combos"); if (combos->empty()) { throw std::runtime_error("run.combos must not be empty"); diff --git a/data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp b/data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp index e78033c..93a29c9 100644 --- a/data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp +++ b/data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp @@ -83,6 +83,11 @@ struct ResultBlock { struct ResultCollection { std::uint64_t collection_id = 0; std::uint64_t monotonic_ns = 0; + // Wall-clock duration of `process_collection()` for this collection, + // measured on the data_processor side with a monotonic clock. Used by the + // Python pipeline to report processing-time metrics. Zero is a valid + // "unmeasured" sentinel for legacy producers. + std::uint64_t processing_duration_ns = 0; std::vector collection_payloads{}; std::vector blocks{}; }; diff --git a/data_acq_and_processing/common_cpp/ipc/src/shared_types.cpp b/data_acq_and_processing/common_cpp/ipc/src/shared_types.cpp index 0b81f01..00408fe 100644 --- a/data_acq_and_processing/common_cpp/ipc/src/shared_types.cpp +++ b/data_acq_and_processing/common_cpp/ipc/src/shared_types.cpp @@ -14,7 +14,7 @@ namespace { constexpr std::uint32_t kRawCollectionMagic = 0x32574152U; // RAW2 constexpr std::uint32_t kPreprocessedCollectionMagic = 0x32525050U; // PRP2 -constexpr std::uint32_t kResultCollectionMagic = 0x314C5352U; // RSL1 +constexpr std::uint32_t kResultCollectionMagic = 0x324C5352U; // RSL2 template concept TriviallySerializable = std::is_trivially_copyable_v; @@ -408,6 +408,7 @@ auto serialize_result_collection(const ResultCollection& collection) -> std::vec writer.write(kResultCollectionMagic); writer.write(collection.collection_id); writer.write(collection.monotonic_ns); + writer.write(collection.processing_duration_ns); writer.write(checked_count_to_u32(collection.collection_payloads.size(), "Collection payload count")); writer.write(checked_count_to_u32(collection.blocks.size(), "Result block count")); @@ -432,6 +433,7 @@ auto deserialize_result_collection(std::span bytes) -> Resul ResultCollection collection{}; collection.collection_id = reader.read(); collection.monotonic_ns = reader.read(); + collection.processing_duration_ns = reader.read(); const auto collection_payload_count = reader.read(); const auto block_count = reader.read(); diff --git a/data_acq_and_processing/processing/data_processor/include/data_processor.hpp b/data_acq_and_processing/processing/data_processor/include/data_processor.hpp index 61e2ba3..813866e 100644 --- a/data_acq_and_processing/processing/data_processor/include/data_processor.hpp +++ b/data_acq_and_processing/processing/data_processor/include/data_processor.hpp @@ -7,6 +7,7 @@ #include #include +#include "locator/tcp_server.hpp" #include "processor_interface.hpp" #include "processing_live_config.hpp" #include "run_config.hpp" @@ -22,7 +23,8 @@ class DataProcessor { const config::RunConfig& config, ipc::ShmRing& preprocessed_ring, ipc::ShmRing& results_ring, - ProcessorRegistry processors + ProcessorRegistry processors, + radar::locator::TcpServer* locator_server = nullptr ); void run(const std::atomic& stop_requested); @@ -38,12 +40,26 @@ class DataProcessor { [[nodiscard]] auto resolve_processor(const ProcessingLiveConfig& live_config) -> ProcessorInterface&; [[nodiscard]] auto should_replay_entire_history(const ProcessingLiveConfig& live_config) const -> bool; + // Merge live config with the latest socket-supplied speed (if any and if + // not suppressed by `ignore_socket_speed`). The returned config is what + // actually drives processing for this tick. + [[nodiscard]] auto resolve_effective_live_config(const ProcessingLiveConfig& live_config) const + -> ProcessingLiveConfig; + + // Convert one processed collection into a locator filter spec, derived + // from the live config and current processor mode. + [[nodiscard]] auto build_locator_filter(const ProcessingLiveConfig& live_config) const + -> radar::locator::FilterParams; + + void publish_locator(const ipc::ResultCollection& collection, const ProcessingLiveConfig& live_config); + const config::RunConfig& config_; ipc::ShmRing& preprocessed_ring_; ipc::ShmRing& results_ring_; ProcessorRegistry processors_{}; std::string default_processor_mode_{}; ProcessingLiveConfigLoader live_config_loader_; + radar::locator::TcpServer* locator_server_ = nullptr; }; [[nodiscard]] auto create_default_processors() -> ProcessorRegistry; diff --git a/data_acq_and_processing/processing/data_processor/include/processing_live_config.hpp b/data_acq_and_processing/processing/data_processor/include/processing_live_config.hpp index 25782c1..8b71f9e 100644 --- a/data_acq_and_processing/processing/data_processor/include/processing_live_config.hpp +++ b/data_acq_and_processing/processing/data_processor/include/processing_live_config.hpp @@ -52,6 +52,17 @@ struct ProcessingLiveConfig { // BP image is computed in the y=imaging_plane_y_m slice of the 3D grid. // Default 0 keeps legacy 1D antenna layouts imaging in the antenna plane. float gpr_imaging_plane_y_m = 0.0F; + // Locator filter parameters. Mode-dependent threshold (legacy_gpr uses + // `legacy_gpr_min_visible_pair_count`, everything else uses + // `gpr_min_visible_score`). Draw limits apply only to non-legacy modes. + float gpr_min_visible_score = 0.0F; + float legacy_gpr_min_visible_pair_count = 0.0F; + std::uint32_t gpr_max_detected_objects_to_draw = 0; + std::uint32_t gpr_draw_top_m_objects = 0; + // When true, the data_processor ignores socket-supplied `vlc` updates and + // keeps using `gpr_speed_m_s` from this file. Mirrored from the GUI's + // "ignore socket speed" checkbox. + bool ignore_socket_speed = false; bool reprocess_current_result = true; std::uint64_t history_command_seq = 0; HistoryCommand history_command = HistoryCommand::None; diff --git a/data_acq_and_processing/processing/data_processor/src/data_processor.cpp b/data_acq_and_processing/processing/data_processor/src/data_processor.cpp index 7aaaf86..f8d95ff 100644 --- a/data_acq_and_processing/processing/data_processor/src/data_processor.cpp +++ b/data_acq_and_processing/processing/data_processor/src/data_processor.cpp @@ -37,14 +37,16 @@ DataProcessor::DataProcessor( const config::RunConfig& config, ipc::ShmRing& preprocessed_ring, ipc::ShmRing& results_ring, - ProcessorRegistry processors + ProcessorRegistry processors, + radar::locator::TcpServer* locator_server ) : config_(config), preprocessed_ring_(preprocessed_ring), results_ring_(results_ring), processors_(std::move(processors)), default_processor_mode_(kDefaultProcessorMode), - live_config_loader_(config.runtime.processing_live_config_path) { + live_config_loader_(config.runtime.processing_live_config_path), + locator_server_(locator_server) { if (processors_.empty()) { throw std::runtime_error("DataProcessor requires at least one processor"); } @@ -61,7 +63,8 @@ void DataProcessor::run(const std::atomic& stop_requested) { std::uint64_t last_applied_history_command_seq = 0; while (!stop_requested.load(std::memory_order_relaxed)) { - const auto live_config = live_config_loader_.refresh_if_needed(); + const auto live_config_raw = live_config_loader_.refresh_if_needed(); + const auto live_config = resolve_effective_live_config(live_config_raw); const auto live_revision = live_config_loader_.revision(); auto& processor = resolve_processor(live_config); @@ -89,6 +92,7 @@ void DataProcessor::run(const std::atomic& stop_requested) { live_config ); publish_result_collection(replay_result, results_ring_); + publish_locator(replay_result, live_config); } } else if (!preprocessed_history.empty()) { const auto replay_result = process_collection( @@ -98,6 +102,7 @@ void DataProcessor::run(const std::atomic& stop_requested) { live_config ); publish_result_collection(replay_result, results_ring_); + publish_locator(replay_result, live_config); } last_replayed_revision = live_revision; } @@ -116,6 +121,7 @@ void DataProcessor::run(const std::atomic& stop_requested) { live_config ); publish_result_collection(result_collection, results_ring_); + publish_locator(result_collection, live_config); continue; } @@ -129,7 +135,13 @@ auto DataProcessor::process_collection( ProcessorInterface& processor, const ProcessingLiveConfig& live_config ) -> ipc::ResultCollection { - return processor.process_collection(config_, preprocessed, previous_collections, live_config); + const auto started_at = std::chrono::steady_clock::now(); + auto result = processor.process_collection(config_, preprocessed, previous_collections, live_config); + const auto finished_at = std::chrono::steady_clock::now(); + result.processing_duration_ns = static_cast( + std::chrono::duration_cast(finished_at - started_at).count() + ); + return result; } auto DataProcessor::resolve_processor(const ProcessingLiveConfig& live_config) -> ProcessorInterface& { @@ -151,6 +163,54 @@ auto DataProcessor::should_replay_entire_history(const ProcessingLiveConfig& liv return requested_mode == "bscan"; } +auto DataProcessor::resolve_effective_live_config(const ProcessingLiveConfig& live_config) const + -> ProcessingLiveConfig { + if (live_config.ignore_socket_speed || locator_server_ == nullptr) { + return live_config; + } + const auto socket_speed = locator_server_->latest_socket_speed(); + if (!socket_speed.has_value()) { + return live_config; + } + ProcessingLiveConfig effective = live_config; + effective.gpr_speed_m_s = static_cast(*socket_speed); + return effective; +} + +auto DataProcessor::build_locator_filter(const ProcessingLiveConfig& live_config) const + -> radar::locator::FilterParams { + const std::string requested_mode = + live_config.processor_mode.empty() ? default_processor_mode_ : live_config.processor_mode; + + radar::locator::FilterParams filter{}; + if (requested_mode == "legacy_gpr") { + filter.min_score = live_config.legacy_gpr_min_visible_pair_count; + // The GUI deliberately disables the "draw top N" capping for legacy + // GPR, so we also skip it on the wire to match observation semantics. + filter.draw_limits.reset(); + } else { + filter.min_score = live_config.gpr_min_visible_score; + if (live_config.gpr_max_detected_objects_to_draw > 0U + && live_config.gpr_draw_top_m_objects > 0U) { + filter.draw_limits = radar::locator::DrawLimits{ + .max_detected_objects = live_config.gpr_max_detected_objects_to_draw, + .draw_top_objects = live_config.gpr_draw_top_m_objects, + }; + } + } + return filter; +} + +void DataProcessor::publish_locator( + const ipc::ResultCollection& collection, + const ProcessingLiveConfig& live_config +) { + if (locator_server_ == nullptr || !locator_server_->is_running()) { + return; + } + locator_server_->publish(collection, build_locator_filter(live_config)); +} + auto create_default_processors() -> ProcessorRegistry { ProcessorRegistry processors{}; { diff --git a/data_acq_and_processing/processing/data_processor/src/main.cpp b/data_acq_and_processing/processing/data_processor/src/main.cpp index 903c257..acb1f23 100644 --- a/data_acq_and_processing/processing/data_processor/src/main.cpp +++ b/data_acq_and_processing/processing/data_processor/src/main.cpp @@ -2,9 +2,11 @@ #include #include #include +#include #include #include "data_processor.hpp" +#include "locator/tcp_server.hpp" #include "run_config.hpp" #include "shm_ring.hpp" @@ -21,6 +23,8 @@ void signal_handler(int /*signal*/) { void install_signal_handlers() { std::signal(SIGINT, signal_handler); std::signal(SIGTERM, signal_handler); + // Writing to a peer-closed socket would otherwise terminate the process. + std::signal(SIGPIPE, SIG_IGN); } [[nodiscard]] auto read_config_path(int argc, char** argv) -> std::string { @@ -34,6 +38,21 @@ void install_signal_handlers() { return config_path; } +[[nodiscard]] auto start_locator_server(const radar::config::RunConfig& config) + -> std::unique_ptr { + if (!config.locator_server.enabled) { + return nullptr; + } + auto server = std::make_unique(config.locator_server); + try { + server->start(); + } catch (const std::exception& exception) { + std::cerr << "data_processor: locator server disabled (" << exception.what() << ")\n"; + return nullptr; + } + return server; +} + } // namespace int main(int argc, char** argv) { @@ -54,11 +73,14 @@ int main(int argc, char** argv) { config.rings.results.slot_size_bytes ); + auto locator_server = start_locator_server(config); + radar::processing::DataProcessor processor( config, preprocessed_ring, results_ring, - radar::processing::create_default_processors() + radar::processing::create_default_processors(), + locator_server.get() ); processor.run(g_stop_requested); return 0; diff --git a/data_acq_and_processing/processing/data_processor/src/processing_live_config.cpp b/data_acq_and_processing/processing/data_processor/src/processing_live_config.cpp index e1fd4dd..84f34ad 100644 --- a/data_acq_and_processing/processing/data_processor/src/processing_live_config.cpp +++ b/data_acq_and_processing/processing/data_processor/src/processing_live_config.cpp @@ -311,6 +311,32 @@ void apply_legacy_gpr_algorithm_alias(ProcessingLiveConfig& config, const std::s } config.gpr_imaging_plane_y_m = static_cast(found->get()); } + if (const auto found = root.find("gpr_min_visible_score"); found != root.end()) { + if (!found->is_number()) { + throw std::runtime_error("processing.gpr_min_visible_score must be number"); + } + config.gpr_min_visible_score = static_cast(found->get()); + } + if (const auto found = root.find("legacy_gpr_min_visible_pair_count"); found != root.end()) { + if (!found->is_number()) { + throw std::runtime_error("processing.legacy_gpr_min_visible_pair_count must be number"); + } + config.legacy_gpr_min_visible_pair_count = static_cast(found->get()); + } + if (const auto found = root.find("gpr_max_detected_objects_to_draw"); found != root.end()) { + config.gpr_max_detected_objects_to_draw = + parse_u32_number(*found, "processing.gpr_max_detected_objects_to_draw"); + } + if (const auto found = root.find("gpr_draw_top_m_objects"); found != root.end()) { + config.gpr_draw_top_m_objects = + parse_u32_number(*found, "processing.gpr_draw_top_m_objects"); + } + if (const auto found = root.find("ignore_socket_speed"); found != root.end()) { + if (!found->is_boolean()) { + throw std::runtime_error("processing.ignore_socket_speed must be bool"); + } + config.ignore_socket_speed = found->get(); + } if (const auto found = root.find("reprocess_current_result"); found != root.end()) { if (!found->is_boolean()) { throw std::runtime_error("processing.reprocess_current_result must be bool"); diff --git a/data_acq_and_processing/processing/locator/include/locator/locator_config.hpp b/data_acq_and_processing/processing/locator/include/locator/locator_config.hpp new file mode 100644 index 0000000..448d1ef --- /dev/null +++ b/data_acq_and_processing/processing/locator/include/locator/locator_config.hpp @@ -0,0 +1,20 @@ +#pragma once + +#include +#include + +namespace radar::locator { + +// Static configuration for the locator TCP server, mirrored from `run.locator_server` +// in `run_config.json`. Values are validated by the config parser before reaching here. +struct LocatorServerConfig { + bool enabled = true; + std::string host = "0.0.0.0"; + std::uint16_t port = 8888; + std::uint32_t device_id = 3; + std::uint32_t protocol_version = 1; + std::uint32_t max_payload_bytes = 64U * 1024U; + std::uint32_t client_queue_size = 32; +}; + +} // namespace radar::locator diff --git a/data_acq_and_processing/processing/locator/include/locator/observation.hpp b/data_acq_and_processing/processing/locator/include/locator/observation.hpp new file mode 100644 index 0000000..44482bc --- /dev/null +++ b/data_acq_and_processing/processing/locator/include/locator/observation.hpp @@ -0,0 +1,14 @@ +#pragma once + +namespace radar::locator { + +// One outbound locator observation: object position relative to the radar in metres. +// `crs` is the cross-range (X) coordinate; `dst` is the range (Z) coordinate. +// Values are kept as float because they are quantised to two decimal places before +// being serialised on the wire. +struct Observation { + float crs = 0.0F; + float dst = 0.0F; +}; + +} // namespace radar::locator diff --git a/data_acq_and_processing/processing/locator/include/locator/payload_builder.hpp b/data_acq_and_processing/processing/locator/include/locator/payload_builder.hpp new file mode 100644 index 0000000..ed13015 --- /dev/null +++ b/data_acq_and_processing/processing/locator/include/locator/payload_builder.hpp @@ -0,0 +1,72 @@ +#pragma once + +#include +#include +#include +#include + +#include "locator/observation.hpp" +#include "shared_types.hpp" + +namespace radar::locator { + +// Inclusive bounds used to clip locator observations to a visible window. +struct VisibleBounds { + float x_min = 0.0F; + float x_max = 0.0F; + float z_min = 0.0F; + float z_max = 0.0F; +}; + +// Limits mirroring the GUI semantics: if the number of detected objects exceeds +// `max_detected_objects`, the result is intentionally empty (matches Python +// reference). Otherwise, at most `draw_top_objects` rows are emitted. +struct DrawLimits { + std::uint32_t max_detected_objects = 0; + std::uint32_t draw_top_objects = 0; +}; + +// Filter parameters for the payload builder. None of these fields couple to GUI +// state; they are resolved by the data_processor from the live processing config +// and passed in explicitly so the builder remains a pure function. +struct FilterParams { + float min_score = 0.0F; + std::optional visible_bounds{}; + std::optional draw_limits{}; +}; + +// Extract observations from one result collection. +// +// Mirrors `python_app/orchestration/gpr_locator.py::locator_observations_from_collection`: +// * Looks for a `gpr_points` TableF32 payload first, then falls back to +// `gpr_region_centers`. Both layouts encode `[x_m, z_m, score, ...]` rows. +// * Drops rows with non-finite coordinates and those below `min_score`. +// * If `visible_bounds` are provided, drops rows outside the inclusive window. +// * If `draw_limits` are provided and the surviving count exceeds +// `max_detected_objects`, returns an empty vector (the GUI's "too many to +// trust" heuristic). Otherwise, keeps the first `draw_top_objects` rows. +// +// The function never throws; malformed payloads degrade to an empty result. +[[nodiscard]] auto observations_from_collection( + const ipc::ResultCollection& collection, + const FilterParams& filter +) -> std::vector; + +// Serialize a list of observations into the JSON payload format expected by +// locator clients: `{"ver": , "tim": "HH:MM:SS.mmm", "sts": 1, "obs": [...]}`. +// Each observation contributes `{"dst": , "crs": }` with two-decimal +// quantisation. +[[nodiscard]] auto build_payload_json( + const std::vector& observations, + std::uint32_t protocol_version, + std::uint32_t status = 1U +) -> std::string; + +// Wrap a JSON payload string into a framed wire packet: +// ``. +[[nodiscard]] auto encode_packet( + const std::string& payload_json, + std::uint32_t device_id +) -> std::vector; + +} // namespace radar::locator diff --git a/data_acq_and_processing/processing/locator/include/locator/tcp_server.hpp b/data_acq_and_processing/processing/locator/include/locator/tcp_server.hpp new file mode 100644 index 0000000..6404695 --- /dev/null +++ b/data_acq_and_processing/processing/locator/include/locator/tcp_server.hpp @@ -0,0 +1,165 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "locator/locator_config.hpp" +#include "locator/payload_builder.hpp" +#include "shared_types.hpp" + +namespace radar::locator { + +// Bounded, in-memory packet queue used by the per-client writer thread. +// Marking the queue as full closes the client (back-pressure by disconnect), +// matching the semantics of the previous Python implementation. +class ClientQueue { + public: + explicit ClientQueue(std::size_t capacity); + + // Push a packet onto the queue. Returns false if the queue is full or has + // been closed; the caller is expected to disconnect the client in that case. + [[nodiscard]] auto try_push(std::vector packet) -> bool; + + // Block until a packet is available or the queue is closed. + // Returns nullopt iff the queue has been closed and is drained. + [[nodiscard]] auto wait_pop() -> std::optional>; + + // Wake any waiter and reject further pushes. Idempotent. + void close(); + + [[nodiscard]] auto is_closed() const -> bool; + + private: + mutable std::mutex mutex_{}; + std::condition_variable not_empty_{}; + std::deque> queue_{}; + std::size_t capacity_; + bool closed_ = false; +}; + +// One connected locator client: owns its socket, writer thread, reader thread, +// and outbound queue. Removed from the server's roster once both threads exit. +class ClientSession { + public: + ClientSession( + int socket_fd, + std::string peer_name, + std::size_t queue_capacity, + std::uint32_t max_payload_bytes + ); + + ~ClientSession(); + + ClientSession(const ClientSession&) = delete; + auto operator=(const ClientSession&) -> ClientSession& = delete; + ClientSession(ClientSession&&) = delete; + auto operator=(ClientSession&&) -> ClientSession& = delete; + + // Start writer/reader threads. The shared `vlc` slot is notified whenever + // an inbound packet carrying a finite `vlc` field arrives. + void start(std::atomic& shared_vlc_slot); + + // Enqueue one outbound packet. Disconnects this session if the queue is + // already full or the writer has stopped. + void enqueue(std::vector packet); + + // Initiate teardown of this client (idempotent): closes the queue and + // shuts the socket so writer/reader threads can exit promptly. + void request_stop(); + + // Join writer/reader threads and release the socket. Must be called from a + // thread other than this session's writer or reader. + void join(); + + [[nodiscard]] auto has_exited() const -> bool; + + [[nodiscard]] auto peer_name() const -> const std::string&; + + private: + void writer_loop(); + void reader_loop(std::atomic& shared_vlc_slot); + + int socket_fd_; + std::string peer_name_; + ClientQueue queue_; + std::uint32_t max_payload_bytes_; + std::atomic stop_requested_{false}; + std::atomic exited_{false}; + std::thread writer_thread_{}; + std::thread reader_thread_{}; +}; + +// Multi-client TCP locator server. +// +// Design contract: +// * Threading: one acceptor thread + two threads per connected client. The +// producer (data_processor) calls publish() synchronously; that call is +// non-blocking and never throws for typical operation. +// * Back-pressure: each client has its own bounded outbound queue. If a +// client is too slow to drain, the next publish() drops it (matches the +// prior Python service). Other clients are unaffected. +// * Latest-snapshot: the most recently published packet is cached and sent +// to every newly connected client before any new packets are forwarded. +// * Lifetime: `start()` may throw on listen failure. `stop()` is idempotent +// and is also invoked from the destructor. +class TcpServer { + public: + explicit TcpServer(LocatorServerConfig config); + + ~TcpServer(); + + TcpServer(const TcpServer&) = delete; + auto operator=(const TcpServer&) -> TcpServer& = delete; + TcpServer(TcpServer&&) = delete; + auto operator=(TcpServer&&) -> TcpServer& = delete; + + // Bind the listening socket and start the acceptor thread. + // Throws std::runtime_error on socket(), bind() or listen() failure. + void start(); + + // Tear down all clients and the acceptor. Idempotent. + void stop(); + + [[nodiscard]] auto is_running() const -> bool; + + // Build and broadcast one locator packet derived from a GPR result + // collection. Never blocks for I/O. Filter parameters are supplied by the + // caller so the builder remains state-free. + void publish(const ipc::ResultCollection& collection, const FilterParams& filter); + + // Most recent `vlc` value received from any connected client, or nullopt + // if no client has ever submitted one since startup. Lock-free. + [[nodiscard]] auto latest_socket_speed() const -> std::optional; + + private: + void acceptor_loop(); + void enroll_client(std::unique_ptr session); + void broadcast_packet(const std::vector& packet); + void reap_finished_clients(); + void cache_latest_packet(std::vector packet); + [[nodiscard]] auto latest_packet_copy() const -> std::optional>; + + LocatorServerConfig config_; + std::atomic running_{false}; + int listen_fd_ = -1; + std::thread acceptor_thread_{}; + + mutable std::mutex clients_mutex_{}; + std::vector> clients_{}; + + mutable std::mutex latest_packet_mutex_{}; + std::optional> latest_packet_{}; + + // Sentinel of "no value yet" is NaN. Lock-free read from data_processor. + std::atomic latest_socket_speed_{}; +}; + +} // namespace radar::locator diff --git a/data_acq_and_processing/processing/locator/src/payload_builder.cpp b/data_acq_and_processing/processing/locator/src/payload_builder.cpp new file mode 100644 index 0000000..a28badd --- /dev/null +++ b/data_acq_and_processing/processing/locator/src/payload_builder.cpp @@ -0,0 +1,208 @@ +#include "locator/payload_builder.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace radar::locator { +namespace { + +using Json = nlohmann::json; + +constexpr const char* kGprPointsName = "gpr_points"; +constexpr const char* kGprRegionCentersName = "gpr_region_centers"; +constexpr std::uint32_t kMinTableColumns = 3U; // [x_m, z_m, score, ...] + +// Locate the first TableF32 payload matching one of the GPR object names. +// Returns nullptr if no usable payload is present in the collection. +[[nodiscard]] auto find_object_table(const ipc::ResultCollection& collection) + -> const ipc::ResultPayload* { + const ipc::ResultPayload* fallback = nullptr; + for (const auto& payload : collection.collection_payloads) { + if (payload.kind != ipc::ResultKind::TableF32) { + continue; + } + if (payload.table_columns < kMinTableColumns) { + continue; + } + if (payload.processing_name == kGprPointsName) { + return &payload; + } + if (fallback == nullptr && payload.processing_name == kGprRegionCentersName) { + fallback = &payload; + } + } + return fallback; +} + +// Quantise to two decimal places. Equivalent to Python's `round(value, 2)` +// but explicit so behaviour does not silently depend on the local C library. +[[nodiscard]] auto quantise_to_centimetres(float value) -> float { + return std::round(value * 100.0F) / 100.0F; +} + +[[nodiscard]] auto passes_basic_filter( + float x_m, + float z_m, + float score, + float min_score +) -> bool { + if (!std::isfinite(x_m) || !std::isfinite(z_m) || !std::isfinite(score)) { + return false; + } + return score >= min_score; +} + +[[nodiscard]] auto passes_visible_bounds( + float x_m, + float z_m, + const VisibleBounds& bounds +) -> bool { + return x_m >= bounds.x_min + && x_m <= bounds.x_max + && z_m >= bounds.z_min + && z_m <= bounds.z_max; +} + +// Render the current wall-clock time as "HH:MM:SS.mmm". Uses localtime to match +// the Python reference behaviour (which calls datetime.now() with no tz info). +[[nodiscard]] auto format_timestamp_now() -> std::string { + using Clock = std::chrono::system_clock; + const auto now = Clock::now(); + const auto seconds = std::chrono::time_point_cast(now); + const auto millis = std::chrono::duration_cast( + now - seconds + ).count(); + + const std::time_t epoch_seconds = Clock::to_time_t(seconds); + std::tm broken_down{}; +#if defined(_WIN32) + localtime_s(&broken_down, &epoch_seconds); +#else + localtime_r(&epoch_seconds, &broken_down); +#endif + + std::ostringstream stream; + stream << std::put_time(&broken_down, "%H:%M:%S") + << '.' << std::setw(3) << std::setfill('0') << millis; + return stream.str(); +} + +void append_u32_little_endian(std::vector& buffer, std::uint32_t value) { + buffer.push_back(static_cast(value & 0xFFU)); + buffer.push_back(static_cast((value >> 8U) & 0xFFU)); + buffer.push_back(static_cast((value >> 16U) & 0xFFU)); + buffer.push_back(static_cast((value >> 24U) & 0xFFU)); +} + +} // namespace + +auto observations_from_collection( + const ipc::ResultCollection& collection, + const FilterParams& filter +) -> std::vector { + const auto* payload = find_object_table(collection); + if (payload == nullptr) { + return {}; + } + + const std::uint32_t columns = payload->table_columns; + if (columns == 0U) { + return {}; + } + const std::size_t row_count = payload->table_values.size() / columns; + if (row_count == 0U) { + return {}; + } + + std::vector visible; + visible.reserve(row_count); + for (std::size_t row = 0; row < row_count; ++row) { + const std::size_t base = row * static_cast(columns); + const float x_m = payload->table_values[base + 0U]; + const float z_m = payload->table_values[base + 1U]; + const float score = payload->table_values[base + 2U]; + + if (!passes_basic_filter(x_m, z_m, score, filter.min_score)) { + continue; + } + if (filter.visible_bounds.has_value() + && !passes_visible_bounds(x_m, z_m, *filter.visible_bounds)) { + continue; + } + visible.push_back({.crs = x_m, .dst = z_m}); + } + + if (!filter.draw_limits.has_value()) { + std::vector quantised; + quantised.reserve(visible.size()); + for (const auto& observation : visible) { + quantised.push_back({ + .crs = quantise_to_centimetres(observation.crs), + .dst = quantise_to_centimetres(observation.dst), + }); + } + return quantised; + } + + const auto& limits = *filter.draw_limits; + if (visible.size() > limits.max_detected_objects) { + return {}; + } + const std::size_t kept = std::min(visible.size(), limits.draw_top_objects); + std::vector result; + result.reserve(kept); + for (std::size_t index = 0; index < kept; ++index) { + result.push_back({ + .crs = quantise_to_centimetres(visible[index].crs), + .dst = quantise_to_centimetres(visible[index].dst), + }); + } + return result; +} + +auto build_payload_json( + const std::vector& observations, + std::uint32_t protocol_version, + std::uint32_t status +) -> std::string { + Json obs_array = Json::array(); + for (const auto& observation : observations) { + obs_array.push_back({ + {"dst", observation.dst}, + {"crs", observation.crs}, + }); + } + + const Json root{ + {"ver", protocol_version}, + {"tim", format_timestamp_now()}, + {"sts", status}, + {"obs", std::move(obs_array)}, + }; + return root.dump(); +} + +auto encode_packet( + const std::string& payload_json, + std::uint32_t device_id +) -> std::vector { + const auto payload_size = static_cast(payload_json.size()); + std::vector packet; + packet.reserve(static_cast(8U) + payload_json.size()); + + append_u32_little_endian(packet, device_id); + append_u32_little_endian(packet, payload_size); + const auto* bytes = reinterpret_cast(payload_json.data()); + packet.insert(packet.end(), bytes, bytes + payload_json.size()); + return packet; +} + +} // namespace radar::locator diff --git a/data_acq_and_processing/processing/locator/src/tcp_server.cpp b/data_acq_and_processing/processing/locator/src/tcp_server.cpp new file mode 100644 index 0000000..e8f119c --- /dev/null +++ b/data_acq_and_processing/processing/locator/src/tcp_server.cpp @@ -0,0 +1,496 @@ +#include "locator/tcp_server.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace radar::locator { +namespace { + +using Json = nlohmann::json; + +constexpr std::size_t kPacketHeaderSize = 8U; // device_id u32 LE + payload_len u32 LE. + +// Best-effort full-write helper: loops over write() until everything is sent +// or an error occurs. Returns false on socket error or peer disconnect. +[[nodiscard]] auto write_all(int socket_fd, const std::uint8_t* data, std::size_t size) -> bool { + std::size_t written = 0; + while (written < size) { + const auto chunk = ::send( + socket_fd, + data + written, + size - written, + MSG_NOSIGNAL + ); + if (chunk < 0) { + if (errno == EINTR) { + continue; + } + return false; + } + if (chunk == 0) { + return false; + } + written += static_cast(chunk); + } + return true; +} + +// Best-effort exact-read helper. Returns false if the peer closed the socket +// or an unrecoverable error occurred before all bytes were read. +[[nodiscard]] auto read_exact(int socket_fd, std::uint8_t* data, std::size_t size) -> bool { + std::size_t consumed = 0; + while (consumed < size) { + const auto chunk = ::recv(socket_fd, data + consumed, size - consumed, 0); + if (chunk < 0) { + if (errno == EINTR) { + continue; + } + return false; + } + if (chunk == 0) { + return false; + } + consumed += static_cast(chunk); + } + return true; +} + +[[nodiscard]] auto decode_u32_little_endian(const std::uint8_t* bytes) -> std::uint32_t { + return static_cast(bytes[0]) + | (static_cast(bytes[1]) << 8U) + | (static_cast(bytes[2]) << 16U) + | (static_cast(bytes[3]) << 24U); +} + +[[nodiscard]] auto format_peer(const sockaddr_storage& addr) -> std::string { + std::array host_buffer{}; + std::array port_buffer{}; + const auto err = ::getnameinfo( + reinterpret_cast(&addr), + sizeof(addr), + host_buffer.data(), + host_buffer.size(), + port_buffer.data(), + port_buffer.size(), + NI_NUMERICHOST | NI_NUMERICSERV + ); + if (err != 0) { + return "unknown"; + } + return std::string(host_buffer.data()) + ':' + port_buffer.data(); +} + +void apply_socket_keepalive(int socket_fd) { + int yes = 1; + (void)::setsockopt(socket_fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)); + (void)::setsockopt(socket_fd, SOL_SOCKET, SO_KEEPALIVE, &yes, sizeof(yes)); +} + +void shutdown_and_close(int& socket_fd) { + if (socket_fd < 0) { + return; + } + (void)::shutdown(socket_fd, SHUT_RDWR); + (void)::close(socket_fd); + socket_fd = -1; +} + +void log_warning(const std::string& message) { + std::cerr << "locator: " << message << '\n'; +} + +} // namespace + +// ----- ClientQueue ---------------------------------------------------------- + +ClientQueue::ClientQueue(std::size_t capacity) : capacity_(std::max(1U, capacity)) {} + +auto ClientQueue::try_push(std::vector packet) -> bool { + { + std::lock_guard guard(mutex_); + if (closed_ || queue_.size() >= capacity_) { + return false; + } + queue_.push_back(std::move(packet)); + } + not_empty_.notify_one(); + return true; +} + +auto ClientQueue::wait_pop() -> std::optional> { + std::unique_lock guard(mutex_); + not_empty_.wait(guard, [this]() { return closed_ || !queue_.empty(); }); + if (queue_.empty()) { + return std::nullopt; + } + auto packet = std::move(queue_.front()); + queue_.pop_front(); + return packet; +} + +void ClientQueue::close() { + { + std::lock_guard guard(mutex_); + if (closed_) { + return; + } + closed_ = true; + } + not_empty_.notify_all(); +} + +auto ClientQueue::is_closed() const -> bool { + std::lock_guard guard(mutex_); + return closed_; +} + +// ----- ClientSession -------------------------------------------------------- + +ClientSession::ClientSession( + int socket_fd, + std::string peer_name, + std::size_t queue_capacity, + std::uint32_t max_payload_bytes +) + : socket_fd_(socket_fd), + peer_name_(std::move(peer_name)), + queue_(queue_capacity), + max_payload_bytes_(max_payload_bytes) {} + +ClientSession::~ClientSession() { + request_stop(); + join(); + shutdown_and_close(socket_fd_); +} + +void ClientSession::start(std::atomic& shared_vlc_slot) { + writer_thread_ = std::thread([this]() { writer_loop(); }); + reader_thread_ = std::thread([this, &shared_vlc_slot]() { reader_loop(shared_vlc_slot); }); +} + +void ClientSession::enqueue(std::vector packet) { + if (stop_requested_.load(std::memory_order_acquire)) { + return; + } + if (!queue_.try_push(std::move(packet))) { + log_warning("disconnecting client " + peer_name_ + " after outbound queue overflow"); + request_stop(); + } +} + +void ClientSession::request_stop() { + if (stop_requested_.exchange(true, std::memory_order_acq_rel)) { + return; + } + queue_.close(); + // Wake any blocking recv() in the reader thread. + if (socket_fd_ >= 0) { + (void)::shutdown(socket_fd_, SHUT_RDWR); + } +} + +void ClientSession::join() { + if (writer_thread_.joinable()) { + writer_thread_.join(); + } + if (reader_thread_.joinable()) { + reader_thread_.join(); + } +} + +auto ClientSession::has_exited() const -> bool { + return exited_.load(std::memory_order_acquire); +} + +auto ClientSession::peer_name() const -> const std::string& { + return peer_name_; +} + +void ClientSession::writer_loop() { + while (!stop_requested_.load(std::memory_order_acquire)) { + auto packet = queue_.wait_pop(); + if (!packet.has_value()) { + break; + } + if (!write_all(socket_fd_, packet->data(), packet->size())) { + break; + } + } + request_stop(); + // Exited flag is set once both threads finish; reader_loop sets it. +} + +void ClientSession::reader_loop(std::atomic& shared_vlc_slot) { + std::array header_buffer{}; + std::vector payload_buffer; + + while (!stop_requested_.load(std::memory_order_acquire)) { + if (!read_exact(socket_fd_, header_buffer.data(), header_buffer.size())) { + break; + } + const auto payload_len = decode_u32_little_endian(header_buffer.data() + 4U); + if (payload_len > max_payload_bytes_) { + log_warning( + "closing client " + peer_name_ + " after payload size " + + std::to_string(payload_len) + " exceeded the configured limit" + ); + break; + } + + payload_buffer.assign(payload_len, std::uint8_t{0}); + if (payload_len > 0U && !read_exact(socket_fd_, payload_buffer.data(), payload_len)) { + break; + } + + try { + const auto json = Json::parse(payload_buffer.begin(), payload_buffer.end()); + if (json.is_object()) { + const auto found = json.find("vlc"); + if (found != json.end() && found->is_number()) { + const double value = found->get(); + if (std::isfinite(value)) { + shared_vlc_slot.store(value, std::memory_order_release); + } + } + } + } catch (const Json::parse_error& error) { + log_warning( + "ignoring malformed packet from " + peer_name_ + ": " + error.what() + ); + } + } + request_stop(); + exited_.store(true, std::memory_order_release); +} + +// ----- TcpServer ------------------------------------------------------------ + +TcpServer::TcpServer(LocatorServerConfig config) : config_(std::move(config)) { + latest_socket_speed_.store( + std::numeric_limits::quiet_NaN(), + std::memory_order_relaxed + ); +} + +TcpServer::~TcpServer() { + stop(); +} + +void TcpServer::start() { + if (running_.exchange(true, std::memory_order_acq_rel)) { + return; + } + + addrinfo hints{}; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + hints.ai_flags = AI_PASSIVE; + + addrinfo* resolved = nullptr; + const auto port_str = std::to_string(config_.port); + const auto gai = ::getaddrinfo( + config_.host.c_str(), + port_str.c_str(), + &hints, + &resolved + ); + if (gai != 0 || resolved == nullptr) { + running_.store(false, std::memory_order_release); + throw std::runtime_error( + "locator: getaddrinfo failed for " + config_.host + ":" + port_str + + " (" + ::gai_strerror(gai) + ")" + ); + } + + int fd = -1; + for (addrinfo* candidate = resolved; candidate != nullptr; candidate = candidate->ai_next) { + fd = ::socket(candidate->ai_family, candidate->ai_socktype, candidate->ai_protocol); + if (fd < 0) { + continue; + } + int yes = 1; + (void)::setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)); + if (::bind(fd, candidate->ai_addr, candidate->ai_addrlen) == 0) { + break; + } + ::close(fd); + fd = -1; + } + ::freeaddrinfo(resolved); + + if (fd < 0) { + running_.store(false, std::memory_order_release); + throw std::runtime_error( + "locator: failed to bind " + config_.host + ":" + port_str + + " (" + std::string(std::strerror(errno)) + ")" + ); + } + + if (::listen(fd, 16) < 0) { + ::close(fd); + running_.store(false, std::memory_order_release); + throw std::runtime_error( + "locator: listen() failed (" + std::string(std::strerror(errno)) + ")" + ); + } + + listen_fd_ = fd; + acceptor_thread_ = std::thread([this]() { acceptor_loop(); }); +} + +void TcpServer::stop() { + if (!running_.exchange(false, std::memory_order_acq_rel)) { + return; + } + + if (listen_fd_ >= 0) { + (void)::shutdown(listen_fd_, SHUT_RDWR); + (void)::close(listen_fd_); + listen_fd_ = -1; + } + + if (acceptor_thread_.joinable()) { + acceptor_thread_.join(); + } + + std::vector> sessions; + { + std::lock_guard guard(clients_mutex_); + sessions = std::move(clients_); + clients_.clear(); + } + for (auto& session : sessions) { + session->request_stop(); + } + for (auto& session : sessions) { + session->join(); + } +} + +auto TcpServer::is_running() const -> bool { + return running_.load(std::memory_order_acquire); +} + +void TcpServer::publish(const ipc::ResultCollection& collection, const FilterParams& filter) { + if (!running_.load(std::memory_order_acquire)) { + return; + } + + const auto observations = observations_from_collection(collection, filter); + const auto payload_json = build_payload_json(observations, config_.protocol_version); + auto packet = encode_packet(payload_json, config_.device_id); + cache_latest_packet(packet); + broadcast_packet(packet); +} + +auto TcpServer::latest_socket_speed() const -> std::optional { + const double value = latest_socket_speed_.load(std::memory_order_acquire); + if (std::isnan(value)) { + return std::nullopt; + } + return value; +} + +void TcpServer::acceptor_loop() { + while (running_.load(std::memory_order_acquire)) { + sockaddr_storage peer_addr{}; + socklen_t peer_len = sizeof(peer_addr); + const int client_fd = ::accept( + listen_fd_, + reinterpret_cast(&peer_addr), + &peer_len + ); + if (client_fd < 0) { + if (errno == EINTR) { + continue; + } + // Listening socket closed during shutdown produces EBADF/EINVAL; bail. + break; + } + + reap_finished_clients(); + + apply_socket_keepalive(client_fd); + auto session = std::make_unique( + client_fd, + format_peer(peer_addr), + config_.client_queue_size, + config_.max_payload_bytes + ); + const auto snapshot = latest_packet_copy(); + if (snapshot.has_value()) { + session->enqueue(*snapshot); + } + session->start(latest_socket_speed_); + enroll_client(std::move(session)); + } +} + +void TcpServer::enroll_client(std::unique_ptr session) { + std::lock_guard guard(clients_mutex_); + clients_.push_back(std::move(session)); +} + +void TcpServer::broadcast_packet(const std::vector& packet) { + std::lock_guard guard(clients_mutex_); + for (auto& client : clients_) { + client->enqueue(packet); + } +} + +void TcpServer::reap_finished_clients() { + std::vector> to_join; + { + std::lock_guard guard(clients_mutex_); + auto first_dead = std::partition( + clients_.begin(), + clients_.end(), + [](const std::unique_ptr& session) { + return !session->has_exited(); + } + ); + for (auto it = first_dead; it != clients_.end(); ++it) { + to_join.push_back(std::move(*it)); + } + clients_.erase(first_dead, clients_.end()); + } + for (auto& session : to_join) { + session->request_stop(); + session->join(); + } +} + +void TcpServer::cache_latest_packet(std::vector packet) { + std::lock_guard guard(latest_packet_mutex_); + latest_packet_ = std::move(packet); +} + +auto TcpServer::latest_packet_copy() const -> std::optional> { + std::lock_guard guard(latest_packet_mutex_); + return latest_packet_; +} + +} // namespace radar::locator diff --git a/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_lifecycle.cpp b/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_lifecycle.cpp index 7946cf2..4cae316 100644 --- a/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_lifecycle.cpp +++ b/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_lifecycle.cpp @@ -2,11 +2,15 @@ #include #include +#include #include +#include #include +#include #include #include #include +#include #include #include @@ -19,6 +23,60 @@ namespace { constexpr std::uint32_t kNativeAcquireMaxAttempts = 3U; constexpr auto kNativeSweepResponseTimeout = std::chrono::milliseconds(1500); +// One synthetic GPR reflector. `range_m` is its physical depth, `reflection` +// is the dimensionless complex reflection coefficient (|Γ| ≤ 1). +struct MockTarget { + float range_m; + float reflection_magnitude; +}; + +// Three reflectors at GPR-typical depths: a strong near-surface scatterer, +// a mid-depth target, and a weak deeper one. The magnitudes are tuned so the +// summed S21 stays within unit modulus across the sweep band. +constexpr std::array kMockTargets = {{ + {0.45F, 0.55F}, + {1.30F, 0.30F}, + {2.90F, 0.18F}, +}}; + +// Group velocity in moderately wet soil (≈ c / sqrt(εr), εr ≈ 4). The choice +// is what maps reflector depth to round-trip phase delay; it is held constant +// to keep the simulator deterministic. +constexpr float kGroundVelocityMps = 1.5e8F; + +// Soil attenuation grows with frequency. Calibrated so a target at 3 m sees +// roughly −20 dB extra loss at 6 GHz on top of geometric spreading. +constexpr float kAttenuationCoeffPerMeterAtRefHz = 0.22F; +constexpr float kAttenuationReferenceHz = 6e9F; + +// Antenna mismatch dominates S11: simulate one shallow reflection right at +// the connector, plus a small amount of cross-coupling from S21 targets. +constexpr float kS11ConnectorReflection = 0.55F; +constexpr float kS11ConnectorRangeM = 0.02F; +constexpr float kS11CrossCouplingFactor = 0.06F; + +// Noise floor in linear voltage units. Real LibreVNA hits ~ −90 dB at 1 kHz +// IFBW; pick something a touch noisier so the GPR processor has to work. +constexpr float kNoiseAmplitudeLinear = 0.004F; + +// Minimum simulated dwell so 0-point or pathological configs don't busy-loop. +constexpr auto kMockMinimumSweepDuration = std::chrono::microseconds(50); + +// Compute the dwell time the mock pretends to spend on the device. Mirrors +// the real LibreVNA contract: per-point dwell ≈ 1 / IFBW. Capped so absurd +// configs (IFBW ≈ 0 or huge sweeps) cannot freeze the producer for hours. +[[nodiscard]] auto mock_target_sweep_duration(const config::RadarSweepSettings& sweep) + -> std::chrono::nanoseconds { + const float points = std::max(1.0F, static_cast(sweep.points)); + const float if_bw = std::max(1.0F, sweep.if_bandwidth_hz); + const double seconds = static_cast(points) / static_cast(if_bw); + const auto duration_ns = std::chrono::nanoseconds( + static_cast(seconds * 1e9) + ); + constexpr auto kHardCap = std::chrono::seconds(5); + return std::clamp(duration_ns, kMockMinimumSweepDuration, kHardCap); +} + [[nodiscard]] auto is_retryable_native_acquire_error(std::string_view message) -> bool { constexpr std::array kRetryableSubstrings = { "Timeout waiting for expected LibreVNA packet type", @@ -109,34 +167,82 @@ auto LibreVnaMinimalDriver::acquire_sweep() -> SweepTrace { } auto LibreVnaMinimalDriver::acquire_mock() -> SweepTrace { + // Synthesise a frequency-domain VNA response containing several discrete + // reflectors at known depths, plus light Gaussian noise. The sweep dwells + // for a realistic duration derived from the configured IF bandwidth so + // downstream stages cannot be flooded faster than a real device would + // produce data. + const auto started_at = std::chrono::steady_clock::now(); + const auto target_duration = mock_target_sweep_duration(settings_.sweep); + SweepTrace trace{}; trace.frequency_hz.reserve(settings_.sweep.points); trace.s11.reserve(settings_.sweep.points); trace.s21.reserve(settings_.sweep.points); - const auto span_hz = settings_.sweep.stop_hz - settings_.sweep.start_hz; - const auto denominator = settings_.sweep.points > 1U ? static_cast(settings_.sweep.points - 1U) : 1.0F; + const float span_hz = settings_.sweep.stop_hz - settings_.sweep.start_hz; + const float denominator = + settings_.sweep.points > 1U ? static_cast(settings_.sweep.points - 1U) : 1.0F; + + // A tiny per-sweep range drift gives the rendered B-scan a visible motion + // signature so the simulator does not look frozen. + const float range_drift_m = + 0.01F * std::sin(0.07F * static_cast(sweep_index_)); + + // Deterministic-per-sweep noise so two consecutive frames look distinct + // but the test stays reproducible for any given sweep index. + std::mt19937 noise_engine( + static_cast(0x9E3779B9ULL ^ sweep_index_) + ); + std::normal_distribution noise_dist(0.0F, kNoiseAmplitudeLinear); for (std::uint32_t point = 0; point < settings_.sweep.points; ++point) { - const auto ratio = static_cast(point) / denominator; - const auto frequency_hz = settings_.sweep.start_hz + span_hz * ratio; - const auto phase = 2.0F * detail::kPi * (frequency_hz / std::max(settings_.mock_signal_hz, 1.0F)) + - static_cast(sweep_index_) * 0.05F; - const auto envelope = 0.6F + 0.4F * std::sin(0.5F * phase); + const float ratio = static_cast(point) / denominator; + const float frequency_hz = settings_.sweep.start_hz + span_hz * ratio; + const float frequency_scale = frequency_hz / kAttenuationReferenceHz; - ipc::Complex32 sample{}; - sample.re = envelope * std::cos(phase); - sample.im = envelope * std::sin(phase); + std::complex s21_total{0.0F, 0.0F}; + std::complex s11_total{0.0F, 0.0F}; - const auto reflection_phase = 0.7F * phase + 0.35F; - const auto reflection_envelope = 0.15F + 0.1F * std::cos(0.25F * phase); - ipc::Complex32 reflection{}; - reflection.re = reflection_envelope * std::cos(reflection_phase); - reflection.im = reflection_envelope * std::sin(reflection_phase); + for (const auto& target : kMockTargets) { + const float range_m = target.range_m + range_drift_m; + // Round-trip phase: 2π·f·(2R/v). + const float round_trip_phase = + 2.0F * detail::kPi * frequency_hz * (2.0F * range_m / kGroundVelocityMps); + // Geometric spreading: 1/(1 + R) keeps near-zero ranges finite. + const float spreading = 1.0F / (1.0F + range_m); + // Frequency-dependent soil attenuation in linear amplitude. + const float attenuation = + std::exp(-kAttenuationCoeffPerMeterAtRefHz * range_m * frequency_scale); + + const std::complex contribution = std::polar( + target.reflection_magnitude * spreading * attenuation, + -round_trip_phase + ); + s21_total += contribution; + s11_total += kS11CrossCouplingFactor * contribution; + } + + // Antenna mismatch dominates the near-field S11 response. + const float antenna_phase = + 2.0F * detail::kPi * frequency_hz * (2.0F * kS11ConnectorRangeM / kGroundVelocityMps); + s11_total += std::polar(kS11ConnectorReflection, -antenna_phase); + + // Independent noise per channel; complex variance ≈ kNoiseAmplitudeLinear². + s21_total += std::complex(noise_dist(noise_engine), noise_dist(noise_engine)); + s11_total += std::complex(noise_dist(noise_engine), noise_dist(noise_engine)); trace.frequency_hz.push_back(frequency_hz); - trace.s11.push_back(reflection); - trace.s21.push_back(sample); + trace.s11.push_back({.re = s11_total.real(), .im = s11_total.imag()}); + trace.s21.push_back({.re = s21_total.real(), .im = s21_total.imag()}); + } + + // Honour the IFBW-derived dwell time. If generation already took longer + // than the simulated device would have needed (huge `points` × CPU jitter) + // we skip the sleep so the producer does not fall further behind. + const auto elapsed = std::chrono::steady_clock::now() - started_at; + if (elapsed < target_duration) { + std::this_thread::sleep_for(target_duration - elapsed); } return trace; diff --git a/python_app/gui/app_window.py b/python_app/gui/app_window.py index 9545035..8e436f7 100644 --- a/python_app/gui/app_window.py +++ b/python_app/gui/app_window.py @@ -27,11 +27,10 @@ from python_app.gui.controllers.app_window_ui_mixin import AppWindowUiMixin from python_app.gui.preprocess_dialog import PreprocessDialog from python_app.models.dataset_model import ResultCollection, SweepCollection from python_app.models.gui_profile_model import GuiProfileModel -from python_app.models.run_config_model import RunConfigModel from python_app.orchestration.config_writer import ConfigWriter from python_app.orchestration.gui_session_state import GuiSessionState, GuiSessionStateStore from python_app.orchestration.live_processing_config import ProcessingLiveConfigWriter -from python_app.orchestration.locator_runtime import LocatorTcpService +from python_app.orchestration.pipeline_metrics import PipelineMetrics from python_app.orchestration.preprocess_assets import VISIBLE_PREPROCESS_ASSET_KEYS, preprocess_asset_model from python_app.orchestration.process_supervisor import ProcessSupervisor from python_app.orchestration.shm_reader import ShmRingReader @@ -83,7 +82,22 @@ class AppWindow( self._supervisor = ProcessSupervisor(self._project_root) self._live_config_writer = ProcessingLiveConfigWriter(runtime_dir / "processing_live.json") self._gui_session_state_store = GuiSessionStateStore(runtime_dir / "gui_session_state.json") - self._locator_service: LocatorTcpService | None = None + # `log_sink` is attached after the runtime log widget exists. + self._pipeline_metrics = PipelineMetrics( + report_every=self._resolve_metrics_report_every() + ) + + @staticmethod + def _resolve_metrics_report_every() -> int: + """Read the metrics flush threshold from env, falling back to 50.""" + raw = os.environ.get("RADAR_SYSTEM_METRICS_REPORT_EVERY", "").strip() + if not raw: + return 50 + try: + value = int(raw) + except ValueError: + return 50 + return value if value >= 1 else 50 def _init_config_profile_state(self) -> None: """Resolve startup profile path, load active profile, and queue fallback notices.""" @@ -112,7 +126,6 @@ class AppWindow( "INFO", f"Loaded legacy run config without GUI defaults: {active_profile_path}", ) - self._locator_service = self._build_locator_service(self._defaults_config) self._remember_active_profile_path(active_profile_path, startup=True) def _init_reader_handles(self) -> None: @@ -217,60 +230,13 @@ class AppWindow( self._log(f"Active config profile: {self._active_profile_path}") self._refresh_preprocess_summary_labels() self._apply_initial_radar_limits() - self._start_locator_service() self._on_processing_mode_changed(self._processing_mode.currentText()) self._write_live_processing_config() + # Defer the log sink wiring until the runtime log widget exists. + self._pipeline_metrics.set_log_sink(self._log) self._timer.start() self._maybe_auto_start_pipeline() - def _start_locator_service(self) -> None: - """Start embedded locator TCP service without failing the GUI.""" - try: - if self._locator_service is None: - self._locator_service = self._build_locator_service(self._defaults_config) - self._locator_service.start() - self._locator_service.publish_empty() - self._log( - f"Locator TCP server listening on " - f"{self._locator_service.host}:{self._locator_service.port}" - ) - except Exception as exc: # noqa: BLE001 - self._log_exception("Failed to start locator TCP server", exc, level="WARN") - - def _build_locator_service(self, config: RunConfigModel) -> LocatorTcpService: - """Create locator service instance from stable run config.""" - locator_server = config.runtime.locator_server - return LocatorTcpService( - host=str(locator_server.host), - port=int(locator_server.port), - device_id=int(locator_server.device_id), - protocol_version=int(locator_server.protocol_version), - max_payload_bytes=int(locator_server.max_payload_bytes), - client_queue_size=int(locator_server.client_queue_size), - logger_name=str(locator_server.logger_name), - ) - - def _reload_locator_service_from_config(self) -> None: - """Rebuild locator service using current stable config and restart if needed.""" - previous_service = self._locator_service - was_running = previous_service is not None and previous_service.is_running() - if previous_service is not None: - previous_service.stop() - - self._locator_service = self._build_locator_service(self._defaults_config) - if not was_running: - return - - try: - self._locator_service.start() - self._locator_service.publish_empty() - self._log( - "Locator TCP server reloaded from config: " - f"{self._locator_service.host}:{self._locator_service.port}" - ) - except Exception as exc: # noqa: BLE001 - self._log_exception("Failed to reload locator TCP server from config", exc, level="WARN") - def _resolve_startup_profile_path(self) -> Path: """Resolve active profile path from session-state or root fallback path.""" env_profile_path = os.environ.get("RADAR_SYSTEM_PROFILE", "").strip() @@ -300,12 +266,36 @@ class AppWindow( return profile_path def _maybe_auto_start_pipeline(self) -> None: - """Schedule pipeline start when requested by launcher environment.""" - auto_start = os.environ.get("RADAR_SYSTEM_AUTO_START", "").strip().lower() - if auto_start not in {"1", "true", "yes", "on"}: + """Schedule pipeline start when requested by launcher environment. + + With `RADAR_SYSTEM_AUTO_APPLY_RADAR=1` the launcher also reproduces the + "Apply Radar" click before "Start". This is the headless deployment + recipe: the GUI configures the device exactly as a human operator + would, then starts the capture pipeline. + """ + auto_start = self._is_truthy_env("RADAR_SYSTEM_AUTO_START") + if not auto_start: return self._log("Auto-start requested by launcher.") - QTimer.singleShot(500, self._start_run) + if self._is_truthy_env("RADAR_SYSTEM_AUTO_APPLY_RADAR"): + QTimer.singleShot(500, self._auto_apply_radar_then_start) + else: + QTimer.singleShot(500, self._start_run) + + def _auto_apply_radar_then_start(self) -> None: + """Apply current radar settings then start the pipeline (headless boot).""" + try: + self._apply_radar_settings() + except Exception as exc: # noqa: BLE001 + self._log_exception("Auto apply-radar failed", exc, level="WARN") + # Hand control back to the event loop so widget updates from + # _apply_radar_settings can flush before _start_run takes over. + QTimer.singleShot(100, self._start_run) + + @staticmethod + def _is_truthy_env(name: str) -> bool: + """Return True when an environment variable is set to a truthy literal.""" + return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"} def _normalize_profile_path(self, path: Path) -> Path: """Return normalized absolute profile path.""" @@ -494,6 +484,8 @@ class AppWindow( def _show_error(self, message: str, *, details: str | None = None) -> None: """Log and present an error in a modal dialog with optional detail text.""" self._log_error(message, details=details) + if self._is_truthy_env("RADAR_SYSTEM_HEADLESS"): + return dialog = QMessageBox(self) dialog.setIcon(QMessageBox.Icon.Critical) dialog.setWindowTitle("Error") @@ -505,6 +497,8 @@ class AppWindow( def _show_exception(self, context: str, exc: Exception) -> None: """Log full exception details and show modal dialog with expandable traceback.""" message, details = self._log_exception(context, exc, level="ERROR") + if self._is_truthy_env("RADAR_SYSTEM_HEADLESS"): + return dialog = QMessageBox(self) dialog.setIcon(QMessageBox.Icon.Critical) dialog.setWindowTitle("Error") @@ -520,9 +514,6 @@ class AppWindow( self._abort_capture_sequence(resume_pipeline=False) # 2) Stop all managed processes/readers. self._stop_all_processes() - # 3) Stop embedded locator service. - if self._locator_service is not None: - self._locator_service.stop() # 3) Close auxiliary dialog windows. if self._preprocess_dialog is not None: self._preprocess_dialog.close() diff --git a/python_app/gui/controllers/app_window_config/live_processing_mixin.py b/python_app/gui/controllers/app_window_config/live_processing_mixin.py index 2346171..6651f80 100644 --- a/python_app/gui/controllers/app_window_config/live_processing_mixin.py +++ b/python_app/gui/controllers/app_window_config/live_processing_mixin.py @@ -89,6 +89,9 @@ class AppWindowLiveProcessingMixin: gpr_background_mean_count=gpr_background_mean_count, gpr_remove_sidelobe_objects_enabled=bool(self._gpr_remove_sidelobe_objects_enabled.isChecked()), gpr_imaging_plane_y_m=float(self._gpr_imaging_plane_y_m.value()), + gpr_min_visible_score=float(self._gpr_min_visible_score.value()), + legacy_gpr_min_visible_pair_count=float(self._legacy_gpr_min_visible_pair_count.value()), + ignore_socket_speed=bool(self._legacy_gpr_ignore_socket_speed_enabled.isChecked()), reprocess_current_result=bool(reprocess_current_result), history_command_seq=int(self._history_command_seq), history_command=str(history_command), @@ -140,6 +143,14 @@ class AppWindowLiveProcessingMixin: except Exception as exc: # noqa: BLE001 self._show_exception("Failed to update live processing settings", exc) + def _on_legacy_gpr_ignore_socket_speed_toggled(self, ignore_socket_speed: bool) -> None: + """Reflect socket/manual speed authority in the GUI and live config.""" + try: + self._legacy_gpr_speed_m_s.setEnabled(bool(ignore_socket_speed)) + self._write_live_processing_config() + except Exception as exc: # noqa: BLE001 + self._show_exception("Failed to update socket-speed mode", exc) + def _on_gpr_visual_settings_changed(self, *_args) -> None: """Redraw current GPR result using updated GUI-only render settings.""" if not self._is_gpr_processing_mode(self._processing_mode.currentText()): @@ -152,22 +163,22 @@ class AppWindowLiveProcessingMixin: self._show_exception("Failed to update GPR render settings", exc) def _on_gpr_locator_threshold_changed(self, *_args) -> None: - """Redraw GPR view and republish locator snapshot after threshold changes.""" + """Redraw GPR view; locator delivery lives in the C++ data_processor.""" self._on_gpr_visual_settings_changed() if not self._is_gpr_processing_mode(self._processing_mode.currentText()): return try: - self._publish_locator_snapshot_from_latest_result() + self._write_live_processing_config() except Exception as exc: # noqa: BLE001 self._show_exception("Failed to update locator GPR threshold", exc) def _on_gpr_locator_window_changed(self, *_args) -> None: - """Redraw GPR view and republish locator snapshot after visible X/Z changes.""" + """Redraw GPR view; locator delivery lives in the C++ data_processor.""" self._on_gpr_visual_settings_changed() if not self._is_gpr_processing_mode(self._processing_mode.currentText()): return try: - self._publish_locator_snapshot_from_latest_result() + self._write_live_processing_config() except Exception as exc: # noqa: BLE001 self._show_exception("Failed to update locator GPR window", exc) @@ -205,10 +216,6 @@ class AppWindowLiveProcessingMixin: self._set_plot_mode(mode) self._set_processing_mode_page(mode) self._on_processing_live_settings_changed() - if self._is_gpr_processing_mode(mode): - self._publish_locator_snapshot_from_latest_result() - elif self._is_gpr_processing_mode(previous_mode) and self._locator_service is not None: - self._locator_service.publish_empty() if mode == "pass_through": self._log( "Processing mode selected: pass_through " diff --git a/python_app/gui/controllers/app_window_config/profile_io_mixin.py b/python_app/gui/controllers/app_window_config/profile_io_mixin.py index 15b92f5..afb5b04 100644 --- a/python_app/gui/controllers/app_window_config/profile_io_mixin.py +++ b/python_app/gui/controllers/app_window_config/profile_io_mixin.py @@ -445,9 +445,11 @@ class AppWindowConfigProfileIOMixin: self._legacy_gpr_start_freq_mhz.setValue(float(gui_state.processing.legacy_gpr.start_freq_mhz)) self._legacy_gpr_stop_freq_mhz.setValue(float(gui_state.processing.legacy_gpr.stop_freq_mhz)) self._legacy_gpr_speed_m_s.setValue(float(gui_state.processing.legacy_gpr.speed_m_s)) - self._legacy_gpr_ignore_socket_speed_enabled.setChecked( - bool(gui_state.processing.legacy_gpr.ignore_socket_speed_enabled) + ignore_socket_speed_enabled = bool( + gui_state.processing.legacy_gpr.ignore_socket_speed_enabled ) + self._legacy_gpr_ignore_socket_speed_enabled.setChecked(ignore_socket_speed_enabled) + self._legacy_gpr_speed_m_s.setEnabled(ignore_socket_speed_enabled) self._legacy_gpr_look_angle_deg.setValue(float(gui_state.processing.legacy_gpr.look_angle_deg)) self._legacy_gpr_background_subtract_enabled.setChecked( bool(gui_state.processing.legacy_gpr.background_subtract_enabled) @@ -494,7 +496,6 @@ class AppWindowConfigProfileIOMixin: self._apply_initial_radar_limits() if self._preprocess_dialog is not None: self._refresh_sets() - self._reload_locator_service_from_config() self._on_processing_mode_changed(gui_state.processing.selected_mode) self._update_history_indicator() self._remember_active_profile_path(profile_path) diff --git a/python_app/gui/controllers/app_window_config/radar_limits_mixin.py b/python_app/gui/controllers/app_window_config/radar_limits_mixin.py index 1e17586..19f18db 100644 --- a/python_app/gui/controllers/app_window_config/radar_limits_mixin.py +++ b/python_app/gui/controllers/app_window_config/radar_limits_mixin.py @@ -3,6 +3,7 @@ from __future__ import annotations from python_app.hardware_full.librevna_service import LibreVnaService +from python_app.hardware_full.matrix_radar_service import create_matrix_radar_service from python_app.hardware_full.single_radar_service import create_single_radar_service @@ -22,6 +23,8 @@ class AppWindowRadarLimitsMixin: return self._apply_radar_limits_to_ui(None) if config.is_multi_device: radar_service = LibreVnaService(serial=config.radar.serial or None) + elif config.is_matrix_radar: + radar_service = create_matrix_radar_service(config) else: radar_service = create_single_radar_service(config) diff --git a/python_app/gui/controllers/app_window_config/state_builders.py b/python_app/gui/controllers/app_window_config/state_builders.py index 09ecac3..259256f 100644 --- a/python_app/gui/controllers/app_window_config/state_builders.py +++ b/python_app/gui/controllers/app_window_config/state_builders.py @@ -399,8 +399,8 @@ class AppWindowConfigStateBuildersMixin: config.runtime.settling_ms = int(self._settling_ms.text().strip()) config.runtime.processing_live_config_path = str(self._live_config_writer.path) - if config.is_multi_device: - if len(config.radar.multi_device.slave_serials) != 2: + if config.is_matrix_radar: + if config.is_multi_device and len(config.radar.multi_device.slave_serials) != 2: raise ValueError("LibreVNA multi-device mode requires exactly two slave serials") config.apply_device_model_constraints() else: diff --git a/python_app/gui/controllers/app_window_pipeline_mixin.py b/python_app/gui/controllers/app_window_pipeline_mixin.py index 15aa44d..18622d9 100644 --- a/python_app/gui/controllers/app_window_pipeline_mixin.py +++ b/python_app/gui/controllers/app_window_pipeline_mixin.py @@ -4,7 +4,6 @@ from __future__ import annotations import time -from PyQt6.QtCore import QSignalBlocker from python_app.gui.runtime.constraints import validate_processing_mode_constraints from python_app.gui.runtime.history import build_run_history_signature, record_result_history @@ -12,7 +11,6 @@ from python_app.hardware_full.kamil_adc_service import apply_kamil_adc_laser_con from python_app.hardware_full.single_radar_service import create_single_radar_service from python_app.models.dataset_model import ComboKey, ResultCollection, SweepCollection from python_app.models.run_config_model import RunConfigModel -from python_app.orchestration.gpr_locator import collection_has_gpr_payloads from python_app.orchestration.preprocess_assets import ( PREPROCESS_ASSET_SPECS, REQUIRED_PREPROCESS_ASSET_KEYS, @@ -42,8 +40,18 @@ class AppWindowPipelineMixin: ) return if self._supervisor.is_running(): - self._show_error("Pipeline is already running", details=self._process_state_details()) - return + # A single-shot capture from a running continuous pipeline must + # restart acquisition with `runtime.continuous=false`; refusing + # here would leave the previous run streaming and the user could + # never reach the single-capture termination state. + if not single_capture: + self._show_error( + "Pipeline is already running", + details=self._process_state_details(), + ) + return + self._log("Stopping continuous pipeline before single capture") + self._stop_run() try: processor_was_running = self._supervisor.is_processor_running() @@ -198,6 +206,9 @@ class AppWindowPipelineMixin: if config.is_multi_device: self._log("Multi-device raw producer will configure all LibreVNA devices") return + if config.is_matrix_radar: + self._log("Matrix raw producer will configure the matrix radar") + return if config.is_kamil_adc: if apply_kamil_adc_laser_control(config): self._log("Kamil ADC laser_control applied via Apply Radar") @@ -280,8 +291,6 @@ class AppWindowPipelineMixin: self._log_error(report.format()) try: - self._drain_locator_speed_updates() - self._drain_locator_log_updates() if self._raw_reader is not None: self._read_all_raw() self._read_all_preprocessed() @@ -294,7 +303,14 @@ class AppWindowPipelineMixin: return return - self._draw_preferred_collection(result_latest=result_latest) + if result_latest is not None: + render_started_ns = time.monotonic_ns() + self._draw_preferred_collection(result_latest=result_latest) + self._pipeline_metrics.record( + "rendering", time.monotonic_ns() - render_started_ns + ) + else: + self._draw_preferred_collection(result_latest=None) except Exception as exc: # noqa: BLE001 signature = (type(exc).__name__, str(exc)) if self._last_reader_error_signature == signature: @@ -347,6 +363,11 @@ class AppWindowPipelineMixin: break self._raw_history.append(collection) latest = collection + # `capture_*_ns` are populated by the C++ sweep_orchestrator with + # wallclocks captured around the actual device read. Pre-orchestrator + # producers leave them zero, in which case PipelineMetrics drops it. + acquisition_ns = int(collection.capture_end_ns) - int(collection.capture_start_ns) + self._pipeline_metrics.record("acquisition", acquisition_ns) if self._single_capture_active and self._single_capture_start_ns is not None: if collection.monotonic_ns >= self._single_capture_start_ns: self._single_capture_seen_raw = True @@ -373,13 +394,9 @@ class AppWindowPipelineMixin: collection = self._result_reader.pop_result_collection() if collection is None: break + self._pipeline_metrics.record("processing", int(collection.processing_duration_ns)) if record_result_history(self._result_history, collection): latest = collection - if ( - self._is_gpr_processing_mode(self._processing_mode.currentText()) - and collection_has_gpr_payloads(collection) - ): - self._publish_locator_snapshot_from_collection(collection) return latest def _drain_rings_once_for_history(self) -> None: @@ -484,60 +501,3 @@ class AppWindowPipelineMixin: self._live_processing_config(), ) - def _drain_locator_speed_updates(self) -> None: - """Apply the newest queued locator speed packet to live processing settings.""" - if self._locator_service is None: - return - speed_m_s = self._locator_service.drain_speed_updates() - if speed_m_s is None: - return - if self._processing_mode.currentText() != "legacy_gpr": - return - if self._legacy_gpr_ignore_socket_speed_enabled.isChecked(): - return - - previous_speed_m_s = float(self._legacy_gpr_speed_m_s.value()) - with QSignalBlocker(self._legacy_gpr_speed_m_s): - self._legacy_gpr_speed_m_s.setValue(float(speed_m_s)) - current_speed_m_s = float(self._legacy_gpr_speed_m_s.value()) - if current_speed_m_s == previous_speed_m_s: - return - - self._write_live_processing_config(reprocess_current_result=False) - - def _drain_locator_log_updates(self) -> None: - """Append queued locator socket traffic messages to the runtime log.""" - if self._locator_service is None: - return - for message in self._locator_service.drain_log_updates(): - self._log(message) - - def _publish_locator_snapshot_from_collection(self, collection: ResultCollection) -> None: - """Publish one locator snapshot from a GPR result collection.""" - if self._locator_service is None: - return - self._locator_service.publish_collection( - collection, - self._gpr_locator_threshold(), - visible_bounds=self._gpr_visible_object_bounds(), - object_draw_limits=self._gpr_draw_limits(), - ) - - def _publish_locator_snapshot_from_latest_result(self) -> None: - """Publish current locator-visible snapshot from latest cached GPR result.""" - if self._locator_service is None: - return - if not self._is_gpr_processing_mode(self._processing_mode.currentText()): - self._locator_service.publish_empty() - return - - if not self._result_history: - self._locator_service.publish_empty() - return - - latest = self._result_history[-1] - if not collection_has_gpr_payloads(latest): - self._locator_service.publish_empty() - return - - self._publish_locator_snapshot_from_collection(latest) diff --git a/python_app/gui/controllers/sections/processing_section.py b/python_app/gui/controllers/sections/processing_section.py index e054e57..bc859a8 100644 --- a/python_app/gui/controllers/sections/processing_section.py +++ b/python_app/gui/controllers/sections/processing_section.py @@ -390,6 +390,12 @@ def build_processing_group(owner) -> QGroupBox: owner._legacy_gpr_ignore_socket_speed_enabled.setChecked( bool(legacy_gpr_defaults.ignore_socket_speed_enabled) ) + # When the box is unchecked, an external TCP client controls the speed + # through the C++ locator server; disable the spinner so the GUI value + # cannot silently win against the live socket value. + owner._legacy_gpr_speed_m_s.setEnabled( + bool(legacy_gpr_defaults.ignore_socket_speed_enabled) + ) owner._legacy_gpr_look_angle_deg = QDoubleSpinBox() owner._legacy_gpr_look_angle_deg.setDecimals(2) @@ -516,6 +522,9 @@ def build_processing_group(owner) -> QGroupBox: owner._legacy_gpr_start_freq_mhz.valueChanged.connect(owner._on_processing_live_settings_changed) owner._legacy_gpr_stop_freq_mhz.valueChanged.connect(owner._on_processing_live_settings_changed) owner._legacy_gpr_speed_m_s.valueChanged.connect(owner._on_processing_live_settings_changed) + owner._legacy_gpr_ignore_socket_speed_enabled.toggled.connect( + owner._on_legacy_gpr_ignore_socket_speed_toggled + ) owner._legacy_gpr_look_angle_deg.valueChanged.connect(owner._on_processing_live_settings_changed) owner._legacy_gpr_background_subtract_enabled.toggled.connect(owner._on_processing_live_settings_changed) owner._legacy_gpr_background_mean_count.valueChanged.connect(owner._on_processing_live_settings_changed) diff --git a/python_app/gui/main.py b/python_app/gui/main.py index 8a20e78..1364525 100644 --- a/python_app/gui/main.py +++ b/python_app/gui/main.py @@ -2,10 +2,13 @@ from __future__ import annotations +import os from pathlib import Path +import signal import sys import pyqtgraph as pg +from PyQt6.QtCore import QTimer from PyQt6.QtWidgets import QApplication # Ensure imports are resolved when started as a script. @@ -17,6 +20,38 @@ from python_app.gui.app_window import AppWindow from python_app.gui.theme import apply_light_theme +def _is_headless() -> bool: + """Return whether the launcher requested a non-interactive deployment.""" + return os.environ.get("RADAR_SYSTEM_HEADLESS", "").strip().lower() in { + "1", + "true", + "yes", + "on", + } + + +def _install_unix_signal_handlers(app: QApplication, window: AppWindow) -> None: + """Route SIGINT and SIGTERM through the Qt event loop into a clean shutdown. + + `window.close()` runs `closeEvent`, which aborts any active capture and + shuts down managed C++ processes; only then does the Qt loop exit. A + short repeating timer keeps the Python interpreter pinned in the event + loop just long enough to deliver pending signals. + """ + + def _request_shutdown(*_args: object) -> None: + window.close() + app.quit() + + for sig in (signal.SIGINT, signal.SIGTERM): + signal.signal(sig, _request_shutdown) + + keepalive_timer = QTimer(app) + keepalive_timer.setInterval(200) + keepalive_timer.timeout.connect(lambda: None) + keepalive_timer.start() + + def main() -> int: """Run Qt event loop and show main radar control window.""" app = QApplication(sys.argv) @@ -24,7 +59,10 @@ def main() -> int: # PyQtGraph foreground controls axis lines, tick text, labels, and titles. pg.setConfigOptions(antialias=True, background="#ffffff", foreground="#ffffff") window = AppWindow(PROJECT_ROOT) - window.showMaximized() + if _is_headless(): + _install_unix_signal_handlers(app, window) + else: + window.showMaximized() return app.exec() diff --git a/python_app/hardware_full/matrix_radar_service.py b/python_app/hardware_full/matrix_radar_service.py index 87a1c96..755482a 100644 --- a/python_app/hardware_full/matrix_radar_service.py +++ b/python_app/hardware_full/matrix_radar_service.py @@ -59,9 +59,11 @@ def create_matrix_radar_service(config: RunConfigModel) -> MatrixRadarService: raise RuntimeError("SN9000 requires radar.driver_mode='native'") from python_app.hardware_full.sn9000_service import Sn9000Service + visa_library = config.radar.visa_library or "@ivi" return Sn9000Service( host=config.radar.remote_host, port=config.radar.remote_port, + visa_library=visa_library, ) raise RuntimeError(f"Unsupported matrix radar model: {model}") diff --git a/python_app/hardware_full/sn9000_service.py b/python_app/hardware_full/sn9000_service.py index b11e135..663e0ac 100644 --- a/python_app/hardware_full/sn9000_service.py +++ b/python_app/hardware_full/sn9000_service.py @@ -60,8 +60,6 @@ class Sn9000Service: if self.timeout_ms <= 0: raise ValueError("SN9000 timeout_ms must be > 0") self.visa_library = str(self.visa_library).strip() or "@ivi" - if self.visa_library == "@py" or self.visa_library.endswith("@py"): - raise ValueError("SN9000 requires an IVI/Vendor VISA backend, not pyvisa-py") @property def resource(self) -> str: @@ -146,6 +144,10 @@ class Sn9000Service: return { "min_frequency_hz": float(instrument.query("SERV:SWE:FREQ:MIN?")), "max_frequency_hz": float(instrument.query("SERV:SWE:FREQ:MAX?")), + # SN9000 SCPI does not expose IFBW capability queries; use the + # documented hardware sequence (1 Hz .. 300 kHz, manual p. 58, 1261). + "min_ifbw_hz": 1.0, + "max_ifbw_hz": 300_000.0, "max_points": int(float(instrument.query("SERV:SWE:POIN?"))), "min_power_dbm": float(instrument.query("SERV:SWE:POW:MIN?")), "max_power_dbm": float(instrument.query("SERV:SWE:POW:MAX?")), @@ -208,18 +210,37 @@ class Sn9000Service: def _query_sweep_s_parameters(self, points: int) -> dict[str, np.ndarray]: instrument = self._require_instrument() + if self._uses_pyvisa_py_backend(): + # pyvisa-py HiSLIP loses synchronization when a single packet aggregates + # *OPC? plus multiple binary blocks, so issue trigger and data queries + # one at a time. The corrected-data buffer holds the last completed + # sweep, so reading each S-parameter sequentially is safe. + instrument.write("TRIG:SING") + self._expect_opc("*OPC?", context="SN9000 sweep") + complex_values: dict[str, np.ndarray] = {} + for parameter_name in _S_PARAMETER_QUERY_ORDER: + instrument.write(f"SENS:DATA:CORR? {parameter_name}") + interleaved = self._read_float32_block( + f"SENS:DATA:CORR? {parameter_name}", points * 2 + ) + complex_values[parameter_name] = self._complex_from_interleaved(interleaved) + return complex_values + data_queries = ";".join(f":SENS:DATA:CORR? {name}" for name in _S_PARAMETER_QUERY_ORDER) instrument.write(f"TRIG:SING;*OPC?;{data_queries}") opc_token = self._read_ascii_token() if opc_token != "1": raise RuntimeError(f"SN9000 sweep returned unexpected *OPC? response: {opc_token!r}") - complex_values: dict[str, np.ndarray] = {} + complex_values = {} for parameter_name in _S_PARAMETER_QUERY_ORDER: interleaved = self._read_float32_block(f"SENS:DATA:CORR? {parameter_name}", points * 2) complex_values[parameter_name] = self._complex_from_interleaved(interleaved) return complex_values + def _uses_pyvisa_py_backend(self) -> bool: + return self.visa_library == "@py" or self.visa_library.endswith("@py") + def _assemble_traces(self, s_parameters: dict[str, np.ndarray]) -> list[TraceData]: frequency_hz = self._require_frequency_axis() traces: list[TraceData] = [] @@ -286,8 +307,26 @@ class Sn9000Service: f"SN9000 response for {context!r} returned {array.size} float32 values, " f"expected {expected_values}" ) + self._drain_trailing_terminators() return array + def _drain_trailing_terminators(self) -> None: + """Consume the SCPI terminator that follows IEEE binary blocks. + + SCPI responses end with `\\n`, which over HiSLIP closes the DataEnd + message group. pyvisa-py's HiSLIP layer needs the terminator drained + before the next request, otherwise it loses message-frame + synchronization on subsequent reads. + """ + instrument = self._require_instrument() + deadline = time.monotonic() + 0.2 + while time.monotonic() < deadline: + try: + instrument.read_bytes(1, break_on_termchar=True) + return + except Exception: + return + def _read_response_bytes(self, count: int) -> bytes: instrument = self._require_instrument() data = instrument.read_bytes(count, break_on_termchar=False) diff --git a/python_app/models/dataset_model.py b/python_app/models/dataset_model.py index bfe27d8..9154144 100644 --- a/python_app/models/dataset_model.py +++ b/python_app/models/dataset_model.py @@ -80,5 +80,8 @@ class ResultCollection: collection_id: int monotonic_ns: int + # Wall-clock nanoseconds spent by the data_processor on `process_collection` + # for this collection. Zero means the producer did not report a measurement. + processing_duration_ns: int = 0 collection_payloads: list[ResultPayload] = field(default_factory=list) blocks: list[ResultBlock] = field(default_factory=list) diff --git a/python_app/models/run_config_codec.py b/python_app/models/run_config_codec.py index 3008f1a..07b54fe 100644 --- a/python_app/models/run_config_codec.py +++ b/python_app/models/run_config_codec.py @@ -93,6 +93,7 @@ def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel: model.radar.remote_port = int(radar_payload.get("remote_port", model.radar.remote_port)) model.radar.driver_mode = str(radar_payload.get("driver_mode", model.radar.driver_mode)) model.radar.mock_signal_hz = float(radar_payload.get("mock_signal_hz", model.radar.mock_signal_hz)) + model.radar.visa_library = str(radar_payload.get("visa_library", model.radar.visa_library)) model.radar.sweep.start_hz = float(sweep_payload.get("start_hz", model.radar.sweep.start_hz)) model.radar.sweep.stop_hz = float(sweep_payload.get("stop_hz", model.radar.sweep.stop_hz)) @@ -393,6 +394,7 @@ def run_config_to_dict(model: RunConfigModel) -> dict[str, Any]: "remote_port": model.radar.remote_port, "driver_mode": model.radar.driver_mode, "mock_signal_hz": model.radar.mock_signal_hz, + "visa_library": model.radar.visa_library, "multi_device": { "slave_serials": list(model.radar.multi_device.slave_serials), "force_external_reference": model.radar.multi_device.force_external_reference, diff --git a/python_app/models/run_config_schema.py b/python_app/models/run_config_schema.py index 41467a4..49daa5c 100644 --- a/python_app/models/run_config_schema.py +++ b/python_app/models/run_config_schema.py @@ -103,6 +103,7 @@ class RadarModel: remote_port: int = 50209 driver_mode: str = "mock" mock_signal_hz: float = 1_000_000.0 + visa_library: str = "" sweep: RadarSweepModel = field(default_factory=RadarSweepModel) multi_device: RadarMultiDeviceModel = field(default_factory=RadarMultiDeviceModel) kamil_adc: KamilAdcModel = field(default_factory=KamilAdcModel) diff --git a/python_app/orchestration/gpr_locator.py b/python_app/orchestration/gpr_locator.py index ad8f4ea..86c75c5 100644 --- a/python_app/orchestration/gpr_locator.py +++ b/python_app/orchestration/gpr_locator.py @@ -1,10 +1,11 @@ -"""Helpers for extracting GPR objects and locator observations from results.""" +"""Helpers for extracting GPR objects from result collections. + +Locator TCP delivery now lives in the C++ data_processor. This module retains +only the inspection helpers that the GUI uses for plotting. +""" from __future__ import annotations -from datetime import datetime -from typing import Any - import numpy as np from python_app.models.dataset_model import ResultCollection, ResultPayload @@ -64,64 +65,3 @@ def gpr_object_rows(collection: ResultCollection) -> np.ndarray: return centers[:, :3] return np.zeros((0, 3), dtype=np.float32) - - -def locator_observations_from_collection( - collection: ResultCollection, - min_score: float, - *, - visible_bounds: tuple[float, float, float, float] | None = None, - object_draw_limits: tuple[int, int] | None = None, -) -> list[dict[str, float]]: - """Build locator observations from GPR rows using score threshold and optional X/Z bounds.""" - rows = gpr_object_rows(collection) - if rows.size == 0: - return [] - - finite_mask = np.all(np.isfinite(rows[:, :3]), axis=1) - visible_mask = finite_mask & (rows[:, 2] >= float(min_score)) - if visible_bounds is not None: - x_min, x_max, z_min, z_max = (float(value) for value in visible_bounds) - visible_mask &= ( - (rows[:, 0] >= x_min) - & (rows[:, 0] <= x_max) - & (rows[:, 1] >= z_min) - & (rows[:, 1] <= z_max) - ) - filtered = rows[visible_mask] - if object_draw_limits is not None and filtered.size > 0: - max_detected_objects, draw_top_objects = object_draw_limits - if filtered.shape[0] > int(max_detected_objects): - filtered = np.zeros((0, filtered.shape[1]), dtype=filtered.dtype) - else: - filtered = filtered[: max(0, int(draw_top_objects))] - - observations: list[dict[str, float]] = [] - for x_m, z_m, _score in filtered: - observations.append( - { - "dst": round(float(z_m), 2), - "crs": round(float(x_m), 2), - } - ) - return observations - - -def build_locator_payload( - observations: list[dict[str, float]], - *, - protocol_version: int, - status: int = 1, -) -> dict[str, Any]: - """Assemble one outbound locator payload from precomputed observations.""" - return { - "ver": int(protocol_version), - "tim": _format_timestamp(), - "sts": int(status), - "obs": observations, - } - - -def _format_timestamp() -> str: - """Return wall-clock timestamp with millisecond precision.""" - return datetime.now().strftime("%H:%M:%S.%f")[:-3] diff --git a/python_app/orchestration/live_processing_config.py b/python_app/orchestration/live_processing_config.py index 79621f1..c8800ae 100644 --- a/python_app/orchestration/live_processing_config.py +++ b/python_app/orchestration/live_processing_config.py @@ -44,6 +44,12 @@ class ProcessingLiveConfig: gpr_background_mean_count: int = 10 gpr_remove_sidelobe_objects_enabled: bool = True gpr_imaging_plane_y_m: float = 0.0 + # Locator filter parameters consumed by the C++ TCP locator server. + gpr_min_visible_score: float = 0.0 + legacy_gpr_min_visible_pair_count: float = 0.0 + # When true, the C++ data_processor ignores socket-supplied vlc updates + # and keeps using `gpr_speed_m_s` from this file. + ignore_socket_speed: bool = False reprocess_current_result: bool = True history_command_seq: int = 0 history_command: str = "none" @@ -95,6 +101,9 @@ class ProcessingLiveConfig: "gpr_background_mean_count": int(self.gpr_background_mean_count), "gpr_remove_sidelobe_objects_enabled": bool(self.gpr_remove_sidelobe_objects_enabled), "gpr_imaging_plane_y_m": float(self.gpr_imaging_plane_y_m), + "gpr_min_visible_score": float(self.gpr_min_visible_score), + "legacy_gpr_min_visible_pair_count": float(self.legacy_gpr_min_visible_pair_count), + "ignore_socket_speed": bool(self.ignore_socket_speed), "reprocess_current_result": bool(self.reprocess_current_result), "history_command_seq": int(self.history_command_seq), "history_command": str(self.history_command), diff --git a/python_app/orchestration/locator_runtime.py b/python_app/orchestration/locator_runtime.py deleted file mode 100644 index 6344f79..0000000 --- a/python_app/orchestration/locator_runtime.py +++ /dev/null @@ -1,460 +0,0 @@ -"""Event-driven locator TCP service fed by already-consumed GUI GPR results.""" - -from __future__ import annotations - -import asyncio -import contextlib -from dataclasses import dataclass -import json -import logging -import math -import queue -import struct -import threading -from typing import Any - -from python_app.models.dataset_model import ResultCollection -from python_app.orchestration.gpr_locator import ( - build_locator_payload, - locator_observations_from_collection, -) - -_PACKET_HEADER_STRUCT = struct.Struct(" bytes: - """Serialize a JSON payload with the protocol binary header.""" - payload_bytes = json.dumps( - payload, - ensure_ascii=True, - separators=(",", ":"), - ).encode("utf-8") - return _PACKET_HEADER_STRUCT.pack(device_id, len(payload_bytes)) + payload_bytes - - -def decode_packet(header_bytes: bytes, payload_bytes: bytes) -> tuple[int, Any]: - """Decode one protocol packet from its binary header and JSON payload.""" - if len(header_bytes) != _PACKET_HEADER_STRUCT.size: - raise ValueError(f"Packet header must be exactly {_PACKET_HEADER_STRUCT.size} bytes long.") - - device_id, payload_length = _PACKET_HEADER_STRUCT.unpack(header_bytes) - if payload_length != len(payload_bytes): - raise ValueError("Payload length does not match the header value.") - - try: - payload = json.loads(payload_bytes.decode("utf-8")) - except UnicodeDecodeError as error: - raise ValueError("Payload is not valid UTF-8.") from error - except json.JSONDecodeError as error: - raise ValueError("Payload is not valid JSON.") from error - - return device_id, payload - - -def parse_vlc(payload: dict[str, Any]) -> float: - """Validate and normalize inbound speed payload.""" - try: - vlc = float(payload["vlc"]) - except (KeyError, TypeError, ValueError) as error: - raise ValueError("Payload field 'vlc' must be numeric.") from error - - if not math.isfinite(vlc): - raise ValueError("Payload field 'vlc' must be finite.") - return vlc - - -def format_payload_for_log(payload: Any) -> str: - """Return compact JSON-ish payload text for logs.""" - return json.dumps(payload, ensure_ascii=True, separators=(",", ":")) - - -def decode_packet_for_log(packet: bytes) -> tuple[int, str]: - """Decode an outbound packet into `(device_id, payload_text)` for logging.""" - if len(packet) < _PACKET_HEADER_STRUCT.size: - raise ValueError("Packet is shorter than the locator header") - header_bytes = packet[: _PACKET_HEADER_STRUCT.size] - payload_bytes = packet[_PACKET_HEADER_STRUCT.size :] - device_id, payload = decode_packet(header_bytes, payload_bytes) - return device_id, format_payload_for_log(payload) - - -def format_peer_name(writer: asyncio.StreamWriter) -> str: - """Return a readable peer address for logs.""" - peer_name = writer.get_extra_info("peername") - if isinstance(peer_name, tuple) and len(peer_name) >= 2: - return f"{peer_name[0]}:{peer_name[1]}" - return str(peer_name or "unknown") - - -async def read_packet_with_limit(reader: asyncio.StreamReader, max_payload_bytes: int) -> tuple[int, Any]: - """Read and decode a single packet using the requested payload limit.""" - header_bytes = await reader.readexactly(_PACKET_HEADER_STRUCT.size) - _, payload_length = _PACKET_HEADER_STRUCT.unpack(header_bytes) - if payload_length > int(max_payload_bytes): - raise ValueError( - "Payload length %d exceeds the %d byte limit." - % (payload_length, int(max_payload_bytes)) - ) - payload_bytes = await reader.readexactly(payload_length) - return decode_packet(header_bytes, payload_bytes) - - -@dataclass(eq=False, slots=True) -class _ClientConnection: - """Runtime state for one connected locator client.""" - - writer: asyncio.StreamWriter - peer_name: str - queue: asyncio.Queue[bytes] - closed: bool = False - - -class LocatorTcpService: - """Background-thread TCP service for locator packets.""" - - def __init__( - self, - host: str, - port: int, - *, - device_id: int, - protocol_version: int, - max_payload_bytes: int, - client_queue_size: int, - logger_name: str, - logger: logging.Logger | None = None, - ) -> None: - """Create a stopped service instance.""" - self._host = host - self._port = int(port) - self._device_id = int(device_id) - self._protocol_version = int(protocol_version) - self._max_payload_bytes = int(max_payload_bytes) - self._logger = logger or logging.getLogger(str(logger_name)) - self._client_queue_size = int(client_queue_size) - self._speed_updates: queue.Queue[float] = queue.Queue() - self._log_updates: queue.Queue[str] = queue.Queue() - self._loop: asyncio.AbstractEventLoop | None = None - self._server: asyncio.AbstractServer | None = None - self._thread: threading.Thread | None = None - self._startup_event = threading.Event() - self._startup_error: Exception | None = None - self._clients: set[_ClientConnection] = set() - self._snapshot_lock = threading.Lock() - self._latest_packet: bytes | None = None - - @property - def host(self) -> str: - """Return bind host.""" - return self._host - - @property - def port(self) -> int: - """Return bind port.""" - return self._port - - def start(self) -> None: - """Start the background event loop and TCP listener.""" - if self.is_running(): - return - - self._startup_event = threading.Event() - self._startup_error = None - self._thread = threading.Thread( - target=self._thread_main, - name="locator-tcp-service", - daemon=True, - ) - self._thread.start() - - if not self._startup_event.wait(timeout=5.0): - raise RuntimeError("Timed out waiting for locator TCP service startup.") - - if self._startup_error is not None: - error = self._startup_error - self.stop() - raise RuntimeError(f"Failed to start locator TCP service: {error}") from error - - def stop(self) -> None: - """Stop listener, disconnect clients, and join the background thread.""" - loop = self._loop - thread = self._thread - - if loop is not None: - with contextlib.suppress(RuntimeError): - loop.call_soon_threadsafe(loop.stop) - - if thread is not None: - thread.join(timeout=5.0) - - self._thread = None - self._loop = None - self._server = None - self._clients.clear() - - def is_running(self) -> bool: - """Return whether the background loop is alive.""" - return self._thread is not None and self._thread.is_alive() and self._loop is not None - - def publish_collection( - self, - collection: ResultCollection, - min_score: float, - *, - visible_bounds: tuple[float, float, float, float] | None = None, - object_draw_limits: tuple[int, int] | None = None, - ) -> None: - """Publish one locator payload derived from a GPR result collection.""" - observations = locator_observations_from_collection( - collection, - min_score, - visible_bounds=visible_bounds, - object_draw_limits=object_draw_limits, - ) - payload = build_locator_payload( - observations, - protocol_version=self._protocol_version, - status=1, - ) - self._publish_packet(encode_packet(payload, device_id=self._device_id)) - - def publish_empty(self) -> None: - """Publish an empty locator snapshot.""" - payload = build_locator_payload( - [], - protocol_version=self._protocol_version, - status=1, - ) - self._publish_packet(encode_packet(payload, device_id=self._device_id)) - - def drain_speed_updates(self) -> float | None: - """Drain queued speed updates and return the newest one, if any.""" - latest: float | None = None - while True: - try: - latest = float(self._speed_updates.get_nowait()) - except queue.Empty: - return latest - - def drain_log_updates(self) -> list[str]: - """Drain queued socket traffic log lines.""" - lines: list[str] = [] - while True: - try: - lines.append(str(self._log_updates.get_nowait())) - except queue.Empty: - return lines - - def _queue_log_update(self, message: str) -> None: - """Queue one socket traffic line for the GUI runtime log.""" - self._log_updates.put(str(message)) - - def _log_socket_traffic(self, message: str) -> None: - """Log socket traffic to both Python logging and the GUI-visible queue.""" - self._logger.info(message) - self._queue_log_update(message) - - def _publish_packet(self, packet: bytes) -> None: - """Store latest packet and broadcast it to all connected clients.""" - with self._snapshot_lock: - self._latest_packet = packet - - loop = self._loop - if loop is None: - return - - with contextlib.suppress(RuntimeError): - loop.call_soon_threadsafe(self._broadcast_packet, packet) - - def _get_latest_packet(self) -> bytes | None: - """Return the latest stored packet snapshot.""" - with self._snapshot_lock: - return self._latest_packet - - def _thread_main(self) -> None: - """Own the event loop and TCP listener lifecycle.""" - loop = asyncio.new_event_loop() - self._loop = loop - asyncio.set_event_loop(loop) - - try: - self._server = loop.run_until_complete( - asyncio.start_server(self._handle_client, self._host, self._port) - ) - except Exception as exc: # noqa: BLE001 - self._startup_error = exc - self._startup_event.set() - self._loop = None - asyncio.set_event_loop(None) - loop.close() - return - - self._startup_event.set() - try: - loop.run_forever() - finally: - with contextlib.suppress(Exception): - loop.run_until_complete(self._shutdown_async()) - asyncio.set_event_loop(None) - loop.close() - self._server = None - self._loop = None - - async def _shutdown_async(self) -> None: - """Close listener and all active client connections.""" - server = self._server - if server is not None: - server.close() - await server.wait_closed() - - clients = list(self._clients) - self._clients.clear() - for client in clients: - client.closed = True - client.writer.close() - - for client in clients: - with contextlib.suppress(BrokenPipeError, ConnectionResetError): - await client.writer.wait_closed() - - pending = [ - task - for task in asyncio.all_tasks() - if task is not asyncio.current_task() - ] - for task in pending: - task.cancel() - for task in pending: - with contextlib.suppress(asyncio.CancelledError, Exception): - await task - - async def _handle_client( - self, - reader: asyncio.StreamReader, - writer: asyncio.StreamWriter, - ) -> None: - """Handle one client until disconnect or protocol failure.""" - peer_name = format_peer_name(writer) - client = _ClientConnection( - writer=writer, - peer_name=peer_name, - queue=asyncio.Queue(maxsize=self._client_queue_size), - ) - self._clients.add(client) - self._logger.info("Locator client connected: %s", peer_name) - - latest_packet = self._get_latest_packet() - if latest_packet is not None: - self._enqueue_packet(client, latest_packet) - - send_task = asyncio.create_task( - self._send_packets(client), - name=f"locator_send:{peer_name}", - ) - receive_task = asyncio.create_task( - self._receive_packets(reader, client), - name=f"locator_receive:{peer_name}", - ) - - done, pending = await asyncio.wait( - {send_task, receive_task}, - return_when=asyncio.FIRST_COMPLETED, - ) - - for task in pending: - task.cancel() - for task in pending: - with contextlib.suppress(asyncio.CancelledError): - await task - - self._clients.discard(client) - client.closed = True - writer.close() - with contextlib.suppress(BrokenPipeError, ConnectionResetError): - await writer.wait_closed() - - for task in done: - exception = task.exception() - if exception is None: - continue - if isinstance(exception, asyncio.IncompleteReadError): - self._logger.info("Locator client closed the connection: %s", peer_name) - continue - if isinstance(exception, (BrokenPipeError, ConnectionResetError)): - self._logger.info("Locator connection lost: %s", peer_name) - continue - if isinstance(exception, ValueError): - self._logger.warning( - "Closing locator client %s after protocol error: %s", - peer_name, - exception, - ) - continue - self._logger.error( - "Unexpected locator client error: %s", - peer_name, - exc_info=(type(exception), exception, exception.__traceback__), - ) - - self._logger.info("Locator client disconnected: %s", peer_name) - - async def _send_packets(self, client: _ClientConnection) -> None: - """Drain one client's outbound queue.""" - while True: - packet = await client.queue.get() - client.writer.write(packet) - await client.writer.drain() - try: - device_id, payload_text = decode_packet_for_log(packet) - self._log_socket_traffic( - "Locator socket sent to %s: device_id=%d payload=%s" - % (client.peer_name, device_id, payload_text) - ) - except ValueError as error: - self._log_socket_traffic( - "Locator socket sent undecodable packet to %s: bytes=%d error=%s" - % (client.peer_name, len(packet), error) - ) - - async def _receive_packets( - self, - reader: asyncio.StreamReader, - client: _ClientConnection, - ) -> None: - """Receive inbound client packets and queue valid speed updates.""" - while True: - device_id, payload = await read_packet_with_limit(reader, self._max_payload_bytes) - payload_text = format_payload_for_log(payload) - if isinstance(payload, dict) and "vlc" in payload: - speed_m_s = parse_vlc(payload) - self._speed_updates.put(speed_m_s) - self._log_socket_traffic( - "Locator socket received from %s: device_id=%d payload=%s speed_m_s=%g" - % (client.peer_name, device_id, payload_text, speed_m_s) - ) - continue - - self._log_socket_traffic( - "Locator socket received from %s: device_id=%d payload=%s" - % (client.peer_name, device_id, payload_text) - ) - - def _broadcast_packet(self, packet: bytes) -> None: - """Enqueue one packet for all connected clients.""" - for client in list(self._clients): - self._enqueue_packet(client, packet) - - def _enqueue_packet(self, client: _ClientConnection, packet: bytes) -> None: - """Enqueue one packet or disconnect a backpressured client.""" - if client.closed: - return - - try: - client.queue.put_nowait(packet) - except asyncio.QueueFull: - client.closed = True - self._logger.warning( - "Disconnecting locator client %s after outbound queue overflow.", - client.peer_name, - ) - client.writer.close() diff --git a/python_app/orchestration/pipeline_metrics.py b/python_app/orchestration/pipeline_metrics.py new file mode 100644 index 0000000..8f4f1f0 --- /dev/null +++ b/python_app/orchestration/pipeline_metrics.py @@ -0,0 +1,104 @@ +"""Rolling pipeline timing metrics emitted to the runtime log. + +Three independent samples are accumulated: + * acquisition — `capture_end_ns - capture_start_ns` from each raw sweep + * processing — `processing_duration_ns` from each result collection + * rendering — wall time of the Python render call + +Each metric flushes an averaged report to a caller-supplied logger as soon as +its rolling buffer reaches `report_every` samples (default 50). Metrics are +strictly read-only: malformed or missing input is silently ignored so a busy +pipeline never blocks on a stray sample. +""" + +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass +from typing import Callable, Iterable + + +@dataclass(frozen=True, slots=True) +class MetricReport: + """Summary of one rolling-window flush. + + All durations are in nanoseconds. `n` is the number of samples that fed the + summary — never less than 1. `min_ns` / `max_ns` mark the extremes of the + window so spikes are visible even when the average stays calm. + """ + + name: str + n: int + avg_ns: int + min_ns: int + max_ns: int + + def format_ms(self) -> str: + """Format the summary as a one-line `ms`-scaled log message.""" + return ( + f"metrics: {self.name} n={self.n} " + f"avg={self.avg_ns / 1_000_000:.2f}ms " + f"min={self.min_ns / 1_000_000:.2f}ms " + f"max={self.max_ns / 1_000_000:.2f}ms" + ) + + +class PipelineMetrics: + """Accumulate per-stage durations and flush averaged reports. + + The caller supplies a `log_sink` (a function taking a single string) that + receives one report line per flushed metric. Wiring `log_sink` to the GUI + log writer keeps metric output co-located with the rest of the runtime + log; routing it to `print` keeps the class trivially unit-testable. + """ + + def __init__( + self, + *, + report_every: int = 50, + log_sink: Callable[[str], None] | None = None, + ) -> None: + """Create a collector with a flush threshold and optional log sink.""" + if report_every < 1: + raise ValueError("report_every must be >= 1") + self._report_every = int(report_every) + self._log_sink = log_sink + self._buffers: dict[str, deque[int]] = {} + + def set_log_sink(self, log_sink: Callable[[str], None] | None) -> None: + """Reassign the log sink (used when the GUI log appears after init).""" + self._log_sink = log_sink + + def record(self, name: str, duration_ns: int) -> MetricReport | None: + """Append one sample. Return a flushed report if the buffer is full.""" + if duration_ns <= 0: + return None + buffer = self._buffers.setdefault(name, deque()) + buffer.append(int(duration_ns)) + if len(buffer) < self._report_every: + return None + + samples = list(buffer) + buffer.clear() + report = self._summarize(name, samples) + if self._log_sink is not None: + self._log_sink(report.format_ms()) + return report + + def reset(self) -> None: + """Discard all buffered samples without emitting a report.""" + self._buffers.clear() + + @staticmethod + def _summarize(name: str, samples: Iterable[int]) -> MetricReport: + """Reduce a sample sequence to one report.""" + sample_list = list(samples) + total = sum(sample_list) + count = len(sample_list) + return MetricReport( + name=name, + n=count, + avg_ns=total // count, + min_ns=min(sample_list), + max_ns=max(sample_list), + ) diff --git a/python_app/orchestration/shm/decoder.py b/python_app/orchestration/shm/decoder.py index 2ca1fec..c0eb49c 100644 --- a/python_app/orchestration/shm/decoder.py +++ b/python_app/orchestration/shm/decoder.py @@ -16,7 +16,7 @@ from python_app.orchestration.shm.binary_cursor import ByteCursor RAW_MAGIC = 0x32574152 PREPROC_MAGIC = 0x32525050 -RESULT_MAGIC = 0x314C5352 +RESULT_MAGIC = 0x324C5352 # RSL2: adds processing_duration_ns after monotonic_ns def decode_trace_collection(payload: bytes, expected_magic: int) -> SweepCollection: @@ -136,6 +136,7 @@ def decode_result_collection(payload: bytes) -> ResultCollection: collection_id = cursor.read_u64() monotonic_ns = cursor.read_u64() + processing_duration_ns = cursor.read_u64() collection_payload_count = cursor.read_u32() block_count = cursor.read_u32() @@ -163,6 +164,7 @@ def decode_result_collection(payload: bytes) -> ResultCollection: return ResultCollection( collection_id=collection_id, monotonic_ns=monotonic_ns, + processing_duration_ns=processing_duration_ns, collection_payloads=collection_payloads, blocks=blocks, ) diff --git a/python_app/scripts/hardware_raw_orchestrator_test.py b/python_app/scripts/hardware_raw_orchestrator_test.py index a06557e..bd7e9f4 100644 --- a/python_app/scripts/hardware_raw_orchestrator_test.py +++ b/python_app/scripts/hardware_raw_orchestrator_test.py @@ -75,8 +75,8 @@ def _prepare_radar_if_needed(config_path: Path, *, strict: bool) -> str | None: config = RunConfigModel.from_dict(config_payload) if config.radar.driver_mode != "native": return "Radar pre-configuration skipped (mock mode)." - if config.is_multi_device: - return "Radar pre-configuration skipped (multi-device producer config)." + if config.is_matrix_radar: + return "Radar pre-configuration skipped (matrix radar producer config)." radar_service = create_single_radar_service(config) if not getattr(radar_service, "driver_available", True): diff --git a/python_app/storage/npz/serialize.py b/python_app/storage/npz/serialize.py index a0097bb..65848af 100644 --- a/python_app/storage/npz/serialize.py +++ b/python_app/storage/npz/serialize.py @@ -10,7 +10,7 @@ from python_app.models.dataset_model import ResultCollection, SweepCollection RAW_MAGIC = 0x32574152 PREPROC_MAGIC = 0x32525050 -RESULT_MAGIC = 0x314C5352 +RESULT_MAGIC = 0x324C5352 # RSL2: adds processing_duration_ns after monotonic_ns def _write_interleaved_complex(buffer: bytearray, values: np.ndarray) -> None: @@ -109,10 +109,11 @@ def serialize_result_collection(collection: ResultCollection) -> bytes: buffer = bytearray() buffer.extend( struct.pack( - " tuple[str, SweepCollection]: """Capture all switch combinations and persist them as calibration set.""" - if config.is_multi_device: + if config.is_matrix_radar: raise RuntimeError( - "LibreVNA multi-device S21 through calibration is not supported by this one-shot full-set helper. " + "Matrix-radar S21 through calibration is not supported by this one-shot full-set helper. " "Use the sequential preprocess capture flow so each virtual combo can be connected through " "and captured explicitly." ) diff --git a/python_app/workflows/multi_radar_capture_workflow.py b/python_app/workflows/multi_radar_capture_workflow.py index 1a38d36..e462d7b 100644 --- a/python_app/workflows/multi_radar_capture_workflow.py +++ b/python_app/workflows/multi_radar_capture_workflow.py @@ -16,7 +16,7 @@ from python_app.models.run_config_model import ComboModel, RunConfigModel from python_app.storage.npz_store import NpzStore from python_app.workflows.radar_config_variants import RadarConfigVariant from python_app.workflows.sequential_capture_workflow import ( - MULTI_DEVICE_MANUAL_CAPTURE_KINDS, + MATRIX_RADAR_MANUAL_CAPTURE_KINDS, SequentialCaptureState, combine_collections_via_median, combine_traces_via_median, @@ -75,8 +75,9 @@ class MultiRadarSequentialCaptureSession: self._radar_variants = list(radar_variants) self._median_sweep_count = int(median_sweep_count) self._is_matrix_radar = base_config.is_matrix_radar - self._is_multi_device = base_config.is_multi_device - self._manual_multi_device_capture = self._is_multi_device and kind in MULTI_DEVICE_MANUAL_CAPTURE_KINDS + self._manual_matrix_radar_capture = ( + self._is_matrix_radar and kind in MATRIX_RADAR_MANUAL_CAPTURE_KINDS + ) self._combos = ( RunConfigModel.build_matrix_radar_virtual_combos() if self._is_matrix_radar @@ -169,7 +170,7 @@ class MultiRadarSequentialCaptureSession: set_name=self._set_name, captured_count=( self._next_index - if self._is_matrix_radar and not self._manual_multi_device_capture + if self._is_matrix_radar and not self._manual_matrix_radar_capture else len(self._captured_batches) ), total_count=len(self._combos), @@ -177,7 +178,7 @@ class MultiRadarSequentialCaptureSession: can_undo=bool(self._captured_batches), is_complete=self.is_complete(), variant_count=len(self._radar_variants), - supports_batch_capture=not self._manual_multi_device_capture, + supports_batch_capture=not self._manual_matrix_radar_capture, ) def capture_current_combo(self) -> MultiRadarCaptureBatch: @@ -205,7 +206,7 @@ class MultiRadarSequentialCaptureSession: f"Matrix radar variant {variant.display_name} returned no traces" ) collections.append(collection) - if self._manual_multi_device_capture: + if self._manual_matrix_radar_capture: per_sweep_traces = [select_trace_for_combo(collection, combo) for collection in collections] trace = combine_traces_via_median(per_sweep_traces) pending_traces_by_radar_key[variant.radar_key] = [trace] @@ -251,7 +252,7 @@ class MultiRadarSequentialCaptureSession: variant_labels=tuple(variant_labels), ) self._captured_batches.append(batch) - if self._is_matrix_radar and not self._manual_multi_device_capture: + if self._is_matrix_radar and not self._manual_matrix_radar_capture: self._next_index = len(self._combos) else: self._next_index += 1 @@ -264,7 +265,7 @@ class MultiRadarSequentialCaptureSession: if not self._captured_batches or self._next_index <= 0: raise RuntimeError("No captured combo is available to undo") - if self._is_matrix_radar and not self._manual_multi_device_capture: + if self._is_matrix_radar and not self._manual_matrix_radar_capture: removed_batch = self._captured_batches[-1] for variant in self._radar_variants: traces = self._traces_by_radar_key[variant.radar_key] diff --git a/python_app/workflows/sequential_capture_workflow.py b/python_app/workflows/sequential_capture_workflow.py index d328ecd..f7413f3 100644 --- a/python_app/workflows/sequential_capture_workflow.py +++ b/python_app/workflows/sequential_capture_workflow.py @@ -15,7 +15,7 @@ from python_app.models.dataset_model import ComboKey, SweepCollection, TraceData from python_app.models.run_config_model import ComboModel, RunConfigModel from python_app.storage.npz_store import NpzStore, radar_key_from_config -MULTI_DEVICE_MANUAL_CAPTURE_KINDS = frozenset({"s21_calibration", "s11_open", "s11_short", "s11_load"}) +MATRIX_RADAR_MANUAL_CAPTURE_KINDS = frozenset({"s21_calibration", "s11_open", "s11_short", "s11_load"}) DEFAULT_CALIBRATION_MEDIAN_SWEEP_COUNT = 5 @@ -58,8 +58,9 @@ class SequentialCaptureSession: self._set_name = set_name self._median_sweep_count = int(median_sweep_count) self._is_matrix_radar = config.is_matrix_radar - self._is_multi_device = config.is_multi_device - self._manual_multi_device_capture = self._is_multi_device and kind in MULTI_DEVICE_MANUAL_CAPTURE_KINDS + self._manual_matrix_radar_capture = ( + self._is_matrix_radar and kind in MATRIX_RADAR_MANUAL_CAPTURE_KINDS + ) self._combos = ( RunConfigModel.build_matrix_radar_virtual_combos() if self._is_matrix_radar @@ -148,7 +149,7 @@ class SequentialCaptureSession: current_combo=current_combo, can_undo=bool(self._traces), is_complete=self.is_complete(), - supports_batch_capture=not self._manual_multi_device_capture, + supports_batch_capture=not self._manual_matrix_radar_capture, ) def capture_current_combo(self) -> TraceData: @@ -166,7 +167,7 @@ class SequentialCaptureSession: if not collection.traces: raise RuntimeError("Matrix radar capture returned no traces") collections.append(collection) - if self._manual_multi_device_capture: + if self._manual_matrix_radar_capture: per_sweep_traces = [select_trace_for_combo(collection, combo) for collection in collections] trace = combine_traces_via_median(per_sweep_traces) self._traces.append(trace) @@ -208,7 +209,7 @@ class SequentialCaptureSession: if not self._traces or self._next_index <= 0: raise RuntimeError("No captured combo is available to undo") - if self._is_matrix_radar and not self._manual_multi_device_capture: + if self._is_matrix_radar and not self._manual_matrix_radar_capture: if len(self._traces) != len(self._combos): raise RuntimeError("Capture session state is inconsistent; matrix radar trace matrix is incomplete") removed_trace = self._traces[-1] diff --git a/run_config.json b/run_config.json index ace723b..47c8cd7 100644 --- a/run_config.json +++ b/run_config.json @@ -1,22 +1,18 @@ { - "radar": { - "model": "librevna", - "serial": "", - "driver_mode": "mock", - "mock_signal_hz": 5000000.0, - "multi_device": { - "slave_serials": [], - "force_external_reference": false, - "recovery_attempts": 3 - }, - "sweep": { - "start_hz": 1000000.0, - "stop_hz": 6000000000.0, - "points": 201, - "if_bandwidth_hz": 50000.0, - "stimulus_power_dbm": -10.0 - } - }, +"radar": { + "model": "sn9000", + "remote_host": "192.168.2.102", + "remote_port": 4880, + "driver_mode": "native", + "visa_library": "@py", + "sweep": { + "start_hz": 1000000.0, + "stop_hz": 6000000000.0, + "points": 201, + "if_bandwidth_hz": 10000.0, + "stimulus_power_dbm": -10.0 + } +}, "switches": { "port1": { "name": "port1", diff --git a/start.sh b/start.sh index b7e82ea..5a12d47 100755 --- a/start.sh +++ b/start.sh @@ -16,6 +16,7 @@ CLEAN_SHM=0 KAMIL_ADC_MODE=0 AUTO_START=0 PRODUCER_ONLY=0 +HEADLESS=0 print_usage() { cat <<'EOF' @@ -25,6 +26,10 @@ Options: --kamil-adc Use the Raspberry Pi Kamil ADC profile --profile PATH Use a specific GUI/run config profile --auto-start Start the GUI pipeline automatically after launch + --headless Run without a display (Qt offscreen platform) and apply + the active radar config, then start the pipeline. Suitable + for unattended Raspberry Pi deployments. Implies + --auto-start. --producer-only Run only the raw producer selected by the profile --skip-build Skip C++ build step --build-only Build C++ binaries and exit @@ -59,6 +64,10 @@ parse_args() { --auto-start) AUTO_START=1 ;; + --headless) + HEADLESS=1 + AUTO_START=1 + ;; --producer-only) PRODUCER_ONLY=1 ;; @@ -140,12 +149,15 @@ ensure_python_dependencies() { exit 1 fi - if ! "${VENV_PYTHON}" -c "${dependency_check}" >/dev/null 2>&1; then - echo "[start.sh] Installing Python dependencies into virtual environment..." - "${VENV_PIP}" install --upgrade pip - "${VENV_PIP}" install -r "${REQUIREMENTS_FILE}" + if "${VENV_PYTHON}" -c "${dependency_check}" >/dev/null 2>&1; then + PYTHON_CMD="${VENV_PYTHON}" + return fi + echo "[start.sh] Installing Python dependencies into virtual environment..." + "${VENV_PIP}" install --upgrade pip + "${VENV_PIP}" install -r "${REQUIREMENTS_FILE}" + if ! "${VENV_PYTHON}" -c "${dependency_check}" >/dev/null 2>&1; then echo "Required Python dependencies are still unavailable in virtual environment: ${PROJECT_ROOT}/.venv" >&2 exit 1 @@ -234,6 +246,10 @@ EOF } build_cpp_binaries() { + if make -C "${PROJECT_ROOT}" -q all >/dev/null 2>&1; then + echo "[start.sh] C++ binaries are up to date; skipping build." + return + fi local jobs jobs="${BUILD_JOBS:-$(nproc)}" echo "[start.sh] Building C++ binaries (jobs=${jobs})..." @@ -265,6 +281,16 @@ run_gui() { export RADAR_SYSTEM_AUTO_START=1 echo "[start.sh] GUI auto-start is enabled." fi + if ((HEADLESS == 1)); then + # Qt offscreen platform lets the GUI controller and its event loop run + # on a machine with no display (typical Raspberry Pi deployment). All + # backend services — supervisor, Python device drivers, SHM readers, + # locator client (vlc) handling — continue to work unchanged. + export QT_QPA_PLATFORM=offscreen + export RADAR_SYSTEM_HEADLESS=1 + export RADAR_SYSTEM_AUTO_APPLY_RADAR=1 + echo "[start.sh] Headless mode: Qt offscreen + auto apply-radar + auto-start." + fi echo "[start.sh] Launching GUI..." exec "${PYTHON_CMD}" "${GUI_ENTRY}"