Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- ToolsAE.py
- # -*- coding: utf-8 -*-
- import os, stat, time, json, hashlib
- import urllib3.request, requests
- import subprocess
- class IamToken(object):
- def __init__(self):
- self.filename = '/var/lib/asterisk/agi-bin/iam.token'
- self.expireTime = 60*60*11
- def get(self):
- self.__iamUpdate()
- with open(self.filename) as f:
- key = f.read()
- return key.rstrip()
- def __iamUpdate(self):
- modTimeIam = int(os.path.getmtime(self.filename))
- curTime = int(time.time())
- if modTimeIam < curTime-self.expireTime:
- iamCmd = subprocess.Popen(['/usr/local/bin/yc','iam', 'create-token'], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- out, err = iamCmd.communicate()
- with open(self.filename, "w") as f:
- f.write(out.rstrip())
- class YandexSpeechSTT(object):
- def __init__(self, folderId, iamToken):
- self.folderId = folderId
- self.iamToken = iamToken
- def __convertAudio(self, fileSrc, fileDst):
- convertCmd = subprocess.Popen(['/usr/bin/opusenc', fileSrc, fileDst], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- output, error = convertCmd.communicate()
- if convertCmd.returncode != 0:
- raise Exception(error.strip())
- os.remove(fileSrc)
- def convertVoiceToText(self, fileSrc, topic):
- import urllib2
- fileDst = '/tmp/' + os.path.splitext(os.path.basename(fileSrc))[0] + '.ogg'
- self.__convertAudio(fileSrc, fileDst)
- with open(fileDst, "rb") as f:
- data = f.read()
- params = "&".join([
- "topic=%s" % topic,
- "folderId=%s" % self.folderId,
- "lang=ru-RU"
- ])
- url = urllib2.Request("https://stt.api.cloud.yandex.net/speech/v1/stt:recognize?%s" % params, data=data)
- url.add_header("Authorization", "Bearer %s" % self.iamToken)
- responseData = urllib2.urlopen(url).read().decode('UTF-8')
- decodedData = json.loads(responseData)
- if decodedData.get("error_code") is None:
- os.remove(fileDst)
- return decodedData.get("result")
- else:
- raise Exception("Error_code: %s, message: %s" % (decodedData.get("error_code"), decodedData.get("error_message")))
- class YandexSpeechTTS(object):
- def __init__(self, folderId, iamToken):
- self.folderId = folderId
- self.iamToken = iamToken
- self.dstPath = '/var/lib/asterisk/sounds/ru/ysc'
- self.dstMd5Path = '/var/lib/asterisk/sounds/ru/ysc/md5'
- def __synthesize(self, text):
- url = 'https://tts.api.cloud.yandex.net/speech/v1/tts:synthesize'
- headers = {'Authorization': 'Bearer ' + self.iamToken,}
- data = {
- 'text': text,
- 'lang': 'ru-RU',
- 'folderId': self.folderId,
- 'format': 'lpcm',
- 'emotion': 'neutral',
- 'speed': '1.0',
- 'voice': 'jane',
- 'sampleRateHertz': 48000,
- }
- resp = requests.post(url, headers=headers, data=data, stream=True, verify=False)
- if resp.status_code != 200:
- raise Exception("Invalid response received: code: %d, message: %s" % (resp.status_code, resp.text))
- for chunk in resp.iter_content(chunk_size=None):
- yield chunk
- def __textToHashFname(self, hash, text):
- with open(self.dstMd5Path + "/" + hash, "w") as f:
- f.write(text)
- def __convertAudio(self, typ, fileSrc, fileDst, text):
- if typ == 'ast':
- convertCmd = subprocess.Popen(['/usr/bin/sox', fileSrc, '-e', 'a-law', '-t', 'RAW', '-r', '8000', '-c', '1', fileDst], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- if typ == 'raw':
- convertCmd = subprocess.Popen(['/usr/bin/sox', '-r', '48000', '-b', '16', '-e', 'signed-integer', '-c', '1', fileSrc, fileDst], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- self.__textToHashFname(textToHash(text),text)
- output, error = convertCmd.communicate()
- if convertCmd.returncode != 0:
- raise Exception(error.strip())
- def ToFile(self, filename, text):
- fileSrc = self.dstPath + "/" + filename + '.raw'
- fileDst = self.dstPath + "/" + filename + '.wav'
- fileAster = self.dstPath + "/" + filename + '.alaw'
- exPath = self.dstMd5Path + "/" + textToHash(text)
- if os.path.isfile(exPath) and os.path.isfile(fileAster):
- return
- with open(fileSrc, "wb") as f:
- for audio_content in self.__synthesize(text):
- f.write(audio_content)
- self.__convertAudio('raw', fileSrc, fileDst, text)
- os.remove(fileSrc)
- self.__convertAudio('ast', fileDst, fileAster, text)
- os.remove(fileDst)
- def textToHash(text):
- return hashlib.md5(text).hexdigest()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement