WIP

daqbufp2/src/test.rs

@@ -28,6 +28,22 @@ fn f32_cmp_near(x: f32, y: f32) -> bool {
     x == y
 }
 
+fn f64_cmp_near(x: f64, y: f64) -> bool {
+    let x = {
+        let mut a = x.to_le_bytes();
+        a[0] &= 0x00;
+        a[1] &= 0x00;
+        f64::from_ne_bytes(a)
+    };
+    let y = {
+        let mut a = y.to_le_bytes();
+        a[0] &= 0x00;
+        a[1] &= 0x00;
+        f64::from_ne_bytes(a)
+    };
+    x == y
+}
+
 fn f32_iter_cmp_near<A, B>(a: A, b: B) -> bool
 where
     A: IntoIterator<Item = f32>,
@@ -50,6 +66,28 @@ where
     }
 }
 
+fn f64_iter_cmp_near<A, B>(a: A, b: B) -> bool
+where
+    A: IntoIterator<Item = f64>,
+    B: IntoIterator<Item = f64>,
+{
+    let mut a = a.into_iter();
+    let mut b = b.into_iter();
+    loop {
+        let x = a.next();
+        let y = b.next();
+        if let (Some(x), Some(y)) = (x, y) {
+            if !f64_cmp_near(x, y) {
+                return false;
+            }
+        } else if x.is_some() || y.is_some() {
+            return false;
+        } else {
+            return true;
+        }
+    }
+}
+
 #[test]
 fn test_f32_iter_cmp_near() {
     let a = [-127.553e17];

daqbufp2/src/test/api4/binnedjson.rs

@@ -2,6 +2,8 @@ use crate::err::ErrConv;
 use crate::nodes::require_test_hosts_running;
 use crate::test::api4::common::fetch_binned_json;
 use crate::test::f32_cmp_near;
+use crate::test::f32_iter_cmp_near;
+use crate::test::f64_iter_cmp_near;
 use chrono::Utc;
 use err::Error;
 use http::StatusCode;
@@ -158,8 +160,9 @@ fn binned_d0_json_02() -> Result<(), Error> {
                 name: "test-gen-f64-dim1-v00".into(),
                 series: None,
             },
-            "1970-01-01T00:20:10.000Z",
+            "1970-01-01T00:20:00Z",
-            "1970-01-01T01:20:45.000Z",
+            "1970-01-01T00:20:10Z",
+            //"1970-01-01T01:20:45.000Z",
             10,
             cluster,
         )
@@ -167,9 +170,30 @@ fn binned_d0_json_02() -> Result<(), Error> {
         debug!("Receveided a response json value: {jsv:?}");
         let res: BinsDim0CollectedResult<f64> = serde_json::from_value(jsv)?;
         // inmem was meant just for functional test, ignores the requested time range
-        assert_eq!(res.ts_anchor_sec(), 1200);
-        assert_eq!(res.len(), 13);
         assert_eq!(res.range_final(), true);
+        assert_eq!(res.len(), 10);
+        assert_eq!(res.ts_anchor_sec(), 1200);
+        let nb = res.len();
+        {
+            let a1: Vec<_> = res.ts1_off_ms().iter().map(|x| *x).collect();
+            let a2: Vec<_> = (0..nb as _).into_iter().map(|x| 300 * 1000 * x).collect();
+            assert_eq!(a1, a2);
+        }
+        {
+            let a1: Vec<_> = res.ts2_off_ms().iter().map(|x| *x).collect();
+            let a2: Vec<_> = (0..nb as _).into_iter().map(|x| 300 * 1000 * (1 + x)).collect();
+            assert_eq!(a1, a2);
+        }
+        {
+            let a1: Vec<_> = res.counts().iter().map(|x| *x).collect();
+            let a2: Vec<_> = (0..nb as _).into_iter().map(|_| 1024).collect();
+            assert_eq!(a1, a2);
+        }
+        {
+            let a1: Vec<_> = res.mins().iter().map(|x| *x).collect();
+            let a2: Vec<_> = (0..nb as _).into_iter().map(|_| 0.1).collect();
+            assert_eq!(f64_iter_cmp_near(a1, a2), true);
+        }
         Ok(())
     };
     taskrun::run(fut)
@@ -420,7 +444,7 @@ async fn get_binned_json(
     let s = String::from_utf8_lossy(&buf);
     let res: JsonValue = serde_json::from_str(&s)?;
     let pretty = serde_json::to_string_pretty(&res)?;
-    trace!("{pretty}");
+    info!("Received from remote:\n{pretty}");
     let t2 = chrono::Utc::now();
     let ms = t2.signed_duration_since(t1).num_milliseconds() as u64;
     // TODO add timeout

daqbufp2/src/test/api4/eventsjson.rs

@@ -71,7 +71,7 @@ fn events_plain_json_01() -> Result<(), Error> {
         assert_eq!(res.pulse_anchor(), 2420);
         let exp = [2420., 2421., 2422., 2423., 2424., 2425.];
         assert_eq!(f32_iter_cmp_near(res.values_to_f32(), exp), true);
-        assert_eq!(res.range_complete(), true);
+        assert_eq!(res.range_final(), true);
         assert_eq!(res.timed_out(), false);
         Ok(())
     };
@@ -95,7 +95,7 @@ fn events_plain_json_02_range_incomplete() -> Result<(), Error> {
         )
         .await?;
         let res: EventsDim0CollectorOutput<i32> = serde_json::from_value(jsv).unwrap();
-        assert_eq!(res.range_complete(), false);
+        assert_eq!(res.range_final(), false);
         assert_eq!(res.timed_out(), false);
         Ok(())
     };

disk/src/decode.rs

