viva_u3v/
stream.rs

1//! USB3 Vision streaming protocol.
2//!
3//! U3V streaming uses USB bulk transfers with a leader/payload/trailer
4//! structure. Unlike GigE Vision (GVSP over UDP), USB bulk transfers are
5//! reliable and ordered — no packet reassembly, bitmap tracking, or resend
6//! logic is needed.
7//!
8//! Frame structure:
9//! ```text
10//! ┌─────────┐   ┌─────────────┐   ┌──────────┐
11//! │ Leader  │ → │ Payload(s)  │ → │ Trailer  │
12//! │ (meta)  │   │ (image data)│   │ (status) │
13//! └─────────┘   └─────────────┘   └──────────┘
14//! ```
15
16use std::sync::Arc;
17use std::time::Duration;
18
19use bytes::Bytes;
20
21use crate::U3vError;
22use crate::usb::UsbTransfer;
23
24/// U3V stream leader prefix: "U3VL" in little-endian.
25const LEADER_PREFIX: u32 = 0x4C56_3355;
26/// U3V stream trailer prefix: "U3VT" in little-endian.
27const TRAILER_PREFIX: u32 = 0x5456_3355;
28
29/// Minimum leader size (prefix + fixed fields).
30const MIN_LEADER_SIZE: usize = 36;
31/// Minimum trailer size (prefix + fixed fields).
32const MIN_TRAILER_SIZE: usize = 16;
33
34/// Default USB bulk read timeout for streaming.
35const STREAM_TIMEOUT: Duration = Duration::from_millis(5000);
36
37/// Parsed U3V stream leader (image metadata).
38#[derive(Debug, Clone)]
39pub struct Leader {
40    /// Payload type (0x0001 = image, 0x0002 = image extended, 0x4001 = chunk).
41    pub payload_type: u16,
42    /// Device timestamp in ticks.
43    pub timestamp: u64,
44    /// PFNC pixel format code.
45    pub pixel_format: u32,
46    /// Image width in pixels.
47    pub width: u32,
48    /// Image height in pixels.
49    pub height: u32,
50    /// X offset (ROI).
51    pub x_offset: u32,
52    /// Y offset (ROI).
53    pub y_offset: u32,
54    /// Horizontal padding bytes per line.
55    pub x_padding: u16,
56}
57
58/// Parsed U3V stream trailer.
59#[derive(Debug, Clone)]
60pub struct Trailer {
61    /// Status of the acquired frame (0 = success).
62    pub status: u32,
63    /// Block ID for this frame.
64    pub block_id: u64,
65    /// Actual number of valid payload bytes.
66    pub valid_payload_size: u64,
67}
68
69/// U3V stream receiver that reads frames from a USB bulk endpoint.
70pub struct U3vStream<T: UsbTransfer> {
71    transport: Arc<T>,
72    ep_in: u8,
73    timeout: Duration,
74    leader_buf: Vec<u8>,
75    trailer_buf: Vec<u8>,
76    payload_buf: Vec<u8>,
77    /// Configured payload size per frame (from SIRM). The device sends
78    /// exactly this many bytes of payload data, potentially across
79    /// multiple USB bulk transfers.
80    payload_size: usize,
81}
82
83impl<T: UsbTransfer> U3vStream<T> {
84    /// Create a new stream receiver.
85    ///
86    /// `max_leader_size`, `max_trailer_size` and `payload_size` come
87    /// from the SIRM registers. `payload_size` is the exact number of
88    /// payload bytes per frame (configured in the SIRM).
89    pub fn new(
90        transport: Arc<T>,
91        ep_in: u8,
92        max_leader_size: usize,
93        max_trailer_size: usize,
94        payload_size: usize,
95    ) -> Self {
96        Self {
97            transport,
98            ep_in,
99            timeout: STREAM_TIMEOUT,
100            leader_buf: vec![0u8; max_leader_size],
101            trailer_buf: vec![0u8; max_trailer_size],
102            payload_buf: vec![0u8; payload_size],
103            payload_size,
104        }
105    }
106
107    /// Override the default bulk read timeout.
108    pub fn set_timeout(&mut self, timeout: Duration) {
109        self.timeout = timeout;
110    }
111
112    /// Receive the next complete frame.
113    ///
114    /// Blocks until a full leader + payload + trailer sequence is read
115    /// from the stream endpoint. Returns the parsed leader, raw payload,
116    /// and trailer.
117    pub fn next_frame(&mut self) -> Result<RawFrame, U3vError> {
118        let leader = self.read_leader()?;
119        let payload = self.read_payload()?;
120        let trailer = self.read_trailer()?;
121
122        Ok(RawFrame {
123            leader,
124            payload,
125            trailer,
126        })
127    }
128
129    fn read_leader(&mut self) -> Result<Leader, U3vError> {
130        let n = self
131            .transport
132            .bulk_read(self.ep_in, &mut self.leader_buf, self.timeout)?;
133        parse_leader(&self.leader_buf[..n])
134    }
135
136    /// Read the full payload, looping over multiple USB bulk transfers
137    /// if the payload is larger than a single transfer.
138    fn read_payload(&mut self) -> Result<Bytes, U3vError> {
139        let mut offset = 0;
140        while offset < self.payload_size {
141            let n = self.transport.bulk_read(
142                self.ep_in,
143                &mut self.payload_buf[offset..self.payload_size],
144                self.timeout,
145            )?;
146            if n == 0 {
147                return Err(U3vError::Protocol(format!(
148                    "zero-length bulk read at payload offset {offset}/{}",
149                    self.payload_size
150                )));
151            }
152            offset += n;
153        }
154        Ok(Bytes::copy_from_slice(
155            &self.payload_buf[..self.payload_size],
156        ))
157    }
158
159    fn read_trailer(&mut self) -> Result<Trailer, U3vError> {
160        let n = self
161            .transport
162            .bulk_read(self.ep_in, &mut self.trailer_buf, self.timeout)?;
163        parse_trailer(&self.trailer_buf[..n])
164    }
165}
166
167/// A raw frame as received from the U3V stream, before conversion to
168/// the transport-agnostic `Frame` type.
169#[derive(Debug)]
170pub struct RawFrame {
171    /// Leader containing image metadata.
172    pub leader: Leader,
173    /// Raw pixel payload.
174    pub payload: Bytes,
175    /// Trailer with status and block ID.
176    pub trailer: Trailer,
177}
178
179// ---------------------------------------------------------------------------
180// Wire format parsing
181// ---------------------------------------------------------------------------
182
183/// Parse a U3V stream leader from raw bytes.
184///
185/// Leader layout (little-endian):
186/// ```text
187/// [0..4]   prefix        0x4C563355 ("U3VL")
188/// [4..6]   reserved
189/// [6..8]   leader_size   (total size in bytes)
190/// [8..10]  reserved
191/// [10..12] payload_type
192/// [12..20] timestamp     (device ticks)
193/// [20..24] pixel_format  (PFNC code)
194/// [24..28] width
195/// [28..32] height
196/// [32..36] x_offset
197/// [36..40] y_offset      (may be absent if leader is short)
198/// [40..42] x_padding
199/// ```
200pub fn parse_leader(buf: &[u8]) -> Result<Leader, U3vError> {
201    if buf.len() < MIN_LEADER_SIZE {
202        return Err(U3vError::Protocol(format!(
203            "leader too short: {} bytes, need at least {MIN_LEADER_SIZE}",
204            buf.len()
205        )));
206    }
207
208    let prefix = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
209    if prefix != LEADER_PREFIX {
210        return Err(U3vError::Protocol(format!(
211            "bad leader prefix: {prefix:#010x}, expected {LEADER_PREFIX:#010x}"
212        )));
213    }
214
215    let payload_type = u16::from_le_bytes([buf[10], buf[11]]);
216    let timestamp = u64::from_le_bytes(buf[12..20].try_into().unwrap());
217    let pixel_format = u32::from_le_bytes(buf[20..24].try_into().unwrap());
218    let width = u32::from_le_bytes(buf[24..28].try_into().unwrap());
219    let height = u32::from_le_bytes(buf[28..32].try_into().unwrap());
220    let x_offset = u32::from_le_bytes(buf[32..36].try_into().unwrap());
221
222    // Optional fields (may be absent in minimal leaders).
223    let y_offset = if buf.len() >= 40 {
224        u32::from_le_bytes(buf[36..40].try_into().unwrap())
225    } else {
226        0
227    };
228    let x_padding = if buf.len() >= 42 {
229        u16::from_le_bytes([buf[40], buf[41]])
230    } else {
231        0
232    };
233
234    Ok(Leader {
235        payload_type,
236        timestamp,
237        pixel_format,
238        width,
239        height,
240        x_offset,
241        y_offset,
242        x_padding,
243    })
244}
245
246/// Parse a U3V stream trailer from raw bytes.
247///
248/// Trailer layout (little-endian):
249/// ```text
250/// [0..4]   prefix              0x54563355 ("U3VT")
251/// [4..6]   reserved
252/// [6..8]   trailer_size
253/// [8..12]  status
254/// [12..20] block_id            (u64)
255/// [20..28] valid_payload_size  (u64, may be absent)
256/// ```
257pub fn parse_trailer(buf: &[u8]) -> Result<Trailer, U3vError> {
258    if buf.len() < MIN_TRAILER_SIZE {
259        return Err(U3vError::Protocol(format!(
260            "trailer too short: {} bytes, need at least {MIN_TRAILER_SIZE}",
261            buf.len()
262        )));
263    }
264
265    let prefix = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
266    if prefix != TRAILER_PREFIX {
267        return Err(U3vError::Protocol(format!(
268            "bad trailer prefix: {prefix:#010x}, expected {TRAILER_PREFIX:#010x}"
269        )));
270    }
271
272    let status = u32::from_le_bytes(buf[8..12].try_into().unwrap());
273    let block_id = if buf.len() >= 20 {
274        u64::from_le_bytes(buf[12..20].try_into().unwrap())
275    } else {
276        0
277    };
278    let valid_payload_size = if buf.len() >= 28 {
279        u64::from_le_bytes(buf[20..28].try_into().unwrap())
280    } else {
281        0
282    };
283
284    Ok(Trailer {
285        status,
286        block_id,
287        valid_payload_size,
288    })
289}
290
291// ---------------------------------------------------------------------------
292// Tests
293// ---------------------------------------------------------------------------
294
295#[cfg(test)]
296mod tests {
297    use super::*;
298    use crate::usb::MockUsbTransfer;
299
300    fn build_leader(width: u32, height: u32, pixel_format: u32, timestamp: u64) -> Vec<u8> {
301        let mut buf = vec![0u8; 42];
302        // Prefix
303        buf[0..4].copy_from_slice(&LEADER_PREFIX.to_le_bytes());
304        // leader_size
305        buf[6..8].copy_from_slice(&42u16.to_le_bytes());
306        // payload_type = image
307        buf[10..12].copy_from_slice(&0x0001u16.to_le_bytes());
308        // timestamp
309        buf[12..20].copy_from_slice(&timestamp.to_le_bytes());
310        // pixel_format
311        buf[20..24].copy_from_slice(&pixel_format.to_le_bytes());
312        // width
313        buf[24..28].copy_from_slice(&width.to_le_bytes());
314        // height
315        buf[28..32].copy_from_slice(&height.to_le_bytes());
316        // x_offset = 0, y_offset = 0, x_padding = 0
317        buf
318    }
319
320    fn build_trailer(status: u32, block_id: u64, valid_payload_size: u64) -> Vec<u8> {
321        let mut buf = vec![0u8; 28];
322        buf[0..4].copy_from_slice(&TRAILER_PREFIX.to_le_bytes());
323        buf[6..8].copy_from_slice(&28u16.to_le_bytes());
324        buf[8..12].copy_from_slice(&status.to_le_bytes());
325        buf[12..20].copy_from_slice(&block_id.to_le_bytes());
326        buf[20..28].copy_from_slice(&valid_payload_size.to_le_bytes());
327        buf
328    }
329
330    #[test]
331    fn parse_leader_valid() {
332        let data = build_leader(640, 480, 0x0108_0001, 12345);
333        let leader = parse_leader(&data).unwrap();
334        assert_eq!(leader.width, 640);
335        assert_eq!(leader.height, 480);
336        assert_eq!(leader.pixel_format, 0x0108_0001); // Mono8
337        assert_eq!(leader.timestamp, 12345);
338        assert_eq!(leader.payload_type, 0x0001);
339        assert_eq!(leader.x_offset, 0);
340        assert_eq!(leader.y_offset, 0);
341        assert_eq!(leader.x_padding, 0);
342    }
343
344    #[test]
345    fn parse_leader_bad_prefix() {
346        let mut data = build_leader(640, 480, 0, 0);
347        data[0] = 0xFF;
348        assert!(parse_leader(&data).is_err());
349    }
350
351    #[test]
352    fn parse_leader_too_short() {
353        let data = vec![0u8; 10];
354        assert!(parse_leader(&data).is_err());
355    }
356
357    #[test]
358    fn parse_trailer_valid() {
359        let data = build_trailer(0, 42, 307200);
360        let trailer = parse_trailer(&data).unwrap();
361        assert_eq!(trailer.status, 0);
362        assert_eq!(trailer.block_id, 42);
363        assert_eq!(trailer.valid_payload_size, 307200);
364    }
365
366    #[test]
367    fn parse_trailer_bad_prefix() {
368        let mut data = build_trailer(0, 0, 0);
369        data[0] = 0xFF;
370        assert!(parse_trailer(&data).is_err());
371    }
372
373    #[test]
374    fn parse_trailer_too_short() {
375        let data = vec![0u8; 8];
376        assert!(parse_trailer(&data).is_err());
377    }
378
379    #[test]
380    fn stream_next_frame() {
381        let mock = Arc::new(MockUsbTransfer::new());
382        let ep_in = 0x82;
383
384        let width = 320u32;
385        let height = 240u32;
386        let pixel_format = 0x0108_0001u32; // Mono8
387        let payload_size = (width * height) as usize;
388
389        // Enqueue leader
390        mock.enqueue_read(ep_in, build_leader(width, height, pixel_format, 99));
391        // Enqueue payload
392        mock.enqueue_read(ep_in, vec![0x80; payload_size]);
393        // Enqueue trailer
394        mock.enqueue_read(ep_in, build_trailer(0, 1, payload_size as u64));
395
396        let mut stream = U3vStream::new(Arc::clone(&mock), ep_in, 64, 64, payload_size);
397        let frame = stream.next_frame().unwrap();
398
399        assert_eq!(frame.leader.width, width);
400        assert_eq!(frame.leader.height, height);
401        assert_eq!(frame.leader.pixel_format, pixel_format);
402        assert_eq!(frame.leader.timestamp, 99);
403        assert_eq!(frame.payload.len(), payload_size);
404        assert_eq!(frame.trailer.status, 0);
405        assert_eq!(frame.trailer.block_id, 1);
406        assert_eq!(frame.trailer.valid_payload_size, payload_size as u64);
407    }
408
409    #[test]
410    fn stream_multiple_frames() {
411        let mock = Arc::new(MockUsbTransfer::new());
412        let ep_in = 0x82;
413
414        for i in 0..3u64 {
415            mock.enqueue_read(ep_in, build_leader(64, 64, 0x0108_0001, i * 1000));
416            mock.enqueue_read(ep_in, vec![0xAA; 64 * 64]);
417            mock.enqueue_read(ep_in, build_trailer(0, i, 64 * 64));
418        }
419
420        let mut stream = U3vStream::new(Arc::clone(&mock), ep_in, 64, 64, 64 * 64);
421
422        for i in 0..3u64 {
423            let frame = stream.next_frame().unwrap();
424            assert_eq!(frame.trailer.block_id, i);
425            assert_eq!(frame.leader.timestamp, i * 1000);
426        }
427    }
428
429    /// Payload split across multiple USB bulk transfers should be reassembled.
430    #[test]
431    fn stream_multi_transfer_payload() {
432        let mock = Arc::new(MockUsbTransfer::new());
433        let ep_in = 0x82;
434        let payload_size = 1024usize;
435
436        mock.enqueue_read(ep_in, build_leader(32, 32, 0x0108_0001, 42));
437        // Split payload into 4 chunks of 256 bytes each.
438        for chunk in 0..4u8 {
439            mock.enqueue_read(ep_in, vec![chunk; 256]);
440        }
441        mock.enqueue_read(ep_in, build_trailer(0, 1, payload_size as u64));
442
443        let mut stream = U3vStream::new(Arc::clone(&mock), ep_in, 64, 64, payload_size);
444        let frame = stream.next_frame().unwrap();
445
446        assert_eq!(frame.payload.len(), payload_size);
447        // Verify each 256-byte chunk has the correct fill byte.
448        for chunk in 0..4u8 {
449            let start = chunk as usize * 256;
450            assert!(
451                frame.payload[start..start + 256]
452                    .iter()
453                    .all(|&b| b == chunk),
454                "chunk {chunk} has wrong data"
455            );
456        }
457    }
458}
viva_u3v/stream.rs

viva_u3v/
stream.rs
