/
/
/
1"""Loudness Analysis provider - measures EBU R128 integrated loudness via FFmpeg ebur128."""
2
3from __future__ import annotations
4
5import contextlib
6import re
7from collections.abc import Iterable
8from dataclasses import dataclass
9from typing import TYPE_CHECKING
10
11from music_assistant_models.config_entries import ConfigEntry
12from music_assistant_models.enums import ConfigEntryType, VolumeNormalizationMode
13
14from music_assistant.constants import LOUDNESS_MEASUREMENT_MIN_LUFS
15from music_assistant.helpers.ffmpeg import FFMpeg
16from music_assistant.helpers.tags import write_replaygain_track_gain
17from music_assistant.models.audio_analysis import AudioAnalysisData, AudioAnalysisError
18from music_assistant.models.audio_analysis_provider import AudioAnalysisProvider
19
20if TYPE_CHECKING:
21 from music_assistant_models.config_entries import ProviderConfig
22 from music_assistant_models.enums import ProviderFeature
23 from music_assistant_models.media_items import AudioFormat
24 from music_assistant_models.provider import ProviderManifest
25 from music_assistant_models.streamdetails import StreamDetails
26
27 from music_assistant.mass import MusicAssistant
28
29MAX_DURATION_SECONDS = 600
30MIN_DURATION_SECONDS = 10
31
32CONF_WRITE_REPLAYGAIN_TAGS = "write_replaygain_tags"
33
34_INTEGRATED_RE = re.compile(r"Integrated loudness:.*?I:\s*(-?\d+(?:\.\d+)?)\s*LUFS", re.DOTALL)
35_LRA_RE = re.compile(r"Loudness range:.*?LRA:\s*(-?\d+(?:\.\d+)?)\s*LU", re.DOTALL)
36_TRUE_PEAK_RE = re.compile(r"True peak:.*?Peak:\s*(-?\d+(?:\.\d+)?)\s*dBFS", re.DOTALL)
37
38
39@dataclass
40class LoudnessSessionData:
41 """Per-session state for a loudness analysis job."""
42
43 ffmpeg: FFMpeg
44 chunks_received: int = 0
45 eof_sent: bool = False
46
47
48class LoudnessAnalysisProvider(AudioAnalysisProvider):
49 """Audio analysis provider that measures EBU R128 integrated loudness."""
50
51 analysis_version: int = 2
52
53 def __init__(
54 self,
55 mass: MusicAssistant,
56 manifest: ProviderManifest,
57 config: ProviderConfig,
58 supported_features: set[ProviderFeature],
59 ) -> None:
60 """Initialize the provider."""
61 super().__init__(mass, manifest, config, supported_features)
62 self._data: dict[str, LoudnessSessionData] = {}
63
64 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
65 """Return config entries for this provider."""
66 return (
67 ConfigEntry(
68 key=CONF_WRITE_REPLAYGAIN_TAGS,
69 type=ConfigEntryType.BOOLEAN,
70 default_value=False,
71 required=False,
72 ),
73 )
74
75 async def process_pcm_chunk(
76 self,
77 session_id: str,
78 pcm_chunk: bytes,
79 ) -> None:
80 """Feed a PCM chunk to the session's ebur128 ffmpeg process."""
81 data = self._data.get(session_id)
82 if not data or data.eof_sent:
83 return
84 data.chunks_received += 1
85 await data.ffmpeg.write(pcm_chunk)
86 if data.chunks_received >= MAX_DURATION_SECONDS:
87 # cap the analysis window for very long streams
88 await self._send_eof(data)
89
90 async def cancel(self, session_id: str) -> None:
91 """Abort an in-progress loudness analysis session."""
92 data = self._data.pop(session_id, None)
93 if data:
94 with contextlib.suppress(OSError):
95 await data.ffmpeg.close()
96 await super().cancel(session_id)
97
98 async def post_analysis(
99 self,
100 streamdetails: StreamDetails,
101 analysis: AudioAnalysisData,
102 ) -> None:
103 """Write the ReplayGain track-gain tag back to the source file when configured."""
104 if not isinstance(streamdetails.path, str) or not streamdetails.path:
105 return
106 if not self.config.get_value(CONF_WRITE_REPLAYGAIN_TAGS):
107 return
108 if analysis.loudness_integrated is None:
109 return
110 # ReplayGain 2.0: track_gain_db = -18 - loudness_lufs
111 track_gain_db = -18.0 - analysis.loudness_integrated
112 ok = await write_replaygain_track_gain(streamdetails.path, track_gain_db)
113 if ok:
114 self.logger.debug(
115 "Wrote ReplayGain tag to %s (gain=%.2f dB)",
116 streamdetails.path,
117 track_gain_db,
118 )
119
120 async def _start_analysis(
121 self,
122 session_id: str,
123 streamdetails: StreamDetails,
124 audio_format: AudioFormat,
125 ) -> bool:
126 """Prepare provider state for a new analysis session."""
127 # skip when nothing here normalizes on our side: the player opted out, or the
128 # source levelled the audio itself and measuring its output would store that
129 # level as the track's own. The nightly background job picks the measurement
130 # up if it is ever needed
131 if streamdetails.volume_normalization_mode in (
132 VolumeNormalizationMode.DISABLED,
133 VolumeNormalizationMode.SOURCE,
134 ):
135 return False
136 ffmpeg = FFMpeg(
137 audio_input="-",
138 input_format=audio_format,
139 output_format=audio_format,
140 audio_output="NULL",
141 filter_params=["ebur128=framelog=verbose:peak=true"],
142 collect_log_history=True,
143 loglevel="info",
144 )
145 await ffmpeg.start()
146 self._data[session_id] = LoudnessSessionData(ffmpeg=ffmpeg)
147 return True
148
149 async def _finalize(self, session_id: str) -> AudioAnalysisData | None:
150 """Persist the final loudness measurement for the session."""
151 data = self._data.pop(session_id, None)
152 if not data:
153 return None
154
155 await self._send_eof(data)
156 try:
157 await data.ffmpeg.wait()
158 except Exception as err:
159 # ffmpeg.wait() can surface process/pipe errors plus anything the ebur128
160 # subprocess raises; broad so a failed measurement degrades to "no result"
161 # rather than crashing finalize.
162 self.logger.debug("Loudness analysis ffmpeg failed: %s", err)
163 await data.ffmpeg.close()
164 return None
165
166 metrics = _parse_ebur128_metrics(data.ffmpeg.log_history)
167 await data.ffmpeg.close()
168
169 session = self._sessions.get(session_id)
170 if session is None:
171 return None
172
173 if data.chunks_received < MIN_DURATION_SECONDS:
174 raise AudioAnalysisError("track too short for loudness measurement")
175
176 loudness, loudness_range, true_peak = metrics
177 if loudness is None:
178 self.logger.debug(
179 "Could not determine loudness of %s from buffer analysis",
180 session.streamdetails.uri,
181 )
182 return None
183
184 if loudness <= LOUDNESS_MEASUREMENT_MIN_LUFS:
185 # ebur128 reports ~-70 LUFS on a near-silent track; below the reliability floor
186 # it would cause huge gain corrections, and the reading is deterministic per file.
187 raise AudioAnalysisError("track too quiet to measure loudness")
188
189 analysis = AudioAnalysisData(
190 loudness_integrated=round(loudness, 2),
191 loudness_range=round(loudness_range, 2) if loudness_range is not None else None,
192 true_peak=round(true_peak, 2) if true_peak is not None else None,
193 )
194 # update in-memory streamdetails so subsequent seeks use the measurement
195 # instead of dynamic normalization
196 session.streamdetails.loudness = round(loudness, 2)
197 self.logger.debug(
198 "Loudness measurement for %s: %s LUFS (LRA=%s LU, peak=%s dBTP)",
199 session.streamdetails.uri,
200 loudness,
201 loudness_range,
202 true_peak,
203 )
204 return analysis
205
206 async def _send_eof(self, data: LoudnessSessionData) -> None:
207 """Signal end-of-input to the session's ffmpeg process (idempotent)."""
208 if data.eof_sent:
209 return
210 data.eof_sent = True
211 with contextlib.suppress(OSError):
212 await data.ffmpeg.write_eof()
213
214
215def _parse_ebur128_metrics(
216 log_lines: Iterable[str],
217) -> tuple[float | None, float | None, float | None]:
218 """Extract (integrated_loudness, loudness_range, true_peak) from an ebur128 log."""
219 log = "\n".join(log_lines)
220 integrated = _match_float(_INTEGRATED_RE, log)
221 lra = _match_float(_LRA_RE, log)
222 true_peak = _match_float(_TRUE_PEAK_RE, log)
223 return integrated, lra, true_peak
224
225
226def _match_float(pattern: re.Pattern[str], text: str) -> float | None:
227 match = pattern.search(text)
228 if not match:
229 return None
230 try:
231 return float(match.group(1))
232 except ValueError:
233 return None
234