Jungfraujoch/tests/JFJochStateMachineTest.cpp

// SPDX-FileCopyrightText: 2024 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
// SPDX-License-Identifier: GPL-3.0-only

#include <catch2/catch_all.hpp>
#include "../broker/JFJochStateMachine.h"
#include "../acquisition_device/HLSSimulatedDevice.h"
#include "../receiver/JFJochReceiverService.h"
#include "../broker/OpenAPIConvert.h"
#include "../broker/ModulePlacementConvert.h"
#include "Dataset_settings.h"
#include <nlohmann/json.hpp>

using namespace std::literals::chrono_literals;

TEST_CASE("JFJochStateMachine_States") {
    Logger logger("JFJochStateMachine_States");
    JFJochServices services(logger);
    DiffractionExperiment experiment;
    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF4M());

    DatasetSettings setup;

    REQUIRE(state_machine.GetStatus().state == JFJochState::Inactive);

    REQUIRE_THROWS(state_machine.Start(setup));
    REQUIRE_THROWS(state_machine.Pedestal());

    REQUIRE_NOTHROW(state_machine.Initialize());
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
    REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);

    REQUIRE_NOTHROW(state_machine.Deactivate());
    REQUIRE(state_machine.GetStatus().state == JFJochState::Inactive);
}

TEST_CASE("JFJochStateMachine_Initialize_DisablesRugnux") {
    Logger logger("JFJochStateMachine_Initialize_DisablesRugnux");
    JFJochServices services(logger);
    DiffractionExperiment experiment;
    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF4M());

    state_machine.SetRugnuxLive(true, 4, 30.0);
    REQUIRE(state_machine.GetRugnuxLive().enable);

    REQUIRE_NOTHROW(state_machine.Initialize());
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());

    const auto s = state_machine.GetRugnuxLive();
    CHECK(!s.enable);
    CHECK(s.model == nullptr);
    CHECK(s.model_size_bytes == 0);
}

TEST_CASE("JFJochStateMachine_State_Pedestal") {
    Logger logger("JFJochStateMachine_State_Pedestal");
    JFJochServices services(logger);
    DiffractionExperiment experiment;
    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF4M());

    DatasetSettings setup;

    state_machine.DebugOnly_SetState(JFJochState::Calibration);

    REQUIRE(state_machine.GetStatus().state == JFJochState::Calibration);

    REQUIRE_THROWS(state_machine.Start(setup));
    REQUIRE_THROWS(state_machine.Pedestal());
    REQUIRE_THROWS(state_machine.Initialize());
    REQUIRE(state_machine.WaitTillMeasurementDone(std::chrono::milliseconds(1)).state == JFJochState::Calibration);
}

TEST_CASE("JFJochStateMachine_State_Measure") {
    Logger logger("JFJochStateMachine_State_Measure");
    JFJochServices services(logger);
    DiffractionExperiment experiment;
    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF4M());

    DatasetSettings setup;

    state_machine.DebugOnly_SetState(JFJochState::Measuring);

    REQUIRE(state_machine.GetStatus().state == JFJochState::Measuring);

    REQUIRE_THROWS(state_machine.Start(setup));
    REQUIRE_THROWS(state_machine.Pedestal());
    REQUIRE_THROWS(state_machine.Initialize());
    REQUIRE(state_machine.WaitTillMeasurementDone(std::chrono::milliseconds(1)).state == JFJochState::Measuring);

    DetectorSettings settings{};
    REQUIRE_THROWS(state_machine.LoadDetectorSettings(settings));
}

TEST_CASE("JFJochStateMachine_State_Error") {
    Logger logger("JFJochStateMachine_State_Error");
    JFJochServices services(logger);
    DiffractionExperiment experiment;
    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF4M());

    DatasetSettings setup;

    state_machine.DebugOnly_SetState(JFJochState::Error,
        "msg1234",
        BrokerStatus::MessageSeverity::Error);

    REQUIRE(state_machine.GetStatus().state == JFJochState::Error);
    REQUIRE(state_machine.GetStatus().message.has_value());
    REQUIRE(state_machine.GetStatus().message == "msg1234");
    REQUIRE(state_machine.GetStatus().message_severity == BrokerStatus::MessageSeverity::Error);

    state_machine.DebugOnly_SetState(JFJochState::Inactive, "msg3456", BrokerStatus::MessageSeverity::Info);
    REQUIRE(state_machine.GetStatus().state == JFJochState::Inactive);
    REQUIRE(state_machine.GetStatus().message.has_value());
    REQUIRE(state_machine.GetStatus().message == "msg3456");
    REQUIRE(state_machine.GetStatus().message_severity == BrokerStatus::MessageSeverity::Info);

    state_machine.DebugOnly_SetState(JFJochState::Error);
    REQUIRE(state_machine.GetStatus().state == JFJochState::Error);
    REQUIRE(!state_machine.GetStatus().message.has_value());

    REQUIRE_THROWS(state_machine.Start(setup));
    REQUIRE_THROWS(state_machine.Pedestal());

    REQUIRE(state_machine.WaitTillMeasurementDone(std::chrono::milliseconds(1)).state == JFJochState::Error);

    DetectorSettings settings;
    REQUIRE_NOTHROW(state_machine.LoadDetectorSettings(settings));

    REQUIRE_NOTHROW(state_machine.Initialize());
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
    REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
}

