pixelflow-filters 0.1.0

//! Select filter validation and executor.

use std::collections::BTreeSet;

use pixelflow_core::{
    Clip, ClipMedia, ConcurrencyClass, DependencyPattern, DynamicDependencyBounds, ErrorCategory,
    ErrorCode, FilterChangeSet, FilterCompatibility, FilterOptionValue, FilterOptions, FilterPlan,
    FilterPlanRequest, Frame, FrameCount, FrameExecutor, FrameRate, FrameRequest, GraphBuilder,
    PixelFlowError, Rational, Result,
};

use crate::{
    OfficialFilterContract, SupportedFormats,
    contracts::{ALL_PLANAR_FAMILIES, ALL_SAMPLE_TYPES},
    options::{single_input_media, single_input_slice},
};

/// Official select filter script name.
pub const FILTER_SELECT: &str = "select";

/// Contract for `select`.
pub const SELECT_CONTRACT: OfficialFilterContract = OfficialFilterContract::new(
    FILTER_SELECT,
    SupportedFormats::new(ALL_PLANAR_FAMILIES, ALL_SAMPLE_TYPES),
    FilterCompatibility::AllowChanges(FilterChangeSet {
        format: false,
        resolution: false,
        frame_count: true,
        frame_rate: true,
    }),
);

/// Generic select options from script or embedding callers.
pub type SelectOptions = FilterOptions;

/// Select mapping validated against finite input media.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ValidatedSelect {
    source_frames: Vec<usize>,
}

impl ValidatedSelect {
    /// Creates validated output-to-source mapping.
    #[must_use]
    pub const fn new(source_frames: Vec<usize>) -> Self {
        Self { source_frames }
    }

    /// Returns output-to-source frame mapping.
    #[must_use]
    pub fn source_frames(&self) -> &[usize] {
        &self.source_frames
    }

    fn source_frame(&self, output_frame: usize) -> Result<usize> {
        self.source_frames
            .get(output_frame)
            .copied()
            .ok_or_else(|| {
                select_error(
                    "filter.select_frame_out_of_range",
                    format!("filter '{FILTER_SELECT}' output frame {output_frame} is out of range"),
                )
            })
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum SelectRate {
    DivideBy(usize),
    DropEvery(usize),
    Preserve,
}

/// Validates select options and derives output media.
pub fn select_output_media(
    input: &ClipMedia,
    options: &SelectOptions,
) -> Result<(ValidatedSelect, ClipMedia)> {
    SELECT_CONTRACT.validate_input_media(input)?;
    reject_unsupported_options(options)?;

    let input_frames = finite_frame_count(input)?;
    let (source_frames, rate) = selected_source_frames(options, input_frames)?;
    let output_media = ClipMedia::new(
        input.format().clone(),
        input.resolution().clone(),
        FrameCount::Finite(source_frames.len()),
        select_frame_rate(input.frame_rate(), rate)?,
    );

    Ok((ValidatedSelect::new(source_frames), output_media))
}

pub(crate) fn plan_select(request: FilterPlanRequest<'_>) -> Result<FilterPlan> {
    let input = single_input_media(request, FILTER_SELECT)?;
    let output_media = select_output_media(input, request.options())?.1;

    Ok(
        FilterPlan::new(output_media, SELECT_CONTRACT.compatibility()).with_schedule(
            DependencyPattern::frame_map(DynamicDependencyBounds::any()),
            ConcurrencyClass::Stateless,
        ),
    )
}

pub(crate) fn select_executor_from_options(
    input_media: &[ClipMedia],
    options: &SelectOptions,
) -> Result<SelectExecutor> {
    let input = single_input_slice(input_media, FILTER_SELECT)?;
    let (selection, _media) = select_output_media(input, options)?;

    Ok(SelectExecutor::new(selection))
}

/// Adds validated select filter node and returns matching render executor.
pub fn add_select_filter(
    builder: &mut GraphBuilder,
    input: Clip,
    input_media: &ClipMedia,
    options: &SelectOptions,
) -> Result<(Clip, SelectExecutor)> {
    let (selection, output_media) = select_output_media(input_media, options)?;
    let output = builder.filter_with_schedule(
        FILTER_SELECT,
        &[input],
        output_media,
        SELECT_CONTRACT.compatibility(),
        DependencyPattern::frame_map(DynamicDependencyBounds::any()),
        ConcurrencyClass::Stateless,
    )?;

    Ok((output, SelectExecutor::new(selection)))
}

/// Render executor for validated temporal select operations.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SelectExecutor {
    selection: ValidatedSelect,
}

impl SelectExecutor {
    /// Creates select executor from validated mapping.
    #[must_use]
    pub const fn new(selection: ValidatedSelect) -> Self {
        Self { selection }
    }

