Ethanfel/Comfyui-JSON-Manager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge nicegui-migration: full NiceGUI web UI

Browse Source
This commit is contained in:
Ethanfel
2026-02-27 22:16:34 +01:00
parent bdcc05f388 8911323832
commit d8597f201a
9 changed files with 2071 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
README.md
View File

@@ -21,13 +21,13 @@
<p align="center">
<img src="https://img.shields.io/badge/License-Apache_2.0-blue.svg" alt="License" />
<img src="https://img.shields.io/badge/Python-3.10%2B-green" alt="Python" />
<img src="https://img.shields.io/badge/Built%20with-Streamlit-red" alt="Streamlit" />
<img src="https://img.shields.io/badge/Built%20with-NiceGUI-FF4B4B" alt="NiceGUI" />
<img src="https://img.shields.io/badge/ComfyUI-Custom%20Nodes-purple" alt="ComfyUI" />
</p>
A visual dashboard for managing, versioning, and batch-processing JSON configuration files used in AI video generation workflows (I2V, VACE). Two parts:
1. **Streamlit Web Interface** &mdash; Dockerized editor for prompts, LoRAs, settings, and branching history
1. **NiceGUI Web Interface** &mdash; Dockerized editor for prompts, LoRAs, settings, and branching history
2. **ComfyUI Custom Nodes** &mdash; Read JSON files directly into workflows, including a dynamic node that auto-discovers keys
---
@@ -86,12 +86,12 @@ Dynamic Node (New)
## Installation
### 1. Unraid / Docker (Streamlit Manager)
### 1. Unraid / Docker (NiceGUI Manager)
```bash
# Repository: python:3.12-slim
# Network: Bridge
# WebUI: http://[IP]:[PORT:8501]
# WebUI: http://[IP]:[PORT:8080]
```
**Path Mappings:**
@@ -103,8 +103,8 @@ Dynamic Node (New)
**Post Arguments:**
```bash
/bin/sh -c "apt-get update && apt-get install -y graphviz && \
pip install streamlit opencv-python-headless graphviz streamlit-agraph && \
cd /app && streamlit run app.py --server.headless true --server.port 8501"
pip install nicegui graphviz requests && \
cd /app && python main.py"
```
### 2. ComfyUI (Custom Nodes)
@@ -274,7 +274,7 @@ The **JSON Loader (Dynamic)** node reads your JSON file and automatically create
## Web Interface Usage
Navigate to your container's IP (e.g., `http://192.168.1.100:8501`).
Navigate to your container's IP (e.g., `http://192.168.1.100:8080`).
**Path navigation** supports case-insensitive matching &mdash; typing `/media/P5/myFolder` will resolve to `/media/p5/MyFolder` automatically.
@@ -315,13 +315,14 @@ ComfyUI-JSON-Manager/
├── json_loader.py # All ComfyUI node classes + /json_manager/get_keys API
├── web/
│ └── json_dynamic.js # Frontend extension for Dynamic node (refresh, show/hide)
├── app.py # Streamlit main entry point & navigator
├── main.py # NiceGUI web UI entry point & navigator
├── state.py # Application state management
├── utils.py # I/O, config, defaults, case-insensitive path resolver
├── history_tree.py # Git-style branching engine
├── tab_batch.py # Batch processor UI
├── tab_timeline.py # Visual timeline UI
├── tab_comfy.py # ComfyUI server monitor
├── tab_raw.py # Raw JSON editor
├── tab_batch_ng.py # Batch processor UI (NiceGUI)
├── tab_timeline_ng.py # Visual timeline UI (NiceGUI)
├── tab_comfy_ng.py # ComfyUI server monitor (NiceGUI)
├── tab_raw_ng.py # Raw JSON editor (NiceGUI)
└── tests/
├── test_json_loader.py
├── test_utils.py

485
main.py Normal file
View File