TEST_CASE("JFJochStateMachine_NoDetectorSetup") {
    Logger logger("JFJochStateMachine_NoDetectorSetup");
    JFJochServices services(logger);
    DiffractionExperiment experiment;
    JFJochStateMachine state_machine(experiment, services, logger);
    REQUIRE_THROWS(state_machine.Initialize());
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
}

TEST_CASE("JFJochStateMachine_AddDetectorSetup") {
    Logger logger("JFJochStateMachine_AddDetectorSetup");
    JFJochServices services(logger);
    DiffractionExperiment experiment;
    JFJochStateMachine state_machine(experiment, services, logger);
    DetectorSetup setup = DetJF4M();
    state_machine.AddDetectorSetup(setup);
    REQUIRE_NOTHROW(state_machine.Initialize());
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
}

TEST_CASE("JFJochStateMachine_AddDetectorSetup_Gain") {
    Logger logger("JFJochStateMachine_AddDetectorSetup_Gain");
    JFJochServices services(logger);
    DiffractionExperiment experiment;
    JFJochStateMachine state_machine(experiment, services, logger);
    DetectorSetup setup = DetJF(4);
    setup.LoadGain({"../../tests/test_data/gainMaps_M049.bin",
                    "../../tests/test_data/gainMaps_M049.bin",
                    "../../tests/test_data/gainMaps_M049.bin",
                    "../../tests/test_data/gainMaps_M049.bin"});
    state_machine.AddDetectorSetup(setup);
    REQUIRE_NOTHROW(state_machine.Initialize());
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
}

TEST_CASE("JFJochStateMachine_AddDetectorSetup_Multiple") {
    Logger logger("JFJochStateMachine_AddDetectorSetup_Multiple");
    JFJochServices services(logger);
    DiffractionExperiment experiment;

    JFJochStateMachine state_machine(experiment, services, logger);
    REQUIRE_NOTHROW(state_machine.AddDetectorSetup(DetJF(4, 1, 0,0,false,"Det1", {"mx1", "mx2", "mx3", "mx4"})));
    REQUIRE_NOTHROW(state_machine.AddDetectorSetup(DetEIGER(2, 1, 0, 0, false, "Det2", {"mx1", "mx2", "mx3", "mx4"})));
    REQUIRE_NOTHROW(state_machine.AddDetectorSetup(DetDECTRIS(23,45, "Det3", {"mx5"})));


    auto dl = state_machine.GetDetectorsList();
    REQUIRE(dl.detector.size() == 3);
    REQUIRE(dl.detector[0].description == "Det1");
    REQUIRE(dl.detector[0].nmodules == 4);
    REQUIRE(dl.detector[0].detector_type == DetectorType::JUNGFRAU);
    REQUIRE(dl.detector[1].description == "Det2");
    REQUIRE(dl.detector[1].nmodules == 2);
    REQUIRE(dl.detector[1].detector_type == DetectorType::EIGER);
    REQUIRE(dl.detector[2].description == "Det3");
    REQUIRE(dl.detector[2].nmodules == 1);
    REQUIRE(dl.detector[2].detector_type == DetectorType::DECTRIS);

    REQUIRE_NOTHROW(state_machine.Initialize());
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
    REQUIRE(state_machine.Experiment().GetModulesNum() == 4);
    REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);

    REQUIRE_THROWS(state_machine.SelectDetector(7));
    REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);

    REQUIRE_NOTHROW(state_machine.SelectDetector(2));
    REQUIRE(state_machine.Experiment().GetModulesNum() == 1);
    REQUIRE(state_machine.Experiment().GetDetectorDescription() == "Det3");
    REQUIRE(state_machine.GetStatus().state == JFJochState::Inactive);
}

TEST_CASE("JFJochStateMachine_LoadDetectorSettings_Error") {
    Logger logger("JFJochStateMachine_LoadDetectorSettings_Error");
    JFJochServices services(logger);
    DiffractionExperiment experiment;
    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF4M());

    DatasetSettings setup;

    state_machine.DebugOnly_SetState(JFJochState::Idle);

    DetectorSettings settings;
    settings.FrameTime(std::chrono::microseconds(1));

    REQUIRE_THROWS(state_machine.LoadDetectorSettings(settings));
    REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
}

namespace {
    // Stands in for a writer that refuses to start the run - e.g. the output file exists and cannot be
    // overwritten. The refusal reaches the broker as an ordinary exception thrown while the start
    // message is being sent, which is what separates it from a critical detector fault.
    class RefusingImagePusher : public ImagePusher {
    public:
        bool refuse = true;
        void StartDataCollection(StartMessage &) override {
            if (refuse)
                throw JFJochException(JFJochExceptionCategory::InputParameterInvalid, "writer_refused_1234");
        }
        bool EndDataCollection(const EndMessage &) override { return true; }
        bool SendImage(const uint8_t *, size_t, int64_t) override { return true; }
        bool SendCalibration(const CompressedImage &) override { return true; }
        std::string PrintSetup() const override { return "RefusingImagePusher"; }
        ImagePusherType GetType() const override { return ImagePusherType::Test; }
    };
}

// An asynchronous start returns before the measurement thread runs, so a failure there has to be
// reported to whoever asks next. Without this the state is a plain Idle, indistinguishable from a
// timeout, and the writer's message is lost.
TEST_CASE("JFJochStateMachine_AsyncStartFailure") {
    Logger logger("JFJochStateMachine_AsyncStartFailure");

    DiffractionExperiment experiment(DetJF(2));
    experiment.Conversion().PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true)
            .ImagesPerTrigger(4).IncidentEnergy_keV(12.4);

    AcquisitionDeviceGroup aq_devices;
    for (int i = 0; i < experiment.GetDataStreamsNum(); i++)
        aq_devices.Add(std::make_unique<HLSSimulatedDevice>(i, 64));

    RefusingImagePusher pusher;
    JFJochReceiverService receiver_service(aq_devices, logger, pusher);

    JFJochServices services(logger);
    services.Receiver(&receiver_service);

    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF(2));
    state_machine.DebugOnly_SetState(JFJochState::Idle);

    DatasetSettings setup;
    setup.ImagesPerTrigger(4).NumTriggers(1);

    // The asynchronous start itself succeeds - it only launches the measurement thread.
    REQUIRE_NOTHROW(state_machine.Start(setup, true));

    // Both wait functions report the failure, and repeatedly: the exception_ptr is not consumed.
    REQUIRE_THROWS_WITH(state_machine.WaitTillNotBusy(std::chrono::seconds(10)),
                        Catch::Matchers::ContainsSubstring("writer_refused_1234"));
    REQUIRE_THROWS_WITH(state_machine.WaitTillNotBusy(std::chrono::seconds(10)),
                        Catch::Matchers::ContainsSubstring("writer_refused_1234"));
    REQUIRE_THROWS_WITH(state_machine.WaitTillMeasurementDone(std::chrono::seconds(10)),
                        Catch::Matchers::ContainsSubstring("writer_refused_1234"));

    // An ordinary failure leaves the detector usable, so the run can be retried from Idle. The
    // status keeps the writer's message, and nothing is left of the run that never happened - a
    // progress figure in particular would show up in the frontend as a stalled acquisition.
    auto status = state_machine.GetStatus();
    REQUIRE(status.state == JFJochState::Idle);
    REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Error);
    REQUIRE_THAT(status.message.value_or(""), Catch::Matchers::ContainsSubstring("writer_refused_1234"));
    REQUIRE_FALSE(status.progress.has_value());

    auto statistics = state_machine.GetMeasurementStatistics();
    REQUIRE(statistics.has_value());
    REQUIRE(statistics->images_collected == 0);
    REQUIRE(statistics->images_sent == 0);
    REQUIRE_FALSE(statistics->collection_efficiency.has_value());

    // A synchronous start reports the same failure directly, and leaves it visible to the wait
    // functions afterwards.
    REQUIRE_THROWS_WITH(state_machine.Start(setup),
                        Catch::Matchers::ContainsSubstring("writer_refused_1234"));
    REQUIRE_THROWS_WITH(state_machine.WaitTillNotBusy(std::chrono::seconds(10)),
                        Catch::Matchers::ContainsSubstring("writer_refused_1234"));

    // A start rejected before the measurement thread is launched must not leave the previous
    // failure behind for the next wait call to report.
    DatasetSettings rejected_setup;
    rejected_setup.ImagesPerTrigger(4).NumTriggers(1)
            .ImageTime(state_machine.Experiment().GetFrameTime() + std::chrono::nanoseconds(1));
    REQUIRE_THROWS(state_machine.Start(rejected_setup, true));
    REQUIRE_NOTHROW(state_machine.WaitTillNotBusy(std::chrono::seconds(10)));

    // Any other operation started from Idle supersedes the pending failure - it must not be
    // attributed to the pedestal, which is allowed from Idle and has nothing to do with it.
    REQUIRE_THROWS_WITH(state_machine.Start(setup),
                        Catch::Matchers::ContainsSubstring("writer_refused_1234"));
    REQUIRE_NOTHROW(state_machine.Pedestal());
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());

    // Nothing is left half-started: once the writer accepts the run, the next start goes through
    // and completes without any intervening re-initialisation.
    pusher.refuse = false;
    REQUIRE_NOTHROW(state_machine.Start(setup));
    REQUIRE(state_machine.GetStatus().state == JFJochState::Measuring);
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
    REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
    REQUIRE(state_machine.GetMeasurementStatistics()->images_sent == 4);

    // Re-initialising discards a pending failure too, so a detector brought back up does not report
    // the failure of the run before it.
    pusher.refuse = true;
    REQUIRE_THROWS_WITH(state_machine.Start(setup),
                        Catch::Matchers::ContainsSubstring("writer_refused_1234"));
    services.Receiver(nullptr);
    REQUIRE_NOTHROW(state_machine.Initialize());
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
    REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
}

