### Build and Serve Media Pipeline (Go)

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

Illustrates how to build and serve a media processing pipeline using `avpipeline.Serve`. This involves creating input, output, and transcoder nodes, connecting them to define the data flow, and starting the pipeline with error monitoring. Key dependencies include avpipeline, codec, kernel, node, processor, observability, and secret packages.

```go
package main

import (
    "context"
    "errors"
    "io"
    "time"

    "github.com/xaionaro-go/avpipeline"
    "github.com/xaionaro-go/avpipeline/codec"
    "github.com/xaionaro-go/avpipeline/kernel"
    "github.com/xaionaro-go/avpipeline/node"
    "github.com/xaionaro-go/avpipeline/processor"
    "github.com/xaionaro-go/observability"
    "github.com/xaionaro-go/secret"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Create nodes
    input, _ := processor.NewInputFromURL(ctx, "rtmp://source/live", secret.New(""), kernel.InputConfig{})
    defer input.Close(ctx)
    inputNode := node.New(input)

    output, _ := processor.NewOutputFromURL(ctx, "rtmp://dest/live", secret.New("key"), kernel.OutputConfig{})
    defer output.Close(ctx)
    outputNode := node.New(output)

    transcoder, _ := processor.NewTranscoder(
        ctx,
        codec.NewNaiveDecoderFactory(ctx, nil),
        codec.NewNaiveEncoderFactory(ctx, &codec.NaiveEncoderFactoryParams{
            VideoCodec: codec.Name("libx264"),
            AudioCodec: codec.NameCopy,
        }),
        nil,
    )
    defer transcoder.Close(ctx)
    transcoderNode := node.New(transcoder)

    // Connect nodes: input -> transcoder -> output
    inputNode.AddPushPacketsTo(ctx, transcoderNode)
    transcoderNode.AddPushPacketsTo(ctx, outputNode)

    // Start the pipeline
    errCh := make(chan node.Error, 10)
    observability.Go(ctx, func(ctx context.Context) {
        defer cancel()
        avpipeline.Serve(ctx, avpipeline.ServeConfig{
            EachNode: node.ServeConfig{
                FrameDropVideo: true,  // drop frames if processing is too slow
                FrameDropAudio: false,
            },
        }, errCh, inputNode)
    })

    // Monitor for errors
    for {
        select {
        case <-ctx.Done():
            return
        case err := <-errCh:
            if errors.Is(err.Err, io.EOF) || errors.Is(err.Err, context.Canceled) {
                continue
            }
            if err.Err != nil {
                panic(err.Err)
            }
        case <-time.After(time.Second):
            stats := inputNode.GetCountersPtr()
            println("Packets sent:", stats.Sent.Packets.ToStats().Count)
        }
    }
}

```

--------------------------------

### Create Input Sources with AVPipeline

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

The processor.NewInputFromURL function initializes media inputs from URLs, files, or devices. It supports custom configurations like timeout settings, video resolution, and format overrides.

```go
package main

import (
    "context"
    "github.com/xaionaro-go/avpipeline/kernel"
    "github.com/xaionaro-go/avpipeline/node"
    "github.com/xaionaro-go/avpipeline/processor"
    "github.com/xaionaro-go/secret"
)

func main() {
    ctx := context.Background()

    // Create input from RTMP stream
    input, err := processor.NewInputFromURL(
        ctx,
        "rtmp://localhost/live/stream",
        secret.New(""),  // optional stream key
        kernel.InputConfig{
            AutoClose:          true,
            IgnoreIncorrectDTS: false,
            IgnoreZeroDuration: false,
            CustomOptions: []kernel.DictionaryItem{
                {Key: "timeout", Value: "5000000"},  // microseconds
            },
        },
    )
    if err != nil {
        panic(err)
    }
    defer input.Close(ctx)

    // Wrap processor in a node for pipeline use
    inputNode := node.New(input)

    // For file input with video_size override
    fileInput, err := processor.NewInputFromURL(
        ctx,
        "/dev/video0",  // webcam
        secret.New(""),
        kernel.InputConfig{
            CustomOptions: []kernel.DictionaryItem{
                {Key: "video_size", Value: "1920x1080"},
                {Key: "framerate", Value: "30"},
                {Key: "f", Value: "v4l2"},
            },
        },
    )
    if err != nil {
        panic(err)
    }
    defer fileInput.Close(ctx)
}
```

--------------------------------

### Build and Run Dynamic Audio/Video Processing Pipeline in Go

Source: https://github.com/xaionaro-go/avpipeline/blob/main/README.md

This Go code snippet demonstrates how to construct and run a dynamic audio/video processing pipeline using the avpipeline package. It sets up input, output, and transcoding nodes, connects them, and includes error handling and status observation. Dependencies include context, logger, processor, secret, kernel, node, codec, types, observability, and json packages.

```go
	ctx, cancelFn := context.WithCancel(ctx)
	defer cancelFn()

	// input node

	logger.Debugf(ctx, "opening '%s' as the input...", fromURL)
	input, err := processor.NewInputFromURL(ctx, fromURL, secret.New(""), kernel.InputConfig{})
	assert(ctx, err == nil, err)
	defer input.Close(ctx)
	inputNode := node.New(input)

	// output node

	logger.Debugf(ctx, "opening '%s' as the output...", toURL)
	output, err := processor.NewOutputFromURL(
		ctx,
		toURL, secret.New(""),
		kernel.OutputConfig{},
	)
	assert(ctx, err == nil, err)
	defer output.Close(ctx)
	outputNode := node.New(output)

	// transcoder node

	hwDevName := codec.HardwareDeviceName(*hwDeviceName)
	transcoder, err := processor.NewTranscoder(
		ctx,
		codec.NewNaiveDecoderFactory(ctx, &codec.NaiveDecoderFactoryParams{
			HardwareDeviceName: hwDeviceName,
		}),
		codec.NewNaiveEncoderFactory(ctx, &codec.NaiveEncoderFactoryParams{
			VideoCodec:         *videoCodec,
			AudioCodec:         codec.CodecNameCopy,
			HardwareDeviceType: 0,
			HardwareDeviceName: hwDeviceName,
			VideoOptions:       types.DictionaryItems{{Key: "bf", Value: "0"}}.ToAstiav(),
		}),
		nil,
	)
	assert(ctx, err == nil, err)
	defer transcoder.Close(ctx)
	logger.Debugf(ctx, "initialized a transcoder to %s (hwdev:%s)...", *videoCodec, hwDevName)
	transcodingNode := node.New(transcoder)

	// route nodes: input -> transcoder -> output

	inputNode.AddPushPacketsTo(transcodingNode)
	transcodingNode.AddPushPacketsTo(outputNode)
	logger.Debugf(ctx, "resulting pipeline: %s", inputNode.String())
	logger.Debugf(ctx, "resulting pipeline (for graphviz):\n%s\n", inputNode.DotString(false))

	// start

	errCh := make(chan node.Error, 10)
	observability.Go(ctx, func() {
		defer cancelFn()
		avpipeline.Serve(ctx, avpipeline.ServeConfig{
			EachNode: node.ServeConfig{
				FrameDrop: *frameDrop,
			},
		}, errCh, inputNode)
	})

	// observe

	statusTicker := time.NewTicker(time.Second)
	defer statusTicker.Stop()
	for {
		select {
		case <-ctx.Done():
			logger.Infof(ctx, "finished")
			return
		case err, ok := <-errCh:
			if !ok {
				return
			}
			if errors.Is(err.Err, context.Canceled) {
				continue
			}
			if errors.Is(err.Err, io.EOF) {
				continue
			}
			if err.Err != nil {
				logger.Fatal(ctx, err)
				return
			}
		case <-statusTicker.C:
			inputStats := inputNode.GetStats()
			inputStatsJSON, err := json.Marshal(inputStats.FramesWrote)
			assert(ctx, err == nil, err)

			outputStats := outputNode.GetStats()
			outputStatsJSON, err := json.Marshal(outputStats.FramesRead)
			assert(ctx, err == nil, err)

			fmt.Printf("input:%s -> output:%s\n", inputStatsJSON, outputStatsJSON)
		}
	}

```

--------------------------------

### Create Transcoder with Hardware Acceleration (Go)

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

Demonstrates creating a transcoder using `processor.NewTranscoder`. It configures hardware-accelerated decoding (VAAPI) and H.264 encoding with specific quality and performance options. Dependencies include astiav, codec, node, processor, quality, and globaltypes packages.

```go
package main

import (
    "context"
    "github.com/asticode/go-astiav"
    "github.com/xaionaro-go/avpipeline/codec"
    "github.com/xaionaro-go/avpipeline/node"
    "github.com/xaionaro-go/avpipeline/processor"
    "github.com/xaionaro-go/avpipeline/quality"
    globaltypes "github.com/xaionaro-go/avpipeline/types"
)

func main() {
    ctx := context.Background()

    // Create decoder factory with hardware acceleration
    decoderFactory := codec.NewNaiveDecoderFactory(ctx, &codec.NaiveDecoderFactoryParams{
        HardwareDeviceType: globaltypes.HardwareDeviceType(astiav.HardwareDeviceTypeVAAPI),
        HardwareDeviceName: "/dev/dri/renderD128",
        PostInitFunc: func(ctx context.Context, d *codec.Decoder) {
            d.SetLowLatency(ctx, true)
        },
    })

    // Create encoder factory with H.264 encoding
    encoderFactory := codec.NewNaiveEncoderFactory(ctx, &codec.NaiveEncoderFactoryParams{
        VideoCodec:         codec.Name("libx264"),
        AudioCodec:         codec.NameCopy,  // passthrough audio
        HardwareDeviceType: 0,
        HardwareDeviceName: "",
        VideoQuality:       quality.ConstantBitrate(4000000),  // 4 Mbps
        VideoResolution:    &codec.Resolution{Width: 1920, Height: 1080},
        VideoAverageFrameRate: astiav.NewRational(30, 1),
        VideoOptions: func() *astiav.Dictionary {
            d := astiav.NewDictionary()
            d.Set("preset", "fast", 0)
            d.Set("tune", "zerolatency", 0)
            d.Set("bf", "0", 0)  // no B-frames
            return d
        }(),
    })

    transcoder, err := processor.NewTranscoder(
        ctx,
        decoderFactory,
        encoderFactory,
        nil,  // optional stream configurer
    )
    if err != nil {
        panic(err)
    }
    defer transcoder.Close(ctx)

    transcoderNode := node.New(transcoder)
    _ = transcoderNode
}

```

--------------------------------

### Configure Transcoder with Passthrough Mode in Go

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

Shows how to initialize a transcoder that supports dynamic switching between transcoding and passthrough modes. This is useful for optimizing CPU usage by bypassing re-encoding when source codecs match output requirements.

```go
package main

import (
    "context"
    "time"

    "github.com/xaionaro-go/avpipeline"
    codectypes "github.com/xaionaro-go/avpipeline/codec/types"
    "github.com/xaionaro-go/avpipeline/kernel"
    "github.com/xaionaro-go/avpipeline/node"
    "github.com/xaionaro-go/avpipeline/preset/transcoderwithpassthrough"
    "github.com/xaionaro-go/avpipeline/preset/transcoderwithpassthrough/types"
    "github.com/xaionaro-go/avpipeline/processor"
    "github.com/xaionaro-go/secret"
)

func main() {
    ctx := context.Background()

    input, _ := processor.NewInputFromURL(ctx, "rtmp://source/live", secret.New(""), kernel.InputConfig{})
    inputKernel := input.Kernel

    output, _ := processor.NewOutputFromURL(ctx, "rtmp://dest/live", secret.New("key"), kernel.OutputConfig{})
    outputNode := node.New(output)

    twp, err := transcoderwithpassthrough.New[struct{}, processor.Abstract](
        ctx,
        inputKernel,
        outputNode,
    )
    if err != nil {
        panic(err)
    }

    err = twp.SetTranscoderConfig(ctx, types.TranscoderConfig{
        Output: types.TranscoderConfigOutput{
            VideoTrackConfigs: []types.VideoTrackConfig{{
                CodecName:      codectypes.Name("libx264"),
                AverageBitRate: 4000000,
                Resolution:     codectypes.Resolution{Width: 1920, Height: 1080},
            }},
            AudioTrackConfigs: []types.AudioTrackConfig{{
                CodecName: codectypes.Name("copy"),
            }},
        },
    })
    if err != nil {
        panic(err)
    }

    err = twp.Start(ctx, types.PassthroughModeSameTracks, avpipeline.ServeConfig{})
    if err != nil {
        panic(err)
    }

    time.AfterFunc(30*time.Second, func() {
        twp.SetTranscoderConfig(ctx, types.TranscoderConfig{
            Output: types.TranscoderConfigOutput{
                VideoTrackConfigs: []types.VideoTrackConfig{{
                    CodecName:      codectypes.Name("copy"),
                    AverageBitRate: 0,
                }},
                AudioTrackConfigs: []types.AudioTrackConfig{{
                    CodecName: codectypes.Name("copy"),
                }},
            },
        })
    })

    twp.Wait(ctx)
}
```

