oximedia-distributed 0.2.1

Distributed encoding coordinator for OxiMedia
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
//! Video segmentation and reassembly for distributed encoding.
//!
//! This module handles:
//! - Video segmentation strategies (time-based, tile-based, GOP-based)
//! - GOP (Group of Pictures) detection
//! - Segment overlap handling
//! - Reassembly and concatenation
//! - Bitrate normalization

#![allow(dead_code)]

use crate::{Result, SplitStrategy};
use std::path::{Path, PathBuf};
use std::time::Duration;
use tracing::{debug, info, warn};

/// Video segment representation
#[derive(Debug, Clone)]
pub struct VideoSegment {
    /// Segment identifier
    pub id: String,

    /// Segment index
    pub index: usize,

    /// Start time in seconds
    pub start_time: f64,

    /// End time in seconds
    pub end_time: f64,

    /// Duration
    pub duration: f64,

    /// Overlap with next segment
    pub overlap: f64,

    /// Source file path
    pub source_path: PathBuf,

    /// Output file path
    pub output_path: Option<PathBuf>,

    /// Segment type
    pub segment_type: SegmentType,
}

/// Segment type
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SegmentType {
    /// Time-based segment
    Time,
    /// Spatial tile
    Tile,
    /// GOP-aligned segment
    GOP,
}

/// Video segmenter
pub struct VideoSegmenter {
    /// Strategy for splitting
    strategy: SplitStrategy,

    /// Segment duration (for time-based)
    segment_duration: Duration,

    /// Overlap duration
    overlap_duration: Duration,

    /// GOP size hint
    gop_size: usize,

    /// Tile grid (width, height)
    tile_grid: (u32, u32),
}

impl VideoSegmenter {
    /// Create a new video segmenter
    #[must_use]
    pub fn new(strategy: SplitStrategy) -> Self {
        Self {
            strategy,
            segment_duration: Duration::from_secs(60),
            overlap_duration: Duration::from_millis(500),
            gop_size: 30,
            tile_grid: (2, 2),
        }
    }

    /// Set segment duration for time-based splitting
    #[must_use]
    pub fn with_duration(mut self, duration: Duration) -> Self {
        self.segment_duration = duration;
        self
    }

    /// Set overlap duration
    #[must_use]
    pub fn with_overlap(mut self, overlap: Duration) -> Self {
        self.overlap_duration = overlap;
        self
    }

    /// Set GOP size hint
    #[must_use]
    pub fn with_gop_size(mut self, gop_size: usize) -> Self {
        self.gop_size = gop_size;
        self
    }

    /// Set tile grid for spatial splitting
    #[must_use]
    pub fn with_tile_grid(mut self, width: u32, height: u32) -> Self {
        self.tile_grid = (width, height);
        self
    }

    /// Split video into segments
    pub async fn split_video(
        &self,
        source: &Path,
        video_info: &VideoInfo,
    ) -> Result<Vec<VideoSegment>> {
        match self.strategy {
            SplitStrategy::SegmentBased => self.split_by_time(source, video_info).await,
            SplitStrategy::TileBased => self.split_by_tiles(source, video_info).await,
            SplitStrategy::GopBased => self.split_by_gop(source, video_info).await,
        }
    }

    /// Split video by time segments
    async fn split_by_time(
        &self,
        source: &Path,
        video_info: &VideoInfo,
    ) -> Result<Vec<VideoSegment>> {
        info!("Splitting video by time segments");

        let mut segments = Vec::new();
        let total_duration = video_info.duration;
        let segment_duration = self.segment_duration.as_secs_f64();
        let overlap = self.overlap_duration.as_secs_f64();

        let num_segments = (total_duration / segment_duration).ceil() as usize;

        for i in 0..num_segments {
            let start_time = i as f64 * segment_duration;
            let mut end_time = (i + 1) as f64 * segment_duration;

            // Adjust last segment
            if end_time > total_duration {
                end_time = total_duration;
            }

            // Add overlap
            let actual_end_time = if i < num_segments - 1 {
                (end_time + overlap).min(total_duration)
            } else {
                end_time
            };

            let segment = VideoSegment {
                id: format!("seg_{i:04}"),
                index: i,
                start_time,
                end_time: actual_end_time,
                duration: actual_end_time - start_time,
                overlap,
                source_path: source.to_path_buf(),
                output_path: None,
                segment_type: SegmentType::Time,
            };

            segments.push(segment);
        }

        info!("Created {} time-based segments", segments.len());
        Ok(segments)
    }

