Source code for pisak.sound_effects
"""
Sound effects player.
"""
import subprocess
import threading
import gi
gi.require_version('Gst', '1.0')
from gi.repository import GObject, Gst
import pisak
from pisak import arg_parser
from pisak import logger
_LOG = logger.get_logger(__name__)
GObject.threads_init()
Gst.init(arg_parser.get_args().args)
[docs]class SoundEffectsPlayer:
"""
Player of some simple sound effects. Audio files are loaded just
once when registered and then they are stored and available in a sounds pool,
throughout the class' entire lifetime.
:param sounds_dict: dictionary of sounds.
"""
def __init__(self, sounds_dict):
super().__init__()
self.sounds = sounds_dict
self._volume = pisak.config.as_int('sound_effects_volume') / 100
self._playbin = Gst.ElementFactory.make('playbin', 'button_sound')
self._playbin.set_property("volume", self.volume)
self._bus = self._playbin.get_bus()
self._bus.connect('message', self._on_message)
@property
def volume(self):
"""
Common volume level, normalized value between 0 and 1.
"""
return self._volume
@volume.setter
def volume(self, value):
vol = float(value)
if 0 <= vol <= 1:
self._volume = vol
self._playbin.set_property("volume", vol)
else:
msg = "Provided value must be between 0 and 1. Received {}."
_LOG.error(msg.format(vol))
[docs] def play(self, sound_name):
"""
Play a sound. Sounds are played in a thread-safe, non-blocking manner.
:param sound_name: name of a previously registered sound.
"""
self.volume = self.volume
self._playbin.set_state(Gst.State.READY)
if sound_name in self.sounds:
self._playbin.set_property('uri', 'file://' + self.sounds[sound_name])
else:
self._playbin.set_property('uri', 'file://' + sound_name)
self._bus.add_signal_watch()
self._playbin.set_state(Gst.State.READY)
self._playbin.set_state(Gst.State.PLAYING)
def _free_resource(self):
self._bus.remove_signal_watch()
self._playbin.set_state(Gst.State.NULL)
msg = 'Resources freed from playbin with file: ' +\
str(self._playbin.get_property('uri'))
_LOG.debug(msg)
def _on_message(self, _bus, message):
if message.type == Gst.MessageType.EOS:
self._free_resource()
elif message.type == Gst.MessageType.ERROR:
_LOG.warning("An error occured while playing file: " +\
str(self._playbin.get_property('uri')))
self._free_resource()
[docs]class Synthesizer:
"""
Speech synthesizer. Uses Milena text-to-speech program.
:param text: text to be read, string.
"""
def __init__(self, text):
self.text = text
self.process = None
@staticmethod
def _sec_converter(seconds):
"""
Converts seconds to minutes and seconds, formats an output properly,
for the Milena program exclusively.
:param seconds: number of seconds, can be string, int or float.
:return: properly formatted number of minutes and seconds, string.
"""
seconds = int(seconds)
minutes = seconds // 60
seconds = seconds - (minutes*60)
return "{0:02d}:{1:02d}".format(minutes, seconds)
[docs] def read(self, timeout=None):
"""
Read the text out loud. Milena is spawned as a subprocess
in a non-blocking mode.
:param timeout: optional, max number of seconds
that the reading should be limited to.
"""
if timeout is None or timeout <= 0:
self.process = subprocess.Popen(["milena_say", self.text])
else:
timeout = self._sec_converter(timeout)
if pisak.arg_parser.get_args().debug:
call = 'milena_say "-S trim 0 {}" {}'
else:
call = 'milena_say "-S trim 0 {}" {} 2>/dev/null'
self.process = subprocess.Popen([call.format(timeout,
self.text)],
shell=True)
[docs] def read_and_call(self, func_to_call, timeout=None):
"""
Read the text out loud and then, when finished,
call some function. Everything is executed
from a separate thread.
:param func_to_call: function to be called.
"""
def worker():
self.read(timeout)
self.process.wait()
func_to_call()
thread = threading.Thread(target=worker, daemon=True)
thread.start()