viva_camctl/
cmd_stream.rs

1use std::net::Ipv4Addr;
2use std::path::PathBuf;
3use std::time::Duration;
4
5use anyhow::{Context, Result, anyhow, bail};
6use tokio::time::{self, Instant, MissedTickBehavior};
7use tracing::{info, warn};
8
9use viva_genicam::pfnc::PixelFormat;
10use viva_genicam::{Frame, FrameStream, StreamBuilder, StreamDest};
11
12use viva_gige::nic::IfaceSelector;
13
14use crate::common::{self, DEFAULT_DISCOVERY_TIMEOUT_MS};
15
16#[derive(Debug, Clone)]
17pub struct StreamArgs {
18    pub ip: Option<Ipv4Addr>,
19    pub index: Option<usize>,
20    pub iface: Option<IfaceSelector>,
21    pub mode: String,
22    pub group: Option<Ipv4Addr>,
23    pub port: u16,
24    /// Explicit GVSP packet size ceiling. Mutually exclusive with [`StreamArgs::auto`].
25    pub packet_size: Option<u32>,
26    /// Set from NIC MTU then path-probe (ADR-0021). Mutually exclusive with
27    /// [`StreamArgs::packet_size`].
28    pub auto: bool,
29    pub save: usize,
30    pub rgb: bool,
31    pub duration_s: u64,
32}
33
34// `await_holding_lock` resolves its lint level at the enclosing coroutine body,
35// so a statement-scoped `#[allow]` on the `let stream = { … }` below has no
36// effect and the attribute has to live here. The single offending guard is
37// documented at the point it is taken; nothing else in this function holds a
38// lock across an await.
39#[allow(clippy::await_holding_lock)]
40pub async fn run(args: StreamArgs) -> Result<()> {
41    let timeout = Duration::from_millis(DEFAULT_DISCOVERY_TIMEOUT_MS);
42    let device = common::select_device(args.ip, args.index, args.iface.as_ref(), timeout).await?;
43    info!(ip = %device.ip, "opening camera for streaming");
44    let mut camera = common::open_camera(&device)
45        .await
46        .context("open camera for stream")?;
47
48    let iface = common::resolve_receive_iface(args.iface.as_ref(), device.ip)?;
49    let host_ip = iface
50        .ipv4()
51        .ok_or_else(|| anyhow!("interface {} has no IPv4 address", iface.name()))?;
52    let mode = parse_mode(&args.mode)?;
53
54    if let StreamMode::Multicast = mode {
55        let group = args
56            .group
57            .ok_or_else(|| anyhow!("multicast mode requires --group"))?;
58        camera
59            .configure_stream_multicast(0, group, args.port)
60            .context("configure multicast destination")?;
61    }
62
63    let dest = match mode {
64        StreamMode::Unicast => StreamDest::Unicast {
65            dst_ip: host_ip,
66            dst_port: args.port,
67        },
68        StreamMode::Multicast => {
69            let group = args
70                .group
71                .ok_or_else(|| anyhow!("multicast mode requires --group"))?;
72            StreamDest::Multicast {
73                group,
74                port: args.port,
75                loopback: false,
76                ttl: 1,
77            }
78        }
79    };
80    // Negotiate the stream through the camera's existing GVCP handle so control
81    // privilege, stream configuration, and acquisition commands all come from
82    // the same application endpoint. This scope also releases the device lock
83    // before the higher-level Camera API is used again below.
84    //
85    // The guard must span the builder's awaits (StreamBuilder borrows the
86    // locked device), and nothing else contends this mutex until streaming
87    // starts, so holding it across the awaits cannot deadlock here. The
88    // `await_holding_lock` allow this needs is on the function; see there.
89    let stream = {
90        let mut stream_device = camera
91            .transport()
92            .lock_device()
93            .context("access camera control channel for stream configuration")?;
94        stream_device
95            .claim_control()
96            .await
97            .context("claim camera control for stream configuration")?;
98
99        let mut builder = StreamBuilder::new(&mut stream_device)
100            .iface(iface.clone())
101            .dest(dest)
102            .rcvbuf_bytes(64 << 20);
103        if args.auto {
104            builder = builder.auto_packet_size();
105        } else if let Some(size) = args.packet_size {
106            builder = builder.packet_size(size);
107        }
108        if args.port != 0 {
109            builder = builder.destination_port(args.port);
110        }
111        builder.build().await.context("negotiate stream")?
112    };
113
114    // Keep the CLI at the completed-frame boundary. FrameStream owns the receive
115    // buffer, platform-specific packet reception, GVSP reassembly, chunk parsing,
116    // and completed-frame statistics.
117    let mut frame_stream = FrameStream::new(stream, None);
118    // This is a shared snapshot handle; FrameStream records each completed frame
119    // exactly once, so the CLI must not call record_frame() again.
120    let stats = frame_stream.stats_handle();
121
122    if let Err(err) = camera.set("TLParamsLocked", "1") {
123        warn!(error = %err, "failed to lock transport-layer parameters");
124    }
125    camera.acquisition_start().context("start acquisition")?;
126    let mut saved_frames = 0usize;
127    let mut frame_index = 0usize;
128    let end_deadline = if args.duration_s > 0 {
129        Some(Instant::now() + Duration::from_secs(args.duration_s))
130    } else {
131        None
132    };
133    let mut interrupted = false;
134    let mut ctrl_c = Box::pin(tokio::signal::ctrl_c());
135    let mut ticker = time::interval(Duration::from_secs(1));
136    ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
137
138    loop {
139        if let Some(deadline) = end_deadline
140            && Instant::now() >= deadline
141        {
142            info!("stream duration elapsed");
143            break;
144        }
145
146        tokio::select! {
147            _ = ticker.tick() => {
148                // No keepalive here: `GigeRegisterIo` runs one for as long as it
149                // exists, so the control channel survives a stream that sends no
150                // GVCP traffic of its own.
151                let snapshot = stats.snapshot();
152                println!(
153                    "[stream] fps={:.1} Mbps={:.2} frames={} drops={} resends={}",
154                    snapshot.avg_fps,
155                    snapshot.avg_mbps,
156                    snapshot.frames,
157                    snapshot.drops,
158                    snapshot.resends,
159                );
160            }
161            _ = &mut ctrl_c => {
162                info!("received ctrl-c; stopping stream");
163                interrupted = true;
164                break;
165            }
166            received = frame_stream.next_frame() => {
167                match received {
168                    Ok(Some(mut frame)) => {
169                        // FrameStream has already counted this frame. Mapping its
170                        // host timestamp here enriches frame metadata only and
171                        // must not trigger another record_frame() call.
172                        if let Some(timestamp) = frame.ts_dev {
173                            frame.ts_host = Some(camera.map_dev_ts(timestamp));
174                        }
175                        frame_index += 1;
176
177                        if saved_frames < args.save {
178                            if let Err(err) = save_frame(&frame, frame_index, args.rgb) {
179                                warn!(error = %err, frame = frame_index, "failed to save frame");
180                            } else {
181                                saved_frames += 1;
182                            }
183                        }
184                    }
185                    Err(err) => {
186                        warn!(error = %err, "stream receiver failed");
187                        break;
188                    }
189                    Ok(None) => break,
190                }
191            }
192        }
193    }
194
195    camera.acquisition_stop().context("stop acquisition")?;
196    if let Err(err) = camera.set("TLParamsLocked", "0") {
197        warn!(error = %err, "failed to unlock transport-layer parameters");
198    }
199    if interrupted {
200        println!("Stream interrupted by user.");
201    }
202    let summary = stats.snapshot();
203    println!(
204        "Summary: frames={} bytes={} drops={} resends={} avg_fps={:.1} avg_mbps={:.2}",
205        summary.frames,
206        summary.bytes,
207        summary.drops,
208        summary.resends,
209        summary.avg_fps,
210        summary.avg_mbps,
211    );
212
213    Ok(())
214}
215
216#[derive(Debug, Clone, Copy, PartialEq, Eq)]
217enum StreamMode {
218    Unicast,
219    Multicast,
220}
221
222fn parse_mode(value: &str) -> Result<StreamMode> {
223    match value.to_ascii_lowercase().as_str() {
224        "unicast" => Ok(StreamMode::Unicast),
225        "multicast" => Ok(StreamMode::Multicast),
226        other => bail!("unknown stream mode '{other}' (expected unicast or multicast)"),
227    }
228}
229
230fn save_frame(frame: &Frame, index: usize, rgb: bool) -> Result<PathBuf> {
231    let (buffer, ext) = if !rgb && frame.pixel_format == PixelFormat::Mono8 {
232        let data = frame.payload.clone();
233        let encoded = common::encode_pgm(frame.width, frame.height, data.as_ref())?;
234        (encoded, "pgm")
235    } else {
236        let rgb_pixels = frame.to_rgb8().context("convert frame to RGB8")?;
237        let encoded = common::encode_ppm(frame.width, frame.height, &rgb_pixels)?;
238        (encoded, "ppm")
239    };
240    let path = PathBuf::from(format!("frame_{index:04}.{ext}"));
241    common::save_image(&buffer, &path)?;
242    info!(file = %path.display(), "saved frame");
243    Ok(path)
244}
viva_camctl/cmd_stream.rs

viva_camctl/
cmd_stream.rs
