582 lines
21 KiB
Python
582 lines
21 KiB
Python
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 = []
|
|
|
|
self.input = ""
|
|
self.cursor = 0
|
|
self.last_was_enter = False
|
|
self.input_mark = None
|
|
self.kill_buffer = ""
|
|
|
|
self.events = queue.Queue()
|
|
self.streaming = False
|
|
self.stream_answer = ""
|
|
self.stream_error = None
|
|
|
|
self.scroll = 0
|
|
self.follow_stream = True
|
|
|
|
# Focus is normally the input. History focus is entered with
|
|
# Ctrl-B followed by Up/Down, tmux-style.
|
|
self.focus = "input"
|
|
self.prefix_ctrl_b = False
|
|
self.prefix_ctrl_k = False
|
|
self.history_cursor = 0
|
|
self.history_col = 0
|
|
self.history_mark = None
|
|
|
|
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)
|
|
curses.init_pair(2, curses.COLOR_GREEN, -1)
|
|
curses.init_pair(3, curses.COLOR_YELLOW, -1)
|
|
curses.init_pair(4, curses.COLOR_WHITE, -1)
|
|
curses.init_pair(5, curses.COLOR_BLACK, curses.COLOR_WHITE)
|
|
|
|
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: "
|
|
wrapped = self.wrap_text(message["content"], max(1, width - len(prefix)))
|
|
if not wrapped:
|
|
wrapped = [""]
|
|
lines.append((message["role"], prefix + wrapped[0]))
|
|
for line in wrapped[1:]:
|
|
lines.append((message["role"], " " + line))
|
|
return lines
|
|
|
|
def input_lines(self, width):
|
|
prefix = "> "
|
|
continuation = " "
|
|
first_width = max(1, width - len(prefix))
|
|
rest_width = max(1, width - len(continuation))
|
|
rendered = []
|
|
cursor_location = (0, len(prefix))
|
|
absolute = 0
|
|
|
|
for logical in self.input.split("\n"):
|
|
local_cursor = self.cursor - absolute
|
|
if local_cursor < 0:
|
|
local_cursor = -1
|
|
if local_cursor > len(logical):
|
|
local_cursor = len(logical)
|
|
|
|
first = True
|
|
remaining = logical
|
|
if remaining == "":
|
|
rendered.append((prefix if first else continuation, ""))
|
|
if local_cursor == 0:
|
|
cursor_location = (len(rendered) - 1, len(prefix))
|
|
else:
|
|
while remaining:
|
|
line_width = first_width if first else rest_width
|
|
piece = remaining[:line_width]
|
|
remaining = remaining[line_width:]
|
|
pfx = prefix if first else continuation
|
|
rendered.append((pfx, piece))
|
|
if 0 <= local_cursor <= len(piece):
|
|
cursor_location = (len(rendered) - 1, len(pfx) + local_cursor)
|
|
local_cursor = -1
|
|
elif local_cursor > len(piece):
|
|
local_cursor -= len(piece)
|
|
first = False
|
|
|
|
absolute += len(logical) + 1
|
|
|
|
if self.cursor == len(self.input) and rendered:
|
|
pfx, text = rendered[-1]
|
|
cursor_location = (len(rendered) - 1, len(pfx) + len(text))
|
|
return rendered, cursor_location
|
|
|
|
def input_visual_height(self, width):
|
|
return max(1, len(self.input_lines(width)[0]))
|
|
|
|
def display_messages(self):
|
|
if self.streaming and self.stream_answer:
|
|
return self.messages + [{"role": "assistant", "content": self.stream_answer}]
|
|
return self.messages
|
|
|
|
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), curses.A_REVERSE)
|
|
|
|
visible_messages = self.display_messages() if messages is None else messages
|
|
chat = self.chat_lines(visible_messages, max(1, width))
|
|
|
|
input_height = min(max(1, height - 4), self.input_visual_height(max(1, width)))
|
|
separator = height - 2 - input_height
|
|
chat_available = max(0, separator - 1)
|
|
max_scroll = max(0, len(chat) - chat_available)
|
|
if self.follow_stream and self.focus == "input":
|
|
self.scroll = max_scroll
|
|
else:
|
|
self.scroll = max(0, min(self.scroll, max_scroll))
|
|
|
|
start = self.scroll
|
|
for y, (role, line) in enumerate(chat[start:start + chat_available], 1):
|
|
attr = curses.color_pair(1 if role == "user" else 2) if curses.has_colors() else 0
|
|
screen.addnstr(y, 0, line, max(0, width), attr)
|
|
|
|
# Draw a cursor in history when history has focus.
|
|
if self.focus == "history" and chat:
|
|
self.history_cursor = max(0, min(self.history_cursor, len(chat) - 1))
|
|
row = self.history_cursor - start + 1
|
|
if 1 <= row < separator:
|
|
line = chat[self.history_cursor][1]
|
|
col = max(0, min(self.history_col, len(line)))
|
|
try:
|
|
screen.chgat(row, 0, min(width, max(1, len(line))), curses.A_REVERSE)
|
|
screen.move(row, min(width - 1, col))
|
|
except curses.error:
|
|
pass
|
|
|
|
screen.hline(separator, 0, curses.ACS_HLINE, width)
|
|
|
|
rendered, (cursor_row, cursor_col) = self.input_lines(max(1, width))
|
|
visible_input = rendered[-input_height:]
|
|
first_row = separator + 1
|
|
for i, (pfx, text) in enumerate(visible_input):
|
|
attr = curses.color_pair(1) if self.focus == "input" and curses.has_colors() else 0
|
|
screen.addnstr(first_row + i, 0, pfx + text, max(0, width), attr)
|
|
|
|
if status:
|
|
screen.addnstr(height - 1, 0, status, max(0, width),
|
|
curses.color_pair(3) if curses.has_colors() else 0)
|
|
|
|
if self.focus == "input":
|
|
offset = max(0, len(rendered) - input_height)
|
|
r = first_row + max(0, cursor_row - offset)
|
|
c = min(max(0, width - 1), cursor_col)
|
|
try:
|
|
screen.move(min(height - 1, r), c)
|
|
except curses.error:
|
|
pass
|
|
screen.refresh()
|
|
|
|
def should_request_title(self):
|
|
users = [m["content"].strip() for m in self.messages if m["role"] == "user"]
|
|
return len(users) >= 2 or 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):
|
|
while True:
|
|
try:
|
|
event, value = self.events.get_nowait()
|
|
except queue.Empty:
|
|
break
|
|
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
|
|
|
|
# ---------- input editor ----------
|
|
def line_col(self):
|
|
before = self.input[:self.cursor]
|
|
line = before.count("\n")
|
|
last = before.rfind("\n")
|
|
col = self.cursor if last < 0 else self.cursor - last - 1
|
|
return line, col
|
|
|
|
def cursor_from_line_col(self, line, col):
|
|
lines = self.input.split("\n")
|
|
line = max(0, min(line, len(lines) - 1))
|
|
return min(sum(len(x) + 1 for x in lines[:line]) + col, sum(len(x) + 1 for x in lines[:line]) + len(lines[line]))
|
|
|
|
def move_input(self, direction):
|
|
if direction == "left" and self.cursor > 0:
|
|
self.cursor -= 1
|
|
elif direction == "right" and self.cursor < len(self.input):
|
|
self.cursor += 1
|
|
elif direction in ("up", "down"):
|
|
line, col = self.line_col()
|
|
self.cursor = self.cursor_from_line_col(line + (1 if direction == "down" else -1), col)
|
|
|
|
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:
|
|
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 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 editor_mark_start(self):
|
|
self.input_mark = self.cursor
|
|
|
|
def editor_mark_end(self):
|
|
if self.input_mark is None:
|
|
return
|
|
a, b = sorted((self.input_mark, self.cursor))
|
|
self.kill_buffer = self.input[a:b]
|
|
self.input_mark = None
|
|
|
|
def delete_marked(self):
|
|
if self.input_mark is None:
|
|
return
|
|
a, b = sorted((self.input_mark, self.cursor))
|
|
self.kill_buffer = self.input[a:b]
|
|
self.input = self.input[:a] + self.input[b:]
|
|
self.cursor = a
|
|
self.input_mark = None
|
|
|
|
def paste(self):
|
|
if self.kill_buffer:
|
|
self.insert(self.kill_buffer)
|
|
|
|
# ---------- history focus / selection ----------
|
|
def history_focus(self, direction):
|
|
lines = self.chat_lines(self.display_messages(), max(1, self.s.getmaxyx()[1]))
|
|
if not lines:
|
|
return
|
|
self.focus = "history"
|
|
if direction == "up":
|
|
self.history_cursor = max(0, self.scroll)
|
|
else:
|
|
height, width = self.s.getmaxyx()
|
|
input_height = min(max(1, height - 4), self.input_visual_height(max(1, width)))
|
|
available = max(1, height - 3 - input_height)
|
|
self.history_cursor = min(len(lines) - 1, self.scroll + available - 1)
|
|
self.history_col = min(self.history_col, len(lines[self.history_cursor][1]))
|
|
self.follow_stream = False
|
|
|
|
def history_move(self, key):
|
|
lines = self.chat_lines(self.display_messages(), max(1, self.s.getmaxyx()[1]))
|
|
if not lines:
|
|
return
|
|
if key == curses.KEY_UP:
|
|
self.history_cursor = max(0, self.history_cursor - 1)
|
|
elif key == curses.KEY_DOWN:
|
|
self.history_cursor = min(len(lines) - 1, self.history_cursor + 1)
|
|
elif key == curses.KEY_LEFT:
|
|
self.history_col = max(0, self.history_col - 1)
|
|
elif key == curses.KEY_RIGHT:
|
|
self.history_col = min(len(lines[self.history_cursor][1]), self.history_col + 1)
|
|
elif key == curses.KEY_PPAGE:
|
|
self.history_cursor = max(0, self.history_cursor - 5)
|
|
elif key == curses.KEY_NPAGE:
|
|
self.history_cursor = min(len(lines) - 1, self.history_cursor + 5)
|
|
self.scroll = max(0, self.history_cursor - 2)
|
|
|
|
def history_mark_start(self):
|
|
self.history_mark = (self.history_cursor, self.history_col)
|
|
|
|
def history_mark_end(self):
|
|
if self.history_mark is None:
|
|
return
|
|
lines = self.chat_lines(self.display_messages(), max(1, self.s.getmaxyx()[1]))
|
|
a = self.history_mark
|
|
b = (self.history_cursor, self.history_col)
|
|
if a > b:
|
|
a, b = b, a
|
|
selected = []
|
|
for i in range(a[0], min(b[0] + 1, len(lines))):
|
|
text = lines[i][1]
|
|
if i == a[0] and i == b[0]:
|
|
selected.append(text[a[1]:b[1]])
|
|
elif i == a[0]:
|
|
selected.append(text[a[1]:])
|
|
elif i == b[0]:
|
|
selected.append(text[:b[1]])
|
|
else:
|
|
selected.append(text)
|
|
self.kill_buffer = "\n".join(selected)
|
|
self.history_mark = None
|
|
self.send_osc52(self.kill_buffer)
|
|
|
|
def send_osc52(self, text):
|
|
# OSC 52 lets compatible terminals copy the selected history text
|
|
# to the system clipboard. Unsupported terminals simply ignore it.
|
|
import base64
|
|
data = base64.b64encode(text.encode("utf-8")).decode("ascii")
|
|
try:
|
|
print(f"\033]52;c;{data}\a", end="", flush=True)
|
|
except Exception:
|
|
pass
|
|
|
|
def return_to_input(self):
|
|
self.focus = "input"
|
|
self.follow_stream = self.scroll == 0
|
|
|
|
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.focus = "input"
|
|
self.follow_stream = True
|
|
self.scroll = 0
|
|
self.start_stream()
|
|
|
|
def joe_key(self, key):
|
|
# A small, useful subset of joe's ^K commands:
|
|
# ^KB = begin block, ^KK = end/copy block, ^KC = copy,
|
|
# ^KY = delete block, ^KU = unmark, ^Y = paste.
|
|
if key in ("b", "B"):
|
|
if self.focus == "input":
|
|
self.editor_mark_start()
|
|
else:
|
|
self.history_mark_start()
|
|
return True
|
|
if key in ("k", "K"):
|
|
if self.focus == "input":
|
|
self.editor_mark_end()
|
|
else:
|
|
self.history_mark_end()
|
|
return True
|
|
if key in ("c", "C"):
|
|
if self.focus == "input" and self.input_mark is not None:
|
|
a, b = sorted((self.input_mark, self.cursor))
|
|
self.kill_buffer = self.input[a:b]
|
|
return True
|
|
if key in ("y", "Y"):
|
|
if self.focus == "input":
|
|
self.paste()
|
|
return True
|
|
if key in ("x", "X"):
|
|
if self.focus == "input":
|
|
self.delete_marked()
|
|
return True
|
|
if key in ("u", "U"):
|
|
self.input_mark = None
|
|
self.history_mark = None
|
|
return True
|
|
return False
|
|
|
|
def handle_key(self, key):
|
|
if self.prefix_ctrl_b:
|
|
self.prefix_ctrl_b = False
|
|
if key == curses.KEY_UP:
|
|
self.history_focus("up")
|
|
return
|
|
if key == curses.KEY_DOWN:
|
|
self.history_focus("down")
|
|
return
|
|
if key == "b":
|
|
self.prefix_ctrl_b = True
|
|
return
|
|
return
|
|
|
|
if key == "\x02": # Ctrl-B, tmux-style prefix
|
|
self.prefix_ctrl_b = True
|
|
return
|
|
|
|
if key in ("\x04", "\x1b"):
|
|
return "quit"
|
|
|
|
if self.focus == "history":
|
|
if key in (curses.KEY_UP, curses.KEY_DOWN, curses.KEY_LEFT, curses.KEY_RIGHT,
|
|
curses.KEY_PPAGE, curses.KEY_NPAGE):
|
|
self.history_move(key)
|
|
return
|
|
if key == "\x11": # Ctrl-Q as a convenient escape to input
|
|
self.return_to_input()
|
|
return
|
|
if key == "\x0b": # Ctrl-K
|
|
self.prefix_ctrl_k = True
|
|
return
|
|
if getattr(self, "prefix_ctrl_k", False):
|
|
self.prefix_ctrl_k = False
|
|
self.joe_key(key)
|
|
return
|
|
# Ctrl-K handling needs to happen before ordinary input.
|
|
if key == "\x0b":
|
|
self.prefix_ctrl_k = True
|
|
return
|
|
return
|
|
|
|
# Input focus.
|
|
multiline = self.input_visual_height(max(1, self.s.getmaxyx()[1])) > 1
|
|
|
|
if key == "\x0b": # Ctrl-K joe prefix
|
|
self.prefix_ctrl_k = True
|
|
return
|
|
|
|
if getattr(self, "prefix_ctrl_k", False):
|
|
self.prefix_ctrl_k = False
|
|
if self.joe_key(key):
|
|
return
|
|
|
|
# Before the input wraps to multiple visual lines, arrows scroll history.
|
|
if not multiline and key in (curses.KEY_UP, curses.KEY_DOWN,
|
|
curses.KEY_LEFT, curses.KEY_RIGHT):
|
|
if key in (curses.KEY_UP, curses.KEY_LEFT):
|
|
self.scroll = min(
|
|
max(0, len(self.chat_lines(self.display_messages(), self.s.getmaxyx()[1]))),
|
|
self.scroll + 1,
|
|
)
|
|
else:
|
|
self.scroll = max(0, self.scroll - 1)
|
|
self.follow_stream = self.scroll == 0
|
|
return
|
|
|
|
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)
|
|
self.follow_stream = self.scroll == 0
|
|
return
|
|
if key == curses.KEY_LEFT:
|
|
self.move_input("left")
|
|
return
|
|
if key == curses.KEY_RIGHT:
|
|
self.move_input("right")
|
|
return
|
|
if key == curses.KEY_UP:
|
|
self.move_input("up")
|
|
return
|
|
if key == curses.KEY_DOWN:
|
|
self.move_input("down")
|
|
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):
|
|
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()
|
|
status = "AI skriver..." if self.streaming else (f"OpenRouter-fel: {self.stream_error}" if self.stream_error else "")
|
|
if self.focus == "history":
|
|
status = "HISTORIK · Ctrl-B↓/↑ för fokusbyte · Ctrl-K B/K markerar"
|
|
elif self.prefix_ctrl_b:
|
|
status = "Ctrl-B · ↑/↓ = historik"
|
|
self.draw(status)
|
|
|
|
try:
|
|
key = self.s.get_wch()
|
|
except curses.error:
|
|
continue
|
|
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())
|