daqbuf/daqbuffer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP on pulse id handling

Browse Source
This commit is contained in:
Dominik Werder
2023-03-09 09:14:22 +01:00
parent 412512a43b
commit 3993fc0432
17 changed files with 1033 additions and 636 deletions
Download Patch File Download Diff File Expand all files Collapse all files

252
items_2/src/binnedcollected.rs Normal file
View File

@@ -0,0 +1,252 @@
use crate::channelevents::ChannelEvents;
use crate::empty::empty_events_dyn_ev;
use crate::ChannelEventsInput;
use crate::Error;
use futures_util::Future;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::Collector;
use items_0::collect_s::ToJsonResult;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::streamitem::StreamItem;
use items_0::TimeBinnable;
use items_0::TimeBinner;
use items_0::Transformer;
use netpod::log::*;
use netpod::transform::Transform;
use netpod::BinnedRange;
use netpod::BinnedRangeEnum;
use netpod::ScalarType;
use netpod::Shape;
use std::pin::Pin;
use std::task::Context;
use std::task::Poll;
use std::time::Duration;
use std::time::Instant;
fn flush_binned(
binner: &mut Box<dyn TimeBinner>,
coll: &mut Option<Box<dyn Collector>>,
force: bool,
) -> Result<(), Error> {
trace!("flush_binned bins_ready_count: {}", binner.bins_ready_count());
if force {
if binner.bins_ready_count() == 0 {
debug!("cycle the binner forced");
binner.cycle();
} else {
debug!("bins ready, do not force");
}
}
if binner.bins_ready_count() > 0 {
let ready = binner.bins_ready();
match ready {
Some(mut ready) => {
trace!("binned_collected ready {ready:?}");
if coll.is_none() {
*coll = Some(ready.as_collectable_mut().new_collector());
}
let cl = coll.as_mut().unwrap();
cl.ingest(ready.as_collectable_mut());
Ok(())
}
None => Err(format!("bins_ready_count but no result").into()),
}
} else {
Ok(())
}
}
#[derive(Debug)]
pub struct BinnedCollectedResult {
pub range_final: bool,
pub did_timeout: bool,
pub result: Box<dyn ToJsonResult>,
}
fn _old_binned_collected(
scalar_type: ScalarType,
shape: Shape,
binrange: BinnedRangeEnum,
transformer: &dyn Transformer,
deadline: Instant,
inp: Pin<Box<dyn Stream<Item = Sitemty<ChannelEvents>> + Send>>,
) -> Result<BinnedCollectedResult, Error> {
event!(Level::TRACE, "binned_collected");
let transprops = transformer.query_transform_properties();
// TODO use a trait to allow check of unfinished data [hcn2956jxhwsf]
// TODO implement continue-at [hcn2956jxhwsf]
// TODO maybe TimeBinner should take all ChannelEvents and handle this?
let empty_item = empty_events_dyn_ev(&scalar_type, &shape)?;
let tmp_item = Ok(StreamItem::DataItem(RangeCompletableItem::Data(ChannelEvents::Events(
empty_item,
))));
let empty_stream = futures_util::stream::once(futures_util::future::ready(tmp_item));
let mut stream = empty_stream.chain(inp);
todo!()
}
enum BinnedCollectedState {
Init,
Run,
Done,
}
pub struct BinnedCollected {
state: BinnedCollectedState,
binrange: BinnedRangeEnum,
scalar_type: ScalarType,
shape: Shape,
do_time_weight: bool,
did_timeout: bool,
range_final: bool,
coll: Option<Box<dyn Collector>>,
binner: Option<Box<dyn TimeBinner>>,
inp: Pin<Box<dyn ChannelEventsInput>>,
}
impl BinnedCollected {
fn self_name() -> &'static str {
"BinnedCollected"
}
pub fn new(
binrange: BinnedRangeEnum,
scalar_type: ScalarType,
shape: Shape,
do_time_weight: bool,
//transformer: &dyn Transformer,
deadline: Instant,
inp: Pin<Box<dyn ChannelEventsInput>>,
) -> Self {
Self {
state: BinnedCollectedState::Init,
binrange,
scalar_type,
shape,
do_time_weight,
did_timeout: false,
range_final: false,
coll: None,
binner: None,
inp,
}
}
fn handle_item(&mut self, item: StreamItem<RangeCompletableItem<ChannelEvents>>) -> Result<(), Error> {
match item {
StreamItem::DataItem(k) => match k {
RangeCompletableItem::RangeComplete => {
self.range_final = true;
}
RangeCompletableItem::Data(k) => match k {
ChannelEvents::Events(events) => {
if self.binner.is_none() {
let bb = events
.as_time_binnable()
.time_binner_new(self.binrange.clone(), self.do_time_weight);
self.binner = Some(bb);
}
let binner = self.binner.as_mut().unwrap();
binner.ingest(events.as_time_binnable());
flush_binned(binner, &mut self.coll, false)?;
}
ChannelEvents::Status(item) => {
trace!("{:?}", item);
}
},
},
StreamItem::Log(item) => {
// TODO collect also errors here?
trace!("{:?}", item);
}
StreamItem::Stats(item) => {
// TODO do something with the stats
trace!("{:?}", item);
}
}
Ok(())
}
fn result(&mut self) -> Result<BinnedCollectedResult, Error> {
if let Some(mut binner) = self.binner.take() {
if self.range_final {
trace!("range_final");
binner.set_range_complete();
} else {
debug!("not range_final");
}
if self.did_timeout {
warn!("did_timeout");
} else {
trace!("not did_timeout");
binner.cycle();
}
flush_binned(&mut binner, &mut self.coll, false)?;
if self.coll.is_none() {
debug!("force a bin");
flush_binned(&mut binner, &mut self.coll, true)?;
} else {
trace!("coll is already some");
}
} else {
error!("no binner, should always have one");
}
let result = match self.coll.take() {
Some(mut coll) => {
let res = coll
.result(None, Some(self.binrange.clone()))
.map_err(|e| format!("{e}"))?;
res
}
None => {
error!("binned_collected nothing collected");
return Err(Error::from(format!("binned_collected nothing collected")));
}
};
let ret = BinnedCollectedResult {
range_final: self.range_final,
did_timeout: self.did_timeout,
result,
};
Ok(ret)
}
}
impl Future for BinnedCollected {
type Output = Result<BinnedCollectedResult, Error>;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
let span = span!(Level::INFO, BinnedCollected::self_name());
let _spg = span.enter();
use Poll::*;
loop {
break match &self.state {
BinnedCollectedState::Init => {
self.state = BinnedCollectedState::Run;
continue;
}
BinnedCollectedState::Run => match self.inp.poll_next_unpin(cx) {
Ready(Some(Ok(item))) => match self.handle_item(item) {
Ok(()) => continue,
Err(e) => {
self.state = BinnedCollectedState::Done;
Ready(Err(e))
}
},
Ready(Some(Err(e))) => {
self.state = BinnedCollectedState::Done;
Ready(Err(e.into()))
}
Ready(None) => {
self.state = BinnedCollectedState::Done;
Ready(self.result())
}
Pending => Pending,
},
BinnedCollectedState::Done => Ready(Err(Error::from(format!("already done")))),
};
}
}
}

99
items_2/src/binsdim0.rs
View File

