Skip to content

preprocess

__all__ = ['Resample', 'Loudnorm', 'Downmix', 'Pan', 'CustomPreprocessor'] module-attribute

Pan = Downmix module-attribute

Preprocessor

Bases: ABC

Source code in muxtools/audio/preprocess.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
class Preprocessor(ABC):
    refresh_metadata = False

    def get_filter(self, caller: Any = None) -> str | None:
        return None

    def get_args(self, caller: Any = None) -> Sequence[str]:
        return []

    def analyze(self, file: AudioFile):
        return None

    @abstractmethod
    def can_run(self, track: Track, preprocessors: Sequence[Any]) -> bool: ...

refresh_metadata = False class-attribute instance-attribute

get_filter(caller=None)

Source code in muxtools/audio/preprocess.py 
22
23
def get_filter(self, caller: Any = None) -> str | None:
    return None

get_args(caller=None)

Source code in muxtools/audio/preprocess.py 
25
26
def get_args(self, caller: Any = None) -> Sequence[str]:
    return []

analyze(file)

Source code in muxtools/audio/preprocess.py 
28
29
def analyze(self, file: AudioFile):
    return None

can_run(track, preprocessors) abstractmethod

Source code in muxtools/audio/preprocess.py 
31
32
@abstractmethod
def can_run(self, track: Track, preprocessors: Sequence[Any]) -> bool: ...

Resample dataclass

Bases: Preprocessor

A FFMPEG Resampling preprocessor. This is used to dither down to 16 bit and resample to 48kHz by default. Uses the sox resampler internally for best results.

Parameters:

Name Type Description Default
dither DitherType

The dither algorithm to use. Uses SoX's default by default.

 TRIANGULAR
depth int | None

The bitdepth to dither to. None will not change the depth. You can technically only choose 16 or 32 as 24 is apparently just 32 with padding and needs specific codec support.

 16
sample_rate int

The sample rate to resample to. Defaults to 48kHz because most encoders support it.

 48000
Source code in muxtools/audio/preprocess.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@dataclass
class Resample(Preprocessor):
    """
    A FFMPEG Resampling preprocessor.
    This is used to dither down to 16 bit and resample to 48kHz by default.
    Uses the sox resampler internally for best results.

    :param dither:          The dither algorithm to use. Uses SoX's default by default.
    :param depth:           The bitdepth to dither to. `None` will not change the depth.
                            You can technically only choose 16 or 32 as 24 is apparently just 32 with padding and needs specific codec support.
    :param sample_rate:     The sample rate to resample to. Defaults to 48kHz because most encoders support it.
    """

    dither: DitherType = DitherType.TRIANGULAR
    depth: int | None = 16
    sample_rate: int = 48000
    refresh_metadata = True

    def can_run(self, track: Track, preprocessors: Sequence[Any]) -> bool:
        # Run if depth or sample rate differ. Also run if loudnorm is being used.
        return (
            (self.depth and getattr(track, "bit_depth", 24) != self.depth)
            or getattr(track, "sampling_rate", 0) != self.sample_rate
            or [p for p in preprocessors if isinstance(p, Loudnorm)]
        )

    def get_args(self, caller: Any = None) -> Sequence[str]:
        if caller:
            debug(
                (
                    f"Resampling to {self.depth} bit and {self.sample_rate / 1000} kHz..."
                    if self.depth
                    else f"Resampling to {self.sample_rate / 1000} kHz..."
                ),
                caller,
            )
        return (
            []
            if not self.depth
            else ["-sample_fmt", f"s{self.depth}"]
            + [
                "-ar",
                str(self.sample_rate),
                "-resampler",
                "soxr",
                "-precision",
                "24",
                "-dither_method",
                self.dither.name.lower(),
            ]
        )

dither: DitherType = DitherType.TRIANGULAR class-attribute instance-attribute

depth: int | None = 16 class-attribute instance-attribute

sample_rate: int = 48000 class-attribute instance-attribute

refresh_metadata = True class-attribute instance-attribute

__init__(dither=DitherType.TRIANGULAR, depth=16, sample_rate=48000, refresh_metadata=True)

can_run(track, preprocessors)

Source code in muxtools/audio/preprocess.py 
53
54
55
56
57
58
59
def can_run(self, track: Track, preprocessors: Sequence[Any]) -> bool:
    # Run if depth or sample rate differ. Also run if loudnorm is being used.
    return (
        (self.depth and getattr(track, "bit_depth", 24) != self.depth)
        or getattr(track, "sampling_rate", 0) != self.sample_rate
        or [p for p in preprocessors if isinstance(p, Loudnorm)]
    )

get_args(caller=None)

Source code in muxtools/audio/preprocess.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def get_args(self, caller: Any = None) -> Sequence[str]:
    if caller:
        debug(
            (
                f"Resampling to {self.depth} bit and {self.sample_rate / 1000} kHz..."
                if self.depth
                else f"Resampling to {self.sample_rate / 1000} kHz..."
            ),
            caller,
        )
    return (
        []
        if not self.depth
        else ["-sample_fmt", f"s{self.depth}"]
        + [
            "-ar",
            str(self.sample_rate),
            "-resampler",
            "soxr",
            "-precision",
            "24",
            "-dither_method",
            self.dither.name.lower(),
        ]
    )

classproperty

Bases: object

Source code in muxtools/audio/preprocess.py 
88
89
90
91
92
93
class classproperty(object):
    def __init__(self, f):
        self.f = classmethod(f)

    def __get__(self, *a):
        return self.f.__get__(*a)()

f = classmethod(f) instance-attribute

__init__(f)

Source code in muxtools/audio/preprocess.py 
89
90
def __init__(self, f):
    self.f = classmethod(f)

__get__(*a)

Source code in muxtools/audio/preprocess.py 
92
93
def __get__(self, *a):
    return self.f.__get__(*a)()

Downmix dataclass

Bases: Preprocessor

A FFMPEG downmixing/pan preprocessor. This essentially just uses the pan filter and offers a few presets.

If you're looking for explanations or other infos feel free to read these threads: https://superuser.com/questions/852400/properly-downmix-5-1-to-stereo-using-ffmpeg https://github.com/mpv-player/mpv/issues/6343

Parameters:

Name Type Description Default
mixing str | None

The Pan filter string. Defaults to the Dave_750 preset. Honestly no recommendations here. Try them all and use what you prefer.

 None
force bool

Force processing even if there are only 2 channels.

 False
Source code in muxtools/audio/preprocess.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
@dataclass
class Downmix(Preprocessor):
    """
    A FFMPEG downmixing/pan preprocessor.
    This essentially just uses the [pan](http://ffmpeg.org/ffmpeg-all.html#pan-1) filter and offers a few presets.

    If you're looking for explanations or other infos feel free to read these threads:
    https://superuser.com/questions/852400/properly-downmix-5-1-to-stereo-using-ffmpeg
    https://github.com/mpv-player/mpv/issues/6343

    :param mixing:      The Pan filter string. Defaults to the Dave_750 preset.
                        Honestly no recommendations here. Try them all and use what you prefer.
    :param force:       Force processing even if there are only 2 channels.
    """

    mixing: str | None = None
    force: bool = False
    refresh_metadata = True

    def can_run(self, track: Track, preprocessors: Sequence[Any]) -> bool:
        return getattr(track, "channel_s", 2) > 2 or self.force

    def get_filter(self, caller: Any = None) -> str:
        if not self.mixing:
            self.mixing = Downmix.Dave_750
        if caller:
            debug("Applying downmix/pan filter...", caller)
        return f"pan={self.mixing}"

    @classproperty
    def ATSC(self) -> str:
        return "stereo|FL<1.0*FL+0.707*FC+0.707*BL+0.707*SL|FR<1.0*FR+0.707*FC+0.707*BR+0.707*SR"

    @classproperty
    def Collier(self) -> str:
        return "stereo|FL=FC+0.30*FL+0.30*BL+0.30*SL|FR=FC+0.30*FR+0.30*BR+0.30*SR"

    @classproperty
    def Dave_750(self) -> str:
        return "stereo|FL=0.5*FC+0.707*FL+0.707*BL+0.707*SL+0.5*LFE|FR=0.5*FC+0.707*FR+0.707*BR+0.707*SR+0.35*LFE"

    @classproperty
    def RFC_7845(self) -> str:
        return "stereo|FL=0.374107*FC+0.529067*FL+0.458186*BL+0.458186*SL+0.264534*BR+0.264534*SR+0.374107*LFE|FR=0.374107*FC+0.529067*FR+0.458186*BR+0.458186*SR+0.264534*BL+0.264534*SL+0.374107*LFE"

mixing: str | None = None class-attribute instance-attribute

force: bool = False class-attribute instance-attribute

refresh_metadata = True class-attribute instance-attribute

__init__(mixing=None, force=False, refresh_metadata=True)

can_run(track, preprocessors)

Source code in muxtools/audio/preprocess.py 
115
116
def can_run(self, track: Track, preprocessors: Sequence[Any]) -> bool:
    return getattr(track, "channel_s", 2) > 2 or self.force

get_filter(caller=None)

Source code in muxtools/audio/preprocess.py 
118
119
120
121
122
123
def get_filter(self, caller: Any = None) -> str:
    if not self.mixing:
        self.mixing = Downmix.Dave_750
    if caller:
        debug("Applying downmix/pan filter...", caller)
    return f"pan={self.mixing}"

ATSC()

Source code in muxtools/audio/preprocess.py 
125
126
127
@classproperty
def ATSC(self) -> str:
    return "stereo|FL<1.0*FL+0.707*FC+0.707*BL+0.707*SL|FR<1.0*FR+0.707*FC+0.707*BR+0.707*SR"

Collier()

Source code in muxtools/audio/preprocess.py 
129
130
131
@classproperty
def Collier(self) -> str:
    return "stereo|FL=FC+0.30*FL+0.30*BL+0.30*SL|FR=FC+0.30*FR+0.30*BR+0.30*SR"

Dave_750()

Source code in muxtools/audio/preprocess.py 
133
134
135
@classproperty
def Dave_750(self) -> str:
    return "stereo|FL=0.5*FC+0.707*FL+0.707*BL+0.707*SL+0.5*LFE|FR=0.5*FC+0.707*FR+0.707*BR+0.707*SR+0.35*LFE"

RFC_7845()

Source code in muxtools/audio/preprocess.py 
137
138
139
@classproperty
def RFC_7845(self) -> str:
    return "stereo|FL=0.374107*FC+0.529067*FL+0.458186*BL+0.458186*SL+0.264534*BR+0.264534*SR+0.374107*LFE|FR=0.374107*FC+0.529067*FR+0.458186*BR+0.458186*SR+0.264534*BL+0.264534*SL+0.374107*LFE"

Loudnorm dataclass

Bases: Preprocessor

A FFMPEG normalization preprocessor according to EBU-R128 standards. It's strongly recommended to also put a Resample preprocessor into the chain as this filter needs to upsample to 192kHz and we don't want to encode that after.

This will do a dynamic pass first to measure various values and then do the proper pass so it might take a while.

Parameters:

Name Type Description Default
i float

The integrated loudness target. Range is -70.0 - -5.0. Default value is -24.0.

 -24.0
lra float

The loudness range target. Range is 1.0 - 50.0. Default value is 7.0.

 7.0
tp float

The maximum true peak. Range is -9.0 - +0.0. Default value is -2.0.

 -2.0
offset float | None

Offset gain. Gain is applied before the true-peak limiter. Will be taken from the analysis in the first pass if None.

 None
Source code in muxtools/audio/preprocess.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
@dataclass
class Loudnorm(Preprocessor):
    """
    A FFMPEG normalization preprocessor according to EBU-R128 standards.
    It's strongly recommended to also put a `Resample` preprocessor into the chain as this filter needs to upsample to 192kHz and we don't want to encode that after.

    This will do a dynamic pass first to measure various values and then do the proper pass so it might take a while.

    :param i:           The integrated loudness target. Range is `-70.0` - `-5.0`. Default value is `-24.0`.
    :param lra:         The loudness range target. Range is `1.0` - `50.0`. Default value is `7.0`.
    :param tp:          The maximum true peak. Range is `-9.0` - `+0.0`. Default value is `-2.0`.
    :param offset:      Offset gain. Gain is applied before the true-peak limiter.
                        Will be taken from the analysis in the first pass if None.
    """

    i: float = -24.0
    lra: float = 7.0
    tp: float = -2.0
    offset: float | None = None

    @dataclass
    class Measurements:
        i: float
        lra: float
        tp: float
        thresh: float
        target_offset: float

    def can_run(self, track: Track, preprocessors: Sequence[Any]) -> bool:
        return True

    def analyze(self, file: AudioFile):
        debug("Analyzing file loudness...", self)
        ffmpeg = get_executable("ffmpeg")
        out_var = "NUL" if os.name == "nt" else "/dev/null"
        args = [
            ffmpeg,
            "-y",
            "-hide_banner",
            "-i",
            str(ensure_path_exists(file.file, self).resolve()),
            "-map",
            "0:a:0",
            "-filter:a",
            "loudnorm=print_format=json",
            "-f",
            "null",
            out_var,
        ]
        out = subprocess.run(args, capture_output=True, text=True, encoding="utf-8", errors="ignore")
        output = (out.stderr or "") + (out.stdout or "")
        output = output.replace("\n", "").replace("\r", "")
        i_match = re.findall(r"input_i.+?(-?\d+(?:\.\d+)?)", output, re.I)
        tp_match = re.findall(r"input_tp.+?(-?\d+(?:\.\d+)?)", output, re.I)
        lra_match = re.findall(r"input_lra.+?(-?\d+(?:\.\d+)?)", output, re.I)
        thresh_match = re.findall(r"input_thresh.+?(-?\d+(?:\.\d+)?)", output, re.I)
        offset_match = re.findall(r"target_offset.+?(-?\d+(?:\.\d+)?)", output, re.I)
        if not all([i_match, tp_match, lra_match, thresh_match, offset_match]):
            raise error("Could not properly measure the input file!", self)

        self.measurements = self.Measurements(
            float(i_match[0]),
            float(lra_match[0]),
            float(tp_match[0]),
            float(thresh_match[0]),
            float(offset_match[0]),
        )

    def get_filter(self, caller: Any = None) -> str | None:
        if caller:
            debug("Applying loudnorm...", caller)
        if not hasattr(self, "measurements"):
            # Ideally shouldn't run into this lmfao
            return ""
        return (
            f"loudnorm=linear=true:i={self.i}:lra={self.lra}:tp={self.tp}:offset={self.offset if self.offset else self.measurements.target_offset}"
            f":measured_I={self.measurements.i}:measured_tp={self.measurements.tp}:measured_LRA={self.measurements.lra}:measured_thresh={self.measurements.thresh}"
        )

i: float = -24.0 class-attribute instance-attribute

lra: float = 7.0 class-attribute instance-attribute

tp: float = -2.0 class-attribute instance-attribute

offset: float | None = None class-attribute instance-attribute

Measurements dataclass

Source code in muxtools/audio/preprocess.py 
165
166
167
168
169
170
171
@dataclass
class Measurements:
    i: float
    lra: float
    tp: float
    thresh: float
    target_offset: float

i: float instance-attribute

lra: float instance-attribute

tp: float instance-attribute

thresh: float instance-attribute

target_offset: float instance-attribute

__init__(i, lra, tp, thresh, target_offset)

__init__(i=-24.0, lra=7.0, tp=-2.0, offset=None)

can_run(track, preprocessors)

Source code in muxtools/audio/preprocess.py 
173
174
def can_run(self, track: Track, preprocessors: Sequence[Any]) -> bool:
    return True

analyze(file)

