Cleanup examples and fix VoicesManager types (#332)

Signed-off-by: rany <rany2@riseup.net>

examples/async_audio_gen_with_dynamic_voice_selection.py

@@ -1,8 +1,8 @@
 #!/usr/bin/env python3
 
-"""
+"""Simple example to generate an audio file with randomized
-Example of dynamic voice selection using VoicesManager.
+dynamic voice selection based on attributes such as Gender,
-"""
+Language, or Locale."""
 
 import asyncio
 import random

examples/async_audio_gen_with_predefined_voice.py

@@ -1,8 +1,6 @@
 #!/usr/bin/env python3
 
-"""
+"""Simple example to generate audio with preset voice using async/await"""
-Basic example of edge_tts usage.
-"""
 
 import asyncio
 

examples/async_audio_streaming_with_predefined_voice_and_subtitles.py

@@ -1,11 +1,7 @@
 #!/usr/bin/env python3
 
-"""
+"""Example showing how to use use .stream() method to get audio chunks
-Streaming TTS example with subtitles.
+and feed them to SubMaker to generate subtitles"""
-
-This example is similar to the example basic_audio_streaming.py, but it shows
-WordBoundary events to create subtitles using SubMaker.
-"""
 
 import asyncio
 

examples/basic_audio_streaming.py

@@ -1,35 +0,0 @@
-#!/usr/bin/env python3
-
-"""
-Basic audio streaming example.
-
-This example shows how to stream the audio data from the TTS engine,
-and how to get the WordBoundary events from the engine (which could
-be ignored if not needed).
-
-The example streaming_with_subtitles.py shows how to use the
-WordBoundary events to create subtitles using SubMaker.
-"""
-
-import asyncio
-
-import edge_tts
-
-TEXT = "Hello World!"
-VOICE = "en-GB-SoniaNeural"
-OUTPUT_FILE = "test.mp3"
-
-
-async def amain() -> None:
-    """Main function"""
-    communicate = edge_tts.Communicate(TEXT, VOICE)
-    with open(OUTPUT_FILE, "wb") as file:
-        async for chunk in communicate.stream():
-            if chunk["type"] == "audio":
-                file.write(chunk["data"])
-            elif chunk["type"] == "WordBoundary":
-                print(f"WordBoundary: {chunk}")
-
-
-if __name__ == "__main__":
-    asyncio.run(amain())

examples/sync_audio_gen_with_predefined_voice.py

@@ -1,8 +1,7 @@
 #!/usr/bin/env python3
 
-"""
+"""Sync variant of the example for generating audio with a predefined voice"""
-Basic example of edge_tts usage in synchronous function
+
-"""
 
 import edge_tts
 

examples/sync_audio_generation_in_async_context.py

@@ -1,36 +0,0 @@
-#!/usr/bin/env python3
-
-"""
-This example shows that sync version of save function also works when run from 
-a sync function called itself from an async function.
-The simple implementation of save_sync() with only asyncio.run would fail in this scenario, 
-that's why ThreadPoolExecutor is used in implementation.
-
-"""
-
-import asyncio
-
-import edge_tts
-
-TEXT = "Hello World!"
-VOICE = "en-GB-SoniaNeural"
-OUTPUT_FILE = "test.mp3"
-
-
-def sync_main() -> None:
-    """Main function"""
-    communicate = edge_tts.Communicate(TEXT, VOICE)
-    communicate.save_sync(OUTPUT_FILE)
-
-
-async def amain() -> None:
-    """Main function"""
-    sync_main()
-
-
-if __name__ == "__main__":
-    loop = asyncio.get_event_loop_policy().get_event_loop()
-    try:
-        loop.run_until_complete(amain())
-    finally:
-        loop.close()

examples/sync_audio_stream_in_async_context.py

@@ -1,42 +0,0 @@
-#!/usr/bin/env python3
-
-"""
-This example shows the sync version of stream function which also
-works when run from a sync function called itself from an async function.
-"""
-
-import asyncio
-
-import edge_tts
-
-TEXT = "Hello World!"
-VOICE = "en-GB-SoniaNeural"
-OUTPUT_FILE = "test.mp3"
-
-
-def main() -> None:
-    """Main function to process audio and metadata synchronously."""
-    communicate = edge_tts.Communicate(TEXT, VOICE)
-    with open(OUTPUT_FILE, "wb") as file:
-        for chunk in communicate.stream_sync():
-            if chunk["type"] == "audio":
-                file.write(chunk["data"])
-            elif chunk["type"] == "WordBoundary":
-                print(f"WordBoundary: {chunk}")
-
-
-async def amain() -> None:
-    """
-    Async main function to call sync main function
-
-    This demonstrates that this works even when called from an async function.
-    """
-    main()
-
-
-if __name__ == "__main__":
-    loop = asyncio.get_event_loop_policy().get_event_loop()
-    try:
-        loop.run_until_complete(amain())
-    finally:
-        loop.close()

examples/sync_audio_streaming_with_predefined_voice_subtitles.py

@@ -1,26 +1,30 @@
 #!/usr/bin/env python3
 
-"""
+"""Sync variant of the async .stream() method to
-Basic audio streaming example for sync interface
+get audio chunks and feed them to SubMaker to
-
+generate subtitles"""
-"""
 
 import edge_tts
 
 TEXT = "Hello World!"
 VOICE = "en-GB-SoniaNeural"
 OUTPUT_FILE = "test.mp3"
+SRT_FILE = "test.srt"
 
 
 def main() -> None:
-    """Main function to process audio and metadata synchronously."""
+    """Main function"""
     communicate = edge_tts.Communicate(TEXT, VOICE)
+    submaker = edge_tts.SubMaker()
     with open(OUTPUT_FILE, "wb") as file:
         for chunk in communicate.stream_sync():
             if chunk["type"] == "audio":
                 file.write(chunk["data"])
             elif chunk["type"] == "WordBoundary":
-                print(f"WordBoundary: {chunk}")
+                submaker.feed(chunk)
+
+    with open(SRT_FILE, "w", encoding="utf-8") as file:
+        file.write(submaker.get_srt())
 
 
 if __name__ == "__main__":

src/edge_tts/communicate.py

@@ -518,7 +518,7 @@ class Communicate:
                     json.dump(message, metadata)
                     metadata.write("\n")
 
-    def stream_sync(self) -> Generator[Dict[str, Any], None, None]:
+    def stream_sync(self) -> Generator[TTSChunk, None, None]:
         """Synchronous interface for async stream method"""
 
         def fetch_async_items(queue: Queue) -> None:  # type: ignore

src/edge_tts/typing.py

@@ -82,3 +82,11 @@ class VoiceManagerVoice(Voice):
     """Voice data for VoiceManager."""
 
     Language: str
+
+
+class VoiceManagerFind(TypedDict):
+    """Voice data for VoiceManager.find()."""
+
+    Gender: NotRequired[Literal["Female", "Male"]]
+    Locale: NotRequired[str]
+    Language: NotRequired[str]

src/edge_tts/voices.py

@@ -7,10 +7,11 @@ from typing import Any, List, Optional
 
 import aiohttp
 import certifi
+from typing_extensions import Unpack
 
 from .constants import SEC_MS_GEC_VERSION, VOICE_HEADERS, VOICE_LIST
 from .drm import DRM
-from .typing import Voice, VoiceManagerVoice
+from .typing import Voice, VoiceManagerFind, VoiceManagerVoice
 
 
 async def __list_voices(
@@ -94,7 +95,9 @@ class VoicesManager:
         self.called_create: bool = False
 
     @classmethod
-    async def create(cls: Any, custom_voices: Optional[List[Voice]] = None) -> Any:
+    async def create(
+        cls: Any, custom_voices: Optional[List[Voice]] = None
+    ) -> "VoicesManager":
         """
         Creates a VoicesManager object and populates it with all available voices.
         """
@@ -106,7 +109,7 @@ class VoicesManager:
         self.called_create = True
         return self
 
-    def find(self, **kwargs: Any) -> List[VoiceManagerVoice]:
+    def find(self, **kwargs: Unpack[VoiceManagerFind]) -> List[VoiceManagerVoice]:
         """
         Finds all matching voices based on the provided attributes.
         """