--------------------------------

### Implement Input Fallback with Go

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

Demonstrates how to create an input source that automatically switches between primary and fallback streams. It uses a slice of input factories to define sources and configures retry intervals and timeout behaviors.

```go
package main

import (
    "context"
    "time"

    "github.com/xaionaro-go/avpipeline/codec"
    "github.com/xaionaro-go/avpipeline/kernel"
    "github.com/xaionaro-go/avpipeline/preset/inputwithfallback"
    "github.com/xaionaro-go/avpipeline/processor"
    "github.com/xaionaro-go/secret"
)

func main() {
    ctx := context.Background()

    inputFactories := []inputwithfallback.InputFactory[*kernel.Input, *codec.NaiveDecoderFactory, struct{}]{ 
        func(ctx context.Context) (*kernel.Input, *codec.NaiveDecoderFactory, struct{}, error) {
            k, err := kernel.NewInputFromURL(ctx, "rtmp://primary/stream", secret.New(""), kernel.InputConfig{})
            return k, nil, struct{}{}, err
        },
        func(ctx context.Context) (*kernel.Input, *codec.NaiveDecoderFactory, struct{}, error) {
            k, err := kernel.NewInputFromURL(ctx, "rtmp://backup1/stream", secret.New(""), kernel.InputConfig{})
            return k, nil, struct{}{}, err
        },
        func(ctx context.Context) (*kernel.Input, *codec.NaiveDecoderFactory, struct{}, error) {
            k, err := kernel.NewInputFromURL(ctx, "testsrc=size=1920x1080:rate=30", secret.New(""), kernel.InputConfig{
                CustomOptions: []kernel.DictionaryItem{{Key: "f", Value: "lavfi"}},
            })
            return k, nil, struct{}{}, err
        },
    }

    inputWithFallback, err := inputwithfallback.New(
        ctx,
        inputFactories,
        inputwithfallback.OptionRetryInterval(5*time.Second),
        inputwithfallback.OptionSwitchKeepUnlessTimeout(2*time.Second),
    )
    if err != nil {
        panic(err)
    }

    outputNode := inputWithFallback.GetOutput()
    _ = outputNode
}
```

--------------------------------

### Create Output Destinations with AVPipeline

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

The processor.NewOutputFromURL function creates media output sinks for streaming protocols or local files. It manages stream negotiation and ensures proper resource cleanup upon closing.

```go
package main

import (
    "context"
    "github.com/xaionaro-go/avpipeline/kernel"
    "github.com/xaionaro-go/avpipeline/node"
    "github.com/xaionaro-go/avpipeline/processor"
    "github.com/xaionaro-go/secret"
)

func main() {
    ctx := context.Background()

    // Create RTMP output
    output, err := processor.NewOutputFromURL(
        ctx,
        "rtmp://live.twitch.tv/app",
        secret.New("your_stream_key"),
        kernel.OutputConfig{
            SendBufferSize: 1024 * 1024,  // 1MB send buffer
            WaitForOutputStreams: &kernel.OutputConfigWaitForOutputStreams{
                MinStreams:      2,  // wait for audio + video
                MinStreamsVideo: 1,
                MinStreamsAudio: 1,
            },
            CustomOptions: []kernel.DictionaryItem{
                {Key: "flvflags", Value: "no_sequence_end"},
            },
        },
    )
    if err != nil {
        panic(err)
    }
    defer output.Close(ctx)

    outputNode := node.New(output)

    // Create file output
    fileOutput, err := processor.NewOutputFromURL(
        ctx,
        "/tmp/output.mp4",
        secret.New(""),
        kernel.OutputConfig{},
    )
    if err != nil {
        panic(err)
    }
    defer fileOutput.Close(ctx)
}
```

--------------------------------

