Add event aggregation and outputs.

pywr-core/src/agg_funcs/mod.rs

@@ -4,7 +4,7 @@ mod py;
 #[cfg(feature = "pyo3")]
 pub use py::PyAggFunc;
 
-use crate::recorders::PeriodValue;
+use crate::recorders::{Event, PeriodValue};
 use thiserror::Error;
 
 #[derive(Error, Debug)]
@@ -85,6 +85,85 @@ impl AggFuncF64 {
         }
     }
 
+    /// Calculate the aggregation over the given slice of `Event`.
+    ///
+    /// This function computes the aggregation based on the duration of each event in fraction days.
+    /// Only completed events (those with a defined end time) are included.
+    /// It returns an `Option<f64>`, which will be `None` if the aggregation cannot be computed (e.g., for `Mean` with no events).
+    pub fn calc_events(&self, events: &[Event]) -> Option<f64> {
+        match self {
+            Self::Sum => Some(
+                events
+                    .iter()
+                    .filter_map(|e| e.duration().map(|d| d.fractional_days()))
+                    .sum(),
+            ),
+            Self::Mean => {
+                let total_duration: f64 = events
+                    .iter()
+                    .filter_map(|e| e.duration().map(|d| d.fractional_days()))
+                    .sum();
+                let count = events.len() as f64;
+                if count == 0.0 {
+                    None
+                } else {
+                    Some(total_duration / count)
+                }
+            }
+            Self::Min => events
+                .iter()
+                .filter_map(|e| e.duration().map(|d| d.fractional_days()))
+                .min_by(|a, b| {
+                    a.partial_cmp(b)
+                        .expect("Failed to calculate minimum of event durations containing a NaN.")
+                }),
+            Self::Max => events
+                .iter()
+                .filter_map(|e| e.duration().map(|d| d.fractional_days()))
+                .max_by(|a, b| {
+                    a.partial_cmp(b)
+                        .expect("Failed to calculate maximum of event durations containing a NaN.")
+                }),
+            Self::CountNonZero => {
+                let count = events.iter().filter(|e| e.end.is_some()).count();
+                Some(count as f64)
+            }
+            Self::CountFunc { func } => {
+                let count = events
+                    .iter()
+                    .filter(|e| e.duration().map(|d| func(d.fractional_days())).unwrap_or(false))
+                    .count();
+                Some(count as f64)
+            }
+            Self::Product => {
+                let product = events
+                    .iter()
+                    .filter_map(|e| e.duration().map(|d| d.fractional_days()))
+                    .product();
+                Some(product)
+            }
+            Self::AnyNonZero { tolerance } => {
+                let any = events.iter().any(|e| {
+                    e.duration()
+                        .map(|d| d.fractional_days().abs() > *tolerance)
+                        .unwrap_or(false)
+                });
+                Some(any as u8 as f64)
+            }
+            #[cfg(feature = "pyo3")]
+            Self::Python(py_func) => {
+                let vals: Vec<f64> = events
+                    .iter()
+                    .filter_map(|e| e.duration().map(|d| d.fractional_days()))
+                    .collect();
+                match py_func.call_f64(vals) {
+                    Ok(result) => Some(result),
+                    Err(e) => panic!("Error in Python aggregation function: {}", e),
+                }
+            }
+        }
+    }
+
     /// Calculate the aggregation of the given iterator of values.
     pub fn calc_iter_f64<'a, V>(&self, values: V) -> Result<f64, AggFuncError>
     where

pywr-core/src/lib.rs

@@ -18,6 +18,7 @@ pub mod models;
 pub mod network;
 pub mod node;
 pub mod parameters;
+pub mod predicate;
 pub mod recorders;
 pub mod scenario;
 pub mod solvers;

pywr-core/src/parameters/mod.rs

@@ -81,7 +81,7 @@ use std::hash::{Hash, Hasher};
 use std::marker::PhantomData;
 use std::ops::Deref;
 use thiserror::Error;
-pub use threshold::{Predicate, ThresholdParameter};
+pub use threshold::ThresholdParameter;
 pub use vector::VectorParameter;
 
 /// Simple parameter index.

pywr-core/src/parameters/multi_threshold.rs

@@ -2,8 +2,9 @@ use crate::metric::MetricF64;
 use crate::network::Network;
 use crate::parameters::errors::{ParameterCalculationError, ParameterSetupError};
 use crate::parameters::{
-    GeneralParameter, Parameter, ParameterMeta, ParameterName, ParameterState, Predicate, downcast_internal_state_mut,
+    GeneralParameter, Parameter, ParameterMeta, ParameterName, ParameterState, downcast_internal_state_mut,
 };
+use crate::predicate::Predicate;
 use crate::scenario::ScenarioIndex;
 use crate::state::State;
 use crate::timestep::Timestep;
