viva_gige/
action.rs

1//! GVCP action command helpers.
2
3use std::collections::HashSet;
4use std::io;
5use std::io::ErrorKind;
6use std::net::{IpAddr, Ipv4Addr, SocketAddr};
7use std::time::{Duration, Instant};
8
9use bytes::{BufMut, BytesMut};
10use tokio::net::UdpSocket;
11use tokio::time;
12use tracing::{debug, info, trace, warn};
13
14use crate::gvcp::{GVCP_PORT, GvcpAckHeader, GvcpRequestHeader, consts};
15
16/// Size of an unscheduled action command payload in bytes.
17///
18/// `device_key` + `group_key` + `group_mask`, per the GigE Vision `ACTION_CMD`
19/// field table (Wireshark's `dissect_action_cmd` reads exactly these three).
20const ACTION_PAYLOAD: usize = 12;
21/// Size of a scheduled action command payload in bytes.
22///
23/// The base payload plus a 64-bit action time at offset 12, present only when
24/// bit 7 of the GVCP flags byte is set.
25const ACTION_PAYLOAD_SCHEDULED: usize = ACTION_PAYLOAD + 8;
26
27/// Parameters used to construct an action command.
28#[derive(Debug, Clone, PartialEq, Eq)]
29pub struct ActionParams {
30    /// Vendor-specific device key used to authorise the action.
31    pub device_key: u32,
32    /// Group key identifying which devices should react to the action.
33    pub group_key: u32,
34    /// Group mask applied to the device key by receivers.
35    pub group_mask: u32,
36    /// Optional scheduled time expressed in device clock ticks.
37    ///
38    /// `Some` selects a scheduled action command: the time is appended to the
39    /// payload and bit 7 of the GVCP flags byte is set. A device that does not
40    /// implement scheduled actions rejects it, so leave this `None` unless the
41    /// camera advertises `GevSupportedOptionScheduledAction`.
42    pub scheduled_time: Option<u64>,
43}
44
45/// Summary of the broadcast performed by [`send_action`].
46#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
47pub struct AckSummary {
48    /// Number of GVCP datagrams transmitted.
49    pub sent: usize,
50    /// Number of distinct acknowledgement sources observed.
51    pub acks: usize,
52}
53
54fn encode_payload(params: &ActionParams) -> BytesMut {
55    let mut buf = BytesMut::with_capacity(ACTION_PAYLOAD_SCHEDULED);
56    buf.put_u32(params.device_key);
57    buf.put_u32(params.group_key);
58    buf.put_u32(params.group_mask);
59    // The action time is not a fixed field padded with zeros: an unscheduled
60    // command is 12 bytes and stops here. Sending the extra 8 bytes anyway
61    // makes the payload length disagree with the flags byte.
62    if let Some(ticks) = params.scheduled_time {
63        buf.put_u64(ticks);
64    }
65    buf
66}
67
68fn parse_ack(buf: &[u8]) -> io::Result<GvcpAckHeader> {
69    if buf.len() < 8 {
70        return Err(io::Error::new(
71            ErrorKind::InvalidData,
72            "acknowledgement shorter than GVCP header",
73        ));
74    }
75    let status = u16::from_be_bytes([buf[0], buf[1]]);
76    let opcode = u16::from_be_bytes([buf[2], buf[3]]);
77    let length = u16::from_be_bytes([buf[4], buf[5]]);
78    let request_id = u16::from_be_bytes([buf[6], buf[7]]);
79    Ok(GvcpAckHeader {
80        status: viva_gencp::StatusCode::from_raw(status),
81        command: opcode,
82        length,
83        request_id,
84    })
85}
86
87fn is_broadcast(addr: &SocketAddr) -> bool {
88    matches!(addr.ip(), IpAddr::V4(ip) if ip == Ipv4Addr::BROADCAST)
89}
90
91/// Send a GVCP action command and collect acknowledgements.
92pub async fn send_action(
93    broadcast: SocketAddr,
94    params: &ActionParams,
95    timeout_ms: u64,
96) -> io::Result<AckSummary> {
97    let destination = SocketAddr::new(broadcast.ip(), GVCP_PORT);
98    let local_ip = match destination.ip() {
99        IpAddr::V4(_) => IpAddr::V4(Ipv4Addr::UNSPECIFIED),
100        IpAddr::V6(_) => {
101            return Err(io::Error::new(
102                ErrorKind::InvalidInput,
103                "IPv6 destinations are not supported for actions",
104            ));
105        }
106    };
107    let socket = UdpSocket::bind(SocketAddr::new(local_ip, 0)).await?;
108    if is_broadcast(&destination) {
109        socket.set_broadcast(true)?;
110    }
111
112    let mut summary = AckSummary::default();
113    let payload = encode_payload(params);
114    let request_id = fastrand::u16(0x8000..=0xFFFE);
115    let mut flags = viva_gencp::CommandFlags::ACK_REQUIRED;
116    if is_broadcast(&destination) {
117        flags |= viva_gencp::CommandFlags::BROADCAST;
118    }
119    if params.scheduled_time.is_some() {
120        flags |= viva_gencp::CommandFlags::SCHEDULED_ACTION;
121    }
122    let header = GvcpRequestHeader {
123        flags,
124        command: consts::ACTION_COMMAND,
125        length: payload.len() as u16,
126        request_id,
127    };
128    let packet = header.encode(&payload);
129    trace!(bytes = packet.len(), %destination, request_id, "sending action command");
130    socket.send_to(&packet, destination).await?;
131    summary.sent = 1;
132
133    let timeout = Duration::from_millis(timeout_ms);
134    if timeout.is_zero() {
135        info!(acks = 0, "action command sent (no wait)");
136        return Ok(summary);
137    }
138
139    let start = Instant::now();
140    let mut buf = vec![0u8; 512];
141    let mut seen = HashSet::new();
142    while let Some(remaining) = timeout.checked_sub(start.elapsed()) {
143        if remaining.is_zero() {
144            break;
145        }
146        match time::timeout(remaining, socket.recv_from(&mut buf)).await {
147            Ok(Ok((len, src))) => {
148                trace!(bytes = len, %src, "received acknowledgement");
149                let header = parse_ack(&buf[..len])?;
150                if header.command != consts::ACTION_ACK {
151                    debug!(
152                        opcode = header.command,
153                        "ignoring unrelated acknowledgement"
154                    );
155                    continue;
156                }
157                if header.request_id != request_id {
158                    debug!(
159                        expected = request_id,
160                        got = header.request_id,
161                        "acknowledgement id mismatch"
162                    );
163                    continue;
164                }
165                if header.status != viva_gencp::StatusCode::Success {
166                    warn!(status = ?header.status, %src, "device reported action failure");
167                    continue;
168                }
169                if seen.insert(src.ip()) {
170                    summary.acks += 1;
171                }
172            }
173            Ok(Err(err)) => {
174                warn!(?err, "error receiving acknowledgement");
175                break;
176            }
177            Err(_) => break,
178        }
179    }
180
181    info!(acks = summary.acks, "action command completed");
182    Ok(summary)
183}
184
185#[cfg(test)]
186mod tests {
187    use super::*;
188
189    fn params() -> ActionParams {
190        ActionParams {
191            device_key: 0x1122_3344,
192            group_key: 0x5566_7788,
193            group_mask: 0xFFFF_0000,
194            scheduled_time: None,
195        }
196    }
197
198    /// The whole datagram, byte for byte, written out from the GigE Vision
199    /// `ACTION_CMD` field table rather than from our own encoder.
200    ///
201    /// This is the fixture that would have caught the 0x0080 collision: the
202    /// opcode is a literal here, so an encoder that emits `READREG` fails
203    /// rather than agreeing with itself (ADR-0019).
204    #[test]
205    fn unscheduled_action_matches_spec_bytes() {
206        let payload = encode_payload(&params());
207        let packet = GvcpRequestHeader {
208            flags: viva_gencp::CommandFlags::ACK_REQUIRED,
209            command: consts::ACTION_COMMAND,
210            length: payload.len() as u16,
211            request_id: 0xBEEF,
212        }
213        .encode(&payload);
214
215        #[rustfmt::skip]
216        let expected: [u8; 20] = [
217            0x42,                   // command key
218            0x01,                   // flags: acknowledge required
219            0x01, 0x00,             // ACTION_CMD — *not* 0x0080 (READREG)
220            0x00, 0x0C,             // length: 12, no action time
221            0xBE, 0xEF,             // request id
222            0x11, 0x22, 0x33, 0x44, // device key
223            0x55, 0x66, 0x77, 0x88, // group key
224            0xFF, 0xFF, 0x00, 0x00, // group mask
225        ];
226        assert_eq!(&packet[..], &expected[..]);
227    }
228
229    /// A scheduled action appends the 64-bit time *and* sets flags bit 7.
230    /// Either one alone is a malformed command.
231    #[test]
232    fn scheduled_action_sets_flag_and_appends_time() {
233        let mut p = params();
234        p.scheduled_time = Some(0x0102_0304_0506_0708);
235        let payload = encode_payload(&p);
236        let mut flags = viva_gencp::CommandFlags::ACK_REQUIRED;
237        flags |= viva_gencp::CommandFlags::SCHEDULED_ACTION;
238        let packet = GvcpRequestHeader {
239            flags,
240            command: consts::ACTION_COMMAND,
241            length: payload.len() as u16,
242            request_id: 0xBEEF,
243        }
244        .encode(&payload);
245
246        #[rustfmt::skip]
247        let expected: [u8; 28] = [
248            0x42,
249            0x81,                   // flags: acknowledge required | scheduled
250            0x01, 0x00,
251            0x00, 0x14,             // length: 20
252            0xBE, 0xEF,
253            0x11, 0x22, 0x33, 0x44,
254            0x55, 0x66, 0x77, 0x88,
255            0xFF, 0xFF, 0x00, 0x00,
256            0x01, 0x02, 0x03, 0x04, // action time, big-endian u64
257            0x05, 0x06, 0x07, 0x08,
258        ];
259        assert_eq!(&packet[..], &expected[..]);
260        assert_eq!(payload.len(), ACTION_PAYLOAD_SCHEDULED);
261    }
262
263    #[test]
264    fn unscheduled_payload_stops_at_twelve_bytes() {
265        assert_eq!(encode_payload(&params()).len(), ACTION_PAYLOAD);
266    }
267
268    /// The two opcodes an action must never be confused with.
269    #[test]
270    fn action_opcodes_do_not_collide_with_register_access() {
271        assert_eq!(consts::ACTION_COMMAND, 0x0100);
272        assert_eq!(consts::ACTION_ACK, 0x0101);
273        assert_ne!(consts::ACTION_COMMAND, 0x0080); // READREG_CMD
274        assert_ne!(consts::ACTION_ACK, 0x0081); // READREG_ACK
275    }
276
277    #[test]
278    fn ack_parser() {
279        let mut buf = BytesMut::with_capacity(8);
280        buf.put_u16(viva_gencp::StatusCode::Success.to_raw());
281        buf.put_u16(consts::ACTION_ACK);
282        buf.put_u16(0);
283        buf.put_u16(0xBEEF);
284        let ack = parse_ack(&buf).expect("ack");
285        assert_eq!(ack.command, consts::ACTION_ACK);
286        assert_eq!(ack.request_id, 0xBEEF);
287    }
288}
viva_gige/action.rs

viva_gige/
action.rs
