<jupyter_start><jupyter_text>Audio Utils<jupyter_code>def RESHAPE(x):
    x = x.reshape((512,512))
    return x

def from_2D_To_3D(x, dv=0):
    x = x[..., np.newaxis]         
    return x

def stft_for_reconstruction(x, fft_size, hopsamp):
    window = np.hanning(fft_size)
    fft_size = int(fft_size)
    hopsamp = int(hopsamp)
    upper = len(x)-fft_size
    return np.array([np.fft.rfft(window*x[i:i+fft_size])
                     for i in range(0,upper, hopsamp)])

def get_signal(in_file, expected_fs=44100):
    fs, y = scipy.io.wavfile.read(in_file)
    num_type = y[0].dtype
    if num_type == 'int16':
        y = y*(1.0/32768)
    elif num_type == 'int32':
        y = y*(1.0/2147483648)
    elif num_type == 'float32':
        # Nothing to do
        pass
    elif num_type == 'uint8':
        raise Exception('8-bit PCM is not supported.')
    else:
        raise Exception('Unknown format.')
    if fs != expected_fs:
        raise Exception('Invalid sample rate.')
    if y.ndim == 1:
        return y
    else:
        return y.mean(axis=1)

def istft_for_reconstruction(X, fft_size, hopsamp):
    fft_size = int(fft_size)
    hopsamp = int(hopsamp)
    window = np.hanning(fft_size)
    time_slices = X.shape[0]
    len_samples = int(time_slices*hopsamp + fft_size)
    x = np.zeros(len_samples)
    for n,i in enumerate(range(0, len(x)-fft_size, hopsamp)):
        x[i:i+fft_size] += window*np.real(np.fft.irfft(X[n]))
    return x

def reconstruct_signal_griffin_lim(magnitude_spectrogram, fft_size, hopsamp, iterations):
    time_slices = magnitude_spectrogram.shape[0]
    len_samples = int(time_slices*hopsamp + fft_size)
    x_reconstruct = np.random.randn(len_samples)
    n = iterations # number of iterations of Griffin-Lim algorithm.
    while n > 0:
        n -= 1
        reconstruction_spectrogram = stft_for_reconstruction(x_reconstruct, fft_size, hopsamp)
        reconstruction_angle = np.angle(reconstruction_spectrogram)
        proposal_spectrogram = magnitude_spectrogram*np.exp(1.0j*reconstruction_angle)
        prev_x = x_reconstruct
        x_reconstruct = istft_for_reconstruction(proposal_spectrogram, fft_size, hopsamp)
        diff = math.sqrt(sum((x_reconstruct - prev_x)**2)/x_reconstruct.size)
        print('Reconstruction iteration: {}/{} RMSE: {} '.format(iterations - n, iterations, diff))

    return x_reconstruct

def array_to_wav(stft_modified_scaled,ITER=300,FFT_SIZE=1022, path_out = 'out.wav'):
    def save_audio_to_file(x, sample_rate, outfile='out.wav'):
        x_max = np.max(abs(x))
        assert x_max <= 1.0, 'Input audio value is out of range. Should be in the range [-1.0, 1.0].'
        x = x*32767.0
        data = array.array('h')
        for i in range(len(x)):
            cur_samp = int(round(x[i]))
            data.append(cur_samp)
        f = wave.open(outfile, 'w')
        f.setparams((1, 2, sample_rate, 0, "NONE", "Uncompressed"))
        f.writeframes(data.tobytes())
        f.close()
    stft_modified_scaled = RESHAPE(stft_modified_scaled)
    hopsamp = FFT_SIZE // 8
    x_reconstruct = reconstruct_signal_griffin_lim(stft_modified_scaled, FFT_SIZE, hopsamp, ITER)
    max_sample = np.max(abs(x_reconstruct))
    if max_sample > 1.0:
        x_reconstruct = x_reconstruct / max_sample
    save_audio_to_file(x_reconstruct, 44100, path_out)

