Skip to content

Instantly share code, notes, and snippets.

@bogged-broker

bogged-broker/audio_platforms_profiles.py

Created December 30, 2025 21:06
Show Gist options

  • Select an option

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

Select an option

Learn more about clone URLs
Save bogged-broker/7310016d33d1019e2fff9a10e2349dec to your computer and use it in GitHub Desktop.
Download ZIP
Raw
audio_platforms_profiles.py
"""
Platform Audio Profiles Module - 5M+ View Inevitability Engine
Defines platform-specific audio constraints and optimization profiles with:
- Real-time trend adaptation via live data ingestion
- RL feedback loop integration for automated optimization
- A/B variant scoring and selection
- Cross-modal synchronization with visuals and captions
- Self-evolving performance learning system
"""
import json
from typing import Dict, List, Optional, Tuple, Any, Callable
from dataclasses import dataclass, asdict, field
from enum import Enum
import numpy as np
from collections import deque
from datetime import datetime, timedelta
import asyncio
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
class TrendSource(Enum):
"""Sources for trending data"""
PLATFORM_API = "platform_api"
WEB_SCRAPER = "web_scraper"
MANUAL_CURATION = "manual_curation"
ML_PREDICTION = "ml_prediction"
MEMORY_MANAGER = "memory_manager"
@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
# Trend decay tracking
last_trend_update: Optional[datetime] = None
trend_confidence: float = 1.0 # 0-1, how confident in current trends
@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 CrossModalSyncProfile:
"""Rules for syncing audio with visual elements"""
visual_audio_sync_bonus: float # Engagement boost when synced (1.0-2.0)
caption_hook_audio_boost: float # Boost when caption appears
scene_transition_audio_weight: float # Audio impact on scene cuts
# Timing rules
audio_leads_visual_ms: int # Audio anticipates visual by this much
beat_on_scene_cut: bool
silence_on_text_overlay: bool
energy_boost_on_visual_climax: bool
bass_hit_on_product_reveal: bool
# Caption integration
duck_audio_during_text_db: float # Lower audio when text appears
boost_voice_during_text_db: float # Boost narration
maintain_beat_during_text: bool
# Visual hook priorities (ordered by importance)
visual_hook_priorities: List[str] = field(default_factory=lambda: [
"product_reveal",
"transformation_moment",
"punchline_delivery",
"emotional_peak",
"call_to_action"
])
@dataclass
class TrendingDataSnapshot:
"""Snapshot of trending audio characteristics"""
timestamp: datetime
source: TrendSource
platform: Platform
# Trending characteristics
trending_bpms: List[int]
trending_hooks: List[Dict[str, Any]]
viral_patterns: List[Dict[str, Any]]
# Engagement metrics
avg_views: float
avg_watch_through: float
avg_loop_rate: float
avg_share_rate: float
# Confidence and decay
confidence_score: float # 0-1, data quality/recency
decay_rate: float # How fast this data becomes stale
expires_at: datetime
@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]
# Real-time tracking
created_at: datetime = field(default_factory=datetime.now)
last_used: Optional[datetime] = None
usage_count: int = 0
success_rate: float = 0.0 # Ratio of successful uses
@dataclass
class AudioVariantScore:
"""Score for an audio variant with detailed breakdown"""
variant_id: str
audio_data: np.ndarray
# Component scores (0-1)
loudness_score: float
frequency_score: float
dynamic_score: float
spatial_score: float
loopability_score: float
hook_score: float
cross_modal_sync_score: float
# Composite scores
composite_score: float # Weighted average
virality_probability: float # 0-1 probability of going viral
# Predictions
predicted_views: float
predicted_view_range: Tuple[float, float]
predicted_watch_through: float
predicted_loop_rate: float
predicted_share_rate: float
# Metadata
virality_tier: ViralityTier
confidence_level: str # "very_high", "high", "medium", "low"
recommendation: str
strategy: str # Optimization strategy used
# RL integration
rl_action_params: Optional[Dict[str, float]] = None
exploration_bonus: float = 0.0
@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
cross_modal_sync: CrossModalSyncProfile
# 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 (auto-updated)
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
# Real-time adaptation
trending_data_queue: deque = field(default_factory=lambda: deque(maxlen=50))
last_profile_update: datetime = field(default_factory=datetime.now)
performance_history: List[Dict[str, Any]] = field(default_factory=list)
# RL integration
rl_action_history: List[Dict[str, Any]] = field(default_factory=list)
rl_reward_history: List[float] = field(default_factory=list)
class TrendIngestionEngine:
"""Engine for ingesting and processing trending audio data"""
def __init__(self):
self.trend_cache: Dict[Platform, deque] = {
platform: deque(maxlen=100) for platform in Platform
}
self.update_callbacks: List[Callable] = []
self.auto_update_enabled = True
self.update_interval_minutes = 30
async def fetch_trending_data(
self,
platform: Platform,
source: TrendSource = TrendSource.PLATFORM_API
) -> TrendingDataSnapshot:
"""
Fetch real-time trending data from specified source.
Args:
platform: Target platform
source: Data source to use
Returns:
TrendingDataSnapshot with current trends
"""
# Simulate API call - in production, this would call actual APIs
await asyncio.sleep(0.1) # Simulate network delay
# Mock trending data (in production: real API responses)
snapshot = TrendingDataSnapshot(
timestamp=datetime.now(),
source=source,
platform=platform,
trending_bpms=self._fetch_trending_bpms(platform),
trending_hooks=self._fetch_trending_hooks(platform),
viral_patterns=self._fetch_viral_patterns(platform),
avg_views=5_800_000.0,
avg_watch_through=0.76,
avg_loop_rate=0.38,
avg_share_rate=0.14,
confidence_score=0.92,
decay_rate=0.05, # 5% decay per hour
expires_at=datetime.now() + timedelta(hours=6)
)
self.trend_cache[platform].append(snapshot)
return snapshot
def _fetch_trending_bpms(self, platform: Platform) -> List[int]:
"""Fetch currently trending BPMs"""
# In production: scrape/API from platform
base_bpms = {
Platform.TIKTOK: [128, 140, 145, 150],
Platform.INSTAGRAM_REELS: [120, 128, 135],
Platform.YOUTUBE_SHORTS: [115, 128, 140]
}
return base_bpms.get(platform, [128, 140])
def _fetch_trending_hooks(self, platform: Platform) -> List[Dict[str, Any]]:
"""Fetch trending audio hook patterns"""
return [
{"type": "bass_drop", "timing": 0.5, "confidence": 0.88},
{"type": "vocal_chop", "timing": 3.0, "confidence": 0.85},
{"type": "synth_build", "timing": 7.0, "confidence": 0.82}
]
def _fetch_viral_patterns(self, platform: Platform) -> List[Dict[str, Any]]:
"""Fetch viral audio patterns from last 24h"""
return [
{
"id": f"{platform.value}_viral_24h_1",
"avg_views": 8_200_000,
"loopability": 0.91,
"shareability": 0.86,
"engagement_mult": 2.6
}
]
def get_weighted_trends(
self,
platform: Platform,
decay_weights: bool = True
) -> TrendingDataSnapshot:
"""
Get weighted average of recent trends with decay.
Args:
platform: Target platform
decay_weights: Apply time-based decay to older trends
Returns:
Aggregated trending snapshot
"""
snapshots = list(self.trend_cache[platform])
if not snapshots:
raise ValueError(f"No trending data available for {platform.value}")
# Calculate weights with time decay
now = datetime.now()
weights = []
for snap in snapshots:
age_hours = (now - snap.timestamp).total_seconds() / 3600
if decay_weights:
weight = snap.confidence_score * np.exp(-snap.decay_rate * age_hours)
else:
weight = snap.confidence_score
weights.append(weight)
weights = np.array(weights)
weights = weights / weights.sum()
# Weighted aggregation
all_bpms = []
for snap, weight in zip(snapshots, weights):
all_bpms.extend(snap.trending_bpms * int(weight * 10))
# Count most frequent BPMs
from collections import Counter
bpm_counts = Counter(all_bpms)
trending_bpms = [bpm for bpm, _ in bpm_counts.most_common(5)]
# Aggregate metrics
avg_views = sum(s.avg_views * w for s, w in zip(snapshots, weights))
avg_wtr = sum(s.avg_watch_through * w for s, w in zip(snapshots, weights))
avg_loop = sum(s.avg_loop_rate * w for s, w in zip(snapshots, weights))
avg_share = sum(s.avg_share_rate * w for s, w in zip(snapshots, weights))
return TrendingDataSnapshot(
timestamp=now,
source=TrendSource.ML_PREDICTION,
platform=platform,
trending_bpms=trending_bpms,
trending_hooks=snapshots[-1].trending_hooks, # Use most recent
viral_patterns=snapshots[-1].viral_patterns,
avg_views=avg_views,
avg_watch_through=avg_wtr,
avg_loop_rate=avg_loop,
avg_share_rate=avg_share,
confidence_score=weights.max(),
decay_rate=0.05,
expires_at=now + timedelta(hours=3)
)
class RLIntegrationInterface:
"""Interface for RL agent to interact with audio profiles"""
def __init__(self, profile_manager: 'PlatformAudioProfileManager'):
self.profile_manager = profile_manager
self.action_history: List[Dict[str, Any]] = []
self.reward_history: List[float] = []
def get_state_vector(
self,
platform: Platform,
audio_characteristics: Optional[Dict[str, float]] = None
) -> np.ndarray:
"""
Get current state as vector for RL agent.
Args:
platform: Target platform
audio_characteristics: Current audio features
Returns:
State vector for RL decision-making
"""
profile = self.profile_manager.get_platform_profile(platform)
state = [
profile.loudness.target_lufs / 20.0, # Normalize to ~0-1
profile.compression.ratio / 10.0,
profile.spatial.stereo_width_percent / 200.0,
profile.predicted_ctr,
profile.predicted_watch_through,
profile.predicted_loop_rate,
profile.predicted_share_rate,
len([p for p in profile.viral_patterns if p.viral_tier == ViralityTier.HOT]) / 10.0,
profile.beat_matching.trend_confidence
]
if audio_characteristics:
state.extend([
audio_characteristics.get("current_lufs", -14.0) / 20.0,
audio_characteristics.get("dynamic_range_db", 8.0) / 20.0,
audio_characteristics.get("loopability", 0.5)
])
return np.array(state, dtype=np.float32)
def get_action_space_bounds(self, platform: Platform) -> Dict[str, Tuple[float, float]]:
"""Get min/max bounds for each action dimension"""
return {
"loudness_adjustment_db": (-3.0, 3.0),
"compression_ratio": (2.0, 8.0),
"stereo_width_percent": (80.0, 180.0),
"low_boost_db": (-3.0, 6.0),
"high_boost_db": (-3.0, 6.0),
"beat_emphasis_db": (0.0, 4.0),
"hook_timing_offset_sec": (-0.5, 0.5)
}
def sample_action(
self,
platform: Platform,
exploration_rate: float = 0.2,
policy_network: Optional[Callable] = None
) -> Dict[str, float]:
"""
Sample action for RL agent (exploration or exploitation).
Args:
platform: Target platform
exploration_rate: Probability of random exploration
policy_network: Optional trained policy function
Returns:
Action dictionary with parameter values
"""
bounds = self.get_action_space_bounds(platform)
if policy_network is None or np.random.random() < exploration_rate:
# Random exploration
action = {
param: np.random.uniform(low, high)
for param, (low, high) in bounds.items()
}
else:
# Use policy network
state = self.get_state_vector(platform)
action_vector = policy_network(state)
action = {
param: float(action_vector[i])
for i, param in enumerate(bounds.keys())
}
self.action_history.append({
"timestamp": datetime.now(),
"platform": platform.value,
"action": action,
"exploration": exploration_rate
})
return action
def compute_reward(
self,
platform: Platform,
actual_metrics: Dict[str, float],
target_views: float = 5_000_000.0
) -> float:
"""
Compute reward signal for RL agent.
Args:
platform: Platform where video was posted
actual_metrics: Actual engagement metrics
target_views: Target view count (default 5M)
Returns:
Reward value between -1.0 and 1.0
"""
profile = self.profile_manager.get_platform_profile(platform)
# Multi-objective reward
weights = {
"views": 0.40,
"watch_through": 0.25,
"loop_rate": 0.20,
"share_rate": 0.15
}
reward_components = []
# Views component (target: 5M+)
if "views" in actual_metrics:
views = actual_metrics["views"]
views_reward = min(views / target_views, 2.0) - 0.5 # -0.5 to 1.5
reward_components.append(weights["views"] * views_reward)
# Watch-through rate
if "watch_through_rate" in actual_metrics:
wtr = actual_metrics["watch_through_rate"]
expected = profile.predicted_watch_through
wtr_reward = (wtr / expected) - 0.5 # -0.5 to 1.5+
reward_components.append(weights["watch_through"] * wtr_reward)
# Loop rate
if "loop_rate" in actual_metrics:
loop = actual_metrics["loop_rate"]
expected = profile.predicted_loop_rate
loop_reward = (loop / expected) - 0.5
reward_components.append(weights["loop_rate"] * loop_reward)
# Share rate
if "share_rate" in actual_metrics:
share = actual_metrics["share_rate"]
expected = profile.predicted_share_rate
share_reward = (share / expected) - 0.5
reward_components.append(weights["share_rate"] * share_reward)
# Total reward
total_reward = sum(reward_components)
normalized_reward = np.tanh(total_reward) # Squash to [-1, 1]
self.reward_history.append(normalized_reward)
return float(normalized_reward)
def update_policy(
self,
platform: Platform,
reward: float,
learning_rate: float = 0.01
):
"""
Update profile based on RL reward (simple policy gradient).
Args:
platform: Platform to update
reward: Reward signal received
learning_rate: How much to adjust parameters
"""
if not self.action_history:
return
last_action = self.action_history[-1]
if reward > 0:
# Positive reward: move profile toward this action
profile = self.profile_manager.get_platform_profile(platform)
if "compression_ratio" in last_action["action"]:
profile.compression.ratio += learning_rate * reward * (
last_action["action"]["compression_ratio"] - profile.compression.ratio
)
if "loudness_adjustment_db" in last_action["action"]:
profile.loudness.target_lufs += learning_rate * reward * (
last_action["action"]["loudness_adjustment_db"]
)
class VariantScoringEngine:
"""Engine for scoring and ranking audio variants"""
def __init__(self, profile_manager: 'PlatformAudioProfileManager'):
self.profile_manager = profile_manager
self.scoring_cache: Dict[str, AudioVariantScore] = {}
def score_single_variant(
self,
audio: np.ndarray,
platform: Platform,
variant_id: str,
strategy: str = "default",
visual_sync_data: Optional[Dict[str, Any]] = None,
niche: Optional[str] = None
) -> AudioVariantScore:
"""
Score a single audio variant comprehensively.
Args:
audio: Audio data array
platform: Target platform
variant_id: Unique variant identifier
strategy: Optimization strategy used
visual_sync_data: Visual hook timing data
niche: Content niche
Returns:
AudioVariantScore with detailed breakdown
"""
profile = self.profile_manager.get_platform_profile(platform)
# 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)
# Cross-modal sync score
cross_modal_score = self._score_cross_modal_sync(
audio,
profile,
visual_sync_data
) if visual_sync_data else 0.75
# Apply niche boost
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
composite_score = (
loudness_score * 0.18 +
frequency_score * 0.16 +
dynamic_score * 0.12 +
spatial_score * 0.08 +
loopability_score * 0.18 +
hook_score * 0.14 +
cross_modal_score * 0.14
) * niche_boost
# Virality probability with algorithm preferences
algo_multiplier = 1.0
if loopability_score > 0.8 and profile.algorithm_preferences.get("favors_looping"):
algo_multiplier *= 1.25
if hook_score > 0.85 and profile.algorithm_preferences.get("rewards_immediate_hook"):
algo_multiplier *= 1.15
virality_prob = min(composite_score * algo_multiplier, 1.0)
# Predict outcomes
base_views = 5_000_000.0
predicted_views = virality_prob * base_views * profile.predicted_ctr * 2.5
if virality_prob >= 0.90:
view_range = (8_000_000, 20_000_000)
confidence = "very_high"
tier = ViralityTier.HOT
elif virality_prob >= 0.80:
view_range = (5_000_000, 12_000_000)
confidence = "high"
tier = ViralityTier.HOT
elif virality_prob >= 0.70:
view_range = (2_000_000, 7_000_000)
confidence = "medium"
tier = ViralityTier.WARM
else:
view_range = (500_000, 3_000_000)
confidence = "low"
tier = ViralityTier.COLD
# Generate recommendation
if virality_prob >= 0.90:
recommendation = "🔥 DEPLOY IMMEDIATELY: Exceptionally high viral potential"
elif virality_prob >= 0.80:
recommendation = "✅ HIGHLY RECOMMENDED: Strong candidate for 5M+ views"
elif virality_prob >= 0.70:
recommendation = "⚠️ TEST RECOMMENDED: Good potential, A/B test advised"
else:
recommendation = "🧪 EXPERIMENTAL: Use for exploration or refinement"
score = AudioVariantScore(
variant_id=variant_id,
audio_data=audio,
loudness_score=loudness_score,
frequency_score=frequency_score,
dynamic_score=dynamic_score,
spatial_score=spatial_score,
loopability_score=loopability_score,
hook_score=hook_score,
cross_modal_sync_score=cross_modal_score,
composite_score=composite_score,
virality_probability=virality_prob,
predicted_views=predicted_views,
predicted_view_range=view_range,
predicted_watch_through=virality_prob * profile.predicted_watch_through,
predicted_loop_rate=virality_prob * profile.predicted_loop_rate,
predicted_share_rate=virality_prob * profile.predicted_share_rate,
virality_tier=tier,
confidence_level=confidence,
recommendation=recommendation,
strategy=strategy
)
self.scoring_cache[variant_id] = score
return score
def score_variants_batch(
self,
variants: List[Tuple[np.ndarray, str]], # (audio, variant_id)
platform: Platform,
parallel: bool = True
) -> List[AudioVariantScore]:
"""
Score multiple variants in batch with optional parallelization.
Args:
variants: List of (audio_array, variant_id) tuples
platform: Target platform
parallel: Use parallel processing if True
Returns:
List of AudioVariantScore objects, sorted by virality_probability
"""
scores = []
for audio, variant_id in variants:
score = self.score_single_variant(
audio,
platform,
variant_id,
strategy=f"batch_{len(scores)}"
)
scores.append(score)
# Sort by virality probability (descending)
scores.sort(key=lambda x: x.virality_probability, reverse=True)
return scores
def select_best_variant(
self,
scores: List[AudioVariantScore],
selection_strategy: str = "highest_virality"
) -> AudioVariantScore:
"""
Select best variant based on strategy.
Args:
scores: List of scored variants
selection_strategy: Selection method
- "highest_virality": Pick highest virality probability
- "balanced": Balance virality with consistency
- "exploration": Favor experimental variants
Returns:
Selected AudioVariantScore
"""
if not scores:
raise ValueError("No variants to select from")
if selection_strategy == "highest_virality":
return scores[0] # Already sorted
elif selection_strategy == "balanced":
# Weight by virality * confidence
weights = [
s.virality_probability * {"very_high": 1.0, "high": 0.9, "medium": 0.7, "low": 0.5}[s.confidence_level]
for s in scores
]
best_idx = np.argmax(weights)
return scores[best_idx]
elif selection_strategy == "exploration":
# Favor WARM/COLD for exploration
exploration_candidates = [s for s in scores if s.virality_tier != ViralityTier.HOT]
if exploration_candidates:
return exploration_candidates[0]
return scores[0]
return scores[0]
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 = 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"""
return 0.82
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))
if 6 <= dynamic_range_db <= 12:
return 1.0
elif dynamic_range_db < 6:
return 0.6
else:
return 0.7
def _score_spatial_quality(self, audio: np.ndarray, profile: PlatformAudioProfile) -> float:
"""Score spatial/stereo quality"""
if audio.ndim == 1:
return 0.9 if profile.spatial.mono_compatibility_required else 0.6
else:
correlation = np.corrcoef(audio[0], audio[1])[0, 1]
if 0.3 <= correlation <= 0.7:
return 1.0
else:
return 0.7
def _calculate_loopability(self, audio: np.ndarray) -> float:
"""Calculate how well audio loops seamlessly"""
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 _score_hook_placement(self, audio: np.ndarray, profile: PlatformAudioProfile) -> float:
"""Score hook timing and placement"""
hook_region = audio[:int(3 * 44100)]
hook_energy = np.mean(hook_region**2)
overall_energy = np.mean(audio**2)
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 _score_cross_modal_sync(
self,
audio: np.ndarray,
profile: PlatformAudioProfile,
visual_sync_data: Optional[Dict[str, Any]]
) -> float:
"""Score cross-modal synchronization quality"""
if not visual_sync_data:
return 0.75
sync_profile = profile.cross_modal_sync
score = 0.5
# Check if beat drops align with visual hooks
visual_hooks = visual_sync_data.get("visual_hook_timestamps", [])
audio_hooks = visual_sync_data.get("audio_hook_timestamps", [])
if visual_hooks and audio_hooks:
# Calculate alignment quality
min_distances = []
for v_hook in visual_hooks:
distances = [abs(v_hook - a_hook) for a_hook in audio_hooks]
min_distances.append(min(distances))
avg_alignment = np.mean(min_distances)
# Good alignment: within 100ms
if avg_alignment <= 0.1:
score = 1.0
elif avg_alignment <= 0.3:
score = 0.85
else:
score = 0.7
return score
class PlatformAudioProfileManager:
"""Main manager for platform-specific audio profiles with full 5M+ inevitability"""
def __init__(self):
self.profiles: Dict[Platform, PlatformAudioProfile] = {}
self.trend_engine = TrendIngestionEngine()
self.rl_interface = RLIntegrationInterface(self)
self.variant_scorer = VariantScoringEngine(self)
self.memory_manager = None
self._initialize_profiles()
self._start_auto_trend_updates()
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,
true_peak_dbtp=-1.0,
short_term_lufs_range=(-16.0, -10.0),
momentary_lufs_max=-8.0,
platform_normalization_lufs=-14.0,
headroom_db=1.0
),
compression=CompressionProfile(
threshold_db=-18.0,
ratio=4.0,
attack_ms=5.0,
release_ms=50.0,
knee_db=6.0,
makeup_gain_db=3.0,
sidechain_filter_hz=80.0
),
frequency_bands=FrequencyBandProfile(
low_cut_hz=30.0,
low_boost_hz=(80.0, 150.0),
mid_presence_hz=(2000.0, 4000.0),
high_shelf_hz=8000.0,
problem_frequencies=[(250.0, 400.0), (3000.0, 3500.0)],
phone_safe_low_hz=50.0,
earbuds_sweet_spot_hz=(200.0, 8000.0)
),
spatial=SpatialProfile(
primary_format="stereo",
mono_compatibility_required=True,
stereo_width_percent=120.0,
center_content_db=3.0,
side_content_limit_db=-6.0,
phone_speaker_mode="optimized",
earbuds_enhancement=True
),
beat_matching=BeatMatchingProfile(
optimal_bpm_range=(120, 150),
viral_bpm_peaks=[128, 140, 145],
beat_drop_timing_sec=[0.5, 3.0, 7.0],
loop_point_requirements=[15.0, 30.0, 60.0],
rhythm_patterns=["trap-hi-hats", "edm-buildup", "drill-pattern"],
sync_to_trending_audio=True
),
cross_modal_sync=CrossModalSyncProfile(
visual_audio_sync_bonus=1.45,
caption_hook_audio_boost=2.5,
scene_transition_audio_weight=1.3,
audio_leads_visual_ms=100,
beat_on_scene_cut=True,
silence_on_text_overlay=False,
energy_boost_on_visual_climax=True,
bass_hit_on_product_reveal=True,
duck_audio_during_text_db=-6.0,
boost_voice_during_text_db=3.0,
maintain_beat_during_text=True
),
max_duration_sec=180.0,
min_duration_sec=3.0,
recommended_duration_sec=15.0,
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
self.profiles[Platform.INSTAGRAM_REELS] = PlatformAudioProfile(
platform=Platform.INSTAGRAM_REELS,
loudness=LoudnessProfile(
target_lufs=-13.0,
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,
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,
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,
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
),
cross_modal_sync=CrossModalSyncProfile(
visual_audio_sync_bonus=1.35,
caption_hook_audio_boost=2.0,
scene_transition_audio_weight=1.25,
audio_leads_visual_ms=80,
beat_on_scene_cut=True,
silence_on_text_overlay=False,
energy_boost_on_visual_climax=True,
bass_hit_on_product_reveal=True,
duck_audio_during_text_db=-5.0,
boost_voice_during_text_db=2.5,
maintain_beat_during_text=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=[],
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
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,
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
),
cross_modal_sync=CrossModalSyncProfile(
visual_audio_sync_bonus=1.30,
caption_hook_audio_boost=1.8,
scene_transition_audio_weight=1.2,
audio_leads_visual_ms=120,
beat_on_scene_cut=True,
silence_on_text_overlay=False,
energy_boost_on_visual_climax=True,
bass_hit_on_product_reveal=False,
duck_audio_during_text_db=-4.0,
boost_voice_during_text_db=2.0,
maintain_beat_during_text=False
),
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=[],
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 _start_auto_trend_updates(self):
"""Start automatic trend updating background process"""
# In production, this would be an async background task
pass
async def update_profile_from_trending_data(
self,
platform: Platform,
force_update: bool = False
):
"""
Update profile with latest trending data.
Args:
platform: Platform to update
force_update: Force update even if recent
"""
profile = self.profiles[platform]
# Check if update needed
time_since_update = datetime.now() - profile.last_profile_update
if not force_update and time_since_update < timedelta(minutes=30):
return
# Fetch trending data
trending = await self.trend_engine.fetch_trending_data(platform)
# Update BPM preferences
profile.beat_matching.viral_bpm_peaks = trending.trending_bpms
profile.beat_matching.last_trend_update = trending.timestamp
profile.beat_matching.trend_confidence = trending.confidence_score
# Update engagement predictions
alpha = 0.2
profile.predicted_ctr = (1 - alpha) * profile.predicted_ctr + alpha * (trending.avg_views / 30_000_000)
profile.predicted_watch_through = (1 - alpha) * profile.predicted_watch_through + alpha * trending.avg_watch_through
profile.predicted_loop_rate = (1 - alpha) * profile.predicted_loop_rate + alpha * trending.avg_loop_rate
profile.predicted_share_rate = (1 - alpha) * profile.predicted_share_rate + alpha * trending.avg_share_rate
# Store trending snapshot
profile.trending_data_queue.append(trending)
profile.last_profile_update = datetime.now()
def get_platform_profile(self, platform: Platform) -> PlatformAudioProfile:
"""Get complete audio profile for specified platform"""
if platform not in self.profiles:
raise ValueError(f"Unsupported platform: {platform}")
return self.profiles[platform]
def generate_optimized_variants(
self,
base_audio: np.ndarray,
platform: Platform,
num_variants: int = 5,
visual_sync_data: Optional[Dict[str, Any]] = None,
niche: Optional[str] = None
) -> List[AudioVariantScore]:
"""
Generate and score multiple optimized audio variants.
Args:
base_audio: Base audio to create variants from
platform: Target platform
num_variants: Number of variants to generate
visual_sync_data: Optional visual hook timing data
niche: Optional content niche
Returns:
List of AudioVariantScore objects, ranked by virality
"""
variants = []
for i in range(num_variants):
variant_audio = base_audio.copy()
strategy = ""
# Different optimization strategies
if i == 0:
# Max loudness aggressive
variant_audio = self._apply_aggressive_compression(variant_audio)
strategy = "max_loudness"
elif i == 1:
# Dynamic emotional preservation
variant_audio = self._preserve_dynamics(variant_audio)
strategy = "emotional_dynamic"
elif i == 2:
# Beat-optimized
variant_audio = self._enhance_beat_alignment(variant_audio, platform)
strategy = "beat_optimized"
elif i == 3:
# Phone speaker optimized
variant_audio = self._optimize_for_phone_speakers(variant_audio, platform)
strategy = "phone_optimized"
else:
# RL-guided exploration
action = self.rl_interface.sample_action(platform, exploration_rate=0.3)
variant_audio = self._apply_rl_action(variant_audio, platform, action)
strategy = "rl_exploration"
# Normalize to platform target
variant_audio, _ = self.normalize_loudness(variant_audio, platform)
variants.append((variant_audio, f"{platform.value}_v{i}_{strategy}"))
# Score all variants
scored_variants = self.variant_scorer.score_variants_batch(variants, platform)
return scored_variants
def select_best_variant_for_deployment(
self,
scored_variants: List[AudioVariantScore],
min_virality_threshold: float = 0.80
) -> Optional[AudioVariantScore]:
"""
Select best variant that meets virality threshold.
Args:
scored_variants: List of scored variants
min_virality_threshold: Minimum virality probability required
Returns:
Best variant above threshold, or None
"""
candidates = [v for v in scored_variants if v.virality_probability >= min_virality_threshold]
if not candidates:
return None
return self.variant_scorer.select_best_variant(candidates, "highest_virality")
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"""
profile = self.profiles[platform]
target_lufs = profile.loudness.target_lufs
current_rms = np.sqrt(np.mean(audio_data**2))
current_lufs = -23.0 + 20 * np.log10(current_rms + 1e-10)
gain_db = target_lufs - current_lufs
gain_linear = 10 ** (gain_db / 20.0)
normalized_audio = audio_data * gain_linear
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)
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 learn_from_video_performance(
self,
platform: Platform,
variant_id: str,
audio_characteristics: Dict[str, Any],
actual_metrics: Dict[str, float]
):
"""
Update profile and RL policy based on actual video performance.
Args:
platform: Platform where video was posted
variant_id: ID of audio variant used
audio_characteristics: Technical audio features
actual_metrics: Actual engagement metrics
"""
profile = self.profiles[platform]
views = actual_metrics.get("views", 0)
# Store performance history
performance_record = {
"timestamp": datetime.now(),
"variant_id": variant_id,
"audio_char": audio_characteristics,
"metrics": actual_metrics
}
profile.performance_history.append(performance_record)
# Create viral pattern if 5M+ views
if views >= 5_000_000:
new_pattern = ViralAudioPattern(
pattern_id=f"{variant_id}_viral_{int(views/1000000)}M",
hook_timing_sec=audio_characteristics.get("hook_timings", [0.5, 3.0]),
emotional_peak_timing_sec=audio_characteristics.get("peak_timings", [7.0]),
energy_curve=audio_characteristics.get("energy_curve", [0.8] * 16),
loopability_score=actual_metrics.get("loop_rate", 0.5),
shareability_score=actual_metrics.get("share_rate", 0.1),
engagement_multiplier=views / 5_000_000,
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]
)
profile.viral_patterns.insert(0, new_pattern)
profile.viral_patterns = profile.viral_patterns[:15]
# Update RL policy
reward = self.rl_interface.compute_reward(platform, actual_metrics)
self.rl_interface.update_policy(platform, reward)
# Update predictions
alpha = 0.15
if "watch_through_rate" in actual_metrics:
profile.predicted_watch_through = (
(1 - alpha) * profile.predicted_watch_through +
alpha * actual_metrics["watch_through_rate"]
)
if "loop_rate" in actual_metrics:
profile.predicted_loop_rate = (
(1 - alpha) * profile.predicted_loop_rate +
alpha * actual_metrics["loop_rate"]
)
# Update compression if successful
if actual_metrics.get("watch_through_rate", 0) > profile.predicted_watch_through * 1.2:
if "compression_ratio" in audio_characteristics:
profile.compression.ratio = (
0.85 * profile.compression.ratio +
0.15 * audio_characteristics["compression_ratio"]
)
def get_cross_modal_sync_rules(self, platform: Platform) -> CrossModalSyncProfile:
"""Get cross-modal synchronization rules for platform"""
return self.profiles[platform].cross_modal_sync
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"""
profile = self.profiles[platform]
sync_rules = profile.cross_modal_sync
aligned_audio = audio_data.copy()
for visual_time in visual_hook_timestamps:
# Add energy boost at visual hook timing
aligned_audio = self._add_audio_emphasis_at_time(
aligned_audio,
visual_time - sync_rules.audio_leads_visual_ms / 1000.0,
emphasis_db=2.5
)
return aligned_audio
def _apply_aggressive_compression(self, audio: np.ndarray) -> np.ndarray:
"""Apply aggressive compression for maximum loudness"""
threshold = 0.25
ratio = 6.0
compressed = np.where(
np.abs(audio) > threshold,
np.sign(audio) * (threshold + (np.abs(audio) - threshold) / ratio),
audio
)
return compressed * 1.5
def _preserve_dynamics(self, audio: np.ndarray) -> np.ndarray:
"""Preserve dynamic range for emotional impact"""
threshold = 0.4
ratio = 2.5
compressed = np.where(
np.abs(audio) > threshold,
np.sign(audio) * (threshold + (np.abs(audio) - threshold) / ratio),
audio
)
return compressed
def _enhance_beat_alignment(self, audio: np.ndarray, platform: Platform) -> np.ndarray:"""
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