import os import shutil import json import tempfile import gradio as gr from pydub import AudioSegment, silence from pydub.effects import low_pass_filter, high_pass_filter from tqdm import tqdm from groq import Groq import logging from langsmith import traceable from langchain.callbacks import LangChainTracer from dotenv import load_dotenv import random from temp_choose import * load_dotenv() logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) logger.addHandler(logging.StreamHandler()) logger.addHandler(logging.FileHandler("log.txt")) @traceable(run_type="llm", name="groq_call") def make_groq_call(stems, song_name, p, section_type=None, bpm=120, bars=16): """ Make a call to the Groq API to get music production instructions. Args: stems (list): List of available stem files song_name (str): Name of the song p (float): Variation parameter (0-1) section_type (str, optional): Specific section to generate variants for bpm (int): Beats per minute bars (int): Number of bars Returns: dict: JSON response with production instructions """ client = Groq(api_key=os.getenv("GROQ_API_KEY")) # Customize prompt based on whether we're generating a full track or section variants if section_type: system_content = """You are a very experienced music producer and analyst, with a deep understanding of music theory and production techniques and arrangement, with specific expertise in creating popular EDM/ dance music. You are given a set of audio stems and asked to create multiple variants of a specific section of a track. Your task is to provide detailed instructions on how to arrange and process the stems for each variant. You will be given audio stems and asked to create multiple variants of a specific section of a track. For each variant, return detailed instructions on how to arrange and process the stems. Be creative and make each variant sound distinct while maintaining a coherent musical style.""" user_content = f"""I need 4 different variants for the {section_type} section of a track named "{song_name}". Available stems: {stems} BPM: {bpm} Length of the section: {bars} bars For each variant, please provide specific instructions on: 1. Which stems to include 2. What audio operations to apply (filters, fades, etc.) 3. How the stems should be arranged Make the variants diverse but coherent, with variation level p={p} (0=minimal variation, 1=maximum variation). Return your response as a JSON object with this structure: {{ "variant1": {{ "stems": ["stem1.wav", "stem2.wav"], "operations": [ {{"stem": "stem1.wav", "operation": "low_pass_filter", "value": 500}}, {{"stem": "stem2.wav", "operation": "fade_in", "value": 1000}} ], "overlay": true, "description": "A brief description of this variant" }}, "variant2": {{ ... }}, "variant3": {{ ... }}, "variant4": {{ ... }} }} """ else: system_content = """You are a very experienced music producer and analyst. For a given audio folder, that has the instruments, are combined to make a loop, and make some variations of it as well. After analyzing the code, you are supposed to return the code containing the different functions for producing the full track.""" user_content = f"""Now, you have a new song {song_name}, which has the following contents: {stems} BPM: {bpm} Bars: {bars} Return the code as per discussed in example. Make proper arrangements, for best groovy music. Create 3 variations and return code in JSON. And make sure to use as many instruments possible in each variation. NOTE: And at least once, the MAIN loop should have ALL stems. Like the variations should be like the main full loop, with some adjustments according to value of p={p} (p will remain in between 0-1; 0 means no variation in loop and 1 means high variation in loop), and then another variation can be a two times repeat of the full loop, with some effects in second time. Make intro better, with some more instruments. The main loop should have ALL wav files, don't exclude any please. Return proper JSON, with all the keys and values. """ completion = client.chat.completions.create( model="gemma2-9b-it", messages=[ {"role": "system", "content": system_content}, {"role": "user", "content": user_content}, ], temperature=1, top_p=1, stream=False, response_format={"type": "json_object"}, stop=None, ) print(completion.choices[0].message.content) return json.loads(completion.choices[0].message.content) def rename_files_remove_spaces(folder): """Rename all files in the folder by removing spaces from filenames""" files_renamed = 0 for file in os.listdir(folder): if " " in file: old_path = os.path.join(folder, file) new_file = file.replace(" ", "") new_path = os.path.join(folder, new_file) # Only rename if the new file doesn't already exist if not os.path.exists(new_path): os.rename(old_path, new_path) print(f"Renamed: {file} → {new_file}") files_renamed += 1 print(f"Total files renamed: {files_renamed}") def load_audio_files(folder): """Load all WAV files from a folder into memory""" files = sorted([f for f in os.listdir(folder) if f.endswith(".wav")]) stems = {} for file in tqdm(files, desc="Loading audio files"): path = os.path.join(folder, file) audio = AudioSegment.from_wav(path) stems[file] = audio return stems def get_stems(folder): """Get a list of all WAV files in a folder""" files = sorted([f for f in os.listdir(folder) if f.endswith(".wav")]) return files def apply_audio_operation(audio, operation, value): """Apply various audio operations to an AudioSegment""" if operation == "low_pass_filter" and isinstance(value, float): return low_pass_filter(audio, value) elif operation == "high_pass_filter" and isinstance(value, float): return high_pass_filter(audio, value) elif operation == "fade_in" and isinstance(value, int): return audio.fade_in(value) elif operation == "fade_out" and isinstance(value, int): return audio.fade_out(value) elif operation == "reverb" and isinstance(value, float): # Simple reverb simulation by adding delayed and attenuated copies result = audio for delay in [50, 100, 150, 200]: attenuated = audio - (value * 10) # Reduce volume based on reverb value delayed = AudioSegment.silent(duration=delay) + attenuated result = result.overlay(delayed) return result elif operation == "delay" and isinstance(value, int): # Simulate delay by adding a delayed copy result = audio delayed = AudioSegment.silent(duration=value) + (audio - 6) # -6dB for the echo return result.overlay(delayed) elif operation == "distortion" and isinstance(value, float): # Simulate distortion by adding some limiting/clipping gain = 1.0 + (value * 5) # Boost the gain based on distortion value return audio + (gain) # Add gain in dB elif operation == "pitch_shift" and isinstance(value, float): # Note: pydub doesn't natively support pitch shifting print(f"Warning: Pitch shift not implemented, value: {value}") return audio elif operation == "volume" and isinstance(value, float): # Adjust volume by dB return audio + value return audio def create_section_from_json(section_config, stems): """Create an audio section based on JSON configuration""" if not section_config: print("No configuration found for section") return AudioSegment.empty() section_stems = [] print(section_config) for stem_name in section_config["stems"]: # First try the original stem name if stem_name in stems: section_stems.append(stems[stem_name]) else: # Try the name without spaces no_spaces_name = stem_name.replace(" ", "") if no_spaces_name in stems: section_stems.append(stems[no_spaces_name]) else: print(f"Warning: Stem {stem_name} not found (with or without spaces)") # Apply operations to stems processed_stems = {name: audio for name, audio in stems.items()} for op in section_config.get("operations", []): stem_name = op["stem"] operation = op["operation"] value = op["value"] # Check both original and no-spaces versions stem_key = None if stem_name in processed_stems: stem_key = stem_name else: no_spaces_name = stem_name.replace(" ", "") if no_spaces_name in processed_stems: stem_key = no_spaces_name if stem_key and operation != "overlay": processed_stems[stem_key] = apply_audio_operation( processed_stems[stem_key], operation, value ) # Collect the processed stems for this section final_stems = [] for stem_name in section_config["stems"]: if stem_name in processed_stems: final_stems.append(processed_stems[stem_name]) else: no_spaces_name = stem_name.replace(" ", "") if no_spaces_name in processed_stems: final_stems.append(processed_stems[no_spaces_name]) # Overlay stems if specified # if section_config.get("overlay", True) and final_stems: result = final_stems[0] for stem in final_stems[1:]: result = result.overlay(stem) # return result # Remove silences longer than 1.5 seconds (1500 ms) silence_thresh = result.dBFS - 16 # silent_chunks = silence.detect_silence( result, min_silence_len=1500, silence_thresh=silence_thresh ) segments = [] prev_end = 0 for start, end in silent_chunks: if prev_end < start: segments.append(result[prev_end:start]) prev_end = end segments.append(result[prev_end:]) result = sum(segments) return result return AudioSegment.empty() def generate_section_variants( stems_folder, audio_stems, section_type, bpm, bars,progress, p=0.5 ): """ Generate multiple variants for a specific section Args: stems_folder (str): Path to folder containing stem files section_type (str): Type of section (intro, verse, chorus, etc.) bpm (int): Beats per minute bars (int): Number of bars p (float): Variation parameter (0-1) Returns: dict: Dictionary of variant audio segments and their descriptions """ stems = get_stems(stems_folder) try: llm_response = make_groq_call( stems, f"{section_type} section", p, section_type=section_type, bpm=bpm, bars=bars, ) except Exception as e: logger.error(f"Error generating variants for {section_type} section: {e}") return {} # Load audio files if not audio_stems: print("No stems loaded.") return {} progress(0.2, desc=f"Generating structure and effects for variants of {section_type}...") # Create each variant # hERE IS tHE validation for the LLM RESPONSE if not isinstance(llm_response, dict): logger.error(f"Invalid LLM response: {llm_response}") # Check if the response is a valid JSON object try: if isinstance(llm_response, str): llm_response = json.loads(llm_response) elif isinstance(llm_response, dict): llm_response = llm_response except json.JSONDecodeError: logger.error(f"Tried Converting the LLM response to JSON, but failed: {llm_response}") variants = {} for variant_key in progress.tqdm(llm_response, desc="Getting variants as per AI arranegments..."): if variant_key.startswith("variant"): variant_config = llm_response[variant_key] logger.info(f" gsv: Variant config: {variant_config}") audio = create_section_from_json(variant_config, audio_stems) description = variant_config.get( "description", f"Variant {variant_key[-1]}" ) list_stems = variant_config.get("stems", []) variants[variant_key] = { "audio": audio, "description": description, "config": variant_config, "stems": list_stems } return variants def create_full_track( sections_folder, audio_stems, selected_variants, crossfade_ms=500 ): """ Create a full track from selected variants Args: sections_folder (dict): Dict mapping section names to their folder paths selected_variants (dict): Dict mapping section names to their selected variant configs crossfade_ms (int): Crossfade duration in milliseconds Returns: AudioSegment: The final track """ final_track = None # Define the order of sections section_order = [ "intro", "buildup", "full_loop", "breakdown", "bridge", "buildup2", "drop2", "breakdown2", "outro", ] # Process each section in order for section_name in section_order: if section_name not in selected_variants: continue # Get the selected variant config variant_config = selected_variants[section_name] # Create audio for this section section_audio = create_section_from_json(variant_config, audio_stems) # Add to final track if final_track is None: final_track = section_audio else: final_track = final_track.append(section_audio, crossfade=crossfade_ms) return final_track def create_intro(llm_answer, stems): """Create intro section from LLM answer""" return create_section_from_json(llm_answer.get("create_intro", {}), stems) def create_variation1(llm_answer, stems): """Create variation1 section from LLM answer""" return create_section_from_json(llm_answer.get("create_variation1", {}), stems) def create_full_loop(llm_answer, stems): """Create full loop section from LLM answer""" return create_section_from_json(llm_answer.get("create_full_loop", {}), stems) def create_variation2(llm_answer, stems): """Create variation2 section from LLM answer""" return create_section_from_json(llm_answer.get("create_variation2", {}), stems) def create_variation3(llm_answer, stems): """Create variation3 section from LLM answer""" return create_section_from_json(llm_answer.get("create_variation3", {}), stems) def create_outro(llm_answer, stems): """Create outro section from LLM answer""" return create_section_from_json(llm_answer.get("create_outro", {}), stems) def calculate_duration(bpm, bars): """Calculate duration in seconds for a given BPM and number of bars""" # Assuming 4/4 time signature (4 beats per bar) beats_per_bar = 4 duration_seconds = (bars * beats_per_bar * 60) / bpm return duration_seconds def get_formatted_duration(seconds): """Format duration in seconds to MM:SS format""" minutes = int(seconds // 60) seconds = int(seconds % 60) return f"{minutes}:{seconds:02d}" def export_section_variants(variants, output_folder, section_name): """Export section variants to audio files""" if not os.path.exists(output_folder): os.makedirs(output_folder) file_paths = {} for variant_key, variant_data in variants.items(): output_path = os.path.join(output_folder, f"{section_name}_{variant_key}.wav") variant_data["audio"].export(output_path, format="wav") file_paths[variant_key] = output_path return file_paths def edm_arrangement_tab(): with gr.Blocks() as iface: gr.Markdown("# 🌚 Interactive EDM Arrangement Tool") with gr.Row(): variation = gr.Radio( choices=list(arrangements.keys()), value="High Energy Flow", label="Choose Arrangement Variation", ) out_plot = gr.Plot(label="Arrangement Diagram", value=load_variation("High Energy Flow")) variation.change(fn=load_variation, inputs=variation, outputs=out_plot) iface.load(fn=load_variation, inputs=[variation], outputs=[out_plot]) with gr.Accordion("🎻 Edit Section Parameters", open=False): for i, (bar, tempo, name, length, curve) in enumerate(arrangement): with gr.Row(): gr.Markdown(f"**{name}**") bar_slider = gr.Slider( minimum=0, maximum=300, value=bar, label="Start Bar" ) tempo_slider = gr.Slider( minimum=20, maximum=100, value=tempo, label="Volume" ) length_slider = gr.Slider( minimum=1, maximum=64, value=length, label="Length" ) curve_selector = gr.Radio( choices=["Flat", "Linear", "-ve Linear"], value=curve, label="Curve Type", ) update_btn = gr.Button("Update") update_btn.click( fn=update_section, inputs=[ gr.Number(value=i, visible=False), bar_slider, tempo_slider, length_slider, curve_selector, ], outputs=[out_plot], ) with gr.Accordion("➕ Insert New Section", open=False): new_index = gr.Number(value=0, label="Insert At Index") new_name = gr.Textbox(label="Section Name", value="New Section") new_bar = gr.Slider(minimum=0, maximum=300, value=0, label="Start Bar") new_tempo = gr.Slider(minimum=20, maximum=100, value=50, label="Volume") new_length = gr.Slider(minimum=1, maximum=64, value=8, label="Length") new_curve = gr.Radio( choices=["Flat", "Linear", "-ve Linear"], value="Flat", label="Curve Type" ) insert_btn = gr.Button("Insert Section") insert_btn.click( fn=insert_section, inputs=[new_index, new_name, new_bar, new_tempo, new_length, new_curve], outputs=[out_plot], ) gr.Markdown("## ✅ Finalise Your Arrangement") final_btn = gr.Button("Finalise and Export JSON") final_output = gr.Textbox(label="Final Arrangement JSON", lines=15) final_btn.click(fn=finalise, outputs=final_output)