Files
Ayzen b88d704126 Port the SAR3D background gate, CLEAN and object boxes from Multi_depthnorm_clean
- BG_GATE_ENABLE: frame peak over its median background (sar3d_background); the
  GUI scales the frame with bg_gate_gain and its cloud threshold.
- SAR_CLEAN_ENABLE: CLEAN of a one-frame aperture on a 4x10x4 cm search grid over
  the whole scene; objects in their own scale (sar3d_objects).
- SAR_MARK_ENABLE: boxes by class around object groups (sar3d_marks), with the
  legend under the views.
- SAR_VOXEL_M as three steps X/Y/Z with the notebook defaults 0.025/0.02/0.02.
- GPU backend: image/download/load/project API, per-plane block reduction for the
  maximum, depth gate without FMA; sar3d host code built with -ffp-contract=off.
- References regenerated from notebook cells 1 and 5, with new CLEAN scenes and a
  synthetic box-rule case; docs updated.
2026-10-16 12:08:57 +03:00

576 lines
28 KiB
C++

#include "channel_bundle_support.hpp"
#include "data_preprocessor.hpp"
#include "locator/tcp_server.hpp"
#include "object_approach_filter.hpp"
#include "processing_live_config.hpp"
#include "runtime_wait.hpp"
#include "shm_ring.hpp"
#include "tty_protocol_writer.h"
#include "switch_sequencer.h"
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <unistd.h>
#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <cstring>
#include <exception>
#include <filesystem>
#include <fstream>
#include <iostream>
#include <limits>
#include <random>
#include <stdexcept>
#include <thread>
#include <nlohmann/json.hpp>
namespace {
using namespace radar;
using Json = nlohmann::json;
void require(bool condition, const char* message) {
if (!condition) {
throw std::runtime_error(message);
}
}
template <typename Callable>
void expect_error(Callable operation) {
try {
operation();
} catch (const std::bad_alloc&) {
throw; // A malformed count must be rejected before allocating.
} catch (const std::exception&) {
return;
}
throw std::runtime_error("Expected operation to reject malformed input");
}
struct Workspace {
std::filesystem::path path;
Workspace() {
char pattern[] = "/tmp/radar-native-XXXXXX";
const auto* created = ::mkdtemp(pattern);
require(created != nullptr, "mkdtemp failed");
path = created;
}
~Workspace() { std::filesystem::remove_all(path); }
};
struct RingName {
std::string name;
explicit RingName(const std::string& suffix = {})
: name("/radar_native_test_" + std::to_string(::getpid()) + suffix) {
::shm_unlink(name.c_str());
}
~RingName() { ::shm_unlink(name.c_str()); }
};
auto save_bundle(const Workspace& workspace, const char* name, const ipc::RawSweepCollection& collection)
-> std::string {
const auto path = workspace.path / name;
const auto bytes = ipc::serialize_raw_collection(collection);
std::ofstream(path, std::ios::binary).write(
reinterpret_cast<const char*>(bytes.data()), static_cast<std::streamsize>(bytes.size()));
return path.string();
}
void write_u32(std::vector<std::uint8_t>& bytes, std::size_t offset, std::uint32_t value) {
require(offset + sizeof(value) <= bytes.size(), "invalid test offset");
std::memcpy(bytes.data() + offset, &value, sizeof(value));
}
void test_wire_format() {
ipc::RawSweepCollection raw{.collection_id = 7, .monotonic_ns = 42,
.capture_start_ns = 21, .capture_end_ns = 40,
.traces = {{.combo = {1, 2}, .frequency_hz = {1, 2},
.s21 = {{5, 6}, {7, 8}},
.capture_start_ns = 25, .capture_end_ns = 39}}};
ipc::ResultCollection results{.collection_id = 7, .monotonic_ns = 0x0102030405060708ULL,
.processing_duration_ns = 17, .settings_revision = 0xfedcba9876543210ULL,
.capture_start_ns = 21, .capture_end_ns = 40, .speed_m_s = -1.5F,
.collection_payloads = {
{.kind = ipc::ResultKind::TraceComplex, .frequency_hz = {1}, .trace = {{2, 3}}},
{.kind = ipc::ResultKind::ScalarF32, .scalar_value = 4},
{.kind = ipc::ResultKind::ImageF32, .image_x_axis = {1, 2},
.image_y_axis = {3}, .image_values = {4, 5}},
{.kind = ipc::ResultKind::TableF32, .table_columns = 2, .table_values = {6, 7}},
{.kind = ipc::ResultKind::ImageF32, .image_y_axis = {1}},
{.kind = ipc::ResultKind::TableF32, .table_columns = 3}},
.blocks = {{.combo = {1, 2}, .payloads = {{.kind = ipc::ResultKind::ScalarF32}}}}};
const auto raw_bytes = ipc::serialize_raw_collection(raw);
const auto pre_bytes = ipc::serialize_preprocessed_collection(raw);
const auto result_bytes = ipc::serialize_result_collection(results);
const std::array<std::uint8_t, 64> result_header{
'R', 'S', 'L', '3', 7, 0, 0, 0, 0, 0, 0, 0,
8, 7, 6, 5, 4, 3, 2, 1, 17, 0, 0, 0, 0, 0, 0, 0,
0x10, 0x32, 0x54, 0x76, 0x98, 0xba, 0xdc, 0xfe,
21, 0, 0, 0, 0, 0, 0, 0, 40, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0xc0, 0xbf, 6, 0, 0, 0, 1, 0, 0, 0,
};
require(std::equal(result_header.begin(), result_header.end(), result_bytes.begin()),
"RSL3 must preserve exact acquisition identity and settings in its 64-byte header");
require(std::memcmp(raw_bytes.data(), "RAW3", 4) == 0 &&
std::memcmp(pre_bytes.data(), "PRP3", 4) == 0,
"trace collections must use the S21-only version 3 format");
// Collection header + one trace header + two 12-byte points + capture trailer.
require(raw_bytes.size() == 24 + 12 + 2 * 12 + 16 + 4 + 16 && pre_bytes.size() == raw_bytes.size(),
"trace wire format must contain only frequency and S21 samples");
auto old_raw = raw_bytes;
auto old_pre = pre_bytes;
write_u32(old_raw, 0, 0x32574152U);
write_u32(old_pre, 0, 0x32525050U);
expect_error([&] { (void)ipc::deserialize_raw_collection(old_raw); });
expect_error([&] { (void)ipc::deserialize_preprocessed_collection(old_pre); });
require(ipc::serialize_raw_collection(ipc::deserialize_raw_collection(raw_bytes)) == raw_bytes,
"raw roundtrip changed data");
require(ipc::serialize_preprocessed_collection(ipc::deserialize_preprocessed_collection(pre_bytes)) == pre_bytes,
"preprocessed roundtrip changed data");
for (const std::size_t tail : {36U, 20U}) {
expect_error([&] { (void)ipc::deserialize_raw_collection(std::span(raw_bytes).first(raw_bytes.size() - tail)); });
expect_error([&] { (void)ipc::deserialize_preprocessed_collection(std::span(pre_bytes).first(pre_bytes.size() - tail)); });
}
// Deterministic random mutations include wire-compatible valid frames as seeds.
std::mt19937 random(0x52414452U);
for (unsigned iteration = 0; iteration < 10'000U; ++iteration) {
auto bytes = iteration % 2U == 0U ? raw_bytes : pre_bytes;
for (unsigned edit = 0; edit < 1U + random() % 5U; ++edit) {
bytes[random() % bytes.size()] = static_cast<std::uint8_t>(random());
}
if (iteration % 4U == 0U) {
bytes.resize(random() % bytes.size());
}
try {
if (iteration % 2U == 0U) { (void)ipc::deserialize_raw_collection(bytes); }
else { (void)ipc::deserialize_preprocessed_collection(bytes); }
} catch (const std::bad_alloc&) { throw; }
catch (const std::runtime_error&) {}
}
}
void test_configuration(const std::string& seed_path, const Workspace& workspace) {
std::ifstream input(seed_path);
const auto seed = Json::parse(input);
const auto config_path = workspace.path / "config.json";
const auto save = [&](const Json& config) { std::ofstream(config_path) << config; };
save(seed);
const auto config = config::load_run_config(config_path.string());
auto no_taps = seed;
no_taps["rings"]["preprocessed_tap"]["name"] = "";
save(no_taps);
const auto disabled = config::load_run_config(config_path.string()).rings;
require(disabled.preprocessed_tap.name.empty(),
"explicit empty tap names must disable the optional rings");
for (const auto* key : {"raw", "preprocessed", "results"}) {
auto invalid = no_taps;
invalid["rings"][key]["name"] = "";
save(invalid);
expect_error([&] { (void)config::load_run_config(config_path.string()); });
}
for (const auto* port : {"port1", "port2"}) {
auto with_driver = seed;
with_driver["switches"][port]["driver"] = "h7992";
save(with_driver);
(void)config::load_run_config(config_path.string());
for (const auto& driver : {Json("hmc349a"), Json("unknown"), Json(7)}) {
with_driver["switches"][port]["driver"] = driver;
save(with_driver);
expect_error([&] { (void)config::load_run_config(config_path.string()); });
}
}
auto with_object_calibration = seed;
with_object_calibration["preprocess"]["s21"]["object_calibration"] =
Json{{"set_name", "reflector"}, {"bundle_path", "corrections.bin"}};
save(with_object_calibration);
const auto selected = config::load_run_config(config_path.string()).preprocess.s21.object_calibration;
require(selected.bundle_path == "corrections.bin",
"object calibration asset did not load");
with_object_calibration["preprocess"]["s21"].erase("object_calibration");
save(with_object_calibration);
require(config::load_run_config(config_path.string()).preprocess.s21.object_calibration.bundle_path.empty(),
"absent object calibration was not disabled");
for (const auto& invalid_asset : {Json(nullptr), Json{{"bundle_path", true}}}) {
with_object_calibration["preprocess"]["s21"]["object_calibration"] = invalid_asset;
save(with_object_calibration);
expect_error([&] { (void)config::load_run_config(config_path.string()); });
}
for (const auto& [pointer, value] : std::array<std::pair<const char*, double>, 3>{{
{"/run/combos/0/input", 0.0000001},
{"/rings/raw/slot_size_bytes", 65}, {"/gpr/rx_geometry/0/x_m", 1e100}}}) {
auto invalid = seed;
invalid[Json::json_pointer(pointer)] = value;
save(invalid);
expect_error([&] { (void)config::load_run_config(config_path.string()); });
}
const auto live_path = workspace.path / "live.json";
std::ofstream(live_path) << Json{{"bscan_gain", 2}, {"settings_revision", 11},
{"sar3d_bg_gate_enabled", false}, {"sar3d_clean_enabled", false},
{"legacy_gpr_mode", false}, {"relative_permittivity", "obsolete"}};
processing::ProcessingLiveConfigLoader loader(live_path.string(), config);
const auto loaded = loader.refresh_if_needed();
require(loaded.bscan_gain == 2 && !loaded.sar3d_bg_gate_enabled && !loaded.sar3d_clean_enabled,
"valid live config did not load");
const auto revision = loader.revision();
for (const auto& invalid : {Json{{"bscan_gain", 1e100}},
Json{{"gpr_input_positions", {0.0000001}}}, Json{{"bscan_gain", true}},
Json{{"processor_mode", "legacy_gpr"}}}) {
std::ofstream(live_path) << invalid;
std::this_thread::sleep_for(std::chrono::milliseconds(110));
const auto current = loader.refresh_if_needed();
require(current.bscan_gain == 2 && loader.revision() == revision,
"invalid live config replaced the last valid settings");
}
}
void test_ring() {
RingName name;
expect_error([&] { (void)ipc::ShmRing::open_or_create(name.name, 2, 65); });
expect_error([&] { (void)ipc::ShmRing::open_or_create(name.name, 0xffffffffU, 0xfffffff8U); });
auto ring = ipc::ShmRing::open_or_create(name.name, 2, 4096);
const auto child = ::fork();
require(child >= 0, "fork failed");
if (child == 0) {
std::array<std::uint64_t, 512> words{};
for (std::uint64_t sequence = 1; sequence <= 20'000U; ++sequence) {
words.fill(sequence);
if (!ring.push({reinterpret_cast<const std::uint8_t*>(words.data()), sizeof(words)})) {
::_exit(2);
}
}
::_exit(0);
}
std::uint64_t previous = 0;
std::vector<std::uint8_t> payload;
int status = 0;
bool finished = false;
const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10);
try {
while (true) {
if (ring.pop(payload)) {
require(payload.size() == 4096, "stress payload changed size");
std::uint64_t sequence = 0;
std::memcpy(&sequence, payload.data(), sizeof(sequence));
require(sequence > previous, "stress ring yielded duplicate or older data");
for (std::size_t offset = 0; offset < payload.size(); offset += sizeof(sequence)) {
std::uint64_t word = 0;
std::memcpy(&word, payload.data() + offset, sizeof(word));
require(word == sequence, "stress ring accepted a torn payload");
}
previous = sequence;
} else if (finished) { break; }
finished = finished || ::waitpid(child, &status, WNOHANG) == child;
require(std::chrono::steady_clock::now() < deadline, "ring stress timed out");
}
require(WIFEXITED(status) && WEXITSTATUS(status) == 0 && previous != 0,
"ring stress writer failed");
} catch (...) {
::kill(child, SIGKILL);
if (!finished) { (void)::waitpid(child, &status, 0); }
throw;
}
}
void test_acquisition_identity() {
processing::ObjectApproachFilter filter;
const std::vector<processing::ObjectApproachFilter::Point> points{{0, 2}};
require(!filter.confirm(points, 1, 1'000'000'000ULL, 0, 0, 2).front(), "first frame confirmed a track");
require(filter.confirm(points, 1, 2'000'000'000ULL, 0, 0, 2).front(),
"reused collection id replaced a distinct acquisition");
require(filter.confirm(points, 1, 2'000'000'000ULL, 0, 0, 2).front(),
"reprocessing duplicated a frame or lost its predecessor");
require(!filter.confirm(points, 7, 1'000'000'000ULL, 0, 0, 2).front(),
"an earlier capture time failed to reset replay history");
}
void test_interruptible_wait() {
std::atomic<bool> stop{false};
std::jthread request_stop([&] {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
stop.store(true);
});
const auto start = std::chrono::steady_clock::now();
require(config::wait_or_stop(0xffffffffU, stop), "large wait ignored stop");
require(std::chrono::steady_clock::now() - start < std::chrono::seconds(1),
"large wait delayed shutdown");
}
void test_cosine_notch_taper() {
// Band 2.0-2.1 GHz with 0.1 GHz shoulders: 1 at the outer edges, 0.5 halfway,
// 0 from the band edges inwards.
const std::vector<float> frequency_hz{1.85e9F, 1.9e9F, 1.95e9F, 2.0e9F, 2.05e9F,
2.1e9F, 2.15e9F, 2.2e9F, 2.25e9F};
const std::vector<float> expected{1.0F, 1.0F, 0.5F, 0.0F, 0.0F, 0.0F, 0.5F, 1.0F, 1.0F};
const auto mask = preprocessing::build_notch_mask(frequency_hz,
{.enabled = true, .bands_hz = {{2.0e9F, 2.1e9F}}, .taper_width_hz = 1e8F, .taper_type = "cosine"});
for (std::size_t index = 0U; index < expected.size(); ++index) {
require(std::abs(mask[index] - expected[index]) < 1e-3F, "cosine notch shoulders are not 1 -> 0 towards the band");
}
}
void test_asset_reload_preserves_data(const Workspace& workspace) {
const auto save = [&](const char* name, float sample) {
return save_bundle(workspace, name, {.traces = {
{.frequency_hz = {1}, .s21 = {{sample, 0}}}}});
};
const auto open = save("open.bin", 1);
const auto nan = save("nan.bin", std::numeric_limits<float>::quiet_NaN());
preprocessing::S21CalibrationBundle s21_calibration;
s21_calibration.load(save("through.bin", 2));
expect_error([&] { s21_calibration.load(nan); });
const std::array<float, 1> frequency{1};
std::array<ipc::Complex32, 1> samples{{{6, 8}}};
s21_calibration.apply({}, frequency, samples);
require(samples[0].re == 3 && samples[0].im == 4, "failed S21 reload changed previous calibration");
const std::array<float, 1> nan_axis{std::numeric_limits<float>::quiet_NaN()};
expect_error([&] { s21_calibration.apply({}, nan_axis, samples); });
preprocessing::S21ReferenceBundle reference;
reference.load(open, s21_calibration, {{}});
expect_error([&] { reference.load(nan, s21_calibration, {{}}); });
ipc::SweepTraceBlock trace{.frequency_hz = {1}, .s21 = {{2, 0}}};
reference.apply(trace);
require(trace.s21[0].re == 1.5F, "failed reference reload changed the calibrated reference");
}
void test_object_calibration(const Workspace& workspace) {
const ipc::SweepTraceBlock first{.combo = {7, 3}, .frequency_hz = {1e9F, 2e9F},
.s21 = {{2, 3}, {4, 5}}, .capture_start_ns = 5, .capture_end_ns = 15};
const ipc::SweepTraceBlock second{.combo = {1, 9}, .frequency_hz = {3e9F, 4e9F, 5e9F},
.s21 = {{6, 7}, {8, 9}, {10, 11}}};
auto factors_first = first;
factors_first.s21 = {{0, 2}, {2, 0}};
auto factors_second = second;
factors_second.s21 = {{1, -1}, {-1, 0}, {0, 1}};
preprocessing::S21ObjectCalibrationBundle correction;
auto trace = first;
correction.apply(trace);
require(ipc::serialize_raw_collection({.traces = {trace}}) == ipc::serialize_raw_collection({.traces = {first}}),
"disabled object calibration changed a trace");
correction.load(save_bundle(workspace, "object.bin", {.traces = {factors_second, factors_first}}));
correction.validate_combos({first.combo, second.combo});
expect_error([&] { correction.validate_combos({{0, 0}}); });
for (auto invalid : {first, second}) {
invalid.combo = {0, 0};
expect_error([&] { correction.apply(invalid); });
}
auto mismatch = second;
mismatch.frequency_hz[1] += 1e8F; // Same endpoints and count; interior grid differs.
expect_error([&] { correction.apply(mismatch); });
mismatch = second;
mismatch.s21.pop_back();
expect_error([&] { correction.apply(mismatch); });
auto nonfinite = factors_first;
nonfinite.s21[0].re = std::numeric_limits<float>::quiet_NaN();
expect_error([&] { correction.load(save_bundle(workspace, "object_nan.bin", {.traces = {nonfinite}})); });
expect_error([&] { correction.load(save_bundle(workspace, "object_duplicate.bin", {.traces = {factors_first, factors_first}})); });
auto expected = first;
expected.s21 = {{-6, 4}, {8, 10}};
trace = first;
correction.apply(trace);
require(ipc::serialize_raw_collection({.traces = {trace}}) == ipc::serialize_raw_collection({.traces = {expected}}),
"object calibration changed timing, pair identity, or multiplication after a failed reload");
trace = second;
correction.apply(trace);
require(trace.s21[0].re == 13 && trace.s21[0].im == 1 && trace.s21[1].re == -8 && trace.s21[1].im == -9 &&
trace.s21[2].re == -11 && trace.s21[2].im == 10, "object calibration mixed pairs or frequency grids");
trace = first;
trace.s21[0] = {std::numeric_limits<float>::max(), std::numeric_limits<float>::max()};
expect_error([&] { correction.apply(trace); });
correction.load("");
trace = first;
correction.apply(trace);
require(trace.s21[0].re == 2 && trace.s21[0].im == 3, "clearing object calibration left stale factors");
}
void test_object_calibration_pipeline(const Workspace& workspace) {
const ipc::SweepTraceBlock trace{.combo = {7, 3}, .frequency_hz = {1e9F, 2e9F},
.s21 = {{2, 4}, {6, 8}}, .capture_start_ns = 5, .capture_end_ns = 15};
auto standard = trace;
standard.s21 = {{2, 0}, {2, 0}};
auto reference = trace;
reference.s21 = {{1, 2}, {2, 2}};
auto factors = trace;
factors.s21 = {{0, 2}, {5, 1}};
config::RunConfig config;
config.preprocess.s21.calibration.bundle_path =
save_bundle(workspace, "pipeline_through.bin", {.traces = {standard}});
config.preprocess.s21.reference.bundle_path =
save_bundle(workspace, "pipeline_reference.bin", {.traces = {reference}});
config.run_combos = {trace.combo};
config.preprocess.s21.object_calibration.bundle_path =
save_bundle(workspace, "pipeline_object.bin", {.traces = {factors}});
config.preprocess.notch = {.enabled = true, .bands_hz = {{2e9F, 2e9F}}, .taper_type = "hard"};
RingName raw_name("_object_raw"), pre_name("_object_pre");
auto raw_ring = ipc::ShmRing::open_or_create(raw_name.name, 2, 4096);
auto pre_ring = ipc::ShmRing::open_or_create(pre_name.name, 2, 4096);
preprocessing::DataPreprocessor preprocessor(config, raw_ring, pre_ring);
require(raw_ring.push(ipc::serialize_raw_collection({.collection_id = 41, .capture_start_ns = 5,
.capture_end_ns = 15, .traces = {trace}})), "test raw ring could not hold a sweep");
std::atomic<bool> stop{false};
std::exception_ptr error;
std::jthread worker([&] { try { preprocessor.run(stop); } catch (...) { error = std::current_exception(); } });
std::vector<std::uint8_t> bytes;
const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
while (bytes.empty() && std::chrono::steady_clock::now() < deadline) {
(void)pre_ring.pop(bytes);
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
stop.store(true);
worker.join();
if (error) { std::rethrow_exception(error); }
require(!bytes.empty(), "object-corrected preprocessed collection was not published");
const auto result = ipc::deserialize_preprocessed_collection(bytes);
const auto& output = result.traces.at(0);
require(result.collection_id == 41 && result.capture_start_ns == 5 && result.capture_end_ns == 15 &&
output.combo == trace.combo && output.capture_start_ns == 5 && output.capture_end_ns == 15,
"object calibration lost capture identity or timing");
require(output.s21[0].re == -2 && output.s21[0].im == 1 && output.s21[1].re == 0 && output.s21[1].im == 0,
"object calibration ran before existing preprocessing or was applied more than once");
}
void test_locator_packet_errors() {
int sockets[2]{};
require(::socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) == 0, "socketpair failed");
std::atomic<double> speed{0};
std::atomic<std::int64_t> at_ns{0};
locator::ClientSession client(sockets[0], "regression", 2, 1024);
client.start(speed, at_ns);
const auto send_json = [&](const std::string& json) {
std::vector<std::uint8_t> bytes(8 + json.size());
write_u32(bytes, 4, static_cast<std::uint32_t>(json.size()));
std::memcpy(bytes.data() + 8, json.data(), json.size());
const auto sent = ::send(sockets[1], bytes.data(), bytes.size(), MSG_NOSIGNAL);
if (sent != static_cast<ssize_t>(bytes.size())) {
throw std::runtime_error(std::string("send test packet failed: ") + std::strerror(errno));
}
};
send_json("{\"vlc\":1e400}");
send_json("{\"vlc\":1e308}");
std::this_thread::sleep_for(std::chrono::milliseconds(50));
require(at_ns.load() == 0 && speed.load() == 0, "invalid socket speed reached the processor");
send_json("{\"vlc\":3.5}");
const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
while (at_ns.load() == 0 && std::chrono::steady_clock::now() < deadline) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
require(speed.load() == 3.5 && at_ns.load() > 0, "locator stopped after oversized JSON number");
client.request_stop();
client.join();
::close(sockets[1]);
require(client.has_exited(), "locator session did not finish both loops");
locator::LocatorServerConfig config{.host = "127.0.0.1", .port = 0};
locator::TcpServer server(config);
for (unsigned iteration = 0; iteration < 30; ++iteration) { server.start(); server.stop(); }
}
void test_tty_sweep_overflow(const Workspace& workspace) {
TtyProtocolWriter writer((workspace.path / "overflow_tty").string(), 32'096);
const int fd = ::open(writer.path().c_str(), O_RDONLY | O_NONBLOCK);
require(fd >= 0, "Cannot open test PTY");
std::vector<std::uint16_t> words(8'000, 42);
for (std::uint64_t id = 1; id <= 100; ++id) {
writer.enqueue_sweep(words, {id, id, id * 100, id * 100 + 50, 1, 2, false});
}
require(writer.stats().ring_overflows > 0, "test did not overflow the queue");
std::vector<std::uint8_t> received;
std::uint64_t last_id = 0;
const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
while (last_id < 100 && std::chrono::steady_clock::now() < deadline) {
std::array<std::uint8_t, 8192> chunk{};
const auto n = ::read(fd, chunk.data(), chunk.size());
if (n > 0) { received.insert(received.end(), chunk.begin(), chunk.begin() + n); }
while (received.size() >= 16'048) {
require(std::memcmp(received.data(), "RPH1", 4) == 0, "overflow truncated a sweep header");
const auto value = [&](std::size_t at, unsigned size) {
std::uint64_t result = 0;
for (unsigned i = 0; i < size; ++i) { result |= std::uint64_t{received[at + i]} << (8U * i); }
return result;
};
require(value(4, 4) == 16'000 && value(8, 8) > last_id, "lost framing or source order");
last_id = value(8, 8);
require(value(16, 8) == last_id && value(24, 8) == last_id * 100 &&
value(32, 8) == last_id * 100 + 50 && value(40, 2) == 1 && value(42, 2) == 2,
"sweep lost source identity, timing, or combo");
require(received[48] == 42 && received[16'047] == 0, "corrupt point payload");
received.erase(received.begin(), received.begin() + 16'048);
}
if (n <= 0) { std::this_thread::sleep_for(std::chrono::milliseconds(1)); }
}
::close(fd);
require(last_id == 100 && received.empty(), "latest complete sweep did not arrive");
}
void test_switch_backlog() {
drivers::H7992MinimalDriverSettings settings{.mode = config::DriverMode::Mock};
rph::SwitchSequencer sequencer({.combos = {{0, 0}, {0, 1}}},
std::make_unique<drivers::H7992MinimalDriver>(settings),
std::make_unique<drivers::H7992MinimalDriver>(settings));
sequencer.open();
sequencer.session_started();
require(!sequencer.on_sweep_end(false), "duration limit requested a new session");
require(sequencer.current_combo().output_pos == 0 && sequencer.cycle_id() == 1,
"duration limit switched hardware before the physical sweep ended");
require(sequencer.on_sweep_end(), "GPIO change must request a new source session");
for (unsigned i = 0; i < 10; ++i) {
require(sequencer.current_dirty(), "queued pre-switch sweep incorrectly clean");
require(!sequencer.on_sweep_end(), "dirty backlog requested another session");
require(sequencer.current_combo().output_pos == 1 && sequencer.cycle_id() == 1,
"dirty backlog advanced the combo or cycle");
}
sequencer.session_started();
require(!sequencer.current_dirty(), "new source session must permit a complete sweep");
sequencer.on_sweep_end();
require(sequencer.current_combo().output_pos == 0 && sequencer.cycle_id() == 2,
"clean cycle did not advance");
rph::SwitchSequencer single({.combos = {{0, 0}}},
std::make_unique<drivers::H7992MinimalDriver>(settings),
std::make_unique<drivers::H7992MinimalDriver>(settings));
single.open(); single.session_started();
require(!single.on_sweep_end() && single.cycle_id() == 2 && !single.current_dirty(),
"unchanged combination restarted its source or stopped advancing cycles");
}
void test_unread_tty_shutdown(const Workspace& workspace) {
TtyProtocolWriter writer((workspace.path / "tty").string(), 160'048);
const std::vector<std::uint16_t> words(80'000, 42);
writer.enqueue_sweep(words, {1, 1, 100, 200});
std::this_thread::sleep_for(std::chrono::milliseconds(50));
const auto start = std::chrono::steady_clock::now();
writer.shutdown();
require(std::chrono::steady_clock::now() - start < std::chrono::seconds(1),
"unread tty blocked shutdown");
writer.shutdown();
}
} // namespace
int main(int argc, char** argv) {
try {
require(argc == 2, "Pass the simulator config fixture path");
Workspace workspace;
test_wire_format();
test_configuration(argv[1], workspace);
test_asset_reload_preserves_data(workspace);
test_object_calibration(workspace);
test_object_calibration_pipeline(workspace);
test_ring();
test_acquisition_identity();
test_interruptible_wait();
test_cosine_notch_taper();
test_tty_sweep_overflow(workspace);
test_switch_backlog();
test_unread_tty_shutdown(workspace);
test_locator_packet_errors();
std::cout << "native regressions passed (10,000 decoder mutations; 20,000 concurrent ring frames)\n";
} catch (const std::exception& error) {
std::cerr << "native regression failed: " << error.what() << '\n';
return 1;
}
}