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

Customize the log out

Browse Source
This commit is contained in:
Dominik Werder
2024-07-03 10:20:39 +02:00
parent 21259e6591
commit 584d977675
15 changed files with 329 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
crates/items_2/src/binnedcollected.rs
View File

@@ -99,6 +99,7 @@ pub struct BinnedCollected {
scalar_type: ScalarType,
shape: Shape,
do_time_weight: bool,
emit_empty_bins: bool,
did_timeout: bool,
range_final: bool,
coll: Option<Box<dyn Collector>>,
@@ -116,6 +117,7 @@ impl BinnedCollected {
scalar_type: ScalarType,
shape: Shape,
do_time_weight: bool,
emit_empty_bins: bool,
//transformer: &dyn Transformer,
deadline: Instant,
inp: Pin<Box<dyn ChannelEventsInput>>,
@@ -126,6 +128,7 @@ impl BinnedCollected {
scalar_type,
shape,
do_time_weight,
emit_empty_bins,
did_timeout: false,
range_final: false,
coll: None,
@@ -143,9 +146,11 @@ impl BinnedCollected {
RangeCompletableItem::Data(k) => match k {
ChannelEvents::Events(mut events) => {
if self.binner.is_none() {
let bb = events
.as_time_binnable_mut()
.time_binner_new(self.binrange.clone(), self.do_time_weight);
let bb = events.as_time_binnable_mut().time_binner_new(
self.binrange.clone(),
self.do_time_weight,
self.emit_empty_bins,
);
self.binner = Some(bb);
}
let binner = self.binner.as_mut().unwrap();

36
crates/items_2/src/binsdim0.rs
View File

@@ -647,8 +647,14 @@ impl<NTY: ScalarOps> TimeBinnableTypeAggregator for BinsDim0Aggregator<NTY> {
}
impl<NTY: ScalarOps> TimeBinnable for BinsDim0<NTY> {
fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Box<dyn TimeBinner> {
fn time_binner_new(
&self,
binrange: BinnedRangeEnum,
do_time_weight: bool,
emit_empty_bins: bool,
) -> Box<dyn TimeBinner> {
// TODO get rid of unwrap
// TODO respect emit_empty_bins
let ret = BinsDim0TimeBinner::<NTY>::new(binrange, do_time_weight).unwrap();
Box::new(ret)
}
@@ -1016,7 +1022,9 @@ fn bins_timebin_fill_empty_00() {
bin_cnt: 5,
});
let do_time_weight = true;
let mut binner = bins.as_time_binnable_ref().time_binner_new(binrange, do_time_weight);
let mut binner = bins
.as_time_binnable_ref()
.time_binner_new(binrange, do_time_weight, false);
binner.ingest(&mut bins);
binner.append_empty_until_end();
let ready = binner.bins_ready();
@@ -1038,7 +1046,9 @@ fn bins_timebin_fill_empty_01() {
bin_cnt: 5,
});
let do_time_weight = true;
let mut binner = bins.as_time_binnable_ref().time_binner_new(binrange, do_time_weight);
let mut binner = bins
.as_time_binnable_ref()
.time_binner_new(binrange, do_time_weight, false);
binner.ingest(&mut bins);
binner.push_in_progress(true);
binner.append_empty_until_end();
@@ -1061,7 +1071,9 @@ fn bins_timebin_push_empty_00() {
bin_cnt: 5,
});
let do_time_weight = true;
let mut binner = bins.as_time_binnable_ref().time_binner_new(binrange, do_time_weight);
let mut binner = bins
.as_time_binnable_ref()
.time_binner_new(binrange, do_time_weight, false);
binner.ingest(&mut bins);
binner.push_in_progress(true);
let ready = binner.bins_ready();
@@ -1083,7 +1095,9 @@ fn bins_timebin_push_empty_01() {
bin_cnt: 5,
});
let do_time_weight = true;
let mut binner = bins.as_time_binnable_ref().time_binner_new(binrange, do_time_weight);
let mut binner = bins
.as_time_binnable_ref()
.time_binner_new(binrange, do_time_weight, false);
binner.ingest(&mut bins);
binner.push_in_progress(true);
binner.push_in_progress(true);
@@ -1109,7 +1123,9 @@ fn bins_timebin_ingest_only_before() {
bin_cnt: 5,
});
let do_time_weight = true;
let mut binner = bins.as_time_binnable_ref().time_binner_new(binrange, do_time_weight);
let mut binner = bins
.as_time_binnable_ref()
.time_binner_new(binrange, do_time_weight, false);
binner.ingest(&mut bins);
binner.push_in_progress(true);
let ready = binner.bins_ready();
@@ -1132,7 +1148,9 @@ fn bins_timebin_ingest_00() {
bin_cnt: 5,
});
let do_time_weight = true;
let mut binner = bins.as_time_binnable_ref().time_binner_new(binrange, do_time_weight);
let mut binner = bins
.as_time_binnable_ref()
.time_binner_new(binrange, do_time_weight, false);
binner.ingest(&mut bins);
binner.push_in_progress(true);
let ready = binner.bins_ready();
@@ -1157,7 +1175,9 @@ fn bins_timebin_ingest_continuous_00() {
bins.push(SEC * 20, SEC * 21, 3, 70, 94, 82.);
//bins.push(SEC * 21, SEC * 22, 5, 71, 93, 86.);
//bins.push(SEC * 23, SEC * 24, 6, 72, 92, 81.);
let mut binner = bins.as_time_binnable_ref().time_binner_new(binrange, do_time_weight);
let mut binner = bins
.as_time_binnable_ref()
.time_binner_new(binrange, do_time_weight, false);
binner.ingest(&mut bins);
//binner.push_in_progress(true);
let ready = binner.bins_ready();

8
crates/items_2/src/binsxbindim0.rs
View File

@@ -607,7 +607,13 @@ impl<NTY: ScalarOps> TimeBinnableTypeAggregator for BinsXbinDim0Aggregator<NTY>
}
impl<NTY: ScalarOps> TimeBinnable for BinsXbinDim0<NTY> {
fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Box<dyn TimeBinner> {
fn time_binner_new(
&self,
binrange: BinnedRangeEnum,
do_time_weight: bool,
emit_empty_bins: bool,
) -> Box<dyn TimeBinner> {
// TODO respect emit_empty_bins
let ret = BinsXbinDim0TimeBinner::<NTY>::new(binrange, do_time_weight);
Box::new(ret)
}

21
crates/items_2/src/channelevents.rs
View File

@@ -820,8 +820,13 @@ impl RangeOverlapInfo for ChannelEvents {
}
impl TimeBinnable for ChannelEvents {
fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Box<dyn TimeBinner> {
let ret = <ChannelEvents as TimeBinnableTy>::time_binner_new(&self, binrange, do_time_weight);
fn time_binner_new(
&self,
binrange: BinnedRangeEnum,
do_time_weight: bool,
emit_empty_bins: bool,
) -> Box<dyn TimeBinner> {
let ret = <ChannelEvents as TimeBinnableTy>::time_binner_new(&self, binrange, do_time_weight, emit_empty_bins);
Box::new(ret)
}
@@ -997,6 +1002,7 @@ pub struct ChannelEventsTimeBinner {
// here we would rather require a simplified current state for binning purpose.
binrange: BinnedRangeEnum,
do_time_weight: bool,
emit_empty_bins: bool,
conn_state: ConnStatus,
binner: Option<Box<dyn TimeBinner>>,
}
@@ -1012,6 +1018,7 @@ impl fmt::Debug for ChannelEventsTimeBinner {
fmt.debug_struct(Self::type_name())
.field("binrange", &self.binrange)
.field("do_time_weight", &self.do_time_weight)
.field("emit_empty_bins", &self.emit_empty_bins)
.field("conn_state", &self.conn_state)
.finish()
}
@@ -1028,7 +1035,7 @@ impl TimeBinnerTy for ChannelEventsTimeBinner {
match item {
ChannelEvents::Events(item) => {
if self.binner.is_none() {
let binner = item.time_binner_new(self.binrange.clone(), self.do_time_weight);
let binner = item.time_binner_new(self.binrange.clone(), self.do_time_weight, self.emit_empty_bins);
self.binner = Some(binner);
}
match self.binner.as_mut() {
@@ -1141,7 +1148,12 @@ impl TimeBinner for ChannelEventsTimeBinner {
impl TimeBinnableTy for ChannelEvents {
type TimeBinner = ChannelEventsTimeBinner;
fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Self::TimeBinner {
fn time_binner_new(
&self,
binrange: BinnedRangeEnum,
do_time_weight: bool,
emit_empty_bins: bool,
) -> Self::TimeBinner {
trace!("TimeBinnableTy for ChannelEvents make ChannelEventsTimeBinner");
// TODO probably wrong?
let (binner, status) = match self {
@@ -1151,6 +1163,7 @@ impl TimeBinnableTy for ChannelEvents {
ChannelEventsTimeBinner {
binrange,
do_time_weight,
emit_empty_bins,
conn_state: status,
binner,
}

34
crates/items_2/src/eventsdim0.rs
View File

@@ -545,7 +545,7 @@ impl<STY: ScalarOps> TimeAggregatorCommonV0Trait for EventsDim0Aggregator<STY> {
&self.range
}
fn common_ingest_unweight_range(&mut self, item: &Self::Input, r: std::ops::Range<usize>) {
fn common_ingest_unweight_range(&mut self, item: &Self::Input, r: core::ops::Range<usize>) {
for (&ts, val) in item.tss.range(r.clone()).zip(item.values.range(r)) {
self.apply_event_unweight(val.clone());
self.count += 1;
@@ -560,7 +560,7 @@ impl<STY: ScalarOps> TimeAggregatorCommonV0Trait for EventsDim0Aggregator<STY> {
self.last_val = Some(item.values[j].clone());
}
fn common_ingest_range(&mut self, item: &Self::Input, r: std::ops::Range<usize>) {
fn common_ingest_range(&mut self, item: &Self::Input, r: core::ops::Range<usize>) {
let beg = self.range.beg_u64();
for (&ts, val) in item.tss.range(r.clone()).zip(item.values.range(r)) {
if ts > beg {
@@ -787,9 +787,14 @@ impl<STY: ScalarOps> TimeBinnableTypeAggregator for EventsDim0Aggregator<STY> {
}
impl<STY: ScalarOps> TimeBinnable for EventsDim0<STY> {
fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Box<dyn TimeBinner> {
fn time_binner_new(
&self,
binrange: BinnedRangeEnum,
do_time_weight: bool,
emit_empty_bins: bool,
) -> Box<dyn TimeBinner> {
// TODO get rid of unwrap
let ret = EventsDim0TimeBinner::<STY>::new(binrange, do_time_weight).unwrap();
let ret = EventsDim0TimeBinner::<STY>::new(binrange, do_time_weight, emit_empty_bins).unwrap();
Box::new(ret)
}
@@ -1027,6 +1032,7 @@ pub struct EventsDim0TimeBinner<STY: ScalarOps> {
agg: EventsDim0Aggregator<STY>,
ready: Option<<EventsDim0Aggregator<STY> as TimeBinnableTypeAggregator>::Output>,
range_final: bool,
emit_empty_bins: bool,
}
impl<STY: ScalarOps> EventsDim0TimeBinner<STY> {
@@ -1034,11 +1040,12 @@ impl<STY: ScalarOps> EventsDim0TimeBinner<STY> {
any::type_name::<Self>()
}
pub fn new(binrange: BinnedRangeEnum, do_time_weight: bool) -> Result<Self, Error> {
pub fn new(binrange: BinnedRangeEnum, do_time_weight: bool, emit_empty_bins: 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 {:?}", Self::type_name(), rng);
let agg = EventsDim0Aggregator::new(rng, do_time_weight);
let ret = Self {
binrange,
@@ -1047,6 +1054,7 @@ impl<STY: ScalarOps> EventsDim0TimeBinner<STY> {
agg,
ready: None,
range_final: false,
emit_empty_bins,
};
Ok(ret)
}
@@ -1121,6 +1129,10 @@ impl<STY: ScalarOps> TimeBinnerCommonV0Trait for EventsDim0TimeBinner<STY> {
}
impl<STY: ScalarOps> TimeBinner for EventsDim0TimeBinner<STY> {
fn ingest(&mut self, item: &mut dyn TimeBinnable) {
TimeBinnerCommonV0Func::ingest(self, item)
}
fn bins_ready_count(&self) -> usize {
TimeBinnerCommonV0Trait::common_bins_ready_count(self)
}
@@ -1132,10 +1144,6 @@ impl<STY: ScalarOps> TimeBinner for EventsDim0TimeBinner<STY> {
}
}
fn ingest(&mut self, item: &mut dyn TimeBinnable) {
TimeBinnerCommonV0Func::ingest(self, item)
}
fn push_in_progress(&mut self, push_empty: bool) {
TimeBinnerCommonV0Func::push_in_progress(self, push_empty)
}
@@ -1308,7 +1316,7 @@ fn binner_00() {
ev1.push(MS * 1200, 3, 1.2f32);
ev1.push(MS * 3200, 3, 3.2f32);
let binrange = BinnedRangeEnum::from_custom(TsNano::from_ns(SEC), 0, 10);
let mut binner = ev1.time_binner_new(binrange, true);
let mut binner = ev1.time_binner_new(binrange, true, false);
binner.ingest(ev1.as_time_binnable_mut());
eprintln!("{:?}", binner);
// TODO add actual asserts
@@ -1322,7 +1330,7 @@ fn binner_01() {
ev1.push(MS * 2100, 3, 2.1);
ev1.push(MS * 2300, 3, 2.3);
let binrange = BinnedRangeEnum::from_custom(TsNano::from_ns(SEC), 0, 10);
let mut binner = ev1.time_binner_new(binrange, true);
let mut binner = ev1.time_binner_new(binrange, true, false);
binner.ingest(ev1.as_time_binnable_mut());
eprintln!("{:?}", binner);
// TODO add actual asserts
@@ -1409,7 +1417,9 @@ fn events_timebin_ingest_continuous_00() {
let mut bins = EventsDim0::<u32>::empty();
bins.push(SEC * 20, 1, 20);
bins.push(SEC * 23, 2, 23);
let mut binner = bins.as_time_binnable_ref().time_binner_new(binrange, do_time_weight);
let mut binner = bins
.as_time_binnable_ref()
.time_binner_new(binrange, do_time_weight, false);
binner.ingest(&mut bins);
//binner.push_in_progress(true);
let ready = binner.bins_ready();

8
crates/items_2/src/eventsdim1.rs
View File

@@ -742,7 +742,13 @@ impl<STY: ScalarOps> TimeBinnableTypeAggregator for EventsDim1Aggregator<STY> {
}
impl<STY: ScalarOps> TimeBinnable for EventsDim1<STY> {
fn time_binner_new(&self, binrange: BinnedRangeEnum, do_time_weight: bool) -> Box<dyn TimeBinner> {
fn time_binner_new(
&self,
binrange: BinnedRangeEnum,
do_time_weight: bool,
emit_empty_bins: bool,
) -> Box<dyn TimeBinner> {
// TODO respect emit_empty_bins
let ret = EventsDim1TimeBinner::<STY>::new(binrange, do_time_weight).unwrap();
Box::new(ret)
}

2
crates/items_2/src/eventsxbindim0.rs
View File

@@ -613,7 +613,9 @@ where
&self,
binrange: BinnedRangeEnum,
do_time_weight: bool,
emit_empty_bins: bool,
) -> Box<dyn items_0::timebin::TimeBinner> {
// TODO respect emit_empty_bins
let ret = EventsXbinDim0TimeBinner::<STY>::new(binrange, do_time_weight).unwrap();
Box::new(ret)
}

17
crates/items_2/src/test.rs
View File

@@ -373,11 +373,13 @@ fn bin_00() {
let binrange = BinnedRangeEnum::covering_range(range.into(), 10).unwrap();
let deadline = Instant::now() + Duration::from_millis(4000);
let do_time_weight = true;
let emit_empty_bins = false;
let res = BinnedCollected::new(
binrange,
ScalarType::F32,
Shape::Scalar,
do_time_weight,
emit_empty_bins,
deadline,
Box::pin(stream),
)
@@ -421,11 +423,13 @@ fn bin_01() {
let stream = Box::pin(stream);
let deadline = Instant::now() + Duration::from_millis(4000);
let do_time_weight = true;
let emit_empty_bins = false;
let res = BinnedCollected::new(
binrange,
ScalarType::F32,
Shape::Scalar,
do_time_weight,
emit_empty_bins,
deadline,
Box::pin(stream),
)
@@ -481,8 +485,17 @@ fn binned_timeout_00() {
let timeout = Duration::from_millis(400);
let deadline = Instant::now() + timeout;
let do_time_weight = true;
let res =
BinnedCollected::new(binrange, ScalarType::F32, Shape::Scalar, do_time_weight, deadline, inp1).await?;
let emit_empty_bins = false;
let res = BinnedCollected::new(
binrange,
ScalarType::F32,
Shape::Scalar,
do_time_weight,
emit_empty_bins,
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);

53
crates/items_2/src/timebin.rs
View File

@@ -12,14 +12,20 @@ use std::ops::Range;
#[allow(unused)]
macro_rules! trace_ingest {
($($arg:tt)*) => {};
($($arg:tt)*) => { trace!($($arg)*); };
($($arg:tt)*) => {
if false {
trace!($($arg)*);
}
};
}
#[allow(unused)]
macro_rules! trace_ingest_item {
($($arg:tt)*) => {};
($($arg:tt)*) => { trace!($($arg)*); };
($($arg:tt)*) => {
if true {
info!($($arg)*);
}
};
}
#[allow(unused)]
@@ -45,23 +51,15 @@ pub trait TimeBinnerCommonV0Trait {
pub struct TimeBinnerCommonV0Func {}
impl TimeBinnerCommonV0Func {
pub fn agg_ingest<B>(binner: &mut B, item: &mut <B as TimeBinnerCommonV0Trait>::Input)
where
B: TimeBinnerCommonV0Trait,
{
//self.agg.ingest(item);
<B as TimeBinnerCommonV0Trait>::common_agg_ingest(binner, item)
}
pub fn ingest<B>(binner: &mut B, item: &mut dyn TimeBinnable)
where
B: TimeBinnerCommonV0Trait,
{
let self_name = B::type_name();
trace_ingest_item!(
"TimeBinner for {} ingest agg.range {:?} item {:?}",
Self::type_name(),
self.agg.range(),
"TimeBinner for {} ingest common_range_current {:?} item {:?}",
self_name,
binner.common_range_current(),
item
);
if item.len() == 0 {
@@ -120,6 +118,14 @@ impl TimeBinnerCommonV0Func {
}
}
fn agg_ingest<B>(binner: &mut B, item: &mut <B as TimeBinnerCommonV0Trait>::Input)
where
B: TimeBinnerCommonV0Trait,
{
//self.agg.ingest(item);
<B as TimeBinnerCommonV0Trait>::common_agg_ingest(binner, item)
}
pub fn push_in_progress<B>(binner: &mut B, push_empty: bool)
where
B: TimeBinnerCommonV0Trait,
@@ -200,20 +206,21 @@ impl ChooseIndicesForTimeBinEvents {
}
pub fn choose_timeweight(beg: u64, end: u64, tss: &VecDeque<u64>) -> (Option<usize>, usize, usize) {
let self_name = "choose_timeweight";
// TODO improve via binary search.
let mut one_before = None;
let mut j = 0;
let mut k = tss.len();
for (i1, &ts) in tss.iter().enumerate() {
if ts >= end {
trace_ingest!("{self_name} ingest {:6} {:20} {:10?} AFTER", i1, ts, val);
trace_ingest!("{self_name} ingest {:6} {:20} AFTER", i1, ts);
// TODO count all the ignored events for stats
k = i1;
break;
} else if ts >= beg {
trace_ingest!("{self_name} ingest {:6} {:20} {:10?} INSIDE", i1, ts, val);
trace_ingest!("{self_name} ingest {:6} {:20} INSIDE", i1, ts);
} else {
trace_ingest!("{self_name} ingest {:6} {:20} {:10?} BEFORE", i1, ts, val);
trace_ingest!("{self_name} ingest {:6} {:20} BEFORE", i1, ts);
one_before = Some(i1);
j = i1 + 1;
}
@@ -223,7 +230,7 @@ impl ChooseIndicesForTimeBinEvents {
}
pub trait TimeAggregatorCommonV0Trait {
type Input: RangeOverlapInfo + ChooseIndicesForTimeBin + 'static;
type Input: WithLen + RangeOverlapInfo + ChooseIndicesForTimeBin + 'static;
type Output: WithLen + Empty + AppendEmptyBin + HasNonemptyFirstBin + 'static;
fn type_name() -> &'static str;
fn common_range_current(&self) -> &SeriesRange;
@@ -240,10 +247,12 @@ impl TimeAggregatorCommonV0Func {
B: TimeAggregatorCommonV0Trait,
{
let self_name = B::type_name();
// TODO
let items_seen = 777;
trace_ingest!(
"{self_name}::ingest_unweight item len {} items_seen {}",
item.len(),
self.items_seen
items_seen
);
let rng = B::common_range_current(binner);
if rng.is_time() {
@@ -265,10 +274,12 @@ impl TimeAggregatorCommonV0Func {
B: TimeAggregatorCommonV0Trait,
{
let self_name = B::type_name();
// TODO
let items_seen = 777;
trace_ingest!(
"{self_name}::ingest_time_weight item len {} items_seen {}",
item.len(),
self.items_seen
items_seen
);
let rng = B::common_range_current(binner);
if rng.is_time() {