# This is part of Kaylee # -- this code is licensed GPLv3 # Copyright 2015-2017 Clayton G. Hobbs # Portions Copyright 2013 Jezra import importlib import queue import sys import subprocess import signal import os.path from gi.repository import GObject, GLib from .recognizer import Recognizer from .util import * from .numbers import NumberParser import kayleevc.plugins class Kaylee: def __init__(self): self.ui = None self.options = {} self.continuous_listen = False # Load configuration self.config = Config() self.options = vars(self.config.options) # Make sure some plugins are configured to be loaded if self.config.plugins is None: print("error: no plugins configured", file=sys.stderr) sys.exit(1) # Load plugins self.plugins = [] for plugin in self.config.plugins.keys(): pmod = importlib.import_module(plugin, 'kayleevc.plugins') pobj = pmod.Plugin(self.config, plugin) self.plugins.append(pobj) # Create a hasher self.hasher = Hasher(self.config) # Create the strings file self.update_voice_commands_if_changed() if self.options['interface']: if self.options['interface'] == "g": from kayleevc.gui import GTKInterface as UI else: print("no GUI defined") sys.exit() self.ui = UI(self.options, self.options['continuous']) self.ui.connect("command", self.process_command) # Can we load the icon resource? icon = self.load_resource("icon_small.png") if icon: self.ui.set_icon_active_asset(icon) # Can we load the icon_inactive resource? icon_inactive = self.load_resource("icon_inactive_small.png") if icon_inactive: self.ui.set_icon_inactive_asset(icon_inactive) if self.options['history']: self.history = [] # Update the language if necessary self.language_updater = LanguageUpdater(self.config) self.language_updater.update_language_if_changed() # Create the recognizer self.recognizer = Recognizer(self.config) # Connect the recognizer's finished signal to the appropriate method self.recognizer.connect('finished', self.recognizer_finished) def update_voice_commands_if_changed(self): """Use hashes to test if the voice commands have changed""" stored_hash = self.hasher['voice_commands'] # Calculate the hash the voice commands have right now hasher = self.hasher.get_hash_object() for plugin in self.plugins: for string in sorted(plugin.corpus_strings): hasher.update(string.encode('utf-8')) # Add a separator to avoid odd behavior hasher.update('\n'.encode('utf-8')) new_hash = hasher.hexdigest() if new_hash != stored_hash: self.create_strings_file() self.hasher['voice_commands'] = new_hash self.hasher.store() def create_strings_file(self): # Open the strings file with open(self.config.strings_file, 'w') as strings: # Add command words to the corpus # FIXME: Doing this twice is a silly thing for plugin in self.plugins: for string in sorted(plugin.corpus_strings): strings.write(string + "\n") # Add number words to the corpus if NumberParser.number_words is not None: for word in NumberParser.number_words: strings.write(word + " ") strings.write("\n") def _log_history(self, plugin, text): """Log the recognized sentence to the history file""" if self.options['history']: self.history.append(f"{plugin.name}: {text}") if len(self.history) > self.options['history']: # Pop off the first item self.history.pop(0) # Open and truncate the history file with open(self.config.history_file, 'w') as hfile: for line in self.history: hfile.write(line + '\n') self._stop_ui(text) def _valid_cmd(self): """Run the valid_sentence_command if it's set""" if self.options['valid_sentence_command']: subprocess.call(self.options['valid_sentence_command'], shell=True) def _invalid_cmd(self, text): """Run the invalid_sentence_command if it's set""" if self.options['invalid_sentence_command']: subprocess.call(self.options['invalid_sentence_command'], shell=True) print(f"no matching command {text}") def recognizer_finished(self, recognizer, text): confidence_heap = queue.PriorityQueue() min_confidence = self.options['minimum_confidence'] # Add handlers to the heap for index, plugin in enumerate(self.plugins): # Get a handler handler = plugin.get_handler(text) # If the handler meets minimum confidence if ((handler is not None) and ((min_confidence is None) or (min_confidence is not None and handler.confidence >= min_confidence))): # Add a triple to the heap, so handlers are sorted first by # their confidence, then by index to break ties confidence_heap.put_nowait((handler, index, plugin)) # Run valid or invalid sentence command if confidence_heap.empty(): self._invalid_cmd(text) else: self._valid_cmd() # Give the command to the handlers that want it while True: # Get information about the handler try: info = confidence_heap.get_nowait() except queue.Empty: # If the queue is empty, we have nothing else to try, so break break handler = info[0] plugin = info[2] # Handle the command try: handler(self.plugin_tts) except kayleevc.plugins.HandlerFailure: # If the handler failed, keep trying continue else: # If the handler succeeded, log it and break self._log_history(plugin, text) break self._stop_ui(text) def plugin_tts(self, text): # Stop listening self.recognizer.pause() # Speak try: subprocess.call(self.options['tts'] + [text]) except KeyError: print('TTS:', text) # Resume listening self.recognizer.listen() def _stop_ui(self, text): # If there is a UI and we are not continuous listen if self.ui: if not self.continuous_listen: # Stop listening self.recognizer.pause() # Let the UI know that there is a finish self.ui.finished(text) def run(self): if self.ui: self.ui.run() else: self.recognizer.listen() def quit(self): sys.exit() def process_command(self, UI, command): print(command) if command == "listen": self.recognizer.listen() elif command == "stop": self.continuous_listen = False self.recognizer.pause() elif command == "continuous_listen": self.continuous_listen = True self.recognizer.listen() elif command == "quit": self.quit() def load_resource(self, string): # TODO: Use the Config object for this path management local_data = os.path.join(os.path.dirname(__file__), '..', 'data') paths = ["/usr/share/kaylee/", "/usr/local/share/kaylee", local_data] for path in paths: resource = os.path.join(path, string) if os.path.exists(resource): return resource # If we get this far, no resource was found return False def run(): # Make our kaylee object kaylee = Kaylee() # Init gobject threads GObject.threads_init() # We want a main loop main_loop = GObject.MainLoop() # Handle sigint signal.signal(signal.SIGINT, signal.SIG_DFL) # Run the kaylee kaylee.run() # Start the main loop try: main_loop.run() except: main_loop.quit() sys.exit()