    /// Split video by spatial tiles
    async fn split_by_tiles(
        &self,
        source: &Path,
        video_info: &VideoInfo,
    ) -> Result<Vec<VideoSegment>> {
        info!("Splitting video by spatial tiles");

        let mut segments = Vec::new();
        let (tile_cols, tile_rows) = self.tile_grid;

        let _tile_width = video_info.width / tile_cols;
        let _tile_height = video_info.height / tile_rows;

        for row in 0..tile_rows {
            for col in 0..tile_cols {
                let index = (row * tile_cols + col) as usize;

                // Tiles span the entire video duration
                let segment = VideoSegment {
                    id: format!("tile_{row}_{col}"),
                    index,
                    start_time: 0.0,
                    end_time: video_info.duration,
                    duration: video_info.duration,
                    overlap: 0.0,
                    source_path: source.to_path_buf(),
                    output_path: None,
                    segment_type: SegmentType::Tile,
                };

                segments.push(segment);
            }
        }

        info!(
            "Created {} tile-based segments ({}x{})",
            segments.len(),
            tile_cols,
            tile_rows
        );
        Ok(segments)
    }

    /// Split video by GOP boundaries
    async fn split_by_gop(
        &self,
        source: &Path,
        video_info: &VideoInfo,
    ) -> Result<Vec<VideoSegment>> {
        info!("Splitting video by GOP boundaries");

        // Detect GOP boundaries
        let gop_boundaries = self.detect_gop_boundaries(source, video_info).await?;

        let mut segments = Vec::new();
        let frame_rate = video_info.frame_rate;

        for (i, window) in gop_boundaries.windows(2).enumerate() {
            let start_frame = window[0];
            let end_frame = window[1];

            let start_time = start_frame as f64 / frame_rate;
            let end_time = end_frame as f64 / frame_rate;

            let segment = VideoSegment {
                id: format!("gop_{i:04}"),
                index: i,
                start_time,
                end_time,
                duration: end_time - start_time,
                overlap: 0.0,
                source_path: source.to_path_buf(),
                output_path: None,
                segment_type: SegmentType::GOP,
            };

            segments.push(segment);
        }

        info!("Created {} GOP-based segments", segments.len());
        Ok(segments)
    }

    /// Detect GOP boundaries in video
    async fn detect_gop_boundaries(
        &self,
        _source: &Path,
        video_info: &VideoInfo,
    ) -> Result<Vec<u64>> {
        debug!("Detecting GOP boundaries");

        // Simplified GOP detection - assume regular GOP structure
        let total_frames = (video_info.duration * video_info.frame_rate) as u64;
        let gop_size = self.gop_size as u64;

        let mut boundaries = vec![0];
        let mut current_frame = gop_size;

        while current_frame < total_frames {
            boundaries.push(current_frame);
            current_frame += gop_size;
        }

        boundaries.push(total_frames);

        debug!("Detected {} GOP boundaries", boundaries.len());
        Ok(boundaries)
    }
}

/// Video information
#[derive(Debug, Clone)]
pub struct VideoInfo {
    /// Video width
    pub width: u32,

    /// Video height
    pub height: u32,

    /// Duration in seconds
    pub duration: f64,

    /// Frame rate
    pub frame_rate: f64,

    /// Bitrate
    pub bitrate: u64,

    /// Codec
    pub codec: String,

    /// Total frames
    pub total_frames: u64,
}

impl VideoInfo {
    /// Create video info from basic parameters
    #[must_use]
    pub fn new(width: u32, height: u32, duration: f64, frame_rate: f64) -> Self {
        Self {
            width,
            height,
            duration,
            frame_rate,
            bitrate: 5_000_000, // Default 5 Mbps
            codec: "h264".to_string(),
            total_frames: (duration * frame_rate) as u64,
        }
    }