@@ -0,0 +1,485 @@
import json
from pathlib import Path
from nicegui import ui
from state import AppState
from utils import (
load_config, save_config, load_snippets, save_snippets,
load_json, save_json, generate_templates, DEFAULTS,
KEY_BATCH_DATA, KEY_SEQUENCE_NUMBER,
resolve_path_case_insensitive,
)
from tab_batch_ng import render_batch_processor
from tab_timeline_ng import render_timeline_tab
from tab_raw_ng import render_raw_editor
from tab_comfy_ng import render_comfy_monitor
@ui.page('/')
def index():
# -- Streamlit dark theme --
ui.dark_mode(True)
ui.colors(primary='#F59E0B')
ui.add_head_html(
'<link rel="stylesheet" href="https://fonts.googleapis.com/css2?family=Inter:wght@400;500;600;700&display=swap">'
)
ui.add_css('''
/* === Dark Theme with Depth Palette === */
:root {
--bg-page: #0B0E14;
--bg-surface-1: #13161E;
--bg-surface-2: #1A1E2A;
--bg-surface-3: #242836;
--border: rgba(255,255,255,0.08);
--text-primary: #EAECF0;
--text-secondary: rgba(234,236,240,0.55);
--accent: #F59E0B;
--accent-subtle: rgba(245,158,11,0.12);
--negative: #EF4444;
}
/* Backgrounds */
body.body--dark,
.q-page.body--dark,
.body--dark .q-page { background: var(--bg-page) !important; }
.body--dark .q-drawer { background: var(--bg-surface-1) !important; }
.body--dark .q-card {
background: var(--bg-surface-2) !important;
border: 1px solid var(--border);
border-radius: 0.75rem;
}
.body--dark .q-tab-panels { background: transparent !important; }
.body--dark .q-tab-panel { background: transparent !important; }
.body--dark .q-expansion-item { background: transparent !important; }
/* Text */
.body--dark { color: var(--text-primary) !important; }
.body--dark .q-field__label { color: var(--text-secondary) !important; }
.body--dark .text-caption { color: var(--text-secondary) !important; }
.body--dark .text-subtitle1,
.body--dark .text-subtitle2 { color: var(--text-primary) !important; }
/* Inputs & textareas */
.body--dark .q-field--outlined .q-field__control {
background: var(--bg-surface-3) !important;
border-radius: 0.5rem !important;
}
.body--dark .q-field--outlined .q-field__control:before {
border-color: var(--border) !important;
border-radius: 0.5rem !important;
}
.body--dark .q-field--outlined.q-field--focused .q-field__control:after {
border-color: var(--accent) !important;
}
.body--dark .q-field__native,
.body--dark .q-field__input { color: var(--text-primary) !important; }
/* Sidebar inputs get page bg */
.body--dark .q-drawer .q-field--outlined .q-field__control {
background: var(--bg-page) !important;
}
/* Buttons */
.body--dark .q-btn--standard { border-radius: 0.5rem !important; }
.body--dark .q-btn--outline {
transition: background 0.15s ease;
}
.body--dark .q-btn--outline:hover {
background: var(--accent-subtle) !important;
}
/* Tabs */
.body--dark .q-tab--active { color: var(--accent) !important; }
.body--dark .q-tab__indicator { background: var(--accent) !important; }
/* Separators */
.body--dark .q-separator { background: var(--border) !important; }
/* Expansion items */
.body--dark .q-expansion-item__content { padding: 12px 16px; }
.body--dark .q-item { border-radius: 0.5rem; }
/* Splitter */
.body--dark .q-splitter__separator { background: var(--border) !important; }
.body--dark .q-splitter__before,
.body--dark .q-splitter__after { padding: 0 8px; }
/* Action row wrap */
.action-row { flex-wrap: wrap !important; gap: 8px !important; }
/* Notifications */
.body--dark .q-notification { border-radius: 0.5rem; }
/* Font */
body { font-family: "Inter", "Source Sans Pro", "Source Sans 3", sans-serif !important; }
/* Surface utility classes (need .body--dark to beat .body--dark .q-card specificity) */
.body--dark .surface-1 { background: var(--bg-surface-1) !important; }
.body--dark .surface-2 { background: var(--bg-surface-2) !important; }
.body--dark .surface-3 { background: var(--bg-surface-3) !important; }
/* Typography utility classes */
.section-header {
font-size: 0.8rem;
font-weight: 600;
text-transform: uppercase;
letter-spacing: 0.05em;
color: var(--text-secondary) !important;
}
.subsection-header {
font-size: 0.85rem;
font-weight: 500;
color: var(--text-primary) !important;
}
/* Scrollbar */
::-webkit-scrollbar { width: 6px; height: 6px; }
::-webkit-scrollbar-track { background: transparent; }
::-webkit-scrollbar-thumb {
background: rgba(255,255,255,0.12);
border-radius: 3px;
}
::-webkit-scrollbar-thumb:hover {
background: rgba(255,255,255,0.2);
}
/* Secondary pane teal accent */
.pane-secondary .q-field--outlined.q-field--focused .q-field__control:after {
border-color: #06B6D4 !important;
}
.pane-secondary .q-btn.bg-primary { background-color: #06B6D4 !important; }
.pane-secondary .section-header { color: rgba(6,182,212,0.7) !important; }
''')
config = load_config()
state = AppState(
config=config,
current_dir=Path(config.get('last_dir', str(Path.cwd()))),
snippets=load_snippets(),
)
dual_pane = {'active': False, 'state': None}
# ------------------------------------------------------------------
# Define helpers FIRST (before sidebar, which needs them)
# ------------------------------------------------------------------
@ui.refreshable
def render_main_content():
max_w = '2400px' if dual_pane['active'] else '1200px'
with ui.column().classes('w-full q-pa-md').style(f'max-width: {max_w}; margin: 0 auto'):
if not state.file_path or not state.file_path.exists():
ui.label('Select a file from the sidebar to begin.').classes(
'text-subtitle1 q-pa-lg')
return
ui.label(f'Editing: {state.file_path.name}').classes('text-h5 q-mb-lg').style('font-weight: 600')
with ui.tabs().classes('w-full').style('border-bottom: 1px solid var(--border)') as tabs:
ui.tab('batch', label='Batch Processor')
ui.tab('timeline', label='Timeline')
ui.tab('raw', label='Raw Editor')
with ui.tab_panels(tabs, value='batch').classes('w-full'):
with ui.tab_panel('batch'):
_render_batch_tab_content()
with ui.tab_panel('timeline'):
render_timeline_tab(state)
with ui.tab_panel('raw'):
render_raw_editor(state)
if state.show_comfy_monitor:
ui.separator()
with ui.expansion('ComfyUI Monitor', icon='dns').classes('w-full'):
render_comfy_monitor(state)
@ui.refreshable
def _render_batch_tab_content():
def on_toggle(e):
dual_pane['active'] = e.value
if e.value and dual_pane['state'] is None:
s2 = state.create_secondary()
s2._render_main = _render_batch_tab_content
dual_pane['state'] = s2
render_main_content.refresh()
ui.switch('Dual Pane', value=dual_pane['active'], on_change=on_toggle)
if not dual_pane['active']:
render_batch_processor(state)
else:
s2 = dual_pane['state']
with ui.row().classes('w-full gap-4'):
with ui.column().classes('col'):
ui.label('Pane A').classes('section-header q-mb-sm')
_render_pane_file_selector(state)
render_batch_processor(state)
with ui.column().classes('col pane-secondary'):
ui.label('Pane B').classes('section-header q-mb-sm')
_render_pane_file_selector(s2)
if s2.file_path and s2.file_path.exists():
render_batch_processor(s2)
else:
ui.label('Select a file above to begin.').classes(
'text-caption q-pa-md')
def _render_pane_file_selector(pane_state: AppState):
if not pane_state.current_dir.exists():
ui.label('Directory not found.').classes('text-warning')
return
json_files = sorted(pane_state.current_dir.glob('*.json'))
json_files = [f for f in json_files if f.name not in (
'.editor_config.json', '.editor_snippets.json')]
file_names = [f.name for f in json_files]
current_val = pane_state.file_path.name if pane_state.file_path else None
def on_select(e):
if not e.value:
return
fp = pane_state.current_dir / e.value
data, mtime = load_json(fp)
pane_state.data_cache = data
pane_state.last_mtime = mtime
pane_state.loaded_file = str(fp)
pane_state.file_path = fp
pane_state.restored_indicator = None
_render_batch_tab_content.refresh()
ui.select(
file_names,
value=current_val,
label='File',
on_change=on_select,
).classes('w-full')
def load_file(file_name: str):
"""Load a JSON file and refresh the main content."""
fp = state.current_dir / file_name
if state.loaded_file == str(fp):
return
data, mtime = load_json(fp)
state.data_cache = data
state.last_mtime = mtime
state.loaded_file = str(fp)
state.file_path = fp
state.restored_indicator = None
if state._main_rendered:
render_main_content.refresh()
# Attach helpers to state so sidebar can call them
state._load_file = load_file
state._render_main = render_main_content
state._main_rendered = False
# ------------------------------------------------------------------
# Sidebar (rendered AFTER helpers are attached)
# ------------------------------------------------------------------
with ui.left_drawer(value=True).classes('q-pa-md').style('width: 320px'):
render_sidebar(state, dual_pane)
# ------------------------------------------------------------------
# Main content area
# ------------------------------------------------------------------
render_main_content()
state._main_rendered = True
# ======================================================================
# Sidebar
# ======================================================================
def render_sidebar(state: AppState, dual_pane: dict):
ui.label('Navigator').classes('text-h6')
# --- Path input + Pin ---
with ui.card().classes('w-full q-pa-md q-mb-md'):
path_input = ui.input(
'Current Path',
value=str(state.current_dir),
).classes('w-full')
def on_path_enter():
p = resolve_path_case_insensitive(path_input.value)
if p is not None and p.is_dir():
state.current_dir = p
if dual_pane['state']:
dual_pane['state'].current_dir = state.current_dir
dual_pane['state'].file_path = None
dual_pane['state'].loaded_file = None
dual_pane['state'].data_cache = {}
state.config['last_dir'] = str(p)
save_config(state.current_dir, state.config['favorites'], state.config)
state.loaded_file = None
state.file_path = None
path_input.set_value(str(p))
render_file_list.refresh()
# Auto-load inside render_file_list already refreshed main content
# if files exist; only refresh here for the empty-directory case.
if not state.loaded_file:
state._render_main.refresh()
path_input.on('keydown.enter', lambda _: on_path_enter())
def pin_folder():
d = str(state.current_dir)
if d not in state.config['favorites']:
state.config['favorites'].append(d)
save_config(state.current_dir, state.config['favorites'], state.config)
render_favorites.refresh()
ui.button('Pin Folder', icon='push_pin', on_click=pin_folder).classes('w-full')
# --- Favorites ---
with ui.card().classes('w-full q-pa-md q-mb-md'):
ui.label('Favorites').classes('section-header')
@ui.refreshable
def render_favorites():
for fav in list(state.config['favorites']):
with ui.row().classes('w-full items-center'):
ui.button(
fav,
on_click=lambda f=fav: _jump_to(f),
).props('flat dense').classes('col')
ui.button(
icon='close',
on_click=lambda f=fav: _unpin(f),
).props('flat dense color=negative')
def _jump_to(fav: str):
state.current_dir = Path(fav)
if dual_pane['state']:
dual_pane['state'].current_dir = state.current_dir
dual_pane['state'].file_path = None
dual_pane['state'].loaded_file = None
dual_pane['state'].data_cache = {}
state.config['last_dir'] = fav
save_config(state.current_dir, state.config['favorites'], state.config)
state.loaded_file = None
state.file_path = None
path_input.set_value(fav)
render_file_list.refresh()
if not state.loaded_file:
state._render_main.refresh()
def _unpin(fav: str):
if fav in state.config['favorites']:
state.config['favorites'].remove(fav)
save_config(state.current_dir, state.config['favorites'], state.config)
render_favorites.refresh()
render_favorites()
# --- Snippet Library ---
with ui.card().classes('w-full q-pa-md q-mb-md'):
ui.label('Snippet Library').classes('section-header')
with ui.expansion('Add New Snippet'):
snip_name_input = ui.input('Name', placeholder='e.g. Cinematic').classes('w-full')
snip_content_input = ui.textarea('Content', placeholder='4k, high quality...').classes('w-full')
def save_snippet():
name = snip_name_input.value
content = snip_content_input.value
if name and content:
state.snippets[name] = content
save_snippets(state.snippets)
snip_name_input.set_value('')
snip_content_input.set_value('')
ui.notify(f"Saved '{name}'")
render_snippet_list.refresh()
ui.button('Save Snippet', on_click=save_snippet).classes('w-full')
@ui.refreshable
def render_snippet_list():
if not state.snippets:
return
ui.label('Click to copy snippet text:').classes('text-caption')
for name, content in list(state.snippets.items()):
with ui.row().classes('w-full items-center'):
async def copy_snippet(c=content):
await ui.run_javascript(
f'navigator.clipboard.writeText({json.dumps(c)})', timeout=3.0)
ui.notify('Copied to clipboard')
ui.button(
f'{name}',
on_click=copy_snippet,
).props('flat dense').classes('col')
ui.button(
icon='delete',
on_click=lambda n=name: _del_snippet(n),
).props('flat dense color=negative')
def _del_snippet(name: str):
if name in state.snippets:
del state.snippets[name]
save_snippets(state.snippets)
render_snippet_list.refresh()
render_snippet_list()
# --- File List ---
with ui.card().classes('w-full q-pa-md q-mb-md'):
@ui.refreshable
def render_file_list():
if not state.current_dir.exists():
ui.label('Directory not found.').classes('text-warning')
return
json_files = sorted(state.current_dir.glob('*.json'))
json_files = [f for f in json_files if f.name not in ('.editor_config.json', '.editor_snippets.json')]
if not json_files:
ui.label('No JSON files in this folder.').classes('text-caption')
ui.button('Generate Templates', on_click=lambda: _gen_templates()).classes('w-full')
return
with ui.expansion('Create New JSON'):
new_fn_input = ui.input('Filename', placeholder='my_prompt_vace').classes('w-full')
def create_new():
fn = new_fn_input.value
if not fn:
return
if not fn.endswith('.json'):
fn += '.json'
path = state.current_dir / fn
first_item = DEFAULTS.copy()
first_item[KEY_SEQUENCE_NUMBER] = 1
save_json(path, {KEY_BATCH_DATA: [first_item]})
new_fn_input.set_value('')
render_file_list.refresh()
ui.button('Create', on_click=create_new).classes('w-full')
ui.label('Select File').classes('subsection-header q-mt-sm')
file_names = [f.name for f in json_files]
current = Path(state.loaded_file).name if state.loaded_file else None
selected = current if current in file_names else (file_names[0] if file_names else None)
ui.radio(
file_names,
value=selected,
on_change=lambda e: state._load_file(e.value) if e.value else None,
).classes('w-full')
# Auto-load first file if nothing loaded yet
if file_names and not state.loaded_file:
state._load_file(file_names[0])
def _gen_templates():
generate_templates(state.current_dir)
render_file_list.refresh()
render_file_list()
# --- Comfy Monitor toggle ---
def on_monitor_toggle(e):
state.show_comfy_monitor = e.value
state._render_main.refresh()
ui.checkbox('Show Comfy Monitor', value=True, on_change=on_monitor_toggle)
ui.run(title='AI Settings Manager', port=8080, reload=True)