def wav_to_array(path_src,ITER = 300, FFT_SIZE = 1022):
    # input_signal = input_signal[:16600] #256x256 #FFT_SIZE: 510 -> #256x256 
    hopsamp = FFT_SIZE // 8
    input_signal = get_signal(path_src, 44100)
    input_signal = input_signal[:65990]
    stft_full = stft_for_reconstruction(input_signal, FFT_SIZE, hopsamp)
    stft_mag = abs(stft_full)**2.0
    scale = 1.0 / np.amax(stft_mag)
    stft_mag *= scale
    stft_modified_scaled = stft_mag / scale
    return stft_modified_scaled**0.5<jupyter_output><empty_output><jupyter_text>Data Loader<jupyter_code>class DataLoader:
    def __init__(self,wav_res=(512, 512)):
        self.img_res = wav_res

    def load_data(self, domain, batch_size=1):
        data_type = '/content/gdrive/My Drive/CycleGAN/Dataset/{}/*'.format(domain)
        path = glob(data_type)

        batch_wav = np.random.choice(path, size=batch_size)

        wav = []
        for wav_path in batch_wav:
            w = self.imread(wav_path)
            wav.append(w)
        wav = np.array(wav)
        return wav

    def load_batch(self, batch_size=1):
        path_A = glob('/content/gdrive/My Drive/CycleGAN/Dataset/Piano/*')
        path_B = glob('/content/gdrive/My Drive/CycleGAN/Dataset/Harpsichord/*')
        
        assert len(path_A) > 0 and len(path_B) > 0, "DATASET EMPTY"  
        
        self.n_batches = int(min(len(path_A), len(path_B)) / batch_size)
        total_samples = self.n_batches * batch_size

        # Sample n_batches * batch_size from each path list so that model sees all
        # samples from both domains
        A = np.random.choice(path_A, total_samples, replace=False)
        B = np.random.choice(path_B, total_samples, replace=False)

        for i in range(self.n_batches-1):
            batch_A = A[i*batch_size:(i+1)*batch_size]
            batch_B = B[i*batch_size:(i+1)*batch_size]
            wav_A, wav_B = [], []
            for w_A, w_B in zip(batch_A, batch_B):
                w_A = self.imread(w_A)
                w_B = self.imread(w_B)
                wav_A.append(w_A)
                wav_B.append(w_B)
            wav_A = np.array(wav_A)
            wav_B = np.array(wav_B)
            


            yield wav_A, wav_B

    def imread(self, path):
      array_wav = wav_to_array(path) # --> 512x512
      a = from_2D_To_3D(array_wav) 
      return a<jupyter_output><empty_output><jupyter_text>CycleGAN<jupyter_code>class CycleGAN:
    def __init__(self, dataset):
        # Input shape
        self.img_rows = 512
        self.img_cols = 512
        self.channels = 1
        self.img_shape = (self.img_rows,self.img_cols, self.channels)
        self.path_checkpoint = '/content/gdrive/My Drive/CycleGAN/Fixed_version/Checkpoints'
        self.data_loader = dataset
        self.loaded_epoch = 0
        
        # Bot Telegram
