Upload 11 files

soprano/soprano/__init__.py

@@ -0,0 +1 @@
+from .tts import SopranoTTS

soprano/soprano/backends/base.py

@@ -0,0 +1,20 @@
+class BaseModel:
+    def infer(self,
+            prompts,
+            top_p=0.95,
+            temperature=0.3,
+            repetition_penalty=1.2):
+        '''
+        Takes a list of prompts and returns the output hidden states
+        '''
+        pass
+
+    def stream_infer(self,
+            prompt,
+            top_p=0.95,
+            temperature=0.3,
+            repetition_penalty=1.2):
+        '''
+        Takes a prompt and returns an iterator of the output hidden states
+        '''
+        pass

soprano/soprano/backends/lmdeploy.py

@@ -0,0 +1,54 @@
+import torch
+from lmdeploy import pipeline, TurbomindEngineConfig, GenerationConfig
+from .base import BaseModel
+
+
+class LMDeployModel(BaseModel):
+    def __init__(self,
+            device='cuda',
+            cache_size_mb=100,
+            **kwargs):
+        assert device == 'cuda', "lmdeploy only supports cuda devices, consider changing device or using a different backend instead."
+        cache_size_ratio = cache_size_mb * 1024**2 / torch.cuda.get_device_properties('cuda').total_memory
+        backend_config = TurbomindEngineConfig(cache_max_entry_count=cache_size_ratio)
+        self.pipeline = pipeline('ekwek/Soprano-80M',
+            log_level='ERROR',
+            backend_config=backend_config)
+
+    def infer(self,
+            prompts,
+            top_p=0.95,
+            temperature=0.3,
+            repetition_penalty=1.2):
+        gen_config=GenerationConfig(output_last_hidden_state='generation',
+            do_sample=True,
+            top_p=top_p,
+            temperature=temperature,
+            repetition_penalty=repetition_penalty,
+            max_new_tokens=512)
+        responses = self.pipeline(prompts, gen_config=gen_config)
+        res = []
+        for response in responses:
+            res.append({
+                'finish_reason': response.finish_reason,
+                'hidden_state': response.last_hidden_state
+            })
+        return res
+
+    def stream_infer(self,
+            prompt,
+            top_p=0.95,
+            temperature=0.3,
+            repetition_penalty=1.2):
+        gen_config=GenerationConfig(output_last_hidden_state='generation',
+            do_sample=True,
+            top_p=top_p,
+            temperature=temperature,
+            repetition_penalty=repetition_penalty,
+            max_new_tokens=512)
+        responses = self.pipeline.stream_infer([prompt], gen_config=gen_config)
+        for response in responses:
+            yield {
+                'finish_reason': response.finish_reason,
+                'hidden_state': response.last_hidden_state
+            }

soprano/soprano/backends/transformers.py

@@ -0,0 +1,68 @@
+import torch
+from transformers import AutoModelForCausalLM, AutoTokenizer
+from .base import BaseModel
+
+
+class TransformersModel(BaseModel):
+    def __init__(self,
+            device='cuda',
+            **kwargs):
+        self.device = device
+        
+        self.model = AutoModelForCausalLM.from_pretrained(
+            'ekwek/Soprano-80M',
+            torch_dtype=torch.bfloat16 if device == 'cuda' else torch.float32,
+            device_map=device
+        )
+        self.tokenizer = AutoTokenizer.from_pretrained('ekwek/Soprano-80M')
+        self.model.eval()
+
+    def infer(self,
+            prompts,
+            top_p=0.95,
+            temperature=0.3,
+            repetition_penalty=1.2):
+        inputs = self.tokenizer(
+            prompts,
+            return_tensors='pt',
+            padding=True,
+            truncation=True,
+            max_length=512,
+        ).to(self.device)
+        
+        with torch.no_grad():
+            outputs = self.model.generate(
+                input_ids=inputs['input_ids'],
+                attention_mask=inputs['attention_mask'],
+                max_new_tokens=512,
+                do_sample=True,
+                top_p=top_p,
+                temperature=temperature,
+                repetition_penalty=repetition_penalty,
+                pad_token_id=self.tokenizer.pad_token_id,
+                return_dict_in_generate=True,
+                output_hidden_states=True,
+            )
+        res = []
+        eos_token_id = self.model.config.eos_token_id
+        for i in range(len(prompts)):
+            seq = outputs.sequences[i]
+            hidden_states = []
+            num_output_tokens = len(outputs.hidden_states)
+            for j in range(num_output_tokens):
+                token = seq[j + seq.size(0) - num_output_tokens]
+                if token != eos_token_id: hidden_states.append(outputs.hidden_states[j][-1][i, -1, :])
+            last_hidden_state = torch.stack(hidden_states).squeeze()
+            finish_reason = 'stop' if seq[-1].item() == eos_token_id else 'length'
+            res.append({
+                'finish_reason': finish_reason,
+                'hidden_state': last_hidden_state
+            })
+        return res
+
+    def stream_infer(self,
+            prompt,
+            top_p=0.95,
+            temperature=0.3,
+            repetition_penalty=1.2):
+        raise NotImplementedError("transformers backend does not currently support streaming, please consider using lmdeploy backend instead.")

