"""Scenario generation for campaign simulations.""" import random from src.models import Campaign class ScenarioGenerator: """Generates campaign scenarios for simulation.""" @staticmethod def generate_random_campaigns( count: int, min_volume: int = 1000, max_volume: int = 10000, time_window_minutes: int = 60, seed: int | None = None, ) -> list[Campaign]: """ Generate random campaigns. Args: count: Number of campaigns to generate min_volume: Minimum campaign volume max_volume: Maximum campaign volume time_window_minutes: Time window for campaign starts (0 to this value) seed: Random seed for reproducibility Returns: List of randomly generated campaigns """ if seed is not None: random.seed(seed) campaigns: list[Campaign] = [] for i in range(count): volume = random.randint(min_volume, max_volume) start_minute = random.randint(0, time_window_minutes) campaign = Campaign( id=f"campaign_{i + 1}", name=f"Campaign {i + 1}", total_volume=volume, start_minute=start_minute, ) campaigns.append(campaign) return campaigns @staticmethod def simultaneous_start(count: int = 3, volume: int = 5000) -> list[Campaign]: """ Generate campaigns that all start at minute 0. Args: count: Number of campaigns volume: Volume for each campaign Returns: List of campaigns starting simultaneously """ campaigns: list[Campaign] = [] for i in range(count): campaign = Campaign( id=f"campaign_{i + 1}", name=f"Campaign {i + 1}", total_volume=volume, start_minute=0, ) campaigns.append(campaign) return campaigns @staticmethod def staggered_start( count: int = 3, volume: int = 5000, interval_minutes: int = 10 ) -> list[Campaign]: """ Generate campaigns that start at regular intervals. Args: count: Number of campaigns volume: Volume for each campaign interval_minutes: Minutes between each campaign start Returns: List of campaigns with staggered starts """ campaigns: list[Campaign] = [] for i in range(count): campaign = Campaign( id=f"campaign_{i + 1}", name=f"Campaign {i + 1}", total_volume=volume, start_minute=i * interval_minutes, ) campaigns.append(campaign) return campaigns @staticmethod def rush_hour( small_count: int = 5, small_volume: int = 2000, large_count: int = 2, large_volume: int = 15000, ) -> list[Campaign]: """ Generate a rush hour scenario with many small and few large campaigns. Args: small_count: Number of small campaigns small_volume: Volume for small campaigns large_count: Number of large campaigns large_volume: Volume for large campaigns Returns: List of campaigns simulating rush hour """ campaigns: list[Campaign] = [] # Add small campaigns all starting at minute 0 for i in range(small_count): campaign = Campaign( id=f"small_{i + 1}", name=f"Small {i + 1}", total_volume=small_volume, start_minute=0, ) campaigns.append(campaign) # Add large campaigns starting slightly after for i in range(large_count): campaign = Campaign( id=f"large_{i + 1}", name=f"Large {i + 1}", total_volume=large_volume, start_minute=random.randint(0, 5), ) campaigns.append(campaign) return campaigns @staticmethod def early_vs_late( early_volume: int = 8000, late_volume: int = 8000, delay_minutes: int = 20 ) -> list[Campaign]: """ Generate scenario with one early campaign and one late campaign. Useful for testing how strategies handle campaigns joining mid-execution. Args: early_volume: Volume for early campaign late_volume: Volume for late campaign delay_minutes: How many minutes late campaign is delayed Returns: List with two campaigns """ return [ Campaign( id="early", name="Early Campaign", total_volume=early_volume, start_minute=0, ), Campaign( id="late", name="Late Campaign", total_volume=late_volume, start_minute=delay_minutes, ), ] @staticmethod def varied_sizes( count: int = 4, ) -> list[Campaign]: """ Generate campaigns with exponentially increasing sizes. Args: count: Number of campaigns Returns: List of campaigns with varied sizes """ campaigns: list[Campaign] = [] base_volume = 1000 for i in range(count): volume = base_volume * (2**i) campaign = Campaign( id=f"campaign_{i + 1}", name=f"Campaign {i + 1} ({volume} emails)", total_volume=volume, start_minute=0, ) campaigns.append(campaign) return campaigns def get_preset_scenarios() -> dict[str, list[Campaign]]: """ Get all preset scenarios. Returns: Dict mapping scenario name to list of campaigns """ generator = ScenarioGenerator() return { "Simultaneous Start (3 campaigns, 5K each)": generator.simultaneous_start( count=3, volume=5000 ), "Staggered Start (3 campaigns, 5K each, 10min apart)": generator.staggered_start( count=3, volume=5000, interval_minutes=10 ), "Rush Hour (5 small + 2 large)": generator.rush_hour( small_count=5, small_volume=2000, large_count=2, large_volume=15000, ), "Early vs Late (8K each, 20min delay)": generator.early_vs_late( early_volume=8000, late_volume=8000, delay_minutes=20 ), "Varied Sizes (4 campaigns, exponential)": generator.varied_sizes(count=4), } def get_scenario_names() -> list[str]: """Get list of available preset scenario names.""" return list(get_preset_scenarios().keys()) def get_scenario_by_name(name: str) -> list[Campaign]: """ Get a preset scenario by name. Args: name: Scenario name Returns: List of campaigns for the scenario Raises: KeyError: If scenario name not found """ scenarios = get_preset_scenarios() if name not in scenarios: raise KeyError(f"Scenario '{name}' not found") return scenarios[name]