First commit
This commit is contained in:
@@ -0,0 +1,239 @@
|
||||
import os
|
||||
import json
|
||||
import librosa
|
||||
import numpy as np
|
||||
from typing import Optional
|
||||
|
||||
|
||||
def analyze_audio(file_path: str) -> dict:
|
||||
"""
|
||||
Phân tích âm thanh: BPM, beat tracking, ước lượng bars.
|
||||
"""
|
||||
# Load audio
|
||||
y, sr = librosa.load(file_path, sr=None)
|
||||
|
||||
# Track beats
|
||||
tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr)
|
||||
|
||||
# Handle tempo which might be scalar or numpy array in different librosa versions
|
||||
if isinstance(tempo, np.ndarray):
|
||||
if tempo.size > 0:
|
||||
bpm = float(tempo[0])
|
||||
else:
|
||||
bpm = 120.0
|
||||
else:
|
||||
bpm = float(tempo)
|
||||
|
||||
# Convert frames to time (seconds)
|
||||
beat_times = librosa.frames_to_time(beat_frames, sr=sr).tolist()
|
||||
|
||||
# Estimate bars (assume 4/4 time signature - grouping every 4 beats)
|
||||
bar_times = [beat_times[i] for i in range(0, len(beat_times), 4)]
|
||||
|
||||
return {
|
||||
"bpm": round(bpm, 2),
|
||||
"beats": [round(t, 4) for t in beat_times],
|
||||
"bars": [round(t, 4) for t in bar_times],
|
||||
"duration": round(float(len(y)) / sr, 4)
|
||||
}
|
||||
|
||||
|
||||
def analyze_audio_advanced(file_path: str) -> dict:
|
||||
"""
|
||||
Phân tích âm thanh nâng cao: BPM, beats, bars, spectral features.
|
||||
Sử dụng librosa để trích xuất đặc trưng âm học chi tiết.
|
||||
"""
|
||||
y, sr = librosa.load(file_path, sr=None)
|
||||
duration = float(len(y)) / sr
|
||||
|
||||
# Beat tracking
|
||||
tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr)
|
||||
|
||||
if isinstance(tempo, np.ndarray):
|
||||
bpm = float(tempo[0]) if tempo.size > 0 else 120.0
|
||||
else:
|
||||
bpm = float(tempo)
|
||||
|
||||
beat_times = librosa.frames_to_time(beat_frames, sr=sr).tolist()
|
||||
bar_times = [beat_times[i] for i in range(0, len(beat_times), 4)]
|
||||
|
||||
# Spectral centroid (brightness)
|
||||
spectral_centroids = librosa.feature.spectral_centroid(y=y, sr=sr)[0]
|
||||
avg_brightness = float(np.mean(spectral_centroids))
|
||||
|
||||
# RMS energy
|
||||
rms = librosa.feature.rms(y=y)[0]
|
||||
avg_energy = float(np.mean(rms))
|
||||
|
||||
# Zero crossing rate
|
||||
zcr = librosa.feature.zero_crossing_rate(y)[0]
|
||||
avg_zcr = float(np.mean(zcr))
|
||||
|
||||
return {
|
||||
"bpm": round(bpm, 2),
|
||||
"beats": [round(t, 4) for t in beat_times],
|
||||
"bars": [round(t, 4) for t in bar_times],
|
||||
"duration": round(duration, 4),
|
||||
"spectral_centroid_avg": round(avg_brightness, 2),
|
||||
"rms_energy_avg": round(avg_energy, 6),
|
||||
"zero_crossing_rate_avg": round(avg_zcr, 6),
|
||||
"sample_rate": sr
|
||||
}
|
||||
|
||||
|
||||
def analyze_structure_with_ai(
|
||||
file_path: str,
|
||||
api_base_url: Optional[str] = None,
|
||||
model: str = "deepseek-chat"
|
||||
) -> dict:
|
||||
"""
|
||||
Phân tích cấu trúc khuôn nhạc bằng AI (OpenAI Compatible API).
|
||||
Gọi API DeepSeek/Ollama để phân đoạn bố cục (Intro, Verse, Chorus, Outro).
|
||||
|
||||
Args:
|
||||
file_path: Đường dẫn tệp âm thanh
|
||||
api_base_url: Base URL của API (mặc định dùng env OPENAI_API_BASE)
|
||||
model: Model name (DeepSeek, Ollama, etc.)
|
||||
|
||||
Note:
|
||||
API key is read exclusively from OPENAI_API_KEY env var.
|
||||
|
||||
Returns:
|
||||
dict: Kết quả phân tích cấu trúc
|
||||
"""
|
||||
# Lấy thông tin phân tích cơ bản trước
|
||||
analysis = analyze_audio_advanced(file_path)
|
||||
|
||||
# Cấu hình API - key chỉ đọc từ env var, không bao giờ truyền qua task queue
|
||||
base_url = api_base_url or os.getenv("OPENAI_API_BASE", "http://localhost:11434/v1")
|
||||
key = os.getenv("OPENAI_API_KEY", "ollama")
|
||||
|
||||
# Chuẩn bị prompt phân tích
|
||||
prompt = f"""Analyze the following audio metadata and suggest the musical structure (sections).
|
||||
|
||||
Audio Analysis:
|
||||
- BPM: {analysis['bpm']}
|
||||
- Duration: {analysis['duration']} seconds
|
||||
- Number of beats: {len(analysis['beats'])}
|
||||
- Number of bars: {len(analysis['bars'])}
|
||||
- Average spectral centroid: {analysis['spectral_centroid_avg']} Hz
|
||||
- Average RMS energy: {analysis['rms_energy_avg']}
|
||||
- Bar timestamps (seconds): {json.dumps(analysis['bars'][:20])}
|
||||
|
||||
Based on this data, estimate the song structure by identifying sections like Intro, Verse, Chorus, Bridge, Outro.
|
||||
Respond ONLY with valid JSON in this exact format:
|
||||
{{
|
||||
"sections": [
|
||||
{{"name": "Intro", "start_time": 0.0, "end_time": 8.5}},
|
||||
{{"name": "Verse", "start_time": 8.5, "end_time": 25.0}},
|
||||
{{"name": "Chorus", "start_time": 25.0, "end_time": 40.0}}
|
||||
]
|
||||
}}"""
|
||||
|
||||
try:
|
||||
import httpx
|
||||
|
||||
response = httpx.post(
|
||||
f"{base_url}/chat/completions",
|
||||
headers={
|
||||
"Authorization": f"Bearer {key}",
|
||||
"Content-Type": "application/json"
|
||||
},
|
||||
json={
|
||||
"model": model,
|
||||
"messages": [
|
||||
{"role": "system", "content": "You are a music analysis AI. Respond only with valid JSON."},
|
||||
{"role": "user", "content": prompt}
|
||||
],
|
||||
"temperature": 0.3
|
||||
},
|
||||
timeout=30.0
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
ai_content = result["choices"][0]["message"]["content"]
|
||||
|
||||
# Parse JSON response
|
||||
try:
|
||||
# Xử lý trường hợp AI trả về text bọc trong ```json ... ```
|
||||
if "```json" in ai_content:
|
||||
ai_content = ai_content.split("```json")[1].split("```")[0]
|
||||
elif "```" in ai_content:
|
||||
ai_content = ai_content.split("```")[1].split("```")[0]
|
||||
|
||||
sections = json.loads(ai_content.strip())
|
||||
except json.JSONDecodeError:
|
||||
sections = {"sections": [], "error": "AI response was not valid JSON"}
|
||||
|
||||
return {
|
||||
**analysis,
|
||||
"ai_structure": sections,
|
||||
"ai_model": model,
|
||||
"ai_status": "success"
|
||||
}
|
||||
else:
|
||||
return {
|
||||
**analysis,
|
||||
"ai_structure": {"sections": []},
|
||||
"ai_model": model,
|
||||
"ai_status": f"API error: {response.status_code}"
|
||||
}
|
||||
|
||||
except ImportError:
|
||||
# httpx not available, fall back to basic heuristic
|
||||
return {
|
||||
**analysis,
|
||||
"ai_structure": _estimate_structure_heuristic(analysis),
|
||||
"ai_model": "heuristic",
|
||||
"ai_status": "httpx not installed, using heuristic"
|
||||
}
|
||||
except Exception as e:
|
||||
# API không khả dụng, dùng heuristic
|
||||
return {
|
||||
**analysis,
|
||||
"ai_structure": _estimate_structure_heuristic(analysis),
|
||||
"ai_model": "heuristic",
|
||||
"ai_status": f"AI unavailable ({str(e)}), using heuristic"
|
||||
}
|
||||
|
||||
|
||||
def _estimate_structure_heuristic(analysis: dict) -> dict:
|
||||
"""
|
||||
Ước lượng cấu trúc bài hát bằng heuristic khi AI không khả dụng.
|
||||
Dựa trên số bars và duration để phân đoạn.
|
||||
"""
|
||||
duration = analysis.get("duration", 0)
|
||||
bars = analysis.get("bars", [])
|
||||
|
||||
if not bars or duration < 10:
|
||||
return {"sections": [{"name": "Full", "start_time": 0.0, "end_time": duration}]}
|
||||
|
||||
sections = []
|
||||
num_bars = len(bars)
|
||||
|
||||
if num_bars >= 8:
|
||||
# Intro: ~first 2 bars
|
||||
intro_end = bars[min(2, num_bars - 1)]
|
||||
sections.append({"name": "Intro", "start_time": 0.0, "end_time": round(intro_end, 4)})
|
||||
|
||||
# Main body
|
||||
if num_bars >= 16:
|
||||
verse_end = bars[min(8, num_bars - 1)]
|
||||
sections.append({"name": "Verse", "start_time": round(intro_end, 4), "end_time": round(verse_end, 4)})
|
||||
|
||||
if num_bars >= 24:
|
||||
chorus_end = bars[min(16, num_bars - 1)]
|
||||
sections.append({"name": "Chorus", "start_time": round(verse_end, 4), "end_time": round(chorus_end, 4)})
|
||||
|
||||
# Outro: last bars to end
|
||||
outro_start = bars[min(num_bars - 2, num_bars - 1)]
|
||||
sections.append({"name": "Outro", "start_time": round(outro_start, 4), "end_time": round(duration, 4)})
|
||||
else:
|
||||
sections.append({"name": "Outro", "start_time": round(verse_end, 4), "end_time": round(duration, 4)})
|
||||
else:
|
||||
sections.append({"name": "Main", "start_time": round(intro_end, 4), "end_time": round(duration, 4)})
|
||||
else:
|
||||
sections.append({"name": "Full", "start_time": 0.0, "end_time": round(duration, 4)})
|
||||
|
||||
return {"sections": sections}
|
||||
@@ -0,0 +1,270 @@
|
||||
import os
|
||||
import tempfile
|
||||
import numpy as np
|
||||
import soundfile as sf
|
||||
from pydub import AudioSegment
|
||||
from app.core.dsp_utils import find_nearest_zero_crossing_file, apply_micro_fade
|
||||
|
||||
def edit_audio_file(config: dict, input_path: str, output_path: str):
|
||||
"""
|
||||
Applies editing commands on the audio file based on config:
|
||||
- cut_start_ms, cut_end_ms (with optional zero_crossing_align)
|
||||
- loop_count
|
||||
- fade_in_ms, fade_out_ms
|
||||
- volume_change_db
|
||||
"""
|
||||
# Load original audio
|
||||
audio = AudioSegment.from_file(input_path)
|
||||
|
||||
# Calculate start and end positions in ms
|
||||
cut_start_ms = config.get("cut_start_ms")
|
||||
cut_end_ms = config.get("cut_end_ms")
|
||||
|
||||
total_len = len(audio)
|
||||
|
||||
start_ms = float(cut_start_ms) if cut_start_ms is not None else 0.0
|
||||
end_ms = float(cut_end_ms) if cut_end_ms is not None else float(total_len)
|
||||
|
||||
# Bound start and end
|
||||
start_ms = max(0.0, min(start_ms, float(total_len)))
|
||||
end_ms = max(start_ms, min(end_ms, float(total_len)))
|
||||
|
||||
# Align to Zero-crossing if requested
|
||||
if config.get("zero_crossing_align", True):
|
||||
# Align start
|
||||
if cut_start_ms is not None:
|
||||
start_sec = start_ms / 1000.0
|
||||
aligned_start_sec = find_nearest_zero_crossing_file(input_path, start_sec)
|
||||
start_ms = aligned_start_sec * 1000.0
|
||||
|
||||
# Align end
|
||||
if cut_end_ms is not None:
|
||||
end_sec = end_ms / 1000.0
|
||||
aligned_end_sec = find_nearest_zero_crossing_file(input_path, end_sec)
|
||||
end_ms = aligned_end_sec * 1000.0
|
||||
|
||||
# Crop the segment
|
||||
segment = audio[start_ms:end_ms]
|
||||
|
||||
# Apply micro-fade (50ms) to ensure smooth boundaries
|
||||
segment = apply_micro_fade(segment, fade_duration_ms=50)
|
||||
|
||||
# Loop the segment
|
||||
loop_count = int(config.get("loop_count", 1))
|
||||
if loop_count > 1:
|
||||
segment = segment * loop_count
|
||||
|
||||
# Apply volume change
|
||||
volume_change_db = float(config.get("volume_change_db", 0.0))
|
||||
if volume_change_db != 0.0:
|
||||
segment = segment + volume_change_db
|
||||
|
||||
# Apply custom fades
|
||||
fade_in_ms = float(config.get("fade_in_ms", 0.0))
|
||||
if fade_in_ms > 0:
|
||||
segment = segment.fade_in(int(fade_in_ms))
|
||||
|
||||
# Apply fade out
|
||||
fade_out_ms = float(config.get("fade_out_ms", 0.0))
|
||||
if fade_out_ms > 0:
|
||||
segment = segment.fade_out(int(fade_out_ms))
|
||||
|
||||
# Ensure parent output directory exists
|
||||
os.makedirs(os.path.dirname(output_path), exist_ok=True)
|
||||
|
||||
# Export as WAV
|
||||
segment.export(output_path, format="wav")
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"output_path": output_path,
|
||||
"duration_ms": len(segment),
|
||||
"aligned_start_ms": round(start_ms, 2),
|
||||
"aligned_end_ms": round(end_ms, 2)
|
||||
}
|
||||
|
||||
def cut_and_loop_segment(
|
||||
file_path: str,
|
||||
start_sec: float,
|
||||
end_sec: float,
|
||||
loop_count: int = 1,
|
||||
fade_in_ms: int = 50,
|
||||
fade_out_ms: int = 50,
|
||||
volume_db_change: float = 0.0
|
||||
) -> AudioSegment:
|
||||
"""
|
||||
Cắt đoạn âm thanh, áp dụng gain, fade-in/out và lặp lại đoạn đó.
|
||||
pydub sử dụng mili-giây (ms) làm đơn vị chuẩn.
|
||||
|
||||
Args:
|
||||
file_path: Đường dẫn tệp âm thanh
|
||||
start_sec: Thời điểm bắt đầu (giây)
|
||||
end_sec: Thời điểm kết thúc (giây)
|
||||
loop_count: Số lần lặp
|
||||
fade_in_ms: Thời gian fade-in (ms)
|
||||
fade_out_ms: Thời gian fade-out (ms)
|
||||
volume_db_change: Thay đổi âm lượng (dB)
|
||||
|
||||
Returns:
|
||||
AudioSegment: Đoạn âm thanh đã xử lý
|
||||
"""
|
||||
# Tải tệp tin âm thanh gốc
|
||||
sound = AudioSegment.from_file(file_path)
|
||||
|
||||
# Chuyển đổi giây sang mili-giây
|
||||
start_ms = int(start_sec * 1000)
|
||||
end_ms = int(end_sec * 1000)
|
||||
|
||||
# Trích xuất phân đoạn (Slicing)
|
||||
clip = sound[start_ms:end_ms]
|
||||
|
||||
# Thay đổi âm lượng nếu có yêu cầu
|
||||
if volume_db_change != 0.0:
|
||||
clip = clip + volume_db_change
|
||||
|
||||
# Áp dụng hiệu ứng mờ đầu và mờ cuối (Fading)
|
||||
if fade_in_ms > 0:
|
||||
clip = clip.fade_in(fade_in_ms)
|
||||
if fade_out_ms > 0:
|
||||
clip = clip.fade_out(fade_out_ms)
|
||||
|
||||
# Tạo chuỗi lặp (Looping)
|
||||
looped_clip = clip * loop_count
|
||||
|
||||
return looped_clip
|
||||
|
||||
def mix_multitrack_session(tracks_meta: list, output_path: str, sample_rate: int = 44100, bit_depth: int = 16):
|
||||
"""
|
||||
Hòa âm đa kênh thống nhất (Multitrack Mixdown).
|
||||
|
||||
Args:
|
||||
tracks_meta: Danh sách cấu hình của từng track:
|
||||
[{"file_path": "...", "volume": 0.8, "muted": False}, ...]
|
||||
output_path: Đường dẫn file xuất
|
||||
sample_rate: Tần số lấy mẫu (Hz)
|
||||
bit_depth: Độ sâu bit (8, 16, hoặc 24)
|
||||
|
||||
Returns:
|
||||
dict: Thông tin kết quả
|
||||
"""
|
||||
master_mix = None
|
||||
|
||||
for track in tracks_meta:
|
||||
if track.get("muted", False):
|
||||
continue
|
||||
|
||||
# Đọc tệp âm thanh của track
|
||||
sound = AudioSegment.from_file(track["file_path"])
|
||||
|
||||
# Áp dụng Gain (chuyển đổi từ tỷ lệ 0.0 -> 1.0 sang decibels dB)
|
||||
gain_raw = track.get("volume", 1.0)
|
||||
if gain_raw <= 0.001:
|
||||
continue # Xem như tắt tiếng hoàn toàn
|
||||
gain_db = 20 * np.log10(gain_raw)
|
||||
sound = sound + gain_db
|
||||
|
||||
# Gộp vào Master Mix
|
||||
if master_mix is None:
|
||||
master_mix = sound
|
||||
else:
|
||||
# overlay tự động đồng bộ thời điểm bắt đầu tại mốc 0ms
|
||||
master_mix = master_mix.overlay(sound, position=0)
|
||||
|
||||
if master_mix is not None:
|
||||
# Đảm bảo thư mục output tồn tại
|
||||
out_dir = os.path.dirname(output_path)
|
||||
if out_dir:
|
||||
os.makedirs(out_dir, exist_ok=True)
|
||||
|
||||
# Xác định format từ đuôi file
|
||||
output_ext = os.path.splitext(output_path)[1].lower().lstrip(".")
|
||||
|
||||
if output_ext in ("mp3", "ogg"):
|
||||
# Xuất trực tiếp qua pydub/FFmpeg cho MP3, OGG
|
||||
master_mix.export(output_path, format=output_ext)
|
||||
else:
|
||||
# Cho WAV: sử dụng soundfile để kiểm soát bit-depth chính xác
|
||||
# Dùng tempfile an toàn thay vì đường dẫn tương đối
|
||||
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
|
||||
temp_wav = tmp.name
|
||||
|
||||
try:
|
||||
master_mix.export(temp_wav, format="wav")
|
||||
y, sr_read = sf.read(temp_wav)
|
||||
|
||||
# Xác định subtype mã hóa bit-depth
|
||||
subtype_map = {8: "PCM_S8", 16: "PCM_16", 24: "PCM_24"}
|
||||
selected_subtype = subtype_map.get(bit_depth, "PCM_16")
|
||||
|
||||
# Ghi tệp WAV chất lượng cao
|
||||
sf.write(output_path, y, sample_rate, subtype=selected_subtype)
|
||||
finally:
|
||||
# Dọn dẹp tệp tạm (luôn thực hiện)
|
||||
if os.path.exists(temp_wav):
|
||||
os.remove(temp_wav)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"output_path": output_path,
|
||||
"duration_ms": len(master_mix),
|
||||
"format": output_ext if output_ext else "wav",
|
||||
"tracks_processed": len([t for t in tracks_meta if not t.get("muted", False)])
|
||||
}
|
||||
else:
|
||||
return {
|
||||
"success": False,
|
||||
"error": "No active tracks to mix"
|
||||
}
|
||||
|
||||
|
||||
def export_audio(input_path: str, output_path: str, format: str = "wav",
|
||||
sample_rate: int = 44100, bit_depth: int = 16):
|
||||
"""
|
||||
Xuất tệp âm thanh sang nhiều định dạng (WAV/MP3/OGG).
|
||||
|
||||
Args:
|
||||
input_path: Đường dẫn file nguồn
|
||||
output_path: Đường dẫn file đích
|
||||
format: Định dạng xuất (wav, mp3, ogg)
|
||||
sample_rate: Tần số lấy mẫu
|
||||
bit_depth: Độ sâu bit (chỉ cho WAV)
|
||||
|
||||
Returns:
|
||||
dict: Thông tin kết quả
|
||||
"""
|
||||
# Guard: prevent overwriting source file
|
||||
if os.path.abspath(input_path) == os.path.abspath(output_path):
|
||||
raise ValueError("output_path must not be the same as input_path")
|
||||
|
||||
sound = AudioSegment.from_file(input_path)
|
||||
|
||||
out_dir = os.path.dirname(output_path)
|
||||
if out_dir:
|
||||
os.makedirs(out_dir, exist_ok=True)
|
||||
|
||||
if format in ("mp3", "ogg"):
|
||||
# Xuất qua pydub/FFmpeg
|
||||
sound.export(output_path, format=format)
|
||||
else:
|
||||
# WAV: dùng soundfile cho bit-depth chính xác
|
||||
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
|
||||
temp_wav = tmp.name
|
||||
|
||||
try:
|
||||
sound.export(temp_wav, format="wav")
|
||||
y, sr_read = sf.read(temp_wav)
|
||||
|
||||
subtype_map = {8: "PCM_S8", 16: "PCM_16", 24: "PCM_24"}
|
||||
selected_subtype = subtype_map.get(bit_depth, "PCM_16")
|
||||
|
||||
sf.write(output_path, y, sample_rate, subtype=selected_subtype)
|
||||
finally:
|
||||
if os.path.exists(temp_wav):
|
||||
os.remove(temp_wav)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"output_path": output_path,
|
||||
"format": format,
|
||||
"duration_ms": len(sound)
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
import numpy as np
|
||||
import librosa
|
||||
from pydub import AudioSegment
|
||||
|
||||
def find_zero_crossing(y: np.ndarray, sr: int, target_time: float, window_seconds: float = 0.04) -> float:
|
||||
"""
|
||||
Tìm điểm zero-crossing gần nhất với mốc thời gian đích (giây) để tránh click/pop.
|
||||
|
||||
Args:
|
||||
y: Mảng biên độ âm thanh (1D numpy array, Mono)
|
||||
sr: Tần số lấy mẫu (Sample Rate)
|
||||
target_time: Vị trí mong muốn cắt (giây)
|
||||
window_seconds: Cửa sổ quét (mặc định 40ms)
|
||||
|
||||
Returns:
|
||||
float: Thời gian của điểm zero-crossing gần nhất (giây)
|
||||
"""
|
||||
if target_time is None or target_time < 0:
|
||||
return target_time
|
||||
|
||||
target_sample = int(target_time * sr)
|
||||
window_samples = int(window_seconds * sr)
|
||||
|
||||
# Xác định giới hạn vùng quét an toàn
|
||||
start_idx = max(0, target_sample - window_samples)
|
||||
end_idx = min(len(y) - 2, target_sample + window_samples)
|
||||
|
||||
if start_idx >= end_idx:
|
||||
return target_time
|
||||
|
||||
# Lấy phân khúc sóng âm trong cửa sổ quét
|
||||
y_window = y[start_idx:end_idx]
|
||||
|
||||
# Tìm các điểm đổi dấu: y[i] * y[i+1] <= 0
|
||||
# Sử dụng np.sign và np.diff để tìm điểm đổi dấu nhanh chóng
|
||||
signs = np.sign(y_window)
|
||||
# Bất kỳ vị trí nào diff != 0 nghĩa là có sự đổi dấu (đi qua điểm 0)
|
||||
zero_crossings = np.where(np.diff(signs) != 0)[0]
|
||||
|
||||
if len(zero_crossings) == 0:
|
||||
return target_time # Không tìm thấy, trả về vị trí gốc
|
||||
|
||||
# Chuyển chỉ số vùng quét về chỉ số mảng tuyệt đối
|
||||
absolute_crossings = zero_crossings + start_idx
|
||||
|
||||
# Tìm điểm gần với target_sample nhất
|
||||
distances = np.abs(absolute_crossings - target_sample)
|
||||
closest_sample_idx = absolute_crossings[np.argmin(distances)]
|
||||
|
||||
# Trả về thời gian tương ứng (giây)
|
||||
return float(closest_sample_idx / sr)
|
||||
|
||||
def find_nearest_zero_crossing_file(file_path: str, target_time_sec: float, search_window_sec: float = 0.04) -> float:
|
||||
"""
|
||||
Tìm điểm zero-crossing từ file âm thanh.
|
||||
Wrapper cho hàm find_zero_crossing để tương thích với code cũ.
|
||||
"""
|
||||
try:
|
||||
# Load mono audio for zero crossing analysis
|
||||
y, sr = librosa.load(file_path, sr=None, mono=True)
|
||||
return find_zero_crossing(y, sr, target_time_sec, search_window_sec)
|
||||
except Exception as e:
|
||||
print(f"Error finding zero crossing: {e}")
|
||||
return target_time_sec
|
||||
|
||||
def find_nearest_zero_crossing(y: np.ndarray, sr: int, target_time_sec: float, search_window_sec: float = 0.04) -> float:
|
||||
"""
|
||||
Tương thích với code cũ. Gọi đến hàm find_zero_crossing mới.
|
||||
"""
|
||||
return find_zero_crossing(y, sr, target_time_sec, search_window_sec)
|
||||
|
||||
def apply_micro_fade(segment: AudioSegment, fade_duration_ms: int = 50) -> AudioSegment:
|
||||
"""
|
||||
Áp dụng micro-fades (fade-in và fade-out) để triệt tiêu click/pop.
|
||||
"""
|
||||
if len(segment) > fade_duration_ms * 2:
|
||||
return segment.fade_in(fade_duration_ms).fade_out(fade_duration_ms)
|
||||
elif len(segment) > fade_duration_ms:
|
||||
return segment.fade_in(fade_duration_ms // 2).fade_out(fade_duration_ms // 2)
|
||||
return segment
|
||||
|
||||
|
||||
def generate_peak_waveform(file_path: str, num_peaks: int = 800) -> dict:
|
||||
"""
|
||||
Tạo dữ liệu peak waveform cho hiển thị đồ thị sóng âm trên Frontend.
|
||||
Dùng để vẽ waveform đồng bộ với Client (thay thế Web Audio API decodeAudioData).
|
||||
|
||||
Args:
|
||||
file_path: Đường dẫn tệp âm thanh
|
||||
num_peaks: Số lượng điểm peak trả về (tương ứng pixel width trên UI)
|
||||
|
||||
Returns:
|
||||
dict: {"peaks": [...], "duration": float, "sample_rate": int}
|
||||
"""
|
||||
# Load mono audio
|
||||
y, sr = librosa.load(file_path, sr=None, mono=True)
|
||||
|
||||
total_samples = len(y)
|
||||
duration = float(total_samples) / sr
|
||||
|
||||
if total_samples == 0:
|
||||
return {
|
||||
"peaks": [],
|
||||
"duration": 0.0,
|
||||
"sample_rate": sr
|
||||
}
|
||||
|
||||
# Tính kích thước mỗi chunk
|
||||
samples_per_peak = max(1, total_samples // num_peaks)
|
||||
|
||||
peaks = []
|
||||
for i in range(0, total_samples, samples_per_peak):
|
||||
chunk = y[i:i + samples_per_peak]
|
||||
if len(chunk) > 0:
|
||||
# Peak = giá trị tuyệt đối lớn nhất trong chunk
|
||||
peak_val = float(np.max(np.abs(chunk)))
|
||||
peaks.append(round(peak_val, 6))
|
||||
|
||||
# Giới hạn đúng số lượng peaks yêu cầu
|
||||
if len(peaks) > num_peaks:
|
||||
peaks = peaks[:num_peaks]
|
||||
|
||||
return {
|
||||
"peaks": peaks,
|
||||
"duration": round(duration, 4),
|
||||
"sample_rate": sr
|
||||
}
|
||||
|
||||
|
||||
def generate_rms_waveform(file_path: str, num_points: int = 800) -> dict:
|
||||
"""
|
||||
Tạo dữ liệu RMS waveform (mượt hơn peak waveform).
|
||||
|
||||
Args:
|
||||
file_path: Đường dẫn tệp âm thanh
|
||||
num_points: Số lượng điểm RMS trả về
|
||||
|
||||
Returns:
|
||||
dict: {"rms": [...], "duration": float, "sample_rate": int}
|
||||
"""
|
||||
y, sr = librosa.load(file_path, sr=None, mono=True)
|
||||
|
||||
total_samples = len(y)
|
||||
duration = float(total_samples) / sr
|
||||
|
||||
if total_samples == 0:
|
||||
return {
|
||||
"rms": [],
|
||||
"duration": 0.0,
|
||||
"sample_rate": sr
|
||||
}
|
||||
|
||||
samples_per_point = max(1, total_samples // num_points)
|
||||
|
||||
rms_values = []
|
||||
for i in range(0, total_samples, samples_per_point):
|
||||
chunk = y[i:i + samples_per_point]
|
||||
if len(chunk) > 0:
|
||||
rms_val = float(np.sqrt(np.mean(chunk ** 2)))
|
||||
rms_values.append(round(rms_val, 6))
|
||||
|
||||
if len(rms_values) > num_points:
|
||||
rms_values = rms_values[:num_points]
|
||||
|
||||
return {
|
||||
"rms": rms_values,
|
||||
"duration": round(duration, 4),
|
||||
"sample_rate": sr
|
||||
}
|
||||
Reference in New Issue
Block a user