Mavis
4adde251ae
- Removed allow_custom_value=True (it was rendering as text input, not select) - Added fetch_lm_studio_models_with_value() wrapper that sets value=first model - demo.load() auto-fetches models on UI load - Removed Measure-Object bug in ps1 launcher - Removed auto-load step in ps1 launcher (user picks model explicitly)
1010 lines
41 KiB
Python
1010 lines
41 KiB
Python
"""Gradio UI для OmniSVG-Lite.
|
||
|
||
Слои:
|
||
1. UI-события → callbacks (on_generate / on_history_select)
|
||
2. callbacks → lm_client + prompts + validator + renderer + history
|
||
3. ошибки ловятся, отображаются в UI, попадают в history со status='failed'
|
||
|
||
Запуск: `python app.py` → http://127.0.0.1:7860
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import os
|
||
import time
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Any, Iterator
|
||
|
||
import gradio as gr # type: ignore # gradio is required to launch UI; tests can mock it
|
||
|
||
from history import DEFAULT_DB_PATH, History, Record
|
||
from incremental_svg import parse_to_valid
|
||
from lm_client import (
|
||
DEFAULT_BASE_URL,
|
||
DEFAULT_MODEL,
|
||
DEFAULT_TIMEOUT_S,
|
||
LMStudioUnavailable,
|
||
chat,
|
||
encode_pil_to_data_url,
|
||
stream_chat,
|
||
validate_image,
|
||
)
|
||
from prompts import build_messages, load_system_prompt
|
||
from renderer import render_png, save_png
|
||
from validator import validate_svg
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Live-режим: настройки и состояние
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Минимальный интервал между yield'ами обновления превью (в секундах).
|
||
# 150 мс ≈ 6-7 обновлений/сек на быстром стриме — глазом воспринимается
|
||
# плавно, не перегружает Gradio/GPU-рендер. Магическое число 0.15 встречается
|
||
# в этом файле именно в этом контексте; используется также в тестах.
|
||
LIVE_THROTTLE_S: float = 0.15
|
||
|
||
# Токен отмены: инкрементируется при каждом новом live-запросе. Активный
|
||
# генератор сравнивает свой токен с текущим; если не совпадает — отменяется
|
||
# (backpressure на случай, если юзер успел отправить новый запрос, пока
|
||
# старый ещё стримился).
|
||
_live_cancel_token: int = 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Логгер и настройки (env можно переопределить)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
log = logging.getLogger("omnisvg")
|
||
if not log.handlers:
|
||
logging.basicConfig(
|
||
level=os.environ.get("LOG_LEVEL", "INFO"),
|
||
format="%(asctime)s %(levelname)-7s %(name)s | %(message)s",
|
||
)
|
||
|
||
DEFAULT_PREVIEW_DIR = Path(
|
||
os.environ.get("OMNISVG_PREVIEW_DIR", str(Path.home() / ".omnisvg_lite" / "previews"))
|
||
)
|
||
PREVIEW_SIZE = {
|
||
"icon": (256, 256),
|
||
"illustration": (512, 512),
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Хелперы
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _format_ts(ts: float) -> str:
|
||
"""Превращает time.time() в человеко-читаемую дату."""
|
||
try:
|
||
return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))
|
||
except Exception: # noqa: BLE001
|
||
return str(ts)
|
||
|
||
|
||
def _check_prompt(prompt: str) -> str:
|
||
"""Возвращает очищенный промпт или бросает ValueError с понятной причиной."""
|
||
if prompt is None:
|
||
raise ValueError("промпт пустой")
|
||
cleaned = prompt.strip()
|
||
if not (1 <= len(cleaned) <= 1000):
|
||
raise ValueError("промпт должен быть от 1 до 1000 символов")
|
||
return cleaned
|
||
|
||
|
||
def _history_to_dataframe(records: list[dict[str, Any]]) -> list[list[Any]]:
|
||
"""Превращает список записей в плоский список строк для gr.Dataframe."""
|
||
rows: list[list[Any]] = []
|
||
for r in records:
|
||
rows.append(
|
||
[
|
||
r["id"],
|
||
_format_ts(r["created_at"]),
|
||
r["mode"],
|
||
r["model"],
|
||
r["n_requested"],
|
||
r["n_returned"],
|
||
r["status"],
|
||
]
|
||
)
|
||
return rows
|
||
|
||
|
||
def _refresh_history_df(limit: int = 20) -> list[list[Any]]:
|
||
with History() as h:
|
||
return _history_to_dataframe(h.list_recent(limit=limit))
|
||
|
||
|
||
# Число output-полей on_generate должно совпадать с сигнатурой ниже.
|
||
_ON_GENERATE_NOUTPUTS = 7
|
||
|
||
|
||
def _empty_result(n_candidates: int) -> tuple[list, list, str, list, str, list, str, str]:
|
||
"""Стандартный «пустой» возврат on_generate для случаев раннего выхода.
|
||
|
||
Gradio ждёт от каждого callback ровно столько значений, сколько объявлено
|
||
в `outputs=[...]`. Когда мы хотим прервать работу через `gr.Warning()` /
|
||
`gr.Error()` (а не через raise), мы ОБЯЗАНЫ вернуть плейсхолдеры для всех
|
||
outputs, иначе Gradio поднимет `IndexError`/warning.
|
||
|
||
Returns:
|
||
Кортеж из 8 элементов: ([] , [] , "" , [] , "" , [] , "" , "")
|
||
— последний "" — пустое поле reasoning_markdown.
|
||
"""
|
||
return ([], [], "", [], "", [], "", "")
|
||
|
||
|
||
def _save_all_previews(
|
||
record_id: int,
|
||
svgs: list[str],
|
||
mode: str,
|
||
) -> list[str]:
|
||
"""Рендерит PNG для каждого валидного SVG и сохраняет на диск.
|
||
|
||
Returns:
|
||
Список путей к PNG в том же порядке, что и svgs. Если рендер упал —
|
||
вместо пути идёт пустая строка.
|
||
"""
|
||
out: list[str] = []
|
||
size = PREVIEW_SIZE.get(mode, (512, 512))
|
||
for i, svg in enumerate(svgs):
|
||
png = render_png(svg, size=size)
|
||
if png is None:
|
||
log.warning("превью #%d пропущено: рендер не удался", i)
|
||
out.append("")
|
||
continue
|
||
try:
|
||
path = save_png(
|
||
png,
|
||
previews_dir=DEFAULT_PREVIEW_DIR,
|
||
record_id=record_id,
|
||
candidate_index=i,
|
||
)
|
||
out.append(str(path))
|
||
except OSError as exc:
|
||
log.warning("не удалось сохранить превью #%d: %s", i, exc)
|
||
out.append("")
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Callbacks
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def on_mode_change(mode: str) -> dict:
|
||
"""Меняет дефолт n_candidates при смене mode. По умолчанию 1 (один экземпляр)."""
|
||
return gr.update(value=1)
|
||
|
||
|
||
def fetch_lm_studio_models(base_url: str, api_key: str) -> tuple[list[str], str]:
|
||
"""Опрашивает LM Studio `/v1/models` и возвращает (список_id, статус).
|
||
|
||
Args:
|
||
base_url: например, http://127.0.0.1:1234/v1
|
||
api_key: bearer-токен (LM Studio игнорирует значение, но требует заголовок).
|
||
|
||
Returns:
|
||
(models, status_message). models — список id моделей, может быть пустым
|
||
при ошибке. status_message — текст для UI ("OK: 12 моделей" / "ошибка: ...").
|
||
"""
|
||
if not base_url or not base_url.strip():
|
||
return ([], "ошибка: пустой URL")
|
||
base = base_url.strip().rstrip("/")
|
||
# Если передали корень без /v1 — добавим
|
||
if not base.endswith("/v1"):
|
||
base = base + "/v1"
|
||
url = base + "/models"
|
||
headers = {"Authorization": f"Bearer {api_key or 'lm-studio'}"}
|
||
try:
|
||
import httpx
|
||
r = httpx.get(url, headers=headers, timeout=10)
|
||
if r.status_code != 200:
|
||
return ([], f"ошибка HTTP {r.status_code}: {r.text[:200]}")
|
||
data = r.json()
|
||
items = data.get("data") or []
|
||
ids = [str(m.get("id")) for m in items if m.get("id")]
|
||
if not ids:
|
||
return ([], "OK, но список пуст")
|
||
return (ids, f"OK: найдено {len(ids)} моделей")
|
||
except Exception as exc: # noqa: BLE001
|
||
return ([], f"ошибка: {type(exc).__name__}: {exc}")
|
||
|
||
|
||
def fetch_lm_studio_models_with_value(base_url: str, api_key: str) -> tuple[gr.update, str]:
|
||
"""Обёртка для UI: возвращает gr.update(choices=..., value=первая модель).
|
||
|
||
Использует fetch_lm_studio_models() для получения списка и устанавливает
|
||
value на ПЕРВУЮ модель из списка, чтобы Gradio не ругался, что value
|
||
не входит в choices.
|
||
"""
|
||
models, status = fetch_lm_studio_models(base_url, api_key)
|
||
if models:
|
||
return (gr.update(choices=models, value=models[0]), status)
|
||
return (gr.update(choices=[], value=None), status)
|
||
|
||
|
||
def on_generate(
|
||
prompt: str,
|
||
mode: str,
|
||
n_candidates: int,
|
||
temperature: float,
|
||
image: Any,
|
||
palette: str,
|
||
model: str,
|
||
base_url: str,
|
||
api_key: str,
|
||
) -> tuple[list[tuple[str, str]], list[dict], str, list[list[Any]], str, list[str], str]:
|
||
"""Обрабатывает клик «Сгенерировать».
|
||
|
||
Returns:
|
||
(gallery, gallery_hidden_value, svg_text_for_code, history_df,
|
||
status_md, preview_paths, error_or_status) — последняя строка для
|
||
ErrorBanner.
|
||
"""
|
||
# 1. UI pre-check
|
||
# ВАЖНО: в Gradio 5.x `gr.Warning` и `gr.Error` — это ФУНКЦИИ, а не
|
||
# исключения. Их нужно ВЫЗЫВАТЬ (а не `raise`). Проверено в Grad 5.37.0:
|
||
# `raise gr.Warning(...)` → TypeError. Если кто-то в будущем вернётся к
|
||
# `raise`, регрессионный тест test_on_generate_no_raise_on_bad_input
|
||
# в tests/test_app.py это поймает.
|
||
clean_prompt, n_candidates, mode = _common_precheck(
|
||
prompt, mode, n_candidates, image
|
||
)
|
||
if clean_prompt is None:
|
||
return _empty_result(n_candidates)
|
||
|
||
model = model or os.environ.get("DEFAULT_MODEL", DEFAULT_MODEL)
|
||
|
||
# 2. Сборка messages
|
||
image_b64: str | None = None
|
||
if image is not None:
|
||
try:
|
||
image_b64 = encode_pil_to_data_url(image, mime="image/png")
|
||
except Exception as exc: # noqa: BLE001
|
||
gr.Warning(f"не удалось закодировать изображение: {exc}")
|
||
return _empty_result(n_candidates)
|
||
|
||
palette_clean = palette.strip() if palette else None
|
||
try:
|
||
messages = build_messages(
|
||
prompt=clean_prompt,
|
||
mode=mode,
|
||
image_b64=image_b64,
|
||
palette=palette_clean,
|
||
n=n_candidates,
|
||
temperature=float(temperature),
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
log.exception("build_messages упал")
|
||
gr.Error(f"ошибка сборки промпта: {exc}")
|
||
return _empty_result(n_candidates)
|
||
|
||
# 3. Запрос в LM Studio
|
||
log.info("генерация: mode=%s n=%d temp=%.2f model=%s", mode, n_candidates, temperature, model)
|
||
try:
|
||
result = chat(
|
||
messages=messages,
|
||
model=model,
|
||
n=n_candidates,
|
||
temperature=float(temperature),
|
||
base_url=(base_url or "").strip() or DEFAULT_BASE_URL,
|
||
api_key=(api_key or "").strip() or "lm-studio",
|
||
timeout_s=float(os.environ.get("REQUEST_TIMEOUT_S", DEFAULT_TIMEOUT_S)),
|
||
)
|
||
except LMStudioUnavailable as exc:
|
||
log.error("LM Studio недоступен: %s", exc)
|
||
# Пишем в history как failed, чтобы пользователь не потерял попытку.
|
||
with History() as h:
|
||
h.add(
|
||
Record(
|
||
prompt=clean_prompt,
|
||
mode=mode,
|
||
model=model,
|
||
n_requested=n_candidates,
|
||
n_returned=0,
|
||
temperature=float(temperature),
|
||
status="failed",
|
||
error_reason=str(exc),
|
||
raw_outputs=[],
|
||
validated_outputs=[],
|
||
previews=[],
|
||
)
|
||
)
|
||
gr.Error(str(exc))
|
||
return _empty_result(n_candidates)
|
||
|
||
raw_texts = result.raw_texts
|
||
log.info("получено %d сырых ответов за %.1fs", len(raw_texts), result.elapsed_s)
|
||
|
||
# 4. Валидация + рендер
|
||
validated: list[str] = []
|
||
invalid_reasons: list[str] = []
|
||
for i, raw in enumerate(raw_texts):
|
||
ok, reason, cleaned = validate_svg(raw, mode=mode)
|
||
if ok:
|
||
validated.append(cleaned)
|
||
else:
|
||
log.warning("кандидат #%d невалиден: %s", i, reason)
|
||
invalid_reasons.append(reason)
|
||
|
||
# 5. Запись в БД (сначала insert, чтобы получить id для превью)
|
||
with History() as h:
|
||
# best_index — MVP-логика: инвертированный размер файла (меньше = лучше).
|
||
best_index: int | None = None
|
||
if validated:
|
||
sizes = [len(s.encode("utf-8")) for s in validated]
|
||
best_index = min(range(len(sizes)), key=lambda i: sizes[i])
|
||
record = Record(
|
||
prompt=clean_prompt,
|
||
mode=mode,
|
||
model=model,
|
||
n_requested=n_candidates,
|
||
n_returned=len(raw_texts),
|
||
temperature=float(temperature),
|
||
status=(
|
||
"ok" if validated and len(validated) == n_candidates
|
||
else "partial" if validated
|
||
else "failed"
|
||
),
|
||
error_reason=None if validated else (
|
||
f"все {n_candidates} кандидатов невалидны: {invalid_reasons}"
|
||
if invalid_reasons else "пустой ответ модели"
|
||
),
|
||
raw_outputs=raw_texts,
|
||
validated_outputs=validated,
|
||
previews=[],
|
||
best_index=best_index,
|
||
)
|
||
record_id = h.add(record)
|
||
|
||
# 6. Превью
|
||
preview_paths = _save_all_previews(record_id, validated, mode)
|
||
if validated and best_index is not None and preview_paths[best_index]:
|
||
pass # пометка best на уровне caption
|
||
# Дозаписываем previews в БД отдельным update-ом, чтобы не светить в Record.
|
||
if any(preview_paths):
|
||
with History() as h:
|
||
conn = h.conn
|
||
import json as _json
|
||
conn.execute(
|
||
"UPDATE generations SET previews = ? WHERE id = ?",
|
||
(_json.dumps(preview_paths, ensure_ascii=False), record_id),
|
||
)
|
||
conn.commit()
|
||
|
||
# 7. Gallery и code-block
|
||
captions: list[tuple[str, str]] = []
|
||
for i, p in enumerate(preview_paths):
|
||
if not p:
|
||
continue
|
||
if best_index is not None and i == best_index:
|
||
captions.append((p, f"★ best — #{i+1}"))
|
||
else:
|
||
captions.append((p, f"#{i+1}"))
|
||
if not captions:
|
||
captions = [] # gallery пустой
|
||
|
||
best_svg = (
|
||
validated[best_index] if (validated and best_index is not None) else ""
|
||
)
|
||
|
||
status_md = (
|
||
f"Сгенерировано {len(validated)}/{n_candidates} за {result.elapsed_s:.1f}с"
|
||
if validated
|
||
else f"Все {n_candidates} кандидатов невалидны. Подробности в history."
|
||
)
|
||
return (
|
||
captions, # gallery
|
||
captions, # hidden (для совместимости, не используется)
|
||
best_svg, # svg code block
|
||
_refresh_history_df(20), # history dataframe
|
||
status_md, # status markdown
|
||
preview_paths, # превью для архива
|
||
"", # error placeholder
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Live-режим: streaming callback
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _save_live_png(png: bytes, path: Path) -> str:
|
||
"""Сохраняет live-preview PNG в `path` (перезаписывает). Возвращает строку-путь или ""."""
|
||
try:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_bytes(png)
|
||
return str(path)
|
||
except OSError as exc:
|
||
log.warning("не удалось сохранить live-превью %s: %s", path, exc)
|
||
return ""
|
||
|
||
|
||
def _common_precheck(
|
||
prompt: str,
|
||
mode: str,
|
||
n_candidates: int,
|
||
image: Any,
|
||
) -> tuple[str | None, int, str]:
|
||
"""Общая валидация входных данных для on_generate / on_generate_live.
|
||
|
||
Возвращает (clean_prompt_or_None, normalized_n_candidates, mode) или
|
||
(None, n_candidates, mode) если валидация упала. При падении колбэк
|
||
УЖЕ вызвал gr.Warning — этого достаточно для UI, возвращаемое значение
|
||
нужно просто чтобы корректно выдать _empty_result.
|
||
"""
|
||
try:
|
||
clean_prompt = _check_prompt(prompt)
|
||
except ValueError as exc:
|
||
gr.Warning(str(exc))
|
||
return (None, n_candidates, mode)
|
||
|
||
if image is not None:
|
||
try:
|
||
validate_image(image)
|
||
except Exception as exc: # noqa: BLE001
|
||
gr.Warning(f"изображение отклонено: {exc}")
|
||
return (None, n_candidates, mode)
|
||
|
||
n_candidates = int(n_candidates)
|
||
if not (1 <= n_candidates <= 8):
|
||
gr.Warning("n_candidates должен быть от 1 до 8")
|
||
return (None, n_candidates, mode)
|
||
if mode not in ("icon", "illustration"):
|
||
gr.Warning(f"неизвестный mode: {mode!r}")
|
||
return (None, n_candidates, mode)
|
||
return (clean_prompt, n_candidates, mode)
|
||
|
||
|
||
def on_generate_live(
|
||
prompt: str,
|
||
mode: str,
|
||
n_candidates: int,
|
||
temperature: float,
|
||
image: Any,
|
||
palette: str,
|
||
model: str,
|
||
base_url: str,
|
||
api_key: str,
|
||
use_live: bool,
|
||
) -> Iterator[tuple]:
|
||
"""Генерирует SVG в live-режиме: стримит токены, обновляет PNG-превью.
|
||
|
||
Args:
|
||
... (те же параметры, что у on_generate)
|
||
use_live: если False — fallback на синхронный on_generate
|
||
(один yield с финальным результатом).
|
||
|
||
Yields:
|
||
Кортежи из 7 элементов (gallery, gallery_state, svg_viewer,
|
||
history_df, status_md, preview_paths, error) — столько же, сколько
|
||
`outputs=[...]` в build_ui(). Между дельта-апдейтами Gallery
|
||
заполняется текущим live-превью; после end-event — финальный
|
||
результат, запись в history, итоговый status.
|
||
|
||
Особенности:
|
||
- Live-режим всегда работает с n=1 (OpenAI не поддерживает n>1 в
|
||
стриме). Если юзер передал n>1, мы тихо понижаем до 1.
|
||
- Throttle: между yield'ами — не менее LIVE_THROTTLE_S (0.15s).
|
||
Это ~6-7 обновлений/сек на быстром стриме.
|
||
- Финальный yield после end-event — ВСЕГДА (даже если throttle
|
||
скипнул последний промежуточный).
|
||
- Backpressure: при новом клике старый стрим отменяется по
|
||
токену `_live_cancel_token`.
|
||
- Ошибка рендера промежуточного SVG (render_png -> None) — это
|
||
нормально, мы её скипаем и продолжаем накапливать буфер.
|
||
"""
|
||
if not use_live:
|
||
# Fallback на синхронный путь — для совместимости со старым
|
||
# контрактом и для случая, когда юзер явно выключил live-стрим.
|
||
result = on_generate(
|
||
prompt, mode, n_candidates, temperature, image, palette,
|
||
model, base_url, api_key,
|
||
)
|
||
yield result
|
||
return
|
||
|
||
# Pre-check (общий с on_generate).
|
||
clean_prompt, n_candidates, mode = _common_precheck(prompt, mode, n_candidates, image)
|
||
if clean_prompt is None:
|
||
yield _empty_result(n_candidates)
|
||
return
|
||
|
||
# Подготовка аргументов для chat / stream_chat.
|
||
model = model or os.environ.get("DEFAULT_MODEL", DEFAULT_MODEL)
|
||
image_b64: str | None = None
|
||
if image is not None:
|
||
try:
|
||
image_b64 = encode_pil_to_data_url(image, mime="image/png")
|
||
except Exception as exc: # noqa: BLE001
|
||
gr.Warning(f"не удалось закодировать изображение: {exc}")
|
||
yield _empty_result(n_candidates)
|
||
return
|
||
|
||
palette_clean = palette.strip() if palette else None
|
||
try:
|
||
messages = build_messages(
|
||
prompt=clean_prompt,
|
||
mode=mode,
|
||
image_b64=image_b64,
|
||
palette=palette_clean,
|
||
n=n_candidates,
|
||
temperature=float(temperature),
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
log.exception("build_messages упал")
|
||
gr.Error(f"ошибка сборки промпта: {exc}")
|
||
yield _empty_result(n_candidates)
|
||
return
|
||
|
||
# Live всегда n=1. Запросили n>1 — понижаем.
|
||
if n_candidates > 1:
|
||
log.warning(
|
||
"on_generate_live: n=%d запрошено, но в stream режиме идём с n=1",
|
||
n_candidates,
|
||
)
|
||
|
||
# Захватываем токен отмены. Если до end-event текущий глобальный
|
||
# токен изменится (пришёл новый запрос) — мы тихо сворачиваемся.
|
||
global _live_cancel_token
|
||
my_token = _live_cancel_token + 1
|
||
_live_cancel_token = my_token
|
||
|
||
log.info("live-генерация: mode=%s temp=%.2f model=%s", mode, temperature, model)
|
||
started = time.monotonic()
|
||
session_id = uuid.uuid4().hex[:8]
|
||
live_path = DEFAULT_PREVIEW_DIR / f"live_{session_id}.png"
|
||
size = PREVIEW_SIZE.get(mode, (256, 256))
|
||
|
||
buffer = ""
|
||
last_update_ts = -1.0 # -1 → первый yield пропускает throttle-проверку
|
||
last_valid_svg = ""
|
||
tokens = 0
|
||
final_model = model
|
||
final_finish_reason = ""
|
||
# qwen3.5 шлёт рассуждения в отдельном поле `reasoning_content`; копим
|
||
# их параллельно с `content` (буфер для SVG) и рисуем в reasoning_md.
|
||
reasoning_text = ""
|
||
|
||
# Первый yield — статус "поехали". Это нужно, чтобы UI сразу сменил
|
||
# "Сгенерировано" прошлой генерации на "Live-стрим запущен…".
|
||
yield (
|
||
[],
|
||
[],
|
||
"",
|
||
gr.update(),
|
||
"Live-стрим запущен…",
|
||
[],
|
||
"",
|
||
"",
|
||
)
|
||
|
||
try:
|
||
events = stream_chat(
|
||
messages=messages,
|
||
model=model,
|
||
n=1,
|
||
temperature=float(temperature),
|
||
base_url=(base_url or "").strip() or DEFAULT_BASE_URL,
|
||
api_key=(api_key or "").strip() or "lm-studio",
|
||
timeout_s=float(os.environ.get("REQUEST_TIMEOUT_S", DEFAULT_TIMEOUT_S)),
|
||
)
|
||
for event in events:
|
||
# Backpressure: если юзер успел отправить новый запрос,
|
||
# глобальный токен уже не наш — выходим.
|
||
if my_token != _live_cancel_token:
|
||
log.info("live-стрим отменён: пришёл более новый запрос")
|
||
return
|
||
|
||
if event.type == "delta":
|
||
# Reasoning-токены (qwen3.5) приходят в `event.reasoning` —
|
||
# копим и обновляем reasoning_md ниже. Контент для SVG
|
||
# идёт в `event.content`. Если оба пустые — игнор.
|
||
had_anything = False
|
||
if event.content:
|
||
buffer += event.content
|
||
tokens += 1
|
||
had_anything = True
|
||
if event.reasoning:
|
||
reasoning_text += event.reasoning
|
||
had_anything = True
|
||
if not had_anything:
|
||
continue
|
||
# Throttle: не чаще одного render-yield на LIVE_THROTTLE_S.
|
||
# Reasoning-обновления пропускаем через throttle наравне с
|
||
# render-yield'ами — иначе UI будет перерисовываться слишком
|
||
# часто (qwen3.5 шлёт 10-20 reasoning-токенов в секунду).
|
||
now = time.monotonic()
|
||
if now - last_update_ts < LIVE_THROTTLE_S:
|
||
continue
|
||
# Парсим + рендерим. render_png может вернуть None на
|
||
# частично валидном SVG — это нормально, пропускаем render-yield,
|
||
# но markdown с reasoning всё равно отдадим, чтобы юзер видел
|
||
# что модель думает.
|
||
svg = parse_to_valid(buffer) if buffer else ""
|
||
if svg:
|
||
last_valid_svg = svg
|
||
png = None
|
||
path_str = ""
|
||
if svg:
|
||
png = render_png(svg, size=size)
|
||
if png is not None:
|
||
path_str = _save_live_png(png, live_path)
|
||
# Готовим caption/gallery только если есть PNG; иначе отдаём
|
||
# пустую галерею, чтобы НЕ стирать предыдущий кадр (Gradio
|
||
# воспринимает [] как "очистить", а нам это не нужно).
|
||
if path_str:
|
||
last_update_ts = now
|
||
caption = (path_str, f"live · {tokens} tok")
|
||
yield (
|
||
[caption],
|
||
[caption],
|
||
svg,
|
||
gr.update(), # history_df не трогаем до финала
|
||
f"Live-стрим: ~{tokens} токенов",
|
||
[path_str],
|
||
"",
|
||
reasoning_text,
|
||
)
|
||
elif reasoning_text:
|
||
# Reasoning идёт, а валидного SVG ещё нет — обновим только
|
||
# reasoning_md, остальные outputs не трогаем.
|
||
yield (
|
||
gr.update(),
|
||
gr.update(),
|
||
gr.update(),
|
||
gr.update(),
|
||
f"Live-стрим: ~{tokens} токенов",
|
||
gr.update(),
|
||
gr.update(),
|
||
reasoning_text,
|
||
)
|
||
elif event.type == "end":
|
||
final_model = event.model or model
|
||
final_finish_reason = event.finish_reason
|
||
break
|
||
except LMStudioUnavailable as exc:
|
||
log.error("LM Studio недоступен в live-режиме: %s", exc)
|
||
with History() as h:
|
||
h.add(
|
||
Record(
|
||
prompt=clean_prompt,
|
||
mode=mode,
|
||
model=model,
|
||
n_requested=n_candidates,
|
||
n_returned=0,
|
||
temperature=float(temperature),
|
||
status="failed",
|
||
error_reason=str(exc),
|
||
raw_outputs=[],
|
||
validated_outputs=[],
|
||
previews=[],
|
||
)
|
||
)
|
||
gr.Error(str(exc))
|
||
yield _empty_result(n_candidates)
|
||
return
|
||
|
||
# ----- ФИНАЛ -----
|
||
# Парсим финальный буфер и сохраняем запись. Этот yield ВСЕГДА
|
||
# выполняется (даже если throttle скипнул последний промежуточный).
|
||
elapsed = time.monotonic() - started
|
||
final_svg = parse_to_valid(buffer) if buffer else ""
|
||
if final_svg:
|
||
last_valid_svg = final_svg
|
||
|
||
# Пытаемся валидировать (как в не-live режиме), чтобы финальный preview
|
||
# был идентичен не-live пути. Если не вышло — fallback на parse_to_valid.
|
||
validated: list[str] = []
|
||
if buffer:
|
||
ok, reason, cleaned = validate_svg(buffer, mode=mode)
|
||
if ok and cleaned:
|
||
validated = [cleaned]
|
||
final_svg = cleaned
|
||
else:
|
||
log.warning("live-финал не прошёл validate_svg: %s", reason)
|
||
# Не валидно по строгим правилам, но parse_to_valid дал что-то
|
||
# рендерабельное — оставляем его, в history пометим "partial".
|
||
if last_valid_svg:
|
||
validated = [last_valid_svg]
|
||
|
||
status_str = "ok" if validated else "failed"
|
||
error_reason: str | None = None if validated else (
|
||
"live-стрим завершён, но SVG не прошёл валидацию" if buffer
|
||
else "пустой ответ модели"
|
||
)
|
||
|
||
with History() as h:
|
||
record = Record(
|
||
prompt=clean_prompt,
|
||
mode=mode,
|
||
model=final_model,
|
||
n_requested=n_candidates,
|
||
n_returned=1,
|
||
temperature=float(temperature),
|
||
status=status_str,
|
||
error_reason=error_reason,
|
||
raw_outputs=[buffer] if buffer else [],
|
||
validated_outputs=validated,
|
||
previews=[],
|
||
best_index=0 if validated else None,
|
||
)
|
||
record_id = h.add(record)
|
||
|
||
# Финальный PNG: сохраняем с привязкой к record_id, чтобы он попал
|
||
# в историю.
|
||
preview_paths: list[str] = []
|
||
if validated:
|
||
png = render_png(validated[0], size=size)
|
||
if png is not None:
|
||
try:
|
||
final_path = save_png(
|
||
png,
|
||
previews_dir=DEFAULT_PREVIEW_DIR,
|
||
record_id=record_id,
|
||
candidate_index=0,
|
||
)
|
||
preview_paths = [str(final_path)]
|
||
with History() as h:
|
||
conn = h.conn
|
||
import json as _json
|
||
conn.execute(
|
||
"UPDATE generations SET previews = ? WHERE id = ?",
|
||
(_json.dumps(preview_paths, ensure_ascii=False), record_id),
|
||
)
|
||
conn.commit()
|
||
except OSError as exc:
|
||
log.warning("не удалось сохранить финальный live-превью: %s", exc)
|
||
|
||
captions: list[tuple[str, str]] = []
|
||
if preview_paths:
|
||
captions = [(preview_paths[0], "★ best — #1")]
|
||
|
||
if validated:
|
||
status_md = f"Сгенерировано {len(validated)}/1 за {elapsed:.1f}с"
|
||
else:
|
||
status_md = f"Live-стрим завершён без валидного SVG за {elapsed:.1f}с"
|
||
|
||
yield (
|
||
captions, # gallery
|
||
captions, # gallery_state
|
||
validated[0] if validated else "", # svg_viewer
|
||
_refresh_history_df(20), # history_df refreshed
|
||
status_md, # status
|
||
preview_paths, # превью для архива
|
||
"", # error
|
||
reasoning_text, # reasoning_md
|
||
)
|
||
|
||
|
||
def on_history_select(
|
||
evt: gr.SelectData,
|
||
history_data: list[list[Any]] | None,
|
||
) -> tuple[list[tuple[str, str]], str, str]:
|
||
"""По клику на строку history подгружает детали записи.
|
||
|
||
Args:
|
||
evt: событие выбора (index, row_payload).
|
||
history_data: текущее содержимое dataframe (для поиска id).
|
||
"""
|
||
if evt is None or not history_data:
|
||
return [], "", ""
|
||
|
||
# В новых версиях Gradio evt.value может быть словарём строки, в старых —
|
||
# индексом. Поддерживаем оба варианта.
|
||
row_idx = None
|
||
if isinstance(evt.index, (list, tuple)) and evt.index:
|
||
row_idx = evt.index[0]
|
||
elif isinstance(evt.index, int):
|
||
row_idx = evt.index
|
||
if row_idx is None or row_idx < 0 or row_idx >= len(history_data):
|
||
return [], "", ""
|
||
|
||
row = history_data[row_idx]
|
||
try:
|
||
record_id = int(row[0])
|
||
except (TypeError, ValueError):
|
||
return [], "", "не удалось извлечь id записи"
|
||
|
||
with History() as h:
|
||
rec = h.get(record_id)
|
||
if rec is None:
|
||
return [], "", f"запись #{record_id} не найдена"
|
||
|
||
previews = rec.get("previews") or []
|
||
captions: list[tuple[str, str]] = []
|
||
for i, p in enumerate(previews):
|
||
if p and Path(p).is_file():
|
||
captions.append((p, f"#{i+1}"))
|
||
validated = rec.get("validated_outputs") or []
|
||
best_idx = rec.get("best_index")
|
||
best_svg = validated[best_idx] if (best_idx is not None and 0 <= best_idx < len(validated)) else (
|
||
validated[0] if validated else ""
|
||
)
|
||
details_md = (
|
||
f"### Запись #{rec['id']}\n"
|
||
f"- **Промпт:** {rec['prompt']}\n"
|
||
f"- **Mode:** {rec['mode']}\n"
|
||
f"- **Model:** {rec['model']}\n"
|
||
f"- **N:** {rec['n_returned']}/{rec['n_requested']}\n"
|
||
f"- **Status:** {rec['status']}"
|
||
)
|
||
return captions, best_svg, details_md
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Сборка UI
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def build_ui() -> gr.Blocks:
|
||
"""Создаёт объект Gradio Blocks."""
|
||
with gr.Blocks(title="OmniSVG-Lite") as demo:
|
||
gr.Markdown(
|
||
"# OmniSVG-Lite\n"
|
||
"Текст/картинка → N SVG-кандидатов через LM Studio."
|
||
)
|
||
|
||
with gr.Row():
|
||
with gr.Column(scale=1):
|
||
prompt_tb = gr.Textbox(
|
||
label="Промпт",
|
||
placeholder="filled magnifying glass",
|
||
lines=4,
|
||
max_lines=8,
|
||
)
|
||
image_in = gr.Image(
|
||
label="Референс-картинка (опц.)",
|
||
type="pil",
|
||
sources=["upload", "clipboard"],
|
||
)
|
||
palette_tb = gr.Textbox(
|
||
label="Палитра (опц.)",
|
||
placeholder="blue and teal",
|
||
lines=1,
|
||
)
|
||
mode_radio = gr.Radio(
|
||
choices=["icon", "illustration"],
|
||
value="icon",
|
||
label="Mode",
|
||
)
|
||
n_slider = gr.Slider(
|
||
minimum=1, maximum=8, step=1, value=1,
|
||
label="Кандидатов",
|
||
)
|
||
temp_slider = gr.Slider(
|
||
minimum=0.0, maximum=1.5, step=0.05, value=0.4,
|
||
label="Temperature",
|
||
)
|
||
with gr.Accordion("LM Studio", open=False):
|
||
base_url_tb = gr.Textbox(
|
||
label="Base URL",
|
||
value=os.environ.get("LM_STUDIO_BASE_URL", DEFAULT_BASE_URL),
|
||
placeholder="http://127.0.0.1:1234/v1",
|
||
lines=1,
|
||
)
|
||
api_key_tb = gr.Textbox(
|
||
label="API token",
|
||
value=os.environ.get("LM_STUDIO_API_KEY", "lm-studio"),
|
||
lines=1,
|
||
)
|
||
refresh_btn = gr.Button("Обновить список моделей", size="sm")
|
||
fetch_status_md = gr.Markdown("")
|
||
model_dd = gr.Dropdown(
|
||
label="Модель",
|
||
choices=[os.environ.get("DEFAULT_MODEL", DEFAULT_MODEL)],
|
||
value=os.environ.get("DEFAULT_MODEL", DEFAULT_MODEL),
|
||
)
|
||
use_live_cb = gr.Checkbox(
|
||
label="Live-стрим (превью в реальном времени)",
|
||
value=True,
|
||
)
|
||
gen_btn = gr.Button("Сгенерировать", variant="primary")
|
||
status_md = gr.Markdown("")
|
||
|
||
with gr.Column(scale=2):
|
||
gallery = gr.Gallery(
|
||
label="PNG-превью",
|
||
columns=3, height=320, object_fit="contain",
|
||
)
|
||
gallery_state = gr.State([])
|
||
# Поле с reasoning-токенами модели (qwen3.5 шлёт рассуждения
|
||
# отдельным `reasoning_content`-полем). Полезно для моделей
|
||
# типа qwen3.5-A3B, которые думают перед тем, как начать писать
|
||
# SVG — без этого юзер не видит, что стрим идёт.
|
||
reasoning_md = gr.Markdown(
|
||
label="Мысли модели (reasoning)",
|
||
value="",
|
||
)
|
||
with gr.Accordion("SVG-код лучшего кандидата", open=False):
|
||
# gr.Code в Gradio 5.37 не имеет "xml" в whitelist языков
|
||
# (есть python/sql/html/markdown/...). Используем "html"
|
||
# — подсветка разметки близка к XML и работает стабильно.
|
||
svg_viewer = gr.Code(label="SVG", language="html")
|
||
with gr.Accordion("История (последние 20)", open=True):
|
||
history_df = gr.Dataframe(
|
||
headers=["id", "created_at", "mode", "model", "n", "returned", "status"],
|
||
datatype=["number", "str", "str", "str", "number", "number", "str"],
|
||
interactive=False,
|
||
wrap=True,
|
||
)
|
||
history_details = gr.Markdown("")
|
||
|
||
# Связи
|
||
mode_radio.change(
|
||
on_mode_change,
|
||
inputs=[mode_radio],
|
||
outputs=[n_slider],
|
||
)
|
||
refresh_btn.click(
|
||
fetch_lm_studio_models_with_value,
|
||
inputs=[base_url_tb, api_key_tb],
|
||
outputs=[model_dd, fetch_status_md],
|
||
)
|
||
# Автозапрос списка моделей при загрузке UI
|
||
demo.load(
|
||
fetch_lm_studio_models_with_value,
|
||
inputs=[base_url_tb, api_key_tb],
|
||
outputs=[model_dd, fetch_status_md],
|
||
)
|
||
gen_btn.click(
|
||
on_generate_live,
|
||
inputs=[
|
||
prompt_tb, mode_radio, n_slider, temp_slider,
|
||
image_in, palette_tb, model_dd, base_url_tb, api_key_tb,
|
||
use_live_cb,
|
||
],
|
||
outputs=[
|
||
gallery,
|
||
gallery_state,
|
||
svg_viewer,
|
||
history_df,
|
||
status_md,
|
||
gr.State([]),
|
||
gr.State(""),
|
||
reasoning_md,
|
||
],
|
||
)
|
||
history_df.select(
|
||
on_history_select,
|
||
inputs=[history_df],
|
||
outputs=[gallery, svg_viewer, history_details],
|
||
)
|
||
|
||
return demo
|
||
|
||
|
||
# Алиас, который просит задача для smoke-теста импорта.
|
||
demo = None
|
||
|
||
|
||
def main() -> None:
|
||
global demo
|
||
log.info(
|
||
"starting model=%s base_url=%s",
|
||
os.environ.get("DEFAULT_MODEL", DEFAULT_MODEL),
|
||
os.environ.get("LM_STUDIO_BASE_URL", DEFAULT_BASE_URL),
|
||
)
|
||
log.info("db path: %s", DEFAULT_DB_PATH)
|
||
log.info("preview dir: %s", DEFAULT_PREVIEW_DIR)
|
||
demo = build_ui()
|
||
port = int(os.environ.get("OMNISVG_PORT", "8788"))
|
||
host = os.environ.get("OMNISVG_HOST", "127.0.0.1")
|
||
log.info("launching Gradio on %s:%d", host, port)
|
||
# allowed_paths нужен, чтобы Gradio отдавал PNG из ~/.omnisvg_lite/previews/
|
||
# иначе InvalidPathError на рендере превью
|
||
demo.launch(
|
||
server_name=host,
|
||
server_port=port,
|
||
allowed_paths=[str(DEFAULT_PREVIEW_DIR), str(Path.cwd())],
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|