    /// Probe video file to get information
    pub async fn probe(_path: &Path) -> Result<Self> {
        // In production, would use FFprobe or similar
        // For now, return mock data
        Ok(Self {
            width: 1920,
            height: 1080,
            duration: 600.0, // 10 minutes
            frame_rate: 30.0,
            bitrate: 5_000_000,
            codec: "h264".to_string(),
            total_frames: 18000,
        })
    }
}

/// Segment reassembler for combining encoded segments
pub struct SegmentReassembler {
    /// Concatenation strategy
    strategy: ConcatenationStrategy,

    /// Bitrate normalization
    normalize_bitrate: bool,

    /// Target bitrate for normalization
    target_bitrate: Option<u64>,
}

impl SegmentReassembler {
    /// Create a new segment reassembler
    #[must_use]
    pub fn new() -> Self {
        Self {
            strategy: ConcatenationStrategy::Concat,
            normalize_bitrate: true,
            target_bitrate: None,
        }
    }

    /// Set concatenation strategy
    #[must_use]
    pub fn with_strategy(mut self, strategy: ConcatenationStrategy) -> Self {
        self.strategy = strategy;
        self
    }

    /// Enable bitrate normalization
    #[must_use]
    pub fn with_normalization(mut self, enable: bool) -> Self {
        self.normalize_bitrate = enable;
        self
    }

    /// Set target bitrate
    #[must_use]
    pub fn with_target_bitrate(mut self, bitrate: u64) -> Self {
        self.target_bitrate = Some(bitrate);
        self
    }

    /// Reassemble segments into final output
    pub async fn reassemble(
        &self,
        segments: &[VideoSegment],
        output: &Path,
    ) -> Result<ReassemblyResult> {
        info!("Reassembling {} segments", segments.len());

        match self.strategy {
            ConcatenationStrategy::Concat => self.concat_segments(segments, output).await,
            ConcatenationStrategy::Blend => self.blend_segments(segments, output).await,
            ConcatenationStrategy::Stitch => self.stitch_tiles(segments, output).await,
        }
    }

    /// Concatenate time-based segments
    async fn concat_segments(
        &self,
        segments: &[VideoSegment],
        output: &Path,
    ) -> Result<ReassemblyResult> {
        info!("Concatenating segments");

        // Sort segments by index
        let mut sorted_segments = segments.to_vec();
        sorted_segments.sort_by_key(|s| s.index);

        // Handle overlaps
        let processed_segments = if sorted_segments.iter().any(|s| s.overlap > 0.0) {
            self.trim_overlaps(&sorted_segments)?
        } else {
            sorted_segments
        };

        // In production, would use FFmpeg concat demuxer
        debug!(
            "Concatenating {} segments to {:?}",
            processed_segments.len(),
            output
        );

        Ok(ReassemblyResult {
            output_path: output.to_path_buf(),
            total_duration: processed_segments.iter().map(|s| s.duration).sum(),
            num_segments: processed_segments.len(),
            final_bitrate: self.target_bitrate.unwrap_or(5_000_000),
        })
    }

    /// Blend segments with overlaps
    async fn blend_segments(
        &self,
        segments: &[VideoSegment],
        output: &Path,
    ) -> Result<ReassemblyResult> {
        info!("Blending segments with overlaps");

        // Sort segments
        let mut sorted_segments = segments.to_vec();
        sorted_segments.sort_by_key(|s| s.index);

        // Process overlapping regions with cross-fade
        for window in sorted_segments.windows(2) {
            let seg1 = &window[0];
            let seg2 = &window[1];

            if seg1.overlap > 0.0 {
                debug!(
                    "Blending overlap between {} and {} ({:.2}s)",
                    seg1.id, seg2.id, seg1.overlap
                );
                // In production, apply cross-fade filter
            }
        }

        Ok(ReassemblyResult {
            output_path: output.to_path_buf(),
            total_duration: segments.iter().map(|s| s.duration - s.overlap).sum::<f64>()
                + segments.last().map_or(0.0, |s| s.overlap),
            num_segments: segments.len(),
            final_bitrate: self.target_bitrate.unwrap_or(5_000_000),
        })
    }

