quaik8/sdb-back
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add VLC service for sound playback and management

Browse Source 
- Implemented VLCService to handle sound playback using VLC.
- Added routes for soundboard management including play, stop, and status.
- Introduced admin routes for sound normalization and scanning.
- Updated user model and services to accommodate new functionalities.
- Enhanced error handling and logging throughout the application.
- Updated dependencies to include python-vlc for sound playback capabilities.
This commit is contained in:
JSC
2025-07-03 21:25:50 +02:00
parent 8f17dd730a
commit 7455811860
20 changed files with 760 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
app/services/sound_normalizer_service.py
View File

@@ -38,7 +38,9 @@ class SoundNormalizerService:
}
@staticmethod
def normalize_sound(sound_id: int, overwrite: bool = False, two_pass: bool = True) -> dict:
def normalize_sound(
sound_id: int, overwrite: bool = False, two_pass: bool = True
) -> dict:
"""Normalize a specific sound file using ffmpeg loudnorm.
Args:
@@ -58,7 +60,9 @@ class SoundNormalizerService:
"error": f"Sound with ID {sound_id} not found",
}
source_path = Path(SoundNormalizerService.SOUNDS_DIR) / sound.filename
source_path = (
Path(SoundNormalizerService.SOUNDS_DIR) / sound.filename
)
if not source_path.exists():
return {
"success": False,
@@ -68,7 +72,10 @@ class SoundNormalizerService:
# Always output as WAV regardless of input format
filename_without_ext = Path(sound.filename).stem
normalized_filename = f"{filename_without_ext}.wav"
normalized_path = Path(SoundNormalizerService.NORMALIZED_DIR) / normalized_filename
normalized_path = (
Path(SoundNormalizerService.NORMALIZED_DIR)
/ normalized_filename
)
normalized_path.parent.mkdir(parents=True, exist_ok=True)
@@ -84,11 +91,15 @@ class SoundNormalizerService:
if two_pass:
result = SoundNormalizerService._normalize_with_ffmpeg(
str(source_path), str(normalized_path),
str(source_path),
str(normalized_path),
)
else:
result = SoundNormalizerService._normalize_with_ffmpeg_single_pass(
str(source_path), str(normalized_path),
result = (
SoundNormalizerService._normalize_with_ffmpeg_single_pass(
str(source_path),
str(normalized_path),
)
)
if result["success"]:
@@ -131,7 +142,9 @@ class SoundNormalizerService:
@staticmethod
def normalize_all_sounds(
overwrite: bool = False, limit: int = None, two_pass: bool = True,
overwrite: bool = False,
limit: int = None,
two_pass: bool = True,
) -> dict:
"""Normalize all soundboard files.
@@ -171,7 +184,9 @@ class SoundNormalizerService:
for sound in sounds:
result = SoundNormalizerService.normalize_sound(
sound.id, overwrite, two_pass,
sound.id,
overwrite,
two_pass,
)
processed += 1
@@ -233,19 +248,19 @@ class SoundNormalizerService:
# FIRST PASS: Analyze the audio to get optimal parameters
logger.debug("Starting first pass (analysis)")
first_pass_result = SoundNormalizerService._run_first_pass(
source_path, params
)
if not first_pass_result["success"]:
return first_pass_result
measured_params = first_pass_result["measured_params"]
# SECOND PASS: Apply normalization using measured parameters
logger.debug("Starting second pass (normalization)")
second_pass_result = SoundNormalizerService._run_second_pass(
source_path, output_path, params, measured_params
)
@@ -281,9 +296,11 @@ class SoundNormalizerService:
return {"success": False, "error": str(e)}
@staticmethod
def _normalize_with_ffmpeg_single_pass(source_path: str, output_path: str) -> dict:
def _normalize_with_ffmpeg_single_pass(
source_path: str, output_path: str
) -> dict:
"""Run ffmpeg loudnorm on a single file using single-pass normalization.
This is the legacy single-pass method for backward compatibility.
Args:
@@ -319,7 +336,9 @@ class SoundNormalizerService:
# Run the ffmpeg process
out, err = ffmpeg.run(
output_stream, capture_stdout=True, capture_stderr=True,
output_stream,
capture_stdout=True,
capture_stderr=True,
)
# Parse loudnorm statistics from stderr
@@ -348,11 +367,11 @@ class SoundNormalizerService:
@staticmethod
def _run_first_pass(source_path: str, params: dict) -> dict:
"""Run first pass of loudnorm to analyze audio characteristics.
Args:
source_path: Path to source audio file
params: Loudnorm target parameters
Returns:
dict: Result with measured parameters and analysis stats
"""
@@ -370,35 +389,36 @@ class SoundNormalizerService:
# Output to null device for analysis
output_stream = ffmpeg.output(
input_stream,
"/dev/null",
af=loudnorm_filter,
f="null"
input_stream, "/dev/null", af=loudnorm_filter, f="null"
)
# Run the first pass
out, err = ffmpeg.run(
output_stream, capture_stdout=True, capture_stderr=True,
output_stream,
capture_stdout=True,
capture_stderr=True,
)
stderr_text = err.decode() if err else ""
# Parse measured parameters from JSON output
measured_params = SoundNormalizerService._parse_measured_params(stderr_text)
measured_params = SoundNormalizerService._parse_measured_params(
stderr_text
)
if not measured_params:
return {
"success": False,
"error": "Failed to parse measured parameters from first pass"
"error": "Failed to parse measured parameters from first pass",
}
# Parse basic stats
stats = SoundNormalizerService._parse_loudnorm_stats(stderr_text)
return {
"success": True,
"measured_params": measured_params,
"stats": stats
"stats": stats,
}
except ffmpeg.Error as e:
@@ -410,15 +430,20 @@ class SoundNormalizerService:
return {"success": False, "error": str(e)}
@staticmethod
def _run_second_pass(source_path: str, output_path: str, target_params: dict, measured_params: dict) -> dict:
def _run_second_pass(
source_path: str,
output_path: str,
target_params: dict,
measured_params: dict,
) -> dict:
"""Run second pass of loudnorm using measured parameters.
Args:
source_path: Path to source audio file
output_path: Path for normalized output file
target_params: Target loudnorm parameters
measured_params: Parameters measured from first pass
Returns:
dict: Result with normalization stats
"""
@@ -452,18 +477,17 @@ class SoundNormalizerService:
# Run the second pass
out, err = ffmpeg.run(
output_stream, capture_stdout=True, capture_stderr=True,
output_stream,
capture_stdout=True,
capture_stderr=True,
)
stderr_text = err.decode() if err else ""
# Parse final statistics
stats = SoundNormalizerService._parse_loudnorm_stats(stderr_text)
return {
"success": True,
"stats": stats
}
return {"success": True, "stats": stats}
except ffmpeg.Error as e:
error_msg = f"Second pass FFmpeg error: {e.stderr.decode() if e.stderr else str(e)}"
@@ -476,23 +500,25 @@ class SoundNormalizerService:
@staticmethod
def _parse_measured_params(stderr_output: str) -> dict:
"""Parse measured parameters from first pass JSON output.
Args:
stderr_output: ffmpeg stderr output containing JSON data
Returns:
dict: Parsed measured parameters, empty if parsing fails
"""
try:
# Find JSON block in stderr output
json_match = re.search(r'\{[^}]*"input_i"[^}]*\}', stderr_output, re.DOTALL)
json_match = re.search(
r'\{[^}]*"input_i"[^}]*\}', stderr_output, re.DOTALL
)
if not json_match:
logger.warning("No JSON block found in first pass output")
return {}
json_str = json_match.group(0)
measured_data = json.loads(json_str)
# Extract required parameters
return {
"input_i": measured_data.get("input_i", 0),
@@ -501,7 +527,7 @@ class SoundNormalizerService:
"input_thresh": measured_data.get("input_thresh", 0),
"target_offset": measured_data.get("target_offset", 0),
}
except (json.JSONDecodeError, KeyError, AttributeError) as e:
logger.warning(f"Failed to parse measured parameters: {e}")
return {}
@@ -625,7 +651,9 @@ class SoundNormalizerService:
sounds = Sound.query.filter_by(type="SDB").all()
for sound in sounds:
original_path = Path(SoundNormalizerService.SOUNDS_DIR) / sound.filename
original_path = (
Path(SoundNormalizerService.SOUNDS_DIR) / sound.filename
)
if original_path.exists():
total_original_size += original_path.stat().st_size
@@ -633,7 +661,10 @@ class SoundNormalizerService:
# Use database field to check if normalized, not file existence
if sound.is_normalized and sound.normalized_filename:
normalized_count += 1
normalized_path = Path(SoundNormalizerService.NORMALIZED_DIR) / sound.normalized_filename
normalized_path = (
Path(SoundNormalizerService.NORMALIZED_DIR)
/ sound.normalized_filename
)
if normalized_path.exists():
total_normalized_size += normalized_path.stat().st_size
@@ -676,7 +707,8 @@ class SoundNormalizerService:
import tempfile
with tempfile.NamedTemporaryFile(
suffix=".wav", delete=False,
suffix=".wav",
delete=False,
) as temp_file:
temp_path = temp_file.name