From b0c3d32e3f274c2a4ae7a95113f6c6d4342454ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Fischer?= Date: Wed, 23 Sep 2026 11:44:17 +0200 Subject: [PATCH] editor replaced (jprompt) multimodal tools added --- jcode.py | 273 ++++++++++++++++++++--- jprompt.py | 643 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 887 insertions(+), 29 deletions(-) create mode 100644 jprompt.py diff --git a/jcode.py b/jcode.py index 59f739f..c79e4b2 100644 --- a/jcode.py +++ b/jcode.py @@ -11,12 +11,13 @@ import re import subprocess import sys import time +import ssl import urllib.request try: - from prompt_toolkit import prompt + from jprompt import prompt # own multiline editor (jprompt.py next to jcode.py) except ImportError: - print("ERROR: 'prompt_toolkit' not installed. Run: pip install prompt_toolkit", + print("ERROR: 'jprompt.py' not found - it must be in the same directory as jcode.py.", file=sys.stderr) sys.exit(1) from html.parser import HTMLParser @@ -26,8 +27,9 @@ from openai import OpenAI # --------------------------------------------------------------- Configuration APP_NAME = "jcode" -APP_VERSION = "0.9.3" -MODEL = "qwen3.6-35b-a3b-uncensored-hauhaucs-aggressive" +APP_VERSION = "0.10.0" +DEFAULT_MODEL = "qwen3.6-35b-a3b-uncensored-hauhaucs-aggressive" +MODEL = DEFAULT_MODEL _LM_STUDIO_URL = "http://localhost:1234/v1" _OLLAMA_URL = "http://localhost:11434/v1" @@ -40,6 +42,19 @@ def _fetch_models(url: str) -> list[str] | None: except Exception: return None +def _pick_model(models: list[str]) -> str: + """Prefers DEFAULT_MODEL (exact, then case-insensitive, then partial match, e.g. + 'publisher/' or '@q4_k_m'); falls back to the first available model.""" + if not models: return DEFAULT_MODEL + want = DEFAULT_MODEL.lower() + for match in (lambda m: m == DEFAULT_MODEL, + lambda m: m.lower() == want, + lambda m: m.lower().split("/")[-1].split("@")[0] == want, + lambda m: want in m.lower()): + for m in models: + if match(m): return m + return models[0] + SETTINGS_FILE = Path("settings.json") def _load_settings() -> dict: """Reads settings.json: providers (OpenAI/Anthropic/Gemini via OpenAI-compatible endpoints) for '/backend '.""" @@ -56,8 +71,7 @@ def _detect_backend() -> tuple[str, str, str]: (_OLLAMA_URL, "ollama", "Ollama")]: models = _fetch_models(url) if models is not None: - if models and MODEL not in models: - MODEL = models[0] + MODEL = _pick_model(models) return url, key, name return _LM_STUDIO_URL, "lm-studio", "LM Studio" @@ -76,19 +90,20 @@ def _switch_backend(name: str) -> None: key, bname = ("ollama", "Ollama") if url == _OLLAMA_URL else ("lm-studio", "LM Studio") models = _fetch_models(url) if models is None: print(f"{RED}\u2717 {bname} not reachable.{RESET}"); return - if models and MODEL not in models: MODEL = models[0] + MODEL = _pick_model(models) BASE_URL, _API_KEY, BACKEND = url, key, bname client = OpenAI(base_url=BASE_URL, api_key=_API_KEY, timeout=REQUEST_TIMEOUT, max_retries=0) print(f"{DIM}Backend: {BACKEND} ({BASE_URL}) | Model: {MODEL}{RESET}") SANDBOX = Path("code").resolve() -MAX_STEPS = 40 +MAX_STEPS = 200 CMD_TIMEOUT = 60 # seconds per command (compiling C/C++/Java takes time) CONTEXT_LIMIT = 256_000 # context window of the model (tokens) COMPACT_THRESHOLD = 250_000 # history is compacted from here on KEEP_RECENT = 8 # most recent messages kept uncompressed -REQUEST_TIMEOUT = 600 # seconds per LLM request (local models are slow) +REQUEST_TIMEOUT = 11600 # seconds per LLM request (local models are slow) RETRY_WAITS = (5, 15, 30, 60) # wait times between retry attempts +REASONING_EFFORT = None # None = backend default; else "high"|"medium"|"low"|"max"|"none" (Ollama /v1) LOG_FILE = Path("logFile.md") _REQ_NR = 0 # running number of LM Studio requests @@ -127,7 +142,10 @@ SYSTEM_PROMPT = ( "You are a precise coding agent. Your working directory is already the " "project directory. ALWAYS give paths relative to it, e.g. 'fib.py' or " "'tests/test_fib.py' - NEVER with a leading 'code/' and NEVER absolute. " - "Use the tools to read, write, list, and edit files and to run commands.\n\n" + "Use the tools to read, write, list, and edit files and to run commands. " + "Images (view_image), screenshots (take_screenshot) and PDF pages " + "(view_pdf_page) are delivered as an image in the message right after the " + "tool result - describe only what you actually see, never guess.\n\n" "Language: reply in the same language as the task (German task -> German answers, English task -> English).\n\n" "How to work:\n" "1. Locate code with search_files and read a file before you edit it. " @@ -443,6 +461,21 @@ class _TextExtractor(HTMLParser): blank = True return "\n".join(out).strip() +def _make_ssl_context() -> ssl.SSLContext: + """CA bundle for https fetches: certifi (pip install certifi) if available, + otherwise the OS/Python default. JCODE_INSECURE_SSL=1 disables verification + (e.g. behind a corporate proxy that re-signs TLS) - use with care.""" + if os.environ.get("JCODE_INSECURE_SSL") == "1": + ctx = ssl.create_default_context() + ctx.check_hostname = False; ctx.verify_mode = ssl.CERT_NONE + return ctx + try: + import certifi # noqa: PLC0415 + return ssl.create_default_context(cafile=certifi.where()) + except Exception: # noqa: BLE001 + return ssl.create_default_context() +_SSL_CONTEXT = _make_ssl_context() + def fetch_url(url: str, max_chars: int = 6000) -> str: """Loads a web page and returns the extracted text content.""" if not url.lower().startswith(("http://", "https://")): @@ -452,7 +485,7 @@ def fetch_url(url: str, max_chars: int = 6000) -> str: url, headers={"User-Agent": f"Mozilla/5.0 ({APP_NAME}/{APP_VERSION})"} ) try: - with urllib.request.urlopen(req, timeout=15) as resp: + with urllib.request.urlopen(req, timeout=15, context=_SSL_CONTEXT) as resp: ctype = resp.headers.get("Content-Type", "") raw = resp.read(2_000_000) # load at most 2 MB charset = resp.headers.get_content_charset() or "utf-8" @@ -474,6 +507,128 @@ def fetch_url(url: str, max_chars: int = 6000) -> str: text = text[:max_chars] + f"\n...[truncated, {len(text)} characters total]..." return f"Content of {url}:\n{text}" +# ----------------------------------------------------------------- Image tools +# Tool results in the OpenAI chat format are text-only. Images are therefore +# queued here and delivered to the model as a separate user message directly +# after the tool results of the current step (works with LM Studio/Ollama for +# vision models such as Qwen3-VL and Gemma). +MAX_IMAGE_SIDE = 1280 # longest edge in pixels sent to the model +MAX_IMAGES_IN_CONTEXT = 4 # older images are replaced by a text stub +IMAGE_TOKENS = 1000 # rough token estimate per image +_PENDING_IMAGES: list[tuple[str, str]] = [] # (label, data URI) + +def _pil(): + try: + from PIL import Image + return Image + except ImportError: + raise RuntimeError("package 'Pillow' not installed - run 'pip install Pillow' with run_command") + +def _attach_image(img, label: str) -> str: + """Scales a PIL image, encodes it as data URI and queues it for the next message.""" + import base64, io + w, h = img.size + if (scale := MAX_IMAGE_SIDE / max(w, h)) < 1: + img = img.resize((round(w * scale), round(h * scale)), _pil().LANCZOS) + buf = io.BytesIO() + if img.mode in ("RGBA", "LA") or "transparency" in img.info: + img.convert("RGBA").save(buf, "PNG"); mime = "image/png" + else: + img.convert("RGB").save(buf, "JPEG", quality=90); mime = "image/jpeg" + uri = f"data:{mime};base64,{base64.b64encode(buf.getvalue()).decode()}" + _PENDING_IMAGES.append((label, uri)) + return f"{label} ({w}x{h}px) attached - it follows in the next message." + +def view_image(path: str) -> str: + """Loads an image file from the sandbox (png/jpg/gif/webp/bmp/svg) for visual analysis.""" + p = _resolve(path) + if not p.is_file(): + return f"File '{path}' not found. Use list_files() to see available files." + Image = _pil() + if p.suffix.lower() == ".svg": + try: + import cairosvg # pip install cairosvg + except ImportError: + return ("ERROR: package 'cairosvg' not installed - run 'pip install cairosvg' " + "with run_command (or read the SVG source with read_file).") + import io + img = Image.open(io.BytesIO(cairosvg.svg2png(url=str(p), output_width=MAX_IMAGE_SIDE))) + else: + img = Image.open(p) + img.load() + return _attach_image(img, f"Image '{path}'") + +def take_screenshot(save_as: str = "", delay: int = 0) -> str: + """Captures the whole screen (e.g. of a running GUI program) for visual analysis.""" + Image = _pil() + time.sleep(max(0, min(int(delay), 30))) + try: + from PIL import ImageGrab + img = ImageGrab.grab() + except Exception as e: # noqa: BLE001 - e.g. Wayland: use a system screenshot tool + import shutil, tempfile + tool = next((c for c in (["grim"], ["gnome-screenshot", "-f"], ["spectacle", "-bno"]) + if shutil.which(c[0])), None) + if tool is None: + return f"ERROR: screenshot failed ({e}) and no grim/gnome-screenshot/spectacle found." + tmp = Path(tempfile.mkdtemp()) / "shot.png" + subprocess.run(tool + [str(tmp)], capture_output=True, timeout=20) + if not tmp.is_file(): + return f"ERROR: screenshot via {tool[0]} failed." + img = Image.open(tmp); img.load() + label = "Screenshot" + if save_as: + p = _resolve(save_as if Path(save_as).suffix else save_as + ".png") + p.parent.mkdir(parents=True, exist_ok=True) + img.save(p); label += f" (saved as '{p.relative_to(SANDBOX)}')" + return _attach_image(img, label) + +def view_pdf_page(path: str, page: int = 1, dpi: int = 110) -> str: + """Renders one page of a PDF from the sandbox as an image for visual analysis.""" + p = _resolve(path) + if not p.is_file(): + return f"File '{path}' not found. Use list_files() to see available files." + try: + try: + import pymupdf as fitz + except ImportError: + import fitz # older pymupdf versions + except ImportError: + return "ERROR: package 'pymupdf' not installed - run 'pip install pymupdf' with run_command." + Image = _pil() + with fitz.open(p) as doc: + n = doc.page_count + if not 1 <= page <= n: + return f"page={page} invalid - '{path}' has {n} page(s)." + pix = doc[page - 1].get_pixmap(dpi=max(50, min(int(dpi), 200))) + img = Image.frombytes("RGB", (pix.width, pix.height), pix.samples) + return _attach_image(img, f"Page {page}/{n} of '{path}'") + +def _image_message() -> dict: + """Builds the user message carrying the queued images and clears the queue.""" + labels = ", ".join(l for l, _ in _PENDING_IMAGES) + content = [{"type": "text", "text": f"[Attached from the tool result(s) above: {labels}]"}] + content += [{"type": "image_url", "image_url": {"url": uri}} for _, uri in _PENDING_IMAGES] + _PENDING_IMAGES.clear() + _log("### Images attached", labels) + return {"role": "user", "content": content} + +def _prune_images(messages: list) -> None: + """Keeps only the MAX_IMAGES_IN_CONTEXT newest images; older ones become text stubs.""" + seen = 0 + for msg in reversed(messages): + for part in reversed(msg["content"]) if isinstance(msg.get("content"), list) else (): + if part.get("type") == "image_url": + seen += 1 + if seen > MAX_IMAGES_IN_CONTEXT: + part.clear(); part.update({"type": "text", "text": "[older image removed from context]"}) + +def _content_text(content) -> str: + """Text view of a message content (image parts become '[image]').""" + if isinstance(content, list): + return " ".join("[image]" if p.get("type") == "image_url" else p.get("text", "") for p in content) + return str(content or "") + TOOL_FUNCS = { "list_files": list_files, "read_file": read_file, @@ -485,6 +640,9 @@ TOOL_FUNCS = { "run_command": run_command, "web_search": web_search, "fetch_url": fetch_url, + "view_image": view_image, + "take_screenshot": take_screenshot, + "view_pdf_page": view_pdf_page, } # Per-tool descriptions and parameter descriptions. The builder below derives @@ -575,6 +733,32 @@ _TOOL_META = { "max_chars": "Max. text length (500-20000), default 6000", }, }, + "view_image": { + "description": ( + "Shows an image file from the project (png, jpg, gif, webp, bmp, svg) " + "to you for visual analysis, e.g. a generated plot, an icon or a " + "mockup. The image arrives in the message after the tool result." + ), + "params": {"path": "Image file, e.g. 'plot.png' or 'assets/logo.svg'"}, + }, + "take_screenshot": { + "description": ( + "Captures the whole screen and shows it to you, e.g. to check a GUI " + "program you started. Start GUI programs in the background first " + "(Linux/macOS: 'python app.py > app.log 2>&1 &', Windows: " + "'start python app.py'), then take the screenshot with a delay." + ), + "params": {"save_as": "Optional file name to keep the screenshot, e.g. 'shot.png'", + "delay": "Seconds to wait before capturing (0-30), default 0"}, + }, + "view_pdf_page": { + "description": ( + "Renders ONE page of a PDF file from the project as an image and " + "shows it to you. The result states the total number of pages." + ), + "params": {"page": "Page number (1-based), default 1", + "dpi": "Render resolution (50-200), default 110"}, + }, } def _build_tool_schema(fn) -> dict: @@ -627,11 +811,12 @@ def splash_screen() -> None: print(f" Sandbox : {SANDBOX}") print(f" Max steps : {MAX_STEPS}") print(" " + "\u2500" * 50) - _submit = "Esc, Enter" if sys.platform == "darwin" else "Alt+Enter" - print(f" Describe a task (multiline, submit with {_submit}). History is kept") - print(" across tasks. '/status' shows usage, '/new' clears the session,") + print(" Describe a task: Enter submits, Ctrl+J starts a new line, Ctrl+C aborts") + print(" the input or a running task, Ctrl+D or Ctrl+C on an empty prompt exits.") + print(" History is kept across tasks. '/status' shows usage, '/new' clears the session,") print(" '/model' sets model, '/models' lists models, '/backend ' switches to") - print(" lm-studio, ollama or a settings.json provider, '/max-turns N' sets step limit.\n") + print(" lm-studio, ollama or a settings.json provider, '/max-turns N' sets step limit,") + print(" '/reasoning high|medium|low|max|none' sets thinking effort (none = off).\n") # Descriptive action label per tool call - WITHOUT code/file content. def _action_label(name: str, args: dict) -> str: @@ -664,6 +849,12 @@ def _action_label(name: str, args: dict) -> str: return f"\U0001f50d Researching \"{args.get('query', '')}\"" if name == "fetch_url": return f"\U0001f310 Loading page {args.get('url', '')}" + if name == "view_image": + return f"\U0001f5bc\ufe0f Viewing {target}" + if name == "take_screenshot": + return f"\U0001f4f8 Screenshot {args.get('save_as') or ''}".rstrip() + if name == "view_pdf_page": + return f"\U0001f4c4 Rendering {target} (page {args.get('page', 1)})" return f"\u2699\ufe0f {name}" # Compact result status - never output file content or code. @@ -724,12 +915,15 @@ LAST_USAGE_TOKENS = 0 # last real usage.total_tokens value from the server def _estimate_tokens(messages: list) -> int: """Rough token estimate (~4 chars/token) as a fallback without usage data.""" - chars = 0 + chars, images = 0, 0 for msg in messages: - chars += len(str(msg.get("content") or "")) + content = msg.get("content") + chars += len(_content_text(content)) + if isinstance(content, list): + images += sum(1 for p in content if p.get("type") == "image_url") for tc in msg.get("tool_calls") or []: chars += len(json.dumps(tc, ensure_ascii=False)) - return chars // 4 + return chars // 4 + images * IMAGE_TOKENS def _compact_cut_index(messages: list) -> int: """Determines the index from which messages are kept. @@ -758,7 +952,7 @@ def _compact_messages(messages: list) -> list: parts = [] for msg in old: role = msg.get("role", "?") - content = str(msg.get("content") or "") + content = _content_text(msg.get("content")) if role == "tool" and len(content) > 400: content = content[:400] + "...[truncated]" calls = ", ".join( @@ -877,7 +1071,7 @@ def run(task: str, messages: list) -> list: The history is preserved across tasks; the compaction therefore works session-wide automatically. Returns the updated history (compaction - creates a new list). + is done in place, so the caller's list stays the live history). """ global _REQ_NR, LAST_USAGE_TOKENS messages.append({"role": "user", "content": task}) @@ -889,18 +1083,19 @@ def run(task: str, messages: list) -> list: # response, otherwise the character heuristic. Compact on overflow. used = max(last_total_tokens, _estimate_tokens(messages)) if used > COMPACT_THRESHOLD: - messages = _compact_messages(messages) + messages[:] = _compact_messages(messages) last_total_tokens = 0 LAST_USAGE_TOKENS = 0 + _prune_images(messages) try: _REQ_NR += 1 _log(f"### Request {_REQ_NR}", f"{len(messages)} messages in context, last: " - f"{str(messages[-1].get('content') or '')[:300]}") + f"{_content_text(messages[-1].get('content'))[:300]}") resp = _chat( - model=MODEL, messages=messages, tools=TOOLS, temperature=0.2 - ) + model=MODEL, messages=messages, tools=TOOLS, temperature=0.2, + **({"reasoning_effort": REASONING_EFFORT} if REASONING_EFFORT else {})) except LLMUnavailable as e: _log("### Error", str(e)) print(f"\n{RED}\u2717 {e}{RESET}") @@ -949,13 +1144,24 @@ def run(task: str, messages: list) -> list: messages.append( {"role": "tool", "tool_call_id": tc.id, "content": str(result)} ) + if _PENDING_IMAGES: + messages.append(_image_message()) print("\u2514\u2500") print("Maximum number of steps reached.") return messages +def _discard_incomplete(messages: list) -> None: + """After Ctrl+C: drops a trailing tool-call round whose tool results are incomplete.""" + i = len(messages) + while i > 0 and messages[i - 1].get("role") == "tool": + i -= 1 + if i > 0 and messages[i - 1].get("role") == "assistant" and messages[i - 1].get("tool_calls"): + if len(messages) - i < len(messages[i - 1]["tool_calls"]): + del messages[i - 1:] + def main() -> None: - global LAST_USAGE_TOKENS, MODEL, MAX_STEPS + global LAST_USAGE_TOKENS, MODEL, MAX_STEPS, REASONING_EFFORT SANDBOX.mkdir(exist_ok=True) if (d := SETTINGS.get("default_provider")): _switch_backend(d) if len(sys.argv) > 1: @@ -967,9 +1173,12 @@ def main() -> None: session = new_session() while True: try: - task = prompt("\nTask> ", multiline=True).strip() - except (EOFError, KeyboardInterrupt): - print("\nSee you soon.") + task = prompt("\nTask>").strip() + except (EOFError, KeyboardInterrupt) as e: + if getattr(e, "text", "").strip(): # Ctrl+C with text: discard it only + print(f"{DIM}(input discarded){RESET}") + continue + print("See you soon.") # Ctrl+D, or Ctrl+C on an empty prompt break if not task: continue @@ -987,9 +1196,15 @@ def main() -> None: _switch_backend(task[8:].strip()); continue if task.lower().startswith("/model"): MODEL = task[6:].strip() or MODEL; print(f"{DIM}Current model: {MODEL}{RESET}"); continue + if task.lower().startswith("/reasoning"): + r = task[10:].strip().lower(); REASONING_EFFORT = r if r in ("high", "medium", "low", "max", "none") else (None if r in ("", "default") else REASONING_EFFORT); print(f"{DIM}Reasoning effort: {REASONING_EFFORT or 'default'}{RESET}" if r in ("high", "medium", "low", "max", "none", "", "default") else f"{RED}\u2717 Use: /reasoning high|medium|low|max|none{RESET}"); continue if task.lower().startswith("/max-turns"): MAX_STEPS = int(n) if (n := task[10:].strip()).isdigit() else MAX_STEPS; print(f"{DIM}Max steps: {MAX_STEPS}{RESET}"); continue - session = run(task, session) + try: + session = run(task, session) + except KeyboardInterrupt: + _discard_incomplete(session) + print(f"\n{RED}\u2717 Task aborted (Ctrl+C) - the session is preserved.{RESET}") if __name__ == "__main__": main() diff --git a/jprompt.py b/jprompt.py new file mode 100644 index 0000000..702cd90 --- /dev/null +++ b/jprompt.py @@ -0,0 +1,643 @@ +#!/usr/bin/env python3 +# MIT License - Copyright (c) 2026 Prof. Dr. Joern Fischer (full text: see README.md) +"""jprompt - a small, dependency-free multiline prompt editor for the terminal. + +The editor mimics the input box of Claude Code: the text starts on its own +line (below an optional header such as "Task>"), wraps at the terminal +width, scrolls vertically when it is taller than the terminal and keeps the +cursor visible at all times. The terminal is read character by character +(cbreak-like raw mode); nothing is echoed by the terminal itself. + +Keys + Enter submit the text + Ctrl+J insert a new line (Shift+Enter / Alt+Enter work too + where the terminal reports them) + Ctrl+C abort the input -> raises Abort (a KeyboardInterrupt + whose .text holds the discarded input, "" if it was empty) + Ctrl+D exit the application -> raises EOFError + Ctrl+V / Cmd+V paste. Terminals normally deliver a paste themselves + (bracketed paste is enabled so newlines in the pasted + text do NOT submit). If the terminal passes Ctrl+V + through, the system clipboard is read as a fallback. + Arrows move the cursor. Up on the first row and Down on the + last row do nothing. + Home/End, Ctrl+A/Ctrl+E, Backspace, Delete, Tab (= 4 spaces) + +Usage + from jprompt import prompt + text = prompt("Task>") # header is printed, editing starts below + +Architecture (every part can be tested without a terminal) + Buffer lines + cursor and all editing operations + Layout wraps the Buffer lines to a width, maps cursor <-> screen cell + Editor Buffer + Layout + rendering (ANSI) into any file-like object + KeyReader turns raw characters into KeyEvents (escape sequences, paste) + prompt() wires everything to the real terminal (raw mode, SIGWINCH) + +Known limitation: terminals that *reflow* old text on resize (iTerm2, +kitty, Windows Terminal) may leave a stale copy of the editor above the +freshly drawn one after a resize; the editor itself and its cursor stay +correct because every redraw is relative to the cursor position. +""" +from __future__ import annotations + +import codecs +import os +import select +import shutil +import signal +import subprocess +import sys +import unicodedata +from dataclasses import dataclass + +try: + import termios +except ImportError: # Windows + termios = None +try: + import msvcrt +except ImportError: + msvcrt = None + +__all__ = ["prompt", "Abort", "Buffer", "Layout", "Editor", "KeyReader", "KeyEvent", "RESIZE"] + +class Abort(KeyboardInterrupt): + """Raised on Ctrl+C; `.text` is the input that was being edited.""" + + def __init__(self, text: str = "") -> None: + super().__init__("input aborted") + self.text = text + +# ------------------------------------------------------------- Text helpers +def char_width(ch: str) -> int: + """Display width of one character (0 for combining marks, 2 for CJK).""" + if unicodedata.combining(ch) or unicodedata.category(ch) in ("Mn", "Me", "Cf"): + return 0 + return 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1 + +def str_width(s: str) -> int: + return sum(char_width(ch) for ch in s) + +def _clean(text: str) -> str: + """Normalises newlines, expands tabs and drops control characters.""" + text = text.replace("\r\n", "\n").replace("\r", "\n").replace("\t", " ") + return "".join(ch for ch in text if ch == "\n" or unicodedata.category(ch) != "Cc") + +# ------------------------------------------------------------------- Buffer +class Buffer: + """Editable text: a list of logical lines plus a cursor (row, col).""" + + def __init__(self, text: str = "") -> None: + self.lines: list[str] = _clean(text).split("\n") + self.row = len(self.lines) - 1 + self.col = len(self.lines[-1]) + + # -- state + @property + def text(self) -> str: + return "\n".join(self.lines) + + @property + def cursor(self) -> tuple[int, int]: + return self.row, self.col + + # -- editing + def insert(self, text: str) -> None: + parts = _clean(text).split("\n") + line = self.lines[self.row] + head, tail = line[: self.col], line[self.col :] + if len(parts) == 1: + self.lines[self.row] = head + parts[0] + tail + self.col += len(parts[0]) + return + self.lines[self.row : self.row + 1] = [head + parts[0], *parts[1:-1], parts[-1] + tail] + self.row += len(parts) - 1 + self.col = len(parts[-1]) + + def backspace(self) -> None: + if self.col > 0: + line = self.lines[self.row] + self.lines[self.row] = line[: self.col - 1] + line[self.col :] + self.col -= 1 + elif self.row > 0: + self.col = len(self.lines[self.row - 1]) + self.lines[self.row - 1] += self.lines.pop(self.row) + self.row -= 1 + + def delete(self) -> None: + line = self.lines[self.row] + if self.col < len(line): + self.lines[self.row] = line[: self.col] + line[self.col + 1 :] + elif self.row + 1 < len(self.lines): + self.lines[self.row] += self.lines.pop(self.row + 1) + + # -- horizontal movement + def left(self) -> None: + if self.col > 0: + self.col -= 1 + elif self.row > 0: + self.row -= 1 + self.col = len(self.lines[self.row]) + + def right(self) -> None: + if self.col < len(self.lines[self.row]): + self.col += 1 + elif self.row + 1 < len(self.lines): + self.row += 1 + self.col = 0 + + def home(self) -> None: + self.col = 0 + + def end(self) -> None: + self.col = len(self.lines[self.row]) + +# ------------------------------------------------------------------- Layout +@dataclass(frozen=True) +class VisualRow: + line: int # index of the logical line + start: int # first character (inclusive) + end: int # last character (exclusive) + last: bool # last visual row of its logical line? + +class Layout: + """Wraps logical lines to `width` display columns. + + A logical line of exactly `width` columns is NOT split: the cursor may sit + at display column `width` (which is why Editor uses terminal width - 1). + A cursor at the end of a non-last visual row is shown at the start of the + next row - the same convention prompt_toolkit uses. + """ + + def __init__(self, lines: list[str], width: int) -> None: + self.lines = lines + self.width = max(1, width) + self.rows: list[VisualRow] = [] + for li, line in enumerate(lines): + start = acc = 0 + for i, ch in enumerate(line): + w = char_width(ch) + if acc + w > self.width and i > start: + self.rows.append(VisualRow(li, start, i, False)) + start, acc = i, 0 + acc += w + self.rows.append(VisualRow(li, start, len(line), True)) + + def row_text(self, r: VisualRow) -> str: + return self.lines[r.line][r.start : r.end] + + def cursor_to_visual(self, row: int, col: int) -> tuple[int, int]: + """(logical row, col) -> (visual row index, display column).""" + for vr, r in enumerate(self.rows): + if r.line == row and (r.start <= col < r.end or (r.last and col == r.end)): + return vr, str_width(self.lines[row][r.start : col]) + return len(self.rows) - 1, 0 # unreachable for a consistent buffer + + def visual_to_cursor(self, vr: int, want_col: int) -> tuple[int, int]: + """(visual row index, wanted display column) -> nearest (row, col).""" + r = self.rows[vr] + line = self.lines[r.line] + limit = r.end if r.last else r.end - 1 # end of a wrapped row belongs to the next row + col, acc = r.start, 0 + while col < limit: + w = char_width(line[col]) + if acc + w > want_col: + break + acc, col = acc + w, col + 1 + return r.line, col + +# ---------------------------------------------------------------- KeyEvents +@dataclass(frozen=True) +class KeyEvent: + kind: str # text, newline, enter, backspace, delete, left, right, up, + # down, home, end, ctrl_c, ctrl_d, resize, ignore + text: str = "" # only for kind == "text" (typed char or pasted block) + +RESIZE = object() # sentinel a key source returns when the window was resized + +# ------------------------------------------------------------------- Editor +class Editor: + """Buffer + rendering. `out` needs write()/flush(); `size()` -> (cols, rows). + + The editor assumes the terminal cursor is at column 0 of the line where + the text should start when render() is first called. All later drawing is + relative to the cursor, so scrolling of the terminal does not matter. + """ + + def __init__(self, out, size, text: str = "") -> None: + self.buf = Buffer(text) + self.out = out + self._size = size + self._want_col: int | None = None # sticky column for Up/Down + self._top = 0 # first visible visual row (vertical scrolling) + self._cur_row = 0 # screen row of the cursor relative to the first drawn row + self._drawn = 0 # number of rows drawn by the last render() + self.hidden_above = 0 # visual rows scrolled out of view (shown as markers) + self.hidden_below = 0 + + # -- geometry + @property + def width(self) -> int: + """Wrap width: one less than the terminal so the cursor never touches the auto-wrap column.""" + return max(1, self._size()[0] - 1) + + @property + def height(self) -> int: + """Max. visible rows: one less than the terminal so a header line above survives.""" + return max(1, self._size()[1] - 1) + + @property + def text(self) -> str: + return self.buf.text + + def layout(self) -> Layout: + return Layout(self.buf.lines, self.width) + + # -- keys + def handle(self, key: KeyEvent) -> bool: + """Applies one key; returns True when the text was submitted.""" + k = key.kind + if k == "ctrl_c": + raise Abort(self.text) + if k == "ctrl_d": + raise EOFError + if k == "enter": + return True + if k == "text": + self.buf.insert(key.text) + elif k == "newline": + self.buf.insert("\n") + elif k == "backspace": + self.buf.backspace() + elif k == "delete": + self.buf.delete() + elif k == "left": + self.buf.left() + elif k == "right": + self.buf.right() + elif k == "home": + self.buf.home() + elif k == "end": + self.buf.end() + elif k == "up": + self.move_up() + elif k == "down": + self.move_down() + if k not in ("up", "down", "ignore"): + self._want_col = None + self.render() + return False + + def move_up(self) -> bool: + return self._vertical(-1) + + def move_down(self) -> bool: + return self._vertical(+1) + + def _vertical(self, delta: int) -> bool: + """Moves one *visual* row; returns False at the top/bottom edge.""" + lay = self.layout() + vr, vc = lay.cursor_to_visual(*self.buf.cursor) + target = vr + delta + if not 0 <= target < len(lay.rows): + return False + if self._want_col is None: + self._want_col = vc + self.buf.row, self.buf.col = lay.visual_to_cursor(target, self._want_col) + return True + + # -- drawing + def _viewport(self, vr: int, n: int) -> tuple[int, int, bool, bool]: + """Chooses the first visible row so the cursor row is on screen. + + Returns (top, rows shown, marker above?, marker below?). Each marker + line ("... N more lines ...") costs one row of the available height, + so the choice is iterated until it is stable. + """ + h, top = self.height, self._top + above = below = False + for _ in range(3): + avail = max(1, h - above - below) + if vr < top: + top = vr + elif vr >= top + avail: + top = vr - avail + 1 + top = max(0, min(top, n - avail)) + if h < 3: # no room for markers + return top, avail, False, False + new = (top > 0, top + avail < n) + if new == (above, below): + return top, avail, above, below + above, below = new + return top, avail, above, below + + def _marker(self, count: int, where: str) -> str: + text = f"\u2026 {count} more line{'s' if count != 1 else ''} {where} \u2026" + if str_width(text) > self.width: + arrow = "\u2191" if where == "above" else "\u2193" + text = f"\u2026{count}{arrow}" + return f"\x1b[2m{text}\x1b[0m" + + def render(self) -> None: + lay = self.layout() + vr, vc = lay.cursor_to_visual(*self.buf.cursor) + n = len(lay.rows) + top, avail, above, below = self._viewport(vr, n) + self._top = top + self.hidden_above, self.hidden_below = top, max(0, n - top - avail) + + lines = [lay.row_text(r) for r in lay.rows[top : top + avail]] + if above: + lines.insert(0, self._marker(self.hidden_above, "above")) + if below: + lines.append(self._marker(self.hidden_below, "below")) + crow = vr - top + int(above) + + parts = [] + if self._cur_row > 0: # back to the first drawn row + parts.append(f"\x1b[{self._cur_row}A") + parts.append("\r\x1b[J") # column 0, clear to end of screen + parts.append("\r\n".join(lines)) + back = len(lines) - 1 - crow + if back > 0: + parts.append(f"\x1b[{back}A") + parts.append("\r") + if vc > 0: + parts.append(f"\x1b[{vc}C") + self.out.write("".join(parts)) + self.out.flush() + self._cur_row, self._drawn = crow, len(lines) + + def finish(self) -> None: + """Moves the cursor below the text and starts a fresh line.""" + down = self._drawn - 1 - self._cur_row + self.out.write((f"\x1b[{down}B" if down > 0 else "") + "\r\n") + self.out.flush() + +# ---------------------------------------------------------------- KeyReader +class KeyReader: + """Decodes characters from a `source` into KeyEvents. + + source.read() blocks; returns one character or RESIZE + source.pending(timeout) True if a character arrives within `timeout` s + """ + + ESC_TIMEOUT = 0.05 + + _CTRL = {"\r": "enter", "\n": "newline", "\x7f": "backspace", "\x08": "backspace", + "\x03": "ctrl_c", "\x04": "ctrl_d", "\x01": "home", "\x05": "end"} + _CSI_FINAL = {"A": "up", "B": "down", "C": "right", "D": "left", "H": "home", "F": "end"} + _CSI_TILDE = {"1": "home", "7": "home", "4": "end", "8": "end", "3": "delete"} + + def __init__(self, source) -> None: + self.source = source + self._resized = False + + def read_key(self) -> KeyEvent: + if self._resized: + self._resized = False + return KeyEvent("resize") + ch = self.source.read() + if ch is RESIZE: + return KeyEvent("resize") + if ch == "\x1b": + return self._escape() + if ch == "\x16": # Ctrl+V passed through by the terminal + text = _clipboard() + return KeyEvent("text", text) if text else KeyEvent("ignore") + if ch == "\t": + return KeyEvent("text", " ") + if ch in self._CTRL: + return KeyEvent(self._CTRL[ch]) + if unicodedata.category(ch) != "Cc": + return KeyEvent("text", ch) + return KeyEvent("ignore") + + def _next(self) -> str: + """Next character inside an escape sequence; remembers resizes for later.""" + while True: + ch = self.source.read() + if ch is RESIZE: + self._resized = True + continue + return ch + + def _escape(self) -> KeyEvent: + if not self.source.pending(self.ESC_TIMEOUT): + return KeyEvent("ignore") # a lone Esc + c = self._next() + if c in ("\r", "\n"): + return KeyEvent("newline") # Alt+Enter + if c == "O": # SS3: cursor keys in application mode + return KeyEvent(self._CSI_FINAL.get(self._next(), "ignore")) + if c != "[": + return KeyEvent("ignore") # Alt+ + seq = "" + while True: + d = self._next() + seq += d + if "\x40" <= d <= "\x7e": + break + if len(seq) > 32: + return KeyEvent("ignore") + return self._csi(seq) + + def _csi(self, seq: str) -> KeyEvent: + final, params = seq[-1], seq[:-1] + if seq == "200~": # bracketed paste + return self._paste() + if final in self._CSI_FINAL: + return KeyEvent(self._CSI_FINAL[final]) + if final == "~": + return KeyEvent(self._CSI_TILDE.get(params.split(";")[0], "ignore")) + if final == "u" and params.split(";")[0] == "13": # kitty protocol: Shift/Ctrl+Enter + return KeyEvent("newline") + return KeyEvent("ignore") + + def _paste(self) -> KeyEvent: + end, buf = "\x1b[201~", [] + while True: + buf.append(self._next()) + if len(buf) >= len(end) and "".join(buf[-len(end):]) == end: + return KeyEvent("text", "".join(buf[: -len(end)])) + +def _clipboard() -> str: + """Best-effort read of the system clipboard (fallback for a raw Ctrl+V).""" + for cmd in (["pbpaste"], ["wl-paste", "-n"], ["xclip", "-selection", "clipboard", "-o"], + ["xsel", "-b", "-o"], ["powershell", "-NoProfile", "-Command", "Get-Clipboard"]): + if shutil.which(cmd[0]): + try: + return subprocess.run(cmd, capture_output=True, text=True, timeout=2).stdout + except Exception: # noqa: BLE001 + continue + return "" + +# -------------------------------------------------------- Terminal plumbing +class _TtySource: + """Character source for a POSIX tty; `wake_fd` (optional) signals resizes.""" + + def __init__(self, fd: int, wake_fd: int | None = None) -> None: + self.fd, self.wake = fd, wake_fd + self._dec = codecs.getincrementaldecoder("utf-8")(errors="replace") + self._buf: list[str] = [] # decoded but not yet consumed characters + + def read(self): + fds = [self.fd] if self.wake is None else [self.fd, self.wake] + while not self._buf: + ready, _, _ = select.select(fds, [], []) + if self.wake in ready: + try: + os.read(self.wake, 64) + except OSError: + pass + return RESIZE + # Read whole chunks: a large paste must be drained quickly, the + # kernel's tty input queue is small (1 KB on macOS). + b = os.read(self.fd, 65536) + if not b: + raise EOFError + self._buf.extend(self._dec.decode(b)) + return self._buf.pop(0) + + def pending(self, timeout: float) -> bool: + if self._buf: + return True + ready, _, _ = select.select([self.fd], [], [], timeout) + return bool(ready) + +class _WinSource: + """Character source for the Windows console (msvcrt); special keys become CSI sequences.""" + + _SPECIAL = {"H": "A", "P": "B", "K": "D", "M": "C", "G": "H", "O": "F", "S": "3~"} + + def __init__(self) -> None: + self._queue: list[str] = [] + + def read(self): + if self._queue: + return self._queue.pop(0) + ch = msvcrt.getwch() + if ch in ("\x00", "\xe0"): + code = self._SPECIAL.get(msvcrt.getwch()) + if code is None: + return self.read() + self._queue.extend("[" + code) + return "\x1b" + return ch + + def pending(self, timeout: float) -> bool: + return bool(self._queue) or msvcrt.kbhit() + +class _RawMode: + """Character-at-a-time input without echo; Enter stays '\\r', Ctrl+J stays '\\n'.""" + + def __init__(self, fd: int) -> None: + self.fd = fd + + def __enter__(self): + if termios is None: + return self + self.old = termios.tcgetattr(self.fd) + new = termios.tcgetattr(self.fd) + new[0] &= ~(termios.ICRNL | termios.INLCR | termios.IXON) # iflag + new[3] &= ~(termios.ECHO | termios.ICANON | termios.ISIG | termios.IEXTEN) # lflag + new[6][termios.VMIN], new[6][termios.VTIME] = 1, 0 + termios.tcsetattr(self.fd, termios.TCSANOW, new) + return self + + def __exit__(self, *exc): + if termios is not None: + termios.tcsetattr(self.fd, termios.TCSADRAIN, self.old) + +class _ResizeWatcher: + """Turns SIGWINCH into a readable byte on `self.fd` (None when unsupported).""" + + def __enter__(self): + self.fd = None + if not hasattr(signal, "SIGWINCH"): + return self + r, w = os.pipe() + try: + os.set_blocking(w, False) + self._old_wake = signal.set_wakeup_fd(w) + self._old_handler = signal.signal(signal.SIGWINCH, lambda *_: None) + self.fd, self._pipe = r, (r, w) + except (ValueError, OSError): # e.g. not in the main thread + os.close(r), os.close(w) + return self + + def __exit__(self, *exc): + if self.fd is None: + return + signal.signal(signal.SIGWINCH, self._old_handler) + signal.set_wakeup_fd(self._old_wake) + for fd in self._pipe: + os.close(fd) + +def _terminal_size() -> tuple[int, int]: + size = shutil.get_terminal_size((80, 24)) + return size.columns, size.lines + +def _enable_windows_vt() -> None: + try: + import ctypes + k32 = ctypes.windll.kernel32 + handle, mode = k32.GetStdHandle(-11), ctypes.c_uint32() + if k32.GetConsoleMode(handle, ctypes.byref(mode)): + k32.SetConsoleMode(handle, mode.value | 0x0004) # ENABLE_VIRTUAL_TERMINAL_PROCESSING + except Exception: # noqa: BLE001 + pass + +def _interactive() -> bool: + try: + return sys.stdin.isatty() and sys.stdout.isatty() and (termios or msvcrt) is not None + except (AttributeError, ValueError): + return False + +# ------------------------------------------------------------------- prompt +def prompt(header: str = "", text: str = "") -> str: + """Prints `header` (if any) and edits `text` on the following line(s). + + Returns the entered text on Enter. Raises Abort (a KeyboardInterrupt with + the current input in `.text`) on Ctrl+C and EOFError on Ctrl+D. Without a + terminal (pipe), one line is read from stdin instead. + """ + if header: + print(header, flush=True) + if not _interactive(): + line = sys.stdin.readline() + if not line: + raise EOFError + return line.rstrip("\r\n") + + out = sys.stdout + editor = Editor(out, _terminal_size, text) + if termios is None: + _enable_windows_vt() + source, fd = _WinSource(), None + else: + fd = sys.stdin.fileno() + with _RawMode(fd), _ResizeWatcher() as watcher: + if termios is not None: + source = _TtySource(fd, watcher.fd) + reader = KeyReader(source) + out.write("\x1b[?25h\x1b[?2004h") # cursor visible, bracketed paste on + try: + editor.render() + while not editor.handle(reader.read_key()): + pass + finally: + editor.finish() + out.write("\x1b[?2004l") + out.flush() + return editor.text + +if __name__ == "__main__": + try: + result = prompt("jprompt demo - Enter submits, Ctrl+J = new line, Ctrl+C aborts, Ctrl+D exits\nText>") + print(f"You entered: {result!r}") + except KeyboardInterrupt: + print("(aborted)") + except EOFError: + print("(exit)")