// A calibration that does not complete must not be reported as a good one. Every step signals
// failure by calling SetState and returning normally, so nothing threw and the sequence used to
// finish with an unconditional "Calibration sequence done"/Success on top of it - leaving the
// broker Idle and apparently ready while holding partial pedestals.
TEST_CASE("JFJochStateMachine_CalibrationFailure") {
    Logger logger("JFJochStateMachine_CalibrationFailure");
    JFJochServices services(logger);
    DiffractionExperiment experiment;
    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF4M());

    // Ask for a G0 pedestal. There is no receiver, so no frames come back and the import fails -
    // the same outcome as a pedestal that collected nothing on real hardware.
    DetectorSettings settings = state_machine.GetDetectorSettings();
    settings.PedestalG0Frames(100);
    REQUIRE_NOTHROW(state_machine.LoadDetectorSettings(settings));

    REQUIRE_NOTHROW(state_machine.Initialize());
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());

    // Error, not Idle: the calibration is undefined, so the detector has to be initialized again
    // rather than looking ready to measure. Error and not Inactive, so that the wait calls report
    // the reason instead of the bodiless answer a deliberate power-off leaves behind.
    auto status = state_machine.GetStatus();
    REQUIRE(status.state == JFJochState::Error);
    REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Error);
    REQUIRE(status.message == "Pedestal not collected properly");

    // ... and a data collection is refused on it, instead of producing mis-converted images.
    DatasetSettings setup;
    REQUIRE_THROWS(state_machine.Start(setup));

    // The success path is unchanged - with no pedestal to take, the sequence still reports success.
    JFJochStateMachine calibrated(experiment, services, logger);
    calibrated.AddDetectorSetup(DetJF4M());
    REQUIRE_NOTHROW(calibrated.Initialize());
    REQUIRE_NOTHROW(calibrated.WaitTillMeasurementDone());
    REQUIRE(calibrated.GetStatus().state == JFJochState::Idle);
    REQUIRE(calibrated.GetStatus().message_severity == BrokerStatus::MessageSeverity::Success);
}

namespace {
    // Holds the run open until the test lets go, so a cancel lands while the collection is running
    // instead of racing its end.
    class GatedPusher : public ImagePusher {
        std::atomic<bool> released{true};
    public:
        void Hold() { released = false; }
        void Release() { released = true; }
        void StartDataCollection(StartMessage &) override {}
        bool EndDataCollection(const EndMessage &) override { return true; }
        bool SendImage(const uint8_t *, size_t, int64_t) override {
            while (!released)
                std::this_thread::sleep_for(1ms);
            return true;
        }
        bool SendCalibration(const CompressedImage &) override { return true; }
        std::string PrintSetup() const override { return "GatedPusher"; }
        ImagePusherType GetType() const override { return ImagePusherType::Test; }
    };

    // A writer that cannot finish the run - a rename that collides, a disk that filled up. It
    // reaches the state machine from Stop(), not from Start(), which is the case that used to be
    // reported as a run without problems.
    class FinalizeFailingPusher : public ImagePusher {
    public:
        void StartDataCollection(StartMessage &) override {}
        bool EndDataCollection(const EndMessage &) override { return true; }
        bool SendImage(const uint8_t *, size_t, int64_t) override { return true; }
        bool SendCalibration(const CompressedImage &) override { return true; }
        std::string Finalize() override {
            throw JFJochException(JFJochExceptionCategory::FileWriteError, "writer_finalize_failed_9876");
        }
        std::string PrintSetup() const override { return "FinalizeFailingPusher"; }
        ImagePusherType GetType() const override { return ImagePusherType::Test; }
    };
}

// /wait_till_done reads the severity of the end-of-run message to decide between 200 and 500, so
// what the state machine puts there is what a script sees. A cancelled collection must not be an
// error: it is incomplete by definition - the packets it never received would otherwise report
// themselves - and calling every cancel a failure would make the endpoint useless.
TEST_CASE("JFJochStateMachine_CancelledRunIsNotAnError") {
    Logger logger("JFJochStateMachine_CancelledRunIsNotAnError");

    DiffractionExperiment experiment(DetJF(2));
    experiment.Conversion().PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true)
            .ImagesPerTrigger(8).IncidentEnergy_keV(12.4);

    AcquisitionDeviceGroup aq_devices;
    for (int i = 0; i < experiment.GetDataStreamsNum(); i++)
        aq_devices.Add(std::make_unique<HLSSimulatedDevice>(i, 64));

    GatedPusher pusher;
    JFJochReceiverService receiver_service(aq_devices, logger, pusher);

    JFJochServices services(logger);
    services.Receiver(&receiver_service);

    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF(2));
    state_machine.DebugOnly_SetState(JFJochState::Idle);

    DatasetSettings setup;
    setup.ImagesPerTrigger(8).NumTriggers(1);

    // A run that finished properly is a Success, so /wait_till_done answers 200.
    REQUIRE_NOTHROW(state_machine.Start(setup));
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
    REQUIRE(state_machine.GetStatus().state == JFJochState::Idle);
    REQUIRE(state_machine.GetStatus().message_severity == BrokerStatus::MessageSeverity::Success);

    // The same run cancelled halfway. The gate keeps it running until the cancel is in.
    pusher.Hold();
    REQUIRE_NOTHROW(state_machine.Start(setup, true));

    for (int i = 0; i < 400 && state_machine.GetStatus().state != JFJochState::Measuring; i++)
        std::this_thread::sleep_for(25ms);
    REQUIRE(state_machine.GetStatus().state == JFJochState::Measuring);

    state_machine.Cancel();
    pusher.Release();

    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());
    auto status = state_machine.GetStatus();
    REQUIRE(status.state == JFJochState::Idle);
    // Info, not Error and not Warning: the collection did what it was told to do.
    REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Info);
    REQUIRE_THAT(status.message.value_or(""), Catch::Matchers::ContainsSubstring("cancelled"));

    // The run really was short - what makes this test worth having is that the incompleteness did
    // not get reported ahead of the cancel.
    REQUIRE(state_machine.GetMeasurementStatistics()->cancelled);
}

// A writer that failed while the run was being finalized arrives from Stop(), which leaves the
// state Idle - the detector is still usable - and says so only in the severity. /wait_till_done
// turns that into a 500 with the writer's message rather than reporting a good collection.
TEST_CASE("JFJochStateMachine_WriterFailureAtEndIsAnError") {
    Logger logger("JFJochStateMachine_WriterFailureAtEndIsAnError");

    DiffractionExperiment experiment(DetJF(2));
    experiment.Conversion().PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true)
            .ImagesPerTrigger(4).IncidentEnergy_keV(12.4);

    AcquisitionDeviceGroup aq_devices;
    for (int i = 0; i < experiment.GetDataStreamsNum(); i++)
        aq_devices.Add(std::make_unique<HLSSimulatedDevice>(i, 64));

    FinalizeFailingPusher pusher;
    JFJochReceiverService receiver_service(aq_devices, logger, pusher);

    JFJochServices services(logger);
    services.Receiver(&receiver_service);

    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF(2));
    state_machine.DebugOnly_SetState(JFJochState::Idle);

    DatasetSettings setup;
    setup.ImagesPerTrigger(4).NumTriggers(1);

    REQUIRE_NOTHROW(state_machine.Start(setup, true));
    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone());

    auto status = state_machine.GetStatus();
    REQUIRE(status.state == JFJochState::Idle);
    REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Error);
    REQUIRE_THAT(status.message.value_or(""),
                 Catch::Matchers::ContainsSubstring("writer_finalize_failed_9876"));
}

namespace {
    // Refuses the run before it starts, the way a writer refuses to overwrite a file that is
    // already there. Records whether the collection was started anyway.
    class PreflightRefusingPusher : public ImagePusher {
    public:
        std::atomic<bool> start_called{false};
        void Preflight(StartMessage &) override {
            throw JFJochException(JFJochExceptionCategory::FileWriteError, "preflight_refused_2468");
        }
        void StartDataCollection(StartMessage &) override { start_called = true; }
        bool EndDataCollection(const EndMessage &) override { return true; }
        bool SendImage(const uint8_t *, size_t, int64_t) override { return true; }
        bool SendCalibration(const CompressedImage &) override { return true; }
        std::string PrintSetup() const override { return "PreflightRefusingPusher"; }
        ImagePusherType GetType() const override { return ImagePusherType::Test; }
    };
}

