feat: CLI history autosave

Implements a thread that auto saves history

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)