### Visualize Pipeline Topology with GraphViz in Go

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

This Go code snippet shows how to generate GraphViz DOT representations of a pipeline's topology for debugging and documentation. It uses the `String()` method for a human-readable representation and `DotString()` to generate DOT format, optionally including statistics. The output is written to files for external rendering. It depends on 'context', 'fmt', 'os', and avpipeline packages.

```go
package main

import (
    "context"
    "fmt"
    "os"

    "github.com/xaionaro-go/avpipeline/node"
    "github.com/xaionaro-go/avpipeline/processor"
)

func visualizePipeline(inputNode *node.Node[processor.Abstract]) {
    // Get human-readable string representation
    pipelineStr := inputNode.String()
    fmt.Println("Pipeline:", pipelineStr)

    // Generate GraphViz DOT format
    dotStr := inputNode.DotString(false)  // false = without stats

    // Write to file for visualization
    os.WriteFile("/tmp/pipeline.dot", []byte(dotStr), 0644)

    // Generate with stats
    dotWithStats := inputNode.DotString(true)
    os.WriteFile("/tmp/pipeline_with_stats.dot", []byte(dotWithStats), 0644)

    // Render with: dot -Tpng /tmp/pipeline.dot -o pipeline.png
}

```

--------------------------------

### Conditional Routing with Packet Filters

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

Demonstrates how to use packet filters and conditions to route packets based on media type, key frame status, and combinations using logical operators (And, Or, Not). This allows for complex routing logic in media processing pipelines.

```go
package main

import (
    "context"

    "github.com/asticode/go-astiav"
    "github.com/xaionaro-go/avpipeline/node"
    packetfiltercondition "github.com/xaionaro-go/avpipeline/node/filter/packetfilter/condition"
    framefiltercondition "github.com/xaionaro-go/avpipeline/node/filter/framefilter/condition"
    "github.com/xaionaro-go/avpipeline/processor"
)

func setupConditionalRouting(
    ctx context.Context,
    inputNode *node.Node[processor.Abstract],
    videoProcessor *node.Node[processor.Abstract],
    audioProcessor *node.Node[processor.Abstract],
    outputNode *node.Node[processor.Abstract],
) {
    // Route video packets to video processor
    inputNode.AddPushPacketsTo(ctx, videoProcessor,
        packetfiltercondition.MediaType(astiav.MediaTypeVideo),
    )

    // Route audio packets to audio processor
    inputNode.AddPushPacketsTo(ctx, audioProcessor,
        packetfiltercondition.MediaType(astiav.MediaTypeAudio),
    )

    // Route only key frames to output (useful for thumbnail extraction)
    inputNode.AddPushPacketsTo(ctx, outputNode,
        packetfiltercondition.And{
            packetfiltercondition.MediaType(astiav.MediaTypeVideo),
            packetfiltercondition.IsKeyFrame(true),
        },
    )

    // Complex condition: video key frames OR all audio
    inputNode.AddPushPacketsTo(ctx, outputNode,
        packetfiltercondition.Or{
            packetfiltercondition.And{
                packetfiltercondition.MediaType(astiav.MediaTypeVideo),
                packetfiltercondition.IsKeyFrame(true),
            },
            packetfiltercondition.MediaType(astiav.MediaTypeAudio),
        },
    )

    // For frame-level filtering (after decoding)
    inputNode.AddPushFramesTo(ctx, videoProcessor,
        framefiltercondition.MediaType(astiav.MediaTypeVideo),
    )
}

```

--------------------------------

### Monitor Pipeline Statistics with Go

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

This Go code snippet demonstrates how to monitor pipeline statistics, including received and sent packets, and calculate the drop rate. It uses `GetCountersPtr()` to access real-time statistics from nodes and their processors, then formats them as JSON for display. It relies on the 'context', 'encoding/json', 'fmt', 'time', and specific avpipeline packages.

