AI駆動型オーディオビジュアル生成

import librosa
import numpy as np
import torch
from torch.utils.data import Dataset, DataLoader

class AudioDataset(Dataset):
    def __init__(self, audio_files, sample_rate=44100, window_size=2048, hop_length=1024):
        self.audio_files = audio_files
        self.sample_rate = sample_rate
        self.window_size = window_size
        self.hop_length = hop_length
        self.spectrograms = []

        for file in self.audio_files:
            y, sr = librosa.load(file, sr=self.sample_rate)
            if y.size > 0 :  # 読み込みエラーを回避
                stft = librosa.stft(y, n_fft=self.window_size, hop_length=self.hop_length, window='hamming')
                spectrogram = np.abs(stft)
                self.spectrograms.append(spectrogram)

        self.spectrograms = np.array(self.spectrograms) # リストをnumpy配列に変換

    def __len__(self):
        return len(self.spectrograms)

    def __getitem__(self, idx):
        return torch.from_numpy(self.spectrograms[idx]).float()

def load_audio_data(data_dir, batch_size=32):
    import os
    audio_files = [os.path.join(data_dir, f) for f in os.listdir(data_dir) if f.endswith('.wav')]
    dataset = AudioDataset(audio_files)
    dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)
    return dataloader

import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim

class Encoder(nn.Module):
    def __init__(self, input_size, hidden_size, latent_size):
        super(Encoder, self).__init__()
        self.fc1 = nn.Linear(input_size, hidden_size)
        self.fc_mu = nn.Linear(hidden_size, latent_size)
        self.fc_logvar = nn.Linear(hidden_size, latent_size)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        mu = self.fc_mu(x)
        logvar = self.fc_logvar(x)
        return mu, logvar

class Decoder(nn.Module):
    def __init__(self, latent_size, hidden_size, output_size):
        super(Decoder, self).__init__()
        self.fc1 = nn.Linear(latent_size, hidden_size)
        self.fc2 = nn.Linear(hidden_size, output_size)

    def forward(self, z):
        z = F.relu(self.fc1(z))
        return torch.sigmoid(self.fc2(z))  # スペクトログラムの値は0~1に正規化されていることを前提とする

class VAE(nn.Module):
    def __init__(self, input_size, hidden_size, latent_size):
        super(VAE, self).__init__()
        self.encoder = Encoder(input_size, hidden_size, latent_size)
        self.decoder = Decoder(latent_size, hidden_size, input_size)

    def reparameterize(self, mu, logvar):
        std = torch.exp(0.5 * logvar)
        eps = torch.randn_like(std)
        return mu + eps * std

    def forward(self, x):
        mu, logvar = self.encoder(x)
        z = self.reparameterize(mu, logvar)
        reconstructed_x = self.decoder(z)
        return reconstructed_x, mu, logvar

def loss_function(recon_x, x, mu, logvar):
    recon_loss = F.mse_loss(recon_x, x, reduction='sum')
    kl_div = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp())
    return recon_loss + kl_div

def train_vae(model, dataloader, num_epochs=100, learning_rate=1e-3):
    optimizer = optim.Adam(model.parameters(), lr=learning_rate)
    for epoch in range(num_epochs):
        for batch_idx, x in enumerate(dataloader):
            optimizer.zero_grad()
            x = x.reshape(x.size(0), -1)
            recon_x, mu, logvar = model(x)
            loss = loss_function(recon_x, x, mu, logvar)
            loss.backward()
            optimizer.step()

        if epoch % 10 == 0 :
           print(f"Epoch: {epoch}, Loss: {loss.item()}")

    return model

def generate_audio(model, latent_size, output_size, num_samples=10):
   with torch.no_grad():
        z = torch.randn(num_samples, latent_size)
        spectrogram = model.decoder(z)

   return spectrogram.reshape(num_samples, 1025, -1).cpu().numpy()

import librosa

def reconstruct_audio(spectrogram, sample_rate=44100, window_size=2048, hop_length=1024, n_iter=100):
    audio = []
    for spec in spectrogram:
        mag_spec = np.abs(spec)
        phase = np.exp(1j * np.angle(librosa.stft(np.random.rand(1, 1000), n_fft=window_size, hop_length=hop_length, window='hamming')))
        y = librosa.istft(mag_spec * phase, hop_length=hop_length, window='hamming', length=None)
        for i in range(n_iter): # Griffin-Limアルゴリズムを使用
            phase = np.exp(1j * np.angle(librosa.stft(y, n_fft=window_size, hop_length=hop_length, window='hamming')))
            y = librosa.istft(mag_spec * phase, hop_length=hop_length, window='hamming', length=None)
        audio.append(y)
    return audio

import moviepy.editor as mp
import os
def create_video(audio_samples, image_samples, fps=30, output_path="output.mp4"):
    """
    音声と画像を同期させて動画を作成する関数
    :param audio_samples: list of np.array, オーディオのサンプルデータ
    :param image_samples: list of np.array, 画像データ
    :param fps: フレームレート
    :param output_path: 出力ファイルパス
    """
    if len(audio_samples) != len(image_samples):
        raise ValueError("オーディオと画像のサンプル数が一致しません。")

    audio_clips = []
    image_clips = []

    for i in range(len(audio_samples)):
        # 音声クリップを作成
        audio_samples[i] = np.array(audio_samples[i])
        audio_clip = mp.AudioClip(lambda t: audio_samples[i][int(t*44100)] if int(t*44100) < len(audio_samples[i]) else 0 , duration=len(audio_samples[i])/44100)
        audio_clips.append(audio_clip)

        # 画像クリップを作成
        image_samples[i] = np.array(image_samples[i])
        image_clip = mp.ImageClip(image_samples[i], duration=len(audio_samples[i])/44100)
        image_clips.append(image_clip)

    # 音声クリップを連結
    audio_final = mp.concatenate_audioclips(audio_clips)

    # 画像クリップを連結
    video_final = mp.concatenate_videoclips(image_clips)

    # 動画に音声を合成
    video_final = video_final.set_audio(audio_final)

    # 出力
    video_final.write_videofile(output_path, fps=fps, codec='libx264', audio_codec='aac')