Add sync versions of stream and save methods (#215)

* Add sync versions of stream and save methods

In order to provide synchronous interface to the library

* Fix save_sync() failing to use metadata_fname and fix typing issues

Signed-off-by: rany <[email protected]>

---------

Signed-off-by: rany <[email protected]>
Co-authored-by: rany <[email protected]>

examples/basic_sync_audio_streaming.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+
+"""
+Basic audio streaming example for sync interface
+
+"""
+
+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}")
+
+
+if __name__ == "__main__":
+ main()

examples/basic_sync_generation.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+
+"""
+Basic example of edge_tts usage in synchronous function
+"""
+
+import edge_tts
+
+TEXT = "Hello World!"
+VOICE = "en-GB-SoniaNeural"
+OUTPUT_FILE = "test.mp3"
+
+
+def main() -> None:
+ """Main function"""
+ communicate = edge_tts.Communicate(TEXT, VOICE)
+ communicate.save_sync(OUTPUT_FILE)
+
+
+if __name__ == "__main__":
+ main()

examples/sync_audio_generation_in_async_context.py

@@ -0,0 +1,36 @@
+#!/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

@@ -0,0 +1,42 @@
+#!/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()

src/edge_tts/communicate.py

@@ -2,13 +2,16 @@
 Communicate package.
 """
 
+import asyncio
+import concurrent.futures
 import json
 import re
 import ssl
 import time
 import uuid
 from contextlib import nullcontext
 from io import TextIOWrapper
+from queue import Queue
 from typing import (
  Any,
  AsyncGenerator,
@@ -498,3 +501,40 @@ async def save(
  ):
  json.dump(message, metadata)
  metadata.write("\n")
+
+ def stream_sync(self) -> Generator[Dict[str, Any], None, None]:
+ """Synchronous interface for async stream method"""
+
+ def fetch_async_items(queue: Queue) -> None: # type: ignore
+ async def get_items() -> None:
+ async for item in self.stream():
+ queue.put(item)
+ queue.put(None)
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ loop.run_until_complete(get_items())
+ loop.close()
+
+ queue: Queue = Queue() # type: ignore
+
+ with concurrent.futures.ThreadPoolExecutor() as executor:
+ executor.submit(fetch_async_items, queue)
+
+ while True:
+ item = queue.get()
+ if item is None:
+ break
+ yield item
+
+ def save_sync(
+ self,
+ audio_fname: Union[str, bytes],
+ metadata_fname: Optional[Union[str, bytes]] = None,
+ ) -> None:
+ """Synchronous interface for async save method."""
+ with concurrent.futures.ThreadPoolExecutor() as executor:
+ future = executor.submit(
+ asyncio.run, self.save(audio_fname, metadata_fname)
+ )
+ future.result()