32
state.py Normal file
View File

@@ -0,0 +1,32 @@
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable
@dataclass
class AppState:
config: dict
current_dir: Path
loaded_file: str | None = None
last_mtime: float = 0
data_cache: dict = field(default_factory=dict)
snippets: dict = field(default_factory=dict)
file_path: Path | None = None
restored_indicator: str | None = None
timeline_selected_nodes: set = field(default_factory=set)
live_toggles: dict = field(default_factory=dict)
show_comfy_monitor: bool = True
# Set at runtime by main.py / tab_comfy_ng.py
_render_main: Any = None
_load_file: Callable | None = None
_main_rendered: bool = False
_live_checkboxes: dict = field(default_factory=dict)
_live_refreshables: dict = field(default_factory=dict)
def create_secondary(self) -> 'AppState':
return AppState(
config=self.config,
current_dir=self.current_dir,
snippets=self.snippets,
)

720
tab_batch_ng.py Normal file
View File

@@ -0,0 +1,720 @@
import copy
import random
from pathlib import Path
from nicegui import ui
from state import AppState
from utils import (
DEFAULTS, save_json, load_json,
KEY_BATCH_DATA, KEY_HISTORY_TREE, KEY_PROMPT_HISTORY, KEY_SEQUENCE_NUMBER,
)
from history_tree import HistoryTree
IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.webp', '.bmp', '.gif'}
SUB_SEGMENT_MULTIPLIER = 1000
FRAME_TO_SKIP_DEFAULT = DEFAULTS['frame_to_skip']
VACE_MODES = [
'End Extend', 'Pre Extend', 'Middle Extend', 'Edge Extend',
'Join Extend', 'Bidirectional Extend', 'Frame Interpolation',
'Replace/Inpaint', 'Video Inpaint', 'Keyframe',
]
VACE_FORMULAS = [
'base + A', 'base + B', 'base + A + B', 'base + A + B',
'base + A + B', 'base + A + B', '(B-1) * step',
'snap(source)', 'snap(source)', 'base + A + B',
]
# --- Sub-segment helpers (same as original) ---
def is_subsegment(seq_num):
return int(seq_num) >= SUB_SEGMENT_MULTIPLIER
def parent_of(seq_num):
seq_num = int(seq_num)
return seq_num // SUB_SEGMENT_MULTIPLIER if is_subsegment(seq_num) else seq_num
def sub_index_of(seq_num):
seq_num = int(seq_num)
return seq_num % SUB_SEGMENT_MULTIPLIER if is_subsegment(seq_num) else 0
def format_seq_label(seq_num):
seq_num = int(seq_num)
if is_subsegment(seq_num):
return f'Sub #{parent_of(seq_num)}.{sub_index_of(seq_num)}'
return f'Sequence #{seq_num}'
def next_sub_segment_number(batch_list, parent_seq_num):
parent_seq_num = int(parent_seq_num)
max_sub = 0
for s in batch_list:
sn = int(s.get(KEY_SEQUENCE_NUMBER, 0))
if is_subsegment(sn) and parent_of(sn) == parent_seq_num:
max_sub = max(max_sub, sub_index_of(sn))
return parent_seq_num * SUB_SEGMENT_MULTIPLIER + max_sub + 1
def max_main_seq_number(batch_list):
"""Highest non-subsegment sequence number in the batch."""
return max(
(int(x.get(KEY_SEQUENCE_NUMBER, 0))
for x in batch_list if not is_subsegment(x.get(KEY_SEQUENCE_NUMBER, 0))),
default=0,
)
def find_insert_position(batch_list, parent_index, parent_seq_num):
parent_seq_num = int(parent_seq_num)
pos = parent_index + 1
while pos < len(batch_list):
sn = int(batch_list[pos].get(KEY_SEQUENCE_NUMBER, 0))
if is_subsegment(sn) and parent_of(sn) == parent_seq_num:
pos += 1
else:
break
return pos
# --- Helper for repetitive dict-bound inputs ---
def dict_input(element_fn, label, seq, key, **kwargs):
"""Create an input element bound to seq[key] via blur and model-value update."""
val = seq.get(key, '')
if isinstance(val, (int, float)):
val = str(val) if element_fn != ui.number else val
el = element_fn(label, value=val, **kwargs)
def _sync(k=key):
seq[k] = el.value
el.on('blur', lambda _: _sync())
el.on('update:model-value', lambda _: _sync())
return el
def dict_number(label, seq, key, default=0, **kwargs):
"""Number input bound to seq[key] via blur and model-value update."""
val = seq.get(key, default)
try:
# Try float first to handle "1.5" strings, then check if it's a clean int
fval = float(val)
val = int(fval) if fval == int(fval) else fval
except (ValueError, TypeError, OverflowError):
val = default
el = ui.number(label, value=val, **kwargs)
def _sync(k=key, d=default):
v = el.value
if v is None:
v = d
elif isinstance(v, float):
try:
v = int(v) if v == int(v) else v
except (OverflowError, ValueError):
v = d
seq[k] = v
el.on('blur', lambda _: _sync())
el.on('update:model-value', lambda _: _sync())
return el
def dict_textarea(label, seq, key, **kwargs):
"""Textarea bound to seq[key] via blur and model-value update."""
el = ui.textarea(label, value=seq.get(key, ''), **kwargs)
def _sync(k=key):
seq[k] = el.value
el.on('blur', lambda _: _sync())
el.on('update:model-value', lambda _: _sync())
return el
# ======================================================================
# Main render function
# ======================================================================
def render_batch_processor(state: AppState):
data = state.data_cache
file_path = state.file_path
if isinstance(data, list):
data = {KEY_BATCH_DATA: data}
state.data_cache = data
is_batch_file = KEY_BATCH_DATA in data
if not is_batch_file:
ui.label('This is a Single file. To use Batch mode, create a copy.').classes(
'text-warning')
def create_batch():
new_name = f'batch_{file_path.name}'
new_path = file_path.parent / new_name
if new_path.exists():
ui.notify(f'File {new_name} already exists!', type='warning')
return
first_item = copy.deepcopy(data)
first_item.pop(KEY_PROMPT_HISTORY, None)
first_item.pop(KEY_HISTORY_TREE, None)
first_item[KEY_SEQUENCE_NUMBER] = 1
new_data = {KEY_BATCH_DATA: [first_item], KEY_HISTORY_TREE: {},
KEY_PROMPT_HISTORY: []}
save_json(new_path, new_data)
ui.notify(f'Created {new_name}', type='positive')
ui.button('Create Batch Copy', icon='content_copy', on_click=create_batch)
return
if state.restored_indicator:
ui.label(f'Editing Restored Version: {state.restored_indicator}').classes(
'text-info q-pa-sm')
batch_list = data.get(KEY_BATCH_DATA, [])
# Source file data for importing
with ui.card().classes('w-full q-pa-md q-mb-lg'):
with ui.expansion('Add New Sequence from Source File', icon='playlist_add').classes('w-full'):
json_files = sorted(state.current_dir.glob('*.json'))
json_files = [f for f in json_files if f.name not in (
'.editor_config.json', '.editor_snippets.json')]
file_options = {f.name: f.name for f in json_files}
src_file_select = ui.select(
file_options,
value=file_path.name,
label='Source File:',
).classes('w-64')
src_seq_select = ui.select([], label='Source Sequence:').classes('w-64')
# Track loaded source data
_src_cache = {'data': None, 'batch': [], 'name': None}
def _update_src():
name = src_file_select.value
if name and name != _src_cache['name']:
src_data, _ = load_json(state.current_dir / name)
_src_cache['data'] = src_data
_src_cache['batch'] = src_data.get(KEY_BATCH_DATA, [])
_src_cache['name'] = name
if _src_cache['batch']:
opts = {i: format_seq_label(s.get(KEY_SEQUENCE_NUMBER, i+1))
for i, s in enumerate(_src_cache['batch'])}
src_seq_select.set_options(opts, value=0)
else:
src_seq_select.set_options({})
src_file_select.on_value_change(lambda _: _update_src())
_update_src()
def _add_sequence(new_item):
new_item[KEY_SEQUENCE_NUMBER] = max_main_seq_number(batch_list) + 1
for k in [KEY_PROMPT_HISTORY, KEY_HISTORY_TREE, 'note', 'loras']:
new_item.pop(k, None)
batch_list.append(new_item)
data[KEY_BATCH_DATA] = batch_list
save_json(file_path, data)
render_sequence_list.refresh()
with ui.row().classes('q-mt-sm'):
def add_empty():
_add_sequence(DEFAULTS.copy())
def add_from_source():
item = copy.deepcopy(DEFAULTS)
src_batch = _src_cache['batch']
sel_idx = src_seq_select.value
if src_batch and sel_idx is not None:
item.update(copy.deepcopy(src_batch[int(sel_idx)]))
elif _src_cache['data']:
item.update(copy.deepcopy(_src_cache['data']))
_add_sequence(item)
ui.button('Add Empty', icon='add', on_click=add_empty)
ui.button('From Source', icon='file_download', on_click=add_from_source)
# --- Standard / LoRA / VACE key sets ---
lora_keys = ['lora 1 high', 'lora 1 low', 'lora 2 high', 'lora 2 low',
'lora 3 high', 'lora 3 low']
standard_keys = {
'general_prompt', 'general_negative', 'current_prompt', 'negative', 'prompt',
'seed', 'cfg', 'camera', 'flf', KEY_SEQUENCE_NUMBER,
'frame_to_skip', 'end_frame', 'transition', 'vace_length',
'input_a_frames', 'input_b_frames', 'reference switch', 'vace schedule',
'reference path', 'video file path', 'reference image path', 'flf image path',
}
standard_keys.update(lora_keys)
def sort_by_number():
batch_list.sort(key=lambda s: int(s.get(KEY_SEQUENCE_NUMBER, 0)))
data[KEY_BATCH_DATA] = batch_list
save_json(file_path, data)
ui.notify('Sorted by sequence number!', type='positive')
render_sequence_list.refresh()
# --- Sequence list + mass update (inside refreshable so they stay in sync) ---
@ui.refreshable
def render_sequence_list():
# Mass update (rebuilt on refresh so checkboxes match current sequences)
_render_mass_update(batch_list, data, file_path, state, render_sequence_list)
with ui.row().classes('w-full items-center'):
ui.label(f'Batch contains {len(batch_list)} sequences.')
ui.button('Sort by Number', icon='sort', on_click=sort_by_number).props('flat')
for i, seq in enumerate(batch_list):
with ui.card().classes('w-full q-mb-sm'):
_render_sequence_card(
i, seq, batch_list, data, file_path, state,
_src_cache, src_seq_select,
standard_keys, render_sequence_list,
)
render_sequence_list()
# --- Save & Snap ---
with ui.card().classes('w-full q-pa-md q-mt-lg'):
with ui.row().classes('w-full items-end q-gutter-md'):
commit_input = ui.input('Change Note (Optional)',
placeholder='e.g. Added sequence 3').classes('col')
def save_and_snap():
data[KEY_BATCH_DATA] = batch_list
tree_data = data.get(KEY_HISTORY_TREE, {})
htree = HistoryTree(tree_data)
snapshot_payload = copy.deepcopy(data)
snapshot_payload.pop(KEY_HISTORY_TREE, None)
note = commit_input.value if commit_input.value else 'Batch Update'
htree.commit(snapshot_payload, note=note)
data[KEY_HISTORY_TREE] = htree.to_dict()
save_json(file_path, data)
state.restored_indicator = None
commit_input.set_value('')
ui.notify('Batch Saved & Snapshot Created!', type='positive')
ui.button('Save & Snap', icon='save', on_click=save_and_snap).props('color=primary')
# ======================================================================
# Single sequence card
# ======================================================================
def _render_sequence_card(i, seq, batch_list, data, file_path, state,
src_cache, src_seq_select, standard_keys,
refresh_list):
def commit(message=None):
data[KEY_BATCH_DATA] = batch_list
save_json(file_path, data)
if message:
ui.notify(message, type='positive')
refresh_list.refresh()
seq_num = seq.get(KEY_SEQUENCE_NUMBER, i + 1)
if is_subsegment(seq_num):
label = f'Sub #{parent_of(seq_num)}.{sub_index_of(seq_num)} ({int(seq_num)})'
else:
label = f'Sequence #{seq_num}'
with ui.expansion(label, icon='movie').classes('w-full'):
# --- Action row ---
with ui.row().classes('w-full q-gutter-sm action-row'):
# Copy from source
def copy_source(idx=i, sn=seq_num):
item = copy.deepcopy(DEFAULTS)
src_batch = src_cache['batch']
sel_idx = src_seq_select.value
if src_batch and sel_idx is not None:
item.update(copy.deepcopy(src_batch[int(sel_idx)]))
elif src_cache['data']:
item.update(copy.deepcopy(src_cache['data']))
item[KEY_SEQUENCE_NUMBER] = sn
item.pop(KEY_PROMPT_HISTORY, None)
item.pop(KEY_HISTORY_TREE, None)
batch_list[idx] = item
commit('Copied!')
ui.button('Copy Src', icon='file_download', on_click=copy_source).props('outline')
# Clone Next
def clone_next(idx=i, sn=seq_num, s=seq):
new_seq = copy.deepcopy(s)
new_seq[KEY_SEQUENCE_NUMBER] = max_main_seq_number(batch_list) + 1
if not is_subsegment(sn):
pos = find_insert_position(batch_list, idx, int(sn))
else:
pos = idx + 1
batch_list.insert(pos, new_seq)
commit('Cloned to Next!')
ui.button('Clone Next', icon='content_copy', on_click=clone_next).props('outline')
# Clone End
def clone_end(s=seq):
new_seq = copy.deepcopy(s)
new_seq[KEY_SEQUENCE_NUMBER] = max_main_seq_number(batch_list) + 1
batch_list.append(new_seq)
commit('Cloned to End!')
ui.button('Clone End', icon='vertical_align_bottom', on_click=clone_end).props('outline')
# Clone Sub
def clone_sub(idx=i, sn=seq_num, s=seq):
new_seq = copy.deepcopy(s)
p_seq = parent_of(sn)
p_idx = idx
if is_subsegment(sn):
for pi, ps in enumerate(batch_list):
if int(ps.get(KEY_SEQUENCE_NUMBER, 0)) == p_seq:
p_idx = pi
break
new_seq[KEY_SEQUENCE_NUMBER] = next_sub_segment_number(batch_list, p_seq)
pos = find_insert_position(batch_list, p_idx, p_seq)
batch_list.insert(pos, new_seq)
commit(f'Created {format_seq_label(new_seq[KEY_SEQUENCE_NUMBER])}!')
ui.button('Clone Sub', icon='link', on_click=clone_sub).props('outline')
ui.element('div').classes('col')
# Delete
def delete(idx=i):
batch_list.pop(idx)
commit()
ui.button(icon='delete', on_click=delete).props('color=negative')
ui.separator()
# --- Prompts + Settings (2-column like Streamlit) ---
with ui.splitter(value=66).classes('w-full') as splitter:
with splitter.before:
dict_textarea('General Prompt', seq, 'general_prompt').classes(
'w-full q-mt-sm').props('outlined rows=2')
dict_textarea('General Negative', seq, 'general_negative').classes(
'w-full q-mt-sm').props('outlined rows=2')
dict_textarea('Specific Prompt', seq, 'current_prompt').classes(
'w-full q-mt-sm').props('outlined rows=10')
dict_textarea('Specific Negative', seq, 'negative').classes(
'w-full q-mt-sm').props('outlined rows=2')
with splitter.after:
# Sequence number
sn_label = (
f'Seq Number (Sub #{parent_of(seq_num)}.{sub_index_of(seq_num)})'
if is_subsegment(seq_num) else 'Sequence Number'
)
sn_input = dict_number(sn_label, seq, KEY_SEQUENCE_NUMBER)
sn_input.props('outlined').classes('w-full')
# Seed + randomize
with ui.row().classes('w-full items-end'):
seed_input = dict_number('Seed', seq, 'seed').classes('col').props('outlined')
def randomize_seed(si=seed_input, s=seq):
new_seed = random.randint(0, 999999999999)
si.set_value(new_seed)
s['seed'] = new_seed
ui.button(icon='casino', on_click=randomize_seed).props('flat')
# CFG
dict_number('CFG', seq, 'cfg', default=DEFAULTS['cfg'],
step=0.5, format='%.1f').props('outlined').classes('w-full')
dict_input(ui.input, 'Camera', seq, 'camera').props('outlined').classes('w-full')
dict_input(ui.input, 'FLF', seq, 'flf').props('outlined').classes('w-full')
dict_number('End Frame', seq, 'end_frame').props('outlined').classes('w-full')
dict_input(ui.input, 'Video File Path', seq, 'video file path').props(
'outlined input-style="direction: rtl"').classes('w-full')
# Image paths with preview
for img_label, img_key in [
('Reference Image Path', 'reference image path'),
('Reference Path', 'reference path'),
('FLF Image Path', 'flf image path'),
]:
with ui.row().classes('w-full items-center'):
inp = dict_input(ui.input, img_label, seq, img_key).classes(
'col').props('outlined input-style="direction: rtl"')
img_path = Path(seq.get(img_key, '')) if seq.get(img_key) else None
if (img_path and img_path.exists() and
img_path.suffix.lower() in IMAGE_EXTENSIONS):
with ui.dialog() as dlg, ui.card():
ui.image(str(img_path)).classes('w-full')
ui.button(icon='visibility', on_click=dlg.open).props('flat dense')
# --- VACE Settings (full width) ---
with ui.expansion('VACE Settings', icon='settings').classes('w-full'):
_render_vace_settings(i, seq, batch_list, data, file_path, refresh_list)
# --- LoRA Settings ---
with ui.expansion('LoRA Settings', icon='style').classes('w-full'):
for lora_idx in range(1, 4):
for tier, tier_label in [('high', 'High'), ('low', 'Low')]:
k = f'lora {lora_idx} {tier}'
raw = str(seq.get(k, ''))
inner = raw.replace('<lora:', '').replace('>', '')
# Split "name:strength" or just "name"
if ':' in inner:
parts = inner.rsplit(':', 1)
lora_name = parts[0]
try:
lora_strength = float(parts[1])
except ValueError:
lora_name = inner
lora_strength = 1.0
else:
lora_name = inner
lora_strength = 1.0
with ui.row().classes('w-full items-center q-gutter-sm'):
ui.label(f'L{lora_idx} {tier_label}').classes(
'text-caption').style('min-width: 55px')
name_input = ui.input(
'Name',
value=lora_name,
).classes('col').props('outlined dense')
strength_input = ui.number(
'Str',
value=lora_strength,
min=0, max=10, step=0.1,
format='%.1f',
).props('outlined dense').style('max-width: 80px')
def _lora_sync(key=k, n_inp=name_input, s_inp=strength_input):
name = n_inp.value or ''
strength = s_inp.value if s_inp.value is not None else 1.0
seq[key] = f'<lora:{name}:{strength:.1f}>' if name else ''
name_input.on('blur', lambda _, s=_lora_sync: s())
name_input.on('update:model-value', lambda _, s=_lora_sync: s())
strength_input.on('blur', lambda _, s=_lora_sync: s())
strength_input.on('update:model-value', lambda _, s=_lora_sync: s())
# --- Custom Parameters ---
ui.label('Custom Parameters').classes('section-header q-mt-md')
custom_keys = [k for k in seq.keys() if k not in standard_keys]
if custom_keys:
for k in custom_keys:
with ui.row().classes('w-full items-center'):
ui.input('Key', value=k).props('readonly outlined dense').classes('w-32')
dict_input(ui.input, 'Value', seq, k).props('outlined dense').classes('col')
def del_custom(key=k):
del seq[key]
commit()
ui.button(icon='delete', on_click=del_custom).props('flat dense color=negative')
with ui.expansion('Add Parameter', icon='add').classes('w-full'):
new_k_input = ui.input('Key').props('outlined dense')
new_v_input = ui.input('Value').props('outlined dense')
def add_param():
k = new_k_input.value
v = new_v_input.value
if k and k not in seq:
seq[k] = v
new_k_input.set_value('')
new_v_input.set_value('')
commit()
ui.button('Add', on_click=add_param).props('flat')
# ======================================================================
# VACE Settings sub-section
# ======================================================================
def _render_vace_settings(i, seq, batch_list, data, file_path, refresh_list):
# VACE Schedule (needed early for both columns)
sched_val = max(0, min(int(seq.get('vace schedule', 1)), len(VACE_MODES) - 1))
# Mode reference dialog
with ui.dialog() as ref_dlg, ui.card():
table_md = (
'| # | Mode | Formula |\n|:--|:-----|:--------|\n'
+ '\n'.join(
f'| **{j}** | {VACE_MODES[j]} | `{VACE_FORMULAS[j]}` |'
for j in range(len(VACE_MODES)))
+ '\n\n*All totals snapped to 4n+1 (1,5,9,...,49,...,81,...)*'
)
ui.markdown(table_md)
with ui.row().classes('w-full q-gutter-md'):
# --- Left column ---
with ui.column().classes('col'):
# Frame to Skip + shift
with ui.row().classes('w-full items-end'):
fts_input = dict_number('Frame to Skip', seq, 'frame_to_skip').classes(
'col').props('outlined')
_original_fts = int(seq.get('frame_to_skip', FRAME_TO_SKIP_DEFAULT))
def shift_fts(idx=i, orig=_original_fts):
new_fts = int(fts_input.value) if fts_input.value is not None else orig
delta = new_fts - orig
if delta == 0:
ui.notify('No change to shift', type='info')
return
shifted = 0
for j in range(idx + 1, len(batch_list)):
batch_list[j]['frame_to_skip'] = int(
batch_list[j].get('frame_to_skip', FRAME_TO_SKIP_DEFAULT)) + delta
shifted += 1
data[KEY_BATCH_DATA] = batch_list
save_json(file_path, data)
ui.notify(f'Shifted {shifted} sequences by {delta:+d}', type='positive')
refresh_list.refresh()
ui.button('Shift', icon='arrow_downward', on_click=shift_fts).props(
'outline').style('height: 40px')
dict_input(ui.input, 'Transition', seq, 'transition').props('outlined').classes(
'w-full q-mt-sm')
# VACE Schedule
with ui.row().classes('w-full items-center q-mt-sm'):
vs_input = dict_number('VACE Schedule', seq, 'vace schedule', default=1,
min=0, max=len(VACE_MODES) - 1).classes('col').props(
'outlined')
mode_label = ui.label(VACE_MODES[sched_val]).classes('text-caption')
ui.button(icon='help', on_click=ref_dlg.open).props('flat dense round')
def update_mode_label(e):
idx = int(e.sender.value) if e.sender.value is not None else 0
idx = max(0, min(idx, len(VACE_MODES) - 1))
mode_label.set_text(VACE_MODES[idx])
vs_input.on('update:model-value', update_mode_label)
# --- Right column ---
with ui.column().classes('col'):
ia_input = dict_number('Input A Frames', seq, 'input_a_frames').props(
'outlined').classes('w-full')
ib_input = dict_number('Input B Frames', seq, 'input_b_frames').props(
'outlined').classes('w-full q-mt-sm')
# VACE Length + output calculation
input_a = int(seq.get('input_a_frames', 16))
input_b = int(seq.get('input_b_frames', 16))
stored_total = int(seq.get('vace_length', 49))
mode_idx = int(seq.get('vace schedule', 1))
if mode_idx == 0:
base_length = max(stored_total - input_a, 1)
elif mode_idx == 1:
base_length = max(stored_total - input_b, 1)
else:
base_length = max(stored_total - input_a - input_b, 1)
with ui.row().classes('w-full items-center q-mt-sm'):
vl_input = ui.number('VACE Length', value=base_length, min=1).classes(
'col').props('outlined')
output_label = ui.label(f'Output: {stored_total}').classes('text-bold')
dict_number('Reference Switch', seq, 'reference switch').props(
'outlined').classes('w-full q-mt-sm')
# Recalculate VACE output when any input changes
def recalc_vace(*_args):
mi = int(vs_input.value) if vs_input.value is not None else 0
ia = int(ia_input.value) if ia_input.value is not None else 16
ib = int(ib_input.value) if ib_input.value is not None else 16
nb = int(vl_input.value) if vl_input.value is not None else 1
if mi == 0:
raw = nb + ia
elif mi == 1:
raw = nb + ib
else:
raw = nb + ia + ib
snapped = ((raw + 2) // 4) * 4 + 1
seq['vace_length'] = snapped
output_label.set_text(f'Output: {snapped}')
for inp in (vs_input, ia_input, ib_input, vl_input):
inp.on('update:model-value', recalc_vace)
# ======================================================================
# Mass Update
# ======================================================================
def _render_mass_update(batch_list, data, file_path, state: AppState, refresh_list=None):
with ui.expansion('Mass Update', icon='sync').classes('w-full'):
if len(batch_list) < 2:
ui.label('Need at least 2 sequences for mass update.').classes('text-caption')
return
source_options = {i: format_seq_label(s.get(KEY_SEQUENCE_NUMBER, i+1))
for i, s in enumerate(batch_list)}
source_select = ui.select(source_options, value=0,
label='Copy from sequence:').classes('w-full')
field_select = ui.select([], multiple=True,
label='Fields to copy:').classes('w-full')
def update_fields(_=None):
idx = source_select.value
if idx is not None and 0 <= idx < len(batch_list):
src = batch_list[idx]
keys = [k for k in src.keys() if k != 'sequence_number']
field_select.set_options(keys)
source_select.on_value_change(update_fields)
update_fields()
ui.label('Apply to:').classes('subsection-header q-mt-md')
select_all_cb = ui.checkbox('Select All')
target_checks = {}
with ui.scroll_area().style('max-height: 250px'):
for idx, s in enumerate(batch_list):
sn = s.get(KEY_SEQUENCE_NUMBER, idx + 1)
cb = ui.checkbox(format_seq_label(sn))
target_checks[idx] = cb
def on_select_all(e):
for cb in target_checks.values():
cb.set_value(e.value)
select_all_cb.on_value_change(on_select_all)
def apply_mass_update():
src_idx = source_select.value
if src_idx is None or src_idx >= len(batch_list):
ui.notify('Source sequence no longer exists', type='warning')
return
selected_keys = field_select.value or []
if not selected_keys:
ui.notify('No fields selected', type='warning')
return
source_seq = batch_list[src_idx]
targets = [idx for idx, cb in target_checks.items()
if cb.value and idx != src_idx and idx < len(batch_list)]
if not targets:
ui.notify('No target sequences selected', type='warning')
return
for idx in targets:
for key in selected_keys:
batch_list[idx][key] = copy.deepcopy(source_seq.get(key))
data[KEY_BATCH_DATA] = batch_list
htree = HistoryTree(data.get(KEY_HISTORY_TREE, {}))
snapshot = copy.deepcopy(data)
snapshot.pop(KEY_HISTORY_TREE, None)
htree.commit(snapshot, f"Mass update: {', '.join(selected_keys)}")
data[KEY_HISTORY_TREE] = htree.to_dict()
save_json(file_path, data)
ui.notify(f'Updated {len(targets)} sequences', type='positive')
if refresh_list:
refresh_list.refresh()
ui.button('Apply Changes', icon='check', on_click=apply_mass_update).props(
'color=primary')

278
tab_comfy_ng.py Normal file
View File

@@ -0,0 +1,278 @@
import asyncio
import html
import time
import urllib.parse
import requests
from nicegui import ui
from state import AppState
from utils import save_config
def render_comfy_monitor(state: AppState):
config = state.config
# --- Global Monitor Settings ---
with ui.expansion('Monitor Settings', icon='settings').classes('w-full'):
with ui.row().classes('w-full items-end'):
viewer_input = ui.input(
'Remote Browser URL',
value=config.get('viewer_url', ''),
placeholder='e.g., http://localhost:5800',
).classes('col')
timeout_slider = ui.slider(
min=0, max=60, step=1,
value=config.get('monitor_timeout', 0),
).classes('col')
ui.label().bind_text_from(timeout_slider, 'value',
backward=lambda v: f'Timeout: {v} min')
def save_monitor_settings():
config['viewer_url'] = viewer_input.value
config['monitor_timeout'] = int(timeout_slider.value)
save_config(state.current_dir, config['favorites'], config)
ui.notify('Monitor settings saved!', type='positive')
ui.button('Save Monitor Settings', icon='save', on_click=save_monitor_settings)
# --- Instance Management ---
if 'comfy_instances' not in config:
config['comfy_instances'] = [
{'name': 'Main Server', 'url': 'http://192.168.1.100:8188'}
]
instances = config['comfy_instances']
@ui.refreshable
def render_instance_tabs():
if not instances:
ui.label('No servers configured. Add one below.')
for idx, inst in enumerate(instances):
with ui.expansion(inst.get('name', f'Server {idx+1}'), icon='dns').classes('w-full'):
_render_single_instance(state, inst, idx, instances, render_instance_tabs)
# Add server section
ui.separator()
ui.label('Add New Server').classes('section-header')
with ui.row().classes('w-full items-end'):
new_name = ui.input('Server Name', placeholder='e.g. Render Node 2').classes('col')
new_url = ui.input('URL', placeholder='http://192.168.1.50:8188').classes('col')
def add_instance():
if new_name.value and new_url.value:
instances.append({'name': new_name.value, 'url': new_url.value})
config['comfy_instances'] = instances
save_config(state.current_dir, config['favorites'], config)
ui.notify('Server Added!', type='positive')
new_name.set_value('')
new_url.set_value('')
render_instance_tabs.refresh()
else:
ui.notify('Please fill in both Name and URL.', type='warning')
ui.button('Add Instance', icon='add', on_click=add_instance)
render_instance_tabs()
# --- Auto-poll timer (every 300s) ---
# Store live_checkbox references so the timer can update them
_live_checkboxes = state._live_checkboxes
_live_refreshables = state._live_refreshables
def poll_all():
timeout_val = config.get('monitor_timeout', 0)
if timeout_val > 0:
for key, start_time in list(state.live_toggles.items()):
if start_time and (time.time() - start_time) > (timeout_val * 60):
state.live_toggles[key] = None
if key in _live_checkboxes:
_live_checkboxes[key].set_value(False)
if key in _live_refreshables:
_live_refreshables[key].refresh()
ui.timer(300, poll_all)
def _fetch_blocking(url, timeout=1.5):
"""Run a blocking GET request; returns (response, error)."""
try:
res = requests.get(url, timeout=timeout)
return res, None
except Exception as e:
return None, e
def _render_single_instance(state: AppState, instance_config: dict, index: int,
all_instances: list, refresh_fn):
config = state.config
url = instance_config.get('url', 'http://127.0.0.1:8188')
name = instance_config.get('name', f'Server {index+1}')
comfy_url = url.rstrip('/')
# --- Settings popover ---
with ui.expansion('Settings', icon='settings'):
name_input = ui.input('Name', value=name).classes('w-full')
url_input = ui.input('URL', value=url).classes('w-full')
def update_server():
all_instances[index]['name'] = name_input.value
all_instances[index]['url'] = url_input.value
config['comfy_instances'] = all_instances
save_config(state.current_dir, config['favorites'], config)
ui.notify('Server config saved!', type='positive')
refresh_fn.refresh()
def remove_server():
all_instances.pop(index)
config['comfy_instances'] = all_instances
save_config(state.current_dir, config['favorites'], config)
ui.notify('Server removed', type='info')
refresh_fn.refresh()
ui.button('Update & Save', icon='save', on_click=update_server).props('color=primary')
ui.button('Remove Server', icon='delete', on_click=remove_server).props('color=negative')
# --- Status Dashboard ---
status_container = ui.row().classes('w-full items-center q-gutter-md')
async def refresh_status():
status_container.clear()
loop = asyncio.get_event_loop()
res, err = await loop.run_in_executor(
None, lambda: _fetch_blocking(f'{comfy_url}/queue'))
with status_container:
if res is not None:
try:
queue_data = res.json()
except (ValueError, Exception):
ui.label('Invalid response from server').classes('text-negative')
return
running_cnt = len(queue_data.get('queue_running', []))
pending_cnt = len(queue_data.get('queue_pending', []))
with ui.card().classes('q-pa-md text-center').style('min-width: 100px'):
ui.label('Status')
ui.label('Online' if running_cnt > 0 else 'Idle').classes(
'text-positive' if running_cnt > 0 else 'text-grey')
with ui.card().classes('q-pa-md text-center').style('min-width: 100px'):
ui.label('Pending')
ui.label(str(pending_cnt))
with ui.card().classes('q-pa-md text-center').style('min-width: 100px'):
ui.label('Running')
ui.label(str(running_cnt))
else:
with ui.card().classes('q-pa-md text-center').style('min-width: 100px'):
ui.label('Status')
ui.label('Offline').classes('text-negative')
ui.label(f'Could not connect to {comfy_url}').classes('text-negative')
# Initial status fetch (non-blocking via button click handler pattern)
ui.timer(0.1, refresh_status, once=True)
ui.button('Refresh Status', icon='refresh', on_click=refresh_status).props('flat dense')
# --- Live View ---
with ui.card().classes('w-full q-pa-md q-mt-md'):
ui.label('Live View').classes('section-header')
toggle_key = f'live_toggle_{index}'
live_checkbox = ui.checkbox('Enable Live Preview', value=False)
# Store reference so poll_all timer can disable it on timeout
state._live_checkboxes[toggle_key] = live_checkbox
@ui.refreshable
def render_live_view():
if not live_checkbox.value:
ui.label('Live Preview is disabled.').classes('text-caption')
return
# Record start time
if toggle_key not in state.live_toggles or state.live_toggles.get(toggle_key) is None:
state.live_toggles[toggle_key] = time.time()
timeout_val = config.get('monitor_timeout', 0)
if timeout_val > 0:
start = state.live_toggles.get(toggle_key, time.time())
remaining = (timeout_val * 60) - (time.time() - start)
if remaining <= 0:
live_checkbox.set_value(False)
state.live_toggles[toggle_key] = None
ui.label('Preview timed out.').classes('text-caption')
return
ui.label(f'Auto-off in: {int(remaining)}s').classes('text-caption')
iframe_h = ui.slider(min=600, max=2500, step=50, value=1000).classes('w-full')
ui.label().bind_text_from(iframe_h, 'value', backward=lambda v: f'Height: {v}px')
viewer_base = config.get('viewer_url', '').strip()
parsed = urllib.parse.urlparse(viewer_base)
if viewer_base and parsed.scheme in ('http', 'https'):
safe_src = html.escape(viewer_base, quote=True)
ui.label(f'Viewing: {viewer_base}').classes('text-caption')
iframe_container = ui.column().classes('w-full')
def update_iframe():
iframe_container.clear()
with iframe_container:
ui.html(
f'<iframe src="{safe_src}" width="100%" height="{int(iframe_h.value)}px"'
f' style="border: 2px solid #666; border-radius: 8px;"></iframe>'
)
iframe_h.on_value_change(lambda _: update_iframe())
update_iframe()
else:
ui.label('No valid viewer URL configured.').classes('text-warning')
state._live_refreshables[toggle_key] = render_live_view
live_checkbox.on_value_change(lambda _: render_live_view.refresh())
render_live_view()
# --- Latest Output ---
with ui.card().classes('w-full q-pa-md q-mt-md'):
ui.label('Latest Output').classes('section-header')
img_container = ui.column().classes('w-full')
async def check_image():
img_container.clear()
loop = asyncio.get_event_loop()
res, err = await loop.run_in_executor(
None, lambda: _fetch_blocking(f'{comfy_url}/history', timeout=2))
with img_container:
if err is not None:
ui.label(f'Error fetching image: {err}').classes('text-negative')
return
try:
history = res.json()
except (ValueError, Exception):
ui.label('Invalid response from server').classes('text-negative')
return
if not history:
ui.label('No history found.').classes('text-caption')
return
last_prompt_id = list(history.keys())[-1]
outputs = history[last_prompt_id].get('outputs', {})
found_img = None
for node_output in outputs.values():
if 'images' in node_output:
for img_info in node_output['images']:
if img_info['type'] == 'output':
found_img = img_info
break
if found_img:
break
if found_img:
params = urllib.parse.urlencode({
'filename': found_img['filename'],
'subfolder': found_img['subfolder'],
'type': found_img['type'],
})
img_url = f'{comfy_url}/view?{params}'
ui.image(img_url).classes('w-full').style('max-width: 600px')
ui.label(f'Last Output: {found_img["filename"]}').classes('text-caption')
else:
ui.label('Last run had no image output.').classes('text-caption')
ui.button('Check Latest Image', icon='image', on_click=check_image).props('flat')

73
tab_raw_ng.py Normal file
View File

@@ -0,0 +1,73 @@
import copy
import json
from nicegui import ui
from state import AppState
from utils import save_json, get_file_mtime, KEY_HISTORY_TREE, KEY_PROMPT_HISTORY
def render_raw_editor(state: AppState):
data = state.data_cache
file_path = state.file_path
with ui.card().classes('w-full q-pa-md'):
ui.label(f'Raw Editor: {file_path.name}').classes('text-h6 q-mb-md')
hide_history = ui.checkbox(
'Hide History (Safe Mode)',
value=True,
)
@ui.refreshable
def render_editor():
# Prepare display data
if hide_history.value:
display_data = copy.deepcopy(data)
display_data.pop(KEY_HISTORY_TREE, None)
display_data.pop(KEY_PROMPT_HISTORY, None)
else:
display_data = data
try:
json_str = json.dumps(display_data, indent=4, ensure_ascii=False)
except Exception as e:
ui.notify(f'Error serializing JSON: {e}', type='negative')
json_str = '{}'
text_area = ui.textarea(
'JSON Content',
value=json_str,
).classes('w-full font-mono').props('outlined rows=30')
def do_save():
try:
input_data = json.loads(text_area.value)
# Merge hidden history back in if safe mode
if hide_history.value:
if KEY_HISTORY_TREE in data:
input_data[KEY_HISTORY_TREE] = data[KEY_HISTORY_TREE]
if KEY_PROMPT_HISTORY in data:
input_data[KEY_PROMPT_HISTORY] = data[KEY_PROMPT_HISTORY]
save_json(file_path, input_data)
data.clear()
data.update(input_data)
state.last_mtime = get_file_mtime(file_path)
ui.notify('Raw JSON Saved Successfully!', type='positive')
render_editor.refresh()
except json.JSONDecodeError as e:
ui.notify(f'Invalid JSON Syntax: {e}', type='negative')
except Exception as e:
ui.notify(f'Unexpected Error: {e}', type='negative')
ui.button('Save Raw Changes', icon='save', on_click=do_save).props(
'color=primary'
).classes('w-full q-mt-md')
hide_history.on_value_change(lambda _: render_editor.refresh())
render_editor()

469
tab_timeline_ng.py Normal file
View File

@@ -0,0 +1,469 @@
import copy
import time
from nicegui import ui
from state import AppState
from history_tree import HistoryTree
from utils import save_json, KEY_BATCH_DATA, KEY_HISTORY_TREE
def _delete_nodes(htree, data, file_path, node_ids):
"""Delete nodes with backup, branch cleanup, and head fallback."""
if 'history_tree_backup' not in data:
data['history_tree_backup'] = []
data['history_tree_backup'].append(copy.deepcopy(htree.to_dict()))
for nid in node_ids:
htree.nodes.pop(nid, None)
for b, tip in list(htree.branches.items()):
if tip in node_ids:
del htree.branches[b]
if htree.head_id in node_ids:
if htree.nodes:
htree.head_id = sorted(htree.nodes.values(),
key=lambda x: x['timestamp'])[-1]['id']
else:
htree.head_id = None
data[KEY_HISTORY_TREE] = htree.to_dict()
save_json(file_path, data)
def _render_selection_picker(all_nodes, htree, state, refresh_fn):
"""Multi-select picker for batch-deleting timeline nodes."""
all_ids = [n['id'] for n in all_nodes]
def fmt_option(nid):
n = htree.nodes[nid]
ts = time.strftime('%b %d %H:%M', time.localtime(n['timestamp']))
note = n.get('note', 'Step')
head = ' (HEAD)' if nid == htree.head_id else ''
return f'{note} - {ts} ({nid[:6]}){head}'
options = {nid: fmt_option(nid) for nid in all_ids}
def on_selection_change(e):
state.timeline_selected_nodes = set(e.value) if e.value else set()
ui.select(
options,
value=list(state.timeline_selected_nodes),
multiple=True,
label='Select nodes to delete:',
on_change=on_selection_change,
).classes('w-full')
with ui.row():
def select_all():
state.timeline_selected_nodes = set(all_ids)
refresh_fn()
def deselect_all():
state.timeline_selected_nodes = set()
refresh_fn()
ui.button('Select All', on_click=select_all).props('flat dense')
ui.button('Deselect All', on_click=deselect_all).props('flat dense')
def _render_graph_or_log(mode, all_nodes, htree, selected_nodes,
selection_mode_on, toggle_select_fn, restore_fn):
"""Render graph visualization or linear log view."""
if mode in ('Horizontal', 'Vertical'):
direction = 'LR' if mode == 'Horizontal' else 'TB'
with ui.card().classes('w-full q-pa-md'):
try:
graph_dot = htree.generate_graph(direction=direction)
_render_graphviz(graph_dot)
except Exception as e:
ui.label(f'Graph Error: {e}').classes('text-negative')
elif mode == 'Linear Log':
ui.label('Chronological list of all snapshots.').classes('text-caption')
for n in all_nodes:
is_head = n['id'] == htree.head_id
is_selected = n['id'] in selected_nodes
card_style = ''
if is_selected:
card_style = 'background: rgba(239, 68, 68, 0.1) !important; border-left: 3px solid var(--negative);'
elif is_head:
card_style = 'background: var(--accent-subtle) !important; border-left: 3px solid var(--accent);'
with ui.card().classes('w-full q-mb-sm').style(card_style):
with ui.row().classes('w-full items-center'):
if selection_mode_on:
ui.checkbox(
'',
value=is_selected,
on_change=lambda e, nid=n['id']: toggle_select_fn(
nid, e.value),
)
icon = 'location_on' if is_head else 'circle'
ui.icon(icon).classes(
'text-primary' if is_head else 'text-grey')
with ui.column().classes('col'):
note = n.get('note', 'Step')
ts = time.strftime('%b %d %H:%M',
time.localtime(n['timestamp']))
label = f'{note} (Current)' if is_head else note
ui.label(label).classes('text-bold')
ui.label(
f'ID: {n["id"][:6]} - {ts}').classes('text-caption')
if not is_head and not selection_mode_on:
ui.button(
'Restore',
icon='restore',
on_click=lambda node=n: restore_fn(node),
).props('flat dense color=primary')
def _render_batch_delete(htree, data, file_path, state, refresh_fn):
"""Render batch delete controls for selected timeline nodes."""
valid = state.timeline_selected_nodes & set(htree.nodes.keys())
state.timeline_selected_nodes = valid
count = len(valid)
if count == 0:
return
ui.label(
f'{count} node{"s" if count != 1 else ""} selected for deletion.'
).classes('text-warning q-mt-md')
def do_batch_delete():
current_valid = state.timeline_selected_nodes & set(htree.nodes.keys())
_delete_nodes(htree, data, file_path, current_valid)
state.timeline_selected_nodes = set()
ui.notify(
f'Deleted {len(current_valid)} node{"s" if len(current_valid) != 1 else ""}!',
type='positive')
refresh_fn()
ui.button(
f'Delete {count} Node{"s" if count != 1 else ""}',
icon='delete',
on_click=do_batch_delete,
).props('color=negative')
def _walk_branch_nodes(htree, tip_id):
"""Walk parent pointers from tip, returning nodes newest-first."""
nodes = []
current = tip_id
while current and current in htree.nodes:
nodes.append(htree.nodes[current])
current = htree.nodes[current].get('parent')
return nodes
def _find_active_branch(htree):
"""Return branch name whose tip == head_id, or None if detached."""
if not htree.head_id:
return None
for b_name, tip_id in htree.branches.items():
if tip_id == htree.head_id:
return b_name
return None
def _render_node_manager(all_nodes, htree, data, file_path, restore_fn, refresh_fn):
"""Render branch-grouped node manager with restore, rename, delete, and preview."""
ui.label('Manage Version').classes('section-header')
# --- State that survives @ui.refreshable ---
active_branch = _find_active_branch(htree)
# Default branch: active branch, or branch whose ancestry contains HEAD
default_branch = active_branch
if not default_branch and htree.head_id:
for b_name, tip_id in htree.branches.items():
for n in _walk_branch_nodes(htree, tip_id):
if n['id'] == htree.head_id:
default_branch = b_name
break
if default_branch:
break
if not default_branch and htree.branches:
default_branch = next(iter(htree.branches))
selected = {'node_id': htree.head_id, 'branch': default_branch}
# --- (a) Branch selector ---
def fmt_branch(b_name):
count = len(_walk_branch_nodes(htree, htree.branches.get(b_name)))
suffix = ' (active)' if b_name == active_branch else ''
return f'{b_name} ({count} nodes){suffix}'
branch_options = {b: fmt_branch(b) for b in htree.branches}
def on_branch_change(e):
selected['branch'] = e.value
tip = htree.branches.get(e.value)
if tip:
selected['node_id'] = tip
render_branch_nodes.refresh()
ui.select(
branch_options,
value=selected['branch'],
label='Branch:',
on_change=on_branch_change,
).classes('w-full')
# --- (b) Node list + (c) Actions panel ---
@ui.refreshable
def render_branch_nodes():
branch_name = selected['branch']
tip_id = htree.branches.get(branch_name)
nodes = _walk_branch_nodes(htree, tip_id) if tip_id else []
if not nodes:
ui.label('No nodes on this branch.').classes('text-caption q-pa-sm')
return
with ui.scroll_area().classes('w-full').style('max-height: 350px'):
for n in nodes:
nid = n['id']
is_head = nid == htree.head_id
is_tip = nid == tip_id
is_selected = nid == selected['node_id']
card_style = ''
if is_selected:
card_style = 'border-left: 3px solid var(--primary);'
elif is_head:
card_style = 'border-left: 3px solid var(--accent);'
with ui.card().classes('w-full q-mb-xs q-pa-xs').style(card_style):
with ui.row().classes('w-full items-center no-wrap'):
icon = 'location_on' if is_head else 'circle'
icon_size = 'sm' if is_head else 'xs'
ui.icon(icon, size=icon_size).classes(
'text-primary' if is_head else 'text-grey')
with ui.column().classes('col q-ml-xs').style('min-width: 0'):
note = n.get('note', 'Step')
ts = time.strftime('%b %d %H:%M',
time.localtime(n['timestamp']))
label_text = note
lbl = ui.label(label_text).classes('text-body2 ellipsis')
if is_head:
lbl.classes('text-bold')
ui.label(f'{ts} \u2022 {nid[:6]}').classes(
'text-caption text-grey')
if is_head:
ui.badge('HEAD', color='amber').props('dense')
if is_tip and not is_head:
ui.badge('tip', color='green', outline=True).props('dense')
def select_node(node_id=nid):
selected['node_id'] = node_id
render_branch_nodes.refresh()
ui.button(icon='check_circle', on_click=select_node).props(
'flat dense round size=sm'
).tooltip('Select this node')
# --- (c) Actions panel ---
sel_id = selected['node_id']
if not sel_id or sel_id not in htree.nodes:
return
sel_node = htree.nodes[sel_id]
sel_note = sel_node.get('note', 'Step')
is_head = sel_id == htree.head_id
ui.separator().classes('q-my-sm')
ui.label(f'Selected: {sel_note} ({sel_id[:6]})').classes(
'text-caption text-bold')
with ui.row().classes('w-full items-end q-gutter-sm'):
if not is_head:
def restore_selected():
if sel_id in htree.nodes:
restore_fn(htree.nodes[sel_id])
ui.button('Restore', icon='restore',
on_click=restore_selected).props('color=primary dense')
# Rename
rename_input = ui.input('Rename Label').classes('col').props('dense')
def rename_node():
if sel_id in htree.nodes and rename_input.value:
htree.nodes[sel_id]['note'] = rename_input.value
data[KEY_HISTORY_TREE] = htree.to_dict()
save_json(file_path, data)
ui.notify('Label updated', type='positive')
refresh_fn()
ui.button('Update Label', on_click=rename_node).props('flat dense')
# Danger zone
with ui.expansion('Danger Zone', icon='warning').classes(
'w-full q-mt-sm').style('border-left: 3px solid var(--negative)'):
ui.label('Deleting a node cannot be undone.').classes('text-warning')
def delete_selected():
if sel_id in htree.nodes:
_delete_nodes(htree, data, file_path, {sel_id})
ui.notify('Node Deleted', type='positive')
refresh_fn()
ui.button('Delete This Node', icon='delete',
on_click=delete_selected).props('color=negative dense')
# Data preview
with ui.expansion('Data Preview', icon='preview').classes('w-full q-mt-sm'):
_render_data_preview(sel_id, htree)
render_branch_nodes()
def render_timeline_tab(state: AppState):
data = state.data_cache
file_path = state.file_path
tree_data = data.get(KEY_HISTORY_TREE, {})
if not tree_data:
ui.label('No history timeline exists. Make some changes in the Editor first!').classes(
'text-subtitle1 q-pa-md')
return
htree = HistoryTree(tree_data)
if state.restored_indicator:
ui.label(f'Editing Restored Version: {state.restored_indicator}').classes(
'text-info q-pa-sm')
# --- View mode + Selection toggle ---
with ui.row().classes('w-full items-center q-gutter-md q-mb-md'):
ui.label('Version History').classes('text-h6 col')
view_mode = ui.toggle(
['Horizontal', 'Vertical', 'Linear Log'],
value='Horizontal',
)
selection_mode = ui.switch('Select to Delete')
@ui.refreshable
def render_timeline():
all_nodes = sorted(htree.nodes.values(), key=lambda x: x['timestamp'], reverse=True)
selected_nodes = state.timeline_selected_nodes if selection_mode.value else set()
if selection_mode.value:
_render_selection_picker(all_nodes, htree, state, render_timeline.refresh)
_render_graph_or_log(
view_mode.value, all_nodes, htree, selected_nodes,
selection_mode.value, _toggle_select, _restore_and_refresh)
if selection_mode.value and state.timeline_selected_nodes:
_render_batch_delete(htree, data, file_path, state, render_timeline.refresh)
with ui.card().classes('w-full q-pa-md q-mt-md'):
_render_node_manager(
all_nodes, htree, data, file_path,
_restore_and_refresh, render_timeline.refresh)
def _toggle_select(nid, checked):
if checked:
state.timeline_selected_nodes.add(nid)
else:
state.timeline_selected_nodes.discard(nid)
render_timeline.refresh()
def _restore_and_refresh(node):
_restore_node(data, node, htree, file_path, state)
# Refresh all tabs (batch, raw, timeline) so they pick up the restored data
state._render_main.refresh()
view_mode.on_value_change(lambda _: render_timeline.refresh())
selection_mode.on_value_change(lambda _: render_timeline.refresh())
render_timeline()
def _render_graphviz(dot_source: str):
"""Render graphviz DOT source as SVG using ui.html."""
try:
import graphviz
src = graphviz.Source(dot_source)
svg = src.pipe(format='svg').decode('utf-8')
ui.html(f'<div style="overflow-x: auto;">{svg}</div>')
except ImportError:
ui.label('Install graphviz Python package for graph rendering.').classes('text-warning')
ui.code(dot_source).classes('w-full')
except Exception as e:
ui.label(f'Graph rendering error: {e}').classes('text-negative')
def _restore_node(data, node, htree, file_path, state: AppState):
"""Restore a history node as the current version."""
node_data = copy.deepcopy(node['data'])
if KEY_BATCH_DATA not in node_data and KEY_BATCH_DATA in data:
del data[KEY_BATCH_DATA]
data.update(node_data)
htree.head_id = node['id']
data[KEY_HISTORY_TREE] = htree.to_dict()
save_json(file_path, data)
label = f"{node.get('note', 'Step')} ({node['id'][:4]})"
state.restored_indicator = label
ui.notify('Restored!', type='positive')
def _render_data_preview(nid, htree):
"""Render a read-only preview of the selected node's data."""
if not nid or nid not in htree.nodes:
ui.label('No node selected.').classes('text-caption')
return
node_data = htree.nodes[nid]['data']
batch_list = node_data.get(KEY_BATCH_DATA, [])
if batch_list and isinstance(batch_list, list) and len(batch_list) > 0:
ui.label(f'This snapshot contains {len(batch_list)} sequences.').classes('text-caption')
for i, seq_data in enumerate(batch_list):
seq_num = seq_data.get('sequence_number', i + 1)
with ui.expansion(f'Sequence #{seq_num}', value=(i == 0)):
_render_preview_fields(seq_data)
else:
_render_preview_fields(node_data)
def _render_preview_fields(item_data: dict):
"""Render read-only preview of prompts, settings, LoRAs."""
with ui.grid(columns=2).classes('w-full'):
ui.textarea('General Positive',
value=item_data.get('general_prompt', '')).props('readonly outlined rows=3')
ui.textarea('General Negative',
value=item_data.get('general_negative', '')).props('readonly outlined rows=3')
val_sp = item_data.get('current_prompt', '') or item_data.get('prompt', '')
ui.textarea('Specific Positive',
value=val_sp).props('readonly outlined rows=3')
ui.textarea('Specific Negative',
value=item_data.get('negative', '')).props('readonly outlined rows=3')
with ui.row().classes('w-full q-gutter-md'):
ui.input('Camera', value=str(item_data.get('camera', 'static'))).props('readonly outlined')
ui.input('FLF', value=str(item_data.get('flf', '0.0'))).props('readonly outlined')
ui.input('Seed', value=str(item_data.get('seed', '-1'))).props('readonly outlined')
with ui.expansion('LoRA Configuration'):
with ui.row().classes('w-full q-gutter-md'):
for lora_idx in range(1, 4):
with ui.column():
ui.input(f'L{lora_idx} Name',
value=item_data.get(f'lora {lora_idx} high', '')).props(
'readonly outlined dense')
ui.input(f'L{lora_idx} Str',
value=str(item_data.get(f'lora {lora_idx} low', ''))).props(
'readonly outlined dense')
vace_keys = ['frame_to_skip', 'vace schedule', 'video file path']
if any(k in item_data for k in vace_keys):
with ui.expansion('VACE / I2V Settings'):
with ui.row().classes('w-full q-gutter-md'):
ui.input('Skip Frames',
value=str(item_data.get('frame_to_skip', ''))).props('readonly outlined')
ui.input('Schedule',
value=str(item_data.get('vace schedule', ''))).props('readonly outlined')
ui.input('Video Path',
value=str(item_data.get('video file path', ''))).props('readonly outlined')

7
tests/test_utils.py
View File

@@ -1,15 +1,8 @@
import json
import os
from pathlib import Path
from unittest.mock import patch
import pytest
# Mock streamlit before importing utils
import sys
from unittest.mock import MagicMock
sys.modules.setdefault("streamlit", MagicMock())
from utils import load_json, save_json, get_file_mtime, ALLOWED_BASE_DIR, DEFAULTS, resolve_path_case_insensitive

4
utils.py
View File

@@ -5,8 +5,6 @@ import time
from pathlib import Path
from typing import Any
import streamlit as st
# --- Magic String Keys ---
KEY_BATCH_DATA = "batch_data"
KEY_HISTORY_TREE = "history_tree"
@@ -145,7 +143,7 @@ def load_json(path: str | Path) -> tuple[dict[str, Any], float]:
data = json.load(f)
return data, path.stat().st_mtime
except Exception as e:
st.error(f"Error loading JSON: {e}")
logger.error(f"Error loading JSON: {e}")
return DEFAULTS.copy(), 0
def save_json(path: str | Path, data: dict[str, Any]) -> None: