prole/knoe/ui/screens/ollama.py

666 lines
25 KiB
Python

"""Ollama LLM configuration screen."""
import configparser
import os
import threading
from pathlib import Path
import tkinter as tk
from tkinter import ttk
from knoe import screen as ui
from knoe.core.ollama_scan import run_ollama_scan
from knoe.core.env import (
DEFAULT_OLLAMA_PORT,
PROJECT_ROOT,
_format_ollama_host,
_parse_ollama_host,
)
from knoe.config import _collect_cfg_vars, _expand_cfg_value
class OllamaScreenMixin:
"""Ollama LLM configuration screen."""
def _render_ollama_config_page(self):
# Letterhead at top right (matching welcome screen theme)
content_width = self.bg_canvas.winfo_width() or 975
right_margin = content_width - 48
ui.canvas_text(
self,
right_margin,
40,
"knoe.dev",
fill="#6e6e73",
font=("SF Pro Text", 32, "bold"),
anchor="ne",
)
ui.canvas_text(
self,
right_margin,
85,
"infrastructure.auto()",
fill="#6e6e73",
font=("SF Pro Text", 18),
anchor="ne",
)
self._render_title("Ollama Server", y=150)
self._render_paragraph(
"Detect and select a network Ollama server. Click Scan to discover servers on the network. "
"Select a server from the browser to save it to prole.cfg.",
y=200,
)
x_label = 48
y = 260
self._canvas_items.append(
ui.canvas_text(
self,
x_label,
y,
"Ollama Server Browser",
fill="#1d1d1f",
font=("SF Pro Text", 12, "bold"),
)
)
refresh_btn = tk.Button(
self.bg_canvas,
text="Scan",
command=self._refresh_ollama_table,
bg="#F5F5DC",
fg="black",
activebackground="#E5E5D5",
highlightbackground="#F5F5DC",
highlightthickness=0,
relief="flat",
font=("SF Pro Text", 10),
padx=10,
pady=4,
)
refresh_window = self.bg_canvas.create_window(
x_label + 190, y - 8, window=refresh_btn, anchor="nw"
)
self._canvas_items.append(refresh_window)
self._overlay_widgets.append(refresh_btn)
self._ollama_status_var = tk.StringVar(value="Ready")
status_item = ui.canvas_text(
self,
x_label + 270,
y,
"Status: Ready",
fill="#6e6e73",
font=("SF Pro Text", 10),
)
self._canvas_items.append(status_item)
def update_status_text(*args):
try:
self.bg_canvas.itemconfig(
status_item, text=f"Status: {self._ollama_status_var.get()}"
)
except Exception:
pass
self._ollama_status_var.trace_add("write", update_status_text)
# Scan console — shows real-time progress from the scan script
console_y = y + 28
console_frame = tk.Frame(
self.bg_canvas,
bg="#1e1e1e",
highlightbackground="#444",
highlightthickness=1,
)
console_window = self.bg_canvas.create_window(
x_label, console_y, window=console_frame, anchor="nw", width=900, height=100
)
self._canvas_items.append(console_window)
self._overlay_widgets.append(console_frame)
console_text = tk.Text(
console_frame,
bg="#1e1e1e",
fg="#00ff00",
font=("Menlo", 9),
relief="flat",
state="disabled",
wrap="word",
borderwidth=0,
highlightthickness=0,
insertbackground="#00ff00",
)
console_scroll = ttk.Scrollbar(
console_frame, orient="vertical", command=console_text.yview
)
console_text.configure(yscrollcommand=console_scroll.set)
console_scroll.pack(side="right", fill="y")
console_text.pack(side="left", fill="both", expand=True)
self._overlay_widgets.append(console_text)
self._overlay_widgets.append(console_scroll)
self._ollama_console = console_text
table_y = console_y + 110
table_frame = tk.Frame(
self.bg_canvas,
bg="white",
highlightbackground="#E0E0E0",
highlightthickness=1,
)
table_window = self.bg_canvas.create_window(
x_label, table_y, window=table_frame, anchor="nw", width=900, height=220
)
self._canvas_items.append(table_window)
self._overlay_widgets.append(table_frame)
# Add an editable Default Model column; available models shown for reference
columns = ("select", "host", "port", "default", "models")
tree = ttk.Treeview(table_frame, columns=columns, show="headings", height=6)
tree.heading("select", text="Select")
tree.heading("host", text="Host")
tree.heading("port", text="Port")
tree.heading("default", text="Default Model")
tree.heading("models", text="Models")
tree.column("select", width=60, anchor="center")
tree.column("host", width=200, anchor="w")
tree.column("port", width=70, anchor="center")
tree.column("default", width=180, anchor="w")
tree.column("models", width=380, anchor="w")
tree.pack(side="left", fill="both", expand=True)
scroll = ttk.Scrollbar(table_frame, orient="vertical", command=tree.yview)
tree.configure(yscrollcommand=scroll.set)
scroll.pack(side="right", fill="y")
self._overlay_widgets.append(tree)
self._overlay_widgets.append(scroll)
self._ollama_server_tree = tree
# Inline combobox to edit Default Model for the selected row
self._ollama_row_model_combo = None
def _place_row_model_combo(item_id):
try:
# Compute bbox for 'default' column
bbox = tree.bbox(item_id, column="default")
if not bbox:
return
x, y, w, h = bbox
# Create lazily
if not self._ollama_row_model_combo:
self._ollama_row_model_combo = ttk.Combobox(tree, state="readonly")
self._ollama_row_model_combo.bind(
"<<ComboboxSelected>>", _on_row_model_change
)
# Identify models by key
vals = tree.item(item_id, "values")
if not vals:
return
host = vals[1]
port = vals[2] or DEFAULT_OLLAMA_PORT
key = f"{host}:{port}"
models = self._ollama_model_options.get(key, [])
try:
self._ollama_row_model_combo.configure(values=models)
except Exception:
pass
# Set current selection
current = self._ollama_default_model_by_key.get(key) or (
models[0] if models else ""
)
if current:
self._ollama_row_model_combo.set(current)
# Place within Treeview
self._ollama_row_model_combo.place(
x=x + 1, y=y + 1, width=max(w - 2, 60), height=h - 2
)
except Exception:
pass
def _on_row_model_change(_evt=None):
try:
if not self._ollama_row_model_combo:
return
sel = tree.selection()
if not sel:
return
item_id = sel[0]
vals = tree.item(item_id, "values")
if not vals:
return
host = vals[1]
port = vals[2] or DEFAULT_OLLAMA_PORT
key = f"{host}:{port}"
choice = self._ollama_row_model_combo.get().strip()
self._ollama_default_model_by_key[key] = choice
# If this row is the active selection for host/port, reflect into self.ollama_model
current_key = self._ollama_current_key()
if current_key == key and choice:
try:
self.ollama_model.set(choice)
except Exception:
pass
# Update the table cell text
new_vals = list(vals)
# Default column index is 3
if len(new_vals) >= 4:
new_vals[3] = choice or ""
tree.item(item_id, values=new_vals)
except Exception:
pass
def on_select(event):
if getattr(self, "_refreshing_ollama_table", False):
return
if getattr(self, "_selecting_ollama_row", False):
return
self._selecting_ollama_row = True
try:
sel = tree.selection()
if not sel:
# Hide editor when nothing selected
if self._ollama_row_model_combo:
self._ollama_row_model_combo.place_forget()
return
vals = tree.item(sel[0], "values")
if not vals:
return
host = vals[1]
port = vals[2] or DEFAULT_OLLAMA_PORT
self.ollama_server_host.set(host)
self.ollama_server_port.set(port)
key = f"{host}:{port}"
# If we have a stored default model for this key, reflect into self.ollama_model for saving
chosen = self._ollama_default_model_by_key.get(key) or ""
if chosen:
try:
self.ollama_model.set(chosen)
except Exception:
pass
# Update selection indicators in-place without full re-render
current_key = key
for item_id in tree.get_children():
v = tree.item(item_id, "values")
if not v:
continue
item_key = f"{v[1]}:{v[2] or DEFAULT_OLLAMA_PORT}"
new_prefix = " [✓] " if item_key == current_key else " [ ] "
if v[0] != new_prefix:
new_vals = list(v)
new_vals[0] = new_prefix
tree.item(item_id, values=new_vals)
# Place inline model editor over this row
try:
_place_row_model_combo(sel[0])
except Exception:
pass
# Auto-save on selection
self._save_ollama_config()
finally:
self._selecting_ollama_row = False
tree.bind("<<TreeviewSelect>>", on_select)
# Also allow double-click on default cell to focus the combobox
def on_double_click(event):
if getattr(self, "_refreshing_ollama_table", False):
return
item_id = tree.identify_row(event.y)
col = tree.identify_column(event.x)
if not item_id:
return
if col in ("#4", "default"):
_place_row_model_combo(item_id)
tree.bind("<Double-1>", on_double_click)
# OLLAMA_HOST preview — read-only, updated on selection
preview_y = table_y + 232
preview_item = ui.canvas_text(
self,
x_label,
preview_y,
"OLLAMA_HOST: (not set)",
fill="#6e6e73",
font=("SF Pro Text", 10),
)
self._canvas_items.append(preview_item)
self._ollama_preview_item = preview_item
def update_preview(*_args):
host = (self.ollama_server_host.get() or "").strip()
port = (self.ollama_server_port.get() or "").strip()
url = _format_ollama_host(host, port)
text = f"OLLAMA_HOST: {url}" if url else "OLLAMA_HOST: (not set)"
try:
self.bg_canvas.itemconfig(preview_item, text=text)
except Exception:
pass
self.ollama_server_host.trace_add("write", update_preview)
self.ollama_server_port.trace_add("write", update_preview)
update_preview()
self._ollama_note_item = ui.canvas_text(
self, x_label, preview_y + 22, "", fill="#6e6e73", font=("SF Pro Text", 10)
)
self._canvas_items.append(self._ollama_note_item)
self._render_ollama_table_from_cache()
self._refresh_ollama_table()
def _ollama_current_key(self) -> str:
host = (self.ollama_server_host.get() or "").strip()
port = (self.ollama_server_port.get() or "").strip() or DEFAULT_OLLAMA_PORT
if not host:
return ""
return f"{host}:{port}"
def _set_ollama_model_values(self, models: list[str]):
combo = getattr(self, "_ollama_model_combo", None)
if not combo:
return
if getattr(self, "_refreshing_ollama_table", False):
return
values = models or []
try:
combo.configure(values=values)
except Exception:
pass
current = (self.ollama_model.get() or "").strip()
if values and (not current or current not in values):
self.ollama_model.set(values[0])
def _render_ollama_table_from_cache(self):
rows = list(getattr(self, "_ollama_servers_cache", []) or [])
notice = getattr(self, "_ollama_last_notice", "")
self._update_ollama_table(rows, notice)
def _update_ollama_table(self, rows: list[dict], notice: str = ""):
tree = getattr(self, "_ollama_server_tree", None)
if not tree:
return
self._ollama_servers_cache = rows
# Map of key -> list[str] available models
self._ollama_model_options = {}
# Map of key -> selected default model
if not hasattr(self, "_ollama_default_model_by_key"):
self._ollama_default_model_by_key = {}
self._ollama_last_notice = notice or ""
for item in tree.get_children():
tree.delete(item)
current_key = self._ollama_current_key()
for row in rows:
host = row.get("host", "")
port = row.get("port", "") or DEFAULT_OLLAMA_PORT
models = row.get("models", []) or []
key = f"{host}:{port}"
self._ollama_model_options[key] = models
# Initialize default selection if missing
if key not in self._ollama_default_model_by_key:
self._ollama_default_model_by_key[key] = models[0] if models else ""
prefix = " [✓] " if key == current_key else " [ ] "
models_preview = ", ".join(models) if models else ""
default_model = self._ollama_default_model_by_key.get(key, "") or ""
try:
tree.insert(
"",
"end",
values=(prefix, host, port, default_model, models_preview),
)
except Exception:
# Fallback if columns not updated yet
tree.insert("", "end", values=(prefix, host, port, models_preview))
# Suppress on_select while programmatically setting selection
self._selecting_ollama_row = True
try:
if current_key:
for item in tree.get_children():
vals = tree.item(item, "values")
if (
vals
and f"{vals[1]}:{vals[2] or DEFAULT_OLLAMA_PORT}" == current_key
):
tree.selection_set(item)
tree.see(item)
break
finally:
self._selecting_ollama_row = False
note_item = getattr(self, "_ollama_note_item", None)
if note_item and self.bg_canvas.winfo_exists():
msg = notice if notice else ""
self.bg_canvas.itemconfig(note_item, text=msg)
models = self._ollama_model_options.get(current_key, [])
if models:
self._set_ollama_model_values(models)
def _ollama_console_append(self, text: str):
"""Append a line to the scan console widget (thread-safe via safe_after)."""
def _append():
console = getattr(self, "_ollama_console", None)
if not console:
return
try:
console.configure(state="normal")
console.insert("end", text + "\n")
console.see("end")
console.configure(state="disabled")
except Exception:
pass
self.safe_after(_append)
def _refresh_ollama_table(self):
if getattr(self, "_refreshing_ollama_table", False):
return
self._refreshing_ollama_table = True
self._ollama_model_options = {}
if getattr(self, "_ollama_status_var", None):
self._ollama_status_var.set("Scanning...")
# Clear console
console = getattr(self, "_ollama_console", None)
if console:
try:
console.configure(state="normal")
console.delete("1.0", "end")
console.configure(state="disabled")
except Exception:
pass
self._ollama_console_append("Starting network scan...")
def worker():
notice = ""
rows: list[dict] = []
try:
rc, rows, notice = run_ollama_scan(
PROJECT_ROOT,
on_progress_line=lambda line: self._ollama_console_append(line),
)
if rc == 0:
if rows:
self.safe_after(
lambda: (
self._ollama_status_var.set("Scan complete")
if getattr(self, "_ollama_status_var", None)
else None
)
)
else:
self.safe_after(
lambda: (
self._ollama_status_var.set("No servers found")
if getattr(self, "_ollama_status_var", None)
else None
)
)
else:
self.safe_after(
lambda: (
self._ollama_status_var.set("Scan failed")
if getattr(self, "_ollama_status_var", None)
else None
)
)
except Exception as e:
notice = f"Scan error: {e}"
self._ollama_console_append(notice)
self.safe_after(
lambda: (
self._ollama_status_var.set("Scan failed")
if getattr(self, "_ollama_status_var", None)
else None
)
)
finally:
def finish():
self._refreshing_ollama_table = False
self._update_ollama_table(rows, notice)
self.safe_after(finish)
threading.Thread(target=worker, daemon=True).start()
def _save_ollama_config(self):
raw_host = (self.ollama_server_host.get() or "").strip()
raw_port = (self.ollama_server_port.get() or "").strip()
raw_model = (self.ollama_model.get() or "").strip()
host = raw_host
port = raw_port
if raw_host:
parsed_host, parsed_port = _parse_ollama_host(raw_host)
if parsed_host:
host = parsed_host
if not port and parsed_port:
port = parsed_port
if host and not port:
port = DEFAULT_OLLAMA_PORT
if host and host != raw_host:
self.ollama_server_host.set(host)
if port and port != raw_port:
self.ollama_server_port.set(port)
data = self.prole_cfg_data.setdefault("Ollama", {})
if host:
data["OLLAMA_SERVER_HOST"] = host
data["OLLAMA_SERVER_PORT"] = port
data["OLLAMA_HOST"] = _format_ollama_host(host, port)
else:
data.pop("OLLAMA_SERVER_HOST", None)
data.pop("OLLAMA_SERVER_PORT", None)
data.pop("OLLAMA_HOST", None)
if raw_model:
data["OLLAMA_MODEL"] = raw_model
else:
data.pop("OLLAMA_MODEL", None)
self._save_prole_cfg()
if getattr(self, "_ollama_status_var", None):
self._ollama_status_var.set("Saved")
def _apply_ollama_defaults(self):
env_host = (os.environ.get("OLLAMA_SERVER_HOST") or "").strip()
env_port = (os.environ.get("OLLAMA_SERVER_PORT") or "").strip()
env_model = (os.environ.get("OLLAMA_MODEL") or "").strip()
env_url = (os.environ.get("OLLAMA_HOST") or "").strip()
if env_url and not env_host:
parsed_host, parsed_port = _parse_ollama_host(env_url)
if parsed_host:
env_host = parsed_host
if parsed_port and not env_port:
env_port = parsed_port
if env_host and not self.ollama_server_host.get().strip():
self.ollama_server_host.set(env_host)
if env_port and not self.ollama_server_port.get().strip():
self.ollama_server_port.set(env_port)
if env_model and not self.ollama_model.get().strip():
self.ollama_model.set(env_model)
cfg_host, cfg_port, cfg_model = self._read_ollama_cfg_values()
if cfg_host and not self.ollama_server_host.get().strip():
self.ollama_server_host.set(cfg_host)
if cfg_port and not self.ollama_server_port.get().strip():
self.ollama_server_port.set(cfg_port)
if cfg_model and not self.ollama_model.get().strip():
self.ollama_model.set(cfg_model)
def _read_ollama_cfg_values(self) -> tuple[str, str, str]:
"""Return (host, port, model) from prole.cfg if present."""
cfg_path = None
try:
if self._cfg_path_override is not None:
cfg_path = self._cfg_path_override
if cfg_path.is_dir():
cfg_path = cfg_path / "prole.cfg"
else:
cfg_path = self._resolve_prole_conf_dir() / "prole.cfg"
except Exception:
cfg_path = None
if not cfg_path or not Path(cfg_path).exists():
return "", "", ""
cfg = configparser.ConfigParser(interpolation=None)
cfg.optionxform = str
try:
cfg.read(cfg_path)
except Exception:
return "", "", ""
cfg_vars = _collect_cfg_vars(cfg)
host_val = ""
port_val = ""
model_val = ""
host_url = ""
if cfg.has_section("Ollama"):
sec = cfg["Ollama"]
host_val = _expand_cfg_value(
sec.get("OLLAMA_SERVER_HOST", host_val), cfg_vars
).strip()
port_val = _expand_cfg_value(
sec.get("OLLAMA_SERVER_PORT", port_val), cfg_vars
).strip()
model_val = _expand_cfg_value(
sec.get("OLLAMA_MODEL", model_val), cfg_vars
).strip()
host_url = _expand_cfg_value(
sec.get("OLLAMA_HOST", host_url), cfg_vars
).strip()
if cfg.has_section("Global"):
sec = cfg["Global"]
if not host_val:
host_val = _expand_cfg_value(
sec.get("OLLAMA_SERVER_HOST", host_val), cfg_vars
).strip()
if not port_val:
port_val = _expand_cfg_value(
sec.get("OLLAMA_SERVER_PORT", port_val), cfg_vars
).strip()
if not model_val:
model_val = _expand_cfg_value(
sec.get("OLLAMA_MODEL", model_val), cfg_vars
).strip()
if not host_url:
host_url = _expand_cfg_value(
sec.get("OLLAMA_HOST", host_url), cfg_vars
).strip()
if host_url and not host_val:
parsed_host, parsed_port = _parse_ollama_host(host_url)
if parsed_host:
host_val = parsed_host
if parsed_port and not port_val:
port_val = parsed_port
return host_val, port_val, model_val