#         self.bot = telepot.Bot('740587014:AAHkwyhU5W4yZROjlXwtvOmxTsNl0CWHwJk')
#         self.chat_id = '-376260969'
        
        # Calculate output shape of D (PatchGAN)
        patch = int(self.img_rows / 2 ** 4)
        self.disc_patch = (patch, patch, 1)

        # Number of filters in the first layer of G and D
        self.gf = 64
        self.df = 64

        # Loss weights
        self.lambda_cycle = 10.0  # Cycle-consistency loss
        self.lambda_id = 0.5 * self.lambda_cycle  # Identity loss
      
        optimizer = Adam(0.0001, 0, 0.9)

        # Build and compile the discriminators
        self.d_A = self.build_discriminator()
        self.d_B = self.build_discriminator()
        self.d_A.compile(loss='mse',
                         optimizer=optimizer,
                         metrics=['accuracy'])
        self.d_B.compile(loss='mse',
                         optimizer=optimizer,
                         metrics=['accuracy'])
        # -------------------------------------------
        # Construct Computational Graph of Generators
        # -------------------------------------------

        # Build the generators
        self.g_AB = self.build_generator()
        self.g_AB.summary()
        self.g_BA = self.build_generator()

        # Input images from both domains
        img_A = Input(shape=self.img_shape)
        img_B = Input(shape=self.img_shape)

        # Translate images to the other domain
        fake_B = self.g_AB(img_A)
        fake_A = self.g_BA(img_B)
        # Translate images back to original domain
        reconstr_A = self.g_BA(fake_B)
        reconstr_B = self.g_AB(fake_A)
        # Identity mapping of images
        img_A_id = self.g_BA(img_A)
        img_B_id = self.g_AB(img_B)

        # For the combined model we will only train the generators
        self.d_A.trainable = False
        self.d_B.trainable = False

        # Discriminators determines validity of translated images
        valid_A = self.d_A(fake_A)
        valid_B = self.d_B(fake_B)

        # Combined model trains generators to fool discriminators
        self.combined = Model(inputs=[img_A, img_B],
                              outputs=[
                                  valid_A, valid_B,        #Discriminator
                                  reconstr_A, reconstr_B,  #partendo da B ritorno in A, Partendo da A ritorno in B
                                  img_A_id, img_B_id       #Discriminator per il generatore reconstr[AB]
                              ])
        self.combined.compile(
            loss=['mse', 'mse','mae', 'mae','mae', 'mae'],
            loss_weights=[1,1,self.lambda_cycle, self.lambda_cycle,self.lambda_id, self.lambda_id],
            optimizer=optimizer )
        
        
        

    def build_generator(self):
        def conv2d(layer_input, filters, f_size=4, stride=2):
            d = Conv2D(filters, kernel_size=f_size, strides=stride, padding='same')(layer_input)
            d = InstanceNormalization()(d)
            d = Activation('relu')(d)
            return d

        def res_block(x, filters=256, use_dropout=False):
            y = conv2d(x, filters, 3, 1)
            if use_dropout:
                y = Dropout(0.5)(y)
            y = conv2d(y, filters, 3, 1)
            return Add()([y, x])

        def deconv2d(layer_input, filters, f_size=4, dropout_rate=0):
            u = Conv2DTranspose(filters, kernel_size=f_size, strides=2, padding='same')(layer_input)
            if dropout_rate:
                u = Dropout(dropout_rate)(u)
            u = InstanceNormalization()(u)
            u = Activation('relu')(u)
            return u

        # Image input
        d0 = Input(shape=self.img_shape)
        # c7s1-64
        d1 = conv2d(d0, self.gf, 7, 1)
        # d128    
        d2 = conv2d(d1, self.gf * 2, 3)
        # d256    
        d3 = conv2d(d2, self.gf * 4, 3)
        # R256 x 9 -> 512x512
        x = res_block(d3)
        x = res_block(x)
        x = res_block(x)
        x = res_block(x)
        x = res_block(x)
        x = res_block(x)
        x = res_block(x)
        x = res_block(x)
        x = res_block(x)
        # u128
        u1 = deconv2d(x, self.gf * 2, 3)
        # u64
        u2 = deconv2d(u1, self.gf, 3)
        # c7s1-3
        output_img = Conv2D(self.channels, kernel_size=7, strides=1, padding='same', activation='tanh')(u2)

        return Model(d0, output_img)

    def build_discriminator(self):
        def d_layer(layer_input, filters, f_size=4, normalization=True):
            d = Conv2D(filters, kernel_size=f_size, strides=2, padding='same')(layer_input)
            if normalization:
                d = InstanceNormalization()(d)
            d = LeakyReLU(alpha=0.2)(d)
            return d

        img = Input(shape=self.img_shape)
        # C64-C128-C256-C512
        d1 = d_layer(img, self.df, normalization=False)
        d2 = d_layer(d1, self.df * 2)
        d3 = d_layer(d2, self.df * 4)
        d4 = d_layer(d3, self.df * 8)
        validity = Conv2D(1, kernel_size=4, strides=1, padding='same')(d4)

        return Model(img, validity)

    def load_network(self, n):
        # returns a compiled model identical to the previous one
        self.combined.load_weights(self.path_checkpoint+"/combined_ep{}.h5".format(n))
        self.loaded_epoch = n + 1

    def train(self, epochs, batch_size=1, sample_interval=50):
        start_time = datetime.datetime.now()

        # Adversarial loss ground truths
        valid = np.ones((batch_size,) + self.disc_patch)
        fake = np.zeros((batch_size,) + self.disc_patch)

        for epoch in range(self.loaded_epoch, epochs):
            for batch_i, (imgs_A, imgs_B) in enumerate(self.data_loader.load_batch(batch_size)):

                #  Train Discriminators

                # Translate images to opposite domain
                fake_B = self.g_AB.predict(imgs_A)
                fake_A = self.g_BA.predict(imgs_B)
                
                # Train the discriminators (original images = real / translated = Fake)
                dA_loss_real = self.d_A.train_on_batch(imgs_A, valid)
                dA_loss_fake = self.d_A.train_on_batch(fake_A, fake)
                dA_loss = 0.5 * np.add(dA_loss_real, dA_loss_fake)
                dB_loss_real = self.d_B.train_on_batch(imgs_B, valid)
                dB_loss_fake = self.d_B.train_on_batch(fake_B, fake)
                dB_loss = 0.5 * np.add(dB_loss_real, dB_loss_fake)
                # Total disciminator loss
                d_loss = 0.5 * np.add(dA_loss, dB_loss)

                # Train the generators
                g_loss = self.combined.train_on_batch([imgs_A, imgs_B],
                                                      [valid, valid,
                                                       imgs_A, imgs_B,
                                                       imgs_A, imgs_B])

                elapsed_time = datetime.datetime.now() - start_time
                info = "[Epoch %d/%d] [Batch %d/%d] [D loss: %f, acc: %3d%%] [G loss: %05f, adv: %05f, recon: %05f, id: %05f] time: %s " \
                    % (epoch, epochs,
                       batch_i, self.data_loader.n_batches,
                       d_loss[0], 100 * d_loss[1],
                       g_loss[0],
                       np.mean(g_loss[1:3]),
                       np.mean(g_loss[3:5]),
                       np.mean(g_loss[5:6]),
                       elapsed_time)
                print( info ) # Plot the progress

                # If at save interval => save generated image samples
