crowncode-backend / app /services /dataset_organizer.py
Rthur2003's picture
feat: remix engine routes dream analysis servisleri ve dataset organization utilities eklendi
e49c006
Raw History Blame Contribute Delete
5.39 kB
"""
Dataset organization service — analyzes an audio file and derives real
acoustic metadata (duration, tempo, key, loudness, energy tier) used to
auto-tag and categorize entries in a dataset, per the "Veri Seti
Organizasyonu" tool advertised on the data-manipulation page.
Deliberately independent from feature_extractor.py (AURIS's 49-feature
AI-detection pipeline) — that module is tuned for AI-vs-human classification
and pulls in a much heavier feature set than dataset tagging needs. Coupling
this to it would make a simple tagging tool fragile to AI-detection changes.
"""
import io
import logging
import librosa
import numpy as np
logger = logging.getLogger(__name__)
PITCH_CLASSES = [
"C", "C#", "D", "D#", "E", "F",
"F#", "G", "G#", "A", "A#", "B",
]
# Krumhansl-Schmuckler key profiles — standard weights for major/minor
# key detection via correlation against the chroma vector.
_MAJOR_PROFILE = np.array(
[6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.39, 3.66, 2.29, 2.88]
)
_MINOR_PROFILE = np.array(
[6.33, 2.68, 3.52, 5.38, 2.60, 3.53, 2.54, 4.75, 3.98, 2.69, 3.34, 3.17]
)
def detect_key(chroma_mean: np.ndarray) -> str:
"""Correlate the mean chroma vector against all 24 rotated key profiles."""
best_score = -np.inf
best_key = "C major"
for shift in range(12):
major_rot = np.roll(_MAJOR_PROFILE, shift)
minor_rot = np.roll(_MINOR_PROFILE, shift)
major_score = float(np.corrcoef(chroma_mean, major_rot)[0, 1])
minor_score = float(np.corrcoef(chroma_mean, minor_rot)[0, 1])
if major_score > best_score:
best_score = major_score
best_key = f"{PITCH_CLASSES[shift]} major"
if minor_score > best_score:
best_score = minor_score
best_key = f"{PITCH_CLASSES[shift]} minor"
return best_key
def _tempo_tag(bpm: float) -> str:
# librosa.beat.beat_track returns ~0 when it can't lock onto a beat at
# all (e.g. a sustained tone or ambient texture with no rhythm) — that's
# "no beat detected", not literally "slow", so tag it separately rather
# than lump it into the slow bucket.
if bpm < 1:
return "beatless"
if bpm < 76:
return "slow"
if bpm < 120:
return "moderate"
if bpm < 150:
return "upbeat"
return "fast"
def _energy_tag(rms_mean: float) -> str:
# RMS on a float32 waveform normalized to [-1, 1]; these thresholds are
# calibrated against typical mixed/mastered music, not raw voice memos.
if rms_mean < 0.03:
return "ambient"
if rms_mean < 0.08:
return "calm"
if rms_mean < 0.15:
return "energetic"
return "intense"
def _loudness_lufs_approx(y: np.ndarray) -> float:
"""
Rough integrated-loudness estimate in LUFS-like dB, using RMS as a
stand-in for full ITU-R BS.1770 K-weighting (that needs a dedicated
filter chain this tool doesn't need for a dataset-tagging heuristic).
"""
rms = float(np.sqrt(np.mean(np.square(y)))) if len(y) else 0.0
if rms <= 0:
return -70.0
return round(float(20 * np.log10(rms)), 1)
class DatasetEntryMetadata:
"""Plain result container — kept dependency-free so routes can shape
the JSON response without importing pydantic into this module."""
def __init__(
self,
duration_sec: float,
tempo_bpm: float,
key: str,
loudness_db: float,
tags: list[str],
) -> None:
self.duration_sec = duration_sec
self.tempo_bpm = tempo_bpm
self.key = key
self.loudness_db = loudness_db
self.tags = tags
def to_dict(self) -> dict:
return {
"durationSec": self.duration_sec,
"tempoBpm": self.tempo_bpm,
"key": self.key,
"loudnessDb": self.loudness_db,
"tags": self.tags,
}
def analyze_for_organization(file_bytes: bytes) -> DatasetEntryMetadata:
"""
Extract real acoustic metadata from an audio file for dataset tagging:
duration, tempo (BPM), musical key, approximate loudness, and a set of
auto-generated descriptive tags (tempo feel, energy level, duration
bucket) suitable for filtering/categorizing a growing dataset.
"""
try:
y, sr = librosa.load(io.BytesIO(file_bytes), sr=22050, mono=True)
except Exception as e:
logger.error(f"Failed to load audio for organization: {e}", exc_info=True)
raise ValueError(f"Could not read audio file: {e}")
if y.size == 0 or float(np.max(np.abs(y))) < 1e-6:
raise ValueError("Audio file is empty or silent")
duration_sec = round(float(librosa.get_duration(y=y, sr=sr)), 2)
tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
tempo_bpm = round(float(np.atleast_1d(tempo)[0]), 1)
chroma = librosa.feature.chroma_cqt(y=y, sr=sr)
key = detect_key(np.mean(chroma, axis=1))
loudness_db = _loudness_lufs_approx(y)
rms_mean = float(np.mean(librosa.feature.rms(y=y)[0]))
tags = [_tempo_tag(tempo_bpm), _energy_tag(rms_mean)]
if duration_sec < 30:
tags.append("short-clip")
elif duration_sec > 240:
tags.append("long-form")
return DatasetEntryMetadata(
duration_sec=duration_sec,
tempo_bpm=tempo_bpm,
key=key,
loudness_db=loudness_db,
tags=tags,
)