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

Start fetching events

Browse Source
This commit is contained in:
Dominik Werder
2021-12-17 18:23:48 +01:00
parent e17edd0533
commit 09724fd540
13 changed files with 412 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
archapp/src/events.rs
+194 -22
View File
@@ -1,40 +1,157 @@
use crate::parse::PbFileReader;
use crate::EventsItem;
use chrono::{TimeZone, Utc};
use err::Error;
use futures_core::Stream;
use items::Framable;
use items::eventvalues::EventValues;
use items::{Framable, RangeCompletableItem, StreamItem};
use netpod::log::*;
use netpod::query::RawEventsQuery;
use netpod::{ArchiverAppliance, Channel, ChannelInfo, NodeConfigCached, Shape};
use netpod::timeunits::DAY;
use netpod::{ArchiverAppliance, Channel, ChannelInfo, ScalarType, Shape};
use serde_json::Value as JsonValue;
use std::path::PathBuf;
use std::pin::Pin;
use tokio::fs::{read_dir, File};
pub async fn make_event_pipe(
_evq: &RawEventsQuery,
_aa: &ArchiverAppliance,
) -> Result<Pin<Box<dyn Stream<Item = Box<dyn Framable>> + Send>>, Error> {
err::todoval()
struct DataFilename {
year: u32,
month: u32,
}
pub async fn channel_info(channel: &Channel, node_config: &NodeConfigCached) -> Result<ChannelInfo, Error> {
fn parse_data_filename(s: &str) -> Result<DataFilename, Error> {
if !s.ends_with(".pb") {
return Err(Error::with_msg_no_trace("not a .pb file"));
}
if s.len() < 12 {
return Err(Error::with_msg_no_trace("filename too short"));
}
let j = &s[s.len() - 11..];
if &j[0..1] != ":" {
return Err(Error::with_msg_no_trace("no colon"));
}
if &j[5..6] != "_" {
return Err(Error::with_msg_no_trace("no underscore"));
}
let year: u32 = j[1..5].parse()?;
let month: u32 = j[6..8].parse()?;
let ret = DataFilename { year, month };
Ok(ret)
}
pub async fn make_event_pipe(
evq: &RawEventsQuery,
aa: &ArchiverAppliance,
) -> Result<Pin<Box<dyn Stream<Item = Box<dyn Framable>> + Send>>, Error> {
info!("make_event_pipe {:?}", evq);
let evq = evq.clone();
let DirAndPrefix { dir, prefix } = directory_for_channel_files(&evq.channel, aa)?;
let channel_info = channel_info(&evq.channel, aa).await?;
//let dtbeg = Utc.timestamp((evq.range.beg / 1000000000) as i64, (evq.range.beg % 1000000000) as u32);
let (tx, rx) = async_channel::bounded(16);
let block1 = async move {
let mut rd = tokio::fs::read_dir(&dir).await?;
while let Some(de) = rd.next_entry().await? {
let s = de.file_name().to_string_lossy().into_owned();
if s.starts_with(&prefix) && s.ends_with(".pb") {
match parse_data_filename(&s) {
Ok(df) => {
let ts0 = Utc.ymd(df.year as i32, df.month, 0).and_hms(0, 0, 0);
let ts1 = ts0.timestamp() as u64 * 1000000000 + ts0.timestamp_subsec_nanos() as u64;
if evq.range.beg < ts1 + DAY * 32 && evq.range.end > ts1 {
let f1 = File::open(de.path()).await?;
info!("opened {:?}", de.path());
let mut pbr = PbFileReader::new(f1).await;
pbr.read_header().await?;
loop {
match pbr.read_msg().await {
Ok(ev) => match ev {
EventsItem::ScalarDouble(h) => {
//
let (x, y) = h
.tss
.into_iter()
.zip(h.values.into_iter())
.filter_map(|(j, k)| {
if j < evq.range.beg || j >= evq.range.end {
None
} else {
Some((j, k))
}
})
.fold((vec![], vec![]), |(mut a, mut b), (j, k)| {
a.push(j);
b.push(k);
(a, b)
});
let b = EventValues { tss: x, values: y };
let b = Ok(StreamItem::DataItem(RangeCompletableItem::Data(b)));
tx.send(Box::new(b) as Box<dyn Framable>).await?;
}
_ => {
//
error!("case not covered");
return Err(Error::with_msg_no_trace("todo"));
}
},
Err(e) => {}
}
}
}
}
Err(e) => {}
}
}
}
Ok::<_, Error>(())
};
let block2 = async move {
match block1.await {
Ok(_) => {
info!("block1 done ok");
}
Err(e) => {
error!("{:?}", e);
}
}
};
tokio::task::spawn(block2);
Ok(Box::pin(rx))
}
struct DirAndPrefix {
dir: PathBuf,
prefix: String,
}
fn directory_for_channel_files(channel: &Channel, aa: &ArchiverAppliance) -> Result<DirAndPrefix, Error> {
// SARUN11/CVME/DBLM546/IOC_CPU_LOAD
// SARUN11-CVME-DBLM546:IOC_CPU_LOAD
let a: Vec<_> = channel.name.split("-").map(|s| s.split(":")).flatten().collect();
let path1 = node_config
.node
.archiver_appliance
.as_ref()
.unwrap()
.data_base_path
.clone();
let path2 = a.iter().take(a.len() - 1).fold(path1, |a, &x| a.join(x));
let path = aa.data_base_path.clone();
let path = a.iter().take(a.len() - 1).fold(path, |a, &x| a.join(x));
let ret = DirAndPrefix {
dir: path,
prefix: a
.last()
.ok_or_else(|| Error::with_msg_no_trace("no prefix in file"))?
.to_string(),
};
Ok(ret)
}
pub async fn channel_info(channel: &Channel, aa: &ArchiverAppliance) -> Result<ChannelInfo, Error> {
let DirAndPrefix { dir, prefix } = directory_for_channel_files(channel, aa)?;
let mut msgs = vec![];
msgs.push(format!("a: {:?}", a));
msgs.push(format!("path2: {}", path2.to_string_lossy()));
let mut rd = tokio::fs::read_dir(&path2).await?;
msgs.push(format!("path: {}", dir.to_string_lossy()));
let mut scalar_type = None;
let mut shape = None;
let mut rd = read_dir(&dir).await?;
while let Some(de) = rd.next_entry().await? {
let s = de.file_name().to_string_lossy().into_owned();
if s.starts_with(a.last().unwrap()) && s.ends_with(".pb") {
if s.starts_with(&prefix) && s.ends_with(".pb") {
msgs.push(s);
let f1 = tokio::fs::File::open(de.path()).await?;
let f1 = File::open(de.path()).await?;
let mut pbr = PbFileReader::new(f1).await;
pbr.read_header().await?;
msgs.push(format!("got header {}", pbr.channel_name()));
@@ -42,6 +159,57 @@ pub async fn channel_info(channel: &Channel, node_config: &NodeConfigCached) ->
match ev {
Ok(item) => {
msgs.push(format!("got event {:?}", item));
shape = Some(match &item {
EventsItem::ScalarByte(_) => Shape::Scalar,
EventsItem::ScalarShort(_) => Shape::Scalar,
EventsItem::ScalarInt(_) => Shape::Scalar,
EventsItem::ScalarFloat(_) => Shape::Scalar,
EventsItem::ScalarDouble(_) => Shape::Scalar,
EventsItem::WaveByte(item) => Shape::Wave(
item.vals
.first()
.ok_or_else(|| Error::with_msg_no_trace("empty event batch"))?
.len() as u32,
),
EventsItem::WaveShort(item) => Shape::Wave(
item.vals
.first()
.ok_or_else(|| Error::with_msg_no_trace("empty event batch"))?
.len() as u32,
),
EventsItem::WaveInt(item) => Shape::Wave(
item.vals
.first()
.ok_or_else(|| Error::with_msg_no_trace("empty event batch"))?
.len() as u32,
),
EventsItem::WaveFloat(item) => Shape::Wave(
item.vals
.first()
.ok_or_else(|| Error::with_msg_no_trace("empty event batch"))?
.len() as u32,
),
EventsItem::WaveDouble(item) => Shape::Wave(
item.vals
.first()
.ok_or_else(|| Error::with_msg_no_trace("empty event batch"))?
.len() as u32,
),
});
// These type mappings are defined by the protobuffer schema.
scalar_type = Some(match item {
EventsItem::ScalarByte(_) => ScalarType::U8,
EventsItem::ScalarShort(_) => ScalarType::I32,
EventsItem::ScalarInt(_) => ScalarType::I32,
EventsItem::ScalarFloat(_) => ScalarType::F32,
EventsItem::ScalarDouble(_) => ScalarType::F64,
EventsItem::WaveByte(_) => ScalarType::U8,
EventsItem::WaveShort(_) => ScalarType::I32,
EventsItem::WaveInt(_) => ScalarType::I32,
EventsItem::WaveFloat(_) => ScalarType::F32,
EventsItem::WaveDouble(_) => ScalarType::F64,
});
break;
}
Err(e) => {
msgs.push(format!("can not read event! {:?}", e));
@@ -50,8 +218,12 @@ pub async fn channel_info(channel: &Channel, node_config: &NodeConfigCached) ->
msgs.push(format!("got header {}", pbr.channel_name()));
}
}
let shape = shape.ok_or_else(|| Error::with_msg("could not determine shape"))?;
let scalar_type = scalar_type.ok_or_else(|| Error::with_msg("could not determine scalar_type"))?;
let ret = ChannelInfo {
shape: Shape::Scalar,
scalar_type,
byte_order: None,
shape,
msg: JsonValue::Array(msgs.into_iter().map(JsonValue::String).collect()),
};
Ok(ret)
archapp/src/parse.rs
+53 -25
View File
@@ -2,11 +2,12 @@ use crate::generated::EPICSEvent::PayloadType;
use crate::{unescape_archapp_msg, EventsItem};
use archapp_xc::*;
use async_channel::{bounded, Receiver};
use chrono::{TimeZone, Utc};
use err::Error;
use items::eventvalues::EventValues;
use items::waveevents::WaveEvents;
use netpod::log::*;
use netpod::NodeConfigCached;
use netpod::{ArchiverAppliance, ChannelConfigQuery, ChannelConfigResponse, NodeConfigCached};
use protobuf::Message;
use serde::Serialize;
use serde_json::Value as JsonValue;
@@ -25,16 +26,32 @@ pub struct PbFileReader {
year: u32,
}
fn parse_scalar_byte(m: &[u8], year: u32) -> Result<EventsItem, Error> {
let msg = crate::generated::EPICSEvent::ScalarByte::parse_from_bytes(m)
.map_err(|_| Error::with_msg(format!("can not parse pb-type {}", "ScalarByte")))?;
let mut t = EventValues::<i32> {
tss: vec![],
values: vec![],
};
let yd = Utc.ymd(year as i32, 1, 1).and_hms(0, 0, 0);
let ts = yd.timestamp() as u64 * 1000000000 + msg.get_secondsintoyear() as u64 * 1000000000 + msg.get_nano() as u64;
let v = msg.get_val().first().map_or(0, |k| *k as i32);
t.tss.push(ts);
t.values.push(v);
Ok(EventsItem::ScalarByte(t))
}
macro_rules! scalar_parse {
($m:expr, $pbt:ident, $eit:ident, $evty:ident) => {{
($m:expr, $year:expr, $pbt:ident, $eit:ident, $evty:ident) => {{
let msg = crate::generated::EPICSEvent::$pbt::parse_from_bytes($m)
.map_err(|_| Error::with_msg(format!("can not parse pb-type {}", stringify!($pbt))))?;
let mut t = EventValues::<$evty> {
tss: vec![],
values: vec![],
};
// TODO Translate by the file-time-offset.
let ts = msg.get_secondsintoyear() as u64;
let yd = Utc.ymd($year as i32, 1, 1).and_hms(0, 0, 0);
let ts =
yd.timestamp() as u64 * 1000000000 + msg.get_secondsintoyear() as u64 * 1000000000 + msg.get_nano() as u64;
let v = msg.get_val();
t.tss.push(ts);
t.values.push(v);
@@ -43,15 +60,16 @@ macro_rules! scalar_parse {
}
macro_rules! wave_parse {
($m:expr, $pbt:ident, $eit:ident, $evty:ident) => {{
($m:expr, $year:expr, $pbt:ident, $eit:ident, $evty:ident) => {{
let msg = crate::generated::EPICSEvent::$pbt::parse_from_bytes($m)
.map_err(|_| Error::with_msg(format!("can not parse pb-type {}", stringify!($pbt))))?;
let mut t = WaveEvents::<$evty> {
tss: vec![],
vals: vec![],
};
// TODO Translate by the file-time-offset.
let ts = msg.get_secondsintoyear() as u64;
let yd = Utc.ymd($year as i32, 1, 1).and_hms(0, 0, 0);
let ts =
yd.timestamp() as u64 * 1000000000 + msg.get_secondsintoyear() as u64 * 1000000000 + msg.get_nano() as u64;
let v = msg.get_val();
t.tss.push(ts);
t.vals.push(v.to_vec());
@@ -59,11 +77,13 @@ macro_rules! wave_parse {
}};
}
const MIN_BUF_FILL: usize = 1024 * 16;
impl PbFileReader {
pub async fn new(file: File) -> Self {
Self {
file,
buf: vec![0; 1024 * 128],
buf: vec![0; MIN_BUF_FILL * 4],
wp: 0,
rp: 0,
channel_name: String::new(),
@@ -93,42 +113,39 @@ impl PbFileReader {
let m = unescape_archapp_msg(&buf[self.rp..k])?;
use PayloadType::*;
let ei = match self.payload_type {
SCALAR_BYTE => {
//scalar_parse!(&m, ScalarByte, ScalarByte, u8)
err::todoval()
}
SCALAR_BYTE => parse_scalar_byte(&m, self.year)?,
SCALAR_ENUM => {
scalar_parse!(&m, ScalarEnum, ScalarInt, i32)
scalar_parse!(&m, self.year, ScalarEnum, ScalarInt, i32)
}
SCALAR_SHORT => {
scalar_parse!(&m, ScalarShort, ScalarShort, i32)
scalar_parse!(&m, self.year, ScalarShort, ScalarShort, i32)
}
SCALAR_INT => {
scalar_parse!(&m, ScalarInt, ScalarInt, i32)
scalar_parse!(&m, self.year, ScalarInt, ScalarInt, i32)
}
SCALAR_FLOAT => {
scalar_parse!(&m, ScalarFloat, ScalarFloat, f32)
scalar_parse!(&m, self.year, ScalarFloat, ScalarFloat, f32)
}
SCALAR_DOUBLE => {
scalar_parse!(&m, ScalarDouble, ScalarDouble, f64)
scalar_parse!(&m, self.year, ScalarDouble, ScalarDouble, f64)
}
WAVEFORM_BYTE => {
wave_parse!(&m, VectorChar, WaveByte, u8)
wave_parse!(&m, self.year, VectorChar, WaveByte, u8)
}
WAVEFORM_SHORT => {
wave_parse!(&m, VectorShort, WaveShort, i32)
wave_parse!(&m, self.year, VectorShort, WaveShort, i32)
}
WAVEFORM_ENUM => {
wave_parse!(&m, VectorEnum, WaveInt, i32)
wave_parse!(&m, self.year, VectorEnum, WaveInt, i32)
}
WAVEFORM_INT => {
wave_parse!(&m, VectorInt, WaveInt, i32)
wave_parse!(&m, self.year, VectorInt, WaveInt, i32)
}
WAVEFORM_FLOAT => {
wave_parse!(&m, VectorFloat, WaveFloat, f32)
wave_parse!(&m, self.year, VectorFloat, WaveFloat, f32)
}
WAVEFORM_DOUBLE => {
wave_parse!(&m, VectorDouble, WaveDouble, f64)
wave_parse!(&m, self.year, VectorDouble, WaveDouble, f64)
}
SCALAR_STRING | WAVEFORM_STRING | V4_GENERIC_BYTES => {
return Err(Error::with_msg(format!("not supported: {:?}", self.payload_type)));
@@ -139,10 +156,10 @@ impl PbFileReader {
}
async fn fill_buf(&mut self) -> Result<(), Error> {
if self.wp - self.rp >= 1024 * 16 {
if self.wp - self.rp >= MIN_BUF_FILL {
return Ok(());
}
if self.rp >= 1024 * 42 {
if self.rp >= self.buf.len() - MIN_BUF_FILL {
let n = self.wp - self.rp;
for i in 0..n {
self.buf[i] = self.buf[self.rp + i];
@@ -397,3 +414,14 @@ pub async fn scan_files_inner(
tokio::spawn(block2);
Ok(rx)
}
pub async fn channel_config(q: &ChannelConfigQuery, aa: &ArchiverAppliance) -> Result<ChannelConfigResponse, Error> {
let ci = crate::events::channel_info(&q.channel, aa).await?;
let ret = ChannelConfigResponse {
channel: q.channel.clone(),
scalar_type: ci.scalar_type,
byte_order: ci.byte_order,
shape: ci.shape,
};
Ok(ret)
}