Swap to HF diffusers

README.md

@@ -8,6 +8,7 @@ sdk_version: 3.27.0
 app_file: app.py
 pinned: false
 license: bigscience-openrail-m
+duplicated_from: haoheliu/audioldm-text-to-audio-generation
 ---
 
 Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -1,197 +1,139 @@
 import gradio as gr
-import numpy as np
-from audioldm import text_to_audio, build_model
+import torch
+from diffusers import AudioLDMPipeline
 from share_btn import community_icon_html, loading_icon_html, share_js
 
-model_id="haoheliu/AudioLDM-S-Full"
+from transformers import AutoProcessor, ClapModel
 
-audioldm = None
-current_model_name = None
 
-# def predict(input, history=[]):
-#     # tokenize the new input sentence
-#     new_user_input_ids = tokenizer.encode(input + tokenizer.eos_token, return_tensors='pt')
+# make Space compatible with CPU duplicates
+if torch.cuda.is_available():
+    device = "cuda"
+    torch_dtype = torch.float16
+else:
+    device = "cpu"
+    torch_dtype = torch.float32
 
-#     # append the new user input tokens to the chat history
-#     bot_input_ids = torch.cat([torch.LongTensor(history), new_user_input_ids], dim=-1)
+# load the diffusers pipeline
+repo_id = "cvssp/audioldm-m-full"
+pipe = AudioLDMPipeline.from_pretrained(repo_id, torch_dtype=torch_dtype).to(device)
+pipe.unet = torch.compile(pipe.unet)
 
-#     # generate a response 
-#     history = model.generate(bot_input_ids, max_length=1000, pad_token_id=tokenizer.eos_token_id).tolist()
+# CLAP model (only required for automatic scoring)
+clap_model = ClapModel.from_pretrained("sanchit-gandhi/clap-htsat-unfused-m-full").to(device)
+processor = AutoProcessor.from_pretrained("sanchit-gandhi/clap-htsat-unfused-m-full")
 
-#     # convert the tokens to text, and then split the responses into lines
-#     response = tokenizer.decode(history[0]).split("<|endoftext|>")
-#     response = [(response[i], response[i+1]) for i in range(0, len(response)-1, 2)]  # convert to tuples of list
-#     return response, history
-  
-def text2audio(text, duration, guidance_scale, random_seed, n_candidates, model_name="audioldm-m-text-ft"):
-    global audioldm, current_model_name
-    
-    if audioldm is None or model_name != current_model_name:
-        audioldm=build_model(model_name=model_name)
-        current_model_name = model_name
-        
-    # print(text, length, guidance_scale)
-    waveform = text_to_audio(
-        latent_diffusion=audioldm,
-        text=text,
-        seed=random_seed,
-        duration=duration,
+generator = torch.Generator(device)
+
+
+def text2audio(text, negative_prompt, duration, guidance_scale, random_seed, n_candidates):
+    if text is None:
+        raise gr.Error("Please provide a text input.")
+
+    waveforms = pipe(
+        text,
+        audio_length_in_s=duration,
         guidance_scale=guidance_scale,
-        n_candidate_gen_per_text=int(n_candidates),
-    )  # [bs, 1, samples]
-    waveform = [
-        gr.make_waveform((16000, wave[0]), bg_image="bg.png") for wave in waveform
-    ]
-    # waveform = [(16000, np.random.randn(16000)), (16000, np.random.randn(16000))]
-    if(len(waveform) == 1):
-      waveform = waveform[0]
-    return waveform
+        negative_prompt=negative_prompt,
+        num_waveforms_per_prompt=n_candidates if n_candidates else 1,
+        generator=generator.manual_seed(int(random_seed)),
+    )["audios"]
+
+    if waveforms.shape[0] > 1:
+        waveform = score_waveforms(text, waveforms)
+    else:
+        waveform = waveforms[0]
+
+    return gr.make_waveform((16000, waveform), bg_image="bg.png")
+
 
-# iface = gr.Interface(fn=text2audio, inputs=[
-#         gr.Textbox(value="A man is speaking in a huge room", max_lines=1),
-#         gr.Slider(2.5, 10, value=5, step=2.5),
-#         gr.Slider(0, 5, value=2.5, step=0.5),
-#         gr.Number(value=42)
-#     ], outputs=[gr.Audio(label="Output", type="numpy"), gr.Audio(label="Output", type="numpy")],
-#                 allow_flagging="never"
-#                      )
-# iface.launch(share=True)
+def score_waveforms(text, waveforms):
+    inputs = processor(text=text, audios=list(waveforms), return_tensors="pt", padding=True)
+    inputs = {key: inputs[key].to(device) for key in inputs}
+    with torch.no_grad():
+        logits_per_text = clap_model(**inputs).logits_per_text  # this is the audio-text similarity score
+        probs = logits_per_text.softmax(dim=-1)  # we can take the softmax to get the label probabilities
+        most_probable = torch.argmax(probs)  # and now select the most likely audio waveform
+    waveform = waveforms[most_probable]
+    return waveform
 
 
 css = """
         a {
-            color: inherit;
-            text-decoration: underline;
-        }
-        .gradio-container {
+            color: inherit; text-decoration: underline;
+        } .gradio-container {
             font-family: 'IBM Plex Sans', sans-serif;
-        }
-        .gr-button {
-            color: white;
-            border-color: #000000;
-            background: #000000;
-        }
-        input[type='range'] {
+        } .gr-button {
+            color: white; border-color: #000000; background: #000000;
+        } input[type='range'] {
             accent-color: #000000;
-        }
-        .dark input[type='range'] {
+        } .dark input[type='range'] {
             accent-color: #dfdfdf;
-        }
-        .container {
-            max-width: 730px;
-            margin: auto;
-            padding-top: 1.5rem;
-        }
-        #gallery {
-            min-height: 22rem;
-            margin-bottom: 15px;
-            margin-left: auto;
-            margin-right: auto;
-            border-bottom-right-radius: .5rem !important;
-            border-bottom-left-radius: .5rem !important;
-        }
-        #gallery>div>.h-full {
+        } .container {
+            max-width: 730px; margin: auto; padding-top: 1.5rem;
+        } #gallery {
+            min-height: 22rem; margin-bottom: 15px; margin-left: auto; margin-right: auto; border-bottom-right-radius:
+            .5rem !important; border-bottom-left-radius: .5rem !important;
+        } #gallery>div>.h-full {
             min-height: 20rem;
-        }
-        .details:hover {
+        } .details:hover {
             text-decoration: underline;
-        }
-        .gr-button {
+        } .gr-button {
             white-space: nowrap;
-        }
-        .gr-button:focus {
-            border-color: rgb(147 197 253 / var(--tw-border-opacity));
-            outline: none;
-            box-shadow: var(--tw-ring-offset-shadow), var(--tw-ring-shadow), var(--tw-shadow, 0 0 #0000);
-            --tw-border-opacity: 1;
-            --tw-ring-offset-shadow: var(--tw-ring-inset) 0 0 0 var(--tw-ring-offset-width) var(--tw-ring-offset-color);
-            --tw-ring-shadow: var(--tw-ring-inset) 0 0 0 calc(3px var(--tw-ring-offset-width)) var(--tw-ring-color);
-            --tw-ring-color: rgb(191 219 254 / var(--tw-ring-opacity));
-            --tw-ring-opacity: .5;
-        }
-        #advanced-btn {
-            font-size: .7rem !important;
-            line-height: 19px;
-            margin-top: 12px;
-            margin-bottom: 12px;
-            padding: 2px 8px;
+        } .gr-button:focus {
+            border-color: rgb(147 197 253 / var(--tw-border-opacity)); outline: none; box-shadow:
+            var(--tw-ring-offset-shadow), var(--tw-ring-shadow), var(--tw-shadow, 0 0 #0000); --tw-border-opacity: 1;
+            --tw-ring-offset-shadow: var(--tw-ring-inset) 0 0 0 var(--tw-ring-offset-width)
+            var(--tw-ring-offset-color); --tw-ring-shadow: var(--tw-ring-inset) 0 0 0 calc(3px
+            var(--tw-ring-offset-width)) var(--tw-ring-color); --tw-ring-color: rgb(191 219 254 /
+            var(--tw-ring-opacity)); --tw-ring-opacity: .5;
+        } #advanced-btn {
+            font-size: .7rem !important; line-height: 19px; margin-top: 12px; margin-bottom: 12px; padding: 2px 8px;
             border-radius: 14px !important;
-        }
-        #advanced-options {
+        } #advanced-options {
             margin-bottom: 20px;
-        }
-        .footer {
-            margin-bottom: 45px;
-            margin-top: 35px;
-            text-align: center;
-            border-bottom: 1px solid #e5e5e5;
-        }
-        .footer>p {
-            font-size: .8rem;
-            display: inline-block;
-            padding: 0 10px;
-            transform: translateY(10px);
-            background: white;
-        }
-        .dark .footer {
+        } .footer {
+            margin-bottom: 45px; margin-top: 35px; text-align: center; border-bottom: 1px solid #e5e5e5;
+        } .footer>p {
+            font-size: .8rem; display: inline-block; padding: 0 10px; transform: translateY(10px); background: white;
+        } .dark .footer {
             border-color: #303030;
-        }
-        .dark .footer>p {
+        } .dark .footer>p {
             background: #0b0f19;
-        }
-        .acknowledgments h4{
-            margin: 1.25em 0 .25em 0;
-            font-weight: bold;
-            font-size: 115%;
-        }
-        #container-advanced-btns{
-            display: flex;
-            flex-wrap: wrap;
-            justify-content: space-between;
-            align-items: center;
-        }
-        .animate-spin {
+        } .acknowledgments h4{
+            margin: 1.25em 0 .25em 0; font-weight: bold; font-size: 115%;
+        } #container-advanced-btns{
+            display: flex; flex-wrap: wrap; justify-content: space-between; align-items: center;
+        } .animate-spin {
             animation: spin 1s linear infinite;
-        }
-        @keyframes spin {
+        } @keyframes spin {
             from {
                 transform: rotate(0deg);
-            }
-            to {
+            } to {
                 transform: rotate(360deg);
             }
-        }
-        #share-btn-container {
-            display: flex; padding-left: 0.5rem !important; padding-right: 0.5rem !important; background-color: #000000; justify-content: center; align-items: center; border-radius: 9999px !important; width: 13rem;
-            margin-top: 10px;
-            margin-left: auto;
-        }
-        #share-btn {
-            all: initial; color: #ffffff;font-weight: 600; cursor:pointer; font-family: 'IBM Plex Sans', sans-serif; margin-left: 0.5rem !important; padding-top: 0.25rem !important; padding-bottom: 0.25rem !important;right:0;
-        }
-        #share-btn * {
+        } #share-btn-container {
+            display: flex; padding-left: 0.5rem !important; padding-right: 0.5rem !important; background-color:
+            #000000; justify-content: center; align-items: center; border-radius: 9999px !important; width: 13rem;
+            margin-top: 10px; margin-left: auto;
+        } #share-btn {
+            all: initial; color: #ffffff;font-weight: 600; cursor:pointer; font-family: 'IBM Plex Sans', sans-serif;
+            margin-left: 0.5rem !important; padding-top: 0.25rem !important; padding-bottom: 0.25rem
+            !important;right:0;
+        } #share-btn * {
             all: unset;
-        }
-        #share-btn-container div:nth-child(-n+2){
-            width: auto !important;
-            min-height: 0px !important;
-        }
-        #share-btn-container .wrap {
+        } #share-btn-container div:nth-child(-n+2){
+            width: auto !important; min-height: 0px !important;
+        } #share-btn-container .wrap {
             display: none !important;
-        }
-        .gr-form{
+        } .gr-form{
             flex: 1 1 50%; border-top-right-radius: 0; border-bottom-right-radius: 0;
-        }
-        #prompt-container{
+        } #prompt-container{
             gap: 0;
-        }
-        #generated_id{
+        } #generated_id{
             min-height: 700px
-        }
-        #setting_id{
-          margin-bottom: 12px;
-          text-align: center;
-          font-weight: 900;
+        } #setting_id{
+          margin-bottom: 12px; text-align: center; font-weight: 900;
         }
 """
 iface = gr.Blocks(css=css)
@@ -202,56 +144,72 @@ with iface:
             <div style="text-align: center; max-width: 700px; margin: 0 auto;">
               <div
                 style="
-                  display: inline-flex;
-                  align-items: center;
-                  gap: 0.8rem;
-                  font-size: 1.75rem;
+                  display: inline-flex; align-items: center; gap: 0.8rem; font-size: 1.75rem;
                 "
               >
                 <h1 style="font-weight: 900; margin-bottom: 7px; line-height: normal;">
                   AudioLDM: Text-to-Audio Generation with Latent Diffusion Models
                 </h1>
-              </div>
-              <p style="margin-bottom: 10px; font-size: 94%">
-                <a href="https://arxiv.org/abs/2301.12503">[Paper]</a>  <a href="https://audioldm.github.io/">[Project page]</a>
+              </div> <p style="margin-bottom: 10px; font-size: 94%">
+                <a href="https://arxiv.org/abs/2301.12503">[Paper]</a> <a href="https://audioldm.github.io/">[Project
+                page]</a> <a href="https://huggingface.co/docs/diffusers/main/en/api/pipelines/audioldm">[🧨
+                Diffusers]</a>
               </p>
             </div>
         """
     )
-    gr.HTML("""
-        <h1 style="font-weight: 900; margin-bottom: 7px;">
-        AudioLDM: Text-to-Audio Generation with Latent Diffusion Models
-        </h1>
-        <p>For faster inference without waiting in queue, you may duplicate the space and upgrade to GPU in settings.
-        <br/>
-        <a href="https://huggingface.co/spaces/haoheliu/audioldm-text-to-audio-generation?duplicate=true">
-        <img style="margin-top: 0em; margin-bottom: 0em" src="https://bit.ly/3gLdBN6" alt="Duplicate Space"></a>
-        <p/>
-    """)
+    gr.HTML(
+        """
+        <p>This is the demo for AudioLDM, powered by 🧨 Diffusers. Demo uses the checkpoint <a
+        href="https://huggingface.co/cvssp/audioldm-m-full"> audioldm-m-full </a>. For faster inference without waiting in
+        queue, you may duplicate the space and upgrade to a GPU in the settings. <br/> <a
+        href="https://huggingface.co/spaces/haoheliu/audioldm-text-to-audio-generation?duplicate=true"> <img
+        style="margin-top: 0em; margin-bottom: 0em" src="https://bit.ly/3gLdBN6" alt="Duplicate Space"></a> <p/>
+    """
+    )
+
     with gr.Group():
         with gr.Box():
-            ############# Input
-            textbox = gr.Textbox(value="A hammer is hitting a wooden surface", max_lines=1, label="Input your text here. Your text is important for the audio quality. Please ensure it is descriptive by using more adjectives.", elem_id="prompt-in")
+            textbox = gr.Textbox(
+                value="A hammer is hitting a wooden surface",
+                max_lines=1,
+                label="Input text",
+                info="Your text is important for the audio quality. Please ensure it is descriptive by using more adjectives.",
+                elem_id="prompt-in",
+            )
+            negative_textbox = gr.Textbox(
+                value="low quality, average quality",
+                max_lines=1,
+                label="Negative prompt",
+                info="Enter a negative prompt not to guide the audio generation. Selecting appropriate negative prompts can improve the audio quality significantly.",
+                elem_id="prompt-in",
+            )
 
             with gr.Accordion("Click to modify detailed configurations", open=False):
-              seed = gr.Number(value=45, label="Change this value (any integer number) will lead to a different generation result.")
-              duration = gr.Slider(2.5, 10, value=5, step=2.5, label="Duration (seconds)")
-              guidance_scale = gr.Slider(0, 4, value=2.5, step=0.5, label="Guidance scale (Large => better quality and relavancy to text; Small => better diversity)")
-              n_candidates = gr.Slider(1, 3, value=3, step=1, label="Automatic quality control. This number control the number of candidates (e.g., generate three audios and choose the best to show you). A Larger value usually lead to better quality with heavier computation")
-              # model_name = gr.Dropdown(
-              #       ["audioldm-m-text-ft", "audioldm-s-text-ft", "audioldm-m-full","audioldm-s-full-v2", "audioldm-s-full", "audioldm-l-full"], value="audioldm-m-full", label="Choose the model to use. audioldm-m-text-ft and audioldm-s-text-ft are recommanded. -s- means small, -m- means medium and -l- means large",
-              #   )
-            ############# Output
-            # outputs=gr.Audio(label="Output", type="numpy")
-            outputs=gr.Video(label="Output", elem_id="output-video")
-            
-            # with gr.Group(elem_id="container-advanced-btns"):
-            #   # advanced_button = gr.Button("Advanced options", elem_id="advanced-btn")
-            #   with gr.Group(elem_id="share-btn-container"):
-            #     community_icon = gr.HTML(community_icon_html, visible=False)
-            #     loading_icon = gr.HTML(loading_icon_html, visible=False)
-            #     share_button = gr.Button("Share to community", elem_id="share-btn", visible=False)
-            # outputs=[gr.Audio(label="Output", type="numpy"), gr.Audio(label="Output", type="numpy")]
+                seed = gr.Number(
+                    value=45,
+                    label="Seed",
+                    info="Change this value (any integer number) will lead to a different generation result.",
+                )
+                duration = gr.Slider(2.5, 10, value=5, step=2.5, label="Duration (seconds)")
+                guidance_scale = gr.Slider(
+                    0,
+                    4,
+                    value=2.5,
+                    step=0.5,
+                    label="Guidance scale",
+                    info="Large => better quality and relevancy to text; Small => better diversity",
+                )
+                n_candidates = gr.Slider(
+                    1,
+                    3,
+                    value=3,
+                    step=1,
+                    label="Number waveforms to generate",
+                    info="Automatic quality control. This number control the number of candidates (e.g., generate three audios and choose the best to show you). A Larger value usually lead to better quality with heavier computation",
+                )
+
+            outputs = gr.Video(label="Output", elem_id="output-video")
             btn = gr.Button("Submit").style(full_width=True)
 
         with gr.Group(elem_id="share-btn-container", visible=False):
@@ -259,51 +217,61 @@ with iface:
             loading_icon = gr.HTML(loading_icon_html)
             share_button = gr.Button("Share to community", elem_id="share-btn")
 
-        # btn.click(text2audio, inputs=[
-        #           textbox, duration, guidance_scale, seed, n_candidates, model_name], outputs=[outputs])
-        btn.click(text2audio, inputs=[
-                  textbox, duration, guidance_scale, seed, n_candidates], outputs=[outputs])
-        
+        btn.click(
+            text2audio,
+            inputs=[textbox, negative_textbox, duration, guidance_scale, seed, n_candidates],
+            outputs=[outputs],
+        )
+
         share_button.click(None, [], [], _js=share_js)
-        gr.HTML('''
+        gr.HTML(
+            """
         <div class="footer" style="text-align: center; max-width: 700px; margin: 0 auto;">
-                    <p>Follow the latest update of AudioLDM on our<a href="https://github.com/haoheliu/AudioLDM" style="text-decoration: underline;" target="_blank"> Github repo</a>
-                    </p>
-                    <br>
-                    <p>Model by <a href="https://twitter.com/LiuHaohe" style="text-decoration: underline;" target="_blank">Haohe Liu</a></p>
-                    <br>
+                    <p>Follow the latest update of AudioLDM on our<a href="https://github.com/haoheliu/AudioLDM"
+                    style="text-decoration: underline;" target="_blank"> Github repo</a> </p> <br> <p>Model by <a
+                    href="https://twitter.com/LiuHaohe" style="text-decoration: underline;" target="_blank">Haohe
+                    Liu</a>. Code and demo by 🤗 Hugging Face.</p> <br>
         </div>
