Merge pull request #84 from pipeless-ai/fix_timestamps

fix(data): Set epoch seconds in frame input_ts

examples/yolo/process.py

@@ -2,27 +2,13 @@
 import time
 
 def hook(frame, context):
-    start = time.time()
     rgb_frame = frame['original']
     model = context['model']
     input_fps = frame['fps']
-    raw_samples = pipeless_kvs_get('samples')
-    samples = int(raw_samples) if raw_samples else 0
-    raw_avg_time = pipeless_kvs_get('avg_process_time')
-    avg_time = float(raw_avg_time) if raw_avg_time else 0
-
-    if should_skip_frame(avg_time, input_fps):
-        print('Skipping frame to maintain real-time')
+    delay = time.time() - frame['input_ts']
+    if delay > 1 / input_fps:
+       print('Skipping frame to maintain real-time')
     else:
         prediction = next(model(rgb_frame, stream=True))
         bboxes = prediction.boxes.data.tolist() if prediction.boxes else []
         frame['inference_output'] = np.array(bboxes, dtype="float32")
-
-    exec_time = time.time() - start
-    new_avg_time = (avg_time * samples + exec_time) / (samples + 1)
-    pipeless_kvs_set('samples', samples + 1)
-    pipeless_kvs_set('avg_process_time', new_avg_time)
-
-def should_skip_frame(avg_execution_time, desired_fps):
-    max_process_time = 1 / desired_fps
-    return avg_execution_time > max_process_time

pipeless/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pipeless-ai"
-version = "1.1.1"
+version = "1.1.2"
 edition = "2021"
 authors = ["Miguel A. Cabrera Minagorri"]
 description = "An open-source computer vision framework to build and deploy applications in minutes"

pipeless/src/data.rs

@@ -13,7 +13,7 @@ pub struct RgbFrame {
     dts: gst::ClockTime,
     duration: gst::ClockTime,
     fps: u8,
-    input_ts: std::time::Instant, // to measure processing performance
+    input_ts: f64, // epoch in seconds
     inference_input: ndarray::ArrayBase<ndarray::OwnedRepr<f32>, ndarray::Dim<ndarray::IxDynImpl>>,
     inference_output: ndarray::ArrayBase<ndarray::OwnedRepr<f32>, ndarray::Dim<ndarray::IxDynImpl>>,
     pipeline_id: uuid::Uuid,
@@ -23,7 +23,7 @@ impl RgbFrame {
         original: ndarray::Array3<u8>,
         width: usize, height: usize,
         pts: gst::ClockTime, dts: gst::ClockTime, duration: gst::ClockTime,
-        fps: u8, input_ts: std::time::Instant,
+        fps: u8, input_ts: f64,
         pipeline_id: uuid::Uuid,
     ) -> Self {
         let modified = original.to_owned();
@@ -45,7 +45,7 @@ impl RgbFrame {
         modified: ndarray::Array3<u8>,
         width: usize, height: usize,
         pts: u64, dts: u64, duration: u64,
-        fps: u8, input_ts: u64,
+        fps: u8, input_ts: f64,
         inference_input: ndarray::ArrayBase<ndarray::OwnedRepr<f32>, ndarray::Dim<ndarray::IxDynImpl>>,
         inference_output: ndarray::ArrayBase<ndarray::OwnedRepr<f32>, ndarray::Dim<ndarray::IxDynImpl>>,
         pipeline_id: &str,
@@ -57,8 +57,7 @@ impl RgbFrame {
             pts: gst::ClockTime::from_mseconds(pts),
             dts: gst::ClockTime::from_mseconds(dts),
             duration: gst::ClockTime::from_mseconds(duration),
-            fps,
-            input_ts: std::time::Instant::now() - std::time::Duration::from_millis(input_ts),
+            fps, input_ts,
             inference_input, inference_output,
             pipeline_id: uuid::Uuid::from_str(pipeline_id).unwrap(),
         }
@@ -108,7 +107,7 @@ impl RgbFrame {
     pub fn get_duration(&self) -> gst::ClockTime {
         self.duration
     }
-    pub fn get_input_ts(&self) -> std::time::Instant {
+    pub fn get_input_ts(&self) -> f64 {
         self.input_ts
     }
     pub fn get_inference_input(&self) -> &ndarray::ArrayBase<ndarray::OwnedRepr<f32>, ndarray::Dim<ndarray::IxDynImpl>> {
@@ -139,7 +138,7 @@ impl Frame {
         original: ndarray::Array3<u8>,
         width: usize, height: usize,
         pts: gst::ClockTime, dts: gst::ClockTime, duration: gst::ClockTime,
-        fps: u8, input_ts: std::time::Instant,
+        fps: u8, input_ts: f64,
         pipeline_id: uuid::Uuid
     ) -> Self {
         let rgb = RgbFrame::new(

pipeless/src/input/pipeline.rs

@@ -51,7 +51,7 @@ fn on_new_sample(
         gst::FlowError::Error
     })?;
 
-    let frame_input_instant = std::time::Instant::now();
+    let frame_input_instant = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs_f64();
     let caps = sample.caps().ok_or_else(|| {
         error!("Unable to get sample capabilities");
         gst::FlowError::Error

pipeless/src/stages/languages/python.rs

@@ -37,7 +37,7 @@ impl IntoPy<Py<PyAny>> for RgbFrame {
         dict.set_item("dts", self.get_dts().mseconds()).unwrap();
         dict.set_item("duration", self.get_duration().mseconds()).unwrap();
         dict.set_item("fps", self.get_fps()).unwrap();
-        dict.set_item("input_ts", self.get_input_ts().elapsed().as_millis()).unwrap();
+        dict.set_item("input_ts", self.get_input_ts()).unwrap();
         dict.set_item("inference_input", self.get_inference_input().to_owned().to_pyarray(py)).unwrap();
         dict.set_item("inference_output", self.get_inference_output().to_owned().to_pyarray(py)).unwrap();
         dict.set_item("pipeline_id", self.get_pipeline_id().to_string()).unwrap();