    /// Returns validated output-to-source mapping.
    #[must_use]
    pub const fn selection(&self) -> &ValidatedSelect {
        &self.selection
    }
}

impl FrameExecutor for SelectExecutor {
    fn prepare(&self, request: FrameRequest<'_>) -> Result<Frame> {
        let source_frame = self.selection.source_frame(request.frame_number())?;
        request.input_frame(0, source_frame)
    }
}

fn reject_unsupported_options(options: &SelectOptions) -> Result<()> {
    const ALLOWED: [&str; 5] = ["every", "offset", "drop_every", "drop_offset", "frames"];
    const PREDICATE_KEYS: [&str; 3] = ["predicate", "where", "condition"];

    for key in options.keys() {
        if PREDICATE_KEYS.contains(&key.as_str()) {
            return Err(select_error(
                "filter.unsupported_select_predicate",
                format!("filter '{FILTER_SELECT}' predicate-based selection is not supported"),
            ));
        }
        if !ALLOWED.contains(&key.as_str()) {
            return Err(select_error(
                "filter.invalid_select",
                format!("filter '{FILTER_SELECT}' option '{key}' is not supported"),
            ));
        }
    }

    Ok(())
}

fn finite_frame_count(input: &ClipMedia) -> Result<usize> {
    match input.frame_count() {
        FrameCount::Finite(frames) => Ok(frames),
        FrameCount::Unknown => Err(select_error(
            "filter.variable_frame_count",
            format!("filter '{FILTER_SELECT}' requires finite input frame count"),
        )),
    }
}

fn selected_source_frames(
    options: &SelectOptions,
    input_frames: usize,
) -> Result<(Vec<usize>, SelectRate)> {
    let every = option_usize(options, "every")?;
    let drop_every = option_usize(options, "drop_every")?;
    let frames = option_string(options, "frames")?;
    let forms = usize::from(every.is_some())
        + usize::from(drop_every.is_some())
        + usize::from(frames.is_some());

    if forms != 1 {
        return Err(select_error(
            "filter.invalid_select",
            format!(
                "filter '{FILTER_SELECT}' requires exactly one of 'every', 'drop_every', or 'frames'"
            ),
        ));
    }

    if let Some(interval) = every {
        let interval = non_zero("every", interval)?;
        reject_form_option(options, "drop_offset", "every")?;
        let offset = option_usize(options, "offset")?.unwrap_or(0);
        if offset >= interval {
            return Err(select_error(
                "filter.invalid_select",
                format!("filter '{FILTER_SELECT}' option 'offset' must be less than 'every'"),
            ));
        }
        return Ok((
            select_every(input_frames, interval, offset),
            SelectRate::DivideBy(interval),
        ));
    }

    if let Some(interval) = drop_every {
        let interval = minimum_two("drop_every", interval)?;
        reject_form_option(options, "offset", "drop_every")?;
        let offset = option_usize(options, "drop_offset")?.unwrap_or(interval - 1);
        if offset >= interval {
            return Err(select_error(
                "filter.invalid_select",
                format!(
                    "filter '{FILTER_SELECT}' option 'drop_offset' must be less than 'drop_every'"
                ),
            ));
        }
        return Ok((
            select_drop_every(input_frames, interval, offset),
            SelectRate::DropEvery(interval),
        ));
    }

    let expression = frames.expect("validated forms include frames expression");
    reject_form_option(options, "offset", "frames")?;
    reject_form_option(options, "drop_offset", "frames")?;
    Ok((
        parse_frame_expression(expression, input_frames)?,
        SelectRate::Preserve,
    ))
}

fn reject_form_option(options: &SelectOptions, option: &str, form: &str) -> Result<()> {
    if options.contains_key(option) {
        return Err(select_error(
            "filter.invalid_select",
            format!("filter '{FILTER_SELECT}' option '{option}' cannot be used with '{form}'"),
        ));
    }

    Ok(())
}

fn option_usize(options: &SelectOptions, name: &str) -> Result<Option<usize>> {
    match options.get(name) {
        None | Some(FilterOptionValue::None) => Ok(None),
        Some(FilterOptionValue::Int(value)) => usize::try_from(*value).map(Some).map_err(|_| {
            select_error(
                "filter.invalid_select",
                format!("filter '{FILTER_SELECT}' option '{name}' must be non-negative"),
            )
        }),
        Some(_) => Err(select_error(
            "filter.invalid_select",
            format!("filter '{FILTER_SELECT}' option '{name}' must be integer"),
        )),
    }
}

fn option_string<'a>(options: &'a SelectOptions, name: &str) -> Result<Option<&'a str>> {
    match options.get(name) {
        None | Some(FilterOptionValue::None) => Ok(None),
        Some(FilterOptionValue::String(value)) => Ok(Some(value.as_str())),
        Some(_) => Err(select_error(
            "filter.invalid_select",
            format!("filter '{FILTER_SELECT}' option '{name}' must be string"),
        )),
    }
}

