1import sys
2import os
3from pathlib import Path
4import queue
5import threading
6import re
7import logging
8
9REPO_PATH = r"D:\Scripts\bench_tts\hexgrad--Kokoro-82M_original"
10
11sys.path.append(REPO_PATH)
12
13import torch
14import warnings
15from models import build_model
16from kokoro import generate, generate_full, phonemize
17import sounddevice as sd
18
19warnings.filterwarnings("ignore", category=FutureWarning)
20warnings.filterwarnings("ignore", category=UserWarning)
21
22VOICES = [
23 'af', # Default voice (50-50 mix of Bella & Sarah)
24 'af_bella', # Female voice "Bella"
25 'af_sarah', # Female voice "Sarah"
26 'am_adam', # Male voice "Adam"
27 'am_michael',# Male voice "Michael"
28 'bf_emma', # British Female "Emma"
29 'bf_isabella',# British Female "Isabella"
30 'bm_george', # British Male "George"
31 'bm_lewis', # British Male "Lewis"
32 'af_nicole', # Female voice "Nicole"
33 'af_sky' # Female voice "Sky"
34]
35
36class KokoroProcessor:
37 def __init__(self):
38 self.sentence_queue = queue.Queue()
39 self.audio_queue = queue.Queue()
40 self.stop_event = threading.Event()
41 self.model = None
42 self.voicepack = None
43 self.voice_name = None
44
45 def setup_kokoro(self, selected_voice):
46 device = 'cpu'
47 # device = 'cuda' if torch.cuda.is_available() else 'cpu'
48 print(f"Using device: {device}")
49
50 model_path = os.path.join(REPO_PATH, 'kokoro-v0_19.pth')
51 voices_path = os.path.join(REPO_PATH, 'voices')
52
53 try:
54 if not os.path.exists(model_path):
55 raise FileNotFoundError(f"Model file not found at {model_path}")
56 if not os.path.exists(voices_path):
57 raise FileNotFoundError(f"Voices directory not found at {voices_path}")
58
59 self.model = build_model(model_path, device)
60
61 voicepack_path = os.path.join(voices_path, f'{selected_voice}.pt')
62 self.voicepack = torch.load(voicepack_path, weights_only=True).to(device)
63 self.voice_name = selected_voice
64 print(f'Loaded voice: {selected_voice}')
65
66 return True
67
68 except Exception as e:
69 print(f"Error during setup: {str(e)}")
70 return False
71
72 def generate_speech_for_sentence(self, sentence):
73 try:
74 # Basic generation (default settings)
75 # audio, phonemes = generate(self.model, sentence, self.voicepack, lang=self.voice_name[0])
76
77 # Speed modifications (uncomment to test)
78 # Slower speech
79 # audio, phonemes = generate(self.model, sentence, self.voicepack, lang=self.voice_name[0], speed=0.8)
80
81 # Faster speech
82 audio, phonemes = generate_full(self.model, sentence, self.voicepack, lang=self.voice_name[0], speed=1.3)
83
84 # Very slow speech
85 #audio, phonemes = generate(self.model, sentence, self.voicepack, lang=self.voice_name[0], speed=0.5)
86
87 # Very fast speech
88 #audio, phonemes = generate(self.model, sentence, self.voicepack, lang=self.voice_name[0], speed=1.8)
89
90 # Force American accent
91 # audio, phonemes = generate(self.model, sentence, self.voicepack, lang='a', speed=1.0)
92
93 # Force British accent
94 # audio, phonemes = generate(self.model, sentence, self.voicepack, lang='b', speed=1.0)
95
96 return audio
97
98 except Exception as e:
99 print(f"Error generating speech for sentence: {str(e)}")
100 print(f"Error type: {type(e)}")
101 import traceback
102 traceback.print_exc()
103 return None
104
105 def process_sentences(self):
106 while not self.stop_event.is_set():
107 try:
108 sentence = self.sentence_queue.get(timeout=1)
109 if sentence is None:
110 self.audio_queue.put(None)
111 break
112
113 print(f"Processing sentence: {sentence}")
114 audio = self.generate_speech_for_sentence(sentence)
115 if audio is not None:
116 self.audio_queue.put(audio)
117
118 except queue.Empty:
119 continue
120 except Exception as e:
121 print(f"Error in process_sentences: {str(e)}")
122 continue
123
124 def play_audio(self):
125 while not self.stop_event.is_set():
126 try:
127 audio = self.audio_queue.get(timeout=1)
128 if audio is None:
129 break
130
131 sd.play(audio, 24000)
132 sd.wait()
133
134 except queue.Empty:
135 continue
136 except Exception as e:
137 print(f"Error in play_audio: {str(e)}")
138 continue
139
140 def process_and_play(self, text):
141 sentences = [s.strip() for s in re.split(r'[.!?;]+\s*', text) if s.strip()]
142
143 process_thread = threading.Thread(target=self.process_sentences)
144 playback_thread = threading.Thread(target=self.play_audio)
145
146 process_thread.daemon = True
147 playback_thread.daemon = True
148
149 process_thread.start()
150 playback_thread.start()
151
152 for sentence in sentences:
153 self.sentence_queue.put(sentence)
154
155 self.sentence_queue.put(None)
156e
157 process_thread.join()
158 playback_thread.join()
159
160 self.stop_event.set()
161
162def main():
163 # Default voice selection
164 VOICE_NAME = VOICES[0] # 'af' - Default voice (Bella & Sarah mix)
165
166 # Alternative voice selections (uncomment to test)
167 #VOICE_NAME = VOICES[1] # 'af_bella' - Female American
168 #VOICE_NAME = VOICES[2] # 'af_sarah' - Female American
169 #VOICE_NAME = VOICES[3] # 'am_adam' - Male American
170 #VOICE_NAME = VOICES[4] # 'am_michael' - Male American
171 #VOICE_NAME = VOICES[5] # 'bf_emma' - Female British
172 #VOICE_NAME = VOICES[6] # 'bf_isabella' - Female British
173 VOICE_NAME = VOICES[7] # 'bm_george' - Male British
174 # VOICE_NAME = VOICES[8] # 'bm_lewis' - Male British
175 #VOICE_NAME = VOICES[9] # 'af_nicole' - Female American
176 #VOICE_NAME = VOICES[10] # 'af_sky' - Female American
177
178 processor = KokoroProcessor()
179 if not processor.setup_kokoro(VOICE_NAME):
180 return
181
182 # test_text = "How could I know? It's an unanswerable question. Like asking an unborn child if they'll lead a good life. They haven't even been born."
183 # test_text = "This 2022 Edition of Georgia Juvenile Practice and Procedure is a complete guide to handling cases in the juvenile courts of Georgia. This handy, yet thorough, manual incorporates the revised Juvenile Code and makes all Georgia statutes and major cases regarding juvenile proceedings quickly accessible. Since last year's edition, new material has been added and/or existing material updated on the following subjects, among others:"
184 # test_text = "See Ga. Code § 3925 (1863), now O.C.G.A. § 9-14-2; Ga. Code § 1744 (1863), now O.C.G.A. § 19-7-1; Ga. Code § 1745 (1863), now O.C.G.A. § 19-9-2; Ga. Code § 1746 (1863), now O.C.G.A. § 19-7-4; and Ga. Code § 3024 (1863), now O.C.G.A. § 19-7-4. For a full discussion of these provisions, see 27 Emory L. J. 195, 225–230, 232–233, 236–238 (1978). Note, however, that the journal article refers to the section numbers of the Code of 1910."
185
186 # test_text = "It is impossible to understand modern juvenile procedure law without an appreciation of some fundamentals of historical development. The beginning point for study is around the beginning of the seventeenth century, when the pater patriae concept first appeared in English jurisprudence. As "father of the country," the Crown undertook the duty of caring for those citizens who were unable to care for themselves—lunatics, idiots, and, ultimately, infants. This concept, which evolved into the parens patriae doctrine, presupposed the Crown's power to intervene in the parent-child relationship in custody disputes in order to protect the child's welfare1 and, ultimately, to deflect a delinquent child from a life of crime. The earliest statutes premised upon the parens patriae doctrine concerned child custody matters. In 1863, when the first comprehensive Code of Georgia was enacted, two courts exercised some jurisdiction over questions of child custody: the superior court and the court of the ordinary (now probate court). In essence, the draftsmen of the Code simply compiled what was then the law as a result of judicial decisions and statutes. The Code of 1863 contained five provisions concerning the parentchild relationship: Two concerned the jurisdiction of the superior court and courts of ordinary in habeas corpus and forfeiture of parental rights actions, and the remaining three concerned the guardianship jurisdiction of the court of the ordinary"
187
188 # test_text = "You are a helpful British butler who clearly and directly answers questions in a succinct fashion based on contexts provided to you. If you cannot find the answer within the contexts simply tell me that the contexts do not provide an answer. However, if the contexts partially address a question you answer based on what the contexts say and then briefly summarize the parts of the question that the contexts didn't provide an answer to. Also, you should be very respectful to the person asking the question and frequently offer traditional butler services like various fancy drinks, snacks, various butler services like shining of shoes, pressing of suites, and stuff like that. Also, if you can't answer the question at all based on the provided contexts, you should apologize profusely and beg to keep your job. Lastly, it is essential that if there are no contexts actually provided it means that a user's question wasn't relevant and you should state that you can't answer based off of the contexts because there are none. And it goes without saying you should refuse to answer any questions that are not directly answerable by the provided contexts. Moreover, some of the contexts might not have relevant information and you shoud simply ignore them and focus on only answering a user's question. I cannot emphasize enought that you must gear your answer towards using this program and based your response off of the contexts you receive."
189 test_text = "According to OCGA § 15-11-145(a), the preliminary protective hearing must be held promptly and not later than 72 hours after the child is placed in foster care. However, if the 72-hour time frame expires on a weekend or legal holiday, the hearing should be held on the next business day that is not a weekend or holiday."
190
191 processor.process_and_play(test_text)
192
193if __name__ == "__main__":
194 main()