New Terminal UI

codebased/GREETING.txt

@@ -1,7 +1,3 @@
-Hello and welcome to Codebased, a suite of tools to make software development fast and painless.
 I'm Max Conradt, founder of Codebased.
-I'm creating Codebased because I love writing software and love people who write software.
 If you ever need any help, text me at +1 (913) 808-7343. For now, I will even help you write your code.
-That's my real phone number, because Codebased will never be anything without its users.
-You can also check out the Discord: https://discord.gg/CA6Jzn2S. I'll be more responsive on Discord because I don't
-look at my phone while I'm working, but am on Discord when I'm working.
+You can also check out the Discord: https://discord.gg/CA6Jzn2S.

codebased/background_worker.py

@@ -0,0 +1,44 @@
+import queue
+import threading
+import time
+from pathlib import Path
+
+from codebased.index import Dependencies, Flags, Config, index_paths
+from codebased.stats import STATS
+
+
+def background_worker(
+        dependencies: Dependencies,
+        flags: Flags,
+        config: Config,
+        shutdown_event: threading.Event,
+        event_queue: queue.Queue[Path],
+):
+    def pre_filter(event: Path) -> bool:
+        return not event.is_relative_to(config.codebased_directory) and not event.is_relative_to(config.root / '.git')
+
+    while not shutdown_event.is_set():
+        # Wait indefinitely for an event.
+        events: list[Path] = [event_queue.get()]
+        start = time.monotonic()
+        loop_timeout = .1
+        while time.monotonic() - start < loop_timeout:
+            try:
+                events.append(event_queue.get(timeout=loop_timeout))
+            except queue.Empty:
+                break
+        try:
+            while not event_queue.empty():
+                events.append(event_queue.get(block=False))
+        except queue.Empty:
+            pass
+        # Don't create events when we write to the index, especially from this thread.
+        events = [event for event in events if pre_filter(event)]
+        if not events:
+            continue
+        if shutdown_event.is_set():
+            break
+        STATS.increment("codebased.background_worker.updates.total")
+        STATS.increment("codebased.background_worker.updates.events", len(events))
+        index_paths(dependencies, config, events, total=False)
+        STATS.increment("codebased.background_worker.updates.index")

codebased/editor.py

@@ -1,39 +1,19 @@
-import curses
 import subprocess
 from pathlib import Path
 from typing import Literal
 
 
-def open_editor(editor: Literal["vi", "idea", "code"], *, file: Path, row: int, column: int):
-    if editor == "vi":
-        # Save the current terminal state
-        curses.def_prog_mode()
-
-        # Exit curses mode temporarily
-        curses.endwin()
-
-        # Clear the screen using ANSI escape sequence
-        print("\033[2J\033[H", end="", flush=True)
-
-        # Open Vim
-        subprocess.run(["vi", str(file), f"+{row}"])
+def suspends(editor: Literal["vi", "idea", "code"]) -> bool:
+    return editor == "vi"
 
-        # Clear the screen again
-        print("\033[2J\033[H", end="", flush=True)
 
-        # Restore the terminal to the state curses left it in
-        curses.reset_prog_mode()
-
-        # Refresh the screen
-        curses.doupdate()
-
-        # Force a redraw of the entire screen
-        stdscr = curses.initscr()
-        stdscr.clear()
-        stdscr.refresh()
+def open_editor(editor: Literal["vi", "idea", "code"], *, file: Path, row: int, column: int):
+    line_number = row + 1
+    if editor == "vi":
+        subprocess.run(["vi", str(file), f"+{line_number}"])
     elif editor == "idea":
-        subprocess.run(["idea", "--line", str(row), str(file)])
+        subprocess.run(["idea", "--line", str(line_number), str(file)])
     elif editor == "code":
-        subprocess.run(["code", "--goto", f"{file}:{row}:{column}"])
+        subprocess.run(["code", "--goto", f"{file}:{line_number}:{column}"])
     else:
-        raise NotImplementedError(editor)
+        raise NotImplementedError(editor)

codebased/filesystem.py

@@ -2,7 +2,6 @@
 
 import dataclasses
 import queue
-import time
 from pathlib import Path
 
 import watchdog.events
@@ -18,16 +17,34 @@ class EventWrapper:
 
 
 class QueueEventHandler(watchdog.events.FileSystemEventHandler):
-    def __init__(self, q: queue.Queue[EventWrapper]):
+    def __init__(self, q: queue.Queue[Path]):
         self.q = q
 
     def on_any_event(self, event: watchdog.events.FileSystemEvent):
-        self.q.put(EventWrapper(event, time.time()))
+        if event.is_directory:
+            return
+        for path in get_paths(event):
+            self.q.put(path)
 
 
-def get_filesystem_events_queue(root: Path) -> queue.Queue[EventWrapper]:
+def get_filesystem_events_queue(root: Path) -> queue.Queue[Path]:
     observer = _OBSERVER
     q = queue.Queue()
     observer.schedule(QueueEventHandler(q), root, recursive=True)
     observer.start()
     return q
+
+
+def get_paths(
+        event: watchdog.events.FileSystemEvent
+) -> list[Path]:
+    abs_paths: list[str] = []
+    if event.event_type == 'moved':
+        abs_paths = [event.src_path, event.dest_path]
+    elif event.event_type == 'created':
+        abs_paths = [event.src_path]
+    elif event.event_type == 'deleted':
+        abs_paths = [event.src_path]
+    elif event.event_type == 'modified':
+        abs_paths = [event.src_path]
+    return [Path(path) for path in abs_paths]