fn non_zero(name: &str, value: usize) -> Result<usize> {
    if value == 0 {
        return Err(select_error(
            "filter.invalid_select",
            format!("filter '{FILTER_SELECT}' option '{name}' must be positive"),
        ));
    }

    Ok(value)
}

fn minimum_two(name: &str, value: usize) -> Result<usize> {
    if value < 2 {
        return Err(select_error(
            "filter.invalid_select",
            format!("filter '{FILTER_SELECT}' option '{name}' must be greater than 1"),
        ));
    }

    Ok(value)
}

fn select_every(input_frames: usize, interval: usize, offset: usize) -> Vec<usize> {
    let mut selected = Vec::new();
    let mut frame = offset;

    while frame < input_frames {
        selected.push(frame);
        let Some(next) = frame.checked_add(interval) else {
            break;
        };
        frame = next;
    }

    selected
}

fn select_drop_every(input_frames: usize, interval: usize, offset: usize) -> Vec<usize> {
    (0..input_frames)
        .filter(|frame| frame % interval != offset)
        .collect()
}

fn parse_frame_expression(expression: &str, input_frames: usize) -> Result<Vec<usize>> {
    if expression.trim().is_empty() {
        return Err(select_error(
            "filter.invalid_select",
            format!("filter '{FILTER_SELECT}' option 'frames' must not be empty"),
        ));
    }

    let mut selected = Vec::new();
    let mut seen = BTreeSet::new();

    for token in expression.split(',') {
        let token = token.trim();
        if token.is_empty() {
            return Err(select_error(
                "filter.invalid_select",
                format!("filter '{FILTER_SELECT}' option 'frames' contains an empty entry"),
            ));
        }

        for frame in parse_frame_token(token)? {
            if frame >= input_frames {
                return Err(select_error(
                    "filter.select_frame_out_of_range",
                    format!(
                        "filter '{FILTER_SELECT}' frame {frame} exceeds input frame count {input_frames}"
                    ),
                ));
            }
            if !seen.insert(frame) {
                return Err(select_error(
                    "filter.duplicate_select_frame",
                    format!("filter '{FILTER_SELECT}' frame {frame} appears more than once"),
                ));
            }
            selected.push(frame);
        }
    }

    Ok(selected)
}

fn parse_frame_token(token: &str) -> Result<Vec<usize>> {
    if let Some((start, end)) = token.split_once("..=") {
        return expand_range(parse_frame_number(start)?, parse_frame_number(end)?);
    }
    if let Some((start, end)) = token.split_once("..") {
        return expand_range(parse_frame_number(start)?, parse_frame_number(end)?);
    }
    if let Some((start, end)) = token.split_once('-') {
        return expand_range(parse_frame_number(start)?, parse_frame_number(end)?);
    }

    Ok(vec![parse_frame_number(token)?])
}

