Mix Analyzer Module

"""
<Module Name> Analyzer - Brief description of what it analyzes
"""
import logging
from typing import TYPE_CHECKING, Any, Dict, List, Optional

import numpy as np

from app.analysis.base import AnalysisResult, BaseAnalyzer

if TYPE_CHECKING:
    from app.analysis.dsp_context import DSPContext

logger = logging.getLogger(__name__)


class <ModuleName>Analyzer(BaseAnalyzer):
    """Analyzer for <description>"""

    name: str = "<module_name>"
    description: str = "<Brief description>"
    version: str = "1.0.0"

    async def analyze(
        self,
        y: np.ndarray,
        sr: int,
        y_full: np.ndarray = None,
        dsp_ctx: "DSPContext" = None,
    ) -> AnalysisResult:
        """
        Analyze audio and return results.

        Args:
            y: Mono audio signal as numpy array
            sr: Sample rate (typically 44100 after downsampling)
            y_full: Full stereo audio if available
            dsp_ctx: Precomputed DSP context with STFT, HPSS, beats, etc.

        Returns:
            AnalysisResult with score, data, issues, recommendations
        """
        issues: List[Dict[str, Any]] = []
        recommendations: List[str] = []
        data: Dict[str, Any] = {}

        # Use DSPContext for expensive computations
        if dsp_ctx is not None:
            stft = dsp_ctx.stft  # Precomputed STFT
            harmonic, percussive = dsp_ctx.hpss  # Precomputed HPSS separation

        # Perform analysis
        # ...

        # Detect issues with severity
        if some_condition:
            issues.append({
                "severity": "medium",  # "low", "medium", or "high"
                "description": "Description of the issue",
                "value": measured_value,
                "target": target_value,
            })
            recommendations.append("Suggestion to fix the issue")

        # Calculate normalized score (0.0 to 1.0)
        score = self._calculate_score_from_issues(issues)

        return AnalysisResult(
            score=score,
            data=data,
            issues=issues,
            recommendations=recommendations,
        )

Step 2: Register in Audio Processing Task

Edit backend/app/workers/tasks/audio_processing.py:

1. Import the analyzer:

from app.analysis.<module_name> import <ModuleName>Analyzer

2. Add to KNOWN_MODULES frozenset:

KNOWN_MODULES: frozenset = frozenset([
    # ... existing modules ...
    "<module_name>",
])

3. Add to _get_analyzers() dict:

_ANALYZERS = {
    # ... existing analyzers ...
    "<module_name>": <ModuleName>Analyzer(),
}

4. Add to appropriate layer:

# For independent modules (most cases):
LAYER_1 = [
    # ... existing modules ...
    "<module_name>",
]

# For modules that need results from Layer 1:
LAYER_2 = ["keywords", "scoring", "reference", "<module_name>"]

# For modules that need all results (rare):
LAYER_3 = ["ai_recommendations", "<module_name>"]

Step 3: Add Database Column

Edit backend/app/models/analysis.py:

1. Add JSONB column:

<module_name>_data: Mapped[Optional[Dict[str, Any]]] = mapped_column(JSONB)

2. Add to _get_results() modules list:

modules = [
    # ... existing modules ...
    "<module_name>",
]

Step 4: Create Migration

cd backend
alembic revision --autogenerate -m "Add <module_name>_data column"
alembic upgrade head

Using DSPContext

The DSPContext (in analysis/dsp_context.py) precomputes expensive operations. Access via dsp_ctx parameter:

# Available precomputed features:
dsp_ctx.stft          # Short-time Fourier Transform
dsp_ctx.hpss          # (harmonic, percussive) separation
dsp_ctx.tempo         # Estimated BPM
dsp_ctx.beats         # Beat frame indices
dsp_ctx.beat_times    # Beat times in seconds
dsp_ctx.mfcc          # Mel-frequency cepstral coefficients
dsp_ctx.chroma        # Chromagram
dsp_ctx.spectral_centroid  # Spectral centroid
dsp_ctx.f0            # Fundamental frequency (pitch)

Special Module Patterns

Modules Needing Other Results

If the module needs results from other modules, receive analysis_results:

async def analyze(
    self,
    y: np.ndarray = None,
    sr: int = None,
    y_full: np.ndarray = None,
    dsp_ctx: "DSPContext" = None,
    analysis_results: Dict[str, Any] = None,  # Results from previous modules
) -> AnalysisResult:
    # Access other module results
    genre = analysis_results.get("genre", {}).get("data", {}).get("primary")
    frequency_score = analysis_results.get("frequency", {}).get("score")

Register in Layer 2 or 3, and update _run_module() in audio_processing.py.

Modules Needing Genre

The reference module pattern for genre dependency:

async def analyze(
    self,
    y: np.ndarray,
    sr: int,
    y_full: np.ndarray = None,
    dsp_ctx: "DSPContext" = None,
    genre: str = None,
    analysis_results: Dict[str, Any] = None,
) -> AnalysisResult:
    # Use genre for reference comparison

Error Handling

Individual module errors should not crash the entire analysis:

async def analyze(self, y, sr, y_full=None, dsp_ctx=None) -> AnalysisResult:
    try:
        # Analysis logic
        pass
    except Exception as e:
        logger.error("Error in %s: %s", self.name, e)
        # Return partial results or empty result
        return AnalysisResult(
            score=None,
            data={"error": str(e)},
            issues=[],
            recommendations=[],
        )

Score Calculation Patterns

Normalization Helper

score = self._normalize_score(measured_value, min_val=0, max_val=1)

Issue-Based Scoring

severity_weights = {"low": 0.05, "medium": 0.1, "high": 0.2}
score = self._calculate_score_from_issues(issues, base_score=1.0)

Custom Weighted Scoring

def _calculate_weighted_score(self, metrics: Dict[str, float]) -> float:
    weights = {
        "metric_a": 0.4,
        "metric_b": 0.3,
        "metric_c": 0.3,
    }
    total = sum(metrics.get(k, 0) * v for k, v in weights.items())
    return max(0.0, min(1.0, total))