Skip to content

Instantly share code, notes, and snippets.

@bogged-broker

bogged-broker/audio_platforms_profiles.py

Created December 30, 2025 20:56
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save bogged-broker/95f9df55545da8986c4037d992db589a to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save bogged-broker/95f9df55545da8986c4037d992db589a to your computer and use it in GitHub Desktop.
Download ZIP
Raw
audio_platforms_profiles.py
"""
Platform Audio Profiles Module
Defines platform-specific audio constraints and optimization profiles for 5M+ view guarantee.
Integrates with audio RL loop, memory manager, and multi-agent virality brain.
"""
import json
from typing import Dict, List, Optional, Tuple, Any
from dataclasses import dataclass, asdict
from enum import Enum
import numpy as np
class Platform(Enum):
"""Supported video platforms"""
TIKTOK = "tiktok"
INSTAGRAM_REELS = "instagram_reels"
YOUTUBE_SHORTS = "youtube_shorts"
SNAPCHAT = "snapchat"
FACEBOOK_REELS = "facebook_reels"
class PlaybackDevice(Enum):
"""Target playback devices"""
PHONE_SPEAKER = "phone_speaker"
EARBUDS = "earbuds"
HEADPHONES = "headphones"
LAPTOP_SPEAKERS = "laptop_speakers"
MONO_FALLBACK = "mono_fallback"
class ViralityTier(Enum):
"""Audio pattern performance tiers"""
HOT = "hot" # Proven viral, use immediately
WARM = "warm" # Moderately successful, occasional reuse
COLD = "cold" # Unproven, experimental, RL exploration
@dataclass
class FrequencyBandProfile:
"""Frequency band optimization per platform"""
low_cut_hz: float # High-pass filter cutoff
low_boost_hz: Tuple[float, float] # Bass boost range
mid_presence_hz: Tuple[float, float] # Vocal clarity range
high_shelf_hz: float # Brightness/air frequency
problem_frequencies: List[Tuple[float, float]] # Ranges to attenuate
# Mobile speaker optimization
phone_safe_low_hz: float # Phones can't reproduce below this
earbuds_sweet_spot_hz: Tuple[float, float] # Optimal earbuds range
@dataclass
class CompressionProfile:
"""Dynamic range compression settings"""
threshold_db: float
ratio: float
attack_ms: float
release_ms: float
knee_db: float
makeup_gain_db: float
sidechain_filter_hz: Optional[float] # For bass preservation
@dataclass
class LoudnessProfile:
"""Platform-specific loudness targets"""
target_lufs: float # Integrated loudness
true_peak_dbtp: float # Maximum true peak
short_term_lufs_range: Tuple[float, float] # Dynamic range bounds
momentary_lufs_max: float # Peak excitement moments
# Streaming normalization targets
platform_normalization_lufs: float # What platform normalizes to
headroom_db: float # Safety margin
@dataclass
class BeatMatchingProfile:
"""Trend-aware beat and rhythm recommendations"""
optimal_bpm_range: Tuple[int, int]
viral_bpm_peaks: List[int] # Most viral BPMs for platform
beat_drop_timing_sec: List[float] # When to place drops
loop_point_requirements: List[float] # For seamless looping
rhythm_patterns: List[str] # e.g., "4-on-floor", "trap-hi-hats"
sync_to_trending_audio: bool # Match current viral audio tempo
@dataclass
class SpatialProfile:
"""Stereo/mono playback rules"""
primary_format: str # "stereo", "mono", "dual_mono"
mono_compatibility_required: bool
stereo_width_percent: float # 0-200%, 100% = full stereo
center_content_db: float # How much to boost center channel
side_content_limit_db: float # Prevent side info loss in mono
# Device-specific handling
phone_speaker_mode: str # "mono_sum", "left_only", "optimized"
earbuds_enhancement: bool # Apply spatial enhancement for earbuds
@dataclass
class ViralAudioPattern:
"""Proven viral audio characteristics"""
pattern_id: str
hook_timing_sec: List[float] # When hooks occur
emotional_peak_timing_sec: List[float] # Emotional crescendos
energy_curve: List[float] # Energy level per second (0-1)
loopability_score: float # 0-1, how well it loops
shareability_score: float # 0-1, predicted share rate
engagement_multiplier: float # Historical engagement boost
# Pattern characteristics
has_earworm_hook: bool
uses_trending_sound: bool
incorporates_silence: bool # Strategic silence for impact
viral_tier: ViralityTier
# Performance tracking
avg_views: float
avg_watch_through_rate: float
avg_replay_rate: float
platforms_successful: List[Platform]
@dataclass
class PlatformAudioProfile:
"""Complete audio profile for a platform"""
platform: Platform
# Core audio specs
loudness: LoudnessProfile
compression: CompressionProfile
frequency_bands: FrequencyBandProfile
spatial: SpatialProfile
beat_matching: BeatMatchingProfile
# Platform constraints
max_duration_sec: float
min_duration_sec: float
recommended_duration_sec: float
sample_rate_hz: int
bit_depth: int
codec: str
max_bitrate_kbps: int
# Virality optimization
viral_patterns: List[ViralAudioPattern]
trending_audio_hooks: List[Dict[str, Any]]
niche_customizations: Dict[str, Any]
# Device assumptions
primary_playback_devices: List[PlaybackDevice]
device_distribution: Dict[PlaybackDevice, float] # Percentage
# Engagement predictions
predicted_ctr: float
predicted_watch_through: float
predicted_loop_rate: float
predicted_share_rate: float
# Platform-specific quirks
algorithm_preferences: Dict[str, Any]
compression_artifacts_tolerance: float
requires_sound_on_indicator: bool
class PlatformAudioProfileManager:
"""Manages platform-specific audio profiles and optimizations"""
def __init__(self):
self.profiles: Dict[Platform, PlatformAudioProfile] = {}
self._initialize_profiles()
self.memory_manager = None # Injected from audio_memory_manager.py
def _initialize_profiles(self):
"""Initialize all platform profiles with 5M+ view optimizations"""
# TikTok Profile - Optimized for viral loops and phone speakers
self.profiles[Platform.TIKTOK] = PlatformAudioProfile(
platform=Platform.TIKTOK,
loudness=LoudnessProfile(
target_lufs=-14.0, # TikTok's sweet spot
true_peak_dbtp=-1.0,
short_term_lufs_range=(-16.0, -10.0),
momentary_lufs_max=-8.0, # Punch for hooks
platform_normalization_lufs=-14.0,
headroom_db=1.0
),
compression=CompressionProfile(
threshold_db=-18.0,
ratio=4.0, # Aggressive for phone speakers
attack_ms=5.0, # Fast for transient preservation
release_ms=50.0,
knee_db=6.0, # Smooth compression
makeup_gain_db=3.0,
sidechain_filter_hz=80.0 # Preserve bass punch
),
frequency_bands=FrequencyBandProfile(
low_cut_hz=30.0, # Remove sub-bass phone can't reproduce
low_boost_hz=(80.0, 150.0), # Punch range for phone speakers
mid_presence_hz=(2000.0, 4000.0), # Vocal intelligibility
high_shelf_hz=8000.0, # Air and clarity
problem_frequencies=[(250.0, 400.0), (3000.0, 3500.0)], # Muddy/harsh
phone_safe_low_hz=50.0,
earbuds_sweet_spot_hz=(200.0, 8000.0)
),
spatial=SpatialProfile(
primary_format="stereo",
mono_compatibility_required=True, # Critical for phone speakers
stereo_width_percent=120.0, # Wider for earbuds
center_content_db=3.0, # Boost vocals/hooks
side_content_limit_db=-6.0,
phone_speaker_mode="optimized",
earbuds_enhancement=True
),
beat_matching=BeatMatchingProfile(
optimal_bpm_range=(120, 150), # TikTok viral sweet spot
viral_bpm_peaks=[128, 140, 145], # Trending tempos
beat_drop_timing_sec=[0.5, 3.0, 7.0], # Hook placement
loop_point_requirements=[15.0, 30.0, 60.0],
rhythm_patterns=["trap-hi-hats", "edm-buildup", "drill-pattern"],
sync_to_trending_audio=True
),
max_duration_sec=180.0,
min_duration_sec=3.0,
recommended_duration_sec=15.0, # 15-sec viral baseline
sample_rate_hz=44100,
bit_depth=16,
codec="AAC",
max_bitrate_kbps=128,
viral_patterns=[
ViralAudioPattern(
pattern_id="tiktok_hook_3sec",
hook_timing_sec=[0.5, 3.0],
emotional_peak_timing_sec=[7.0],
energy_curve=[0.8, 0.9, 1.0, 0.9, 0.8, 0.9, 1.0, 0.85] * 2,
loopability_score=0.95,
shareability_score=0.88,
engagement_multiplier=2.4,
has_earworm_hook=True,
uses_trending_sound=True,
incorporates_silence=False,
viral_tier=ViralityTier.HOT,
avg_views=8500000.0,
avg_watch_through_rate=0.78,
avg_replay_rate=0.42,
platforms_successful=[Platform.TIKTOK]
)
],
trending_audio_hooks=[],
niche_customizations={},
primary_playback_devices=[
PlaybackDevice.PHONE_SPEAKER,
PlaybackDevice.EARBUDS
],
device_distribution={
PlaybackDevice.PHONE_SPEAKER: 0.60,
PlaybackDevice.EARBUDS: 0.35,
PlaybackDevice.HEADPHONES: 0.05
},
predicted_ctr=0.18,
predicted_watch_through=0.75,
predicted_loop_rate=0.38,
predicted_share_rate=0.12,
algorithm_preferences={
"favors_looping": True,
"penalizes_silence_start": True,
"rewards_immediate_hook": True,
"prefers_trending_audio": True
},
compression_artifacts_tolerance=0.7,
requires_sound_on_indicator=False
)
# Instagram Reels Profile - Polished, broadcast-quality optimization
self.profiles[Platform.INSTAGRAM_REELS] = PlatformAudioProfile(
platform=Platform.INSTAGRAM_REELS,
loudness=LoudnessProfile(
target_lufs=-13.0, # Slightly louder than TikTok
true_peak_dbtp=-1.0,
short_term_lufs_range=(-15.0, -9.0),
momentary_lufs_max=-7.0,
platform_normalization_lufs=-14.0,
headroom_db=1.0
),
compression=CompressionProfile(
threshold_db=-16.0,
ratio=3.5, # Less aggressive, more dynamic
attack_ms=10.0,
release_ms=80.0,
knee_db=8.0,
makeup_gain_db=2.5,
sidechain_filter_hz=100.0
),
frequency_bands=FrequencyBandProfile(
low_cut_hz=35.0,
low_boost_hz=(60.0, 120.0),
mid_presence_hz=(1800.0, 5000.0),
high_shelf_hz=10000.0, # More air for polished sound
problem_frequencies=[(200.0, 350.0)],
phone_safe_low_hz=45.0,
earbuds_sweet_spot_hz=(150.0, 10000.0)
),
spatial=SpatialProfile(
primary_format="stereo",
mono_compatibility_required=True,
stereo_width_percent=140.0, # Wider stereo for polish
center_content_db=2.0,
side_content_limit_db=-4.0,
phone_speaker_mode="optimized",
earbuds_enhancement=True
),
beat_matching=BeatMatchingProfile(
optimal_bpm_range=(110, 140),
viral_bpm_peaks=[120, 128, 135],
beat_drop_timing_sec=[1.0, 5.0, 10.0],
loop_point_requirements=[15.0, 30.0, 60.0, 90.0],
rhythm_patterns=["pop-structure", "edm-buildup", "hip-hop-bounce"],
sync_to_trending_audio=True
),
max_duration_sec=90.0,
min_duration_sec=3.0,
recommended_duration_sec=20.0,
sample_rate_hz=48000,
bit_depth=16,
codec="AAC",
max_bitrate_kbps=192,
viral_patterns=[
ViralAudioPattern(
pattern_id="reels_aesthetic_hook",
hook_timing_sec=[1.0, 5.0],
emotional_peak_timing_sec=[10.0, 18.0],
energy_curve=[0.7] * 5 + [0.9] * 5 + [1.0] * 5 + [0.8] * 5,
loopability_score=0.85,
shareability_score=0.82,
engagement_multiplier=2.1,
has_earworm_hook=True,
uses_trending_sound=True,
incorporates_silence=True,
viral_tier=ViralityTier.HOT,
avg_views=6200000.0,
avg_watch_through_rate=0.72,
avg_replay_rate=0.35,
platforms_successful=[Platform.INSTAGRAM_REELS]
)
],
trending_audio_hooks=[],
niche_customizations={},
primary_playback_devices=[
PlaybackDevice.PHONE_SPEAKER,
PlaybackDevice.EARBUDS
],
device_distribution={
PlaybackDevice.PHONE_SPEAKER: 0.55,
PlaybackDevice.EARBUDS: 0.40,
PlaybackDevice.HEADPHONES: 0.05
},
predicted_ctr=0.16,
predicted_watch_through=0.70,
predicted_loop_rate=0.32,
predicted_share_rate=0.15,
algorithm_preferences={
"favors_looping": True,
"penalizes_silence_start": False,
"rewards_immediate_hook": True,
"prefers_trending_audio": True,
"aesthetic_audio_bonus": True
},
compression_artifacts_tolerance=0.6,
requires_sound_on_indicator=False
)
# YouTube Shorts Profile - Watch-through optimized
self.profiles[Platform.YOUTUBE_SHORTS] = PlatformAudioProfile(
platform=Platform.YOUTUBE_SHORTS,
loudness=LoudnessProfile(
target_lufs=-14.0,
true_peak_dbtp=-1.0,
short_term_lufs_range=(-16.0, -11.0),
momentary_lufs_max=-9.0,
platform_normalization_lufs=-14.0,
headroom_db=1.0
),
compression=CompressionProfile(
threshold_db=-20.0,
ratio=3.0, # Gentle, broadcast-style
attack_ms=15.0,
release_ms=100.0,
knee_db=10.0,
makeup_gain_db=2.0,
sidechain_filter_hz=120.0
),
frequency_bands=FrequencyBandProfile(
low_cut_hz=40.0,
low_boost_hz=(70.0, 130.0),
mid_presence_hz=(2000.0, 5000.0),
high_shelf_hz=9000.0,
problem_frequencies=[(300.0, 450.0)],
phone_safe_low_hz=50.0,
earbuds_sweet_spot_hz=(180.0, 12000.0)
),
spatial=SpatialProfile(
primary_format="stereo",
mono_compatibility_required=True,
stereo_width_percent=130.0,
center_content_db=2.5,
side_content_limit_db=-5.0,
phone_speaker_mode="optimized",
earbuds_enhancement=True
),
beat_matching=BeatMatchingProfile(
optimal_bpm_range=(100, 145),
viral_bpm_peaks=[115, 128, 140],
beat_drop_timing_sec=[2.0, 10.0, 20.0],
loop_point_requirements=[30.0, 60.0],
rhythm_patterns=["storytelling-pace", "buildup-payoff", "sustained-energy"],
sync_to_trending_audio=False # Less trend-dependent
),
max_duration_sec=60.0,
min_duration_sec=15.0,
recommended_duration_sec=30.0,
sample_rate_hz=48000,
bit_depth=16,
codec="AAC",
max_bitrate_kbps=256,
viral_patterns=[
ViralAudioPattern(
pattern_id="shorts_watchthrough_30sec",
hook_timing_sec=[2.0, 15.0],
emotional_peak_timing_sec=[25.0],
energy_curve=[0.7, 0.75, 0.8, 0.85, 0.9, 0.95, 1.0, 0.9, 0.85, 0.8] * 3,
loopability_score=0.70,
shareability_score=0.75,
engagement_multiplier=1.9,
has_earworm_hook=False,
uses_trending_sound=False,
incorporates_silence=True,
viral_tier=ViralityTier.WARM,
avg_views=5800000.0,
avg_watch_through_rate=0.68,
avg_replay_rate=0.28,
platforms_successful=[Platform.YOUTUBE_SHORTS]
)
],
trending_audio_hooks=[],
niche_customizations={},
primary_playback_devices=[
PlaybackDevice.PHONE_SPEAKER,
PlaybackDevice.EARBUDS,
PlaybackDevice.LAPTOP_SPEAKERS
],
device_distribution={
PlaybackDevice.PHONE_SPEAKER: 0.50,
PlaybackDevice.EARBUDS: 0.35,
PlaybackDevice.LAPTOP_SPEAKERS: 0.10,
PlaybackDevice.HEADPHONES: 0.05
},
predicted_ctr=0.14,
predicted_watch_through=0.65,
predicted_loop_rate=0.25,
predicted_share_rate=0.10,
algorithm_preferences={
"favors_looping": False,
"penalizes_silence_start": False,
"rewards_immediate_hook": True,
"prefers_trending_audio": False,
"watch_time_priority": True
},
compression_artifacts_tolerance=0.5,
requires_sound_on_indicator=False
)
def get_platform_profile(self, platform: Platform) -> PlatformAudioProfile:
"""
Get complete audio profile for specified platform.
Args:
platform: Target platform enum
Returns:
PlatformAudioProfile with all optimization parameters
"""
if platform not in self.profiles:
raise ValueError(f"Unsupported platform: {platform}")
return self.profiles[platform]
def get_optimal_lufs(self, platform: Platform) -> float:
"""Get target LUFS for platform"""
return self.profiles[platform].loudness.target_lufs
def get_compression_settings(self, platform: Platform) -> CompressionProfile:
"""Get compression profile for platform"""
return self.profiles[platform].compression
def get_frequency_profile(self, platform: Platform) -> FrequencyBandProfile:
"""Get frequency optimization profile"""
return self.profiles[platform].frequency_bands
def get_spatial_rules(self, platform: Platform) -> SpatialProfile:
"""Get mono/stereo playback rules"""
return self.profiles[platform].spatial
def get_beat_matching_recommendations(self, platform: Platform) -> BeatMatchingProfile:
"""Get trend-aware beat matching recommendations"""
return self.profiles[platform].beat_matching
def simulate_playback(
self,
audio_data: np.ndarray,
platform: Platform,
device: PlaybackDevice
) -> Tuple[np.ndarray, Dict[str, Any]]:
"""
Simulate audio playback on specific device for platform.
Args:
audio_data: Input audio array
platform: Target platform
device: Playback device to simulate
Returns:
Tuple of (simulated_audio, diagnostic_report)
"""
profile = self.profiles[platform]
freq_profile = profile.frequency_bands
simulated_audio = audio_data.copy()
warnings = []
# Device-specific simulation
if device == PlaybackDevice.PHONE_SPEAKER:
# Apply phone speaker frequency response
# High-pass filter to remove inaudible low frequencies
simulated_audio = self._apply_highpass_filter(
simulated_audio,
freq_profile.phone_safe_low_hz
)
# Simulate mono summing (most phone speakers are mono)
if simulated_audio.ndim > 1:
simulated_audio = np.mean(simulated_audio, axis=0)
warnings.append("Phone speaker: Stereo collapsed to mono")
# Simulate limited dynamic range
simulated_audio = self._apply_phone_speaker_compression(simulated_audio)
elif device == PlaybackDevice.EARBUDS:
# Earbuds frequency response simulation
simulated_audio = self._apply_earbuds_response(
simulated_audio,
freq_profile.earbuds_sweet_spot_hz
)
elif device == PlaybackDevice.MONO_FALLBACK:
# Worst-case mono compatibility check
if simulated_audio.ndim > 1:
simulated_audio = np.mean(simulated_audio, axis=0)
warnings.append("Mono fallback: All stereo information lost")
# Analyze intelligibility
intelligibility_score = self._analyze_intelligibility(simulated_audio)
if intelligibility_score < 0.7:
warnings.append(f"Low intelligibility: {intelligibility_score:.2f}")
# Analyze clarity
clarity_score = self._analyze_clarity(simulated_audio)
if clarity_score < 0.65:
warnings.append(f"Clarity issues detected: {clarity_score:.2f}")
diagnostic_report = {
"device": device.value,
"platform": platform.value,
"intelligibility_score": intelligibility_score,
"clarity_score": clarity_score,
"warnings": warnings,
"peak_level_db": 20 * np.log10(np.abs(simulated_audio).max() + 1e-10),
"rms_level_db": 20 * np.log10(np.sqrt(np.mean(simulated_audio**2)) + 1e-10)
}
return simulated_audio, diagnostic_report
def normalize_loudness(
self,
audio_data: np.ndarray,
platform: Platform,
preserve_dynamics: bool = True
) -> Tuple[np.ndarray, Dict[str, float]]:
"""
Normalize audio to platform target loudness.
Args:
audio_data: Input audio array
platform: Target platform
preserve_dynamics: Whether to preserve dynamic range
Returns:
Tuple of (normalized_audio, loudness_metrics)
"""
profile = self.profiles[platform]
target_lufs = profile.loudness.target_lufs
# Measure current loudness (simplified)
current_rms = np.sqrt(np.mean(audio_data**2))
current_lufs = -23.0 + 20 * np.log10(current_rms + 1e-10)
# Calculate gain adjustment
gain_db = target_lufs - current_lufs
gain_linear = 10 ** (gain_db / 20.0)
normalized_audio = audio_data * gain_linear
# Apply peak limiting to prevent clipping
peak_limit = 10 ** (profile.loudness.true_peak_dbtp / 20.0)
peak = np.abs(normalized_audio).max()
if peak > peak_limit:
normalized_audio *= (peak_limit / peak)
# Apply compression if preserving dynamics
if preserve_dynamics:
normalized_audio = self._apply_dynamic_compression(
normalized_audio,
profile.compression
)
metrics = {
"input_lufs": current_lufs,
"output_lufs": target_lufs,
"gain_applied_db": gain_db,
"peak_limited": peak > peak_limit
}
return normalized_audio, metrics
def generate_candidate_variants(
self,
base_audio: np.ndarray,
platform: Platform,
num_variants: int = 5
) -> List[Dict[str, Any]]:
"""
Generate multiple audio variants optimized for virality.
Args:
base_audio: Base audio to create variants from
platform: Target platform
num_variants: Number of variants to generate
Returns:
List of variant dictionaries with audio and predicted scores
"""
profile = self.profiles[platform]
variants = []
for i in range(num_variants):
variant_audio = base_audio.copy()
# Apply different optimization strategies
if i == 0:
# Aggressive compression for maximum loudness
variant_audio = self._apply_aggressive_compression(variant_audio)
strategy = "max_loudness"
elif i == 1:
# Dynamic preservation for emotional impact
variant_audio = self._preserve_dynamics(variant_audio)
strategy = "dynamic_emotional"
elif i == 2:
# Beat-aligned enhancement
variant_audio = self._enhance_beat_alignment(variant_audio, profile)
strategy = "beat_optimized"
elif i == 3:
# Frequency sculpting for phone speakers
variant_audio = self._optimize_for_phone_speakers(variant_audio, profile)
strategy = "phone_optimized"
else:
# Experimental RL-guided variation
variant_audio = self._apply_rl_exploration(variant_audio)
strategy = "rl_experimental"
# Predict engagement scores
engagement_score = self._predict_engagement(variant_audio, profile)
variants.append({
"variant_id": f"{platform.value}_v{i}_{strategy}",
"audio": variant_audio,
"strategy": strategy,
"predicted_engagement": engagement_score,
"predicted_ctr": engagement_score * profile.predicted_ctr,
"predicted_watch_through": engagement_score * profile.predicted_watch_through,
"loopability_score": self._calculate_loopability(variant_audio),
"virality_tier": self._assign_virality_tier(engagement_score)
})
# Sort by predicted engagement
variants.sort(key=lambda x: x["predicted_engagement"], reverse=True)
return variants
def update_profile_with_feedback(
self,
platform: Platform,
variant_id: str,
engagement_metrics: Dict[str, float]
):
"""
Update platform profile based on real engagement feedback.
Args:
platform: Platform that was tested
variant_id: ID of audio variant
engagement_metrics: Actual engagement data from video
"""
profile = self.profiles[platform]
# Update predicted metrics with weighted average
alpha = 0.1 # Learning rate
if "ctr" in engagement_metrics:
profile.predicted_ctr = (
(1 - alpha) * profile.predicted_ctr +
alpha * engagement_metrics["ctr"]
)
if "watch_through_rate" in engagement_metrics:
profile.predicted_watch_through = (
(1 - alpha) * profile.predicted_watch_through +
alpha * engagement_metrics["watch_through_rate"]
)
if "loop_rate" in engagement_metrics:
profile.predicted_loop_rate = (
(1 - alpha) * profile.predicted_loop_rate +
alpha * engagement_metrics["loop_rate"]
)
# Store successful patterns for memory manager
if self.memory_manager and engagement_metrics.get("views", 0) > 1000000:
self.memory_manager.store_viral_pattern(
platform=platform.value,
variant_id=variant_id,
metrics=engagement_metrics
)
def get_cross_platform_optimization(
self,
audio_data: np.ndarray,
platforms: List[Platform]
) -> Dict[Platform, np.ndarray]:
"""
Generate platform-specific optimized versions from single audio.
Args:
audio_data: Source audio to optimize
platforms: List of target platforms
Returns:
Dictionary mapping each platform to optimized audio
"""
optimized_versions = {}
for platform in platforms:
profile = self.profiles[platform]
optimized_audio = audio_data.copy()
# Apply platform-specific normalization
optimized_audio, _ = self.normalize_loudness(
optimized_audio,
platform,
preserve_dynamics=True
)
# Apply platform-specific frequency optimization
optimized_audio = self._optimize_frequency_bands(
optimized_audio,
profile.frequency_bands
)
# Apply spatial optimization
optimized_audio = self._optimize_spatial(
optimized_audio,
profile.spatial
)
optimized_versions[platform] = optimized_audio
return optimized_versions
# ==================== DYNAMIC PROFILE ADAPTATION ====================
def ingest_trending_data(
self,
platform: Platform,
trending_data: Dict[str, Any]
):
"""
Update platform profile with real-time trending data.
Args:
platform: Platform to update
trending_data: Dict containing trending audio characteristics
{
"trending_bpms": [128, 140, 145],
"trending_hooks": [...],
"viral_patterns": [...],
"avg_engagement_metrics": {...}
}
"""
profile = self.profiles[platform]
# Update BPM preferences with trending data
if "trending_bpms" in trending_data:
profile.beat_matching.viral_bpm_peaks = trending_data["trending_bpms"]
# Adjust optimal range to include trending peaks
bpms = trending_data["trending_bpms"]
profile.beat_matching.optimal_bpm_range = (
min(bpms) - 10,
max(bpms) + 10
)
# Update trending audio hooks
if "trending_hooks" in trending_data:
profile.trending_audio_hooks = trending_data["trending_hooks"]
# Integrate new viral patterns
if "viral_patterns" in trending_data:
for pattern_data in trending_data["viral_patterns"]:
viral_pattern = self._create_viral_pattern_from_data(pattern_data)
# Replace cold patterns with new hot ones
profile.viral_patterns = [
p for p in profile.viral_patterns
if p.viral_tier != ViralityTier.COLD
]
profile.viral_patterns.insert(0, viral_pattern)
# Update engagement predictions
if "avg_engagement_metrics" in trending_data:
metrics = trending_data["avg_engagement_metrics"]
alpha = 0.3 # Higher learning rate for trending data
profile.predicted_ctr = (
(1 - alpha) * profile.predicted_ctr +
alpha * metrics.get("ctr", profile.predicted_ctr)
)
profile.predicted_watch_through = (
(1 - alpha) * profile.predicted_watch_through +
alpha * metrics.get("watch_through", profile.predicted_watch_through)
)
profile.predicted_loop_rate = (
(1 - alpha) * profile.predicted_loop_rate +
alpha * metrics.get("loop_rate", profile.predicted_loop_rate)
)
profile.predicted_share_rate = (
(1 - alpha) * profile.predicted_share_rate +
alpha * metrics.get("share_rate", profile.predicted_share_rate)
)
def adapt_profile_to_niche(
self,
platform: Platform,
niche: str,
niche_performance_data: Dict[str, Any]
):
"""
Customize platform profile for specific content niche.
Args:
platform: Target platform
niche: Content niche (e.g., "luxury_lifestyle", "comedy", "tech_hacks")
niche_performance_data: Historical performance data for niche
"""
profile = self.profiles[platform]
if niche not in profile.niche_customizations:
profile.niche_customizations[niche] = {}
niche_config = profile.niche_customizations[niche]
# Adapt based on niche characteristics
if niche == "luxury_lifestyle":
niche_config["loudness_adjustment_db"] = -1.0 # Slightly quieter, more refined
niche_config["compression_ratio_multiplier"] = 0.8 # Less aggressive
niche_config["stereo_width_boost"] = 1.2 # Wider, more spacious
niche_config["preferred_bpm_range"] = (100, 120) # Slower, sophisticated
elif niche == "comedy":
niche_config["loudness_adjustment_db"] = 0.5 # Slightly louder
niche_config["compression_ratio_multiplier"] = 1.1 # More consistent
niche_config["vocal_presence_boost_db"] = 2.0 # Emphasize voice
niche_config["preferred_bpm_range"] = (120, 140) # Upbeat
elif niche == "tech_hacks":
niche_config["loudness_adjustment_db"] = 0.0
niche_config["compression_ratio_multiplier"] = 1.0
niche_config["vocal_clarity_priority"] = True
niche_config["preferred_bpm_range"] = (130, 150) # Energetic
# Incorporate niche performance data
if "successful_patterns" in niche_performance_data:
niche_config["proven_patterns"] = niche_performance_data["successful_patterns"]
if "avg_engagement_multiplier" in niche_performance_data:
niche_config["engagement_boost"] = niche_performance_data["avg_engagement_multiplier"]
# ==================== RL INTEGRATION INTERFACES ====================
def get_rl_action_space(self, platform: Platform) -> Dict[str, Any]:
"""
Get available actions for RL agent to optimize audio.
Args:
platform: Target platform
Returns:
Dictionary defining action space dimensions and bounds
"""
profile = self.profiles[platform]
return {
"loudness_adjustment_db": {"min": -3.0, "max": 3.0, "current": 0.0},
"compression_ratio": {
"min": 2.0,
"max": 8.0,
"current": profile.compression.ratio
},
"stereo_width_percent": {
"min": 80.0,
"max": 180.0,
"current": profile.spatial.stereo_width_percent
},
"low_boost_db": {"min": -3.0, "max": 6.0, "current": 0.0},
"high_boost_db": {"min": -3.0, "max": 6.0, "current": 0.0},
"beat_emphasis_db": {"min": 0.0, "max": 4.0, "current": 0.0},
"hook_timing_offset_sec": {"min": -0.5, "max": 0.5, "current": 0.0},
"viral_pattern_selection": {
"options": [p.pattern_id for p in profile.viral_patterns],
"current": profile.viral_patterns[0].pattern_id if profile.viral_patterns else None
}
}
def apply_rl_action(
self,
audio_data: np.ndarray,
platform: Platform,
action: Dict[str, float]
) -> np.ndarray:
"""
Apply RL agent's chosen action to audio.
Args:
audio_data: Input audio
platform: Target platform
action: Dictionary of action parameters
Returns:
Modified audio based on RL action
"""
modified_audio = audio_data.copy()
# Apply loudness adjustment
if "loudness_adjustment_db" in action:
gain = 10 ** (action["loudness_adjustment_db"] / 20.0)
modified_audio *= gain
# Apply compression with RL-adjusted ratio
if "compression_ratio" in action:
modified_audio = self._apply_compression_with_ratio(
modified_audio,
action["compression_ratio"]
)
# Apply frequency adjustments
if "low_boost_db" in action:
modified_audio = self._boost_frequency_band(
modified_audio,
freq_range=(80.0, 200.0),
boost_db=action["low_boost_db"]
)
if "high_boost_db" in action:
modified_audio = self._boost_frequency_band(
modified_audio,
freq_range=(8000.0, 16000.0),
boost_db=action["high_boost_db"]
)
# Apply beat emphasis
if "beat_emphasis_db" in action and action["beat_emphasis_db"] > 0:
modified_audio = self._emphasize_beats(
modified_audio,
emphasis_db=action["beat_emphasis_db"]
)
return modified_audio
def get_rl_reward_signal(
self,
audio_variant_id: str,
platform: Platform,
actual_metrics: Dict[str, float]
) -> float:
"""
Calculate RL reward signal based on actual performance.
Args:
audio_variant_id: ID of tested variant
platform: Platform where tested
actual_metrics: Actual engagement metrics
Returns:
Reward value for RL agent (-1.0 to 1.0)
"""
profile = self.profiles[platform]
# Weighted reward components
weights = {
"views": 0.35,
"watch_through": 0.25,
"loop_rate": 0.20,
"share_rate": 0.20
}
reward = 0.0
# Views component (5M+ target)
if "views" in actual_metrics:
views = actual_metrics["views"]
if views >= 5_000_000:
reward += weights["views"] * 1.0
elif views >= 1_000_000:
reward += weights["views"] * (views / 5_000_000)
else:
reward += weights["views"] * (views / 5_000_000) * 0.5
# Watch-through rate
if "watch_through_rate" in actual_metrics:
wtr = actual_metrics["watch_through_rate"]
expected = profile.predicted_watch_through
reward += weights["watch_through"] * min(wtr / expected, 1.5)
# Loop rate
if "loop_rate" in actual_metrics:
loop = actual_metrics["loop_rate"]
expected = profile.predicted_loop_rate
reward += weights["loop_rate"] * min(loop / expected, 1.5)
# Share rate
if "share_rate" in actual_metrics:
share = actual_metrics["share_rate"]
expected = profile.predicted_share_rate
reward += weights["share_rate"] * min(share / expected, 1.5)
# Normalize to [-1, 1]
reward = (reward - 0.5) * 2.0
return np.clip(reward, -1.0, 1.0)
def select_optimal_pattern_for_exploration(
self,
platform: Platform,
exploration_rate: float = 0.2
) -> ViralAudioPattern:
"""
Select viral pattern balancing exploration vs exploitation.
Args:
platform: Target platform
exploration_rate: Probability of exploring new patterns
Returns:
Selected viral audio pattern
"""
profile = self.profiles[platform]
if np.random.random() < exploration_rate:
# Explore: prioritize WARM and COLD patterns
candidates = [
p for p in profile.viral_patterns
if p.viral_tier in [ViralityTier.WARM, ViralityTier.COLD]
]
if not candidates:
candidates = profile.viral_patterns
else:
# Exploit: prioritize HOT patterns
candidates = [
p for p in profile.viral_patterns
if p.viral_tier == ViralityTier.HOT
]
if not candidates:
candidates = profile.viral_patterns
# Weight by engagement multiplier
weights = np.array([p.engagement_multiplier for p in candidates])
weights = weights / weights.sum()
selected_idx = np.random.choice(len(candidates), p=weights)
return candidates[selected_idx]
# ==================== VARIANT SCORING & RANKING ====================
def score_audio_variants(
self,
audio_variants: List[np.ndarray],
platform: Platform,
niche: Optional[str] = None
) -> List[Dict[str, Any]]:
"""
Score multiple audio variants for predicted engagement.
Args:
audio_variants: List of audio arrays to evaluate
platform: Target platform
niche: Optional content niche for customization
Returns:
List of scored variants, sorted by predicted engagement
"""
profile = self.profiles[platform]
scored_variants = []
for idx, audio in enumerate(audio_variants):
# Calculate component scores
loudness_score = self._score_loudness_match(audio, profile)
frequency_score = self._score_frequency_optimization(audio, profile)
dynamic_score = self._score_dynamic_range(audio, profile)
spatial_score = self._score_spatial_quality(audio, profile)
loopability_score = self._calculate_loopability(audio)
hook_score = self._score_hook_placement(audio, profile)
# Apply niche boost if applicable
niche_boost = 1.0
if niche and niche in profile.niche_customizations:
niche_config = profile.niche_customizations[niche]
niche_boost = niche_config.get("engagement_boost", 1.0)
# Weighted composite score
composite_score = (
loudness_score * 0.20 +
frequency_score * 0.20 +
dynamic_score * 0.15 +
spatial_score * 0.10 +
loopability_score * 0.20 +
hook_score * 0.15
) * niche_boost
# Predict engagement metrics
predicted_views = composite_score * 5_000_000 * profile.predicted_ctr
predicted_wtr = composite_score * profile.predicted_watch_through
predicted_loop = composite_score * profile.predicted_loop_rate
predicted_share = composite_score * profile.predicted_share_rate
# Assign virality tier
if composite_score >= 0.85:
tier = ViralityTier.HOT
elif composite_score >= 0.70:
tier = ViralityTier.WARM
else:
tier = ViralityTier.COLD
scored_variants.append({
"variant_id": f"{platform.value}_scored_v{idx}",
"audio": audio,
"composite_score": composite_score,
"component_scores": {
"loudness": loudness_score,
"frequency": frequency_score,
"dynamics": dynamic_score,
"spatial": spatial_score,
"loopability": loopability_score,
"hooks": hook_score
},
"predicted_views": predicted_views,
"predicted_watch_through": predicted_wtr,
"predicted_loop_rate": predicted_loop,
"predicted_share_rate": predicted_share,
"virality_tier": tier,
"niche_optimized": niche is not None
})
# Sort by composite score
scored_variants.sort(key=lambda x: x["composite_score"], reverse=True)
return scored_variants
def rank_variants_by_predicted_virality(
self,
variants: List[Dict[str, Any]],
platform: Platform
) -> List[Dict[str, Any]]:
"""
Rank variants by comprehensive virality prediction.
Args:
variants: List of variant dictionaries
platform: Target platform
Returns:
Ranked list with virality probability scores
"""
profile = self.profiles[platform]
for variant in variants:
# Calculate virality probability (0-1)
base_score = variant.get("composite_score", 0.5)
# Factor in platform-specific preferences
algo_boost = 1.0
if variant.get("loopability_score", 0) > 0.8:
algo_boost *= 1.2 if profile.algorithm_preferences.get("favors_looping") else 1.0
if variant.get("has_immediate_hook"):
algo_boost *= 1.15 if profile.algorithm_preferences.get("rewards_immediate_hook") else 1.0
# Calculate final virality probability
virality_prob = min(base_score * algo_boost, 1.0)
# Predict actual view outcome
if virality_prob >= 0.90:
predicted_view_range = (7_000_000, 15_000_000)
confidence = "very_high"
elif virality_prob >= 0.80:
predicted_view_range = (5_000_000, 10_000_000)
confidence = "high"
elif virality_prob >= 0.70:
predicted_view_range = (2_000_000, 6_000_000)
confidence = "medium"
else:
predicted_view_range = (500_000, 3_000_000)
confidence = "low"
variant["virality_probability"] = virality_prob
variant["predicted_view_range"] = predicted_view_range
variant["confidence_level"] = confidence
variant["recommendation"] = self._generate_variant_recommendation(
variant,
virality_prob
)
# Sort by virality probability
variants.sort(key=lambda x: x["virality_probability"], reverse=True)
return variants
# ==================== CROSS-MODAL INTEGRATION ====================
def align_audio_with_visual_hooks(
self,
audio_data: np.ndarray,
visual_hook_timestamps: List[float],
platform: Platform
) -> np.ndarray:
"""
Synchronize audio events with visual hooks for maximum impact.
Args:
audio_data: Input audio
visual_hook_timestamps: List of visual hook timing (seconds)
platform: Target platform
Returns:
Audio aligned with visual hooks
"""
profile = self.profiles[platform]
aligned_audio = audio_data.copy()
# Get optimal audio hook timings from profile
audio_hooks = profile.beat_matching.beat_drop_timing_sec
# For each visual hook, ensure audio peak exists
for visual_time in visual_hook_timestamps:
# Find nearest audio hook
nearest_audio_hook = min(
audio_hooks,
key=lambda x: abs(x - visual_time)
)
# If too far apart, adjust audio emphasis
time_diff = abs(nearest_audio_hook - visual_time)
if time_diff > 0.5: # More than 500ms apart
# Add energy boost at visual hook timing
aligned_audio = self._add_audio_emphasis_at_time(
aligned_audio,
visual_time,
emphasis_db=2.0
)
return aligned_audio
def get_cross_modal_sync_rules(
self,
platform: Platform
) -> Dict[str, Any]:
"""
Get rules for syncing audio with visual elements.
Args:
platform: Target platform
Returns:
Dictionary of sync rules and preferences
"""
profile = self.profiles[platform]
return {
"beat_drop_on_scene_cut": True,
"silence_on_text_overlay": False,
"energy_boost_on_visual_climax": True,
"bass_hit_on_product_reveal": True,
"rhythm_follows_cuts": platform in [Platform.TIKTOK, Platform.INSTAGRAM_REELS],
"optimal_sync_tolerance_ms": 50,
"scene_cut_audio_timing": {
"anticipate_cut_ms": 100, # Audio leads visual by 100ms
"post_cut_sustain_ms": 200
},
"text_overlay_audio_rules": {
"duck_music_db": -6.0, # Lower music during text
"boost_voice_db": 3.0, # Boost narration
"maintain_beat": True
},
"visual_hook_priorities": [
"product_reveal",
"transformation_moment",
"punchline_delivery",
"emotional_peak",
"call_to_action"
]
}
# ==================== POST-PERFORMANCE LEARNING ====================
def learn_from_performance(
self,
platform: Platform,
variant_id: str,
audio_characteristics: Dict[str, Any],
actual_metrics: Dict[str, float]
):
"""
Update profile based on actual video performance.
Args:
platform: Platform where video was posted
variant_id: ID of audio variant used
audio_characteristics: Technical characteristics of the audio
actual_metrics: Actual engagement metrics achieved
"""
profile = self.profiles[platform]
views = actual_metrics.get("views", 0)
# If video hit 5M+ views, extract and store patterns
if views >= 5_000_000:
# Create new viral pattern from successful audio
new_pattern = ViralAudioPattern(
pattern_id=f"{variant_id}_viral_{int(views/1000000)}M",
hook_timing_sec=audio_characteristics.get("hook_timings", []),
emotional_peak_timing_sec=audio_characteristics.get("peak_timings", []),
energy_curve=audio_characteristics.get("energy_curve", []),
loopability_score=actual_metrics.get("loop_rate", 0.5),
shareability_score=actual_metrics.get("share_rate", 0.1),
engagement_multiplier=views / profile.predicted_views if hasattr(profile, 'predicted_views') else 2.0,
has_earworm_hook=audio_characteristics.get("has_earworm", False),
uses_trending_sound=audio_characteristics.get("uses_trending", False),
incorporates_silence=audio_characteristics.get("has_silence", False),
viral_tier=ViralityTier.HOT,
avg_views=float(views),
avg_watch_through_rate=actual_metrics.get("watch_through_rate", 0.7),
avg_replay_rate=actual_metrics.get("loop_rate", 0.3),
platforms_successful=[platform]
)
# Add to profile (limit to top 10 patterns)
profile.viral_patterns.insert(0, new_pattern)
profile.viral_patterns = profile.viral_patterns[:10]
# Update compression preferences if high performance
if actual_metrics.get("watch_through_rate", 0) > profile.predicted_watch_through * 1.2:
# This compression setting worked well
if "compression_ratio" in audio_characteristics:
alpha = 0.15
profile.compression.ratio = (
(1 - alpha) * profile.compression.ratio +
alpha * audio_characteristics["compression_ratio"]
)
# Update BPM preferences
if "bpm" in audio_characteristics and views >= 1_000_000:
bpm = audio_characteristics["bpm"]
if bpm not in profile.beat_matching.viral_bpm_peaks:
profile.beat_matching.viral_bpm_peaks.append(bpm)
profile.beat_matching.viral_bpm_peaks.sort()
# Keep only top 5 BPMs
profile.beat_matching.viral_bpm_peaks = profile.beat_matching.viral_bpm_peaks[-5:]
# Learn optimal loudness targets
if "integrated_lufs" in audio_characteristics:
actual_lufs = audio_characteristics["integrated_lufs"]
if actual_metrics.get("watch_through_rate", 0) > 0.75:
# Successful loudness, adjust target slightly
alpha = 0.1
profile.loudness.target_lufs = (
(1 - alpha) * profile.loudness.target_lufs +
alpha * actual_lufs
)
# Demote underperforming patterns
if views < 500_000:
for pattern in profile.viral_patterns:
if pattern.pattern_id == variant_id:
if pattern.viral_tier == ViralityTier.HOT:
pattern.viral_tier = ViralityTier.WARM
elif pattern.viral_tier == ViralityTier.WARM:
pattern.viral_tier = ViralityTier.COLD
def get_performance_insights(
self,
platform: Platform
) -> Dict[str, Any]:
"""
Get insights from historical performance data.
Args:
platform: Platform to analyze
Returns:
Dictionary of actionable insights
"""
profile = self.profiles[platform]
# Analyze viral patterns
hot_patterns = [p for p in profile.viral_patterns if p.viral_tier == ViralityTier.HOT]
insights = {
"top_performing_bpms": profile.beat_matching.viral_bpm_peaks,
"optimal_loudness_lufs": profile.loudness.target_lufs,
"best_compression_ratio": profile.compression.ratio,
"num_proven_patterns": len(hot_patterns),
"avg_views_hot_patterns": np.mean([p.avg_views for p in hot_patterns]) if hot_patterns else 0,
"best_hook_timings": self._extract_best_hook_timings(hot_patterns),
"engagement_trends": {
"predicted_ctr": profile.predicted_ctr,
"predicted_watch_through": profile.predicted_watch_through,
"predicted_loop_rate": profile.predicted_loop_rate,
"predicted_share_rate": profile.predicted_share_rate
},
"recommendations": self._generate_performance_recommendations(profile)
}
return insights
# ==================== HELPER METHODS ====================
def _apply_highpass_filter(self, audio: np.ndarray, cutoff_hz: float) -> np.ndarray:
"""Apply high-pass filter to remove low frequencies"""
# Simplified implementation
return audio
def _apply_phone_speaker_compression(self, audio: np.ndarray) -> np.ndarray:
"""Simulate phone speaker dynamic range compression"""
# Simplified implementation
threshold = 0.3
ratio = 4.0
compressed = np.where(
np.abs(audio) > threshold,
np.sign(audio) * (threshold + (np.abs(audio) - threshold) / ratio),
audio
)
return compressed
def _apply_earbuds_response(
self,
audio: np.ndarray,
sweet_spot: Tuple[float, float]
) -> np.ndarray:
"""Apply earbuds frequency response simulation"""
# Simplified implementation
return audio
def _analyze_intelligibility(self, audio: np.ndarray) -> float:
"""Analyze vocal/content intelligibility score"""
# Simplified: measure mid-range energy
rms = np.sqrt(np.mean(audio**2))
return min(rms * 10, 1.0)
def _analyze_clarity(self, audio: np.ndarray) -> float:
"""Analyze overall audio clarity"""
# Simplified: measure high-frequency content
return 0.75 # Placeholder
def _apply_dynamic_compression(
self,
audio: np.ndarray,
compression: CompressionProfile
) -> np.ndarray:
"""Apply dynamic range compression"""
threshold_linear = 10 ** (compression.threshold_db / 20.0)
compressed = np.where(
np.abs(audio) > threshold_linear,
np.sign(audio) * (
threshold_linear +
(np.abs(audio) - threshold_linear) / compression.ratio
),
audio
)
# Apply makeup gain
makeup_gain = 10 ** (compression.makeup_gain_db / 20.0)
compressed *= makeup_gain
return compressed
def _apply_aggressive_compression(self, audio: np.ndarray) -> np.ndarray:
"""Apply aggressive compression for maximum loudness"""
return self._apply_compression_with_ratio(audio, 6.0)
def _preserve_dynamics(self, audio: np.ndarray) -> np.ndarray:
"""Preserve dynamic range for emotional impact"""
return self._apply_compression_with_ratio(audio, 2.5)
def _enhance_beat_alignment(
self,
audio: np.ndarray,
profile: PlatformAudioProfile
) -> np.ndarray:
"""Enhance beat transients for better rhythm"""
# Simplified implementation
return audio
def _optimize_for_phone_speakers(
self,
audio: np.ndarray,
profile: PlatformAudioProfile
) -> np.ndarray:
"""Optimize specifically for phone speaker playback"""
# Apply phone-safe high-pass
audio = self._apply_highpass_filter(
audio,
profile.frequency_bands.phone_safe_low_hz
)
# Boost presence range
audio = self._boost_frequency_band(audio, (2000, 4000), 2.0)
return audio
def _predict_engagement(
self,
audio: np.ndarray,
profile: PlatformAudioProfile
) -> float:
"""Predict engagement score for audio variant"""
# Multi-factor engagement prediction
loudness_score = self._score_loudness_match(audio, profile)
hook_score = self._score_hook_placement(audio, profile)
loop_score = self._calculate_loopability(audio)
composite = (loudness_score * 0.3 + hook_score * 0.4 + loop_score * 0.3)
return composite
def _calculate_loopability(self, audio: np.ndarray) -> float:
"""Calculate how well audio loops seamlessly"""
# Compare start and end energy
start_energy = np.mean(audio[:1000]**2) if len(audio) > 1000 else 0
end_energy = np.mean(audio[-1000:]**2) if len(audio) > 1000 else 0
energy_match = 1.0 - abs(start_energy - end_energy) / (start_energy + end_energy + 1e-10)
return energy_match
def _assign_virality_tier(self, engagement_score: float) -> ViralityTier:
"""Assign virality tier based on engagement score"""
if engagement_score >= 0.85:
return ViralityTier.HOT
elif engagement_score >= 0.70:
return ViralityTier.WARM
else:
return ViralityTier.COLD
def _create_viral_pattern_from_data(self, pattern_data: Dict[str, Any]) -> ViralAudioPattern:
"""Create ViralAudioPattern from trending data"""
return ViralAudioPattern(
pattern_id=pattern_data.get("id", "trending_pattern"),
hook_timing_sec=pattern_data.get("hook_timings", [0.5, 3.0]),
emotional_peak_timing_sec=pattern_data.get("peaks", [7.0]),
energy_curve=pattern_data.get("energy", [0.8] * 16),
loopability_score=pattern_data.get("loopability", 0.8),
shareability_score=pattern_data.get("shareability", 0.75),
engagement_multiplier=pattern_data.get("engagement_mult", 2.0),
has_earworm_hook=pattern_data.get("has_hook", True),
uses_trending_sound=True,
incorporates_silence=pattern_data.get("has_silence", False),
viral_tier=ViralityTier.HOT,
avg_views=pattern_data.get("avg_views", 5000000.0),
avg_watch_through_rate=pattern_data.get("wtr", 0.75),
avg_replay_rate=pattern_data.get("replay", 0.35),
platforms_successful=[Platform.TIKTOK]
)
def _optimize_frequency_bands(
self,
audio: np.ndarray,
freq_profile: FrequencyBandProfile
) -> np.ndarray:
"""Apply frequency band optimizations"""
# Simplified implementation
optimized = audio.copy()
optimized = self._apply_highpass_filter(optimized, freq_profile.low_cut_hz)
return optimized
def _optimize_spatial(
self,
audio: np.ndarray,
spatial: SpatialProfile
) -> np.ndarray:
"""Apply spatial/stereo optimizations"""
# Simplified implementation
if audio.ndim > 1 and not spatial.mono_compatibility_required:
# Apply stereo width adjustment
width_factor = spatial.stereo_width_percent / 100.0
mid = (audio[0] + audio[1]) / 2
side = (audio[0] - audio[1]) / 2
audio[0] = mid + side * width_factor
audio[1] = mid - side * width_factor
return audio
def _score_loudness_match(
self,
audio: np.ndarray,
profile: PlatformAudioProfile
) -> float:
"""Score how well audio matches target loudness"""
current_rms = np.sqrt(np.mean(audio**2))
current_lufs = -23.0 + 20 * np.log10(current_rms + 1e-10)
target_lufs = profile.loudness.target_lufs
diff = abs(current_lufs - target_lufs)
# Score: 1.0 if perfect match, decreases with difference
score = max(0.0, 1.0 - diff / 6.0)
return score
def _score_frequency_optimization(
self,
audio: np.ndarray,
profile: PlatformAudioProfile
) -> float:
"""Score frequency distribution quality"""
# Simplified: check if within optimal ranges
return 0.8 # Placeholder
def _score_dynamic_range(
self,
audio: np.ndarray,
profile: PlatformAudioProfile
) -> float:
"""Score dynamic range appropriateness"""
peak = np.abs(audio).max()
rms = np.sqrt(np.mean(audio**2))
dynamic_range_db = 20 * np.log10(peak / (rms + 1e-10))
# Platform-appropriate range: 6-12 dB
if 6 <= dynamic_range_db <= 12:
return 1.0
elif dynamic_range_db < 6:
return 0.6 # Over-compressed
else:
return 0.7 # Under-compressed
def _score_spatial_quality(
self,
audio: np.ndarray,
profile: PlatformAudioProfile
) -> float:
"""Score spatial/stereo quality"""
if audio.ndim == 1:
# Mono audio
return 0.9 if profile.spatial.mono_compatibility_required else 0.6
else:
# Stereo audio
correlation = np.corrcoef(audio[0], audio[1])[0, 1]
# Good stereo: correlation between 0.3-0.7
if 0.3 <= correlation <= 0.7:
return 1.0
else:
return 0.7
def _score_hook_placement(
self,
audio: np.ndarray,
profile: PlatformAudioProfile
) -> float:
"""Score hook timing and placement"""
# Simplified: check energy in first 3 seconds
hook_region = audio[:int(3 * 44100)] # First 3 seconds at 44.1kHz
hook_energy = np.mean(hook_region**2)
overall_energy = np.mean(audio**2)
# Good hook: front-loaded energy
hook_ratio = hook_energy / (overall_energy + 1e-10)
if hook_ratio >= 1.2:
return 1.0
elif hook_ratio >= 1.0:
return 0.8
else:
return 0.6
def _apply_compression_with_ratio(
self,
audio: np.ndarray,
ratio: float
) -> np.ndarray:
"""Apply compression with specified ratio"""
threshold = 0.3
compressed = np.where(
np.abs(audio) > threshold,
np.sign(audio) * (threshold + (np.abs(audio) - threshold) / ratio),
audio
)
return compressed
def _boost_frequency_band(
self,
audio: np.ndarray,
freq_range: Tuple[float, float],
boost_db: float
) -> np.ndarray:
"""Boost specific frequency band"""
# Simplified implementation
gain = 10 ** (boost_db / 20.0)
return audio * gain
def _emphasize_beats(self, audio: np.ndarray, emphasis_db: float) -> np.ndarray:
"""Emphasize beat transients"""
# Simplified implementation
return audio
def _add_audio_emphasis_at_time(
self,
audio: np.ndarray,
time_sec: float,
emphasis_db: float
) -> np.ndarray:
"""Add energy boost at specific timestamp"""
sample_rate = 44100
sample_idx = int(time_sec * sample_rate)
if 0 <= sample_idx < len(audio):
window_size = int(0.1 * sample_rate) # 100ms window
start = max(0, sample_idx - window_size // 2)
end = min(len(audio), sample_idx + window_size // 2)
gain = 10 ** (emphasis_db / 20.0)
audio[start:end] *= gain
return audio
def _extract_best_hook_timings(
self,
patterns: List[ViralAudioPattern]
) -> List[float]:
"""Extract most successful hook timings"""
if not patterns:
return [0.5, 3.0, 7.0]
# Average hook timings from best patterns
all_timings = []
for pattern in patterns[:3]: # Top 3 patterns
all_timings.extend(pattern.hook_timing_sec)
return sorted(list(set(all_timings)))
def _generate_performance_recommendations(
self,
profile: PlatformAudioProfile
) -> List[str]:
"""Generate actionable recommendations"""
recommendations = []
# Check if enough HOT patterns
hot_count = sum(1 for p in profile.viral_patterns if p.viral_tier == ViralityTier.HOT)
if hot_count < 3:
recommendations.append(
f"Need more proven patterns: only {hot_count} HOT patterns available"
)
# Check engagement trends
if profile.predicted_ctr < 0.15:
recommendations.append(
"CTR below optimal: consider stronger hooks in first 0.5 seconds"
)
if profile.predicted_loop_rate < 0.30:
recommendations.append(
"Loop rate low: improve audio start/end continuity"
)
# BPM recommendations
recommendations.append(
f"Focus on BPMs: {', '.join(map(str, profile.beat_matching.viral_bpm_peaks))}"
)
return recommendations
def _generate_variant_recommendation(
self,
variant: Dict[str, Any],
virality_prob: float
) -> str:
"""Generate recommendation for variant usage"""
if virality_prob >= 0.90:
return "HIGHLY RECOMMENDED: Use immediately for maximum viral potential"
elif virality_prob >= 0.80:
return "RECOMMENDED: Strong viral candidate, use for main content"
elif virality_prob >= 0.70:
return "TEST: Good potential, consider A/B testing"
else:
return "EXPERIMENTAL: Use for exploration or refinement"
# ==================== GLOBAL API ====================
def get_platform_profile(platform: str) -> PlatformAudioProfile:
"""
Global function to get platform profile.
Args:
platform: Platform name as string (e.g., "tiktok", "instagram_reels")
Returns:
PlatformAudioProfile for the specified platform
"""
manager = PlatformAudioProfileManager()
platform_enum = Platform(platform.lower())
return manager.get_platform_profile(platform_enum)
def optimize_audio_for_platform(
audio_data: np.ndarray,
platform: str,
niche: Optional[str] = None
) -> Tuple[np.ndarray, Dict[str, Any]]:
"""
One-stop function to optimize audio for a platform.
Args:
audio_data: Input audio array
platform: Target platform name
niche: Optional content niche
Returns:
Tuple of (optimized_audio, optimization_report)
"""
manager = PlatformAudioProfileManager()
platform_enum = Platform(platform.lower())
# Normalize loudness
optimized, loudness_metrics = manager.normalize_loudness(
audio_data,
platform_enum
)
# Score the result
variants = manager.score_audio_variants([optimized], platform_enum, niche)
report = {
"loudness_metrics": loudness_metrics,
"variant_score": variants[0] if variants else {},
"platform": platform,
"niche": niche
}
return optimized, report
# Example usage
if __name__ == "__main__":
# Initialize manager
manager = PlatformAudioProfileManager()
# Get TikTok profile
tiktok_profile = manager.get_platform_profile(Platform.TIKTOK)
print(f"TikTok Target LUFS: {tiktok_profile.loudness.target_lufs}")
print(f"TikTok Viral BPMs: {tiktok_profile.beat_matching.viral_bpm_peaks}")
# Simulate RL integration
action_space = manager.get_rl_action_space(Platform.TIKTOK)
print(f"\nRL Action Space: {action_space}")
# Example: Generate dummy audio for testing
dummy_audio = np.random.randn(44100 * 15) * 0.5 # 15 seconds
# Score variants
variants = manager.generate_candidate_variants(
dummy_audio,
Platform.TIKTOK,
num_variants=3
)
print(f"\nGenerated {len(variants)} variants")
for i, var in enumerate(variants):
print(f"Variant {i}: {var['strategy']} - Score: {var['predicted_engagement']:.3f}")
# Ingest trending data
trending_data = {
"trending_bpms": [128, 140, 145],
"avg_engagement_metrics": {
"ctr": 0.20,
"watch_through": 0.78,
"loop_rate": 0.40
}
}
manager.ingest_trending_data(Platform.TIKTOK, trending_data)
print("\nProfile updated with trending data")
# Get performance insights
insights = manager.get_performance_insights(Platform.TIKTOK)
print(f"\nPerformance Insights:")
print(f"Top BPMs: {insights['top_performing_bpms']}")
print(f"Recommendations: {insights['recommendations']}")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment