From 8c356a000cf16f17bc8c21890ccecc4c8a2dd2c6 Mon Sep 17 00:00:00 2001 From: rany2 Date: Thu, 5 Jan 2023 00:07:08 +0200 Subject: [PATCH] Slightly cleanup some more --- src/edge_tts/communicate.py | 25 ++++++++----------------- src/edge_tts/exceptions.py | 2 ++ src/edge_tts/list_voices.py | 7 +++---- src/edge_tts/util.py | 16 ++++++++++------ 4 files changed, 23 insertions(+), 27 deletions(-) diff --git a/src/edge_tts/communicate.py b/src/edge_tts/communicate.py index 5c04c06..0f416e0 100644 --- a/src/edge_tts/communicate.py +++ b/src/edge_tts/communicate.py @@ -12,7 +12,8 @@ from xml.sax.saxutils import escape import aiohttp -from edge_tts.exceptions import * +from edge_tts.exceptions import (NoAudioReceived, UnexpectedResponse, + UnknownResponse) from .constants import WSS_URL @@ -207,7 +208,6 @@ class Communicate: ValueError: If the voice is not valid. """ self.text = text - self.boundary_type = 1 self.codec = "audio-24khz-48kbitrate-mono-mp3" self.voice = voice # Possible values for voice are: @@ -284,7 +284,7 @@ class Communicate: # download indicates whether we should be expecting audio data, # this is so what we avoid getting binary data from the websocket # and falsely thinking it's audio data. - download = False + download_audio = False # audio_was_received indicates whether we have received audio data # from the websocket. This is so we can raise an exception if we @@ -332,12 +332,12 @@ class Communicate: "Path" in parameters and parameters["Path"] == "turn.start" ): - download = True + download_audio = True elif ( "Path" in parameters and parameters["Path"] == "turn.end" ): - download = False + download_audio = False break elif ( "Path" in parameters @@ -376,15 +376,6 @@ class Communicate: "Path" in parameters and parameters["Path"] == "response" ): - # TODO: implement this: - """ - X-RequestId:xxxxxxxxxxxxxxxxxxxxxxxxx - Content-Type:application/json; charset=utf-8 - Path:response - - {"context":{"serviceTag":"yyyyyyyyyyyyyyyyyyy"},"audio": - {"type":"inline","streamId":"zzzzzzzzzzzzzzzzz"}} - """ pass else: raise UnknownResponse( @@ -392,7 +383,7 @@ class Communicate: + received.data ) elif received.type == aiohttp.WSMsgType.BINARY: - if download: + if download_audio: yield { "type": "audio", "data": b"Path:audio\r\n".join( @@ -402,12 +393,12 @@ class Communicate: audio_was_received = True else: raise UnexpectedResponse( - "The service sent a binary message, but we are not expecting one." + "We received a binary message, but we are not expecting one." ) if not audio_was_received: raise NoAudioReceived( - "No audio was received from the service. Please verify that your parameters are correct." + "No audio was received. Please verify that your parameters are correct." ) async def save( diff --git a/src/edge_tts/exceptions.py b/src/edge_tts/exceptions.py index c37c55a..dd2668c 100644 --- a/src/edge_tts/exceptions.py +++ b/src/edge_tts/exceptions.py @@ -1,3 +1,5 @@ +"""Exceptions for the Edge TTS project.""" + class UnknownResponse(Exception): """Raised when an unknown response is received from the server.""" diff --git a/src/edge_tts/list_voices.py b/src/edge_tts/list_voices.py index f1d50a3..1fba22b 100644 --- a/src/edge_tts/list_voices.py +++ b/src/edge_tts/list_voices.py @@ -1,5 +1,5 @@ """ -list_voices package. +list_voices package for edge_tts. """ import json @@ -9,13 +9,12 @@ import aiohttp from .constants import VOICE_LIST -async def list_voices(proxy=None): +async def list_voices(*, proxy=None): """ List all available voices and their attributes. This pulls data from the URL used by Microsoft Edge to return a list of - all available voices. However many more experimental voices are available - than are listed here. (See https://aka.ms/csspeech/voicenames) + all available voices. Returns: dict: A dictionary of voice attributes. diff --git a/src/edge_tts/util.py b/src/edge_tts/util.py index 7f55ed5..8c403a3 100644 --- a/src/edge_tts/util.py +++ b/src/edge_tts/util.py @@ -10,7 +10,8 @@ import sys from edge_tts import Communicate, SubMaker, list_voices -async def _list_voices(proxy): +async def _print_voices(proxy): + """Print all available voices.""" for idx, voice in enumerate(await list_voices(proxy=proxy)): if idx != 0: print() @@ -22,7 +23,8 @@ async def _list_voices(proxy): print(f"{key}: {voice[key]}") -async def _tts(args): +async def _run_tts(args): + """Run TTS after parsing arguments from command line.""" tts = await Communicate( args.text, args.voice, @@ -33,6 +35,7 @@ async def _tts(args): try: media_file = None if args.write_media: + # pylint: disable=consider-using-with media_file = open(args.write_media, "wb") subs = SubMaker(args.overlapping) @@ -55,7 +58,7 @@ async def _tts(args): media_file.close() -async def _main(): +async def _async_main(): parser = argparse.ArgumentParser(description="Microsoft Edge TTS") group = parser.add_mutually_exclusive_group(required=True) group.add_argument("-t", "--text", help="what TTS will say") @@ -108,7 +111,7 @@ async def _main(): args = parser.parse_args() if args.list_voices: - await _list_voices(args.proxy) + await _print_voices(args.proxy) sys.exit(0) if args.text is not None or args.file is not None: @@ -123,11 +126,12 @@ async def _main(): with open(args.file, "r", encoding="utf-8") as file: args.text = file.read() - await _tts(args) + await _run_tts(args) def main(): - asyncio.get_event_loop().run_until_complete(_main()) + """Run the main function using asyncio.""" + asyncio.get_event_loop().run_until_complete(_async_main()) if __name__ == "__main__":