Source code in muxtools/audio/preprocess.py 
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
def analyze(self, file: AudioFile):
    debug("Analyzing file loudness...", self)
    ffmpeg = get_executable("ffmpeg")
    out_var = "NUL" if os.name == "nt" else "/dev/null"
    args = [
        ffmpeg,
        "-y",
        "-hide_banner",
        "-i",
        str(ensure_path_exists(file.file, self).resolve()),
        "-map",
        "0:a:0",
        "-filter:a",
        "loudnorm=print_format=json",
        "-f",
        "null",
        out_var,
    ]
    out = subprocess.run(args, capture_output=True, text=True, encoding="utf-8", errors="ignore")
    output = (out.stderr or "") + (out.stdout or "")
    output = output.replace("\n", "").replace("\r", "")
    i_match = re.findall(r"input_i.+?(-?\d+(?:\.\d+)?)", output, re.I)
    tp_match = re.findall(r"input_tp.+?(-?\d+(?:\.\d+)?)", output, re.I)
    lra_match = re.findall(r"input_lra.+?(-?\d+(?:\.\d+)?)", output, re.I)
    thresh_match = re.findall(r"input_thresh.+?(-?\d+(?:\.\d+)?)", output, re.I)
    offset_match = re.findall(r"target_offset.+?(-?\d+(?:\.\d+)?)", output, re.I)
    if not all([i_match, tp_match, lra_match, thresh_match, offset_match]):
        raise error("Could not properly measure the input file!", self)

    self.measurements = self.Measurements(
        float(i_match[0]),
        float(lra_match[0]),
        float(tp_match[0]),
        float(thresh_match[0]),
        float(offset_match[0]),
    )

get_filter(caller=None)

Source code in muxtools/audio/preprocess.py 
213
214
215
216
217
218
219
220
221
222
def get_filter(self, caller: Any = None) -> str | None:
    if caller:
        debug("Applying loudnorm...", caller)
    if not hasattr(self, "measurements"):
        # Ideally shouldn't run into this lmfao
        return ""
    return (
        f"loudnorm=linear=true:i={self.i}:lra={self.lra}:tp={self.tp}:offset={self.offset if self.offset else self.measurements.target_offset}"
        f":measured_I={self.measurements.i}:measured_tp={self.measurements.tp}:measured_LRA={self.measurements.lra}:measured_thresh={self.measurements.thresh}"
    )

CustomPreprocessor dataclass

Bases: Preprocessor

A custom preprocessor class to pass arbitrary filters or arguments to ffmpeg.

Parameters:

Name Type Description Default
filt str | None

Audio filter to append to the filterchain. Don't include any flags or whatever. It should look like this afade=t=in:ss=0:d=15

 None
args str | Sequence[str] | None

Other args you may want to pass to ffmpeg.

 None
Source code in muxtools/audio/preprocess.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
@dataclass
class CustomPreprocessor(Preprocessor):
    """
    A custom preprocessor class to pass arbitrary filters or arguments to ffmpeg.

    :param filt:        Audio filter to append to the filterchain. Don't include any flags or whatever.
                        It should look like this `afade=t=in:ss=0:d=15`
    :param args:        Other args you may want to pass to ffmpeg.
    """

    filt: str | None = None
    args: str | Sequence[str] | None = None

    def can_run(self, track: Track, preprocessors: Sequence[Any]) -> bool:
        return True

    def get_filter(self, caller: Any = None) -> str | None:
        return self.filt

    def get_args(self, caller: Any = None) -> Sequence[str]:
        if isinstance(self.args, str) and not isinstance(self.args, Sequence):
            self.args = [self.args]
        return list(self.args) if self.args else []

filt: str | None = None class-attribute instance-attribute

args: str | Sequence[str] | None = None class-attribute instance-attribute

__init__(filt=None, args=None)

can_run(track, preprocessors)

Source code in muxtools/audio/preprocess.py 
238
239
def can_run(self, track: Track, preprocessors: Sequence[Any]) -> bool:
    return True

get_filter(caller=None)

Source code in muxtools/audio/preprocess.py 
241
242
def get_filter(self, caller: Any = None) -> str | None:
    return self.filt

get_args(caller=None)

Source code in muxtools/audio/preprocess.py 
244
245
246
247
def get_args(self, caller: Any = None) -> Sequence[str]:
    if isinstance(self.args, str) and not isinstance(self.args, Sequence):
        self.args = [self.args]
    return list(self.args) if self.args else []