"""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 knoe.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( "<>", _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("<>", 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("", 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.knoe_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_knoe_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 knoe.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 / "knoe.cfg" else: cfg_path = self._resolve_knoe_conf_dir() / "knoe.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