#                 if batch_i % sample_interval == 0:
#                     self.bot.sendMessage(self.chat_id, info)       
#                     self.sample_wav(epoch, batch_i)
                    
                
            #AD OGNI FINE EPOCA
            self.sample_wav(epoch)
            self.combined.save(self.path_checkpoint+"/combined_ep{}.h5".format(epoch))

    def sample_wav(self, epoch):
      dir_name = "/content/gdrive/My Drive/CycleGAN/Fixed_version/Result/Training/Epoch-{}".format(epoch)
      os.mkdir(dir_name)

      wav_A = self.data_loader.load_data(domain="Piano", batch_size=1)
      wav_B = self.data_loader.load_data(domain="Harpsichord", batch_size=1)
      
      # Translate images to the other domain
      fake_B = self.g_AB.predict(wav_A)
      fake_A = self.g_BA.predict(wav_B)
      
      # Translate back to original domain
      reconstr_A = self.g_BA.predict(fake_B)
      reconstr_B = self.g_AB.predict(fake_A)
      
      
      print("----- [Epoch {}] SAVING -----".format(epoch))
      #ORIGINAL
#       print("[Original] wav_A -> ",wav_A[0].shape, "wav_B -> ",wav_B[0].shape )
      array_to_wav(wav_A[0], path_out = dir_name+'/piano_original.wav')
      array_to_wav(wav_B[0], path_out = dir_name+'/harp_original.wav')
      #FAKE
#       print("[Fake] fake_A -> ",fake_A[0].shape, "fake_B -> ",fake_B[0].shape )
      array_to_wav(fake_A[0], path_out = dir_name+'/(Piano - Harp) harp_fake.wav')
      array_to_wav(fake_B[0], path_out = dir_name+'/(Harp - Piano)piano_fake.wav')
      #RECONSTR
#       print("[Reconstruction] fake_A -> ",fake_A[0].shape, "fake_B -> ",fake_B[0].shape )
      array_to_wav(reconstr_A[0], path_out = dir_name+'/piano_reconstr.wav')
      array_to_wav(reconstr_B[0], path_out = dir_name+'/harp_reconstr.wav')<jupyter_output><empty_output><jupyter_text>Load DataLoader<jupyter_code>ds = DataLoader()<jupyter_output><empty_output><jupyter_text>CycleGAN<jupyter_code>cyclegan = CycleGAN(ds)
#cyclegan.load_network(32) #Comincia da QUI<jupyter_output>__________________________________________________________________________________________________
Layer (type)                    Output Shape         Param #     Connected to                     
==================================================================================================
input_9 (InputLayer)            (None, 512, 512, 1)  0                                            
__________________________________________________________________________________________________
conv2d_65 (Conv2D)              (None, 512, 512, 64) 3200        input_9[0][0]                    
__________________________________________________________________________________________________
instance_normalization_59 (Inst (None, 512, 512, 64) 2           conv2d_65[0][0]                  
__________________________________________________________________________________________________
activation_47 (Activation)      (None, 512, 512, 64) 0           instance_normalization_59[0][0]  
__________[...]<jupyter_text>Start train<jupyter_code>cyclegan.train(epochs=50, batch_size=1, sample_interval=500)<jupyter_output>WARNING:tensorflow:From /usr/local/lib/python3.6/dist-packages/tensorflow/python/ops/math_ops.py:3066: to_int32 (from tensorflow.python.ops.math_ops) is deprecated and will be removed in a future version.
Instructions for updating:
Use tf.cast instead.
{'fraction_non_alphanumeric': 0.11931054630441133, 'fraction_numerical': 0.025182588372772423, 'mean_word_length': 2.2052434456928838, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 11, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '264584', 'license': 'Apache-2.0', 'n_tokens_mistral': 6089, 'n_tokens_neox': 5709, 'n_words': 1073, 'path': 'TimbreTron.ipynb', 'repo_name': 'edivadiranatnom/Machine-Learning-Project', 'size': '17115'}