thread safe slu

README

@@ -33,21 +33,16 @@ DONE - send action through osc
 DONE - show an action performed message (message log with timing?)
 DONE click section = select that section
 DONE click action = perform action 
+DONE add thread for slu
+DONE remove section changer UI
 
 events = click action or words to resynchronize ?
 click line = synchronize to that line
 click action = synchronize to the next line
-
-add global timer which shows elapsed time
+insert timer in main ui, use it for logger
+add logger
 change xml view to reflect already performed actions, already recognized text
 move slu to asr
-
 make selector a proper window
-
-add thread for slu
-
 allow sequence advance in slu, add UI for that
-
-remove section changer UI
-
 add global keybindings (1-9 for sections, y/n)...

main.py

@@ -25,7 +25,7 @@ signal.signal(signal.SIGINT, signal.SIG_DFL)
 
 # import local stuff
 import confirm, asr, actions, xmlview
-import levenstein, SLU, osc
+import levenstein, slu, osc
 
 class ScriptedASR(Gtk.Window):
     def __init__(self, xml_filename, asr_config_file, osc_host, osc_port):
@@ -56,7 +56,7 @@ class ScriptedASR(Gtk.Window):
             found = re.search('section(\d+)\.fst$', section_fst)
             if found:
                 section_id = int(found.group(1))
-                self.slu[section_id] = SLU.SLU(prefix % 'dico_word.txt', prefix % 'dico_action.txt', section_fst, prefix % 'clean_tail.fst')
+                self.slu[section_id - 1] = slu.SLU(prefix % 'dico_word.txt', prefix % 'dico_action.txt', section_fst, prefix % 'clean_tail.fst')
 
         self.add(vbox)
         self.show_all()
@@ -80,17 +80,20 @@ class ScriptedASR(Gtk.Window):
                 self.current_line = i
                 self.lines[self.current_line].highlight(True)
 
+    def slu_finished(self, model, slu_output):
+        for action_id in range(self.previous_actions, model.num_actions()):
+            action = model.get_action(action_id)
+            actions.perform_action(actions.Action(action))
+
     def hyp_changed(self, hyp):
         #hyp = ' '.join(hyp).replace('[noise]', ' ').split()
         words = hyp[-1].strip().replace('_', ' ').split()
-        section_id = int(self.xmlview.current_section.name) - 1
+        section_id = self.xmlview.get_section()
+        print section_id
         if section_id in self.slu:
-            slu = self.slu[section_id]
-            previous_actions = slu.num_actions()
-            slu.process(words)
-            for action_id in range(previous_actions, slu.num_actions()):
-                action = slu.get_action(action_id)
-                actions.perform_action(actions.Action(action))
+            model = self.slu[section_id]
+            self.previous_actions = model.num_actions()
+            model.process(words, self.slu_finished)
 
         #if self.current_line >= len(self.lines) - 1:
         #    print "FINISHED"

SLU.py → slu.py

+from gi.repository import GObject, GLib
+import threading
 from ctypes import *
 
 _backend = None
+_semaphore = None
 
 class SLU:
     #/src.new/rocio_slu -word "$prefix"_dico_word.txt -action "$prefix"_dico_action.txt -fstmodel "$prefix".fst -fstclean "$prefix"_clean_tail.fst
     def __init__(self, word_lexicon, action_lexicon, model_fst, cleaner_fst, library='slu/src.new/librocio_slu.so'):
-        global _backend
+        global _backend, _semaphore
         if _backend == None:
             _backend = cdll.LoadLibrary(library)
 
@@ -29,25 +32,48 @@ class SLU:
             _backend.free_slu.argtypes = [c_void_p]
             _backend.free_slu.restype = None
 
+            _semaphore = threading.Semaphore()
+        _semaphore.acquire()
+        thread = threading.Thread(target=self._init_thread, args=[word_lexicon, action_lexicon, model_fst, cleaner_fst])
+        thread.daemon = True
+        thread.start()
+
+    def _init_thread(self, word_lexicon, action_lexicon, model_fst, cleaner_fst):
+        global _backend, _semaphore
         self.slu = _backend.init_slu(word_lexicon, action_lexicon, model_fst, cleaner_fst)
+        _semaphore.release()
+
+    def process(self, words, callback):
+        thread = threading.Thread(target=self._process_thread, args=[words, callback])
+        thread.daemon = True
+        thread.start()
 
-    def process(self, words):
+    def _process_thread(self, words, callback):
+        global _backend, _semaphore
         c_words = (c_char_p * len(words))(*words)
-        return _backend.run_slu(self.slu, c_words, len(words), self.num_actions())
+        _semaphore.acquire()
+        output = _backend.run_slu(self.slu, c_words, len(words), self.num_actions())
+        _semaphore.release()
+        GLib.idle_add(callback, self, output)
 
     def num_actions(self):
-        return _backend.num_actions(self.slu)
+        global _backend, _semaphore
+        output = _backend.num_actions(self.slu)
+        return output
 
     def get_action(self, index):
-        return _backend.get_action(self.slu, index)
+        global _backend, _semaphore
+        output = _backend.get_action(self.slu, index)
+        return output
 
     def shutdown(self):
+        global _backend, _semaphore
         _backend.free_slu(self.slu)
 
 if __name__ == '__main__':
     prefix = 'slu/automate/homeostasis_25nov_%s'
     slu = SLU(prefix % 'dico_word.txt', prefix % 'dico_action.txt', prefix % 'section6.fst', prefix % 'clean_tail.fst')
     print 'before'
-    slu.process(open('slu/homeostasis_25nov.asr/sect6.ref').read().strip().split())
+    slu.process(open('slu/homeostasis_25nov.asr/sect6.ref').read().strip().split(), lambda x: sys.stdout.write('%s\n' % x))
     print 'after'
     slu.shutdown()