-        ''')
-        gr.Examples([
-            ["A hammer is hitting a wooden surface", 5, 2.5, 45, 3, "audioldm-m-full"],
-            ["Peaceful and calming ambient music with singing bowl and other instruments.", 5, 2.5, 45, 3, "audioldm-m-full"],
-            ["A man is speaking in a small room.", 5, 2.5, 45, 3, "audioldm-m-full"],
-            ["A female is speaking followed by footstep sound", 5, 2.5, 45, 3, "audioldm-m-full"],
-            ["Wooden table tapping sound followed by water pouring sound.", 5, 2.5, 45, 3, "audioldm-m-full"],
-        ],
+        """
+        )
+        gr.Examples(
+            [
+                ["A hammer is hitting a wooden surface", "low quality, average quality", 5, 2.5, 45, 3],
+                ["Peaceful and calming ambient music with singing bowl and other instruments.", "low quality, average quality", 5, 2.5, 45, 3],
+                ["A man is speaking in a small room.", "low quality, average quality", 5, 2.5, 45, 3],
+                ["A female is speaking followed by footstep sound", "low quality, average quality", 5, 2.5, 45, 3],
+                ["Wooden table tapping sound followed by water pouring sound.", "low quality, average quality", 5, 2.5, 45, 3],
+            ],
             fn=text2audio,
-            # inputs=[textbox, duration, guidance_scale, seed, n_candidates, model_name],
-            inputs=[textbox, duration, guidance_scale, seed, n_candidates],
+            inputs=[textbox, negative_textbox, duration, guidance_scale, seed, n_candidates],
             outputs=[outputs],
             cache_examples=True,
         )
-        gr.HTML('''
-                <div class="acknowledgements">
-                <p>Essential Tricks for Enhancing the Quality of Your Generated Audio</p>
-                <p>1. Try to use more adjectives to describe your sound. For example: "A man is speaking clearly and slowly in a large room" is better than "A man is speaking". This can make sure AudioLDM understands what you want.</p>
-                <p>2. Try to use different random seeds, which can affect the generation quality significantly sometimes.</p>
-                <p>3. It's better to use general terms like 'man' or 'woman' instead of specific names for individuals or abstract objects that humans may not be familiar with, such as 'mummy'.</p>
-                </div>
-                ''')
+        gr.HTML(
+            """
+                <div class="acknowledgements"> <p>Essential Tricks for Enhancing the Quality of Your Generated
+                Audio</p> <p>1. Try to use more adjectives to describe your sound. For example: "A man is speaking
+                clearly and slowly in a large room" is better than "A man is speaking". This can make sure AudioLDM
+                understands what you want.</p> <p>2. Try to use different random seeds, which can affect the generation
+                quality significantly sometimes.</p> <p>3. It's better to use general terms like 'man' or 'woman'
+                instead of specific names for individuals or abstract objects that humans may not be familiar with,
+                such as 'mummy'.</p> <p>4. Using a negative prompt to not guide the diffusion process can improve the
+                audio quality significantly. Try using negative prompts like 'low quality'.</p> </div>
+                """
+        )
         with gr.Accordion("Additional information", open=False):
             gr.HTML(
                 """
                 <div class="acknowledgments">
-                    <p> We build the model with data from <a href="http://research.google.com/audioset/">AudioSet</a>, <a href="https://freesound.org/">Freesound</a> and <a href="https://sound-effects.bbcrewind.co.uk/">BBC Sound Effect library</a>. We share this demo based on the <a href="https://assets.publishing.service.gov.uk/government/uploads/system/uploads/attachment_data/file/375954/Research.pdf">UK copyright exception</a> of data for academic research. </p>
+                    <p> We build the model with data from <a href="http://research.google.com/audioset/">AudioSet</a>,
+                    <a href="https://freesound.org/">Freesound</a> and <a
+                    href="https://sound-effects.bbcrewind.co.uk/">BBC Sound Effect library</a>. We share this demo
+                    based on the <a
+                    href="https://assets.publishing.service.gov.uk/government/uploads/system/uploads/attachment_data/file/375954/Research.pdf">UK
+                    copyright exception</a> of data for academic research. </p>
                             </div>
                         """
             )
 # <p>This demo is strictly for research demo purpose only. For commercial use please <a href="haoheliu@gmail.com">contact us</a>.</p>
 
 iface.queue(max_size=10).launch(debug=True)
-# iface.launch(debug=True, share=True)

audioldm/__init__.py

@@ -1,3 +0,0 @@
-from .ldm import LatentDiffusion
-from .utils import seed_everything
-from .pipeline import *

audioldm/audio/audio_processing.py

@@ -1,100 +0,0 @@
-import torch
-import numpy as np
-import librosa.util as librosa_util
-from scipy.signal import get_window
-
-
-def window_sumsquare(
-    window,
-    n_frames,
-    hop_length,
-    win_length,
-    n_fft,
-    dtype=np.float32,
-    norm=None,
-):
-    """
-    # from librosa 0.6
-    Compute the sum-square envelope of a window function at a given hop length.
-
-    This is used to estimate modulation effects induced by windowing
-    observations in short-time fourier transforms.
-
-    Parameters
-    ----------
-    window : string, tuple, number, callable, or list-like
-        Window specification, as in `get_window`
-
-    n_frames : int > 0
-        The number of analysis frames
-
-    hop_length : int > 0
-        The number of samples to advance between frames
-
-    win_length : [optional]
-        The length of the window function.  By default, this matches `n_fft`.
-
-    n_fft : int > 0
-        The length of each analysis frame.
-
-    dtype : np.dtype
-        The data type of the output
-
-    Returns
-    -------
-    wss : np.ndarray, shape=`(n_fft + hop_length * (n_frames - 1))`
-        The sum-squared envelope of the window function
-    """
-    if win_length is None:
-        win_length = n_fft
-
-    n = n_fft + hop_length * (n_frames - 1)
-    x = np.zeros(n, dtype=dtype)
-
-    # Compute the squared window at the desired length
-    win_sq = get_window(window, win_length, fftbins=True)
-    win_sq = librosa_util.normalize(win_sq, norm=norm) ** 2
-    win_sq = librosa_util.pad_center(win_sq, n_fft)
-
-    # Fill the envelope
-    for i in range(n_frames):
-        sample = i * hop_length
-        x[sample : min(n, sample + n_fft)] += win_sq[: max(0, min(n_fft, n - sample))]
-    return x
-
-
-def griffin_lim(magnitudes, stft_fn, n_iters=30):
-    """
-    PARAMS
-    ------
-    magnitudes: spectrogram magnitudes
-    stft_fn: STFT class with transform (STFT) and inverse (ISTFT) methods
-    """
-
-    angles = np.angle(np.exp(2j * np.pi * np.random.rand(*magnitudes.size())))
-    angles = angles.astype(np.float32)
-    angles = torch.autograd.Variable(torch.from_numpy(angles))
-    signal = stft_fn.inverse(magnitudes, angles).squeeze(1)
-
-    for i in range(n_iters):
-        _, angles = stft_fn.transform(signal)
-        signal = stft_fn.inverse(magnitudes, angles).squeeze(1)
-    return signal
-
-
-def dynamic_range_compression(x, normalize_fun=torch.log, C=1, clip_val=1e-5):
-    """
-    PARAMS
-    ------
-    C: compression factor
-    """
-    return normalize_fun(torch.clamp(x, min=clip_val) * C)
-
-
-def dynamic_range_decompression(x, C=1):
-    """
-    PARAMS
-    ------
-    C: compression factor used to compress
-    """
-    return torch.exp(x) / C

audioldm/audio/stft.py

@@ -1,180 +0,0 @@
-import torch
-import torch.nn.functional as F
-import numpy as np
-from scipy.signal import get_window
-from librosa.util import pad_center, tiny
-from librosa.filters import mel as librosa_mel_fn
-
-from audioldm.audio.audio_processing import (
-    dynamic_range_compression,
-    dynamic_range_decompression,
-    window_sumsquare,
-)
-
-
-class STFT(torch.nn.Module):
-    """adapted from Prem Seetharaman's https://github.com/pseeth/pytorch-stft"""
-
-    def __init__(self, filter_length, hop_length, win_length, window="hann"):
-        super(STFT, self).__init__()
-        self.filter_length = filter_length
-        self.hop_length = hop_length
-        self.win_length = win_length
-        self.window = window
-        self.forward_transform = None
-        scale = self.filter_length / self.hop_length
-        fourier_basis = np.fft.fft(np.eye(self.filter_length))
-
-        cutoff = int((self.filter_length / 2 + 1))
-        fourier_basis = np.vstack(
-            [np.real(fourier_basis[:cutoff, :]), np.imag(fourier_basis[:cutoff, :])]
-        )
-
-        forward_basis = torch.FloatTensor(fourier_basis[:, None, :])
-        inverse_basis = torch.FloatTensor(
-            np.linalg.pinv(scale * fourier_basis).T[:, None, :]
-        )
-
-        if window is not None:
-            assert filter_length >= win_length
-            # get window and zero center pad it to filter_length
-            fft_window = get_window(window, win_length, fftbins=True)
-            fft_window = pad_center(fft_window, filter_length)
-            fft_window = torch.from_numpy(fft_window).float()
-
-            # window the bases
-            forward_basis *= fft_window
-            inverse_basis *= fft_window
-
-        self.register_buffer("forward_basis", forward_basis.float())
-        self.register_buffer("inverse_basis", inverse_basis.float())
-
-    def transform(self, input_data):
-        num_batches = input_data.size(0)
-        num_samples = input_data.size(1)
-
-        self.num_samples = num_samples
-
-        # similar to librosa, reflect-pad the input
-        input_data = input_data.view(num_batches, 1, num_samples)
-        input_data = F.pad(
-            input_data.unsqueeze(1),
-            (int(self.filter_length / 2), int(self.filter_length / 2), 0, 0),
-            mode="reflect",
-        )
-        input_data = input_data.squeeze(1)
-
-        forward_transform = F.conv1d(
-            input_data,
-            torch.autograd.Variable(self.forward_basis, requires_grad=False),
-            stride=self.hop_length,
-            padding=0,
-        ).cpu()
-
-        cutoff = int((self.filter_length / 2) + 1)
-        real_part = forward_transform[:, :cutoff, :]
-        imag_part = forward_transform[:, cutoff:, :]
-
-        magnitude = torch.sqrt(real_part**2 + imag_part**2)
-        phase = torch.autograd.Variable(torch.atan2(imag_part.data, real_part.data))
-
-        return magnitude, phase
-
-    def inverse(self, magnitude, phase):
-        recombine_magnitude_phase = torch.cat(
-            [magnitude * torch.cos(phase), magnitude * torch.sin(phase)], dim=1
-        )
-
-        inverse_transform = F.conv_transpose1d(
-            recombine_magnitude_phase,
-            torch.autograd.Variable(self.inverse_basis, requires_grad=False),
-            stride=self.hop_length,
-            padding=0,
-        )
-
-        if self.window is not None:
-            window_sum = window_sumsquare(
-                self.window,
-                magnitude.size(-1),
-                hop_length=self.hop_length,
-                win_length=self.win_length,
-                n_fft=self.filter_length,
-                dtype=np.float32,
-            )
-            # remove modulation effects
-            approx_nonzero_indices = torch.from_numpy(
-                np.where(window_sum > tiny(window_sum))[0]
-            )
-            window_sum = torch.autograd.Variable(
-                torch.from_numpy(window_sum), requires_grad=False
-            )
-            window_sum = window_sum
-            inverse_transform[:, :, approx_nonzero_indices] /= window_sum[
-                approx_nonzero_indices
-            ]
-
-            # scale by hop ratio
-            inverse_transform *= float(self.filter_length) / self.hop_length
-
-        inverse_transform = inverse_transform[:, :, int(self.filter_length / 2) :]
-        inverse_transform = inverse_transform[:, :, : -int(self.filter_length / 2) :]
-
-        return inverse_transform
-
-    def forward(self, input_data):
-        self.magnitude, self.phase = self.transform(input_data)
-        reconstruction = self.inverse(self.magnitude, self.phase)
-        return reconstruction
-
-
-class TacotronSTFT(torch.nn.Module):
-    def __init__(
-        self,
-        filter_length,
-        hop_length,
-        win_length,
-        n_mel_channels,
-        sampling_rate,
-        mel_fmin,
-        mel_fmax,
-    ):
-        super(TacotronSTFT, self).__init__()
-        self.n_mel_channels = n_mel_channels
-        self.sampling_rate = sampling_rate
-        self.stft_fn = STFT(filter_length, hop_length, win_length)
-        mel_basis = librosa_mel_fn(
-            sampling_rate, filter_length, n_mel_channels, mel_fmin, mel_fmax
-        )
-        mel_basis = torch.from_numpy(mel_basis).float()
-        self.register_buffer("mel_basis", mel_basis)
-
-    def spectral_normalize(self, magnitudes, normalize_fun):
-        output = dynamic_range_compression(magnitudes, normalize_fun)
-        return output
-
-    def spectral_de_normalize(self, magnitudes):
-        output = dynamic_range_decompression(magnitudes)
-        return output
-
-    def mel_spectrogram(self, y, normalize_fun=torch.log):
-        """Computes mel-spectrograms from a batch of waves
-        PARAMS
-        ------
-        y: Variable(torch.FloatTensor) with shape (B, T) in range [-1, 1]
-
-        RETURNS
-        -------
-        mel_output: torch.FloatTensor of shape (B, n_mel_channels, T)
-        """
-        assert torch.min(y.data) >= -1, torch.min(y.data)
-        assert torch.max(y.data) <= 1, torch.max(y.data)
-
-        magnitudes, phases = self.stft_fn.transform(y)
-        magnitudes = magnitudes.data
-        mel_output = torch.matmul(self.mel_basis, magnitudes)
-        mel_output = self.spectral_normalize(mel_output, normalize_fun)
-        energy = torch.norm(magnitudes, dim=1)
-
-        log_magnitudes = self.spectral_normalize(magnitudes, normalize_fun)
-
-        return mel_output, log_magnitudes, energy

audioldm/audio/tools.py

@@ -1,33 +0,0 @@
-import torch
-import numpy as np
-
-
-def get_mel_from_wav(audio, _stft):
-    audio = torch.clip(torch.FloatTensor(audio).unsqueeze(0), -1, 1)
-    audio = torch.autograd.Variable(audio, requires_grad=False)
-    melspec, log_magnitudes_stft, energy = _stft.mel_spectrogram(audio)
-    melspec = torch.squeeze(melspec, 0).numpy().astype(np.float32)
-    log_magnitudes_stft = (
-        torch.squeeze(log_magnitudes_stft, 0).numpy().astype(np.float32)
-    )
-    energy = torch.squeeze(energy, 0).numpy().astype(np.float32)
-    return melspec, log_magnitudes_stft, energy
-
-
-# def inv_mel_spec(mel, out_filename, _stft, griffin_iters=60):
-#     mel = torch.stack([mel])
-#     mel_decompress = _stft.spectral_de_normalize(mel)
-#     mel_decompress = mel_decompress.transpose(1, 2).data.cpu()
-#     spec_from_mel_scaling = 1000
-#     spec_from_mel = torch.mm(mel_decompress[0], _stft.mel_basis)
-#     spec_from_mel = spec_from_mel.transpose(0, 1).unsqueeze(0)
-#     spec_from_mel = spec_from_mel * spec_from_mel_scaling
-
-#     audio = griffin_lim(
-#         torch.autograd.Variable(spec_from_mel[:, :, :-1]), _stft._stft_fn, griffin_iters
-#     )
-
-#     audio = audio.squeeze()
-#     audio = audio.cpu().numpy()
-#     audio_path = out_filename
-#     write(audio_path, _stft.sampling_rate, audio)

audioldm/clap/encoders.py

@@ -1,170 +0,0 @@
-import torch
-import torch.nn as nn
-from audioldm.clap.open_clip import create_model
-from audioldm.clap.training.data import get_audio_features
-import torchaudio
-from transformers import RobertaTokenizer
-import torch.nn.functional as F
-
-
-class CLAPAudioEmbeddingClassifierFreev2(nn.Module):
-    def __init__(
-        self,
-        pretrained_path="",
-        key="class",
-        sampling_rate=16000,
-        embed_mode="audio",
-        amodel = "HTSAT-tiny",
-        unconditional_prob=0.1,
-        random_mute=False,
-        max_random_mute_portion=0.5,
-        training_mode=True,
-    ):
-        super().__init__()
-
-        self.key = key
-        self.device = "cpu"
-        self.precision = "fp32"
-        self.amodel = amodel
-        self.tmodel = "roberta"  # the best text encoder in our training
-        self.enable_fusion = False  # False if you do not want to use the fusion model
-        self.fusion_type = "aff_2d"
-        self.pretrained = pretrained_path
-        self.embed_mode = embed_mode
-        self.embed_mode_orig = embed_mode
-        self.sampling_rate = sampling_rate
-        self.unconditional_prob = unconditional_prob
-        self.random_mute = random_mute
-        self.tokenize = RobertaTokenizer.from_pretrained("roberta-base")
-        self.max_random_mute_portion = max_random_mute_portion
-        self.training_mode = training_mode
-        self.model, self.model_cfg = create_model(
-            self.amodel,
-            self.tmodel,
-            self.pretrained,
-            precision=self.precision,
-            device=self.device,
-            enable_fusion=self.enable_fusion,
-            fusion_type=self.fusion_type,
-        )
-        for p in self.model.parameters():
-            p.requires_grad = False
-
-        self.model.eval()
-
-    def get_unconditional_condition(self, batchsize):
-        self.unconditional_token = self.model.get_text_embedding(
-            self.tokenizer(["", ""])
-        )[0:1]
-        return torch.cat([self.unconditional_token.unsqueeze(0)] * batchsize, dim=0)
-
-    def batch_to_list(self, batch):
-        ret = []
-        for i in range(batch.size(0)):
-            ret.append(batch[i])
-        return ret
-
-    def make_decision(self, probability):
-        if float(torch.rand(1)) < probability:
-            return True
-        else:
-            return False
-
-    def random_uniform(self, start, end):
-        val = torch.rand(1).item()
-        return start + (end - start) * val
-
-    def _random_mute(self, waveform):
-        # waveform: [bs, t-steps]
-        t_steps = waveform.size(-1)
-        for i in range(waveform.size(0)):
-            mute_size = int(
-                self.random_uniform(0, end=int(t_steps * self.max_random_mute_portion))
-            )
-            mute_start = int(self.random_uniform(0, t_steps - mute_size))
-            waveform[i, mute_start : mute_start + mute_size] = 0
-        return waveform
-
-    def cos_similarity(self, waveform, text):
-        # waveform: [bs, t_steps]
-        with torch.no_grad():
-            self.embed_mode = "audio"
-            audio_emb = self(waveform.cuda())
-            self.embed_mode = "text"
-            text_emb = self(text)
-            similarity = F.cosine_similarity(audio_emb, text_emb, dim=2)
-            return similarity.squeeze()
-
-    def forward(self, batch, key=None):
-        # If you want this conditioner to be unconditional, set self.unconditional_prob = 1.0
-        # If you want this conditioner to be fully conditional, set self.unconditional_prob = 0.0
-        if self.model.training == True and not self.training_mode:
-            print(
-                "The pretrained CLAP model should always be in eval mode. Reloading model just in case you change the parameters."
-            )
-            self.model, self.model_cfg = create_model(
-                self.amodel,
-                self.tmodel,
-                self.pretrained,
-                precision=self.precision,
-                device="cuda",
-                enable_fusion=self.enable_fusion,
-                fusion_type=self.fusion_type,
-            )
-            for p in self.model.parameters():
-                p.requires_grad = False
-            self.model.eval()
-
-        # the 'fusion' truncate mode can be changed to 'rand_trunc' if run in unfusion mode
-        if self.embed_mode == "audio":
-            with torch.no_grad():
-                audio_dict_list = []
-                assert (
-                    self.sampling_rate == 16000
-                ), "We only support 16000 sampling rate"
-                if self.random_mute:
-                    batch = self._random_mute(batch)
-                # batch: [bs, 1, t-samples]
-                batch = torchaudio.functional.resample(
-                    batch, orig_freq=self.sampling_rate, new_freq=48000
-                )
-                for waveform in self.batch_to_list(batch):
-                    audio_dict = {}
-                    audio_dict = get_audio_features(
-                        audio_dict,
-                        waveform,
-                        480000,
-                        data_truncating="fusion",
-                        data_filling="repeatpad",
-                        audio_cfg=self.model_cfg["audio_cfg"],
-                    )
-                    audio_dict_list.append(audio_dict)
-                # [bs, 512]
-                embed = self.model.get_audio_embedding(audio_dict_list)
-        elif self.embed_mode == "text":
-            with torch.no_grad():
-                # the 'fusion' truncate mode can be changed to 'rand_trunc' if run in unfusion mode
-                text_data = self.tokenizer(batch)
-                embed = self.model.get_text_embedding(text_data)
-
-        embed = embed.unsqueeze(1)
-        self.unconditional_token = self.model.get_text_embedding(
-            self.tokenizer(["", ""])
-        )[0:1]
-
-        for i in range(embed.size(0)):
-            if self.make_decision(self.unconditional_prob):
-                embed[i] = self.unconditional_token
-
-        # [bs, 1, 512]
-        return embed.detach()
-
-    def tokenizer(self, text):
-        result = self.tokenize(
-            text,
-            padding="max_length",
-            truncation=True,
-            max_length=512,
-            return_tensors="pt",
-        )
-        return {k: v.squeeze(0) for k, v in result.items()}

audioldm/clap/open_clip/__init__.py

@@ -1,25 +0,0 @@
-from .factory import (
-    list_models,
-    create_model,
-    create_model_and_transforms,
-    add_model_config,
-)
-from .loss import ClipLoss, gather_features, LPLoss, lp_gather_features, LPMetrics
-from .model import (
-    CLAP,
-    CLAPTextCfg,
-    CLAPVisionCfg,
-    CLAPAudioCfp,
-    convert_weights_to_fp16,
-    trace_model,
-)
-from .openai import load_openai_model, list_openai_models
-from .pretrained import (
-    list_pretrained,
-    list_pretrained_tag_models,
-    list_pretrained_model_tags,
-    get_pretrained_url,
-    download_pretrained,
-)
-from .tokenizer import SimpleTokenizer, tokenize
-from .transform import image_transform

audioldm/clap/open_clip/bert.py

@@ -1,40 +0,0 @@
-from transformers import BertTokenizer, BertModel
-
-tokenizer = BertTokenizer.from_pretrained("bert-base-uncased")
-model = BertModel.from_pretrained("bert-base-uncased")
-text = "Replace me by any text you'd like."
-
-
-def bert_embeddings(text):
-    # text = "Replace me by any text you'd like."
-    encoded_input = tokenizer(text, return_tensors="pt")
-    output = model(**encoded_input)
-    return output
-
-
-from transformers import RobertaTokenizer, RobertaModel
-
-tokenizer = RobertaTokenizer.from_pretrained("roberta-base")
-model = RobertaModel.from_pretrained("roberta-base")
-text = "Replace me by any text you'd like."
-
-
-def Roberta_embeddings(text):
-    # text = "Replace me by any text you'd like."
-    encoded_input = tokenizer(text, return_tensors="pt")
-    output = model(**encoded_input)
-    return output
-
-
-from transformers import BartTokenizer, BartModel
-
-tokenizer = BartTokenizer.from_pretrained("facebook/bart-base")
-model = BartModel.from_pretrained("facebook/bart-base")
-text = "Replace me by any text you'd like."
-
-
-def bart_embeddings(text):
-    # text = "Replace me by any text you'd like."
-    encoded_input = tokenizer(text, return_tensors="pt")
-    output = model(**encoded_input)
-    return output

audioldm/clap/open_clip/bpe_simple_vocab_16e6.txt.gz

@@ -1,3 +0,0 @@
-version https://git-lfs.github.com/spec/v1
-oid sha256:924691ac288e54409236115652ad4aa250f48203de50a9e4722a6ecd48d6804a
-size 1356917

audioldm/clap/open_clip/factory.py

@@ -1,277 +0,0 @@
-import json
-import logging
-import os
-import pathlib
-import re
-from copy import deepcopy
-from pathlib import Path
-
-import torch
-
-from .model import CLAP, convert_weights_to_fp16
-from .openai import load_openai_model
-from .pretrained import get_pretrained_url, download_pretrained
-from .transform import image_transform
-
-_MODEL_CONFIG_PATHS = [Path(__file__).parent / f"model_configs/"]
-_MODEL_CONFIGS = {}  # directory (model_name: config) of model architecture configs
-
-
-def _natural_key(string_):
-    return [int(s) if s.isdigit() else s for s in re.split(r"(\d+)", string_.lower())]
-
-
-def _rescan_model_configs():
-    global _MODEL_CONFIGS
-
-    config_ext = (".json",)
-    config_files = []
-    for config_path in _MODEL_CONFIG_PATHS:
-        if config_path.is_file() and config_path.suffix in config_ext:
-            config_files.append(config_path)
-        elif config_path.is_dir():
-            for ext in config_ext:
-                config_files.extend(config_path.glob(f"*{ext}"))
-
-    for cf in config_files:
-        if os.path.basename(cf)[0] == ".":
-            continue  # Ignore hidden files
-
-        with open(cf, "r") as f:
-            model_cfg = json.load(f)
-            if all(a in model_cfg for a in ("embed_dim", "audio_cfg", "text_cfg")):
-                _MODEL_CONFIGS[cf.stem] = model_cfg
-
-    _MODEL_CONFIGS = {
-        k: v
-        for k, v in sorted(_MODEL_CONFIGS.items(), key=lambda x: _natural_key(x[0]))
-    }
-
-
-_rescan_model_configs()  # initial populate of model config registry
-
-
-def load_state_dict(checkpoint_path: str, map_location="cpu", skip_params=True):
-    checkpoint = torch.load(checkpoint_path, map_location=map_location)
-    if isinstance(checkpoint, dict) and "state_dict" in checkpoint:
-        state_dict = checkpoint["state_dict"]
-    else:
-        state_dict = checkpoint
-    if skip_params:
-        if next(iter(state_dict.items()))[0].startswith("module"):
-            state_dict = {k[7:]: v for k, v in state_dict.items()}
-    # for k in state_dict:
-    #     if k.startswith('transformer'):
-    #         v = state_dict.pop(k)
-    #         state_dict['text_branch.' + k[12:]] = v
-    return state_dict
-
-
-def create_model(
-    amodel_name: str,
-    tmodel_name: str,
-    pretrained: str = "",
-    precision: str = "fp32",
-    device: torch.device = torch.device("cpu"),
-    jit: bool = False,
-    force_quick_gelu: bool = False,
-    openai_model_cache_dir: str = os.path.expanduser("~/.cache/clip"),
-    skip_params=True,
-    pretrained_audio: str = "",
-    pretrained_text: str = "",
-    enable_fusion: bool = False,
-    fusion_type: str = "None"
-    # pretrained_image: bool = False,
-):
-    amodel_name = amodel_name.replace(
-        "/", "-"
-    )  # for callers using old naming with / in ViT names
-    pretrained_orig = pretrained
-    pretrained = pretrained.lower()
-    if pretrained == "openai":
-        if amodel_name in _MODEL_CONFIGS:
-            logging.info(f"Loading {amodel_name} model config.")
-            model_cfg = deepcopy(_MODEL_CONFIGS[amodel_name])
-        else:
-            logging.error(
-                f"Model config for {amodel_name} not found; available models {list_models()}."
-            )
-            raise RuntimeError(f"Model config for {amodel_name} not found.")
-
-        logging.info(f"Loading pretrained ViT-B-16 text encoder from OpenAI.")
-        # Hard Code in model name
-        model_cfg["text_cfg"]["model_type"] = tmodel_name
-        model = load_openai_model(
-            "ViT-B-16",
-            model_cfg,
-            device=device,
-            jit=jit,
-            cache_dir=openai_model_cache_dir,
-            enable_fusion=enable_fusion,
-            fusion_type=fusion_type,
-        )
-        # See https://discuss.pytorch.org/t/valueerror-attemting-to-unscale-fp16-gradients/81372
-        if precision == "amp" or precision == "fp32":
-            model = model.float()
-    else:
-        if amodel_name in _MODEL_CONFIGS:
-            logging.info(f"Loading {amodel_name} model config.")
-            model_cfg = deepcopy(_MODEL_CONFIGS[amodel_name])
-        else:
-            logging.error(
-                f"Model config for {amodel_name} not found; available models {list_models()}."
-            )
-            raise RuntimeError(f"Model config for {amodel_name} not found.")
-
-        if force_quick_gelu:
-            # override for use of QuickGELU on non-OpenAI transformer models
-            model_cfg["quick_gelu"] = True
-
-        # if pretrained_image:
-        #     if 'timm_amodel_name' in model_cfg.get('vision_cfg', {}):
-        #         # pretrained weight loading for timm models set via vision_cfg
-        #         model_cfg['vision_cfg']['timm_model_pretrained'] = True
-        #     else:
-        #         assert False, 'pretrained image towers currently only supported for timm models'
-        model_cfg["text_cfg"]["model_type"] = tmodel_name
-        model_cfg["enable_fusion"] = enable_fusion
-        model_cfg["fusion_type"] = fusion_type
-        model = CLAP(**model_cfg)
-
-        if pretrained:
-            checkpoint_path = ""
-            url = get_pretrained_url(amodel_name, pretrained)
-            if url:
-                checkpoint_path = download_pretrained(url, root=openai_model_cache_dir)
-            elif os.path.exists(pretrained_orig):
-                checkpoint_path = pretrained_orig
-            if checkpoint_path:
-                logging.info(
-                    f"Loading pretrained {amodel_name}-{tmodel_name} weights ({pretrained})."
-                )
-                ckpt = load_state_dict(checkpoint_path, skip_params=True)
-                model.load_state_dict(ckpt)
-                param_names = [n for n, p in model.named_parameters()]
-                # for n in param_names:
-                #     print(n, "\t", "Loaded" if n in ckpt else "Unloaded")
-            else:
-                logging.warning(
-                    f"Pretrained weights ({pretrained}) not found for model {amodel_name}."
-                )
-                raise RuntimeError(
-                    f"Pretrained weights ({pretrained}) not found for model {amodel_name}."
-                )
-
-        if pretrained_audio:
-            if amodel_name.startswith("PANN"):
-                if "Cnn14_mAP" in pretrained_audio:  # official checkpoint
-                    audio_ckpt = torch.load(pretrained_audio, map_location="cpu")
-                    audio_ckpt = audio_ckpt["model"]
-                    keys = list(audio_ckpt.keys())
-                    for key in keys:
-                        if (
-                            "spectrogram_extractor" not in key
-                            and "logmel_extractor" not in key
-                        ):
-                            v = audio_ckpt.pop(key)
-                            audio_ckpt["audio_branch." + key] = v
-                elif os.path.basename(pretrained_audio).startswith(
-                    "PANN"
-                ):  # checkpoint trained via HTSAT codebase
-                    audio_ckpt = torch.load(pretrained_audio, map_location="cpu")
-                    audio_ckpt = audio_ckpt["state_dict"]
-                    keys = list(audio_ckpt.keys())
-                    for key in keys:
-                        if key.startswith("sed_model"):
-                            v = audio_ckpt.pop(key)
-                            audio_ckpt["audio_branch." + key[10:]] = v
-                elif os.path.basename(pretrained_audio).startswith(
-                    "finetuned"
-                ):  # checkpoint trained via linear probe codebase
-                    audio_ckpt = torch.load(pretrained_audio, map_location="cpu")
-                else:
-                    raise ValueError("Unknown audio checkpoint")
-            elif amodel_name.startswith("HTSAT"):
-                if "HTSAT_AudioSet_Saved" in pretrained_audio:  # official checkpoint
-                    audio_ckpt = torch.load(pretrained_audio, map_location="cpu")
-                    audio_ckpt = audio_ckpt["state_dict"]
-                    keys = list(audio_ckpt.keys())
-                    for key in keys:
-                        if key.startswith("sed_model") and (
-                            "spectrogram_extractor" not in key
-                            and "logmel_extractor" not in key
-                        ):
-                            v = audio_ckpt.pop(key)
-                            audio_ckpt["audio_branch." + key[10:]] = v
-                elif os.path.basename(pretrained_audio).startswith(
-                    "HTSAT"
-                ):  # checkpoint trained via HTSAT codebase
-                    audio_ckpt = torch.load(pretrained_audio, map_location="cpu")
-                    audio_ckpt = audio_ckpt["state_dict"]
-                    keys = list(audio_ckpt.keys())
-                    for key in keys:
-                        if key.startswith("sed_model"):
-                            v = audio_ckpt.pop(key)
-                            audio_ckpt["audio_branch." + key[10:]] = v
-                elif os.path.basename(pretrained_audio).startswith(
-                    "finetuned"
-                ):  # checkpoint trained via linear probe codebase
-                    audio_ckpt = torch.load(pretrained_audio, map_location="cpu")
-                else:
-                    raise ValueError("Unknown audio checkpoint")
-            else:
-                raise f"this audio encoder pretrained checkpoint is not support"
-
-            model.load_state_dict(audio_ckpt, strict=False)
-            logging.info(
-                f"Loading pretrained {amodel_name} weights ({pretrained_audio})."
-            )
-            param_names = [n for n, p in model.named_parameters()]
-            for n in param_names:
-                print(n, "\t", "Loaded" if n in audio_ckpt else "Unloaded")
-
-        model.to(device=device)
-        if precision == "fp16":
-            assert device.type != "cpu"
-            convert_weights_to_fp16(model)
-
-        if jit:
-            model = torch.jit.script(model)
-
-    return model, model_cfg
-
-
-def create_model_and_transforms(
-    model_name: str,
-    pretrained: str = "",
-    precision: str = "fp32",
-    device: torch.device = torch.device("cpu"),
-    jit: bool = False,
-    force_quick_gelu: bool = False,
-    # pretrained_image: bool = False,
-):
-    model = create_model(
-        model_name,
-        pretrained,
-        precision,
-        device,
-        jit,
-        force_quick_gelu=force_quick_gelu,
-        # pretrained_image=pretrained_image
-    )
-    preprocess_train = image_transform(model.visual.image_size, is_train=True)
-    preprocess_val = image_transform(model.visual.image_size, is_train=False)
-    return model, preprocess_train, preprocess_val
-
-
-def list_models():
-    """enumerate available model architectures based on config files"""
-    return list(_MODEL_CONFIGS.keys())
-
-
-def add_model_config(path):
-    """add model config path or file and update registry"""
-    if not isinstance(path, Path):
-        path = Path(path)
-    _MODEL_CONFIG_PATHS.append(path)
-    _rescan_model_configs()

audioldm/clap/open_clip/feature_fusion.py

@@ -1,192 +0,0 @@
-"""
-Feature Fusion for Varible-Length Data Processing
-AFF/iAFF is referred and modified from https://github.com/YimianDai/open-aff/blob/master/aff_pytorch/aff_net/fusion.py
-According to the paper: Yimian Dai et al, Attentional Feature Fusion, IEEE Winter Conference on Applications of Computer Vision, WACV 2021
-"""
-
-import torch
-import torch.nn as nn
-
-
-class DAF(nn.Module):
-    """
-    直接相加 DirectAddFuse
-    """
-
-    def __init__(self):
-        super(DAF, self).__init__()
-
-    def forward(self, x, residual):
-        return x + residual
-
-
-class iAFF(nn.Module):
-    """
-    多特征融合 iAFF
-    """
-
-    def __init__(self, channels=64, r=4, type="2D"):
-        super(iAFF, self).__init__()
-        inter_channels = int(channels // r)
-
-        if type == "1D":
-            # 本地注意力
-            self.local_att = nn.Sequential(
-                nn.Conv1d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv1d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(channels),
-            )
-
-            # 全局注意力
-            self.global_att = nn.Sequential(
-                nn.AdaptiveAvgPool1d(1),
-                nn.Conv1d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv1d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(channels),
-            )
-
-            # 第二次本地注意力
-            self.local_att2 = nn.Sequential(
-                nn.Conv1d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv1d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(channels),
-            )
-            # 第二次全局注意力
-            self.global_att2 = nn.Sequential(
-                nn.AdaptiveAvgPool1d(1),
-                nn.Conv1d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv1d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(channels),
-            )
-        elif type == "2D":
-            # 本地注意力
-            self.local_att = nn.Sequential(
-                nn.Conv2d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv2d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(channels),
-            )
-
-            # 全局注意力
-            self.global_att = nn.Sequential(
-                nn.AdaptiveAvgPool2d(1),
-                nn.Conv2d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv2d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(channels),
-            )
-
-            # 第二次本地注意力
-            self.local_att2 = nn.Sequential(
-                nn.Conv2d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv2d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(channels),
-            )
-            # 第二次全局注意力
-            self.global_att2 = nn.Sequential(
-                nn.AdaptiveAvgPool2d(1),
-                nn.Conv2d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv2d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(channels),
-            )
-        else:
-            raise f"the type is not supported"
-
-        self.sigmoid = nn.Sigmoid()
-
-    def forward(self, x, residual):
-        flag = False
-        xa = x + residual
-        if xa.size(0) == 1:
-            xa = torch.cat([xa, xa], dim=0)
-            flag = True
-        xl = self.local_att(xa)
-        xg = self.global_att(xa)
-        xlg = xl + xg
-        wei = self.sigmoid(xlg)
-        xi = x * wei + residual * (1 - wei)
-
-        xl2 = self.local_att2(xi)
-        xg2 = self.global_att(xi)
-        xlg2 = xl2 + xg2
-        wei2 = self.sigmoid(xlg2)
-        xo = x * wei2 + residual * (1 - wei2)
-        if flag:
-            xo = xo[0].unsqueeze(0)
-        return xo
-
-
-class AFF(nn.Module):
-    """
-    多特征融合 AFF
-    """
-
-    def __init__(self, channels=64, r=4, type="2D"):
-        super(AFF, self).__init__()
-        inter_channels = int(channels // r)
-
-        if type == "1D":
-            self.local_att = nn.Sequential(
-                nn.Conv1d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv1d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(channels),
-            )
-            self.global_att = nn.Sequential(
-                nn.AdaptiveAvgPool1d(1),
-                nn.Conv1d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv1d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm1d(channels),
-            )
-        elif type == "2D":
-            self.local_att = nn.Sequential(
-                nn.Conv2d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv2d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(channels),
-            )
-            self.global_att = nn.Sequential(
-                nn.AdaptiveAvgPool2d(1),
-                nn.Conv2d(channels, inter_channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(inter_channels),
-                nn.ReLU(inplace=True),
-                nn.Conv2d(inter_channels, channels, kernel_size=1, stride=1, padding=0),
-                nn.BatchNorm2d(channels),
-            )
-        else:
-            raise f"the type is not supported."
-
-        self.sigmoid = nn.Sigmoid()
-
-    def forward(self, x, residual):
-        flag = False
-        xa = x + residual
-        if xa.size(0) == 1:
-            xa = torch.cat([xa, xa], dim=0)
-            flag = True
-        xl = self.local_att(xa)
-        xg = self.global_att(xa)
-        xlg = xl + xg
-        wei = self.sigmoid(xlg)
-        xo = 2 * x * wei + 2 * residual * (1 - wei)
-        if flag:
-            xo = xo[0].unsqueeze(0)
-        return xo

audioldm/clap/open_clip/htsat.py

@@ -1,1308 +0,0 @@
-# Ke Chen
-# knutchen@ucsd.edu
-# HTS-AT: A HIERARCHICAL TOKEN-SEMANTIC AUDIO TRANSFORMER FOR SOUND CLASSIFICATION AND DETECTION
-# Some layers designed on the model
-# below codes are based and referred from https://github.com/microsoft/Swin-Transformer
-# Swin Transformer for Computer Vision: https://arxiv.org/pdf/2103.14030.pdf
-
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-from itertools import repeat
-import collections.abc
-import math
-import warnings
-
-from torch.nn.init import _calculate_fan_in_and_fan_out
-import torch.utils.checkpoint as checkpoint
-
-import random
-
-from torchlibrosa.stft import Spectrogram, LogmelFilterBank
-from torchlibrosa.augmentation import SpecAugmentation
-
-from itertools import repeat
-from .utils import do_mixup, interpolate
-
-from .feature_fusion import iAFF, AFF, DAF
-
-# from PyTorch internals
-def _ntuple(n):
-    def parse(x):
-        if isinstance(x, collections.abc.Iterable):
-            return x
-        return tuple(repeat(x, n))
-
-    return parse
-
-
-to_1tuple = _ntuple(1)
-to_2tuple = _ntuple(2)
-to_3tuple = _ntuple(3)
-to_4tuple = _ntuple(4)
-to_ntuple = _ntuple
-
-
-def drop_path(x, drop_prob: float = 0.0, training: bool = False):
-    """Drop paths (Stochastic Depth) per sample (when applied in main path of residual blocks).
-    This is the same as the DropConnect impl I created for EfficientNet, etc networks, however,
-    the original name is misleading as 'Drop Connect' is a different form of dropout in a separate paper...
-    See discussion: https://github.com/tensorflow/tpu/issues/494#issuecomment-532968956 ... I've opted for
-    changing the layer and argument names to 'drop path' rather than mix DropConnect as a layer name and use
-    'survival rate' as the argument.
-    """
-    if drop_prob == 0.0 or not training:
-        return x
-    keep_prob = 1 - drop_prob
-    shape = (x.shape[0],) + (1,) * (
-        x.ndim - 1
-    )  # work with diff dim tensors, not just 2D ConvNets
-    random_tensor = keep_prob + torch.rand(shape, dtype=x.dtype, device=x.device)
-    random_tensor.floor_()  # binarize
-    output = x.div(keep_prob) * random_tensor
-    return output
-
-
-class DropPath(nn.Module):
-    """Drop paths (Stochastic Depth) per sample  (when applied in main path of residual blocks)."""
-
-    def __init__(self, drop_prob=None):
-        super(DropPath, self).__init__()
-        self.drop_prob = drop_prob
-
-    def forward(self, x):
-        return drop_path(x, self.drop_prob, self.training)
-
-
-class PatchEmbed(nn.Module):
-    """2D Image to Patch Embedding"""
-
-    def __init__(
-        self,
-        img_size=224,
-        patch_size=16,
-        in_chans=3,
-        embed_dim=768,
-        norm_layer=None,
-        flatten=True,
-        patch_stride=16,
-        enable_fusion=False,
-        fusion_type="None",
-    ):
-        super().__init__()
-        img_size = to_2tuple(img_size)
-        patch_size = to_2tuple(patch_size)
-        patch_stride = to_2tuple(patch_stride)
-        self.img_size = img_size
-        self.patch_size = patch_size
-        self.patch_stride = patch_stride
-        self.grid_size = (
-            img_size[0] // patch_stride[0],
-            img_size[1] // patch_stride[1],
-        )
-        self.num_patches = self.grid_size[0] * self.grid_size[1]
-        self.flatten = flatten
-        self.in_chans = in_chans
-        self.embed_dim = embed_dim
-
-        self.enable_fusion = enable_fusion
-        self.fusion_type = fusion_type
-
-        padding = (
-            (patch_size[0] - patch_stride[0]) // 2,
-            (patch_size[1] - patch_stride[1]) // 2,
-        )
-
-        if (self.enable_fusion) and (self.fusion_type == "channel_map"):
-            self.proj = nn.Conv2d(
-                in_chans * 4,
-                embed_dim,
-                kernel_size=patch_size,
-                stride=patch_stride,
-                padding=padding,
-            )
-        else:
-            self.proj = nn.Conv2d(
-                in_chans,
-                embed_dim,
-                kernel_size=patch_size,
-                stride=patch_stride,
-                padding=padding,
-            )
-        self.norm = norm_layer(embed_dim) if norm_layer else nn.Identity()
-
-        if (self.enable_fusion) and (
-            self.fusion_type in ["daf_2d", "aff_2d", "iaff_2d"]
-        ):
-            self.mel_conv2d = nn.Conv2d(
-                in_chans,
-                embed_dim,
-                kernel_size=(patch_size[0], patch_size[1] * 3),
-                stride=(patch_stride[0], patch_stride[1] * 3),
-                padding=padding,
-            )
-            if self.fusion_type == "daf_2d":
-                self.fusion_model = DAF()
-            elif self.fusion_type == "aff_2d":
-                self.fusion_model = AFF(channels=embed_dim, type="2D")
-            elif self.fusion_type == "iaff_2d":
-                self.fusion_model = iAFF(channels=embed_dim, type="2D")
-
-    def forward(self, x, longer_idx=None):
-        if (self.enable_fusion) and (
-            self.fusion_type in ["daf_2d", "aff_2d", "iaff_2d"]
-        ):
-            global_x = x[:, 0:1, :, :]
-
-            # global processing
-            B, C, H, W = global_x.shape
-            assert (
-                H == self.img_size[0] and W == self.img_size[1]
-            ), f"Input image size ({H}*{W}) doesn't match model ({self.img_size[0]}*{self.img_size[1]})."
-            global_x = self.proj(global_x)
-            TW = global_x.size(-1)
-            if len(longer_idx) > 0:
-                # local processing
-                local_x = x[longer_idx, 1:, :, :].contiguous()
-                B, C, H, W = local_x.shape
-                local_x = local_x.view(B * C, 1, H, W)
-                local_x = self.mel_conv2d(local_x)
-                local_x = local_x.view(
-                    B, C, local_x.size(1), local_x.size(2), local_x.size(3)
-                )
-                local_x = local_x.permute((0, 2, 3, 1, 4)).contiguous().flatten(3)
-                TB, TC, TH, _ = local_x.size()
-                if local_x.size(-1) < TW:
-                    local_x = torch.cat(
-                        [
-                            local_x,
-                            torch.zeros(
-                                (TB, TC, TH, TW - local_x.size(-1)),
-                                device=global_x.device,
-                            ),
-                        ],
-                        dim=-1,
-                    )
-                else:
-                    local_x = local_x[:, :, :, :TW]
-
-                global_x[longer_idx] = self.fusion_model(global_x[longer_idx], local_x)
-            x = global_x
-        else:
-            B, C, H, W = x.shape
-            assert (
-                H == self.img_size[0] and W == self.img_size[1]
-            ), f"Input image size ({H}*{W}) doesn't match model ({self.img_size[0]}*{self.img_size[1]})."
-            x = self.proj(x)
-
-        if self.flatten:
-            x = x.flatten(2).transpose(1, 2)  # BCHW -> BNC
-        x = self.norm(x)
-        return x
-
-
-class Mlp(nn.Module):
-    """MLP as used in Vision Transformer, MLP-Mixer and related networks"""
-
-    def __init__(
-        self,
-        in_features,
-        hidden_features=None,
-        out_features=None,
-        act_layer=nn.GELU,
-        drop=0.0,
-    ):
-        super().__init__()
-        out_features = out_features or in_features
-        hidden_features = hidden_features or in_features
-        self.fc1 = nn.Linear(in_features, hidden_features)
-        self.act = act_layer()
-        self.fc2 = nn.Linear(hidden_features, out_features)
-        self.drop = nn.Dropout(drop)
-
-    def forward(self, x):
-        x = self.fc1(x)
-        x = self.act(x)
-        x = self.drop(x)
-        x = self.fc2(x)
-        x = self.drop(x)
-        return x
-
-
-def _no_grad_trunc_normal_(tensor, mean, std, a, b):
-    # Cut & paste from PyTorch official master until it's in a few official releases - RW
-    # Method based on https://people.sc.fsu.edu/~jburkardt/presentations/truncated_normal.pdf
-    def norm_cdf(x):
-        # Computes standard normal cumulative distribution function
-        return (1.0 + math.erf(x / math.sqrt(2.0))) / 2.0
-
-    if (mean < a - 2 * std) or (mean > b + 2 * std):
-        warnings.warn(
-            "mean is more than 2 std from [a, b] in nn.init.trunc_normal_. "
-            "The distribution of values may be incorrect.",
-            stacklevel=2,
-        )
-
-    with torch.no_grad():
-        # Values are generated by using a truncated uniform distribution and
-        # then using the inverse CDF for the normal distribution.
-        # Get upper and lower cdf values
-        l = norm_cdf((a - mean) / std)
-        u = norm_cdf((b - mean) / std)
-
-        # Uniformly fill tensor with values from [l, u], then translate to
-        # [2l-1, 2u-1].
-        tensor.uniform_(2 * l - 1, 2 * u - 1)
-
-        # Use inverse cdf transform for normal distribution to get truncated
-        # standard normal
-        tensor.erfinv_()
-
-        # Transform to proper mean, std
-        tensor.mul_(std * math.sqrt(2.0))
-        tensor.add_(mean)
-
-        # Clamp to ensure it's in the proper range
-        tensor.clamp_(min=a, max=b)
-        return tensor
-
-
-def trunc_normal_(tensor, mean=0.0, std=1.0, a=-2.0, b=2.0):
-    # type: (Tensor, float, float, float, float) -> Tensor
-    r"""Fills the input Tensor with values drawn from a truncated
-    normal distribution. The values are effectively drawn from the
-    normal distribution :math:`\mathcal{N}(\text{mean}, \text{std}^2)`
-    with values outside :math:`[a, b]` redrawn until they are within
-    the bounds. The method used for generating the random values works
-    best when :math:`a \leq \text{mean} \leq b`.
-    Args:
-        tensor: an n-dimensional `torch.Tensor`
-        mean: the mean of the normal distribution
-        std: the standard deviation of the normal distribution
-        a: the minimum cutoff value
-        b: the maximum cutoff value
-    Examples:
-        >>> w = torch.empty(3, 5)
-        >>> nn.init.trunc_normal_(w)
-    """
-    return _no_grad_trunc_normal_(tensor, mean, std, a, b)
-
-
-def variance_scaling_(tensor, scale=1.0, mode="fan_in", distribution="normal"):
-    fan_in, fan_out = _calculate_fan_in_and_fan_out(tensor)
-    if mode == "fan_in":
-        denom = fan_in
-    elif mode == "fan_out":
-        denom = fan_out
-    elif mode == "fan_avg":
-        denom = (fan_in + fan_out) / 2
-
-    variance = scale / denom
-
-    if distribution == "truncated_normal":
-        # constant is stddev of standard normal truncated to (-2, 2)
-        trunc_normal_(tensor, std=math.sqrt(variance) / 0.87962566103423978)
-    elif distribution == "normal":
-        tensor.normal_(std=math.sqrt(variance))
-    elif distribution == "uniform":
-        bound = math.sqrt(3 * variance)
-        tensor.uniform_(-bound, bound)
-    else:
-        raise ValueError(f"invalid distribution {distribution}")
-
-
-def lecun_normal_(tensor):
-    variance_scaling_(tensor, mode="fan_in", distribution="truncated_normal")
-
-
-def window_partition(x, window_size):
-    """
-    Args:
-        x: (B, H, W, C)
-        window_size (int): window size
-    Returns:
-        windows: (num_windows*B, window_size, window_size, C)
-    """
-    B, H, W, C = x.shape
-    x = x.view(B, H // window_size, window_size, W // window_size, window_size, C)
-    windows = (
-        x.permute(0, 1, 3, 2, 4, 5).contiguous().view(-1, window_size, window_size, C)
-    )
-    return windows
-
-
-def window_reverse(windows, window_size, H, W):
-    """
-    Args:
-        windows: (num_windows*B, window_size, window_size, C)
-        window_size (int): Window size
-        H (int): Height of image
-        W (int): Width of image
-    Returns:
-        x: (B, H, W, C)
-    """
-    B = int(windows.shape[0] / (H * W / window_size / window_size))
-    x = windows.view(
-        B, H // window_size, W // window_size, window_size, window_size, -1
-    )
-    x = x.permute(0, 1, 3, 2, 4, 5).contiguous().view(B, H, W, -1)
-    return x
-
-
-class WindowAttention(nn.Module):
-    r"""Window based multi-head self attention (W-MSA) module with relative position bias.
-    It supports both of shifted and non-shifted window.
-    Args:
-        dim (int): Number of input channels.
-        window_size (tuple[int]): The height and width of the window.
-        num_heads (int): Number of attention heads.
-        qkv_bias (bool, optional):  If True, add a learnable bias to query, key, value. Default: True
-        qk_scale (float | None, optional): Override default qk scale of head_dim ** -0.5 if set
-        attn_drop (float, optional): Dropout ratio of attention weight. Default: 0.0
-        proj_drop (float, optional): Dropout ratio of output. Default: 0.0
-    """
-
-    def __init__(
-        self,
-        dim,
-        window_size,
-        num_heads,
-        qkv_bias=True,
-        qk_scale=None,
-        attn_drop=0.0,
-        proj_drop=0.0,
-    ):
-
-        super().__init__()
-        self.dim = dim
-        self.window_size = window_size  # Wh, Ww
-        self.num_heads = num_heads
-        head_dim = dim // num_heads
-        self.scale = qk_scale or head_dim**-0.5
-
-        # define a parameter table of relative position bias
-        self.relative_position_bias_table = nn.Parameter(
-            torch.zeros((2 * window_size[0] - 1) * (2 * window_size[1] - 1), num_heads)
-        )  # 2*Wh-1 * 2*Ww-1, nH
-
-        # get pair-wise relative position index for each token inside the window
-        coords_h = torch.arange(self.window_size[0])
-        coords_w = torch.arange(self.window_size[1])
-        coords = torch.stack(torch.meshgrid([coords_h, coords_w]))  # 2, Wh, Ww
-        coords_flatten = torch.flatten(coords, 1)  # 2, Wh*Ww
-        relative_coords = (
-            coords_flatten[:, :, None] - coords_flatten[:, None, :]
-        )  # 2, Wh*Ww, Wh*Ww
-        relative_coords = relative_coords.permute(
-            1, 2, 0
-        ).contiguous()  # Wh*Ww, Wh*Ww, 2
-        relative_coords[:, :, 0] += self.window_size[0] - 1  # shift to start from 0
-        relative_coords[:, :, 1] += self.window_size[1] - 1
-        relative_coords[:, :, 0] *= 2 * self.window_size[1] - 1
-        relative_position_index = relative_coords.sum(-1)  # Wh*Ww, Wh*Ww
-        self.register_buffer("relative_position_index", relative_position_index)
-
-        self.qkv = nn.Linear(dim, dim * 3, bias=qkv_bias)
-        self.attn_drop = nn.Dropout(attn_drop)
-        self.proj = nn.Linear(dim, dim)
-        self.proj_drop = nn.Dropout(proj_drop)
-
-        trunc_normal_(self.relative_position_bias_table, std=0.02)
-        self.softmax = nn.Softmax(dim=-1)
-
-    def forward(self, x, mask=None):
-        """
-        Args:
-            x: input features with shape of (num_windows*B, N, C)
-            mask: (0/-inf) mask with shape of (num_windows, Wh*Ww, Wh*Ww) or None
-        """
-        B_, N, C = x.shape
-        qkv = (
-            self.qkv(x)
-            .reshape(B_, N, 3, self.num_heads, C // self.num_heads)
-            .permute(2, 0, 3, 1, 4)
-        )
-        q, k, v = (
-            qkv[0],
-            qkv[1],
-            qkv[2],
-        )  # make torchscript happy (cannot use tensor as tuple)
-
-        q = q * self.scale
-        attn = q @ k.transpose(-2, -1)
-
-        relative_position_bias = self.relative_position_bias_table[
-            self.relative_position_index.view(-1)
-        ].view(
-            self.window_size[0] * self.window_size[1],
-            self.window_size[0] * self.window_size[1],
-            -1,
-        )  # Wh*Ww,Wh*Ww,nH
-        relative_position_bias = relative_position_bias.permute(
-            2, 0, 1
-        ).contiguous()  # nH, Wh*Ww, Wh*Ww
-        attn = attn + relative_position_bias.unsqueeze(0)
-
-        if mask is not None:
-            nW = mask.shape[0]
-            attn = attn.view(B_ // nW, nW, self.num_heads, N, N) + mask.unsqueeze(
-                1
-            ).unsqueeze(0)
-            attn = attn.view(-1, self.num_heads, N, N)
-            attn = self.softmax(attn)
-        else:
-            attn = self.softmax(attn)
-
-        attn = self.attn_drop(attn)
-
-        x = (attn @ v).transpose(1, 2).reshape(B_, N, C)
-        x = self.proj(x)
-        x = self.proj_drop(x)
-        return x, attn
-
-    def extra_repr(self):
-        return f"dim={self.dim}, window_size={self.window_size}, num_heads={self.num_heads}"
-
-
-# We use the model based on Swintransformer Block, therefore we can use the swin-transformer pretrained model
-class SwinTransformerBlock(nn.Module):
-    r"""Swin Transformer Block.
-    Args:
-        dim (int): Number of input channels.
-        input_resolution (tuple[int]): Input resulotion.
-        num_heads (int): Number of attention heads.
-        window_size (int): Window size.
-        shift_size (int): Shift size for SW-MSA.
-        mlp_ratio (float): Ratio of mlp hidden dim to embedding dim.
-        qkv_bias (bool, optional): If True, add a learnable bias to query, key, value. Default: True
-        qk_scale (float | None, optional): Override default qk scale of head_dim ** -0.5 if set.
-        drop (float, optional): Dropout rate. Default: 0.0
-        attn_drop (float, optional): Attention dropout rate. Default: 0.0
-        drop_path (float, optional): Stochastic depth rate. Default: 0.0
-        act_layer (nn.Module, optional): Activation layer. Default: nn.GELU
-        norm_layer (nn.Module, optional): Normalization layer.  Default: nn.LayerNorm
-    """
-
-    def __init__(
-        self,
-        dim,
-        input_resolution,
-        num_heads,
-        window_size=7,
-        shift_size=0,
-        mlp_ratio=4.0,
-        qkv_bias=True,
-        qk_scale=None,
-        drop=0.0,
-        attn_drop=0.0,
-        drop_path=0.0,
-        act_layer=nn.GELU,
-        norm_layer=nn.LayerNorm,
-        norm_before_mlp="ln",
-    ):
-        super().__init__()
-        self.dim = dim
-        self.input_resolution = input_resolution
-        self.num_heads = num_heads
-        self.window_size = window_size
-        self.shift_size = shift_size
-        self.mlp_ratio = mlp_ratio
-        self.norm_before_mlp = norm_before_mlp
-        if min(self.input_resolution) <= self.window_size:
-            # if window size is larger than input resolution, we don't partition windows
-            self.shift_size = 0
-            self.window_size = min(self.input_resolution)
-        assert (
-            0 <= self.shift_size < self.window_size
-        ), "shift_size must in 0-window_size"
-
-        self.norm1 = norm_layer(dim)
-        self.attn = WindowAttention(
-            dim,
-            window_size=to_2tuple(self.window_size),
-            num_heads=num_heads,
-            qkv_bias=qkv_bias,
-            qk_scale=qk_scale,
-            attn_drop=attn_drop,
-            proj_drop=drop,
-        )
-
-        self.drop_path = DropPath(drop_path) if drop_path > 0.0 else nn.Identity()
-        if self.norm_before_mlp == "ln":
-            self.norm2 = nn.LayerNorm(dim)
-        elif self.norm_before_mlp == "bn":
-            self.norm2 = lambda x: nn.BatchNorm1d(dim)(x.transpose(1, 2)).transpose(
-                1, 2
-            )
-        else:
-            raise NotImplementedError
-        mlp_hidden_dim = int(dim * mlp_ratio)
-        self.mlp = Mlp(
-            in_features=dim,
-            hidden_features=mlp_hidden_dim,
-            act_layer=act_layer,
-            drop=drop,
-        )
-
-        if self.shift_size > 0:
-            # calculate attention mask for SW-MSA
-            H, W = self.input_resolution
-            img_mask = torch.zeros((1, H, W, 1))  # 1 H W 1
-            h_slices = (
-                slice(0, -self.window_size),
-                slice(-self.window_size, -self.shift_size),
-                slice(-self.shift_size, None),
-            )
-            w_slices = (
-                slice(0, -self.window_size),
-                slice(-self.window_size, -self.shift_size),
-                slice(-self.shift_size, None),
-            )
-            cnt = 0
-            for h in h_slices:
-                for w in w_slices:
-                    img_mask[:, h, w, :] = cnt
-                    cnt += 1
-
-            mask_windows = window_partition(
-                img_mask, self.window_size
-            )  # nW, window_size, window_size, 1
-            mask_windows = mask_windows.view(-1, self.window_size * self.window_size)
-            attn_mask = mask_windows.unsqueeze(1) - mask_windows.unsqueeze(2)
-            attn_mask = attn_mask.masked_fill(
-                attn_mask != 0, float(-100.0)
-            ).masked_fill(attn_mask == 0, float(0.0))
-        else:
-            attn_mask = None
-
-        self.register_buffer("attn_mask", attn_mask)
-
-    def forward(self, x):
-        # pdb.set_trace()
-        H, W = self.input_resolution
-        # print("H: ", H)
-        # print("W: ", W)
-        # pdb.set_trace()
-        B, L, C = x.shape
-        # assert L == H * W, "input feature has wrong size"
-
-        shortcut = x
-        x = self.norm1(x)
-        x = x.view(B, H, W, C)
-
-        # cyclic shift
-        if self.shift_size > 0:
-            shifted_x = torch.roll(
-                x, shifts=(-self.shift_size, -self.shift_size), dims=(1, 2)
-            )
-        else:
-            shifted_x = x
-
-        # partition windows
-        x_windows = window_partition(
-            shifted_x, self.window_size
-        )  # nW*B, window_size, window_size, C
-        x_windows = x_windows.view(
-            -1, self.window_size * self.window_size, C
-        )  # nW*B, window_size*window_size, C
-
-        # W-MSA/SW-MSA
-        attn_windows, attn = self.attn(
-            x_windows, mask=self.attn_mask
-        )  # nW*B, window_size*window_size, C
-
-        # merge windows
-        attn_windows = attn_windows.view(-1, self.window_size, self.window_size, C)
-        shifted_x = window_reverse(attn_windows, self.window_size, H, W)  # B H' W' C
-
-        # reverse cyclic shift
-        if self.shift_size > 0:
-            x = torch.roll(
-                shifted_x, shifts=(self.shift_size, self.shift_size), dims=(1, 2)
-            )
-        else:
-            x = shifted_x
-        x = x.view(B, H * W, C)
-
-        # FFN
-        x = shortcut + self.drop_path(x)
-        x = x + self.drop_path(self.mlp(self.norm2(x)))
-
-        return x, attn
-
-    def extra_repr(self):
-        return (
-            f"dim={self.dim}, input_resolution={self.input_resolution}, num_heads={self.num_heads}, "
-            f"window_size={self.window_size}, shift_size={self.shift_size}, mlp_ratio={self.mlp_ratio}"
-        )
-
-
-class PatchMerging(nn.Module):
-    r"""Patch Merging Layer.
-    Args:
-        input_resolution (tuple[int]): Resolution of input feature.
-        dim (int): Number of input channels.
-        norm_layer (nn.Module, optional): Normalization layer.  Default: nn.LayerNorm
-    """
-
-    def __init__(self, input_resolution, dim, norm_layer=nn.LayerNorm):
-        super().__init__()
-        self.input_resolution = input_resolution
-        self.dim = dim
-        self.reduction = nn.Linear(4 * dim, 2 * dim, bias=False)
-        self.norm = norm_layer(4 * dim)
-
-    def forward(self, x):
-        """
-        x: B, H*W, C
-        """
-        H, W = self.input_resolution
-        B, L, C = x.shape
-        assert L == H * W, "input feature has wrong size"
-        assert H % 2 == 0 and W % 2 == 0, f"x size ({H}*{W}) are not even."
-
-        x = x.view(B, H, W, C)
-
-        x0 = x[:, 0::2, 0::2, :]  # B H/2 W/2 C
-        x1 = x[:, 1::2, 0::2, :]  # B H/2 W/2 C
-        x2 = x[:, 0::2, 1::2, :]  # B H/2 W/2 C
-        x3 = x[:, 1::2, 1::2, :]  # B H/2 W/2 C
-        x = torch.cat([x0, x1, x2, x3], -1)  # B H/2 W/2 4*C
-        x = x.view(B, -1, 4 * C)  # B H/2*W/2 4*C
-
-        x = self.norm(x)
-        x = self.reduction(x)
-
-        return x
-
-    def extra_repr(self):
-        return f"input_resolution={self.input_resolution}, dim={self.dim}"
-
-
-class BasicLayer(nn.Module):
-    """A basic Swin Transformer layer for one stage.
-    Args:
-        dim (int): Number of input channels.
-        input_resolution (tuple[int]): Input resolution.
-        depth (int): Number of blocks.
-        num_heads (int): Number of attention heads.
-        window_size (int): Local window size.
-        mlp_ratio (float): Ratio of mlp hidden dim to embedding dim.
-        qkv_bias (bool, optional): If True, add a learnable bias to query, key, value. Default: True
-        qk_scale (float | None, optional): Override default qk scale of head_dim ** -0.5 if set.
-        drop (float, optional): Dropout rate. Default: 0.0
-        attn_drop (float, optional): Attention dropout rate. Default: 0.0
-        drop_path (float | tuple[float], optional): Stochastic depth rate. Default: 0.0
-        norm_layer (nn.Module, optional): Normalization layer. Default: nn.LayerNorm
-        downsample (nn.Module | None, optional): Downsample layer at the end of the layer. Default: None
-        use_checkpoint (bool): Whether to use checkpointing to save memory. Default: False.
-    """
-
-    def __init__(
-        self,
-        dim,
-        input_resolution,
-        depth,
-        num_heads,
-        window_size,
-        mlp_ratio=4.0,
-        qkv_bias=True,
-        qk_scale=None,
-        drop=0.0,
-        attn_drop=0.0,
-        drop_path=0.0,
-        norm_layer=nn.LayerNorm,
-        downsample=None,
-        use_checkpoint=False,
-        norm_before_mlp="ln",
-    ):
-
-        super().__init__()
-        self.dim = dim
-        self.input_resolution = input_resolution
-        self.depth = depth
-        self.use_checkpoint = use_checkpoint
-
-        # build blocks
-        self.blocks = nn.ModuleList(
-            [
-                SwinTransformerBlock(
-                    dim=dim,
-                    input_resolution=input_resolution,
-                    num_heads=num_heads,
-                    window_size=window_size,
-                    shift_size=0 if (i % 2 == 0) else window_size // 2,
-                    mlp_ratio=mlp_ratio,
-                    qkv_bias=qkv_bias,
-                    qk_scale=qk_scale,
-                    drop=drop,
-                    attn_drop=attn_drop,
-                    drop_path=drop_path[i]
-                    if isinstance(drop_path, list)
-                    else drop_path,
-                    norm_layer=norm_layer,
-                    norm_before_mlp=norm_before_mlp,
-                )
-                for i in range(depth)
-            ]
-        )
-
-        # patch merging layer
-        if downsample is not None:
-            self.downsample = downsample(
-                input_resolution, dim=dim, norm_layer=norm_layer
-            )
-        else:
-            self.downsample = None
-
-    def forward(self, x):
-        attns = []
-        for blk in self.blocks:
-            if self.use_checkpoint:
-                x = checkpoint.checkpoint(blk, x)
-            else:
-                x, attn = blk(x)
-                if not self.training:
-                    attns.append(attn.unsqueeze(0))
-        if self.downsample is not None:
-            x = self.downsample(x)
-        if not self.training:
-            attn = torch.cat(attns, dim=0)
-            attn = torch.mean(attn, dim=0)
-        return x, attn
-
-    def extra_repr(self):
-        return f"dim={self.dim}, input_resolution={self.input_resolution}, depth={self.depth}"
-
-
-# The Core of HTSAT
-class HTSAT_Swin_Transformer(nn.Module):
-    r"""HTSAT based on the Swin Transformer
-    Args:
-        spec_size (int | tuple(int)): Input Spectrogram size. Default 256
-        patch_size (int | tuple(int)): Patch size. Default: 4
-        path_stride (iot | tuple(int)): Patch Stride for Frequency and Time Axis. Default: 4
-        in_chans (int): Number of input image channels. Default: 1 (mono)
-        num_classes (int): Number of classes for classification head. Default: 527
-        embed_dim (int): Patch embedding dimension. Default: 96
-        depths (tuple(int)): Depth of each HTSAT-Swin Transformer layer.
-        num_heads (tuple(int)): Number of attention heads in different layers.
-        window_size (int): Window size. Default: 8
-        mlp_ratio (float): Ratio of mlp hidden dim to embedding dim. Default: 4
-        qkv_bias (bool): If True, add a learnable bias to query, key, value. Default: True
-        qk_scale (float): Override default qk scale of head_dim ** -0.5 if set. Default: None
-        drop_rate (float): Dropout rate. Default: 0
-        attn_drop_rate (float): Attention dropout rate. Default: 0
-        drop_path_rate (float): Stochastic depth rate. Default: 0.1
-        norm_layer (nn.Module): Normalization layer. Default: nn.LayerNorm.
-        ape (bool): If True, add absolute position embedding to the patch embedding. Default: False
-        patch_norm (bool): If True, add normalization after patch embedding. Default: True
-        use_checkpoint (bool): Whether to use checkpointing to save memory. Default: False
-        config (module): The configuration Module from config.py
-    """
-
-    def __init__(
-        self,
-        spec_size=256,
-        patch_size=4,
-        patch_stride=(4, 4),
-        in_chans=1,
-        num_classes=527,
-        embed_dim=96,
-        depths=[2, 2, 6, 2],
-        num_heads=[4, 8, 16, 32],
-        window_size=8,
-        mlp_ratio=4.0,
-        qkv_bias=True,
-        qk_scale=None,
-        drop_rate=0.0,
-        attn_drop_rate=0.0,
-        drop_path_rate=0.1,
-        norm_layer=nn.LayerNorm,
-        ape=False,
-        patch_norm=True,
-        use_checkpoint=False,
-        norm_before_mlp="ln",
-        config=None,
-        enable_fusion=False,
-        fusion_type="None",
-        **kwargs,
-    ):
-        super(HTSAT_Swin_Transformer, self).__init__()
-
-        self.config = config
-        self.spec_size = spec_size
-        self.patch_stride = patch_stride
-        self.patch_size = patch_size
-        self.window_size = window_size
-        self.embed_dim = embed_dim
-        self.depths = depths
-        self.ape = ape
-        self.in_chans = in_chans
-        self.num_classes = num_classes
-        self.num_heads = num_heads
-        self.num_layers = len(self.depths)
-        self.num_features = int(self.embed_dim * 2 ** (self.num_layers - 1))
-
-        self.drop_rate = drop_rate
-        self.attn_drop_rate = attn_drop_rate
-        self.drop_path_rate = drop_path_rate
-
-        self.qkv_bias = qkv_bias
-        self.qk_scale = None
-
-        self.patch_norm = patch_norm
-        self.norm_layer = norm_layer if self.patch_norm else None
-        self.norm_before_mlp = norm_before_mlp
-        self.mlp_ratio = mlp_ratio
-
-        self.use_checkpoint = use_checkpoint
-
-        self.enable_fusion = enable_fusion
-        self.fusion_type = fusion_type
-
-        #  process mel-spec ; used only once
-        self.freq_ratio = self.spec_size // self.config.mel_bins
-        window = "hann"
-        center = True
-        pad_mode = "reflect"
-        ref = 1.0
-        amin = 1e-10
-        top_db = None
-        self.interpolate_ratio = 32  # Downsampled ratio
-        # Spectrogram extractor
-        self.spectrogram_extractor = Spectrogram(
-            n_fft=config.window_size,
-            hop_length=config.hop_size,
-            win_length=config.window_size,
-            window=window,
-            center=center,
-            pad_mode=pad_mode,
-            freeze_parameters=True,
-        )
-        # Logmel feature extractor
-        self.logmel_extractor = LogmelFilterBank(
-            sr=config.sample_rate,
-            n_fft=config.window_size,
-            n_mels=config.mel_bins,
-            fmin=config.fmin,
-            fmax=config.fmax,
-            ref=ref,
-            amin=amin,
-            top_db=top_db,
-            freeze_parameters=True,
-        )
-        # Spec augmenter
-        self.spec_augmenter = SpecAugmentation(
-            time_drop_width=64,
-            time_stripes_num=2,
-            freq_drop_width=8,
-            freq_stripes_num=2,
-        )  # 2 2
-        self.bn0 = nn.BatchNorm2d(self.config.mel_bins)
-
-        # split spctrogram into non-overlapping patches
-        self.patch_embed = PatchEmbed(
-            img_size=self.spec_size,
-            patch_size=self.patch_size,
-            in_chans=self.in_chans,
-            embed_dim=self.embed_dim,
-            norm_layer=self.norm_layer,
-            patch_stride=patch_stride,
-            enable_fusion=self.enable_fusion,
-            fusion_type=self.fusion_type,
-        )
-
-        num_patches = self.patch_embed.num_patches
-        patches_resolution = self.patch_embed.grid_size
-        self.patches_resolution = patches_resolution
-
-        # absolute position embedding
-        if self.ape:
-            self.absolute_pos_embed = nn.Parameter(
-                torch.zeros(1, num_patches, self.embed_dim)
-            )
-            trunc_normal_(self.absolute_pos_embed, std=0.02)
-
-        self.pos_drop = nn.Dropout(p=self.drop_rate)
-
-        # stochastic depth
-        dpr = [
-            x.item() for x in torch.linspace(0, self.drop_path_rate, sum(self.depths))
-        ]  # stochastic depth decay rule
-
-        # build layers
-        self.layers = nn.ModuleList()
-        for i_layer in range(self.num_layers):
-            layer = BasicLayer(
-                dim=int(self.embed_dim * 2**i_layer),
-                input_resolution=(
-                    patches_resolution[0] // (2**i_layer),
-                    patches_resolution[1] // (2**i_layer),
-                ),
-                depth=self.depths[i_layer],
-                num_heads=self.num_heads[i_layer],
-                window_size=self.window_size,
-                mlp_ratio=self.mlp_ratio,
-                qkv_bias=self.qkv_bias,
-                qk_scale=self.qk_scale,
-                drop=self.drop_rate,
-                attn_drop=self.attn_drop_rate,
-                drop_path=dpr[
-                    sum(self.depths[:i_layer]) : sum(self.depths[: i_layer + 1])
-                ],
-                norm_layer=self.norm_layer,
-                downsample=PatchMerging if (i_layer < self.num_layers - 1) else None,
-                use_checkpoint=use_checkpoint,
-                norm_before_mlp=self.norm_before_mlp,
-            )
-            self.layers.append(layer)
-
-        self.norm = self.norm_layer(self.num_features)
-        self.avgpool = nn.AdaptiveAvgPool1d(1)
-        self.maxpool = nn.AdaptiveMaxPool1d(1)
-
-        SF = (
-            self.spec_size
-            // (2 ** (len(self.depths) - 1))
-            // self.patch_stride[0]
-            // self.freq_ratio
-        )
-        self.tscam_conv = nn.Conv2d(
-            in_channels=self.num_features,
-            out_channels=self.num_classes,
-            kernel_size=(SF, 3),
-            padding=(0, 1),
-        )
-        self.head = nn.Linear(num_classes, num_classes)
-
-        if (self.enable_fusion) and (
-            self.fusion_type in ["daf_1d", "aff_1d", "iaff_1d"]
-        ):
-            self.mel_conv1d = nn.Sequential(
-                nn.Conv1d(64, 64, kernel_size=5, stride=3, padding=2),
-                nn.BatchNorm1d(64),
-            )
-            if self.fusion_type == "daf_1d":
-                self.fusion_model = DAF()
-            elif self.fusion_type == "aff_1d":
-                self.fusion_model = AFF(channels=64, type="1D")
-            elif self.fusion_type == "iaff_1d":
-                self.fusion_model = iAFF(channels=64, type="1D")
-
-        self.apply(self._init_weights)
-
-    def _init_weights(self, m):
-        if isinstance(m, nn.Linear):
-            trunc_normal_(m.weight, std=0.02)
-            if isinstance(m, nn.Linear) and m.bias is not None:
-                nn.init.constant_(m.bias, 0)
-        elif isinstance(m, nn.LayerNorm):
-            nn.init.constant_(m.bias, 0)
-            nn.init.constant_(m.weight, 1.0)
-
-    @torch.jit.ignore
-    def no_weight_decay(self):
-        return {"absolute_pos_embed"}
-
-    @torch.jit.ignore
-    def no_weight_decay_keywords(self):
-        return {"relative_position_bias_table"}
-
-    def forward_features(self, x, longer_idx=None):
-        # A deprecated optimization for using a hierarchical output from different blocks
-
-        frames_num = x.shape[2]
-        x = self.patch_embed(x, longer_idx=longer_idx)
-        if self.ape:
-            x = x + self.absolute_pos_embed
-        x = self.pos_drop(x)
-        for i, layer in enumerate(self.layers):
-            x, attn = layer(x)
-        # for x
-        x = self.norm(x)
-        B, N, C = x.shape
-        SF = frames_num // (2 ** (len(self.depths) - 1)) // self.patch_stride[0]
-        ST = frames_num // (2 ** (len(self.depths) - 1)) // self.patch_stride[1]
-        x = x.permute(0, 2, 1).contiguous().reshape(B, C, SF, ST)
-        B, C, F, T = x.shape
-        # group 2D CNN
-        c_freq_bin = F // self.freq_ratio
-        x = x.reshape(B, C, F // c_freq_bin, c_freq_bin, T)
-        x = x.permute(0, 1, 3, 2, 4).contiguous().reshape(B, C, c_freq_bin, -1)
-        # get latent_output
-        fine_grained_latent_output = torch.mean(x, dim=2)
-        fine_grained_latent_output = interpolate(
-            fine_grained_latent_output.permute(0, 2, 1).contiguous(),
-            8 * self.patch_stride[1],
-        )
-
-        latent_output = self.avgpool(torch.flatten(x, 2))
-        latent_output = torch.flatten(latent_output, 1)
-
-        # display the attention map, if needed
-
-        x = self.tscam_conv(x)
-        x = torch.flatten(x, 2)  # B, C, T
-
-        fpx = interpolate(
-            torch.sigmoid(x).permute(0, 2, 1).contiguous(), 8 * self.patch_stride[1]
-        )
-
-        x = self.avgpool(x)
-        x = torch.flatten(x, 1)
-
-        output_dict = {
-            "framewise_output": fpx,  # already sigmoided
-            "clipwise_output": torch.sigmoid(x),
-            "fine_grained_embedding": fine_grained_latent_output,
-            "embedding": latent_output,
-        }
-
-        return output_dict
-
-    def crop_wav(self, x, crop_size, spe_pos=None):
-        time_steps = x.shape[2]
-        tx = torch.zeros(x.shape[0], x.shape[1], crop_size, x.shape[3]).to(x.device)
-        for i in range(len(x)):
-            if spe_pos is None:
-                crop_pos = random.randint(0, time_steps - crop_size - 1)
-            else:
-                crop_pos = spe_pos
-            tx[i][0] = x[i, 0, crop_pos : crop_pos + crop_size, :]
-        return tx
-
-    # Reshape the wavform to a img size, if you want to use the pretrained swin transformer model
-    def reshape_wav2img(self, x):
-        B, C, T, F = x.shape
-        target_T = int(self.spec_size * self.freq_ratio)
-        target_F = self.spec_size // self.freq_ratio
-        assert (
-            T <= target_T and F <= target_F
-        ), "the wav size should less than or equal to the swin input size"
-        # to avoid bicubic zero error
-        if T < target_T:
-            x = nn.functional.interpolate(
-                x, (target_T, x.shape[3]), mode="bicubic", align_corners=True
-            )
-        if F < target_F:
-            x = nn.functional.interpolate(
-                x, (x.shape[2], target_F), mode="bicubic", align_corners=True
-            )
-        x = x.permute(0, 1, 3, 2).contiguous()
-        x = x.reshape(
-            x.shape[0],
-            x.shape[1],
-            x.shape[2],
-            self.freq_ratio,
-            x.shape[3] // self.freq_ratio,
-        )
-        # print(x.shape)
-        x = x.permute(0, 1, 3, 2, 4).contiguous()
-        x = x.reshape(x.shape[0], x.shape[1], x.shape[2] * x.shape[3], x.shape[4])
-        return x
-
-    # Repeat the wavform to a img size, if you want to use the pretrained swin transformer model
-    def repeat_wat2img(self, x, cur_pos):
-        B, C, T, F = x.shape
-        target_T = int(self.spec_size * self.freq_ratio)
-        target_F = self.spec_size // self.freq_ratio
-        assert (
-            T <= target_T and F <= target_F
-        ), "the wav size should less than or equal to the swin input size"
-        # to avoid bicubic zero error
-        if T < target_T:
-            x = nn.functional.interpolate(
-                x, (target_T, x.shape[3]), mode="bicubic", align_corners=True
-            )
-        if F < target_F:
-            x = nn.functional.interpolate(
-                x, (x.shape[2], target_F), mode="bicubic", align_corners=True
-            )
-        x = x.permute(0, 1, 3, 2).contiguous()  # B C F T
-        x = x[:, :, :, cur_pos : cur_pos + self.spec_size]
-        x = x.repeat(repeats=(1, 1, 4, 1))
-        return x
-
-    def forward(
-        self, x: torch.Tensor, mixup_lambda=None, infer_mode=False, device=None
-    ):  # out_feat_keys: List[str] = None):
-
-        if self.enable_fusion and x["longer"].sum() == 0:
-            # if no audio is longer than 10s, then randomly select one audio to be longer
-            x["longer"][torch.randint(0, x["longer"].shape[0], (1,))] = True
-
-        if not self.enable_fusion:
-            x = x["waveform"].to(device=device, non_blocking=True)
-            x = self.spectrogram_extractor(x)  # (batch_size, 1, time_steps, freq_bins)
-            x = self.logmel_extractor(x)  # (batch_size, 1, time_steps, mel_bins)
-            x = x.transpose(1, 3)
-            x = self.bn0(x)
-            x = x.transpose(1, 3)
-            if self.training:
-                x = self.spec_augmenter(x)
-
-            if self.training and mixup_lambda is not None:
-                x = do_mixup(x, mixup_lambda)
-
-            x = self.reshape_wav2img(x)
-            output_dict = self.forward_features(x)
-        else:
-            longer_list = x["longer"].to(device=device, non_blocking=True)
-            x = x["mel_fusion"].to(device=device, non_blocking=True)
-            x = x.transpose(1, 3)
-            x = self.bn0(x)
-            x = x.transpose(1, 3)
-            longer_list_idx = torch.where(longer_list)[0]
-            if self.fusion_type in ["daf_1d", "aff_1d", "iaff_1d"]:
-                new_x = x[:, 0:1, :, :].clone().contiguous()
-                if len(longer_list_idx) > 0:
-                    # local processing
-                    fusion_x_local = x[longer_list_idx, 1:, :, :].clone().contiguous()
-                    FB, FC, FT, FF = fusion_x_local.size()
-                    fusion_x_local = fusion_x_local.view(FB * FC, FT, FF)
-                    fusion_x_local = torch.permute(
-                        fusion_x_local, (0, 2, 1)
-                    ).contiguous()
-                    fusion_x_local = self.mel_conv1d(fusion_x_local)
-                    fusion_x_local = fusion_x_local.view(
-                        FB, FC, FF, fusion_x_local.size(-1)
-                    )
-                    fusion_x_local = (
-                        torch.permute(fusion_x_local, (0, 2, 1, 3))
-                        .contiguous()
-                        .flatten(2)
-                    )
-                    if fusion_x_local.size(-1) < FT:
-                        fusion_x_local = torch.cat(
-                            [
-                                fusion_x_local,
-                                torch.zeros(
-                                    (FB, FF, FT - fusion_x_local.size(-1)),
-                                    device=device,
-                                ),
-                            ],
-                            dim=-1,
-                        )
-                    else:
-                        fusion_x_local = fusion_x_local[:, :, :FT]
-                    # 1D fusion
-                    new_x = new_x.squeeze(1).permute((0, 2, 1)).contiguous()
-                    new_x[longer_list_idx] = self.fusion_model(
-                        new_x[longer_list_idx], fusion_x_local
-                    )
-                    x = new_x.permute((0, 2, 1)).contiguous()[:, None, :, :]
-                else:
-                    x = new_x
-
-            elif self.fusion_type in ["daf_2d", "aff_2d", "iaff_2d", "channel_map"]:
-                x = x  # no change
-
-            if self.training:
-                x = self.spec_augmenter(x)
-            if self.training and mixup_lambda is not None:
-                x = do_mixup(x, mixup_lambda)
-
-            x = self.reshape_wav2img(x)
-            output_dict = self.forward_features(x, longer_idx=longer_list_idx)
-
-        # if infer_mode:
-        #     # in infer mode. we need to handle different length audio input
-        #     frame_num = x.shape[2]
-        #     target_T = int(self.spec_size * self.freq_ratio)
-        #     repeat_ratio = math.floor(target_T / frame_num)
-        #     x = x.repeat(repeats=(1,1,repeat_ratio,1))
-        #     x = self.reshape_wav2img(x)
-        #     output_dict = self.forward_features(x)
-        # else:
-        #     if x.shape[2] > self.freq_ratio * self.spec_size:
-        #         if self.training:
-        #             x = self.crop_wav(x, crop_size=self.freq_ratio * self.spec_size)
-        #             x = self.reshape_wav2img(x)
-        #             output_dict = self.forward_features(x)
-        #         else:
-        #             # Change: Hard code here
-        #             overlap_size = (x.shape[2] - 1) // 4
-        #             output_dicts = []
-        #             crop_size = (x.shape[2] - 1) // 2
-        #             for cur_pos in range(0, x.shape[2] - crop_size - 1, overlap_size):
-        #                 tx = self.crop_wav(x, crop_size = crop_size, spe_pos = cur_pos)
-        #                 tx = self.reshape_wav2img(tx)
-        #                 output_dicts.append(self.forward_features(tx))
-        #             clipwise_output = torch.zeros_like(output_dicts[0]["clipwise_output"]).float().to(x.device)
-        #             framewise_output = torch.zeros_like(output_dicts[0]["framewise_output"]).float().to(x.device)
-        #             for d in output_dicts:
-        #                 clipwise_output += d["clipwise_output"]
-        #                 framewise_output += d["framewise_output"]
-        #             clipwise_output  = clipwise_output / len(output_dicts)
-        #             framewise_output = framewise_output / len(output_dicts)
-        #             output_dict = {
-        #                 'framewise_output': framewise_output,
-        #                 'clipwise_output': clipwise_output
-        #             }
-        #     else: # this part is typically used, and most easy one
-        #         x = self.reshape_wav2img(x)
-        #         output_dict = self.forward_features(x)
-        # x = self.head(x)
-
-        # We process the data in the dataloader part, in that here we only consider the input_T < fixed_T
-
-        return output_dict
-
-
-def create_htsat_model(audio_cfg, enable_fusion=False, fusion_type="None"):
-    try:
-
-        assert audio_cfg.model_name in [
-            "tiny",
-            "base",
-            "large",
-        ], "model name for HTS-AT is wrong!"
-        if audio_cfg.model_name == "tiny":
-            model = HTSAT_Swin_Transformer(
-                spec_size=256,
-                patch_size=4,
-                patch_stride=(4, 4),
-                num_classes=audio_cfg.class_num,
-                embed_dim=96,
-                depths=[2, 2, 6, 2],
-                num_heads=[4, 8, 16, 32],
-                window_size=8,
-                config=audio_cfg,
-                enable_fusion=enable_fusion,
-                fusion_type=fusion_type,
-            )
-        elif audio_cfg.model_name == "base":
-            model = HTSAT_Swin_Transformer(
-                spec_size=256,
-                patch_size=4,
-                patch_stride=(4, 4),
-                num_classes=audio_cfg.class_num,
-                embed_dim=128,
-                depths=[2, 2, 12, 2],
-                num_heads=[4, 8, 16, 32],
-                window_size=8,
-                config=audio_cfg,
-                enable_fusion=enable_fusion,
-                fusion_type=fusion_type,
-            )
-        elif audio_cfg.model_name == "large":
-            model = HTSAT_Swin_Transformer(
-                spec_size=256,
-                patch_size=4,
-                patch_stride=(4, 4),
-                num_classes=audio_cfg.class_num,
-                embed_dim=256,
-                depths=[2, 2, 12, 2],
-                num_heads=[4, 8, 16, 32],
-                window_size=8,
-                config=audio_cfg,
-                enable_fusion=enable_fusion,
-                fusion_type=fusion_type,
-            )
-
-        return model
-    except:
-        raise RuntimeError(
-            f"Import Model for {audio_cfg.model_name} not found, or the audio cfg parameters are not enough."
-        )

audioldm/clap/open_clip/linear_probe.py

@@ -1,66 +0,0 @@
-import numpy as np
-import torch.nn.functional as F
-from torch import nn
-from .model import MLPLayers
-
-
-class LinearProbe(nn.Module):
-    def __init__(self, model, mlp, freeze, in_ch, out_ch, act=None):
-        """
-        Args:
-            model: nn.Module
-            mlp: bool, if True, then use the MLP layer as the linear probe module
-            freeze: bool, if Ture, then freeze all the CLAP model's layers when training the linear probe
-            in_ch: int, the output channel from CLAP model
-            out_ch: int, the output channel from linear probe (class_num)
-            act: torch.nn.functional, the activation function before the loss function
-        """
-        super().__init__()
-        in_ch = 512
-        self.clap_model = model
-        self.clap_model.text_branch = None  # to save memory
-        self.freeze = freeze
-        if mlp:
-            self.lp_layer = MLPLayers(units=[in_ch, in_ch * 2, out_ch])
-        else:
-            self.lp_layer = nn.Linear(in_ch, out_ch)
-
-        if self.freeze:
-            for param in self.clap_model.parameters():
-                param.requires_grad = False
-
-        if act == "None":
-            self.act = None
-        elif act == "relu":
-            self.act = nn.ReLU()
-        elif act == "elu":
-            self.act = nn.ELU()
-        elif act == "prelu":
-            self.act = nn.PReLU(num_parameters=in_ch)
-        elif act == "softmax":
-            self.act = nn.Softmax(dim=-1)
-        elif act == "sigmoid":
-            self.act = nn.Sigmoid()
-
-    def forward(self, x, mix_lambda=None, device=None):
-        """
-        Args:
-            x: waveform, torch.tensor [batch, t_samples] / batch of mel_spec and longer list
-            mix_lambda: torch.tensor [batch], the mixup lambda
-        Returns:
-            class_prob: torch.tensor [batch, class_num]
-
-        """
-        # batchnorm cancel grandient
-        if self.freeze:
-            self.clap_model.eval()
-
-        x = self.clap_model.audio_projection(
-            self.clap_model.audio_branch(x, mixup_lambda=mix_lambda, device=device)[
-                "embedding"
-            ]
-        )
-        out = self.lp_layer(x)
-        if self.act is not None:
-            out = self.act(out)
-        return out

audioldm/clap/open_clip/loss.py

@@ -1,398 +0,0 @@
-from multiprocessing.sharedctypes import Value
-import torch
-import torch.distributed.nn
-from torch import distributed as dist, nn as nn
-from torch.nn import functional as F
-import numpy as np
-from sklearn.metrics import average_precision_score, roc_auc_score, accuracy_score
-
-try:
-    import horovod.torch as hvd
-except ImportError:
-    hvd = None
-
-
-def gather_features(
-    audio_features,
-    text_features,
-    audio_features_mlp=None,
-    text_features_mlp=None,
-    local_loss=False,
-    gather_with_grad=False,
-    rank=0,
-    world_size=1,
-    use_horovod=False,
-    mlp_loss=False,
-):
-    if use_horovod:
-        assert hvd is not None, "Please install horovod"
-        if gather_with_grad:
-            all_audio_features = hvd.allgather(audio_features)
-            all_text_features = hvd.allgather(text_features)
-            if mlp_loss:
-                all_audio_features_mlp = hvd.allgather(audio_features_mlp)
-                all_text_features_mlp = hvd.allgather(text_features_mlp)
-        else:
-            with torch.no_grad():
-                all_audio_features = hvd.allgather(audio_features)
-                all_text_features = hvd.allgather(text_features)
-                if mlp_loss:
-                    all_audio_features_mlp = hvd.allgather(audio_features_mlp)
-                    all_text_features_mlp = hvd.allgather(text_features_mlp)
-            if not local_loss:
-                # ensure grads for local rank when all_* features don't have a gradient
-                gathered_audio_features = list(
-                    all_audio_features.chunk(world_size, dim=0)
-                )
-                gathered_text_features = list(
-                    all_text_features.chunk(world_size, dim=0)
-                )
-                gathered_audio_features[rank] = audio_features
-                gathered_text_features[rank] = text_features
-                all_audio_features = torch.cat(gathered_audio_features, dim=0)
-                all_text_features = torch.cat(gathered_text_features, dim=0)
-                if mlp_loss:
-                    gathered_audio_features_mlp = list(
-                        all_audio_features_mlp.chunk(world_size, dim=0)
-                    )
-                    gathered_text_features_mlp = list(
-                        all_text_features_mlp.chunk(world_size, dim=0)
-                    )
-                    gathered_audio_features_mlp[rank] = audio_features_mlp
-                    gathered_text_features_mlp[rank] = text_features_mlp
-                    all_audio_features_mlp = torch.cat(
-                        gathered_audio_features_mlp, dim=0
-                    )
-                    all_text_features_mlp = torch.cat(gathered_text_features_mlp, dim=0)
-    else:
-        # We gather tensors from all gpus
-        if gather_with_grad:
-            all_audio_features = torch.cat(
-                torch.distributed.nn.all_gather(audio_features), dim=0
-            )
-            all_text_features = torch.cat(
-                torch.distributed.nn.all_gather(text_features), dim=0
-            )
-            if mlp_loss:
-                all_audio_features_mlp = torch.cat(
-                    torch.distributed.nn.all_gather(audio_features_mlp), dim=0
-                )
-                all_text_features_mlp = torch.cat(
-                    torch.distributed.nn.all_gather(text_features_mlp), dim=0
-                )
-        else:
-            gathered_audio_features = [
-                torch.zeros_like(audio_features) for _ in range(world_size)
-            ]
-            gathered_text_features = [
-                torch.zeros_like(text_features) for _ in range(world_size)
-            ]
-            dist.all_gather(gathered_audio_features, audio_features)
-            dist.all_gather(gathered_text_features, text_features)
-            if mlp_loss:
-                gathered_audio_features_mlp = [
-                    torch.zeros_like(audio_features_mlp) for _ in range(world_size)
-                ]
-                gathered_text_features_mlp = [
-                    torch.zeros_like(text_features_mlp) for _ in range(world_size)
-                ]
-                dist.all_gather(gathered_audio_features_mlp, audio_features_mlp)
-                dist.all_gather(gathered_text_features_mlp, text_features_mlp)
-            if not local_loss:
-                # ensure grads for local rank when all_* features don't have a gradient
-                gathered_audio_features[rank] = audio_features
-                gathered_text_features[rank] = text_features
-                if mlp_loss:
-                    gathered_audio_features_mlp[rank] = audio_features_mlp
-                    gathered_text_features_mlp[rank] = text_features_mlp
-
-            all_audio_features = torch.cat(gathered_audio_features, dim=0)
-            all_text_features = torch.cat(gathered_text_features, dim=0)
-            if mlp_loss:
-                all_audio_features_mlp = torch.cat(gathered_audio_features_mlp, dim=0)
-                all_text_features_mlp = torch.cat(gathered_text_features_mlp, dim=0)
-    if mlp_loss:
-        return (
-            all_audio_features,
-            all_text_features,
-            all_audio_features_mlp,
-            all_text_features_mlp,
-        )
-    else:
-        return all_audio_features, all_text_features
-
-
-class ClipLoss(nn.Module):
-    def __init__(
-        self,
-        local_loss=False,
-        gather_with_grad=False,
-        cache_labels=False,
-        rank=0,
-        world_size=1,
-        use_horovod=False,
-        mlp_loss=False,
-        weight_loss_kappa=0,
-    ):
-        super().__init__()
-        self.local_loss = local_loss
-        self.gather_with_grad = gather_with_grad
-        self.cache_labels = cache_labels
-        self.rank = rank
-        self.world_size = world_size
-        self.use_horovod = use_horovod
-        self.mlp_loss = mlp_loss
-        self.weighted_loss = bool(weight_loss_kappa != 0)
-        self.weight_loss_kappa = weight_loss_kappa
-        # cache state
-        self.prev_num_logits = 0
-        self.labels = {}
-
-    def forward(
-        self,
-        audio_features,
-        text_features,
-        logit_scale_a,
-        logit_scale_t=None,
-        audio_features_mlp=None,
-        text_features_mlp=None,
-    ):
-        device = audio_features.device
-        if self.mlp_loss:
-            if self.world_size > 1:
-                (
-                    all_audio_features,
-                    all_text_features,
-                    all_audio_features_mlp,
-                    all_text_features_mlp,
-                ) = gather_features(
-                    audio_features=audio_features,
-                    text_features=text_features,
-                    audio_features_mlp=audio_features_mlp,
-                    text_features_mlp=text_features_mlp,
-                    local_loss=self.local_loss,
-                    gather_with_grad=self.gather_with_grad,
-                    rank=self.rank,
-                    world_size=self.world_size,
-                    use_horovod=self.use_horovod,
-                    mlp_loss=self.mlp_loss,
-                )
-                if self.local_loss:
-                    a_logits_per_audio = (
-                        logit_scale_a * audio_features @ all_text_features_mlp.T
-                    )
-                    a_logits_per_text = (
-                        logit_scale_a * text_features_mlp @ all_audio_features.T
-                    )
-                    t_logits_per_audio = (
-                        logit_scale_t * audio_features_mlp @ all_text_features.T
-                    )
-                    t_logits_per_text = (
-                        logit_scale_t * text_features @ all_audio_features_mlp.T
-                    )
-                else:
-                    a_logits_per_audio = (
-                        logit_scale_a * all_audio_features @ all_text_features_mlp.T
-                    )
-                    a_logits_per_text = a_logits_per_audio.T
-                    t_logits_per_audio = (
-                        logit_scale_t * all_audio_features_mlp @ all_text_features.T
-                    )
-                    t_logits_per_text = t_logits_per_audio.T
-            else:
-                a_logits_per_audio = (
-                    logit_scale_a * audio_features @ text_features_mlp.T
-                )
-                a_logits_per_text = logit_scale_a * text_features_mlp @ audio_features.T
-                t_logits_per_audio = (
-                    logit_scale_t * audio_features_mlp @ text_features.T
-                )
-                t_logits_per_text = logit_scale_t * text_features @ audio_features_mlp.T
-
-            # calculated ground-truth and cache if enabled
-            num_logits = a_logits_per_audio.shape[0]
-            if self.prev_num_logits != num_logits or device not in self.labels:
-                labels = torch.arange(num_logits, device=device, dtype=torch.long)
-                if self.world_size > 1 and self.local_loss:
-                    labels = labels + num_logits * self.rank
-                if self.cache_labels:
-                    self.labels[device] = labels
-                    self.prev_num_logits = num_logits
-            else:
-                labels = self.labels[device]
-
-            if not self.weighted_loss:
-                total_loss = (
-                    F.cross_entropy(a_logits_per_audio, labels)
-                    + F.cross_entropy(a_logits_per_text, labels)
-                    + F.cross_entropy(t_logits_per_audio, labels)
-                    + F.cross_entropy(t_logits_per_text, labels)
-                ) / 4
-            else:
-                audio_weight = (audio_features @ audio_features.T).detach()
-                audio_weight = (
-                    torch.exp(
-                        torch.sum(audio_weight, axis=1)
-                        / (self.weight_loss_kappa * len(audio_weight))
-                    )
-                ).detach()
-                text_weight = (text_features @ text_features.T).detach()
-                text_weight = (
-                    torch.exp(
-                        torch.sum(text_weight, axis=1)
-                        / (self.weight_loss_kappa * len(text_features))
-                    )
-                ).detach()
-                total_loss = (
-                    F.cross_entropy(a_logits_per_audio, labels, weight=audio_weight)
-                    + F.cross_entropy(a_logits_per_text, labels, weight=audio_weight)
-                    + F.cross_entropy(t_logits_per_audio, labels, weight=text_weight)
-                    + F.cross_entropy(t_logits_per_text, labels, weight=text_weight)
-                ) / 4
-        else:
-            if self.world_size > 1:
-                all_audio_features, all_text_features = gather_features(
-                    audio_features=audio_features,
-                    text_features=text_features,
-                    local_loss=self.local_loss,
-                    gather_with_grad=self.gather_with_grad,
-                    rank=self.rank,
-                    world_size=self.world_size,
-                    use_horovod=self.use_horovod,
-                    mlp_loss=self.mlp_loss,
-                )
-
-                if self.local_loss:
-                    logits_per_audio = (
-                        logit_scale_a * audio_features @ all_text_features.T
-                    )
-                    logits_per_text = (
-                        logit_scale_a * text_features @ all_audio_features.T
-                    )
-                else:
-                    logits_per_audio = (
-                        logit_scale_a * all_audio_features @ all_text_features.T
-                    )
-                    logits_per_text = logits_per_audio.T
-            else:
-                logits_per_audio = logit_scale_a * audio_features @ text_features.T
-                logits_per_text = logit_scale_a * text_features @ audio_features.T
-
-            # calculated ground-truth and cache if enabled
-            num_logits = logits_per_audio.shape[0]
-            if self.prev_num_logits != num_logits or device not in self.labels:
-                labels = torch.arange(num_logits, device=device, dtype=torch.long)
-                if self.world_size > 1 and self.local_loss:
-                    labels = labels + num_logits * self.rank
-                if self.cache_labels:
-                    self.labels[device] = labels
-                    self.prev_num_logits = num_logits
-            else:
-                labels = self.labels[device]
-            if not self.weighted_loss:
-                total_loss = (
-                    F.cross_entropy(logits_per_audio, labels)
-                    + F.cross_entropy(logits_per_text, labels)
-                ) / 2
-            else:
-                audio_weight = (all_audio_features @ all_audio_features.T).detach()
-                audio_weight = (
-                    torch.exp(
-                        torch.sum(audio_weight, axis=1)
-                        / (self.weight_loss_kappa * len(all_audio_features))
-                    )
-                ).detach()
-                text_weight = (all_text_features @ all_text_features.T).detach()
-                text_weight = (
-                    torch.exp(
-                        torch.sum(text_weight, axis=1)
-                        / (self.weight_loss_kappa * len(all_text_features))
-                    )
-                ).detach()
-                total_loss = (
-                    F.cross_entropy(logits_per_audio, labels, weight=text_weight)
-                    + F.cross_entropy(logits_per_text, labels, weight=audio_weight)
-                ) / 2
-        return total_loss
-
-
-def lp_gather_features(pred, target, world_size=1, use_horovod=False):
-    if use_horovod:
-        assert hvd is not None, "Please install horovod"
-        with torch.no_grad():
-            all_preds = hvd.allgather(pred)
-            all_targets = hvd.allgath(target)
-    else:
-        gathered_preds = [torch.zeros_like(pred) for _ in range(world_size)]
-        gathered_targets = [torch.zeros_like(target) for _ in range(world_size)]
-
-        dist.all_gather(gathered_preds, pred)
-        dist.all_gather(gathered_targets, target)
-        all_preds = torch.cat(gathered_preds, dim=0)
-        all_targets = torch.cat(gathered_targets, dim=0)
-
-    return all_preds, all_targets
-
-
-def get_map(pred, target):
-    pred = torch.sigmoid(pred).numpy()
-    target = target.numpy()
-    return np.mean(average_precision_score(target, pred, average=None))
-
-
-def get_acc(pred, target):
-    pred = torch.argmax(pred, 1).numpy()
-    target = torch.argmax(target, 1).numpy()
-    return accuracy_score(target, pred)
-
-
-def get_mauc(pred, target):
-    pred = torch.sigmoid(pred).numpy()
-    target = target.numpy()
-    return np.mean(roc_auc_score(target, pred, average=None))
-
-
-class LPMetrics(object):
-    def __init__(self, metric_names=["map", "acc", "mauc"]):
-        self.metrics = []
-        for name in metric_names:
-            self.metrics.append(self.get_metric(name))
-        self.metric_names = metric_names
-
-    def get_metric(self, name):
-        if name == "map":
-            return get_map
-        elif name == "acc":
-            return get_acc
-        elif name == "mauc":
-            return get_mauc
-        else:
-            raise ValueError(f"the metric should be at least one of [map, acc, mauc]")
-
-    def evaluate_mertics(self, pred, target):
-        metric_dict = {}
-        for i in range(len(self.metric_names)):
-            metric_dict[self.metric_names[i]] = self.metrics[i](pred, target)
-        return metric_dict
-
-
-def calc_celoss(pred, target):
-    target = torch.argmax(target, 1).long()
-    return nn.CrossEntropyLoss()(pred, target)
-
-
-class LPLoss(nn.Module):
-    def __init__(self, loss_name):
-        super().__init__()
-        if loss_name == "bce":
-            self.loss_func = nn.BCEWithLogitsLoss()
-        elif loss_name == "ce":
-            self.loss_func = calc_celoss
-        elif loss_name == "mse":
-            self.loss_func = nn.MSELoss()
-        else:
-            raise ValueError(f"the loss func should be at least one of [bce, ce, mse]")
-
-    def forward(self, pred, target):
-        loss = self.loss_func(pred, target)
-        return loss

audioldm/clap/open_clip/model.py

@@ -1,936 +0,0 @@
-""" CLAP Model
-
-Adapted from CLIP: https://github.com/openai/CLIP. Originally MIT License, Copyright (c) 2021 OpenAI.
-Adapted to the Audio Task.
-"""
-
-from collections import OrderedDict
-from dataclasses import dataclass
-from email.mime import audio
-from typing import Tuple, Union, Callable, Optional
-
-import numpy as np
-import torch
-import torch.nn.functional as F
-from torch import nn
-
-from .timm_model import TimmModel
-import logging
-from .utils import freeze_batch_norm_2d
-
-from .pann_model import create_pann_model
-from .htsat import create_htsat_model
-from transformers import BertModel, RobertaModel, BartModel
-from transformers.tokenization_utils_base import BatchEncoding
-
-
-class MLPLayers(nn.Module):
-    def __init__(self, units=[512, 512, 512], nonlin=nn.ReLU(), dropout=0.1):
-        super(MLPLayers, self).__init__()
-        self.nonlin = nonlin
-        self.dropout = dropout
-
-        sequence = []
-        for u0, u1 in zip(units[:-1], units[1:]):
-            sequence.append(nn.Linear(u0, u1))
-            sequence.append(self.nonlin)
-            sequence.append(nn.Dropout(self.dropout))
-        sequence = sequence[:-2]
-
-        self.sequential = nn.Sequential(*sequence)
-
-    def forward(self, X):
-        X = self.sequential(X)
-        return X
-
-
-class Bottleneck(nn.Module):
-    expansion = 4
-
-    def __init__(self, inplanes, planes, stride=1):
-        super().__init__()
-
-        # all conv layers have stride 1. an avgpool is performed after the second convolution when stride > 1
-        self.conv1 = nn.Conv2d(inplanes, planes, 1, bias=False)
-        self.bn1 = nn.BatchNorm2d(planes)
-
-        self.conv2 = nn.Conv2d(planes, planes, 3, padding=1, bias=False)
-        self.bn2 = nn.BatchNorm2d(planes)
-
-        self.avgpool = nn.AvgPool2d(stride) if stride > 1 else nn.Identity()
-
-        self.conv3 = nn.Conv2d(planes, planes * self.expansion, 1, bias=False)
-        self.bn3 = nn.BatchNorm2d(planes * self.expansion)
-
-        self.relu = nn.ReLU(inplace=True)
-        self.downsample = None
-        self.stride = stride
-
-        if stride > 1 or inplanes != planes * Bottleneck.expansion:
-            # downsampling layer is prepended with an avgpool, and the subsequent convolution has stride 1
-            self.downsample = nn.Sequential(
-                OrderedDict(
-                    [
-                        ("-1", nn.AvgPool2d(stride)),
-                        (
-                            "0",
-                            nn.Conv2d(
-                                inplanes,
-                                planes * self.expansion,
-                                1,
-                                stride=1,
-                                bias=False,
-                            ),
-                        ),
-                        ("1", nn.BatchNorm2d(planes * self.expansion)),
-                    ]
-                )
-            )
-
-    def forward(self, x: torch.Tensor):
-        identity = x
-
-        out = self.relu(self.bn1(self.conv1(x)))
-        out = self.relu(self.bn2(self.conv2(out)))
-        out = self.avgpool(out)
-        out = self.bn3(self.conv3(out))
-
-        if self.downsample is not None:
-            identity = self.downsample(x)
-
-        out += identity
-        out = self.relu(out)
-        return out
-
-
-class AttentionPool2d(nn.Module):
-    def __init__(
-        self, spacial_dim: int, embed_dim: int, num_heads: int, output_dim: int = None
-    ):
-        super().__init__()
-        self.positional_embedding = nn.Parameter(
-            torch.randn(spacial_dim**2 + 1, embed_dim) / embed_dim**0.5
-        )
-        self.k_proj = nn.Linear(embed_dim, embed_dim)
-        self.q_proj = nn.Linear(embed_dim, embed_dim)
-        self.v_proj = nn.Linear(embed_dim, embed_dim)
-        self.c_proj = nn.Linear(embed_dim, output_dim or embed_dim)
-        self.num_heads = num_heads
-
-    def forward(self, x):
-        x = x.reshape(x.shape[0], x.shape[1], x.shape[2] * x.shape[3]).permute(
-            2, 0, 1
-        )  # NCHW -> (HW)NC
-        x = torch.cat([x.mean(dim=0, keepdim=True), x], dim=0)  # (HW+1)NC
-        x = x + self.positional_embedding[:, None, :].to(x.dtype)  # (HW+1)NC
-        x, _ = F.multi_head_attention_forward(
-            query=x,
-            key=x,
-            value=x,
-            embed_dim_to_check=x.shape[-1],
-            num_heads=self.num_heads,
-            q_proj_weight=self.q_proj.weight,
-            k_proj_weight=self.k_proj.weight,
-            v_proj_weight=self.v_proj.weight,
-            in_proj_weight=None,
-            in_proj_bias=torch.cat(
-                [self.q_proj.bias, self.k_proj.bias, self.v_proj.bias]
-            ),
-            bias_k=None,
-            bias_v=None,
-            add_zero_attn=False,
-            dropout_p=0,
-            out_proj_weight=self.c_proj.weight,
-            out_proj_bias=self.c_proj.bias,
-            use_separate_proj_weight=True,
-            training=self.training,
-            need_weights=False,
-        )
-
-        return x[0]
-
-
-class ModifiedResNet(nn.Module):
-    """
-    A ResNet class that is similar to torchvision's but contains the following changes:
-    - There are now 3 "stem" convolutions as opposed to 1, with an average pool instead of a max pool.
-    - Performs anti-aliasing strided convolutions, where an avgpool is prepended to convolutions with stride > 1
-    - The final pooling layer is a QKV attention instead of an average pool
-    """
-
-    def __init__(self, layers, output_dim, heads, image_size=224, width=64):
-        super().__init__()
-        self.output_dim = output_dim
-        self.image_size = image_size
-
-        # the 3-layer stem
-        self.conv1 = nn.Conv2d(
-            3, width // 2, kernel_size=3, stride=2, padding=1, bias=False
-        )
-        self.bn1 = nn.BatchNorm2d(width // 2)
-        self.conv2 = nn.Conv2d(
-            width // 2, width // 2, kernel_size=3, padding=1, bias=False
-        )
-        self.bn2 = nn.BatchNorm2d(width // 2)
-        self.conv3 = nn.Conv2d(width // 2, width, kernel_size=3, padding=1, bias=False)
-        self.bn3 = nn.BatchNorm2d(width)
-        self.avgpool = nn.AvgPool2d(2)
-        self.relu = nn.ReLU(inplace=True)
-
-        # residual layers
-        self._inplanes = width  # this is a *mutable* variable used during construction
-        self.layer1 = self._make_layer(width, layers[0])
-        self.layer2 = self._make_layer(width * 2, layers[1], stride=2)
-        self.layer3 = self._make_layer(width * 4, layers[2], stride=2)
-        self.layer4 = self._make_layer(width * 8, layers[3], stride=2)
-
-        embed_dim = width * 32  # the ResNet feature dimension
-        self.attnpool = AttentionPool2d(image_size // 32, embed_dim, heads, output_dim)
-
-        self.init_parameters()
-
-    def _make_layer(self, planes, blocks, stride=1):
-        layers = [Bottleneck(self._inplanes, planes, stride)]
-
-        self._inplanes = planes * Bottleneck.expansion
-        for _ in range(1, blocks):
-            layers.append(Bottleneck(self._inplanes, planes))
-
-        return nn.Sequential(*layers)
-
-    def init_parameters(self):
-        if self.attnpool is not None:
-            std = self.attnpool.c_proj.in_features**-0.5
-            nn.init.normal_(self.attnpool.q_proj.weight, std=std)
-            nn.init.normal_(self.attnpool.k_proj.weight, std=std)
-            nn.init.normal_(self.attnpool.v_proj.weight, std=std)
-            nn.init.normal_(self.attnpool.c_proj.weight, std=std)
-
-        for resnet_block in [self.layer1, self.layer2, self.layer3, self.layer4]:
-            for name, param in resnet_block.named_parameters():
-                if name.endswith("bn3.weight"):
-                    nn.init.zeros_(param)
-
-    def lock(self, unlocked_groups=0, freeze_bn_stats=False):
-        assert (
-            unlocked_groups == 0
-        ), "partial locking not currently supported for this model"
-        for param in self.parameters():
-            param.requires_grad = False
-        if freeze_bn_stats:
-            freeze_batch_norm_2d(self)
-
-    def stem(self, x):
-        for conv, bn in [
-            (self.conv1, self.bn1),
-            (self.conv2, self.bn2),
-            (self.conv3, self.bn3),
-        ]:
-            x = self.relu(bn(conv(x)))
-        x = self.avgpool(x)
-        return x
-
-    def forward(self, x):
-        x = self.stem(x)
-        x = self.layer1(x)
-        x = self.layer2(x)
-        x = self.layer3(x)
-        x = self.layer4(x)
-        x = self.attnpool(x)
-
-        return x
-
-
-class LayerNorm(nn.LayerNorm):
-    """Subclass torch's LayerNorm to handle fp16."""
-
-    def forward(self, x: torch.Tensor):
-        orig_type = x.dtype
-        x = F.layer_norm(x, self.normalized_shape, self.weight, self.bias, self.eps)
-        return x.to(orig_type)
-
-
-class QuickGELU(nn.Module):
-    # NOTE This is slower than nn.GELU or nn.SiLU and uses more GPU memory
-    def forward(self, x: torch.Tensor):
-        return x * torch.sigmoid(1.702 * x)
-
-
-class ResidualAttentionBlock(nn.Module):
-    def __init__(self, d_model: int, n_head: int, act_layer: Callable = nn.GELU):
-        super().__init__()
-
-        self.attn = nn.MultiheadAttention(d_model, n_head)
-        self.ln_1 = LayerNorm(d_model)
-        self.mlp = nn.Sequential(
-            OrderedDict(
-                [
-                    ("c_fc", nn.Linear(d_model, d_model * 4)),
-                    ("gelu", act_layer()),
-                    ("c_proj", nn.Linear(d_model * 4, d_model)),
-                ]
-            )
-        )
-        self.ln_2 = LayerNorm(d_model)
-
-    def attention(self, x: torch.Tensor, attn_mask: Optional[torch.Tensor] = None):
-        return self.attn(x, x, x, need_weights=False, attn_mask=attn_mask)[0]
-
-    def forward(self, x: torch.Tensor, attn_mask: Optional[torch.Tensor] = None):
-        x = x + self.attention(self.ln_1(x), attn_mask=attn_mask)
-        x = x + self.mlp(self.ln_2(x))
-        return x
-
-
-class Transformer(nn.Module):
-    def __init__(
-        self, width: int, layers: int, heads: int, act_layer: Callable = nn.GELU
-    ):
-        super().__init__()
-        self.width = width
-        self.layers = layers
-        self.resblocks = nn.ModuleList(
-            [
-                ResidualAttentionBlock(width, heads, act_layer=act_layer)
-                for _ in range(layers)
-            ]
-        )
-
-    def forward(self, x: torch.Tensor, attn_mask: Optional[torch.Tensor] = None):
-        for r in self.resblocks:
-            x = r(x, attn_mask=attn_mask)
-        return x
-
-
-class VisualTransformer(nn.Module):
-    def __init__(
-        self,
-        image_size: int,
-        patch_size: int,
-        width: int,
-        layers: int,
-        heads: int,
-        output_dim: int,
-        act_layer: Callable = nn.GELU,
-    ):
-        super().__init__()
-        self.image_size = image_size
-        self.output_dim = output_dim
-        self.conv1 = nn.Conv2d(
-            in_channels=3,
-            out_channels=width,
-            kernel_size=patch_size,
-            stride=patch_size,
-            bias=False,
-        )
-
-        scale = width**-0.5
-        self.class_embedding = nn.Parameter(scale * torch.randn(width))
-        self.positional_embedding = nn.Parameter(
-            scale * torch.randn((image_size // patch_size) ** 2 + 1, width)
-        )
-        self.ln_pre = LayerNorm(width)
-
-        self.text_branch = Transformer(width, layers, heads, act_layer=act_layer)
-
-        self.ln_post = LayerNorm(width)
-        self.proj = nn.Parameter(scale * torch.randn(width, output_dim))
-
-    def lock(self, unlocked_groups=0, freeze_bn_stats=False):
-        assert (
-            unlocked_groups == 0
-        ), "partial locking not currently supported for this model"
-        for param in self.parameters():
-            param.requires_grad = False
-
-    def forward(self, x: torch.Tensor):
-        x = self.conv1(x)  # shape = [*, width, grid, grid]
-        x = x.reshape(x.shape[0], x.shape[1], -1)  # shape = [*, width, grid ** 2]
-        x = x.permute(0, 2, 1)  # shape = [*, grid ** 2, width]
-        x = torch.cat(
-            [
-                self.class_embedding.to(x.dtype)
-                + torch.zeros(
-                    x.shape[0], 1, x.shape[-1], dtype=x.dtype, device=x.device
-                ),
-                x,
-            ],
-            dim=1,
-        )  # shape = [*, grid ** 2 + 1, width]
-        x = x + self.positional_embedding.to(x.dtype)
-        x = self.ln_pre(x)
-
-        x = x.permute(1, 0, 2)  # NLD -> LND
-        x = self.text_branch(x)
-        x = x.permute(1, 0, 2)  # LND -> NLD
-
-        x = self.ln_post(x[:, 0, :])
-
-        if self.proj is not None:
-            x = x @ self.proj
-
-        return x
-
-
-@dataclass
-class CLAPVisionCfg:
-    layers: Union[Tuple[int, int, int, int], int] = 12
-    width: int = 768
-    patch_size: int = 16
-    image_size: Union[Tuple[int, int], int] = 224
-    timm_model_name: str = (
-        None  # a valid model name overrides layers, width, patch_size
-    )
-    timm_model_pretrained: bool = (
-        False  # use (imagenet) pretrained weights for named model
-    )
-    timm_pool: str = (
-        "avg"  # feature pooling for timm model ('abs_attn', 'rot_attn', 'avg', '')
-    )
-    timm_proj: str = (
-        "linear"  # linear projection for timm model output ('linear', 'mlp', '')
-    )
-
-
-# Audio Config Class
-@dataclass
-class CLAPAudioCfp:
-    model_type: str = "PANN"
-    model_name: str = "Cnn14"
-    sample_rate: int = 48000
-    # Param
-    audio_length: int = 1024
-    window_size: int = 1024
-    hop_size: int = 1024
-    fmin: int = 50
-    fmax: int = 14000
-    class_num: int = 527
-    mel_bins: int = 64
-    clip_samples: int = 480000
-
-
-@dataclass
-class CLAPTextCfg:
-    context_length: int
-    vocab_size: int
-    width: int
-    heads: int
-    layers: int
-    model_type: str
-
-
-class CLAP(nn.Module):
-    def __init__(
-        self,
-        embed_dim: int,
-        audio_cfg: CLAPAudioCfp,
-        text_cfg: CLAPTextCfg,
-        quick_gelu: bool = False,
-        enable_fusion: bool = False,
-        fusion_type: str = "None",
-        joint_embed_shape: int = 512,
-        mlp_act: str = "relu",
-    ):
-        super().__init__()
-        if isinstance(audio_cfg, dict):
-            audio_cfg = CLAPAudioCfp(**audio_cfg)
-        if isinstance(text_cfg, dict):
-            text_cfg = CLAPTextCfg(**text_cfg)
-
-        self.audio_cfg = audio_cfg
-        self.text_cfg = text_cfg
-        self.enable_fusion = enable_fusion
-        self.fusion_type = fusion_type
-        self.joint_embed_shape = joint_embed_shape
-        self.mlp_act = mlp_act
-
-        self.context_length = text_cfg.context_length
-
-        # OpenAI models are pretrained w/ QuickGELU but native nn.GELU is both faster and more
-        # memory efficient in recent PyTorch releases (>= 1.10).
-        # NOTE: timm models always use native GELU regardless of quick_gelu flag.
-        act_layer = QuickGELU if quick_gelu else nn.GELU
-
-        if mlp_act == "relu":
-            mlp_act_layer = nn.ReLU()
-        elif mlp_act == "gelu":
-            mlp_act_layer = nn.GELU()
-        else:
-            raise NotImplementedError
-
-        # audio branch
-        # audio branch parameters
-        if audio_cfg.model_type == "PANN":
-            self.audio_branch = create_pann_model(audio_cfg, enable_fusion, fusion_type)
-        elif audio_cfg.model_type == "HTSAT":
-            self.audio_branch = create_htsat_model(
-                audio_cfg, enable_fusion, fusion_type
-            )
-        else:
-            logging.error(f"Model config for {audio_cfg.model_type} not found")
-            raise RuntimeError(f"Model config for {audio_cfg.model_type} not found.")
-
-        # text branch
-        # text branch parameters
-        if text_cfg.model_type == "transformer":
-            self.text_branch = Transformer(
-                width=text_cfg.width,
-                layers=text_cfg.layers,
-                heads=text_cfg.heads,
-                act_layer=act_layer,
-            )
-            self.vocab_size = text_cfg.vocab_size
-            self.token_embedding = nn.Embedding(text_cfg.vocab_size, text_cfg.width)
-            self.positional_embedding = nn.Parameter(
-                torch.empty(self.context_length, text_cfg.width)
-            )
-            self.ln_final = LayerNorm(text_cfg.width)
-            self.text_transform = MLPLayers(
-                units=[
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                ],
-                dropout=0.1,
-            )
-            self.text_projection = nn.Sequential(
-                nn.Linear(text_cfg.width, self.joint_embed_shape),
-                mlp_act_layer,
-                nn.Linear(self.joint_embed_shape, self.joint_embed_shape),
-            )
-        elif text_cfg.model_type == "bert":
-            self.text_branch = BertModel.from_pretrained("bert-base-uncased")
-            self.text_transform = MLPLayers(
-                units=[
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                ],
-                dropout=0.1,
-            )
-            self.text_projection = nn.Sequential(
-                nn.Linear(768, self.joint_embed_shape),
-                mlp_act_layer,
-                nn.Linear(self.joint_embed_shape, self.joint_embed_shape),
-            )
-        elif text_cfg.model_type == "roberta":
-            self.text_branch = RobertaModel.from_pretrained("roberta-base")
-            self.text_transform = MLPLayers(
-                units=[
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                ],
-                dropout=0.1,
-            )
-            self.text_projection = nn.Sequential(
-                nn.Linear(768, self.joint_embed_shape),
-                mlp_act_layer,
-                nn.Linear(self.joint_embed_shape, self.joint_embed_shape),
-            )
-        elif text_cfg.model_type == "bart":
-            self.text_branch = BartModel.from_pretrained("facebook/bart-base")
-            self.text_transform = MLPLayers(
-                units=[
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                    self.joint_embed_shape,
-                ],
-                dropout=0.1,
-            )
-            self.text_projection = nn.Sequential(
-                nn.Linear(768, self.joint_embed_shape),
-                mlp_act_layer,
-                nn.Linear(self.joint_embed_shape, self.joint_embed_shape),
-            )
-        else:
-            logging.error(f"Model config for {text_cfg.model_type} not found")
-            raise RuntimeError(f"Model config for {text_cfg.model_type} not found.")
-        self.text_branch_type = text_cfg.model_type
-        # text branch parameters
-
-        # audio branch parameters
-        self.audio_transform = MLPLayers(
-            units=[
-                self.joint_embed_shape,
-                self.joint_embed_shape,
-                self.joint_embed_shape,
-            ],
-            dropout=0.1,
-        )
-
-        # below here is text branch parameters
-
-        # ============================================================================================================
-        self.audio_projection = nn.Sequential(
-            nn.Linear(embed_dim, self.joint_embed_shape),
-            mlp_act_layer,
-            nn.Linear(self.joint_embed_shape, self.joint_embed_shape),
-        )
-
-        self.logit_scale_a = nn.Parameter(torch.ones([]) * np.log(1 / 0.07))
-        self.logit_scale_t = nn.Parameter(torch.ones([]) * np.log(1 / 0.07))
-        self.register_buffer("attn_mask", self.build_attention_mask(), persistent=False)
-
-        self.init_text_branch_parameters()
-
-    def init_text_branch_parameters(self):
-        if self.text_branch_type == "transformer":
-            nn.init.normal_(self.token_embedding.weight, std=0.02)
-            nn.init.normal_(self.positional_embedding, std=0.01)
-            proj_std = (self.text_branch.width**-0.5) * (
-                (2 * self.text_branch.layers) ** -0.5
-            )
-            attn_std = self.text_branch.width**-0.5
-            fc_std = (2 * self.text_branch.width) ** -0.5
-            for block in self.text_branch.resblocks:
-                nn.init.normal_(block.attn.in_proj_weight, std=attn_std)
-                nn.init.normal_(block.attn.out_proj.weight, std=proj_std)
-                nn.init.normal_(block.mlp.c_fc.weight, std=fc_std)
-                nn.init.normal_(block.mlp.c_proj.weight, std=proj_std)
-        if self.text_branch_type == "bert" or self.text_branch_type == "roberta":
-            width = self.text_branch.embeddings.word_embeddings.weight.shape[-1]
-        elif self.text_branch_type == "bart":
-            width = self.text_branch.shared.weight.shape[-1]
-        else:
-            width = self.text_branch.width
-        nn.init.constant_(self.logit_scale_a, np.log(1 / 0.07))
-        nn.init.constant_(self.logit_scale_t, np.log(1 / 0.07))
-
-        # deprecated
-        # if hasattr(self.visual, 'init_parameters'):
-        # self.visual.init_parameters()
-
-        # if self.text_projection is not None:
-        #     nn.init.normal_(self.text_projection, std=width**-0.5)
-
-    def build_attention_mask(self):
-        # lazily create causal attention mask, with full attention between the vision tokens
-        # pytorch uses additive attention mask; fill with -inf
-        mask = torch.empty(self.context_length, self.context_length)
-        mask.fill_(float("-inf"))
-        mask.triu_(1)  # zero out the lower diagonal
-        return mask
-
-    def encode_audio(self, audio, device):
-        return self.audio_branch(
-            audio, mixup_lambda=None, device=device
-        )  # mix lambda needs to add
-
-    # def list_of_dict_of_tensor2dict_of_tensor(self, x, device):
-    #     tmp = {}
-    #     for k in x[0].keys():
-    #         tmp[k] = []
-    #         for i in range(len(x)):
-    #             tmp[k].append(x[i][k][:77])
-    #     for k in x[0].keys():
-    #         tmp[k] = torch.tensor(tmp[k]).to(device=device, non_blocking=True)
-    #     return tmp
-
-    def encode_text(self, text, device):
-        if self.text_branch_type == "transformer":
-            text = text.to(device=device, non_blocking=True)
-            x = self.token_embedding(text)  # [batch_size, n_ctx, d_model]
-
-            x = x + self.positional_embedding
-            x = x.permute(1, 0, 2)  # NLD -> LND
-            x = self.text_branch(x, attn_mask=self.attn_mask)
-            x = x.permute(1, 0, 2)  # LND -> NLD
-            x = self.ln_final(x)
-
-            # x.shape = [batch_size, n_ctx, transformer.width]
-            # take features from the eot embedding (eot_token is the highest number in each sequence)
-            x = self.text_projection(x[torch.arange(x.shape[0]), text.argmax(dim=-1)])
-        elif self.text_branch_type == "bert":
-            # text = self.list_of_dict_of_tensor2dict_of_tensor(text, device)
-            # text = BatchEncoding(text)
-            x = self.text_branch(
-                input_ids=text["input_ids"].to(device=device, non_blocking=True),
-                attention_mask=text["attention_mask"].to(
-                    device=device, non_blocking=True
-                ),
-                token_type_ids=text["token_type_ids"].to(
-                    device=device, non_blocking=True
-                ),
-            )["pooler_output"]
-            x = self.text_projection(x)
-        elif self.text_branch_type == "roberta":
-            x = self.text_branch(
-                input_ids=text["input_ids"].to(device=device, non_blocking=True),
-                attention_mask=text["attention_mask"].to(
-                    device=device, non_blocking=True
-                ),
-            )["pooler_output"]
-            x = self.text_projection(x)
-        elif self.text_branch_type == "bart":
-            x = torch.mean(
-                self.text_branch(
-                    input_ids=text["input_ids"].to(device=device, non_blocking=True),
-                    attention_mask=text["attention_mask"].to(
-                        device=device, non_blocking=True
-                    ),
-                )["encoder_last_hidden_state"],
-                axis=1,
-            )
-            x = self.text_projection(x)
-        else:
-            logging.error(f"Model type {self.text_branch_type} not found")
-            raise RuntimeError(f"Model type {self.text_branch_type} not found.")
-        return x
-
-    def forward(self, audio, text, device=None):
-        """Forward audio and text into the CLAP
-
-        Parameters
-        ----------
-        audio: torch.Tensor (batch_size, audio_length)
-            the time-domain audio input / the batch of mel_spec and longer list.
-        text: torch.Tensor () // need to add
-            the text token input
-        """
-        if device is None:
-            if audio is not None:
-                device = audio.device
-            elif text is not None:
-                device = text.device
-        if audio is None and text is None:
-            # a hack to get the logit scale
-            return self.logit_scale_a.exp(), self.logit_scale_t.exp()
-        elif audio is None:
-            return self.encode_text(text, device=device)
-        elif text is None:
-            return self.audio_projection(
-                self.encode_audio(audio, device=device)["embedding"]
-            )
-        audio_features = self.audio_projection(
-            self.encode_audio(audio, device=device)["embedding"]
-        )
-        audio_features = F.normalize(audio_features, dim=-1)
-
-        text_features = self.encode_text(text, device=device)
-        # print("text_features", text_features)
-        # print("text_features.shape", text_features.shape)
-        # print("text_features.type", type(text_features))
-        text_features = F.normalize(text_features, dim=-1)
-
-        audio_features_mlp = self.audio_transform(audio_features)
-        text_features_mlp = self.text_transform(text_features)
-        # Four outputs: audio features (basic & MLP), text features (basic & MLP)
-        return (
-            audio_features,
-            text_features,
-            audio_features_mlp,
-            text_features_mlp,
-            self.logit_scale_a.exp(),
-            self.logit_scale_t.exp(),
-        )
-
-    def get_logit_scale(self):
-        return self.logit_scale_a.exp(), self.logit_scale_t.exp()
-
-    def get_text_embedding(self, data):
-        """Get the text embedding from the model
-
-        Parameters
-        ----------
-        data: torch.Tensor
-            a tensor of text embedding
-
-        Returns
-        ----------
-        text_embed: torch.Tensor
-            a tensor of text_embeds (N, D)
-
-        """
-        device = next(self.parameters()).device
-        for k in data:
-            data[k] = data[k].to(device)
-            if(len(data[k].size()) < 2):
-                data[k] = data[k].unsqueeze(0)
-        text_embeds = self.encode_text(data, device=device)
-        text_embeds = F.normalize(text_embeds, dim=-1)
-
-        return text_embeds
-
-    def get_audio_embedding(self, data):
-        """Get the audio embedding from the model
-
-        Parameters
-        ----------
-        data: a list of dict
-            the audio input dict list from 'get_audio_feature' method
-
-        Returns
-        ----------
-        audio_embed: torch.Tensor
-            a tensor of audio_embeds (N, D)
-
-        """
-        device = next(self.parameters()).device
-        input_dict = {}
-        keys = data[0].keys()
-        for k in keys:
-            input_dict[k] = torch.cat([d[k].unsqueeze(0) for d in data], dim=0).to(
-                device
-            )
-
-        audio_embeds = self.audio_projection(
-            self.encode_audio(input_dict, device=device)["embedding"]
-        )
-        audio_embeds = F.normalize(audio_embeds, dim=-1)
-
-        return audio_embeds
-
-    def audio_infer(self, audio, hopsize=None, device=None):
-        """Forward one audio and produce the audio embedding
-
-        Parameters
-        ----------
-        audio:  (audio_length)
-            the time-domain audio input, notice that it must be only one input
-        hopsize: int
-            the overlap hopsize as the sliding window
-
-        Returns
-        ----------
-        output_dict: {
-            key: [n, (embedding_shape)] if "HTS-AT"
-            or
-            key: [(embedding_shape)] if "PANN"
-        }
-            the list of key values of the audio branch
-
-        """
-
-        assert not self.training, "the inference mode must be run at eval stage"
-        output_dict = {}
-        # PANN
-        if self.audio_cfg.model_type == "PANN":
-            audio_input = audio.unsqueeze(dim=0)
-            output_dict[key] = self.encode_audio(audio_input, device=device)[
-                key
-            ].squeeze(dim=0)
-        elif self.audio_cfg.model_type == "HTSAT":
-            # repeat
-            audio_len = len(audio)
-            k = self.audio_cfg.clip_samples // audio_len
-            if k > 1:
-                audio = audio.repeat(k)
-                audio_len = len(audio)
-
-            if hopsize is None:
-                hopsize = min(hopsize, audio_len)
-
-            if audio_len > self.audio_cfg.clip_samples:
-                audio_input = [
-                    audio[pos : pos + self.audio_cfg.clip_samples].clone()
-                    for pos in range(
-                        0, audio_len - self.audio_cfg.clip_samples, hopsize
-                    )
-                ]
-                audio_input.append(audio[-self.audio_cfg.clip_samples :].clone())
-                audio_input = torch.stack(audio_input)
-                output_dict[key] = self.encode_audio(audio_input, device=device)[key]
-            else:
-                audio_input = audio.unsqueeze(dim=0)
-                output_dict[key] = self.encode_audio(audio_input, device=device)[
-                    key
-                ].squeeze(dim=0)
-
-        return output_dict
-
-
-def convert_weights_to_fp16(model: nn.Module):
-    """Convert applicable model parameters to fp16"""
-
-    def _convert_weights_to_fp16(l):
-        if isinstance(l, (nn.Conv1d, nn.Conv2d, nn.Linear)):
-            l.weight.data = l.weight.data.half()
-            if l.bias is not None:
-                l.bias.data = l.bias.data.half()
-
-        if isinstance(l, nn.MultiheadAttention):
-            for attr in [
-                *[f"{s}_proj_weight" for s in ["in", "q", "k", "v"]],
-                "in_proj_bias",
-                "bias_k",
-                "bias_v",
-            ]:
-                tensor = getattr(l, attr)
-                if tensor is not None:
-                    tensor.data = tensor.data.half()
-
-        for name in ["text_projection", "proj"]:
-            if hasattr(l, name):
-                attr = getattr(l, name)
-                if attr is not None:
-                    attr.data = attr.data.half()
-
-    model.apply(_convert_weights_to_fp16)
-
-
-# Ignore the state dict of the vision part
-def build_model_from_openai_state_dict(
-    state_dict: dict, model_cfg, enable_fusion: bool = False, fusion_type: str = "None"
-):
-
-    embed_dim = model_cfg["embed_dim"]
-    audio_cfg = model_cfg["audio_cfg"]
-    text_cfg = model_cfg["text_cfg"]
-    context_length = state_dict["positional_embedding"].shape[0]
-    vocab_size = state_dict["token_embedding.weight"].shape[0]
-    transformer_width = state_dict["ln_final.weight"].shape[0]
-    transformer_heads = transformer_width // 64
-    transformer_layers = len(
-        set(
-            k.split(".")[2]
-            for k in state_dict
-            if k.startswith(f"transformer.resblocks")
-        )
-    )
-
-    audio_cfg = CLAPAudioCfp(**audio_cfg)
-    text_cfg = CLAPTextCfg(**text_cfg)
-
-    model = CLAP(
-        embed_dim,
-        audio_cfg=audio_cfg,
-        text_cfg=text_cfg,
-        quick_gelu=True,  # OpenAI models were trained with QuickGELU
-        enable_fusion=enable_fusion,
-        fusion_type=fusion_type,
-    )
-    state_dict["logit_scale_a"] = state_dict["logit_scale"]
-    state_dict["logit_scale_t"] = state_dict["logit_scale"]
-    pop_keys = list(state_dict.keys())[::]
-    # pop the visual branch saved weights
-    for key in pop_keys:
-        if key.startswith("visual."):
-            state_dict.pop(key, None)
-
-    for key in ["logit_scale", "input_resolution", "context_length", "vocab_size"]:
-        state_dict.pop(key, None)
-
-    # not use fp16
-    # convert_weights_to_fp16(model)
-    model.load_state_dict(state_dict, strict=False)
-    return model.eval()
-
-
-def trace_model(model, batch_size=256, device=torch.device("cpu")):
-    model.eval()
-    audio_length = model.audio_cfg.audio_length
-    example_audio = torch.ones((batch_size, audio_length), device=device)
-    example_text = torch.zeros(
-        (batch_size, model.context_length), dtype=torch.int, device=device
-    )
-    model = torch.jit.trace_module(
-        model,
-        inputs=dict(
-            forward=(example_audio, example_text),
-            encode_text=(example_text,),
-            encode_image=(example_audio,),
-        ),
-    )
-    model.audio_cfg.audio_length = audio_length  # Question: what does this do?
-    return model

audioldm/clap/open_clip/model_configs/HTSAT-base.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 1024,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 480000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1024,
-        "hop_size": 480,
-        "fmin": 50,
-        "fmax": 14000,
-        "class_num": 527,
-        "model_type": "HTSAT",
-        "model_name": "base"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/HTSAT-large.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 2048,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 480000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1024,
-        "hop_size": 480,
-        "fmin": 50,
-        "fmax": 14000,
-        "class_num": 527,
-        "model_type": "HTSAT",
-        "model_name": "large"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/HTSAT-tiny-win-1536.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 768,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 480000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1536,
-        "hop_size": 480,
-        "fmin": 50,
-        "fmax": 14000,
-        "class_num": 527,
-        "model_type": "HTSAT",
-        "model_name": "tiny"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/HTSAT-tiny.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 768,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 480000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1024,
-        "hop_size": 480,
-        "fmin": 50,
-        "fmax": 14000,
-        "class_num": 527,
-        "model_type": "HTSAT",
-        "model_name": "tiny"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/PANN-10.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 1024,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 480000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1024,
-        "hop_size": 480,
-        "fmin": 50,
-        "fmax": 14000,
-        "class_num": 527,
-        "model_type": "PANN",
-        "model_name": "Cnn10"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/PANN-14-fmax-18k.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 2048,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 480000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1024,
-        "hop_size": 480,
-        "fmin": 50,
-        "fmax": 18000,
-        "class_num": 527,
-        "model_type": "PANN",
-        "model_name": "Cnn14"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/PANN-14-fmax-8k-20s.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 2048,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 960000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1024,
-        "hop_size": 360,
-        "fmin": 50,
-        "fmax": 8000,
-        "class_num": 527,
-        "model_type": "PANN",
-        "model_name": "Cnn14"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/PANN-14-tiny-transformer.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 2048,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 480000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1024,
-        "hop_size": 480,
-        "fmin": 50,
-        "fmax": 14000,
-        "class_num": 527,
-        "model_type": "PANN",
-        "model_name": "Cnn14"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 4
-    }
-}

audioldm/clap/open_clip/model_configs/PANN-14-win-1536.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 2048,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 480000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1536,
-        "hop_size": 480,
-        "fmin": 50,
-        "fmax": 14000,
-        "class_num": 527,
-        "model_type": "PANN",
-        "model_name": "Cnn14"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/PANN-14.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 2048,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 480000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1024,
-        "hop_size": 480,
-        "fmin": 50,
-        "fmax": 14000,
-        "class_num": 527,
-        "model_type": "PANN",
-        "model_name": "Cnn14"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/PANN-6.json

@@ -1,23 +0,0 @@
-{
-    "embed_dim": 512,
-    "audio_cfg": {
-        "audio_length": 1024,
-        "clip_samples": 480000,
-        "mel_bins": 64,
-        "sample_rate": 48000,
-        "window_size": 1024,
-        "hop_size": 480,
-        "fmin": 50,
-        "fmax": 14000,
-        "class_num": 527,
-        "model_type": "PANN",
-        "model_name": "Cnn6"
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/RN101-quickgelu.json

@@ -1,22 +0,0 @@
-{
-    "embed_dim": 512,
-    "quick_gelu": true,
-    "vision_cfg": {
-        "image_size": 224,
-        "layers": [
-            3,
-            4,
-            23,
-            3
-        ],
-        "width": 64,
-        "patch_size": null
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/RN101.json

@@ -1,21 +0,0 @@
-{
-    "embed_dim": 512,
-    "vision_cfg": {
-        "image_size": 224,
-        "layers": [
-            3,
-            4,
-            23,
-            3
-        ],
-        "width": 64,
-        "patch_size": null
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/RN50-quickgelu.json

@@ -1,22 +0,0 @@
-{
-    "embed_dim": 1024,
-    "quick_gelu": true,
-    "vision_cfg": {
-        "image_size": 224,
-        "layers": [
-            3,
-            4,
-            6,
-            3
-        ],
-        "width": 64,
-        "patch_size": null
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/RN50.json

@@ -1,21 +0,0 @@
-{
-    "embed_dim": 1024,
-    "vision_cfg": {
-        "image_size": 224,
-        "layers": [
-            3,
-            4,
-            6,
-            3
-        ],
-        "width": 64,
-        "patch_size": null
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/RN50x16.json

@@ -1,21 +0,0 @@
-{
-    "embed_dim": 768,
-    "vision_cfg": {
-        "image_size": 384,
-        "layers": [
-            6,
-            8,
-            18,
-            8
-        ],
-        "width": 96,
-        "patch_size": null
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 768,
-        "heads": 12,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/RN50x4.json

@@ -1,21 +0,0 @@
-{
-    "embed_dim": 640,
-    "vision_cfg": {
-        "image_size": 288,
-        "layers": [
-            4,
-            6,
-            10,
-            6
-        ],
-        "width": 80,
-        "patch_size": null
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 640,
-        "heads": 10,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/ViT-B-16.json

@@ -1,16 +0,0 @@
-{
-    "embed_dim": 512,
-    "vision_cfg": {
-        "image_size": 224,
-        "layers": 12,
-        "width": 768,
-        "patch_size": 16
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/ViT-B-32-quickgelu.json

@@ -1,17 +0,0 @@
-{
-    "embed_dim": 512,
-    "quick_gelu": true,
-    "vision_cfg": {
-        "image_size": 224,
-        "layers": 12,
-        "width": 768,
-        "patch_size": 32
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/ViT-B-32.json

@@ -1,16 +0,0 @@
-{
-    "embed_dim": 512,
-    "vision_cfg": {
-        "image_size": 224,
-        "layers": 12,
-        "width": 768,
-        "patch_size": 32
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 512,
-        "heads": 8,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/model_configs/ViT-L-14.json

@@ -1,16 +0,0 @@
-{
-    "embed_dim": 768,
-    "vision_cfg": {
-        "image_size": 224,
-        "layers": 24,
-        "width": 1024,
-        "patch_size": 14
-    },
-    "text_cfg": {
-        "context_length": 77,
-        "vocab_size": 49408,
-        "width": 768,
-        "heads": 12,
-        "layers": 12
-    }
-}

audioldm/clap/open_clip/openai.py

@@ -1,156 +0,0 @@
-""" OpenAI pretrained model functions
-
-Adapted from https://github.com/openai/CLIP. Originally MIT License, Copyright (c) 2021 OpenAI.
-"""
-
-import os
-import warnings
-from typing import Union, List
-
-import torch
-
-from .model import build_model_from_openai_state_dict
-from .pretrained import (
-    get_pretrained_url,
-    list_pretrained_tag_models,
-    download_pretrained,
-)
-
-__all__ = ["list_openai_models", "load_openai_model"]
-
-
-def list_openai_models() -> List[str]:
-    """Returns the names of available CLIP models"""
-    return list_pretrained_tag_models("openai")
-
-
-def load_openai_model(
-    name: str,
-    model_cfg,
-    device: Union[str, torch.device] = "cuda" if torch.cuda.is_available() else "cpu",
-    jit=True,
-    cache_dir=os.path.expanduser("~/.cache/clip"),
-    enable_fusion: bool = False,
-    fusion_type: str = "None",
-):
-    """Load a CLIP model, preserve its text pretrained part, and set in the CLAP model
-
-    Parameters
-    ----------
-    name : str
-        A model name listed by `clip.available_models()`, or the path to a model checkpoint containing the state_dict
-    device : Union[str, torch.device]
-        The device to put the loaded model
-    jit : bool
-        Whether to load the optimized JIT model (default) or more hackable non-JIT model.
-
-    Returns
-    -------
-    model : torch.nn.Module
-        The CLAP model
-    preprocess : Callable[[PIL.Image], torch.Tensor]
-        A torchvision transform that converts a PIL image into a tensor that the returned model can take as its input
-    """
-    if get_pretrained_url(name, "openai"):
-        model_path = download_pretrained(
-            get_pretrained_url(name, "openai"), root=cache_dir
-        )
-    elif os.path.isfile(name):
-        model_path = name
-    else:
-        raise RuntimeError(
-            f"Model {name} not found; available models = {list_openai_models()}"
-        )
-
-    try:
-        # loading JIT archive
-        model = torch.jit.load(model_path, map_location=device if jit else "cpu").eval()
-        state_dict = None
-    except RuntimeError:
-        # loading saved state dict
-        if jit:
-            warnings.warn(
-                f"File {model_path} is not a JIT archive. Loading as a state dict instead"
-            )
-            jit = False
-        state_dict = torch.load(model_path, map_location="cpu")
-
-    if not jit:
-        try:
-            model = build_model_from_openai_state_dict(
-                state_dict or model.state_dict(), model_cfg, enable_fusion, fusion_type
-            ).to(device)
-        except KeyError:
-            sd = {k[7:]: v for k, v in state_dict["state_dict"].items()}
-            model = build_model_from_openai_state_dict(
-                sd, model_cfg, enable_fusion, fusion_type
-            ).to(device)
-
-        if str(device) == "cpu":
-            model.float()
-        return model
-
-    # patch the device names
-    device_holder = torch.jit.trace(
-        lambda: torch.ones([]).to(torch.device(device)), example_inputs=[]
-    )
-    device_node = [
-        n
-        for n in device_holder.graph.findAllNodes("prim::Constant")
-        if "Device" in repr(n)
-    ][-1]
-
-    def patch_device(module):
-        try:
-            graphs = [module.graph] if hasattr(module, "graph") else []
-        except RuntimeError:
-            graphs = []
-
-        if hasattr(module, "forward1"):
-            graphs.append(module.forward1.graph)
-
-        for graph in graphs:
-            for node in graph.findAllNodes("prim::Constant"):
-                if "value" in node.attributeNames() and str(node["value"]).startswith(
-                    "cuda"
-                ):
-                    node.copyAttributes(device_node)
-
-    model.apply(patch_device)
-    patch_device(model.encode_audio)
-    patch_device(model.encode_text)
-
-    # patch dtype to float32 on CPU
-    if str(device) == "cpu":
-        float_holder = torch.jit.trace(
-            lambda: torch.ones([]).float(), example_inputs=[]
-        )
-        float_input = list(float_holder.graph.findNode("aten::to").inputs())[1]
-        float_node = float_input.node()
-
-        def patch_float(module):
-            try:
-                graphs = [module.graph] if hasattr(module, "graph") else []
-            except RuntimeError:
-                graphs = []
-
-            if hasattr(module, "forward1"):
-                graphs.append(module.forward1.graph)
-
-            for graph in graphs:
-                for node in graph.findAllNodes("aten::to"):
-                    inputs = list(node.inputs())
-                    for i in [
-                        1,
-                        2,
-                    ]:  # dtype can be the second or third argument to aten::to()
-                        if inputs[i].node()["value"] == 5:
-                            inputs[i].node().copyAttributes(float_node)
-
-        model.apply(patch_float)
-        patch_float(model.encode_audio)
-        patch_float(model.encode_text)
-        model.float()
-
-    model.audio_branch.audio_length = model.audio_cfg.audio_length
-    return model

audioldm/clap/open_clip/pann_model.py

@@ -1,703 +0,0 @@
-# PANNs: Large-Scale Pretrained Audio Neural Networks for Audio Pattern Recognition
-# Reference from https://github.com/qiuqiangkong/audioset_tagging_cnn
-# Some layers are re-designed for CLAP
-import os
-
-os.environ["NUMBA_CACHE_DIR"] = "/tmp/"
-
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-from torchlibrosa.stft import Spectrogram, LogmelFilterBank
-from torchlibrosa.augmentation import SpecAugmentation
-
-from .utils import do_mixup, interpolate, pad_framewise_output
-from .feature_fusion import iAFF, AFF, DAF
-
-
-def init_layer(layer):
-    """Initialize a Linear or Convolutional layer."""
-    nn.init.xavier_uniform_(layer.weight)
-
-    if hasattr(layer, "bias"):
-        if layer.bias is not None:
-            layer.bias.data.fill_(0.0)
-
-def init_bn(bn):
-    """Initialize a Batchnorm layer."""
-    bn.bias.data.fill_(0.0)
-    bn.weight.data.fill_(1.0)
-
-
-class ConvBlock(nn.Module):
-    def __init__(self, in_channels, out_channels):
-
-        super(ConvBlock, self).__init__()
-
-        self.conv1 = nn.Conv2d(
-            in_channels=in_channels,
-            out_channels=out_channels,
-            kernel_size=(3, 3),
-            stride=(1, 1),
-            padding=(1, 1),
-            bias=False,
-        )
-
-        self.conv2 = nn.Conv2d(
-            in_channels=out_channels,
-            out_channels=out_channels,
-            kernel_size=(3, 3),
-            stride=(1, 1),
-            padding=(1, 1),
-            bias=False,
-        )
-
-        self.bn1 = nn.BatchNorm2d(out_channels)
-        self.bn2 = nn.BatchNorm2d(out_channels)
-
-        self.init_weight()
-
-    def init_weight(self):
-        init_layer(self.conv1)
-        init_layer(self.conv2)
-        init_bn(self.bn1)
-        init_bn(self.bn2)
-
-    def forward(self, input, pool_size=(2, 2), pool_type="avg"):
-
-        x = input
-        x = F.relu_(self.bn1(self.conv1(x)))
-        x = F.relu_(self.bn2(self.conv2(x)))
-        if pool_type == "max":
-            x = F.max_pool2d(x, kernel_size=pool_size)
-        elif pool_type == "avg":
-            x = F.avg_pool2d(x, kernel_size=pool_size)
-        elif pool_type == "avg+max":
-            x1 = F.avg_pool2d(x, kernel_size=pool_size)
-            x2 = F.max_pool2d(x, kernel_size=pool_size)
-            x = x1 + x2
-        else:
-            raise Exception("Incorrect argument!")
-
-        return x
-
-
-class ConvBlock5x5(nn.Module):
-    def __init__(self, in_channels, out_channels):
-
-        super(ConvBlock5x5, self).__init__()
-
-        self.conv1 = nn.Conv2d(
-            in_channels=in_channels,
-            out_channels=out_channels,
-            kernel_size=(5, 5),
-            stride=(1, 1),
-            padding=(2, 2),
-            bias=False,
-        )
-
-        self.bn1 = nn.BatchNorm2d(out_channels)
-
-        self.init_weight()
-
-    def init_weight(self):
-        init_layer(self.conv1)
-        init_bn(self.bn1)
-
-    def forward(self, input, pool_size=(2, 2), pool_type="avg"):
-
-        x = input
-        x = F.relu_(self.bn1(self.conv1(x)))
-        if pool_type == "max":
-            x = F.max_pool2d(x, kernel_size=pool_size)
-        elif pool_type == "avg":
-            x = F.avg_pool2d(x, kernel_size=pool_size)
-        elif pool_type == "avg+max":
-            x1 = F.avg_pool2d(x, kernel_size=pool_size)
-            x2 = F.max_pool2d(x, kernel_size=pool_size)
-            x = x1 + x2
-        else:
-            raise Exception("Incorrect argument!")
-
-        return x
-
-
-class AttBlock(nn.Module):
-    def __init__(self, n_in, n_out, activation="linear", temperature=1.0):
-        super(AttBlock, self).__init__()
-
-        self.activation = activation
-        self.temperature = temperature
-        self.att = nn.Conv1d(
-            in_channels=n_in,
-            out_channels=n_out,
-            kernel_size=1,
-            stride=1,
-            padding=0,
-            bias=True,
-        )
-        self.cla = nn.Conv1d(
-            in_channels=n_in,
-            out_channels=n_out,
-            kernel_size=1,
-            stride=1,
-            padding=0,
-            bias=True,
-        )
-
-        self.bn_att = nn.BatchNorm1d(n_out)
-        self.init_weights()
-
-    def init_weights(self):
-        init_layer(self.att)
-        init_layer(self.cla)
-        init_bn(self.bn_att)
-
-    def forward(self, x):
-        # x: (n_samples, n_in, n_time)
-        norm_att = torch.softmax(torch.clamp(self.att(x), -10, 10), dim=-1)
-        cla = self.nonlinear_transform(self.cla(x))
-        x = torch.sum(norm_att * cla, dim=2)
-        return x, norm_att, cla
-
-    def nonlinear_transform(self, x):
-        if self.activation == "linear":
-            return x
-        elif self.activation == "sigmoid":
-            return torch.sigmoid(x)
-
-
-class Cnn14(nn.Module):
-    def __init__(
-        self,
-        sample_rate,
-        window_size,
-        hop_size,
-        mel_bins,
-        fmin,
-        fmax,
-        classes_num,
-        enable_fusion=False,
-        fusion_type="None",
-    ):
-
-        super(Cnn14, self).__init__()
-
-        window = "hann"
-        center = True
-        pad_mode = "reflect"
-        ref = 1.0
-        amin = 1e-10
-        top_db = None
-
-        self.enable_fusion = enable_fusion
-        self.fusion_type = fusion_type
-
-        # Spectrogram extractor
-        self.spectrogram_extractor = Spectrogram(
-            n_fft=window_size,
-            hop_length=hop_size,
-            win_length=window_size,
-            window=window,
-            center=center,
-            pad_mode=pad_mode,
-            freeze_parameters=True,
-        )
-
-        # Logmel feature extractor
-        self.logmel_extractor = LogmelFilterBank(
-            sr=sample_rate,
-            n_fft=window_size,
-            n_mels=mel_bins,
-            fmin=fmin,
-            fmax=fmax,
-            ref=ref,
-            amin=amin,
-            top_db=top_db,
-            freeze_parameters=True,
-        )
-
-        # Spec augmenter
-        self.spec_augmenter = SpecAugmentation(
-            time_drop_width=64,
-            time_stripes_num=2,
-            freq_drop_width=8,
-            freq_stripes_num=2,
-        )
-
-        self.bn0 = nn.BatchNorm2d(64)
-
-        if (self.enable_fusion) and (self.fusion_type == "channel_map"):
-            self.conv_block1 = ConvBlock(in_channels=4, out_channels=64)
-        else:
-            self.conv_block1 = ConvBlock(in_channels=1, out_channels=64)
-        self.conv_block2 = ConvBlock(in_channels=64, out_channels=128)
-        self.conv_block3 = ConvBlock(in_channels=128, out_channels=256)
-        self.conv_block4 = ConvBlock(in_channels=256, out_channels=512)
-        self.conv_block5 = ConvBlock(in_channels=512, out_channels=1024)
-        self.conv_block6 = ConvBlock(in_channels=1024, out_channels=2048)
-
-        self.fc1 = nn.Linear(2048, 2048, bias=True)
-        self.fc_audioset = nn.Linear(2048, classes_num, bias=True)
-
-        if (self.enable_fusion) and (
-            self.fusion_type in ["daf_1d", "aff_1d", "iaff_1d"]
-        ):
-            self.mel_conv1d = nn.Sequential(
-                nn.Conv1d(64, 64, kernel_size=5, stride=3, padding=2),
-                nn.BatchNorm1d(64),  # No Relu
-            )
-            if self.fusion_type == "daf_1d":
-                self.fusion_model = DAF()
-            elif self.fusion_type == "aff_1d":
-                self.fusion_model = AFF(channels=64, type="1D")
-            elif self.fusion_type == "iaff_1d":
-                self.fusion_model = iAFF(channels=64, type="1D")
-
-        if (self.enable_fusion) and (
-            self.fusion_type in ["daf_2d", "aff_2d", "iaff_2d"]
-        ):
-            self.mel_conv2d = nn.Sequential(
-                nn.Conv2d(1, 64, kernel_size=(5, 5), stride=(6, 2), padding=(2, 2)),
-                nn.BatchNorm2d(64),
-                nn.ReLU(inplace=True),
-            )
-
-            if self.fusion_type == "daf_2d":
-                self.fusion_model = DAF()
-            elif self.fusion_type == "aff_2d":
-                self.fusion_model = AFF(channels=64, type="2D")
-            elif self.fusion_type == "iaff_2d":
-                self.fusion_model = iAFF(channels=64, type="2D")
-        self.init_weight()
-
-    def init_weight(self):
-        init_bn(self.bn0)
-        init_layer(self.fc1)
-        init_layer(self.fc_audioset)
-
-    def forward(self, input, mixup_lambda=None, device=None):
-        """
-        Input: (batch_size, data_length)"""
-
-        if self.enable_fusion and input["longer"].sum() == 0:
-            # if no audio is longer than 10s, then randomly select one audio to be longer
-            input["longer"][torch.randint(0, input["longer"].shape[0], (1,))] = True
-
-        if not self.enable_fusion:
-            x = self.spectrogram_extractor(
-                input["waveform"].to(device=device, non_blocking=True)
-            )  # (batch_size, 1, time_steps, freq_bins)
-            x = self.logmel_extractor(x)  # (batch_size, 1, time_steps, mel_bins)
-
-            x = x.transpose(1, 3)
-            x = self.bn0(x)
-            x = x.transpose(1, 3)
-        else:
-            longer_list = input["longer"].to(device=device, non_blocking=True)
-            x = input["mel_fusion"].to(device=device, non_blocking=True)
-            longer_list_idx = torch.where(longer_list)[0]
-            x = x.transpose(1, 3)
-            x = self.bn0(x)
-            x = x.transpose(1, 3)
-            if self.fusion_type in ["daf_1d", "aff_1d", "iaff_1d"]:
-                new_x = x[:, 0:1, :, :].clone().contiguous()
-                # local processing
-                if len(longer_list_idx) > 0:
-                    fusion_x_local = x[longer_list_idx, 1:, :, :].clone().contiguous()
-                    FB, FC, FT, FF = fusion_x_local.size()
-                    fusion_x_local = fusion_x_local.view(FB * FC, FT, FF)
-                    fusion_x_local = torch.permute(
-                        fusion_x_local, (0, 2, 1)
-                    ).contiguous()
-                    fusion_x_local = self.mel_conv1d(fusion_x_local)
-                    fusion_x_local = fusion_x_local.view(
-                        FB, FC, FF, fusion_x_local.size(-1)
-                    )
-                    fusion_x_local = (
-                        torch.permute(fusion_x_local, (0, 2, 1, 3))
-                        .contiguous()
-                        .flatten(2)
-                    )
-                    if fusion_x_local.size(-1) < FT:
-                        fusion_x_local = torch.cat(
-                            [
-                                fusion_x_local,
-                                torch.zeros(
-                                    (FB, FF, FT - fusion_x_local.size(-1)),
-                                    device=device,
-                                ),
-                            ],
-                            dim=-1,
-                        )
-                    else:
-                        fusion_x_local = fusion_x_local[:, :, :FT]
-                    # 1D fusion
-                    new_x = new_x.squeeze(1).permute((0, 2, 1)).contiguous()
-                    new_x[longer_list_idx] = self.fusion_model(
-                        new_x[longer_list_idx], fusion_x_local
-                    )
-                    x = new_x.permute((0, 2, 1)).contiguous()[:, None, :, :]
-                else:
-                    x = new_x
-            elif self.fusion_type in ["daf_2d", "aff_2d", "iaff_2d", "channel_map"]:
-                x = x  # no change
-
-        if self.training:
-            x = self.spec_augmenter(x)
-        # Mixup on spectrogram
-        if self.training and mixup_lambda is not None:
-            x = do_mixup(x, mixup_lambda)
-        if (self.enable_fusion) and (
-            self.fusion_type in ["daf_2d", "aff_2d", "iaff_2d"]
-        ):
-            global_x = x[:, 0:1, :, :]
-
-            # global processing
-            B, C, H, W = global_x.shape
-            global_x = self.conv_block1(global_x, pool_size=(2, 2), pool_type="avg")
-            if len(longer_list_idx) > 0:
-                local_x = x[longer_list_idx, 1:, :, :].contiguous()
-                TH = global_x.size(-2)
-                # local processing
-                B, C, H, W = local_x.shape
-                local_x = local_x.view(B * C, 1, H, W)
-                local_x = self.mel_conv2d(local_x)
-                local_x = local_x.view(
-                    B, C, local_x.size(1), local_x.size(2), local_x.size(3)
-                )
-                local_x = local_x.permute((0, 2, 1, 3, 4)).contiguous().flatten(2, 3)
-                TB, TC, _, TW = local_x.size()
-                if local_x.size(-2) < TH:
-                    local_x = torch.cat(
-                        [
-                            local_x,
-                            torch.zeros(
-                                (TB, TC, TH - local_x.size(-2), TW),
-                                device=global_x.device,
-                            ),
-                        ],
-                        dim=-2,
-                    )
-                else:
-                    local_x = local_x[:, :, :TH, :]
-
-                global_x[longer_list_idx] = self.fusion_model(
-                    global_x[longer_list_idx], local_x
-                )
-            x = global_x
-        else:
-            x = self.conv_block1(x, pool_size=(2, 2), pool_type="avg")
-
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block2(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block3(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block4(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block5(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block6(x, pool_size=(1, 1), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = torch.mean(x, dim=3)
-
-        latent_x1 = F.max_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x2 = F.avg_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x = latent_x1 + latent_x2
-        latent_x = latent_x.transpose(1, 2)
-        latent_x = F.relu_(self.fc1(latent_x))
-        latent_output = interpolate(latent_x, 32)
-
-        (x1, _) = torch.max(x, dim=2)
-        x2 = torch.mean(x, dim=2)
-        x = x1 + x2
-        x = F.dropout(x, p=0.5, training=self.training)
-        x = F.relu_(self.fc1(x))
-        embedding = F.dropout(x, p=0.5, training=self.training)
-        clipwise_output = torch.sigmoid(self.fc_audioset(x))
-
-        output_dict = {
-            "clipwise_output": clipwise_output,
-            "embedding": embedding,
-            "fine_grained_embedding": latent_output,
-        }
-        return output_dict
-
-
-class Cnn6(nn.Module):
-    def __init__(
-        self,
-        sample_rate,
-        window_size,
-        hop_size,
-        mel_bins,
-        fmin,
-        fmax,
-        classes_num,
-        enable_fusion=False,
-        fusion_type="None",
-    ):
-
-        super(Cnn6, self).__init__()
-
-        window = "hann"
-        center = True
-        pad_mode = "reflect"
-        ref = 1.0
-        amin = 1e-10
-        top_db = None
-
-        self.enable_fusion = enable_fusion
-        self.fusion_type = fusion_type
-
-        # Spectrogram extractor
-        self.spectrogram_extractor = Spectrogram(
-            n_fft=window_size,
-            hop_length=hop_size,
-            win_length=window_size,
-            window=window,
-            center=center,
-            pad_mode=pad_mode,
-            freeze_parameters=True,
-        )
-
-        # Logmel feature extractor
-        self.logmel_extractor = LogmelFilterBank(
-            sr=sample_rate,
-            n_fft=window_size,
-            n_mels=mel_bins,
-            fmin=fmin,
-            fmax=fmax,
-            ref=ref,
-            amin=amin,
-            top_db=top_db,
-            freeze_parameters=True,
-        )
-
-        # Spec augmenter
-        self.spec_augmenter = SpecAugmentation(
-            time_drop_width=64,
-            time_stripes_num=2,
-            freq_drop_width=8,
-            freq_stripes_num=2,
-        )
-
-        self.bn0 = nn.BatchNorm2d(64)
-
-        self.conv_block1 = ConvBlock5x5(in_channels=1, out_channels=64)
-        self.conv_block2 = ConvBlock5x5(in_channels=64, out_channels=128)
-        self.conv_block3 = ConvBlock5x5(in_channels=128, out_channels=256)
-        self.conv_block4 = ConvBlock5x5(in_channels=256, out_channels=512)
-
-        self.fc1 = nn.Linear(512, 512, bias=True)
-        self.fc_audioset = nn.Linear(512, classes_num, bias=True)
-
-        self.init_weight()
-
-    def init_weight(self):
-        init_bn(self.bn0)
-        init_layer(self.fc1)
-        init_layer(self.fc_audioset)
-
-    def forward(self, input, mixup_lambda=None, device=None):
-        """
-        Input: (batch_size, data_length)"""
-
-        x = self.spectrogram_extractor(input)  # (batch_size, 1, time_steps, freq_bins)
-        x = self.logmel_extractor(x)  # (batch_size, 1, time_steps, mel_bins)
-
-        x = x.transpose(1, 3)
-        x = self.bn0(x)
-        x = x.transpose(1, 3)
-
-        if self.training:
-            x = self.spec_augmenter(x)
-
-        # Mixup on spectrogram
-        if self.training and mixup_lambda is not None:
-            x = do_mixup(x, mixup_lambda)
-
-        x = self.conv_block1(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block2(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block3(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block4(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = torch.mean(x, dim=3)
-
-        latent_x1 = F.max_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x2 = F.avg_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x = latent_x1 + latent_x2
-        latent_x = latent_x.transpose(1, 2)
-        latent_x = F.relu_(self.fc1(latent_x))
-        latent_output = interpolate(latent_x, 16)
-
-        (x1, _) = torch.max(x, dim=2)
-        x2 = torch.mean(x, dim=2)
-        x = x1 + x2
-        x = F.dropout(x, p=0.5, training=self.training)
-        x = F.relu_(self.fc1(x))
-        embedding = F.dropout(x, p=0.5, training=self.training)
-        clipwise_output = torch.sigmoid(self.fc_audioset(x))
-
-        output_dict = {
-            "clipwise_output": clipwise_output,
-            "embedding": embedding,
-            "fine_grained_embedding": latent_output,
-        }
-
-        return output_dict
-
-
-class Cnn10(nn.Module):
-    def __init__(
-        self,
-        sample_rate,
-        window_size,
-        hop_size,
-        mel_bins,
-        fmin,
-        fmax,
-        classes_num,
-        enable_fusion=False,
-        fusion_type="None",
-    ):
-
-        super(Cnn10, self).__init__()
-
-        window = "hann"
-        center = True
-        pad_mode = "reflect"
-        ref = 1.0
-        amin = 1e-10
-        top_db = None
-
-        self.enable_fusion = enable_fusion
-        self.fusion_type = fusion_type
-
-        # Spectrogram extractor
-        self.spectrogram_extractor = Spectrogram(
-            n_fft=window_size,
-            hop_length=hop_size,
-            win_length=window_size,
-            window=window,
-            center=center,
-            pad_mode=pad_mode,
-            freeze_parameters=True,
-        )
-
-        # Logmel feature extractor
-        self.logmel_extractor = LogmelFilterBank(
-            sr=sample_rate,
-            n_fft=window_size,
-            n_mels=mel_bins,
-            fmin=fmin,
-            fmax=fmax,
-            ref=ref,
-            amin=amin,
-            top_db=top_db,
-            freeze_parameters=True,
-        )
-
-        # Spec augmenter
-        self.spec_augmenter = SpecAugmentation(
-            time_drop_width=64,
-            time_stripes_num=2,
-            freq_drop_width=8,
-            freq_stripes_num=2,
-        )
-
-        self.bn0 = nn.BatchNorm2d(64)
-
-        self.conv_block1 = ConvBlock(in_channels=1, out_channels=64)
-        self.conv_block2 = ConvBlock(in_channels=64, out_channels=128)
-        self.conv_block3 = ConvBlock(in_channels=128, out_channels=256)
-        self.conv_block4 = ConvBlock(in_channels=256, out_channels=512)
-        self.conv_block5 = ConvBlock(in_channels=512, out_channels=1024)
-
-        self.fc1 = nn.Linear(1024, 1024, bias=True)
-        self.fc_audioset = nn.Linear(1024, classes_num, bias=True)
-
-        self.init_weight()
-
-    def init_weight(self):
-        init_bn(self.bn0)
-        init_layer(self.fc1)
-        init_layer(self.fc_audioset)
-
-    def forward(self, input, mixup_lambda=None, device=None):
-        """
-        Input: (batch_size, data_length)"""
-
-        x = self.spectrogram_extractor(input)  # (batch_size, 1, time_steps, freq_bins)
-        x = self.logmel_extractor(x)  # (batch_size, 1, time_steps, mel_bins)
-
-        x = x.transpose(1, 3)
-        x = self.bn0(x)
-        x = x.transpose(1, 3)
-
-        if self.training:
-            x = self.spec_augmenter(x)
-
-        # Mixup on spectrogram
-        if self.training and mixup_lambda is not None:
-            x = do_mixup(x, mixup_lambda)
-
-        x = self.conv_block1(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block2(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block3(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block4(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = self.conv_block5(x, pool_size=(2, 2), pool_type="avg")
-        x = F.dropout(x, p=0.2, training=self.training)
-        x = torch.mean(x, dim=3)
-
-        latent_x1 = F.max_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x2 = F.avg_pool1d(x, kernel_size=3, stride=1, padding=1)
-        latent_x = latent_x1 + latent_x2
-        latent_x = latent_x.transpose(1, 2)
-        latent_x = F.relu_(self.fc1(latent_x))
-        latent_output = interpolate(latent_x, 32)
-
-        (x1, _) = torch.max(x, dim=2)
-        x2 = torch.mean(x, dim=2)
-        x = x1 + x2
-        x = F.dropout(x, p=0.5, training=self.training)
-        x = F.relu_(self.fc1(x))
-        embedding = F.dropout(x, p=0.5, training=self.training)
-        clipwise_output = torch.sigmoid(self.fc_audioset(x))
-
-        output_dict = {
-            "clipwise_output": clipwise_output,
-            "embedding": embedding,
-            "fine_grained_embedding": latent_output,
-        }
-
-        return output_dict
-
-
-def create_pann_model(audio_cfg, enable_fusion=False, fusion_type="None"):
-    try:
-        ModelProto = eval(audio_cfg.model_name)
-        model = ModelProto(
-            sample_rate=audio_cfg.sample_rate,
-            window_size=audio_cfg.window_size,
-            hop_size=audio_cfg.hop_size,
-            mel_bins=audio_cfg.mel_bins,
-            fmin=audio_cfg.fmin,
-            fmax=audio_cfg.fmax,
-            classes_num=audio_cfg.class_num,
-            enable_fusion=enable_fusion,
-            fusion_type=fusion_type,
-        )
-        return model
-    except:
-        raise RuntimeError(
-            f"Import Model for {audio_cfg.model_name} not found, or the audio cfg parameters are not enough."
-        )

audioldm/clap/open_clip/pretrained.py

@@ -1,167 +0,0 @@
-import hashlib
-import os
-import urllib
-import warnings
-
-from tqdm import tqdm
-
-_RN50 = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/afeb0e10f9e5a86da6080e35cf09123aca3b358a0c3e3b6c78a7b63bc04b6762/RN50.pt",
-    yfcc15m="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/rn50-quickgelu-yfcc15m-455df137.pt",
-    cc12m="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/rn50-quickgelu-cc12m-f000538c.pt",
-)
-
-_RN50_quickgelu = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/afeb0e10f9e5a86da6080e35cf09123aca3b358a0c3e3b6c78a7b63bc04b6762/RN50.pt",
-    yfcc15m="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/rn50-quickgelu-yfcc15m-455df137.pt",
-    cc12m="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/rn50-quickgelu-cc12m-f000538c.pt",
-)
-
-_RN101 = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/8fa8567bab74a42d41c5915025a8e4538c3bdbe8804a470a72f30b0d94fab599/RN101.pt",
-    yfcc15m="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/rn101-quickgelu-yfcc15m-3e04b30e.pt",
-)
-
-_RN101_quickgelu = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/8fa8567bab74a42d41c5915025a8e4538c3bdbe8804a470a72f30b0d94fab599/RN101.pt",
-    yfcc15m="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/rn101-quickgelu-yfcc15m-3e04b30e.pt",
-)
-
-_RN50x4 = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/7e526bd135e493cef0776de27d5f42653e6b4c8bf9e0f653bb11773263205fdd/RN50x4.pt",
-)
-
-_RN50x16 = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/52378b407f34354e150460fe41077663dd5b39c54cd0bfd2b27167a4a06ec9aa/RN50x16.pt",
-)
-
-_RN50x64 = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/be1cfb55d75a9666199fb2206c106743da0f6468c9d327f3e0d0a543a9919d9c/RN50x64.pt",
-)
-
-_VITB32 = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/40d365715913c9da98579312b702a82c18be219cc2a73407c4526f58eba950af/ViT-B-32.pt",
-    laion400m_e31="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/vit_b_32-quickgelu-laion400m_e31-d867053b.pt",
-    laion400m_e32="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/vit_b_32-quickgelu-laion400m_e32-46683a32.pt",
-    laion400m_avg="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/vit_b_32-quickgelu-laion400m_avg-8a00ab3c.pt",
-)
-
-_VITB32_quickgelu = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/40d365715913c9da98579312b702a82c18be219cc2a73407c4526f58eba950af/ViT-B-32.pt",
-    laion400m_e31="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/vit_b_32-quickgelu-laion400m_e31-d867053b.pt",
-    laion400m_e32="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/vit_b_32-quickgelu-laion400m_e32-46683a32.pt",
-    laion400m_avg="https://github.com/mlfoundations/open_clip/releases/download/v0.2-weights/vit_b_32-quickgelu-laion400m_avg-8a00ab3c.pt",
-)
-
-_VITB16 = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/5806e77cd80f8b59890b7e101eabd078d9fb84e6937f9e85e4ecb61988df416f/ViT-B-16.pt",
-)
-
-_VITL14 = dict(
-    openai="https://openaipublic.azureedge.net/clip/models/b8cca3fd41ae0c99ba7e8951adf17d267cdb84cd88be6f7c2e0eca1737a03836/ViT-L-14.pt",
-)
-
-_PRETRAINED = {
-    "RN50": _RN50,
-    "RN50-quickgelu": _RN50_quickgelu,
-    "RN101": _RN101,
-    "RN101-quickgelu": _RN101_quickgelu,
-    "RN50x4": _RN50x4,
-    "RN50x16": _RN50x16,
-    "ViT-B-32": _VITB32,
-    "ViT-B-32-quickgelu": _VITB32_quickgelu,
-    "ViT-B-16": _VITB16,
-    "ViT-L-14": _VITL14,
-}
-
-
-def list_pretrained(as_str: bool = False):
-    """returns list of pretrained models
-    Returns a tuple (model_name, pretrain_tag) by default or 'name:tag' if as_str == True
-    """
-    return [
-        ":".join([k, t]) if as_str else (k, t)
-        for k in _PRETRAINED.keys()
-        for t in _PRETRAINED[k].keys()
-    ]
-
-
-def list_pretrained_tag_models(tag: str):
-    """return all models having the specified pretrain tag"""
-    models = []
-    for k in _PRETRAINED.keys():
-        if tag in _PRETRAINED[k]:
-            models.append(k)
-    return models
-
-
-def list_pretrained_model_tags(model: str):
-    """return all pretrain tags for the specified model architecture"""
-    tags = []
-    if model in _PRETRAINED:
-        tags.extend(_PRETRAINED[model].keys())
-    return tags
-
-
-def get_pretrained_url(model: str, tag: str):
-    if model not in _PRETRAINED:
-        return ""
-    model_pretrained = _PRETRAINED[model]
-    if tag not in model_pretrained:
-        return ""
-    return model_pretrained[tag]
-
-
-def download_pretrained(url: str, root: str = os.path.expanduser("~/.cache/clip")):
-    os.makedirs(root, exist_ok=True)
-    filename = os.path.basename(url)
-
-    if "openaipublic" in url:
-        expected_sha256 = url.split("/")[-2]
-    else:
-        expected_sha256 = ""
-
-    download_target = os.path.join(root, filename)
-
-    if os.path.exists(download_target) and not os.path.isfile(download_target):
-        raise RuntimeError(f"{download_target} exists and is not a regular file")
-
-    if os.path.isfile(download_target):
-        if expected_sha256:
-            if (
-                hashlib.sha256(open(download_target, "rb").read()).hexdigest()
-                == expected_sha256
-            ):
-                return download_target
-            else:
-                warnings.warn(
-                    f"{download_target} exists, but the SHA256 checksum does not match; re-downloading the file"
-                )
-        else:
-            return download_target
-
-    with urllib.request.urlopen(url) as source, open(download_target, "wb") as output:
-        with tqdm(
-            total=int(source.info().get("Content-Length")),
-            ncols=80,
-            unit="iB",
-            unit_scale=True,
-        ) as loop:
-            while True:
-                buffer = source.read(8192)
-                if not buffer:
-                    break
-
-                output.write(buffer)
-                loop.update(len(buffer))
-
-    if (
-        expected_sha256
-        and hashlib.sha256(open(download_target, "rb").read()).hexdigest()
-        != expected_sha256
-    ):
-        raise RuntimeError(
-            f"Model has been downloaded but the SHA256 checksum does not not match"
-        )
-
-    return download_target

audioldm/clap/open_clip/timm_model.py

@@ -1,112 +0,0 @@
-""" timm model adapter
-
-Wraps timm (https://github.com/rwightman/pytorch-image-models) models for use as a vision tower in CLIP model.
-"""
-from collections import OrderedDict
-
-import torch.nn as nn
-
-try:
-    import timm
-    from timm.models.layers import Mlp, to_2tuple
-    from timm.models.layers.attention_pool2d import RotAttentionPool2d
-    from timm.models.layers.attention_pool2d import (
-        AttentionPool2d as AbsAttentionPool2d,
-    )
-except ImportError as e:
-    timm = None
-
-from .utils import freeze_batch_norm_2d
-
-
-class TimmModel(nn.Module):
-    """timm model adapter
-    # FIXME this adapter is a work in progress, may change in ways that break weight compat
-    """
-
-    def __init__(
-        self,
-        model_name,
-        embed_dim,
-        image_size=224,
-        pool="avg",
-        proj="linear",
-        drop=0.0,
-        pretrained=False,
-    ):
-        super().__init__()
-        if timm is None:
-            raise RuntimeError("Please `pip install timm` to use timm models.")
-
-        self.image_size = to_2tuple(image_size)
-        self.trunk = timm.create_model(model_name, pretrained=pretrained)
-        feat_size = self.trunk.default_cfg.get("pool_size", None)
-        feature_ndim = 1 if not feat_size else 2
-        if pool in ("abs_attn", "rot_attn"):
-            assert feature_ndim == 2
-            # if attn pooling used, remove both classifier and default pool
-            self.trunk.reset_classifier(0, global_pool="")
-        else:
-            # reset global pool if pool config set, otherwise leave as network default
-            reset_kwargs = dict(global_pool=pool) if pool else {}
-            self.trunk.reset_classifier(0, **reset_kwargs)
-        prev_chs = self.trunk.num_features
-
-        head_layers = OrderedDict()
-        if pool == "abs_attn":
-            head_layers["pool"] = AbsAttentionPool2d(
-                prev_chs, feat_size=feat_size, out_features=embed_dim
-            )
-            prev_chs = embed_dim
-        elif pool == "rot_attn":
-            head_layers["pool"] = RotAttentionPool2d(prev_chs, out_features=embed_dim)
-            prev_chs = embed_dim
-        else:
-            assert proj, "projection layer needed if non-attention pooling is used."
-
-        # NOTE attention pool ends with a projection layer, so proj should usually be set to '' if such pooling is used
-        if proj == "linear":
-            head_layers["drop"] = nn.Dropout(drop)
-            head_layers["proj"] = nn.Linear(prev_chs, embed_dim)
-        elif proj == "mlp":
-            head_layers["mlp"] = Mlp(prev_chs, 2 * embed_dim, embed_dim, drop=drop)
-
-        self.head = nn.Sequential(head_layers)
-
-    def lock(self, unlocked_groups=0, freeze_bn_stats=False):
-        """lock modules
-        Args:
-            unlocked_groups (int): leave last n layer groups unlocked (default: 0)
-        """
-        if not unlocked_groups:
-            # lock full model
-            for param in self.trunk.parameters():
-                param.requires_grad = False
-            if freeze_bn_stats:
-                freeze_batch_norm_2d(self.trunk)
-        else:
-            # NOTE: partial freeze requires latest timm (master) branch and is subject to change
-            try:
-                # FIXME import here until API stable and in an official release
-                from timm.models.helpers import group_parameters, group_modules
-            except ImportError:
-                raise RuntimeError(
-                    "Please install latest timm `pip install git+https://github.com/rwightman/pytorch-image-models`"
-                )
-            matcher = self.trunk.group_matcher()
-            gparams = group_parameters(self.trunk, matcher)
-            max_layer_id = max(gparams.keys())
-            max_layer_id = max_layer_id - unlocked_groups
-            for group_idx in range(max_layer_id + 1):
-                group = gparams[group_idx]
-                for param in group:
-                    self.trunk.get_parameter(param).requires_grad = False
-            if freeze_bn_stats:
-                gmodules = group_modules(self.trunk, matcher, reverse=True)
-                gmodules = {k for k, v in gmodules.items() if v <= max_layer_id}
-                freeze_batch_norm_2d(self.trunk, gmodules)
-
-    def forward(self, x):
-        x = self.trunk(x)
-        x = self.head(x)
-        return x

audioldm/clap/open_clip/tokenizer.py

@@ -1,197 +0,0 @@
-""" CLIP tokenizer
-
-Copied from https://github.com/openai/CLIP. Originally MIT License, Copyright (c) 2021 OpenAI.
-"""
-import gzip
-import html
-import os
-from functools import lru_cache
-from typing import Union, List
-
-import ftfy
-import regex as re
-import torch
-
-
-@lru_cache()
-def default_bpe():
-    return os.path.join(
-        os.path.dirname(os.path.abspath(__file__)), "bpe_simple_vocab_16e6.txt.gz"
-    )
-
-
-@lru_cache()
-def bytes_to_unicode():
-    """
-    Returns list of utf-8 byte and a corresponding list of unicode strings.
-    The reversible bpe codes work on unicode strings.
-    This means you need a large # of unicode characters in your vocab if you want to avoid UNKs.
-    When you're at something like a 10B token dataset you end up needing around 5K for decent coverage.
-    This is a signficant percentage of your normal, say, 32K bpe vocab.
-    To avoid that, we want lookup tables between utf-8 bytes and unicode strings.
-    And avoids mapping to whitespace/control characters the bpe code barfs on.
-    """
-    bs = (
-        list(range(ord("!"), ord("~") + 1))
-        + list(range(ord("¡"), ord("¬") + 1))
-        + list(range(ord("®"), ord("ÿ") + 1))
-    )
-    cs = bs[:]
-    n = 0
-    for b in range(2**8):
-        if b not in bs:
-            bs.append(b)
-            cs.append(2**8 + n)
-            n += 1
-    cs = [chr(n) for n in cs]
-    return dict(zip(bs, cs))
-
-
-def get_pairs(word):
-    """Return set of symbol pairs in a word.
-    Word is represented as tuple of symbols (symbols being variable-length strings).
-    """
-    pairs = set()
-    prev_char = word[0]
-    for char in word[1:]:
-        pairs.add((prev_char, char))
-        prev_char = char
-    return pairs
-
-
-def basic_clean(text):
-    text = ftfy.fix_text(text)
-    text = html.unescape(html.unescape(text))
-    return text.strip()
-
-
-def whitespace_clean(text):
-    text = re.sub(r"\s+", " ", text)
-    text = text.strip()
-    return text
-
-
-class SimpleTokenizer(object):
-    def __init__(self, bpe_path: str = default_bpe(), special_tokens=None):
-        self.byte_encoder = bytes_to_unicode()
-        self.byte_decoder = {v: k for k, v in self.byte_encoder.items()}
-        merges = gzip.open(bpe_path).read().decode("utf-8").split("\n")
-        merges = merges[1 : 49152 - 256 - 2 + 1]
-        merges = [tuple(merge.split()) for merge in merges]
-        vocab = list(bytes_to_unicode().values())
-        vocab = vocab + [v + "</w>" for v in vocab]
-        for merge in merges:
-            vocab.append("".join(merge))
-        if not special_tokens:
-            special_tokens = ["<start_of_text>", "<end_of_text>"]
-        else:
-            special_tokens = ["<start_of_text>", "<end_of_text>"] + special_tokens
-        vocab.extend(special_tokens)
-        self.encoder = dict(zip(vocab, range(len(vocab))))
-        self.decoder = {v: k for k, v in self.encoder.items()}
-        self.bpe_ranks = dict(zip(merges, range(len(merges))))
-        self.cache = {t: t for t in special_tokens}
-        special = "|".join(special_tokens)
-        self.pat = re.compile(
-            special + r"""|'s|'t|'re|'ve|'m|'ll|'d|[\p{L}]+|[\p{N}]|[^\s\p{L}\p{N}]+""",
-            re.IGNORECASE,
-        )
-
-        self.vocab_size = len(self.encoder)
-        self.all_special_ids = [self.encoder[t] for t in special_tokens]
-
-    def bpe(self, token):
-        if token in self.cache:
-            return self.cache[token]
-        word = tuple(token[:-1]) + (token[-1] + "</w>",)
-        pairs = get_pairs(word)
-
-        if not pairs:
-            return token + "</w>"
-
-        while True:
-            bigram = min(pairs, key=lambda pair: self.bpe_ranks.get(pair, float("inf")))
-            if bigram not in self.bpe_ranks:
-                break
-            first, second = bigram
-            new_word = []
-            i = 0
-            while i < len(word):
-                try:
-                    j = word.index(first, i)
-                    new_word.extend(word[i:j])
-                    i = j
-                except:
-                    new_word.extend(word[i:])
-                    break
-
-                if word[i] == first and i < len(word) - 1 and word[i + 1] == second:
-                    new_word.append(first + second)
-                    i += 2
-                else:
-                    new_word.append(word[i])
-                    i += 1
-            new_word = tuple(new_word)
-            word = new_word
-            if len(word) == 1:
-                break
-            else:
-                pairs = get_pairs(word)
-        word = " ".join(word)
-        self.cache[token] = word
-        return word
-
-    def encode(self, text):
-        bpe_tokens = []
-        text = whitespace_clean(basic_clean(text)).lower()
-        for token in re.findall(self.pat, text):
-            token = "".join(self.byte_encoder[b] for b in token.encode("utf-8"))
-            bpe_tokens.extend(
-                self.encoder[bpe_token] for bpe_token in self.bpe(token).split(" ")
-            )
-        return bpe_tokens
-
-    def decode(self, tokens):
-        text = "".join([self.decoder[token] for token in tokens])
-        text = (
-            bytearray([self.byte_decoder[c] for c in text])
-            .decode("utf-8", errors="replace")
-            .replace("</w>", " ")
-        )
-        return text
-
-
-_tokenizer = SimpleTokenizer()
-
-
-def tokenize(
-    texts: Union[str, List[str]], context_length: int = 77
-) -> torch.LongTensor:
-    """
-    Returns the tokenized representation of given input string(s)
-
-    Parameters
-    ----------
-    texts : Union[str, List[str]]
-        An input string or a list of input strings to tokenize
-    context_length : int
-        The context length to use; all CLIP models use 77 as the context length
-
-    Returns
-    -------
-    A two-dimensional tensor containing the resulting tokens, shape = [number of input strings, context_length]
-    """
-    if isinstance(texts, str):
-        texts = [texts]
-
-    sot_token = _tokenizer.encoder["<start_of_text>"]
-    eot_token = _tokenizer.encoder["<end_of_text>"]
-    all_tokens = [[sot_token] + _tokenizer.encode(text) + [eot_token] for text in texts]
-    result = torch.zeros(len(all_tokens), context_length, dtype=torch.long)
-
-    for i, tokens in enumerate(all_tokens):
-        if len(tokens) > context_length:
-            tokens = tokens[:context_length]  # Truncate
-        result[i, : len(tokens)] = torch.tensor(tokens)
-
-    return result

audioldm/clap/open_clip/transform.py

@@ -1,45 +0,0 @@
-from torchvision.transforms import (
-    Normalize,
-    Compose,
-    RandomResizedCrop,
-    InterpolationMode,
-    ToTensor,
-    Resize,
-    CenterCrop,
-)
-
-
-def _convert_to_rgb(image):
-    return image.convert("RGB")
-
-
-def image_transform(
-    image_size: int,
-    is_train: bool,
-    mean=(0.48145466, 0.4578275, 0.40821073),
-    std=(0.26862954, 0.26130258, 0.27577711),
-):
-    normalize = Normalize(mean=mean, std=std)
-    if is_train:
-        return Compose(
-            [
-                RandomResizedCrop(
-                    image_size,
-                    scale=(0.9, 1.0),
-                    interpolation=InterpolationMode.BICUBIC,
-                ),
-                _convert_to_rgb,
-                ToTensor(),
-                normalize,
-            ]
-        )
-    else:
-        return Compose(
-            [
-                Resize(image_size, interpolation=InterpolationMode.BICUBIC),
-                CenterCrop(image_size),
-                _convert_to_rgb,
-                ToTensor(),
-                normalize,
-            ]
-        )

audioldm/clap/open_clip/utils.py

@@ -1,361 +0,0 @@
-import numpy as np
-import torch
-from torch import nn as nn
-from torchvision.ops.misc import FrozenBatchNorm2d
-import logging
-# import h5py
-from tqdm import tqdm
-import random
-import json
-import os
-import pathlib
-
-# TODO: (yusong) this not a good place to store those information and does not scale. Need to be fixed later.
-dataset_split = {
-    "audiocaps": ["train", "valid", "test"],
-    "audioset": ["balanced_train", "unbalanced_train", "eval"],
-    "BBCSoundEffects": ["train", "test"],
-    "Clotho": ["train", "test", "valid"],
-    "free_to_use_sounds": ["train", "test"],
-    "paramount_motion": ["train", "test"],
-    "sonniss_game_effects": ["train", "test"],
-    "wesoundeffects": ["train", "test"],
-    "MACS": ["train", "test"],
-    "freesound": ["train", "test"],
-    "FSD50K": ["train", "test", "valid"],
-    "fsd50k_class_label": ["train", "test", "valid"],
-    "esc50": ["train", "test"],
-    "audiostock": ["train", "test"],
-    "freesound_no_overlap_noesc50": ["train", "test"],
-    "epidemic_sound_effects": ["train", "test"],
-    "VGGSound": ["train", "test"],
-    "urbansound8k_class_label": ["train", "test"],
-    "audioset_t5": ["balanced_train", "unbalanced_train", "eval"],
-    "epidemic_sound_effects_t5": ["train", "test"],
-    "WavText5K": ["train", "test"],
-    "esc50_no_overlap": ["train", "test"],
-    "usd8k_no_overlap": ["train", "test"],
-    "fsd50k_200_class_label": ["train", "test", "valid"],
-}
-
-
-def freeze_batch_norm_2d(module, module_match={}, name=""):
-    """
-    Converts all `BatchNorm2d` and `SyncBatchNorm` layers of provided module into `FrozenBatchNorm2d`. If `module` is
-    itself an instance of either `BatchNorm2d` or `SyncBatchNorm`, it is converted into `FrozenBatchNorm2d` and
-    returned. Otherwise, the module is walked recursively and submodules are converted in place.
-
-    Args:
-        module (torch.nn.Module): Any PyTorch module.
-        module_match (dict): Dictionary of full module names to freeze (all if empty)
-        name (str): Full module name (prefix)
-
-    Returns:
-        torch.nn.Module: Resulting module
-
-    Inspired by https://github.com/pytorch/pytorch/blob/a5895f85be0f10212791145bfedc0261d364f103/torch/nn/modules/batchnorm.py#L762
-    """
-    res = module
-    is_match = True
-    if module_match:
-        is_match = name in module_match
-    if is_match and isinstance(
-        module, (nn.modules.batchnorm.BatchNorm2d, nn.modules.batchnorm.SyncBatchNorm)
-    ):
-        res = FrozenBatchNorm2d(module.num_features)
-        res.num_features = module.num_features
-        res.affine = module.affine
-        if module.affine:
-            res.weight.data = module.weight.data.clone().detach()
-            res.bias.data = module.bias.data.clone().detach()
-        res.running_mean.data = module.running_mean.data
-        res.running_var.data = module.running_var.data
-        res.eps = module.eps
-    else:
-        for child_name, child in module.named_children():
-            full_child_name = ".".join([name, child_name]) if name else child_name
-            new_child = freeze_batch_norm_2d(child, module_match, full_child_name)
-            if new_child is not child:
-                res.add_module(child_name, new_child)
-    return res
-
-
-def exist(dataset_name, dataset_type):
-    """
-    Check if dataset exists
-    """
-    if dataset_type in dataset_split[dataset_name]:
-        return True
-    else:
-        return False
-
-
-def get_tar_path_from_dataset_name(
-    dataset_names, dataset_types, islocal, dataset_path, proportion=1, full_dataset=None
-):
-    """
-    Get tar path from dataset name and type
-    """
-    output = []
-    for n in dataset_names:
-        if full_dataset is not None and n in full_dataset:
-            current_dataset_types = dataset_split[n]
-        else:
-            current_dataset_types = dataset_types
-        for s in current_dataset_types:
-            tmp = []
-            if islocal:
-                sizefilepath_ = f"{dataset_path}/{n}/{s}/sizes.json"
-                if not os.path.exists(sizefilepath_):
-                    sizefilepath_ = f"./json_files/{n}/{s}/sizes.json"
-            else:
-                sizefilepath_ = f"./json_files/{n}/{s}/sizes.json"
-            if not os.path.exists(sizefilepath_):
-                continue
-            sizes = json.load(open(sizefilepath_, "r"))
-            for k in sizes.keys():
-                if islocal:
-                    tmp.append(f"{dataset_path}/{n}/{s}/{k}")
-                else:
-                    tmp.append(
-                        f"pipe:aws s3 --cli-connect-timeout 0 cp s3://s-laion-audio/webdataset_tar/{n}/{s}/{k} -"
-                    )
-            if proportion != 1:
-                tmp = random.sample(tmp, int(proportion * len(tmp)))
-            output.append(tmp)
-    return sum(output, [])
-
-
-def get_tar_path_from_txts(txt_path, islocal, proportion=1):
-    """
-    Get tar path from txt path
-    """
-    if isinstance(txt_path, (list, tuple)):
-        return sum(
-            [
-                get_tar_path_from_txts(
-                    txt_path[i], islocal=islocal, proportion=proportion
-                )
-                for i in range(len(txt_path))
-            ],
-            [],
-        )
-    if isinstance(txt_path, str):
-        with open(txt_path) as f:
-            lines = f.readlines()
-        if islocal:
-            lines = [
-                lines[i]
-                .split("\n")[0]
-                .replace("pipe:aws s3 cp s3://s-laion-audio/", "/mnt/audio_clip/")
-                for i in range(len(lines))
-            ]
-        else:
-            lines = [
-                lines[i].split("\n")[0].replace(".tar", ".tar -")
-                for i in range(len(lines))
-            ]
-        if proportion != 1:
-            print("Sampling tars with proportion of {}".format(proportion))
-            lines = random.sample(lines, int(proportion * len(lines)))
-        return lines
-
-
-def get_mix_lambda(mixup_alpha, batch_size):
-    mixup_lambdas = [
-        np.random.beta(mixup_alpha, mixup_alpha, 1)[0] for _ in range(batch_size)
-    ]
-    return np.array(mixup_lambdas).astype(np.float32)
-
-
-def do_mixup(x, mixup_lambda):
-    """
-    Args:
-      x: (batch_size , ...)
-      mixup_lambda: (batch_size,)
-    Returns:
-      out: (batch_size, ...)
-    """
-    out = (
-        x.transpose(0, -1) * mixup_lambda
-        + torch.flip(x, dims=[0]).transpose(0, -1) * (1 - mixup_lambda)
-    ).transpose(0, -1)
-    return out
-
-
-def interpolate(x, ratio):
-    """Interpolate data in time domain. This is used to compensate the
-    resolution reduction in downsampling of a CNN.
-
-    Args:
-      x: (batch_size, time_steps, classes_num)
-      ratio: int, ratio to interpolate
-    Returns:
-      upsampled: (batch_size, time_steps * ratio, classes_num)
-    """
-    (batch_size, time_steps, classes_num) = x.shape
-    upsampled = x[:, :, None, :].repeat(1, 1, ratio, 1)
-    upsampled = upsampled.reshape(batch_size, time_steps * ratio, classes_num)
-    return upsampled
-
-
-def pad_framewise_output(framewise_output, frames_num):
-    """Pad framewise_output to the same length as input frames. The pad value
-    is the same as the value of the last frame.
-    Args:
-      framewise_output: (batch_size, frames_num, classes_num)
-      frames_num: int, number of frames to pad
-    Outputs:
-      output: (batch_size, frames_num, classes_num)
-    """
-    pad = framewise_output[:, -1:, :].repeat(
-        1, frames_num - framewise_output.shape[1], 1
-    )
-    """tensor for padding"""
-
-    output = torch.cat((framewise_output, pad), dim=1)
-    """(batch_size, frames_num, classes_num)"""
-
-
-# def process_ipc(index_path, classes_num, filename):
-#     # load data
-#     logging.info("Load Data...............")
-#     ipc = [[] for _ in range(classes_num)]
-#     with h5py.File(index_path, "r") as f:
-#         for i in tqdm(range(len(f["target"]))):
-#             t_class = np.where(f["target"][i])[0]
-#             for t in t_class:
-#                 ipc[t].append(i)
-#     print(ipc)
-#     np.save(filename, ipc)
-#     logging.info("Load Data Succeed...............")
-
-
-def save_to_dict(s, o_={}):
-    sp = s.split(": ")
-    o_.update({sp[0]: float(sp[1])})
-    return o_
-
-
-def get_data_from_log(txt_path):
-    """
-    Output dictionary from out.txt log file
-    """
-    with open(txt_path) as f:
-        lines = f.readlines()
-    val_data = {}
-    train_data = {}
-    train_losses = []
-    train_losses_epoch = []
-    for i in range(len(lines)):
-        if "| INFO |" in lines[i]:
-            if "Eval Epoch" in lines[i]:
-                if "val_loss" in lines[i]:
-                    # float(regex.sub("", lines[310].split("	")[-1]).replace(" ", ""))
-                    line = lines[i].split("Eval Epoch: ")[-1]
-                    num_epoch = int(line.split("	")[0].split(" ")[0])
-                    d = {
-                        line.split("	")[0]
-                        .split(" ")[1]
-                        .replace(":", ""): float(line.split("	")[0].split(" ")[-1])
-                    }
-                    for i in range(1, len(line.split("	"))):
-                        d = save_to_dict(line.split("	")[i], d)
-                    val_data[num_epoch] = d
-            elif "Train Epoch" in lines[i]:
-                num_epoch = int(lines[i].split("Train Epoch: ")[1][0])
-                loss = float(lines[i].split("Loss: ")[-1].split(" (")[0])
-                train_losses.append(loss)
-                train_losses_epoch.append(num_epoch)
-    for i in range(len(train_losses)):
-        train_data[i] = {
-            "num_epoch": train_losses_epoch[i],
-            "train_loss": train_losses[i],
-        }
-    return train_data, val_data
-
-
-def save_p(obj, filename):
-    import pickle
-
-    try:
-        from deepdiff import DeepDiff
-    except:
-        os.system("pip install deepdiff")
-        from deepdiff import DeepDiff
-    with open(filename, "wb") as file:
-        pickle.dump(obj, file, protocol=pickle.HIGHEST_PROTOCOL)  # highest protocol
-    with open(filename, "rb") as file:
-        z = pickle.load(file)
-    assert (
-        DeepDiff(obj, z, ignore_string_case=True) == {}
-    ), "there is something wrong with the saving process"
-    return
-
-
-def load_p(filename):
-    import pickle
-
-    with open(filename, "rb") as file:
-        z = pickle.load(file)
-    return z
-
-
-def save_json(data, name="data.json"):
-    import json
-
-    with open(name, "w") as fp:
-        json.dump(data, fp)
-    return
-
-
-def load_json(name):
-    import json
-
-    with open(name, "r") as fp:
-        data = json.load(fp)
-    return data
-
-
-from multiprocessing import Process, Manager
-from multiprocessing import Process, Value, Array
-from ctypes import c_wchar
-
-
-def load_class_label(path):
-    # https://stackoverflow.com/questions/48004243/how-to-share-large-read-only-dictionary-list-across-processes-in-multiprocessing
-    # https://stackoverflow.com/questions/45693949/storing-strings-in-a-multiprocessing-sharedctypes-array
-    out = None
-    if path is not None:
-        if pathlib.Path(path).suffix in [".pkl", ".pickle"]:
-            out = load_p(path)
-        elif pathlib.Path(path).suffix in [".json", ".txt"]:
-            out = load_json(path)
-        elif pathlib.Path(path).suffix in [".npy", ".npz"]:
-            out = np.load(path)
-        elif pathlib.Path(path).suffix in [".csv"]:
-            import pandas as pd
-
-            out = pd.read_csv(path)
-    return out
-    # if out is None:
-    #     return None
-    # else:
-    #     key = Array(c_wchar, '\n'.join(list(out.keys())), lock=False)
-    #     val = Array('i', out.values(), lock=False)
-    #     return (key, val)
-
-
-from torch import optim
-
-
-def get_optimizer(params, lr, betas, eps, momentum, optimizer_name):
-    if optimizer_name.lower() == "adamw":
-        optimizer = optim.AdamW(params, lr=lr, betas=betas, eps=eps)
-    elif optimizer_name.lower() == "sgd":
-        optimizer = optim.SGD(params, lr=lr, momentum=momentum)
-    elif optimizer_name.lower() == "adam":
-        optimizer = optim.Adam(params, lr=lr, betas=betas, eps=eps)
-    else:
-        raise ValueError("optimizer name is not correct")
-    return optimizer

audioldm/clap/open_clip/version.py

@@ -1 +0,0 @@
-__version__ = "0.2.1"

audioldm/clap/training/audioset_textmap.npy

@@ -1,3 +0,0 @@
-version https://git-lfs.github.com/spec/v1
-oid sha256:bada103070d92f9eadd33e1b4f45ec8583f59080ef218c966b43294bd4c86d5b
-size 84448

audioldm/clap/training/data.py

@@ -1,977 +0,0 @@
-import ast
-import json
-import logging
-import math
-import os
-import random
-# import h5py
-from dataclasses import dataclass
-from audioldm.clap.training.params import parse_args
-# import braceexpand
-import numpy as np
-import pandas as pd
-import torch
-import torch.nn as nn
-import torch.nn.functional as F
-import torchvision.datasets as datasets
-import torchvision.transforms
-# import webdataset as wds
-from PIL import Image
-from torch.utils.data import Dataset, DataLoader, SubsetRandomSampler
-from torch.utils.data.distributed import DistributedSampler
-from functools import partial
-import soundfile as sf
-import io
-from pathlib import Path
-# import wget
-
-from audioldm.clap.open_clip.utils import (
-    get_tar_path_from_dataset_name,
-    dataset_split,
-)
-from audioldm.clap.open_clip.utils import load_p, load_class_label
-import copy
-
-try:
-    import horovod.torch as hvd
-except ImportError:
-    hvd = None
-
-try:
-    import torchaudio
-except ImportError:
-    torchaudio = None
-
-from audioldm.clap.open_clip import tokenize
-
-
-def tokenizer(text):
-    return tokenize(text).squeeze(0)
-
-
-from transformers import RobertaTokenizer
-
-tokenize = RobertaTokenizer.from_pretrained("roberta-base")
-
-
-def tokenizer(text):
-    result = tokenize(
-        text,
-        padding="max_length",
-        truncation=True,
-        max_length=77,
-        return_tensors="pt",
-    )
-    return {k: v.squeeze(0) for k, v in result.items()}
-
-
-# initizlied the audioset map
-_AUDIOSET_MAP_PATH = os.path.join(Path(__file__).parent, "audioset_textmap.npy")
-_AUDIOSET_MAP = np.load(_AUDIOSET_MAP_PATH, allow_pickle=True)
-
-
-def int16_to_float32(x):
-    return (x / 32767.0).astype(np.float32)
-
-
-def float32_to_int16(x):
-    x = np.clip(x, a_min=-1.0, a_max=1.0)
-    return (x * 32767.0).astype(np.int16)
-
-
-# For Toy Dataset
-# class ToyDataset(Dataset):
-#     def __init__(self, index_path, ipc, config, eval_mode=False):
-#         """Toy Dataset for testing the audioset input with text labels
-#         Parameters
-#         ----------
-#             index_path: str
-#                 the link to the h5 file of each audio
-#             idc: str
-#                 the link to the npy file, the number of samples in each class
-#             config: dict
-#                 the audio cfg file
-#            eval_model (bool): to indicate if the dataset is a testing dataset
-#         """
-#         self.audio_cfg = config["audio_cfg"]
-#         self.text_cfg = config["text_cfg"]
-#         self.fp = h5py.File(index_path, "r")
-#         self.ipc = np.load(ipc, allow_pickle=True)
-#         self.total_size = len(self.fp["audio_name"])
-#         self.classes_num = self.audio_cfg["class_num"]
-#         self.eval_mode = eval_mode
-
-#         if not eval_mode:
-#             self.generate_queue()
-#         else:
-#             self.queue = []
-#             for i in range(self.total_size):
-#                 target = self.fp["target"][i]
-#                 if np.sum(target) > 0:
-#                     self.queue.append(i)
-#             self.total_size = len(self.queue)
-#         logging.info("total dataset size: %d" % (self.total_size))
-#         logging.info("class num: %d" % (self.classes_num))
-
-#     def time_shifting(self, x):
-#         frame_num = len(x)
-#         shift_len = random.randint(0, frame_num - 1)
-#         new_sample = np.concatenate([x[shift_len:], x[:shift_len]], axis=0)
-#         return new_sample
-
-#     def generate_queue(self):
-#         self.queue = []
-#         while len(self.queue) < self.total_size:
-#             class_set = [*range(self.classes_num)]
-#             random.shuffle(class_set)
-#             self.queue += [
-#                 self.ipc[d][random.randint(0, len(self.ipc[d]) - 1)] for d in class_set
-#             ]
-#         self.queue = self.queue[: self.total_size]
-
-#         logging.info("queue regenerated:%s" % (self.queue[-5:]))
-
-#     def crop_wav(self, x):
-#         crop_size = self.audio_cfg["crop_size"]
-#         crop_pos = random.randint(0, len(x) - crop_size - 1)
-#         return x[crop_pos : crop_pos + crop_size]
-
-#     def prompt_text(self, target):
-#         events = _AUDIOSET_MAP[np.where(target > 0)]
-#         event_text = "The sounds of " + ", ".join(events[:-1]) + " and " + events[-1]
-#         text = tokenize(event_text)[0]
-#         return text
-
-#     def __getitem__(self, index):
-#         """Load waveform, text, and target of an audio clip
-
-#         Parameters
-#         ----------
-#             index: int
-#                 the index number
-#         Return
-#         ------
-#             output: dict {
-#                 "hdf5_path": str,
-#                 "index_in_hdf5": int,
-#                 "audio_name": str,
-#                 "waveform": list (audio_length,),
-#                 "target": list (class_num, ),
-#                 "text": torch.tensor (context_length,)
-#             }
-#                 the output dictionary
-#         """
-#         s_index = self.queue[index]
-
-#         audio_name = self.fp["audio_name"][s_index].decode()
-#         # Hardcode here CHANGE
-#         hdf5_path = (
-#             self.fp["hdf5_path"][s_index]
-#             .decode()
-#             .replace(
-#                 "../workspace",
-#                 "/home/la/kechen/Research/ke_zsasp/workspace",
-#             )
-#         )
-#         r_idx = self.fp["index_in_hdf5"][s_index]
-#         target = self.fp["target"][s_index].astype(np.float32)
-#         text = self.prompt_text(target)
-#         with h5py.File(hdf5_path, "r") as f:
-#             waveform = int16_to_float32(f["waveform"][r_idx])[
-#                 : self.audio_cfg["clip_samples"]
-#             ]
-#         assert (
-#             len(waveform) == self.audio_cfg["clip_samples"]
-#         ), "The sample length is not match"
-#         # Time shift
-#         # if (self.config.enable_time_shift) and (not self.eval_mode):
-#         #     waveform = self.time_shifting(waveform)
-#         # # Label Enhance
-#         # if (self.config.crop_size is not None) and (not self.eval_mode):
-#         #     waveform = self.crop_wav(waveform)
-#         # # the label enhance rate is fixed 0.5
-#         # if (self.config.enable_label_enhance) and (not self.eval_mode) and random.random() < 0.5:
-#         #     kidx = np.where(target)[0]
-#         #     for k in kidx:
-#         #         for add_key in self.class_map[k][1]:
-#         #             target[add_key] = 1.0
-#         #         if len(self.class_map[k][2]) > 0:
-#         #             add_key = random.choice(self.class_map[k][2])
-#         #             target[add_key] = 1.0
-
-#         # missing the text input
-#         mel_spec = get_mel(torch.from_numpy(waveform), self.audio_cfg)[None, :, :]
-#         mel_spec = (
-#             torch.cat(
-#                 [mel_spec, mel_spec.clone(), mel_spec.clone(), mel_spec.clone()], dim=0
-#             )
-#             .cpu()
-#             .numpy()
-#         )
-#         longer = random.choice([True, False])
-#         if longer == False:
-#             mel_spec[1:, :, :] = 0.0
-#         data_dict = {
-#             "hdf5_path": hdf5_path,
-#             "index_in_hdf5": r_idx,
-#             "audio_name": audio_name,
-#             "waveform": waveform,
-#             "class_label": target,
-#             "text": text,
-#             "longer": longer,
-#             "mel_fusion": mel_spec,
-#         }
-#         return data_dict
-
-#     def __len__(self):
-#         return self.total_size
-
-
-class CsvDataset(Dataset):
-    def __init__(self, input_filename, transforms, img_key, caption_key, sep="\t"):
-        logging.debug(f"Loading csv data from {input_filename}.")
-        df = pd.read_csv(input_filename, sep=sep)
-
-        self.images = df[img_key].tolist()
-        self.captions = df[caption_key].tolist()
-        self.transforms = transforms
-        logging.debug("Done loading data.")
-
-    def __len__(self):
-        return len(self.captions)
-
-    def __getitem__(self, idx):
-        images = self.transforms(Image.open(str(self.images[idx])))
-        texts = tokenize([str(self.captions[idx])])[0]
-        return images, texts
-
-
-@dataclass
-class DataInfo:
-    dataloader: DataLoader
-    sampler: DistributedSampler
-
-
-def preprocess_txt(text):
-    return tokenize([str(text)])[0]
-
-
-def get_dataset_size(shards, sizefilepath_=None, is_local=True):
-    if isinstance(shards, list):
-        size_list = []
-        for s in shards:
-            size_list.append(
-                get_dataset_size(s, sizefilepath_=sizefilepath_, is_local=is_local)[0]
-            )
-    else:
-        if not is_local:
-            for n in dataset_split.keys():
-                if n in shards.split("/"):
-                    break
-            for s in dataset_split[n]:
-                if s in shards.split("/"):
-                    break
-            sizefilepath_ = f"./json_files/{n}/{s}/sizes.json"
-        shards_list = list(braceexpand.braceexpand(shards))
-        dir_path = os.path.dirname(shards)
-        if sizefilepath_ is not None:
-            sizes = json.load(open(sizefilepath_, "r"))
-            total_size = sum(
-                [
-                    int(sizes[os.path.basename(shard.replace(".tar -", ".tar"))])
-                    for shard in shards_list
-                ]
-            )
-        else:
-            sizes_filename = os.path.join(dir_path, "sizes.json")
-            len_filename = os.path.join(dir_path, "__len__")
-            if os.path.exists(sizes_filename):
-                sizes = json.load(open(sizes_filename, "r"))
-                total_size = sum(
-                    [int(sizes[os.path.basename(shard)]) for shard in shards_list]
-                )
-            elif os.path.exists(len_filename):
-                # FIXME this used to be eval(open(...)) but that seemed rather unsafe
-                total_size = ast.literal_eval(open(len_filename, "r").read())
-            else:
-                raise Exception(
-                    "Cannot find sizes file for dataset. Please specify the path to the file."
-                )
-                # total_size = None  # num samples undefined
-                # some common dataset sizes (at time of authors last download)
-                # cc3m-train: 2905954
-                # cc12m: 10968539
-                # LAION-400m: 407332084
-        num_shards = len(shards_list)
-    if isinstance(shards, list):
-        return sum(size_list), len(shards)
-    else:
-        return total_size, num_shards
-
-
-def get_imagenet(args, preprocess_fns, split):
-    assert split in ["train", "val", "v2"]
-    is_train = split == "train"
-    preprocess_train, preprocess_val = preprocess_fns
-
-    if split == "v2":
-        from imagenetv2_pytorch import ImageNetV2Dataset
-
-        dataset = ImageNetV2Dataset(location=args.imagenet_v2, transform=preprocess_val)
-    else:
-        if is_train:
-            data_path = args.imagenet_train
-            preprocess_fn = preprocess_train
-        else:
-            data_path = args.imagenet_val
-            preprocess_fn = preprocess_val
-        assert data_path
-
-        dataset = datasets.ImageFolder(data_path, transform=preprocess_fn)
-
-    if is_train:
-        idxs = np.zeros(len(dataset.targets))
-        target_array = np.array(dataset.targets)
-        k = 50
-        for c in range(1000):
-            m = target_array == c
-            n = len(idxs[m])
-            arr = np.zeros(n)
-            arr[:k] = 1
-            np.random.shuffle(arr)
-            idxs[m] = arr
-
-        idxs = idxs.astype("int")
-        sampler = SubsetRandomSampler(np.where(idxs)[0])
-    else:
-        sampler = None
-
-    dataloader = torch.utils.data.DataLoader(
-        dataset,
-        batch_size=args.batch_size,
-        num_workers=args.workers,
-        sampler=sampler,
-    )
-
-    return DataInfo(dataloader, sampler)
-
-
-def count_samples(dataloader):
-    os.environ["WDS_EPOCH"] = "0"
-    n_elements, n_batches = 0, 0
-    for images, texts in dataloader:
-        n_batches += 1
-        n_elements += len(images)
-        assert len(images) == len(texts)
-    return n_elements, n_batches
-
-
-def filter_no_caption(sample):
-    return "txt" in sample
-
-
-def log_and_continue(exn):
-    """Call in an exception handler to ignore any exception, isssue a warning, and continue."""
-    logging.warning(f"Handling webdataset error ({repr(exn)}). Ignoring.")
-    return True
-
-
-_SHARD_SHUFFLE_SIZE = 2000
-_SHARD_SHUFFLE_INITIAL = 500
-_SAMPLE_SHUFFLE_SIZE = 5000
-_SAMPLE_SHUFFLE_INITIAL = 1000
-
-
-def sample_prop(sizefile, inputs, proportion, is_local=True):
-    """
-    Sample a proportion of the data.
-    """
-    file_path_dict = {
-        os.path.split(inputs[i])[1]: os.path.split(inputs[i])[0]
-        for i in range(len(inputs))
-    }
-    sampled_filepath_dict = {}
-    sampled_size_dict = {}
-    if not is_local:
-        if os.path.exists("sizes.json"):
-            os.remove("sizes.json")
-        wget.download(sizefile, "sizes.json")
-        sizefile = "sizes.json"
-    with open(sizefile, "r", encoding="UTF-8") as f:
-        load_dict = json.load(f)
-    L = int(len(file_path_dict) * proportion)
-    subkeys = random.sample(file_path_dict.keys(), L)
-    for k in subkeys:
-        sampled_size_dict[k] = load_dict[k]
-        sampled_filepath_dict[k] = file_path_dict[k]
-    return (
-        sum(sampled_size_dict.values()),
-        L,
-        [os.path.join(v, k) for k, v in sampled_filepath_dict.items()],
-        sampled_size_dict,
-    )
-
-
-def get_mel(audio_data, audio_cfg):
-    # mel shape: (n_mels, T)
-    mel = torchaudio.transforms.MelSpectrogram(
-        sample_rate=audio_cfg["sample_rate"],
-        n_fft=audio_cfg["window_size"],
-        win_length=audio_cfg["window_size"],
-        hop_length=audio_cfg["hop_size"],
-        center=True,
-        pad_mode="reflect",
-        power=2.0,
-        norm=None,
-        onesided=True,
-        n_mels=64,
-        f_min=audio_cfg["fmin"],
-        f_max=audio_cfg["fmax"],
-    ).to(audio_data.device)
-    mel = mel(audio_data)
-    # Align to librosa:
-    # librosa_melspec = librosa.feature.melspectrogram(
-    #     waveform,
-    #     sr=audio_cfg['sample_rate'],
-    #     n_fft=audio_cfg['window_size'],
-    #     hop_length=audio_cfg['hop_size'],
-    #     win_length=audio_cfg['window_size'],
-    #     center=True,
-    #     pad_mode="reflect",
-    #     power=2.0,
-    #     n_mels=64,
-    #     norm=None,
-    #     htk=True,
-    #     f_min=audio_cfg['fmin'],
-    #     f_max=audio_cfg['fmax']
-    # )
-    # we use log mel spectrogram as input
-    mel = torchaudio.transforms.AmplitudeToDB(top_db=None)(mel)
-    return mel.T  # (T, n_mels)
-
-
-def get_audio_features(
-    sample, audio_data, max_len, data_truncating, data_filling, audio_cfg
-):
-    """
-    Calculate and add audio features to sample.
-    Sample: a dict containing all the data of current sample.
-    audio_data: a tensor of shape (T) containing audio data.
-    max_len: the maximum length of audio data.
-    data_truncating: the method of truncating data.
-    data_filling: the method of filling data.
-    audio_cfg: a dict containing audio configuration. Comes from model_cfg['audio_cfg'].
-    """
-    with torch.no_grad():
-        if len(audio_data) > max_len:
-            if data_truncating == "rand_trunc":
-                longer = torch.tensor([True])
-            elif data_truncating == "fusion":
-                # fusion
-                mel = get_mel(audio_data, audio_cfg)
-                # split to three parts
-                chunk_frames = (
-                    max_len // audio_cfg["hop_size"] + 1
-                )  # the +1 related to how the spectrogram is computed
-                total_frames = mel.shape[0]
-                if chunk_frames == total_frames:
-                    # there is a corner case where the audio length is
-                    # larger than max_len but smaller than max_len+hop_size.
-                    # In this case, we just use the whole audio.
-                    mel_fusion = torch.stack([mel, mel, mel, mel], dim=0)
-                    sample["mel_fusion"] = mel_fusion
-                    longer = torch.tensor([False])
-                else:
-                    ranges = np.array_split(
-                        list(range(0, total_frames - chunk_frames + 1)), 3
-                    )
-                    # print('total_frames-chunk_frames:', total_frames-chunk_frames,
-                    #       'len(audio_data):', len(audio_data),
-                    #       'chunk_frames:', chunk_frames,
-                    #       'total_frames:', total_frames)
-                    if len(ranges[1]) == 0:
-                        # if the audio is too short, we just use the first chunk
-                        ranges[1] = [0]
-                    if len(ranges[2]) == 0:
-                        # if the audio is too short, we just use the first chunk
-                        ranges[2] = [0]
-                    # randomly choose index for each part
-                    idx_front = np.random.choice(ranges[0])
-                    idx_middle = np.random.choice(ranges[1])
-                    idx_back = np.random.choice(ranges[2])
-                    # select mel
-                    mel_chunk_front = mel[idx_front : idx_front + chunk_frames, :]
-                    mel_chunk_middle = mel[idx_middle : idx_middle + chunk_frames, :]
-                    mel_chunk_back = mel[idx_back : idx_back + chunk_frames, :]
-
-                    # shrink the mel
-                    mel_shrink = torchvision.transforms.Resize(size=[chunk_frames, 64])(
-                        mel[None]
-                    )[0]
-                    # logging.info(f"mel_shrink.shape: {mel_shrink.shape}")
-
-                    # stack
-                    mel_fusion = torch.stack(
-                        [mel_chunk_front, mel_chunk_middle, mel_chunk_back, mel_shrink],
-                        dim=0,
-                    )
-                    sample["mel_fusion"] = mel_fusion
-                    longer = torch.tensor([True])
-            else:
-                raise NotImplementedError(
-                    f"data_truncating {data_truncating} not implemented"
-                )
-            # random crop to max_len (for compatibility)
-            overflow = len(audio_data) - max_len
-            idx = np.random.randint(0, overflow + 1)
-            audio_data = audio_data[idx : idx + max_len]
-
-        else:  # padding if too short
-            if len(audio_data) < max_len:  # do nothing if equal
-                if data_filling == "repeatpad":
-                    n_repeat = int(max_len / len(audio_data))
-                    audio_data = audio_data.repeat(n_repeat)
-                    # audio_data = audio_data.unsqueeze(0).unsqueeze(0).unsqueeze(0)
-                    # audio_data = F.interpolate(audio_data,size=max_len,mode="bicubic")[0,0,0]
-                    audio_data = F.pad(
-                        audio_data,
-                        (0, max_len - len(audio_data)),
-                        mode="constant",
-                        value=0,
-                    )
-                elif data_filling == "pad":
-                    audio_data = F.pad(
-                        audio_data,
-                        (0, max_len - len(audio_data)),
-                        mode="constant",
-                        value=0,
-                    )
-                elif data_filling == "repeat":
-                    n_repeat = int(max_len / len(audio_data))
-                    audio_data = audio_data.repeat(n_repeat + 1)[:max_len]
-                else:
-                    raise NotImplementedError(
-                        f"data_filling {data_filling} not implemented"
-                    )
-            if data_truncating == "fusion":
-                mel = get_mel(audio_data, audio_cfg)
-                mel_fusion = torch.stack([mel, mel, mel, mel], dim=0)
-                sample["mel_fusion"] = mel_fusion
-            longer = torch.tensor([False])
-
-    sample["longer"] = longer
-    sample["waveform"] = audio_data
-
-    return sample
-
-
-def preprocess(
-    sample,
-    audio_ext,
-    text_ext,
-    max_len,
-    audio_cfg,
-    class_index_dict=None,
-    data_filling="pad",
-    data_truncating="rand_trunc",
-    text_augment_selection=None,
-):
-    """
-    Preprocess a single sample for wdsdataloader.
-    """
-    audio_data, orig_sr = sf.read(io.BytesIO(sample[audio_ext]))
-    audio_data = int16_to_float32(float32_to_int16(audio_data))
-    audio_data = torch.tensor(audio_data).float()
-
-    # TODO: (yusong) to be include in the future
-    # # if torchaudio not installed, use soundfile to load audio
-    # if torchaudio is None:
-    #     audio_data, orig_sr = sf.read(io.BytesIO(sample[audio_ext]))
-    #     audio_data = torch.tensor(audio_data).float()
-    # else:
-    #     # https://github.com/webdataset/webdataset/blob/main/webdataset/autodecode.py
-    #     with tempfile.TemporaryDirectory() as dirname:
-    #         os.makedirs(dirname, exist_ok=True)
-    #         fname = os.path.join(dirname, f"file.flac")
-    #         with open(fname, "wb") as stream:
-    #             stream.write(sample[audio_ext])
-    #         audio_data, orig_sr = torchaudio.load(fname)
-    #         audio_data = audio_data[0, :].float()
-
-    sample = get_audio_features(
-        sample, audio_data, max_len, data_truncating, data_filling, audio_cfg
-    )
-    del sample[audio_ext]
-
-    try:
-        json_dict_raw = json.loads(sample[text_ext].decode("utf-8"))
-    except:
-        print("sample[__url__]:", sample["__url__"])
-
-    # For selecting augmented text from dataset
-    if text_augment_selection is None or text_augment_selection == "none":
-        texts = json_dict_raw["text"]
-    elif text_augment_selection == "all":
-        if "text_augment_all" in json_dict_raw.keys():
-            texts = json_dict_raw["text_augment_all"]
-        else:
-            texts = json_dict_raw["text"]
-    elif text_augment_selection == "augment_only":
-        if "text_augment_all" in json_dict_raw.keys():
-            if json_dict_raw["text_augment_t5"] is None:
-                texts = json_dict_raw["text"]
-            else:
-                texts = json_dict_raw["text_augment_t5"]
-        else:
-            texts = json_dict_raw["text"]
-    else:
-        raise NotImplementedError(
-            f"text_augment_selection {text_augment_selection} not implemented"
-        )
-    sample["full_text"] = texts
-
-    if isinstance(texts, list) and isinstance(texts[0], str) and len(texts) > 1:
-        texts = random.choice(texts)
-    sample["raw_text"] = texts
-    sample["text"] = tokenizer(texts)  # text shape: [num_token]
-    if class_index_dict is not None:
-        # https://stackoverflow.com/questions/48004243/how-to-share-large-read-only-dictionary-list-across-processes-in-multiprocessing
-        # https://stackoverflow.com/questions/45693949/storing-strings-in-a-multiprocessing-sharedctypes-array
-        # key, val = class_index_dict
-        # key = key[:].split('\n')
-        # _dict = {k: v for k, v in zip(key, val)}
-        sample["class_label"] = np.zeros(len(class_index_dict.keys()))
-        for x in json_dict_raw["tag"]:
-            sample["class_label"][class_index_dict[x]] = 1
-        sample["class_label"] = torch.tensor(sample["class_label"]).float()
-    del sample[text_ext]
-    sample["audio_name"] = sample["__key__"].split("/")[-1] + "." + audio_ext
-    sample["text_name"] = sample["__key__"].split("/")[-1] + "." + text_ext
-    sample["audio_orig_sr"] = orig_sr
-    return sample
-
-
-def collate_fn(batch):
-    """
-    Collate function for wdsdataloader.
-    batch: a list of dict, each dict is a sample
-    """
-    # concatenate values in each dictionary. if it is a tensor, concatenate. if it is a list, extend.
-    batch_dict = {}
-    for k in batch[0].keys():
-        if isinstance(batch[0][k], dict):  # dealwith bert tokenizer output
-            batch_dict[k] = {}
-            for kk in batch[0][k].keys():
-                tmp = []
-                for i in range(len(batch)):
-                    tmp.append(batch[i][k][kk])
-                batch_dict[k][kk] = torch.vstack(tmp)
-        elif isinstance(batch[0][k], torch.Tensor):
-            batch_dict[k] = torch.stack([sample[k] for sample in batch])
-        elif isinstance(batch[0][k], np.ndarray):
-            batch_dict[k] = torch.tensor(np.stack([sample[k] for sample in batch]))
-        else:
-            batch_dict[k] = [sample[k] for sample in batch]
-    return batch_dict
-
-
-def get_wds_dataset(
-    args,
-    model_cfg,
-    is_train,
-    audio_ext="flac",
-    text_ext="json",
-    max_len=480000,
-    proportion=1.0,
-    sizefilepath_=None,
-    is_local=None,
-):
-    """
-    Get a dataset for wdsdataloader.
-    """
-    if is_local is None and (not args.remotedata is None):
-        is_local = not args.remotedata
-
-    input_shards = args.train_data if is_train else args.val_data
-    assert input_shards is not None
-
-    if not sizefilepath_ is None:
-        sizefilepath = sizefilepath_
-    else:
-        sizefilepath = os.path.join(os.path.dirname(input_shards[0]), "sizes.json")
-
-    if proportion != 1.0:
-        num_samples, num_shards, input_shards, _ = sample_prop(
-            sizefilepath, input_shards, proportion, is_local=is_local
-        )
-    else:
-        num_samples, num_shards = get_dataset_size(
-            input_shards, sizefilepath_=sizefilepath_, is_local=is_local
-        )
-
-    if not num_samples:
-        if is_train:
-            num_samples = args.train_num_samples
-            if not num_samples:
-                raise RuntimeError(
-                    "Currently, number of dataset samples must be specified for training dataset. "
-                    "Please specify via `--train-num-samples` if no dataset length info present."
-                )
-        else:
-            num_samples = (
-                args.val_num_samples or 0
-            )  # eval will just exhaust the iterator if not specified
-
-    pipeline = [wds.SimpleShardList(input_shards)]
-    # at this point we have an iterator over all the shards
-    # TODO: (yusong): add a if statement of distributed. If not, we don't need to split_by_node
-    if is_train or args.parallel_eval:
-        pipeline.extend(
-            [
-                wds.detshuffle(
-                    bufsize=_SHARD_SHUFFLE_SIZE,
-                    initial=_SHARD_SHUFFLE_INITIAL,
-                    seed=args.seed,
-                ),
-                wds.split_by_node,
-                wds.split_by_worker,
-                # at this point, we have an iterator over the shards assigned to each worker at each node
-                wds.tarfile_to_samples(handler=log_and_continue),
-                wds.shuffle(
-                    bufsize=_SAMPLE_SHUFFLE_SIZE,
-                    initial=_SAMPLE_SHUFFLE_INITIAL,
-                    rng=random.Random(args.seed),
-                ),
-                # wds.repeatedly,  # FIXME determine if this is beneficial
-            ]
-        )
-    else:
-        pipeline.extend(
-            [
-                wds.split_by_worker,
-                # at this point, we have an iterator over the shards assigned to each worker
-                wds.tarfile_to_samples(handler=log_and_continue),
-            ]
-        )
-    pipeline.append(
-        wds.map(
-            partial(
-                preprocess,
-                audio_ext=audio_ext,
-                text_ext=text_ext,
-                max_len=max_len,
-                audio_cfg=model_cfg["audio_cfg"],
-                class_index_dict=copy.deepcopy(args.class_index_dict),
-                data_filling=args.data_filling,
-                data_truncating=args.data_truncating,
-                text_augment_selection=args.text_augment_selection,
-            )
-        ),
-    )
-
-    pipeline.append(
-        wds.batched(
-            args.batch_size,
-            partial=not (is_train or args.parallel_eval),
-            collation_fn=collate_fn,
-        )
-    )
-
-    dataset = wds.DataPipeline(*pipeline)
-    if is_train or args.parallel_eval:
-        # (yusong): Currently parallel evaluation will be not precise as we are repeat the last few samples.
-        # (yusong): See comments below.
-        # roll over and repeat a few samples to get same number of full batches on each node
-        global_batch_size = args.batch_size * args.world_size
-        num_batches = math.ceil(num_samples / global_batch_size)
-        num_workers = max(1, args.workers)
-        num_worker_batches = math.ceil(
-            num_batches / num_workers
-        )  # per dataloader worker
-        num_batches = num_worker_batches * num_workers
-        num_samples = num_batches * global_batch_size
-        dataset = dataset.with_epoch(
-            num_worker_batches
-        )  # each worker is iterating over this
-    else:
-        # last batches are partial, eval is done on single (master) node
-        num_batches = math.ceil(num_samples / args.batch_size)
-
-    kwargs = {}
-    if args.horovod:  # multi-node training on summit
-        kwargs["multiprocessing_context"] = "forkserver"
-
-    dataloader = wds.WebLoader(
-        dataset, batch_size=None, shuffle=False, num_workers=args.workers, **kwargs
-    )
-
-    # FIXME not clear which approach is better, with_epoch before vs after dataloader?
-    # hoping to resolve via https://github.com/webdataset/webdataset/issues/169
-    # if is_train:
-    #     # roll over and repeat a few samples to get same number of full batches on each node
-    #     global_batch_size = args.batch_size * args.world_size
-    #     num_batches = math.ceil(num_samples / global_batch_size)
-    #     num_workers = max(1, args.workers)
-    #     num_batches = math.ceil(num_batches / num_workers) * num_workers
-    #     num_samples = num_batches * global_batch_size
-    #     dataloader = dataloader.with_epoch(num_batches)
-    # else:
-    #     # last batches are partial, eval is done on single (master) node
-    #     num_batches = math.ceil(num_samples / args.batch_size)
-
-    # add meta-data to dataloader instance for convenience
-    dataloader.num_batches = num_batches
-    dataloader.num_samples = num_samples
-
-    return DataInfo(dataloader, None)
-
-
-def wds_batch_list2dict(
-    batch,
-    keys=[
-        "__url__",
-        "__key__",
-        "waveform",
-        "text",
-        "raw_text",
-        "audio_name",
-        "text_name",
-        "audio_orig_sr",
-    ],
-):
-    """
-    Return a dictionary of the batch, with keys as the names of the fields.
-    """
-    assert len(keys) == len(
-        batch
-    ), "batch must have same number of keys as keys argument"
-    return {keys[i]: batch[i] for i in range(len(batch))}
-
-
-def get_csv_dataset(args, preprocess_fn, is_train):
-    input_filename = args.train_data if is_train else args.val_data
-    assert input_filename
-    dataset = CsvDataset(
-        input_filename,
-        preprocess_fn,
-        img_key=args.csv_img_key,
-        caption_key=args.csv_caption_key,
-        sep=args.csv_separator,
-    )
-    num_samples = len(dataset)
-    sampler = DistributedSampler(dataset) if args.distributed and is_train else None
-    shuffle = is_train and sampler is None
-
-    dataloader = DataLoader(
-        dataset,
-        batch_size=args.batch_size,
-        shuffle=shuffle,
-        num_workers=args.workers,
-        pin_memory=True,
-        sampler=sampler,
-        drop_last=is_train,
-    )
-    dataloader.num_samples = num_samples
-    dataloader.num_batches = len(dataloader)
-
-    return DataInfo(dataloader, sampler)
-
-
-def get_toy_dataset(args, model_cfg, is_train):
-    index_path = args.train_data if is_train else args.val_data
-    ipc_path = args.train_ipc if is_train else args.val_ipc
-    assert index_path and ipc_path
-    eval_mode = not is_train
-    dataset = ToyDataset(index_path, ipc_path, model_cfg, eval_mode=eval_mode)
-
-    num_samples = len(dataset)
-    sampler = (
-        DistributedSampler(dataset, shuffle=False)
-        if args.distributed and is_train
-        else None
-    )
-
-    dataloader = DataLoader(
-        dataset,
-        batch_size=args.batch_size,
-        shuffle=False,
-        num_workers=args.workers,
-        sampler=sampler,
-        drop_last=is_train,
-    )
-    dataloader.num_samples = num_samples
-    dataloader.num_batches = len(dataloader)
-
-    return DataInfo(dataloader, sampler)
-
-
-def get_dataset_fn(data_path, dataset_type):
-    if dataset_type == "webdataset":
-        return get_wds_dataset
-    elif dataset_type == "csv":
-        return get_csv_dataset
-    elif dataset_type == "auto":
-        ext = data_path.split(".")[-1]
-        if ext in ["csv", "tsv"]:
-            return get_csv_dataset
-        elif ext in ["tar"]:
-            return get_wds_dataset
-        else:
-            raise ValueError(
-                f"Tried to figure out dataset type, but failed for extention {ext}."
-            )
-    elif dataset_type == "toy":
-        return get_toy_dataset
-    else:
-        raise ValueError(f"Unsupported dataset type: {dataset_type}")
-
-
-def get_data(args, model_cfg):
-    data = {}
-
-    args.class_index_dict = load_class_label(args.class_label_path)
-
-    if args.datasetinfos is None:
-        args.datasetinfos = ["train", "unbalanced_train", "balanced_train"]
-    if args.dataset_type == "webdataset":
-        args.train_data = get_tar_path_from_dataset_name(
-            args.datasetnames,
-            args.datasetinfos,
-            islocal=not args.remotedata,
-            proportion=args.dataset_proportion,
-            dataset_path=args.datasetpath,
-            full_dataset=args.full_train_dataset,
-        )
-
-        if args.full_train_dataset is None:
-            args.full_train_dataset = []
-        if args.exclude_eval_dataset is None:
-            args.exclude_eval_dataset = []
-        excluded_eval_datasets = args.full_train_dataset + args.exclude_eval_dataset
-
-        val_dataset_names = (
-            [n for n in args.datasetnames if n not in excluded_eval_datasets]
-            if excluded_eval_datasets
-            else args.datasetnames
-        )
-        args.val_dataset_names = val_dataset_names
-        args.val_data = get_tar_path_from_dataset_name(
-            val_dataset_names,
-            ["valid", "test", "eval"],
-            islocal=not args.remotedata,
-            proportion=1,
-            dataset_path=args.datasetpath,
-            full_dataset=None,
-        )
-
-    if args.train_data:
-        data["train"] = get_dataset_fn(args.train_data, args.dataset_type)(
-            args, model_cfg, is_train=True
-        )
-
-    if args.val_data:
-        data["val"] = get_dataset_fn(args.val_data, args.dataset_type)(
-            args, model_cfg, is_train=False
-        )
-
-    return data