// The pre-flight is the first thing a measurement does, so a run the writer will not accept is
// refused before the receiver is built and before the detector is armed - which is what makes the
// refusal free: there is no series left armed and waiting for a trigger to tear down.
TEST_CASE("JFJochStateMachine_PreflightRefusesBeforeStarting") {
    Logger logger("JFJochStateMachine_PreflightRefusesBeforeStarting");

    DiffractionExperiment experiment(DetJF(2));
    experiment.Conversion().PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true)
            .ImagesPerTrigger(4).IncidentEnergy_keV(12.4).FilePrefix("preflight_refused");

    AcquisitionDeviceGroup aq_devices;
    for (int i = 0; i < experiment.GetDataStreamsNum(); i++)
        aq_devices.Add(std::make_unique<HLSSimulatedDevice>(i, 64));

    PreflightRefusingPusher pusher;
    JFJochReceiverService receiver_service(aq_devices, logger, pusher);

    JFJochServices services(logger);
    services.Receiver(&receiver_service);

    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF(2));
    state_machine.DebugOnly_SetState(JFJochState::Idle);

    DatasetSettings setup;
    setup.ImagesPerTrigger(4).NumTriggers(1).FilePrefix("preflight_refused");

    // A synchronous start reports the writer's own message.
    REQUIRE_THROWS_WITH(state_machine.Start(setup),
                        Catch::Matchers::ContainsSubstring("preflight_refused_2468"));
    REQUIRE_FALSE(pusher.start_called);

    // An ordinary failure, so the detector stays usable and the run can be retried from Idle.
    auto status = state_machine.GetStatus();
    REQUIRE(status.state == JFJochState::Idle);
    REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Error);

    // The asynchronous start returns, and the wait functions report it.
    REQUIRE_NOTHROW(state_machine.Start(setup, true));
    REQUIRE_THROWS_WITH(state_machine.WaitTillNotBusy(10s),
                        Catch::Matchers::ContainsSubstring("preflight_refused_2468"));
    REQUIRE_THROWS_WITH(state_machine.WaitTillMeasurementDone(10s),
                        Catch::Matchers::ContainsSubstring("preflight_refused_2468"));
    REQUIRE_FALSE(pusher.start_called);
}

namespace {
    class NullPusher : public ImagePusher {
    public:
        void StartDataCollection(StartMessage &) override {}
        bool EndDataCollection(const EndMessage &) override { return true; }
        bool SendImage(const uint8_t *, size_t, int64_t) override { return true; }
        bool SendCalibration(const CompressedImage &) override { return true; }
        std::string PrintSetup() const override { return "NullPusher"; }
        ImagePusherType GetType() const override { return ImagePusherType::Test; }
    };
}

// Cancelling the dark-mask collection - the window between the receiver being started and Stop()
// returning, which is where a calibration spends its time - must abandon the sequence. The mask
// analysis always hands back a full-size array, every pixel that saw no frame reading as good, so
// the size check the sequence relies on accepts a mask measured on nothing; and the dark mask is a
// single step, so no later cancel_sequence check catches it. The calibration used to end Idle with
// "Calibration sequence done" on a mask it never measured.
TEST_CASE("JFJochStateMachine_DarkMaskCancelIsNotASuccess") {
    Logger logger("JFJochStateMachine_DarkMaskCancelIsNotASuccess");

    DarkMaskSettings mask_settings;
    mask_settings.NumberOfFrames(100).MaxCounts(1).MaxFramesWithCounts(5);

    // Nothing listens on the stream address, so no start message ever arrives and the collection
    // stays in the window under test until it is cancelled.
    DiffractionExperiment experiment(DetDECTRIS(1024, 1024, "Test", "127.0.0.1"));
    experiment.ImportDarkMaskSettings(mask_settings);

    AcquisitionDeviceGroup aq_devices;
    NullPusher pusher;
    JFJochReceiverService receiver_service(aq_devices, logger, pusher);

    JFJochServices services(logger);
    services.Receiver(&receiver_service);

    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetDECTRIS(1024, 1024, "Test", "127.0.0.1"));
    state_machine.DebugOnly_SetState(JFJochState::Idle);

    REQUIRE_NOTHROW(state_machine.Pedestal());

    for (int i = 0; i < 400 && state_machine.GetStatus().state != JFJochState::Calibration; i++)
        std::this_thread::sleep_for(25ms);
    REQUIRE(state_machine.GetStatus().state == JFJochState::Calibration);

    state_machine.Cancel();

    REQUIRE_NOTHROW(state_machine.WaitTillMeasurementDone(std::chrono::seconds(30)));
    auto status = state_machine.GetStatus();
    REQUIRE(status.state != JFJochState::Idle);
    REQUIRE(status.message_severity == BrokerStatus::MessageSeverity::Error);
    REQUIRE_THAT(status.message.value_or(""), Catch::Matchers::ContainsSubstring("cancelled"));
}

namespace {
    // Keeps the start message of the run and refuses it, so nothing is collected.
    class StartMessageCapture : public ImagePusher {
    public:
        StartMessage start;
        void Preflight(StartMessage &message) override {
            start = message;
            throw JFJochException(JFJochExceptionCategory::FileWriteError, "captured");
        }
        void StartDataCollection(StartMessage &) override {}
        bool EndDataCollection(const EndMessage &) override { return true; }
        bool SendImage(const uint8_t *, size_t, int64_t) override { return true; }
        bool SendCalibration(const CompressedImage &) override { return true; }
        std::string PrintSetup() const override { return "StartMessageCapture"; }
        ImagePusherType GetType() const override { return ImagePusherType::Test; }
    };
}