@@ -304,6 +304,10 @@ pub struct EventsDynStream {
 }
 
 impl EventsDynStream {
+    pub fn type_name() -> &'static str {
+        std::any::type_name::<Self>()
+    }
+
     pub fn new(
         scalar_type: ScalarType,
         shape: Shape,
@@ -405,7 +409,7 @@ impl Stream for EventsDynStream {
         use Poll::*;
         loop {
             break if self.complete {
-                panic!("poll_next on complete")
+                panic!("{} poll_next on complete", Self::type_name())
             } else if self.done {
                 self.complete = true;
                 Ready(None)

disk/src/disk.rs

@@ -202,6 +202,10 @@ pub struct FileContentStream {
 }
 
 impl FileContentStream {
+    pub fn type_name() -> &'static str {
+        std::any::type_name::<Self>()
+    }
+
     pub fn new(file: File, disk_io_tune: DiskIoTune) -> Self {
         Self {
             file,
@@ -223,7 +227,7 @@ impl Stream for FileContentStream {
         use Poll::*;
         loop {
             break if self.complete {
-                panic!("poll_next on complete")
+                panic!("{} poll_next on complete", Self::type_name())
             } else if self.done {
                 self.complete = true;
                 Ready(None)
@@ -385,6 +389,10 @@ pub struct FileContentStream2 {
 }
 
 impl FileContentStream2 {
+    pub fn type_name() -> &'static str {
+        std::any::type_name::<Self>()
+    }
+
     pub fn new(file: File, disk_io_tune: DiskIoTune) -> Self {
         let file = Box::pin(file);
         Self {
@@ -412,7 +420,7 @@ impl Stream for FileContentStream2 {
         use Poll::*;
         loop {
             break if self.complete {
-                panic!("poll_next on complete")
+                panic!("{} poll_next on complete", Self::type_name())
             } else if self.done {
                 self.complete = true;
                 Ready(None)

disk/src/eventblobs.rs

@@ -52,6 +52,10 @@ pub struct EventChunkerMultifile {
 }
 
 impl EventChunkerMultifile {
+    pub fn type_name() -> &'static str {
+        std::any::type_name::<Self>()
+    }
+
     pub fn new(
         range: NanoRange,
         channel_config: SfDbChConf,
@@ -104,7 +108,7 @@ impl Stream for EventChunkerMultifile {
             break if let Some(item) = self.log_queue.pop_front() {
                 Ready(Some(Ok(StreamItem::Log(item))))
             } else if self.complete {
-                panic!("EventChunkerMultifile  poll_next on complete");
+                panic!("{} poll_next on complete", Self::type_name());
             } else if self.done_emit_range_final {
                 self.complete = true;
                 Ready(None)

items_0/src/overlap.rs

@@ -31,7 +31,6 @@ macro_rules! impl_range_overlap_info_events {
                 if range.is_time() {
                     if let Some(max) = HasTimestampDeque::timestamp_max(self) {
                         max < range.beg_u64()
-                        //<Self as RangeOverlapCmp>::range_overlap_cmp_beg(max, range.beg_u64())
                     } else {
                         true
                     }

items_0/src/timebin.rs

@@ -1,3 +1,5 @@
+pub mod timebinimpl;
+
 use crate::collect_s::Collectable;
 use crate::collect_s::Collector;
 use crate::collect_s::ToJsonResult;
@@ -7,6 +9,7 @@ use crate::AsAnyRef;
 use crate::Events;
 use crate::TypeName;
 use crate::WithLen;
+use err::Error;
 use netpod::log::*;
 use netpod::range::evrange::SeriesRange;
 use netpod::BinnedRangeEnum;
@@ -298,3 +301,7 @@ impl TimeBinnableTy for Box<dyn TimeBinnable> {
         TimeBinnerDynStruct::new(binrange, do_time_weight, binner)
     }
 }
+
+pub trait TimeBinnerIngest: fmt::Debug + TypeName + Send {
+    fn ingest_inrange(&mut self, item: &mut dyn TimeBinnable) -> Result<(), Error>;
+}

items_0/src/timebin/timebinimpl.rs

@@ -0,0 +1,199 @@
+#![allow(unused)]
+use crate::timebin::TimeBinnable;
+use crate::timebin::TimeBinned;
+use crate::timebin::TimeBinner;
+use crate::timebin::TimeBinnerIngest;
+use crate::TypeName;
+use netpod::log::*;
+use netpod::range::evrange::NanoRange;
+
+#[allow(unused)]
+macro_rules! trace2 {
+    ($($arg:tt)*) => { trace!($($arg)*) };
+}
+
+#[allow(unused)]
+macro_rules! trace_ingest {
+    ($($arg:tt)*) => { trace!($($arg)*) };
+}
+
+#[cfg(DISABLED)]
+impl<T> TimeBinner for T
+where
+    T: TimeBinnerIngest,
+{
+    fn bins_ready_count(&self) -> usize {
+        match &self.ready {
+            Some(k) => k.len(),
+            None => 0,
+        }
+    }
+
+    fn bins_ready(&mut self) -> Option<Box<dyn TimeBinned>> {
+        match self.ready.take() {
+            Some(k) => Some(Box::new(k)),
+            None => None,
+        }
+    }
+
+    fn ingest(&mut self, item: &mut dyn TimeBinnable) {
+        trace2!(
+            "TimeBinner for {} ingest  agg.range {:?}  item {:?}",
+            self.type_name(),
+            self.agg.range(),
+            item
+        );
+        if item.len() == 0 {
+            // Return already here, RangeOverlapInfo would not give much sense.
+            return;
+        }
+        // TODO optimize by remembering at which event array index we have arrived.
+        // That needs modified interfaces which can take and yield the start and latest index.
+        loop {
+            while item.starts_after(self.agg.range()) {
+                trace!(
+                    "{}  IGNORE ITEM  AND CYCLE  BECAUSE item.starts_after",
+                    self.type_name()
+                );
+                self.cycle();
+                if self.rng.is_none() {
+                    warn!("{}  no more bin in edges B", self.type_name());
+                    return;
+                }
+            }
+            if item.ends_before(self.agg.range()) {
+                trace!("{}  IGNORE ITEM  BECAUSE ends_before", self.type_name());
+                return;
+            } else {
+                if self.rng.is_none() {
+                    trace!("{}  no more bin in edges D", self.type_name());
+                    return;
+                } else {
+                    match TimeBinnerIngest::ingest_inrange(self, item) {
+                        Ok(()) => {
+                            if item.ends_after(self.agg.range()) {
+                                trace_ingest!("{}  FED ITEM, ENDS AFTER.", self.type_name());
+                                self.cycle();
+                                if self.rng.is_none() {
+                                    warn!("{}  no more bin in edges C", self.type_name());
+                                    return;
+                                } else {
+                                    trace_ingest!("{}  FED ITEM, CYCLED, CONTINUE.", self.type_name());
+                                }
+                            } else {
+                                trace_ingest!("{}  FED ITEM.", self.type_name());
+                                break;
+                            }
+                        }
+                        Err(e) => {
+                            error!("{}::ingest  {}", self.type_name(), e);
+                        }
+                    }
+                    /*
+                    // Move to TimeBinnerIngest
+                    if let Some(item) = item
+                        .as_any_ref()
+                        // TODO make statically sure that we attempt to cast to the correct type here:
+                        .downcast_ref::<<EventsDim0Aggregator<STY> as TimeBinnableTypeAggregator>::Input>()
+                    {
+                        // TODO collect statistics associated with this request:
+                        trace_ingest!("{self_name}  FEED THE ITEM...");
+                        self.agg.ingest(item);
+                    } else {
+                        error!("{self_name}::ingest  unexpected item type");
+                    };
+                    */
+                }
+            }
+        }
+    }
+
+    fn push_in_progress(&mut self, push_empty: bool) {
+        trace!("{}::push_in_progress  push_empty {push_empty}", self.type_name());
+        // TODO expand should be derived from AggKind. Is it still required after all?
+        // TODO here, the expand means that agg will assume that the current value is kept constant during
+        // the rest of the time range.
+        if self.rng.is_none() {
+        } else {
+            let expand = true;
+            let range_next = self.next_bin_range();
+            self.rng = range_next.clone();
+            let mut bins = if let Some(range_next) = range_next {
+                self.agg.result_reset(range_next, expand)
+            } else {
+                // Acts as placeholder
+                let range_next = NanoRange {
+                    beg: u64::MAX - 1,
+                    end: u64::MAX,
+                };
+                self.agg.result_reset(range_next.into(), expand)
+            };
+            if bins.len() != 1 {
+                error!("{}::push_in_progress  bins.len() {}", self.type_name(), bins.len());
+                return;
+            } else {
+                if push_empty || bins.counts[0] != 0 {
+                    match self.ready.as_mut() {
+                        Some(ready) => {
+                            ready.append_all_from(&mut bins);
+                        }
+                        None => {
+                            self.ready = Some(bins);
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    fn cycle(&mut self) {
+        trace!("{}::cycle", self.type_name());
+        // TODO refactor this logic.
+        let n = self.bins_ready_count();
+        self.push_in_progress(true);
+        if self.bins_ready_count() == n {
+            let range_next = self.next_bin_range();
+            self.rng = range_next.clone();
+            if let Some(range) = range_next {
+                /*
+                TODO Move out to trait.
+                let mut bins = BinsDim0::empty();
+                if range.is_time() {
+                    bins.append_zero(range.beg_u64(), range.end_u64());
+                } else {
+                    error!("TODO  {self_name}::cycle  is_pulse");
+                }
+                match self.ready.as_mut() {
+                    Some(ready) => {
+                        ready.append_all_from(&mut bins);
+                    }
+                    None => {
+                        self.ready = Some(bins);
+                    }
+                }
+                */
+                if self.bins_ready_count() <= n {
+                    error!("{}::cycle  failed to push a zero bin", self.type_name());
+                }
+            } else {
+                warn!(
+                    "{}::cycle  no in-progress bin pushed, but also no more bin to add as zero-bin",
+                    self.type_name()
+                );
+            }
+        }
+    }
+
+    fn set_range_complete(&mut self) {
+        self.range_final = true;
+    }
+
+    fn empty(&self) -> Box<dyn TimeBinned> {
+        /*
+        TODO factor out to trait.
+        let ret = <EventsDim0Aggregator<STY> as TimeBinnableTypeAggregator>::Output::empty();
+        */
+        let ret = todo!();
+        Box::new(ret)
+    }
+}

items_2/src/binsxbindim0.rs

@@ -48,7 +48,7 @@ macro_rules! trace4 {
 pub struct BinsXbinDim0<NTY> {
     ts1s: VecDeque<u64>,
     ts2s: VecDeque<u64>,
-    counts: VecDeque<u64>,
+    pub counts: VecDeque<u64>,
     mins: VecDeque<NTY>,
     maxs: VecDeque<NTY>,
     avgs: VecDeque<f32>,

items_2/src/eventsdim0.rs

@@ -257,7 +257,7 @@ impl<STY: ScalarOps> EventsDim0CollectorOutput<STY> {
         self.values.iter().map(|x| x.as_prim_f32_b()).collect()
     }
 
-    pub fn range_complete(&self) -> bool {
+    pub fn range_final(&self) -> bool {
         self.range_final
     }
 
@@ -457,10 +457,10 @@ impl<STY: ScalarOps> EventsDim0Aggregator<STY> {
         format!("{}<{}>", any::type_name::<Self>(), any::type_name::<STY>())
     }
 
-    pub fn new(binrange: SeriesRange, do_time_weight: bool) -> Self {
+    pub fn new(range: SeriesRange, do_time_weight: bool) -> Self {
-        let int_ts = binrange.beg_u64();
+        let int_ts = range.beg_u64();
         Self {
-            range: binrange,
+            range,
             count: 0,
             min: STY::zero_b(),
             max: STY::zero_b(),
@@ -521,11 +521,7 @@ impl<STY: ScalarOps> EventsDim0Aggregator<STY> {
             if px > pxbeg {
                 self.apply_min_max(v2);
             }
-            let w = if self.do_time_weight {
+            let w = (px - self.int_ts) as f32 * 1e-9;
-                (px - self.int_ts) as f32 * 1e-9
-            } else {
-                1.
-            };
             if vf.is_nan() {
             } else {
                 self.sum += vf * w;
@@ -570,9 +566,9 @@ impl<STY: ScalarOps> EventsDim0Aggregator<STY> {
                 let val = item.values[i1].clone();
                 if ts < self.int_ts {
                     trace_ingest!("{self_name} ingest  {:6}  {:20}  {:10?}  BEFORE", i1, ts, val);
-                    self.events_ignored_count += 1;
                     self.last_seen_ts = ts;
                     self.last_seen_val = Some(val);
+                    self.events_ignored_count += 1;
                 } else if ts >= self.range.end_u64() {
                     trace_ingest!("{self_name} ingest  {:6}  {:20}  {:10?}  AFTER", i1, ts, val);
                     self.events_ignored_count += 1;
@@ -640,8 +636,10 @@ impl<STY: ScalarOps> EventsDim0Aggregator<STY> {
     fn result_reset_time_weight(&mut self, range: SeriesRange) -> BinsDim0<STY> {
         // TODO check callsite for correct expand status.
         debug!("result_reset_time_weight calls apply_event_time_weight");
+        let range_beg = self.range.beg_u64();
+        let range_end = self.range.end_u64();
         if self.range.is_time() {
-            self.apply_event_time_weight(self.range.end_u64(), self.range.beg_u64());
+            self.apply_event_time_weight(range_end, range_beg);
         } else {
             error!("TODO result_reset_time_weight");
             err::todoval()
@@ -658,8 +656,8 @@ impl<STY: ScalarOps> EventsDim0Aggregator<STY> {
         };
         let ret = if self.range.is_time() {
             BinsDim0 {
-                ts1s: [self.range.beg_u64()].into(),
+                ts1s: [range_beg].into(),
-                ts2s: [self.range.end_u64()].into(),
+                ts2s: [range_end].into(),
                 counts: [self.count].into(),
                 mins: [min].into(),
                 maxs: [max].into(),
@@ -670,7 +668,7 @@ impl<STY: ScalarOps> EventsDim0Aggregator<STY> {
             error!("TODO result_reset_time_weight");
             err::todoval()
         };
-        self.int_ts = range.beg_u64();
+        self.int_ts = range_beg;
         self.range = range;
         self.count = 0;
         self.sum = 0.;

items_2/src/eventsxbindim0.rs

@@ -15,6 +15,9 @@ use items_0::overlap::HasTimestampDeque;
 use items_0::overlap::RangeOverlapCmp;
 use items_0::scalar_ops::ScalarOps;
 use items_0::timebin::TimeBinnable;
+use items_0::timebin::TimeBinned;
+use items_0::timebin::TimeBinner;
+use items_0::timebin::TimeBinnerTy;
 use items_0::AsAnyMut;
 use items_0::AsAnyRef;
 use items_0::Empty;
@@ -25,7 +28,9 @@ use items_0::TypeName;
 use items_0::WithLen;
 use netpod::is_false;
 use netpod::log::*;
+use netpod::range::evrange::NanoRange;
 use netpod::range::evrange::SeriesRange;
+use netpod::timeunits::SEC;
 use netpod::BinnedRangeEnum;
 use serde::Deserialize;
 use serde::Serialize;
@@ -35,6 +40,18 @@ use std::collections::VecDeque;
 use std::fmt;
 use std::mem;
 
+#[allow(unused)]
+macro_rules! trace_ingest {
+    (e$($arg:tt)*) => {};
+    ($($arg:tt)*) => { trace!($($arg)*) };
+}
+
+#[allow(unused)]
+macro_rules! trace2 {
+    (e$($arg:tt)*) => {};
+    ($($arg:tt)*) => { trace!($($arg)*) };
+}
+
 #[derive(Clone, PartialEq, Serialize, Deserialize)]
 pub struct EventsXbinDim0<NTY> {
     pub tss: VecDeque<u64>,
@@ -75,28 +92,43 @@ impl<STY> TypeName for EventsXbinDim0<STY> {
     }
 }
 
-impl<NTY> fmt::Debug for EventsXbinDim0<NTY>
+impl<STY> fmt::Debug for EventsXbinDim0<STY>
 where
-    NTY: fmt::Debug,
+    STY: fmt::Debug,
 {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
-        fmt.debug_struct("EventsXbinDim0")
+        if false {
-            .field("tss", &self.tss)
+            write!(
-            .field("pulses", &self.pulses)
+                fmt,
-            .field("mins", &self.mins)
+                "{} {{ count {}  ts {:?}  vals {:?} }}",
-            .field("maxs", &self.maxs)
+                self.type_name(),
-            .field("avgs", &self.avgs)
+                self.tss.len(),
-            .finish()
+                self.tss.iter().map(|x| x / SEC).collect::<Vec<_>>(),
+                self.avgs,
+            )
+        } else {
+            write!(
+                fmt,
+                "{} {{ count {}  ts {:?} .. {:?}  vals {:?} .. {:?} }}",
+                self.type_name(),
+                self.tss.len(),
+                self.tss.front().map(|x| x / SEC),
+                self.tss.back().map(|x| x / SEC),
+                self.avgs.front(),
+                self.avgs.back(),
+            )
+        }
     }
 }
 
-impl<NTY> ByteEstimate for EventsXbinDim0<NTY> {
+impl<STY> ByteEstimate for EventsXbinDim0<STY> {
     fn byte_estimate(&self) -> u64 {
-        todo!("byte_estimate")
+        let stylen = mem::size_of::<STY>();
+        (self.len() * (8 + 8 + 2 * stylen + 4)) as u64
     }
 }
 
-impl<NTY> Empty for EventsXbinDim0<NTY> {
+impl<STY> Empty for EventsXbinDim0<STY> {
     fn empty() -> Self {
         Self {
             tss: VecDeque::new(),
@@ -108,25 +140,25 @@ impl<NTY> Empty for EventsXbinDim0<NTY> {
     }
 }
 
-impl<NTY> AsAnyRef for EventsXbinDim0<NTY>
+impl<STY> AsAnyRef for EventsXbinDim0<STY>
 where
-    NTY: ScalarOps,
+    STY: ScalarOps,
 {
     fn as_any_ref(&self) -> &dyn Any {
         self
     }
 }
 
-impl<NTY> AsAnyMut for EventsXbinDim0<NTY>
+impl<STY> AsAnyMut for EventsXbinDim0<STY>
 where
-    NTY: ScalarOps,
+    STY: ScalarOps,
 {
     fn as_any_mut(&mut self) -> &mut dyn Any {
         self
     }
 }
 
-impl<NTY> WithLen for EventsXbinDim0<NTY> {
+impl<STY> WithLen for EventsXbinDim0<STY> {
     fn len(&self) -> usize {
         self.tss.len()
     }
@@ -336,12 +368,221 @@ impl<STY: ScalarOps> Events for EventsXbinDim0<STY> {
     }
 }
 
-impl<NTY> TimeBinnableType for EventsXbinDim0<NTY>
+#[derive(Debug)]
+pub struct EventsXbinDim0TimeBinner<STY: ScalarOps> {
+    binrange: BinnedRangeEnum,
+    rix: usize,
+    rng: Option<SeriesRange>,
+    agg: EventsXbinDim0Aggregator<STY>,
+    ready: Option<<EventsXbinDim0Aggregator<STY> as TimeBinnableTypeAggregator>::Output>,
+    range_final: bool,
+}
+
+impl<STY: ScalarOps> EventsXbinDim0TimeBinner<STY> {
+    fn type_name() -> &'static str {
+        any::type_name::<Self>()
+    }
+
+    fn new(binrange: BinnedRangeEnum, do_time_weight: bool) -> Result<Self, Error> {
+        trace!("{}::new  binrange {:?}", Self::type_name(), binrange);
+        let rng = binrange
+            .range_at(0)
+            .ok_or_else(|| Error::with_msg_no_trace("empty binrange"))?;
+        trace!("{}::new  rng {rng:?}", Self::type_name());
+        let agg = EventsXbinDim0Aggregator::new(rng, do_time_weight);
+        trace!("{}  agg range {:?}", Self::type_name(), agg.range());
+        let ret = Self {
+            binrange,
+            rix: 0,
+            rng: Some(agg.range.clone()),
+            agg,
+            ready: None,
+            range_final: false,
+        };
+        Ok(ret)
+    }
+
+    fn next_bin_range(&mut self) -> Option<SeriesRange> {
+        self.rix += 1;
+        if let Some(rng) = self.binrange.range_at(self.rix) {
+            trace!("{}  next_bin_range {:?}", Self::type_name(), rng);
+            Some(rng)
+        } else {
+            trace!("{}  next_bin_range None", Self::type_name());
+            None
+        }
+    }
+}
+
+impl<STY: ScalarOps> TimeBinner for EventsXbinDim0TimeBinner<STY> {
+    fn bins_ready_count(&self) -> usize {
+        match &self.ready {
+            Some(k) => k.len(),
+            None => 0,
+        }
+    }
+
+    fn bins_ready(&mut self) -> Option<Box<dyn TimeBinned>> {
+        match self.ready.take() {
+            Some(k) => Some(Box::new(k)),
+            None => None,
+        }
+    }
+
+    fn ingest(&mut self, item: &mut dyn TimeBinnable) {
+        trace2!(
+            "TimeBinner for {} ingest  agg.range {:?}  item {:?}",
+            Self::type_name(),
+            self.agg.range(),
+            item
+        );
+        if item.len() == 0 {
+            // Return already here, RangeOverlapInfo would not give much sense.
+            return;
+        }
+        // TODO optimize by remembering at which event array index we have arrived.
+        // That needs modified interfaces which can take and yield the start and latest index.
+        loop {
+            while item.starts_after(self.agg.range()) {
+                trace!(
+                    "{}  IGNORE ITEM  AND CYCLE  BECAUSE item.starts_after",
+                    Self::type_name()
+                );
+                self.cycle();
+                if self.rng.is_none() {
+                    warn!("{}  no more bin in edges B", Self::type_name());
+                    return;
+                }
+            }
+            if item.ends_before(self.agg.range()) {
+                trace!(
+                    "{}  IGNORE ITEM  BECAUSE ends_before  {:?}  {:?}",
+                    Self::type_name(),
+                    self.agg.range(),
+                    item
+                );
+                return;
+            } else {
+                if self.rng.is_none() {
+                    trace!("{}  no more bin in edges D", Self::type_name());
+                    return;
+                } else {
+                    if let Some(item) = item
+                        .as_any_ref()
+                        // TODO make statically sure that we attempt to cast to the correct type here:
+                        .downcast_ref::<<EventsXbinDim0Aggregator<STY> as TimeBinnableTypeAggregator>::Input>()
+                    {
+                        // TODO collect statistics associated with this request:
+                        trace_ingest!("{}  FEED THE ITEM...", Self::type_name());
+                        self.agg.ingest(item);
+                        if item.ends_after(self.agg.range()) {
+                            trace_ingest!("{}  FED ITEM, ENDS AFTER.", Self::type_name());
+                            self.cycle();
+                            if self.rng.is_none() {
+                                warn!("{}  no more bin in edges C", Self::type_name());
+                                return;
+                            } else {
+                                trace_ingest!("{}  FED ITEM, CYCLED, CONTINUE.", Self::type_name());
+                            }
+                        } else {
+                            trace_ingest!("{}  FED ITEM.", Self::type_name());
+                            break;
+                        }
+                    } else {
+                        error!("{}::ingest  unexpected item type", Self::type_name());
+                    };
+                }
+            }
+        }
+    }
+
+    fn push_in_progress(&mut self, push_empty: bool) {
+        trace!("{}::push_in_progress  push_empty {push_empty}", Self::type_name());
+        // TODO expand should be derived from AggKind. Is it still required after all?
+        // TODO here, the expand means that agg will assume that the current value is kept constant during
+        // the rest of the time range.
+        if self.rng.is_none() {
+        } else {
+            let expand = true;
+            let range_next = self.next_bin_range();
+            trace!("\n+++++\n+++++\n{}  range_next {:?}", Self::type_name(), range_next);
+            self.rng = range_next.clone();
+            let mut bins = if let Some(range_next) = range_next {
+                self.agg.result_reset(range_next, expand)
+            } else {
+                // Acts as placeholder
+                let range_next = NanoRange {
+                    beg: u64::MAX - 1,
+                    end: u64::MAX,
+                };
+                self.agg.result_reset(range_next.into(), expand)
+            };
+            if bins.len() != 1 {
+                error!("{}::push_in_progress  bins.len() {}", Self::type_name(), bins.len());
+                return;
+            } else {
+                if push_empty || bins.counts[0] != 0 {
+                    match self.ready.as_mut() {
+                        Some(ready) => {
+                            ready.append_all_from(&mut bins);
+                        }
+                        None => {
+                            self.ready = Some(bins);
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    fn cycle(&mut self) {
+        trace!("{}::cycle", Self::type_name());
+        // TODO refactor this logic.
+        let n = self.bins_ready_count();
+        self.push_in_progress(true);
+        if self.bins_ready_count() == n {
+            let range_next = self.next_bin_range();
+            self.rng = range_next.clone();
+            if let Some(range) = range_next {
+                let mut bins = BinsXbinDim0::empty();
+                if range.is_time() {
+                    bins.append_zero(range.beg_u64(), range.end_u64());
+                } else {
+                    error!("TODO  {}::cycle  is_pulse", Self::type_name());
+                }
+                match self.ready.as_mut() {
+                    Some(ready) => {
+                        ready.append_all_from(&mut bins);
+                    }
+                    None => {
+                        self.ready = Some(bins);
+                    }
+                }
+                if self.bins_ready_count() <= n {
+                    error!("failed to push a zero bin");
+                }
+            } else {
+                warn!("cycle: no in-progress bin pushed, but also no more bin to add as zero-bin");
+            }
+        }
+    }
+
+    fn set_range_complete(&mut self) {
+        self.range_final = true;
+    }
+
+    fn empty(&self) -> Box<dyn TimeBinned> {
+        let ret = <EventsXbinDim0Aggregator<STY> as TimeBinnableTypeAggregator>::Output::empty();
+        Box::new(ret)
+    }
+}
+
+impl<STY> TimeBinnableType for EventsXbinDim0<STY>
 where
-    NTY: ScalarOps,
+    STY: ScalarOps,
 {
-    type Output = BinsXbinDim0<NTY>;
+    type Output = BinsXbinDim0<STY>;
-    type Aggregator = EventsXbinDim0Aggregator<NTY>;
+    type Aggregator = EventsXbinDim0Aggregator<STY>;
 
     fn aggregator(range: SeriesRange, x_bin_count: usize, do_time_weight: bool) -> Self::Aggregator {
         let name = any::type_name::<Self>();
@@ -353,79 +594,83 @@ where
     }
 }
 
-impl<NTY> TimeBinnable for EventsXbinDim0<NTY>
+impl<STY> TimeBinnable for EventsXbinDim0<STY>
 where
-    NTY: ScalarOps,
+    STY: ScalarOps,
 {
     fn time_binner_new(
         &self,
         binrange: BinnedRangeEnum,
         do_time_weight: bool,
     ) -> Box<dyn items_0::timebin::TimeBinner> {
-        todo!()
+        let ret = EventsXbinDim0TimeBinner::<STY>::new(binrange, do_time_weight).unwrap();
+        Box::new(ret)
     }
 
     fn to_box_to_json_result(&self) -> Box<dyn ToJsonResult> {
-        todo!()
+        let k = serde_json::to_value(self).unwrap();
+        Box::new(k) as _
     }
 }
 
-pub struct EventsXbinDim0Aggregator<NTY>
+#[derive(Debug)]
+pub struct EventsXbinDim0Aggregator<STY>
 where
-    NTY: ScalarOps,
+    STY: ScalarOps,
 {
     range: SeriesRange,
     count: u64,
-    min: NTY,
+    min: STY,
-    max: NTY,
+    max: STY,
     sumc: u64,
     sum: f32,
     int_ts: u64,
     last_ts: u64,
-    last_avg: Option<f32>,
+    last_vals: Option<(STY, STY, f32)>,
-    last_min: Option<NTY>,
-    last_max: Option<NTY>,
     do_time_weight: bool,
 }
 
-impl<NTY> EventsXbinDim0Aggregator<NTY>
+impl<STY> EventsXbinDim0Aggregator<STY>
 where
-    NTY: ScalarOps,
+    STY: ScalarOps,
 {
+    pub fn type_name() -> &'static str {
+        std::any::type_name::<Self>()
+    }
+
     pub fn new(range: SeriesRange, do_time_weight: bool) -> Self {
+        let int_ts = range.beg_u64();
         Self {
-            int_ts: todo!(),
             range,
             count: 0,
-            min: NTY::zero_b(),
+            min: STY::zero_b(),
-            max: NTY::zero_b(),
+            max: STY::zero_b(),
             sumc: 0,
             sum: 0f32,
+            int_ts,
             last_ts: 0,
-            last_avg: None,
+            last_vals: None,
-            last_min: None,
-            last_max: None,
             do_time_weight,
         }
     }
 
-    fn apply_min_max(&mut self, min: NTY, max: NTY) {
+    fn apply_min_max(&mut self, min: &STY, max: &STY) {
         if self.count == 0 {
-            self.min = min;
+            self.min = min.clone();
-            self.max = max;
+            self.max = max.clone();
         } else {
-            if min < self.min {
+            if *min < self.min {
-                self.min = min;
+                self.min = min.clone();
             }
-            if max > self.max {
+            if *max > self.max {
-                self.max = max;
+                self.max = max.clone();
             }
         }
     }
 
-    fn apply_event_unweight(&mut self, avg: f32, min: NTY, max: NTY) {
+    fn apply_event_unweight(&mut self, avg: f32, min: STY, max: STY) {
         //debug!("apply_event_unweight");
-        self.apply_min_max(min, max);
+        self.apply_min_max(&min, &max);
         let vf = avg;
         if vf.is_nan() {
         } else {
@@ -434,24 +679,31 @@ where
         }
     }
 
-    fn apply_event_time_weight(&mut self, ts: u64) {
+    fn apply_event_time_weight(&mut self, px: u64, pxbeg: u64) {
-        //debug!("apply_event_time_weight");
+        trace_ingest!(
-        /*if let (Some(avg), Some(min), Some(max)) = (self.last_avg, &self.last_min, &self.last_max) {
+            "apply_event_time_weight  px {}  pxbeg {}  count {}",
-            let min2 = min.clone();
+            px,
-            let max2 = max.clone();
+            pxbeg,
-            self.apply_min_max(min2, max2);
+            self.count
-            let w = (ts - self.int_ts) as f32 / self.range.delta() as f32;
+        );
-            if avg.is_nan() {
+        if let Some((min, max, avg)) = self.last_vals.as_ref() {
-            } else {
+            let vf = *avg;
-                self.sum += avg * w;
+            if px > pxbeg {
+                let min = min.clone();
+                let max = max.clone();
+                self.apply_min_max(&min, &max);
             }
-            self.sumc += 1;
+            let w = (px - self.int_ts) as f32 * 1e-9;
-            self.int_ts = ts;
+            if vf.is_nan() {
-        }*/
+            } else {
-        todo!()
+                self.sum += vf * w;
+                self.sumc += 1;
+            }
+            self.int_ts = px;
+        }
     }
 
-    fn ingest_unweight(&mut self, item: &EventsXbinDim0<NTY>) {
+    fn ingest_unweight(&mut self, item: &EventsXbinDim0<STY>) {
         /*for i1 in 0..item.tss.len() {
             let ts = item.tss[i1];
             let avg = item.avgs[i1];
@@ -467,32 +719,39 @@ where
         todo!()
     }
 
-    fn ingest_time_weight(&mut self, item: &EventsXbinDim0<NTY>) {
+    fn ingest_time_weight(&mut self, item: &EventsXbinDim0<STY>) {
-        /*for i1 in 0..item.tss.len() {
+        trace!(
-            let ts = item.tss[i1];
+            "{} ingest_time_weight  range {:?}  int_ts {:?}",
-            let avg = item.avgs[i1];
+            Self::type_name(),
-            let min = item.mins[i1].clone();
+            self.range,
-            let max = item.maxs[i1].clone();
+            self.int_ts
+        );
+        let range_end = self.range.end_u64();
+        for (((&ts, min), max), avg) in item
+            .tss
+            .iter()
+            .zip(item.mins.iter())
+            .zip(item.maxs.iter())
+            .zip(item.avgs.iter())
+        {
             if ts < self.int_ts {
                 self.last_ts = ts;
-                self.last_avg = Some(avg);
+                self.last_vals = Some((min.clone(), max.clone(), avg.clone()));
-                self.last_min = Some(min);
+                //self.events_ignored_count += 1;
-                self.last_max = Some(max);
+            } else if ts >= self.range.end_u64() {
-            } else if ts >= self.range.end {
+                //self.events_ignored_count += 1;
                 return;
             } else {
-                self.apply_event_time_weight(ts);
+                self.apply_event_time_weight(ts, self.range.beg_u64());
                 self.count += 1;
                 self.last_ts = ts;
-                self.last_avg = Some(avg);
+                self.last_vals = Some((min.clone(), max.clone(), avg.clone()));
-                self.last_min = Some(min);
+                //self.events_taken_count += 1;
-                self.last_max = Some(max);
             }
-        }*/
+        }
-        todo!()
     }
 
-    fn result_reset_unweight(&mut self, range: SeriesRange, _expand: bool) -> BinsXbinDim0<NTY> {
+    fn result_reset_unweight(&mut self, range: SeriesRange, _expand: bool) -> BinsXbinDim0<STY> {
         /*let avg = if self.sumc == 0 {
             0f32
         } else {
@@ -517,32 +776,43 @@ where
         todo!()
     }
 
-    fn result_reset_time_weight(&mut self, range: SeriesRange, expand: bool) -> BinsXbinDim0<NTY> {
+    fn result_reset_time_weight(&mut self, range: SeriesRange, expand: bool) -> BinsXbinDim0<STY> {
+        trace!("{} result_reset_time_weight", Self::type_name());
         // TODO check callsite for correct expand status.
-        /*if true || expand {
+        if self.range.is_time() {
-            self.apply_event_time_weight(self.range.end);
+            self.apply_event_time_weight(self.range.end_u64(), self.range.beg_u64());
+        } else {
+            error!("TODO result_reset_time_weight");
+            err::todoval()
         }
-        let avg = {
+        let range_beg = self.range.beg_u64();
-            let sc = self.range.delta() as f32 * 1e-9;
+        let range_end = self.range.end_u64();
-            self.sum / sc
+        let (min, max, avg) = if self.sumc > 0 {
+            let avg = self.sum / (self.range.delta_u64() as f32 * 1e-9);
+            (self.min.clone(), self.max.clone(), avg)
+        } else {
+            let (min, max, avg) = match &self.last_vals {
+                Some((min, max, avg)) => (min.clone(), max.clone(), avg.clone()),
+                None => (STY::zero_b(), STY::zero_b(), 0.),
+            };
+            (min, max, avg)
         };
         let ret = BinsXbinDim0::from_content(
-            [self.range.beg].into(),
+            [range_beg].into(),
-            [self.range.end].into(),
+            [range_end].into(),
             [self.count].into(),
-            [self.min.clone()].into(),
+            [min.clone()].into(),
-            [self.max.clone()].into(),
+            [max.clone()].into(),
             [avg].into(),
         );
-        self.int_ts = range.beg;
+        self.int_ts = range_beg;
         self.range = range;
         self.count = 0;
-        self.min = NTY::zero_b();
+        self.sum = 0.;
-        self.max = NTY::zero_b();
-        self.sum = 0f32;
         self.sumc = 0;
-        ret*/
+        self.min = STY::zero_b();
-        todo!()
+        self.max = STY::zero_b();
+        ret
     }
 }
 
@@ -558,7 +828,7 @@ where
     }
 
     fn ingest(&mut self, item: &Self::Input) {
-        debug!("ingest");
+        debug!("{} ingest", Self::type_name());
         if self.do_time_weight {
             self.ingest_time_weight(item)
         } else {

items_2/src/items_2.rs

@@ -15,6 +15,7 @@ pub mod streams;
 #[cfg(test)]
 pub mod test;
 pub mod testgen;
+pub mod timebin;
 pub mod transform;
 
 use channelevents::ChannelEvents;
@@ -22,12 +23,12 @@ use chrono::DateTime;
 use chrono::TimeZone;
 use chrono::Utc;
 use futures_util::Stream;
+use items_0::overlap::RangeOverlapInfo;
 use items_0::streamitem::Sitemty;
 use items_0::transform::EventTransform;
 use items_0::Empty;
 use items_0::Events;
 use items_0::MergeError;
-use items_0::overlap::RangeOverlapInfo;
 use merger::Mergeable;
 use netpod::range::evrange::SeriesRange;
 use netpod::timeunits::*;

items_2/src/timebin.rs

@@ -0,0 +1 @@
+

netpod/src/netpod.rs

@@ -1863,7 +1863,13 @@ pub trait RetStreamExt: Stream {
 pub struct OnlyFirstError<T> {
     inp: T,
     errored: bool,
-    completed: bool,
+    complete: bool,
+}
+
+impl<T> OnlyFirstError<T> {
+    pub fn type_name() -> &'static str {
+        std::any::type_name::<Self>()
+    }
 }
 
 impl<T, I, E> Stream for OnlyFirstError<T>
@@ -1874,11 +1880,11 @@ where
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;
-        if self.completed {
+        if self.complete {
-            panic!("poll_next on completed");
+            panic!("{} poll_next on complete", Self::type_name())
         }
         if self.errored {
-            self.completed = true;
+            self.complete = true;
             return Ready(None);
         }
         match self.inp.poll_next_unpin(cx) {
@@ -1888,7 +1894,7 @@ where
                 Ready(Some(Err(e)))
             }
             Ready(None) => {
-                self.completed = true;
+                self.complete = true;
                 Ready(None)
             }
             Pending => Pending,
@@ -1904,7 +1910,7 @@ where
         OnlyFirstError {
             inp: self,
             errored: false,
-            completed: false,
+            complete: false,
         }
     }
 }

netpod/src/streamext.rs

@@ -33,7 +33,7 @@ where
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;
         if self.completed {
-            panic!("poll_next on completed");
+            panic!("SCC poll_next on completed");
         } else if self.errored {
             self.completed = true;
             Ready(None)

streams/src/frames/inmem.rs

@@ -47,6 +47,10 @@ impl<T> InMemoryFrameAsyncReadStream<T>
 where
     T: AsyncRead + Unpin,
 {
+    pub fn type_name() -> &'static str {
+        std::any::type_name::<Self>()
+    }
+
     pub fn new(inp: T, bufcap: usize) -> Self {
         Self {
             inp,
@@ -162,7 +166,7 @@ where
         let _spanguard = span.enter();
         loop {
             break if self.complete {
-                panic!("poll_next on complete")
+                panic!("{} poll_next on complete", Self::type_name())
             } else if self.done {
                 self.complete = true;
                 Ready(None)

streams/src/generators.rs

@@ -8,6 +8,7 @@ use items_0::streamitem::Sitemty;
 use items_0::streamitem::StreamItem;
 use items_0::Appendable;
 use items_0::Empty;
+use items_0::WithLen;
 use items_2::channelevents::ChannelEvents;
 use items_2::eventsdim0::EventsDim0;
 use items_2::eventsdim1::EventsDim1;
@@ -243,10 +244,10 @@ impl GenerateF64V00 {
             let mut value = Vec::new();
             let pi = PI;
             for i in 0..21 {
-                let x = ((-pi + (2. * pi / 20.) * i as f64).cos() + 1.) * ampl;
+                let x = ((-pi + (2. * pi / 20.) * i as f64).cos() + 1.1) * ampl;
                 value.push(x);
             }
-            if true {
+            if false {
                 info!(
                     "v01  node {}  made event  ts {}  pulse {}  value {:?}",
                     self.node_ix, ts, pulse, value
@@ -256,6 +257,7 @@ impl GenerateF64V00 {
             ts += self.dts;
         }
         self.ts = ts;
+        info!("generated  len {}", item.len());
         let w = ChannelEvents::Events(Box::new(item) as _);
         let w = sitem_data(w);
         w

streams/src/rangefilter2.rs

@@ -39,14 +39,14 @@ where
     S: Stream<Item = Sitemty<ITY>> + Unpin,
     ITY: Mergeable,
 {
-    const fn selfname() -> &'static str {
+    pub fn type_name() -> &'static str {
-        "RangeFilter2"
+        std::any::type_name::<Self>()
     }
 
     pub fn new(inp: S, range: NanoRange, one_before_range: bool) -> Self {
-        trace!(
+        info!(
-            "{}::new  range: {:?}  one_before_range: {:?}",
+            "----------------------------\n------------------------\n------------------------\n{}::new  range: {:?}  one_before_range: {:?}",
-            Self::selfname(),
+            Self::type_name(),
             range,
             one_before_range
         );
@@ -152,7 +152,11 @@ where
         use Poll::*;
         loop {
             break if self.complete {
-                Ready(Some(Err(Error::with_msg_no_trace("poll_next on complete"))))
+                error!("{} poll_next on complete", Self::type_name());
+                Ready(Some(Err(Error::with_msg_no_trace(format!(
+                    "{} poll_next on complete",
+                    Self::type_name()
+                )))))
             } else if self.done {
                 self.complete = true;
                 Ready(None)
@@ -219,7 +223,7 @@ where
     ITY: Mergeable,
 {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        f.debug_struct(Self::selfname()).finish()
+        f.debug_struct(Self::type_name()).finish()
     }
 }
 
@@ -229,6 +233,6 @@ where
     ITY: Mergeable,
 {
     fn drop(&mut self) {
-        debug!("drop {} {:?}", Self::selfname(), self);
+        debug!("drop {} {:?}", Self::type_name(), self);
     }
 }

streams/src/timebin.rs

@@ -167,6 +167,7 @@ where
                 StreamItem::Stats(item) => Ok(Break(Ready(Ok(StreamItem::Stats(item))))),
             },
             Err(e) => {
+                error!("received error item: {e}");
                 self.done = true;
                 Err(e)
             }
@@ -179,11 +180,11 @@ where
         use ControlFlow::*;
         use Poll::*;
         trace2!("=================   handle_none");
-        let self_range_complete = self.range_final;
+        let self_range_final = self.range_final;
         if let Some(binner) = self.binner.as_mut() {
             trace!("bins ready count before finish {}", binner.bins_ready_count());
             // TODO rework the finish logic
-            if self_range_complete {
+            if self_range_final {
                 binner.set_range_complete();
             }
             binner.push_in_progress(false);
@@ -194,6 +195,7 @@ where
             } else {
                 warn!("bins ready but got nothing");
                 if let Some(bins) = binner.empty() {
+                    self.done_data = true;
                     Ok(Break(Ready(sitem_data(bins))))
                 } else {
                     let e = Error::with_msg_no_trace("bins ready, but nothing, can not even create empty B");
@@ -242,7 +244,7 @@ where
         trace2!("=================   POLL");
         loop {
             break if self.complete {
-                panic!("poll on complete")
+                panic!("TimeBinnedStream poll on complete")
             } else if self.done {
                 self.complete = true;
                 Ready(None)