1use std::collections::HashMap;
5
6use crate::column::ColumnData;
7use crate::error::{PondError, Result};
8use crate::reshape::At;
9use crate::sequence::Sequence;
10use crate::series::{Column, ColumnKind, IntervalKeys, Key, Labels, TimeSeries};
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
14pub enum AlignMethod {
15 #[default]
17 Hold,
18 Linear,
22}
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
28pub struct Align {
29 pub method: AlignMethod,
31 pub sample: At,
33 pub range: Option<(i64, i64)>,
37}
38
39#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
41pub enum Pick {
42 First,
44 #[default]
46 Last,
47 Nearest,
50}
51
52#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
56pub struct Materialize {
57 pub pick: Pick,
59 pub sample: At,
61 pub range: Option<(i64, i64)>,
63}
64
65#[derive(Debug, Clone, PartialEq, Eq, Default)]
67pub enum Keep {
68 First,
70 #[default]
72 Last,
73 Error,
75 Drop,
77 Min(String),
80 Max(String),
82}
83
84#[derive(Clone, Copy)]
86enum Take {
87 Nothing,
88 Row(usize),
89 Between(usize, usize, f64),
91}
92
93impl Take {
94 fn row(self) -> Option<usize> {
96 match self {
97 Take::Nothing => None,
98 Take::Row(i) | Take::Between(i, _, _) => Some(i),
99 }
100 }
101}
102
103impl TimeSeries {
104 pub fn align(&self, sequence: &Sequence, options: Align) -> Result<TimeSeries> {
146 const CALL: &str = "align";
147 self.check_new_key_name(CALL, "interval")?;
148 let Some((begin, end)) = self.grid_range(CALL, options.range)? else {
149 return Ok(self.empty_grid("interval", Key::Interval(no_buckets())));
150 };
151 if options.method == AlignMethod::Linear && !matches!(self.key(), Key::Time(_)) {
152 return Err(PondError::Unsupported {
153 call: CALL,
154 message: "linear alignment needs a time-keyed series (pond-ts has the same limit); use AlignMethod::Hold".into(),
155 });
156 }
157 let buckets = sequence.buckets_sampled(begin, end, options.sample)?;
158 let points = sample_points(&buckets, options.sample);
159 let (takes, rows) = match options.method {
162 AlignMethod::Hold => (Vec::new(), self.hold_rows(points)),
163 AlignMethod::Linear => {
164 let takes = self.linear_takes(points);
165 let holds = self
166 .columns()
167 .iter()
168 .any(|c| !matches!(c.data(), ColumnData::Number(_)));
169 let rows = if holds {
170 takes.iter().map(|t| t.row()).collect()
171 } else {
172 Vec::new()
173 };
174 (takes, rows)
175 }
176 };
177 let columns = self
178 .columns()
179 .iter()
180 .map(|c| {
181 let data = match c.data() {
182 ColumnData::Number(n) if options.method == AlignMethod::Linear => {
183 let v = n.values();
184 let present = |i: usize| n.get(i);
185 ColumnData::from(
186 takes
187 .iter()
188 .map(|t| match *t {
189 Take::Nothing => f64::NAN,
190 Take::Row(i) => v[i],
191 Take::Between(p, q, ratio) => match (present(p), present(q)) {
192 (Some(a), Some(b)) => a + (b - a) * ratio,
193 _ => v[p],
194 },
195 })
196 .collect::<Vec<f64>>(),
197 )
198 }
199 data => data.take(&rows),
200 };
201 Column::from_data(c.name().to_string(), data, false)
202 })
203 .collect();
204 Ok(TimeSeries::from_parts(
205 self.name().to_string(),
206 "interval".to_string(),
207 Key::Interval(buckets),
208 columns,
209 ))
210 }
211
212 pub fn materialize(&self, sequence: &Sequence, options: Materialize) -> Result<TimeSeries> {
238 const CALL: &str = "materialize";
239 self.check_new_key_name(CALL, "time")?;
240 let Some((begin, end)) = self.grid_range(CALL, options.range)? else {
241 return Ok(self.empty_grid("time", Key::Time(vec![])));
242 };
243 let buckets = sequence.buckets_sampled(begin, end, options.sample)?;
244 let mut time = Vec::with_capacity(buckets.begin.len());
245 for (i, p2) in sample_points(&buckets, options.sample).enumerate() {
246 if p2 % 2 != 0 {
247 return Err(PondError::Unsupported {
248 call: CALL,
249 message: format!(
250 "the middle of bucket {i} ({} to {} ms) is half a millisecond; pond-rs keeps whole milliseconds. Use At::Begin or At::End, or a step of an even number of milliseconds",
251 buckets.begin[i], buckets.end[i]
252 ),
253 });
254 }
255 time.push(i64::try_from(p2 / 2).map_err(|_| PondError::InvalidData {
256 call: CALL,
257 message: format!("bucket {i}'s time is out of range"),
258 })?);
259 }
260 let starts = self.key().begin();
261 let n = starts.len();
262 let mut cursor = 0;
263 let mut rows = Vec::with_capacity(time.len());
264 for (i, p2) in sample_points(&buckets, options.sample).enumerate() {
265 let (b, e) = (buckets.begin[i], buckets.end[i]);
266 while cursor < n && starts[cursor] < b {
267 cursor += 1;
268 }
269 let mut pick = None;
270 match options.pick {
271 Pick::First => {
272 if cursor < n && starts[cursor] < e {
273 pick = Some(cursor);
274 }
275 }
276 Pick::Last => {
277 let mut s = cursor;
278 while s < n && starts[s] < e {
279 pick = Some(s);
280 s += 1;
281 }
282 }
283 Pick::Nearest => {
284 let mut best = u128::MAX;
285 let mut s = cursor;
286 while s < n && starts[s] < e {
287 let d = (2 * i128::from(starts[s]) - p2).unsigned_abs();
288 if d < best {
289 pick = Some(s);
290 best = d;
291 }
292 s += 1;
293 }
294 }
295 }
296 rows.push(pick);
297 }
298 let columns = self
299 .columns()
300 .iter()
301 .map(|c| Column::from_data(c.name().to_string(), c.data().take(&rows), false))
302 .collect();
303 Ok(TimeSeries::from_parts(
304 self.name().to_string(),
305 "time".to_string(),
306 Key::Time(time),
307 columns,
308 ))
309 }
310
311 pub fn dedupe(&self, keep: Keep) -> Result<TimeSeries> {
332 const CALL: &str = "dedupe";
333 let by = match &keep {
334 Keep::Min(c) | Keep::Max(c) => {
335 let column = self.column_index(CALL, c)?;
336 let kind = self.columns()[column].kind();
337 if !matches!(kind, ColumnKind::Number | ColumnKind::Decimal) {
338 return Err(PondError::WrongKind {
339 call: CALL,
340 column: c.clone(),
341 kind: kind.as_str(),
342 reason: "Keep::Min and Keep::Max need a number or decimal column".into(),
343 });
344 }
345 Some(&self.columns()[column])
346 }
347 _ => None,
348 };
349 let mut kept = Vec::with_capacity(self.len());
350 let mut repeated = false;
351 let mut resolve = |group: &[usize]| -> Result<()> {
352 if group.len() == 1 {
353 kept.push(group[0]);
354 return Ok(());
355 }
356 repeated = true;
357 match &keep {
358 Keep::First => kept.push(group[0]),
359 Keep::Last => kept.push(group[group.len() - 1]),
360 Keep::Drop => {}
361 Keep::Error => {
362 return Err(PondError::InvalidData {
363 call: CALL,
364 message: format!(
365 "{} events at {}. Specify a different 'keep' policy or fix upstream.",
366 group.len(),
367 self.describe_key(group[0])
368 ),
369 });
370 }
371 Keep::Min(_) | Keep::Max(_) => {
372 let wants_min = matches!(keep, Keep::Min(_));
373 let column = by.expect("checked above");
374 kept.push(best_row(column.data(), group, wants_min));
375 }
376 }
377 Ok(())
378 };
379 self.for_each_key_group(&mut resolve)?;
380 if !repeated {
381 return Ok(self.clone());
382 }
383 Ok(self.take_rows(&kept))
384 }
385
386 fn for_each_key_group(&self, visit: &mut impl FnMut(&[usize]) -> Result<()>) -> Result<()> {
389 let key = self.key();
390 let begin = key.begin();
391 let end = key.end();
392 let n = begin.len();
393 let mut run = 0;
394 let mut group = Vec::new();
395 while run < n {
396 let same = |i: usize| begin[i] == begin[run] && end.is_none_or(|e| e[i] == e[run]);
398 let mut stop = run + 1;
399 while stop < n && same(stop) {
400 stop += 1;
401 }
402 match key {
403 Key::Interval(k) if stop - run > 1 => {
406 let mut groups: Vec<Vec<usize>> = Vec::new();
407 let mut at: HashMap<LabelKey, usize> = HashMap::new();
408 for i in run..stop {
409 let g = *at.entry(LabelKey::of(&k.label, i)).or_insert_with(|| {
410 groups.push(Vec::new());
411 groups.len() - 1
412 });
413 groups[g].push(i);
414 }
415 for g in &groups {
416 visit(g)?;
417 }
418 }
419 _ => {
420 group.clear();
421 group.extend(run..stop);
422 visit(&group)?;
423 }
424 }
425 run = stop;
426 }
427 Ok(())
428 }
429
430 fn describe_key(&self, i: usize) -> String {
432 match self.key() {
433 Key::Time(t) => format!("{} ({})", iso_time(t[i]), t[i]),
434 Key::TimeRange(k) => format!("key \"{}:{}\"", k.begin[i], k.end[i]),
435 Key::Interval(k) => {
436 let label = match &k.label {
437 Labels::Number(v) => js_number(v[i]),
438 Labels::String(c) => c.get(i).unwrap_or_default().to_string(),
439 };
440 format!("key \"{}:{}:{label}\"", k.begin[i], k.end[i])
441 }
442 }
443 }
444
445 fn check_new_key_name(&self, call: &'static str, name: &str) -> Result<()> {
447 if self.columns().iter().any(|c| c.name() == name) {
448 return Err(PondError::InvalidData {
449 call,
450 message: format!(
451 "a value column is already named '{name}', the name the new key takes; rename it first"
452 ),
453 });
454 }
455 Ok(())
456 }
457
458 pub(crate) fn grid_range(
462 &self,
463 call: &'static str,
464 range: Option<(i64, i64)>,
465 ) -> Result<Option<(i64, i64)>> {
466 if let Some((begin, end)) = range {
467 if end < begin {
468 return Err(PondError::InvalidData {
469 call,
470 message: format!("range ends ({end}) before it begins ({begin})"),
471 });
472 }
473 return Ok(Some((begin, end)));
474 }
475 if self.is_empty() {
476 return Ok(None);
477 }
478 let key = self.key();
479 let begin = key.begin();
480 let end = match key.end() {
481 None => begin[begin.len() - 1],
482 Some(e) => *e.iter().max().expect("not empty"),
483 };
484 Ok(Some((begin[0], end)))
485 }
486
487 fn empty_grid(&self, key_name: &str, key: Key) -> TimeSeries {
489 let columns = self
490 .columns()
491 .iter()
492 .map(|c| Column::from_data(c.name().to_string(), c.data().take(&[]), false))
493 .collect();
494 TimeSeries::from_parts(self.name().to_string(), key_name.to_string(), key, columns)
495 }
496
497 fn hold_rows(&self, points: impl Iterator<Item = i128>) -> Vec<Option<usize>> {
500 let key = self.key();
501 let begin = key.begin();
502 let n = begin.len();
503 let zero_interval = |i: usize| match key {
507 Key::Interval(k) => k.begin[i] == k.end[i],
508 _ => false,
509 };
510 let is_time = matches!(key, Key::Time(_));
511 let mut idx = 0;
512 points
513 .map(|p2| {
514 while idx < n {
515 let b2 = 2 * i128::from(begin[idx]);
516 if b2 < p2 || (b2 == p2 && zero_interval(idx)) {
517 idx += 1;
518 } else {
519 break;
520 }
521 }
522 if is_time && idx < n && 2 * i128::from(begin[idx]) == p2 {
525 Some(idx)
526 } else {
527 idx.checked_sub(1)
528 }
529 })
530 .collect()
531 }
532
533 fn linear_takes(&self, points: impl Iterator<Item = i128>) -> Vec<Take> {
535 let begin = self.key().begin();
536 let n = begin.len();
537 let mut idx = 0;
538 points
539 .map(|p2| {
540 while idx < n && 2 * i128::from(begin[idx]) < p2 {
541 idx += 1;
542 }
543 if idx < n && 2 * i128::from(begin[idx]) == p2 {
544 return Take::Row(idx);
545 }
546 if idx == 0 {
547 return Take::Nothing;
548 }
549 if idx == n {
550 return Take::Row(n - 1);
551 }
552 let (p, q) = (idx - 1, idx);
553 let t = p2 as f64 / 2.0;
555 let (bp, bq) = (begin[p] as f64, begin[q] as f64);
556 Take::Between(p, q, (t - bp) / (bq - bp))
557 })
558 .collect()
559 }
560}
561
562fn sample_points(buckets: &IntervalKeys, at: At) -> impl Iterator<Item = i128> + '_ {
564 buckets
565 .begin
566 .iter()
567 .zip(&buckets.end)
568 .map(move |(&b, &e)| match at {
569 At::Begin => 2 * i128::from(b),
570 At::Center => i128::from(b) + i128::from(e),
571 At::End => 2 * i128::from(e),
572 })
573}
574
575pub(crate) fn no_buckets() -> IntervalKeys {
576 IntervalKeys {
577 label: Vec::<f64>::new().into(),
578 begin: vec![],
579 end: vec![],
580 }
581}
582
583fn best_row(data: &ColumnData, group: &[usize], min: bool) -> usize {
586 let better = |a: std::cmp::Ordering| {
587 if min {
588 a == std::cmp::Ordering::Less
589 } else {
590 a == std::cmp::Ordering::Greater
591 }
592 };
593 let mut best = group[0];
594 match data {
595 ColumnData::Number(c) => {
596 let mut best_v = c.get(best);
597 for &i in &group[1..] {
598 let Some(v) = c.get(i) else { continue };
599 if best_v.is_none_or(|b| v.partial_cmp(&b).is_some_and(better)) {
600 best = i;
601 best_v = Some(v);
602 }
603 }
604 }
605 ColumnData::Decimal(c) => {
606 let mut best_v = c.get(best).map(|d| d.value);
607 for &i in &group[1..] {
608 let Some(v) = c.get(i).map(|d| d.value) else {
609 continue;
610 };
611 if best_v.is_none_or(|b| better(v.cmp(&b))) {
612 best = i;
613 best_v = Some(v);
614 }
615 }
616 }
617 _ => unreachable!("checked to be number or decimal"),
618 }
619 best
620}
621
622#[derive(Hash, PartialEq, Eq)]
625enum LabelKey {
626 Number(u64),
627 Text(Option<String>),
628}
629
630impl LabelKey {
631 fn of(labels: &Labels, i: usize) -> Self {
632 match labels {
633 Labels::Number(v) => LabelKey::Number((v[i] + 0.0).to_bits()),
634 Labels::String(c) => LabelKey::Text(c.get(i).map(str::to_string)),
635 }
636 }
637}
638
639pub(crate) fn iso_time(ms: i64) -> String {
642 let days = ms.div_euclid(86_400_000);
643 let in_day = ms.rem_euclid(86_400_000);
644 let z = days + 719_468;
647 let era = z.div_euclid(146_097);
648 let doe = z.rem_euclid(146_097);
649 let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
650 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
651 let mp = (5 * doy + 2) / 153;
652 let day = doy - (153 * mp + 2) / 5 + 1;
653 let month = if mp < 10 { mp + 3 } else { mp - 9 };
654 let year = yoe + era * 400 + i64::from(month <= 2);
655 let year = if (0..=9999).contains(&year) {
656 format!("{year:04}")
657 } else {
658 format!(
659 "{}{:06}",
660 if year < 0 { '-' } else { '+' },
661 year.unsigned_abs()
662 )
663 };
664 format!(
665 "{year}-{month:02}-{day:02}T{:02}:{:02}:{:02}.{:03}Z",
666 in_day / 3_600_000,
667 in_day / 60_000 % 60,
668 in_day / 1_000 % 60,
669 in_day % 1_000
670 )
671}
672
673pub(crate) fn js_number(x: f64) -> String {
677 if x == 0.0 {
678 return "0".into();
679 }
680 if !x.is_finite() {
681 return if x.is_nan() {
682 "NaN".into()
683 } else if x > 0.0 {
684 "Infinity".into()
685 } else {
686 "-Infinity".into()
687 };
688 }
689 let sci = format!("{:e}", x.abs());
691 let (mantissa, exp) = sci.split_once('e').expect("exponent form");
692 let digits: String = mantissa.chars().filter(|c| *c != '.').collect();
693 let k = digits.len() as i32;
694 let n = exp.parse::<i32>().expect("exponent") + 1; let sign = if x < 0.0 { "-" } else { "" };
696 let body = if k <= n && n <= 21 {
697 format!("{digits}{}", "0".repeat((n - k) as usize))
698 } else if 0 < n && n <= 21 {
699 format!("{}.{}", &digits[..n as usize], &digits[n as usize..])
700 } else if -6 < n && n <= 0 {
701 format!("0.{}{digits}", "0".repeat((-n) as usize))
702 } else {
703 let e = n - 1;
704 let exp = if e < 0 {
705 format!("-{}", -e)
706 } else {
707 format!("+{e}")
708 };
709 if k == 1 {
710 format!("{digits}e{exp}")
711 } else {
712 format!("{}.{}e{exp}", &digits[..1], &digits[1..])
713 }
714 };
715 format!("{sign}{body}")
716}
717
718#[cfg(test)]
719mod tests {
720 use super::*;
721
722 use crate::column::DecimalColumn;
723 use crate::error::PondError;
724
725 fn s(times: Vec<i64>, columns: Vec<Column>) -> TimeSeries {
726 TimeSeries::new("s", times, columns).unwrap()
727 }
728
729 #[test]
730 fn materialize_refuses_a_half_millisecond() {
731 let x = s(vec![0, 5], vec![Column::required("v", vec![1.0, 2.0])]);
732 let odd = Sequence::every_ms(3).unwrap();
733 let centre = Materialize {
734 sample: At::Center,
735 ..Materialize::default()
736 };
737 let e = x.materialize(&odd, centre).unwrap_err();
738 assert!(matches!(e, PondError::Unsupported { .. }), "{e}");
739 assert!(e.to_string().contains("half a millisecond"), "{e}");
740 let even = Sequence::every_ms(4).unwrap();
742 assert_eq!(x.materialize(&even, centre).unwrap().key().begin(), &[2]);
743 let aligned = x
745 .align(
746 &odd,
747 Align {
748 sample: At::Center,
749 method: AlignMethod::Linear,
750 ..Align::default()
751 },
752 )
753 .unwrap();
754 let v = aligned.column("v").unwrap().numbers().unwrap();
756 assert_eq!(v.values(), &[1.3, 1.9]);
757 }
758
759 #[test]
760 fn keep_min_max_need_a_number_or_decimal_column() {
761 let x = s(
762 vec![0, 0],
763 vec![
764 Column::required("v", vec![1.0, 2.0]),
765 Column::required("t", vec!["a", "b"]),
766 ],
767 );
768 let e = x.dedupe(Keep::Min("t".into())).unwrap_err();
769 assert!(matches!(e, PondError::WrongKind { .. }), "{e}");
770 let e = x.dedupe(Keep::Max("nope".into())).unwrap_err();
771 assert!(matches!(e, PondError::UnknownColumn { .. }), "{e}");
772 let single = s(vec![0], vec![Column::required("t", vec!["a"])]);
774 assert!(single.dedupe(Keep::Min("t".into())).is_err());
775 }
776
777 #[test]
778 fn decimals_hold_under_linear_and_compare_exactly() {
779 let d = DecimalColumn::from_values(vec![100, 300, 299], 2).unwrap();
780 let x = s(vec![0, 2_000, 2_000], vec![Column::required("d", d)]);
781 let aligned = x
782 .align(
783 &Sequence::every("1s").unwrap(),
784 Align {
785 method: AlignMethod::Linear,
786 ..Align::default()
787 },
788 )
789 .unwrap();
790 let d = aligned.column("d").unwrap().decimals().unwrap();
791 let got: Vec<i128> = d.iter().map(|v| v.unwrap().value).collect();
793 assert_eq!(got, vec![100, 100, 300]);
794 let min = x.dedupe(Keep::Min("d".into())).unwrap();
795 let got: Vec<i128> = min
796 .column("d")
797 .unwrap()
798 .decimals()
799 .unwrap()
800 .iter()
801 .map(|v| v.unwrap().value)
802 .collect();
803 assert_eq!(got, vec![100, 299]);
804 }
805
806 #[test]
807 fn dedupe_error_names_the_key_as_pond_ts_does() {
808 let x = s(
809 vec![7_700, 7_700],
810 vec![Column::required("v", vec![1.0, 2.0])],
811 );
812 let e = x.dedupe(Keep::Error).unwrap_err().to_string();
813 assert_eq!(
814 e,
815 "dedupe: 2 events at 1970-01-01T00:00:07.700Z (7700). \
816 Specify a different 'keep' policy or fix upstream."
817 );
818 let keys = IntervalKeys {
819 label: vec![1.5, 1.5].into(),
820 begin: vec![0, 0],
821 end: vec![10, 10],
822 };
823 let x = TimeSeries::from_intervals("i", keys, vec![]).unwrap();
824 let e = x.dedupe(Keep::Error).unwrap_err().to_string();
825 assert_eq!(
826 e,
827 "dedupe: 2 events at key \"0:10:1.5\". \
828 Specify a different 'keep' policy or fix upstream."
829 );
830 }
831
832 #[test]
833 fn options_are_checked() {
834 let x = s(vec![0], vec![Column::required("v", vec![1.0])]);
835 let every = Sequence::every("1s").unwrap();
836 let reversed = Align {
837 range: Some((5, 0)),
838 ..Align::default()
839 };
840 assert!(x.align(&every, reversed).is_err());
841 let narrow = Materialize {
843 range: Some((1, 999)),
844 ..Materialize::default()
845 };
846 assert!(x.materialize(&every, narrow).unwrap().is_empty());
847 }
848
849 #[test]
850 fn js_number_matches_javascript() {
851 for (x, js) in [
852 (1.0, "1"),
853 (1.5, "1.5"),
854 (-0.25, "-0.25"),
855 (0.1 + 0.2, "0.30000000000000004"),
856 (123456789.0, "123456789"),
857 (1e21, "1e+21"),
858 (1.5e21, "1.5e+21"),
859 (1e20, "100000000000000000000"),
860 (1e-6, "0.000001"),
861 (1e-7, "1e-7"),
862 (1.25e-7, "1.25e-7"),
863 (-0.0, "0"),
864 (5e-324, "5e-324"),
865 (f64::MAX, "1.7976931348623157e+308"),
866 ] {
867 assert_eq!(js_number(x), js, "{x}");
868 }
869 }
870
871 #[test]
872 fn iso_time_matches_javascript() {
873 assert_eq!(iso_time(0), "1970-01-01T00:00:00.000Z");
874 assert_eq!(iso_time(-1), "1969-12-31T23:59:59.999Z");
875 assert_eq!(iso_time(1_792_157_400_123), "2026-10-16T13:30:00.123Z");
876 assert_eq!(
877 iso_time(8_640_000_000_000_000),
878 "+275760-09-13T00:00:00.000Z"
879 );
880 assert_eq!(iso_time(-62_198_755_200_000), "-000001-01-01T00:00:00.000Z");
881 assert_eq!(iso_time(951_782_400_000), "2000-02-29T00:00:00.000Z");
882 assert_eq!(iso_time(-2_208_988_800_001), "1899-12-31T23:59:59.999Z");
883 assert_eq!(iso_time(253_402_300_800_000), "+010000-01-01T00:00:00.000Z");
884 assert_eq!(
885 iso_time(-8_640_000_000_000_000),
886 "-271821-04-20T00:00:00.000Z"
887 );
888 }
889}