ferroui_dynamic_data/aggregation/
aggregation_ex.rs

1//! `AggregationEx` (`Aggregation/AggregationEx.cs`): change sets shaped for aggregations
2//! (`ForAggregation`), the invalidation of an aggregating sequence (`InvalidateWhen`) and the
3//! accumulator of the aggregations (`Accumulate`), with the traits that carry every operator of
4//! the namespace `DynamicData.Aggregation`.
5//!
6//! Upstream's static classes `AggregationEx`, `AvgEx`, `CountEx`, `MaxEx`, `StdDevEx` and `SumEx`
7//! extend three kinds of sequences. The port has one trait per kind, each implemented for that
8//! kind only:
9//! - [`AggregationCacheEx`] for the sequences of change sets of caches (any [`cache::IChangeSet`]);
10//! - [`AggregationListEx`] for those of lists (any [`list::IChangeSet`]);
11//! - [`AggregateChangeSetEx`] for the sequences of aggregate change sets
12//!   ([`AggregateChangeSets`]);
13//! - and [`InvalidateWhenEx`] for every sequence.
14//!
15//! Naming (DEVIATIONS.md, DynamicData): `Count`, `IsEmpty` and `Sum` are names of operators of
16//! `ferroui_reactive::ObservableExt`, which applies to the same sequences, so they are
17//! `count_cache`, `is_empty_cache`, `sum_cache` (caches), `count_list`, `is_empty_list`,
18//! `sum_list` (lists) and `count_aggregate`, `sum_aggregate` (aggregate change sets). `Avg`,
19//! `Maximum`, `Minimum`, `StdDev` and `IsNotEmpty` keep their names. The default arguments of
20//! upstream (`emptyValue`) are the `_with` forms.
21//!
22//! [`cache::IChangeSet`]: crate::cache::IChangeSet
23//! [`list::IChangeSet`]: crate::list::IChangeSet
24
25use super::{
26    cache_aggregate, list_aggregate, AggregateChangeSets, AggregateNumber, AggregateType, AggregateValue, IAggregateChangeSet, MaxOrMin, StdDevValue,
27};
28use crate::cache::ObservableCacheEx;
29use crate::list::ObservableListEx;
30use ferroui_reactive::ObservableExt;
31use ferroui_system::IObservable;
32use std::hash::Hash;
33use std::rc::Rc;
34
35/// `Accumulate<TObject, TResult>(IObservable<IAggregateChangeSet<TObject>>, ..)`: from `seed`,
36/// `add_action` of the accessed value of each added item and `remove_action` of each removed one;
37/// one value per change set.
38pub(crate) fn accumulate<T: 'static, R: Clone + 'static>(
39    source: AggregateChangeSets<T>,
40    seed: R,
41    accessor: impl Fn(&T) -> R + 'static,
42    add_action: impl Fn(R, R) -> R + 'static,
43    remove_action: impl Fn(R, R) -> R + 'static,
44) -> Rc<dyn IObservable<R>> {
45    source.scan_with(seed, move |state: R, changes: Rc<dyn IAggregateChangeSet<T>>| {
46        changes.items().fold(state, |current, aggregate_item| match aggregate_item.aggregate_type() {
47            AggregateType::Add => add_action(current, accessor(aggregate_item.item())),
48            AggregateType::Remove => remove_action(current, accessor(aggregate_item.item())),
49        })
50    })
51}
52
53/// The aggregations of caches (the overloads of `AggregationEx`, `AvgEx`, `CountEx`, `MaxEx`,
54/// `StdDevEx` and `SumEx` on `IObservable<IChangeSet<TObject, TKey>>`).
55pub trait AggregationCacheEx<T: Clone + 'static, K: Eq + Hash + Clone + 'static> {
56    /// The change sets as aggregate change sets: an add, a remove, or an update as the remove of
57    /// the previous item and the add of the current one; refreshes and moves are left out
58    /// (`ForAggregation<TObject, TKey>()`).
59    fn for_aggregation(&self) -> AggregateChangeSets<T>;
60
61    /// The average of the values of the items after each change set, zero when there are none
62    /// (`Avg(Func<TObject, TValue>)`).
63    fn avg<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>>;
64
65    /// [`avg`](Self::avg) with the value when there are no items
66    /// (`Avg(Func<TObject, TValue>, TValue emptyValue)`).
67    fn avg_with<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static, empty_value: V::Number) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>>;
68
69    /// The count of the items after each change set (`Count<TObject, TKey>()`, and
70    /// `Count<TObject>(IObservable<IDistinctChangeSet<TObject>>)`).
71    fn count_cache(&self) -> Rc<dyn IObservable<usize>>;
72
73    /// True at once, then whether there are no items after each change set (`IsEmpty`).
74    fn is_empty_cache(&self) -> Rc<dyn IObservable<bool>>;
75
76    /// False at once, then whether there are items after each change set (`IsNotEmpty`).
77    fn is_not_empty(&self) -> Rc<dyn IObservable<bool>>;
78
79    /// The greatest value of the items, `R::default()` when there are none, each time it changes
80    /// (`Maximum(Func<TObject, TResult>)`).
81    fn maximum<R: Copy + PartialOrd + Default + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static) -> Rc<dyn IObservable<R>>;
82
83    /// [`maximum`](Self::maximum) with the value when there are no items
84    /// (`Maximum(Func<TObject, TResult>, TResult emptyValue)`).
85    fn maximum_with<R: Copy + PartialOrd + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static, empty_value: R) -> Rc<dyn IObservable<R>>;
86
87    /// The least value of the items, `R::default()` when there are none, each time it changes
88    /// (`Minimum(Func<TObject, TResult>)`).
89    fn minimum<R: Copy + PartialOrd + Default + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static) -> Rc<dyn IObservable<R>>;
90
91    /// [`minimum`](Self::minimum) with the value when there are no items
92    /// (`Minimum(Func<TObject, TResult>, TResult emptyValue)`).
93    fn minimum_with<R: Copy + PartialOrd + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static, empty_value: R) -> Rc<dyn IObservable<R>>;
94
95    /// The sample standard deviation of the values of the items after each change set,
96    /// `fallback_value` for fewer than two items (`StdDev(Func<TObject, TValue>, TValue
97    /// fallbackValue)`).
98    fn std_dev<V: StdDevValue>(&self, value_selector: impl Fn(&T) -> V + 'static, fallback_value: V) -> Rc<dyn IObservable<f64>>;
99
100    /// The sum of the values of the items after each change set (`Sum(Func<TObject, TValue>)`).
101    fn sum_cache<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<V::Number>>;
102}
103
104impl<T: Clone + 'static, K: Eq + Hash + Clone + 'static, C: crate::cache::IChangeSet<T, K>> AggregationCacheEx<T, K> for Rc<dyn IObservable<C>> {
105    fn for_aggregation(&self) -> AggregateChangeSets<T> {
106        ObservableCacheEx::<T, K>::change_sets(self).select(cache_aggregate::<T, K>)
107    }
108
109    fn avg<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>> {
110        AggregationCacheEx::<T, K>::avg_with(self, value_selector, <V::Number as AggregateNumber>::ZERO)
111    }
112
113    fn avg_with<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static, empty_value: V::Number) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>> {
114        super::avg_ex::avg(AggregationCacheEx::<T, K>::for_aggregation(self), value_selector, empty_value)
115    }
116
117    fn count_cache(&self) -> Rc<dyn IObservable<usize>> {
118        super::count_ex::count(AggregationCacheEx::<T, K>::for_aggregation(self))
119    }
120
121    fn is_empty_cache(&self) -> Rc<dyn IObservable<bool>> {
122        super::count_ex::is_empty(AggregationCacheEx::<T, K>::for_aggregation(self))
123    }
124
125    fn is_not_empty(&self) -> Rc<dyn IObservable<bool>> {
126        super::count_ex::is_not_empty(AggregationCacheEx::<T, K>::for_aggregation(self))
127    }
128
129    fn maximum<R: Copy + PartialOrd + Default + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static) -> Rc<dyn IObservable<R>> {
130        AggregationCacheEx::<T, K>::maximum_with(self, value_selector, R::default())
131    }
132
133    fn maximum_with<R: Copy + PartialOrd + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static, empty_value: R) -> Rc<dyn IObservable<R>> {
134        let source = super::max_ex::cache_changes_and_collection(ObservableCacheEx::<T, K>::change_sets(self));
135        super::max_ex::calculate(source, Rc::new(value_selector), MaxOrMin::Max, empty_value)
136    }
137
138    fn minimum<R: Copy + PartialOrd + Default + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static) -> Rc<dyn IObservable<R>> {
139        AggregationCacheEx::<T, K>::minimum_with(self, value_selector, R::default())
140    }
141
142    fn minimum_with<R: Copy + PartialOrd + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static, empty_value: R) -> Rc<dyn IObservable<R>> {
143        let source = super::max_ex::cache_changes_and_collection(ObservableCacheEx::<T, K>::change_sets(self));
144        super::max_ex::calculate(source, Rc::new(value_selector), MaxOrMin::Min, empty_value)
145    }
146
147    fn std_dev<V: StdDevValue>(&self, value_selector: impl Fn(&T) -> V + 'static, fallback_value: V) -> Rc<dyn IObservable<f64>> {
148        super::std_dev_ex::std_dev(AggregationCacheEx::<T, K>::for_aggregation(self), value_selector, fallback_value)
149    }
150
151    fn sum_cache<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<V::Number>> {
152        super::sum_ex::sum(AggregationCacheEx::<T, K>::for_aggregation(self), value_selector)
153    }
154}
155
156/// The aggregations of lists (the overloads of `AggregationEx`, `AvgEx`, `CountEx`, `MaxEx`,
157/// `StdDevEx` and `SumEx` on `IObservable<IChangeSet<T>>`).
158pub trait AggregationListEx<T: Clone + PartialEq + 'static> {
159    /// The change sets as aggregate change sets: adds and ranges of adds, removes and ranges of
160    /// removes and clears, or a replace as the remove of the previous item and the add of the
161    /// current one; refreshes and moves are left out (`ForAggregation<TObject>()`).
162    fn for_aggregation(&self) -> AggregateChangeSets<T>;
163
164    /// The average of the values of the items after each change set, zero when there are none
165    /// (`Avg(Func<T, TValue>)`).
166    fn avg<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>>;
167
168    /// [`avg`](Self::avg) with the value when there are no items
169    /// (`Avg(Func<T, TValue>, TValue emptyValue)`).
170    fn avg_with<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static, empty_value: V::Number) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>>;
171
172    /// The count of the items after each change set (`Count<TObject>()`).
173    fn count_list(&self) -> Rc<dyn IObservable<usize>>;
174
175    /// True at once, then whether there are no items after each change set (`IsEmpty`).
176    fn is_empty_list(&self) -> Rc<dyn IObservable<bool>>;
177
178    /// False at once, then whether there are items after each change set (`IsNotEmpty`).
179    fn is_not_empty(&self) -> Rc<dyn IObservable<bool>>;
180
181    /// The greatest value of the items, `R::default()` when there are none, each time it changes
182    /// (`Maximum(Func<TObject, TResult>)`).
183    fn maximum<R: Copy + PartialOrd + Default + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static) -> Rc<dyn IObservable<R>>;
184
185    /// [`maximum`](Self::maximum) with the value when there are no items
186    /// (`Maximum(Func<TObject, TResult>, TResult emptyValue)`).
187    fn maximum_with<R: Copy + PartialOrd + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static, empty_value: R) -> Rc<dyn IObservable<R>>;
188
189    /// The least value of the items, `R::default()` when there are none, each time it changes
190    /// (`Minimum(Func<TObject, TResult>)`).
191    fn minimum<R: Copy + PartialOrd + Default + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static) -> Rc<dyn IObservable<R>>;
192
193    /// [`minimum`](Self::minimum) with the value when there are no items
194    /// (`Minimum(Func<TObject, TResult>, TResult emptyValue)`).
195    fn minimum_with<R: Copy + PartialOrd + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static, empty_value: R) -> Rc<dyn IObservable<R>>;
196
197    /// The sample standard deviation of the values of the items after each change set,
198    /// `fallback_value` for fewer than two items (`StdDev(Func<T, TValue>, TValue fallbackValue)`).
199    fn std_dev<V: StdDevValue>(&self, value_selector: impl Fn(&T) -> V + 'static, fallback_value: V) -> Rc<dyn IObservable<f64>>;
200
201    /// The sum of the values of the items after each change set (`Sum(Func<T, TValue>)`).
202    fn sum_list<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<V::Number>>;
203}
204
205impl<T: Clone + PartialEq + 'static, C: crate::list::IChangeSet<T>> AggregationListEx<T> for Rc<dyn IObservable<C>> {
206    fn for_aggregation(&self) -> AggregateChangeSets<T> {
207        ObservableListEx::<T>::change_sets(self).select(list_aggregate::<T>)
208    }
209
210    fn avg<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>> {
211        AggregationListEx::<T>::avg_with(self, value_selector, <V::Number as AggregateNumber>::ZERO)
212    }
213
214    fn avg_with<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static, empty_value: V::Number) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>> {
215        super::avg_ex::avg(AggregationListEx::<T>::for_aggregation(self), value_selector, empty_value)
216    }
217
218    fn count_list(&self) -> Rc<dyn IObservable<usize>> {
219        super::count_ex::count(AggregationListEx::<T>::for_aggregation(self))
220    }
221
222    fn is_empty_list(&self) -> Rc<dyn IObservable<bool>> {
223        super::count_ex::is_empty(AggregationListEx::<T>::for_aggregation(self))
224    }
225
226    fn is_not_empty(&self) -> Rc<dyn IObservable<bool>> {
227        super::count_ex::is_not_empty(AggregationListEx::<T>::for_aggregation(self))
228    }
229
230    fn maximum<R: Copy + PartialOrd + Default + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static) -> Rc<dyn IObservable<R>> {
231        AggregationListEx::<T>::maximum_with(self, value_selector, R::default())
232    }
233
234    fn maximum_with<R: Copy + PartialOrd + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static, empty_value: R) -> Rc<dyn IObservable<R>> {
235        let source = super::max_ex::list_changes_and_collection(ObservableListEx::<T>::change_sets(self));
236        super::max_ex::calculate(source, Rc::new(value_selector), MaxOrMin::Max, empty_value)
237    }
238
239    fn minimum<R: Copy + PartialOrd + Default + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static) -> Rc<dyn IObservable<R>> {
240        AggregationListEx::<T>::minimum_with(self, value_selector, R::default())
241    }
242
243    fn minimum_with<R: Copy + PartialOrd + 'static>(&self, value_selector: impl Fn(&T) -> R + 'static, empty_value: R) -> Rc<dyn IObservable<R>> {
244        let source = super::max_ex::list_changes_and_collection(ObservableListEx::<T>::change_sets(self));
245        super::max_ex::calculate(source, Rc::new(value_selector), MaxOrMin::Min, empty_value)
246    }
247
248    fn std_dev<V: StdDevValue>(&self, value_selector: impl Fn(&T) -> V + 'static, fallback_value: V) -> Rc<dyn IObservable<f64>> {
249        super::std_dev_ex::std_dev(AggregationListEx::<T>::for_aggregation(self), value_selector, fallback_value)
250    }
251
252    fn sum_list<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<V::Number>> {
253        super::sum_ex::sum(AggregationListEx::<T>::for_aggregation(self), value_selector)
254    }
255}
256
257/// The aggregations of aggregate change sets (the overloads of `AvgEx`, `CountEx`, `StdDevEx` and
258/// `SumEx` on `IObservable<IAggregateChangeSet<T>>`), to which the forms of caches and lists
259/// forward.
260pub trait AggregateChangeSetEx<T: 'static> {
261    /// The average of the values of the items after each change set, zero when there are none
262    /// (`Avg(Func<T, TValue>)`).
263    fn avg<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>>;
264
265    /// [`avg`](Self::avg) with the value when there are no items
266    /// (`Avg(Func<T, TValue>, TValue emptyValue)`).
267    fn avg_with<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static, empty_value: V::Number) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>>;
268
269    /// The count of the items after each change set (`Count<TObject>()`).
270    fn count_aggregate(&self) -> Rc<dyn IObservable<usize>>;
271
272    /// The sample standard deviation of the values of the items after each change set,
273    /// `fallback_value` for fewer than two items (`StdDev(Func<T, TValue>, TValue fallbackValue)`).
274    fn std_dev<V: StdDevValue>(&self, value_selector: impl Fn(&T) -> V + 'static, fallback_value: V) -> Rc<dyn IObservable<f64>>;
275
276    /// The sum of the values of the items after each change set (`Sum(Func<T, TValue>)`).
277    fn sum_aggregate<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<V::Number>>;
278}
279
280impl<T: 'static> AggregateChangeSetEx<T> for AggregateChangeSets<T> {
281    fn avg<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>> {
282        AggregateChangeSetEx::<T>::avg_with(self, value_selector, <V::Number as AggregateNumber>::ZERO)
283    }
284
285    fn avg_with<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static, empty_value: V::Number) -> Rc<dyn IObservable<<V::Number as AggregateNumber>::Average>> {
286        super::avg_ex::avg(self.clone(), value_selector, empty_value)
287    }
288
289    fn count_aggregate(&self) -> Rc<dyn IObservable<usize>> {
290        super::count_ex::count(self.clone())
291    }
292
293    fn std_dev<V: StdDevValue>(&self, value_selector: impl Fn(&T) -> V + 'static, fallback_value: V) -> Rc<dyn IObservable<f64>> {
294        super::std_dev_ex::std_dev(self.clone(), value_selector, fallback_value)
295    }
296
297    fn sum_aggregate<V: AggregateValue>(&self, value_selector: impl Fn(&T) -> V + 'static) -> Rc<dyn IObservable<V::Number>> {
298        super::sum_ex::sum(self.clone(), value_selector)
299    }
300}
301
302/// The invalidation of an aggregating sequence (`InvalidateWhen`), for every sequence.
303pub trait InvalidateWhenEx<T> {
304    /// The sequence, subscribed again each time `invalidate` notifies, without repeating a value:
305    /// used when an item changes inline, so that an aggregation is computed again
306    /// (`InvalidateWhen(IObservable<Unit>)` and `InvalidateWhen<T, TTrigger>(IObservable<TTrigger?>)`,
307    /// which are one method: the trigger's values are not read).
308    fn invalidate_when<U: 'static>(&self, invalidate: Rc<dyn IObservable<U>>) -> Rc<dyn IObservable<T>>;
309}
310
311impl<T: Clone + PartialEq + 'static> InvalidateWhenEx<T> for Rc<dyn IObservable<T>> {
312    fn invalidate_when<U: 'static>(&self, invalidate: Rc<dyn IObservable<U>>) -> Rc<dyn IObservable<T>> {
313        let source = self.clone();
314        invalidate
315            .select(|_: U| ())
316            .start_with([()])
317            .select(move |_: ()| source.clone())
318            .switch::<T>()
319            .distinct_until_changed()
320    }
321}
ferroui_dynamic_data/aggregation/aggregation_ex.rs

ferroui_dynamic_data/aggregation/
aggregation_ex.rs
