Upload username_transformer.py

username_transformer.py

@@ -0,0 +1,539 @@
+# -*- coding: utf-8 -*-
+"""Username_Transformer
+
+Automatically generated by Colab.
+
+Original file is located at
+    https://colab.research.google.com/drive/1iae8ZzCuKYOPmMyTibAh7hVzwjbrW4Pe
+"""
+
+# Commented out IPython magic to ensure Python compatibility.
+# Install PyTorch
+# %pip install torch torchvision torchaudio
+
+# Install other dependencies
+# %pip install numpy pandas nltk elevenlabs requests
+
+import torch
+import torch.nn as nn
+from torch.utils.data import Dataset, DataLoader
+import numpy as np
+import nltk
+import re
+from collections import Counter
+from tqdm import tqdm
+import requests
+from nltk.corpus import cmudict
+import os
+import pandas as pd
+
+# allow cuDNN benchmark to pick fastest model
+import torch.backends.cudnn as cudnn
+cudnn.benchmark = True
+
+nltk.download('cmudict')
+
+cmu_dict = cmudict.dict()
+
+url = "https://raw.githubusercontent.com/danielmiessler/SecLists/master/Usernames/xato-net-10-million-usernames.txt"
+
+try:
+  response = requests.get(url)
+  response.raise_for_status()  # Raise an exception for bad status codes
+
+  usernames = response.text.splitlines()
+  print(f"Downloaded {len(usernames)} usernames.")
+
+except requests.exceptions.RequestException as e:
+  print(f"Error downloading usernames: {e}")
+  usernames = []
+
+def normalize_username(username):
+    # Convert to lowercase
+    username = username.lower()
+    # Replace numbers with words
+    num_to_word = {
+        '0': ' zero ', '1': ' one ', '2': ' two ', '3': ' three ',
+        '4': ' four ', '5': ' five ', '6': ' six ', '7': ' seven ',
+        '8': ' eight ', '9': ' nine '
+    }
+    for num, word in num_to_word.items():
+        username = username.replace(num, word)
+    # Replace special characters with spaces
+    username = re.sub(r'[\W_]+', ' ', username)
+    # Remove extra spaces
+    username = re.sub(r'\s+', ' ', username).strip()
+    return username
+
+def get_phonemes(word):
+    phonemes_list = cmu_dict.get(word)
+    if phonemes_list:
+        return phonemes_list[0]  # Use the first pronunciation
+    else:
+        return None  # Only show usernames that have correct phonemes
+
+def username_to_phonemes(username):
+    normalized = normalize_username(username)
+    words = normalized.split()
+    phonemes = []
+    for word in words:
+        phoneme = get_phonemes(word)
+        if phoneme:
+            phonemes.extend(phoneme)
+        # else:
+        #     print(f"Warning: Unable to find phonemes for word: {word}")
+    return phonemes
+
+input_sequences = []
+target_sequences = []
+
+for username in usernames:
+    input_seq = list(normalize_username(username))
+    target_seq = username_to_phonemes(username)
+    if target_seq:
+      input_sequences.append(input_seq)
+      target_sequences.append(target_seq)
+
+# Character Vocabulary
+char_counter = Counter([char for seq in input_sequences for char in seq])
+char_list = ['<pad>'] + sorted(char_counter.keys())
+char_vocab = {char: idx for idx, char in enumerate(char_list)}
+
+# Phoneme Vocabulary
+phoneme_counter = Counter([phoneme for seq in target_sequences for phoneme in seq])
+phoneme_list = ['<pad>', '<sos>', '<eos>'] + sorted(phoneme_counter.keys())
+phoneme_vocab = {phoneme: idx for idx, phoneme in enumerate(phoneme_list)}
+
+def encode_sequence(seq, vocab, max_len, add_special_tokens=False):
+    encoded = [vocab.get(token, vocab['<pad>']) for token in seq]
+    if add_special_tokens:
+        encoded = [vocab['<sos>']] + encoded + [vocab['<eos>']]
+    # Trim or pad the sequence to max_len
+    encoded = encoded[:max_len] + [vocab['<pad>']] * max(0, max_len - len(encoded))
+    return encoded
+
+
+max_input_len = max(len(seq) for seq in input_sequences)
+max_target_len = max(len(seq) for seq in target_sequences) + 2  # For <sos> and <eos>
+
+encoded_inputs = [encode_sequence(seq, char_vocab, max_input_len) for seq in input_sequences]
+encoded_targets = [encode_sequence(seq, phoneme_vocab, max_target_len, True) for seq in target_sequences]
+
+class UsernameDataset(Dataset):
+    def __init__(self, inputs, targets):
+        self.inputs = torch.tensor(inputs, dtype=torch.long)
+        self.targets = torch.tensor(targets, dtype=torch.long)
+
+    def __len__(self):
+        return len(self.inputs)
+
+    def __getitem__(self, idx):
+        return self.inputs[idx], self.targets[idx]
+
+dataset = UsernameDataset(encoded_inputs, encoded_targets)
+data_loader = DataLoader(dataset, batch_size=512, shuffle=True)
+
+# Function to decode sequences
+def decode_sequence(encoded_seq, vocab):
+    idx_to_token = {idx: token for token, idx in vocab.items()}
+    decoded_seq = [idx_to_token.get(idx, '<unk>') for idx in encoded_seq]
+    return decoded_seq
+
+# Create lists to store decoded usernames and pronunciations
+usernames = []
+pronunciations = []
+
+# Iterate through the dataset and decode sequences
+for input_seq, target_seq in dataset:
+    username = ''.join(decode_sequence(input_seq.tolist(), char_vocab))
+    pronunciation = ' '.join(decode_sequence(target_seq.tolist(), phoneme_vocab))
+    usernames.append(username)
+    pronunciations.append(pronunciation)
+
+# Create a Pandas DataFrame
+df = pd.DataFrame({'username': usernames, 'pronunciation': pronunciations})
+
+# Export to CSV
+df.to_csv('username_pronunciation.csv', index=False)
+
+class Encoder(nn.Module):
+    def __init__(self, input_dim, emb_dim, hid_dim):
+        super().__init__()
+        self.embedding = nn.Embedding(input_dim, emb_dim, padding_idx=char_vocab['<pad>'])
+        self.gru = nn.GRU(emb_dim, hid_dim, batch_first=True)
+
+    def forward(self, src):
+        embedded = self.embedding(src)
+        outputs, hidden = self.gru(embedded)
+        return outputs, hidden
+
+class Attention(nn.Module):
+    def __init__(self, hid_dim):
+        super().__init__()
+        self.attn = nn.Linear(hid_dim * 2, hid_dim)
+        self.v = nn.Linear(hid_dim, 1, bias=False)
+
+    def forward(self, hidden, encoder_outputs):
+        src_len = encoder_outputs.shape[1]
+        hidden = hidden.repeat(1, src_len, 1)
+        energy = torch.tanh(self.attn(torch.cat((hidden, encoder_outputs), dim=2)))
+        attention = self.v(energy).squeeze(2)
+        return torch.softmax(attention, dim=1)
+
+class Decoder(nn.Module):
+    def __init__(self, output_dim, emb_dim, hid_dim, attention):
+        super().__init__()
+        self.output_dim = output_dim
+        self.attention = attention
+        self.embedding = nn.Embedding(output_dim, emb_dim, padding_idx=phoneme_vocab['<pad>'])
+        self.gru = nn.GRU(emb_dim + hid_dim, hid_dim, batch_first=True)
+        self.fc_out = nn.Linear(hid_dim * 2, output_dim)
+
+    def forward(self, input, hidden, encoder_outputs):
+        input = input.unsqueeze(1)
+        embedded = self.embedding(input)
+        a = self.attention(hidden.permute(1, 0, 2), encoder_outputs)
+        a = a.unsqueeze(1)
+        weighted = torch.bmm(a, encoder_outputs)
+        rnn_input = torch.cat((embedded, weighted), dim=2)
+        output, hidden = self.gru(rnn_input, hidden)
+        output = torch.cat((output.squeeze(1), weighted.squeeze(1)), dim=1)
+        prediction = self.fc_out(output)
+        return prediction, hidden
+
+class Seq2Seq(nn.Module):
+    def __init__(self, encoder, decoder, device):
+        super().__init__()
+        self.encoder = encoder
+        self.decoder = decoder
+        self.device = device
+
+    def forward(self, src, trg, teacher_forcing_ratio=0.5):
+        batch_size = src.shape[0]
+        trg_len = trg.shape[1]
+        trg_vocab_size = self.decoder.output_dim
+
+        outputs = torch.zeros(batch_size, trg_len, trg_vocab_size).to(self.device)
+        encoder_outputs, hidden = self.encoder(src)
+        input = trg[:, 0]
+
+        for t in range(1, trg_len):
+            output, hidden = self.decoder(input, hidden, encoder_outputs)
+            outputs[:, t] = output
+            top1 = output.argmax(1)
+            teacher_force = np.random.random() < teacher_forcing_ratio
+            input = trg[:, t] if teacher_force else top1
+        return outputs
+
+def get_latest_checkpoint(directory):
+    # Get a list of all files in the directory
+    files = os.listdir(directory)
+
+    # Filter the list to only include g2p{n}.pth files
+    checkpoint_files = [f for f in files if re.match(r'g2p\d+\.pth', f)]
+
+    # Extract the numbers from the filenames
+    checkpoint_numbers = [int(re.search(r'g2p(\d+)\.pth', f).group(1)) for f in checkpoint_files]
+    print(checkpoint_numbers)
+
+    # Sort the files by their numbers
+    sorted_files = sorted(zip(checkpoint_numbers, checkpoint_files))
+
+    # Get the latest file (last element in the sorted list)
+    if sorted_files:
+        latest_file = sorted_files[-1][1]
+        latest_checkpoint_path = os.path.join(directory, latest_file)
+        return latest_checkpoint_path
+    else:
+        return None
+
+def get_next_version(directory):
+    files = os.listdir(directory)
+
+    # Filter the list to only include g2p{n}.pth files
+    checkpoint_files = [f for f in files if re.match(r'g2p\d+\.pth', f)]
+
+    # Extract the numbers from the filenames
+    checkpoint_numbers = [int(re.search(r'g2p(\d+)\.pth', f).group(1)) for f in checkpoint_files]
+    print(checkpoint_numbers)
+
+    # Sort the files by their numbers
+    sorted_files = sorted(zip(checkpoint_numbers, checkpoint_files))
+    if sorted_files:
+        latest_version = sorted_files[-1][0]
+        print(f"Latest version: {sorted_files[-1]}")
+        return latest_version + 1
+    else:
+        return 1  # Start with version 1 if no checkpoints exist
+
+def save_checkpoint(model, directory, version):
+    filename = f"g2p{version}.pth"
+    filepath = os.path.join(directory, filename)
+    torch.save(model.state_dict(), filepath)
+    print(f"Model saved to {filepath}")
+
+# Get the latest checkpoint file path
+directory = '/content/drive/MyDrive/AI/username_g2p/'
+latest_checkpoint_file = get_latest_checkpoint(directory)
+
+if latest_checkpoint_file:
+    print(f"Latest checkpoint file: {latest_checkpoint_file}")
+else:
+    print("No checkpoint files found.")
+
+print(get_next_version(directory))
+
+INPUT_DIM = len(char_vocab)
+OUTPUT_DIM = len(phoneme_vocab)
+ENC_EMB_DIM = 64
+DEC_EMB_DIM = 64
+HID_DIM = 128
+
+attn = Attention(HID_DIM)
+enc = Encoder(INPUT_DIM, ENC_EMB_DIM, HID_DIM)
+dec = Decoder(OUTPUT_DIM, DEC_EMB_DIM, HID_DIM, attn)
+
+device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+model = Seq2Seq(enc, dec, device).to(device)
+optimizer = torch.optim.Adam(model.parameters())
+criterion = nn.CrossEntropyLoss(ignore_index=phoneme_vocab['<pad>'])
+
+# Path to your checkpoint file
+checkpoint_file = latest_checkpoint_file if latest_checkpoint_file else 'g2p1.pth'
+
+# Check if the checkpoint file exists
+if os.path.exists(checkpoint_file):
+    # Load the checkpoint
+    print(f"Loading checkpoint from {checkpoint_file}")
+    model.load_state_dict(torch.load(checkpoint_file))
+else:
+    print(f"Checkpoint file not found. Using default initialization.")
+
+print(device)
+
+# Verify input sequences
+max_input_idx = max([max(seq) for seq in encoded_inputs])
+print(f'Max input index: {max_input_idx}, Input vocab size: {INPUT_DIM}')
+
+# Verify target sequences
+max_target_idx = max([max(seq) for seq in encoded_targets])
+print(f'Max target index: {max_target_idx}, Output vocab size: {OUTPUT_DIM}')
+
+def train(model, loader, optimizer, criterion, clip):
+    model.train()
+    epoch_loss = 0
+
+    for src, trg in tqdm(loader, desc="Training Batches"):
+        src, trg = src.to(device), trg.to(device)
+        optimizer.zero_grad()
+        output = model(src, trg)
+        output_dim = output.shape[-1]
+        output = output[:, 1:].reshape(-1, output_dim)
+        trg = trg[:, 1:].reshape(-1)
+        loss = criterion(output, trg)
+        loss.backward()
+        torch.nn.utils.clip_grad_norm_(model.parameters(), clip)
+        optimizer.step()
+        epoch_loss += loss.item()
+
+    return epoch_loss / len(loader)
+
+N_EPOCHS = 1
+CLIP = 1
+
+for epoch in range(N_EPOCHS):
+    loss = train(model, data_loader, optimizer, criterion, CLIP)
+    print(f'Epoch: {epoch+1}, Loss: {loss:.4f}')
+
+# Get the next version number
+next_version = get_next_version(directory)
+
+# Save the model with the new version number
+save_checkpoint(model, directory, next_version)
+
+def predict(model, username):
+    model.eval()
+    with torch.no_grad():
+        normalized = normalize_username(username)
+        input_seq = encode_sequence(list(normalized), char_vocab, max_input_len)
+        src = torch.tensor([input_seq], dtype=torch.long).to(device)
+        encoder_outputs, hidden = model.encoder(src)
+        input_token = torch.tensor([phoneme_vocab['<sos>']], dtype=torch.long).to(device)
+        outputs = []
+
+        for _ in range(max_target_len):
+            output, hidden = model.decoder(input_token, hidden, encoder_outputs)
+            top1 = output.argmax(1)
+            if top1.item() == phoneme_vocab['<eos>']:
+                break
+            outputs.append(top1.item())
+            input_token = top1
+
+        idx_to_phoneme = {idx: phoneme for phoneme, idx in phoneme_vocab.items()}
+        predicted_phonemes = [idx_to_phoneme[idx] for idx in outputs]
+        return ' '.join(predicted_phonemes)
+
+test_username = 'supercalafragalisticexpialadocous'
+test_username = 'barnabassacket'
+pronunciation = predict(model, test_username)
+print(f'Username: {test_username}')
+print(f'Pronunciation: {pronunciation}')
+
+# from https://github.com/margonaut/CMU-to-IPA-Converter/blob/master/cmu_ipa_mapping.rb
+CMU_IPA_MAPPING = {
+    "B": "b",
+    "CH": "ʧ",
+    "D": "d",
+    "DH": "ð",
+    "F": "f",
+    "G": "g",
+    "HH": "h",
+    "JH": "ʤ",
+    "K": "k",
+    "L": "l",
+    "M": "m",
+    "N": "n",
+    "NG": "ŋ",
+    "P": "p",
+    "R": "r",
+    "S": "s",
+    "SH": "ʃ",
+    "T": "t",
+    "TH": "θ",
+    "V": "v",
+    "W": "w",
+    "Y": "j",
+    "Z": "z",
+    "ZH": "ʒ",
+    "AA0": "ɑ",
+    "AA1": "ɑ",
+    "AA2": "ɑ",
+    "AE0": "æ",
+    "AE1": "æ",
+    "AE2": "æ",
+    "AH0": "ə",
+    "AH1": "ʌ",
+    "AH2": "ʌ",
+    "AO0": "ɔ",
+    "AO1": "ɔ",
+    "AO2": "ɔ",
+    "EH0": "ɛ",
+    "EH1": "ɛ",
+    "EH2": "ɛ",
+    "ER0": "ɚ",
+    "ER1": "ɝ",
+    "ER2": "ɝ",
+    "IH0": "ɪ",
+    "IH1": "ɪ",
+    "IH2": "ɪ",
+    "IY0": "i",
+    "IY1": "i",
+    "IY2": "i",
+    "UH0": "ʊ",
+    "UH1": "ʊ",
+    "UH2": "ʊ",
+    "UW0": "u",
+    "UW1": "u",
+    "UW2": "u",
+    "AW0": "aʊ",
+    "AW1": "aʊ",
+    "AW2": "aʊ",
+    "AY0": "aɪ",
+    "AY1": "aɪ",
+    "AY2": "aɪ",
+    "EY0": "eɪ",
+    "EY1": "eɪ",
+    "EY2": "eɪ",
+    "OW0": "oʊ",
+    "OW1": "oʊ",
+    "OW2": "oʊ",
+    "OY0": "ɔɪ",
+    "OY1": "ɔɪ",
+    "OY2": "ɔɪ"
+}
+
+pronunciation = predict(model, test_username)
+ipa_sequence = ''.join([CMU_IPA_MAPPING.get(phoneme, phoneme) for phoneme in pronunciation.split()])
+print(f'Username: {test_username}')
+print(f'Pronunciation: {ipa_sequence}')
+
+ssml_template = """<phoneme alphabet="{alphabet}" ph="{phonetics}">{text}</phoneme>"""
+
+class Alphabets:
+  IPA = "ipa"
+  CMU = "cmu-arpabet"
+
+print(ssml_template.format(alphabet=Alphabets.IPA, phonetics="ˈæktʃuəli", text="actually"))
+
+from google.colab import userdata
+eleven_labs_key = userdata.get('ELEVENLABS')
+
+from elevenlabs import save
+from elevenlabs.client import ElevenLabs
+from IPython.display import Audio, display
+
+sound_file = 'test.mp3'
+
+def build_eleven_labs_query(username: str):
+  client = ElevenLabs(
+    api_key=eleven_labs_key,
+  )
+
+  audio = client.generate(
+    text=ssml_template.format(
+        alphabet=Alphabets.CMU,
+        phonetics=predict(model, username),
+        text=username
+    ),
+    voice="Rachel",
+    model="eleven_flash_v2"
+  )
+  save(audio, sound_file)
+
+build_eleven_labs_query(test_username)
+
+
+display(Audio(sound_file, autoplay=True))
+
+# prompt: get the parameters of a pytorch model
+
+import torch
+
+# Assuming 'model' is your Seq2Seq model instance
+# Replace with your actual model if named differently
+
+# Method 1: Using model.named_parameters()
+for name, param in model.named_parameters():
+    print(f"Parameter Name: {name}, Shape: {param.shape}")
+
+# Method 2: Using model.parameters() (without parameter names)
+for param in model.parameters():
+    print(f"Parameter Shape: {param.shape}")
+
+
+print(f"Model Parameters: {sum(p.numel() for p in model.parameters())}")
+
+# prompt: visualize the weights
+
+import matplotlib.pyplot as plt
+import numpy as np
+
+# Assuming 'model' is your Seq2Seq model instance
+# Replace with your actual model if named differently
+
+# Collect parameter shapes and names
+parameter_shapes = []
+parameter_names = []
+for name, param in model.named_parameters():
+    parameter_shapes.append(np.prod(param.shape))
+    parameter_names.append(name)
+
+# Create a bar chart
+plt.figure(figsize=(10, 6))
+plt.bar(parameter_names, parameter_shapes)
+plt.xlabel("Parameter Name")
+plt.ylabel("Number of Weights")
+plt.title("Distribution of Weights in the Model")
+plt.xticks(rotation=90) # Rotate x-axis labels for better readability
+plt.tight_layout()
+plt.show()