fn parse_frame_number(value: &str) -> Result<usize> {
    value.trim().parse::<usize>().map_err(|_| {
        select_error(
            "filter.invalid_select",
            format!("filter '{FILTER_SELECT}' frame expression contains invalid number '{value}'"),
        )
    })
}

fn expand_range(start: usize, end: usize) -> Result<Vec<usize>> {
    if start > end {
        return Err(select_error(
            "filter.invalid_select",
            format!("filter '{FILTER_SELECT}' frame ranges must be ascending"),
        ));
    }

    Ok((start..=end).collect())
}

fn select_frame_rate(input_rate: FrameRate, rate: SelectRate) -> Result<FrameRate> {
    let input_rate = match input_rate {
        FrameRate::Cfr(rate) if rate.numerator > 0 && rate.denominator > 0 => rate,
        FrameRate::Cfr(_) => {
            return Err(select_error(
                "filter.invalid_select",
                format!("filter '{FILTER_SELECT}' requires positive input frame rate"),
            ));
        }
        FrameRate::Unknown => {
            return Err(select_error(
                "filter.variable_frame_rate",
                format!("filter '{FILTER_SELECT}' requires constant input frame rate"),
            ));
        }
    };

    match rate {
        SelectRate::Preserve => Ok(FrameRate::Cfr(input_rate)),
        SelectRate::DivideBy(interval) => Ok(FrameRate::Cfr(Rational {
            numerator: input_rate.numerator,
            denominator: checked_rate_mul(input_rate.denominator, interval)?,
        })),
        SelectRate::DropEvery(interval) => Ok(FrameRate::Cfr(Rational {
            numerator: checked_rate_mul(input_rate.numerator, interval - 1)?,
            denominator: checked_rate_mul(input_rate.denominator, interval)?,
        })),
    }
}

fn checked_rate_mul(value: i64, factor: usize) -> Result<i64> {
    let factor = i64::try_from(factor).map_err(|_| {
        select_error(
            "filter.invalid_select",
            format!("filter '{FILTER_SELECT}' frame-rate multiplier is too large"),
        )
    })?;

    value.checked_mul(factor).ok_or_else(|| {
        select_error(
            "filter.invalid_select",
            format!("filter '{FILTER_SELECT}' frame-rate calculation overflowed"),
        )
    })
}

