From 2fac5fcec73e4570ae03cb39c41a024140a18c52 Mon Sep 17 00:00:00 2001 From: Tiago Quintino Date: Sun, 18 Oct 2026 21:34:38 +0100 Subject: [PATCH 1/2] statistics-mtg2: accumulate the accumulated parameters under their own paramId Add typeOfStatisticalProcessing=1 (accumulate) mappings with paramIn equal to paramOut for 19 accumulated parameters, so that statistics-mtg2 with strict mapping can accumulate per-step increments sent under the paramId of the accumulation: water, kg m**-2: 228228, 228143, 3062, 228144, 260259, 231002, 231010, 231012 energy, J m**-2: 146, 147, 176, 177, 169, 175, 178, 179, 212 stress, N m**-2 s: 180, 181 Without the entries the action throws "Mapping for param=228143 and typeOfStatisticalProcessing=1 is undefined!". Values are not converted: the increments must already be in the units of the paramId. The entries are appended as manually added mappings, since the generator does not produce them. Each id was checked against the parameter database: an accumulation (typeOfStatisticalProcessing 1) with the units above. Tests: ParamMapping covers the 19 entries with strict mapping on, the sum of the increments in the units of the input, and that other operations and the legacy IFS ids in metres (143, 142, 228) stay undefined. --- .../statistics_param_mappings.yaml | 84 ++++++++++ .../action/statistics-mtg2/ParamMapping.cc | 155 ++++++++++++++++-- 2 files changed, 222 insertions(+), 17 deletions(-) diff --git a/share/multio/param-mappings/statistics_param_mappings.yaml b/share/multio/param-mappings/statistics_param_mappings.yaml index 7e03e33d3..4c35a4821 100644 --- a/share/multio/param-mappings/statistics_param_mappings.yaml +++ b/share/multio/param-mappings/statistics_param_mappings.yaml @@ -4866,3 +4866,87 @@ paramOut: 266909 - typeOfStatisticalProcessing: 6 paramOut: 267909 + +# Manually added mappings: accumulation of the IFS accumulated parameters. +# The input is the per-step increment of an accumulated parameter, sent under the paramId of the +# accumulation itself, so the accumulation keeps its paramId (paramIn == paramOut). The values are +# not converted and must already be in the units of the paramId: kg m**-2 (water: 228228, 228143, +# 3062, 228144, 260259, 231002, 231010, 231012), J m**-2 (146, 147, 176, 177, 169, 175, 178, 179, +# 212) or N m**-2 s (180, 181). The generator above does not produce these entries: keep them when +# this file is regenerated. +- paramIn: 228228 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 228228 +- paramIn: 228143 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 228143 +- paramIn: 3062 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 3062 +- paramIn: 228144 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 228144 +- paramIn: 260259 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 260259 +- paramIn: 231002 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 231002 +- paramIn: 231010 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 231010 +- paramIn: 231012 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 231012 +- paramIn: 146 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 146 +- paramIn: 147 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 147 +- paramIn: 176 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 176 +- paramIn: 177 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 177 +- paramIn: 169 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 169 +- paramIn: 175 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 175 +- paramIn: 178 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 178 +- paramIn: 179 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 179 +- paramIn: 212 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 212 +- paramIn: 180 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 180 +- paramIn: 181 + mappings: + - typeOfStatisticalProcessing: 1 + paramOut: 181 diff --git a/tests/multio/action/statistics-mtg2/ParamMapping.cc b/tests/multio/action/statistics-mtg2/ParamMapping.cc index 96465c3e8..3aab6be20 100644 --- a/tests/multio/action/statistics-mtg2/ParamMapping.cc +++ b/tests/multio/action/statistics-mtg2/ParamMapping.cc @@ -78,24 +78,54 @@ std::vector mappings = { // Ocean (levtype o3d) --- original param not verified }; -std::int64_t testParameterMapping(std::int64_t param, std::string op) { - const auto plan - = "{ \"name\": \"statistics_param_mapping_" + std::to_string(param) + "_" + op - + "_test\", " - "\"actions\": [ { " - "\"type\": \"statistics-mtg2\", " - "\"output-frequency\": \"1h\", " - "\"operations\": [ \"" - + op - + "\" ], " - "\"options\": { \"initial-condition-present\": \"true\", \"disable-strict-mapping\": \"true\" } }," - "{ \"type\": \"debug-sink\" } ] }"; - - auto env = MultioTestEnvironment(plan); - EXPECT_EQUAL(env.debugSink().size(), 0); - sendSimulationStart(env, 20200721); +// Accumulated parameters, accumulated under their own paramId (levtype sfc); input and output paramId are identical +std::vector accumulationMappings = { + // Water, kg m**-2 + {228228, "accumulate", 228228}, + {228143, "accumulate", 228143}, + {3062, "accumulate", 3062}, + {228144, "accumulate", 228144}, + {260259, "accumulate", 260259}, + {231002, "accumulate", 231002}, + {231010, "accumulate", 231010}, + {231012, "accumulate", 231012}, + + // Energy, J m**-2 + {146, "accumulate", 146}, + {147, "accumulate", 147}, + {176, "accumulate", 176}, + {177, "accumulate", 177}, + {169, "accumulate", 169}, + {175, "accumulate", 175}, + {178, "accumulate", 178}, + {179, "accumulate", 179}, + {212, "accumulate", 212}, - // Send step 0 and 1 message with the input parameter ID + // Stress, N m**-2 s + {180, "accumulate", 180}, + {181, "accumulate", 181}, +}; + +std::string makePlan(std::int64_t param, const std::string& op, bool strict, + const std::string& outputFrequency = "1h") { + return "{ \"name\": \"statistics_param_mapping_" + std::to_string(param) + "_" + op + + "_test\", " + "\"actions\": [ { " + "\"type\": \"statistics-mtg2\", " + "\"output-frequency\": \"" + + outputFrequency + + "\", " + "\"operations\": [ \"" + + op + + "\" ], " + "\"options\": { \"initial-condition-present\": \"true\", \"disable-strict-mapping\": \"" + + std::string(strict ? "false" : "true") + + "\" } }," + "{ \"type\": \"debug-sink\" } ] }"; +} + +// Send step 0 and 1 messages with the input parameter ID +void sendStepZeroAndOne(MultioTestEnvironment& env, std::int64_t param) { for (int64_t step = 0; step <= 1; ++step) { auto md = Metadata({{"param", param}, {"levtype", "sfc"}, @@ -109,6 +139,31 @@ std::int64_t testParameterMapping(std::int64_t param, std::string op) { auto msg = Message({Message::Tag::Field, {}, {}, std::move(md)}, std::move(pl)); EXPECT_NO_THROW(env.process(std::move(msg))); } +} + +// The mapping is applied when the statistics message is emitted; with strict mapping on, an undefined mapping throws. +bool statisticsMappingIsRejected(std::int64_t param, const std::string& op) { + auto env = MultioTestEnvironment(makePlan(param, op, true)); + sendSimulationStart(env, 20200721); + sendStepZeroAndOne(env, param); + try { + env.process({{Message::Tag::Flush, {}, {}, {{"flushKind", "last-step"}}}}); + } + catch (const eckit::Exception& e) { + // The failure-aware action wraps the exception: look for the mapping error in the nested message + std::ostringstream os; + os << e; + return os.str().find("is undefined!") != std::string::npos; + } + return false; +} + +std::int64_t testParameterMapping(std::int64_t param, std::string op, bool strict = false) { + auto env = MultioTestEnvironment(makePlan(param, op, strict)); + EXPECT_EQUAL(env.debugSink().size(), 0); + sendSimulationStart(env, 20200721); + + sendStepZeroAndOne(env, param); EXPECT_EQUAL(env.debugSink().size(), 0); // Send a flush last-step to trigger emitting the statistics message @@ -134,6 +189,72 @@ CASE("statistics parameter mappings") { } } +// The mappings above are applied with strict mapping on, i.e. without `disable-strict-mapping` +CASE("statistics parameter mappings of the IFS accumulated parameters, strict mapping") { + for (StatisticsParamMapping mapping : accumulationMappings) { + std::int64_t inParam = std::get<0>(mapping); + std::string inOp = std::get<1>(mapping); + std::int64_t outParam = std::get<2>(mapping); + + std::ostringstream os; + os << "strict param=" << inParam << ", operation=" << inOp << " -> param=" << outParam; + + SECTION(os.str()) { + EXPECT_EQUAL(testParameterMapping(inParam, inOp, true), outParam); + } + } +} + +// The accumulated parameters are only mapped for accumulation, and only under their own paramId: other operations and +// the legacy IFS paramIds in metres (143, 142, ...) stay undefined. +CASE("statistics parameter mappings of the accumulated parameters, unmapped combinations are rejected") { + const std::vector rejected = { + {228143, "average", 0}, {146, "average", 0}, {180, "average", 0}, + {143, "accumulate", 0}, {142, "accumulate", 0}, {228, "accumulate", 0}, + }; + for (const auto& [param, op, ignored] : rejected) { + std::ostringstream os; + os << "strict param=" << param << ", operation=" << op; + + SECTION(os.str()) { + EXPECT(statisticsMappingIsRejected(param, op)); + } + } +} + +// The accumulation is the plain sum of the increments, in the units of the input: nothing is converted. +CASE("statistics accumulation of the accumulated parameters keeps the units of the increments") { + for (std::int64_t param : {228143, 3062, 146, 180}) { + SECTION("strict param=" + std::to_string(param)) { + auto env = MultioTestEnvironment(makePlan(param, "accumulate", true, "3h")); + sendSimulationStart(env, 20200721); + + // Step 0 is the anchor of the window (initial condition); steps 1..3 carry the increments 1, 2 and 3 + for (std::int64_t step = 0; step <= 3; ++step) { + auto md = Metadata({{"param", param}, + {"levtype", "sfc"}, + {"grid", "O80"}, + {"date", 20200721}, + {"time", 0000}, + {"step", step}, + {"misc-precision", "double"}}); + setTimingMetadata(md, 3600); + auto data = static_cast(step); + auto pl = eckit::Buffer{&data, sizeof(double)}; + EXPECT_NO_THROW(env.process(Message({Message::Tag::Field, {}, {}, std::move(md)}, std::move(pl)))); + } + + EXPECT_NO_THROW(env.process({{Message::Tag::Flush, {}, {}, {{"flushKind", "last-step"}}}})); + EXPECT_EQUAL(env.debugSink().size(), 2); + + EXPECT_EQUAL(sizeof(double), env.debugSink().front().size()); + auto sum = static_cast(env.debugSink().front().payload().data()); + EXPECT_EQUAL(1.0 + 2.0 + 3.0, *sum); + EXPECT_EQUAL(param, env.debugSink().front().metadata().get("param")); + } + } +} + } // namespace multio::test::statistics_mtg2 int main(int argc, char** argv) { From 8e8461185b0f7940916274acc8ee2f1ad7ee5591 Mon Sep 17 00:00:00 2001 From: Tiago Quintino Date: Mon, 19 Oct 2026 04:46:40 +0100 Subject: [PATCH 2/2] statistics-mtg2: add window-origin forecast-start A statistic can now accumulate from the simulation start and be emitted at every output-frequency, instead of being closed and reset after each output-frequency window. This gives the running accumulations 0-3, 0-6, 0-9 h from hourly increments without a last-step flush per output step. With `window-origin: forecast-start` the window is (simulation start, next emission point] and is never reset. A sample on the emission point is emitted at once with the absolute current step, timespan "fs" and the elapsed length as distance; the window end then moves to the next emission point. Completeness is judged against the elapsed span. A last-step flush emits only the tail after the last emission. The default (window-start) is unchanged. Supported: accumulate, average, maximum, minimum; output-frequency in hours; forward-offset windows; instantaneous input. Other combinations are rejected when the action is built. The restart format is unchanged. --- src/multio/action/statistics-mtg2/CONTEXT.md | 20 + .../action/statistics-mtg2/OperationWindow.cc | 49 +- .../action/statistics-mtg2/OperationWindow.h | 15 +- .../action/statistics-mtg2/PeriodUpdaters.h | 3 + .../action/statistics-mtg2/Statistics.cc | 103 +++- .../action/statistics-mtg2/Statistics.h | 1 + .../statistics-mtg2/TemporalStatistics.cc | 4 + .../statistics-mtg2/TemporalStatistics.h | 2 + .../statistics-mtg2/cfg/StatisticsOptions.cc | 19 + .../statistics-mtg2/cfg/StatisticsOptions.h | 12 + .../action/statistics-mtg2/CMakeLists.txt | 14 + .../statistics-mtg2/ForecastStartWindow.cc | 444 ++++++++++++++++++ 12 files changed, 679 insertions(+), 7 deletions(-) create mode 100644 tests/multio/action/statistics-mtg2/ForecastStartWindow.cc diff --git a/src/multio/action/statistics-mtg2/CONTEXT.md b/src/multio/action/statistics-mtg2/CONTEXT.md index 65c18aef5..c808e0b7e 100644 --- a/src/multio/action/statistics-mtg2/CONTEXT.md +++ b/src/multio/action/statistics-mtg2/CONTEXT.md @@ -110,6 +110,26 @@ For a newly created forward-offset window, a message exactly on the nominal lowe This lower-bound behavior is structural and must not depend on `initial-condition-present`. +## Forecast-Start Window Origin + +```yaml +output-frequency: "3h" +options: + window-origin: forecast-start # default: window-start +``` + +The default `window-start` is the behaviour described above: a calendar window of `output-frequency` that is emitted when a field outside it arrives and then reset. + +With `forecast-start` the statistic accumulates from the simulation start (the first-step flush) and is never reset; `output-frequency` is only the emission cadence: + +- the window is `(simulation start, next emission point]`; a sample on the emission point is emitted immediately (no later field or flush needed), and the end then moves to the next emission point; +- the emitted `step` is the absolute current step, `timespan` is `fs` when the window starts at the forecast epoch (otherwise the elapsed duration), and `misc-distanceFromPreviousStepInSeconds` is the elapsed length; +- completeness is judged against the elapsed span (first sample seen, every sample at its declared distance from the previous one); a missing sample makes all later emissions of the field incomplete (suppressed unless `emit-incomplete-statistics`); +- a field after the next emission point is an error: the input must contain every emission point; +- a `last-step` flush emits the open tail (samples after the last emission) once, a run ending on an emission point has no tail; +- `lastFlush_` of the window is the last emitted point; the restart format is unchanged, the origin comes from the options when a window is loaded, and a restored window keeps the original forecast start (a run without restored state accumulates from its own start); +- accepted: operations `accumulate`, `average`, `maximum`, `minimum` (their `compute` leaves the state untouched), `output-frequency` in hours, `window-type: forward-offset`, instantaneous input only, simulation start on a whole hour. + ## Initial Condition Option The active option is: diff --git a/src/multio/action/statistics-mtg2/OperationWindow.cc b/src/multio/action/statistics-mtg2/OperationWindow.cc index 97d48e1f1..9e6acb907 100644 --- a/src/multio/action/statistics-mtg2/OperationWindow.cc +++ b/src/multio/action/statistics-mtg2/OperationWindow.cc @@ -62,6 +62,24 @@ eckit::DateTime yyyymmdd_hhmmss2DateTime(uint64_t yyyymmdd, uint64_t hhmmss) { OperationWindow make_window(const std::unique_ptr& periodUpdater, const StatisticsConfiguration& cfg, const eckit::DateTime& simulationStart) { eckit::DateTime epochPoint{cfg.epoch()}; + + if (cfg.options().windowOrigin() == WindowOrigin::ForecastStart) { + // The window opens exactly at the simulation start and ends at the first emission point. It is never advanced + // by a field: later emission points are reached by OperationWindow::extendEnd. + if (simulationStart.time().minutes() != 0 || simulationStart.time().seconds() != 0) { + std::ostringstream os; + os << "window-origin forecast-start requires a simulation start on a whole hour, got " << simulationStart; + throw eckit::UserError{os.str(), Here()}; + } + eckit::DateTime endPoint{periodUpdater->computeWinEndTime(simulationStart)}; + return OperationWindow{epochPoint, + simulationStart, + eckit::DateTime{cfg.curr()}, + endPoint, + cfg.options().windowType(), + WindowOrigin::ForecastStart}; + } + eckit::DateTime startPoint{periodUpdater->computeWinStartTime(simulationStart)}; eckit::DateTime endPoint{periodUpdater->computeWinEndTime(startPoint)}; @@ -99,6 +117,7 @@ OperationWindow::OperationWindow(std::shared_ptr& IOmanager, const count_{0}, counts_{}, windowType_{WindowType::ForwardOffset}, + windowOrigin_{opt.windowOrigin()}, firstPoint_{}, declaredDistanceHistogram_{}, observedDistanceHistogram_{}, @@ -110,7 +129,7 @@ OperationWindow::OperationWindow(std::shared_ptr& IOmanager, const OperationWindow::OperationWindow(const eckit::DateTime& epochPoint, const eckit::DateTime& startPoint, const eckit::DateTime& creationPoint, const eckit::DateTime& endPoint, - WindowType windowType) : + WindowType windowType, WindowOrigin windowOrigin) : epochPoint_{epochPoint}, startPoint_{startPoint}, creationPoint_{creationPoint}, @@ -121,6 +140,7 @@ OperationWindow::OperationWindow(const eckit::DateTime& epochPoint, const eckit: count_{0}, counts_{}, windowType_{windowType}, + windowOrigin_{windowOrigin}, firstPoint_{}, declaredDistanceHistogram_{}, observedDistanceHistogram_{}, @@ -221,6 +241,15 @@ void OperationWindow::updateWindow(const eckit::DateTime& startPoint, const ecki return; } +void OperationWindow::extendEnd(const eckit::DateTime& endPoint) { + if (endPoint <= endPoint_) { + std::ostringstream os; + os << *this << " : new end " << endPoint << " does not extend the window end " << endPoint_ << std::endl; + throw eckit::SeriousBug(os.str(), Here()); + } + endPoint_ = endPoint; +} + bool OperationWindow::isWithin(const eckit::DateTime& dt) const { bool ret; if (windowType_ == WindowType::ForwardOffset) { @@ -283,6 +312,18 @@ long OperationWindow::timeSpanInSeconds() const { return long(endPoint_ - startPoint_); } +long OperationWindow::emissionSpanInSeconds() const { + return windowOrigin_ == WindowOrigin::ForecastStart ? long(currPoint_ - startPoint_) : timeSpanInSeconds(); +} + +bool OperationWindow::hasUnemittedSamples() const { + return count_ > 0 && currPoint_ > lastFlush_; +} + +WindowOrigin OperationWindow::windowOrigin() const { + return windowOrigin_; +} + long OperationWindow::lastPointsDiffInSeconds() const { return long(currPoint_ - prevPoint_); } @@ -401,6 +442,10 @@ bool OperationWindow::isComplete() const { if (!firstPoint_ || !contiguous_) { return false; } + if (windowOrigin_ == WindowOrigin::ForecastStart) { + // Judged against the elapsed span (start .. current point), not against the next emission point + return true; + } if (windowType_ == WindowType::ForwardOffset) { return currPoint_ == endPoint_; } @@ -452,7 +497,7 @@ const std::map& OperationWindow::declaredDistanceHist std::map OperationWindow::observedDistanceHistogram() const { auto histogram = observedDistanceHistogram_; - if (firstPoint_) { + if (firstPoint_ && windowOrigin_ != WindowOrigin::ForecastStart) { const auto trailingDistance = static_cast(endPoint_ - currPoint_); if (trailingDistance > 0) { histogram[trailingDistance]++; diff --git a/src/multio/action/statistics-mtg2/OperationWindow.h b/src/multio/action/statistics-mtg2/OperationWindow.h index 72bea15df..ac3210e78 100644 --- a/src/multio/action/statistics-mtg2/OperationWindow.h +++ b/src/multio/action/statistics-mtg2/OperationWindow.h @@ -22,7 +22,8 @@ class OperationWindow { OperationWindow(std::shared_ptr& IOmanager, const StatisticsOptions& opt); OperationWindow(const eckit::DateTime& epochPoint, const eckit::DateTime& startPoint, - const eckit::DateTime& creationPoint, const eckit::DateTime& endPoint, WindowType windowType); + const eckit::DateTime& creationPoint, const eckit::DateTime& endPoint, WindowType windowType, + WindowOrigin windowOrigin = WindowOrigin::WindowStart); long count() const; const std::vector& counts() const; @@ -33,6 +34,9 @@ class OperationWindow { void updateData(const eckit::DateTime& currentPoint, std::int64_t distanceFromPreviousStepInSeconds); void updateWindow(const eckit::DateTime& startPoint, const eckit::DateTime& endPoint); + /// Forecast-start windows only: move the end to the next emission point without touching the accumulated state + void extendEnd(const eckit::DateTime& endPoint); + void dump(std::shared_ptr& IOmanager, const StatisticsOptions& opt) const; void load(std::shared_ptr& IOmanager, const StatisticsOptions& opt); @@ -44,6 +48,9 @@ class OperationWindow { long timeSpanInSeconds() const; long timeSpanInHours() const; + /// Length reported with an emission: the nominal window length, or for a forecast-start window the elapsed + /// length from the window start to the current point + long emissionSpanInSeconds() const; long lastPointsDiffInSeconds() const; util::DateTimeDiff lastPointsDiff() const; @@ -83,6 +90,9 @@ class OperationWindow { bool isUniform() const; std::string incompleteReason() const; WindowType windowType() const; + WindowOrigin windowOrigin() const; + /// Forecast-start windows only: samples have been received since the last emission + bool hasUnemittedSamples() const; std::int64_t lastDeclaredDistance() const; const std::map& declaredDistanceHistogram() const; std::map observedDistanceHistogram() const; @@ -91,6 +101,7 @@ class OperationWindow { std::string stepRangeInHours(const eckit::DateTime& refPoint) const; + /// Records the current point as the last emitted point void updateFlush(); size_t restartSize() const; @@ -107,6 +118,8 @@ class OperationWindow { long count_; mutable std::vector counts_; WindowType windowType_; + // Not part of the restart state: taken from the options when a window is created or loaded + WindowOrigin windowOrigin_; std::optional firstPoint_; std::map declaredDistanceHistogram_; std::map observedDistanceHistogram_; diff --git a/src/multio/action/statistics-mtg2/PeriodUpdaters.h b/src/multio/action/statistics-mtg2/PeriodUpdaters.h index 358123a90..f8b3deba0 100644 --- a/src/multio/action/statistics-mtg2/PeriodUpdaters.h +++ b/src/multio/action/statistics-mtg2/PeriodUpdaters.h @@ -15,6 +15,9 @@ namespace multio::action::statistics_mtg2 { +/// Split an output frequency such as "3h" into its span (3) and its period kind ("h") +void parsePeriodGrammar(std::string const& output_freq, long& span, std::string& periodKind); + std::unique_ptr make_period_updater(std::string const& output_freq, const StatisticsConfiguration& cfg); std::unique_ptr load_period_updater(std::shared_ptr& IOmanager, const StatisticsOptions& opt); diff --git a/src/multio/action/statistics-mtg2/Statistics.cc b/src/multio/action/statistics-mtg2/Statistics.cc index 7beeb27d2..1e3d5c951 100644 --- a/src/multio/action/statistics-mtg2/Statistics.cc +++ b/src/multio/action/statistics-mtg2/Statistics.cc @@ -128,6 +128,52 @@ dm::TimeDuration absoluteStep(std::int64_t seconds) { } return dm::TimeDuration{std::chrono::seconds{seconds}}; } + +// Length of the emitted statistic as a MARS timespan. A forecast-start window that opened at the forecast epoch is +// reported as 'fs' (from start, step = elapsed length); a forecast-start window opened later (a run restarted without +// restored state) is reported with its elapsed duration, so it never claims data from before its start. +dm::TimeSpan emittedTimespan(const OperationWindow& window) { + if (window.windowOrigin() == WindowOrigin::WindowStart) { + return dm::TimeSpan{dm::TimeDuration{std::chrono::hours{window.timeSpanInHours()}}}; + } + if (window.startPoint() == window.epochPoint()) { + return dm::TimeSpan::fromStart(); + } + return dm::TimeSpan{absoluteStep(window.emissionSpanInSeconds())}; +} + +// A forecast-start window is never reset, so only operations whose compute() leaves the accumulated state untouched and +// whose result over the elapsed span is well defined are accepted. +void validateForecastStartConfiguration(const StatisticsOptions& opt, const std::vector& operations, + const std::string& outputFrequency) { + if (opt.windowOrigin() != WindowOrigin::ForecastStart) { + return; + } + + long span; + std::string periodKind; + parsePeriodGrammar(outputFrequency, span, periodKind); + if (periodKind != "h" && periodKind != "hour") { + throw eckit::UserError{ + "window-origin forecast-start requires an output-frequency in hours, got " + outputFrequency, Here()}; + } + if (opt.windowType() != WindowType::ForwardOffset) { + throw eckit::UserError{"window-origin forecast-start requires window-type forward-offset", Here()}; + } + for (const auto& op : operations) { + if (op != "accumulate" && op != "average" && op != "maximum" && op != "minimum") { + throw eckit::UserError{ + "window-origin forecast-start supports the operations accumulate, average, maximum " + "and minimum, got " + + op, + Here()}; + } + } + if (opt.outputTimeReference() == OutputTimeReference::StartOfWindow) { + throw eckit::UserError{"window-origin forecast-start does not support output-time-reference start-of-window", + Here()}; + } +} } // namespace Statistics::Statistics(const ComponentConfiguration& compConf) : @@ -140,7 +186,9 @@ Statistics::Statistics(const ComponentConfiguration& compConf) : outputFrequency_{compConf.parsedConfig().getString("output-frequency")}, paramMapping_{StatisticsParamMapping::makeStatisticsParamMapping()}, operationMapping_{StatisticsOperationMapping::makeStatisticsOperationMapping()}, - IOmanager_{StatisticsIOFactory::instance().build(opt_.restartLib(), opt_.restartPath(), opt_.restartPrefix())} {} + IOmanager_{StatisticsIOFactory::instance().build(opt_.restartLib(), opt_.restartPath(), opt_.restartPrefix())} { + validateForecastStartConfiguration(opt_, operations_, outputFrequency_); +} void Statistics::handleSimulationStart(const FlushMetadataKeys& flush) { if (!flush.date.isSet() || !flush.time.isSet() || !flush.step.isSet()) { @@ -400,6 +448,14 @@ void Statistics::executeImpl(message::Message msg) { StatisticsConfiguration cfg{msg, opt_}; std::string key = cfg.key(); + if (opt_.windowOrigin() == WindowOrigin::ForecastStart && cfg.isStatistical()) { + throw eckit::UserError{ + "window-origin forecast-start only accepts instantaneous fields, got a field with " + "timespan or stattype: " + + msg.metadata().toString(), + Here()}; + } + // TODO: if (!simulationStart_) { throw eckit::SeriousBug("Field received before simulation-start flush", Here()); @@ -459,6 +515,11 @@ void Statistics::executeImpl(message::Message msg) { throw eckit::SeriousBug(os.str(), Here()); } + if (opt_.windowOrigin() == WindowOrigin::ForecastStart) { + updateForecastStartWindow(ts, msg, cfg); + return; + } + // Decide to emit statistics while (ts.isOutsideWindow(msg, cfg)) { emitStatistics(ts, msg.source(), msg.destination(), true); @@ -474,8 +535,39 @@ void Statistics::executeImpl(message::Message msg) { return; } +// A forecast-start window is emitted as soon as a sample reaches the next emission point and keeps accumulating +void Statistics::updateForecastStartWindow(TemporalStatistics& ts, message::Message& msg, + const StatisticsConfiguration& cfg) { + if (ts.isOutsideWindow(msg, cfg)) { + std::ostringstream os; + os << "Field at " << cfg.curr() << " is after the next emission point " << ts.cwin().endPoint() + << " of the forecast-start window: input is required at every emission point"; + throw eckit::UserError{os.str(), Here()}; + } + + ts.updateData(msg, cfg); + + if (ts.cwin().currPoint() != ts.cwin().endPoint()) { + return; + } + emitStatistics(ts, msg.source(), msg.destination()); + ts.win().updateFlush(); + ts.advanceEnd(); + if (opt_.debug()) { + logWindowEvent("WIN_ADVANCE", ts); + } +} + void Statistics::emitAllStatistics(message::Peer source, message::Peer destination) { for (auto& [key, ts] : fieldStats_) { + if (opt_.windowOrigin() == WindowOrigin::ForecastStart) { + // Only the tail after the last emission point is still open + if (ts->cwin().hasUnemittedSamples()) { + emitStatistics(*ts, source, destination); + ts->win().updateFlush(); + } + continue; + } emitStatistics(*ts, source, destination); } } @@ -590,8 +682,7 @@ void Statistics::emitStatistics(TemporalStatistics& ts, message::Peer source, me auto opname = (*it)->operation(); if (opname != "instant") { if (currentLoop == 1) { - const std::int64_t timespan = ts.cwin().timeSpanInHours(); - dm::dumpEntry(dm::TIMESPAN, dm::TIMESPAN.makeEntry(timespan), md); + dm::dumpEntry(dm::TIMESPAN, dm::TIMESPAN.makeEntry(emittedTimespan(ts.cwin())), md); dm::dumpEntry(dm::TimeIncrementInSeconds, dm::TimeIncrementInSeconds.makeEntry(cfg.outputStepInSeconds()), md); paramMapping_.applyMapping(md, opname, !opt_.disableStrictMapping()); @@ -633,7 +724,7 @@ void Statistics::emitStatistics(TemporalStatistics& ts, message::Peer source, me } const auto outputDistance - = opname == "instant" ? ts.cwin().lastDeclaredDistance() : ts.cwin().timeSpanInSeconds(); + = opname == "instant" ? ts.cwin().lastDeclaredDistance() : ts.cwin().emissionSpanInSeconds(); dm::dumpEntry(dm::DistanceFromPreviousStepInSeconds, dm::DistanceFromPreviousStepInSeconds.makeEntry(outputDistance), md); @@ -644,6 +735,10 @@ void Statistics::emitStatistics(TemporalStatistics& ts, message::Peer source, me break; } case OutputTimeReference::StartOfWindow: { + if (ts.cwin().windowOrigin() == WindowOrigin::ForecastStart) { + throw eckit::UserError{ + "window-origin forecast-start does not support output-time-reference start-of-window", Here()}; + } const auto step = absoluteStep(ts.win().currPointInSeconds(ts.win().creationPoint())); dm::dumpEntry(dm::STEP, dm::STEP.makeEntry(step), md); // We explicitly take the creation point - alternative would be the start point. diff --git a/src/multio/action/statistics-mtg2/Statistics.h b/src/multio/action/statistics-mtg2/Statistics.h index 0bdb1c70a..463803503 100644 --- a/src/multio/action/statistics-mtg2/Statistics.h +++ b/src/multio/action/statistics-mtg2/Statistics.h @@ -71,6 +71,7 @@ class Statistics : public ChainedAction { std::map> fieldStats_; + void updateForecastStartWindow(TemporalStatistics& ts, message::Message& msg, const StatisticsConfiguration& cfg); void emitAllStatistics(message::Peer source, message::Peer destination); void emitStatistics(TemporalStatistics& ts, message::Peer source, message::Peer destination, bool finalizeSuppressed = false); diff --git a/src/multio/action/statistics-mtg2/TemporalStatistics.cc b/src/multio/action/statistics-mtg2/TemporalStatistics.cc index a0a0300b6..a8e49b51e 100644 --- a/src/multio/action/statistics-mtg2/TemporalStatistics.cc +++ b/src/multio/action/statistics-mtg2/TemporalStatistics.cc @@ -93,6 +93,10 @@ void TemporalStatistics::updateWindow(const message::Message& msg, const Statist return; } +void TemporalStatistics::advanceEnd() { + window_.extendEnd(periodUpdater_->updateWinEndTime(window_.endPoint())); +} + bool TemporalStatistics::isOutsideWindow(message::Message& msg, const StatisticsConfiguration& cfg) { LOG_DEBUG_LIB(::multio::LibMultio) << cfg.options().logPrefix() << " *** Check outside Window " << std::endl; return !window_.isWithin(currentDateTime(msg, cfg)); diff --git a/src/multio/action/statistics-mtg2/TemporalStatistics.h b/src/multio/action/statistics-mtg2/TemporalStatistics.h index 9cec51a4e..75d1b5f69 100644 --- a/src/multio/action/statistics-mtg2/TemporalStatistics.h +++ b/src/multio/action/statistics-mtg2/TemporalStatistics.h @@ -31,6 +31,8 @@ class TemporalStatistics { void updateData(message::Message& msg, const StatisticsConfiguration& cfg); void updateWindow(const message::Message& msg, const StatisticsConfiguration& cfg); + /// Forecast-start windows: move the window end to the next emission point, keeping the accumulated state + void advanceEnd(); void dump(std::shared_ptr& IOmanager, const StatisticsOptions& opt) const; diff --git a/src/multio/action/statistics-mtg2/cfg/StatisticsOptions.cc b/src/multio/action/statistics-mtg2/cfg/StatisticsOptions.cc index 1d6d3fa5f..2a63be538 100644 --- a/src/multio/action/statistics-mtg2/cfg/StatisticsOptions.cc +++ b/src/multio/action/statistics-mtg2/cfg/StatisticsOptions.cc @@ -90,6 +90,20 @@ WindowType parseWindowType(const eckit::LocalConfiguration& cfg) { throw eckit::UserError(os.str(), Here()); } +WindowOrigin parseWindowOrigin(const eckit::LocalConfiguration& cfg) { + const auto windowOrigin = cfg.getString("window-origin", "window-start"); + if (windowOrigin == "window-start") { + return WindowOrigin::WindowStart; + } + if (windowOrigin == "forecast-start") { + return WindowOrigin::ForecastStart; + } + + std::ostringstream os; + os << "Invalid window origin :: " << windowOrigin << " (expected window-start or forecast-start)" << std::endl; + throw eckit::UserError(os.str(), Here()); +} + std::optional parseValueCountThreshold(const eckit::LocalConfiguration& cfg) { const auto threshold = cfg.getLong("value-count-threshold", -1); @@ -172,6 +186,7 @@ StatisticsOptions::StatisticsOptions(const eckit::LocalConfiguration& cfg) : restartLib_{parseRestartLib(cfg)}, logPrefix_{parseLogPrefix(cfg)}, windowType_{parseWindowType(cfg)}, + windowOrigin_{parseWindowOrigin(cfg)}, valueCountThreshold_{parseValueCountThreshold(cfg)}, disableStrictMapping_{parseDisableStrictMapping(cfg)}, disableSquashing_{parseDisableSquashing(cfg)}, @@ -215,6 +230,10 @@ WindowType StatisticsOptions::windowType() const { return windowType_; } +WindowOrigin StatisticsOptions::windowOrigin() const { + return windowOrigin_; +} + std::optional StatisticsOptions::valueCountThreshold() const { return valueCountThreshold_; } diff --git a/src/multio/action/statistics-mtg2/cfg/StatisticsOptions.h b/src/multio/action/statistics-mtg2/cfg/StatisticsOptions.h index 35695f145..ab48bc7a1 100644 --- a/src/multio/action/statistics-mtg2/cfg/StatisticsOptions.h +++ b/src/multio/action/statistics-mtg2/cfg/StatisticsOptions.h @@ -16,6 +16,16 @@ enum class WindowType : std::int64_t BackwardOffset = 1, }; +/// Where the statistics window starts. +/// WindowStart: the window is a calendar period of `output-frequency` that is closed and reset after each emission +/// ForecastStart: the window opens at the simulation start and is never reset; `output-frequency` is the cadence at +/// which the running statistic over the whole forecast so far is emitted +enum class WindowOrigin : std::int64_t +{ + WindowStart = 0, + ForecastStart = 1, +}; + enum class OutputTimeReference : std::int64_t { StartOfForecast = 0, @@ -40,6 +50,7 @@ class StatisticsOptions { const std::string logPrefix_; const WindowType windowType_; + const WindowOrigin windowOrigin_; const std::optional valueCountThreshold_; @@ -67,6 +78,7 @@ class StatisticsOptions { const std::string& logPrefix() const; WindowType windowType() const; + WindowOrigin windowOrigin() const; std::optional valueCountThreshold() const; diff --git a/tests/multio/action/statistics-mtg2/CMakeLists.txt b/tests/multio/action/statistics-mtg2/CMakeLists.txt index 99a9780bf..6ef1c9d12 100644 --- a/tests/multio/action/statistics-mtg2/CMakeLists.txt +++ b/tests/multio/action/statistics-mtg2/CMakeLists.txt @@ -71,4 +71,18 @@ ecbuild_add_test( LIBS multio-action-debug-sink multio-action-statistics-mtg2 ) +ecbuild_add_test( + TARGET ${PREFIX}_forecast_start_window_delete_dir + COMMAND ${CMAKE_COMMAND} + ARGS -E remove_directory ${CMAKE_CURRENT_BINARY_DIR}/ForecastStartDump +) + +ecbuild_add_test( + TARGET ${PREFIX}_forecast_start_window + TEST_REQUIRES ${PREFIX}_forecast_start_window_delete_dir + SOURCES ForecastStartWindow.cc + NO_AS_NEEDED + LIBS multio-action-debug-sink multio-action-statistics-mtg2 +) + add_subdirectory(operations) diff --git a/tests/multio/action/statistics-mtg2/ForecastStartWindow.cc b/tests/multio/action/statistics-mtg2/ForecastStartWindow.cc new file mode 100644 index 000000000..736997a04 --- /dev/null +++ b/tests/multio/action/statistics-mtg2/ForecastStartWindow.cc @@ -0,0 +1,444 @@ +/* + * (C) Copyright 2026- ECMWF. + * + * This software is licensed under the terms of the Apache Licence Version 2.0 + * which can be obtained at http://www.apache.org/licenses/LICENSE-2.0. + * In applying this licence, ECMWF does not waive the privileges and immunities + * granted to it by virtue of its status as an intergovernmental organisation nor + * does it submit to any jurisdiction. + */ + +/// Tests of the statistics-mtg2 option `window-origin: forecast-start`: the statistic accumulates from the simulation +/// start and the running value is emitted at every `output-frequency`, with `timespan=fs`, instead of being closed and +/// reset after each output-frequency window. + +#include +#include + +#include "eckit/io/Buffer.h" +#include "eckit/testing/Test.h" + +#include "multio/message/Message.h" +#include "multio/message/Metadata.h" + +#include "../../MultioTestEnvironment.h" +#include "StatisticsTestHelpers.h" + +namespace multio::test::statistics_mtg2 { + +using multio::message::Message; +using multio::message::Metadata; +using multio::test::MultioTestEnvironment; + +namespace { + +constexpr std::int64_t DATE = 20260101; +constexpr std::int64_t PARAM = 228143; // total precipitation kg m**-2: accumulated under its own paramId + +// The increment received at hour h is h, so the accumulation over 0..t is t(t+1)/2: 0-3 -> 6, 0-6 -> 21, 0-9 -> 45 +double increment(std::int64_t hour) { + return static_cast(hour); +} + +double accumulation(std::int64_t hours) { + return static_cast(hours * (hours + 1) / 2); +} + +// `options` is a list of `"key": value,` entries (each with a trailing comma) +std::string makePlan(const std::string& operations, const std::string& options, const std::string& frequency = "3h") { + return R"json({ + "name": "forecast-start-window", + "actions": [ + { + "type": "statistics-mtg2", + "output-frequency": ")json" + + frequency + R"json(", + "operations": [ )json" + + operations + R"json( ], + "options": { )json" + + options + R"json( "debug": false } + }, + { "type": "debug-sink" } + ] + })json"; +} + +const std::string FORECAST_START = R"json( "window-origin": "forecast-start",)json"; +const std::string IC_PRESENT = R"json( "initial-condition-present": true,)json"; +const std::string EMIT_INCOMPLETE = R"json( "emit-incomplete-statistics": true,)json"; + +std::string forecastStart(const std::string& more = "") { + return IC_PRESENT + FORECAST_START + more; +} + +Metadata fieldMetadata(std::int64_t param, std::int64_t step) { + Metadata md{{{"param", param}, + {"levtype", "sfc"}, + {"grid", "N80"}, + {"date", DATE}, + {"time", 0}, + {"step", step}, + {"misc-precision", "double"}}}; + setTimingMetadata(md, 3600); + return md; +} + +void sendField(MultioTestEnvironment& env, std::int64_t step, std::int64_t param = PARAM) { + const double value = increment(step); + env.process({{Message::Tag::Field, {}, {}, fieldMetadata(param, step)}, eckit::Buffer{&value, sizeof(value)}}); +} + +void flushLastStep(MultioTestEnvironment& env) { + env.process({{Message::Tag::Flush, {}, {}, {{"flushKind", "last-step"}}}}); +} + +struct Emitted { + double value; + std::int64_t param; + std::int64_t step; + std::string timespan; + std::int64_t distanceInSeconds; +}; + +// Removes and describes the field at the front of the sink +Emitted popEmitted(MultioTestEnvironment& env) { + auto& sink = env.debugSink(); + EXPECT(!sink.empty()); + EXPECT(sink.front().tag() == Message::Tag::Field); + const auto& md = sink.front().metadata(); + // timespan is the string "fs" or a number of hours + const auto timespan = md.get("timespan").visit([](const auto& v) -> std::string { + using Value = std::decay_t; + if constexpr (std::is_same_v) { + return v; + } + else if constexpr (std::is_integral_v) { + return std::to_string(v); + } + else { + return "unexpected type"; + } + }); + Emitted e{*static_cast(sink.front().payload().data()), md.get("param"), + md.get("step"), timespan, md.get("misc-distanceFromPreviousStepInSeconds")}; + sink.pop(); + return e; +} + +// The flush the simulation start left in the sink is already consumed by sendSimulationStart +void sendHours(MultioTestEnvironment& env, std::int64_t first, std::int64_t last, std::vector& emitted) { + for (std::int64_t step = first; step <= last; ++step) { + sendField(env, step); + while (!env.debugSink().empty()) { + emitted.push_back(popEmitted(env)); + } + } +} + +} // namespace + +CASE("forecast-start emits the running accumulation at every output step") { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart())}; + sendSimulationStart(env, DATE); + + std::vector emitted; + // Step 0 is the initial condition; the increments of hours 1..9 follow + for (std::int64_t step = 0; step <= 9; ++step) { + sendField(env, step); + const bool outputStep = step > 0 && step % 3 == 0; + // The accumulation is available as soon as the output step has been received: no later field, no flush + const std::size_t expectedSize = outputStep ? 1 : 0; + EXPECT_EQUAL(env.debugSink().size(), expectedSize); + if (outputStep) { + emitted.push_back(popEmitted(env)); + } + } + + EXPECT_EQUAL(emitted.size(), 3); + const std::vector hours{3, 6, 9}; + for (std::size_t i = 0; i < emitted.size(); ++i) { + const auto t = hours[i]; + EXPECT_EQUAL(emitted[i].value, accumulation(t)); + EXPECT_EQUAL(emitted[i].param, PARAM); + EXPECT_EQUAL(emitted[i].step, t); + EXPECT_EQUAL(emitted[i].timespan, std::string{"fs"}); + EXPECT_EQUAL(emitted[i].distanceInSeconds, t * 3600); + } + // 0-3: 1+2+3, 0-6: 1+..+6, 0-9: 1+..+9 + EXPECT_EQUAL(emitted[0].value, 6.0); + EXPECT_EQUAL(emitted[1].value, 21.0); + EXPECT_EQUAL(emitted[2].value, 45.0); + + // The forecast ended on an output step: nothing is left open, a last-step flush adds only the flush itself + EXPECT_NO_THROW(flushLastStep(env)); + EXPECT_EQUAL(env.debugSink().size(), 1); + EXPECT(env.debugSink().front().tag() == Message::Tag::Flush); +} + +CASE("forecast-start with an output frequency of one hour emits every step") { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart(), "1h")}; + sendSimulationStart(env, DATE); + sendField(env, 0); + + std::vector emitted; + sendHours(env, 1, 4, emitted); + EXPECT_EQUAL(emitted.size(), 4); + for (std::size_t i = 0; i < emitted.size(); ++i) { + const auto t = static_cast(i) + 1; + EXPECT_EQUAL(emitted[i].value, accumulation(t)); + EXPECT_EQUAL(emitted[i].step, t); + EXPECT_EQUAL(emitted[i].distanceInSeconds, t * 3600); + } +} + +CASE("forecast-start emits the open tail once when the forecast ends between output steps") { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart())}; + sendSimulationStart(env, DATE); + sendField(env, 0); + + std::vector emitted; + sendHours(env, 1, 7, emitted); + // 0-3 and 0-6 were emitted on the way, hour 7 is still open + EXPECT_EQUAL(emitted.size(), 2); + EXPECT_EQUAL(emitted[1].value, accumulation(6)); + + EXPECT_NO_THROW(flushLastStep(env)); + EXPECT_EQUAL(env.debugSink().size(), 2); + const auto tail = popEmitted(env); + EXPECT_EQUAL(tail.value, accumulation(7)); + EXPECT_EQUAL(tail.step, 7); + EXPECT_EQUAL(tail.timespan, std::string{"fs"}); + EXPECT_EQUAL(tail.distanceInSeconds, 7 * 3600); + EXPECT(env.debugSink().front().tag() == Message::Tag::Flush); + env.debugSink().pop(); + + // A repeated flush has nothing left to emit + EXPECT_NO_THROW(flushLastStep(env)); + EXPECT_EQUAL(env.debugSink().size(), 1); + EXPECT(env.debugSink().front().tag() == Message::Tag::Flush); +} + +CASE("forecast-start without a last-step flush leaves the tail unemitted") { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart())}; + sendSimulationStart(env, DATE); + sendField(env, 0); + std::vector emitted; + sendHours(env, 1, 4, emitted); + EXPECT_EQUAL(emitted.size(), 1); + EXPECT_EQUAL(emitted[0].step, 3); +} + +CASE("forecast-start works without the initial condition") { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", FORECAST_START)}; + sendSimulationStart(env, DATE); + std::vector emitted; + sendHours(env, 1, 6, emitted); + EXPECT_EQUAL(emitted.size(), 2); + EXPECT_EQUAL(emitted[0].value, accumulation(3)); + EXPECT_EQUAL(emitted[1].value, accumulation(6)); +} + +CASE("forecast-start average, maximum and minimum are running statistics over the elapsed span") { + // The increments are 1, 2, 3, ...: over 1..t the average is (t+1)/2, the maximum t and the minimum 1 + auto env = MultioTestEnvironment{ + makePlan(R"("average", "maximum", "minimum")", forecastStart(R"json( "disable-strict-mapping": true,)json"))}; + sendSimulationStart(env, DATE); + sendField(env, 0, 130); + + for (std::int64_t step = 1; step <= 6; ++step) { + const double value = increment(step); + env.process({{Message::Tag::Field, {}, {}, fieldMetadata(130, step)}, eckit::Buffer{&value, sizeof(value)}}); + if (step % 3 != 0) { + EXPECT_EQUAL(env.debugSink().size(), 0); + continue; + } + EXPECT_EQUAL(env.debugSink().size(), 3); + const auto average = popEmitted(env); + const auto maximum = popEmitted(env); + const auto minimum = popEmitted(env); + EXPECT_EQUAL(average.value, (static_cast(step) + 1.0) / 2.0); + EXPECT_EQUAL(maximum.value, static_cast(step)); + EXPECT_EQUAL(minimum.value, 1.0); + for (const auto& e : {average, maximum, minimum}) { + EXPECT_EQUAL(e.step, step); + EXPECT_EQUAL(e.timespan, std::string{"fs"}); + EXPECT_EQUAL(e.distanceInSeconds, step * 3600); + } + } +} + +CASE("forecast-start accumulates every field independently") { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart())}; + sendSimulationStart(env, DATE); + for (const auto param : {228143, 3062}) { + sendField(env, 0, param); + } + std::vector params; + for (std::int64_t step = 1; step <= 3; ++step) { + for (const auto param : {228143, 3062}) { + sendField(env, step, param); + } + } + EXPECT_EQUAL(env.debugSink().size(), 2); + for (int i = 0; i < 2; ++i) { + const auto e = popEmitted(env); + EXPECT_EQUAL(e.value, 6.0); + EXPECT_EQUAL(e.step, 3); + params.push_back(e.param); + } + EXPECT_EQUAL(params[0], 228143); + EXPECT_EQUAL(params[1], 3062); +} + +CASE("forecast-start window opened after the forecast epoch reports its elapsed duration, not fs") { + // A run that starts at step 6 without restored state accumulates from step 6: it must not claim 0-9 + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart())}; + sendSimulationStart(env, DATE, 0, 6); + sendField(env, 6); + + std::vector emitted; + sendHours(env, 7, 9, emitted); + EXPECT_EQUAL(emitted.size(), 1); + EXPECT_EQUAL(emitted[0].value, increment(7) + increment(8) + increment(9)); + EXPECT_EQUAL(emitted[0].step, 9); + EXPECT_EQUAL(emitted[0].timespan, std::string{"3"}); + EXPECT_EQUAL(emitted[0].distanceInSeconds, 3 * 3600); +} + +CASE("forecast-start suppresses accumulations with a missing hour unless incomplete statistics are requested") { + { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart())}; + sendSimulationStart(env, DATE); + sendField(env, 0); + sendField(env, 1); + sendField(env, 3); // hour 2 is missing: the sum 0-3 is wrong + EXPECT_EQUAL(env.debugSink().size(), 0); + // the gap is permanent, later emissions are suppressed too + sendField(env, 4); + sendField(env, 5); + sendField(env, 6); + EXPECT_EQUAL(env.debugSink().size(), 0); + } + { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart(EMIT_INCOMPLETE))}; + sendSimulationStart(env, DATE); + sendField(env, 0); + sendField(env, 1); + sendField(env, 3); + EXPECT_EQUAL(env.debugSink().size(), 1); + EXPECT_EQUAL(popEmitted(env).value, increment(1) + increment(3)); + } +} + +CASE("forecast-start rejects input that skips an output step") { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart())}; + sendSimulationStart(env, DATE); + sendField(env, 0); + sendField(env, 1); + sendField(env, 2); + EXPECT_THROWS(sendField(env, 4)); +} + +CASE("forecast-start rejects unsupported configurations") { + // Operations that re-baseline on compute + EXPECT_THROWS(MultioTestEnvironment{makePlan(R"("difference")", forecastStart())}); + EXPECT_THROWS(MultioTestEnvironment{makePlan(R"("instant")", forecastStart())}); + // Calendar frequencies + EXPECT_THROWS(MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart(), "1d")}); + // Backward offset windows + EXPECT_THROWS(MultioTestEnvironment{ + makePlan(R"("accumulate")", forecastStart(R"json( "window-type": "backward-offset",)json"))}); + EXPECT_THROWS(MultioTestEnvironment{ + makePlan(R"("accumulate")", forecastStart(R"json( "output-time-reference": "start-of-window",)json"))}); + // Unknown origin + EXPECT_THROWS(MultioTestEnvironment{makePlan(R"("accumulate")", R"json( "window-origin": "elsewhere",)json")}); + + // Statistical input + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart())}; + sendSimulationStart(env, DATE); + auto md = fieldMetadata(PARAM, 1); + md.set("timespan", 1); + md.set("misc-timeIncrementInSeconds", 3600); + const double value = 1.0; + EXPECT_THROWS(env.process({{Message::Tag::Field, {}, {}, std::move(md)}, eckit::Buffer{&value, sizeof(value)}})); +} + +CASE("window-start is the default and unchanged: the window is closed by the next field and reset") { + for (const std::string& origin : {std::string{}, std::string{R"json( "window-origin": "window-start",)json"}}) { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", IC_PRESENT + origin)}; + sendSimulationStart(env, DATE); + sendField(env, 0); + + std::vector emitted; + sendHours(env, 1, 3, emitted); + // Nothing before a field outside the window arrives + EXPECT_EQUAL(emitted.size(), 0); + sendHours(env, 4, 6, emitted); + // Window 0-3 closes on hour 4 with its nominal length; 3-6 stays open (and complete) until hour 7 + EXPECT_EQUAL(emitted.size(), 1); + EXPECT_EQUAL(emitted[0].value, accumulation(3)); + EXPECT_EQUAL(emitted[0].step, 3); + EXPECT_EQUAL(emitted[0].timespan, std::string{"3"}); + EXPECT_EQUAL(emitted[0].distanceInSeconds, 3 * 3600); + + sendHours(env, 7, 7, emitted); + EXPECT_EQUAL(emitted.size(), 2); + // The second window is a 3-6 window: the reset accumulation is 4+5+6, not 1..6 + EXPECT_EQUAL(emitted[1].value, increment(4) + increment(5) + increment(6)); + EXPECT_EQUAL(emitted[1].step, 6); + EXPECT_EQUAL(emitted[1].timespan, std::string{"3"}); + } +} + +CASE("forecast-start restart continues the accumulation from the original forecast start") { + const std::string restartOptions = R"json( "write-restart": true, "read-restart": false, + "restart-path": ".", "restart-prefix": "ForecastStartDump", "restart-lib": "eckit_codec", + "restart-time": "latest",)json"; + const std::string readOptions = R"json( "write-restart": false, "read-restart": true, + "restart-path": ".", "restart-prefix": "ForecastStartDump", "restart-lib": "eckit_codec", + "restart-time": "latest",)json"; + + { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart(restartOptions))}; + sendSimulationStart(env, DATE); + sendField(env, 0); + std::vector emitted; + sendHours(env, 1, 4, emitted); + // 0-3 was emitted before the restart, hour 4 is part of the open tail stored in the restart + EXPECT_EQUAL(emitted.size(), 1); + EXPECT_EQUAL(emitted[0].value, accumulation(3)); + + EXPECT_NO_THROW(env.process({{Message::Tag::Flush, {}, {}, {{"flushKind", "step-and-restart"}, {"step", 4}}}})); + EXPECT_EQUAL(env.debugSink().size(), 1); + EXPECT(env.debugSink().front().tag() == Message::Tag::Flush); + env.debugSink().pop(); + } + + { + auto env = MultioTestEnvironment{makePlan(R"("accumulate")", forecastStart(readOptions))}; + sendSimulationStart(env, DATE); + + std::vector emitted; + // Hour 4 may be sent again by the solver after the restart: it is recognised as already accumulated + sendHours(env, 4, 7, emitted); + EXPECT_EQUAL(emitted.size(), 1); + // 0-6 contains the hours accumulated before the restart: the origin is the original forecast start + EXPECT_EQUAL(emitted[0].value, accumulation(6)); + EXPECT_EQUAL(emitted[0].step, 6); + EXPECT_EQUAL(emitted[0].timespan, std::string{"fs"}); + EXPECT_EQUAL(emitted[0].distanceInSeconds, 6 * 3600); + + EXPECT_NO_THROW(flushLastStep(env)); + EXPECT_EQUAL(env.debugSink().size(), 2); + const auto tail = popEmitted(env); + EXPECT_EQUAL(tail.value, accumulation(7)); + EXPECT_EQUAL(tail.step, 7); + EXPECT_EQUAL(tail.timespan, std::string{"fs"}); + } +} + +} // namespace multio::test::statistics_mtg2 + +int main(int argc, char** argv) { + return eckit::testing::run_tests(argc, argv); +}