/
/
/
1"""Helper functions for DSP filters."""
2
3import math
4import os
5from dataclasses import dataclass, field
6from typing import TYPE_CHECKING
7
8from music_assistant_models.dsp import (
9 AudioChannel,
10 BalanceFilter,
11 CompressorFilter,
12 ConvolutionFilter,
13 CrossfeedFilter,
14 DSPFilter,
15 GainFilter,
16 HighLowPassFilter,
17 HighLowPassMode,
18 ParametricEQBandType,
19 ParametricEQFilter,
20 SafetyLimiterFilter,
21 StereoWidthFilter,
22 ToneControlFilter,
23 TransposeFilter,
24)
25
26from music_assistant.constants import DSP_IR_ID_RE
27
28if TYPE_CHECKING:
29 from music_assistant_models.media_items.audio_format import AudioFormat
30
31# ruff: noqa: PLR0915
32
33
34@dataclass(slots=True)
35class ComplexFilterInput:
36 """
37 An extra audio source feeding a ComplexFilter.
38
39 :param path: Audio source to read, either a file path or a URL.
40 :param filters: Optional chain applied to the input before the body consumes
41 it (e.g. "aresample=48000").
42 :param input_args: Optional FFmpeg options for reading this input, placed
43 before its ``-i`` on top of the ones every input already gets
44 (e.g. ["-stream_loop", "-1"]).
45 """
46
47 path: str
48 filters: str = ""
49 input_args: list[str] = field(default_factory=list)
50
51
52@dataclass(slots=True)
53class ComplexFilter:
54 """
55 A DSP filter fragment that pulls in one or more extra audio inputs.
56
57 Represents a chain entry that cannot be expressed as a plain single-input
58 filter string, such as an FFmpeg ``afir`` convolution that needs an
59 impulse-response input.
60
61 :param body: The filter consuming the main input followed by each extra
62 input in order (e.g. "afir=irnorm=1").
63 :param inputs: Extra audio sources for ``body``, in the order it consumes them.
64 """
65
66 body: str
67 inputs: list[ComplexFilterInput] = field(default_factory=list)
68
69
70def filter_to_ffmpeg_params(
71 dsp_filter: DSPFilter, input_format: AudioFormat, *, ir_dir: str
72) -> list[str | ComplexFilter]:
73 """
74 Convert a DSP filter model to FFmpeg filter parameters.
75
76 :param dsp_filter: DSP filter configuration.
77 :param input_format: Input audio format (sample rate/channels).
78 :param ir_dir: Directory holding convolution impulse responses, used to resolve a
79 ConvolutionFilter's ir_id to a file path.
80 :return: Ordered chain of FFmpeg filter strings and/or ComplexFilter fragments.
81 """
82 filter_params: list[str | ComplexFilter] = []
83
84 if isinstance(dsp_filter, ParametricEQFilter):
85 has_per_channel_preamp = any(value != 0 for value in dsp_filter.per_channel_preamp.values())
86 if dsp_filter.preamp and dsp_filter.preamp != 0 and not has_per_channel_preamp:
87 filter_params.append(f"volume={dsp_filter.preamp}dB")
88 # "volume" is handled for the whole audio stream only, so we'll use the pan filter instead
89 elif has_per_channel_preamp:
90 channel_config = []
91 all_channels = [AudioChannel.FL, AudioChannel.FR]
92 for channel_id in all_channels:
93 # Get gain for this channel, default to 0 if not specified
94 gain_db = dsp_filter.per_channel_preamp.get(channel_id, 0)
95 # Apply both the overall preamp and the per-channel preamp
96 total_gain_db = (
97 dsp_filter.preamp + gain_db if dsp_filter.preamp is not None else gain_db
98 )
99 if total_gain_db != 0:
100 # Convert dB to linear gain
101 gain = 10 ** (total_gain_db / 20)
102 channel_config.append(f"{channel_id}={gain}*{channel_id}")
103 else:
104 channel_config.append(f"{channel_id}={channel_id}")
105
106 # Could potentially also be expanded for more than 2 channels
107 filter_params.append("pan=stereo|" + "|".join(channel_config))
108 for b in dsp_filter.bands:
109 if not b.enabled:
110 continue
111 channels = ""
112 if b.channel != AudioChannel.ALL:
113 channels = f":c={b.channel}"
114 # From https://webaudio.github.io/Audio-EQ-Cookbook/audio-eq-cookbook.html
115
116 f_s = input_format.sample_rate
117 f_0 = b.frequency
118 db_gain = b.gain
119 q = b.q
120
121 a = math.sqrt(10 ** (db_gain / 20))
122 w_0 = 2 * math.pi * f_0 / f_s
123 alpha = math.sin(w_0) / (2 * q)
124
125 if b.type == ParametricEQBandType.PEAK:
126 b0 = 1 + alpha * a
127 b1 = -2 * math.cos(w_0)
128 b2 = 1 - alpha * a
129 a0 = 1 + alpha / a
130 a1 = -2 * math.cos(w_0)
131 a2 = 1 - alpha / a
132
133 filter_params.append(
134 f"biquad=b0={b0}:b1={b1}:b2={b2}:a0={a0}:a1={a1}:a2={a2}{channels}"
135 )
136 elif b.type == ParametricEQBandType.LOW_SHELF:
137 b0 = a * ((a + 1) - (a - 1) * math.cos(w_0) + 2 * math.sqrt(a) * alpha)
138 b1 = 2 * a * ((a - 1) - (a + 1) * math.cos(w_0))
139 b2 = a * ((a + 1) - (a - 1) * math.cos(w_0) - 2 * math.sqrt(a) * alpha)
140 a0 = (a + 1) + (a - 1) * math.cos(w_0) + 2 * math.sqrt(a) * alpha
141 a1 = -2 * ((a - 1) + (a + 1) * math.cos(w_0))
142 a2 = (a + 1) + (a - 1) * math.cos(w_0) - 2 * math.sqrt(a) * alpha
143
144 filter_params.append(
145 f"biquad=b0={b0}:b1={b1}:b2={b2}:a0={a0}:a1={a1}:a2={a2}{channels}"
146 )
147 elif b.type == ParametricEQBandType.HIGH_SHELF:
148 b0 = a * ((a + 1) + (a - 1) * math.cos(w_0) + 2 * math.sqrt(a) * alpha)
149 b1 = -2 * a * ((a - 1) + (a + 1) * math.cos(w_0))
150 b2 = a * ((a + 1) + (a - 1) * math.cos(w_0) - 2 * math.sqrt(a) * alpha)
151 a0 = (a + 1) - (a - 1) * math.cos(w_0) + 2 * math.sqrt(a) * alpha
152 a1 = 2 * ((a - 1) - (a + 1) * math.cos(w_0))
153 a2 = (a + 1) - (a - 1) * math.cos(w_0) - 2 * math.sqrt(a) * alpha
154
155 filter_params.append(
156 f"biquad=b0={b0}:b1={b1}:b2={b2}:a0={a0}:a1={a1}:a2={a2}{channels}"
157 )
158 elif b.type == ParametricEQBandType.HIGH_PASS:
159 filter_params.append(
160 _pass_biquad_params(high_pass=True, w_0=w_0, alpha=alpha, channels=channels)
161 )
162 elif b.type == ParametricEQBandType.LOW_PASS:
163 filter_params.append(
164 _pass_biquad_params(high_pass=False, w_0=w_0, alpha=alpha, channels=channels)
165 )
166 elif b.type == ParametricEQBandType.NOTCH:
167 b0 = 1
168 b1 = -2 * math.cos(w_0)
169 b2 = 1
170 a0 = 1 + alpha
171 a1 = -2 * math.cos(w_0)
172 a2 = 1 - alpha
173
174 filter_params.append(
175 f"biquad=b0={b0}:b1={b1}:b2={b2}:a0={a0}:a1={a1}:a2={a2}{channels}"
176 )
177 if isinstance(dsp_filter, ToneControlFilter):
178 # A basic 3-band equalizer
179 if dsp_filter.bass_level != 0:
180 filter_params.append(
181 f"equalizer=frequency=100:width=200:width_type=h:gain={dsp_filter.bass_level}"
182 )
183 if dsp_filter.mid_level != 0:
184 filter_params.append(
185 f"equalizer=frequency=900:width=1800:width_type=h:gain={dsp_filter.mid_level}"
186 )
187 if dsp_filter.treble_level != 0:
188 filter_params.append(
189 f"equalizer=frequency=9000:width=18000:width_type=h:gain={dsp_filter.treble_level}"
190 )
191 if isinstance(dsp_filter, GainFilter) and dsp_filter.gain != 0:
192 filter_params.append(f"volume={dsp_filter.gain}dB")
193 if isinstance(dsp_filter, BalanceFilter) and dsp_filter.balance != 0:
194 # balance is a stereo operation; on a non-stereo source the FL/FR pan
195 # expression would output silence, so only apply it to stereo streams
196 if input_format.channels == 2:
197 # attenuate only the channel opposite the slider direction, so there is
198 # no positive gain and thus no clipping risk
199 attenuation = (100 - abs(dsp_filter.balance)) / 100
200 if dsp_filter.balance > 0:
201 filter_params.append(f"pan=stereo|FL={attenuation}*FL|FR=FR")
202 else:
203 filter_params.append(f"pan=stereo|FL=FL|FR={attenuation}*FR")
204 if isinstance(dsp_filter, TransposeFilter) and dsp_filter.semitones != 0:
205 # rubberband expects a frequency ratio rather than a number of semitones
206 pitch = 2 ** (dsp_filter.semitones / 12)
207 # preserving formants keeps voices natural instead of chipmunk-like; revisit
208 # these quality options if they prove too costly on low powered hardware
209 filter_params.append(
210 f"rubberband=pitch={pitch}:formant=preserved:pitchq=quality:window=long"
211 )
212 if isinstance(dsp_filter, SafetyLimiterFilter):
213 # user placed safety limiter; level=false keeps it a transparent
214 # ceiling (no auto make-up), latency=true realigns the lookahead buffer
215 filter_params.append(
216 f"alimiter=limit={dsp_filter.ceiling}dB:level=false:asc=true:latency=true"
217 )
218 # a unity ratio compresses nothing, leaving the make-up gain as the only effect
219 if isinstance(dsp_filter, CompressorFilter) and (
220 dsp_filter.ratio != 1.0 or dsp_filter.makeup != 0
221 ):
222 # acompressor knee is threshold/sqrt(knee)..threshold*sqrt(knee), so a knee
223 # width of N dB maps to a linear knee factor of 10**(N/20)
224 knee = 10 ** (dsp_filter.knee / 20)
225 filter_params.append(
226 f"acompressor=threshold={dsp_filter.threshold}dB:ratio={dsp_filter.ratio}"
227 f":attack={dsp_filter.attack}:release={dsp_filter.release}"
228 f":knee={knee}:makeup={dsp_filter.makeup}dB"
229 )
230
231 if isinstance(dsp_filter, HighLowPassFilter):
232 # A high/low-pass of a given slope is a cascade of second-order Butterworth
233 # sections, each adding 12 dB/octave, at the same cutoff but different Q.
234 # slope is validated to 12, 24 or 48 dB/octave, so order is 2, 4 or 8.
235 high_pass = dsp_filter.mode == HighLowPassMode.HIGH_PASS
236 order = dsp_filter.slope // 6
237 w_0 = 2 * math.pi * dsp_filter.frequency / input_format.sample_rate
238 sin_w0 = math.sin(w_0)
239 for section in range(order // 2):
240 # Butterworth pole Q for this section
241 q = 1 / (2 * math.cos(math.pi * (2 * section + 1) / (2 * order)))
242 alpha = sin_w0 / (2 * q)
243 filter_params.append(_pass_biquad_params(high_pass=high_pass, w_0=w_0, alpha=alpha))
244
245 if isinstance(dsp_filter, StereoWidthFilter) and dsp_filter.width != 1.0:
246 # width scales the side (L-R) signal; a non-stereo source has no side
247 # component, so only apply it to stereo streams
248 if input_format.channels == 2:
249 # c=0 disables the internal hard clipping extrastereo applies by default,
250 # which would clamp a widened signal before any later filter or the output
251 # gain can bring it down. The float internal format exists for that headroom
252 filter_params.append(f"extrastereo=m={dsp_filter.width}:c=0")
253
254 if isinstance(dsp_filter, CrossfeedFilter):
255 # crossfeed blends the left and right channels for headphone listening
256 # and is only meaningful on stereo streams
257 if input_format.channels == 2:
258 # crossfeed is turned off by disabling the filter, never by a zero strength
259 # level_in defaults to 0.9, which would attenuate every crossfed stream
260 filter_params.append(
261 f"crossfeed=strength={dsp_filter.strength}:range={dsp_filter.soundstage}:level_in=1"
262 )
263
264 # the id rule also rejects the empty "no impulse response selected" value
265 if isinstance(dsp_filter, ConvolutionFilter) and DSP_IR_ID_RE.match(dsp_filter.ir_id):
266 # afir takes the impulse response as its second input; irnorm holds it at
267 # unity gain, which afir's deprecated gtype no longer does
268 ir_input = ComplexFilterInput(
269 path=os.path.join(ir_dir, f"{dsp_filter.ir_id}.wav"),
270 filters=f"aresample={input_format.sample_rate}",
271 )
272 filter_params.append(ComplexFilter(body="afir=irnorm=1", inputs=[ir_input]))
273 if dsp_filter.gain != 0:
274 filter_params.append(f"volume={dsp_filter.gain}dB")
275
276 return filter_params
277
278
279def _pass_biquad_params(*, high_pass: bool, w_0: float, alpha: float, channels: str = "") -> str:
280 """
281 Build the FFmpeg biquad parameters for one high-pass or low-pass section.
282
283 :param high_pass: True for a high-pass section, False for a low-pass section.
284 :param w_0: Normalised angular cutoff frequency, 2*pi*frequency/sample_rate.
285 :param alpha: Cookbook alpha term, sin(w_0)/(2*Q).
286 :param channels: Optional FFmpeg channel selector suffix, e.g. ":c=FL".
287 """
288 cos_w0 = math.cos(w_0)
289 if high_pass:
290 b0 = (1 + cos_w0) / 2
291 b1 = -(1 + cos_w0)
292 b2 = (1 + cos_w0) / 2
293 else:
294 b0 = (1 - cos_w0) / 2
295 b1 = 1 - cos_w0
296 b2 = (1 - cos_w0) / 2
297 a0 = 1 + alpha
298 a1 = -2 * cos_w0
299 a2 = 1 - alpha
300 return f"biquad=b0={b0}:b1={b1}:b2={b2}:a0={a0}:a1={a1}:a2={a2}{channels}"
301