first commit
This commit is contained in:
Joakim
531
src/ai/tui.py
Normal file
531
src/ai/tui.py
Normal file
@@ -0,0 +1,531 @@
|
||||
import curses
|
||||
import queue
|
||||
import textwrap
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
from .config import load
|
||||
from .history import load as load_chat, save, chat_path
|
||||
from .openrouter import OpenRouter
|
||||
|
||||
|
||||
class App:
|
||||
def __init__(self, screen, chat_file):
|
||||
self.s = screen
|
||||
self.s.keypad(True)
|
||||
self.s.timeout(50)
|
||||
|
||||
self.cfg, self.instructions, self.memories = load(Path.cwd())
|
||||
self.api = OpenRouter(self.cfg["api_key"], self.cfg["model"])
|
||||
|
||||
self.path = None
|
||||
self.title = "Ny chat"
|
||||
self.messages = []
|
||||
|
||||
# Multiline editor.
|
||||
self.input = ""
|
||||
self.cursor = 0
|
||||
self.preferred_column = None
|
||||
self.last_was_enter = False
|
||||
|
||||
# Streaming worker.
|
||||
self.events = queue.Queue()
|
||||
self.streaming = False
|
||||
self.stream_answer = ""
|
||||
self.stream_error = None
|
||||
|
||||
# Viewport.
|
||||
self.scroll = 0
|
||||
self.follow_stream = True
|
||||
|
||||
if chat_file:
|
||||
path = Path(chat_file).expanduser()
|
||||
if not path.is_absolute():
|
||||
path = Path.cwd() / path
|
||||
|
||||
if path.exists():
|
||||
self.path = path
|
||||
self.title, self.messages = load_chat(path)
|
||||
|
||||
def setup_colors(self):
|
||||
if not curses.has_colors():
|
||||
return
|
||||
|
||||
curses.start_color()
|
||||
curses.use_default_colors()
|
||||
curses.init_pair(1, curses.COLOR_CYAN, -1) # You
|
||||
curses.init_pair(2, curses.COLOR_GREEN, -1) # AI
|
||||
curses.init_pair(3, curses.COLOR_YELLOW, -1) # status
|
||||
curses.init_pair(4, curses.COLOR_WHITE, -1)
|
||||
|
||||
def wrap_text(self, text, width):
|
||||
result = []
|
||||
|
||||
for paragraph in text.split("\n"):
|
||||
if paragraph == "":
|
||||
result.append("")
|
||||
else:
|
||||
result.extend(
|
||||
textwrap.wrap(
|
||||
paragraph,
|
||||
width=max(1, width),
|
||||
break_long_words=True,
|
||||
break_on_hyphens=False,
|
||||
) or [""]
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
def chat_lines(self, messages, width):
|
||||
lines = []
|
||||
|
||||
for message in messages:
|
||||
prefix = "You: " if message["role"] == "user" else "AI: "
|
||||
continuation = " "
|
||||
|
||||
first_width = max(1, width - len(prefix))
|
||||
rest_width = max(1, width - len(continuation))
|
||||
|
||||
wrapped = self.wrap_text(message["content"], first_width)
|
||||
|
||||
if not wrapped:
|
||||
wrapped = [""]
|
||||
|
||||
lines.append(
|
||||
(message["role"], prefix + wrapped[0])
|
||||
)
|
||||
|
||||
for line in wrapped[1:]:
|
||||
lines.append(
|
||||
(message["role"], continuation + line)
|
||||
)
|
||||
|
||||
return lines
|
||||
|
||||
def input_lines(self, width):
|
||||
prefix = "> "
|
||||
continuation = " "
|
||||
available_first = max(1, width - len(prefix))
|
||||
available_rest = max(1, width - len(continuation))
|
||||
|
||||
logical_lines = self.input.split("\n")
|
||||
if not logical_lines:
|
||||
logical_lines = [""]
|
||||
|
||||
rendered = []
|
||||
cursor_locations = []
|
||||
|
||||
cursor_pos = self.cursor
|
||||
absolute = 0
|
||||
|
||||
for logical in logical_lines:
|
||||
if absolute + len(logical) < cursor_pos:
|
||||
absolute += len(logical) + 1
|
||||
continue
|
||||
|
||||
local_cursor = cursor_pos - absolute
|
||||
local_cursor = max(0, min(local_cursor, len(logical)))
|
||||
|
||||
first = True
|
||||
remaining = logical
|
||||
|
||||
if remaining == "":
|
||||
rendered.append((prefix if first else continuation, ""))
|
||||
cursor_locations.append(
|
||||
(len(rendered) - 1, len(prefix) + local_cursor)
|
||||
)
|
||||
else:
|
||||
while remaining:
|
||||
width_for_line = available_first if first else available_rest
|
||||
piece = remaining[:width_for_line]
|
||||
remaining = remaining[width_for_line:]
|
||||
|
||||
rendered.append(
|
||||
(
|
||||
prefix if first else continuation,
|
||||
piece,
|
||||
)
|
||||
)
|
||||
|
||||
if first and local_cursor <= len(piece):
|
||||
cursor_locations.append(
|
||||
(len(rendered) - 1, len(prefix) + local_cursor)
|
||||
)
|
||||
local_cursor = -1
|
||||
elif not first and local_cursor >= 0:
|
||||
if local_cursor <= len(piece):
|
||||
cursor_locations.append(
|
||||
(len(rendered) - 1, len(continuation) + local_cursor)
|
||||
)
|
||||
local_cursor = -1
|
||||
else:
|
||||
local_cursor -= len(piece)
|
||||
|
||||
first = False
|
||||
|
||||
absolute += len(logical) + 1
|
||||
|
||||
if self.cursor == len(self.input) and not cursor_locations:
|
||||
# Cursor at the end.
|
||||
last = len(rendered) - 1 if rendered else 0
|
||||
text = rendered[last][1] if rendered else ""
|
||||
prefix_len = len(rendered[last][0]) if rendered else len(prefix)
|
||||
cursor_locations = [(last, prefix_len + len(text))]
|
||||
|
||||
return rendered, cursor_locations[0] if cursor_locations else (0, len(prefix))
|
||||
|
||||
def total_input_height(self, width):
|
||||
rendered, _ = self.input_lines(width)
|
||||
return max(1, len(rendered))
|
||||
|
||||
def draw(self, status="", messages=None):
|
||||
screen = self.s
|
||||
height, width = screen.getmaxyx()
|
||||
screen.erase()
|
||||
|
||||
self.setup_colors()
|
||||
|
||||
screen.addnstr(
|
||||
0,
|
||||
0,
|
||||
f" AI · {self.title} ",
|
||||
max(0, width - 1),
|
||||
curses.A_REVERSE,
|
||||
)
|
||||
|
||||
visible_messages = self.messages if messages is None else messages
|
||||
chat = self.chat_lines(visible_messages, max(1, width))
|
||||
|
||||
input_height = min(
|
||||
max(1, height - 4),
|
||||
self.total_input_height(max(1, width)),
|
||||
)
|
||||
|
||||
separator_row = height - 2 - input_height
|
||||
chat_top = 1
|
||||
chat_available = max(0, separator_row - chat_top)
|
||||
|
||||
max_scroll = max(0, len(chat) - chat_available)
|
||||
|
||||
if self.follow_stream:
|
||||
start = max_scroll
|
||||
self.scroll = max_scroll
|
||||
else:
|
||||
start = max(0, min(self.scroll, max_scroll))
|
||||
|
||||
for y, (role, line) in enumerate(
|
||||
chat[start:start + chat_available],
|
||||
chat_top,
|
||||
):
|
||||
if role == "user":
|
||||
attr = curses.color_pair(1) if curses.has_colors() else 0
|
||||
else:
|
||||
attr = curses.color_pair(2) if curses.has_colors() else 0
|
||||
|
||||
screen.addnstr(
|
||||
y,
|
||||
0,
|
||||
line,
|
||||
max(0, width),
|
||||
attr,
|
||||
)
|
||||
|
||||
screen.hline(
|
||||
separator_row,
|
||||
0,
|
||||
curses.ACS_HLINE,
|
||||
width,
|
||||
)
|
||||
|
||||
input_rendered, cursor_location = self.input_lines(max(1, width))
|
||||
|
||||
# Only show the last input lines if the editor is taller than the
|
||||
# available terminal area.
|
||||
input_rendered = input_rendered[-input_height:]
|
||||
|
||||
first_input_row = separator_row + 1
|
||||
|
||||
for index, (prefix, text) in enumerate(input_rendered):
|
||||
row = first_input_row + index
|
||||
screen.addnstr(
|
||||
row,
|
||||
0,
|
||||
prefix + text,
|
||||
max(0, width),
|
||||
)
|
||||
|
||||
# Status is shown only when there is room below the input.
|
||||
if status and height > 1:
|
||||
screen.addnstr(
|
||||
height - 1,
|
||||
0,
|
||||
status,
|
||||
max(0, width),
|
||||
curses.color_pair(3) if curses.has_colors() else 0,
|
||||
)
|
||||
|
||||
cursor_row, cursor_col = cursor_location
|
||||
visible_first = max(0, len(self.input_lines(max(1, width))[0]) - input_height)
|
||||
cursor_row -= visible_first
|
||||
cursor_row = max(0, min(input_height - 1, cursor_row))
|
||||
|
||||
screen.move(
|
||||
min(height - 1, first_input_row + cursor_row),
|
||||
min(max(0, width - 1), cursor_col),
|
||||
)
|
||||
|
||||
screen.refresh()
|
||||
|
||||
def should_request_title(self):
|
||||
users = [
|
||||
m["content"].strip()
|
||||
for m in self.messages
|
||||
if m["role"] == "user"
|
||||
]
|
||||
|
||||
if len(users) >= 2:
|
||||
return True
|
||||
|
||||
return bool(users and len(users[0]) >= 100)
|
||||
|
||||
def start_stream(self):
|
||||
self.streaming = True
|
||||
self.stream_answer = ""
|
||||
self.stream_error = None
|
||||
|
||||
messages = list(self.messages)
|
||||
context = self.instructions + self.memories
|
||||
|
||||
def worker():
|
||||
try:
|
||||
for chunk in self.api.stream(messages, context):
|
||||
self.events.put(("chunk", chunk))
|
||||
self.events.put(("done", None))
|
||||
except Exception as exc:
|
||||
self.events.put(("error", exc))
|
||||
|
||||
threading.Thread(target=worker, daemon=True).start()
|
||||
|
||||
def process_events(self):
|
||||
changed = False
|
||||
|
||||
while True:
|
||||
try:
|
||||
event, value = self.events.get_nowait()
|
||||
except queue.Empty:
|
||||
break
|
||||
|
||||
changed = True
|
||||
|
||||
if event == "chunk":
|
||||
self.stream_answer += value
|
||||
|
||||
elif event == "done":
|
||||
if self.stream_answer:
|
||||
self.messages.append(
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": self.stream_answer,
|
||||
}
|
||||
)
|
||||
|
||||
self.streaming = False
|
||||
|
||||
if self.path is None and self.should_request_title():
|
||||
try:
|
||||
title = self.api.title(self.messages)
|
||||
if title and title != "UNTITLED":
|
||||
self.title = title
|
||||
self.path = chat_path(Path.cwd(), title)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if self.path:
|
||||
save(
|
||||
self.path,
|
||||
self.title,
|
||||
self.messages,
|
||||
self.cfg["model"],
|
||||
)
|
||||
|
||||
elif event == "error":
|
||||
self.stream_error = str(value)
|
||||
self.streaming = False
|
||||
|
||||
return changed
|
||||
|
||||
def display_messages(self):
|
||||
if self.streaming and self.stream_answer:
|
||||
return self.messages + [
|
||||
{"role": "assistant", "content": self.stream_answer}
|
||||
]
|
||||
|
||||
return self.messages
|
||||
|
||||
# ---------- Multiline editor ----------
|
||||
|
||||
def line_col(self):
|
||||
before = self.input[:self.cursor]
|
||||
line = before.count("\n")
|
||||
last_newline = before.rfind("\n")
|
||||
col = self.cursor if last_newline < 0 else self.cursor - last_newline - 1
|
||||
return line, col
|
||||
|
||||
def cursor_from_line_col(self, target_line, target_col):
|
||||
lines = self.input.split("\n")
|
||||
target_line = max(0, min(target_line, len(lines) - 1))
|
||||
pos = sum(len(x) + 1 for x in lines[:target_line])
|
||||
return min(pos + target_col, pos + len(lines[target_line]))
|
||||
|
||||
def move_left(self):
|
||||
if self.cursor > 0:
|
||||
self.cursor -= 1
|
||||
|
||||
def move_right(self):
|
||||
if self.cursor < len(self.input):
|
||||
self.cursor += 1
|
||||
|
||||
def move_vertical(self, direction):
|
||||
line, col = self.line_col()
|
||||
target = line + direction
|
||||
lines = self.input.split("\n")
|
||||
|
||||
if target < 0 or target >= len(lines):
|
||||
return
|
||||
|
||||
self.cursor = self.cursor_from_line_col(target, col)
|
||||
|
||||
def home(self):
|
||||
line, _ = self.line_col()
|
||||
self.cursor = self.cursor_from_line_col(line, 0)
|
||||
|
||||
def end(self):
|
||||
line, _ = self.line_col()
|
||||
lines = self.input.split("\n")
|
||||
self.cursor = self.cursor_from_line_col(line, len(lines[line]))
|
||||
|
||||
def insert(self, text):
|
||||
self.input = self.input[:self.cursor] + text + self.input[self.cursor:]
|
||||
self.cursor += len(text)
|
||||
|
||||
def backspace(self):
|
||||
if self.cursor > 0:
|
||||
self.input = self.input[:self.cursor - 1] + self.input[self.cursor:]
|
||||
self.cursor -= 1
|
||||
|
||||
def delete(self):
|
||||
if self.cursor < len(self.input):
|
||||
self.input = self.input[:self.cursor] + self.input[self.cursor + 1:]
|
||||
|
||||
def submit(self):
|
||||
text = self.input.strip()
|
||||
|
||||
if not text:
|
||||
return
|
||||
|
||||
self.input = ""
|
||||
self.cursor = 0
|
||||
self.last_was_enter = False
|
||||
|
||||
self.messages.append({"role": "user", "content": text})
|
||||
self.follow_stream = True
|
||||
self.scroll = 0
|
||||
self.start_stream()
|
||||
|
||||
def handle_key(self, key):
|
||||
if key in ("\x04", "\x1b"):
|
||||
return "quit"
|
||||
|
||||
# Scroll the chat while AI is streaming or idle.
|
||||
if key == curses.KEY_PPAGE:
|
||||
self.follow_stream = False
|
||||
self.scroll += 5
|
||||
return
|
||||
|
||||
if key == curses.KEY_NPAGE:
|
||||
self.scroll = max(0, self.scroll - 5)
|
||||
if self.scroll == 0:
|
||||
self.follow_stream = True
|
||||
return
|
||||
|
||||
# Arrow keys are normally editor navigation. When Ctrl is used,
|
||||
# curses may expose them differently on various terminals.
|
||||
if key == curses.KEY_LEFT:
|
||||
self.move_left()
|
||||
return
|
||||
|
||||
if key == curses.KEY_RIGHT:
|
||||
self.move_right()
|
||||
return
|
||||
|
||||
if key == curses.KEY_UP:
|
||||
self.move_vertical(-1)
|
||||
return
|
||||
|
||||
if key == curses.KEY_DOWN:
|
||||
self.move_vertical(1)
|
||||
return
|
||||
|
||||
if key == curses.KEY_HOME:
|
||||
self.home()
|
||||
return
|
||||
|
||||
if key == curses.KEY_END:
|
||||
self.end()
|
||||
return
|
||||
|
||||
if key in (curses.KEY_BACKSPACE, 127, 8):
|
||||
self.backspace()
|
||||
self.last_was_enter = False
|
||||
return
|
||||
|
||||
if key == curses.KEY_DC:
|
||||
self.delete()
|
||||
self.last_was_enter = False
|
||||
return
|
||||
|
||||
if key in ("\n", "\r", curses.KEY_ENTER):
|
||||
# Double Enter sends. The first Enter inserts a newline.
|
||||
if self.last_was_enter and self.input.endswith("\n"):
|
||||
self.input = self.input[:-1]
|
||||
self.cursor = min(self.cursor, len(self.input))
|
||||
self.submit()
|
||||
else:
|
||||
self.insert("\n")
|
||||
self.last_was_enter = True
|
||||
return
|
||||
|
||||
if isinstance(key, str) and key.isprintable():
|
||||
self.insert(key)
|
||||
self.last_was_enter = False
|
||||
|
||||
def run(self):
|
||||
curses.curs_set(1)
|
||||
self.setup_colors()
|
||||
self.draw()
|
||||
|
||||
while True:
|
||||
self.process_events()
|
||||
|
||||
if self.streaming:
|
||||
status = "AI skriver..."
|
||||
elif self.stream_error:
|
||||
status = f"OpenRouter-fel: {self.stream_error}"
|
||||
else:
|
||||
status = ""
|
||||
|
||||
self.draw(status, self.display_messages())
|
||||
|
||||
key = self.s.get_wch()
|
||||
|
||||
if key == -1:
|
||||
continue
|
||||
|
||||
result = self.handle_key(key)
|
||||
|
||||
if result == "quit":
|
||||
return
|
||||
|
||||
|
||||
def run(chat_file=None):
|
||||
curses.wrapper(lambda screen: App(screen, chat_file).run())
|
||||
Reference in New Issue
Block a user