feat: CLI history autosave

Implements a thread that auto saves history

Author: teocns

interpreter/terminal_interface/terminal_interface.py

@@ -3,11 +3,6 @@
 If you were to build a frontend this would be a way to do it.
 """
 
-try:
-    import readline
-except ImportError:
-    pass
-
 import os
 import platform
 import random
@@ -27,21 +22,7 @@
 from .utils.find_image_path import find_image_path
 from .utils.cli_input import cli_input
 
-# Add examples to the readline history
-examples = [
-    "How many files are on my desktop?",
-    "What time is it in Seattle?",
-    "Make me a simple Pomodoro app.",
-    "Open Chrome and go to YouTube.",
-    "Can you set my system to light mode?",
-]
-random.shuffle(examples)
-try:
-    for example in examples:
-        readline.add_history(example)
-except:
-    # If they don't have readline, that's fine
-    pass
+from .utils.history import history_autosaver
 
 
 def terminal_interface(interpreter, message):
@@ -72,18 +53,14 @@ def terminal_interface(interpreter, message):
     active_block = None
     voice_subprocess = None
 
+    history = history_autosaver()  # Will stop when goes out of context
     while True:
         try:
             if interactive:
                 ### This is the primary input for Open Interpreter.
 
-                try:
-                    # This lets users hit the up arrow key for past messages
-                    readline.add_history(message)
-                except:
-                    # If the user doesn't have readline (may be the case on windows), that's fine
-                    pass
                 message = cli_input("> ", interpreter.multi_line).strip()
+                history.add(message)  # Maybe move this in cli_input?
 
         except KeyboardInterrupt:
             # Exit gracefully

interpreter/terminal_interface/utils/history.py

@@ -0,0 +1,77 @@
+import logging
+import os
+import queue
+import threading
+
+from .oi_dir import oi_dir
+
+log = logging.getLogger(__name__)
+
+
+# We probably want to decouple HistoryAutosave from oi_dir
+def history_autosaver(hist_filepath=os.path.join(oi_dir, "terminal.hist")):
+    try:
+        import readline
+
+        class HistoryAutosaver:
+            # Messages queue
+            _q: queue.Queue
+            _thread: threading.Thread
+
+            def __init__(self):
+                self._q = queue.Queue()
+                self._run()
+
+            def add(self, msg, blocking=False):
+                if blocking:
+                    readline.write_history_file(hist_filepath)
+                else:
+                    self._q.put(msg)
+
+            def _load(self):
+                try:
+                    readline.read_history_file(hist_filepath)
+                except FileNotFoundError:
+                    pass
+
+            def _run(self):
+                readline.set_auto_history(True)  # Maybe redundant
+                self._thread = threading.Thread(target=self._loop, daemon=True)
+                self._thread.start()
+
+            def _loop(self):
+                readline.read_history_file(hist_filepath)
+                while True:
+                    log.debug("Waiting for history to write")
+                    msg = self._q.get()
+                    if msg is None:
+                        break
+                    try:
+                        readline.append_history_file(1, hist_filepath)
+                    except FileNotFoundError:
+                        readline.write_history_file(hist_filepath)
+                    log.debug("History written to " + hist_filepath)
+
+            def __del__(self):
+                # Is this redundant?
+                try:
+                    log.debug("Closing history manager")
+                    self._thread.join()
+                except:
+                    pass
+
+    except ImportError:
+        log.warning("readline module not found, history autosave disabled")
+
+        class HistoryAutosaver:
+            pass
+
+    return HistoryAutosaver()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
+    handle = history_autosaver()
+    while True:
+        cmd = input("")
+        handle.add(cmd)