fn select_error(code: &'static str, message: impl Into<String>) -> PixelFlowError {
    PixelFlowError::new(ErrorCategory::Format, ErrorCode::new(code), message)
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use pixelflow_core::{
        ClipMedia, DependencyPattern, DynamicDependencyBounds, ErrorCategory, ErrorCode,
        FilterOptionValue, Frame, FrameCount, FrameExecutor, FrameRate, FrameRequest, GraphBuilder,
        MetadataValue, NodeKind, Rational, RenderEngine, RenderExecutorMap, RenderOptions,
        WorkerPoolConfig,
    };

    use crate::testkit::{fixed_media, synthetic_u8_frame, with_frame_number_metadata};

    use super::{SelectOptions, select_output_media};

    fn options(
        entries: impl IntoIterator<Item = (&'static str, FilterOptionValue)>,
    ) -> SelectOptions {
        entries
            .into_iter()
            .map(|(name, value)| (name.to_owned(), value))
            .collect()
    }

    #[test]
    fn select_every_validates_mapping_count_and_infers_frame_rate() {
        let input = fixed_media("gray8", 8, 6);

        let (select, output) = select_output_media(
            &input,
            &options([
                ("every", FilterOptionValue::Int(3)),
                ("offset", FilterOptionValue::Int(1)),
            ]),
        )
        .expect("regular every pattern should validate");

        assert_eq!(select.source_frames(), &[1, 4, 7, 10, 13, 16, 19, 22]);
        assert_eq!(output.frame_count(), FrameCount::Finite(8));
        assert_eq!(
            output.frame_rate(),
            FrameRate::Cfr(Rational {
                numerator: 24_000,
                denominator: 3_003,
            })
        );
    }

    #[test]
    fn select_drop_every_validates_mapping_count_and_infers_average_rate() {
        let input = fixed_media("gray8", 8, 6);

        let (select, output) = select_output_media(
            &input,
            &options([("drop_every", FilterOptionValue::Int(5))]),
        )
        .expect("drop pattern should validate");

        assert_eq!(
            select.source_frames(),
            &[
                0, 1, 2, 3, 5, 6, 7, 8, 10, 11, 12, 13, 15, 16, 17, 18, 20, 21, 22, 23
            ]
        );
        assert_eq!(output.frame_count(), FrameCount::Finite(20));
        assert_eq!(
            output.frame_rate(),
            FrameRate::Cfr(Rational {
                numerator: 96_000,
                denominator: 5_005,
            })
        );
    }

    #[test]
    fn select_frames_expression_preserves_order_and_input_rate() {
        let input = fixed_media("gray8", 8, 6);

        let (select, output) = select_output_media(
            &input,
            &options([(
                "frames",
                FilterOptionValue::String("5, 1, 8-10, 12..=14".to_owned()),
            )]),
        )
        .expect("explicit expression should validate");

        assert_eq!(select.source_frames(), &[5, 1, 8, 9, 10, 12, 13, 14]);
        assert_eq!(output.frame_count(), FrameCount::Finite(8));
        assert_eq!(output.frame_rate(), input.frame_rate());
    }

    #[test]
    fn select_rejects_invalid_forms_before_render() {
        let input = fixed_media("gray8", 8, 6);

        for bad_options in [
            options([]),
            options([
                ("every", FilterOptionValue::Int(2)),
                ("frames", FilterOptionValue::String("0".to_owned())),
            ]),
            options([("every", FilterOptionValue::Int(0))]),
            options([
                ("every", FilterOptionValue::Int(2)),
                ("offset", FilterOptionValue::Int(2)),
            ]),
            options([
                ("every", FilterOptionValue::Int(2)),
                ("drop_offset", FilterOptionValue::Int(1)),
            ]),
            options([("drop_every", FilterOptionValue::Int(-1))]),
            options([("drop_every", FilterOptionValue::Int(1))]),
            options([
                ("drop_every", FilterOptionValue::Int(3)),
                ("offset", FilterOptionValue::Int(1)),
            ]),
            options([("frames", FilterOptionValue::String(String::new()))]),
            options([
                ("frames", FilterOptionValue::String("0".to_owned())),
                ("offset", FilterOptionValue::Int(1)),
            ]),
            options([("frames", FilterOptionValue::String("3-1".to_owned()))]),
            options([("frames", FilterOptionValue::String("1,1".to_owned()))]),
            options([("frames", FilterOptionValue::String("24".to_owned()))]),
            options([("unknown", FilterOptionValue::Bool(true))]),
        ] {
            let error =
                select_output_media(&input, &bad_options).expect_err("bad select should fail");
            assert_eq!(error.category(), ErrorCategory::Format);
        }
    }

    #[test]
    fn select_rejects_phase_two_predicate_forms() {
        let input = fixed_media("gray8", 8, 6);

        for key in ["predicate", "where", "condition"] {
            let error = select_output_media(
                &input,
                &options([(key, FilterOptionValue::String("n % 2 == 0".to_owned()))]),
            )
            .expect_err("predicate select should be reserved for Phase 2");

            assert_eq!(error.category(), ErrorCategory::Format);
            assert_eq!(
                error.code(),
                ErrorCode::new("filter.unsupported_select_predicate")
            );
            assert!(error.message().contains("not supported"));
            assert!(!error.message().contains("Phase 2"));
        }
    }

    #[test]
    fn select_rejects_unknown_input_frame_count_before_render() {
        let input = fixed_media("gray8", 8, 6);
        let unknown_count = ClipMedia::new(
            input.format().clone(),
            input.resolution().clone(),
            FrameCount::Unknown,
            input.frame_rate(),
        );

        let error = select_output_media(
            &unknown_count,
            &options([("every", FilterOptionValue::Int(2))]),
        )
        .expect_err("unknown frame count should fail before render");

        assert_eq!(error.category(), ErrorCategory::Format);
        assert_eq!(error.code(), ErrorCode::new("filter.variable_frame_count"));
    }

    #[test]
    fn select_rejects_unknown_input_frame_rate_before_render() {
        let input = fixed_media("gray8", 8, 6);
        let unknown_rate = ClipMedia::new(
            input.format().clone(),
            input.resolution().clone(),
            input.frame_count(),
            FrameRate::Unknown,
        );

        let error = select_output_media(
            &unknown_rate,
            &options([("every", FilterOptionValue::Int(2))]),
        )
        .expect_err("unknown frame rate should fail before render");

        assert_eq!(error.category(), ErrorCategory::Format);
        assert_eq!(error.code(), ErrorCode::new("filter.variable_frame_rate"));
    }

    #[test]
    fn select_rejects_non_positive_input_frame_rate_before_render() {
        let input = fixed_media("gray8", 8, 6);
        let invalid_rate = ClipMedia::new(
            input.format().clone(),
            input.resolution().clone(),
            input.frame_count(),
            FrameRate::Cfr(Rational {
                numerator: 0,
                denominator: 1,
            }),
        );

        let error = select_output_media(
            &invalid_rate,
            &options([("every", FilterOptionValue::Int(2))]),
        )
        .expect_err("non-positive frame rate should fail before render");

        assert_eq!(error.category(), ErrorCategory::Format);
        assert_eq!(error.code(), ErrorCode::new("filter.invalid_select"));
    }

    #[test]
    fn add_select_filter_creates_filter_node_with_frame_map_schedule() {
        let input_media = fixed_media("gray8", 8, 6);
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", input_media.clone());

        let (selected, executor) = super::add_select_filter(
            &mut builder,
            source,
            &input_media,
            &options([("frames", FilterOptionValue::String("2, 0".to_owned()))]),
        )
        .expect("select node should build");
        let graph = builder.build();
        let node = graph
            .node(selected.node_id())
            .expect("select node should exist");

        assert_eq!(executor.selection().source_frames(), &[2, 0]);
        assert_eq!(node.media().frame_count(), FrameCount::Finite(2));
        assert_eq!(node.media().frame_rate(), input_media.frame_rate());
        assert!(matches!(
            node.kind(),
            NodeKind::Filter { dependencies, .. }
                if dependencies
                    == &DependencyPattern::frame_map(DynamicDependencyBounds::any())
        ));
    }

    #[test]
    fn select_executor_maps_output_frames_to_source_frames() {
        struct NumberedSource;

        impl FrameExecutor for NumberedSource {
            fn prepare(&self, request: FrameRequest<'_>) -> pixelflow_core::Result<Frame> {
                let frame = synthetic_u8_frame("gray8", 2, 2, |_plane, x, y| {
                    u8::try_from(request.frame_number() + x + y).expect("fixture sample fits u8")
                })?;
                with_frame_number_metadata(&frame, request.frame_number())
            }
        }

        let input_media = fixed_media("gray8", 2, 2);
        let mut builder = GraphBuilder::new();
        let source = builder.source("source", input_media.clone());
        let (selected, executor) = super::add_select_filter(
            &mut builder,
            source,
            &input_media,
            &options([("frames", FilterOptionValue::String("5,1,3".to_owned()))]),
        )
        .expect("select node should build");
        builder.set_output(selected);
        let graph = builder.build();

        let mut executors = RenderExecutorMap::new();
        executors.insert(source.node_id(), Arc::new(NumberedSource));
        executors.insert(selected.node_id(), Arc::new(executor));

        let frames: Vec<_> = RenderEngine::new(WorkerPoolConfig::new(1))
            .render_ordered(graph, executors, RenderOptions::default())
            .expect("render should start")
            .map(|frame| frame.expect("selected frame should render"))
            .collect();

        let selected_numbers: Vec<_> = frames
            .iter()
            .map(|frame| match frame.metadata().get("core:frame_number") {
                Some(MetadataValue::Int(value)) => *value,
                other => panic!("expected frame_number metadata, got {other:?}"),
            })
            .collect();
        assert_eq!(selected_numbers, [5, 1, 3]);
    }
}