// A /start carrying a module placement (rugnux's <prefix>_module_placement.json) puts the modules where it says into
// the start message - and so into the NXmx module groups - whatever order it lists them in; one naming a
// module the detector does not have is refused.
TEST_CASE("JFJochStateMachine_StartModulePlacement") {
    Logger logger("JFJochStateMachine_StartModulePlacement");

    DiffractionExperiment experiment(DetJF(2));
    experiment.Conversion().PedestalG0Frames(0).NumTriggers(1).UseInternalPacketGenerator(true)
            .ImagesPerTrigger(4).IncidentEnergy_keV(12.4).FilePrefix("module_placement");

    AcquisitionDeviceGroup aq_devices;
    for (int i = 0; i < experiment.GetDataStreamsNum(); i++)
        aq_devices.Add(std::make_unique<HLSSimulatedDevice>(i, 64));
    StartMessageCapture pusher;
    JFJochReceiverService receiver_service(aq_devices, logger, pusher);
    JFJochServices services(logger);
    services.Receiver(&receiver_service);
    JFJochStateMachine state_machine(experiment, services, logger);
    state_machine.AddDetectorSetup(DetJF(2));
    state_machine.DebugOnly_SetState(JFJochState::Idle);

    const auto &geom = experiment.GetDetectorSetup().GetGeometry();
    const auto assembled = ModulesAsImaged(geom).Get();
    REQUIRE(assembled.size() == 2);
    // A module is named by its readout pixel 0 and axes in the stored image, after mirror_y.
    for (int64_t m = 0; m < 2; m++) {
        CHECK(PlacementOf(assembled[m]).x0 == geom.GetX0(m));
        CHECK(PlacementOf(assembled[m]).y0 == geom.GetY0(m));
    }
    // Module 1 moved by (+0.4, -1.3) px and turned slightly about its pixel 0, module 0 as assembled - listed 1 first.
    const auto entry = [](const DetectorModule &m, double dx, double dy, double turn) {
        const ModulePlacement p = PlacementOf(m);
        const auto turned = [turn](const std::array<double, 2> &v) {
            return std::vector<double>{std::cos(turn) * v[0] - std::sin(turn) * v[1],
                                       std::sin(turn) * v[0] + std::cos(turn) * v[1]};
        };
        return nlohmann::json{{"x0", p.x0}, {"y0", p.y0},
                              {"fast_axis", DirectionName(p.fast)}, {"slow_axis", DirectionName(p.slow)},
                              {"x0_plane", p.x0_plane + dx}, {"y0_plane", p.y0_plane + dy},
                              {"fast_step", turned(p.fast_step)}, {"slow_step", turned(p.slow_step)}};
    };
    nlohmann::json body = {{"beam_x_pxl", 500}, {"beam_y_pxl", 500}, {"detector_distance_mm", 100},
                           {"incident_energy_keV", 12.4}, {"images_per_trigger", 4},
                           {"file_prefix", "module_placement"}};
    body["module_placement"] = {entry(assembled[1], 0.4, -1.3, 0.001), entry(assembled[0], 0, 0, 0)};

    org::openapitools::server::model::Dataset_settings model;
    from_json(body, model);
    std::stringstream msg;
    REQUIRE(model.validate(msg));

    REQUIRE_THROWS_WITH(state_machine.Start(Convert(model)), Catch::Matchers::ContainsSubstring("captured"));
    const auto &modules = pusher.start.detector_modules;
    REQUIRE(modules.size() == 2);
    CHECK(modules[0].x0 == assembled[0].x0);
    CHECK(modules[0].ox == Catch::Approx(assembled[0].ox));
    CHECK(modules[1].x0 == assembled[1].x0);
    CHECK(modules[1].y0 == assembled[1].y0);
    CHECK(modules[1].fast == assembled[1].fast);
    CHECK(modules[1].slow == assembled[1].slow);
    const ModulePlacement placed = PlacementOf(modules[1]);
    CHECK(placed.x0_plane == Catch::Approx(PlacementOf(assembled[1]).x0_plane + 0.4));
    CHECK(placed.y0_plane == Catch::Approx(PlacementOf(assembled[1]).y0_plane - 1.3));
    CHECK(modules[1].fy == Catch::Approx(std::sin(0.001)));

    // Without it, the modules as assembled.
    body.erase("module_placement");
    model = {};
    from_json(body, model);
    REQUIRE_THROWS_WITH(state_machine.Start(Convert(model)), Catch::Matchers::ContainsSubstring("captured"));
    REQUIRE(pusher.start.detector_modules.size() == 2);
    CHECK(pusher.start.detector_modules[1].oy == Catch::Approx(assembled[1].oy));

    const auto refused = [&](const nlohmann::json &list, const std::string &why) {
        body["module_placement"] = list;
        model = {};
        from_json(body, model);
        REQUIRE_THROWS_WITH(state_machine.Start(Convert(model)), Catch::Matchers::ContainsSubstring(why));
    };
    // A module the detector does not have: pixel 0 one row off, or the readout axes the other way.
    auto off = entry(assembled[1], 0, 0, 0);
    off["y0"] = off["y0"].get<int64_t>() + 1;
    refused(nlohmann::json::array({off, entry(assembled[0], 0, 0, 0)}), "is not a module of this detector");
    auto reversed = entry(assembled[1], 0, 0, 0);
    reversed["slow_axis"] = reversed["slow_axis"] == "Yn" ? "Yp" : "Yn";
    refused(nlohmann::json::array({reversed, entry(assembled[0], 0, 0, 0)}), "is not a module of this detector");
    // One placed twice, one left out.
    refused(nlohmann::json::array({entry(assembled[0], 0, 0, 0), entry(assembled[0], 0, 0, 0)}), "is placed twice");
    refused(nlohmann::json::array({entry(assembled[0], 0, 0, 0)}), "is not placed");
}

