From ee95e02b7eb36e6d30c3a3df90dfaab2ef51e9bc Mon Sep 17 00:00:00 2001 From: James Tomlinson Date: Tue, 23 Apr 2024 17:05:22 +0100 Subject: [PATCH 1/5] refactor: Move periodic aggregator to its own sub-module. --- pywr-core/src/recorders/aggregator/mod.rs | 542 +----------------- .../src/recorders/aggregator/periodic.rs | 541 +++++++++++++++++ 2 files changed, 543 insertions(+), 540 deletions(-) create mode 100644 pywr-core/src/recorders/aggregator/periodic.rs diff --git a/pywr-core/src/recorders/aggregator/mod.rs b/pywr-core/src/recorders/aggregator/mod.rs index fa75d11b..d78ad907 100644 --- a/pywr-core/src/recorders/aggregator/mod.rs +++ b/pywr-core/src/recorders/aggregator/mod.rs @@ -1,541 +1,3 @@ -use crate::timestep::PywrDuration; -use chrono::{Datelike, Duration, NaiveDate, NaiveDateTime, NaiveTime}; -use std::num::NonZeroUsize; +mod periodic; -#[derive(Clone, Debug)] -pub enum AggregationFrequency { - Monthly, - Annual, - Days(NonZeroUsize), -} - -impl AggregationFrequency { - fn is_date_in_period(&self, period_start: &NaiveDateTime, date: &NaiveDateTime) -> bool { - match self { - Self::Monthly => (period_start.year() == date.year()) && (period_start.month() == date.month()), - Self::Annual => period_start.year() == date.year(), - Self::Days(days) => { - let period_end = *period_start + Duration::days(days.get() as i64); - (period_start <= date) && (date < &period_end) - } - } - } - - fn start_of_next_period(&self, current_date: &NaiveDateTime) -> NaiveDateTime { - match self { - Self::Monthly => { - let current_month = current_date.month(); - // Increment the year if we're in December - let year = if current_month == 12 { - current_date.year() + 1 - } else { - current_date.year() - }; - let next_month = (current_month % 12) + 1; - // 1st of the next month - // SAFETY: This should be safe to unwrap as it will always create a valid date unless - // we are at the limit of dates that are representable. - let date = NaiveDate::from_ymd_opt(year, next_month, 1).unwrap(); - NaiveDateTime::new(date, NaiveTime::default()) - } - Self::Annual => { - // 1st of January in the next year - // SAFETY: This should be safe to unwrap as it will always create a valid date unless - // we are at the limit of dates that are representable. - let date = NaiveDate::from_ymd_opt(current_date.year() + 1, 1, 1).unwrap(); - NaiveDateTime::new(date, NaiveTime::default()) - } - Self::Days(days) => *current_date + Duration::days(days.get() as i64), - } - } - - /// Split the value representing a period into multiple ['PeriodValue'] that do not cross the - /// boundary of the given period. - fn split_value_into_periods(&self, value: PeriodValue) -> Vec> { - let mut sub_values = Vec::new(); - - let mut current_date = value.start; - let end_date = value.duration + value.start; - - while current_date < end_date { - let start_of_next_period = self.start_of_next_period(¤t_date); - - let current_duration = if start_of_next_period <= end_date { - start_of_next_period - current_date - } else { - end_date - current_date - }; - - sub_values.push(PeriodValue { - start: current_date, - duration: current_duration.into(), - value: value.value, - }); - - current_date = start_of_next_period; - } - - sub_values - } -} - -#[derive(Clone, Debug)] -pub enum AggregationFunction { - Sum, - Mean, - Min, - Max, - CountNonZero, - CountFunc { func: fn(f64) -> bool }, -} - -impl AggregationFunction { - /// Calculate the aggregation of the given values. - pub fn calc_period_values(&self, values: &[PeriodValue]) -> Option { - 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 { - 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) - } - } - } -} - -#[derive(Default, Debug, Clone)] -struct PeriodicAggregatorState { - current_values: Option>>, -} - -impl PeriodicAggregatorState { - fn process_value( - &mut self, - value: PeriodValue, - agg_freq: &AggregationFrequency, - agg_func: &AggregationFunction, - ) -> Option> { - if let Some(current_values) = self.current_values.as_mut() { - // SAFETY: The current_values vector is guaranteed to contain at least one value. - let current_period_start = current_values - .first() - .expect("Aggregation state contains no values when at least one is expected.") - .start; - - // Determine if the value is in the current period - if agg_freq.is_date_in_period(¤t_period_start, &value.start) { - // New value in the current aggregation period; just append it. - current_values.push(value); - - None - } else { - // New value is part of a different period (assume the next one). - - // Calculate the aggregated value of the previous period. - let agg_period = if let Some(agg_value) = agg_func.calc_period_values(current_values) { - let agg_duration = value.start - current_period_start; - Some(PeriodValue::new(current_period_start, agg_duration.into(), agg_value)) - } else { - None - }; - - // Reset the state for the next period - current_values.clear(); - current_values.push(value); - - // Finally return the aggregated value from the previous period - agg_period - } - } else { - // No previous values defined; just append the value - self.current_values = Some(vec![value]); - - None - } - } - - fn process_value_no_period(&mut self, value: PeriodValue) { - if let Some(current_values) = self.current_values.as_mut() { - current_values.push(value); - } else { - self.current_values = Some(vec![value]); - } - } - - fn calc_aggregation(&self, agg_func: &AggregationFunction) -> Option> { - if let Some(current_values) = &self.current_values { - if let Some(agg_value) = agg_func.calc_period_values(current_values) { - // SAFETY: The current_values vector is guaranteed to contain at least one value. - let current_period_start = current_values - .first() - .expect("Aggregation state contains no values when at least one is expected.") - .start; - - let current_period_end = current_values - .last() - .expect("Aggregation state contains no values when at least one is expected.") - .start; - let current_period_duration = current_period_end - current_period_start; - Some(PeriodValue::new( - current_period_start, - current_period_duration.into(), - agg_value, - )) - } else { - None - } - } else { - None - } - } -} - -#[derive(Clone, Debug)] -struct PeriodicAggregator { - frequency: Option, - function: AggregationFunction, -} - -#[derive(Debug, Copy, Clone)] -pub struct PeriodValue { - pub start: NaiveDateTime, - pub duration: PywrDuration, - pub value: T, -} - -impl PeriodValue { - pub fn new(start: NaiveDateTime, duration: PywrDuration, value: T) -> Self { - Self { start, duration, value } - } - - /// The end of the period. - pub fn end(&self) -> NaiveDateTime { - self.duration + self.start - } -} - -impl PeriodValue> { - pub fn index(&self, index: usize) -> PeriodValue - where - T: Copy, - { - PeriodValue { - start: self.start, - duration: self.duration, - value: self.value[index], - } - } - pub fn len(&self) -> usize { - self.value.len() - } -} - -impl From<&[PeriodValue]> for PeriodValue> -where - T: Copy, -{ - fn from(values: &[PeriodValue]) -> Self { - let start = values.first().expect("Empty vector of period values.").start; - let duration = values.last().expect("Empty vector of period values.").duration; - - let value = values.iter().map(|v| v.value).collect(); - Self { start, duration, value } - } -} - -impl PeriodicAggregator { - fn setup(&self) -> PeriodicAggregatorState { - PeriodicAggregatorState::default() - } - - /// Append a new value to the aggregator. - /// - /// The new value should sequentially follow from the previously processed values. If the - /// value completes a new aggregation period then a value representing that aggregation is - /// returned. - fn process_value( - &self, - current_state: &mut PeriodicAggregatorState, - value: PeriodValue, - ) -> Option> { - // Split the given period into separate periods that align with the aggregation period. - let mut agg_value = None; - - if let Some(period) = &self.frequency { - for v in period.split_value_into_periods(value) { - let av = current_state.process_value(v, period, &self.function); - if av.is_some() { - if agg_value.is_some() { - panic!("Multiple aggregated values yielded from aggregator. This indicates that the given value spans multiple aggregation periods which is not supported.") - } - agg_value = av; - } - } - } else { - current_state.process_value_no_period(value); - } - agg_value - } - - fn calc_aggregation(&self, state: &PeriodicAggregatorState) -> Option> { - state.calc_aggregation(&self.function) - } -} - -#[derive(Debug, Clone)] -pub struct AggregatorState { - state: PeriodicAggregatorState, - child: Option>, -} - -#[derive(Clone, Debug)] -pub struct Aggregator { - agg: PeriodicAggregator, - child: Option>, -} - -impl Aggregator { - pub fn new(period: Option, function: AggregationFunction, child: Option) -> Self { - Self { - agg: PeriodicAggregator { - frequency: period, - function, - }, - child: child.map(Box::new), - } - } - - pub fn setup(&self) -> AggregatorState { - AggregatorState { - state: self.agg.setup(), - child: self.child.as_ref().map(|c| Box::new(c.setup())), - } - } - - /// Append a new value to the aggregator. - pub fn append_value(&self, state: &mut AggregatorState, value: PeriodValue) -> Option> { - let agg_value = match (&self.child, state.child.as_mut()) { - (Some(child), Some(child_state)) => child.append_value(child_state, value), - (None, None) => Some(value), - (None, Some(_)) => panic!("Aggregator state contains a child state when none is expected."), - (Some(_), None) => panic!("Aggregator state does not contain a child state when one is expected."), - }; - - if let Some(agg_value) = agg_value { - self.agg.process_value(&mut state.state, agg_value) - } else { - None - } - } - - /// Compute the final aggregation value from the current state. - /// - /// This will also compute the final aggregation value from the child aggregators if any exists. - /// This includes aggregation calculations over partial or unfinished periods. - pub fn finalise(&self, state: &mut AggregatorState) -> Option> { - let final_child_value = match (&self.child, state.child.as_mut()) { - (Some(child), Some(child_state)) => child.finalise(child_state), - (None, None) => None, - (None, Some(_)) => panic!("Aggregator state contains a child state when none is expected."), - (Some(_), None) => panic!("Aggregator state does not contain a child state when one is expected."), - }; - - // If there is a final value from the child aggregator then process it - if let Some(final_child_value) = final_child_value { - let _ = self.agg.process_value(&mut state.state, final_child_value); - } - - // Finally, compute the aggregation of the current state - self.agg.calc_aggregation(&state.state) - } - - /// Create the initial default state for the aggregator. - pub fn default_state(&self) -> AggregatorState { - let state = PeriodicAggregatorState::default(); - let child = self.child.as_ref().map(|c| Box::new(c.default_state())); - AggregatorState { state, child } - } -} - -#[cfg(test)] -mod tests { - use super::{AggregationFrequency, AggregationFunction, Aggregator, PeriodicAggregator, PeriodicAggregatorState}; - use crate::recorders::aggregator::PeriodValue; - use chrono::{Datelike, NaiveDate, TimeDelta}; - use float_cmp::assert_approx_eq; - - #[test] - fn test_periodic_aggregator() { - let agg = PeriodicAggregator { - frequency: Some(AggregationFrequency::Monthly), - function: AggregationFunction::Sum, - }; - - let mut state = PeriodicAggregatorState::default(); - - let start = NaiveDate::from_ymd_opt(2023, 1, 30) - .unwrap() - .and_hms_opt(0, 0, 0) - .unwrap(); - let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 1.0)); - assert!(agg_value.is_none()); - - let start = NaiveDate::from_ymd_opt(2023, 1, 31) - .unwrap() - .and_hms_opt(0, 0, 0) - .unwrap(); - let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 1.0)); - assert!(agg_value.is_none()); - - let start = NaiveDate::from_ymd_opt(2023, 2, 1) - .unwrap() - .and_hms_opt(0, 0, 0) - .unwrap(); - let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 1.0)); - assert!(agg_value.is_some()); - - let start = NaiveDate::from_ymd_opt(2023, 2, 2) - .unwrap() - .and_hms_opt(0, 0, 0) - .unwrap(); - let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 1.0)); - assert!(agg_value.is_none()); - } - - #[test] - fn test_nested_aggregator() { - let model_agg = PeriodicAggregator { - frequency: None, - function: AggregationFunction::Max, - }; - - let annual_agg = PeriodicAggregator { - frequency: Some(AggregationFrequency::Annual), - function: AggregationFunction::Min, - }; - - // Setup an aggregator to calculate the max of the annual minimum values - let max_annual_min = Aggregator { - agg: model_agg, - child: Some(Box::new(Aggregator { - agg: annual_agg, - child: None, - })), - }; - - let mut state = max_annual_min.default_state(); - - let mut date = NaiveDate::from_ymd_opt(2023, 1, 1) - .unwrap() - .and_hms_opt(0, 0, 0) - .unwrap(); - for _i in 0..365 * 3 { - let value = PeriodValue::new(date, TimeDelta::days(1).into(), date.year() as f64); - let _agg_value = max_annual_min.append_value(&mut state, value); - date += TimeDelta::days(1); - } - - let final_value = max_annual_min.finalise(&mut state); - - if let Some(final_value) = final_value { - assert_approx_eq!(f64, final_value.value, 2025.0); - } else { - panic!("Final value is None!") - } - } - - #[test] - fn test_sub_daily_aggregation() { - let values = vec![ - PeriodValue::new( - NaiveDate::from_ymd_opt(2023, 1, 1) - .unwrap() - .and_hms_opt(0, 0, 0) - .unwrap(), - TimeDelta::hours(1).into(), - 2.0, - ), - PeriodValue::new( - NaiveDate::from_ymd_opt(2023, 1, 1) - .unwrap() - .and_hms_opt(1, 0, 0) - .unwrap(), - TimeDelta::hours(2).into(), - 1.0, - ), - PeriodValue::new( - NaiveDate::from_ymd_opt(2023, 1, 1) - .unwrap() - .and_hms_opt(3, 0, 0) - .unwrap(), - TimeDelta::hours(1).into(), - 3.0, - ), - ]; - - let agg_value = AggregationFunction::Mean.calc_period_values(values.as_slice()).unwrap(); - assert_approx_eq!(f64, agg_value, 7.0 / 4.0); - - let agg_value = AggregationFunction::Sum.calc_period_values(values.as_slice()).unwrap(); - let expected = 2.0 * (1.0 / 24.0) + 1.0 * (2.0 / 24.0) + 3.0 * (1.0 / 24.0); - assert_approx_eq!(f64, agg_value, expected); - } -} +pub use periodic::{AggregationFrequency, AggregationFunction, Aggregator, AggregatorState, PeriodValue}; diff --git a/pywr-core/src/recorders/aggregator/periodic.rs b/pywr-core/src/recorders/aggregator/periodic.rs new file mode 100644 index 00000000..97eda1a8 --- /dev/null +++ b/pywr-core/src/recorders/aggregator/periodic.rs @@ -0,0 +1,541 @@ +use crate::timestep::PywrDuration; +use chrono::{Datelike, Duration, NaiveDate, NaiveDateTime, NaiveTime}; +use std::num::NonZeroUsize; + +#[derive(Clone, Debug)] +pub enum AggregationFrequency { + Monthly, + Annual, + Days(NonZeroUsize), +} + +impl AggregationFrequency { + fn is_date_in_period(&self, period_start: &NaiveDateTime, date: &NaiveDateTime) -> bool { + match self { + Self::Monthly => (period_start.year() == date.year()) && (period_start.month() == date.month()), + Self::Annual => period_start.year() == date.year(), + Self::Days(days) => { + let period_end = *period_start + Duration::days(days.get() as i64); + (period_start <= date) && (date < &period_end) + } + } + } + + fn start_of_next_period(&self, current_date: &NaiveDateTime) -> NaiveDateTime { + match self { + Self::Monthly => { + let current_month = current_date.month(); + // Increment the year if we're in December + let year = if current_month == 12 { + current_date.year() + 1 + } else { + current_date.year() + }; + let next_month = (current_month % 12) + 1; + // 1st of the next month + // SAFETY: This should be safe to unwrap as it will always create a valid date unless + // we are at the limit of dates that are representable. + let date = NaiveDate::from_ymd_opt(year, next_month, 1).unwrap(); + NaiveDateTime::new(date, NaiveTime::default()) + } + Self::Annual => { + // 1st of January in the next year + // SAFETY: This should be safe to unwrap as it will always create a valid date unless + // we are at the limit of dates that are representable. + let date = NaiveDate::from_ymd_opt(current_date.year() + 1, 1, 1).unwrap(); + NaiveDateTime::new(date, NaiveTime::default()) + } + Self::Days(days) => *current_date + Duration::days(days.get() as i64), + } + } + + /// Split the value representing a period into multiple ['PeriodValue'] that do not cross the + /// boundary of the given period. + fn split_value_into_periods(&self, value: PeriodValue) -> Vec> { + let mut sub_values = Vec::new(); + + let mut current_date = value.start; + let end_date = value.duration + value.start; + + while current_date < end_date { + let start_of_next_period = self.start_of_next_period(¤t_date); + + let current_duration = if start_of_next_period <= end_date { + start_of_next_period - current_date + } else { + end_date - current_date + }; + + sub_values.push(PeriodValue { + start: current_date, + duration: current_duration.into(), + value: value.value, + }); + + current_date = start_of_next_period; + } + + sub_values + } +} + +#[derive(Clone, Debug)] +pub enum AggregationFunction { + Sum, + Mean, + Min, + Max, + CountNonZero, + CountFunc { func: fn(f64) -> bool }, +} + +impl AggregationFunction { + /// Calculate the aggregation of the given values. + pub fn calc_period_values(&self, values: &[PeriodValue]) -> Option { + 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 { + 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) + } + } + } +} + +#[derive(Default, Debug, Clone)] +struct PeriodicAggregatorState { + current_values: Option>>, +} + +impl PeriodicAggregatorState { + fn process_value( + &mut self, + value: PeriodValue, + agg_freq: &AggregationFrequency, + agg_func: &AggregationFunction, + ) -> Option> { + if let Some(current_values) = self.current_values.as_mut() { + // SAFETY: The current_values vector is guaranteed to contain at least one value. + let current_period_start = current_values + .first() + .expect("Aggregation state contains no values when at least one is expected.") + .start; + + // Determine if the value is in the current period + if agg_freq.is_date_in_period(¤t_period_start, &value.start) { + // New value in the current aggregation period; just append it. + current_values.push(value); + + None + } else { + // New value is part of a different period (assume the next one). + + // Calculate the aggregated value of the previous period. + let agg_period = if let Some(agg_value) = agg_func.calc_period_values(current_values) { + let agg_duration = value.start - current_period_start; + Some(PeriodValue::new(current_period_start, agg_duration.into(), agg_value)) + } else { + None + }; + + // Reset the state for the next period + current_values.clear(); + current_values.push(value); + + // Finally return the aggregated value from the previous period + agg_period + } + } else { + // No previous values defined; just append the value + self.current_values = Some(vec![value]); + + None + } + } + + fn process_value_no_period(&mut self, value: PeriodValue) { + if let Some(current_values) = self.current_values.as_mut() { + current_values.push(value); + } else { + self.current_values = Some(vec![value]); + } + } + + fn calc_aggregation(&self, agg_func: &AggregationFunction) -> Option> { + if let Some(current_values) = &self.current_values { + if let Some(agg_value) = agg_func.calc_period_values(current_values) { + // SAFETY: The current_values vector is guaranteed to contain at least one value. + let current_period_start = current_values + .first() + .expect("Aggregation state contains no values when at least one is expected.") + .start; + + let current_period_end = current_values + .last() + .expect("Aggregation state contains no values when at least one is expected.") + .start; + let current_period_duration = current_period_end - current_period_start; + Some(PeriodValue::new( + current_period_start, + current_period_duration.into(), + agg_value, + )) + } else { + None + } + } else { + None + } + } +} + +#[derive(Clone, Debug)] +struct PeriodicAggregator { + frequency: Option, + function: AggregationFunction, +} + +#[derive(Debug, Copy, Clone)] +pub struct PeriodValue { + pub start: NaiveDateTime, + pub duration: PywrDuration, + pub value: T, +} + +impl PeriodValue { + pub fn new(start: NaiveDateTime, duration: PywrDuration, value: T) -> Self { + Self { start, duration, value } + } + + /// The end of the period. + pub fn end(&self) -> NaiveDateTime { + self.duration + self.start + } +} + +impl PeriodValue> { + pub fn index(&self, index: usize) -> PeriodValue + where + T: Copy, + { + PeriodValue { + start: self.start, + duration: self.duration, + value: self.value[index], + } + } + pub fn len(&self) -> usize { + self.value.len() + } +} + +impl From<&[PeriodValue]> for PeriodValue> +where + T: Copy, +{ + fn from(values: &[PeriodValue]) -> Self { + let start = values.first().expect("Empty vector of period values.").start; + let duration = values.last().expect("Empty vector of period values.").duration; + + let value = values.into_iter().map(|v| v.value).collect(); + Self { start, duration, value } + } +} + +impl PeriodicAggregator { + fn setup(&self) -> PeriodicAggregatorState { + PeriodicAggregatorState::default() + } + + /// Append a new value to the aggregator. + /// + /// The new value should sequentially follow from the previously processed values. If the + /// value completes a new aggregation period then a value representing that aggregation is + /// returned. + fn process_value( + &self, + current_state: &mut PeriodicAggregatorState, + value: PeriodValue, + ) -> Option> { + // Split the given period into separate periods that align with the aggregation period. + let mut agg_value = None; + + if let Some(period) = &self.frequency { + for v in period.split_value_into_periods(value) { + let av = current_state.process_value(v, period, &self.function); + if av.is_some() { + if agg_value.is_some() { + panic!("Multiple aggregated values yielded from aggregator. This indicates that the given value spans multiple aggregation periods which is not supported.") + } + agg_value = av; + } + } + } else { + current_state.process_value_no_period(value); + } + agg_value + } + + fn calc_aggregation(&self, state: &PeriodicAggregatorState) -> Option> { + state.calc_aggregation(&self.function) + } +} + +#[derive(Debug, Clone)] +pub struct AggregatorState { + state: PeriodicAggregatorState, + child: Option>, +} + +#[derive(Clone, Debug)] +pub struct Aggregator { + agg: PeriodicAggregator, + child: Option>, +} + +impl Aggregator { + pub fn new(period: Option, function: AggregationFunction, child: Option) -> Self { + Self { + agg: PeriodicAggregator { + frequency: period, + function, + }, + child: child.map(Box::new), + } + } + + pub fn setup(&self) -> AggregatorState { + AggregatorState { + state: self.agg.setup(), + child: self.child.as_ref().map(|c| Box::new(c.setup())), + } + } + + /// Append a new value to the aggregator. + pub fn append_value(&self, state: &mut AggregatorState, value: PeriodValue) -> Option> { + let agg_value = match (&self.child, state.child.as_mut()) { + (Some(child), Some(child_state)) => child.append_value(child_state, value), + (None, None) => Some(value), + (None, Some(_)) => panic!("Aggregator state contains a child state when none is expected."), + (Some(_), None) => panic!("Aggregator state does not contain a child state when one is expected."), + }; + + if let Some(agg_value) = agg_value { + self.agg.process_value(&mut state.state, agg_value) + } else { + None + } + } + + /// Compute the final aggregation value from the current state. + /// + /// This will also compute the final aggregation value from the child aggregators if any exists. + /// This includes aggregation calculations over partial or unfinished periods. + pub fn finalise(&self, state: &mut AggregatorState) -> Option> { + let final_child_value = match (&self.child, state.child.as_mut()) { + (Some(child), Some(child_state)) => child.finalise(child_state), + (None, None) => None, + (None, Some(_)) => panic!("Aggregator state contains a child state when none is expected."), + (Some(_), None) => panic!("Aggregator state does not contain a child state when one is expected."), + }; + + // If there is a final value from the child aggregator then process it + if let Some(final_child_value) = final_child_value { + let _ = self.agg.process_value(&mut state.state, final_child_value); + } + + // Finally, compute the aggregation of the current state + self.agg.calc_aggregation(&state.state) + } + + /// Create the initial default state for the aggregator. + pub fn default_state(&self) -> AggregatorState { + let state = PeriodicAggregatorState::default(); + let child = self.child.as_ref().map(|c| Box::new(c.default_state())); + AggregatorState { state, child } + } +} + +#[cfg(test)] +mod tests { + use super::{AggregationFrequency, AggregationFunction, Aggregator, PeriodicAggregator, PeriodicAggregatorState}; + use crate::recorders::aggregator::PeriodValue; + use chrono::{Datelike, NaiveDate, TimeDelta}; + use float_cmp::assert_approx_eq; + + #[test] + fn test_periodic_aggregator() { + let agg = PeriodicAggregator { + frequency: Some(AggregationFrequency::Monthly), + function: AggregationFunction::Sum, + }; + + let mut state = PeriodicAggregatorState::default(); + + let start = NaiveDate::from_ymd_opt(2023, 1, 30) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 1.0)); + assert!(agg_value.is_none()); + + let start = NaiveDate::from_ymd_opt(2023, 1, 31) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 1.0)); + assert!(agg_value.is_none()); + + let start = NaiveDate::from_ymd_opt(2023, 2, 1) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 1.0)); + assert!(agg_value.is_some()); + + let start = NaiveDate::from_ymd_opt(2023, 2, 2) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 1.0)); + assert!(agg_value.is_none()); + } + + #[test] + fn test_nested_aggregator() { + let model_agg = PeriodicAggregator { + frequency: None, + function: AggregationFunction::Max, + }; + + let annual_agg = PeriodicAggregator { + frequency: Some(AggregationFrequency::Annual), + function: AggregationFunction::Min, + }; + + // Setup an aggregator to calculate the max of the annual minimum values + let max_annual_min = Aggregator { + agg: model_agg, + child: Some(Box::new(Aggregator { + agg: annual_agg, + child: None, + })), + }; + + let mut state = max_annual_min.default_state(); + + let mut date = NaiveDate::from_ymd_opt(2023, 1, 1) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + for _i in 0..365 * 3 { + let value = PeriodValue::new(date, TimeDelta::days(1).into(), date.year() as f64); + let _agg_value = max_annual_min.append_value(&mut state, value); + date = date + TimeDelta::days(1); + } + + let final_value = max_annual_min.finalise(&mut state); + + if let Some(final_value) = final_value { + assert_approx_eq!(f64, final_value.value, 2025.0); + } else { + panic!("Final value is None!") + } + } + + #[test] + fn test_sub_daily_aggregation() { + let values = vec![ + PeriodValue::new( + NaiveDate::from_ymd_opt(2023, 1, 1) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(), + TimeDelta::hours(1).into(), + 2.0, + ), + PeriodValue::new( + NaiveDate::from_ymd_opt(2023, 1, 1) + .unwrap() + .and_hms_opt(1, 0, 0) + .unwrap(), + TimeDelta::hours(2).into(), + 1.0, + ), + PeriodValue::new( + NaiveDate::from_ymd_opt(2023, 1, 1) + .unwrap() + .and_hms_opt(3, 0, 0) + .unwrap(), + TimeDelta::hours(1).into(), + 3.0, + ), + ]; + + let agg_value = AggregationFunction::Mean.calc_period_values(values.as_slice()).unwrap(); + assert_approx_eq!(f64, agg_value, 7.0 / 4.0); + + let agg_value = AggregationFunction::Sum.calc_period_values(values.as_slice()).unwrap(); + let expected = 2.0 * (1.0 / 24.0) + 1.0 * (2.0 / 24.0) + 3.0 * (1.0 / 24.0); + assert_approx_eq!(f64, agg_value, expected); + } +} From b946948f702ae9caf7d5e5323812344d27569352 Mon Sep 17 00:00:00 2001 From: James Tomlinson Date: Wed, 24 Apr 2024 12:53:27 +0100 Subject: [PATCH 2/5] feat: WIP Event aggregator. --- pywr-core/src/parameters/threshold.rs | 20 ++-- pywr-core/src/recorders/aggregator/event.rs | 110 ++++++++++++++++++++ pywr-core/src/recorders/aggregator/mod.rs | 1 + 3 files changed, 124 insertions(+), 7 deletions(-) create mode 100644 pywr-core/src/recorders/aggregator/event.rs diff --git a/pywr-core/src/parameters/threshold.rs b/pywr-core/src/parameters/threshold.rs index fdea3d09..846b241b 100644 --- a/pywr-core/src/parameters/threshold.rs +++ b/pywr-core/src/parameters/threshold.rs @@ -17,6 +17,18 @@ pub enum Predicate { 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, + } + } +} + impl FromStr for Predicate { type Err = PywrError; @@ -94,13 +106,7 @@ impl GeneralParameter for ThresholdParameter { let threshold = self.threshold.get_value(model, state)?; let value = self.metric.get_value(model, state)?; - let active = match self.predicate { - 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, - }; + let active = self.predicate.apply(value, threshold); if active { // Update the internal state to remember we've been triggered! diff --git a/pywr-core/src/recorders/aggregator/event.rs b/pywr-core/src/recorders/aggregator/event.rs new file mode 100644 index 00000000..a3bd6525 --- /dev/null +++ b/pywr-core/src/recorders/aggregator/event.rs @@ -0,0 +1,110 @@ +use crate::parameters::Predicate; +use crate::recorders::aggregator::PeriodValue; +use chrono::NaiveDateTime; + +#[derive(Default)] +enum EventState { + #[default] + Ended, + Started(NaiveDateTime), +} + +pub struct Event { + start: NaiveDateTime, + end: Option, +} + +#[derive(Default)] +pub struct EventAggregatorState { + current: EventState, +} + +pub struct EventAggregator { + predicate: Predicate, + threshold: f64, +} + +impl EventAggregator { + fn setup(&self) -> EventAggregatorState { + EventAggregatorState::default() + } + + /// Process a new value and return an event if one has completed. + fn process_value(&self, current_state: &mut EventAggregatorState, value: PeriodValue) -> Option { + let active_now = self.predicate.apply(value.value, self.threshold); + + let (new_current, event) = match (¤t_state.current, active_now) { + (EventState::Ended, true) => { + // Start a new event + (EventState::Started(value.start), None) + } + (EventState::Started(started), false) => { + // End the current event + let event = Event { + start: *started, + end: Some(value.start), + }; + + (EventState::Ended, Some(event)) + } + (EventState::Started(started), true) => { + // Continue the current event + (EventState::Started(*started), None) + } + (EventState::Ended, false) => { + // No event to continue + (EventState::Ended, None) + } + }; + + current_state.current = new_current; + + event + } +} + +#[cfg(test)] +mod tests { + use super::{EventAggregator, EventAggregatorState}; + use crate::parameters::Predicate; + use crate::recorders::aggregator::PeriodValue; + use chrono::{NaiveDate, TimeDelta}; + + #[test] + fn test_periodic_aggregator() { + let agg = EventAggregator { + predicate: Predicate::GreaterThan, + threshold: 1.0, + }; + + let mut state = EventAggregatorState::default(); + + let start = NaiveDate::from_ymd_opt(2023, 1, 30) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 3.0)); + assert!(agg_value.is_none()); + + let start = NaiveDate::from_ymd_opt(2023, 1, 31) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 3.0)); + assert!(agg_value.is_none()); + + let start = NaiveDate::from_ymd_opt(2023, 2, 1) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 1.0)); + assert!(agg_value.is_some()); + + let start = NaiveDate::from_ymd_opt(2023, 2, 2) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + let agg_value = agg.process_value(&mut state, PeriodValue::new(start, TimeDelta::days(1).into(), 1.0)); + assert!(agg_value.is_none()); + } +} diff --git a/pywr-core/src/recorders/aggregator/mod.rs b/pywr-core/src/recorders/aggregator/mod.rs index d78ad907..e384dc9a 100644 --- a/pywr-core/src/recorders/aggregator/mod.rs +++ b/pywr-core/src/recorders/aggregator/mod.rs @@ -1,3 +1,4 @@ +mod event; mod periodic; pub use periodic::{AggregationFrequency, AggregationFunction, Aggregator, AggregatorState, PeriodValue}; From 8023d37c33a869d51ceb0b27f0020ea9c1875d4f Mon Sep 17 00:00:00 2001 From: James Tomlinson Date: Mon, 17 Feb 2025 16:07:27 +0000 Subject: [PATCH 3/5] WIP --- pywr-core/src/parameters/threshold.rs | 1 + pywr-core/src/recorders/aggregator/event.rs | 11 +- pywr-core/src/recorders/aggregator/mod.rs | 208 +++++++++++++++++- .../src/recorders/aggregator/periodic.rs | 145 ++---------- pywr-core/src/recorders/metric_set.rs | 8 +- pywr-core/src/recorders/mod.rs | 2 +- pywr-schema/src/metric_sets/mod.rs | 4 +- 7 files changed, 235 insertions(+), 144 deletions(-) diff --git a/pywr-core/src/parameters/threshold.rs b/pywr-core/src/parameters/threshold.rs index 846b241b..d4794e5e 100644 --- a/pywr-core/src/parameters/threshold.rs +++ b/pywr-core/src/parameters/threshold.rs @@ -9,6 +9,7 @@ use crate::timestep::Timestep; use crate::PywrError; use std::str::FromStr; +#[derive(Debug, Clone)] pub enum Predicate { LessThan, GreaterThan, diff --git a/pywr-core/src/recorders/aggregator/event.rs b/pywr-core/src/recorders/aggregator/event.rs index a3bd6525..ccdf4e36 100644 --- a/pywr-core/src/recorders/aggregator/event.rs +++ b/pywr-core/src/recorders/aggregator/event.rs @@ -2,7 +2,7 @@ use crate::parameters::Predicate; use crate::recorders::aggregator::PeriodValue; use chrono::NaiveDateTime; -#[derive(Default)] +#[derive(Default, Clone, Debug)] enum EventState { #[default] Ended, @@ -14,23 +14,24 @@ pub struct Event { end: Option, } -#[derive(Default)] +#[derive(Default, Debug, Clone)] pub struct EventAggregatorState { current: EventState, } +#[derive(Debug, Clone)] pub struct EventAggregator { predicate: Predicate, threshold: f64, } impl EventAggregator { - fn setup(&self) -> EventAggregatorState { + pub fn setup(&self) -> EventAggregatorState { EventAggregatorState::default() } /// Process a new value and return an event if one has completed. - fn process_value(&self, current_state: &mut EventAggregatorState, value: PeriodValue) -> Option { + pub fn process_value(&self, current_state: &mut EventAggregatorState, value: PeriodValue) -> Option { let active_now = self.predicate.apply(value.value, self.threshold); let (new_current, event) = match (¤t_state.current, active_now) { @@ -71,7 +72,7 @@ mod tests { use chrono::{NaiveDate, TimeDelta}; #[test] - fn test_periodic_aggregator() { + fn test_event_aggregator() { let agg = EventAggregator { predicate: Predicate::GreaterThan, threshold: 1.0, diff --git a/pywr-core/src/recorders/aggregator/mod.rs b/pywr-core/src/recorders/aggregator/mod.rs index e384dc9a..b0f9c3db 100644 --- a/pywr-core/src/recorders/aggregator/mod.rs +++ b/pywr-core/src/recorders/aggregator/mod.rs @@ -1,4 +1,210 @@ mod event; mod periodic; -pub use periodic::{AggregationFrequency, AggregationFunction, Aggregator, AggregatorState, PeriodValue}; +use event::{Event, EventAggregator, EventAggregatorState}; +pub use periodic::{AggregationFrequency, AggregationFunction, PeriodValue}; +use periodic::{PeriodicAggregator, PeriodicAggregatorState}; + +#[derive(Debug, Clone)] +pub enum AggregatorState { + Periodic(PeriodicAggregatorState), + Event(EventAggregatorState), +} + +impl AggregatorState { + fn as_periodic(&self) -> Option<&PeriodicAggregatorState> { + match self { + AggregatorState::Periodic(state) => Some(state), + _ => None, + } + } + + fn as_periodic_mut(&mut self) -> Option<&mut PeriodicAggregatorState> { + match self { + AggregatorState::Periodic(state) => Some(state), + _ => None, + } + } + + fn as_event(&self) -> Option<&EventAggregatorState> { + match self { + AggregatorState::Event(state) => Some(state), + _ => None, + } + } + + fn as_event_mut(&mut self) -> Option<&mut EventAggregatorState> { + match self { + AggregatorState::Event(state) => Some(state), + _ => None, + } + } +} + +#[derive(Debug, Clone)] +pub struct NestedAggregatorState { + state: AggregatorState, + child: Option>, +} + +pub enum AggregatorValue { + Periodic(PeriodValue), + Event(Event), +} + +impl From for AggregatorValue { + fn from(event: Event) -> Self { + AggregatorValue::Event(event) + } +} + +impl From> for AggregatorValue { + fn from(value: PeriodValue) -> Self { + AggregatorValue::Periodic(value) + } +} + +#[derive(Debug, Clone)] +pub enum Aggregator { + Periodic(PeriodicAggregator), + Event(EventAggregator), +} + +impl From for AggregatorState { + fn from(state: PeriodicAggregatorState) -> Self { + AggregatorState::Periodic(state) + } +} + +impl From for AggregatorState { + fn from(state: EventAggregatorState) -> Self { + AggregatorState::Event(state) + } +} + +impl Aggregator { + fn setup(&self) -> AggregatorState { + match self { + Aggregator::Periodic(_) => PeriodicAggregatorState::default().into(), + Aggregator::Event(_) => EventAggregatorState::default().into(), + } + } + + fn process_value(&self, state: &mut AggregatorState, value: PeriodValue) -> Option { + match self { + Aggregator::Periodic(agg) => agg.process_value(state.as_periodic_mut().unwrap(), value).into(), + Aggregator::Event(agg) => agg.process_value(state.as_event_mut().unwrap(), value).into(), + } + } + + fn calc_aggregation(&self, state: &AggregatorState) -> Option> { + match self { + Aggregator::Periodic(agg) => agg.calc_aggregation(state.as_periodic().unwrap()), + Aggregator::Event(_) => None, + } + } +} + +#[derive(Clone, Debug)] +pub struct NestedAggregator { + aggregator: Aggregator, + child: Option>, +} + +impl NestedAggregator { + pub fn new(aggregator: Aggregator, child: Option) -> Self { + Self { + aggregator, + child: child.map(Box::new), + } + } + + /// Create the initial default state for the aggregator. + pub fn setup(&self) -> NestedAggregatorState { + NestedAggregatorState { + state: self.aggregator.setup(), + child: self.child.as_ref().map(|c| Box::new(c.setup())), + } + } + + /// Append a new value to the aggregator. + pub fn append_value(&self, state: &mut NestedAggregatorState, value: AggregatorValue) -> Option> { + let agg_value = match (&self.child, state.child.as_mut()) { + (Some(child), Some(child_state)) => child.append_value(child_state, value), + (None, None) => Some(value), + (None, Some(_)) => panic!("Aggregator state contains a child state when none is expected."), + (Some(_), None) => panic!("Aggregator state does not contain a child state when one is expected."), + }; + + if let Some(agg_value) = agg_value { + self.aggregator.process_value(&mut state.state, agg_value) + } else { + None + } + } + + /// Compute the final aggregation value from the current state. + /// + /// This will also compute the final aggregation value from the child aggregators if any exists. + /// This includes aggregation calculations over partial or unfinished periods. + pub fn finalise(&self, state: &mut NestedAggregatorState) -> Option> { + let final_child_value = match (&self.child, state.child.as_mut()) { + (Some(child), Some(child_state)) => child.finalise(child_state), + (None, None) => None, + (None, Some(_)) => panic!("Aggregator state contains a child state when none is expected."), + (Some(_), None) => panic!("Aggregator state does not contain a child state when one is expected."), + }; + + // If there is a final value from the child aggregator then process it + if let Some(final_child_value) = final_child_value { + let _ = self.aggregator.process_value(&mut state.state, final_child_value); + } + + // Finally, compute the aggregation of the current state + self.aggregator.calc_aggregation(&state.state) + } +} + +#[cfg(test)] +mod tests { + use super::{AggregationFrequency, AggregationFunction, Aggregator, NestedAggregator, PeriodicAggregator}; + use crate::recorders::aggregator::PeriodValue; + use chrono::{Datelike, NaiveDate, TimeDelta}; + use float_cmp::assert_approx_eq; + + #[test] + fn test_nested_aggregator() { + let model_agg = PeriodicAggregator::new(None, AggregationFunction::Max); + + let annual_agg = PeriodicAggregator::new(Some(AggregationFrequency::Annual), AggregationFunction::Min); + + // Setup an aggregator to calculate the max of the annual minimum values + let max_annual_min = NestedAggregator { + aggregator: Aggregator::Periodic(model_agg), + child: Some(Box::new(NestedAggregator { + aggregator: Aggregator::Periodic(annual_agg), + child: None, + })), + }; + + let mut state = max_annual_min.setup(); + + let mut date = NaiveDate::from_ymd_opt(2023, 1, 1) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(); + for _i in 0..365 * 3 { + let value = PeriodValue::new(date, TimeDelta::days(1).into(), date.year() as f64); + let _agg_value = max_annual_min.append_value(&mut state, value); + date += TimeDelta::days(1); + } + + let final_value = max_annual_min.finalise(&mut state); + + if let Some(final_value) = final_value { + assert_approx_eq!(f64, final_value.value, 2025.0); + } else { + panic!("Final value is None!") + } + } +} diff --git a/pywr-core/src/recorders/aggregator/periodic.rs b/pywr-core/src/recorders/aggregator/periodic.rs index 97eda1a8..1e8e0508 100644 --- a/pywr-core/src/recorders/aggregator/periodic.rs +++ b/pywr-core/src/recorders/aggregator/periodic.rs @@ -162,7 +162,7 @@ impl AggregationFunction { } #[derive(Default, Debug, Clone)] -struct PeriodicAggregatorState { +pub struct PeriodicAggregatorState { current_values: Option>>, } @@ -248,12 +248,6 @@ impl PeriodicAggregatorState { } } -#[derive(Clone, Debug)] -struct PeriodicAggregator { - frequency: Option, - function: AggregationFunction, -} - #[derive(Debug, Copy, Clone)] pub struct PeriodValue { pub start: NaiveDateTime, @@ -296,14 +290,20 @@ where let start = values.first().expect("Empty vector of period values.").start; let duration = values.last().expect("Empty vector of period values.").duration; - let value = values.into_iter().map(|v| v.value).collect(); + let value = values.iter().map(|v| v.value).collect(); Self { start, duration, value } } } +#[derive(Clone, Debug)] +pub struct PeriodicAggregator { + frequency: Option, + function: AggregationFunction, +} + impl PeriodicAggregator { - fn setup(&self) -> PeriodicAggregatorState { - PeriodicAggregatorState::default() + pub fn new(frequency: Option, function: AggregationFunction) -> Self { + Self { frequency, function } } /// Append a new value to the aggregator. @@ -311,7 +311,7 @@ impl PeriodicAggregator { /// The new value should sequentially follow from the previously processed values. If the /// value completes a new aggregation period then a value representing that aggregation is /// returned. - fn process_value( + pub fn process_value( &self, current_state: &mut PeriodicAggregatorState, value: PeriodValue, @@ -335,91 +335,16 @@ impl PeriodicAggregator { agg_value } - fn calc_aggregation(&self, state: &PeriodicAggregatorState) -> Option> { + pub fn calc_aggregation(&self, state: &PeriodicAggregatorState) -> Option> { state.calc_aggregation(&self.function) } } -#[derive(Debug, Clone)] -pub struct AggregatorState { - state: PeriodicAggregatorState, - child: Option>, -} - -#[derive(Clone, Debug)] -pub struct Aggregator { - agg: PeriodicAggregator, - child: Option>, -} - -impl Aggregator { - pub fn new(period: Option, function: AggregationFunction, child: Option) -> Self { - Self { - agg: PeriodicAggregator { - frequency: period, - function, - }, - child: child.map(Box::new), - } - } - - pub fn setup(&self) -> AggregatorState { - AggregatorState { - state: self.agg.setup(), - child: self.child.as_ref().map(|c| Box::new(c.setup())), - } - } - - /// Append a new value to the aggregator. - pub fn append_value(&self, state: &mut AggregatorState, value: PeriodValue) -> Option> { - let agg_value = match (&self.child, state.child.as_mut()) { - (Some(child), Some(child_state)) => child.append_value(child_state, value), - (None, None) => Some(value), - (None, Some(_)) => panic!("Aggregator state contains a child state when none is expected."), - (Some(_), None) => panic!("Aggregator state does not contain a child state when one is expected."), - }; - - if let Some(agg_value) = agg_value { - self.agg.process_value(&mut state.state, agg_value) - } else { - None - } - } - - /// Compute the final aggregation value from the current state. - /// - /// This will also compute the final aggregation value from the child aggregators if any exists. - /// This includes aggregation calculations over partial or unfinished periods. - pub fn finalise(&self, state: &mut AggregatorState) -> Option> { - let final_child_value = match (&self.child, state.child.as_mut()) { - (Some(child), Some(child_state)) => child.finalise(child_state), - (None, None) => None, - (None, Some(_)) => panic!("Aggregator state contains a child state when none is expected."), - (Some(_), None) => panic!("Aggregator state does not contain a child state when one is expected."), - }; - - // If there is a final value from the child aggregator then process it - if let Some(final_child_value) = final_child_value { - let _ = self.agg.process_value(&mut state.state, final_child_value); - } - - // Finally, compute the aggregation of the current state - self.agg.calc_aggregation(&state.state) - } - - /// Create the initial default state for the aggregator. - pub fn default_state(&self) -> AggregatorState { - let state = PeriodicAggregatorState::default(); - let child = self.child.as_ref().map(|c| Box::new(c.default_state())); - AggregatorState { state, child } - } -} - #[cfg(test)] mod tests { - use super::{AggregationFrequency, AggregationFunction, Aggregator, PeriodicAggregator, PeriodicAggregatorState}; + use super::{AggregationFrequency, AggregationFunction, PeriodicAggregator, PeriodicAggregatorState}; use crate::recorders::aggregator::PeriodValue; - use chrono::{Datelike, NaiveDate, TimeDelta}; + use chrono::{NaiveDate, TimeDelta}; use float_cmp::assert_approx_eq; #[test] @@ -460,48 +385,6 @@ mod tests { assert!(agg_value.is_none()); } - #[test] - fn test_nested_aggregator() { - let model_agg = PeriodicAggregator { - frequency: None, - function: AggregationFunction::Max, - }; - - let annual_agg = PeriodicAggregator { - frequency: Some(AggregationFrequency::Annual), - function: AggregationFunction::Min, - }; - - // Setup an aggregator to calculate the max of the annual minimum values - let max_annual_min = Aggregator { - agg: model_agg, - child: Some(Box::new(Aggregator { - agg: annual_agg, - child: None, - })), - }; - - let mut state = max_annual_min.default_state(); - - let mut date = NaiveDate::from_ymd_opt(2023, 1, 1) - .unwrap() - .and_hms_opt(0, 0, 0) - .unwrap(); - for _i in 0..365 * 3 { - let value = PeriodValue::new(date, TimeDelta::days(1).into(), date.year() as f64); - let _agg_value = max_annual_min.append_value(&mut state, value); - date = date + TimeDelta::days(1); - } - - let final_value = max_annual_min.finalise(&mut state); - - if let Some(final_value) = final_value { - assert_approx_eq!(f64, final_value.value, 2025.0); - } else { - panic!("Final value is None!") - } - } - #[test] fn test_sub_daily_aggregation() { let values = vec![ diff --git a/pywr-core/src/recorders/metric_set.rs b/pywr-core/src/recorders/metric_set.rs index b269dbee..264e3900 100644 --- a/pywr-core/src/recorders/metric_set.rs +++ b/pywr-core/src/recorders/metric_set.rs @@ -1,6 +1,6 @@ use crate::metric::MetricF64; use crate::network::Network; -use crate::recorders::aggregator::{Aggregator, AggregatorState, PeriodValue}; +use crate::recorders::aggregator::{NestedAggregator, NestedAggregatorState, PeriodValue}; use crate::scenario::ScenarioIndex; use crate::state::State; use crate::timestep::Timestep; @@ -83,7 +83,7 @@ pub struct MetricSetState { // Populated with any yielded values from the last processing. current_values: Option>>, // If the metric set aggregates then this state tracks the aggregation of each metric - aggregation_states: Option>, + aggregation_states: Option>, } impl MetricSetState { @@ -96,12 +96,12 @@ impl MetricSetState { #[derive(Clone, Debug)] pub struct MetricSet { name: String, - aggregator: Option, + aggregator: Option, metrics: Vec, } impl MetricSet { - pub fn new(name: &str, aggregator: Option, metrics: Vec) -> Self { + pub fn new(name: &str, aggregator: Option, metrics: Vec) -> Self { Self { name: name.to_string(), aggregator, diff --git a/pywr-core/src/recorders/mod.rs b/pywr-core/src/recorders/mod.rs index 76c3fcb0..117d71b5 100644 --- a/pywr-core/src/recorders/mod.rs +++ b/pywr-core/src/recorders/mod.rs @@ -12,7 +12,7 @@ use crate::scenario::ScenarioIndex; use crate::state::State; use crate::timestep::Timestep; use crate::PywrError; -pub use aggregator::{AggregationFrequency, AggregationFunction, Aggregator}; +pub use aggregator::{AggregationFrequency, AggregationFunction, NestedAggregator}; pub use csv::{CsvLongFmtOutput, CsvLongFmtRecord, CsvWideFmtOutput}; use float_cmp::{approx_eq, ApproxEq, F64Margin}; pub use hdf::HDF5Recorder; diff --git a/pywr-schema/src/metric_sets/mod.rs b/pywr-schema/src/metric_sets/mod.rs index f6cff5d5..b341b9d7 100644 --- a/pywr-schema/src/metric_sets/mod.rs +++ b/pywr-schema/src/metric_sets/mod.rs @@ -75,9 +75,9 @@ pub struct MetricAggregator { } #[cfg(feature = "core")] -impl From for pywr_core::recorders::Aggregator { +impl From for pywr_core::recorders::NestedAggregator { fn from(value: MetricAggregator) -> Self { - pywr_core::recorders::Aggregator::new( + pywr_core::recorders::NestedAggregator::new( value.freq.map(|p| p.into()), value.func.into(), value.child.map(|a| (*a).into()), From 57ee42acf6316abcc74fc2a366bf89dc18f39128 Mon Sep 17 00:00:00 2001 From: James Tomlinson Date: Thu, 27 Mar 2025 11:46:35 +0000 Subject: [PATCH 4/5] wip: Further WIP on event outputs. --- pywr-core/src/lib.rs | 4 + pywr-core/src/models/multi.rs | 12 +- pywr-core/src/models/simple.rs | 2 + pywr-core/src/network.rs | 3 +- pywr-core/src/parameters/mod.rs | 2 +- pywr-core/src/parameters/threshold.rs | 38 +--- pywr-core/src/predicate.rs | 38 ++++ pywr-core/src/recorders/aggregator/event.rs | 13 +- pywr-core/src/recorders/aggregator/mod.rs | 86 ++++++-- .../src/recorders/aggregator/periodic.rs | 39 +++- pywr-core/src/recorders/csv.rs | 112 +++++++--- pywr-core/src/recorders/hdf.rs | 1 + pywr-core/src/recorders/memory.rs | 208 ++++++++++++++---- pywr-core/src/recorders/metric_set.rs | 65 ++++-- pywr-core/src/recorders/mod.rs | 7 +- pywr-core/src/timestep.rs | 4 +- pywr-schema/src/lib.rs | 1 + pywr-schema/src/metric_sets/mod.rs | 64 +++++- pywr-schema/src/parameters/thresholds.rs | 42 +--- pywr-schema/src/predicate.rs | 42 ++++ .../src/timeseries/align_and_resample.rs | 4 +- pywr-schema/tests/csv2.json | 13 +- pywr-schema/tests/csv3.json | 26 ++- pywr-schema/tests/memory1.json | 26 ++- 24 files changed, 605 insertions(+), 247 deletions(-) create mode 100644 pywr-core/src/predicate.rs create mode 100644 pywr-schema/src/predicate.rs diff --git a/pywr-core/src/lib.rs b/pywr-core/src/lib.rs index 2835e677..c4ec5e3f 100644 --- a/pywr-core/src/lib.rs +++ b/pywr-core/src/lib.rs @@ -14,6 +14,7 @@ use crate::parameters::{ use crate::recorders::{AggregationError, MetricSetIndex, RecorderIndex}; use crate::state::MultiValue; use crate::virtual_storage::VirtualStorageIndex; +pub use predicate::Predicate; #[cfg(feature = "pyo3")] use pyo3::{ create_exception, @@ -31,6 +32,7 @@ pub mod models; pub mod network; pub mod node; pub mod parameters; +mod predicate; pub mod recorders; pub mod scenario; pub mod solvers; @@ -213,6 +215,8 @@ pub enum PywrError { CannotSimplifyMetric, #[error("Negative factor is not allowed")] NegativeFactor, + #[error("Event value is not supported in wide format")] + EventValueInWideFormat, } // Python errors diff --git a/pywr-core/src/models/multi.rs b/pywr-core/src/models/multi.rs index 939edaf1..79108cc6 100644 --- a/pywr-core/src/models/multi.rs +++ b/pywr-core/src/models/multi.rs @@ -387,7 +387,11 @@ impl MultiNetworkModel { for (idx, entry) in self.networks.iter().enumerate() { let sub_model_ms_states = state.states.get_mut(idx).unwrap().all_metric_set_internal_states_mut(); let sub_model_recorder_states = state.recorder_states.get_mut(idx).unwrap(); - entry.network.finalise(sub_model_ms_states, sub_model_recorder_states)?; + entry.network.finalise( + self.domain.scenarios.indices(), + sub_model_ms_states, + sub_model_recorder_states, + )?; } // End the global timer and print the run statistics timings.finish(count); @@ -439,7 +443,11 @@ impl MultiNetworkModel { for (idx, entry) in self.networks.iter().enumerate() { let sub_model_ms_states = state.states.get_mut(idx).unwrap().all_metric_set_internal_states_mut(); let sub_model_recorder_states = state.recorder_states.get_mut(idx).unwrap(); - entry.network.finalise(sub_model_ms_states, sub_model_recorder_states)?; + entry.network.finalise( + self.domain.scenarios.indices(), + sub_model_ms_states, + sub_model_recorder_states, + )?; } // End the global timer and print the run statistics timings.finish(count); diff --git a/pywr-core/src/models/simple.rs b/pywr-core/src/models/simple.rs index 67a7bc78..19ed26e9 100644 --- a/pywr-core/src/models/simple.rs +++ b/pywr-core/src/models/simple.rs @@ -248,6 +248,7 @@ impl Model { } self.network.finalise( + self.domain.scenarios.indices(), state.state.all_metric_set_internal_states_mut(), &mut state.recorder_state, )?; @@ -306,6 +307,7 @@ impl Model { } self.network.finalise( + self.domain.scenarios.indices(), state.state.all_metric_set_internal_states_mut(), &mut state.recorder_state, )?; diff --git a/pywr-core/src/network.rs b/pywr-core/src/network.rs index 17c58a6e..6132ead8 100644 --- a/pywr-core/src/network.rs +++ b/pywr-core/src/network.rs @@ -345,6 +345,7 @@ impl Network { pub fn finalise( &self, + scenario_indices: &[ScenarioIndex], metric_set_states: &mut [Vec], recorder_internal_states: &mut [Option>], ) -> Result<(), PywrError> { @@ -358,7 +359,7 @@ impl Network { // Setup recorders for (recorder, internal_state) in self.recorders.iter().zip(recorder_internal_states) { - recorder.finalise(self, metric_set_states, internal_state)?; + recorder.finalise(scenario_indices, self, metric_set_states, internal_state)?; } Ok(()) diff --git a/pywr-core/src/parameters/mod.rs b/pywr-core/src/parameters/mod.rs index 4de2fbe8..e4112b4e 100644 --- a/pywr-core/src/parameters/mod.rs +++ b/pywr-core/src/parameters/mod.rs @@ -68,7 +68,7 @@ use std::fmt; use std::fmt::{Display, Formatter}; use std::marker::PhantomData; use std::ops::Deref; -pub use threshold::{Predicate, ThresholdParameter}; +pub use threshold::ThresholdParameter; pub use vector::VectorParameter; /// Simple parameter index. diff --git a/pywr-core/src/parameters/threshold.rs b/pywr-core/src/parameters/threshold.rs index d4794e5e..790270db 100644 --- a/pywr-core/src/parameters/threshold.rs +++ b/pywr-core/src/parameters/threshold.rs @@ -3,47 +3,11 @@ use crate::network::Network; use crate::parameters::{ downcast_internal_state_mut, GeneralParameter, Parameter, ParameterMeta, ParameterName, ParameterState, }; +use crate::predicate::Predicate; use crate::scenario::ScenarioIndex; use crate::state::State; use crate::timestep::Timestep; use crate::PywrError; -use std::str::FromStr; - -#[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, - } - } -} - -impl FromStr for Predicate { - type Err = PywrError; - - fn from_str(name: &str) -> Result { - match name { - "<" => Ok(Self::LessThan), - ">" => Ok(Self::GreaterThan), - "=" => Ok(Self::EqualTo), - "<=" => Ok(Self::LessThanOrEqualTo), - ">=" => Ok(Self::GreaterThanOrEqualTo), - _ => Err(PywrError::InvalidAggregationFunction(name.to_string())), - } - } -} pub struct ThresholdParameter { meta: ParameterMeta, diff --git a/pywr-core/src/predicate.rs b/pywr-core/src/predicate.rs new file mode 100644 index 00000000..762bc990 --- /dev/null +++ b/pywr-core/src/predicate.rs @@ -0,0 +1,38 @@ +use crate::PywrError; +use std::str::FromStr; + +#[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, + } + } +} + +impl FromStr for Predicate { + type Err = PywrError; + + fn from_str(name: &str) -> Result { + match name { + "<" => Ok(Self::LessThan), + ">" => Ok(Self::GreaterThan), + "=" => Ok(Self::EqualTo), + "<=" => Ok(Self::LessThanOrEqualTo), + ">=" => Ok(Self::GreaterThanOrEqualTo), + _ => Err(PywrError::InvalidAggregationFunction(name.to_string())), + } + } +} diff --git a/pywr-core/src/recorders/aggregator/event.rs b/pywr-core/src/recorders/aggregator/event.rs index ccdf4e36..0cebbdc5 100644 --- a/pywr-core/src/recorders/aggregator/event.rs +++ b/pywr-core/src/recorders/aggregator/event.rs @@ -1,4 +1,4 @@ -use crate::parameters::Predicate; +use crate::predicate::Predicate; use crate::recorders::aggregator::PeriodValue; use chrono::NaiveDateTime; @@ -9,9 +9,10 @@ enum EventState { Started(NaiveDateTime), } +#[derive(Debug, Clone, Copy)] pub struct Event { - start: NaiveDateTime, - end: Option, + pub start: NaiveDateTime, + pub end: Option, } #[derive(Default, Debug, Clone)] @@ -26,6 +27,10 @@ pub struct EventAggregator { } impl EventAggregator { + pub fn new(predicate: Predicate, threshold: f64) -> Self { + Self { predicate, threshold } + } + pub fn setup(&self) -> EventAggregatorState { EventAggregatorState::default() } @@ -67,8 +72,8 @@ impl EventAggregator { #[cfg(test)] mod tests { use super::{EventAggregator, EventAggregatorState}; - use crate::parameters::Predicate; use crate::recorders::aggregator::PeriodValue; + use crate::Predicate; use chrono::{NaiveDate, TimeDelta}; #[test] diff --git a/pywr-core/src/recorders/aggregator/mod.rs b/pywr-core/src/recorders/aggregator/mod.rs index b0f9c3db..ff13e7fd 100644 --- a/pywr-core/src/recorders/aggregator/mod.rs +++ b/pywr-core/src/recorders/aggregator/mod.rs @@ -1,9 +1,11 @@ mod event; mod periodic; -use event::{Event, EventAggregator, EventAggregatorState}; -pub use periodic::{AggregationFrequency, AggregationFunction, PeriodValue}; -use periodic::{PeriodicAggregator, PeriodicAggregatorState}; +use crate::recorders::metric_set::MetricSetOutputInfo; +use crate::timestep::TimeDomain; +pub use event::{Event, EventAggregator, EventAggregatorState}; +use periodic::PeriodicAggregatorState; +pub use periodic::{AggregationFrequency, AggregationFunction, PeriodValue, PeriodicAggregator}; #[derive(Debug, Clone)] pub enum AggregatorState { @@ -26,13 +28,6 @@ impl AggregatorState { } } - fn as_event(&self) -> Option<&EventAggregatorState> { - match self { - AggregatorState::Event(state) => Some(state), - _ => None, - } - } - fn as_event_mut(&mut self) -> Option<&mut EventAggregatorState> { match self { AggregatorState::Event(state) => Some(state), @@ -47,6 +42,7 @@ pub struct NestedAggregatorState { child: Option>, } +#[derive(Debug, Clone)] pub enum AggregatorValue { Periodic(PeriodValue), Event(Event), @@ -70,6 +66,18 @@ pub enum Aggregator { Event(EventAggregator), } +impl From for Aggregator { + fn from(agg: PeriodicAggregator) -> Self { + Aggregator::Periodic(agg) + } +} + +impl From for Aggregator { + fn from(agg: EventAggregator) -> Self { + Aggregator::Event(agg) + } +} + impl From for AggregatorState { fn from(state: PeriodicAggregatorState) -> Self { AggregatorState::Periodic(state) @@ -92,17 +100,30 @@ impl Aggregator { fn process_value(&self, state: &mut AggregatorState, value: PeriodValue) -> Option { match self { - Aggregator::Periodic(agg) => agg.process_value(state.as_periodic_mut().unwrap(), value).into(), - Aggregator::Event(agg) => agg.process_value(state.as_event_mut().unwrap(), value).into(), + Aggregator::Periodic(agg) => agg + .process_value(state.as_periodic_mut().unwrap(), value) + .map(|v| v.into()), + Aggregator::Event(agg) => agg + .process_value(state.as_event_mut().unwrap(), value) + .map(|v| v.into()), } } - fn calc_aggregation(&self, state: &AggregatorState) -> Option> { + fn calc_aggregation(&self, state: &AggregatorState) -> Option { match self { - Aggregator::Periodic(agg) => agg.calc_aggregation(state.as_periodic().unwrap()), + Aggregator::Periodic(agg) => agg.calc_aggregation(state.as_periodic().unwrap()).map(|v| v.into()), Aggregator::Event(_) => None, } } + + fn output_info(&self, time_domain: &TimeDomain) -> MetricSetOutputInfo { + match self { + Aggregator::Periodic(agg) => MetricSetOutputInfo::Periodic { + num_periods: agg.number_of_periods(time_domain), + }, + Aggregator::Event(_) => MetricSetOutputInfo::Event, + } + } } #[derive(Clone, Debug)] @@ -119,6 +140,10 @@ impl NestedAggregator { } } + pub fn output_info(&self, time_domain: &TimeDomain) -> MetricSetOutputInfo { + self.aggregator.output_info(time_domain) + } + /// Create the initial default state for the aggregator. pub fn setup(&self) -> NestedAggregatorState { NestedAggregatorState { @@ -128,7 +153,7 @@ impl NestedAggregator { } /// Append a new value to the aggregator. - pub fn append_value(&self, state: &mut NestedAggregatorState, value: AggregatorValue) -> Option> { + pub fn append_value(&self, state: &mut NestedAggregatorState, value: AggregatorValue) -> Option { let agg_value = match (&self.child, state.child.as_mut()) { (Some(child), Some(child_state)) => child.append_value(child_state, value), (None, None) => Some(value), @@ -137,7 +162,12 @@ impl NestedAggregator { }; if let Some(agg_value) = agg_value { - self.aggregator.process_value(&mut state.state, agg_value) + match agg_value { + AggregatorValue::Periodic(value) => self.aggregator.process_value(&mut state.state, value), + AggregatorValue::Event(_event) => { + panic!("It is not possible to process an event value in a nested aggregator. The event aggregator should be the top level aggregator.") + } + } } else { None } @@ -147,7 +177,7 @@ impl NestedAggregator { /// /// This will also compute the final aggregation value from the child aggregators if any exists. /// This includes aggregation calculations over partial or unfinished periods. - pub fn finalise(&self, state: &mut NestedAggregatorState) -> Option> { + pub fn finalise(&self, state: &mut NestedAggregatorState) -> Option { let final_child_value = match (&self.child, state.child.as_mut()) { (Some(child), Some(child_state)) => child.finalise(child_state), (None, None) => None, @@ -156,8 +186,15 @@ impl NestedAggregator { }; // If there is a final value from the child aggregator then process it - if let Some(final_child_value) = final_child_value { - let _ = self.aggregator.process_value(&mut state.state, final_child_value); + if let Some(agg_value) = final_child_value { + match agg_value { + AggregatorValue::Periodic(value) => { + let _ = self.aggregator.process_value(&mut state.state, value); + } + AggregatorValue::Event(_event) => { + panic!("It is not possible to process an event value in a nested aggregator. The event aggregator should be the top level aggregator.") + } + } } // Finally, compute the aggregation of the current state @@ -167,7 +204,9 @@ impl NestedAggregator { #[cfg(test)] mod tests { - use super::{AggregationFrequency, AggregationFunction, Aggregator, NestedAggregator, PeriodicAggregator}; + use super::{ + AggregationFrequency, AggregationFunction, Aggregator, AggregatorValue, NestedAggregator, PeriodicAggregator, + }; use crate::recorders::aggregator::PeriodValue; use chrono::{Datelike, NaiveDate, TimeDelta}; use float_cmp::assert_approx_eq; @@ -195,14 +234,17 @@ mod tests { .unwrap(); for _i in 0..365 * 3 { let value = PeriodValue::new(date, TimeDelta::days(1).into(), date.year() as f64); - let _agg_value = max_annual_min.append_value(&mut state, value); + let _agg_value = max_annual_min.append_value(&mut state, value.into()); date += TimeDelta::days(1); } let final_value = max_annual_min.finalise(&mut state); if let Some(final_value) = final_value { - assert_approx_eq!(f64, final_value.value, 2025.0); + match final_value { + AggregatorValue::Periodic(value) => assert_approx_eq!(f64, value.value, 2025.0), + _ => panic!("Final value is not a PeriodValue!"), + } } else { panic!("Final value is None!") } diff --git a/pywr-core/src/recorders/aggregator/periodic.rs b/pywr-core/src/recorders/aggregator/periodic.rs index 1e8e0508..fd372b05 100644 --- a/pywr-core/src/recorders/aggregator/periodic.rs +++ b/pywr-core/src/recorders/aggregator/periodic.rs @@ -1,4 +1,4 @@ -use crate::timestep::PywrDuration; +use crate::timestep::{PywrDuration, TimeDomain}; use chrono::{Datelike, Duration, NaiveDate, NaiveDateTime, NaiveTime}; use std::num::NonZeroUsize; @@ -10,6 +10,29 @@ pub enum AggregationFrequency { } impl AggregationFrequency { + /// Number of periods in the given time domain. + fn number_of_periods(&self, time_domain: &TimeDomain) -> usize { + match self { + Self::Monthly => { + let start = time_domain.first().date; + let end = time_domain.last().date; + let n_years = (end.year() - start.year()) as u32; + (n_years * 12 + end.month() - start.month()) as usize + } + Self::Annual => { + let start = time_domain.first().date; + let end = time_domain.last().date; + (end.year() - start.year()) as usize + } + Self::Days(days) => { + let start = time_domain.first().date; + let end = time_domain.last().date; + let n_days = end.signed_duration_since(start).num_days(); + (n_days / days.get() as i64) as usize + } + } + } + fn is_date_in_period(&self, period_start: &NaiveDateTime, date: &NaiveDateTime) -> bool { match self { Self::Monthly => (period_start.year() == date.year()) && (period_start.month() == date.month()), @@ -161,6 +184,12 @@ impl AggregationFunction { } } +/// State of the periodic aggregator. +/// +/// This state stores the current values, if any, that are yielded from the aggregation on the +/// given time-step. Periodic output is consistent for each metric, and therefore is stored +/// as a vec of [`PeriodValue`]s that represents the aggregated value over a period of time for all +/// metrics. #[derive(Default, Debug, Clone)] pub struct PeriodicAggregatorState { current_values: Option>>, @@ -338,6 +367,14 @@ impl PeriodicAggregator { pub fn calc_aggregation(&self, state: &PeriodicAggregatorState) -> Option> { state.calc_aggregation(&self.function) } + + /// Expected number of periods in the given time domain. + pub fn number_of_periods(&self, time_domain: &TimeDomain) -> usize { + match &self.frequency { + Some(frequency) => frequency.number_of_periods(time_domain), + None => 1, + } + } } #[cfg(test)] diff --git a/pywr-core/src/recorders/csv.rs b/pywr-core/src/recorders/csv.rs index 7d91088d..5e29821c 100644 --- a/pywr-core/src/recorders/csv.rs +++ b/pywr-core/src/recorders/csv.rs @@ -1,6 +1,7 @@ use super::{MetricSetState, PywrError, Recorder, RecorderMeta, Timestep}; use crate::models::ModelDomain; use crate::network::Network; +use crate::recorders::aggregator::AggregatorValue; use crate::recorders::metric_set::MetricSetIndex; use crate::scenario::ScenarioIndex; use crate::state::State; @@ -46,15 +47,35 @@ impl CsvWideFmtOutput { .get(*self.metric_set_idx.deref()) .ok_or(PywrError::MetricSetIndexNotFound(self.metric_set_idx))?; - if let Some(current_values) = metric_set_state.current_values() { - let values = current_values + // If the metric set has values then turn them into a row. + if metric_set_state.has_some_values() { + let values = metric_set_state + .current_values() .iter() - .map(|v| format!("{:.2}", v.value)) - .collect::>(); + .map(|maybe_v| match maybe_v { + Some(v) => match v { + AggregatorValue::Periodic(p) => Ok(format!("{:.2}", p.value)), + AggregatorValue::Event(_) => Err(PywrError::EventValueInWideFormat), + }, + None => Ok("".to_string()), // Missing value + }) + .collect::, _>>()?; // If the row is empty, add the start time if row.is_empty() { - row.push(current_values.first().unwrap().start.to_string()) + // Find the first non-None value and use that as the start time + let start = metric_set_state + .current_values() + .iter() + .find_map(|maybe_v| { + maybe_v.as_ref().and_then(|v| match v { + AggregatorValue::Periodic(p) => Some(p.start.to_string()), + AggregatorValue::Event(_) => None, + }) + }) + .unwrap_or_else(|| "unknown".to_string()); + + row.push(start) } row.extend(values); @@ -165,6 +186,7 @@ impl Recorder for CsvWideFmtOutput { fn finalise( &self, + _scenario_indices: &[ScenarioIndex], _network: &Network, metric_set_states: &[Vec], internal_state: &mut Option>, @@ -186,7 +208,7 @@ impl Recorder for CsvWideFmtOutput { } #[derive(Debug, Serialize, Deserialize)] -pub struct CsvLongFmtRecord { +pub struct CsvLongFmtValueRecord { time_start: NaiveDateTime, time_end: NaiveDateTime, scenario_index: usize, @@ -196,6 +218,16 @@ pub struct CsvLongFmtRecord { value: f64, } +#[derive(Debug, Serialize, Deserialize)] +pub struct CsvLongFmtEventRecord { + time_start: NaiveDateTime, + time_end: Option, + scenario_index: usize, + metric_set: String, + name: String, + attribute: String, +} + /// Output the values from a several [`MetricSet`]s to a CSV file in long format. /// /// The long format contains a row for each value produced by the metric set. This is useful @@ -237,34 +269,53 @@ impl CsvLongFmtOutput { .get(*metric_set_idx.deref()) .ok_or(PywrError::MetricSetIndexNotFound(*metric_set_idx))?; - if let Some(current_values) = metric_set_state.current_values() { - let metric_set = network.get_metric_set(*metric_set_idx)?; + let metric_set = network.get_metric_set(*metric_set_idx)?; - for (metric, value) in metric_set.iter_metrics().zip(current_values.iter()) { + for (metric, maybe_value) in metric_set.iter_metrics().zip(metric_set_state.current_values()) { + if let Some(value) = maybe_value { let name = metric.name().to_string(); let attribute = metric.attribute().to_string(); - let value_scaled = if let Some(decimal_places) = self.decimal_places { - let scale = 10.0_f64.powi(decimal_places.get() as i32); - (value.value * scale).round() / scale - } else { - value.value - }; - - let record = CsvLongFmtRecord { - time_start: value.start, - time_end: value.end(), - scenario_index: scenario_idx, - metric_set: metric_set.name().to_string(), - name, - attribute, - value: value_scaled, - }; - - internal - .writer - .serialize(record) - .map_err(|e| PywrError::CSVError(e.to_string()))?; + match value { + AggregatorValue::Periodic(value) => { + let value_scaled = if let Some(decimal_places) = self.decimal_places { + let scale = 10.0_f64.powi(decimal_places.get() as i32); + (value.value * scale).round() / scale + } else { + value.value + }; + + let record = CsvLongFmtValueRecord { + time_start: value.start, + time_end: value.end(), + scenario_index: scenario_idx, + metric_set: metric_set.name().to_string(), + name, + attribute, + value: value_scaled, + }; + + internal + .writer + .serialize(record) + .map_err(|e| PywrError::CSVError(e.to_string()))?; + } + AggregatorValue::Event(event) => { + let record = CsvLongFmtEventRecord { + time_start: event.start, + time_end: event.end, + scenario_index: scenario_idx, + metric_set: metric_set.name().to_string(), + name, + attribute, + }; + + internal + .writer + .serialize(record) + .map_err(|e| PywrError::CSVError(e.to_string()))?; + } + } } } } @@ -310,6 +361,7 @@ impl Recorder for CsvLongFmtOutput { fn finalise( &self, + _scenario_indices: &[ScenarioIndex], network: &Network, metric_set_states: &[Vec], internal_state: &mut Option>, diff --git a/pywr-core/src/recorders/hdf.rs b/pywr-core/src/recorders/hdf.rs index 858e2279..7a5a11cd 100644 --- a/pywr-core/src/recorders/hdf.rs +++ b/pywr-core/src/recorders/hdf.rs @@ -133,6 +133,7 @@ impl Recorder for HDF5Recorder { fn finalise( &self, + _scenario_indices: &[ScenarioIndex], _network: &Network, _metric_set_states: &[Vec], internal_state: &mut Option>, diff --git a/pywr-core/src/recorders/memory.rs b/pywr-core/src/recorders/memory.rs index fd7399e1..b14c79fe 100644 --- a/pywr-core/src/recorders/memory.rs +++ b/pywr-core/src/recorders/memory.rs @@ -1,11 +1,13 @@ use crate::models::ModelDomain; use crate::network::Network; -use crate::recorders::aggregator::PeriodValue; +use crate::recorders::aggregator::{AggregatorValue, Event, PeriodValue}; +use crate::recorders::metric_set::MetricSetOutputInfo; use crate::recorders::{AggregationFunction, MetricSetIndex, MetricSetState, Recorder, RecorderMeta}; use crate::scenario::ScenarioIndex; use crate::state::State; use crate::timestep::Timestep; use crate::PywrError; +use chrono::NaiveDateTime; use std::any::Any; use std::ops::Deref; use thiserror::Error; @@ -122,27 +124,16 @@ impl Aggregation { } } -/// Internal state for the memory recorder. +/// Periodic internal state for the memory recorder. /// /// This is a 3D array, where the first dimension is the scenario, the second dimension is the time, -/// and the third dimension is the metric. -struct InternalState { +/// and the third dimension is the metric. It is used for storing periodic output data which +/// produces a value for every scenario at the same time. +struct PeriodicInternalState { data: Vec>>>, } -impl InternalState { - fn new(num_scenarios: usize) -> Self { - let mut data: Vec>>> = Vec::with_capacity(num_scenarios); - - for _ in 0..num_scenarios { - // We can't use `Vec::with_capacity` here because we don't know the number of - // periods that will be recorded. - data.push(Vec::new()) - } - - Self { data } - } - +impl PeriodicInternalState { /// Aggregate over the saved data to a single value using the provided aggregation functions. /// /// This method will first aggregation over the metrics, then over time, and finally over the scenarios. @@ -192,6 +183,125 @@ impl InternalState { } } +struct MemoryEvent { + start: NaiveDateTime, + end: Option, + metric_index: usize, +} + +impl MemoryEvent { + fn from_event(event: Event, metric_index: usize) -> MemoryEvent { + MemoryEvent { + start: event.start, + end: event.end, + metric_index, + } + } +} + +/// Event internal state for the memory recorder. +/// +/// This is a nested vector of events where the outer vec is the length of the scenarios, +/// and the inner vector are the events for that scenario. +struct EventInternalState { + events: Vec>, +} + +/// Internal state for the memory recorder. +/// +/// The variant used depends on the type of data produced by the aggregator. +enum InternalState { + Periodic(PeriodicInternalState), + Events(EventInternalState), +} + +impl InternalState { + fn new_periodic(num_scenarios: usize, num_periods: Option) -> Self { + let mut data: Vec>>> = Vec::with_capacity(num_scenarios); + + for _ in 0..num_scenarios { + data.push(Vec::with_capacity(num_periods.unwrap_or_default())) + } + + Self::Periodic(PeriodicInternalState { data }) + } + + fn new_event(num_scenarios: usize) -> Self { + let events: Vec<_> = Vec::with_capacity(num_scenarios); + + Self::Events(EventInternalState { events }) + } + + /// Aggregate over the saved data to a single value using the provided aggregation functions. + /// + /// This method will first aggregation over the metrics, then over time, and finally over the scenarios. + fn aggregate_metric_time_scenario(&self, aggregation: &Aggregation) -> Result { + match self { + Self::Periodic(state) => state.aggregate_metric_time_scenario(aggregation), + Self::Events(_) => todo!("Cannot aggregate events over time and scenarios."), + } + } + + /// Aggregate over the saved data to a single value using the provided aggregation functions. + /// + /// This method will first aggregation over time, then over the metrics, and finally over the scenarios. + fn aggregate_time_metric_scenario(&self, aggregation: &Aggregation) -> Result { + match self { + Self::Periodic(state) => state.aggregate_time_metric_scenario(aggregation), + Self::Events(_) => todo!("Cannot aggregate events over time and scenarios."), + } + } + + fn append_value(&mut self, scenario_index: &ScenarioIndex, values: &[Option]) { + match self { + Self::Periodic(state) => { + let scenario_data = state + .data + .get_mut(scenario_index.index) + .expect("No scenario data found"); + + // Find the first non-None value and use that as the start time + let (start, duration) = values + .iter() + .find_map(|maybe_v| { + maybe_v.as_ref().and_then(|v| match v { + AggregatorValue::Periodic(p) => Some((p.start, p.duration)), + AggregatorValue::Event(_) => None, + }) + }) + .unwrap_or_else(|| panic!("Could not determine time-step information.")); + + let period_values = values + .iter() + .map(|maybe_v| match maybe_v { + Some(v) => match v { + AggregatorValue::Periodic(v) => v.value, + AggregatorValue::Event(_) => panic!("Cannot append event values to periodic data."), + }, + None => panic!("No value found for metric."), + }) + .collect::>(); + + scenario_data.push(PeriodValue::new(start, duration, period_values)); + } + Self::Events(state) => { + let scenario_data = state + .events + .get_mut(scenario_index.index) + .expect("No scenario data found"); + + for (metric_idx, value) in values.iter().enumerate() { + match value { + Some(AggregatorValue::Event(e)) => scenario_data.push(MemoryEvent::from_event(*e, metric_idx)), + Some(AggregatorValue::Periodic(_)) => panic!("Cannot append periodic values to event data."), + None => panic!("No value found for metric."), + } + } + } + } + } +} + #[derive(Default, Copy, Clone)] pub enum AggregationOrder { #[default] @@ -230,16 +340,23 @@ impl Recorder for MemoryRecorder { &self.meta } - fn setup(&self, domain: &ModelDomain, _network: &Network) -> Result>, PywrError> { - let data = InternalState::new(domain.scenarios().len()); + fn setup(&self, domain: &ModelDomain, network: &Network) -> Result>, PywrError> { + let metric_set = network.get_metric_set(self.metric_set_idx)?; + + let state = match metric_set.output_info(domain.time()) { + MetricSetOutputInfo::Periodic { num_periods } => { + InternalState::new_periodic(domain.scenarios().len(), Some(num_periods)) + } + MetricSetOutputInfo::Event => InternalState::new_event(domain.scenarios().len()), + }; - Ok(Some(Box::new(data))) + Ok(Some(Box::new(state))) } fn save( &self, _timestep: &Timestep, - _scenario_indices: &[ScenarioIndex], + scenario_indices: &[ScenarioIndex], _model: &Network, _state: &[State], metric_set_states: &[Vec], @@ -253,14 +370,14 @@ impl Recorder for MemoryRecorder { None => panic!("No internal state defined when one was expected! :("), }; - // Iterate through all of the scenario's state - for (ms_scenario_states, scenario_data) in metric_set_states.iter().zip(internal_state.data.iter_mut()) { + // Iterate through all the scenario's state + for (scenario_index, ms_scenario_states) in scenario_indices.iter().zip(metric_set_states.iter()) { let metric_set_state = ms_scenario_states .get(*self.metric_set_idx.deref()) .ok_or(PywrError::MetricSetIndexNotFound(self.metric_set_idx))?; - if let Some(current_values) = metric_set_state.current_values() { - scenario_data.push(current_values.into()); + if metric_set_state.has_some_values() { + internal_state.append_value(scenario_index, metric_set_state.current_values()); } } @@ -269,6 +386,7 @@ impl Recorder for MemoryRecorder { fn finalise( &self, + scenario_indices: &[ScenarioIndex], _network: &Network, metric_set_states: &[Vec], internal_state: &mut Option>, @@ -281,14 +399,14 @@ impl Recorder for MemoryRecorder { None => panic!("No internal state defined when one was expected! :("), }; - // Iterate through all of the scenario's state - for (ms_scenario_states, scenario_data) in metric_set_states.iter().zip(internal_state.data.iter_mut()) { + // Iterate through all the scenario's state + for (scenario_index, ms_scenario_states) in scenario_indices.iter().zip(metric_set_states.iter()) { let metric_set_state = ms_scenario_states .get(*self.metric_set_idx.deref()) .ok_or(PywrError::MetricSetIndexNotFound(self.metric_set_idx))?; - if let Some(current_values) = metric_set_state.current_values() { - scenario_data.push(current_values.into()); + if metric_set_state.has_some_values() { + internal_state.append_value(scenario_index, metric_set_state.current_values()); } } @@ -332,7 +450,7 @@ mod tests { fn test_aggregation_orders() { let num_scenarios = 2; let num_metrics = 3; - let mut state = InternalState::new(num_scenarios); + let mut state = InternalState::new_periodic(num_scenarios, None); let mut rng = ChaCha8Rng::seed_from_u64(0); let dist: Normal = Normal::new(0.0, 1.0).unwrap(); @@ -343,24 +461,26 @@ mod tests { let mut count_non_zero_by_metric = vec![0.0; num_metrics]; time_domain.timesteps().iter().for_each(|timestep| { - state.data.iter_mut().for_each(|scenario_data| { - let metric_data = (&mut rng).sample_iter(&dist).take(num_metrics).collect::>(); + if let InternalState::Periodic(state) = &mut state { + state.data.iter_mut().for_each(|scenario_data| { + let metric_data = (&mut rng).sample_iter(&dist).take(num_metrics).collect::>(); - // Compute the expected values - if metric_data.iter().sum::() > 0.0 { - count_non_zero_max += 1.0; - } - // ... and by metric - metric_data.iter().enumerate().for_each(|(i, v)| { - if *v > 0.0 { - count_non_zero_by_metric[i] += 1.0; + // Compute the expected values + if metric_data.iter().sum::() > 0.0 { + count_non_zero_max += 1.0; } - }); + // ... and by metric + metric_data.iter().enumerate().for_each(|(i, v)| { + if *v > 0.0 { + count_non_zero_by_metric[i] += 1.0; + } + }); - let metric_data = PeriodValue::new(timestep.date, timestep.duration, metric_data); + let metric_data = PeriodValue::new(timestep.date, timestep.duration, metric_data); - scenario_data.push(metric_data); - }); + scenario_data.push(metric_data); + }); + } }); let agg = Aggregation::new( diff --git a/pywr-core/src/recorders/metric_set.rs b/pywr-core/src/recorders/metric_set.rs index 264e3900..0b918b34 100644 --- a/pywr-core/src/recorders/metric_set.rs +++ b/pywr-core/src/recorders/metric_set.rs @@ -1,9 +1,9 @@ use crate::metric::MetricF64; use crate::network::Network; -use crate::recorders::aggregator::{NestedAggregator, NestedAggregatorState, PeriodValue}; +use crate::recorders::aggregator::{AggregatorValue, NestedAggregator, NestedAggregatorState, PeriodValue}; use crate::scenario::ScenarioIndex; use crate::state::State; -use crate::timestep::Timestep; +use crate::timestep::{TimeDomain, Timestep}; use crate::PywrError; use std::fmt; use std::fmt::{Display, Formatter}; @@ -80,18 +80,35 @@ impl Display for MetricSetIndex { #[derive(Debug, Clone)] pub struct MetricSetState { - // Populated with any yielded values from the last processing. - current_values: Option>>, - // If the metric set aggregates then this state tracks the aggregation of each metric + /// Populated with any yielded values from the last processing. One entry per + /// metric in the set. + current_values: Vec>, + /// If the metric set aggregates then this state tracks the aggregation of each metric aggregation_states: Option>, } impl MetricSetState { - pub fn current_values(&self) -> Option<&[PeriodValue]> { - self.current_values.as_deref() + /// Returns the current values for the metrics in the set. There is an entry for each metric + /// in the set, which will be `None` if no value was yielded for that metric. + pub fn current_values(&self) -> &[Option] { + self.current_values.as_slice() + } + + /// Helper method to determine if there are any values in the current state. + pub fn has_some_values(&self) -> bool { + self.current_values.iter().any(|v| v.is_some()) } } +/// Information about the type of output expected from a [`MetricSet`]. +pub enum MetricSetOutputInfo { + Periodic { + // The number of time periods expected in the output + num_periods: usize, + }, + Event, +} + /// A set of metrics with an optional aggregator #[derive(Clone, Debug)] pub struct MetricSet { @@ -120,7 +137,7 @@ impl MetricSet { /// Setup a new [`MetricSetState`] for this [`MetricSet`]. pub fn setup(&self) -> MetricSetState { MetricSetState { - current_values: None, + current_values: vec![None; self.metrics.len()], aggregation_states: self .aggregator .as_ref() @@ -128,6 +145,16 @@ impl MetricSet { } } + pub fn output_info(&self, time_domain: &TimeDomain) -> MetricSetOutputInfo { + match &self.aggregator { + Some(aggregator) => aggregator.output_info(time_domain), + None => MetricSetOutputInfo::Periodic { + // Without an aggregator the output will be on per time-step. + num_periods: time_domain.len(), + }, + } + } + pub fn save( &self, timestep: &Timestep, @@ -162,24 +189,14 @@ impl MetricSet { // Use a for loop instead of using an iterator because we need to execute the // `append_value` method on all aggregators. for (value, current_state) in values.iter().zip(aggregation_states.iter_mut()) { - if let Some(agg_value) = aggregator.append_value(current_state, *value) { - agg_values.push(agg_value); - } - } + let agg_value = (*value).into(); - let agg_values = if agg_values.is_empty() { - None - } else if agg_values.len() == values.len() { - Some(agg_values) - } else { - // This should never happen because the aggregator should either yield no values - // or the same number of values as the input metrics. - unreachable!("Some values were aggregated and some were not!"); - }; + agg_values.push(aggregator.append_value(current_state, agg_value)); + } internal_state.current_values = agg_values; } else { - internal_state.current_values = Some(values); + internal_state.current_values = values.into_iter().map(|v| Some(v.into())).collect(); } Ok(()) @@ -195,11 +212,11 @@ impl MetricSet { let final_values = aggregation_states .iter_mut() .map(|current_state| aggregator.finalise(current_state)) - .collect::>>(); + .collect::>(); internal_state.current_values = final_values; } else { - internal_state.current_values = None; + internal_state.current_values = vec![None; self.metrics.len()]; } } } diff --git a/pywr-core/src/recorders/mod.rs b/pywr-core/src/recorders/mod.rs index 117d71b5..97ec7069 100644 --- a/pywr-core/src/recorders/mod.rs +++ b/pywr-core/src/recorders/mod.rs @@ -12,8 +12,10 @@ use crate::scenario::ScenarioIndex; use crate::state::State; use crate::timestep::Timestep; use crate::PywrError; -pub use aggregator::{AggregationFrequency, AggregationFunction, NestedAggregator}; -pub use csv::{CsvLongFmtOutput, CsvLongFmtRecord, CsvWideFmtOutput}; +pub use aggregator::{ + AggregationFrequency, AggregationFunction, Aggregator, EventAggregator, NestedAggregator, PeriodicAggregator, +}; +pub use csv::{CsvLongFmtOutput, CsvWideFmtOutput}; use float_cmp::{approx_eq, ApproxEq, F64Margin}; pub use hdf::HDF5Recorder; pub use memory::{Aggregation, AggregationError, AggregationOrder, MemoryRecorder}; @@ -87,6 +89,7 @@ pub trait Recorder: Send + Sync { } fn finalise( &self, + _scenario_indices: &[ScenarioIndex], _network: &Network, _metric_set_states: &[Vec], _internal_state: &mut Option>, diff --git a/pywr-core/src/timestep.rs b/pywr-core/src/timestep.rs index 26c9539e..25f18b85 100644 --- a/pywr-core/src/timestep.rs +++ b/pywr-core/src/timestep.rs @@ -262,11 +262,11 @@ impl TimeDomain { self.timesteps.len() } - pub fn first_timestep(&self) -> &Timestep { + pub fn first(&self) -> &Timestep { self.timesteps.first().expect("No time-steps defined.") } - pub fn last_timestep(&self) -> &Timestep { + pub fn last(&self) -> &Timestep { self.timesteps.last().expect("No time-steps defined.") } diff --git a/pywr-schema/src/lib.rs b/pywr-schema/src/lib.rs index d265ad8a..57e852e6 100644 --- a/pywr-schema/src/lib.rs +++ b/pywr-schema/src/lib.rs @@ -13,6 +13,7 @@ pub mod model; pub mod nodes; pub mod outputs; pub mod parameters; +mod predicate; pub mod timeseries; mod v1; mod visit; diff --git a/pywr-schema/src/metric_sets/mod.rs b/pywr-schema/src/metric_sets/mod.rs index b341b9d7..2a192545 100644 --- a/pywr-schema/src/metric_sets/mod.rs +++ b/pywr-schema/src/metric_sets/mod.rs @@ -5,6 +5,7 @@ use crate::metric::Metric; use crate::model::LoadArgs; #[cfg(feature = "core")] use crate::parameters::{Parameter, PythonReturnType}; +use crate::predicate::Predicate; use pywr_schema_macros::PywrVisitPaths; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -65,23 +66,64 @@ impl From for pywr_core::recorders::AggregationFrequency { /// If the metric set has a child aggregator then the aggregation will be performed over the /// aggregated values of the child aggregator. #[derive(Deserialize, Serialize, Clone, JsonSchema)] -pub struct MetricAggregator { +#[serde(deny_unknown_fields)] +pub struct PeriodicMetricAggregator { /// Optional aggregation frequency. pub freq: Option, /// Aggregation function to apply over metric values. pub func: MetricAggFunc, - /// Optional child aggregator. - pub child: Option>, } #[cfg(feature = "core")] -impl From for pywr_core::recorders::NestedAggregator { +impl From for pywr_core::recorders::PeriodicAggregator { + fn from(value: PeriodicMetricAggregator) -> Self { + pywr_core::recorders::PeriodicAggregator::new(value.freq.map(|p| p.into()), value.func.into()) + } +} + +#[derive(Deserialize, Serialize, Clone, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct EventMetricAggregator { + pub predicate: Predicate, + pub threshold: f64, +} + +#[cfg(feature = "core")] +impl From for pywr_core::recorders::EventAggregator { + fn from(value: EventMetricAggregator) -> Self { + pywr_core::recorders::EventAggregator::new(value.predicate.into(), value.threshold) + } +} + +#[derive(Deserialize, Serialize, Clone, JsonSchema)] +#[serde(tag = "type")] +pub enum MetricAggregator { + Periodic(PeriodicMetricAggregator), + Event(EventMetricAggregator), +} + +#[cfg(feature = "core")] +impl From for pywr_core::recorders::Aggregator { fn from(value: MetricAggregator) -> Self { - pywr_core::recorders::NestedAggregator::new( - value.freq.map(|p| p.into()), - value.func.into(), - value.child.map(|a| (*a).into()), - ) + match value { + MetricAggregator::Periodic(p) => pywr_core::recorders::Aggregator::Periodic(p.into()), + MetricAggregator::Event(e) => pywr_core::recorders::Aggregator::Event(e.into()), + } + } +} + +#[derive(Deserialize, Serialize, Clone, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct NestedMetricAggregator { + pub parent: MetricAggregator, + /// Optional child aggregator. + pub child: Option>, +} + +#[cfg(feature = "core")] +impl From for pywr_core::recorders::NestedAggregator { + fn from(value: NestedMetricAggregator) -> Self { + pywr_core::recorders::NestedAggregator::new(value.parent.into(), value.child.map(|a| (*a).into())) } } @@ -90,6 +132,7 @@ impl From for pywr_core::recorders::NestedAggregator { /// The filters allow the default metrics for all nodes and/or parameters in a model /// to be added to a metric set. #[derive(Deserialize, Serialize, Clone, JsonSchema, Default)] +#[serde(deny_unknown_fields)] pub struct MetricSetFilters { #[serde(default)] all_nodes: bool, @@ -143,10 +186,11 @@ impl MetricSetFilters { /// Metrics added by the filters will be appended to any metrics specified for the metric attribute, /// if they are not a duplication. #[derive(Deserialize, Serialize, Clone, JsonSchema)] +#[serde(deny_unknown_fields)] pub struct MetricSet { pub name: String, pub metrics: Option>, - pub aggregator: Option, + pub aggregator: Option, #[serde(default)] pub filters: MetricSetFilters, } diff --git a/pywr-schema/src/parameters/thresholds.rs b/pywr-schema/src/parameters/thresholds.rs index 77c43910..4b6feddc 100644 --- a/pywr-schema/src/parameters/thresholds.rs +++ b/pywr-schema/src/parameters/thresholds.rs @@ -5,6 +5,7 @@ use crate::metric::{Metric, NodeReference}; #[cfg(feature = "core")] use crate::model::LoadArgs; use crate::parameters::{ConversionData, ParameterMeta}; +use crate::predicate::Predicate; use crate::v1::{try_convert_parameter_attr, IntoV2, TryFromV1}; use crate::ConversionError; #[cfg(feature = "core")] @@ -12,49 +13,10 @@ use pywr_core::parameters::{ParameterName, ParameterType}; use pywr_schema_macros::PywrVisitAll; use pywr_v1_schema::parameters::{ NodeThresholdParameter as NodeThresholdParameterV1, ParameterThresholdParameter as ParameterThresholdParameterV1, - Predicate as PredicateV1, StorageThresholdParameter as StorageThresholdParameterV1, + StorageThresholdParameter as StorageThresholdParameterV1, }; use schemars::JsonSchema; -#[derive(serde::Deserialize, serde::Serialize, Debug, Clone, Copy, JsonSchema, PywrVisitAll, strum_macros::Display)] -pub enum Predicate { - #[serde(alias = "<")] - LT, - #[serde(alias = ">")] - GT, - #[serde(alias = "==")] - EQ, - #[serde(alias = "<=")] - LE, - #[serde(alias = ">=")] - GE, -} - -impl From for Predicate { - fn from(v1: PredicateV1) -> Self { - match v1 { - PredicateV1::LT => Predicate::LT, - PredicateV1::GT => Predicate::GT, - PredicateV1::EQ => Predicate::EQ, - PredicateV1::LE => Predicate::LE, - PredicateV1::GE => Predicate::GE, - } - } -} - -#[cfg(feature = "core")] -impl From for pywr_core::parameters::Predicate { - fn from(p: Predicate) -> Self { - match p { - Predicate::LT => pywr_core::parameters::Predicate::LessThan, - Predicate::GT => pywr_core::parameters::Predicate::GreaterThan, - Predicate::EQ => pywr_core::parameters::Predicate::EqualTo, - Predicate::LE => pywr_core::parameters::Predicate::LessThanOrEqualTo, - Predicate::GE => pywr_core::parameters::Predicate::GreaterThanOrEqualTo, - } - } -} - /// A parameter that compares a metric against a threshold metric /// /// The metrics are compared using the given predicate and the result is returned as an index. If the comparison diff --git a/pywr-schema/src/predicate.rs b/pywr-schema/src/predicate.rs new file mode 100644 index 00000000..813aa7f5 --- /dev/null +++ b/pywr-schema/src/predicate.rs @@ -0,0 +1,42 @@ +use pywr_schema_macros::PywrVisitAll; +use pywr_v1_schema::parameters::Predicate as PredicateV1; +use schemars::JsonSchema; + +#[derive(serde::Deserialize, serde::Serialize, Debug, Clone, Copy, JsonSchema, PywrVisitAll, strum_macros::Display)] +pub enum Predicate { + #[serde(alias = "<")] + LT, + #[serde(alias = ">")] + GT, + #[serde(alias = "==")] + EQ, + #[serde(alias = "<=")] + LE, + #[serde(alias = ">=")] + GE, +} + +impl From for Predicate { + fn from(v1: PredicateV1) -> Self { + match v1 { + PredicateV1::LT => Predicate::LT, + PredicateV1::GT => Predicate::GT, + PredicateV1::EQ => Predicate::EQ, + PredicateV1::LE => Predicate::LE, + PredicateV1::GE => Predicate::GE, + } + } +} + +#[cfg(feature = "core")] +impl From for pywr_core::Predicate { + fn from(p: Predicate) -> Self { + match p { + Predicate::LT => pywr_core::Predicate::LessThan, + Predicate::GT => pywr_core::Predicate::GreaterThan, + Predicate::EQ => pywr_core::Predicate::EqualTo, + Predicate::LE => pywr_core::Predicate::LessThanOrEqualTo, + Predicate::GE => pywr_core::Predicate::GreaterThanOrEqualTo, + } + } +} diff --git a/pywr-schema/src/timeseries/align_and_resample.rs b/pywr-schema/src/timeseries/align_and_resample.rs index 803192d4..6c385343 100644 --- a/pywr-schema/src/timeseries/align_and_resample.rs +++ b/pywr-schema/src/timeseries/align_and_resample.rs @@ -90,13 +90,13 @@ pub fn align_and_resample( } fn slice_start(df: DataFrame, time_col: &str, domain: &ModelDomain) -> Result { - let start = domain.time().first_timestep().date; + let start = domain.time().first().date; let df = df.clone().lazy().filter(col(time_col).gt_eq(lit(start))).collect()?; Ok(df) } fn slice_end(df: DataFrame, time_col: &str, domain: &ModelDomain) -> Result { - let end = domain.time().last_timestep().date; + let end = domain.time().last().date; let df = df.clone().lazy().filter(col(time_col).lt_eq(lit(end))).collect()?; Ok(df) } diff --git a/pywr-schema/tests/csv2.json b/pywr-schema/tests/csv2.json index bc52ac68..2dfd0c0f 100644 --- a/pywr-schema/tests/csv2.json +++ b/pywr-schema/tests/csv2.json @@ -65,11 +65,14 @@ { "name": "nodes", "aggregator": { - "freq": { - "type": "Monthly" - }, - "func": { - "type": "Mean" + "parent": { + "type": "Periodic", + "freq": { + "type": "Monthly" + }, + "func": { + "type": "Mean" + } } }, "metrics": [ diff --git a/pywr-schema/tests/csv3.json b/pywr-schema/tests/csv3.json index 30cc37d8..a071f790 100644 --- a/pywr-schema/tests/csv3.json +++ b/pywr-schema/tests/csv3.json @@ -65,11 +65,14 @@ { "name": "nodes-monthly-mean", "aggregator": { - "freq": { - "type": "Monthly" - }, - "func": { - "type": "Mean" + "parent": { + "type": "Periodic", + "freq": { + "type": "Monthly" + }, + "func": { + "type": "Mean" + } } }, "metrics": [ @@ -82,11 +85,14 @@ { "name": "nodes-annual-mean", "aggregator": { - "freq": { - "type": "Annual" - }, - "func": { - "type": "Mean" + "parent": { + "type": "Periodic", + "freq": { + "type": "Annual" + }, + "func": { + "type": "Mean" + } } }, "metrics": [ diff --git a/pywr-schema/tests/memory1.json b/pywr-schema/tests/memory1.json index 6f347439..a00ac40b 100644 --- a/pywr-schema/tests/memory1.json +++ b/pywr-schema/tests/memory1.json @@ -65,19 +65,25 @@ { "name": "nodes", "aggregator": { - "freq": { - "type": "Annual" - }, - "func": { - "type": "CountNonZero" - }, - "child": { + "parent": { + "type": "Periodic", "freq": { - "type": "Days", - "days": 4 + "type": "Annual" }, "func": { - "type": "Min" + "type": "CountNonZero" + } + }, + "child": { + "parent": { + "type": "Periodic", + "freq": { + "type": "Days", + "days": 4 + }, + "func": { + "type": "Min" + } } } }, From f17cc20f7af993e842ccd3dd14d5e5b20491bafe Mon Sep 17 00:00:00 2001 From: James Tomlinson Date: Fri, 6 Jun 2025 16:34:59 +0100 Subject: [PATCH 5/5] feat: Further WIP on aggregation of events. --- pywr-core/src/network.rs | 2 +- .../src/recorders/aggregator/agg_func.rs | 140 ++++++++++++++++++ pywr-core/src/recorders/aggregator/event.rs | 7 + pywr-core/src/recorders/aggregator/mod.rs | 4 +- .../src/recorders/aggregator/periodic.rs | 83 +---------- pywr-core/src/recorders/csv.rs | 8 +- pywr-core/src/recorders/hdf.rs | 1 - pywr-core/src/recorders/memory.rs | 137 +++++++++++++++-- pywr-core/src/recorders/mod.rs | 1 - pywr-core/src/test_utils.rs | 13 +- 10 files changed, 297 insertions(+), 99 deletions(-) create mode 100644 pywr-core/src/recorders/aggregator/agg_func.rs diff --git a/pywr-core/src/network.rs b/pywr-core/src/network.rs index 808f6ecd..c0c97bf0 100644 --- a/pywr-core/src/network.rs +++ b/pywr-core/src/network.rs @@ -365,7 +365,7 @@ impl Network { // Setup recorders for (recorder, internal_state) in self.recorders.iter().zip(recorder_internal_states) { - recorder.finalise(scenario_indices, self, scenario_indices, metric_set_states, internal_state)?; + recorder.finalise(self, scenario_indices, metric_set_states, internal_state)?; } Ok(()) diff --git a/pywr-core/src/recorders/aggregator/agg_func.rs b/pywr-core/src/recorders/aggregator/agg_func.rs new file mode 100644 index 00000000..c4d3ac7c --- /dev/null +++ b/pywr-core/src/recorders/aggregator/agg_func.rs @@ -0,0 +1,140 @@ +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` and applies the aggregation function to the values. + /// It returns an `Option`, 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]) -> Option { + 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) + } + } + } + + /// 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`, 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 { + match self { + AggregationFunction::Sum => Some( + events + .iter() + .filter_map(|e| e.duration().map(|d| d.fractional_days())) + .sum(), + ), + AggregationFunction::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) + } + } + AggregationFunction::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.") + }), + AggregationFunction::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.") + }), + AggregationFunction::CountNonZero => { + let count = events.iter().filter(|e| e.end.is_some()).count(); + Some(count as f64) + } + AggregationFunction::CountFunc { func } => { + let count = events + .iter() + .filter(|e| e.duration().map(|d| func(d.fractional_days())).unwrap_or(false)) + .count(); + Some(count as f64) + } + } + } + + pub fn calc_f64(&self, values: &[f64]) -> Option { + 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) + } + } + } +} diff --git a/pywr-core/src/recorders/aggregator/event.rs b/pywr-core/src/recorders/aggregator/event.rs index 0cebbdc5..ddf47bdc 100644 --- a/pywr-core/src/recorders/aggregator/event.rs +++ b/pywr-core/src/recorders/aggregator/event.rs @@ -1,5 +1,6 @@ use crate::predicate::Predicate; use crate::recorders::aggregator::PeriodValue; +use crate::timestep::PywrDuration; use chrono::NaiveDateTime; #[derive(Default, Clone, Debug)] @@ -15,6 +16,12 @@ pub struct Event { pub end: Option, } +impl Event { + pub fn duration(&self) -> Option { + self.end.map(|end| (end - self.start).into()) + } +} + #[derive(Default, Debug, Clone)] pub struct EventAggregatorState { current: EventState, diff --git a/pywr-core/src/recorders/aggregator/mod.rs b/pywr-core/src/recorders/aggregator/mod.rs index ff13e7fd..9388cf8f 100644 --- a/pywr-core/src/recorders/aggregator/mod.rs +++ b/pywr-core/src/recorders/aggregator/mod.rs @@ -1,11 +1,13 @@ +mod agg_func; mod event; mod periodic; use crate::recorders::metric_set::MetricSetOutputInfo; use crate::timestep::TimeDomain; +pub use agg_func::AggregationFunction; pub use event::{Event, EventAggregator, EventAggregatorState}; use periodic::PeriodicAggregatorState; -pub use periodic::{AggregationFrequency, AggregationFunction, PeriodValue, PeriodicAggregator}; +pub use periodic::{AggregationFrequency, PeriodValue, PeriodicAggregator}; #[derive(Debug, Clone)] pub enum AggregatorState { diff --git a/pywr-core/src/recorders/aggregator/periodic.rs b/pywr-core/src/recorders/aggregator/periodic.rs index fd372b05..ceb395ef 100644 --- a/pywr-core/src/recorders/aggregator/periodic.rs +++ b/pywr-core/src/recorders/aggregator/periodic.rs @@ -1,3 +1,4 @@ +use crate::recorders::AggregationFunction; use crate::timestep::{PywrDuration, TimeDomain}; use chrono::{Datelike, Duration, NaiveDate, NaiveDateTime, NaiveTime}; use std::num::NonZeroUsize; @@ -102,88 +103,6 @@ impl AggregationFrequency { } } -#[derive(Clone, Debug)] -pub enum AggregationFunction { - Sum, - Mean, - Min, - Max, - CountNonZero, - CountFunc { func: fn(f64) -> bool }, -} - -impl AggregationFunction { - /// Calculate the aggregation of the given values. - pub fn calc_period_values(&self, values: &[PeriodValue]) -> Option { - 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 { - 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) - } - } - } -} - /// State of the periodic aggregator. /// /// This state stores the current values, if any, that are yielded from the aggregation on the diff --git a/pywr-core/src/recorders/csv.rs b/pywr-core/src/recorders/csv.rs index d20d9405..055317ea 100644 --- a/pywr-core/src/recorders/csv.rs +++ b/pywr-core/src/recorders/csv.rs @@ -209,7 +209,8 @@ pub struct CsvLongFmtValueRecord { pub struct CsvLongFmtEventRecord { time_start: NaiveDateTime, time_end: Option, - scenario_index: usize, + simulation_id: usize, + label: String, metric_set: String, name: String, attribute: String, @@ -277,7 +278,7 @@ impl CsvLongFmtOutput { time_start: value.start, time_end: value.end(), simulation_id: scenario_index.simulation_id(), - label: scenario_index.label(), + label: scenario_index.label(), metric_set: metric_set.name().to_string(), name, attribute, @@ -293,7 +294,8 @@ impl CsvLongFmtOutput { let record = CsvLongFmtEventRecord { time_start: event.start, time_end: event.end, - scenario_index: scenario_index, + simulation_id: scenario_index.simulation_id(), + label: scenario_index.label(), metric_set: metric_set.name().to_string(), name, attribute, diff --git a/pywr-core/src/recorders/hdf.rs b/pywr-core/src/recorders/hdf.rs index c325fce8..ed10b3dc 100644 --- a/pywr-core/src/recorders/hdf.rs +++ b/pywr-core/src/recorders/hdf.rs @@ -133,7 +133,6 @@ impl Recorder for HDF5Recorder { fn finalise( &self, - _scenario_indices: &[ScenarioIndex], _network: &Network, _scenario_indices: &[ScenarioIndex], _metric_set_states: &[Vec], diff --git a/pywr-core/src/recorders/memory.rs b/pywr-core/src/recorders/memory.rs index 32320889..41c5f56c 100644 --- a/pywr-core/src/recorders/memory.rs +++ b/pywr-core/src/recorders/memory.rs @@ -9,6 +9,7 @@ use crate::timestep::Timestep; use crate::PywrError; use chrono::NaiveDateTime; use std::any::Any; +use std::collections::HashMap; use std::ops::Deref; use thiserror::Error; use tracing::warn; @@ -19,6 +20,8 @@ pub enum AggregationError { AggregationFunctionNotDefined, #[error("Aggregation function failed.")] AggregationFunctionFailed, + #[error("Invalid aggregation order: {0}")] + InvalidOrder(String), } pub struct Aggregation { @@ -122,6 +125,29 @@ impl Aggregation { Ok(agg_value) } + + /// Apply the time aggregation function to the provided events. + fn apply_time_func_events(&self, events: &[Event]) -> Result { + let agg_value = if events.len() == 1 { + if self.time.is_some() { + warn!("Aggregation function defined for time, but not used.") + } + events + .first() + .expect("No events found in time series") + .duration() + .map(|d| d.fractional_days()) + .ok_or(AggregationError::AggregationFunctionFailed)? + } else { + self.time + .as_ref() + .ok_or(AggregationError::AggregationFunctionNotDefined)? + .calc_events(events) + .ok_or(AggregationError::AggregationFunctionFailed)? + }; + + Ok(agg_value) + } } /// Periodic internal state for the memory recorder. @@ -183,6 +209,7 @@ impl PeriodicInternalState { } } +#[derive(Copy, Clone)] struct MemoryEvent { start: NaiveDateTime, end: Option, @@ -199,6 +226,15 @@ impl MemoryEvent { } } +impl From for Event { + fn from(me: MemoryEvent) -> Self { + Event { + start: me.start, + end: me.end, + } + } +} + /// Event internal state for the memory recorder. /// /// This is a nested vector of events where the outer vec is the length of the scenarios, @@ -207,6 +243,42 @@ struct EventInternalState { events: Vec>, } +impl EventInternalState { + /// Aggregate over the saved data to a single value using the provided aggregation functions. + /// + /// This method will first aggregation over time, then over the metrics, and finally over the scenarios. + fn aggregate_time_metric_scenario(&self, aggregation: &Aggregation) -> Result { + let scenario_data: Vec = self + .events + .iter() + .map(|events| { + // Accumulate the events for each metric + let mut events_by_metric: HashMap> = HashMap::new(); + + for event in events { + events_by_metric + .entry(event.metric_index) + .or_default() + .push((*event).into()); + } + + // Aggregate each metric over time first. + // NB, these are not necessarily in order of the metric index. + // Some metrics may not have any events. + let metric_ts: Vec = events_by_metric + .values() + .map(|metric_events| aggregation.apply_time_func_events(metric_events)) + .collect::>()?; + + // Now aggregate over the metrics + aggregation.apply_metric_func_f64(&metric_ts) + }) + .collect::>()?; + + aggregation.apply_scenario_func(&scenario_data) + } +} + /// Internal state for the memory recorder. /// /// The variant used depends on the type of data produced by the aggregator. @@ -227,7 +299,11 @@ impl InternalState { } fn new_event(num_scenarios: usize) -> Self { - let events: Vec<_> = Vec::with_capacity(num_scenarios); + let mut events: Vec<_> = Vec::with_capacity(num_scenarios); + + for _ in 0..num_scenarios { + events.push(Vec::new()); + } Self::Events(EventInternalState { events }) } @@ -238,7 +314,9 @@ impl InternalState { fn aggregate_metric_time_scenario(&self, aggregation: &Aggregation) -> Result { match self { Self::Periodic(state) => state.aggregate_metric_time_scenario(aggregation), - Self::Events(_) => todo!("Cannot aggregate events over time and scenarios."), + Self::Events(_) => Err(AggregationError::InvalidOrder( + "Cannot aggregate over events by metric first. Events must be aggregated by time first.".to_string(), + )), } } @@ -248,7 +326,7 @@ impl InternalState { fn aggregate_time_metric_scenario(&self, aggregation: &Aggregation) -> Result { match self { Self::Periodic(state) => state.aggregate_time_metric_scenario(aggregation), - Self::Events(_) => todo!("Cannot aggregate events over time and scenarios."), + Self::Events(state) => state.aggregate_time_metric_scenario(aggregation), } } @@ -257,7 +335,7 @@ impl InternalState { Self::Periodic(state) => { let scenario_data = state .data - .get_mut(scenario_index.index) + .get_mut(scenario_index.simulation_id()) .expect("No scenario data found"); // Find the first non-None value and use that as the start time @@ -287,7 +365,7 @@ impl InternalState { Self::Events(state) => { let scenario_data = state .events - .get_mut(scenario_index.index) + .get_mut(scenario_index.simulation_id()) .expect("No scenario data found"); for (metric_idx, value) in values.iter().enumerate() { @@ -386,9 +464,8 @@ impl Recorder for MemoryRecorder { fn finalise( &self, - scenario_indices: &[ScenarioIndex], _network: &Network, - _scenario_indices: &[ScenarioIndex], + scenario_indices: &[ScenarioIndex], metric_set_states: &[Vec], internal_state: &mut Option>, ) -> Result<(), PywrError> { @@ -438,10 +515,11 @@ impl Recorder for MemoryRecorder { #[cfg(test)] mod tests { use super::{Aggregation, InternalState}; - use crate::recorders::aggregator::PeriodValue; + use crate::recorders::aggregator::{AggregatorValue, Event, PeriodValue}; use crate::recorders::AggregationFunction; - use crate::test_utils::default_timestepper; + use crate::test_utils::{default_timestepper, test_scenario_domain}; use crate::timestep::TimeDomain; + use chrono::NaiveDate; use float_cmp::assert_approx_eq; use rand::{Rng, SeedableRng}; use rand_chacha::ChaCha8Rng; @@ -495,4 +573,45 @@ mod tests { let agg_value = state.aggregate_time_metric_scenario(&agg).expect("Aggregation failed"); assert_approx_eq!(f64, agg_value, count_non_zero_by_metric.iter().sum()); } + + #[test] + fn test_memory_event_aggregation() { + let num_scenarios = 2; + let num_metrics = 3; + + let mut state = InternalState::new_event(num_scenarios); + + let scenario_domain = test_scenario_domain(num_scenarios); + + for scenario_index in scenario_domain.indices() { + for event_index in 0..4 { + // Create an event with a known start and end time + let start = NaiveDate::from_ymd_opt(2016, event_index + 1, 8).unwrap(); + let end = start + chrono::Duration::days(event_index as i64 + 1); + + let events: Vec<_> = (0..num_metrics) + .map(|_| { + let e = Event { + start: start.into(), + end: Some(end.into()), + }; + Some(AggregatorValue::Event(e)) + }) + .collect(); + + state.append_value(scenario_index, &events); + } + } + + // This should be the total duration of all the events + let agg = Aggregation::new( + Some(AggregationFunction::Sum), + Some(AggregationFunction::Sum), + Some(AggregationFunction::Sum), + ); + + let expected_total_duration = num_scenarios as f64 * num_metrics as f64 * (1.0 + 2.0 + 3.0 + 4.0); + let agg_value = state.aggregate_time_metric_scenario(&agg).expect("Aggregation failed"); + assert_approx_eq!(f64, agg_value, expected_total_duration); + } } diff --git a/pywr-core/src/recorders/mod.rs b/pywr-core/src/recorders/mod.rs index a2627537..bc54afa4 100644 --- a/pywr-core/src/recorders/mod.rs +++ b/pywr-core/src/recorders/mod.rs @@ -89,7 +89,6 @@ pub trait Recorder: Send + Sync { } fn finalise( &self, - _scenario_indices: &[ScenarioIndex], _network: &Network, _scenario_indices: &[ScenarioIndex], _metric_set_states: &[Vec], diff --git a/pywr-core/src/test_utils.rs b/pywr-core/src/test_utils.rs index 621a4db0..f27f00ab 100644 --- a/pywr-core/src/test_utils.rs +++ b/pywr-core/src/test_utils.rs @@ -6,7 +6,7 @@ use crate::network::Network; use crate::node::StorageInitialVolume; use crate::parameters::{AggFunc, AggregatedParameter, Array2Parameter, ConstantParameter, GeneralParameter}; use crate::recorders::AssertionRecorder; -use crate::scenario::{ScenarioDomainBuilder, ScenarioGroupBuilder}; +use crate::scenario::{ScenarioDomain, ScenarioDomainBuilder, ScenarioGroupBuilder}; #[cfg(feature = "cbc")] use crate::solvers::CbcSolver; #[cfg(feature = "ipm-ocl")] @@ -54,6 +54,17 @@ pub fn default_model() -> Model { Model::new(domain, network) } +/// Create a test scenario domain with a single scenario group containing the specified number of scenarios. +pub fn test_scenario_domain(num_scenarios: usize) -> ScenarioDomain { + let mut scenario_builder = ScenarioDomainBuilder::default(); + let scenario_group = ScenarioGroupBuilder::new("test-scenario", num_scenarios) + .build() + .unwrap(); + scenario_builder = scenario_builder.with_group(scenario_group).unwrap(); + + scenario_builder.build().expect("Failed to build Scenario domain.") +} + /// Create a simple test network with three nodes. pub fn simple_network(network: &mut Network, inflow_scenario_index: usize, num_inflow_scenarios: usize) { let input_node = network.add_input_node("input", None).unwrap();