@@ -1,23 +1,39 @@
use crate::{ts_offs_from_abs, ts_offs_from_abs_with_anchor};
use crate::{IsoDateTime, RangeOverlapInfo};
use crate::{TimeBinnable, TimeBinnableType, TimeBinnableTypeAggregator, TimeBinner};
use chrono::{TimeZone, Utc};
use crate::ts_offs_from_abs;
use crate::ts_offs_from_abs_with_anchor;
use crate::IsoDateTime;
use crate::RangeOverlapInfo;
use crate::TimeBinnableType;
use crate::TimeBinnableTypeAggregator;
use chrono::TimeZone;
use chrono::Utc;
use err::Error;
use items_0::collect_s::{Collectable, CollectableType, CollectorType, ToJsonResult};
use items_0::collect_s::Collectable;
use items_0::collect_s::CollectableType;
use items_0::collect_s::CollectorType;
use items_0::collect_s::ToJsonResult;
use items_0::scalar_ops::ScalarOps;
use items_0::AppendEmptyBin;
use items_0::AsAnyMut;
use items_0::AsAnyRef;
use items_0::Empty;
use items_0::TimeBinnable;
use items_0::TimeBinned;
use items_0::TimeBinner;
use items_0::TimeBins;
use items_0::WithLen;
use items_0::{AppendEmptyBin, AsAnyRef};
use items_0::{AsAnyMut, Empty};
use netpod::SeriesRange;
use netpod::log::*;
use netpod::timeunits::SEC;
use netpod::BinnedRange;
use netpod::BinnedRangeEnum;
use netpod::NanoRange;
use netpod::{log::*, BinnedRange};
use num_traits::Zero;
use serde::{Deserialize, Serialize};
use serde::Deserialize;
use serde::Serialize;
use std::any::Any;
use std::collections::VecDeque;
use std::{fmt, mem};
use std::fmt;
use std::mem;
#[allow(unused)]
macro_rules! trace4 {
@@ -166,26 +182,59 @@ impl<NTY> WithLen for BinsDim0<NTY> {
}
impl<NTY> RangeOverlapInfo for BinsDim0<NTY> {
fn ends_before(&self, range: NanoRange) -> bool {
if let Some(&max) = self.ts2s.back() {
max <= range.beg
fn ends_before(&self, range: &SeriesRange) -> bool {
if range.is_time() {
if let Some(&max) = self.ts2s.back() {
max <= range.beg_u64()
} else {
true
}
} else if range.is_pulse() {
if let Some(&max) = self.pulses.back() {
max <= range.beg_u64()
} else {
true
}
} else {
error!("unexpected");
true
}
}
fn ends_after(&self, range: NanoRange) -> bool {
if let Some(&max) = self.ts2s.back() {
max > range.end
fn ends_after(&self, range: &SeriesRange) -> bool {
if range.is_time() {
if let Some(&max) = self.ts2s.back() {
max > range.end_u64()
} else {
true
}
} else if range.is_pulse() {
if let Some(&max) = self.pulses.back() {
max > range.end_u64()
} else {
true
}
} else {
true
error!("unexpected");
false
}
}
fn starts_after(&self, range: NanoRange) -> bool {
if let Some(&min) = self.ts1s.front() {
min >= range.end
fn starts_after(&self, range: &SeriesRange) -> bool {
if range.is_time() {
if let Some(&min) = self.ts1s.front() {
min >= range.end_u64()
} else {
true
}
} else if range.is_pulse() {
if let Some(&min) = self.pulses.front() {
min >= range.end_u64()
} else {
true
}
} else {
error!("unexpected");
true
}
}
@@ -386,7 +435,7 @@ impl<NTY: ScalarOps> CollectorType for BinsDim0Collector<NTY> {
self.timed_out = true;
}
fn result(&mut self, _range: Option<NanoRange>, binrange: Option<BinnedRange>) -> Result<Self::Output, Error> {
fn result(&mut self, _range: Option<NanoRange>, binrange: Option<BinnedRangeEnum>) -> Result<Self::Output, Error> {
let bin_count_exp = if let Some(r) = &binrange {
r.bin_count() as u32
} else {
@@ -517,7 +566,7 @@ where
fn result(
&mut self,
range: Option<NanoRange>,
binrange: Option<BinnedRange>,
binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn items_0::collect_c::Collected>, Error> {
match CollectorType::result(self, range, binrange) {
Ok(res) => Ok(Box::new(res)),
@@ -604,8 +653,8 @@ impl<NTY: ScalarOps> TimeBinnableTypeAggregator for BinsDim0Aggregator<NTY> {
}
impl<NTY: ScalarOps> TimeBinnable for BinsDim0<NTY> {
fn time_binner_new(&self, edges: Vec<u64>, do_time_weight: bool) -> Box<dyn TimeBinner> {
let ret = BinsDim0TimeBinner::<NTY>::new(edges.into(), do_time_weight);
fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Box<dyn TimeBinner> {
let ret = BinsDim0TimeBinner::<NTY>::new(binrange, do_time_weight);
Box::new(ret)
}
@@ -623,7 +672,7 @@ pub struct BinsDim0TimeBinner<NTY: ScalarOps> {
}
impl<NTY: ScalarOps> BinsDim0TimeBinner<NTY> {
fn new(edges: VecDeque<u64>, do_time_weight: bool) -> Self {
fn new(binrange: BinnedRangeEnum, do_time_weight: bool) -> Self {
Self {
edges,
do_time_weight,

40
items_2/src/binsxbindim0.rs
View File

@@ -1,23 +1,37 @@
use crate::{ts_offs_from_abs, ts_offs_from_abs_with_anchor};
use crate::{IsoDateTime, RangeOverlapInfo};
use crate::{TimeBinnable, TimeBinnableType, TimeBinnableTypeAggregator, TimeBinner};
use crate::ts_offs_from_abs;
use crate::ts_offs_from_abs_with_anchor;
use crate::IsoDateTime;
use crate::RangeOverlapInfo;
use crate::TimeBinnableType;
use crate::TimeBinnableTypeAggregator;
use chrono::{TimeZone, Utc};
use err::Error;
use items_0::collect_s::{Collectable, CollectableType, CollectorType, ToJsonResult};
use items_0::collect_s::Collectable;
use items_0::collect_s::CollectableType;
use items_0::collect_s::CollectorType;
use items_0::collect_s::ToJsonResult;
use items_0::scalar_ops::ScalarOps;
use items_0::AppendEmptyBin;
use items_0::AsAnyMut;
use items_0::AsAnyRef;
use items_0::Empty;
use items_0::TimeBinnable;
use items_0::TimeBinned;
use items_0::TimeBinner;
use items_0::TimeBins;
use items_0::WithLen;
use items_0::{AppendEmptyBin, AsAnyMut, AsAnyRef};
use netpod::log::*;
use netpod::timeunits::SEC;
use netpod::BinnedRange;
use netpod::BinnedRangeEnum;
use netpod::NanoRange;
use netpod::{log::*, BinnedRange};
use num_traits::Zero;
use serde::{Deserialize, Serialize};
use serde::Deserialize;
use serde::Serialize;
use std::any::Any;
use std::collections::VecDeque;
use std::{fmt, mem};
use std::fmt;
use std::mem;
#[allow(unused)]
macro_rules! trace4 {
@@ -387,7 +401,7 @@ impl<NTY: ScalarOps> CollectorType for BinsXbinDim0Collector<NTY> {
self.timed_out = true;
}
fn result(&mut self, _range: Option<NanoRange>, binrange: Option<BinnedRange>) -> Result<Self::Output, Error> {
fn result(&mut self, _range: Option<NanoRange>, binrange: Option<BinnedRangeEnum>) -> Result<Self::Output, Error> {
let bin_count_exp = if let Some(r) = &binrange {
r.bin_count() as u32
} else {
@@ -517,7 +531,7 @@ where
fn result(
&mut self,
range: Option<NanoRange>,
binrange: Option<BinnedRange>,
binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn items_0::collect_c::Collected>, Error> {
match CollectorType::result(self, range, binrange) {
Ok(res) => Ok(Box::new(res)),
@@ -604,8 +618,8 @@ impl<NTY: ScalarOps> TimeBinnableTypeAggregator for BinsXbinDim0Aggregator<NTY>
}
impl<NTY: ScalarOps> TimeBinnable for BinsXbinDim0<NTY> {
fn time_binner_new(&self, edges: Vec<u64>, do_time_weight: bool) -> Box<dyn TimeBinner> {
let ret = BinsXbinDim0TimeBinner::<NTY>::new(edges.into(), do_time_weight);
fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Box<dyn TimeBinner> {
let ret = BinsXbinDim0TimeBinner::<NTY>::new(binrange, do_time_weight);
Box::new(ret)
}
@@ -623,7 +637,7 @@ pub struct BinsXbinDim0TimeBinner<NTY: ScalarOps> {
}
impl<NTY: ScalarOps> BinsXbinDim0TimeBinner<NTY> {
fn new(edges: VecDeque<u64>, do_time_weight: bool) -> Self {
fn new(binrange: BinnedRangeEnum, do_time_weight: bool) -> Self {
Self {
edges,
do_time_weight,

7
items_2/src/channelevents.rs
View File

@@ -9,6 +9,7 @@ use items_0::AsAnyMut;
use items_0::AsAnyRef;
use netpod::log::*;
use netpod::BinnedRange;
use netpod::BinnedRangeEnum;
use netpod::NanoRange;
use serde::Deserialize;
use serde::Serialize;
@@ -607,7 +608,7 @@ pub struct ChannelEventsTimeBinner {
edges: Vec<u64>,
do_time_weight: bool,
conn_state: ConnStatus,
binner: Option<Box<dyn crate::TimeBinner>>,
binner: Option<Box<dyn items_0::TimeBinner>>,
}
impl fmt::Debug for ChannelEventsTimeBinner {
@@ -719,7 +720,7 @@ impl AsAnyMut for ChannelEventsCollectorOutput {
}
}
impl crate::ToJsonResult for ChannelEventsCollectorOutput {
impl items_0::collect_s::ToJsonResult for ChannelEventsCollectorOutput {
fn to_json_result(&self) -> Result<Box<dyn items_0::collect_s::ToJsonBytes>, err::Error> {
todo!()
}
@@ -783,7 +784,7 @@ impl items_0::collect_c::Collector for ChannelEventsCollector {
fn result(
&mut self,
range: Option<NanoRange>,
binrange: Option<BinnedRange>,
binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn items_0::collect_c::Collected>, err::Error> {
match self.coll.as_mut() {
Some(coll) => {

54
items_2/src/empty.rs Normal file
View File

@@ -0,0 +1,54 @@
use crate::eventsdim0::EventsDim0;
use crate::eventsdim1::EventsDim1;
use crate::Error;
use items_0::Empty;
use items_0::Events;
use netpod::log::*;
use netpod::ScalarType;
use netpod::Shape;
pub fn empty_events_dyn_ev(scalar_type: &ScalarType, shape: &Shape) -> Result<Box<dyn Events>, Error> {
let ret: Box<dyn Events> = match shape {
Shape::Scalar => {
use ScalarType::*;
type K<T> = EventsDim0<T>;
match scalar_type {
U8 => Box::new(K::<u8>::empty()),
U16 => Box::new(K::<u16>::empty()),
U32 => Box::new(K::<u32>::empty()),
U64 => Box::new(K::<u64>::empty()),
I8 => Box::new(K::<i8>::empty()),
I16 => Box::new(K::<i16>::empty()),
I32 => Box::new(K::<i32>::empty()),
I64 => Box::new(K::<i64>::empty()),
F32 => Box::new(K::<f32>::empty()),
F64 => Box::new(K::<f64>::empty()),
BOOL => Box::new(K::<bool>::empty()),
STRING => Box::new(K::<String>::empty()),
}
}
Shape::Wave(..) => {
use ScalarType::*;
type K<T> = EventsDim1<T>;
match scalar_type {
U8 => Box::new(K::<u8>::empty()),
U16 => Box::new(K::<u16>::empty()),
U32 => Box::new(K::<u32>::empty()),
U64 => Box::new(K::<u64>::empty()),
I8 => Box::new(K::<i8>::empty()),
I16 => Box::new(K::<i16>::empty()),
I32 => Box::new(K::<i32>::empty()),
I64 => Box::new(K::<i64>::empty()),
F32 => Box::new(K::<f32>::empty()),
F64 => Box::new(K::<f64>::empty()),
BOOL => Box::new(K::<bool>::empty()),
STRING => Box::new(K::<String>::empty()),
}
}
Shape::Image(..) => {
error!("TODO empty_events_dyn_ev {scalar_type:?} {shape:?}");
err::todoval()
}
};
Ok(ret)
}

388
items_2/src/eventsdim0.rs
View File

@@ -1,25 +1,27 @@
use crate::binsdim0::BinsDim0;
use crate::IsoDateTime;
use crate::RangeOverlapInfo;
use crate::TimeBinnable;
use crate::TimeBinnableType;
use crate::TimeBinnableTypeAggregator;
use crate::TimeBinner;
use err::Error;
use items_0::Appendable;
use items_0::scalar_ops::ScalarOps;
use items_0::Appendable;
use items_0::AsAnyMut;
use items_0::AsAnyRef;
use items_0::Empty;
use items_0::Events;
use items_0::EventsNonObj;
use items_0::TimeBinnable;
use items_0::TimeBinner;
use items_0::WithLen;
use netpod::log::*;
use netpod::timeunits::SEC;
use netpod::BinnedRange;
use netpod::BinnedRangeEnum;
use netpod::NanoRange;
use netpod::SeriesRange;
use serde::Deserialize;
use serde::Serialize;
use std::any;
use std::any::Any;
use std::collections::VecDeque;
use std::fmt;
@@ -124,26 +126,59 @@ impl<NTY> WithLen for EventsDim0<NTY> {
}
impl<NTY: ScalarOps> RangeOverlapInfo for EventsDim0<NTY> {
fn ends_before(&self, range: NanoRange) -> bool {
if let Some(&max) = self.tss.back() {
max < range.beg
fn ends_before(&self, range: &SeriesRange) -> bool {
if range.is_time() {
if let Some(&max) = self.tss.back() {
max < range.beg_u64()
} else {
true
}
} else if range.is_pulse() {
if let Some(&max) = self.pulses.back() {
max < range.beg_u64()
} else {
true
}
} else {
error!("unexpected");
true
}
}
fn ends_after(&self, range: NanoRange) -> bool {
if let Some(&max) = self.tss.back() {
max >= range.end
fn ends_after(&self, range: &SeriesRange) -> bool {
if range.is_time() {
if let Some(&max) = self.tss.back() {
max >= range.end_u64()
} else {
true
}
} else if range.is_pulse() {
if let Some(&max) = self.pulses.back() {
max >= range.end_u64()
} else {
true
}
} else {
true
error!("unexpected");
false
}
}
fn starts_after(&self, range: NanoRange) -> bool {
if let Some(&min) = self.tss.front() {
min >= range.end
fn starts_after(&self, range: &SeriesRange) -> bool {
if range.is_time() {
if let Some(&min) = self.tss.front() {
min >= range.end_u64()
} else {
true
}
} else if range.is_pulse() {
if let Some(&min) = self.pulses.front() {
min >= range.end_u64()
} else {
true
}
} else {
error!("unexpected");
true
}
}
@@ -156,8 +191,8 @@ where
type Output = BinsDim0<NTY>;
type Aggregator = EventsDim0Aggregator<NTY>;
fn aggregator(range: NanoRange, x_bin_count: usize, do_time_weight: bool) -> Self::Aggregator {
let self_name = std::any::type_name::<Self>();
fn aggregator(range: SeriesRange, x_bin_count: usize, do_time_weight: bool) -> Self::Aggregator {
let self_name = any::type_name::<Self>();
debug!(
"TimeBinnableType for {self_name} aggregator() range {:?} x_bin_count {} do_time_weight {}",
range, x_bin_count, do_time_weight
@@ -334,7 +369,7 @@ impl<NTY: ScalarOps> items_0::collect_s::CollectorType for EventsDim0Collector<N
self.timed_out = true;
}
fn result(&mut self, range: Option<NanoRange>, _binrange: Option<BinnedRange>) -> Result<Self::Output, Error> {
fn result(&mut self, range: Option<NanoRange>, _binrange: Option<BinnedRangeEnum>) -> Result<Self::Output, Error> {
// If we timed out, we want to hint the client from where to continue.
// This is tricky: currently, client can not request a left-exclusive range.
// We currently give the timestamp of the last event plus a small delta.
@@ -418,7 +453,7 @@ impl<NTY: ScalarOps> items_0::collect_c::Collector for EventsDim0Collector<NTY>
fn result(
&mut self,
range: Option<NanoRange>,
binrange: Option<BinnedRange>,
binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn items_0::collect_c::Collected>, err::Error> {
match items_0::collect_s::CollectorType::result(self, range, binrange) {
Ok(x) => Ok(Box::new(x)),
@@ -428,7 +463,7 @@ impl<NTY: ScalarOps> items_0::collect_c::Collector for EventsDim0Collector<NTY>
}
pub struct EventsDim0Aggregator<NTY> {
range: NanoRange,
range: SeriesRange,
count: u64,
min: NTY,
max: NTY,
@@ -455,10 +490,10 @@ impl<NTY> Drop for EventsDim0Aggregator<NTY> {
}
impl<NTY: ScalarOps> EventsDim0Aggregator<NTY> {
pub fn new(range: NanoRange, do_time_weight: bool) -> Self {
let int_ts = range.beg;
pub fn new(binrange: SeriesRange, do_time_weight: bool) -> Self {
let int_ts = binrange.beg_u64();
Self {
range,
range: binrange,
count: 0,
min: NTY::zero_b(),
max: NTY::zero_b(),
@@ -511,15 +546,15 @@ impl<NTY: ScalarOps> EventsDim0Aggregator<NTY> {
}
}
fn apply_event_time_weight(&mut self, ts: u64) {
fn apply_event_time_weight(&mut self, px: u64, pxbeg: u64) {
if let Some(v) = &self.last_seen_val {
let vf = v.as_prim_f32_b();
let v2 = v.clone();
if ts > self.range.beg {
if px > pxbeg {
self.apply_min_max(v2);
}
let w = if self.do_time_weight {
(ts - self.int_ts) as f32 * 1e-9
(px - self.int_ts) as f32 * 1e-9
} else {
1.
};
@@ -528,73 +563,80 @@ impl<NTY: ScalarOps> EventsDim0Aggregator<NTY> {
self.sum += vf * w;
self.sumc += 1;
}
self.int_ts = ts;
self.int_ts = px;
} else {
debug!(
"apply_event_time_weight NO VALUE {}",
ts as i64 - self.range.beg as i64
);
debug!("apply_event_time_weight NO VALUE");
}
}
fn ingest_unweight(&mut self, item: &<Self as TimeBinnableTypeAggregator>::Input) {
trace!("TODO check again result_reset_unweight");
err::todo();
for i1 in 0..item.tss.len() {
let ts = item.tss[i1];
let val = item.values[i1].clone();
if ts < self.range.beg {
self.events_ignored_count += 1;
} else if ts >= self.range.end {
self.events_ignored_count += 1;
return;
} else {
self.apply_event_unweight(val);
self.count += 1;
self.events_taken_count += 1;
if self.range.is_time() {
for i1 in 0..item.tss.len() {
let ts = item.tss[i1];
let val = item.values[i1].clone();
if ts < self.range.beg_u64() {
self.events_ignored_count += 1;
} else if ts >= self.range.end_u64() {
self.events_ignored_count += 1;
return;
} else {
self.apply_event_unweight(val);
self.count += 1;
self.events_taken_count += 1;
}
}
} else {
error!("TODO ingest_unweight");
err::todo();
}
}
fn ingest_time_weight(&mut self, item: &<Self as TimeBinnableTypeAggregator>::Input) {
let self_name = std::any::type_name::<Self>();
let self_name = any::type_name::<Self>();
trace!("{self_name}::ingest_time_weight item len {}", item.len());
for i1 in 0..item.tss.len() {
let ts = item.tss[i1];
let val = item.values[i1].clone();
trace!("{self_name} ingest {:6} {:20} {:10?}", i1, ts, val);
if ts < self.int_ts {
if self.last_seen_val.is_none() {
info!(
"ingest_time_weight event before range, only set last ts {} val {:?}",
ts, val
);
if self.range.is_time() {
for i1 in 0..item.tss.len() {
let ts = item.tss[i1];
let val = item.values[i1].clone();
trace!("{self_name} ingest {:6} {:20} {:10?}", i1, ts, val);
if ts < self.int_ts {
if self.last_seen_val.is_none() {
info!(
"ingest_time_weight event before range, only set last ts {} val {:?}",
ts, val
);
}
self.events_ignored_count += 1;
self.last_seen_ts = ts;
self.last_seen_val = Some(val);
} else if ts >= self.range.end_u64() {
self.events_ignored_count += 1;
return;
} else {
if false && self.last_seen_val.is_none() {
// TODO no longer needed or?
info!(
"call apply_min_max without last val, use current instead {} {:?}",
ts, val
);
self.apply_min_max(val.clone());
}
self.apply_event_time_weight(ts, self.range.beg_u64());
self.count += 1;
self.last_seen_ts = ts;
self.last_seen_val = Some(val);
self.events_taken_count += 1;
}
self.events_ignored_count += 1;
self.last_seen_ts = ts;
self.last_seen_val = Some(val);
} else if ts >= self.range.end {
self.events_ignored_count += 1;
return;
} else {
if false && self.last_seen_val.is_none() {
// TODO no longer needed or?
info!(
"call apply_min_max without last val, use current instead {} {:?}",
ts, val
);
self.apply_min_max(val.clone());
}
self.apply_event_time_weight(ts);
self.count += 1;
self.last_seen_ts = ts;
self.last_seen_val = Some(val);
self.events_taken_count += 1;
}
} else {
error!("TODO ingest_unweight");
err::todo();
}
}
fn result_reset_unweight(&mut self, range: NanoRange, _expand: bool) -> BinsDim0<NTY> {
fn result_reset_unweight(&mut self, range: SeriesRange) -> BinsDim0<NTY> {
trace!("TODO check again result_reset_unweight");
err::todo();
let (min, max, avg) = if self.sumc > 0 {
@@ -607,15 +649,20 @@ impl<NTY: ScalarOps> EventsDim0Aggregator<NTY> {
};
(g.clone(), g.clone(), g.as_prim_f32_b())
};
let ret = BinsDim0 {
ts1s: [self.range.beg].into(),
ts2s: [self.range.end].into(),
counts: [self.count].into(),
mins: [min].into(),
maxs: [max].into(),
avgs: [avg].into(),
let ret = if self.range.is_time() {
BinsDim0 {
ts1s: [self.range.beg_u64()].into(),
ts2s: [self.range.end_u64()].into(),
counts: [self.count].into(),
mins: [min].into(),
maxs: [max].into(),
avgs: [avg].into(),
}
} else {
error!("TODO result_reset_unweight");
err::todoval()
};
self.int_ts = range.beg;
self.int_ts = range.beg_u64();
self.range = range;
self.count = 0;
self.sum = 0f32;
@@ -624,16 +671,17 @@ impl<NTY: ScalarOps> EventsDim0Aggregator<NTY> {
ret
}
fn result_reset_time_weight(&mut self, range: NanoRange, expand: bool) -> BinsDim0<NTY> {
fn result_reset_time_weight(&mut self, range: SeriesRange) -> BinsDim0<NTY> {
// TODO check callsite for correct expand status.
if expand {
debug!("result_reset_time_weight calls apply_event_time_weight");
self.apply_event_time_weight(self.range.end);
debug!("result_reset_time_weight calls apply_event_time_weight");
if self.range.is_time() {
self.apply_event_time_weight(self.range.end_u64(), self.range.beg_u64());
} else {
debug!("result_reset_time_weight NO EXPAND");
error!("TODO result_reset_time_weight");
err::todoval()
}
let (min, max, avg) = if self.sumc > 0 {
let avg = self.sum / (self.range.delta() as f32 * 1e-9);
let avg = self.sum / (self.range.delta_u64() as f32 * 1e-9);
(self.min.clone(), self.max.clone(), avg)
} else {
let g = match &self.last_seen_val {
@@ -642,15 +690,20 @@ impl<NTY: ScalarOps> EventsDim0Aggregator<NTY> {
};
(g.clone(), g.clone(), g.as_prim_f32_b())
};
let ret = BinsDim0 {
ts1s: [self.range.beg].into(),
ts2s: [self.range.end].into(),
counts: [self.count].into(),
mins: [min].into(),
maxs: [max].into(),
avgs: [avg].into(),
let ret = if self.range.is_time() {
BinsDim0 {
ts1s: [self.range.beg_u64()].into(),
ts2s: [self.range.end_u64()].into(),
counts: [self.count].into(),
mins: [min].into(),
maxs: [max].into(),
avgs: [avg].into(),
}
} else {
error!("TODO result_reset_time_weight");
err::todoval()
};
self.int_ts = range.beg;
self.int_ts = range.beg_u64();
self.range = range;
self.count = 0;
self.sum = 0.;
@@ -666,7 +719,7 @@ impl<NTY: ScalarOps> TimeBinnableTypeAggregator for EventsDim0Aggregator<NTY> {
type Input = EventsDim0<NTY>;
type Output = BinsDim0<NTY>;
fn range(&self) -> &NanoRange {
fn range(&self) -> &SeriesRange {
&self.range
}
@@ -686,19 +739,19 @@ impl<NTY: ScalarOps> TimeBinnableTypeAggregator for EventsDim0Aggregator<NTY> {
}
}
fn result_reset(&mut self, range: NanoRange, expand: bool) -> Self::Output {
trace!("result_reset {} {}", range.beg, range.end);
fn result_reset(&mut self, range: SeriesRange, _expand: bool) -> Self::Output {
trace!("result_reset {:?}", range);
if self.do_time_weight {
self.result_reset_time_weight(range, expand)
self.result_reset_time_weight(range)
} else {
self.result_reset_unweight(range, expand)
self.result_reset_unweight(range)
}
}
}
impl<NTY: ScalarOps> TimeBinnable for EventsDim0<NTY> {
fn time_binner_new(&self, edges: Vec<u64>, do_time_weight: bool) -> Box<dyn TimeBinner> {
let ret = EventsDim0TimeBinner::<NTY>::new(edges.into(), do_time_weight).unwrap();
fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Box<dyn TimeBinner> {
let ret = EventsDim0TimeBinner::<NTY>::new(binrange, do_time_weight).unwrap();
Box::new(ret)
}
@@ -873,28 +926,29 @@ impl<STY: ScalarOps> Events for EventsDim0<STY> {
}
pub struct EventsDim0TimeBinner<NTY: ScalarOps> {
edges: VecDeque<u64>,
binrange: BinnedRangeEnum,
rix: usize,
rng: Option<SeriesRange>,
agg: EventsDim0Aggregator<NTY>,
ready: Option<<EventsDim0Aggregator<NTY> as TimeBinnableTypeAggregator>::Output>,
range_complete: bool,
}
impl<NTY: ScalarOps> EventsDim0TimeBinner<NTY> {
fn new(edges: VecDeque<u64>, do_time_weight: bool) -> Result<Self, Error> {
if edges.len() < 2 {
return Err(Error::with_msg_no_trace(format!("need at least 2 edges")));
}
let self_name = std::any::type_name::<Self>();
trace!("{self_name}::new edges {edges:?}");
let agg = EventsDim0Aggregator::new(
NanoRange {
beg: edges[0],
end: edges[1],
},
do_time_weight,
);
fn self_name() -> &'static str {
any::type_name::<Self>()
}
fn new(binrange: BinnedRangeEnum, do_time_weight: bool) -> Result<Self, Error> {
trace!("{}::new binrange {binrange:?}", Self::self_name());
let rng = binrange
.range_at(0)
.ok_or_else(|| Error::with_msg_no_trace("empty binrange"))?;
let agg = EventsDim0Aggregator::new(rng, do_time_weight);
let ret = Self {
edges,
binrange,
rix: 0,
rng: Some(agg.range.clone()),
agg,
ready: None,
range_complete: false,
@@ -902,18 +956,13 @@ impl<NTY: ScalarOps> EventsDim0TimeBinner<NTY> {
Ok(ret)
}
fn next_bin_range(&mut self) -> Option<NanoRange> {
let self_name = std::any::type_name::<Self>();
if self.edges.len() >= 3 {
self.edges.pop_front();
let ret = NanoRange {
beg: self.edges[0],
end: self.edges[1],
};
trace!("{self_name} next_bin_range {} {}", ret.beg, ret.end);
Some(ret)
fn next_bin_range(&mut self) -> Option<SeriesRange> {
let self_name = any::type_name::<Self>();
self.rix += 1;
if let Some(rng) = self.binrange.range_at(self.rix) {
trace!("{self_name} next_bin_range {:?}", rng);
Some(rng)
} else {
self.edges.clear();
trace!("{self_name} next_bin_range None");
None
}
@@ -936,37 +985,29 @@ impl<NTY: ScalarOps> TimeBinner for EventsDim0TimeBinner<NTY> {
}
fn ingest(&mut self, item: &dyn TimeBinnable) {
let self_name = std::any::type_name::<Self>();
trace2!(
"TimeBinner for EventsDim0TimeBinner {:?}\n{:?}\n------------------------------------",
self.edges.iter().take(2).collect::<Vec<_>>(),
item
);
let self_name = any::type_name::<Self>();
trace2!("TimeBinner for {self_name} {:?}", item);
if item.len() == 0 {
// Return already here, RangeOverlapInfo would not give much sense.
return;
}
if self.edges.len() < 2 {
warn!("{self_name} no more bin in edges A");
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().clone()) {
while item.starts_after(self.agg.range()) {
trace!("{self_name} IGNORE ITEM AND CYCLE BECAUSE item.starts_after");
self.cycle();
if self.edges.len() < 2 {
if self.rng.is_none() {
warn!("{self_name} no more bin in edges B");
return;
}
}
if item.ends_before(self.agg.range().clone()) {
trace!("{self_name} IGNORE ITEM BECAUSE ends_before\n------------- -----------");
if item.ends_before(self.agg.range()) {
trace!("{self_name} IGNORE ITEM BECAUSE ends_before");
return;
} else {
if self.edges.len() < 2 {
trace!("{self_name} edge list exhausted");
if self.rng.is_none() {
trace!("{self_name} no more bin in edges D");
return;
} else {
if let Some(item) = item
@@ -977,10 +1018,10 @@ impl<NTY: ScalarOps> TimeBinner for EventsDim0TimeBinner<NTY> {
// TODO collect statistics associated with this request:
trace!("{self_name} FEED THE ITEM...");
self.agg.ingest(item);
if item.ends_after(self.agg.range().clone()) {
if item.ends_after(self.agg.range()) {
trace!("{self_name} FED ITEM, ENDS AFTER.");
self.cycle();
if self.edges.len() < 2 {
if self.rng.is_none() {
warn!("{self_name} no more bin in edges C");
return;
} else {
@@ -991,7 +1032,7 @@ impl<NTY: ScalarOps> TimeBinner for EventsDim0TimeBinner<NTY> {
break;
}
} else {
panic!("{self_name} not correct item type");
error!("{self_name}::ingest unexpected item type");
};
}
}
@@ -999,35 +1040,38 @@ impl<NTY: ScalarOps> TimeBinner for EventsDim0TimeBinner<NTY> {
}
fn push_in_progress(&mut self, push_empty: bool) {
let self_name = std::any::type_name::<Self>();
let self_name = any::type_name::<Self>();
trace!("{self_name}::push_in_progress");
// 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.edges.len() >= 2 {
if self.rng.is_none() {
} else {
let expand = true;
let range_next = if let Some(x) = self.next_bin_range() {
Some(x)
} else {
None
};
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, expand)
self.agg.result_reset(range_next.into(), expand)
};
assert_eq!(bins.len(), 1);
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);
if bins.len() != 1 {
error!("{self_name}::push_in_progress bins.len() {}", 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);
}
}
}
}
@@ -1035,15 +1079,21 @@ impl<NTY: ScalarOps> TimeBinner for EventsDim0TimeBinner<NTY> {
}
fn cycle(&mut self) {
let self_name = std::any::type_name::<Self>();
let self_name = any::type_name::<Self>();
trace!("{self_name}::cycle");
// TODO refactor this logic.
let n = self.bins_ready_count();
self.push_in_progress(true);
if self.bins_ready_count() == n {
if let Some(range) = self.next_bin_range() {
let range_next = self.next_bin_range();
self.rng = range_next.clone();
if let Some(range) = range_next {
let mut bins = BinsDim0::<NTY>::empty();
bins.append_zero(range.beg, range.end);
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);
@@ -1101,7 +1151,7 @@ impl items_0::collect_c::CollectorDyn for EventsDim0CollectorDyn {
fn result(
&mut self,
_range: Option<NanoRange>,
_binrange: Option<BinnedRange>,
_binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn items_0::collect_c::Collected>, err::Error> {
todo!()
}
@@ -1133,7 +1183,7 @@ impl<NTY: ScalarOps> items_0::collect_c::CollectorDyn for EventsDim0Collector<NT
fn result(
&mut self,
range: Option<NanoRange>,
binrange: Option<BinnedRange>,
binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn items_0::collect_c::Collected>, err::Error> {
items_0::collect_s::CollectorType::result(self, range, binrange)
.map(|x| Box::new(x) as _)

19
items_2/src/eventsdim1.rs
View File

@@ -1,10 +1,8 @@
use crate::binsdim0::BinsDim0;
use crate::IsoDateTime;
use crate::RangeOverlapInfo;
use crate::TimeBinnable;
use crate::TimeBinnableType;
use crate::TimeBinnableTypeAggregator;
use crate::TimeBinner;
use err::Error;
use items_0::scalar_ops::ScalarOps;
use items_0::Appendable;
@@ -13,10 +11,13 @@ use items_0::AsAnyRef;
use items_0::Empty;
use items_0::Events;
use items_0::EventsNonObj;
use items_0::TimeBinnable;
use items_0::TimeBinner;
use items_0::WithLen;
use netpod::log::*;
use netpod::timeunits::SEC;
use netpod::BinnedRange;
use netpod::BinnedRangeEnum;
use netpod::NanoRange;
use serde::Deserialize;
use serde::Serialize;
@@ -295,7 +296,7 @@ impl<NTY: ScalarOps> items_0::collect_s::CollectorType for EventsDim1Collector<N
self.timed_out = true;
}
fn result(&mut self, range: Option<NanoRange>, _binrange: Option<BinnedRange>) -> Result<Self::Output, Error> {
fn result(&mut self, range: Option<NanoRange>, _binrange: Option<BinnedRangeEnum>) -> Result<Self::Output, Error> {
// If we timed out, we want to hint the client from where to continue.
// This is tricky: currently, client can not request a left-exclusive range.
// We currently give the timestamp of the last event plus a small delta.
@@ -366,7 +367,7 @@ impl<NTY: ScalarOps> items_0::collect_c::Collector for EventsDim1Collector<NTY>
fn result(
&mut self,
range: Option<NanoRange>,
binrange: Option<BinnedRange>,
binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn items_0::collect_c::Collected>, err::Error> {
match items_0::collect_s::CollectorType::result(self, range, binrange) {
Ok(x) => Ok(Box::new(x)),
@@ -651,8 +652,8 @@ impl<NTY: ScalarOps> TimeBinnableTypeAggregator for EventsDim1Aggregator<NTY> {
}
impl<NTY: ScalarOps> TimeBinnable for EventsDim1<NTY> {
fn time_binner_new(&self, edges: Vec<u64>, do_time_weight: bool) -> Box<dyn TimeBinner> {
let ret = EventsDim1TimeBinner::<NTY>::new(edges.into(), do_time_weight).unwrap();
fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Box<dyn TimeBinner> {
let ret = EventsDim1TimeBinner::<NTY>::new(binrange, do_time_weight).unwrap();
Box::new(ret)
}
@@ -834,7 +835,7 @@ pub struct EventsDim1TimeBinner<NTY: ScalarOps> {
}
impl<NTY: ScalarOps> EventsDim1TimeBinner<NTY> {
fn new(edges: VecDeque<u64>, do_time_weight: bool) -> Result<Self, Error> {
fn new(binrange: BinnedRangeEnum, do_time_weight: bool) -> Result<Self, Error> {
if edges.len() < 2 {
return Err(Error::with_msg_no_trace(format!("need at least 2 edges")));
}
@@ -1055,7 +1056,7 @@ impl items_0::collect_c::CollectorDyn for EventsDim1CollectorDyn {
fn result(
&mut self,
_range: Option<NanoRange>,
_binrange: Option<BinnedRange>,
_binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn items_0::collect_c::Collected>, err::Error> {
todo!()
}
@@ -1087,7 +1088,7 @@ impl<NTY: ScalarOps> items_0::collect_c::CollectorDyn for EventsDim1Collector<NT
fn result(
&mut self,
range: Option<NanoRange>,
binrange: Option<BinnedRange>,
binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn items_0::collect_c::Collected>, err::Error> {
items_0::collect_s::CollectorType::result(self, range, binrange)
.map(|x| Box::new(x) as _)

5
items_2/src/eventsxbindim0.rs
View File

@@ -11,6 +11,7 @@ use items_0::Empty;
use items_0::WithLen;
use netpod::log::*;
use netpod::BinnedRange;
use netpod::BinnedRangeEnum;
use netpod::NanoRange;
use serde::Deserialize;
use serde::Serialize;
@@ -447,7 +448,7 @@ where
self.timed_out = true;
}
fn result(&mut self, range: Option<NanoRange>, _binrange: Option<BinnedRange>) -> Result<Self::Output, Error> {
fn result(&mut self, range: Option<NanoRange>, _binrange: Option<BinnedRangeEnum>) -> Result<Self::Output, Error> {
use std::mem::replace;
let continue_at = if self.timed_out {
if let Some(ts) = self.vals.tss.back() {
@@ -521,7 +522,7 @@ where
fn result(
&mut self,
_range: Option<NanoRange>,
_binrange: Option<BinnedRange>,
_binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn items_0::collect_c::Collected>, Error> {
todo!()
}

246
items_2/src/items_2.rs
View File

@@ -1,6 +1,8 @@
pub mod binnedcollected;
pub mod binsdim0;
pub mod binsxbindim0;
pub mod channelevents;
pub mod empty;
pub mod eventfull;
pub mod eventsdim0;
pub mod eventsdim1;
@@ -23,32 +25,21 @@ use chrono::Utc;
use futures_util::FutureExt;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::Collector;
use items_0::collect_s::ToJsonResult;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::streamitem::StreamItem;
use items_0::Empty;
use items_0::Events;
use items_0::RangeOverlapInfo;
use items_0::TimeBinnable;
use items_0::TimeBinner;
use netpod::log::*;
use netpod::timeunits::*;
use netpod::transform::Transform;
use netpod::AggKind;
use netpod::BinnedRange;
use netpod::NanoRange;
use netpod::ScalarType;
use netpod::SeriesRange;
use netpod::Shape;
use serde::Deserialize;
use serde::Serialize;
use serde::Serializer;
use std::collections::VecDeque;
use std::fmt;
use std::pin::Pin;
use std::time::Duration;
use std::time::Instant;
pub fn bool_is_false(x: &bool) -> bool {
*x == false
@@ -212,241 +203,20 @@ impl crate::merger::Mergeable for Box<dyn Events> {
pub trait TimeBinnableType: Send + Unpin + RangeOverlapInfo + Empty {
type Output: TimeBinnableType;
type Aggregator: TimeBinnableTypeAggregator<Input = Self, Output = Self::Output> + Send + Unpin;
fn aggregator(range: NanoRange, bin_count: usize, do_time_weight: bool) -> Self::Aggregator;
fn aggregator(range: SeriesRange, bin_count: usize, do_time_weight: bool) -> Self::Aggregator;
}
pub trait TimeBinnableTypeAggregator: Send {
type Input: TimeBinnableType;
type Output: TimeBinnableType;
fn range(&self) -> &NanoRange;
fn range(&self) -> &SeriesRange;
fn ingest(&mut self, item: &Self::Input);
fn result_reset(&mut self, range: NanoRange, expand: bool) -> Self::Output;
fn result_reset(&mut self, range: SeriesRange, expand: bool) -> Self::Output;
}
pub fn empty_events_dyn_ev(
scalar_type: &ScalarType,
shape: &Shape,
transform: &Transform,
) -> Result<Box<dyn Events>, Error> {
let ret: Box<dyn Events> = match shape {
Shape::Scalar => match transform {
_ if true => {
use ScalarType::*;
type K<T> = eventsdim0::EventsDim0<T>;
match scalar_type {
U8 => Box::new(K::<u8>::empty()),
U16 => Box::new(K::<u16>::empty()),
U32 => Box::new(K::<u32>::empty()),
U64 => Box::new(K::<u64>::empty()),
I8 => Box::new(K::<i8>::empty()),
I16 => Box::new(K::<i16>::empty()),
I32 => Box::new(K::<i32>::empty()),
I64 => Box::new(K::<i64>::empty()),
F32 => Box::new(K::<f32>::empty()),
F64 => Box::new(K::<f64>::empty()),
BOOL => Box::new(K::<bool>::empty()),
STRING => Box::new(K::<String>::empty()),
}
}
_ if true => Box::new(eventsdim0::EventsDim0::<i64>::empty()),
_ => {
error!("TODO empty_events_dyn_ev {transform:?} {scalar_type:?} {shape:?}");
err::todoval()
}
},
Shape::Wave(..) => match transform {
_ if true => {
use ScalarType::*;
type K<T> = eventsdim1::EventsDim1<T>;
match scalar_type {
U8 => Box::new(K::<u8>::empty()),
U16 => Box::new(K::<u16>::empty()),
U32 => Box::new(K::<u32>::empty()),
U64 => Box::new(K::<u64>::empty()),
I8 => Box::new(K::<i8>::empty()),
I16 => Box::new(K::<i16>::empty()),
I32 => Box::new(K::<i32>::empty()),
I64 => Box::new(K::<i64>::empty()),
F32 => Box::new(K::<f32>::empty()),
F64 => Box::new(K::<f64>::empty()),
BOOL => Box::new(K::<bool>::empty()),
STRING => Box::new(K::<String>::empty()),
}
}
_ if true => Box::new(eventsdim0::EventsDim0::<i64>::empty()),
_ => {
error!("TODO empty_events_dyn_ev {transform:?} {scalar_type:?} {shape:?}");
err::todoval()
}
},
Shape::Image(..) => {
error!("TODO empty_events_dyn_ev {transform:?} {scalar_type:?} {shape:?}");
err::todoval()
}
};
Ok(ret)
}
pub trait ChannelEventsInput: Stream<Item = Sitemty<ChannelEvents>> + items_0::Transformer + Send {}
pub fn empty_binned_dyn_tb(scalar_type: &ScalarType, shape: &Shape, transform: &Transform) -> Box<dyn TimeBinnable> {
error!("TODO empty_binned_dyn_tb");
todo!()
}
fn flush_binned(
binner: &mut Box<dyn TimeBinner>,
coll: &mut Option<Box<dyn Collector>>,
force: bool,
) -> Result<(), Error> {
trace!("flush_binned bins_ready_count: {}", binner.bins_ready_count());
if force {
if binner.bins_ready_count() == 0 {
debug!("cycle the binner forced");
binner.cycle();
} else {
debug!("bins ready, do not force");
}
}
if binner.bins_ready_count() > 0 {
let ready = binner.bins_ready();
match ready {
Some(mut ready) => {
trace!("binned_collected ready {ready:?}");
if coll.is_none() {
*coll = Some(ready.as_collectable_mut().new_collector());
}
let cl = coll.as_mut().unwrap();
cl.ingest(ready.as_collectable_mut());
Ok(())
}
None => Err(format!("bins_ready_count but no result").into()),
}
} else {
Ok(())
}
}
// TODO remove.
// Compare with items_2::test::bin01
pub async fn binned_collected(
scalar_type: ScalarType,
shape: Shape,
agg_kind: AggKind,
binrange: BinnedRange,
timeout: Duration,
inp: Pin<Box<dyn Stream<Item = Sitemty<ChannelEvents>> + Send>>,
) -> Result<Box<dyn ToJsonResult>, Error> {
event!(Level::TRACE, "binned_collected");
let transform = Transform::default_time_binned();
let edges = binrange.edges();
if edges.len() < 2 {
return Err(format!("binned_collected but edges.len() {}", edges.len()).into());
}
let ts_edges_max = *edges.last().unwrap();
let deadline = Instant::now() + timeout;
let mut did_timeout = false;
// TODO use a trait to allow check of unfinished data [hcn2956jxhwsf]
#[allow(unused)]
let bin_count_exp = edges.len().max(2) as u32 - 1;
let do_time_weight = agg_kind.do_time_weighted();
// TODO maybe TimeBinner should take all ChannelEvents and handle this?
let mut did_range_complete = false;
let mut coll = None;
let mut binner = None;
let empty_item = empty_events_dyn_ev(&scalar_type, &shape, &transform)?;
let tmp_item = Ok(StreamItem::DataItem(RangeCompletableItem::Data(ChannelEvents::Events(
empty_item,
))));
let empty_stream = futures_util::stream::once(futures_util::future::ready(tmp_item));
let mut stream = empty_stream.chain(inp);
loop {
let item = futures_util::select! {
k = stream.next().fuse() => {
if let Some(k) = k {
k?
}else {
break;
}
},
_ = tokio::time::sleep_until(deadline.into()).fuse() => {
did_timeout = true;
break;
}
};
match item {
StreamItem::DataItem(k) => match k {
RangeCompletableItem::RangeComplete => {
did_range_complete = true;
}
RangeCompletableItem::Data(k) => match k {
ChannelEvents::Events(events) => {
if events.starts_after(NanoRange {
beg: 0,
end: ts_edges_max,
}) {
} else {
if binner.is_none() {
let bb = events.as_time_binnable().time_binner_new(edges.clone(), do_time_weight);
binner = Some(bb);
}
let binner = binner.as_mut().unwrap();
binner.ingest(events.as_time_binnable());
flush_binned(binner, &mut coll, false)?;
}
}
ChannelEvents::Status(item) => {
trace!("{:?}", item);
}
},
},
StreamItem::Log(item) => {
// TODO collect also errors here?
trace!("{:?}", item);
}
StreamItem::Stats(item) => {
// TODO do something with the stats
trace!("{:?}", item);
}
}
}
if let Some(mut binner) = binner {
if did_range_complete {
trace!("did_range_complete");
binner.set_range_complete();
} else {
debug!("range not complete");
}
if did_timeout {
warn!("timeout");
} else {
trace!("cycle the binner");
binner.cycle();
}
trace!("flush binned");
flush_binned(&mut binner, &mut coll, false)?;
if coll.is_none() {
debug!("force a bin");
flush_binned(&mut binner, &mut coll, true)?;
} else {
trace!("coll is already some");
}
} else {
error!("no binner, should always have one");
}
match coll {
Some(mut coll) => {
let res = coll.result(None, Some(binrange)).map_err(|e| format!("{e}"))?;
tokio::time::sleep(Duration::from_millis(2000)).await;
Ok(res)
}
None => {
error!("binned_collected nothing collected");
let item = empty_binned_dyn_tb(&scalar_type, &shape, &transform);
let ret = item.to_box_to_json_result();
tokio::time::sleep(Duration::from_millis(2000)).await;
Ok(ret)
}
}
}
impl<T> ChannelEventsInput for T where T: Stream<Item = Sitemty<ChannelEvents>> + items_0::Transformer + Send {}
pub fn runfut<T, F>(fut: F) -> Result<T, err::Error>
where

6
items_2/src/merger.rs
View File

@@ -409,3 +409,9 @@ where
}
}
}
impl<T> items_0::Transformer for Merger<T> {
fn query_transform_properties(&self) -> items_0::TransformProperties {
todo!()
}
}

166
items_2/src/streams.rs
View File

@@ -1 +1,167 @@
use futures_util::Future;
use futures_util::FutureExt;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::TransformProperties;
use items_0::Transformer;
use std::collections::VecDeque;
use std::pin::Pin;
use std::task::Context;
use std::task::Poll;
pub struct Enumerate2<T> {
inp: T,
cnt: usize,
}
impl<T> Enumerate2<T> {
pub fn new(inp: T) -> Self
where
T: Transformer,
{
Self { inp, cnt: 0 }
}
}
impl<T> Stream for Enumerate2<T>
where
T: Stream + Unpin,
{
type Item = (usize, <T as Stream>::Item);
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
match self.inp.poll_next_unpin(cx) {
Ready(Some(item)) => {
let i = self.cnt;
self.cnt += 1;
Ready(Some((i, item)))
}
Ready(None) => Ready(None),
Pending => Pending,
}
}
}
impl<T> Transformer for Enumerate2<T> {
fn query_transform_properties(&self) -> TransformProperties {
todo!()
}
}
pub struct Then2<T, F, Fut> {
inp: T,
f: F,
fut: Option<Pin<Box<Fut>>>,
}
impl<T, F, Fut> Then2<T, F, Fut>
where
T: Stream,
F: FnMut(<T as Stream>::Item) -> Fut,
Fut: Future,
{
pub fn new(inp: T, f: F) -> Self {
Self { inp, f, fut: None }
}
fn prepare_fut(&mut self, item: <T as Stream>::Item) {
self.fut = Some(Box::pin((self.f)(item)));
}
}
impl<T, F, Fut> Stream for Then2<T, F, Fut>
where
T: Stream + Unpin,
Fut: Future,
{
type Item = <Fut as Future>::Output;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
loop {
break if let Some(fut) = &mut self.fut {
match fut.poll_unpin(cx) {
Ready(item) => {
self.fut = None;
Ready(Some(item))
}
Pending => Pending,
}
} else {
match self.inp.poll_next_unpin(cx) {
Ready(Some(item)) => {
continue;
}
Ready(None) => Ready(None),
Pending => Pending,
}
};
}
}
}
impl<T, F, Fut> Transformer for Then2<T, F, Fut> {
fn query_transform_properties(&self) -> TransformProperties {
todo!()
}
}
pub trait TransformerExt {
fn enumerate2(self) -> Enumerate2<Self>
where
Self: Transformer + Sized;
fn then2<F, Fut>(self, f: F) -> Then2<Self, F, Fut>
where
Self: Transformer + Stream + Sized,
F: FnMut(<Self as Stream>::Item) -> Fut,
Fut: Future;
}
impl<T> TransformerExt for T {
fn enumerate2(self) -> Enumerate2<Self>
where
Self: Transformer + Sized,
{
Enumerate2::new(self)
}
fn then2<F, Fut>(self, f: F) -> Then2<Self, F, Fut>
where
Self: Transformer + Stream + Sized,
F: FnMut(<Self as Stream>::Item) -> Fut,
Fut: Future,
{
Then2::new(self, f)
}
}
pub struct VecStream<T> {
inp: VecDeque<T>,
}
impl<T> VecStream<T> {
pub fn new(inp: VecDeque<T>) -> Self {
Self { inp }
}
}
impl<T> Stream for VecStream<T> {
type Item = T;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
if let Some(item) = self.inp.pop_front() {
Ready(Some(item))
} else {
Ready(None)
}
}
}
impl<T> Transformer for VecStream<T> {
fn query_transform_properties(&self) -> TransformProperties {
todo!()
}
}

144
items_2/src/test.rs
View File

@@ -1,4 +1,4 @@
use crate::binned_collected;
use crate::binnedcollected::BinnedCollected;
use crate::binsdim0::BinsDim0CollectedResult;
use crate::channelevents::ConnStatus;
use crate::channelevents::ConnStatusEvent;
@@ -6,6 +6,8 @@ use crate::eventsdim0::EventsDim0;
use crate::merger::Mergeable;
use crate::merger::Merger;
use crate::runfut;
use crate::streams::TransformerExt;
use crate::streams::VecStream;
use crate::testgen::make_some_boxed_d0_f32;
use crate::ChannelEvents;
use crate::Error;
@@ -24,10 +26,12 @@ use netpod::log::*;
use netpod::timeunits::*;
use netpod::AggKind;
use netpod::BinnedRange;
use netpod::BinnedRangeEnum;
use netpod::NanoRange;
use netpod::ScalarType;
use netpod::Shape;
use std::time::Duration;
use std::time::Instant;
#[test]
fn items_move_events() {
@@ -326,94 +330,24 @@ fn bin01() {
let inp1 = futures_util::stream::iter(inp1);
let inp1 = Box::pin(inp1);
let inp2 = Box::pin(futures_util::stream::empty()) as _;
let mut stream = crate::merger::Merger::new(vec![inp1, inp2], 32);
let mut coll = None;
let mut binner = None;
let stream = crate::merger::Merger::new(vec![inp1, inp2], 32);
let range = NanoRange {
beg: SEC * 0,
end: SEC * 100,
};
let binrange = BinnedRange::covering_range(range, 10).unwrap();
let edges = binrange.edges();
// TODO implement continue-at [hcn2956jxhwsf]
#[allow(unused)]
let bin_count_exp = (edges.len() - 1) as u32;
let binrange = BinnedRangeEnum::covering_range(range.into(), 10).unwrap();
let deadline = Instant::now() + Duration::from_millis(4000);
let do_time_weight = true;
while let Some(item) = stream.next().await {
let item = item?;
match item {
StreamItem::DataItem(item) => match item {
RangeCompletableItem::RangeComplete => todo!(),
RangeCompletableItem::Data(item) => match item {
ChannelEvents::Events(events) => {
if binner.is_none() {
let bb = events.as_time_binnable().time_binner_new(edges.clone(), do_time_weight);
binner = Some(bb);
}
let binner = binner.as_mut().unwrap();
binner.ingest(events.as_time_binnable());
eprintln!("bins_ready_count: {}", binner.bins_ready_count());
if binner.bins_ready_count() > 0 {
let ready = binner.bins_ready();
match ready {
Some(mut ready) => {
eprintln!("ready {ready:?}");
if coll.is_none() {
coll = Some(ready.as_collectable_mut().new_collector());
}
let cl = coll.as_mut().unwrap();
cl.ingest(ready.as_collectable_mut());
}
None => {
return Err(format!("bins_ready_count but no result").into());
}
}
}
}
ChannelEvents::Status(_) => {
eprintln!("TODO Status");
}
},
},
StreamItem::Log(_) => {
eprintln!("TODO Log");
}
StreamItem::Stats(_) => {
eprintln!("TODO Stats");
}
}
}
if let Some(mut binner) = binner {
binner.cycle();
// TODO merge with the same logic above in the loop.
if binner.bins_ready_count() > 0 {
let ready = binner.bins_ready();
match ready {
Some(mut ready) => {
eprintln!("ready {ready:?}");
if coll.is_none() {
coll = Some(ready.as_collectable_mut().new_collector());
}
let cl = coll.as_mut().unwrap();
cl.ingest(ready.as_collectable_mut());
}
None => {
return Err(format!("bins_ready_count but no result").into());
}
}
}
}
match coll {
Some(mut coll) => {
let res = coll.result(None, Some(binrange.clone())).map_err(|e| format!("{e}"))?;
//let res = res.to_json_result().map_err(|e| format!("{e}"))?;
//let res = res.to_json_bytes().map_err(|e| format!("{e}"))?;
eprintln!("res {res:?}");
}
None => {
panic!();
}
}
let res = BinnedCollected::new(
binrange,
ScalarType::F32,
Shape::Scalar,
do_time_weight,
deadline,
Box::pin(stream),
)
.await?;
// TODO assert
Ok::<_, Error>(())
};
runfut(fut).unwrap();
@@ -448,19 +382,20 @@ fn bin02() {
beg: TSBASE + SEC * 1,
end: TSBASE + SEC * 10,
};
let binrange = BinnedRange::covering_range(range, 9).map_err(|e| format!("{e}"))?;
assert_eq!(binrange.edges().len(), 10);
let binrange = BinnedRangeEnum::covering_range(range.into(), 9).map_err(|e| format!("{e}"))?;
let stream = Box::pin(stream);
let collected = binned_collected(
let deadline = Instant::now() + Duration::from_millis(4000);
let do_time_weight = true;
let res = BinnedCollected::new(
binrange,
ScalarType::F32,
Shape::Scalar,
AggKind::TimeWeightedScalar,
binrange,
Duration::from_millis(2000),
stream,
do_time_weight,
deadline,
Box::pin(stream),
)
.await?;
eprintln!("collected {:?}", collected);
eprintln!("res {:?}", res);
Ok::<_, Error>(())
};
runfut(fut).unwrap();
@@ -488,8 +423,8 @@ fn binned_timeout_01() {
events_vec1.push(Ok(StreamItem::DataItem(RangeCompletableItem::Data(cev))));
}
events_vec1.push(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete)));
let inp1 = events_vec1;
let inp1 = futures_util::stream::iter(inp1).enumerate().then(|(i, k)| async move {
let inp1 = VecStream::new(events_vec1.into_iter().collect());
let inp1 = inp1.enumerate2().then2(|(i, k)| async move {
if i == 5 {
let _ = tokio::time::sleep(Duration::from_millis(10000)).await;
}
@@ -500,21 +435,16 @@ fn binned_timeout_01() {
beg: TSBASE + SEC * 1,
end: TSBASE + SEC * 10,
};
let binrange = BinnedRange::covering_range(range, 9)?;
let binrange = BinnedRangeEnum::covering_range(range.into(), 9)?;
eprintln!("edges1: {:?}", edges);
eprintln!("edges2: {:?}", binrange.edges());
let inp1 = Box::pin(inp1) as _;
//eprintln!("edges2: {:?}", binrange.edges());
let inp1 = Box::pin(inp1);
let timeout = Duration::from_millis(400);
let res = binned_collected(
ScalarType::F32,
Shape::Scalar,
AggKind::TimeWeightedScalar,
binrange,
timeout,
inp1,
)
.await?;
let r2: &BinsDim0CollectedResult<f32> = res.as_any_ref().downcast_ref().expect("res seems wrong type");
let deadline = Instant::now() + Duration::from_millis(4000);
let do_time_weight = true;
let res =
BinnedCollected::new(binrange, ScalarType::F32, Shape::Scalar, do_time_weight, deadline, inp1).await?;
let r2: &BinsDim0CollectedResult<f32> = res.result.as_any_ref().downcast_ref().expect("res seems wrong type");
eprintln!("rs: {r2:?}");
assert_eq!(SEC * r2.ts_anchor_sec(), TSBASE + SEC);
assert_eq!(r2.counts(), &[10, 10, 10]);