Skip to content

File: Broken/Externals/FFmpeg.py

Broken.Externals.FFmpeg

FFmpegModuleBase

Bases: BrokenModel, ABC

Source code in Broken/Externals/FFmpeg.py
40
41
42
43
44
45
46
47
48
class FFmpegModuleBase(BrokenModel, ABC):
    model_config = ConfigDict(
        use_attribute_docstrings=True,
        validate_assignment=True,
    )

    @abstractmethod
    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        ...

model_config

model_config = ConfigDict(
    use_attribute_docstrings=True, validate_assignment=True
)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
46
47
48
@abstractmethod
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    ...

FFmpegInputPath

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
55
56
57
58
59
60
class FFmpegInputPath(FFmpegModuleBase):
    type: Annotated[Literal["path"], BrokenTyper.exclude()] = "path"
    path: Path

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        return ("-i", self.path)

type

type: Annotated[Literal["path"], BrokenTyper.exclude()] = (
    "path"
)

path

path: Path

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
59
60
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    return ("-i", self.path)

FFmpegInputPipe

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
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
class FFmpegInputPipe(FFmpegModuleBase):
    type: Annotated[Literal["pipe"], BrokenTyper.exclude()] = "pipe"

    class Format(str, BrokenEnum):
        Rawvideo   = "rawvideo"
        Image2Pipe = "image2pipe"
        Null       = "null"

    format: Annotated[Optional[Format],
        Option("--format", "-f")] = \
        Field(Format.Rawvideo)

    class PixelFormat(str, BrokenEnum):
        YUV420P = "yuv420p"
        YUV444P = "yuv444p"
        RGB24   = "rgb24"
        RGBA    = "rgba"

    pixel_format: Annotated[PixelFormat,
        Option("--pixel-format", "-p")] = \
        Field(PixelFormat.RGB24)

    width: int = Field(1920, gt=0)
    height: int = Field(1080, gt=0)
    framerate: float = Field(60.0, ge=1.0)

    @field_validator("framerate", mode="plain")
    def validate_framerate(cls, value: Union[float, str]) -> float:
        return eval(str(value))

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-f", denum(self.format))
        yield ("-s", f"{self.width}x{self.height}")
        yield ("-pix_fmt", denum(self.pixel_format))
        yield ("-r", self.framerate)
        yield ("-i", "-")

type

type: Annotated[Literal["pipe"], BrokenTyper.exclude()] = (
    "pipe"
)

Format

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
66
67
68
69
class Format(str, BrokenEnum):
    Rawvideo   = "rawvideo"
    Image2Pipe = "image2pipe"
    Null       = "null"
Rawvideo
Rawvideo = 'rawvideo'
Image2Pipe
Image2Pipe = 'image2pipe'
Null
Null = 'null'

format

format: Annotated[
    Optional[Format], Option("--format", "-f")
] = Field(Format.Rawvideo)

PixelFormat

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
75
76
77
78
79
class PixelFormat(str, BrokenEnum):
    YUV420P = "yuv420p"
    YUV444P = "yuv444p"
    RGB24   = "rgb24"
    RGBA    = "rgba"
YUV420P
YUV420P = 'yuv420p'
YUV444P
YUV444P = 'yuv444p'
RGB24
RGB24 = 'rgb24'
RGBA
RGBA = 'rgba'

pixel_format

pixel_format: Annotated[
    PixelFormat, Option("--pixel-format", "-p")
] = Field(PixelFormat.RGB24)

width

width: int = Field(1920, gt=0)

height

height: int = Field(1080, gt=0)

framerate

framerate: float = Field(60.0, ge=1.0)

validate_framerate

validate_framerate(value: Union[float, str]) -> float
Source code in Broken/Externals/FFmpeg.py
89
90
91
@field_validator("framerate", mode="plain")
def validate_framerate(cls, value: Union[float, str]) -> float:
    return eval(str(value))

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
93
94
95
96
97
98
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-f", denum(self.format))
    yield ("-s", f"{self.width}x{self.height}")
    yield ("-pix_fmt", denum(self.pixel_format))
    yield ("-r", self.framerate)
    yield ("-i", "-")

FFmpegInputType

FFmpegInputType: TypeAlias = Union[
    FFmpegInputPath, FFmpegInputPipe
]

FFmpegOutputPath

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
class FFmpegOutputPath(FFmpegModuleBase):
    type: Annotated[Literal["path"], BrokenTyper.exclude()] = "path"

    overwrite: Annotated[bool,
        Option("--overwrite", "-y", " /--no-overwrite", " /-n")] = \
        Field(True)

    path: Annotated[Path,
        typer.Argument(help="The output file path")] = \
        Field(...)

    class PixelFormat(str, BrokenEnum):
        YUV420P = "yuv420p"
        YUV444P = "yuv444p"

    pixel_format: Annotated[Optional[PixelFormat],
        Option("--pixel-format", "-p")] = \
        Field(PixelFormat.YUV420P)

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-pix_fmt", denum(self.pixel_format))
        yield (self.path, self.overwrite*"-y")

type

type: Annotated[Literal["path"], BrokenTyper.exclude()] = (
    "path"
)

overwrite

overwrite: Annotated[
    bool,
    Option("--overwrite", "-y", " /--no-overwrite", " /-n"),
] = Field(True)

path

path: Annotated[
    Path, typer.Argument(help="The output file path")
] = Field(...)

PixelFormat

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
119
120
121
class PixelFormat(str, BrokenEnum):
    YUV420P = "yuv420p"
    YUV444P = "yuv444p"
YUV420P
YUV420P = 'yuv420p'
YUV444P
YUV444P = 'yuv444p'

pixel_format

pixel_format: Annotated[
    Optional[PixelFormat], Option("--pixel-format", "-p")
] = Field(PixelFormat.YUV420P)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
127
128
129
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-pix_fmt", denum(self.pixel_format))
    yield (self.path, self.overwrite*"-y")

FFmpegOutputPipe

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
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
class FFmpegOutputPipe(FFmpegModuleBase):
    type: Annotated[Literal["pipe"], BrokenTyper.exclude()] = "pipe"

    class Format(str, BrokenEnum):
        Rawvideo   = "rawvideo"
        Image2Pipe = "image2pipe"
        Matroska   = "matroska"
        Mpegts     = "mpegts"
        Null       = "null"

    format: Annotated[Optional[Format],
        Option("--format", "-f")] = \
        Field(None)

    class PixelFormat(str, BrokenEnum):
        RGB24 = "rgb24"
        RGBA  = "rgba"

    pixel_format: Annotated[Optional[PixelFormat],
        Option("--pixel-format", "-p")] = \
        Field(None)

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-f", denum(self.format))
        yield every("-pix_fmt", denum(self.pixel_format))
        yield "-"

type

type: Annotated[Literal["pipe"], BrokenTyper.exclude()] = (
    "pipe"
)

Format

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
135
136
137
138
139
140
class Format(str, BrokenEnum):
    Rawvideo   = "rawvideo"
    Image2Pipe = "image2pipe"
    Matroska   = "matroska"
    Mpegts     = "mpegts"
    Null       = "null"
Rawvideo
Rawvideo = 'rawvideo'
Image2Pipe
Image2Pipe = 'image2pipe'
Matroska
Matroska = 'matroska'
Mpegts
Mpegts = 'mpegts'
Null
Null = 'null'

format

format: Annotated[
    Optional[Format], Option("--format", "-f")
] = Field(None)

PixelFormat

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
146
147
148
class PixelFormat(str, BrokenEnum):
    RGB24 = "rgb24"
    RGBA  = "rgba"
RGB24
RGB24 = 'rgb24'
RGBA
RGBA = 'rgba'

pixel_format

pixel_format: Annotated[
    Optional[PixelFormat], Option("--pixel-format", "-p")
] = Field(None)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
154
155
156
157
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-f", denum(self.format))
    yield every("-pix_fmt", denum(self.pixel_format))
    yield "-"

FFmpegOutputType

FFmpegOutputType = Union[FFmpegOutputPipe, FFmpegOutputPath]

FFmpegVideoCodecH264

Bases: FFmpegModuleBase