```go
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "github.com/xaionaro-go/avpipeline/node"
    nodetypes "github.com/xaionaro-go/avpipeline/node/types"
    "github.com/xaionaro-go/avpipeline/processor"
)

func monitorPipeline(
    ctx context.Context,
    inputNode *node.Node[processor.Abstract],
    transcoderNode *node.Node[processor.Abstract],
    outputNode *node.Node[processor.Abstract],
) {
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            // Get node-level counters
            inputCounters := inputNode.GetCountersPtr()
            outputCounters := outputNode.GetCountersPtr()

            // Get processor-level counters
            inputProcCounters := inputNode.GetProcessor().CountersPtr()
            outputProcCounters := outputNode.GetProcessor().CountersPtr()

            // Combine into full statistics
            inputStats := nodetypes.ToStatistics(inputCounters, inputProcCounters)
            outputStats := nodetypes.ToStatistics(outputCounters, outputProcCounters)

            // Convert to JSON for display
            inputJSON, _ := json.Marshal(map[string]interface{}{
                "received": inputStats.Received,
                "sent":     inputStats.Sent,
            })
            outputJSON, _ := json.Marshal(map[string]interface{}{
                "received": outputStats.Received,
                "sent":     outputStats.Sent,
            })

            fmt.Printf("Input: %s\n", inputJSON)
            fmt.Printf("Output: %s\n", outputJSON)

            // Calculate drop rate
            if inputStats.Sent.Packets.Count > 0 {
                dropRate := float64(inputStats.Missed.Packets.Count) / float64(inputStats.Sent.Packets.Count) * 100
                fmt.Printf("Drop rate: %.2f%%\n", dropRate)
            }
        }
    }
}

```

--------------------------------

### Bitrate Limiting Filter for Video

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

Creates a node that limits the video bitrate by dropping packets when the output exceeds a target bitrate. It supports dynamic adjustment of the target bitrate and uses an averaging window for smoother control.

```go
package main

import (
    "context"
    "time"

    "github.com/xaionaro-go/avpipeline/kernel"
    "github.com/xaionaro-go/avpipeline/node"
    packetcondition "github.com/xaionaro-go/avpipeline/packet/condition"
    "github.com/xaionaro-go/avpipeline/packetorframe/filter/limitvideobitrate"
    "github.com/xaionaro-go/avpipeline/processor"
)

func createBitrateLimiter(ctx context.Context) *node.Node[*processor.FromKernel[*kernel.PacketFilter]] {
    // Create a bitrate limiter: 4 Mbps with 1 second averaging window
    bitrateFilter := limitvideobitrate.New(
        ctx,
        4_000_000,       // target bitrate in bits/second
        time.Second,     // averaging period
    )

    // Wrap in a packet filter kernel
    packetFilter := kernel.NewPacketFilter(
        packetcondition.PacketOrFrame{bitrateFilter},
        nil,
    )

    // Create node
    filterNode := node.NewFromKernel(
        ctx,
        packetFilter,
        processor.OptionQueueSizeInputPacket(600),
        processor.OptionQueueSizeOutputPacket(10),
    )

    // Dynamically adjust bitrate
    go func() {
        time.Sleep(30 * time.Second)
        bitrateFilter.AverageBitRate = 2_000_000  // reduce to 2 Mbps
    }()

    return filterNode
}

```

--------------------------------

### Frame Rate Reduction Filter

Source: https://context7.com/xaionaro-go/avpipeline/llms.txt

Implements a filter to reduce the frame rate of video streams. It allows passing through non-video frames while applying a specified frame rate reduction to video frames, controlled by a target frame rate.

```go
package main

import (
    "context"

    "github.com/asticode/go-astiav"
    framecondition "github.com/xaionaro-go/avpipeline/frame/condition"
    mathcondition "github.com/xaionaro-go/avpipeline/math/condition"
    "github.com/xaionaro-go/avpipeline/packetorframe/filter/reduceframerate"
    globaltypes "github.com/xaionaro-go/avpipeline/types"
)

func createFrameRateFilter() framecondition.Condition {
    // Reduce frame rate to 50% (e.g., 60fps -> 30fps)
    targetFrameRate := globaltypes.RationalFromApproxFloat64(0.5)

    filter := framecondition.Or{
        // Pass through all non-video frames
        framecondition.Not{framecondition.MediaType(astiav.MediaTypeVideo)},
        // Apply frame rate reduction to video
        framecondition.PacketOrFrame{
            reduceframerate.New(mathcondition.GetterStatic[globaltypes.Rational]{
                StaticValue: targetFrameRate,
            }),
        },
    }

    return filter
}

// Apply to a transcoder:
// transcoderKernel.Filter = createFrameRateFilter()

```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.