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

WIP

Browse Source
This commit is contained in:
Dominik Werder
2024-11-04 12:07:15 +01:00
parent 66538ced79
commit 5543389c1e
9 changed files with 202 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/netpod/src/netpod.rs
+84 -74
View File
@@ -181,7 +181,33 @@ impl CmpZero for usize {
}
}
#[derive(Debug, err::ThisError)]
#[derive(Debug, ThisError)]
#[cstm(name = "Netpod")]
pub enum NetpodError {
UnknownSeriesKind(i64),
BadInt(#[from] std::num::ParseIntError),
ChronoParse(#[from] chrono::ParseError),
HumantimeDurationParse(#[from] humantime::DurationError),
MissingQueryParameters,
MissingSeries,
MissingBackend,
MissingTimerange,
BadTimerange,
NoSeriesNoName,
BadScalarTypeIndex(i64),
UnsupportedDtype(u8),
JsonParse(#[from] serde_json::Error),
BadScalarTypeVariant(String),
BadScalarTypeCaId(u16),
ScalarTypeNotInCa,
MissingScalarType,
MissingShape,
MissingBinningScheme,
BadCacheUsage(String),
TimelikeBinWidthImpossibleForPulseRange,
}
#[derive(Debug, ThisError)]
#[cstm(name = "AsyncChannelError")]
pub enum AsyncChannelError {
Send,
@@ -212,12 +238,12 @@ impl SeriesKind {
}
}
pub fn from_db_i16(x: i16) -> Result<Self, Error> {
pub fn from_db_i16(x: i16) -> Result<Self, NetpodError> {
let ret = match x {
1 => Self::ChannelStatus,
2 => Self::ChannelData,
3 => Self::CaStatus,
_ => return Err(Error::with_msg_no_trace("bad SeriesKind value")),
_ => return Err(NetpodError::UnknownSeriesKind(x as i64)),
};
Ok(ret)
}
@@ -230,12 +256,14 @@ impl Default for SeriesKind {
}
impl FromUrl for SeriesKind {
fn from_url(url: &Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(url: &Url) -> Result<Self, Self::Error> {
let pairs = get_url_query_pairs(url);
Self::from_pairs(&pairs)
}
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
let ret = pairs
.get("seriesKind")
.and_then(|x| match x.as_str() {
@@ -360,7 +388,7 @@ pub trait HasScalarType {
}
impl ScalarType {
pub fn from_dtype_index(ix: u8) -> Result<Self, Error> {
pub fn from_dtype_index(ix: u8) -> Result<Self, NetpodError> {
use ScalarType::*;
let g = match ix {
0 => BOOL,
@@ -377,8 +405,7 @@ impl ScalarType {
12 => F64,
13 => STRING,
15 => Enum,
6 => return Err(Error::with_msg(format!("CHARACTER not supported"))),
_ => return Err(Error::with_msg(format!("unknown dtype code: {:?}", ix))),
_ => return Err(NetpodError::UnsupportedDtype(ix)),
};
Ok(g)
}
@@ -402,7 +429,7 @@ impl ScalarType {
}
}
pub fn from_variant_str(s: &str) -> Result<Self, Error> {
pub fn from_variant_str(s: &str) -> Result<Self, NetpodError> {
use ScalarType::*;
let ret = match s {
"u8" => U8,
@@ -418,12 +445,7 @@ impl ScalarType {
"bool" => BOOL,
"string" => STRING,
"enum" => Enum,
_ => {
return Err(Error::with_msg_no_trace(format!(
"from_bsread_str can not understand bsread {:?}",
s
)))
}
_ => return Err(NetpodError::BadScalarTypeVariant(s.into())),
};
Ok(ret)
}
@@ -447,7 +469,7 @@ impl ScalarType {
}
}
pub fn from_bsread_str(s: &str) -> Result<Self, Error> {
pub fn from_bsread_str(s: &str) -> Result<Self, NetpodError> {
use ScalarType::*;
let ret = match s {
"uint8" => U8,
@@ -465,17 +487,12 @@ impl ScalarType {
"bool" => BOOL,
"string" => STRING,
"enum" => Enum,
_ => {
return Err(Error::with_msg_no_trace(format!(
"from_bsread_str can not understand bsread {:?}",
s
)))
}
_ => return Err(NetpodError::BadScalarTypeVariant(s.into())),
};
Ok(ret)
}
pub fn from_ca_id(k: u16) -> Result<Self, Error> {
pub fn from_ca_id(k: u16) -> Result<Self, NetpodError> {
use ScalarType::*;
let ret = match k {
0 => STRING,
@@ -485,17 +502,12 @@ impl ScalarType {
4 => I8,
5 => I32,
6 => F64,
_ => {
return Err(Error::with_msg_no_trace(format!(
"from_ca_id can not understand {:?}",
k
)))
}
_ => return Err(NetpodError::BadScalarTypeCaId(k)),
};
Ok(ret)
}
pub fn to_ca_id(&self) -> Result<u16, Error> {
pub fn to_ca_id(&self) -> Result<u16, NetpodError> {
use ScalarType::*;
let ret = match self {
I8 => 4,
@@ -505,12 +517,12 @@ impl ScalarType {
F64 => 6,
STRING => 0,
Enum => 3,
_ => return Err(Error::with_msg_no_trace(format!("can not represent {self:?} as CA id"))),
_ => return Err(NetpodError::ScalarTypeNotInCa),
};
Ok(ret)
}
pub fn from_archeng_db_str(s: &str) -> Result<Self, Error> {
pub fn from_archeng_db_str(s: &str) -> Result<Self, NetpodError> {
use ScalarType::*;
let ret = match s {
"I8" => I8,
@@ -519,19 +531,14 @@ impl ScalarType {
"I64" => I64,
"F32" => F32,
"F64" => F64,
_ => {
return Err(Error::with_msg_no_trace(format!(
"from_archeng_db_str can not understand {:?}",
s
)))
}
_ => return Err(NetpodError::BadScalarTypeVariant(s.into())),
};
Ok(ret)
}
pub fn from_scylla_i32(k: i32) -> Result<Self, Error> {
pub fn from_scylla_i32(k: i32) -> Result<Self, NetpodError> {
if k < 0 || k > u8::MAX as i32 {
return Err(Error::with_public_msg_no_trace(format!("bad scalar type index {k}")));
return Err(NetpodError::BadScalarTypeIndex(k as i64));
}
Self::from_dtype_index(k as u8)
}
@@ -583,7 +590,7 @@ impl ScalarType {
self.index() as i32
}
pub fn from_url_str(s: &str) -> Result<Self, Error> {
pub fn from_url_str(s: &str) -> Result<Self, NetpodError> {
let ret = serde_json::from_str(&format!("\"{s}\""))?;
Ok(ret)
}
@@ -1120,17 +1127,16 @@ impl fmt::Display for SfDbChannel {
}
impl FromUrl for SfDbChannel {
fn from_url(url: &Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(url: &Url) -> Result<Self, Self::Error> {
let pairs = get_url_query_pairs(url);
Self::from_pairs(&pairs)
}
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
let ret = SfDbChannel {
backend: pairs
.get("backend")
.ok_or_else(|| Error::with_public_msg_no_trace("missing backend"))?
.into(),
backend: pairs.get("backend").ok_or_else(|| NetpodError::MissingBackend)?.into(),
name: pairs
.get("channelName")
.map(String::from)
@@ -1142,9 +1148,7 @@ impl FromUrl for SfDbChannel {
kind: SeriesKind::from_pairs(pairs)?,
};
if ret.name.is_empty() && ret.series.is_none() {
return Err(Error::with_public_msg_no_trace(format!(
"Missing one of channelName or seriesId parameters."
)));
return Err(NetpodError::NoSeriesNoName);
}
Ok(ret)
}
@@ -1207,21 +1211,20 @@ impl DaqbufSeries {
}
impl FromUrl for DaqbufSeries {
fn from_url(url: &Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(url: &Url) -> Result<Self, Self::Error> {
let pairs = get_url_query_pairs(url);
Self::from_pairs(&pairs)
}
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
let ret = DaqbufSeries {
series: pairs
.get("seriesId")
.ok_or_else(|| Error::with_public_msg_no_trace("missing seriesId"))
.ok_or_else(|| NetpodError::MissingSeries)
.map(|x| x.parse::<u64>())??,
backend: pairs
.get("backend")
.ok_or_else(|| Error::with_public_msg_no_trace("missing backend"))?
.into(),
backend: pairs.get("backend").ok_or_else(|| NetpodError::MissingBackend)?.into(),
name: pairs
.get("channelName")
.map(String::from)
@@ -1593,7 +1596,7 @@ impl Shape {
}
}
pub fn from_url_str(s: &str) -> Result<Self, Error> {
pub fn from_url_str(s: &str) -> Result<Self, NetpodError> {
let ret = serde_json::from_str(s)?;
Ok(ret)
}
@@ -2304,11 +2307,13 @@ impl PreBinnedPatchCoordEnum {
}
impl FromUrl for PreBinnedPatchCoordEnum {
fn from_url(_url: &Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(_url: &Url) -> Result<Self, Self::Error> {
todo!()
}
fn from_pairs(_pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
fn from_pairs(_pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
todo!()
}
}
@@ -2442,7 +2447,7 @@ impl BinnedRange<TsNano> {
}
}
pub fn covering_range_time(range: NanoRange, bin_len_req: DtMs) -> Result<Self, Error> {
pub fn covering_range_time(range: NanoRange, bin_len_req: DtMs) -> Result<Self, NetpodError> {
let opts = <TsNano as Dim0Index>::binned_bin_len_opts();
let bin_len_req = if bin_len_req.ms() < opts[0].ms() {
DtMs::from_ms_u64(opts[0].ms())
@@ -2598,12 +2603,10 @@ impl BinnedRangeEnum {
}
/// Cover at least the given range while selecting the bin width which best fits the requested bin width.
pub fn covering_range_time(range: SeriesRange, bin_len_req: DtMs) -> Result<Self, Error> {
pub fn covering_range_time(range: SeriesRange, bin_len_req: DtMs) -> Result<Self, NetpodError> {
match range {
SeriesRange::TimeRange(k) => Ok(Self::Time(BinnedRange::covering_range_time(k, bin_len_req)?)),
SeriesRange::PulseRange(_) => Err(Error::with_msg_no_trace(format!(
"timelike bin width not possible for a pulse range"
))),
SeriesRange::PulseRange(_) => Err(NetpodError::TimelikeBinWidthImpossibleForPulseRange),
}
}
@@ -3222,11 +3225,13 @@ impl Default for DiskIoTune {
}
impl FromUrl for DiskIoTune {
fn from_url(url: &Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(url: &Url) -> Result<Self, Self::Error> {
Self::from_pairs(&get_url_query_pairs(url))
}
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
let read_sys = pairs
.get("ReadSys")
.map(|x| x.as_str().into())
@@ -3263,7 +3268,7 @@ pub struct ChannelSearchQuery {
}
impl ChannelSearchQuery {
pub fn from_url(url: &Url) -> Result<Self, Error> {
pub fn from_url(url: &Url) -> Result<Self, NetpodError> {
let pairs = get_url_query_pairs(url);
let ret = Self {
backend: pairs.get("backend").map(Into::into),
@@ -3358,9 +3363,10 @@ pub trait HasTimeout {
}
pub trait FromUrl: Sized {
fn from_url(url: &Url) -> Result<Self, Error>;
type Error;
fn from_url(url: &Url) -> Result<Self, Self::Error>;
// TODO put this in separate trait, because some implementors need url path segments to construct.
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error>;
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error>;
}
pub trait AppendToUrl {
@@ -3379,12 +3385,14 @@ impl AppendToUrl for MapQuery {
}
impl FromUrl for MapQuery {
fn from_url(url: &Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(url: &Url) -> Result<Self, Self::Error> {
let pairs = get_url_query_pairs(url);
Self::from_pairs(&pairs)
}
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
Ok(pairs.clone())
}
}
@@ -3434,12 +3442,14 @@ impl HasTimeout for ChannelConfigQuery {
}
impl FromUrl for ChannelConfigQuery {
fn from_url(url: &Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(url: &Url) -> Result<Self, Self::Error> {
let pairs = get_url_query_pairs(url);
Self::from_pairs(&pairs)
}
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
let beg_date = pairs
.get("begDate")
.map(String::from)
crates/netpod/src/query.rs
+25 -23
View File
@@ -10,6 +10,7 @@ use crate::FromUrl;
use crate::HasBackend;
use crate::HasTimeout;
use crate::NanoRange;
use crate::NetpodError;
use crate::PulseRange;
use crate::SfDbChannel;
use crate::ToNanos;
@@ -45,7 +46,7 @@ impl CacheUsage {
}
// Missing query parameter is not an error
pub fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Option<Self>, Error> {
pub fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Option<Self>, NetpodError> {
pairs
.get("cacheUsage")
.map(|k| {
@@ -58,7 +59,7 @@ impl CacheUsage {
} else if k == "v0nocache" {
Ok(Some(CacheUsage::V0NoCache))
} else {
Err(Error::with_msg(format!("unexpected cacheUsage {:?}", k)))?
Err(NetpodError::BadCacheUsage(k.clone()))?
}
})
.unwrap_or(Ok(None))
@@ -109,27 +110,28 @@ pub struct TimeRangeQuery {
range: NanoRange,
}
fn parse_time(v: &str) -> Result<DateTime<Utc>, Error> {
fn parse_time(v: &str) -> Result<DateTime<Utc>, NetpodError> {
if let Ok(x) = v.parse() {
Ok(x)
} else {
if v.ends_with("ago") {
let d = humantime::parse_duration(&v[..v.len() - 3])
.map_err(|_| Error::with_public_msg_no_trace(format!("can not parse {v}")))?;
let d = humantime::parse_duration(&v[..v.len() - 3]).map_err(|_| NetpodError::BadTimerange)?;
Ok(Utc::now() - d)
} else {
Err(Error::with_public_msg_no_trace(format!("can not parse {v}")))
Err(NetpodError::BadTimerange)
}
}
}
impl FromUrl for TimeRangeQuery {
fn from_url(url: &Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(url: &Url) -> Result<Self, Self::Error> {
let pairs = get_url_query_pairs(url);
Self::from_pairs(&pairs)
}
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
if let (Some(beg), Some(end)) = (pairs.get("begDate"), pairs.get("endDate")) {
let ret = Self {
range: NanoRange {
@@ -147,7 +149,7 @@ impl FromUrl for TimeRangeQuery {
};
Ok(ret)
} else {
Err(Error::with_public_msg_no_trace("missing date range"))
Err(NetpodError::MissingTimerange)
}
}
}
@@ -198,12 +200,14 @@ pub struct PulseRangeQuery {
}
impl FromUrl for PulseRangeQuery {
fn from_url(url: &Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(url: &Url) -> Result<Self, Self::Error> {
let pairs = get_url_query_pairs(url);
Self::from_pairs(&pairs)
}
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
if let (Some(beg), Some(end)) = (pairs.get("begPulse"), pairs.get("endPulse")) {
let ret = Self {
range: PulseRange {
@@ -213,7 +217,7 @@ impl FromUrl for PulseRangeQuery {
};
Ok(ret)
} else {
Err(Error::with_public_msg_no_trace("missing pulse range"))
Err(NetpodError::MissingQueryParameters)
}
}
}
@@ -261,7 +265,7 @@ pub fn binning_scheme_append_to_url(agg_kind: &AggKind, url: &mut Url) {
}
// Absent AggKind is not considered an error.
pub fn agg_kind_from_binning_scheme(pairs: &BTreeMap<String, String>) -> Result<Option<AggKind>, Error> {
pub fn agg_kind_from_binning_scheme(pairs: &BTreeMap<String, String>) -> Result<Option<AggKind>, NetpodError> {
let key = "binningScheme";
if let Some(s) = pairs.get(key) {
let ret = if s == "eventBlobs" {
@@ -278,7 +282,7 @@ pub fn agg_kind_from_binning_scheme(pairs: &BTreeMap<String, String>) -> Result<
} else if s == "pulseIdDiff" {
AggKind::PulseIdDiff
} else {
return Err(Error::with_msg("can not extract binningScheme"));
return Err(NetpodError::MissingBinningScheme);
};
Ok(Some(ret))
} else {
@@ -327,18 +331,16 @@ impl HasTimeout for ChannelStateEventsQuery {
}
impl FromUrl for ChannelStateEventsQuery {
fn from_url(url: &Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(url: &Url) -> Result<Self, Self::Error> {
let pairs = get_url_query_pairs(url);
Self::from_pairs(&pairs)
}
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
let beg_date = pairs
.get("begDate")
.ok_or_else(|| Error::with_msg_no_trace("missing begDate"))?;
let end_date = pairs
.get("endDate")
.ok_or_else(|| Error::with_msg_no_trace("missing endDate"))?;
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
let beg_date = pairs.get("begDate").ok_or_else(|| NetpodError::MissingTimerange)?;
let end_date = pairs.get("endDate").ok_or_else(|| NetpodError::MissingTimerange)?;
let ret = Self {
channel: SfDbChannel::from_pairs(&pairs)?,
range: NanoRange {
@@ -347,7 +349,7 @@ impl FromUrl for ChannelStateEventsQuery {
},
};
let self_name = std::any::type_name::<Self>();
info!("{self_name}::from_url {ret:?}");
debug!("{self_name}::from_url {ret:?}");
Ok(ret)
}
}
crates/netpod/src/query/prebinned.rs
+7 -4
View File
@@ -5,11 +5,11 @@ use crate::AggKind;
use crate::AppendToUrl;
use crate::ByteSize;
use crate::FromUrl;
use crate::NetpodError;
use crate::PreBinnedPatchCoordEnum;
use crate::ScalarType;
use crate::SfDbChannel;
use crate::Shape;
use err::Error;
use std::collections::BTreeMap;
use url::Url;
@@ -48,7 +48,7 @@ impl PreBinnedQuery {
}
}
pub fn from_url(url: &Url) -> Result<Self, Error> {
pub fn from_url(url: &Url) -> Result<Self, NetpodError> {
let mut pairs = BTreeMap::new();
for (j, k) in url.query_pairs() {
pairs.insert(j.to_string(), k.to_string());
@@ -56,11 +56,11 @@ impl PreBinnedQuery {
let pairs = pairs;
let scalar_type = pairs
.get("scalarType")
.ok_or_else(|| Error::with_msg("missing scalarType"))
.ok_or_else(|| NetpodError::MissingScalarType)
.map(|x| ScalarType::from_url_str(&x))??;
let shape = pairs
.get("shape")
.ok_or_else(|| Error::with_msg("missing shape"))
.ok_or_else(|| NetpodError::MissingShape)
.map(|x| Shape::from_url_str(&x))??;
let ret = Self {
patch: PreBinnedPatchCoordEnum::from_pairs(&pairs)?,
@@ -119,6 +119,9 @@ impl PreBinnedQuery {
impl AppendToUrl for PreBinnedQuery {
fn append_to_url(&self, url: &mut Url) {
if false {
panic!("remove, not in use");
}
self.patch.append_to_url(url);
self.channel.append_to_url(url);
self.shape.append_to_url(url);
crates/netpod/src/range/evrange.rs
+6 -3
View File
@@ -4,6 +4,7 @@ use crate::timeunits::SEC;
use crate::AppendToUrl;
use crate::Dim0Kind;
use crate::FromUrl;
use crate::NetpodError;
use crate::TsNano;
use chrono::DateTime;
use chrono::TimeZone;
@@ -189,18 +190,20 @@ impl From<PulseRange> for SeriesRange {
}
impl FromUrl for SeriesRange {
fn from_url(url: &url::Url) -> Result<Self, Error> {
type Error = NetpodError;
fn from_url(url: &url::Url) -> Result<Self, Self::Error> {
let pairs = crate::get_url_query_pairs(url);
Self::from_pairs(&pairs)
}
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Error> {
fn from_pairs(pairs: &BTreeMap<String, String>) -> Result<Self, Self::Error> {
let ret = if let Ok(x) = TimeRangeQuery::from_pairs(pairs) {
SeriesRange::TimeRange(x.into())
} else if let Ok(x) = PulseRangeQuery::from_pairs(pairs) {
SeriesRange::PulseRange(x.into())
} else {
return Err(Error::with_public_msg_no_trace("no time range in url"));
return Err(NetpodError::MissingTimerange);
};
Ok(ret)
}