    /// Stitch spatial tiles back together
    async fn stitch_tiles(
        &self,
        segments: &[VideoSegment],
        output: &Path,
    ) -> Result<ReassemblyResult> {
        info!("Stitching {} tiles", segments.len());

        // In production, would use FFmpeg xstack filter
        debug!("Stitching tiles to {:?}", output);

        Ok(ReassemblyResult {
            output_path: output.to_path_buf(),
            total_duration: segments.first().map_or(0.0, |s| s.duration),
            num_segments: segments.len(),
            final_bitrate: self.target_bitrate.unwrap_or(5_000_000),
        })
    }

    /// Trim overlapping regions from segments
    fn trim_overlaps(&self, segments: &[VideoSegment]) -> Result<Vec<VideoSegment>> {
        let mut trimmed = Vec::new();

        for (i, segment) in segments.iter().enumerate() {
            let mut seg = segment.clone();

            // Trim the overlap from the end (except last segment)
            if i < segments.len() - 1 {
                seg.duration -= seg.overlap;
                seg.end_time -= seg.overlap;
            }

            trimmed.push(seg);
        }

        Ok(trimmed)
    }

    /// Normalize bitrates across segments
    pub fn normalize_bitrates(&self, segments: &[VideoSegment]) -> Result<Vec<BitrateAdjustment>> {
        if !self.normalize_bitrate {
            return Ok(Vec::new());
        }

        let target = self.target_bitrate.unwrap_or(5_000_000);
        info!("Normalizing bitrates to {} bps", target);

        // Create adjustments for each segment
        let adjustments: Vec<BitrateAdjustment> = segments
            .iter()
            .map(|s| BitrateAdjustment {
                segment_id: s.id.clone(),
                original_bitrate: 5_000_000, // Would be detected from encoded file
                target_bitrate: target,
                adjustment_factor: target as f64 / 5_000_000.0,
            })
            .collect();

        Ok(adjustments)
    }
}

impl Default for SegmentReassembler {
    fn default() -> Self {
        Self::new()
    }
}

/// Concatenation strategy
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConcatenationStrategy {
    /// Simple concatenation
    Concat,
    /// Blend overlapping regions
    Blend,
    /// Stitch spatial tiles
    Stitch,
}

/// Reassembly result
#[derive(Debug, Clone)]
pub struct ReassemblyResult {
    /// Output file path
    pub output_path: PathBuf,

    /// Total duration
    pub total_duration: f64,

    /// Number of segments
    pub num_segments: usize,

    /// Final bitrate
    pub final_bitrate: u64,
}

/// Bitrate adjustment information
#[derive(Debug, Clone)]
pub struct BitrateAdjustment {
    pub segment_id: String,
    pub original_bitrate: u64,
    pub target_bitrate: u64,
    pub adjustment_factor: f64,
}

/// GOP (Group of Pictures) detector
pub struct GOPDetector {
    /// Minimum GOP size
    min_gop_size: usize,

    /// Maximum GOP size
    max_gop_size: usize,

    /// Scene change threshold
    scene_threshold: f64,
}

impl GOPDetector {
    /// Create a new GOP detector
    #[must_use]
    pub fn new() -> Self {
        Self {
            min_gop_size: 10,
            max_gop_size: 300,
            scene_threshold: 0.3,
        }
    }

    /// Set GOP size constraints
    #[must_use]
    pub fn with_gop_range(mut self, min: usize, max: usize) -> Self {
        self.min_gop_size = min;
        self.max_gop_size = max;
        self
    }

    /// Set scene change threshold
    #[must_use]
    pub fn with_scene_threshold(mut self, threshold: f64) -> Self {
        self.scene_threshold = threshold;
        self
    }

