Struct Pipeline

Source

pub struct Pipeline<R: Reducer> { /* private fields */ }

Expand description

Event driver that wraps a reducer.

Implementations§

Source §

impl<R: Reducer> Pipeline<R>

Source

pub fn new(reducer: R) -> Self

Create it with asynchronous hook execution and KeepAll for every kind.

Examples found in repository ?

examples/hooks.rs (line 71)

49fn main() -> Result<(), String> {
50    let out = std::env::args().nth(1).map(PathBuf::from).unwrap_or_else(|| std::env::temp_dir().join("cumulus3d_hooks"));
51    let src = out.with_extension("src");
52    if !src.join("gps_ref.txt").exists() {
53        common::make_dataset(&src);
54    }
55    if out.exists() {
56        std::fs::remove_dir_all(&out).map_err(|e| e.to_string())?;
57    }
58    let layout = Layout::discover(&src, 1)?;
59    let npos = layout.frames.len();
60
61    let mut cfg = SessionConfig::new(src.join("images"));
62    cfg.gps = read_gps_file(src.join("gps_ref.txt")).map_err(|e| e.to_string())?;
63    (cfg.span, cfg.overlap, cfg.total_positions, cfg.seed) = (4, 1, Some(npos), Some(7));
64    if cumulus3d_cuda::is_available() {
65        let mut d = DenseConfig::new(make_pm_backend("cuda"), parse_profile("fast")?);
66        d.undistort.max_image_size = 160;
67        cfg.dense = Some(d);
68    }
69
70    // Hook: print event text, zone point-cloud point counts, warnings and errors; plus the default output hook (files).
71    let pipeline = Pipeline::new(Session::new(cfg))
72        .on_any(|e: &Event| {
73            if let Some(t) = e.timeline_text() {
74                println!("[event v{}] {t}", e.meta().version);
75            }
76        })
77        .on(EventKind::ZonePreview, |e: &Event| {
78            if let Event::ZonePreview { range, cloud, model, .. } = e {
79                println!("  초벌 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
80            }
81        })
82        .on(EventKind::ZoneRefined, |e: &Event| {
83            if let Event::ZoneRefined { range, cloud, model, .. } = e {
84                println!("  정밀 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
85            }
86        })
87        .on_message(|k, m| eprintln!("  {k:?}: {m}"));
88    let mut pipeline = sinks::attach(pipeline, &out, &SinkOptions::default()).map_err(|e| e.to_string())?;
89
90    for p in 0..npos {
91        pipeline.push(Input::Frames(FrameSet::new(CAMS.iter().map(|c| (*c, layout.name(c, p))))));
92    }
93    let (_, summary) = pipeline.finish();
94    println!("이벤트 {} 개, 훅 호출 {} 번, 훅 패닉 {}", summary.events, summary.hook_calls, summary.hook_panics);
95
96    for f in ["timeline.txt", "run.log", "snapshots/manifest.json", "snapshots/timeline.txt", "DONE"] {
97        let p = out.join(f);
98        println!("{} {}", if p.exists() { "있음" } else { "없음" }, p.display());
99    }
100    Ok(())
101}

Source

pub fn sync(self, on: bool) -> Self

Set synchronous mode. Decide before the first push. Switching to synchronous first drains the asynchronous queues accumulated so far.

Source

pub fn policy(self, kind: EventKind, p: QueuePolicy) -> Self

Set the per-kind queue policy (only meaningful in asynchronous mode).

Source

pub fn on<F>(self, kind: EventKind, f: F) -> Self
where F: FnMut(&Event) + Send + 'static,

Register a hook for a specific kind. Hooks of the same kind run in registration order.

Examples found in repository ?

examples/hooks.rs (lines 77-81)

49fn main() -> Result<(), String> {
50    let out = std::env::args().nth(1).map(PathBuf::from).unwrap_or_else(|| std::env::temp_dir().join("cumulus3d_hooks"));
51    let src = out.with_extension("src");
52    if !src.join("gps_ref.txt").exists() {
53        common::make_dataset(&src);
54    }
55    if out.exists() {
56        std::fs::remove_dir_all(&out).map_err(|e| e.to_string())?;
57    }
58    let layout = Layout::discover(&src, 1)?;
59    let npos = layout.frames.len();
60
61    let mut cfg = SessionConfig::new(src.join("images"));
62    cfg.gps = read_gps_file(src.join("gps_ref.txt")).map_err(|e| e.to_string())?;
63    (cfg.span, cfg.overlap, cfg.total_positions, cfg.seed) = (4, 1, Some(npos), Some(7));
64    if cumulus3d_cuda::is_available() {
65        let mut d = DenseConfig::new(make_pm_backend("cuda"), parse_profile("fast")?);
66        d.undistort.max_image_size = 160;
67        cfg.dense = Some(d);
68    }
69
70    // Hook: print event text, zone point-cloud point counts, warnings and errors; plus the default output hook (files).
71    let pipeline = Pipeline::new(Session::new(cfg))
72        .on_any(|e: &Event| {
73            if let Some(t) = e.timeline_text() {
74                println!("[event v{}] {t}", e.meta().version);
75            }
76        })
77        .on(EventKind::ZonePreview, |e: &Event| {
78            if let Event::ZonePreview { range, cloud, model, .. } = e {
79                println!("  초벌 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
80            }
81        })
82        .on(EventKind::ZoneRefined, |e: &Event| {
83            if let Event::ZoneRefined { range, cloud, model, .. } = e {
84                println!("  정밀 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
85            }
86        })
87        .on_message(|k, m| eprintln!("  {k:?}: {m}"));
88    let mut pipeline = sinks::attach(pipeline, &out, &SinkOptions::default()).map_err(|e| e.to_string())?;
89
90    for p in 0..npos {
91        pipeline.push(Input::Frames(FrameSet::new(CAMS.iter().map(|c| (*c, layout.name(c, p))))));
92    }
93    let (_, summary) = pipeline.finish();
94    println!("이벤트 {} 개, 훅 호출 {} 번, 훅 패닉 {}", summary.events, summary.hook_calls, summary.hook_panics);
95
96    for f in ["timeline.txt", "run.log", "snapshots/manifest.json", "snapshots/timeline.txt", "DONE"] {
97        let p = out.join(f);
98        println!("{} {}", if p.exists() { "있음" } else { "없음" }, p.display());
99    }
100    Ok(())
101}

Source

pub fn on_any<F>(self, f: F) -> Self
where F: FnMut(&Event) + Send + 'static,

Register a hook for all events (preserves the overall order).

Examples found in repository ?

examples/hooks.rs (lines 72-76)

49fn main() -> Result<(), String> {
50    let out = std::env::args().nth(1).map(PathBuf::from).unwrap_or_else(|| std::env::temp_dir().join("cumulus3d_hooks"));
51    let src = out.with_extension("src");
52    if !src.join("gps_ref.txt").exists() {
53        common::make_dataset(&src);
54    }
55    if out.exists() {
56        std::fs::remove_dir_all(&out).map_err(|e| e.to_string())?;
57    }
58    let layout = Layout::discover(&src, 1)?;
59    let npos = layout.frames.len();
60
61    let mut cfg = SessionConfig::new(src.join("images"));
62    cfg.gps = read_gps_file(src.join("gps_ref.txt")).map_err(|e| e.to_string())?;
63    (cfg.span, cfg.overlap, cfg.total_positions, cfg.seed) = (4, 1, Some(npos), Some(7));
64    if cumulus3d_cuda::is_available() {
65        let mut d = DenseConfig::new(make_pm_backend("cuda"), parse_profile("fast")?);
66        d.undistort.max_image_size = 160;
67        cfg.dense = Some(d);
68    }
69
70    // Hook: print event text, zone point-cloud point counts, warnings and errors; plus the default output hook (files).
71    let pipeline = Pipeline::new(Session::new(cfg))
72        .on_any(|e: &Event| {
73            if let Some(t) = e.timeline_text() {
74                println!("[event v{}] {t}", e.meta().version);
75            }
76        })
77        .on(EventKind::ZonePreview, |e: &Event| {
78            if let Event::ZonePreview { range, cloud, model, .. } = e {
79                println!("  초벌 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
80            }
81        })
82        .on(EventKind::ZoneRefined, |e: &Event| {
83            if let Event::ZoneRefined { range, cloud, model, .. } = e {
84                println!("  정밀 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
85            }
86        })
87        .on_message(|k, m| eprintln!("  {k:?}: {m}"));
88    let mut pipeline = sinks::attach(pipeline, &out, &SinkOptions::default()).map_err(|e| e.to_string())?;
89
90    for p in 0..npos {
91        pipeline.push(Input::Frames(FrameSet::new(CAMS.iter().map(|c| (*c, layout.name(c, p))))));
92    }
93    let (_, summary) = pipeline.finish();
94    println!("이벤트 {} 개, 훅 호출 {} 번, 훅 패닉 {}", summary.events, summary.hook_calls, summary.hook_panics);
95
96    for f in ["timeline.txt", "run.log", "snapshots/manifest.json", "snapshots/timeline.txt", "DONE"] {
97        let p = out.join(f);
98        println!("{} {}", if p.exists() { "있음" } else { "없음" }, p.display());
99    }
100    Ok(())
101}

Source

pub fn on_frame<F>(self, f: F) -> Self
where F: FnMut(&Arc<Frame>) + Send + 'static,

Frame hook: receives each decoded input frame, one camera at a time, before the position is processed. The session emits frames only if SessionConfig::decode_frames is on.

Source

pub fn on_zone_preview<F>(self, f: F) -> Self
where F: FnMut(ZoneRange, &Arc<PointCloud>) + Send + 'static,

Zone preview point-cloud hook.

Source

pub fn on_zone_refined<F>(self, f: F) -> Self
where F: FnMut(ZoneRange, &Arc<PointCloud>) + Send + 'static,

Zone refined point-cloud hook.

Source

pub fn on_snapshot<F>(self, f: F) -> Self
where F: FnMut(usize, &Arc<PointCloud>) + Send + 'static,

Snapshot hook (index, composite point cloud).

Source

pub fn on_position_done<F>(self, f: F) -> Self
where F: FnMut(usize, usize, usize) + Send + 'static,

Position-done hook (position, registered count, expected count).

Source

pub fn on_message<F>(self, f: F) -> Self
where F: FnMut(EventKind, &str) + Send + 'static,

Warning/error message hook (those emitted by the reducer; hook panic errors go only to the subscription channel).

Examples found in repository ?

examples/hooks.rs (line 87)

49fn main() -> Result<(), String> {
50    let out = std::env::args().nth(1).map(PathBuf::from).unwrap_or_else(|| std::env::temp_dir().join("cumulus3d_hooks"));
51    let src = out.with_extension("src");
52    if !src.join("gps_ref.txt").exists() {
53        common::make_dataset(&src);
54    }
55    if out.exists() {
56        std::fs::remove_dir_all(&out).map_err(|e| e.to_string())?;
57    }
58    let layout = Layout::discover(&src, 1)?;
59    let npos = layout.frames.len();
60
61    let mut cfg = SessionConfig::new(src.join("images"));
62    cfg.gps = read_gps_file(src.join("gps_ref.txt")).map_err(|e| e.to_string())?;
63    (cfg.span, cfg.overlap, cfg.total_positions, cfg.seed) = (4, 1, Some(npos), Some(7));
64    if cumulus3d_cuda::is_available() {
65        let mut d = DenseConfig::new(make_pm_backend("cuda"), parse_profile("fast")?);
66        d.undistort.max_image_size = 160;
67        cfg.dense = Some(d);
68    }
69
70    // Hook: print event text, zone point-cloud point counts, warnings and errors; plus the default output hook (files).
71    let pipeline = Pipeline::new(Session::new(cfg))
72        .on_any(|e: &Event| {
73            if let Some(t) = e.timeline_text() {
74                println!("[event v{}] {t}", e.meta().version);
75            }
76        })
77        .on(EventKind::ZonePreview, |e: &Event| {
78            if let Event::ZonePreview { range, cloud, model, .. } = e {
79                println!("  초벌 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
80            }
81        })
82        .on(EventKind::ZoneRefined, |e: &Event| {
83            if let Event::ZoneRefined { range, cloud, model, .. } = e {
84                println!("  정밀 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
85            }
86        })
87        .on_message(|k, m| eprintln!("  {k:?}: {m}"));
88    let mut pipeline = sinks::attach(pipeline, &out, &SinkOptions::default()).map_err(|e| e.to_string())?;
89
90    for p in 0..npos {
91        pipeline.push(Input::Frames(FrameSet::new(CAMS.iter().map(|c| (*c, layout.name(c, p))))));
92    }
93    let (_, summary) = pipeline.finish();
94    println!("이벤트 {} 개, 훅 호출 {} 번, 훅 패닉 {}", summary.events, summary.hook_calls, summary.hook_panics);
95
96    for f in ["timeline.txt", "run.log", "snapshots/manifest.json", "snapshots/timeline.txt", "DONE"] {
97        let p = out.join(f);
98        println!("{} {}", if p.exists() { "있음" } else { "없음" }, p.display());
99    }
100    Ok(())
101}

Channel that receives all events (including hook panic errors). It is released automatically if the receiver is dropped.

Source

pub fn push(&mut self, input: R::Input) -> usize

Feed one input to the reducer and emit the resulting events. Returns the number of events emitted. Reducer::prelude events enter the hook queue before processing, so asynchronous hooks run concurrently with the processing.

Examples found in repository ?

examples/hooks.rs (line 91)

49fn main() -> Result<(), String> {
50    let out = std::env::args().nth(1).map(PathBuf::from).unwrap_or_else(|| std::env::temp_dir().join("cumulus3d_hooks"));
51    let src = out.with_extension("src");
52    if !src.join("gps_ref.txt").exists() {
53        common::make_dataset(&src);
54    }
55    if out.exists() {
56        std::fs::remove_dir_all(&out).map_err(|e| e.to_string())?;
57    }
58    let layout = Layout::discover(&src, 1)?;
59    let npos = layout.frames.len();
60
61    let mut cfg = SessionConfig::new(src.join("images"));
62    cfg.gps = read_gps_file(src.join("gps_ref.txt")).map_err(|e| e.to_string())?;
63    (cfg.span, cfg.overlap, cfg.total_positions, cfg.seed) = (4, 1, Some(npos), Some(7));
64    if cumulus3d_cuda::is_available() {
65        let mut d = DenseConfig::new(make_pm_backend("cuda"), parse_profile("fast")?);
66        d.undistort.max_image_size = 160;
67        cfg.dense = Some(d);
68    }
69
70    // Hook: print event text, zone point-cloud point counts, warnings and errors; plus the default output hook (files).
71    let pipeline = Pipeline::new(Session::new(cfg))
72        .on_any(|e: &Event| {
73            if let Some(t) = e.timeline_text() {
74                println!("[event v{}] {t}", e.meta().version);
75            }
76        })
77        .on(EventKind::ZonePreview, |e: &Event| {
78            if let Event::ZonePreview { range, cloud, model, .. } = e {
79                println!("  초벌 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
80            }
81        })
82        .on(EventKind::ZoneRefined, |e: &Event| {
83            if let Event::ZoneRefined { range, cloud, model, .. } = e {
84                println!("  정밀 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
85            }
86        })
87        .on_message(|k, m| eprintln!("  {k:?}: {m}"));
88    let mut pipeline = sinks::attach(pipeline, &out, &SinkOptions::default()).map_err(|e| e.to_string())?;
89
90    for p in 0..npos {
91        pipeline.push(Input::Frames(FrameSet::new(CAMS.iter().map(|c| (*c, layout.name(c, p))))));
92    }
93    let (_, summary) = pipeline.finish();
94    println!("이벤트 {} 개, 훅 호출 {} 번, 훅 패닉 {}", summary.events, summary.hook_calls, summary.hook_panics);
95
96    for f in ["timeline.txt", "run.log", "snapshots/manifest.json", "snapshots/timeline.txt", "DONE"] {
97        let p = out.join(f);
98        println!("{} {}", if p.exists() { "있음" } else { "없음" }, p.display());
99    }
100    Ok(())
101}

Source

pub fn poll(&mut self) -> usize

Collect finished background work and emit it (does not wait).

Source

pub fn command(&mut self, c: Command) -> usize

Pass a command to the reducer and emit the resulting events.

Source

pub fn flush(&mut self)

Wait until all hook executions queued so far have finished.

Source

pub fn reducer(&self) -> &R

Reducer reference.

Source

pub fn reducer_mut(&mut self) -> &mut R

Mutable reducer reference (events are not emitted).

Source

pub fn finish(self) -> (R, Summary)

Collect and emit the remaining background work, drain the hook queues, close the workers and return the summary. The subscription channels are closed when the pipeline ends.

Examples found in repository ?

examples/hooks.rs (line 93)

49fn main() -> Result<(), String> {
50    let out = std::env::args().nth(1).map(PathBuf::from).unwrap_or_else(|| std::env::temp_dir().join("cumulus3d_hooks"));
51    let src = out.with_extension("src");
52    if !src.join("gps_ref.txt").exists() {
53        common::make_dataset(&src);
54    }
55    if out.exists() {
56        std::fs::remove_dir_all(&out).map_err(|e| e.to_string())?;
57    }
58    let layout = Layout::discover(&src, 1)?;
59    let npos = layout.frames.len();
60
61    let mut cfg = SessionConfig::new(src.join("images"));
62    cfg.gps = read_gps_file(src.join("gps_ref.txt")).map_err(|e| e.to_string())?;
63    (cfg.span, cfg.overlap, cfg.total_positions, cfg.seed) = (4, 1, Some(npos), Some(7));
64    if cumulus3d_cuda::is_available() {
65        let mut d = DenseConfig::new(make_pm_backend("cuda"), parse_profile("fast")?);
66        d.undistort.max_image_size = 160;
67        cfg.dense = Some(d);
68    }
69
70    // Hook: print event text, zone point-cloud point counts, warnings and errors; plus the default output hook (files).
71    let pipeline = Pipeline::new(Session::new(cfg))
72        .on_any(|e: &Event| {
73            if let Some(t) = e.timeline_text() {
74                println!("[event v{}] {t}", e.meta().version);
75            }
76        })
77        .on(EventKind::ZonePreview, |e: &Event| {
78            if let Event::ZonePreview { range, cloud, model, .. } = e {
79                println!("  초벌 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
80            }
81        })
82        .on(EventKind::ZoneRefined, |e: &Event| {
83            if let Event::ZoneRefined { range, cloud, model, .. } = e {
84                println!("  정밀 {}: 조밀 점 {} / 희소 점 {}", range.zone, cloud.len(), model.num_points3d());
85            }
86        })
87        .on_message(|k, m| eprintln!("  {k:?}: {m}"));
88    let mut pipeline = sinks::attach(pipeline, &out, &SinkOptions::default()).map_err(|e| e.to_string())?;
89
90    for p in 0..npos {
91        pipeline.push(Input::Frames(FrameSet::new(CAMS.iter().map(|c| (*c, layout.name(c, p))))));
92    }
93    let (_, summary) = pipeline.finish();
94    println!("이벤트 {} 개, 훅 호출 {} 번, 훅 패닉 {}", summary.events, summary.hook_calls, summary.hook_panics);
95
96    for f in ["timeline.txt", "run.log", "snapshots/manifest.json", "snapshots/timeline.txt", "DONE"] {
97        let p = out.join(f);
98        println!("{} {}", if p.exists() { "있음" } else { "없음" }, p.display());
99    }
100    Ok(())
101}

Auto Trait Implementations§

§

impl<R> UnsafeUnpin for Pipeline<R>
where R: UnsafeUnpin,

Blanket Implementations§

Source §

impl<T> Any for T
where T: 'static + ?Sized,

Source §

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more

Source §

impl<T> Borrow<T> for T
where T: ?Sized,

Source §

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more

Source §

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source §

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more

Source §

impl<T> ByRef<T> for T

Source §

fn by_ref(&self) -> &T

Source §

impl<ST, DT> CastableFrom<ST, Initialized, Initialized> for DT
where ST: ?Sized, DT: ?Sized,

Source §

impl<ST, DT> CastableFrom<ST, Uninit, Uninit> for DT
where ST: ?Sized, DT: ?Sized,

Source §

impl<T> DistributionExt for T
where T: ?Sized,

Source §

fn rand<T>(&self, rng: &mut (impl Rng + ?Sized)) -> T
where Self: Distribution<T>,

Source §

impl<T> From<T> for T

Source §

fn from(t: T) -> T

Returns the argument unchanged.

Source §

impl<T, U> Imply<T> for U
where T: ?Sized, U: ?Sized,

Source §

impl<T, U> Into for T
where U: From<T>,

Source §

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source §

impl<T> IntoEither for T

Source §

fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ

Converts self into a Left variant of Either<Self, Self> if into_left is true. Converts self into a Right variant of Either<Self, Self> otherwise. Read more

Source §

fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
where F: FnOnce(&Self) -> bool,

Converts self into a Left variant of Either<Self, Self> if into_left(&self) returns true. Converts self into a Right variant of Either<Self, Self> otherwise. Read more

Source §