// The module placement file rugnux writes is a dataset_settings fragment: merged into a /start body it
// parses as the broker parses it, and reads back to the same modules.
TEST_CASE("ModulePlacement_FileIsDatasetSettingsFragment") {
    DetectorModule a = DetectorModule::AsImaged(0, 0, 1030, 514);
    // A mirrored module: its readout pixel 0 is the top row of its box, its slow axis runs -y.
    DetectorModule b = DetectorModule::AsImaged(0, 550, 1030, 514);
    b.slow = DetectorModuleGeometry::Direction::Yneg;
    b.ox += 0.25;
    b.oy -= 1.5;
    b.fx = std::cos(0.002);
    b.fy = std::sin(0.002);
    b.sx = -std::sin(0.002);
    b.sy = std::cos(0.002);
    const std::string text = ModulePlacementJson(DetectorModules({a, b}));
    CHECK(text.find(R"("slow_axis": "Yn")") != std::string::npos);

    const auto back = ParseModulePlacement(text);
    REQUIRE(back.size() == 2);
    CHECK(back[0].x0 == 0);
    CHECK(back[0].y0 == 0);
    CHECK(back[0].x0_plane == 0.0);
    CHECK(back[1].x0 == 0);
    CHECK(back[1].y0 == 550 + 513);
    CHECK(back[1].fast == DetectorModuleGeometry::Direction::Xpos);
    CHECK(back[1].slow == DetectorModuleGeometry::Direction::Yneg);
    CHECK(back[1].x0_plane == Catch::Approx(b.ox + 513 * b.sx));
    CHECK(back[1].y0_plane == Catch::Approx(b.oy + 513 * b.sy));
    CHECK(back[1].fast_step[1] == Catch::Approx(b.fy));
    CHECK(back[1].slow_step[0] == Catch::Approx(-b.sx));
    CHECK(back[1].slow_step[1] == Catch::Approx(-b.sy));
    // Read back onto the same detector, the modules are where they were.
    DetectorModule b_assembled = DetectorModule::AsImaged(0, 550, 1030, 514);
    b_assembled.slow = DetectorModuleGeometry::Direction::Yneg;
    const auto placed = ModulesPlaced(DetectorModules({b_assembled, DetectorModule::AsImaged(0, 0, 1030, 514)}), back);
    CHECK(placed[0].ox == Catch::Approx(b.ox));
    CHECK(placed[0].oy == Catch::Approx(b.oy));
    CHECK(placed[0].sx == Catch::Approx(b.sx));
    CHECK(placed[0].sy == Catch::Approx(b.sy));

    nlohmann::json body = {{"beam_x_pxl", 500}, {"beam_y_pxl", 500}, {"detector_distance_mm", 100},
                           {"incident_energy_keV", 12.4}, {"images_per_trigger", 1}};
    body.update(nlohmann::json::parse(text));
    org::openapitools::server::model::Dataset_settings model;
    from_json(body, model);
    std::stringstream msg;
    REQUIRE(model.validate(msg));
    const auto modules = Convert(model).GetModulePlacement();
    REQUIRE(modules.size() == 2);
    CHECK(modules[1].y0 == back[1].y0);
    CHECK(modules[1].slow == DetectorModuleGeometry::Direction::Yneg);
    CHECK(modules[1].y0_plane == back[1].y0_plane);

    // The keys of the earlier box shape are not a placement, nor is a step of three numbers.
    CHECK_THROWS(ParseModulePlacement(R"({"module_placement": [{"box_x": 0, "box_y": 0, "box_width": 1030,
        "box_height": 514, "plane_x": 0, "plane_y": 0, "x_step_x": 1, "x_step_y": 0, "y_step_x": 0, "y_step_y": 1}]})"));
    CHECK_THROWS(ParseModulePlacement(R"({"module_placement": [{"x0": 0, "y0": 0, "fast_axis": "Xp", "slow_axis": "Yp",
        "x0_plane": 0, "y0_plane": 0, "fast_step": [1, 0, 0], "slow_step": [0, 1]}]})"));
}