@@ -100,7 +101,8 @@ impl GeneralParameter<u64> for MultiThresholdParameter {
 mod tests {
     use super::MultiThresholdParameter;
     use crate::metric::MetricF64;
-    use crate::parameters::{Array1Parameter, Predicate};
+    use crate::parameters::Array1Parameter;
+    use crate::predicate::Predicate;
     use crate::test_utils::{run_and_assert_parameter_u64, simple_model};
     use ndarray::{Array1, Array2, Axis, concatenate};
 

pywr-core/src/parameters/threshold.rs

@@ -4,31 +4,11 @@ use crate::parameters::errors::{ParameterCalculationError, ParameterSetupError};
 use crate::parameters::{
     GeneralParameter, Parameter, ParameterMeta, ParameterName, ParameterState, downcast_internal_state_mut,
 };
+use crate::predicate::Predicate;
 use crate::scenario::ScenarioIndex;
 use crate::state::State;
 use crate::timestep::Timestep;
 
-pub enum Predicate {
-    LessThan,
-    GreaterThan,
-    EqualTo,
-    LessThanOrEqualTo,
-    GreaterThanOrEqualTo,
-}
-
-impl Predicate {
-    /// Apply the predicate to a value and a threshold.
-    pub fn apply(&self, value: f64, threshold: f64) -> bool {
-        match self {
-            Predicate::LessThan => value < threshold,
-            Predicate::GreaterThan => value > threshold,
-            Predicate::EqualTo => (value - threshold).abs() < 1E-6, // TODO make this a global constant
-            Predicate::LessThanOrEqualTo => value <= threshold,
-            Predicate::GreaterThanOrEqualTo => value >= threshold,
-        }
-    }
-}
-
 pub struct ThresholdParameter {
     meta: ParameterMeta,
     metric: MetricF64,

pywr-core/src/predicate.rs

@@ -0,0 +1,20 @@
+#[derive(Debug, Clone)]
+pub enum Predicate {
+    LessThan,
+    GreaterThan,
+    EqualTo,
+    LessThanOrEqualTo,
+    GreaterThanOrEqualTo,
+}
+
+impl Predicate {
+    pub fn apply(&self, a: f64, b: f64) -> bool {
+        match self {
+            Predicate::LessThan => a < b,
+            Predicate::GreaterThan => a > b,
+            Predicate::EqualTo => (a - b).abs() < 1E-6, // TODO make this a global constant
+            Predicate::LessThanOrEqualTo => a <= b,
+            Predicate::GreaterThanOrEqualTo => a >= b,
+        }
+    }
+}

pywr-core/src/recorders/aggregator/agg_func.rs

@@ -0,0 +1,87 @@
+use crate::recorders::aggregator::{Event, PeriodValue};
+
+#[derive(Clone, Debug)]
+pub enum AggregationFunction {
+    Sum,
+    Mean,
+    Min,
+    Max,
+    CountNonZero,
+    CountFunc { func: fn(f64) -> bool },
+}
+
+impl AggregationFunction {
+    /// Calculate the aggregation of the given `PeriodValue`.
+    ///
+    /// This function takes a slice of `PeriodValue<f64>` and applies the aggregation function to the values.
+    /// It returns an `Option<f64>`, which will be `None` if the aggregation cannot be computed (e.g., for `Mean` with no values).
+    ///
+    pub fn calc_period_values(&self, values: &[PeriodValue<f64>]) -> Option<f64> {
+        match self {
+            AggregationFunction::Sum => Some(values.iter().map(|v| v.value * v.duration.fractional_days()).sum()),
+            AggregationFunction::Mean => {
+                let ndays: f64 = values.iter().map(|v| v.duration.fractional_days()).sum();
+                if ndays == 0.0 {
+                    None
+                } else {
+                    let sum: f64 = values.iter().map(|v| v.value * v.duration.fractional_days()).sum();
+
+                    Some(sum / ndays)
+                }
+            }
+            AggregationFunction::Min => values.iter().map(|v| v.value).min_by(|a, b| {
+                a.partial_cmp(b)
+                    .expect("Failed to calculate minimum of values containing a NaN.")
+            }),
+            AggregationFunction::Max => values.iter().map(|v| v.value).max_by(|a, b| {
+                a.partial_cmp(b)
+                    .expect("Failed to calculate maximum of values containing a NaN.")
+            }),
+            AggregationFunction::CountNonZero => {
+                let count = values.iter().filter(|v| v.value != 0.0).count();
+                Some(count as f64)
+            }
+            AggregationFunction::CountFunc { func } => {
+                let count = values.iter().filter(|v| func(v.value)).count();
+                Some(count as f64)
+            }
+        }
+    }
+
+    pub fn calc_f64(&self, values: &[f64]) -> Option<f64> {
+        match self {
+            AggregationFunction::Sum => Some(values.iter().sum()),
+            AggregationFunction::Mean => {
+                let ndays: i64 = values.len() as i64;
+                if ndays == 0 {
+                    None
+                } else {
+                    let sum: f64 = values.iter().sum();
+                    Some(sum / ndays as f64)
+                }
+            }
+            AggregationFunction::Min => values
+                .iter()
+                .min_by(|a, b| {
+                    a.partial_cmp(b)
+                        .expect("Failed to calculate minimum of values containing a NaN.")
+                })
+                .copied(),
+            AggregationFunction::Max => values
+                .iter()
+                .max_by(|a, b| {
+                    a.partial_cmp(b)
+                        .expect("Failed to calculate maximum of values containing a NaN.")
+                })
+                .copied(),
+            AggregationFunction::CountNonZero => {
+                let count = values.iter().filter(|v| **v != 0.0).count();
+                Some(count as f64)
+            }
+            AggregationFunction::CountFunc { func } => {
+                let count = values.iter().filter(|v| func(**v)).count();
+                Some(count as f64)
+            }
+        }
+    }
+}