zencodec_media_testkit/
toy.rs

1//! Reference implementations: a toy container (TIVF: a 20-byte header and
2//! length-prefixed timed packets) and a toy codec (a packet holds 0..3 grey
3//! pictures and may re-present the previous one).
4//!
5//! They exercise every rule of the media contracts: output later than input
6//! and 0..N frames per packet, shared pictures, backpressure, drain, reset,
7//! configuration epochs, seeking with an index scan, and pulled muxer output
8//! with or without patches. The conformance checks are validated against
9//! them, as `zencodec-testkit`'s checks are against its reference codec.
10//!
11//! TIVF header: `b"TIVF"`, width `u16`, height `u16`, time base numerator
12//! `u32`, denominator `u32`, packet count `u32` (0 = unknown), all little
13//! endian. Packet: payload length `u32`, pts `i64`, flags `u8` (bit 0:
14//! random access), payload. Toy payload: picture count `u8` (0..=3),
15//! show-existing `u8`, then the grey planes.
16
17use alloc::collections::VecDeque;
18use alloc::sync::Arc;
19use alloc::vec;
20use alloc::vec::Vec;
21use core::cmp::Ordering;
22use core::fmt;
23
24use zencodec::encode::OutputBuffer;
25
26use zencodec::time::{TimeBase, Timestamp};
27use zencodec_media::container::{
28    AccessPattern, DemuxEvent, DemuxProgress, Demuxer, Input, InputRequest, MuxFinish, MuxHint,
29    MuxTiming, Muxer, Patch, SeekMode, SeekResult, SeekScope, SeekStatus, SeekTarget,
30};
31use zencodec_media::frame::{
32    ChromaSiting, Frame, FrameGeometry, FrameInterpretation, FrameTiming, OwnedPlane, OwnedPlanes,
33    PlaneData, PlaneRole, PlaneSamples, Presentation, SampleEncoding, Subsampling,
34};
35use zencodec_media::limits::MediaLimits;
36use zencodec_media::packet::{Packet, RandomAccess};
37use zencodec_media::session::{
38    CodecConfig, ProtocolViolation, Receive, SendStatus, SessionError, UnsupportedMedia,
39    VideoDecoder, VideoEncoder, VideoStreamInfo,
40};
41use zencodec_media::track::{CodecId, CodecTag, TrackInfo, TrackKind, VideoTrackInfo};
42
43/// The TIVF file signature.
44pub const MAGIC: &[u8; 4] = b"TIVF";
45/// TIVF header length.
46pub const HEADER_LEN: usize = 20;
47/// TIVF packet header length.
48pub const PACKET_HEADER_LEN: usize = 13;
49const COUNT_OFFSET: u64 = 16;
50
51/// Errors of the toy container and codec.
52#[derive(Clone, Copy, Debug, PartialEq, Eq)]
53#[non_exhaustive]
54pub enum ToyError {
55    /// Not a TIVF stream.
56    BadMagic,
57    /// The stream ends inside a record.
58    Truncated,
59    /// A zero time base.
60    BadTimeBase,
61    /// A malformed toy payload.
62    BadPayload,
63    /// No presentation satisfies the seek.
64    NoSuchFrame,
65    /// No such track.
66    NoSuchTrack,
67    /// Seek before the headers were read.
68    HeadersFirst,
69    /// Frame geometry differs from the configuration.
70    ShapeMismatch,
71    /// TIVF holds exactly one video track.
72    WrongTrackCount,
73}
74
75impl fmt::Display for ToyError {
76    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
77        f.write_str(match self {
78            Self::BadMagic => "not a TIVF stream",
79            Self::Truncated => "stream ends inside a record",
80            Self::BadTimeBase => "zero time base",
81            Self::BadPayload => "malformed toy packet payload",
82            Self::NoSuchFrame => "no presentation satisfies the seek",
83            Self::NoSuchTrack => "no such track",
84            Self::HeadersFirst => "headers have not been read",
85            Self::ShapeMismatch => "frame geometry does not match the configuration",
86            Self::WrongTrackCount => "toy container holds exactly one video track",
87        })
88    }
89}
90impl core::error::Error for ToyError {}
91
92fn u16_at(b: &[u8], i: usize) -> u16 {
93    u16::from_le_bytes([b[i], b[i + 1]])
94}
95fn u32_at(b: &[u8], i: usize) -> u32 {
96    u32::from_le_bytes([b[i], b[i + 1], b[i + 2], b[i + 3]])
97}
98fn i64_at(b: &[u8], i: usize) -> i64 {
99    let mut a = [0u8; 8];
100    a.copy_from_slice(&b[i..i + 8]);
101    i64::from_le_bytes(a)
102}
103
104#[derive(Clone, Copy, Debug)]
105struct IndexEntry {
106    offset: u64,
107    pts: i64,
108    random_access: bool,
109    ordinal: u64,
110}
111
112#[derive(Clone, Copy, Debug, PartialEq, Eq)]
113enum State {
114    Header,
115    Packets,
116    Scanning,
117    Done,
118}
119
120/// Sans-I/O TIVF demuxer. Sequential by default; `seek` builds a bounded
121/// index by scanning packet headers with `Range` requests.
122pub struct ToyDemuxer {
123    limits: MediaLimits,
124    state: State,
125    tracks: Vec<TrackInfo>,
126    requests: Vec<InputRequest>,
127    time_base: Option<TimeBase>,
128    pos: u64,
129    ordinal: u64,
130    declared: u32,
131    emitted: u64,
132    index: Vec<IndexEntry>,
133    index_complete: bool,
134    scan_pos: u64,
135    scan_ordinal: u64,
136    pending_seek: Option<SeekTarget>,
137}
138
139impl ToyDemuxer {
140    /// A demuxer enforcing `limits`.
141    pub fn new(limits: MediaLimits) -> Self {
142        Self {
143            limits,
144            state: State::Header,
145            tracks: Vec::new(),
146            requests: Vec::new(),
147            time_base: None,
148            pos: 0,
149            ordinal: 0,
150            declared: 0,
151            emitted: 0,
152            index: Vec::new(),
153            index_complete: false,
154            scan_pos: HEADER_LEN as u64,
155            scan_ordinal: 0,
156            pending_seek: None,
157        }
158    }
159
160    fn need(&mut self, r: InputRequest) -> Result<DemuxProgress, SessionError<ToyError>> {
161        self.requests.push(r);
162        Ok(DemuxProgress::need(0))
163    }
164
165    fn note(&mut self, entry: IndexEntry) {
166        if !self.index_complete && self.index.len() as u64 == entry.ordinal {
167            self.index.push(entry);
168            self.scan_pos = entry.offset;
169            self.scan_ordinal = entry.ordinal;
170        }
171    }
172
173    fn pick(&self, target: &SeekTarget) -> Result<IndexEntry, SessionError<ToyError>> {
174        let tb = self
175            .time_base
176            .ok_or(SessionError::Codec(ToyError::HeadersFirst))?;
177        let candidates = self
178            .index
179            .iter()
180            .filter(|e| target.scope == SeekScope::Any || e.random_access);
181        let t = target.time;
182        // Distance on a common denominator, exact in i128.
183        let value = |e: &IndexEntry| {
184            i128::from(e.pts) * i128::from(tb.numerator()) * i128::from(t.base().denominator())
185        };
186        let want =
187            i128::from(t.ticks()) * i128::from(t.base().numerator()) * i128::from(tb.denominator());
188        let chosen = match target.mode {
189            SeekMode::Nearest => candidates.min_by(|a, b| {
190                let da = (value(a) - want).abs();
191                let db = (value(b) - want).abs();
192                da.cmp(&db).then(a.pts.cmp(&b.pts))
193            }),
194            SeekMode::AtOrBefore => candidates
195                .filter(|e| value(e) <= want)
196                .max_by_key(|e| e.pts),
197            SeekMode::AtOrAfter => candidates
198                .filter(|e| value(e) >= want)
199                .min_by_key(|e| e.pts),
200            _ => return Err(SessionError::Unsupported(UnsupportedMedia::Seek)),
201        };
202        chosen
203            .copied()
204            .ok_or(SessionError::Codec(ToyError::NoSuchFrame))
205    }
206
207    fn position(&mut self, e: IndexEntry) -> SeekResult {
208        self.pos = e.offset;
209        self.ordinal = e.ordinal;
210        self.state = State::Packets;
211        SeekResult::new(
212            Timestamp::new(e.pts, self.time_base.expect("headers")),
213            e.ordinal,
214            if e.random_access {
215                RandomAccess::Yes
216            } else {
217                RandomAccess::No
218            },
219            0,
220        )
221    }
222}
223
224impl Demuxer for ToyDemuxer {
225    type Error = ToyError;
226
227    fn step(
228        &mut self,
229        input: Input<'_>,
230        stop: Option<&dyn zencodec::Pulse>,
231    ) -> Result<DemuxProgress, SessionError<ToyError>> {
232        let stop = stop.unwrap_or(&zencodec::NoPulse);
233        stop.check()?;
234        self.requests.clear();
235        match self.state {
236            State::Header => {
237                if input.offset != 0 {
238                    return self.need(InputRequest::at_least(0, HEADER_LEN));
239                }
240                let b = input.bytes;
241                if b.len() < HEADER_LEN {
242                    if input.reaches_end() {
243                        return Err(SessionError::Codec(ToyError::Truncated));
244                    }
245                    return self.need(InputRequest::at_least(0, HEADER_LEN));
246                }
247                if &b[0..4] != MAGIC {
248                    return Err(SessionError::Codec(ToyError::BadMagic));
249                }
250                let (w, h) = (u32::from(u16_at(b, 4)), u32::from(u16_at(b, 6)));
251                let tb = TimeBase::new(u32_at(b, 8), u32_at(b, 12))
252                    .map_err(|_| SessionError::Codec(ToyError::BadTimeBase))?;
253                self.declared = u32_at(b, 16);
254                let mut track = TrackInfo::video(
255                    0,
256                    CodecId::Other(CodecTag::new("toy")),
257                    tb,
258                    VideoTrackInfo::new(w, h),
259                );
260                if self.declared != 0 {
261                    track = track.with_declared_packets(u64::from(self.declared));
262                }
263                let track = track
264                    .build()
265                    .map_err(|_| SessionError::Codec(ToyError::ShapeMismatch))?;
266                self.tracks = vec![track];
267                self.time_base = Some(tb);
268                self.state = State::Packets;
269                self.pos = HEADER_LEN as u64;
270                Ok(DemuxProgress::new(HEADER_LEN, DemuxEvent::Headers))
271            }
272            State::Packets => {
273                if input.offset != self.pos {
274                    return self.need(InputRequest::at_least(self.pos, PACKET_HEADER_LEN));
275                }
276                let b = input.bytes;
277                if b.is_empty() && input.reaches_end() {
278                    self.state = State::Done;
279                    self.index_complete = true;
280                    if self.declared != 0 && self.emitted != u64::from(self.declared) {
281                        return Err(SessionError::Codec(ToyError::Truncated));
282                    }
283                    return Ok(DemuxProgress::new(0, DemuxEvent::Done));
284                }
285                if b.len() < PACKET_HEADER_LEN {
286                    if input.reaches_end() {
287                        return Err(SessionError::Codec(ToyError::Truncated));
288                    }
289                    return self.need(InputRequest::at_least(self.pos, PACKET_HEADER_LEN));
290                }
291                let len = u32_at(b, 0) as usize;
292                self.limits
293                    .check_packet_bytes(len as u64)
294                    .map_err(SessionError::Limit)?;
295                let total = PACKET_HEADER_LEN + len;
296                if b.len() < total {
297                    if input.reaches_end() {
298                        return Err(SessionError::Codec(ToyError::Truncated));
299                    }
300                    return self.need(InputRequest::at_least(self.pos, total));
301                }
302                let pts = i64_at(b, 4);
303                let flags = b[12];
304                let tb = self.time_base.expect("headers");
305                let packet = Packet::from_vec(b[PACKET_HEADER_LEN..total].to_vec())
306                    .map_err(|_| SessionError::Codec(ToyError::BadPayload))?
307                    .with_pts(Timestamp::new(pts, tb))
308                    .with_random_access(if flags & 1 != 0 {
309                        RandomAccess::Yes
310                    } else {
311                        RandomAccess::No
312                    })
313                    .with_track(0)
314                    .with_ordinal(self.ordinal)
315                    .with_input_offset(self.pos);
316                self.note(IndexEntry {
317                    offset: self.pos,
318                    pts,
319                    random_access: flags & 1 != 0,
320                    ordinal: self.ordinal,
321                });
322                self.pos += total as u64;
323                self.scan_pos = self.scan_pos.max(self.pos);
324                self.scan_ordinal = self.scan_ordinal.max(self.ordinal + 1);
325                self.ordinal += 1;
326                self.emitted += 1;
327                Ok(DemuxProgress::new(total, DemuxEvent::Packet(packet)))
328            }
329            State::Scanning => {
330                let Some(total) = input.total_len else {
331                    return self.need(InputRequest::Len);
332                };
333                if self.scan_pos >= total {
334                    self.index_complete = true;
335                    let target = self.pending_seek.take().expect("scanning implies a target");
336                    let entry = self.pick(&target)?;
337                    let result = self.position(entry);
338                    return Ok(DemuxProgress::new(0, DemuxEvent::Seeked(result)));
339                }
340                if input.offset != self.scan_pos || input.bytes.len() < PACKET_HEADER_LEN {
341                    if input.offset == self.scan_pos && input.reaches_end() {
342                        return Err(SessionError::Codec(ToyError::Truncated));
343                    }
344                    return self.need(InputRequest::range(self.scan_pos, PACKET_HEADER_LEN as u64));
345                }
346                let b = input.bytes;
347                let len = u32_at(b, 0) as u64;
348                let entry = IndexEntry {
349                    offset: self.scan_pos,
350                    pts: i64_at(b, 4),
351                    random_access: b[12] & 1 != 0,
352                    ordinal: self.scan_ordinal,
353                };
354                if self.index.len() as u64 == entry.ordinal {
355                    self.index.push(entry);
356                }
357                self.scan_pos += PACKET_HEADER_LEN as u64 + len;
358                self.scan_ordinal += 1;
359                if self.scan_pos >= total {
360                    // The scan reached the end: every packet is indexed.
361                    self.index_complete = true;
362                    let target = self.pending_seek.take().expect("scanning implies a target");
363                    let entry = self.pick(&target)?;
364                    let result = self.position(entry);
365                    return Ok(DemuxProgress::new(
366                        PACKET_HEADER_LEN,
367                        DemuxEvent::Seeked(result),
368                    ));
369                }
370                self.requests
371                    .push(InputRequest::range(self.scan_pos, PACKET_HEADER_LEN as u64));
372                Ok(DemuxProgress::need(PACKET_HEADER_LEN))
373            }
374            State::Done => Ok(DemuxProgress::new(0, DemuxEvent::Done)),
375        }
376    }
377
378    fn requested(&self) -> &[InputRequest] {
379        &self.requests
380    }
381
382    fn tracks(&self) -> &[TrackInfo] {
383        &self.tracks
384    }
385
386    fn access_pattern(&self) -> AccessPattern {
387        if self.state == State::Scanning {
388            AccessPattern::RandomAccess
389        } else {
390            AccessPattern::Sequential
391        }
392    }
393
394    fn seek(&mut self, target: &SeekTarget) -> Result<SeekStatus, SessionError<ToyError>> {
395        if self.state == State::Header {
396            return Err(SessionError::Codec(ToyError::HeadersFirst));
397        }
398        if target.track != 0 {
399            return Err(SessionError::Codec(ToyError::NoSuchTrack));
400        }
401        if self.index_complete {
402            let e = self.pick(target)?;
403            return Ok(SeekStatus::Positioned(self.position(e)));
404        }
405        self.pending_seek = Some(*target);
406        self.state = State::Scanning;
407        Ok(SeekStatus::Scanning)
408    }
409}
410
411/// TIVF muxer. `write_count` means the header's packet count is patched at
412/// finish, which a forward-only destination cannot accept.
413pub struct ToyMuxer {
414    out: OutputBuffer,
415    begun: bool,
416    write_count: bool,
417    time_base: Option<TimeBase>,
418    count: u32,
419}
420
421impl ToyMuxer {
422    /// A muxer; with `write_count`, the header's packet count is patched at
423    /// `finish`.
424    pub fn new(write_count: bool) -> Self {
425        Self {
426            out: OutputBuffer::new(),
427            begun: false,
428            write_count,
429            time_base: None,
430            count: 0,
431        }
432    }
433}
434
435impl Muxer for ToyMuxer {
436    type Error = ToyError;
437
438    fn begin(
439        &mut self,
440        tracks: &[TrackInfo],
441        hint: MuxHint,
442    ) -> Result<MuxTiming, SessionError<ToyError>> {
443        if self.begun {
444            return Err(SessionError::Protocol(ProtocolViolation::BeginTwice));
445        }
446        let [track] = tracks else {
447            return Err(SessionError::Codec(ToyError::WrongTrackCount));
448        };
449        let Some(video) = track
450            .video_info()
451            .filter(|_| track.kind() == TrackKind::Video)
452        else {
453            return Err(SessionError::Codec(ToyError::WrongTrackCount));
454        };
455        if self.write_count && hint.forward_only {
456            return Err(SessionError::Unsupported(
457                UnsupportedMedia::ForwardOnlyLayout,
458            ));
459        }
460        if video.coded_width > u32::from(u16::MAX) || video.coded_height > u32::from(u16::MAX) {
461            return Err(SessionError::Codec(ToyError::ShapeMismatch));
462        }
463        let tb = track.time_base();
464        self.out.push(MAGIC);
465        self.out.push(&(video.coded_width as u16).to_le_bytes());
466        self.out.push(&(video.coded_height as u16).to_le_bytes());
467        self.out.push(&tb.numerator().to_le_bytes());
468        self.out.push(&tb.denominator().to_le_bytes());
469        self.out.push(&0u32.to_le_bytes());
470        self.time_base = Some(tb);
471        self.begun = true;
472        Ok(if self.write_count {
473            MuxTiming::streaming_with_patches()
474        } else {
475            MuxTiming::streaming()
476        })
477    }
478
479    fn write_packet(
480        &mut self,
481        mut packet: Packet,
482        stop: Option<&dyn zencodec::Pulse>,
483    ) -> Result<(), SessionError<ToyError>> {
484        let stop = stop.unwrap_or(&zencodec::NoPulse);
485        stop.check()?;
486        if !self.begun {
487            return Err(SessionError::Protocol(ProtocolViolation::NoBegin));
488        }
489        let pts = packet
490            .pts()
491            .ok_or(SessionError::Protocol(ProtocolViolation::MissingTimestamp))?;
492        if Some(pts.base()) != self.time_base {
493            return Err(SessionError::Protocol(ProtocolViolation::ClockMismatch));
494        }
495        let payload = packet
496            .take()
497            .ok_or(SessionError::Protocol(ProtocolViolation::EmptyPacket))?;
498        let bytes = payload.as_bytes();
499        self.out.push(&(bytes.len() as u32).to_le_bytes());
500        self.out.push(&pts.ticks().to_le_bytes());
501        self.out
502            .push(&[u8::from(packet.random_access() == RandomAccess::Yes)]);
503        self.out.push(bytes);
504        self.count += 1;
505        Ok(())
506    }
507
508    fn pending_output(&self) -> &[u8] {
509        self.out.pending()
510    }
511    fn consume_output(&mut self, n: usize) {
512        self.out.consume(n)
513    }
514    fn take_output(&mut self, into: &mut Vec<u8>) {
515        self.out.take_into(into)
516    }
517
518    fn finish(
519        self,
520        stop: Option<&dyn zencodec::Pulse>,
521    ) -> Result<MuxFinish, SessionError<ToyError>> {
522        let stop = stop.unwrap_or(&zencodec::NoPulse);
523        stop.check()?;
524        if !self.begun {
525            return Err(SessionError::Protocol(ProtocolViolation::NoBegin));
526        }
527        let patches = if self.write_count {
528            vec![Patch {
529                offset: COUNT_OFFSET,
530                bytes: self.count.to_le_bytes().to_vec(),
531            }]
532        } else {
533            Vec::new()
534        };
535        Ok(MuxFinish::new(self.out.into_remaining(), patches))
536    }
537}
538
539/// Toy decoder. Pictures come out one packet late (a one-packet reorder),
540/// so output trails input; a show-existing flag re-presents the previous
541/// picture. Holds at most `max_queued_frames` before pushing back.
542pub struct ToyDecoder {
543    config: Option<CodecConfig>,
544    interpretation: Option<Arc<FrameInterpretation>>,
545    stream: Option<VideoStreamInfo>,
546    hold: VecDeque<Frame>,
547    ready: VecDeque<Frame>,
548    last: Option<Frame>,
549    ended: bool,
550    max_queued: usize,
551}
552
553impl ToyDecoder {
554    /// A decoder holding at most `max_queued_frames` (at least 1) frames.
555    pub fn new(max_queued_frames: usize) -> Self {
556        Self {
557            config: None,
558            interpretation: None,
559            stream: None,
560            hold: VecDeque::new(),
561            ready: VecDeque::new(),
562            last: None,
563            ended: false,
564            max_queued: max_queued_frames.max(1),
565        }
566    }
567}
568
569impl VideoDecoder for ToyDecoder {
570    type Error = ToyError;
571
572    fn configure(
573        &mut self,
574        config: &CodecConfig,
575        stop: Option<&dyn zencodec::Pulse>,
576    ) -> Result<(), SessionError<ToyError>> {
577        let stop = stop.unwrap_or(&zencodec::NoPulse);
578        stop.check()?;
579        let video = config
580            .video
581            .ok_or(SessionError::Unsupported(UnsupportedMedia::PixelFormat))?;
582        let interp = Arc::new(
583            FrameInterpretation::new(
584                FrameGeometry::full(video.coded_width, video.coded_height),
585                Subsampling::Mono,
586                ChromaSiting::Unknown,
587                SampleEncoding::U8,
588                vec![PlaneRole::Luma],
589            )
590            .with_config_epoch(config.epoch),
591        );
592        let mut stream = VideoStreamInfo::new(Arc::clone(&interp));
593        stream.reorder_depth = Some(1);
594        self.stream = Some(stream);
595        self.interpretation = Some(interp);
596        self.config = Some(config.clone());
597        Ok(())
598    }
599
600    fn send(
601        &mut self,
602        packet: &mut Packet,
603        stop: Option<&dyn zencodec::Pulse>,
604    ) -> Result<SendStatus, SessionError<ToyError>> {
605        let stop = stop.unwrap_or(&zencodec::NoPulse);
606        stop.check()?;
607        if self.ended {
608            return Err(SessionError::Protocol(ProtocolViolation::SendAfterEnd));
609        }
610        let (Some(config), Some(interp)) = (&self.config, &self.interpretation) else {
611            return Err(SessionError::Protocol(ProtocolViolation::NotConfigured));
612        };
613        if packet.is_taken() {
614            return Err(SessionError::Protocol(ProtocolViolation::EmptyPacket));
615        }
616        if packet.config_epoch() != config.epoch {
617            return Err(SessionError::Protocol(ProtocolViolation::EpochMismatch));
618        }
619        if !self.ready.is_empty() && self.ready.len() + self.hold.len() >= self.max_queued {
620            return Ok(SendStatus::ReceivePending);
621        }
622        let video = config.video.expect("configured with video");
623        let (w, h) = (video.coded_width as usize, video.coded_height as usize);
624        let b = packet.bytes();
625        if b.len() < 2 {
626            return Err(SessionError::Codec(ToyError::BadPayload));
627        }
628        let pictures = b[0] as usize;
629        let show_existing = b[1] != 0;
630        if pictures > 3 || b.len() != 2 + pictures * w * h {
631            return Err(SessionError::Codec(ToyError::BadPayload));
632        }
633        let mut timing =
634            FrameTiming::default().with_packet(packet.ordinal(), packet.input_offset());
635        timing.pts = packet.pts();
636        timing.duration_ticks = packet.duration_ticks();
637        // Pictures of the previous packet are released now: a one-packet reorder.
638        let released: Vec<Frame> = self.hold.drain(..).collect();
639        self.ready.extend(released);
640        for i in 0..pictures {
641            let start = 2 + i * w * h;
642            let data = b[start..start + w * h].to_vec();
643            let planes = OwnedPlanes::new(
644                vec![OwnedPlane {
645                    data: PlaneData::U8(data),
646                    width: w,
647                    height: h,
648                    stride_bytes: w,
649                }],
650                SampleEncoding::U8,
651            )
652            .map_err(|_| SessionError::Codec(ToyError::ShapeMismatch))?;
653            let mut presentation = Presentation::default();
654            presentation.keyframe = i == 0 && packet.random_access() == RandomAccess::Yes;
655            let frame = Frame::new(Arc::new(planes), Arc::clone(interp), timing, presentation);
656            self.last = Some(frame.clone());
657            self.hold.push_back(frame);
658        }
659        if show_existing {
660            let prev = self
661                .last
662                .as_ref()
663                .ok_or(SessionError::Codec(ToyError::BadPayload))?;
664            let again = prev.represent(timing, Presentation::SHOW_EXISTING);
665            self.hold.push_back(again);
666        }
667        packet.take();
668        Ok(SendStatus::Accepted)
669    }
670
671    fn receive(
672        &mut self,
673        stop: Option<&dyn zencodec::Pulse>,
674    ) -> Result<Receive<Frame>, SessionError<ToyError>> {
675        let stop = stop.unwrap_or(&zencodec::NoPulse);
676        stop.check()?;
677        if let Some(f) = self.ready.pop_front() {
678            return Ok(Receive::Ready(f));
679        }
680        if self.ended {
681            return Ok(match self.hold.pop_front() {
682                Some(f) => Receive::Ready(f),
683                None => Receive::EndOfStream,
684            });
685        }
686        Ok(Receive::NeedInput)
687    }
688
689    fn end_input(&mut self) -> Result<(), SessionError<ToyError>> {
690        self.ended = true;
691        Ok(())
692    }
693
694    fn reset(&mut self) -> Result<(), SessionError<ToyError>> {
695        self.hold.clear();
696        self.ready.clear();
697        self.last = None;
698        self.ended = false;
699        Ok(())
700    }
701
702    fn stream_info(&self) -> Option<&VideoStreamInfo> {
703        self.stream.as_ref()
704    }
705}
706
707/// Toy encoder: one gray picture per packet, `lookahead` frames held back,
708/// a random-access point every `keyint` packets.
709pub struct ToyEncoder {
710    config: CodecConfig,
711    lookahead: usize,
712    keyint: u64,
713    queue: VecDeque<Frame>,
714    out: VecDeque<Packet>,
715    max_out: usize,
716    ended: bool,
717    ordinal: u64,
718}
719
720impl ToyEncoder {
721    /// An encoder of `width` x `height` grey pictures on `time_base`,
722    /// holding `lookahead` frames and marking every `keyint`-th packet as a
723    /// random-access point.
724    pub fn new(
725        width: u32,
726        height: u32,
727        time_base: TimeBase,
728        lookahead: usize,
729        keyint: u64,
730    ) -> Self {
731        Self {
732            config: CodecConfig::new(CodecId::Other(CodecTag::new("toy")), time_base)
733                .with_video(VideoTrackInfo::new(width, height)),
734            lookahead,
735            keyint: keyint.max(1),
736            queue: VecDeque::new(),
737            out: VecDeque::new(),
738            max_out: 2,
739            ended: false,
740            ordinal: 0,
741        }
742    }
743
744    /// Frames held before their packet comes out.
745    pub fn lookahead(&self) -> usize {
746        self.lookahead
747    }
748
749    fn encode(&mut self, frame: &Frame) -> Result<Packet, SessionError<ToyError>> {
750        let video = self.config.video.expect("video config");
751        let (w, h) = (video.coded_width as usize, video.coded_height as usize);
752        let mapped = frame
753            .map()
754            .map_err(|_| SessionError::Codec(ToyError::ShapeMismatch))?;
755        let plane = mapped
756            .plane(0)
757            .ok_or(SessionError::Codec(ToyError::ShapeMismatch))?;
758        if plane.width() != w || plane.height() != h {
759            return Err(SessionError::Codec(ToyError::ShapeMismatch));
760        }
761        let mut payload = Vec::with_capacity(2 + w * h);
762        payload.push(1);
763        payload.push(0);
764        for y in 0..h {
765            match plane.row(y).expect("row in range") {
766                PlaneSamples::U8(r) => payload.extend_from_slice(r),
767                PlaneSamples::U16(_) => return Err(SessionError::Codec(ToyError::ShapeMismatch)),
768            }
769        }
770        let pts = frame
771            .timing()
772            .pts
773            .ok_or(SessionError::Protocol(ProtocolViolation::MissingTimestamp))?;
774        let mut p = Packet::from_vec(payload)
775            .map_err(|_| SessionError::Codec(ToyError::BadPayload))?
776            .with_track(0)
777            .with_pts(pts)
778            .with_ordinal(self.ordinal)
779            .with_random_access(if self.ordinal.is_multiple_of(self.keyint) {
780                RandomAccess::Yes
781            } else {
782                RandomAccess::No
783            });
784        if let Some(d) = frame.timing().duration_ticks {
785            p = p.with_duration_ticks(d);
786        }
787        self.ordinal += 1;
788        Ok(p)
789    }
790}
791
792impl VideoEncoder for ToyEncoder {
793    type Error = ToyError;
794
795    fn send(
796        &mut self,
797        frame: &Frame,
798        stop: Option<&dyn zencodec::Pulse>,
799    ) -> Result<SendStatus, SessionError<ToyError>> {
800        let stop = stop.unwrap_or(&zencodec::NoPulse);
801        stop.check()?;
802        if self.ended {
803            return Err(SessionError::Protocol(ProtocolViolation::SendAfterEnd));
804        }
805        let video = self.config.video.expect("video config");
806        let g = &frame.interpretation().geometry;
807        if g.coded_width != video.coded_width
808            || g.coded_height != video.coded_height
809            || frame.interpretation().subsampling != Subsampling::Mono
810        {
811            return Err(SessionError::Codec(ToyError::ShapeMismatch));
812        }
813        let pts = frame
814            .timing()
815            .pts
816            .ok_or(SessionError::Protocol(ProtocolViolation::MissingTimestamp))?;
817        if pts.base() != self.config.time_base {
818            return Err(SessionError::Protocol(ProtocolViolation::ClockMismatch));
819        }
820        if self.out.len() >= self.max_out {
821            return Ok(SendStatus::ReceivePending);
822        }
823        self.queue.push_back(frame.clone());
824        while self.queue.len() > self.lookahead {
825            let f = self.queue.pop_front().expect("nonempty");
826            let p = self.encode(&f)?;
827            self.out.push_back(p);
828        }
829        Ok(SendStatus::Accepted)
830    }
831
832    fn receive(
833        &mut self,
834        stop: Option<&dyn zencodec::Pulse>,
835    ) -> Result<Receive<Packet>, SessionError<ToyError>> {
836        let stop = stop.unwrap_or(&zencodec::NoPulse);
837        stop.check()?;
838        if let Some(p) = self.out.pop_front() {
839            return Ok(Receive::Ready(p));
840        }
841        if self.ended {
842            return Ok(match self.queue.pop_front() {
843                Some(f) => Receive::Ready(self.encode(&f)?),
844                None => Receive::EndOfStream,
845            });
846        }
847        Ok(Receive::NeedInput)
848    }
849
850    fn end_input(&mut self) -> Result<(), SessionError<ToyError>> {
851        self.ended = true;
852        Ok(())
853    }
854
855    fn output_config(&self) -> Option<&CodecConfig> {
856        Some(&self.config)
857    }
858}
859
860/// Build a toy packet payload.
861pub fn toy_payload(pictures: &[&[u8]], show_existing: bool) -> Vec<u8> {
862    let mut v = Vec::new();
863    v.push(pictures.len() as u8);
864    v.push(u8::from(show_existing));
865    for p in pictures {
866        v.extend_from_slice(p);
867    }
868    v
869}
870
871impl PartialOrd for IndexEntry {
872    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
873        Some(self.pts.cmp(&other.pts))
874    }
875}
876impl PartialEq for IndexEntry {
877    fn eq(&self, other: &Self) -> bool {
878        self.pts == other.pts
879    }
880}
zencodec_media_testkit/toy.rs

zencodec_media_testkit/
toy.rs
