HOTWORDS file parsed as yaml and accepts @dot.notation suffix

whisper-2023/hotwords.py

@@ -7,6 +7,7 @@ import sys
 from os import environ
 from os import kill
 import requests
+import yaml
 
 def main():
    managerToParserQ = queue.Queue(maxsize=1)
@@ -119,6 +120,53 @@ class Parser(threading.Thread):
       r = sr.Recognizer()
       return r.recognize_whisper(clip, language="english")
 
+def load_dot_notation(v, s):
+   items = s.replace("[]", ".[]").split(".")
+   return _load_dot_notation(v, items)
+
+def _load_dot_notation(v, items):
+   for i in range(len(items)):
+      k = items[i]
+      if not k:
+         continue
+      if k == "[]":
+         if isinstance(v, list):
+            result = []
+            for j in v:
+               subresult = _load_dot_notation(j, items[i+1:])
+               if isinstance(subresult, list):
+                  result.extend(subresult)
+               else:
+                  result.append(subresult)
+            return result
+         else:
+            result = []
+            for j in v.values():
+               subresult = _load_dot_notation(j, items[i+1:])
+               if isinstance(subresult, list):
+                  result.extend(subresult)
+               else:
+                  result.append(subresult)
+            return result
+      else:
+         if isinstance(v, list):
+            v = v[int(k)]
+         else:
+            v = v[k]
+   return v
+
+def test_load_dot_notation():
+   assert("a" == load_dot_notation("a", "."))
+   assert(["a"] == load_dot_notation(["a"], "."))
+   assert("b" == load_dot_notation({"a":"b"}, ".a"))
+   assert("c" == load_dot_notation({"a":{"b":"c"}}, ".a.b"))
+   assert("c" == load_dot_notation({"a":{"b":["c"]}}, ".a.b.0"))
+   assert(["c","d"] == load_dot_notation({"a":{"b":"c"}, "a2":{"b":"d"}}, ".[].b"))
+   assert(["c","d"] == load_dot_notation({"a":{"b":["c"], "b2":["d"]}}, ".a.[].0"))
+   assert(["c","d"] == load_dot_notation({"a":{"b":["c"], "b2":["d"]}}, ".a[].0"))
+   assert(["c","d"] == load_dot_notation(["c", "d"], "."))
+   assert(["c","d"] == load_dot_notation(["c", "d"], "[]"))
+test_load_dot_notation()
 
 class Reactor(threading.Thread):
    def __init__(self, inq, outq):
@@ -126,7 +174,7 @@ class Reactor(threading.Thread):
       self.inq = inq
       self.outq = outq
       self.load_hotwords = Reactor.new_load_hotwords()
-      [i for i in self.load_hotwords()]
+      log(f"hotwords: {self.load_hotwords()}")
 
    def new_load_hotwords():
       p = environ.get("HOTWORDS", None)
@@ -136,15 +184,24 @@ class Reactor(threading.Thread):
          return load_nothing
 
       try:
-         def load_hotwords_in_file():
+         if "@" in p:
-            with open(p, "r") as f:
+            def load_hotwords_in_yaml_file():
-               return ["".join(i.strip().lower().split()) for i in f.readlines()]
+               with open(p.split("@")[0], "r") as f:
-         load_hotwords_in_file()
+                  v = yaml.safe_load(f)
-         return load_hotwords_in_file
+                  v = load_dot_notation(v, p.split("@")[-1])
-      except Exception:
+                  return ["".join(i.strip().lower().split()) for i in v if i]
-         pass
+            load_hotwords_in_yaml_file()
+            return load_hotwords_in_yaml_file
+         else:
+            def load_hotwords_in_file():
+               with open(p, "r") as f:
+                  return ["".join(i.strip().lower().split()) for i in f.readlines()]
+            load_hotwords_in_file()
+            return load_hotwords_in_file
+      except Exception as e:
+         log(f"$HOTWORDS {p} is not a file: {e}")
 
-      hotwords = ["".join(i.lower().strip().split()) for i in p.split(",")]
+      hotwords = ["".join(i.lower().strip().split()) for i in p.split("\/\/")]
       log(f'$HOTWORDS: {hotwords}')
       def load_hotwords_as_literal():
          return hotwords