Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Aggregated temporal representation (4.5.19).
3 :
4 : use super::*;
5 :
6 : /// Aggregated representation (4.5.19): attr → `{type, <method>: [[v,start,end]]}`.
7 : /// Aggregation datatype class per 4.5.19.1 (Tables -1, -2, -3). Booleans
8 : /// count as numbers (1/0, table NOTE); a JSON String, a DateTime and a Date
9 : /// share the ordered min/max column; a Time additionally supports avg.
10 : #[derive(Clone, Copy, PartialEq, Debug)]
11 : pub(super) enum AggrClass {
12 : Number,
13 : Text,
14 : /// 4.6.3 DateTime or Date: ordered, and Table 4.5.19.1-2 gives it no
15 : /// arithmetic.
16 : Instant,
17 : TimeOfDay,
18 : List,
19 : Opaque,
20 : Relationship,
21 : }
22 :
23 : /// The 4.6.3 datatype a Property instance's value carries, in either
24 : /// representation C.6 gives for one: a JSON-LD typed value
25 : /// (`{"@type": "DateTime", "@value": …}`), or a string whose `valueType`
26 : /// names the datatype and is coerced to its URI on the way in (4.5.2.2).
27 : /// Table 4.5.19.1-2 applies to the datatype, not to the spelling.
28 174 : pub(super) fn value_datatype(inst: &Value) -> Option<&str> {
29 56 : fn term(s: &str) -> &str {
30 56 : s.strip_prefix(antares_jsonld::NGSI_LD_BASE).unwrap_or(s)
31 56 : }
32 174 : if let Some(vt) = inst.get("valueType").and_then(Value::as_str) {
33 4 : if inst.get("value").is_some_and(Value::is_string) {
34 4 : return Some(term(vt));
35 0 : }
36 170 : }
37 170 : inst.get("value")?
38 170 : .get("@type")
39 170 : .and_then(Value::as_str)
40 170 : .map(term)
41 174 : }
42 :
43 : /// The lexical form of a value: the string itself, or the `@value` of a
44 : /// JSON-LD typed value.
45 168 : pub(super) fn lexical_of(v: &Value) -> Option<&str> {
46 168 : v.as_str()
47 168 : .or_else(|| v.get("@value").and_then(Value::as_str))
48 168 : }
49 :
50 : /// The key the ordered classes compare by. A JSON String and a Date are
51 : /// compared as written — 4.6.3 fixes the width of every component of a Date,
52 : /// so lexicographical order is chronological — a DateTime by its canonical
53 : /// instant, since an optional seconds fraction is written before the `Z` it
54 : /// follows and sorts ahead of it, and a Time by its second of the day, at a
55 : /// fixed width so one string comparison serves all four.
56 128 : pub(super) fn order_key(class: AggrClass, v: &Value) -> Option<String> {
57 128 : let s = lexical_of(v)?;
58 128 : Some(match class {
59 56 : AggrClass::Instant => antares_model::dt_key(s),
60 16 : AggrClass::TimeOfDay => format!("{:013.6}", seconds_of_day(s)?),
61 56 : _ => s.to_owned(),
62 : })
63 128 : }
64 :
65 186 : pub(super) fn classify_instance(inst: &Value) -> AggrClass {
66 186 : if inst.get("object").is_some() {
67 8 : return AggrClass::Relationship;
68 178 : }
69 178 : if inst.get("valueList").is_some() || inst.get("objectList").is_some() {
70 0 : return AggrClass::List;
71 178 : }
72 178 : if inst.get("vocab").is_some()
73 174 : || inst.get("languageMap").is_some()
74 174 : || inst.get("json").is_some()
75 : {
76 : // URI / JSON-object valued kinds: only counting methods apply
77 4 : return AggrClass::Opaque;
78 174 : }
79 174 : match value_datatype(inst) {
80 56 : Some("DateTime" | "Date") => return AggrClass::Instant,
81 20 : Some("Time") => return AggrClass::TimeOfDay,
82 118 : _ => {}
83 : }
84 118 : match inst.get("value") {
85 102 : Some(Value::Number(_)) | Some(Value::Bool(_)) => AggrClass::Number,
86 16 : Some(Value::String(_)) => AggrClass::Text,
87 0 : Some(Value::Array(_)) => AggrClass::List,
88 0 : _ => AggrClass::Opaque,
89 : }
90 186 : }
91 :
92 : /// Table 4.5.19.1 eligibility: which methods apply to which datatype class.
93 276 : pub(super) fn aggr_eligible(class: AggrClass, method: &str) -> bool {
94 276 : match method {
95 276 : "totalCount" | "distinctCount" => true,
96 208 : "min" | "max" => !matches!(class, AggrClass::Opaque | AggrClass::Relationship),
97 112 : "avg" => matches!(
98 76 : class,
99 : AggrClass::Number | AggrClass::List | AggrClass::TimeOfDay
100 : ),
101 36 : "sum" => matches!(class, AggrClass::Number | AggrClass::List),
102 16 : "stddev" | "sumsq" => matches!(class, AggrClass::Number),
103 0 : _ => false,
104 : }
105 276 : }
106 :
107 : /// `HH:MM:SS[.f]` → seconds of day (4.6.3 Time is UTC with optional `Z`).
108 88 : pub(super) fn seconds_of_day(s: &str) -> Option<f64> {
109 88 : let t = s.strip_suffix('Z').unwrap_or(s);
110 88 : let b = t.as_bytes();
111 88 : if b.len() < 8 || b.get(2) != Some(&b':') || b.get(5) != Some(&b':') {
112 20 : return None;
113 68 : }
114 68 : let h: f64 = t.get(0..2)?.parse().ok()?;
115 64 : let m: f64 = t.get(3..5)?.parse().ok()?;
116 64 : let sec: f64 = t.get(6..)?.parse().ok()?;
117 64 : (h < 24.0 && m < 60.0 && sec < 62.0).then_some(h * 3600.0 + m * 60.0 + sec)
118 88 : }
119 :
120 : /// The raw member an instance carries its data under.
121 1616 : pub(super) fn raw_of(inst: &Value) -> Option<&Value> {
122 1664 : for k in [
123 1616 : "value",
124 1616 : "object",
125 1616 : "valueList",
126 1616 : "objectList",
127 1616 : "vocab",
128 1616 : "languageMap",
129 1616 : "json",
130 1616 : ] {
131 1664 : if let Some(v) = inst.get(k) {
132 1616 : return Some(v);
133 48 : }
134 : }
135 0 : None
136 1616 : }
137 :
138 : /// Numeric view of one raw value for the class (None ⇒ excluded from
139 : /// numeric methods; List aggregates SIZES per Table 4.5.19.1-1).
140 2898 : pub(super) fn numeric_of(class: AggrClass, v: &Value) -> Option<f64> {
141 2898 : match class {
142 2662 : AggrClass::Number => match v {
143 0 : Value::Bool(b) => Some(if *b { 1.0 } else { 0.0 }),
144 2662 : _ => v.as_f64(),
145 : },
146 8 : AggrClass::List => v.as_array().map(|a| a.len() as f64),
147 40 : AggrClass::TimeOfDay => lexical_of(v).and_then(seconds_of_day),
148 188 : _ => None,
149 : }
150 2898 : }
151 :
152 182 : pub(super) fn render_aggregated(
153 182 : w: &Windowed,
154 182 : tq: Option<&TemporalQ>,
155 182 : r: &TRepr,
156 182 : ctx: &Context,
157 182 : timeprop: &str,
158 182 : ) -> Result<Map<String, Value>, NgsiError> {
159 182 : let mut out = Map::new();
160 200 : for (k, instances) in &w.attrs {
161 186 : let held = w.held.contains(k);
162 186 : if let Some(attr_out) = aggregate_attr(k, instances, held, tq, r, ctx, timeprop)? {
163 142 : out.insert(ctx.compact_iri(k), Value::Object(attr_out));
164 142 : }
165 : }
166 138 : Ok(out)
167 182 : }
168 :
169 : type Moment = chrono::DateTime<chrono::FixedOffset>;
170 : /// One period: its start and its (exclusive) end.
171 : type Bucket = (Moment, Moment);
172 : /// The instances of one period, by their time.
173 : type Rows<'a> = Vec<(Moment, &'a Value)>;
174 :
175 : /// One Attribute's 4.5.19 member, `None` when no instance carries a time.
176 : /// `held`: the Attribute records `onChange`.
177 186 : fn aggregate_attr(
178 186 : k: &str,
179 186 : instances: &[Value],
180 186 : held: bool,
181 186 : tq: Option<&TemporalQ>,
182 186 : r: &TRepr,
183 186 : ctx: &Context,
184 186 : timeprop: &str,
185 186 : ) -> Result<Option<Map<String, Value>>, NgsiError> {
186 186 : let (mut times, class) = timed_rows(instances, timeprop);
187 : // Set alongside the first instance, and `times` is non-empty then.
188 186 : let Some(class) = class.filter(|_| !times.is_empty()) else {
189 0 : return Ok(None);
190 : };
191 186 : check_methods(k, class, held, r, ctx)?;
192 144 : times.sort_by_key(|(t, _)| *t);
193 144 : let Some(whole) = whole_range(×, tq) else {
194 0 : return Ok(None);
195 : };
196 144 : let Some(&(earliest, _)) = times.first() else {
197 0 : return Ok(None);
198 : };
199 144 : let anchor = tq
200 144 : .and_then(|tq| chrono::DateTime::parse_from_rfc3339(&tq.time_at).ok())
201 144 : .unwrap_or(earliest);
202 : // the row carried in from before the window holds from its start
203 144 : if held {
204 1206 : for (t, _) in &mut times {
205 1206 : *t = (*t).max(whole.0);
206 1206 : }
207 128 : }
208 22786 : let bucket_of = |t: Moment| -> Bucket {
209 22786 : match r.aggr_period {
210 208 : AggrPeriod::Whole => whole,
211 22558 : AggrPeriod::Seconds(sc) => seconds_bucket(anchor, sc, t),
212 20 : AggrPeriod::Months(m, sc) => months_bucket(anchor, m, sc, t),
213 : }
214 22786 : };
215 144 : let Some(buckets) = bucketed(×, held, whole, &bucket_of) else {
216 2 : return Err(NgsiError::BadRequestData(format!(
217 2 : "aggrPeriodDuration splits the window of {} into more than {MAX_HELD_BUCKETS} \
218 2 : periods; ask a longer period or a shorter window",
219 2 : ctx.compact_iri(k)
220 2 : )));
221 : };
222 142 : Ok(Some(render_buckets(&buckets, class, held, whole.1, r)))
223 186 : }
224 :
225 : /// The instances that carry `timeprop`, with their raw values, and the
226 : /// aggregation class of the first of them.
227 186 : fn timed_rows<'a>(instances: &'a [Value], timeprop: &str) -> (Rows<'a>, Option<AggrClass>) {
228 186 : let mut times: Rows<'a> = Vec::new();
229 186 : let mut class: Option<AggrClass> = None;
230 1616 : for inst in instances {
231 1616 : let Some(t) = inst
232 1616 : .get(timeprop)
233 1616 : .and_then(Value::as_str)
234 1616 : .and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
235 : else {
236 0 : continue;
237 : };
238 1616 : if class.is_none() {
239 186 : class = Some(classify_instance(inst));
240 1430 : }
241 1616 : let Some(raw) = raw_of(inst) else { continue };
242 1616 : times.push((t, raw));
243 : }
244 186 : (times, class)
245 186 : }
246 :
247 : /// 5.7.4.4 p.211: "If an aggregated temporal representation is
248 : /// requested and any of the requested Attributes is not eligible for
249 : /// at least one of the aggregation methods specified in the request
250 : /// parameters, then an error of type InvalidRequest shall be raised."
251 : /// An `onChange` Attribute leaves out the rows
252 : /// that repeat a value, and a left-out row is not a zero.
253 186 : fn check_methods(
254 186 : k: &str,
255 186 : class: AggrClass,
256 186 : held: bool,
257 186 : r: &TRepr,
258 186 : ctx: &Context,
259 186 : ) -> Result<(), NgsiError> {
260 276 : for method in &r.aggr_methods {
261 276 : if !aggr_eligible(class, method) {
262 38 : return Err(NgsiError::InvalidRequest(format!(
263 38 : "attribute {} ({class:?}-valued) is not eligible for \
264 38 : aggregation method {method} (4.5.19.1, 5.7.4.4)",
265 38 : ctx.compact_iri(k)
266 38 : )));
267 238 : }
268 : }
269 148 : if let Some(method) = r
270 148 : .aggr_methods
271 148 : .iter()
272 238 : .find(|m| held && matches!(m.as_str(), "sum" | "totalCount"))
273 : {
274 4 : return Err(NgsiError::InvalidRequest(format!(
275 4 : "aggregation-not-meaningful: {method} over {}, which records a row only \
276 4 : when its value changes",
277 4 : ctx.compact_iri(k)
278 4 : )));
279 144 : }
280 144 : Ok(())
281 186 : }
282 :
283 : /// 4.5.19.1: "A duration of 0 second (e.g. expressed as "PT0S" or
284 : /// "P0D") is valid and is interpreted as a duration spanning the whole
285 : /// time range specified by the temporal query." The query names one
286 : /// edge of that range; 4.11 leaves the other open for `before` and
287 : /// `after`, so the data closes the open one. `times` is sorted.
288 144 : fn whole_range(times: &Rows<'_>, tq: Option<&TemporalQ>) -> Option<Bucket> {
289 144 : let at = |s: &str| chrono::DateTime::parse_from_rfc3339(s).ok();
290 144 : let (&(first, _), &(last_at, _)) = (times.first()?, times.last()?);
291 144 : let last = last_at
292 144 : .checked_add_signed(chrono::Duration::seconds(1))
293 144 : .unwrap_or(last_at);
294 46 : Some(match tq {
295 76 : Some(q) if q.timerel == "before" => (first, at(&q.time_at).unwrap_or(last)),
296 68 : Some(q) if q.timerel == "between" => (
297 22 : at(&q.time_at).unwrap_or(first),
298 22 : q.end_time_at.as_deref().and_then(at).unwrap_or(last),
299 22 : ),
300 46 : Some(q) if q.timerel == "after" => (at(&q.time_at).unwrap_or(first), last),
301 68 : _ => (first, last),
302 : })
303 144 : }
304 :
305 : /// The `sc`-second period from `anchor` holding `t`. Checked throughout:
306 : /// an offset no representable date can hold puts the instant in one final
307 : /// open-ended bucket instead of panicking.
308 22558 : fn seconds_bucket(anchor: Moment, sc: i64, t: Moment) -> Bucket {
309 22558 : let idx = t.signed_duration_since(anchor).num_seconds().div_euclid(sc);
310 22558 : let bucket = idx
311 22558 : .checked_mul(sc)
312 22558 : .and_then(chrono::Duration::try_seconds)
313 22558 : .and_then(|off| anchor.checked_add_signed(off))
314 22558 : .and_then(|start| {
315 22558 : chrono::Duration::try_seconds(sc)
316 22558 : .and_then(|w| start.checked_add_signed(w))
317 22558 : .map(|end| (start, end))
318 22558 : });
319 22558 : match bucket {
320 22558 : Some(b) => b,
321 0 : None => (anchor, chrono::DateTime::<chrono::Utc>::MAX_UTC.into()),
322 : }
323 22558 : }
324 :
325 : /// The period of `m` months and `sc` seconds from `anchor` holding `t`.
326 20 : fn months_bucket(anchor: Moment, m: u32, sc: i64, t: Moment) -> Bucket {
327 : use chrono::Datelike;
328 : // start of the k-th period, O(1) in k. Negative k are
329 : // the periods BEFORE the anchor, which is what a
330 : // `before` query is made of.
331 76 : let step = |k: i64| -> Option<Moment> {
332 76 : let n = k.unsigned_abs().checked_mul(u64::from(m))?;
333 76 : let n = chrono::Months::new(u32::try_from(n).ok()?);
334 76 : let base = if k < 0 {
335 40 : anchor.checked_sub_months(n)?
336 : } else {
337 36 : anchor.checked_add_months(n)?
338 : };
339 76 : let off = chrono::Duration::try_seconds(k.checked_mul(sc)?)?;
340 76 : base.checked_add_signed(off)
341 76 : };
342 : // The whole-month distance ignores the day and the time
343 : // of day, and one period is at least one month, so it
344 : // brackets the exact index within one step: binary-search
345 : // between it and the anchor instead of walking there,
346 : // which is O(log) however far the instant is.
347 20 : let approx = i64::from(t.year())
348 20 : .saturating_sub(i64::from(anchor.year()))
349 20 : .saturating_mul(12)
350 20 : .saturating_add(i64::from(t.month()))
351 20 : .saturating_sub(i64::from(anchor.month()))
352 20 : .div_euclid(i64::from(m));
353 20 : let (mut lo, mut hi) = (
354 20 : approx.min(0).saturating_sub(1),
355 20 : approx.max(0).saturating_add(1),
356 20 : );
357 56 : while hi.saturating_sub(lo) > 1 {
358 36 : let mid = lo.saturating_add(hi.saturating_sub(lo).div_euclid(2));
359 36 : if step(mid).is_some_and(|s| s <= t) {
360 24 : lo = mid;
361 24 : } else {
362 12 : hi = mid;
363 12 : }
364 : }
365 : // saturate instead of panic: a huge month period or a
366 : // far-future timeAt overflows chrono's range — treat the
367 : // remainder as one open-ended bucket
368 20 : match (step(lo), step(hi)) {
369 20 : (Some(start), Some(end)) if start <= t => (start, end),
370 0 : (Some(start), None) if start <= t => {
371 0 : (start, chrono::DateTime::<chrono::Utc>::MAX_UTC.into())
372 : }
373 0 : _ => (anchor, chrono::DateTime::<chrono::Utc>::MAX_UTC.into()),
374 : }
375 20 : }
376 :
377 : /// The sorted rows in their periods. An onChange attribute held its value
378 : /// through a period no row fell in: that period answers the value, as
379 : /// `every` would have, up to the window's end, never past now, and at most
380 : /// MAX_HELD_BUCKETS of them: a far edge and PT1S is refused (`None`), not
381 : /// allocated.
382 144 : fn bucketed<'a>(
383 144 : times: &Rows<'a>,
384 144 : held: bool,
385 144 : whole: Bucket,
386 144 : bucket_of: &dyn Fn(Moment) -> Bucket,
387 144 : ) -> Option<Vec<(Bucket, Rows<'a>)>> {
388 144 : let mut buckets: Vec<(Bucket, Rows)> = Vec::new();
389 144 : let edge = whole.1.min(chrono::Utc::now().fixed_offset());
390 144 : let mut refused = false;
391 1290 : let mut fill = |buckets: &mut Vec<(Bucket, Rows)>, until: Moment| {
392 21254 : while let Some(end) = buckets.last().map(|((_, end), _)| *end) {
393 21238 : let gap = bucket_of(end);
394 21238 : if end >= until || end >= edge || gap.0 < end {
395 1204 : break;
396 20034 : }
397 20034 : if buckets.len() >= MAX_HELD_BUCKETS {
398 2 : refused = true;
399 2 : break;
400 20032 : }
401 20032 : buckets.push((gap, Vec::new()));
402 : }
403 1222 : };
404 1548 : for &(t, v) in times {
405 1548 : let b = bucket_of(t);
406 1548 : if held {
407 1206 : fill(&mut buckets, b.0);
408 1392 : }
409 1548 : match buckets.last_mut() {
410 1404 : Some((bb, rows)) if bb.0 == b.0 => rows.push((t, v)),
411 1098 : _ => buckets.push((b, vec![(t, v)])),
412 : }
413 : }
414 144 : if held {
415 16 : fill(&mut buckets, whole.1);
416 128 : }
417 144 : (!refused).then_some(buckets)
418 144 : }
419 :
420 : /// The member of one Attribute: its type and, per method, one
421 : /// `[value, start, end]` per period. `window_end` caps a held value's weight.
422 142 : fn render_buckets(
423 142 : buckets: &[(Bucket, Rows<'_>)],
424 142 : class: AggrClass,
425 142 : held: bool,
426 142 : window_end: Moment,
427 142 : r: &TRepr,
428 142 : ) -> Map<String, Value> {
429 2464 : let fmt = |d: Moment| d.format("%Y-%m-%dT%H:%M:%SZ").to_string();
430 : // the value the last bucket with a row ended on holds into the next one
431 142 : let carried: Vec<Option<&Value>> = buckets
432 142 : .iter()
433 1130 : .scan(None, |last, (_, rows)| {
434 1130 : let before = *last;
435 1130 : if let Some(&(_, v)) = rows.last() {
436 1094 : *last = Some(v);
437 1094 : }
438 1130 : Some(before)
439 1130 : })
440 142 : .collect();
441 142 : let mut attr_out = Map::new();
442 : // 4.5.19.0: the member is labelled "Property" for Properties and
443 : // "Relationship" for Relationships.
444 142 : let label = if class == AggrClass::Relationship {
445 6 : "Relationship"
446 : } else {
447 136 : "Property"
448 : };
449 142 : attr_out.insert("type".into(), Value::String(label.into()));
450 232 : for method in &r.aggr_methods {
451 232 : let rows: Vec<Value> = buckets
452 232 : .iter()
453 232 : .zip(&carried)
454 1232 : .map(|(((bs, be), rows), &before)| {
455 1232 : let weighted = (held && method == "avg")
456 1232 : .then(|| time_weighted_avg(class, rows, before, *bs, (*be).min(window_end)))
457 1232 : .flatten();
458 1232 : let val = weighted.unwrap_or_else(|| {
459 278 : let vals: Vec<&Value> = match (rows.is_empty(), before) {
460 0 : (true, Some(v)) => vec![v],
461 278 : _ => rows.iter().map(|&(_, v)| v).collect(),
462 : };
463 278 : aggregate_bucket(method, class, &vals)
464 278 : });
465 1232 : Value::Array(vec![val, Value::String(fmt(*bs)), Value::String(fmt(*be))])
466 1232 : })
467 232 : .collect();
468 232 : attr_out.insert(method.clone(), Value::Array(rows));
469 : }
470 142 : attr_out
471 142 : }
472 :
473 : /// The mean of a bucket from `start` to `end` whose `rows` (time, value) each
474 : /// hold until the next, `before` holding until the first;
475 : /// `None` for a class with no number or a bucket no value held for a moment.
476 974 : pub(super) fn time_weighted_avg(
477 974 : class: AggrClass,
478 974 : rows: &[(chrono::DateTime<chrono::FixedOffset>, &Value)],
479 974 : before: Option<&Value>,
480 974 : start: chrono::DateTime<chrono::FixedOffset>,
481 974 : end: chrono::DateTime<chrono::FixedOffset>,
482 974 : ) -> Option<Value> {
483 974 : if class == AggrClass::TimeOfDay {
484 4 : return None;
485 970 : }
486 970 : let held = before
487 970 : .map(|v| (start, v))
488 970 : .into_iter()
489 970 : .chain(rows.iter().copied());
490 970 : let until = rows
491 970 : .iter()
492 970 : .skip(usize::from(before.is_none()))
493 970 : .map(|&(t, _)| t)
494 970 : .chain([end]);
495 970 : let (mut weighted, mut span) = (0.0, 0.0);
496 2158 : for ((from, v), to) in held.zip(until) {
497 2158 : let secs = to.signed_duration_since(from).num_milliseconds() as f64 / 1000.0;
498 2158 : if let (Some(x), true) = (numeric_of(class, v), secs > 0.0) {
499 2154 : weighted += x * secs;
500 2154 : span += secs;
501 2154 : }
502 : }
503 970 : (span > 0.0).then(|| serde_json::json!(weighted / span))
504 974 : }
505 :
506 : /// One bucket, one method — per-class semantics from Tables 4.5.19.1-1/2/3.
507 : /// Never emits an out-of-range float (the old fold seeded with
508 : /// f64::INFINITY, which serde_json serializes as null).
509 346 : pub(super) fn aggregate_bucket(method: &str, class: AggrClass, vals: &[&Value]) -> Value {
510 740 : let nums: Vec<f64> = vals.iter().filter_map(|v| numeric_of(class, v)).collect();
511 346 : let finite = |x: f64| {
512 180 : if x.is_finite() {
513 176 : serde_json::json!(x)
514 : } else {
515 4 : Value::Null
516 : }
517 180 : };
518 346 : match method {
519 346 : "totalCount" => serde_json::json!(vals.len()),
520 284 : "distinctCount" => {
521 : // Relationship: "count of distinct relationship TARGETS" — an
522 : // object may be a URI or an array of URIs, so flatten first.
523 24 : let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
524 48 : for v in vals {
525 48 : let items: Vec<&Value> = match (class, v) {
526 4 : (AggrClass::Relationship, Value::Array(a)) => a.iter().collect(),
527 44 : _ => vec![*v],
528 : };
529 52 : for it in items {
530 52 : seen.insert(it.to_string());
531 52 : }
532 : }
533 24 : serde_json::json!(seen.len())
534 : }
535 260 : "min" | "max" => match class {
536 : // ordered classes: the first or last value in the order the
537 : // tables give the datatype, returned as it was written
538 : AggrClass::Text | AggrClass::Instant | AggrClass::TimeOfDay => {
539 56 : let mut keyed: Vec<(String, &Value)> = vals
540 56 : .iter()
541 128 : .filter_map(|v| order_key(class, v).map(|k| (k, *v)))
542 56 : .collect();
543 88 : keyed.sort_by(|a, b| a.0.cmp(&b.0));
544 56 : let pick = if method == "min" {
545 28 : keyed.first()
546 : } else {
547 28 : keyed.last()
548 : };
549 56 : pick.map_or(Value::Null, |(_, v)| (*v).clone())
550 : }
551 : _ => {
552 80 : let it = nums.iter().copied();
553 80 : let picked = if method == "min" {
554 76 : it.fold(None, |a: Option<f64>, v| Some(a.map_or(v, |x| x.min(v))))
555 : } else {
556 132 : it.fold(None, |a: Option<f64>, v| Some(a.map_or(v, |x| x.max(v))))
557 : };
558 80 : picked.map_or(Value::Null, &finite)
559 : }
560 : },
561 124 : "sum" => finite(nums.iter().sum::<f64>()),
562 112 : "avg" => {
563 96 : if nums.is_empty() {
564 4 : Value::Null
565 92 : } else if class == AggrClass::TimeOfDay {
566 4 : let mean = nums.iter().sum::<f64>() / nums.len() as f64;
567 4 : let (h, m, sec) = (
568 4 : (mean / 3600.0) as u32,
569 4 : ((mean % 3600.0) / 60.0) as u32,
570 4 : (mean % 60.0) as u32,
571 4 : );
572 : // 4.6.3: a Time is `hh:mm:ssZ`, and its JSON-LD type is what
573 : // tells a reader it is one — written bare it reads back as a
574 : // JSON String, which Table 4.5.19.1-1 gives no average.
575 4 : serde_json::json!({
576 4 : "@type": "Time",
577 4 : "@value": format!("{h:02}:{m:02}:{sec:02}Z"),
578 : })
579 : } else {
580 88 : finite(nums.iter().sum::<f64>() / nums.len() as f64)
581 : }
582 : }
583 16 : "stddev" => {
584 8 : if nums.is_empty() {
585 4 : Value::Null
586 : } else {
587 4 : let n = nums.len() as f64;
588 4 : let mean = nums.iter().sum::<f64>() / n;
589 12 : finite((nums.iter().map(|v| (v - mean).powi(2)).sum::<f64>() / n).sqrt())
590 : }
591 : }
592 12 : "sumsq" => finite(nums.iter().map(|v| v * v).sum::<f64>()),
593 4 : _ => Value::Null,
594 : }
595 346 : }
|
