140 lines
5.4 KiB
Python
140 lines
5.4 KiB
Python
from .dumas_image_nodes import (
|
|
NODE_CLASS_MAPPINGS as IMAGE_NODE_CLASS_MAPPINGS,
|
|
NODE_DISPLAY_NAME_MAPPINGS as IMAGE_NODE_DISPLAY_NAME_MAPPINGS,
|
|
_next_counter,
|
|
resolve_serve_token,
|
|
)
|
|
from .dumas_json_nodes import (
|
|
NODE_CLASS_MAPPINGS as JSON_NODE_CLASS_MAPPINGS,
|
|
NODE_DISPLAY_NAME_MAPPINGS as JSON_NODE_DISPLAY_NAME_MAPPINGS,
|
|
)
|
|
|
|
NODE_CLASS_MAPPINGS = {}
|
|
NODE_CLASS_MAPPINGS.update(JSON_NODE_CLASS_MAPPINGS)
|
|
NODE_CLASS_MAPPINGS.update(IMAGE_NODE_CLASS_MAPPINGS)
|
|
|
|
NODE_DISPLAY_NAME_MAPPINGS = {}
|
|
NODE_DISPLAY_NAME_MAPPINGS.update(JSON_NODE_DISPLAY_NAME_MAPPINGS)
|
|
NODE_DISPLAY_NAME_MAPPINGS.update(IMAGE_NODE_DISPLAY_NAME_MAPPINGS)
|
|
|
|
WEB_DIRECTORY = "./js"
|
|
|
|
try:
|
|
import asyncio
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from aiohttp import web
|
|
from server import PromptServer
|
|
|
|
_DUMAS_DIALOG_LOCK = threading.Lock()
|
|
|
|
def _dumas_dialog_available():
|
|
return sys.platform == "win32" and shutil.which("powershell") is not None
|
|
|
|
def _dumas_dialog_windows(start_path):
|
|
ps = (
|
|
"Add-Type -AssemblyName System.Windows.Forms;"
|
|
"$r='';"
|
|
"$o=New-Object System.Windows.Forms.Form;"
|
|
"$o.TopMost=$true;$o.ShowInTaskbar=$false;$o.FormBorderStyle='None';"
|
|
"$o.Width=1;$o.Height=1;$o.Opacity=0;$o.StartPosition='CenterScreen';"
|
|
"$o.Add_Shown({"
|
|
"$o.Activate();"
|
|
"$d=New-Object System.Windows.Forms.FolderBrowserDialog;"
|
|
'$d.Description="Select a folder";$d.ShowNewFolderButton=$true;'
|
|
"if($env:DUMAS_START){try{$d.SelectedPath=$env:DUMAS_START}catch{}};"
|
|
"if($d.ShowDialog($o) -eq [System.Windows.Forms.DialogResult]::OK){$script:r=$d.SelectedPath};"
|
|
"$o.Close()"
|
|
"});"
|
|
"[void]$o.ShowDialog();"
|
|
"[Console]::Out.Write($r)"
|
|
)
|
|
env = dict(os.environ)
|
|
env["DUMAS_START"] = start_path or ""
|
|
result = subprocess.run(
|
|
["powershell", "-NoProfile", "-STA", "-Command", ps],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=300,
|
|
env=env,
|
|
creationflags=0x08000000,
|
|
check=False,
|
|
)
|
|
return (result.stdout or "").strip()
|
|
|
|
def _dumas_native_folder_dialog(start_path=""):
|
|
if not _DUMAS_DIALOG_LOCK.acquire(blocking=False):
|
|
return None
|
|
try:
|
|
if sys.platform == "win32":
|
|
return _dumas_dialog_windows(start_path)
|
|
return ""
|
|
finally:
|
|
try:
|
|
_DUMAS_DIALOG_LOCK.release()
|
|
except Exception:
|
|
pass
|
|
|
|
@PromptServer.instance.routes.get("/dumas/api/save_image/file")
|
|
async def dumas_save_image_file(request):
|
|
path = resolve_serve_token(request.query.get("t", ""))
|
|
if not path or not os.path.isfile(path):
|
|
return web.Response(status=404, text="unknown or expired preview token")
|
|
return web.FileResponse(path)
|
|
|
|
@PromptServer.instance.routes.get("/dumas/api/save_image/next_counter")
|
|
async def dumas_save_image_next_counter(request):
|
|
folder_raw = str(request.query.get("folder", "") or "").strip()
|
|
name = str(request.query.get("name", "") or "")
|
|
try:
|
|
digits = max(1, min(8, int(request.query.get("digits", "3"))))
|
|
except Exception:
|
|
digits = 3
|
|
|
|
def _scan():
|
|
import folder_paths
|
|
|
|
output_dir = folder_paths.get_output_directory()
|
|
base = os.path.abspath(folder_raw) if folder_raw else output_dir
|
|
parts = [part for part in name.replace("\\", "/").split("/") if part]
|
|
if not parts:
|
|
return 1, ""
|
|
frame_dir = os.path.join(base, *parts[:-1])
|
|
counter = _next_counter(frame_dir, parts[-1])
|
|
resolved_name = parts[-1].replace("%counter%", str(counter).zfill(digits))
|
|
return counter, "/".join(parts[:-1] + [resolved_name])
|
|
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
counter, resolved = await loop.run_in_executor(None, _scan)
|
|
return web.json_response({"ok": True, "counter": counter, "resolved": resolved})
|
|
except Exception as exc:
|
|
return web.json_response({"ok": False, "message": str(exc), "counter": 1, "resolved": ""})
|
|
|
|
@PromptServer.instance.routes.get("/dumas/api/pick_directory")
|
|
async def dumas_pick_directory(request):
|
|
try:
|
|
if not _dumas_dialog_available():
|
|
return web.json_response({"ok": False, "message": "Native folder picker is unavailable here.", "unavailable": True})
|
|
|
|
start_path = str(request.query.get("path", "") or "").strip()
|
|
if start_path and not os.path.isdir(start_path):
|
|
start_path = ""
|
|
|
|
loop = asyncio.get_running_loop()
|
|
selected = await loop.run_in_executor(None, _dumas_native_folder_dialog, start_path)
|
|
if selected is None:
|
|
return web.json_response({"ok": False, "busy": True})
|
|
if not selected:
|
|
return web.json_response({"ok": False, "cancelled": True})
|
|
return web.json_response({"ok": True, "path": selected})
|
|
except Exception as exc:
|
|
return web.json_response({"ok": False, "message": str(exc)})
|
|
except Exception:
|
|
pass
|
|
|
|
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
|