soprano/soprano/tts.py

@@ -0,0 +1,188 @@
+from .vocos.decoder import SopranoDecoder
+from .utils.text import clean_text
+import torch
+import re
+from unidecode import unidecode
+from scipy.io import wavfile
+from huggingface_hub import hf_hub_download
+import os
+import time
+
+
+class SopranoTTS:
+    def __init__(self,
+            backend='auto',
+            device='cuda',
+            cache_size_mb=10,
+            decoder_batch_size=1):
+        RECOGNIZED_DEVICES = ['cuda']
+        RECOGNIZED_BACKENDS = ['auto', 'lmdeploy', 'transformers']
+        assert device in RECOGNIZED_DEVICES, f"unrecognized device {device}, device must be in {RECOGNIZED_DEVICES}"
+        if backend == 'auto':
+            if device == 'cpu':
+                backend = 'transformers'
+            else:
+                try:
+                    import lmdeploy
+                    backend = 'lmdeploy'
+                except ImportError:
+                    backend='transformers'
+            print(f"Using backend {backend}.")
+        assert backend in RECOGNIZED_BACKENDS, f"unrecognized backend {backend}, backend must be in {RECOGNIZED_BACKENDS}"
+
+        if backend == 'lmdeploy':
+            from .backends.lmdeploy import LMDeployModel
+            self.pipeline = LMDeployModel(device=device, cache_size_mb=cache_size_mb)
+        elif backend == 'transformers':
+            from .backends.transformers import TransformersModel
+            self.pipeline = TransformersModel(device=device)
+
+        self.decoder = SopranoDecoder().cuda()
+        decoder_path = hf_hub_download(repo_id='ekwek/Soprano-80M', filename='decoder.pth')
+        self.decoder.load_state_dict(torch.load(decoder_path))
+        self.decoder_batch_size=decoder_batch_size
+        self.RECEPTIVE_FIELD = 4 # Decoder receptive field
+        self.TOKEN_SIZE = 2048 # Number of samples per audio token
+
+        self.infer("Hello world!") # warmup
+
+    def _preprocess_text(self, texts, min_length=30):
+        '''
+        adds prompt format and sentence/part index
+        Enforces a minimum sentence length by merging short sentences.
+        '''
+        res = []
+        for text_idx, text in enumerate(texts):
+            text = text.strip()
+            cleaned_text = clean_text(text)
+            sentences = re.split(r"(?<=[.!?])\s+", cleaned_text)
+            processed = []
+            for sentence in sentences:
+                processed.append({
+                    "text": sentence,
+                    "text_idx": text_idx,
+                })
+
+            if min_length > 0 and len(processed) > 1:
+                merged = []
+                i = 0
+                while i < len(processed):
+                    cur = processed[i]
+                    if len(cur["text"]) < min_length:
+                        if merged: merged[-1]["text"] = (merged[-1]["text"] + " " + cur["text"]).strip()
+                        else:
+                            if i + 1 < len(processed): processed[i + 1]["text"] = (cur["text"] + " " + processed[i + 1]["text"]).strip()
+                            else: merged.append(cur)
+                    else: merged.append(cur)
+                    i += 1
+                processed = merged
+            sentence_idxes = {}
+            for item in processed:
+                if item['text_idx'] not in sentence_idxes: sentence_idxes[item['text_idx']] = 0
+                res.append((f'[STOP][TEXT]{item["text"]}[START]', item["text_idx"], sentence_idxes[item['text_idx']]))
+                sentence_idxes[item['text_idx']] += 1
+        return res
+
+    def infer(self,
+            text,
+            out_path=None,
+            top_p=0.95,
+            temperature=0.3,
+            repetition_penalty=1.2):
+        results = self.infer_batch([text],
+            top_p=top_p,
+            temperature=temperature,
+            repetition_penalty=repetition_penalty,
+            out_dir=None)[0]
+        if out_path:
+            wavfile.write(out_path, 32000, results.cpu().numpy())
+        return results
+
+    def infer_batch(self,
+            texts,
+            out_dir=None,
+            top_p=0.95,
+            temperature=0.3,
+            repetition_penalty=1.2):
+        sentence_data = self._preprocess_text(texts)
+        prompts = list(map(lambda x: x[0], sentence_data))
+        responses = self.pipeline.infer(prompts,
+            top_p=top_p,
+            temperature=temperature,
+            repetition_penalty=repetition_penalty)
+        hidden_states = []
+        for i, response in enumerate(responses):
+            if response['finish_reason'] != 'stop':
+                print(f"Warning: some sentences did not complete generation, likely due to hallucination.")
+            hidden_state = response['hidden_state']
+            hidden_states.append(hidden_state)
+        combined = list(zip(hidden_states, sentence_data))
+        combined.sort(key=lambda x: -x[0].size(0))
+        hidden_states, sentence_data = zip(*combined)
+
+        num_texts = len(texts)
+        audio_concat = [[] for _ in range(num_texts)]
+        for sentence in sentence_data:
+            audio_concat[sentence[1]].append(None)
+        for idx in range(0, len(hidden_states), self.decoder_batch_size):
+            batch_hidden_states = []
+            lengths = list(map(lambda x: x.size(0), hidden_states[idx:idx+self.decoder_batch_size]))
+            N = len(lengths)
+            for i in range(N):
+                batch_hidden_states.append(torch.cat([
+                    torch.zeros((1, 512, lengths[0]-lengths[i]), device='cuda'),
+                    hidden_states[idx+i].unsqueeze(0).transpose(1,2).cuda().to(torch.float32),
+                ], dim=2))
+            batch_hidden_states = torch.cat(batch_hidden_states)
+            with torch.no_grad():
+                audio = self.decoder(batch_hidden_states)
+            
+            for i in range(N):
+                text_id = sentence_data[idx+i][1]
+                sentence_id = sentence_data[idx+i][2]
+                audio_concat[text_id][sentence_id] = audio[i].squeeze()[-(lengths[i]*self.TOKEN_SIZE-self.TOKEN_SIZE):]
+        audio_concat = [torch.cat(x).cpu() for x in audio_concat]
+        
+        if out_dir:
+            os.makedirs(out_dir, exist_ok=True)
+            for i in range(len(audio_concat)):
+                wavfile.write(f"{out_dir}/{i}.wav", 32000, audio_concat[i].cpu().numpy())
+        return audio_concat
+
+    def infer_stream(self,
+            text,
+            chunk_size=1,
+            top_p=0.95,
+            temperature=0.3,
+            repetition_penalty=1.2):
+        start_time = time.time()
+        sentence_data = self._preprocess_text([text])
+
+        first_chunk = True
+        for sentence, _, _ in sentence_data:
+            responses = self.pipeline.stream_infer(sentence,
+                top_p=top_p,
+                temperature=temperature,
+                repetition_penalty=repetition_penalty)
+            hidden_states_buffer = []
+            chunk_counter = chunk_size
+            for token in responses:
+                finished = token['finish_reason'] is not None
+                if not finished: hidden_states_buffer.append(token['hidden_state'][-1])
+                hidden_states_buffer = hidden_states_buffer[-(2*self.RECEPTIVE_FIELD+chunk_size):]
+                if finished or len(hidden_states_buffer) >= self.RECEPTIVE_FIELD + chunk_size:
+                    if finished or chunk_counter == chunk_size:
+                        batch_hidden_states = torch.stack(hidden_states_buffer)
+                        inp = batch_hidden_states.unsqueeze(0).transpose(1, 2).cuda().to(torch.float32)
+                        with torch.no_grad():
+                            audio = self.decoder(inp)[0]
+                        if finished:
+                            audio_chunk = audio[-((self.RECEPTIVE_FIELD+chunk_counter-1)*self.TOKEN_SIZE-self.TOKEN_SIZE):]
+                        else:
+                            audio_chunk = audio[-((self.RECEPTIVE_FIELD+chunk_size)*self.TOKEN_SIZE-self.TOKEN_SIZE):-(self.RECEPTIVE_FIELD*self.TOKEN_SIZE-self.TOKEN_SIZE)]
+                        chunk_counter = 0
+                        if first_chunk:
+                            print(f"Streaming latency: {1000*(time.time()-start_time):.2f} ms")
+                            first_chunk = False
+                        yield audio_chunk.cpu()
+                    chunk_counter += 1