Use [bold orange3][link=https://www.videolan.org/developers/x264.html]VideoLAN's[/link][/] [blue][link=https://trac.ffmpeg.org/wiki/Encode/H.264]libx264[/link][/]

Source code in Broken/Externals/FFmpeg.py
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
class FFmpegVideoCodecH264(FFmpegModuleBase):
    """Use [bold orange3][link=https://www.videolan.org/developers/x264.html]VideoLAN's[/link][/] [blue][link=https://trac.ffmpeg.org/wiki/Encode/H.264]libx264[/link][/]"""
    type: Annotated[Literal["h264"], BrokenTyper.exclude()] = "h264"

    class Preset(str, BrokenEnum):
        None_     = None
        UltraFast = "ultrafast"
        SuperFast = "superfast"
        VeryFast  = "veryfast"
        Faster    = "faster"
        Fast      = "fast"
        Medium    = "medium"
        Slow      = "slow"
        Slower    = "slower"
        VerySlow  = "veryslow"

    preset: Annotated[Optional[Preset],
        Option("--preset", "-p")] = \
        Field(Preset.Slow)
    """How much time to spend on encoding. Slower options gives better compression
    [blue link=https://trac.ffmpeg.org/wiki/Encode/H.264#Preset]→ Documentation[/]"""

    class Tune(str, BrokenEnum):
        None_       = None
        Film        = "film"
        Animation   = "animation"
        Grain       = "grain"
        StillImage  = "stillimage"
        FastDecode  = "fastdecode"
        ZeroLatency = "zerolatency"

    tune: Annotated[Optional[Tune],
        Option("--tune", "-t")] = \
        Field(None)
    """Tune x264 to keep and optimize for certain aspects of the input media
    [blue link=https://trac.ffmpeg.org/wiki/Encode/H.264#Tune]→ Documentation[/]"""

    class Profile(str, BrokenEnum):
        None_    = None
        Baseline = "baseline"
        Main     = "main"
        High     = "high"
        High10   = "high10"
        High422  = "high422"
        High444p = "high444p"

    profile: Annotated[Optional[Profile],
        Option("--profile", "-p")] = \
        Field(None)
    """How many features the encoder can use, the playback device must support them
    [blue link=https://trac.ffmpeg.org/wiki/Encode/H.264#Profile]→ Documentation[/]"""

    faststart: Annotated[bool,
        Option("--faststart", " /--no-faststart", hidden=True)] = \
        Field(True)
    """Move the index (moov atom) to the beginning of the file for faster initial playback"""

    rgb: Annotated[bool,
        Option("--rgb", " /--yuv")] = \
        Field(False)
    """Use RGB colorspace instead of YUV"""

    crf: int = Field(20, ge=0, le=51)
    """Constant Rate Factor. 0 is lossless, 51 is the worst quality
    [blue link=https://trac.ffmpeg.org/wiki/Encode/H.264#a1.ChooseaCRFvalue]→ Documentation[/]"""

    crf: Annotated[int,
        Option("--crf", "-c", min=0, max=51)] = \
        Field(20, ge=0, le=51)
    """Constant Rate Factor. 0 is lossless, 51 is the worst quality
    [blue link=https://trac.ffmpeg.org/wiki/Encode/H.264#a1.ChooseaCRFvalue]→ Documentation[/]"""

    bitrate: Annotated[Optional[int],
        Option("--bitrate", "-b", min=0)] = \
        Field(None, ge=0)
    """Bitrate in kilobits per second, the higher the better quality and file size"""

    x264params: Annotated[Optional[list[str]],
        Option("--x264-params", hidden=True)] = \
        Field(None)
    """Additional options to pass to x264"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-c:v", "libx264rgb" if self.rgb else "libx264")
        yield every("-movflags", "+faststart"*self.faststart)
        yield every("-profile", denum(self.profile))
        yield every("-preset", denum(self.preset))
        yield every("-tune", denum(self.tune))
        yield every("-b:v", self.bitrate)
        yield every("-crf", self.crf)
        yield every("-x264opts", ":".join(self.x264params or []))

type

type: Annotated[Literal["h264"], BrokenTyper.exclude()] = (
    "h264"
)

Preset

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
172
173
174
175
176
177
178
179
180
181
182
class Preset(str, BrokenEnum):
    None_     = None
    UltraFast = "ultrafast"
    SuperFast = "superfast"
    VeryFast  = "veryfast"
    Faster    = "faster"
    Fast      = "fast"
    Medium    = "medium"
    Slow      = "slow"
    Slower    = "slower"
    VerySlow  = "veryslow"
None_
None_ = None
UltraFast
UltraFast = 'ultrafast'
SuperFast
SuperFast = 'superfast'
VeryFast
VeryFast = 'veryfast'
Faster
Faster = 'faster'
Fast
Fast = 'fast'
Medium
Medium = 'medium'
Slow
Slow = 'slow'
Slower
Slower = 'slower'
VerySlow
VerySlow = 'veryslow'

preset

preset: Annotated[
    Optional[Preset], Option("--preset", "-p")
] = Field(Preset.Slow)

How much time to spend on encoding. Slower options gives better compression [blue link=https://trac.ffmpeg.org/wiki/Encode/H.264#Preset]→ Documentation[/]

Tune

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
190
191
192
193
194
195
196
197
class Tune(str, BrokenEnum):
    None_       = None
    Film        = "film"
    Animation   = "animation"
    Grain       = "grain"
    StillImage  = "stillimage"
    FastDecode  = "fastdecode"
    ZeroLatency = "zerolatency"
None_
None_ = None
Film
Film = 'film'
Animation
Animation = 'animation'
Grain
Grain = 'grain'
StillImage
StillImage = 'stillimage'
FastDecode
FastDecode = 'fastdecode'
ZeroLatency
ZeroLatency = 'zerolatency'

tune

tune: Annotated[Optional[Tune], Option("--tune", "-t")] = (
    Field(None)
)

Tune x264 to keep and optimize for certain aspects of the input media [blue link=https://trac.ffmpeg.org/wiki/Encode/H.264#Tune]→ Documentation[/]

Profile

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
205
206
207
208
209
210
211
212
class Profile(str, BrokenEnum):
    None_    = None
    Baseline = "baseline"
    Main     = "main"
    High     = "high"
    High10   = "high10"
    High422  = "high422"
    High444p = "high444p"
None_
None_ = None
Baseline
Baseline = 'baseline'
Main
Main = 'main'
High
High = 'high'
High10
High10 = 'high10'
High422
High422 = 'high422'
High444p
High444p = 'high444p'

profile

profile: Annotated[
    Optional[Profile], Option("--profile", "-p")
] = Field(None)

How many features the encoder can use, the playback device must support them [blue link=https://trac.ffmpeg.org/wiki/Encode/H.264#Profile]→ Documentation[/]

faststart

faststart: Annotated[
    bool,
    Option("--faststart", " /--no-faststart", hidden=True),
] = Field(True)

Move the index (moov atom) to the beginning of the file for faster initial playback

rgb

rgb: Annotated[bool, Option("--rgb", " /--yuv")] = Field(
    False
)

Use RGB colorspace instead of YUV

crf

crf: Annotated[
    int, Option("--crf", "-c", min=0, max=51)
] = Field(20, ge=0, le=51)

Constant Rate Factor. 0 is lossless, 51 is the worst quality [blue link=https://trac.ffmpeg.org/wiki/Encode/H.264#a1.ChooseaCRFvalue]→ Documentation[/]

bitrate

bitrate: Annotated[
    Optional[int], Option("--bitrate", "-b", min=0)
] = Field(None, ge=0)

Bitrate in kilobits per second, the higher the better quality and file size

x264params

x264params: Annotated[
    Optional[list[str]],
    Option("--x264-params", hidden=True),
] = Field(None)

Additional options to pass to x264

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
250
251
252
253
254
255
256
257
258
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-c:v", "libx264rgb" if self.rgb else "libx264")
    yield every("-movflags", "+faststart"*self.faststart)
    yield every("-profile", denum(self.profile))
    yield every("-preset", denum(self.preset))
    yield every("-tune", denum(self.tune))
    yield every("-b:v", self.bitrate)
    yield every("-crf", self.crf)
    yield every("-x264opts", ":".join(self.x264params or []))

FFmpegVideoCodecH264_NVENC

Bases: FFmpegModuleBase

Use [bold green][link=https://en.wikipedia.org/wiki/Nvidia_NVENC]NVIDIA[/link][/] [blue][link=https://trac.ffmpeg.org/wiki/HWAccelIntro]NVENC H.264[/link][/]

Source code in Broken/Externals/FFmpeg.py
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
class FFmpegVideoCodecH264_NVENC(FFmpegModuleBase):
    """Use [bold green][link=https://en.wikipedia.org/wiki/Nvidia_NVENC]NVIDIA[/link][/] [blue][link=https://trac.ffmpeg.org/wiki/HWAccelIntro]NVENC H.264[/link][/]"""
    type: Annotated[Literal["h264-nvenc"], BrokenTyper.exclude()] = "h264-nvenc"

    class Preset(str, BrokenEnum):
        None_                     = None
        HighQuality2Passes        = "slow"
        HighQuality1Pass          = "medium"
        HighPerformance1Pass      = "fast"
        HighPerformance           = "hp"
        HighQuality               = "hq"
        Balanced                  = "bd"
        LowLatency                = "ll"
        LowLatencyHighQuality     = "llhq"
        LowLatencyHighPerformance = "llhp"
        Lossless                  = "lossless"
        LosslessHighPerformance   = "losslesshp"
        Fastest                   = "p1"
        Faster                    = "p2"
        Fast                      = "p3"
        Medium                    = "p4"
        Slow                      = "p5"
        Slower                    = "p6"
        Slowest                   = "p7"

    preset: Annotated[Optional[Preset],
        Option("--preset", "-p")] = \
        Field(Preset.Medium)
    """How much time to spend on encoding. Slower options gives better compression"""

    class Tune(str, BrokenEnum):
        None_           = None
        HighQuality     = "hq"
        LowLatency      = "ll"
        UltraLowLatency = "ull"
        Lossless        = "lossless"

    tune: Annotated[Optional[Tune],
        Option("--tune", "-t")] = \
        Field(Tune.HighQuality)
    """Tune the encoder for a specific tier of performance"""

    class Profile(str, BrokenEnum):
        None_    = None
        Baseline = "baseline"
        Main     = "main"
        High     = "high"
        High444p = "high444p"

    profile: Annotated[Optional[Profile],
        Option("--profile", "-p")] = \
        Field(Profile.High)
    """How many features the encoder can use, the playback device must support them"""

    class RateControl(str, BrokenEnum):
        None_ = None
        ConstantQuality = "constqp"
        VariableBitrate = "vbr"
        ConstantBitrate = "cbr"

    rate_control: Annotated[Optional[RateControl],
        Option("--rc", "-r", hidden=True)] = \
        Field(RateControl.VariableBitrate)
    """Rate control mode of the bitrate"""

    rc_lookahead: Annotated[Optional[int],
        Option("--rc-lookahead", "-l", hidden=True, min=0)] = \
        Field(32, ge=0)
    """Number of frames to look ahead for the rate control"""

    cbr: Annotated[bool,
        Option("--cbr", "-c", " /--no-cbr", " /-nc", hidden=True)] = \
        Field(False)
    """Enable Constant Bitrate mode"""

    gpu: Annotated[Optional[int],
        Option("--gpu", "-g", min=-1)] = \
        Field(-1, ge=-1)
    """Use the Nth NVENC capable GPU for encoding, -1 to pick the first device available"""

    cq: Annotated[Optional[int],
        Option("--cq", "-q", min=0)] = \
        Field(25, ge=0)
    """(VBR) Similar to CRF, 0 is automatic, 1 is 'lossless', 51 is the worst quality"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-c:v", "h264_nvenc")
        yield every("-b:v", 0)
        yield every("-preset", denum(self.preset))
        yield every("-tune", denum(self.tune))
        yield every("-profile:v", denum(self.profile))
        yield every("-rc", denum(self.rate_control))
        yield every("-rc-lookahead", self.rc_lookahead)
        yield every("-cbr", int(self.cbr))
        yield every("-cq", self.cq)
        yield every("-gpu", self.gpu)

type

type: Annotated[
    Literal["h264-nvenc"], BrokenTyper.exclude()
] = "h264-nvenc"

Preset

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
class Preset(str, BrokenEnum):
    None_                     = None
    HighQuality2Passes        = "slow"
    HighQuality1Pass          = "medium"
    HighPerformance1Pass      = "fast"
    HighPerformance           = "hp"
    HighQuality               = "hq"
    Balanced                  = "bd"
    LowLatency                = "ll"
    LowLatencyHighQuality     = "llhq"
    LowLatencyHighPerformance = "llhp"
    Lossless                  = "lossless"
    LosslessHighPerformance   = "losslesshp"
    Fastest                   = "p1"
    Faster                    = "p2"
    Fast                      = "p3"
    Medium                    = "p4"
    Slow                      = "p5"
    Slower                    = "p6"
    Slowest                   = "p7"
None_
None_ = None
HighQuality2Passes
HighQuality2Passes = 'slow'
HighQuality1Pass
HighQuality1Pass = 'medium'
HighPerformance1Pass
HighPerformance1Pass = 'fast'
HighPerformance
HighPerformance = 'hp'
HighQuality
HighQuality = 'hq'
Balanced
Balanced = 'bd'
LowLatency
LowLatency = 'll'
LowLatencyHighQuality
LowLatencyHighQuality = 'llhq'
LowLatencyHighPerformance
LowLatencyHighPerformance = 'llhp'
Lossless
Lossless = 'lossless'
LosslessHighPerformance
LosslessHighPerformance = 'losslesshp'
Fastest
Fastest = 'p1'
Faster
Faster = 'p2'
Fast
Fast = 'p3'
Medium
Medium = 'p4'
Slow
Slow = 'p5'
Slower
Slower = 'p6'
Slowest
Slowest = 'p7'

preset

preset: Annotated[
    Optional[Preset], Option("--preset", "-p")
] = Field(Preset.Medium)

How much time to spend on encoding. Slower options gives better compression

Tune

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
292
293
294
295
296
297
class Tune(str, BrokenEnum):
    None_           = None
    HighQuality     = "hq"
    LowLatency      = "ll"
    UltraLowLatency = "ull"
    Lossless        = "lossless"
None_
None_ = None
HighQuality
HighQuality = 'hq'
LowLatency
LowLatency = 'll'
UltraLowLatency
UltraLowLatency = 'ull'
Lossless
Lossless = 'lossless'

tune

tune: Annotated[Optional[Tune], Option("--tune", "-t")] = (
    Field(Tune.HighQuality)
)

Tune the encoder for a specific tier of performance

Profile

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
304
305
306
307
308
309
class Profile(str, BrokenEnum):
    None_    = None
    Baseline = "baseline"
    Main     = "main"
    High     = "high"
    High444p = "high444p"
None_
None_ = None
Baseline
Baseline = 'baseline'
Main
Main = 'main'
High
High = 'high'
High444p
High444p = 'high444p'

profile

profile: Annotated[
    Optional[Profile], Option("--profile", "-p")
] = Field(Profile.High)

How many features the encoder can use, the playback device must support them

RateControl

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
316
317
318
319
320
class RateControl(str, BrokenEnum):
    None_ = None
    ConstantQuality = "constqp"
    VariableBitrate = "vbr"
    ConstantBitrate = "cbr"
None_
None_ = None
ConstantQuality
ConstantQuality = 'constqp'
VariableBitrate
VariableBitrate = 'vbr'
ConstantBitrate
ConstantBitrate = 'cbr'

rate_control

rate_control: Annotated[
    Optional[RateControl], Option("--rc", "-r", hidden=True)
] = Field(RateControl.VariableBitrate)

Rate control mode of the bitrate

rc_lookahead

rc_lookahead: Annotated[
    Optional[int],
    Option("--rc-lookahead", "-l", hidden=True, min=0),
] = Field(32, ge=0)

Number of frames to look ahead for the rate control

cbr

cbr: Annotated[
    bool,
    Option(
        "--cbr", "-c", " /--no-cbr", " /-nc", hidden=True
    ),
] = Field(False)

Enable Constant Bitrate mode

gpu

gpu: Annotated[
    Optional[int], Option("--gpu", "-g", min=-1)
] = Field(-1, ge=-1)

Use the Nth NVENC capable GPU for encoding, -1 to pick the first device available

cq

cq: Annotated[
    Optional[int], Option("--cq", "-q", min=0)
] = Field(25, ge=0)

(VBR) Similar to CRF, 0 is automatic, 1 is 'lossless', 51 is the worst quality

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
347
348
349
350
351
352
353
354
355
356
357
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-c:v", "h264_nvenc")
    yield every("-b:v", 0)
    yield every("-preset", denum(self.preset))
    yield every("-tune", denum(self.tune))
    yield every("-profile:v", denum(self.profile))
    yield every("-rc", denum(self.rate_control))
    yield every("-rc-lookahead", self.rc_lookahead)
    yield every("-cbr", int(self.cbr))
    yield every("-cq", self.cq)
    yield every("-gpu", self.gpu)

FFmpegVideoCodecH264_QSV

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
360
361
class FFmpegVideoCodecH264_QSV(FFmpegModuleBase):
    ... # Todo

FFmpegVideoCodecH264_AMF

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
364
365
class FFmpegVideoCodecH264_AMF(FFmpegModuleBase):
    ... # Todo

FFmpegVideoCodecH265

Bases: FFmpegModuleBase

Use [bold orange3][link=https://www.videolan.org/developers/x265.html]VideoLAN's[/link][/] [blue][link=https://trac.ffmpeg.org/wiki/Encode/H.265]libx265[/link][/]

Source code in Broken/Externals/FFmpeg.py
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
class FFmpegVideoCodecH265(FFmpegModuleBase):
    """Use [bold orange3][link=https://www.videolan.org/developers/x265.html]VideoLAN's[/link][/] [blue][link=https://trac.ffmpeg.org/wiki/Encode/H.265]libx265[/link][/]"""
    type: Annotated[Literal["h265"], BrokenTyper.exclude()] = "h265"

    crf: Annotated[Optional[int],
        Option("--crf", "-c", min=0, max=51)] = \
        Field(25, ge=0, le=51)
    """Constant Rate Factor (perceptual quality). 0 is lossless, 51 is the worst quality"""

    bitrate: Annotated[Optional[int],
        Option("--bitrate", "-b", min=0)] = \
        Field(None, ge=1)
    """Bitrate in kilobits per second, the higher the better quality and file size"""

    class Preset(str, BrokenEnum):
        None_     = None
        UltraFast = "ultrafast"
        SuperFast = "superfast"
        VeryFast  = "veryfast"
        Faster    = "faster"
        Fast      = "fast"
        Medium    = "medium"
        Slow      = "slow"
        Slower    = "slower"
        VerySlow  = "veryslow"

    preset: Annotated[Optional[Preset],
        Option("--preset", "-p")] = \
        Field(Preset.Slow)
    """How much time to spend on encoding. Slower options gives better compression"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-c:v", "libx265")
        yield every("-preset", denum(self.preset))
        yield every("-crf", self.crf)
        yield every("-b:v", self.bitrate)

type

type: Annotated[Literal["h265"], BrokenTyper.exclude()] = (
    "h265"
)

crf

crf: Annotated[
    Optional[int], Option("--crf", "-c", min=0, max=51)
] = Field(25, ge=0, le=51)

Constant Rate Factor (perceptual quality). 0 is lossless, 51 is the worst quality

bitrate

bitrate: Annotated[
    Optional[int], Option("--bitrate", "-b", min=0)
] = Field(None, ge=1)

Bitrate in kilobits per second, the higher the better quality and file size

Preset

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
383
384
385
386
387
388
389
390
391
392
393
class Preset(str, BrokenEnum):
    None_     = None
    UltraFast = "ultrafast"
    SuperFast = "superfast"
    VeryFast  = "veryfast"
    Faster    = "faster"
    Fast      = "fast"
    Medium    = "medium"
    Slow      = "slow"
    Slower    = "slower"
    VerySlow  = "veryslow"
None_
None_ = None
UltraFast
UltraFast = 'ultrafast'
SuperFast
SuperFast = 'superfast'
VeryFast
VeryFast = 'veryfast'
Faster
Faster = 'faster'
Fast
Fast = 'fast'
Medium
Medium = 'medium'
Slow
Slow = 'slow'
Slower
Slower = 'slower'
VerySlow
VerySlow = 'veryslow'

preset

preset: Annotated[
    Optional[Preset], Option("--preset", "-p")
] = Field(Preset.Slow)

How much time to spend on encoding. Slower options gives better compression

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
400
401
402
403
404
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-c:v", "libx265")
    yield every("-preset", denum(self.preset))
    yield every("-crf", self.crf)
    yield every("-b:v", self.bitrate)

FFmpegVideoCodecH265_NVENC

Bases: FFmpegVideoCodecH265

Use [bold green][link=https://en.wikipedia.org/wiki/Nvidia_NVENC]NVIDIA[/link][/] [blue][link=https://trac.ffmpeg.org/wiki/HWAccelIntro]NVENC H.265[/link][/]

Source code in Broken/Externals/FFmpeg.py
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
class FFmpegVideoCodecH265_NVENC(FFmpegVideoCodecH265):
    """Use [bold green][link=https://en.wikipedia.org/wiki/Nvidia_NVENC]NVIDIA[/link][/] [blue][link=https://trac.ffmpeg.org/wiki/HWAccelIntro]NVENC H.265[/link][/]"""
    type: Annotated[Literal["h265-nvenc"], BrokenTyper.exclude()] = "h265-nvenc"

    class Preset(str, BrokenEnum):
        HighQuality2Passes        = "slow"
        HighQuality1Pass          = "medium"
        HighPerformance1Pass      = "fast"
        HighPerformance           = "hp"
        HighQuality               = "hq"
        Balanced                  = "bd"
        LowLatency                = "ll"
        LowLatencyHighQuality     = "llhq"
        LowLatencyHighPerformance = "llhp"
        Lossless                  = "lossless"
        LosslessHighPerformance   = "losslesshp"
        Fastest                   = "p1"
        Faster                    = "p2"
        Fast                      = "p3"
        Medium                    = "p4"
        Slow                      = "p5"
        Slower                    = "p6"
        Slowest                   = "p7"

    preset: Annotated[Preset,
        Option("--preset", "-p")] = \
        Field(Preset.Medium)
    """How much time to spend on encoding. Slower options gives better compression"""

    class Tune(str, BrokenEnum):
        None_           = None
        HighQuality     = "hq"
        LowLatency      = "ll"
        UltraLowLatency = "ull"
        Lossless        = "lossless"

    tune: Annotated[Optional[Tune],
        Option("--tune", "-t")] = \
        Field(Tune.HighQuality)

    class Profile(str, BrokenEnum):
        None_  = None
        Main   = "main"
        Main10 = "main10"
        ReXT   = "rext"

    profile: Annotated[Optional[Profile],
        Option("--profile", "-p")] = \
        Field(Profile.Main)

    class Tier(str, BrokenEnum):
        None_ = None
        Main  = "main"
        High  = "high"

    tier: Annotated[Optional[Tier],
        Option("--tier", "-t")] = \
        Field(Tier.High)

    class RateControl(str, BrokenEnum):
        None_           = None
        ConstantQuality = "constqp"
        VariableBitrate = "vbr"
        ConstantBitrate = "cbr"

    rate_control: Annotated[Optional[RateControl],
        Option("--rc", "-r", hidden=True)] = \
        Field(RateControl.VariableBitrate)

    rc_lookahead: Annotated[Optional[int],
        Option("--rc-lookahead", "-l", hidden=True)] = \
        Field(10, ge=1)
    """Number of frames to look ahead for the rate control"""

    cbr: Annotated[bool,
        Option("--cbr", "-c", " /--vbr", " /-v", hidden=True)] = \
        Field(False)
    """Use Constant Bitrate mode"""

    gpu: Annotated[Optional[int],
        Option("--gpu", "-g", min=-1)] = \
        Field(-1, ge=-1)
    """Use the Nth NVENC capable GPU for encoding, -1 to pick the first device available"""

    cq: Annotated[int,
        Option("--cq", "-q", min=0)] = \
        Field(25, ge=0)
    """(VBR) Similar to CRF, 0 is automatic, 1 is 'lossless', 51 is the worst quality"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-c:v", "hevc_nvenc")
        yield every("-preset", denum(self.preset))
        yield every("-tune", denum(self.tune))
        yield every("-profile:v", denum(self.profile))
        yield every("-tier", denum(self.tier))
        yield every("-rc", denum(self.rate_control))
        yield every("-rc-lookahead", self.rc_lookahead)
        yield every("-cbr", int(self.cbr))
        yield every("-cq", self.cq)
        yield every("-gpu", self.gpu)

type

type: Annotated[
    Literal["h265-nvenc"], BrokenTyper.exclude()
] = "h265-nvenc"

Preset

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
class Preset(str, BrokenEnum):
    HighQuality2Passes        = "slow"
    HighQuality1Pass          = "medium"
    HighPerformance1Pass      = "fast"
    HighPerformance           = "hp"
    HighQuality               = "hq"
    Balanced                  = "bd"
    LowLatency                = "ll"
    LowLatencyHighQuality     = "llhq"
    LowLatencyHighPerformance = "llhp"
    Lossless                  = "lossless"
    LosslessHighPerformance   = "losslesshp"
    Fastest                   = "p1"
    Faster                    = "p2"
    Fast                      = "p3"
    Medium                    = "p4"
    Slow                      = "p5"
    Slower                    = "p6"
    Slowest                   = "p7"
HighQuality2Passes
HighQuality2Passes = 'slow'
HighQuality1Pass
HighQuality1Pass = 'medium'
HighPerformance1Pass
HighPerformance1Pass = 'fast'
HighPerformance
HighPerformance = 'hp'
HighQuality
HighQuality = 'hq'
Balanced
Balanced = 'bd'
LowLatency
LowLatency = 'll'
LowLatencyHighQuality
LowLatencyHighQuality = 'llhq'
LowLatencyHighPerformance
LowLatencyHighPerformance = 'llhp'
Lossless
Lossless = 'lossless'
LosslessHighPerformance
LosslessHighPerformance = 'losslesshp'
Fastest
Fastest = 'p1'
Faster
Faster = 'p2'
Fast
Fast = 'p3'
Medium
Medium = 'p4'
Slow
Slow = 'p5'
Slower
Slower = 'p6'
Slowest
Slowest = 'p7'

preset

preset: Annotated[Preset, Option("--preset", "-p")] = Field(
    Preset.Medium
)

How much time to spend on encoding. Slower options gives better compression

Tune

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
437
438
439
440
441
442
class Tune(str, BrokenEnum):
    None_           = None
    HighQuality     = "hq"
    LowLatency      = "ll"
    UltraLowLatency = "ull"
    Lossless        = "lossless"
None_
None_ = None
HighQuality
HighQuality = 'hq'
LowLatency
LowLatency = 'll'
UltraLowLatency
UltraLowLatency = 'ull'
Lossless
Lossless = 'lossless'

tune

tune: Annotated[Optional[Tune], Option("--tune", "-t")] = (
    Field(Tune.HighQuality)
)

Profile

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
448
449
450
451
452
class Profile(str, BrokenEnum):
    None_  = None
    Main   = "main"
    Main10 = "main10"
    ReXT   = "rext"
None_
None_ = None
Main
Main = 'main'
Main10
Main10 = 'main10'
ReXT
ReXT = 'rext'

profile

profile: Annotated[
    Optional[Profile], Option("--profile", "-p")
] = Field(Profile.Main)

Tier

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
458
459
460
461
class Tier(str, BrokenEnum):
    None_ = None
    Main  = "main"
    High  = "high"
None_
None_ = None
Main
Main = 'main'
High
High = 'high'

tier

tier: Annotated[Optional[Tier], Option("--tier", "-t")] = (
    Field(Tier.High)
)

RateControl

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
467
468
469
470
471
class RateControl(str, BrokenEnum):
    None_           = None
    ConstantQuality = "constqp"
    VariableBitrate = "vbr"
    ConstantBitrate = "cbr"
None_
None_ = None
ConstantQuality
ConstantQuality = 'constqp'
VariableBitrate
VariableBitrate = 'vbr'
ConstantBitrate
ConstantBitrate = 'cbr'

rate_control

rate_control: Annotated[
    Optional[RateControl], Option("--rc", "-r", hidden=True)
] = Field(RateControl.VariableBitrate)

rc_lookahead

rc_lookahead: Annotated[
    Optional[int],
    Option("--rc-lookahead", "-l", hidden=True),
] = Field(10, ge=1)

Number of frames to look ahead for the rate control

cbr

cbr: Annotated[
    bool,
    Option("--cbr", "-c", " /--vbr", " /-v", hidden=True),
] = Field(False)

Use Constant Bitrate mode

gpu

gpu: Annotated[
    Optional[int], Option("--gpu", "-g", min=-1)
] = Field(-1, ge=-1)

Use the Nth NVENC capable GPU for encoding, -1 to pick the first device available

cq

cq: Annotated[int, Option("--cq", "-q", min=0)] = Field(
    25, ge=0
)

(VBR) Similar to CRF, 0 is automatic, 1 is 'lossless', 51 is the worst quality

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
497
498
499
500
501
502
503
504
505
506
507
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-c:v", "hevc_nvenc")
    yield every("-preset", denum(self.preset))
    yield every("-tune", denum(self.tune))
    yield every("-profile:v", denum(self.profile))
    yield every("-tier", denum(self.tier))
    yield every("-rc", denum(self.rate_control))
    yield every("-rc-lookahead", self.rc_lookahead)
    yield every("-cbr", int(self.cbr))
    yield every("-cq", self.cq)
    yield every("-gpu", self.gpu)

FFmpegVideoCodecH265_QSV

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
510
511
class FFmpegVideoCodecH265_QSV(FFmpegModuleBase):
    ... # Todo

FFmpegVideoCodecH265_AMF

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
514
515
class FFmpegVideoCodecH265_AMF(FFmpegModuleBase):
    ... # Todo

FFmpegVideoCodecVP9

Bases: FFmpegModuleBase

Use [blue][link=https://trac.ffmpeg.org/wiki/Encode/VP9]libvpx-vp9[/link][/] for VP9 encoding

Source code in Broken/Externals/FFmpeg.py
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
class FFmpegVideoCodecVP9(FFmpegModuleBase):
    """Use [blue][link=https://trac.ffmpeg.org/wiki/Encode/VP9]libvpx-vp9[/link][/] for VP9 encoding"""
    type: Annotated[Literal["vp9"], BrokenTyper.exclude()] = "vp9"

    crf: Annotated[int,
        Option("--crf", "-c", min=1, max=63)] = \
        Field(30, ge=1, le=64)
    """Constant Rate Factor (0-63). Lower values mean better quality, recommended (15-31)
    [blue link=https://trac.ffmpeg.org/wiki/Encode/VP9#constantq]→ Documentation[/]"""

    speed: Annotated[int,
        Option("--speed", "-s", min=1, max=6)] = \
        Field(4, ge=1, le=6)
    """Speed level (0-6). Higher values yields faster encoding but innacuracies in rate control
    [blue link=https://trac.ffmpeg.org/wiki/Encode/VP9#CPUUtilizationSpeed]→ Documentation[/]"""

    class Deadline(str, BrokenEnum):
        Good     = "good"
        Best     = "best"
        Realtime = "realtime"

    deadline: Annotated[Deadline,
        Option("--deadline", "-d")] = \
        Field(Deadline.Good)
    """Tweak the encoding time philosophy for better quality or faster encoding
    [blue link=https://trac.ffmpeg.org/wiki/Encode/VP9#DeadlineQuality]→ Documentation[/]"""

    row_multithreading: Annotated[bool,
        Option("--row-multithreading", "-rmt", " /--no-row-multithreading", " /-rmt")] = \
        Field(True)
    """Faster encodes by splitting rows into multiple threads
    [blue link=https://trac.ffmpeg.org/wiki/Encode/VP9#rowmt]→ Documentation[/]"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-c:v", "libvpx-vp9")
        yield ("-crf", self.crf)
        yield ("-b:v", 0)
        yield ("-deadline", denum(self.deadline))
        yield ("-cpu-used", self.speed)
        yield ("-row-mt", "1") * self.row_multithreading

type

type: Annotated[Literal["vp9"], BrokenTyper.exclude()] = (
    "vp9"
)

crf

crf: Annotated[
    int, Option("--crf", "-c", min=1, max=63)
] = Field(30, ge=1, le=64)

Constant Rate Factor (0-63). Lower values mean better quality, recommended (15-31) [blue link=https://trac.ffmpeg.org/wiki/Encode/VP9#constantq]→ Documentation[/]

speed

speed: Annotated[
    int, Option("--speed", "-s", min=1, max=6)
] = Field(4, ge=1, le=6)

Speed level (0-6). Higher values yields faster encoding but innacuracies in rate control [blue link=https://trac.ffmpeg.org/wiki/Encode/VP9#CPUUtilizationSpeed]→ Documentation[/]

Deadline

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
535
536
537
538
class Deadline(str, BrokenEnum):
    Good     = "good"
    Best     = "best"
    Realtime = "realtime"
Good
Good = 'good'
Best
Best = 'best'
Realtime
Realtime = 'realtime'

deadline

deadline: Annotated[
    Deadline, Option("--deadline", "-d")
] = Field(Deadline.Good)

Tweak the encoding time philosophy for better quality or faster encoding [blue link=https://trac.ffmpeg.org/wiki/Encode/VP9#DeadlineQuality]→ Documentation[/]

row_multithreading

row_multithreading: Annotated[
    bool,
    Option(
        "--row-multithreading",
        "-rmt",
        " /--no-row-multithreading",
        " /-rmt",
    ),
] = Field(True)

Faster encodes by splitting rows into multiple threads [blue link=https://trac.ffmpeg.org/wiki/Encode/VP9#rowmt]→ Documentation[/]

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
552
553
554
555
556
557
558
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-c:v", "libvpx-vp9")
    yield ("-crf", self.crf)
    yield ("-b:v", 0)
    yield ("-deadline", denum(self.deadline))
    yield ("-cpu-used", self.speed)
    yield ("-row-mt", "1") * self.row_multithreading

FFmpegVideoCodecAV1_SVT

Bases: FFmpegModuleBase

Use [bold orange3][link=https://gitlab.com/AOMediaCodec/SVT-AV1]AOM's[/link][/] [blue][link=https://www.ffmpeg.org/ffmpeg-all.html#libsvtav1]SVT-AV1[/link][/]

Source code in Broken/Externals/FFmpeg.py
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
class FFmpegVideoCodecAV1_SVT(FFmpegModuleBase):
    """Use [bold orange3][link=https://gitlab.com/AOMediaCodec/SVT-AV1]AOM's[/link][/] [blue][link=https://www.ffmpeg.org/ffmpeg-all.html#libsvtav1]SVT-AV1[/link][/]"""
    type: Annotated[Literal["libsvtav1"], BrokenTyper.exclude()] = "libsvtav1"

    crf: Annotated[int,
        Option("--crf", "-c", min=1, max=63)] = \
        Field(25, ge=1, le=63)
    """Constant Rate Factor (0-63). Lower values mean better quality
    [blue link=https://trac.ffmpeg.org/wiki/Encode/AV1#CRF]→ Documentation[/]"""

    preset: Annotated[int,
        Option("--preset", "-p", min=1, max=8)] = \
        Field(3, ge=1, le=8)
    """The speed of the encoding, 0 is slowest, 8 is fastest. Decreases compression efficiency.
    [blue link=https://trac.ffmpeg.org/wiki/Encode/AV1#Presetsandtunes]→ Documentation[/]"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-c:v", "libsvtav1")
        yield ("-crf", self.crf)
        yield ("-preset", self.preset)
        yield ("-svtav1-params", "tune=0")

type

type: Annotated[
    Literal["libsvtav1"], BrokenTyper.exclude()
] = "libsvtav1"

crf

crf: Annotated[
    int, Option("--crf", "-c", min=1, max=63)
] = Field(25, ge=1, le=63)

Constant Rate Factor (0-63). Lower values mean better quality [blue link=https://trac.ffmpeg.org/wiki/Encode/AV1#CRF]→ Documentation[/]

preset

preset: Annotated[
    int, Option("--preset", "-p", min=1, max=8)
] = Field(3, ge=1, le=8)

The speed of the encoding, 0 is slowest, 8 is fastest. Decreases compression efficiency. [blue link=https://trac.ffmpeg.org/wiki/Encode/AV1#Presetsandtunes]→ Documentation[/]

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
578
579
580
581
582
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-c:v", "libsvtav1")
    yield ("-crf", self.crf)
    yield ("-preset", self.preset)
    yield ("-svtav1-params", "tune=0")

FFmpegVideoCodecAV1_RAV1E

Bases: FFmpegModuleBase

Use [bold orange3][link=https://github.com/xiph/rav1e]Xiph's[/link][/] [blue][link=https://www.ffmpeg.org/ffmpeg-all.html#librav1e]RAV1E AV1[/link][/]

Source code in Broken/Externals/FFmpeg.py
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
class FFmpegVideoCodecAV1_RAV1E(FFmpegModuleBase):
    """Use [bold orange3][link=https://github.com/xiph/rav1e]Xiph's[/link][/] [blue][link=https://www.ffmpeg.org/ffmpeg-all.html#librav1e]RAV1E AV1[/link][/]"""
    type: Annotated[Literal["librav1e"], BrokenTyper.exclude()] = "librav1e"

    qp: Annotated[int,
        Option("--qp", "-q", min=-1)] = \
        Field(80, ge=-1)
    """Constant quantizer mode (from -1 to 255). Smaller values are higher quality"""

    speed: Annotated[int,
        Option("--speed", "-s", min=1, max=10)] = \
        Field(4, ge=1, le=10)
    """What speed preset to use (from -1 to 10) (default -1)"""

    tile_rows: Annotated[int,
        Option("--tile-rows", "-tr", min=-1)] = \
        Field(4, ge=-1)
    """Number of tile rows to encode with (from -1 to I64_MAX) (default 0)"""

    tile_columns: Annotated[int,
        Option("--tile-columns", "-tc", min=-1)] = \
        Field(4, ge=-1)
    """Number of tile columns to encode with (from -1 to I64_MAX) (default 0)"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-c:v", "librav1e")
        yield ("-qp", self.qp)
        yield ("-speed", self.speed)
        yield ("-tile-rows", self.tile_rows)
        yield ("-tile-columns", self.tile_columns)

type

type: Annotated[
    Literal["librav1e"], BrokenTyper.exclude()
] = "librav1e"

qp

qp: Annotated[int, Option("--qp", "-q", min=-1)] = Field(
    80, ge=-1
)

Constant quantizer mode (from -1 to 255). Smaller values are higher quality

speed

speed: Annotated[
    int, Option("--speed", "-s", min=1, max=10)
] = Field(4, ge=1, le=10)

What speed preset to use (from -1 to 10) (default -1)

tile_rows

tile_rows: Annotated[
    int, Option("--tile-rows", "-tr", min=-1)
] = Field(4, ge=-1)

Number of tile rows to encode with (from -1 to I64_MAX) (default 0)

tile_columns

tile_columns: Annotated[
    int, Option("--tile-columns", "-tc", min=-1)
] = Field(4, ge=-1)

Number of tile columns to encode with (from -1 to I64_MAX) (default 0)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
610
611
612
613
614
615
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-c:v", "librav1e")
    yield ("-qp", self.qp)
    yield ("-speed", self.speed)
    yield ("-tile-rows", self.tile_rows)
    yield ("-tile-columns", self.tile_columns)

FFmpegVideoCodecAV1_NVENC

Bases: FFmpegModuleBase

Use [bold green][link=https://en.wikipedia.org/wiki/Nvidia_NVENC]NVIDIA[/link][/] [blue][link=https://trac.ffmpeg.org/wiki/Encode/AV1]NVENC AV1[/link][/] dim light_coral[/]

Source code in Broken/Externals/FFmpeg.py
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
class FFmpegVideoCodecAV1_NVENC(FFmpegModuleBase):
    """Use [bold green][link=https://en.wikipedia.org/wiki/Nvidia_NVENC]NVIDIA[/link][/] [blue][link=https://trac.ffmpeg.org/wiki/Encode/AV1]NVENC AV1[/link][/] [dim light_coral](RTX 4000+ GPU)[/]"""
    type: Annotated[Literal["av1-nvenc"], BrokenTyper.exclude()] = "av1-nvenc"

    class Preset(str, BrokenEnum):
        Default              = "default"
        HighQuality2Passes   = "slow"
        HighQuality1Pass     = "medium"
        HighPerformance1Pass = "fast"
        Fastest              = "p1"
        Faster               = "p2"
        Fast                 = "p3"
        Medium               = "p4"
        Slow                 = "p5"
        Slower               = "p6"
        Slowest              = "p7"

    preset: Annotated[Preset,
        Option("--preset", "-p")] = \
        Field(Preset.Slow)
    """How much time to spend on encoding. Slower options gives better compression"""

    class Tune(str, BrokenEnum):
        HighQuality      = "hq"
        LowLatency       = "ll"
        UltraLowLatency  = "ull"
        Lossless         = "lossless"

    tune: Annotated[Optional[Tune],
        Option("--tune", "-t")] = \
        Field(Tune.HighQuality)
    """Tune the encoder for a specific tier of performance"""

    class RateControl(str, BrokenEnum):
        None_           = None
        ConstantQuality = "constqp"
        VariableBitrate = "vbr"
        ConstantBitrate = "cbr"

    rate_control: Annotated[Optional[RateControl],
        Option("--rc", "-r", hidden=True)] = \
        Field(RateControl.VariableBitrate)

    class Multipass(str, BrokenEnum):
        None_    = None
        Disabled = "disabled"
        Quarter  = "qres"
        Full     = "fullres"

    multipass: Annotated[Optional[Multipass],
        Option("--multipass", "-m", hidden=True)] = \
        Field(Multipass.Full)

    tile_rows: Annotated[Optional[int],
        Option("--tile-rows", "-tr", min=1, max=64)] = \
        Field(2, ge=1, le=64)
    """Number of encoding tile rows, similar to -row-mt"""

    tile_columns: Annotated[Optional[int],
        Option("--tile-columns", "-tc", min=1, max=64)] = \
        Field(2, ge=1, le=64)
    """Number of encoding tile columns, similar to -col-mt"""

    rc_lookahead: Annotated[Optional[int],
        Option("--rc-lookahead", "-l", hidden=True)] = \
        Field(10, ge=1)
    """Number of frames to look ahead for the rate control"""

    gpu: Annotated[Optional[int],
        Option("--gpu", "-g", min=-1)] = \
        Field(-1, ge=-1)
    """Use the Nth NVENC capable GPU for encoding, -1 to pick the first device available"""

    cq: int = Field(25, ge=1)
    """Set the Constant Quality factor in a Variable Bitrate mode (similar to -crf)"""

    cq: Annotated[Optional[int],
        Option("--cq", "-q", min=0)] = \
        Field(25, ge=0)
    """Set the Constant Quality factor in a Variable Bitrate mode (similar to -crf)"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-c:v", "av1_nvenc")
        yield every("-preset", denum(self.preset))
        yield every("-tune", denum(self.tune))
        yield every("-rc", denum(self.rate_control))
        yield every("-rc-lookahead", self.rc_lookahead)
        yield every("-cq", self.cq)
        yield every("-gpu", self.gpu)

type

type: Annotated[
    Literal["av1-nvenc"], BrokenTyper.exclude()
] = "av1-nvenc"

Preset

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
623
624
625
626
627
628
629
630
631
632
633
634
class Preset(str, BrokenEnum):
    Default              = "default"
    HighQuality2Passes   = "slow"
    HighQuality1Pass     = "medium"
    HighPerformance1Pass = "fast"
    Fastest              = "p1"
    Faster               = "p2"
    Fast                 = "p3"
    Medium               = "p4"
    Slow                 = "p5"
    Slower               = "p6"
    Slowest              = "p7"
Default
Default = 'default'
HighQuality2Passes
HighQuality2Passes = 'slow'
HighQuality1Pass
HighQuality1Pass = 'medium'
HighPerformance1Pass
HighPerformance1Pass = 'fast'
Fastest
Fastest = 'p1'
Faster
Faster = 'p2'
Fast
Fast = 'p3'
Medium
Medium = 'p4'
Slow
Slow = 'p5'
Slower
Slower = 'p6'
Slowest
Slowest = 'p7'

preset

preset: Annotated[Preset, Option("--preset", "-p")] = Field(
    Preset.Slow
)

How much time to spend on encoding. Slower options gives better compression

Tune

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
641
642
643
644
645
class Tune(str, BrokenEnum):
    HighQuality      = "hq"
    LowLatency       = "ll"
    UltraLowLatency  = "ull"
    Lossless         = "lossless"
HighQuality
HighQuality = 'hq'
LowLatency
LowLatency = 'll'
UltraLowLatency
UltraLowLatency = 'ull'
Lossless
Lossless = 'lossless'

tune

tune: Annotated[Optional[Tune], Option("--tune", "-t")] = (
    Field(Tune.HighQuality)
)

Tune the encoder for a specific tier of performance

RateControl

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
652
653
654
655
656
class RateControl(str, BrokenEnum):
    None_           = None
    ConstantQuality = "constqp"
    VariableBitrate = "vbr"
    ConstantBitrate = "cbr"
None_
None_ = None
ConstantQuality
ConstantQuality = 'constqp'
VariableBitrate
VariableBitrate = 'vbr'
ConstantBitrate
ConstantBitrate = 'cbr'

rate_control

rate_control: Annotated[
    Optional[RateControl], Option("--rc", "-r", hidden=True)
] = Field(RateControl.VariableBitrate)

Multipass

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
662
663
664
665
666
class Multipass(str, BrokenEnum):
    None_    = None
    Disabled = "disabled"
    Quarter  = "qres"
    Full     = "fullres"
None_
None_ = None
Disabled
Disabled = 'disabled'
Quarter
Quarter = 'qres'
Full
Full = 'fullres'

multipass

multipass: Annotated[
    Optional[Multipass],
    Option("--multipass", "-m", hidden=True),
] = Field(Multipass.Full)

tile_rows

tile_rows: Annotated[
    Optional[int],
    Option("--tile-rows", "-tr", min=1, max=64),
] = Field(2, ge=1, le=64)

Number of encoding tile rows, similar to -row-mt

tile_columns

tile_columns: Annotated[
    Optional[int],
    Option("--tile-columns", "-tc", min=1, max=64),
] = Field(2, ge=1, le=64)

Number of encoding tile columns, similar to -col-mt

rc_lookahead

rc_lookahead: Annotated[
    Optional[int],
    Option("--rc-lookahead", "-l", hidden=True),
] = Field(10, ge=1)

Number of frames to look ahead for the rate control

gpu

gpu: Annotated[
    Optional[int], Option("--gpu", "-g", min=-1)
] = Field(-1, ge=-1)

Use the Nth NVENC capable GPU for encoding, -1 to pick the first device available

cq

cq: Annotated[
    Optional[int], Option("--cq", "-q", min=0)
] = Field(25, ge=0)

Set the Constant Quality factor in a Variable Bitrate mode (similar to -crf)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
700
701
702
703
704
705
706
707
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-c:v", "av1_nvenc")
    yield every("-preset", denum(self.preset))
    yield every("-tune", denum(self.tune))
    yield every("-rc", denum(self.rate_control))
    yield every("-rc-lookahead", self.rc_lookahead)
    yield every("-cq", self.cq)
    yield every("-gpu", self.gpu)

FFmpegVideoCodecAV1_QSV

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
710
711
class FFmpegVideoCodecAV1_QSV(FFmpegModuleBase):
    ... # Todo

FFmpegVideoCodecAV1_AMF

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
714
715
class FFmpegVideoCodecAV1_AMF(FFmpegModuleBase):
    ... # Todo

FFmpegVideoCodecRawvideo

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
718
719
720
721
722
class FFmpegVideoCodecRawvideo(FFmpegModuleBase):
    type: Annotated[Literal["rawvideo"], BrokenTyper.exclude()] = "rawvideo"

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-c:v", "rawvideo")

type

type: Annotated[
    Literal["rawvideo"], BrokenTyper.exclude()
] = "rawvideo"

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
721
722
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-c:v", "rawvideo")

FFmpegVideoCodecNoVideo

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
725
726
727
728
729
class FFmpegVideoCodecNoVideo(FFmpegModuleBase):
    type: Annotated[Literal["null"], BrokenTyper.exclude()] = "null"

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-c:v", "null")

type

type: Annotated[Literal["null"], BrokenTyper.exclude()] = (
    "null"
)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
728
729
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-c:v", "null")

FFmpegVideoCodecCopy

Bases: FFmpegModuleBase

Source code in Broken/Externals/FFmpeg.py
732
733
734
735
736
class FFmpegVideoCodecCopy(FFmpegModuleBase):
    type: Annotated[Literal["copy"], BrokenTyper.exclude()] = "copy"

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-c:v", "copy")

type

type: Annotated[Literal["copy"], BrokenTyper.exclude()] = (
    "copy"
)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
735
736
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-c:v", "copy")

FFmpegAudioCodecAAC

Bases: FFmpegModuleBase

Use the [blue][link=https://trac.ffmpeg.org/wiki/Encode/AAC]Advanced Audio Codec (AAC)[/link][/]

Source code in Broken/Externals/FFmpeg.py
755
756
757
758
759
760
761
762
763
764
765
766
class FFmpegAudioCodecAAC(FFmpegModuleBase):
    """Use the [blue][link=https://trac.ffmpeg.org/wiki/Encode/AAC]Advanced Audio Codec (AAC)[/link][/]"""
    type: Annotated[Literal["aac"], BrokenTyper.exclude()] = "aac"

    bitrate: Annotated[int,
        Option("--bitrate", "-b", min=1)] = \
        Field(192, ge=1)
    """Bitrate in kilobits per second. This value is shared between all audio channels"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-c:a", "aac")
        yield every("-b:a", f"{self.bitrate}k")

type

type: Annotated[Literal["aac"], BrokenTyper.exclude()] = (
    "aac"
)

bitrate

bitrate: Annotated[
    int, Option("--bitrate", "-b", min=1)
] = Field(192, ge=1)

Bitrate in kilobits per second. This value is shared between all audio channels

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
764
765
766
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-c:a", "aac")
    yield every("-b:a", f"{self.bitrate}k")

FFmpegAudioCodecMP3

Bases: FFmpegModuleBase

Use the [blue][link=https://trac.ffmpeg.org/wiki/Encode/MP3]MPEG Audio Layer 3 (MP3)[/link][/]

Source code in Broken/Externals/FFmpeg.py
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
class FFmpegAudioCodecMP3(FFmpegModuleBase):
    """Use the [blue][link=https://trac.ffmpeg.org/wiki/Encode/MP3]MPEG Audio Layer 3 (MP3)[/link][/]"""
    type: Annotated[Literal["mp3"], BrokenTyper.exclude()] = "mp3"

    bitrate: Annotated[int,
        Option("--bitrate", "-b", min=1)] = \
        Field(192, ge=1)
    """Bitrate in kilobits per second. This value is shared between all audio channels"""

    qscale: Annotated[int,
        Option("--qscale", "-q", min=1)] = \
        Field(2, ge=1)
    """Quality scale, 0-9, Variable Bitrate"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-c:a", "libmp3lame")
        yield every("-b:a", f"{self.bitrate}k")
        yield every("-qscale:a", self.qscale)

type

type: Annotated[Literal["mp3"], BrokenTyper.exclude()] = (
    "mp3"
)

bitrate

bitrate: Annotated[
    int, Option("--bitrate", "-b", min=1)
] = Field(192, ge=1)

Bitrate in kilobits per second. This value is shared between all audio channels

qscale

qscale: Annotated[int, Option("--qscale", "-q", min=1)] = (
    Field(2, ge=1)
)

Quality scale, 0-9, Variable Bitrate

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
783
784
785
786
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-c:a", "libmp3lame")
    yield every("-b:a", f"{self.bitrate}k")
    yield every("-qscale:a", self.qscale)

FFmpegAudioCodecOpus

Bases: FFmpegModuleBase

Use the [blue][link=https://en.wikipedia.org/wiki/Opus_(audio_format)]Opus[/link][/] audio codec

Source code in Broken/Externals/FFmpeg.py
789
790
791
792
793
794
795
796
797
798
799
800
class FFmpegAudioCodecOpus(FFmpegModuleBase):
    """Use the [blue][link=https://en.wikipedia.org/wiki/Opus_(audio_format)]Opus[/link][/] audio codec"""
    type: Annotated[Literal["opus"], BrokenTyper.exclude()] = "opus"

    bitrate: Annotated[int,
        Option("--bitrate", "-b", min=1)] = \
        Field(192, ge=1)
    """Bitrate in kilobits per second. This value is shared between all audio channels"""

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-c:a", "libopus")
        yield every("-b:a", f"{self.bitrate}k")

type

type: Annotated[Literal["opus"], BrokenTyper.exclude()] = (
    "opus"
)

bitrate

bitrate: Annotated[
    int, Option("--bitrate", "-b", min=1)
] = Field(192, ge=1)

Bitrate in kilobits per second. This value is shared between all audio channels

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
798
799
800
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-c:a", "libopus")
    yield every("-b:a", f"{self.bitrate}k")

FFmpegAudioCodecFLAC

Bases: FFmpegModuleBase

Use the [blue][link=https://xiph.org/flac/]Free Lossless Audio Codec (FLAC)[/link][/]

Source code in Broken/Externals/FFmpeg.py
803
804
805
806
807
808
class FFmpegAudioCodecFLAC(FFmpegModuleBase):
    """Use the [blue][link=https://xiph.org/flac/]Free Lossless Audio Codec (FLAC)[/link][/]"""
    type: Annotated[Literal["flac"], BrokenTyper.exclude()] = "flac"

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield every("-c:a", "flac")

type

type: Annotated[Literal["flac"], BrokenTyper.exclude()] = (
    "flac"
)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
807
808
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield every("-c:a", "flac")

FFmpegAudioCodecCopy

Bases: FFmpegModuleBase

Copy the inputs' audio streams to the output

Source code in Broken/Externals/FFmpeg.py
811
812
813
814
815
816
class FFmpegAudioCodecCopy(FFmpegModuleBase):
    """Copy the inputs' audio streams to the output"""
    type: Annotated[Literal["copy"], BrokenTyper.exclude()] = "copy"

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-c:a", "copy")

type

type: Annotated[Literal["copy"], BrokenTyper.exclude()] = (
    "copy"
)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
815
816
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-c:a", "copy")

FFmpegAudioCodecNone

Bases: FFmpegModuleBase

Remove all audio tracks from the output

Source code in Broken/Externals/FFmpeg.py
819
820
821
822
823
824
class FFmpegAudioCodecNone(FFmpegModuleBase):
    """Remove all audio tracks from the output"""
    type: Annotated[Literal["none"], BrokenTyper.exclude()] = "none"

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-an")

type

type: Annotated[Literal["none"], BrokenTyper.exclude()] = (
    "none"
)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
823
824
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-an")

FFmpegAudioCodecEmpty

Bases: FFmpegModuleBase

Adds a silent stereo audio track

Source code in Broken/Externals/FFmpeg.py
827
828
829
830
831
832
833
834
835
836
837
838
class FFmpegAudioCodecEmpty(FFmpegModuleBase):
    """Adds a silent stereo audio track"""
    type: Annotated[Literal["empty"], BrokenTyper.exclude()] = "empty"

    samplerate: Annotated[float,
        Option("--samplerate", "-r", min=1)] = \
        Field(44100, ge=1)

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-f", "lavfi")
        yield ("-t", ffmpeg.time) * bool(ffmpeg.time)
        yield ("-i", f"anullsrc=channel_layout=stereo:sample_rate={self.samplerate}")

type

type: Annotated[Literal["empty"], BrokenTyper.exclude()] = (
    "empty"
)

samplerate

samplerate: Annotated[
    float, Option("--samplerate", "-r", min=1)
] = Field(44100, ge=1)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
835
836
837
838
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-f", "lavfi")
    yield ("-t", ffmpeg.time) * bool(ffmpeg.time)
    yield ("-i", f"anullsrc=channel_layout=stereo:sample_rate={self.samplerate}")

FFmpegPCM

Bases: BrokenEnum

Raw pcm formats ffmpeg -formats | grep PCM

Source code in Broken/Externals/FFmpeg.py
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
class FFmpegPCM(BrokenEnum):
    """Raw pcm formats `ffmpeg -formats | grep PCM`"""
    PCM_FLOAT_32_BITS_BIG_ENDIAN       = "pcm_f32be"
    PCM_FLOAT_32_BITS_LITTLE_ENDIAN    = "pcm_f32le"
    PCM_FLOAT_64_BITS_BIG_ENDIAN       = "pcm_f64be"
    PCM_FLOAT_64_BITS_LITTLE_ENDIAN    = "pcm_f64le"
    PCM_SIGNED_16_BITS_BIG_ENDIAN      = "pcm_s16be"
    PCM_SIGNED_16_BITS_LITTLE_ENDIAN   = "pcm_s16le"
    PCM_SIGNED_24_BITS_BIG_ENDIAN      = "pcm_s24be"
    PCM_SIGNED_24_BITS_LITTLE_ENDIAN   = "pcm_s24le"
    PCM_SIGNED_32_BITS_BIG_ENDIAN      = "pcm_s32be"
    PCM_SIGNED_32_BITS_LITTLE_ENDIAN   = "pcm_s32le"
    PCM_UNSIGNED_16_BITS_BIG_ENDIAN    = "pcm_u16be"
    PCM_UNSIGNED_16_BITS_LITTLE_ENDIAN = "pcm_u16le"
    PCM_UNSIGNED_24_BITS_BIG_ENDIAN    = "pcm_u24be"
    PCM_UNSIGNED_24_BITS_LITTLE_ENDIAN = "pcm_u24le"
    PCM_UNSIGNED_32_BITS_BIG_ENDIAN    = "pcm_u32be"
    PCM_UNSIGNED_32_BITS_LITTLE_ENDIAN = "pcm_u32le"
    PCM_UNSIGNED_8_BITS                = "pcm_u8"
    PCM_SIGNED_8_BITS                  = "pcm_s8"

    @property
    @functools.lru_cache
    def size(self) -> int:
        return int(''.join(filter(str.isdigit, self.value)))//8

    @property
    @functools.lru_cache
    def endian(self) -> str:
        return ("<" if ("le" in self.value) else ">")

    @property
    @functools.lru_cache
    def dtype(self) -> numpy.dtype:
        return numpy.dtype(f"{self.endian}{self.value[4]}{self.size}")

PCM_FLOAT_32_BITS_BIG_ENDIAN

PCM_FLOAT_32_BITS_BIG_ENDIAN = 'pcm_f32be'

PCM_FLOAT_32_BITS_LITTLE_ENDIAN

PCM_FLOAT_32_BITS_LITTLE_ENDIAN = 'pcm_f32le'

PCM_FLOAT_64_BITS_BIG_ENDIAN

PCM_FLOAT_64_BITS_BIG_ENDIAN = 'pcm_f64be'

PCM_FLOAT_64_BITS_LITTLE_ENDIAN

PCM_FLOAT_64_BITS_LITTLE_ENDIAN = 'pcm_f64le'

PCM_SIGNED_16_BITS_BIG_ENDIAN

PCM_SIGNED_16_BITS_BIG_ENDIAN = 'pcm_s16be'

PCM_SIGNED_16_BITS_LITTLE_ENDIAN

PCM_SIGNED_16_BITS_LITTLE_ENDIAN = 'pcm_s16le'

PCM_SIGNED_24_BITS_BIG_ENDIAN

PCM_SIGNED_24_BITS_BIG_ENDIAN = 'pcm_s24be'

PCM_SIGNED_24_BITS_LITTLE_ENDIAN

PCM_SIGNED_24_BITS_LITTLE_ENDIAN = 'pcm_s24le'

PCM_SIGNED_32_BITS_BIG_ENDIAN

PCM_SIGNED_32_BITS_BIG_ENDIAN = 'pcm_s32be'

PCM_SIGNED_32_BITS_LITTLE_ENDIAN

PCM_SIGNED_32_BITS_LITTLE_ENDIAN = 'pcm_s32le'

PCM_UNSIGNED_16_BITS_BIG_ENDIAN

PCM_UNSIGNED_16_BITS_BIG_ENDIAN = 'pcm_u16be'

PCM_UNSIGNED_16_BITS_LITTLE_ENDIAN

PCM_UNSIGNED_16_BITS_LITTLE_ENDIAN = 'pcm_u16le'

PCM_UNSIGNED_24_BITS_BIG_ENDIAN

PCM_UNSIGNED_24_BITS_BIG_ENDIAN = 'pcm_u24be'

PCM_UNSIGNED_24_BITS_LITTLE_ENDIAN

PCM_UNSIGNED_24_BITS_LITTLE_ENDIAN = 'pcm_u24le'

PCM_UNSIGNED_32_BITS_BIG_ENDIAN

PCM_UNSIGNED_32_BITS_BIG_ENDIAN = 'pcm_u32be'

PCM_UNSIGNED_32_BITS_LITTLE_ENDIAN

PCM_UNSIGNED_32_BITS_LITTLE_ENDIAN = 'pcm_u32le'

PCM_UNSIGNED_8_BITS

PCM_UNSIGNED_8_BITS = 'pcm_u8'

PCM_SIGNED_8_BITS

PCM_SIGNED_8_BITS = 'pcm_s8'

size

size: int

endian

endian: str

dtype

dtype: numpy.dtype

FFmpegAudioCodecPCM

Bases: FFmpegModuleBase

Raw pcm formats ffmpeg -formats | grep PCM

Source code in Broken/Externals/FFmpeg.py
878
879
880
881
882
883
884
885
886
887
class FFmpegAudioCodecPCM(FFmpegModuleBase):
    """Raw pcm formats `ffmpeg -formats | grep PCM`"""
    type: Annotated[Literal["pcm"], BrokenTyper.exclude()] = "pcm"

    format: Annotated[FFmpegPCM,
        Option("--format", "-f")] = \
        Field(FFmpegPCM.PCM_FLOAT_32_BITS_LITTLE_ENDIAN)

    def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
        yield ("-c:a", self.format.value, "-f", self.format.value.removeprefix("pcm_"))

type

type: Annotated[Literal["pcm"], BrokenTyper.exclude()] = (
    "pcm"
)

format

format: Annotated[FFmpegPCM, Option("--format", "-f")] = (
    Field(FFmpegPCM.PCM_FLOAT_32_BITS_LITTLE_ENDIAN)
)

command

command(ffmpeg: BrokenFFmpeg) -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
886
887
def command(self, ffmpeg: BrokenFFmpeg) -> Iterable[str]:
    yield ("-c:a", self.format.value, "-f", self.format.value.removeprefix("pcm_"))

FFmpegFilterBase

Bases: BrokenModel, ABC

Source code in Broken/Externals/FFmpeg.py
903
904
905
906
907
908
909
910
class FFmpegFilterBase(BrokenModel, ABC):

    def __str__(self) -> str:
        return self.string()

    @abstractmethod
    def string(self) -> Iterable[str]:
        ...

__str__

__str__() -> str
Source code in Broken/Externals/FFmpeg.py
905
906
def __str__(self) -> str:
    return self.string()

string

string() -> Iterable[str]
Source code in Broken/Externals/FFmpeg.py
908
909
910
@abstractmethod
def string(self) -> Iterable[str]:
    ...

FFmpegFilterScale

Bases: FFmpegFilterBase

Source code in Broken/Externals/FFmpeg.py
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
class FFmpegFilterScale(FFmpegFilterBase):
    type: Annotated[Literal["scale"], BrokenTyper.exclude()] = "scale"

    width: int = Field(gt=0)
    height: int = Field(gt=0)

    class Resample(str, BrokenEnum):
        Bilinear   = "bilinear"
        Nearest    = "neighbor"
        Bicubic    = "bicubic"
        Gaussian   = "gauss"
        Sinc       = "sinc"
        Lanczos    = "lanczos"
        Spline     = "spline"

    resample: Annotated[Resample,
        Option("--resample", "-r")] = \
        Field(Resample.Lanczos)

    def string(self) -> str:
        return f"scale={self.width}x{self.height}:flags={denum(self.resample)}"

type

type: Annotated[Literal["scale"], BrokenTyper.exclude()] = (
    "scale"
)

width

width: int = Field(gt=0)

height

height: int = Field(gt=0)

Resample

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
919
920
921
922
923
924
925
926
class Resample(str, BrokenEnum):
    Bilinear   = "bilinear"
    Nearest    = "neighbor"
    Bicubic    = "bicubic"
    Gaussian   = "gauss"
    Sinc       = "sinc"
    Lanczos    = "lanczos"
    Spline     = "spline"
Bilinear
Bilinear = 'bilinear'
Nearest
Nearest = 'neighbor'
Bicubic
Bicubic = 'bicubic'
Gaussian
Gaussian = 'gauss'
Sinc
Sinc = 'sinc'
Lanczos
Lanczos = 'lanczos'
Spline
Spline = 'spline'

resample

resample: Annotated[
    Resample, Option("--resample", "-r")
] = Field(Resample.Lanczos)

string

string() -> str
Source code in Broken/Externals/FFmpeg.py
932
933
def string(self) -> str:
    return f"scale={self.width}x{self.height}:flags={denum(self.resample)}"

FFmpegFilterVerticalFlip

Bases: FFmpegFilterBase

Source code in Broken/Externals/FFmpeg.py
936
937
938
939
940
class FFmpegFilterVerticalFlip(FFmpegFilterBase):
    type: Annotated[Literal["vflip"], BrokenTyper.exclude()] = "vflip"

    def string(self) -> str:
        return "vflip"

type

type: Annotated[Literal["vflip"], BrokenTyper.exclude()] = (
    "vflip"
)

string

string() -> str
Source code in Broken/Externals/FFmpeg.py
939
940
def string(self) -> str:
    return "vflip"

FFmpegFilterCustom

Bases: FFmpegFilterBase

Source code in Broken/Externals/FFmpeg.py
943
944
945
946
947
948
949
class FFmpegFilterCustom(FFmpegFilterBase):
    type: Annotated[Literal["custom"], BrokenTyper.exclude()] = "custom"

    content: str

    def string(self) -> str:
        return self.content

type

type: Annotated[
    Literal["custom"], BrokenTyper.exclude()
] = "custom"

content

content: str

string

string() -> str
Source code in Broken/Externals/FFmpeg.py
948
949
def string(self) -> str:
    return self.content

FFmpegFilterType

FFmpegFilterType: TypeAlias = Union[
    FFmpegFilterScale,
    FFmpegFilterVerticalFlip,
    FFmpegFilterCustom,
]

BrokenFFmpeg

Bases: BrokenModel, BrokenFluent

đź’Ž Your premium FFmpeg class, serializable, sane defaults, safety

Source code in Broken/Externals/FFmpeg.py
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
class BrokenFFmpeg(BrokenModel, BrokenFluent):
    """đź’Ž Your premium FFmpeg class, serializable, sane defaults, safety"""

    # -------------------------------------------|
    # Re-export classes on BrokenFFmpeg.*

    class Input:
        Path = FFmpegInputPath
        Pipe = FFmpegInputPipe

    class Output:
        Path = FFmpegOutputPath
        Pipe = FFmpegOutputPipe

    class VideoCodec:
        H264       = FFmpegVideoCodecH264
        H264_NVENC = FFmpegVideoCodecH264_NVENC
        H265       = FFmpegVideoCodecH265
        H265_NVENC = FFmpegVideoCodecH265_NVENC
        VP9        = FFmpegVideoCodecVP9
        AV1_SVT    = FFmpegVideoCodecAV1_SVT
        AV1_NVENC  = FFmpegVideoCodecAV1_NVENC
        AV1_RAV1E  = FFmpegVideoCodecAV1_RAV1E
        Rawvideo   = FFmpegVideoCodecRawvideo
        NoVideo    = FFmpegVideoCodecNoVideo
        Copy       = FFmpegVideoCodecCopy

    class AudioCodec:
        AAC   = FFmpegAudioCodecAAC
        MP3   = FFmpegAudioCodecMP3
        Opus  = FFmpegAudioCodecOpus
        FLAC  = FFmpegAudioCodecFLAC
        Copy  = FFmpegAudioCodecCopy
        None_ = FFmpegAudioCodecNone
        Empty = FFmpegAudioCodecEmpty
        PCM   = FFmpegAudioCodecPCM

    class Filter:
        Scale        = FFmpegFilterScale
        VerticalFlip = FFmpegFilterVerticalFlip
        Custom       = FFmpegFilterCustom

    # -------------------------------------------|

    hide_banner: bool = True
    """Hides the compilation information of FFmpeg from the output"""

    shortest: bool = False
    """Ends the video at the shortest input's duration

    [**FFmpeg docs**](https://ffmpeg.org/ffmpeg.html#toc-Advanced-options)
    """

    stream_loop: int = Field(0)
    """Loops the input stream N times to the right. Zero '0' is no loop, one '1' doubles the length"""

    time: float = Field(0.0)
    """If greater than zero, stops encoding at the specified time. `-t` option of FFmpeg"""

    def set_time(self, time: float) -> Self:
        self.time = time
        return self

    vsync: Literal["auto", "passthrough", "cfr", "vfr"] = Field("cfr")
    """
    The video's framerate mode, applied to all subsequent output targets. `-vsync` option of FFmpeg

    - `auto`: FFmpeg default, choses between constant and variable framerate based on muxer support
    - `cfr`: Constant Frame Rate, where frames are droped or duped to precisely match frametimes
    - `vfr`: Variable Frame Rate, static frames are kept, no two frames have the same timestemp
    - `passthrough`: The frames are passed through without modification on their timestamp

    [**FFmpeg docs**](https://ffmpeg.org/ffmpeg.html#Advanced-options)
    """

    class LogLevel(str, BrokenEnum):
        Error   = "error"
        Info    = "info"
        Verbose = "verbose"
        Debug   = "debug"
        Warning = "warning"
        Panic   = "panic"
        Fatal   = "fatal"

    loglevel: Annotated[LogLevel,
        Option("--loglevel", "-log")] = \
        Field(LogLevel.Info)

    class HardwareAcceleration(str, BrokenEnum):
        Auto   = "auto"
        CUDA   = "cuda"
        NVDEC  = "nvdec"
        Vulkan = "vulkan"

    hwaccel: Annotated[Optional[HardwareAcceleration],
        Option("--hwaccel", "-hw")] = \
        Field(None)
    """
    What device to bootstrap, for decoding with hardware acceleration. In practice, it's only useful
    when decoding from a source video file, might cause overhead on pipe input mode

    - `auto`: Finds up the best device to use, more often than not nvdec or cuvid

    TODO: Add the required initializers on the final command per option

    [**FFmpeg docs**](https://trac.ffmpeg.org/wiki/HWAccelIntro)
    """

    threads: Optional[int] = Field(None, ge=0)
    """
    The number of threads the codecs should use (). Generally speaking, more threads drastically
    improves performance, at the cost of worse quality and compression ratios. It's not that bad though. Some
    codecs might not use all available CPU threads. '0' finds the optimal amount automatically

    [**FFmpeg docs**](https://ffmpeg.org/ffmpeg-codecs.html#toc-Codec-Options)
    """

    inputs: list[FFmpegInputType] = Field(default_factory=list)
    """A list of inputs for FFmpeg"""

    filters: list[FFmpegFilterType] = Field(default_factory=list)

    outputs: list[FFmpegOutputType] = Field(default_factory=list)
    """A list of outputs. Yes, FFmpeg natively supports multi-encoding targets"""

    video_codec: Optional[FFmpegVideoCodecType] = Field(default_factory=FFmpegVideoCodecH264)
    """The video codec to use and its configuration"""

    audio_codec: Optional[FFmpegAudioCodecType] = Field(None)
    """The audio codec to use and its configuration"""

    def quiet(self) -> Self:
        self.hide_banner = True
        self.loglevel = "error"
        return self

    # ---------------------------------------------------------------------------------------------|
    # Recycling

    def clear_inputs(self) -> Self:
        self.inputs = list()
        return self

    def clear_filters(self) -> Self:
        self.filters = list()
        return self

    def clear_outputs(self) -> Self:
        self.outputs = list()
        return self

    def clear_video_codec(self) -> Self:
        self.video_codec = None
        return self

    def clear_audio_codec(self) -> Self:
        self.audio_codec = None
        return self

    def clear(self,
        inputs: bool=True,
        filters: bool=True,
        outputs: bool=True,
        video_codec: bool=True,
        audio_codec: bool=True,
    ) -> Self:
        if inputs:      self.clear_inputs()
        if filters:     self.clear_filters()
        if outputs:     self.clear_outputs()
        if video_codec: self.clear_video_codec()
        if audio_codec: self.clear_audio_codec()
        return self

    # ---------------------------------------------------------------------------------------------|
    # Wrappers for all classes

    # Inputs and Outputs

    def add_input(self, input: FFmpegInputType) -> Self:
        self.inputs.append(input)
        return self

    @functools.wraps(FFmpegInputPath)
    def input(self, path: Path, **options) -> Self:
        return self.add_input(FFmpegInputPath(path=path, **options))

    @functools.wraps(FFmpegInputPipe)
    def pipe_input(self, **options) -> Self:
        return self.add_input(FFmpegInputPipe(**options))

    def typer_inputs(self, typer: BrokenTyper) -> None:
        with typer.panel("📦 (FFmpeg) Input"):
            typer.command(FFmpegInputPath, post=self.add_input, name="ipath")
            typer.command(FFmpegInputPipe, post=self.add_input, name="ipipe")

    def add_output(self, output: FFmpegOutputType) -> Self:
        self.outputs.append(output)
        return self

    @functools.wraps(FFmpegOutputPath)
    def output(self, path: Path, **options) -> Self:
        return self.add_output(FFmpegOutputPath(path=path, **options))

    @functools.wraps(FFmpegOutputPipe)
    def pipe_output(self, **options) -> Self:
        return self.add_output(FFmpegOutputPipe(**options))

    def typer_outputs(self, typer: BrokenTyper) -> None:
        with typer.panel("📦 (FFmpeg) Output"):
            typer.command(FFmpegOutputPath, post=self.add_output, name="opath")
            typer.command(FFmpegOutputPipe, post=self.add_output, name="opipe")

    # Video codecs

    def set_video_codec(self, codec: FFmpegVideoCodecType) -> Self:
        self.video_codec = codec
        return self

    @functools.wraps(FFmpegVideoCodecH264)
    def h264(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecH264(**options))

    @functools.wraps(FFmpegVideoCodecH264_NVENC)
    def h264_nvenc(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecH264_NVENC(**options))

    @functools.wraps(FFmpegVideoCodecH265)
    def h265(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecH265(**options))

    @functools.wraps(FFmpegVideoCodecH265_NVENC)
    def h265_nvenc(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecH265_NVENC(**options))

    @functools.wraps(FFmpegVideoCodecVP9)
    def vp9(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecVP9(**options))

    @functools.wraps(FFmpegVideoCodecAV1_SVT)
    def av1_svt(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecAV1_SVT(**options))

    @functools.wraps(FFmpegVideoCodecAV1_NVENC)
    def av1_nvenc(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecAV1_NVENC(**options))

    @functools.wraps(FFmpegVideoCodecAV1_RAV1E)
    def av1_rav1e(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecAV1_RAV1E(**options))

    @functools.wraps(FFmpegVideoCodecRawvideo)
    def rawvideo(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecRawvideo(**options))

    @functools.wraps(FFmpegVideoCodecCopy)
    def copy_video(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecCopy(**options))

    @functools.wraps(FFmpegVideoCodecNoVideo)
    def no_video(self, **options) -> Self:
        return self.set_video_codec(FFmpegVideoCodecNoVideo(**options))

    def typer_vcodecs(self, typer: BrokenTyper) -> None:
        with typer.panel("📦 (Exporting) Video encoder"):
            typer.command(FFmpegVideoCodecH264,       post=self.set_video_codec, name="h264")
            typer.command(FFmpegVideoCodecH264_NVENC, post=self.set_video_codec, name="h264-nvenc")
            typer.command(FFmpegVideoCodecH265,       post=self.set_video_codec, name="h265")
            typer.command(FFmpegVideoCodecH265_NVENC, post=self.set_video_codec, name="h265-nvenc")
            typer.command(FFmpegVideoCodecVP9,        post=self.set_video_codec, name="vp9", hidden=True)
            typer.command(FFmpegVideoCodecAV1_SVT,    post=self.set_video_codec, name="av1-svt")
            typer.command(FFmpegVideoCodecAV1_NVENC,  post=self.set_video_codec, name="av1-nvenc")
            typer.command(FFmpegVideoCodecAV1_RAV1E,  post=self.set_video_codec, name="av1-rav1e")
            typer.command(FFmpegVideoCodecRawvideo,   post=self.set_video_codec, name="rawvideo", hidden=True)
            typer.command(FFmpegVideoCodecCopy,       post=self.set_video_codec, name="video-copy", hidden=True)
            typer.command(FFmpegVideoCodecNoVideo,    post=self.set_video_codec, name="video-none", hidden=True)

    # Audio codecs

    def set_audio_codec(self, codec: FFmpegAudioCodecType) -> Self:
        self.audio_codec = codec
        return self

    @functools.wraps(FFmpegAudioCodecAAC)
    def aac(self, **options) -> Self:
        return self.set_audio_codec(FFmpegAudioCodecAAC(**options))

    @functools.wraps(FFmpegAudioCodecMP3)
    def mp3(self, **options) -> Self:
        return self.set_audio_codec(FFmpegAudioCodecMP3(**options))

    @functools.wraps(FFmpegAudioCodecOpus)
    def opus(self, **options) -> Self:
        return self.set_audio_codec(FFmpegAudioCodecOpus(**options))

    @functools.wraps(FFmpegAudioCodecFLAC)
    def flac(self, **options) -> Self:
        return self.set_audio_codec(FFmpegAudioCodecFLAC(**options))

    @functools.wraps(FFmpegAudioCodecPCM)
    def pcm(self, format: FFmpegAudioCodecPCM="pcm_f32le") -> Self:
        return self.set_audio_codec(FFmpegAudioCodecPCM(format=format))

    @functools.wraps(FFmpegAudioCodecCopy)
    def copy_audio(self, **options) -> Self:
        return self.set_audio_codec(FFmpegAudioCodecCopy(**options))

    @functools.wraps(FFmpegAudioCodecNone)
    def no_audio(self, **options) -> Self:
        return self.set_audio_codec(FFmpegAudioCodecNone(**options))

    @functools.wraps(FFmpegAudioCodecEmpty)
    def empty_audio(self, **options) -> Self:
        return self.set_audio_codec(FFmpegAudioCodecEmpty(**options))

    def typer_acodecs(self, typer: BrokenTyper) -> None:
        with typer.panel("📦 (Exporting) Audio encoder"):
            typer.command(FFmpegAudioCodecAAC,   post=self.set_audio_codec, name="aac")
            typer.command(FFmpegAudioCodecMP3,   post=self.set_audio_codec, name="mp3")
            typer.command(FFmpegAudioCodecOpus,  post=self.set_audio_codec, name="opus")
            typer.command(FFmpegAudioCodecFLAC,  post=self.set_audio_codec, name="flac")
            typer.command(FFmpegAudioCodecCopy,  post=self.set_audio_codec, name="audio-copy")
            typer.command(FFmpegAudioCodecNone,  post=self.set_audio_codec, name="audio-none")
            typer.command(FFmpegAudioCodecEmpty, post=self.set_audio_codec, name="audio-empty")

    # Filters

    def add_filter(self, filter: FFmpegFilterType) -> Self:
        self.filters.append(filter)
        return self

    @functools.wraps(FFmpegFilterScale)
    def scale(self, **options) -> Self:
        return self.add_filter(FFmpegFilterScale(**options))

    @functools.wraps(FFmpegFilterVerticalFlip)
    def vflip(self, **options) -> Self:
        return self.add_filter(FFmpegFilterVerticalFlip(**options))

    @functools.wraps(FFmpegFilterCustom)
    def filter(self, content: str) -> Self:
        return self.add_filter(FFmpegFilterCustom(content=content))

    def typer_filters(self, typer: BrokenTyper) -> None:
        with typer.panel("📦 (FFmpeg) Filters"):
            typer.command(FFmpegFilterScale,        post=self.add_filter, name="scale")
            typer.command(FFmpegFilterVerticalFlip, post=self.add_filter, name="vflip")
            typer.command(FFmpegFilterCustom,       post=self.add_filter, name="filter")

    # ---------------------------------------------------------------------------------------------|
    # Command building and running

    @property
    def command(self) -> list[str]:
        BrokenFFmpeg.install()

        if (not self.inputs):
            raise ValueError("At least one input is required for FFmpeg")
        if (not self.outputs):
            raise ValueError("At least one output is required for FFmpeg")

        command = deque()

        def extend(*objects: Union[FFmpegModuleBase, Iterable[FFmpegModuleBase]]):
            for item in flatten(objects):
                if isinstance(item, FFmpegModuleBase):
                    command.extend(flatten(item.command(self)))
                else:
                    command.append(item)

        extend(shutil.which("ffmpeg"))
        extend("-hide_banner"*self.hide_banner)
        extend("-loglevel", denum(self.loglevel))
        extend(every("-threads", self.threads))
        extend(("-hwaccel", denum(self.hwaccel))*bool(self.hwaccel))
        extend(("-stream_loop", self.stream_loop)*bool(self.stream_loop))
        extend(self.inputs)
        extend(("-t", self.time)*bool(self.time))
        extend("-shortest"*self.shortest)

        # Note: https://trac.ffmpeg.org/wiki/Creating%20multiple%20outputs
        for output in self.outputs:
            extend(self.audio_codec)
            extend(self.video_codec)
            extend(every("-vf", ",".join(map(str, self.filters))))
            extend(output)

        return list(map(str, map(denum, flatten(command))))

    def run(self, **options) -> subprocess.CompletedProcess:
        return shell(self.command, **options)

    def popen(self, **options) -> subprocess.Popen:
        return shell(self.command, Popen=True, **options)

    # ---------------------------------------------------------------------------------------------|
    # High level functions

    @staticmethod
    def install(raises: bool=True) -> None:
        if all(map(BrokenPath.which, ("ffmpeg", "ffprobe"))):
            return None

        if (not BrokenPlatform.OnMacOS):
            log.info("FFmpeg wasn't found on System Path, will download a BtbN's Build")
            BrokenPath.get_external(''.join((
                "https://github.com/BtbN/FFmpeg-Builds/releases/download/latest/",
                "ffmpeg-master-latest-",
                BrokenPlatform.System.value.replace("windows", "win"),
                BrokenPlatform.Arch.replace("amd64", "64"),
                ("-gpl.zip" if BrokenPlatform.OnWindows else "-gpl.tar.xz")
            )))
        else:
            log.info("FFmpeg wasn't found on System Path, will download a EverMeet's Build")
            ffprobe = BrokenPath.get_external("https://evermeet.cx/ffmpeg/getrelease/ffprobe/zip", redirect=True)
            ffmpeg  = BrokenPath.get_external("https://evermeet.cx/ffmpeg/getrelease/zip", redirect=True)

            # Shutil doesn't preserve executable attributes
            for file in ffprobe.rglob("ffprobe*"):
                BrokenPath.make_executable(file)
            for file in ffmpeg.rglob("ffmpeg*"):
                BrokenPath.make_executable(file)

        # Ensure the binaries are available
        if raises and (not all(map(BrokenPath.which, ("ffmpeg", "ffprobe")))):
            raise FileNotFoundError("FFmpeg wasn't found on the system after an attempt to download it")

    # # Video

    @staticmethod
    def loop(path: Path, *, times: int=1, output: Path=None, echo: bool=True) -> Path:
        """Loop a video N-1 times, output to a new file or replace the original"""
        if not (path := BrokenPath.get(path, exists=True)):
            return None
        if ((times := times - 1) <= 0):
            return path
        BrokenFFmpeg.install()

        # Optional override or inline
        output = (output or path)
        looped = output.with_stem(f"{output.stem}-{times}-loops")
        log.info(f"Looping video ({path}) {times}x times to ({output})", echo=echo)

        # Fastest way to loop a video, no re-encoding
        (BrokenFFmpeg(stream_loop=(times - 1)).quiet().copy_audio().copy_video()
            .input(path).output(looped, pixel_format=None).run())

        # Replace the original file or move to target
        return looped.replace(output)

    @staticmethod
    @functools.lru_cache
    def get_video_resolution(path: Path, *, echo: bool=True) -> Optional[tuple[int, int]]:
        """Get the resolution of a video in a smart way"""
        if not (path := BrokenPath.get(path, exists=True)):
            return None
        BrokenFFmpeg.install()
        log.minor(f"Getting Video Resolution of ({path})", echo=echo)
        import PIL
        return PIL.Image.open(io.BytesIO(shell(
            shutil.which("ffmpeg"), "-hide_banner", "-loglevel", "error",
            "-i", path, "-vframes", "1", "-f", "image2pipe", "-",
            stdout=PIPE, echo=echo
        ).stdout), formats=["jpeg"]).size

    @staticmethod
    def iter_video_frames(path: Path, *, skip: int=0, echo: bool=True) -> Optional[Iterable[numpy.ndarray]]:
        """Generator for every frame of the video as numpy arrays, FAST!"""
        if not (path := BrokenPath.get(path, exists=True)):
            return None
        BrokenFFmpeg.install()
        (width, height) = BrokenFFmpeg.get_video_resolution(path)
        log.minor(f"Streaming Video Frames from file ({path}) @ ({width}x{height})", echo=echo)
        ffmpeg = (BrokenFFmpeg(vsync="cfr")
            .quiet()
            .input(path=path)
            .filter(content=f"select='gte(n\\,{skip})'")
            .rawvideo()
            .no_audio()
            .pipe_output(
                pixel_format="rgb24",
                format="rawvideo",
            )
        ).popen(stdout=PIPE, echo=echo)

        # Keep reading frames until we run out, each pixel is 3 bytes !
        while (raw := ffmpeg.stdout.read(width * height * 3)):
            yield numpy.frombuffer(raw, dtype=numpy.uint8).reshape((height, width, 3))

    @staticmethod
    def is_valid_video(path: Path, *, echo: bool=True) -> bool:
        if not (path := BrokenPath.get(path, exists=True)):
            return False
        BrokenFFmpeg.install()
        return (shell(
            shutil.which("ffmpeg"),
            "-hide_banner", "-loglevel", "error",
            "-i", path, "-f", "null", "-", echo=echo,
            stderr=DEVNULL, stdout=DEVNULL
        ).returncode == 0)

    @staticmethod
    @functools.lru_cache
    def get_video_total_frames(path: Path, *, echo: bool=True) -> Optional[int]:
        """Count the total frames of a video by decode voiding and parsing stats output"""
        if not (path := BrokenPath.get(path, exists=True)):
            return None
        BrokenFFmpeg.install()
        with Halo(log.minor(f"Getting total frames of video ({path}) by decoding every frame, might take a while..")):
            return int(re.compile(r"frame=\s*(\d+)").findall((
                BrokenFFmpeg(vsync="cfr")
                .input(path=path)
                .pipe_output(format="null")
            ).run(stderr=PIPE, echo=echo).stderr.decode())[-1])

    @staticmethod
    @functools.lru_cache
    def get_video_duration(path: Path, *, echo: bool=True) -> Optional[float]:
        if not (path := BrokenPath.get(path, exists=True)):
            return None
        BrokenFFmpeg.install()
        log.minor(f"Getting Video Duration of file ({path})", echo=echo)
        return float(shell(
            BrokenPath.which("ffprobe"),
            "-i", path,
            "-show_entries", "format=duration",
            "-v", "quiet", "-of", "csv=p=0",
            output=True, echo=echo
        ))

    @staticmethod
    @functools.lru_cache
    def get_video_framerate(path: Path, *, precise: bool=False, echo: bool=True) -> Optional[float]:
        if not (path := BrokenPath.get(path, exists=True)):
            return None
        BrokenFFmpeg.install()
        log.minor(f"Getting Video Framerate of file ({path})", echo=echo)
        if precise:
            A = BrokenFFmpeg.get_video_total_frames(path, echo=False)
            B = BrokenFFmpeg.get_video_duration(path, echo=False)
            return (A/B)
        else:
            return float(flatten(eval(shell(
                BrokenPath.which("ffprobe"),
                "-i", path,
                "-show_entries", "stream=r_frame_rate",
                "-v", "quiet", "-of", "csv=p=0",
                output=True, echo=echo
            ).splitlines()[0]))[0])

    # # Audio

    @staticmethod
    @functools.lru_cache
    def get_audio_samplerate(path: Path, *, stream: int=0, echo: bool=True) -> Optional[int]:
        if not (path := BrokenPath.get(path, exists=True)):
            return None
        BrokenFFmpeg.install()
        log.minor(f"Getting Audio Samplerate of file ({path})", echo=echo)
        return int(shell(
            BrokenPath.which("ffprobe"),
            "-i", path,
            "-show_entries", "stream=sample_rate",
            "-v", "quiet", "-of", "csv=p=0",
            output=True, echo=echo
        ).strip().splitlines()[stream])

    @staticmethod
    @functools.lru_cache
    def get_audio_channels(path: Path, *, stream: int=0, echo: bool=True) -> Optional[int]:
        if not (path := BrokenPath.get(path, exists=True)):
            return None
        BrokenFFmpeg.install()
        log.minor(f"Getting Audio Channels of file ({path})", echo=echo)
        return int(shell(
            BrokenPath.which("ffprobe"),
            "-i", path,
            "-show_entries", "stream=channels",
            "-v", "quiet", "-of", "csv=p=0",
            output=True, echo=echo
        ).strip().splitlines()[stream])

    @staticmethod
    def get_audio_duration(path: Path, *, echo: bool=True) -> Optional[float]:
        if not (path := BrokenPath.get(path, exists=True)):
            return None
        try:
            generator = BrokenAudioReader(path=path, chunk=10).stream
            while next(generator) is not None: ...
        except StopIteration as result:
            return result.value

    @staticmethod
    def get_audio_numpy(path: Path, *, echo: bool=True) -> Optional[numpy.ndarray]:
        if not (path := BrokenPath.get(path, exists=True)):
            return None
        BrokenFFmpeg.install()
        log.minor(f"Getting Audio as Numpy Array of file ({path})", echo=echo)
        return numpy.concatenate(list(BrokenAudioReader(path=path, chunk=10).stream))

Input

Source code in Broken/Externals/FFmpeg.py
966
967
968
class Input:
    Path = FFmpegInputPath
    Pipe = FFmpegInputPipe
Path
Pipe

Output

Source code in Broken/Externals/FFmpeg.py
970
971
972
class Output:
    Path = FFmpegOutputPath
    Pipe = FFmpegOutputPipe
Path
Pipe

VideoCodec

Source code in Broken/Externals/FFmpeg.py
974
975
976
977
978
979
980
981
982
983
984
985
class VideoCodec:
    H264       = FFmpegVideoCodecH264
    H264_NVENC = FFmpegVideoCodecH264_NVENC
    H265       = FFmpegVideoCodecH265
    H265_NVENC = FFmpegVideoCodecH265_NVENC
    VP9        = FFmpegVideoCodecVP9
    AV1_SVT    = FFmpegVideoCodecAV1_SVT
    AV1_NVENC  = FFmpegVideoCodecAV1_NVENC
    AV1_RAV1E  = FFmpegVideoCodecAV1_RAV1E
    Rawvideo   = FFmpegVideoCodecRawvideo
    NoVideo    = FFmpegVideoCodecNoVideo
    Copy       = FFmpegVideoCodecCopy
H264
H264_NVENC
H265
H265_NVENC
AV1_SVT
AV1_NVENC
AV1_RAV1E
Rawvideo
NoVideo
Copy

AudioCodec

Source code in Broken/Externals/FFmpeg.py
987
988
989
990
991
992
993
994
995
class AudioCodec:
    AAC   = FFmpegAudioCodecAAC
    MP3   = FFmpegAudioCodecMP3
    Opus  = FFmpegAudioCodecOpus
    FLAC  = FFmpegAudioCodecFLAC
    Copy  = FFmpegAudioCodecCopy
    None_ = FFmpegAudioCodecNone
    Empty = FFmpegAudioCodecEmpty
    PCM   = FFmpegAudioCodecPCM

Filter

Source code in Broken/Externals/FFmpeg.py
 997
 998
 999
1000
class Filter:
    Scale        = FFmpegFilterScale
    VerticalFlip = FFmpegFilterVerticalFlip
    Custom       = FFmpegFilterCustom
Scale
VerticalFlip
VerticalFlip = FFmpegFilterVerticalFlip
Custom

hide_banner

hide_banner: bool = True

Hides the compilation information of FFmpeg from the output

shortest

shortest: bool = False

Ends the video at the shortest input's duration

FFmpeg docs

stream_loop

stream_loop: int = Field(0)

Loops the input stream N times to the right. Zero '0' is no loop, one '1' doubles the length

time

time: float = Field(0.0)

If greater than zero, stops encoding at the specified time. -t option of FFmpeg

set_time

set_time(time: float) -> Self
Source code in Broken/Externals/FFmpeg.py
1019
1020
1021
def set_time(self, time: float) -> Self:
    self.time = time
    return self

vsync

vsync: Literal["auto", "passthrough", "cfr", "vfr"] = Field(
    "cfr"
)

The video's framerate mode, applied to all subsequent output targets. -vsync option of FFmpeg

  • auto: FFmpeg default, choses between constant and variable framerate based on muxer support
  • cfr: Constant Frame Rate, where frames are droped or duped to precisely match frametimes
  • vfr: Variable Frame Rate, static frames are kept, no two frames have the same timestemp
  • passthrough: The frames are passed through without modification on their timestamp

FFmpeg docs

LogLevel

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
1035
1036
1037
1038
1039
1040
1041
1042
class LogLevel(str, BrokenEnum):
    Error   = "error"
    Info    = "info"
    Verbose = "verbose"
    Debug   = "debug"
    Warning = "warning"
    Panic   = "panic"
    Fatal   = "fatal"
Error
Error = 'error'
Info
Info = 'info'
Verbose
Verbose = 'verbose'
Debug
Debug = 'debug'
Warning
Warning = 'warning'
Panic
Panic = 'panic'
Fatal
Fatal = 'fatal'

loglevel

loglevel: Annotated[
    LogLevel, Option("--loglevel", "-log")
] = Field(LogLevel.Info)

HardwareAcceleration

Bases: str, BrokenEnum

Source code in Broken/Externals/FFmpeg.py
1048
1049
1050
1051
1052
class HardwareAcceleration(str, BrokenEnum):
    Auto   = "auto"
    CUDA   = "cuda"
    NVDEC  = "nvdec"
    Vulkan = "vulkan"
Auto
Auto = 'auto'
CUDA
CUDA = 'cuda'
NVDEC
NVDEC = 'nvdec'
Vulkan
Vulkan = 'vulkan'

hwaccel

hwaccel: Annotated[
    Optional[HardwareAcceleration],
    Option("--hwaccel", "-hw"),
] = Field(None)

What device to bootstrap, for decoding with hardware acceleration. In practice, it's only useful when decoding from a source video file, might cause overhead on pipe input mode

  • auto: Finds up the best device to use, more often than not nvdec or cuvid

TODO: Add the required initializers on the final command per option

FFmpeg docs

threads

threads: Optional[int] = Field(None, ge=0)

The number of threads the codecs should use (). Generally speaking, more threads drastically improves performance, at the cost of worse quality and compression ratios. It's not that bad though. Some codecs might not use all available CPU threads. '0' finds the optimal amount automatically

FFmpeg docs

inputs

inputs: list[FFmpegInputType] = Field(default_factory=list)

A list of inputs for FFmpeg

filters

filters: list[FFmpegFilterType] = Field(
    default_factory=list
)

outputs

outputs: list[FFmpegOutputType] = Field(
    default_factory=list
)

A list of outputs. Yes, FFmpeg natively supports multi-encoding targets

video_codec

video_codec: Optional[FFmpegVideoCodecType] = Field(
    default_factory=FFmpegVideoCodecH264
)

The video codec to use and its configuration

audio_codec

audio_codec: Optional[FFmpegAudioCodecType] = Field(None)

The audio codec to use and its configuration

quiet

quiet() -> Self
Source code in Broken/Externals/FFmpeg.py
1091
1092
1093
1094
def quiet(self) -> Self:
    self.hide_banner = True
    self.loglevel = "error"
    return self

clear_inputs

clear_inputs() -> Self
Source code in Broken/Externals/FFmpeg.py
1099
1100
1101
def clear_inputs(self) -> Self:
    self.inputs = list()
    return self

clear_filters

clear_filters() -> Self
Source code in Broken/Externals/FFmpeg.py
1103
1104
1105
def clear_filters(self) -> Self:
    self.filters = list()
    return self

clear_outputs

clear_outputs() -> Self
Source code in Broken/Externals/FFmpeg.py
1107
1108
1109
def clear_outputs(self) -> Self:
    self.outputs = list()
    return self

clear_video_codec

clear_video_codec() -> Self
Source code in Broken/Externals/FFmpeg.py
1111
1112
1113
def clear_video_codec(self) -> Self:
    self.video_codec = None
    return self

clear_audio_codec

clear_audio_codec() -> Self
Source code in Broken/Externals/FFmpeg.py
1115
1116
1117
def clear_audio_codec(self) -> Self:
    self.audio_codec = None
    return self

clear

clear(
    inputs: bool = True,
    filters: bool = True,
    outputs: bool = True,
    video_codec: bool = True,
    audio_codec: bool = True,
) -> Self
Source code in Broken/Externals/FFmpeg.py
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
def clear(self,
    inputs: bool=True,
    filters: bool=True,
    outputs: bool=True,
    video_codec: bool=True,
    audio_codec: bool=True,
) -> Self:
    if inputs:      self.clear_inputs()
    if filters:     self.clear_filters()
    if outputs:     self.clear_outputs()
    if video_codec: self.clear_video_codec()
    if audio_codec: self.clear_audio_codec()
    return self

add_input

add_input(input: FFmpegInputType) -> Self
Source code in Broken/Externals/FFmpeg.py
1138
1139
1140
def add_input(self, input: FFmpegInputType) -> Self:
    self.inputs.append(input)
    return self

input

input(path: Path, **options) -> Self
Source code in Broken/Externals/FFmpeg.py
1142
1143
1144
@functools.wraps(FFmpegInputPath)
def input(self, path: Path, **options) -> Self:
    return self.add_input(FFmpegInputPath(path=path, **options))

pipe_input

pipe_input(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1146
1147
1148
@functools.wraps(FFmpegInputPipe)
def pipe_input(self, **options) -> Self:
    return self.add_input(FFmpegInputPipe(**options))

typer_inputs

typer_inputs(typer: BrokenTyper) -> None
Source code in Broken/Externals/FFmpeg.py
1150
1151
1152
1153
def typer_inputs(self, typer: BrokenTyper) -> None:
    with typer.panel("📦 (FFmpeg) Input"):
        typer.command(FFmpegInputPath, post=self.add_input, name="ipath")
        typer.command(FFmpegInputPipe, post=self.add_input, name="ipipe")

add_output

add_output(output: FFmpegOutputType) -> Self
Source code in Broken/Externals/FFmpeg.py
1155
1156
1157
def add_output(self, output: FFmpegOutputType) -> Self:
    self.outputs.append(output)
    return self

output

output(path: Path, **options) -> Self
Source code in Broken/Externals/FFmpeg.py
1159
1160
1161
@functools.wraps(FFmpegOutputPath)
def output(self, path: Path, **options) -> Self:
    return self.add_output(FFmpegOutputPath(path=path, **options))

pipe_output

pipe_output(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1163
1164
1165
@functools.wraps(FFmpegOutputPipe)
def pipe_output(self, **options) -> Self:
    return self.add_output(FFmpegOutputPipe(**options))

typer_outputs

typer_outputs(typer: BrokenTyper) -> None
Source code in Broken/Externals/FFmpeg.py
1167
1168
1169
1170
def typer_outputs(self, typer: BrokenTyper) -> None:
    with typer.panel("📦 (FFmpeg) Output"):
        typer.command(FFmpegOutputPath, post=self.add_output, name="opath")
        typer.command(FFmpegOutputPipe, post=self.add_output, name="opipe")

set_video_codec

set_video_codec(codec: FFmpegVideoCodecType) -> Self
Source code in Broken/Externals/FFmpeg.py
1174
1175
1176
def set_video_codec(self, codec: FFmpegVideoCodecType) -> Self:
    self.video_codec = codec
    return self

h264

h264(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1178
1179
1180
@functools.wraps(FFmpegVideoCodecH264)
def h264(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecH264(**options))

h264_nvenc

h264_nvenc(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1182
1183
1184
@functools.wraps(FFmpegVideoCodecH264_NVENC)
def h264_nvenc(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecH264_NVENC(**options))

h265

h265(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1186
1187
1188
@functools.wraps(FFmpegVideoCodecH265)
def h265(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecH265(**options))

h265_nvenc

h265_nvenc(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1190
1191
1192
@functools.wraps(FFmpegVideoCodecH265_NVENC)
def h265_nvenc(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecH265_NVENC(**options))

vp9

vp9(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1194
1195
1196
@functools.wraps(FFmpegVideoCodecVP9)
def vp9(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecVP9(**options))

av1_svt

av1_svt(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1198
1199
1200
@functools.wraps(FFmpegVideoCodecAV1_SVT)
def av1_svt(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecAV1_SVT(**options))

av1_nvenc

av1_nvenc(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1202
1203
1204
@functools.wraps(FFmpegVideoCodecAV1_NVENC)
def av1_nvenc(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecAV1_NVENC(**options))

av1_rav1e

av1_rav1e(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1206
1207
1208
@functools.wraps(FFmpegVideoCodecAV1_RAV1E)
def av1_rav1e(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecAV1_RAV1E(**options))

rawvideo

rawvideo(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1210
1211
1212
@functools.wraps(FFmpegVideoCodecRawvideo)
def rawvideo(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecRawvideo(**options))

copy_video

copy_video(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1214
1215
1216
@functools.wraps(FFmpegVideoCodecCopy)
def copy_video(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecCopy(**options))

no_video

no_video(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1218
1219
1220
@functools.wraps(FFmpegVideoCodecNoVideo)
def no_video(self, **options) -> Self:
    return self.set_video_codec(FFmpegVideoCodecNoVideo(**options))

typer_vcodecs

typer_vcodecs(typer: BrokenTyper) -> None
Source code in Broken/Externals/FFmpeg.py
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
def typer_vcodecs(self, typer: BrokenTyper) -> None:
    with typer.panel("📦 (Exporting) Video encoder"):
        typer.command(FFmpegVideoCodecH264,       post=self.set_video_codec, name="h264")
        typer.command(FFmpegVideoCodecH264_NVENC, post=self.set_video_codec, name="h264-nvenc")
        typer.command(FFmpegVideoCodecH265,       post=self.set_video_codec, name="h265")
        typer.command(FFmpegVideoCodecH265_NVENC, post=self.set_video_codec, name="h265-nvenc")
        typer.command(FFmpegVideoCodecVP9,        post=self.set_video_codec, name="vp9", hidden=True)
        typer.command(FFmpegVideoCodecAV1_SVT,    post=self.set_video_codec, name="av1-svt")
        typer.command(FFmpegVideoCodecAV1_NVENC,  post=self.set_video_codec, name="av1-nvenc")
        typer.command(FFmpegVideoCodecAV1_RAV1E,  post=self.set_video_codec, name="av1-rav1e")
        typer.command(FFmpegVideoCodecRawvideo,   post=self.set_video_codec, name="rawvideo", hidden=True)
        typer.command(FFmpegVideoCodecCopy,       post=self.set_video_codec, name="video-copy", hidden=True)
        typer.command(FFmpegVideoCodecNoVideo,    post=self.set_video_codec, name="video-none", hidden=True)

set_audio_codec

set_audio_codec(codec: FFmpegAudioCodecType) -> Self
Source code in Broken/Externals/FFmpeg.py
1238
1239
1240
def set_audio_codec(self, codec: FFmpegAudioCodecType) -> Self:
    self.audio_codec = codec
    return self

aac

aac(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1242
1243
1244
@functools.wraps(FFmpegAudioCodecAAC)
def aac(self, **options) -> Self:
    return self.set_audio_codec(FFmpegAudioCodecAAC(**options))

mp3

mp3(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1246
1247
1248
@functools.wraps(FFmpegAudioCodecMP3)
def mp3(self, **options) -> Self:
    return self.set_audio_codec(FFmpegAudioCodecMP3(**options))

opus

opus(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1250
1251
1252
@functools.wraps(FFmpegAudioCodecOpus)
def opus(self, **options) -> Self:
    return self.set_audio_codec(FFmpegAudioCodecOpus(**options))

flac

flac(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1254
1255
1256
@functools.wraps(FFmpegAudioCodecFLAC)
def flac(self, **options) -> Self:
    return self.set_audio_codec(FFmpegAudioCodecFLAC(**options))

pcm

pcm(format: FFmpegAudioCodecPCM = 'pcm_f32le') -> Self
Source code in Broken/Externals/FFmpeg.py
1258
1259
1260
@functools.wraps(FFmpegAudioCodecPCM)
def pcm(self, format: FFmpegAudioCodecPCM="pcm_f32le") -> Self:
    return self.set_audio_codec(FFmpegAudioCodecPCM(format=format))

copy_audio

copy_audio(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1262
1263
1264
@functools.wraps(FFmpegAudioCodecCopy)
def copy_audio(self, **options) -> Self:
    return self.set_audio_codec(FFmpegAudioCodecCopy(**options))

no_audio

no_audio(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1266
1267
1268
@functools.wraps(FFmpegAudioCodecNone)
def no_audio(self, **options) -> Self:
    return self.set_audio_codec(FFmpegAudioCodecNone(**options))

empty_audio

empty_audio(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1270
1271
1272
@functools.wraps(FFmpegAudioCodecEmpty)
def empty_audio(self, **options) -> Self:
    return self.set_audio_codec(FFmpegAudioCodecEmpty(**options))

typer_acodecs

typer_acodecs(typer: BrokenTyper) -> None
Source code in Broken/Externals/FFmpeg.py
1274
1275
1276
1277
1278
1279
1280
1281
1282
def typer_acodecs(self, typer: BrokenTyper) -> None:
    with typer.panel("📦 (Exporting) Audio encoder"):
        typer.command(FFmpegAudioCodecAAC,   post=self.set_audio_codec, name="aac")
        typer.command(FFmpegAudioCodecMP3,   post=self.set_audio_codec, name="mp3")
        typer.command(FFmpegAudioCodecOpus,  post=self.set_audio_codec, name="opus")
        typer.command(FFmpegAudioCodecFLAC,  post=self.set_audio_codec, name="flac")
        typer.command(FFmpegAudioCodecCopy,  post=self.set_audio_codec, name="audio-copy")
        typer.command(FFmpegAudioCodecNone,  post=self.set_audio_codec, name="audio-none")
        typer.command(FFmpegAudioCodecEmpty, post=self.set_audio_codec, name="audio-empty")

add_filter

add_filter(filter: FFmpegFilterType) -> Self
Source code in Broken/Externals/FFmpeg.py
1286
1287
1288
def add_filter(self, filter: FFmpegFilterType) -> Self:
    self.filters.append(filter)
    return self

scale

scale(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1290
1291
1292
@functools.wraps(FFmpegFilterScale)
def scale(self, **options) -> Self:
    return self.add_filter(FFmpegFilterScale(**options))

vflip

vflip(**options) -> Self
Source code in Broken/Externals/FFmpeg.py
1294
1295
1296
@functools.wraps(FFmpegFilterVerticalFlip)
def vflip(self, **options) -> Self:
    return self.add_filter(FFmpegFilterVerticalFlip(**options))

filter

filter(content: str) -> Self
Source code in Broken/Externals/FFmpeg.py
1298
1299
1300
@functools.wraps(FFmpegFilterCustom)
def filter(self, content: str) -> Self:
    return self.add_filter(FFmpegFilterCustom(content=content))

typer_filters

typer_filters(typer: BrokenTyper) -> None
Source code in Broken/Externals/FFmpeg.py
1302
1303
1304
1305
1306
def typer_filters(self, typer: BrokenTyper) -> None:
    with typer.panel("📦 (FFmpeg) Filters"):
        typer.command(FFmpegFilterScale,        post=self.add_filter, name="scale")
        typer.command(FFmpegFilterVerticalFlip, post=self.add_filter, name="vflip")
        typer.command(FFmpegFilterCustom,       post=self.add_filter, name="filter")

command

command: list[str]

run

run(**options) -> subprocess.CompletedProcess
Source code in Broken/Externals/FFmpeg.py
1348
1349
def run(self, **options) -> subprocess.CompletedProcess:
    return shell(self.command, **options)

popen

popen(**options) -> subprocess.Popen
Source code in Broken/Externals/FFmpeg.py
1351
1352
def popen(self, **options) -> subprocess.Popen:
    return shell(self.command, Popen=True, **options)

install

install(raises: bool = True) -> None
Source code in Broken/Externals/FFmpeg.py
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
@staticmethod
def install(raises: bool=True) -> None:
    if all(map(BrokenPath.which, ("ffmpeg", "ffprobe"))):
        return None

    if (not BrokenPlatform.OnMacOS):
        log.info("FFmpeg wasn't found on System Path, will download a BtbN's Build")
        BrokenPath.get_external(''.join((
            "https://github.com/BtbN/FFmpeg-Builds/releases/download/latest/",
            "ffmpeg-master-latest-",
            BrokenPlatform.System.value.replace("windows", "win"),
            BrokenPlatform.Arch.replace("amd64", "64"),
            ("-gpl.zip" if BrokenPlatform.OnWindows else "-gpl.tar.xz")
        )))
    else:
        log.info("FFmpeg wasn't found on System Path, will download a EverMeet's Build")
        ffprobe = BrokenPath.get_external("https://evermeet.cx/ffmpeg/getrelease/ffprobe/zip", redirect=True)
        ffmpeg  = BrokenPath.get_external("https://evermeet.cx/ffmpeg/getrelease/zip", redirect=True)

        # Shutil doesn't preserve executable attributes
        for file in ffprobe.rglob("ffprobe*"):
            BrokenPath.make_executable(file)
        for file in ffmpeg.rglob("ffmpeg*"):
            BrokenPath.make_executable(file)

    # Ensure the binaries are available
    if raises and (not all(map(BrokenPath.which, ("ffmpeg", "ffprobe")))):
        raise FileNotFoundError("FFmpeg wasn't found on the system after an attempt to download it")

loop

loop(
    path: Path,
    *,
    times: int = 1,
    output: Path = None,
    echo: bool = True
) -> Path

Loop a video N-1 times, output to a new file or replace the original

Source code in Broken/Externals/FFmpeg.py
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
@staticmethod
def loop(path: Path, *, times: int=1, output: Path=None, echo: bool=True) -> Path:
    """Loop a video N-1 times, output to a new file or replace the original"""
    if not (path := BrokenPath.get(path, exists=True)):
        return None
    if ((times := times - 1) <= 0):
        return path
    BrokenFFmpeg.install()

    # Optional override or inline
    output = (output or path)
    looped = output.with_stem(f"{output.stem}-{times}-loops")
    log.info(f"Looping video ({path}) {times}x times to ({output})", echo=echo)

    # Fastest way to loop a video, no re-encoding
    (BrokenFFmpeg(stream_loop=(times - 1)).quiet().copy_audio().copy_video()
        .input(path).output(looped, pixel_format=None).run())

    # Replace the original file or move to target
    return looped.replace(output)

get_video_resolution

get_video_resolution(
    path: Path, *, echo: bool = True
) -> Optional[tuple[int, int]]

Get the resolution of a video in a smart way

Source code in Broken/Externals/FFmpeg.py
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
@staticmethod
@functools.lru_cache
def get_video_resolution(path: Path, *, echo: bool=True) -> Optional[tuple[int, int]]:
    """Get the resolution of a video in a smart way"""
    if not (path := BrokenPath.get(path, exists=True)):
        return None
    BrokenFFmpeg.install()
    log.minor(f"Getting Video Resolution of ({path})", echo=echo)
    import PIL
    return PIL.Image.open(io.BytesIO(shell(
        shutil.which("ffmpeg"), "-hide_banner", "-loglevel", "error",
        "-i", path, "-vframes", "1", "-f", "image2pipe", "-",
        stdout=PIPE, echo=echo
    ).stdout), formats=["jpeg"]).size

iter_video_frames

iter_video_frames(
    path: Path, *, skip: int = 0, echo: bool = True
) -> Optional[Iterable[numpy.ndarray]]

Generator for every frame of the video as numpy arrays, FAST!

Source code in Broken/Externals/FFmpeg.py
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
@staticmethod
def iter_video_frames(path: Path, *, skip: int=0, echo: bool=True) -> Optional[Iterable[numpy.ndarray]]:
    """Generator for every frame of the video as numpy arrays, FAST!"""
    if not (path := BrokenPath.get(path, exists=True)):
        return None
    BrokenFFmpeg.install()
    (width, height) = BrokenFFmpeg.get_video_resolution(path)
    log.minor(f"Streaming Video Frames from file ({path}) @ ({width}x{height})", echo=echo)
    ffmpeg = (BrokenFFmpeg(vsync="cfr")
        .quiet()
        .input(path=path)
        .filter(content=f"select='gte(n\\,{skip})'")
        .rawvideo()
        .no_audio()
        .pipe_output(
            pixel_format="rgb24",
            format="rawvideo",
        )
    ).popen(stdout=PIPE, echo=echo)

    # Keep reading frames until we run out, each pixel is 3 bytes !
    while (raw := ffmpeg.stdout.read(width * height * 3)):
        yield numpy.frombuffer(raw, dtype=numpy.uint8).reshape((height, width, 3))

is_valid_video

is_valid_video(path: Path, *, echo: bool = True) -> bool
Source code in Broken/Externals/FFmpeg.py
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
@staticmethod
def is_valid_video(path: Path, *, echo: bool=True) -> bool:
    if not (path := BrokenPath.get(path, exists=True)):
        return False
    BrokenFFmpeg.install()
    return (shell(
        shutil.which("ffmpeg"),
        "-hide_banner", "-loglevel", "error",
        "-i", path, "-f", "null", "-", echo=echo,
        stderr=DEVNULL, stdout=DEVNULL
    ).returncode == 0)

get_video_total_frames

get_video_total_frames(
    path: Path, *, echo: bool = True
) -> Optional[int]

Count the total frames of a video by decode voiding and parsing stats output

Source code in Broken/Externals/FFmpeg.py
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
@staticmethod
@functools.lru_cache
def get_video_total_frames(path: Path, *, echo: bool=True) -> Optional[int]:
    """Count the total frames of a video by decode voiding and parsing stats output"""
    if not (path := BrokenPath.get(path, exists=True)):
        return None
    BrokenFFmpeg.install()
    with Halo(log.minor(f"Getting total frames of video ({path}) by decoding every frame, might take a while..")):
        return int(re.compile(r"frame=\s*(\d+)").findall((
            BrokenFFmpeg(vsync="cfr")
            .input(path=path)
            .pipe_output(format="null")
        ).run(stderr=PIPE, echo=echo).stderr.decode())[-1])

get_video_duration

get_video_duration(
    path: Path, *, echo: bool = True
) -> Optional[float]
Source code in Broken/Externals/FFmpeg.py
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
@staticmethod
@functools.lru_cache
def get_video_duration(path: Path, *, echo: bool=True) -> Optional[float]:
    if not (path := BrokenPath.get(path, exists=True)):
        return None
    BrokenFFmpeg.install()
    log.minor(f"Getting Video Duration of file ({path})", echo=echo)
    return float(shell(
        BrokenPath.which("ffprobe"),
        "-i", path,
        "-show_entries", "format=duration",
        "-v", "quiet", "-of", "csv=p=0",
        output=True, echo=echo
    ))

get_video_framerate

get_video_framerate(
    path: Path, *, precise: bool = False, echo: bool = True
) -> Optional[float]
Source code in Broken/Externals/FFmpeg.py
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
@staticmethod
@functools.lru_cache
def get_video_framerate(path: Path, *, precise: bool=False, echo: bool=True) -> Optional[float]:
    if not (path := BrokenPath.get(path, exists=True)):
        return None
    BrokenFFmpeg.install()
    log.minor(f"Getting Video Framerate of file ({path})", echo=echo)
    if precise:
        A = BrokenFFmpeg.get_video_total_frames(path, echo=False)
        B = BrokenFFmpeg.get_video_duration(path, echo=False)
        return (A/B)
    else:
        return float(flatten(eval(shell(
            BrokenPath.which("ffprobe"),
            "-i", path,
            "-show_entries", "stream=r_frame_rate",
            "-v", "quiet", "-of", "csv=p=0",
            output=True, echo=echo
        ).splitlines()[0]))[0])

get_audio_samplerate

get_audio_samplerate(
    path: Path, *, stream: int = 0, echo: bool = True
) -> Optional[int]
Source code in Broken/Externals/FFmpeg.py
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
@staticmethod
@functools.lru_cache
def get_audio_samplerate(path: Path, *, stream: int=0, echo: bool=True) -> Optional[int]:
    if not (path := BrokenPath.get(path, exists=True)):
        return None
    BrokenFFmpeg.install()
    log.minor(f"Getting Audio Samplerate of file ({path})", echo=echo)
    return int(shell(
        BrokenPath.which("ffprobe"),
        "-i", path,
        "-show_entries", "stream=sample_rate",
        "-v", "quiet", "-of", "csv=p=0",
        output=True, echo=echo
    ).strip().splitlines()[stream])

get_audio_channels

get_audio_channels(
    path: Path, *, stream: int = 0, echo: bool = True
) -> Optional[int]
Source code in Broken/Externals/FFmpeg.py
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
@staticmethod
@functools.lru_cache
def get_audio_channels(path: Path, *, stream: int=0, echo: bool=True) -> Optional[int]:
    if not (path := BrokenPath.get(path, exists=True)):
        return None
    BrokenFFmpeg.install()
    log.minor(f"Getting Audio Channels of file ({path})", echo=echo)
    return int(shell(
        BrokenPath.which("ffprobe"),
        "-i", path,
        "-show_entries", "stream=channels",
        "-v", "quiet", "-of", "csv=p=0",
        output=True, echo=echo
    ).strip().splitlines()[stream])

get_audio_duration

get_audio_duration(
    path: Path, *, echo: bool = True
) -> Optional[float]
Source code in Broken/Externals/FFmpeg.py
1541
1542
1543
1544
1545
1546
1547
1548
1549
@staticmethod
def get_audio_duration(path: Path, *, echo: bool=True) -> Optional[float]:
    if not (path := BrokenPath.get(path, exists=True)):
        return None
    try:
        generator = BrokenAudioReader(path=path, chunk=10).stream
        while next(generator) is not None: ...
    except StopIteration as result:
        return result.value

get_audio_numpy

get_audio_numpy(
    path: Path, *, echo: bool = True
) -> Optional[numpy.ndarray]
Source code in Broken/Externals/FFmpeg.py
1551
1552
1553
1554
1555
1556
1557
@staticmethod
def get_audio_numpy(path: Path, *, echo: bool=True) -> Optional[numpy.ndarray]:
    if not (path := BrokenPath.get(path, exists=True)):
        return None
    BrokenFFmpeg.install()
    log.minor(f"Getting Audio as Numpy Array of file ({path})", echo=echo)
    return numpy.concatenate(list(BrokenAudioReader(path=path, chunk=10).stream))

BrokenAudioReader

Source code in Broken/Externals/FFmpeg.py
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
@define
class BrokenAudioReader:
    path: Path

    format: FFmpegPCM = FFmpegPCM.PCM_FLOAT_32_BITS_LITTLE_ENDIAN
    """The audio format to output the contents of the audio file"""

    bytes_per_sample: int = None
    """Bytes per individual sample"""

    dtype: numpy.dtype = None
    """Numpy dtype out of self.format"""

    channels: int = None
    """The number of audio channels in the file"""

    samplerate: int = None
    """The sample rate of the audio file"""

    chunk: Seconds = 0.1
    """The amount of seconds to yield data at a time"""

    read: int = 0
    """Total number of bytes read from the audio file"""

    ffmpeg: Popen = None
    """The FFmpeg reader process"""

    @property
    def block_size(self) -> int:
        return (self.bytes_per_sample * self.channels)

    @property
    def bytes_per_second(self) -> int:
        return (self.block_size * self.samplerate)

    @property
    def time(self) -> float:
        return (self.read / self.bytes_per_second)

    @property
    def stream(self) -> Generator[numpy.ndarray, float, None]:
        if not (path := BrokenPath.get(self.path, exists=True)):
            return None
        self.path = path

        # Get audio file attributes
        self.channels   = BrokenFFmpeg.get_audio_channels(self.path, echo=False)
        self.samplerate = BrokenFFmpeg.get_audio_samplerate(self.path, echo=False)
        self.format = FFmpegPCM.get(self.format)
        self.bytes_per_sample = self.format.size
        self.dtype = self.format.dtype
        self.read = 0

        # Note: Stderr to null as we might not read all the audio, won't log errors
        self.ffmpeg = (
            BrokenFFmpeg()
            .quiet()
            .input(path=self.path)
            .pcm(self.format.value)
            .no_video()
            .output("-")
        ).popen(stdout=PIPE, stderr=PIPE)

        """
        The following code is wrong:

        ```python
        while (data := ffmpeg.stdout.read(chunk*samplerate)):
            yield (...)
        ```

        Reason being:
        • Small reads yields time imprecision on sample domain vs time domain
        • Must keep track of theoretical time and real time of the read
        """
        target = 0

        while True:
            target += self.chunk

            # Calculate the length of the next read to best match the target time,
            # but do not carry over temporal conversion errors
            length = (target - self.time) * self.bytes_per_second
            length = nearest(length, self.block_size, cast=int)
            length = max(length, self.block_size)
            data   = self.ffmpeg.stdout.read(length)
            if len(data) == 0: break

            # Increment precise time and read time
            yield numpy.frombuffer(data, dtype=self.dtype).reshape(-1, self.channels)
            self.read += len(data)

        # Allow to catch total duration on GeneratorExit
        return self.time

path

path: Path

format

The audio format to output the contents of the audio file

bytes_per_sample

bytes_per_sample: int = None

Bytes per individual sample

dtype

dtype: numpy.dtype = None

Numpy dtype out of self.format

channels

channels: int = None

The number of audio channels in the file

samplerate

samplerate: int = None

The sample rate of the audio file

chunk

chunk: Seconds = 0.1

The amount of seconds to yield data at a time

read

read: int = 0

Total number of bytes read from the audio file

ffmpeg

ffmpeg: Popen = None

The FFmpeg reader process

block_size

block_size: int

bytes_per_second

bytes_per_second: int

time

time: float

stream

stream: Generator[numpy.ndarray, float, None]