    /// Detect GOP structure in video
    pub async fn detect_gops(&self, _video_path: &Path) -> Result<Vec<GOPInfo>> {
        // In production, would analyze video frames
        // For now, return mock data
        let gops = vec![
            GOPInfo {
                start_frame: 0,
                end_frame: 30,
                keyframe_positions: vec![0],
                num_p_frames: 29,
                num_b_frames: 0,
            },
            GOPInfo {
                start_frame: 30,
                end_frame: 60,
                keyframe_positions: vec![30],
                num_p_frames: 29,
                num_b_frames: 0,
            },
        ];

        Ok(gops)
    }

    /// Validate GOP alignment for segments
    pub fn validate_alignment(&self, segments: &[VideoSegment], gops: &[GOPInfo]) -> Result<bool> {
        for segment in segments {
            if segment.segment_type != SegmentType::GOP {
                continue;
            }

            // Check if segment boundaries align with GOPs
            let start_aligned = gops
                .iter()
                .any(|g| (g.start_frame as f64 - segment.start_time * 30.0).abs() < 0.1);

            let end_aligned = gops
                .iter()
                .any(|g| (g.end_frame as f64 - segment.end_time * 30.0).abs() < 0.1);

            if !start_aligned || !end_aligned {
                warn!("Segment {} not aligned with GOP boundaries", segment.id);
                return Ok(false);
            }
        }

        Ok(true)
    }
}

impl Default for GOPDetector {
    fn default() -> Self {
        Self::new()
    }
}

/// GOP information
#[derive(Debug, Clone)]
pub struct GOPInfo {
    /// Start frame
    pub start_frame: u64,

    /// End frame
    pub end_frame: u64,

    /// Keyframe positions
    pub keyframe_positions: Vec<u64>,

    /// Number of P-frames
    pub num_p_frames: usize,

    /// Number of B-frames
    pub num_b_frames: usize,
}

impl GOPInfo {
    /// Get GOP size
    #[must_use]
    pub fn size(&self) -> u64 {
        self.end_frame - self.start_frame
    }

    /// Check if frame is a keyframe
    #[must_use]
    pub fn is_keyframe(&self, frame: u64) -> bool {
        self.keyframe_positions.contains(&frame)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    #[test]
    fn test_time_based_segmentation() {
        let segmenter = VideoSegmenter::new(SplitStrategy::SegmentBased)
            .with_duration(Duration::from_secs(60))
            .with_overlap(Duration::from_millis(500));

        assert_eq!(segmenter.segment_duration.as_secs(), 60);
        assert_eq!(segmenter.overlap_duration.as_millis(), 500);
    }

    #[test]
    fn test_video_info_creation() {
        let info = VideoInfo::new(1920, 1080, 600.0, 30.0);
        assert_eq!(info.width, 1920);
        assert_eq!(info.height, 1080);
        assert_eq!(info.duration, 600.0);
        assert_eq!(info.total_frames, 18000);
    }

    #[test]
    fn test_segment_reassembler() {
        let reassembler = SegmentReassembler::new()
            .with_normalization(true)
            .with_target_bitrate(5_000_000);

        assert!(reassembler.normalize_bitrate);
        assert_eq!(reassembler.target_bitrate, Some(5_000_000));
    }

    #[test]
    fn test_gop_detector() {
        let detector = GOPDetector::new().with_gop_range(10, 300);

        assert_eq!(detector.min_gop_size, 10);
        assert_eq!(detector.max_gop_size, 300);
    }

    #[test]
    fn test_gop_info() {
        let gop = GOPInfo {
            start_frame: 0,
            end_frame: 30,
            keyframe_positions: vec![0, 15, 30],
            num_p_frames: 27,
            num_b_frames: 0,
        };

        assert_eq!(gop.size(), 30);
        assert!(gop.is_keyframe(0));
        assert!(gop.is_keyframe(15));
        assert!(!gop.is_keyframe(10));
    }

    #[test]
    fn test_concatenation_strategies() {
        assert_eq!(ConcatenationStrategy::Concat, ConcatenationStrategy::Concat);
        assert_ne!(ConcatenationStrategy::Concat, ConcatenationStrategy::Blend);
    }
}