acapela.py 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100
  1. import requests
  2. import re
  3. from core import FileManager
  4. from core.TTS.TTSModule import TTSModule, FailToLoadSoundFile, MissingTTSParameter
  5. import logging
  6. logging.basicConfig()
  7. logger = logging.getLogger("kalliope")
  8. TTS_URL = "http://www.acapela-group.com/demo-tts/DemoHTML5Form_V2_fr.php"
  9. TTS_CONTENT_TYPE = "audio/mpeg"
  10. TTS_TIMEOUT_SEC = 30
  11. class TCPTimeOutError(Exception):
  12. """
  13. This error is raised when the TCP connection has been lost. Probably due to a low internet
  14. connection while trying to access the remote API.
  15. """
  16. pass
  17. class Acapela(TTSModule):
  18. def __init__(self, **kwargs):
  19. super(Acapela, self).__init__(**kwargs)
  20. self.voice = kwargs.get('voice', None)
  21. if self.voice is None:
  22. raise MissingTTSParameter("voice parameter is required by the Acapela TTS")
  23. def say(self, words):
  24. """
  25. :param words: The sentence to say
  26. """
  27. self.generate_and_play(words, self._generate_audio_file)
  28. def _generate_audio_file(self):
  29. """
  30. Generic method used as a Callback in TTSModule
  31. - must provided the audio file and write it on the disk
  32. .. raises:: FailToLoadSoundFile, TCPTimeOutError
  33. """
  34. # Prepare payload
  35. payload = self.get_payload()
  36. try:
  37. # Get the mp3 URL from the page
  38. url = Acapela.get_audio_link(TTS_URL, payload)
  39. # getting the mp3
  40. r = requests.get(url, params=payload, stream=True, timeout=TTS_TIMEOUT_SEC)
  41. content_type = r.headers['Content-Type']
  42. logger.debug("Acapela : Trying to get url: %s response code: %s and content-type: %s",
  43. r.url,
  44. r.status_code,
  45. content_type)
  46. # Verify the response status code and the response content type
  47. if r.status_code != requests.codes.ok or content_type != TTS_CONTENT_TYPE:
  48. raise FailToLoadSoundFile("Acapela : Fail while trying to remotely access the audio file")
  49. # OK we get the audio we can write the sound file
  50. FileManager.write_in_file(self.file_path, r.content)
  51. except:
  52. raise TCPTimeOutError("TCP timeout, the connection to the remote API has been lost")
  53. def get_payload(self):
  54. """
  55. Generic method used load the payload used to acces the remote api
  56. :return: Payload to use to access the remote api
  57. """
  58. return {
  59. "MyLanguages": self.language,
  60. "MySelectedVoice": self.voice,
  61. "MyTextForTTS": self.words,
  62. "t": "1",
  63. "SendToVaaS": ""
  64. }
  65. @staticmethod
  66. def get_audio_link(url, payload, timeout_expected=TTS_TIMEOUT_SEC):
  67. """
  68. Return the audio link
  69. :param url: the url to access
  70. :param payload: the payload to use to acces the remote api
  71. :param timeout_expected: timeout before the post request is cancel
  72. :return: the audio link
  73. :rtype: String
  74. """
  75. r = requests.post(url, payload, timeout=timeout_expected)
  76. data = r.content
  77. return re.search("(?P<url>https?://[^\s]+).mp3", data).group(0)