Viewing file:  espeechfactory.py (15.34 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# Orca
#
# Copyright 2005-2008 Google Inc.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Library General Public
# License as published by the Free Software Foundation; either
# version 2 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Library General Public License for more details.
#
# You should have received a copy of the GNU Library General Public
# License along with this library; if not, write to the
# Free Software Foundation, Inc., Franklin Street, Fifth Floor,
# Boston MA 02110-1301 USA.
#
"""Python wrapper for Emacspeak speech servers.
The emacspeak TTS server provides a simple but powerful and
well-tested speech-server abstraction. That server is implemented
as an external program (typically in TCL). This wrapper class
provides Python access to available Emacspeak speech servers.
Initially, this class will provide Python access to the TTS
server commands. Over time, this module may also implement
functionality present in Emacspeak's Lisp layer ---specifically,
higher level TTS functionality provided by the following
emacspeak modules:
0) dtk-speak.el
1) emacspeak-pronounce.el
2) accs-structure.el
"""
__id__ = "$Id$"
__author__ = "T. V. Raman"
__version__ = "$Revision$"
__date__ = "$Date$"
__copyright__ = "Copyright (c) 2005-2008 Google Inc."
__license__ = "LGPL"
__all__ = ['Speaker']
import os
import re
import debug
import settings
import speechserver
from orca_i18n import _ # for gettext support
class SpeechServer(speechserver.SpeechServer):
"""Provides speech server abstraction.
Class Variables:
location -- specifies directory where Emacspeak
speech servers are installed.
config -- dictionary of default settings.
Speaker objects can be initialized with the following parameters:
engine -- TTS server to instantiate. Default: outloud
host -- Host that runs server. Default: localhost
settings -- Dictionary of default settings.
"""
location = "/usr/share/emacs/site-lisp/emacspeak/servers"
config = {'splitcaps' : 1,
'rate' : 70,
'capitalize' : 0,
'allcaps' : 0,
'punctuations' : 'all'
}
# Dictionary of known running servers. The key is the name and
# the value is the SpeechServer instance. We do this to enforce a
# singleton instance of any given server.
#
__activeServers = {}
# A regexp to match chunks of text which will be sent to speech server
# NB: this must always match, the question is how many characters.
_speechChunkRegexp = ('.{0,120}?[\n,.;:!?)][]}"]?\\s+'
'|.{1,120}(?:\\Z|\\s+)'
'|.{1,120}')
def getFactoryName():
"""Returns a localized name describing this factory."""
# Translators: this refers to the speech synthesis services
# provided by the separate emacspeak utility available at
# http://emacspeak.sourceforge.net/.
#
return _("Emacspeak Speech Services")
getFactoryName = staticmethod(getFactoryName)
def getSpeechServers():
"""Gets available speech servers as a list. The caller
is responsible for calling the shutdown() method of each
speech server returned.
"""
haveNewServers = False
serversConf = file(os.path.join(SpeechServer.location, '.servers'))
for name in serversConf:
name = name.strip()
if name == '' or name[0] == '#':
continue
if name not in SpeechServer.__activeServers:
try:
SpeechServer.__activeServers[name] = SpeechServer(name)
haveNewServers = True
except:
debug.printException(debug.LEVEL_WARNING)
serversConf.close()
# Check whether some servers have died and remove those from the list
if haveNewServers:
from time import sleep
sleep(1)
for server in SpeechServer.__activeServers.values():
if not server.process or server.process.poll() is not None:
server.shutdown()
return SpeechServer.__activeServers.values()
getSpeechServers = staticmethod(getSpeechServers)
def getSpeechServer(info=['outloud', 'outloud']):
"""Gets a given SpeechServer based upon the info.
See SpeechServer.getInfo() for more info.
"""
if info[0] in SpeechServer.__activeServers:
return SpeechServer.__activeServers[info[0]]
else:
try:
return SpeechServer(info[0])
except:
debug.printException(debug.LEVEL_SEVERE)
return None
getSpeechServer = staticmethod(getSpeechServer)
def shutdownActiveServers():
"""Cleans up and shuts down this factory.
"""
for server in SpeechServer.__activeServers.values():
server.shutdown()
shutdownActiveServers = staticmethod(shutdownActiveServers)
def __init__ (self,
engine='outloud',
host='localhost',
initial=config):
"""Launches speech engine."""
speechserver.SpeechServer.__init__(self)
self._engine = engine
self._output = None
self.process = None
e = __import__(_getcodes(engine),
globals(),
locals(),
[''])
self.getvoice = e.getvoice
self.getrate = e.getrate
self.getvoicelist = e.getvoicelist
self._specialChars = e.makeSpecialCharMap()
if host == 'localhost':
self._server = os.path.join(SpeechServer.location, self._engine)
else:
self._server = os.path.join(SpeechServer.location,
"ssh-%s" % self._engine)
# Start the process and close its output channels since they were
# closed implicitly in previous version of espeechfactory.py.
from subprocess import (Popen, PIPE)
proc = Popen(self._server, close_fds=True,
stdin=PIPE, stdout=PIPE, stderr=PIPE)
proc.stdout.close()
proc.stderr.close()
proc.stdout = proc.stderr = None
self._output = proc.stdin
self.process = proc
self._speechChunk = re.compile(self._speechChunkRegexp, re.DOTALL)
self._settings = {}
if initial:
self._settings.update(initial)
self.configure(self._settings)
def configure(self, engineSettings):
"""Configure engine with settings."""
for k in engineSettings.keys():
if hasattr(self, k) and callable(getattr(self, k)):
getattr(self, k)(engineSettings[k])
def settings(self):
return self._settings
def getInfo(self):
"""Returns [driverName, serverId]
"""
return [self._engine, self._engine]
def getVoiceFamilies(self):
"""Returns a list of speechserver.VoiceFamily instances
representing all the voice families known by the speech server.
"""
families = []
try:
for voice in self.getvoicelist():
props = {
speechserver.VoiceFamily.NAME : voice
}
families.append(speechserver.VoiceFamily(props))
except:
debug.printException(debug.LEVEL_SEVERE)
return families
def _quoteSpecialChars(self, text):
"""Replaces all special characters in text by their replacements
according to self._specialChars.
"""
for char, name in self._specialChars:
text = text.replace(char, name)
return text
def queueText(self, text="", acss=None):
"""Adds the text to the queue.
Arguments:
- text: text to be spoken
- acss: acss.ACSS instance; if None,
the default voice settings will be used.
Otherwise, the acss settings will be
used to augment/override the default
voice settings.
Output is produced by the next call to speak.
"""
if not settings.enableSpeech:
return
# A command to send to the speech server queueing text
cmd = "q { %s }\n"
if acss:
try:
code = self.getvoice(acss)
cmd = "q {%s %%s %s}\n" % (code[0], code[1])
except:
debug.printException(debug.LEVEL_WARNING)
# Split text into chunks and each chunk queue separately
for chunk in (t.group() for t in self._speechChunk.finditer(text)):
self._output.write(cmd % self._quoteSpecialChars(chunk))
def queueTone(self, pitch=440, duration=50):
"""Adds a tone to the queue.
Output is produced by the next call to speak.
"""
self._output.write("t %s %s\n " % (pitch, duration))
def queueSilence( self, duration=50):
"""Adds silence to the queue.
Output is produced by the next call to speak.
"""
self._output.write("sh %s" % duration)
def speakCharacter(self, character, acss=None):
"""Speaks a single character immediately.
Arguments:
- character: text to be spoken
- acss: acss.ACSS instance; if None,
the default voice settings will be used.
Otherwise, the acss settings will be
used to augment/override the default
voice settings.
"""
if character in '{\\}':
character = self._quoteSpecialChars(character)
self._output.write("l {%s}\n" % character)
self._output.flush()
def speakUtterances(self, utteranceList, acss=None, interrupt=True):
"""Speaks the given list of utterances immediately.
Arguments:
- utteranceList: list of strings to be spoken
- acss: acss.ACSS instance; if None,
the default voice settings will be used.
Otherwise, the acss settings will be
used to augment/override the default
voice settings.
- interrupt: if True, stop any speech currently in progress.
"""
for utt in utteranceList:
self.queueText(str(utt), acss)
self._output.write("d\n")
self._output.flush()
def speak(self, text="", acss=None, interrupt=True):
"""Speaks all queued text immediately. If text is not None,
it is added to the queue before speaking.
Arguments:
- text: optional text to add to the queue before speaking
- acss: acss.ACSS instance; if None,
the default voice settings will be used.
Otherwise, the acss settings will be
used to augment/override the default
voice settings.
- interrupt: if True, stops any speech in progress before
speaking the text
"""
# If the user has speech turned off, just return.
#
if not settings.enableSpeech:
return
self.queueText(text, acss)
self._output.write("d\n")
self._output.flush()
def increaseSpeechPitch(self, step=0.5):
"""Increases the speech pitch."""
self._settings['average-pitch'] += step
def decreaseSpeechPitch(self, step=0.5):
"""Decreases the speech pitch."""
self._settings['average-pitch'] -= step
def increaseSpeechRate(self, step=5):
"""Increases the speech rate.
"""
self._settings['rate'] += step
self._output.write("tts_set_speech_rate %s\n" \
% self.getrate(self._settings['rate']))
self._output.flush()
def decreaseSpeechRate(self, step=5):
"""Decreases the speech rate.
"""
self._settings['rate'] -= step
self._output.write("tts_set_speech_rate %s\n" \
% self.getrate(self._settings['rate']))
self._output.flush()
def stop(self):
"""Stops ongoing speech and flushes the queue."""
self._output.write("s\n")
self._output.flush()
def shutdown(self):
"""Shuts down the speech engine."""
if self.process is None:
return
if self.process.poll() is None:
from signal import SIGKILL
os.kill(self.process.pid, SIGKILL)
self.process.wait()
self._output.close()
self._output = self.process = None
if self is SpeechServer.__activeServers.get(self._engine, None):
del SpeechServer.__activeServers[self._engine]
def reset(self, text=None, acss=None):
"""Resets the speech engine."""
self._output.write("tts_reset\n")
self._output.flush()
def version(self):
"""Speak TTS version info."""
self._output.write("version\n")
def punctuations(self, mode):
"""Set punctuation mode."""
if mode in ['all', 'some', 'none']:
self._settings['punctuations'] = mode
self._output.write("tts_set_punctuations %s\n" % mode)
self._output.flush()
def rate(self, r):
"""Set speech rate."""
self._settings['rate'] = r
self._output.write("tts_set_speech_rate %s\n" % self.getrate(r))
self._output.flush()
def splitcaps(self, flag):
"""Set splitcaps mode. 1 turns on, 0 turns off"""
flag = bool(flag) and 1 or 0
self._settings['splitcaps'] = flag
self._output.write("tts_split_caps %s\n" % flag)
self._output.flush()
def capitalize(self, flag):
"""Set capitalization mode. 1 turns on, 0 turns off"""
flag = bool(flag) and 1 or 0
self._settings['capitalize'] = flag
self._output.write("tts_capitalize %s\n" % flag)
self._output.flush()
def allcaps(self, flag):
"""Set allcaps mode. 1 turns on, 0 turns off"""
flag = bool(flag) and 1 or 0
self._settings['allcaps'] = flag
self._output.write("tts_allcaps_beep %s\n" % flag)
self._output.flush()
def __del__(self):
"Shuts down the speech engine."
self.shutdown()
def _getcodes(engine):
"""Helper function that fetches synthesizer codes for a
specified engine."""
if engine not in _codeTable:
raise Exception("No code table for %s" % engine)
return _codeTable[engine]
_codeTable = {
'dtk-exp' : 'dectalk',
'dtk-mv' : 'dectalk',
'dtk-soft' : 'dectalk',
'outloud' : 'outloud',
}
def _test():
"""Self test."""
import time
import acss
s = SpeechServer()
a = acss.ACSS()
s.punctuations('some')
s.queueText("This is an initial test.")
s.queueText("Next, we'll test audio formatted output.")
for d in ['average-pitch', 'pitch-range',
'richness', 'stress']:
for i in range(0, 10, 2):
a[d] = i
s.queueText("Set %s to %i. " % (d, i), a)
del a[d]
s.queueText("Reset %s." % d, a)
s.speak()
print "sleeping while waiting for speech to complete."
time.sleep(40)
s.shutdown()
if __name__ == "__main__":
_test()
|