soprano/soprano/utils/text.py

@@ -0,0 +1,388 @@
+"""
+Normalize input text to a format that Soprano recognizes.
+Adapted from https://github.com/neonbjb/tortoise-tts/blob/main/tortoise/utils/tokenizer.py
+"""
+import os
+import re
+
+import inflect
+from unidecode import unidecode
+
+
+_inflect = inflect.engine()
+
+####################################################################################################
+# Abbreviations
+
+_abbreviations = [(re.compile('\\b%s\\.' % x[0], re.IGNORECASE), x[1]) for x in [
+    ('mrs', 'misuss'),
+    ('ms', 'miss'),
+    ('mr', 'mister'),
+    ('dr', 'doctor'),
+    ('st', 'saint'),
+    ('co', 'company'),
+    ('jr', 'junior'),
+    ('maj', 'major'),
+    ('gen', 'general'),
+    ('drs', 'doctors'),
+    ('rev', 'reverend'),
+    ('lt', 'lieutenant'),
+    ('hon', 'honorable'),
+    ('sgt', 'sergeant'),
+    ('capt', 'captain'),
+    ('esq', 'esquire'),
+    ('ltd', 'limited'),
+    ('col', 'colonel'),
+    ('ft', 'fort'),
+]]
+_cased_abbreviations = [(re.compile('\\b%s\\b' % x[0]), x[1]) for x in [
+    ('TTS', 'text to speech'),
+    ('Hz', 'hertz'),
+    ('kHz', 'kilohertz'),
+    ('KBs', 'kilobytes'),
+    ('KB', 'kilobyte'),
+    ('MBs', 'megabytes'),
+    ('MB', 'megabyte'),
+    ('GBs', 'gigabytes'),
+    ('GB', 'gigabyte'),
+    ('TBs', 'terabytes'),
+    ('TB', 'terabyte'),
+    ('APIs', 'a p i\'s'),
+    ('API', 'a p i'),
+    ('CLIs', 'c l i\'s'),
+    ('CLI', 'c l i'),
+    ('CPUs', 'c p u\'s'),
+    ('CPU', 'c p u'),
+    ('GPUs', 'g p u\'s'),
+    ('GPU', 'g p u'),
+    ('Ave', 'avenue'),
+]]
+
+def expand_abbreviations(text):
+    for regex, replacement in _abbreviations + _cased_abbreviations:
+        text = re.sub(regex, replacement, text)
+    return text
+
+####################################################################################################
+# Numbers
+
+_num_prefix_re = re.compile(r'#\d')
+_num_suffix_re = re.compile(r'\d(K|M|B|T)', re.IGNORECASE)
+_num_letter_split_re = re.compile(r'(\d[a-z]|[a-z]\d)', re.IGNORECASE)
+
+_comma_number_re = re.compile(r'(\d[\d\,]+\d)')
+_date_re = re.compile(r'(^|[^/])(\d\d?[/-]\d\d?[/-]\d\d(?:\d\d)?)($|[^/])')
+_phone_number_re = re.compile(r'(\(?\d{3}\)?[-.\s]\d{3}[-.\s]?\d{4})')
+_time_re = re.compile(r'(\d\d?:\d\d(?::\d\d)?)')
+_pounds_re = re.compile(r'£([\d\,]*\d+)')
+_dollars_re = re.compile(r'\$([\d\.\,]*\d+)')
+_decimal_number_re = re.compile(r'(\d+(?:\.\d+)+)')
+_multiply_re = re.compile(r'(\d\s?\*\s?\d)')
+_divide_re = re.compile(r'(\d\s?/\s?\d)')
+_add_re = re.compile(r'(\d\s?\+\s?\d)')
+_subtract_re = re.compile(r'(\d?\s?-\s?\d)') # also does negative numbers
+_fraction_re = re.compile(r'(\d+(?:/\d+)+)')
+_ordinal_re = re.compile(r'\d+(st|nd|rd|th)')
+_number_re = re.compile(r'\d+')
+
+def _expand_num_prefix(m):
+    match = m.group(0)
+    return f"number {match[1]}"
+
+def _expand_num_suffix(m):
+    match = m.group(0)
+    if match[1].upper() == 'K': return f"{match[0]} thousand"
+    elif match[1].upper() == 'M': return f"{match[0]} million"
+    elif match[1].upper() == 'B': return f"{match[0]} billion"
+    elif match[1].upper() == 'T': return f"{match[0]} trillion"
+    return match # unexpected format
+
+def _split_alphanumeric(m):
+    match = m.group(1)
+    return f"{match[0]} {match[1]}"
+
+def _remove_commas(m):
+    return m.group(1).replace(',', '')
+
+def _expand_date(m):
+    match = m.group(2)
+    match = re.split('[./-]', match)
+    return m.group(1) + ' dash '.join(match) + m.group(3)
+    
+def _expand_phone_number(m):
+    match = m.group(1)
+    match = re.sub(r'\D', '', match)
+    assert len(match) == 10
+    match = f"{' '.join(list(match[:3]))}, {' '.join(list(match[3:6]))}, {' '.join(list(match[6:]))}"
+    return match
+    
+def _expand_time(m):
+    match = m.group(1)
+    match = match.split(':')
+    if len(match) == 2:
+        hours, minutes = match
+        if minutes == '00':
+            if int(hours) == 0:
+                return '0'
+            elif int(hours) > 12: return f"{hours} minutes"
+            return f"{hours} o'clock"
+        elif minutes.startswith('0'):
+            minutes = f'oh {minutes[1:]}'
+        return f"{hours} {minutes}"
+    else:
+        hours, minutes, seconds = match
+        if int(hours) != 0:
+            return f"{hours} {'oh oh' if minutes == '00' else f'oh {minutes}' if minutes.startswith('0') else {minutes}} {'' if seconds == '00' else f'oh {seconds}' if seconds.startswith('0') else seconds}"
+        elif minutes != '00':
+            return f"{minutes} {'oh oh' if seconds == '00' else f'oh {seconds}' if seconds.startswith('0') else seconds}"
+        else:
+            return seconds
+
+def _expand_dollars(m):
+    match = m.group(1)
+    parts = match.split('.')
+    if len(parts) > 2:
+        return match + ' dollars'  # Unexpected format
+    dollars = int(parts[0]) if parts[0] else 0
+    cents = int(parts[1]) if len(parts) > 1 and parts[1] else 0
+    if dollars and cents:
+        dollar_unit = 'dollar' if dollars == 1 else 'dollars'
+        cent_unit = 'cent' if cents == 1 else 'cents'
+        return '%s %s, %s %s' % (dollars, dollar_unit, cents, cent_unit)
+    elif dollars:
+        dollar_unit = 'dollar' if dollars == 1 else 'dollars'
+        return '%s %s' % (dollars, dollar_unit)
+    elif cents:
+        cent_unit = 'cent' if cents == 1 else 'cents'
+        return '%s %s' % (cents, cent_unit)
+    else:
+        return 'zero dollars'
+
+def _expand_decimal_point(m):
+    match = m.group(1)
+    match = match.split('.')
+    return match[0] + ' point ' + ' point '.join(' '.join(list(match[i])) for i in range(1, len(match)))
+
+def _expand_fraction(m):
+    match = m.group(1)
+    match = match.split('/')
+    return ' over '.join(match) if len(match)==2 else ' slash '.join(match)
+    
+def _expand_multiply(m):
+    return ' times '.join(m.group(1).split('*'))
+    
+def _expand_divide(m):
+    return ' over '.join(m.group(1).split('/'))
+    
+def _expand_add(m):
+    return ' plus '.join(m.group(1).split('+'))
+    
+def _expand_subtract(m):
+    return ' minus '.join(m.group(1).split('-'))
+    
+def _expand_ordinal(m):
+    return _inflect.number_to_words(m.group(0), andword='')
+
+def _expand_number(m):
+    num = int(m.group(0))
+    if num > 1000 and num < 3000:
+        if num == 2000:
+            return 'two thousand'
+        elif num > 2000 and num < 2010:
+            return 'two thousand ' + _inflect.number_to_words(num % 100)
+        elif num % 100 == 0:
+            return _inflect.number_to_words(num // 100) + ' hundred'
+        else:
+            return _inflect.number_to_words(num, andword='', zero='oh', group=2).replace(', ', ' ')
+    else:
+        return _inflect.number_to_words(num, andword='')
+
+def normalize_numbers(text):
+    text = re.sub(_num_prefix_re, _expand_num_prefix, text)
+    text = re.sub(_num_suffix_re, _expand_num_suffix, text)
+    for _ in range(2): # need to do this twice to find all matches
+        text = re.sub(_num_letter_split_re, _split_alphanumeric, text)
+    text = re.sub(_comma_number_re, _remove_commas, text)
+    text = re.sub(_date_re, _expand_date, text)
+    text = re.sub(_phone_number_re, _expand_phone_number, text)
+    text = re.sub(_time_re, _expand_time, text)
+    text = re.sub(_pounds_re, r'\1 pounds', text)
+    text = re.sub(_dollars_re, _expand_dollars, text)
+    text = re.sub(_decimal_number_re, _expand_decimal_point, text)
+    text = re.sub(_multiply_re, _expand_multiply, text)
+    text = re.sub(_divide_re, _expand_divide, text)
+    text = re.sub(_add_re, _expand_add, text)
+    text = re.sub(_subtract_re, _expand_subtract, text)
+
+    text = re.sub(_fraction_re, _expand_fraction, text)
+    text = re.sub(_ordinal_re, _expand_ordinal, text)
+    text = re.sub(_number_re, _expand_number, text)
+    return text
+
+####################################################################################################
+# Special characters & other patterns
+
+_special_characters = [(re.compile(x[0]), x[1]) for x in [
+    ('@', ' at '),
+    ('&', ' and '),
+    ('%', ' percent '),
+    (':', '.'),
+    (';', ','),
+    (r'\+', ' plus '),
+    (r'\\', ' backslash '),
+    ('~', ' about '),
+    ('(^| )<3', ' heart '),
+    ('<=', ' less than or equal to '),
+    ('>=', ' greater than or equal to '),
+    ('<', ' less than '),
+    ('>', ' greater than '),
+    ('=', ' equals '),
+    ('/', ' slash '),
+    ('_', ' '),
+]]
+_link_header_re = re.compile(r'(https?://)')
+_dash_re = re.compile(r'(. - .)')
+_dot_re = re.compile(r'([A-Z]\.[A-Z])', re.IGNORECASE)
+_parentheses_re = re.compile(r'[\(\[\{].*[\)\]\}](.|$)')
+
+def expand_special_characters(text):
+    for regex, replacement in _special_characters:
+        text = re.sub(regex, replacement, text)
+    return text
+
+def _expand_link_header(m):
+    return 'h t t p s colon slash slash '
+
+def _expand_dash(m):
+    match = m.group(0)
+    return f"{match[0]}, {match[4]}"
+
+def _expand_dot(m):
+    match = m.group(0)
+    return f"{match[0]} dot {match[2]}"
+
+def _expand_parantheses(m):
+    match = m.group(0)
+    match = re.sub(r'[\(\[\{]', ', ', match)
+    match = re.sub(r'[\)\]\}][^$.!?,]', ', ', match)
+    match = re.sub(r'[\)\]\}]', '', match)
+    return match
+
+def normalize_special(text):
+    text = re.sub(_link_header_re, _expand_link_header, text)
+    text = re.sub(_dash_re, _expand_dash, text)
+    text = re.sub(_dot_re, _expand_dot, text)
+    text = re.sub(_parentheses_re, _expand_parantheses, text)
+    return text
+
+####################################################################################################
+# Misc
+
+def lowercase(text):
+    return text.lower()
+
+def convert_to_ascii(text):
+    return unidecode(text)
+
+def normalize_newlines(text):
+    text = text.split('\n')
+    for i in range(len(text)):
+        if not text[i]: continue
+        text[i] = text[i].strip()
+        if text[i][-1] not in '.!?':
+            text[i] = f"{text[i]}."
+    return ' '.join(text)
+
+def remove_unknown_characters(text):
+    text = re.sub(r"[^A-Za-z !\$%&'\*\+,-./0123456789<>\?_]", "", text)
+    text = re.sub(r"[<>/_+]", "", text)
+    return text
+
+def collapse_whitespace(text):
+    text = re.sub(r'\s+', ' ', text)
+    text = re.sub(r' [.\?!,]', lambda m: m.group(0)[1], text)
+    return text
+
+def dedup_punctuation(text):
+    text = re.sub(r"\.\.\.+", "[ELLIPSIS]", text)
+    text = re.sub(r",+", ",", text)
+    text = re.sub(r"[\.,]*\.[\.,]*", ".", text)
+    text = re.sub(r"[\.,!]*![\.,!]*", "!", text)
+    text = re.sub(r"[\.,!\?]*\?[\.,!\?]*", "?", text)
+    text = re.sub("[ELLIPSIS]", "...", text)
+    return text
+
+def clean_text(text):
+    text = convert_to_ascii(text)
+    text = normalize_newlines(text)
+    text = normalize_numbers(text)
+    text = normalize_special(text)
+    text = expand_abbreviations(text)
+    text = expand_special_characters(text)
+    text = lowercase(text)
+    text = remove_unknown_characters(text)
+    text = collapse_whitespace(text)
+    text = dedup_punctuation(text)
+    return text
+
+
+if __name__ == '__main__':
+    print(normalize_numbers('1,2,3,456,176'))
+    print(normalize_numbers('123,456,789'))
+    print(normalize_numbers('123,456,789th'))
+    print(normalize_numbers('123-456-7890'))
+    print(normalize_numbers('111-111-1111'))
+    print(normalize_numbers('(111) 111-1111'))
+    print(normalize_numbers('A(111) 111-1111'))
+    print(normalize_numbers('A (111) 111-1111'))
+    print(normalize_numbers('$2.47'))
+    print(normalize_numbers('$247'))
+    print(normalize_numbers('$0.27'))
+    print(normalize_numbers('$1.00'))
+    print(normalize_numbers('£20'))
+    for i in range(1990, 2030):
+        print(normalize_numbers(str(i)))
+    print(normalize_numbers('2656'))
+    print(normalize_numbers('1024'))
+    print(normalize_numbers('2.47023'))
+    print(normalize_numbers('20.47023'))
+    print(normalize_numbers('1.17.1.1'))
+    print(normalize_numbers('111.111.1111'))
+    print(normalize_numbers('1/1/2025'))
+    print(normalize_numbers('1-1-2025'))
+    print(normalize_numbers('1-1-25'))
+    print(normalize_numbers('A 1/1/11 A'))
+    print(normalize_numbers('A 1/1 A'))
+    print(normalize_numbers('1/1'))
+    print(normalize_numbers('1/10'))
+    print(normalize_numbers('1/1/10'))
+    print(normalize_numbers('11/1/1/10'))
+
+    print(normalize_numbers('0:00'))
+    print(normalize_numbers('12:00'))
+    print(normalize_numbers('13:00'))
+    print(normalize_numbers('8:00'))
+    print(normalize_numbers('8:05'))
+    print(normalize_numbers('8:15'))
+    print(normalize_numbers('0:00:00'))
+    print(normalize_numbers('00:01:10'))
+    print(normalize_numbers('00:10:01'))
+    print(normalize_numbers('01:01:01'))
+    print(normalize_numbers('00:01:00'))
+    print(normalize_numbers('01:00:00'))
+
+    print(normalize_numbers('-1 + 2 * 3 - 4 / 5'))
+    print(normalize_numbers('-1+2*3-5/4/25'))
+
+    print(normalize_numbers('100x1'))
+    print(normalize_numbers('100k'))
+    print(normalize_numbers('100m'))
+    print(normalize_numbers('100b'))
+    print(normalize_numbers('100t'))
+    
+    print(normalize_numbers('#1'))
+    
+    print(normalize_numbers('12:00'))
+    print(normalize_numbers('11:59'))
+    print(normalize_numbers('01:00'))
+    print(normalize_numbers('0100'))

soprano/soprano/vocos/decoder.py

@@ -0,0 +1,45 @@
+import torch
+from torch import nn
+
+from .models import VocosBackbone
+from .heads import ISTFTHead
+
+
+class SopranoDecoder(nn.Module):
+    def __init__(self,
+                 num_input_channels=512,
+                 decoder_num_layers=8,
+                 decoder_dim=512,
+                 decoder_intermediate_dim=None,
+                 hop_length=512,
+                 n_fft=2048,
+                 upscale=4,
+                 dw_kernel=3,
+                ):
+        super().__init__()
+        self.decoder_initial_channels = num_input_channels
+        self.num_layers = decoder_num_layers
+        self.dim = decoder_dim
+        self.intermediate_dim = decoder_intermediate_dim if decoder_intermediate_dim else decoder_dim*3
+        self.hop_length = hop_length
+        self.n_fft = n_fft
+        self.upscale = upscale
+        self.dw_kernel = dw_kernel
+        
+        self.decoder = VocosBackbone(input_channels=self.decoder_initial_channels,
+                                      dim=self.dim,
+                                      intermediate_dim=self.intermediate_dim,
+                                      num_layers=self.num_layers,
+                                      input_kernel_size=dw_kernel,
+                                      dw_kernel_size=dw_kernel,
+                                      )
+        self.head = ISTFTHead(dim=self.dim,
+                            n_fft=self.n_fft,
+                            hop_length=self.hop_length)
+
+    def forward(self, x):
+        T = x.size(2)
+        x = torch.nn.functional.interpolate(x, size=self.upscale*(T-1)+1, mode='linear', align_corners=True)
+        x = self.decoder(x)
+        reconstructed = self.head(x)
+        return reconstructed

soprano/soprano/vocos/heads.py

@@ -0,0 +1,50 @@
+import torch
+from torch import nn
+from .spectral_ops import ISTFT
+
+
+class ISTFTHead(nn.Module):
+    """
+    ISTFT Head module for predicting STFT complex coefficients.
+
+    Args:
+        dim (int): Hidden dimension of the model.
+        n_fft (int): Size of Fourier transform.
+        hop_length (int): The distance between neighboring sliding window frames, which should align with
+                          the resolution of the input features.
+        padding (str, optional): Type of padding. Options are "center" or "same". Defaults to "same".
+    """
+
+    def __init__(self, dim: int, n_fft: int, hop_length: int, padding: str = "center"):
+        super().__init__()
+        out_dim = n_fft + 2
+        self.out = torch.nn.Linear(dim, out_dim)
+        self.istft = ISTFT(n_fft=n_fft, hop_length=hop_length, win_length=n_fft, padding=padding)
+
+    @torch.compiler.disable
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        """
+        Forward pass of the ISTFTHead module.
+
+        Args:
+            x (Tensor): Input tensor of shape (B, L, H), where B is the batch size,
+                        L is the sequence length, and H denotes the model dimension.
+
+        Returns:
+            Tensor: Reconstructed time-domain audio signal of shape (B, T), where T is the length of the output signal.
+        """
+        x = self.out(x.transpose(1,2)).transpose(1, 2)
+        mag, p = x.chunk(2, dim=1)
+        mag = torch.exp(mag)
+        mag = torch.clip(mag, max=1e2)  # safeguard to prevent excessively large magnitudes
+        # wrapping happens here. These two lines produce real and imaginary value
+        x = torch.cos(p)
+        y = torch.sin(p)
+        # recalculating phase here does not produce anything new
+        # only costs time
+        # phase = torch.atan2(y, x)
+        # S = mag * torch.exp(phase * 1j)
+        # better directly produce the complex value 
+        S = mag * (x + 1j * y)
+        audio = self.istft(S)
+        return audio

soprano/soprano/vocos/models.py

@@ -0,0 +1,61 @@
+from typing import Optional
+
+import torch
+from torch import nn
+
+from .modules import ConvNeXtBlock
+
+class VocosBackbone(nn.Module):
+    """
+    Vocos backbone module built with ConvNeXt blocks. Supports additional conditioning with Adaptive Layer Normalization
+
+    Args:
+        input_channels (int): Number of input features channels.
+        dim (int): Hidden dimension of the model.
+        intermediate_dim (int): Intermediate dimension used in ConvNeXtBlock.
+        num_layers (int): Number of ConvNeXtBlock layers.
+        layer_scale_init_value (float, optional): Initial value for layer scaling. Defaults to `1 / num_layers`.
+    """
+
+    def __init__(
+        self,
+        input_channels: int,
+        dim: int,
+        intermediate_dim: int,
+        num_layers: int,
+        input_kernel_size: int = 9,
+        dw_kernel_size: int = 9,
+        layer_scale_init_value: Optional[float] = None,
+        pad: str = 'zeros',
+    ):
+        super().__init__()
+        self.embed = nn.Conv1d(input_channels, dim, kernel_size=input_kernel_size, padding=input_kernel_size//2, padding_mode=pad)
+        self.norm = nn.LayerNorm(dim, eps=1e-6)
+        self.convnext = nn.ModuleList(
+            [
+                ConvNeXtBlock(
+                    dim=dim,
+                    intermediate_dim=intermediate_dim,
+                    dw_kernel_size=dw_kernel_size,
+                    layer_scale_init_value=layer_scale_init_value or 1 / num_layers**0.5,
+                )
+                for _ in range(num_layers)
+            ]
+        )
+        self.final_layer_norm = nn.LayerNorm(dim, eps=1e-6)
+        self.apply(self._init_weights)
+
+    def _init_weights(self, m):
+        if isinstance(m, (nn.Conv1d, nn.Linear)):
+            nn.init.trunc_normal_(m.weight, std=0.02)
+            if m.bias is not None: nn.init.constant_(m.bias, 0)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        x = self.embed(x) # (B, C, L)
+        x = self.norm(x.transpose(1, 2))
+        x = x.transpose(1, 2)
+        for conv_block in self.convnext:
+            x = conv_block(x)
+        x = self.final_layer_norm(x.transpose(1, 2))
+        x = x.transpose(1, 2)
+        return x

soprano/soprano/vocos/modules.py

@@ -0,0 +1,47 @@
+import torch
+from torch import nn
+
+
+class ConvNeXtBlock(nn.Module):
+    """ConvNeXt Block adapted from https://github.com/facebookresearch/ConvNeXt to 1D audio signal.
+
+    Args:
+        dim (int): Number of input channels.
+        intermediate_dim (int): Dimensionality of the intermediate layer.
+        layer_scale_init_value (float, optional): Initial value for the layer scale. None means no scaling.
+            Defaults to None.
+    """
+
+    def __init__(
+        self,
+        dim: int,
+        intermediate_dim: int,
+        layer_scale_init_value: float,
+        dw_kernel_size: int = 9,
+    ):
+        super().__init__()
+        self.dwconv = nn.Conv1d(dim, dim, kernel_size=dw_kernel_size, padding=dw_kernel_size//2, groups=dim)  # depthwise conv
+        self.norm = nn.LayerNorm(dim, eps=1e-6)
+        self.pwconv1 = nn.Linear(dim, intermediate_dim)  # pointwise/1x1 convs, implemented with linear layers
+        self.act = nn.GELU()
+        self.pwconv2 = nn.Linear(intermediate_dim, dim)
+        self.gamma = (
+            nn.Parameter(layer_scale_init_value * torch.ones(dim), requires_grad=True)
+            if layer_scale_init_value > 0
+            else None
+        )
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        residual = x
+        x = self.dwconv(x)
+        x = x.transpose(1, 2)  # (B, C, T) -> (B, T, C)
+        x = self.norm(x)
+        x = self.pwconv1(x)
+        x = self.act(x)
+        x = self.pwconv2(x)
+        if self.gamma is not None:
+            x = self.gamma * x
+        x = x.transpose(1, 2)  # (B, T, C) -> (B, C, T)
+
+        x = residual + x
+        return x

soprano/soprano/vocos/spectral_ops.py

@@ -0,0 +1,74 @@
+import torch
+from torch import nn
+
+class ISTFT(nn.Module):
+    """
+    Custom implementation of ISTFT since torch.istft doesn't allow custom padding (other than `center=True`) with
+    windowing. This is because the NOLA (Nonzero Overlap Add) check fails at the edges.
+    See issue: https://github.com/pytorch/pytorch/issues/62323
+    Specifically, in the context of neural vocoding we are interested in "same" padding analogous to CNNs.
+    The NOLA constraint is met as we trim padded samples anyway.
+
+    Args:
+        n_fft (int): Size of Fourier transform.
+        hop_length (int): The distance between neighboring sliding window frames.
+        win_length (int): The size of window frame and STFT filter.
+        padding (str, optional): Type of padding. Options are "center" or "same". Defaults to "same".
+    """
+
+    def __init__(self, n_fft: int, hop_length: int, win_length: int, padding: str = "same"):
+        super().__init__()
+        if padding not in ["center", "same"]:
+            raise ValueError("Padding must be 'center' or 'same'.")
+        self.padding = padding
+        self.n_fft = n_fft
+        self.hop_length = hop_length
+        self.win_length = win_length
+        window = torch.hann_window(win_length).to('cuda')
+        self.register_buffer("window", window)
+
+    def forward(self, spec: torch.Tensor) -> torch.Tensor:
+        """
+        Compute the Inverse Short Time Fourier Transform (ISTFT) of a complex spectrogram.
+
+        Args:
+            spec (Tensor): Input complex spectrogram of shape (B, N, T), where B is the batch size,
+                            N is the number of frequency bins, and T is the number of time frames.
+
+        Returns:
+            Tensor: Reconstructed time-domain signal of shape (B, L), where L is the length of the output signal.
+        """
+        if self.padding == "center":
+            spec[:,0] = 0 # fixes some strange bug where first/last freqs don't matter when bs<16 which causes exploding gradients
+            spec[:,-1] = 0
+            # Fallback to pytorch native implementation
+            return torch.istft(spec, self.n_fft, self.hop_length, self.win_length, self.window, center=True)
+        elif self.padding == "same":
+            pad = (self.win_length - self.hop_length) // 2
+        else:
+            raise ValueError("Padding must be 'center' or 'same'.")
+
+        assert spec.dim() == 3, "Expected a 3D tensor as input"
+        B, N, T = spec.shape
+
+        # Inverse FFT
+        ifft = torch.fft.irfft(spec, self.n_fft, dim=1, norm="backward")
+        ifft = ifft * self.window[None, :, None]
+
+        # Overlap and Add
+        output_size = (T - 1) * self.hop_length + self.win_length
+        y = torch.nn.functional.fold(
+            ifft, output_size=(1, output_size), kernel_size=(1, self.win_length), stride=(1, self.hop_length),
+        )[:, 0, 0, pad:-pad]
+
+        # Window envelope
+        window_sq = self.window.square().expand(1, T, -1).transpose(1, 2)
+        window_envelope = torch.nn.functional.fold(
+            window_sq, output_size=(1, output_size), kernel_size=(1, self.win_length), stride=(1, self.hop_length),
+        ).squeeze()[pad:-pad]
+
+        # Normalize
+        assert (window_envelope > 1e-11).all()
+        y = y / window_envelope
+
+        return y