File size: 4,983 Bytes
26c0314
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
"""
Creator Studio Multitrack Mixer — real DSP, not a mock.

Takes 2-6 uploaded tracks, applies per-track gain (dB) and stereo pan,
aligns them to a common length (shorter tracks are zero-padded — silence,
not looped, since these are independent parts of one arrangement rather
than a loopable texture like remix_engine's crossfade blend), sums them
into a single stereo mix, and peak-normalizes to prevent clipping.

Equal-power panning (constant-power law) is used so a track panned to
either side doesn't lose perceived loudness relative to center.
"""

import io
import logging

import librosa
import numpy as np
import soundfile as sf

from app.schemas import MultitrackChannelSettings, MultitrackOptions

logger = logging.getLogger(__name__)

MAX_CHANNELS = 6
MIN_CHANNELS = 2


class MultitrackResult:
    """Plain result container, mirroring RemixResult's pattern."""

    def __init__(self, audio: io.BytesIO, analysis: dict) -> None:
        self.audio = audio
        self.analysis = analysis


def _db_to_linear(db: float) -> float:
    return float(10.0 ** (db / 20.0))


def _equal_power_pan(mono: np.ndarray, pan: float) -> np.ndarray:
    """Pan a mono signal to stereo using the constant-power law.
    pan=-1 -> full left, 0 -> center (both channels at -3dB), 1 -> full right."""
    angle = (pan + 1.0) * (np.pi / 4.0)  # maps [-1, 1] -> [0, pi/2]
    left_gain = float(np.cos(angle))
    right_gain = float(np.sin(angle))
    stereo = np.stack([mono * left_gain, mono * right_gain], axis=1)
    return stereo


def mix_tracks(
    track_bytes_list: list[bytes],
    options: MultitrackOptions,
) -> MultitrackResult:
    if len(track_bytes_list) < MIN_CHANNELS:
        raise ValueError(f"At least {MIN_CHANNELS} tracks are required for a multitrack mix")
    if len(track_bytes_list) > MAX_CHANNELS:
        raise ValueError(f"At most {MAX_CHANNELS} tracks are supported per mix")

    channels = options.channels
    if len(channels) != len(track_bytes_list):
        # Missing settings default to unity gain, centered, unmuted —
        # a caller can send fewer channel configs than files and get sane
        # defaults for the rest rather than a validation error.
        channels = list(channels) + [
            MultitrackChannelSettings() for _ in range(len(track_bytes_list) - len(channels))
        ]

    sr = 22050
    loaded: list[np.ndarray] = []
    for i, raw in enumerate(track_bytes_list):
        try:
            y, _ = librosa.load(io.BytesIO(raw), sr=sr, mono=True)
        except Exception as e:
            logger.error(f"Failed to load track {i} for multitrack mix: {e}", exc_info=True)
            raise ValueError(f"Could not read audio for track {i + 1}: {e}")
        if y.size == 0:
            raise ValueError(f"Track {i + 1} is empty")
        loaded.append(y)

    max_len = max(len(y) for y in loaded)
    channel_reports = []
    mix = np.zeros((max_len, 2), dtype=np.float64)

    for i, (y, settings) in enumerate(zip(loaded, channels)):
        duration_sec = round(float(len(y) / sr), 2)

        if settings.muted:
            channel_reports.append({
                "index": i,
                "duration_sec": duration_sec,
                "applied_gain_db": settings.gain_db,
                "pan": settings.pan,
                "muted": True,
                "peak_level": 0.0,
            })
            continue

        # Zero-pad to common length — independent arrangement parts should
        # not loop into each other the way a two-track crossfade blend would.
        padded = np.pad(y, (0, max_len - len(y)))
        gained = padded * _db_to_linear(settings.gain_db)
        stereo = _equal_power_pan(gained, settings.pan)
        mix += stereo

        peak = float(np.max(np.abs(gained))) if gained.size else 0.0
        channel_reports.append({
            "index": i,
            "duration_sec": duration_sec,
            "applied_gain_db": settings.gain_db,
            "pan": settings.pan,
            "muted": False,
            "peak_level": round(peak, 4),
        })

    output_peak = float(np.max(np.abs(mix))) if mix.size else 0.0
    clipping_prevented = False
    if options.normalize_output and output_peak > 1.0:
        mix = mix / output_peak * 0.98
        clipping_prevented = True
        output_peak = 0.98
    elif output_peak > 1.0:
        # Caller explicitly disabled normalization — hard-clip rather than
        # let soundfile silently wrap/distort on write.
        mix = np.clip(mix, -1.0, 1.0)
        clipping_prevented = True
        output_peak = 1.0

    out_buffer = io.BytesIO()
    sf.write(out_buffer, mix.astype(np.float32), sr, format="WAV")
    out_buffer.seek(0)

    analysis = {
        "channels": channel_reports,
        "output_duration_sec": round(float(max_len / sr), 2),
        "output_peak_level": round(output_peak, 4),
        "clipping_prevented": clipping_prevented,
    }
    return MultitrackResult(audio=out_buffer, analysis=analysis)