|
import types |
|
from server import PromptServer |
|
from aiohttp import web |
|
from asyncio import sleep, run |
|
import json |
|
|
|
|
|
|
|
class AnyType(str): |
|
def __ne__(self, __value: object) -> bool: |
|
return False |
|
|
|
|
|
PY_CODE = AnyType("*") |
|
IDEs_DICT = {} |
|
|
|
|
|
@PromptServer.instance.routes.post("/alekpet/check_js_complete") |
|
async def check_js_complete(request): |
|
json_data = await request.json() |
|
unique_id = json_data.get("unique_id", None) |
|
result_code = json_data.get("result_code", None) |
|
|
|
if ( |
|
unique_id is not None |
|
and unique_id in IDEs_DICT |
|
and result_code |
|
and result_code is not None |
|
): |
|
IDEs_DICT[unique_id].js_result = result_code |
|
IDEs_DICT[unique_id].js_complete = True |
|
return web.json_response({"status": "Ok"}) |
|
|
|
return web.json_response({"status": "Error"}) |
|
|
|
|
|
async def wait_js_complete(unique_id, time_out=40): |
|
for _ in range(time_out): |
|
if ( |
|
hasattr(IDEs_DICT[unique_id], "js_complete") |
|
and IDEs_DICT[unique_id].js_complete == True |
|
and IDEs_DICT[unique_id].js_result is not None |
|
): |
|
IDEs_DICT[unique_id].js_complete = False |
|
return True |
|
|
|
await sleep(0.1) |
|
|
|
return False |
|
|
|
|
|
class IDENode: |
|
def __init__(self): |
|
self.js_complete = False |
|
self.js_result = None |
|
|
|
@classmethod |
|
def INPUT_TYPES(s): |
|
|
|
return { |
|
"optional": {}, |
|
"required": { |
|
"language": ( |
|
(["python", "javascript"]), |
|
{"default": "python"}, |
|
), |
|
"pycode": ( |
|
"PYCODE", |
|
{ |
|
"default": """# !!! Attention, do not insert unverified code !!! |
|
# ---- Example code ---- |
|
# Globals inputs variables: var1, var2, var3, user variables ... |
|
from time import strftime |
|
def runCode(): |
|
nowDataTime = strftime("%Y-%m-%d %H:%M:%S") |
|
return f"Hello ComfyUI with us today {nowDataTime}!" |
|
result = runCode()""" |
|
}, |
|
), |
|
}, |
|
"hidden": {"unique_id": "UNIQUE_ID"}, |
|
} |
|
|
|
RETURN_TYPES = (PY_CODE,) |
|
RETURN_NAMES = ("any",) |
|
FUNCTION = "exec_py" |
|
|
|
CATEGORY = "AlekPet Nodes/experiments" |
|
|
|
def exec_py(self, pycode, language, unique_id, **kwargs): |
|
if unique_id not in IDEs_DICT: |
|
IDEs_DICT[unique_id] = self |
|
|
|
if language == "python": |
|
my_namespace = types.SimpleNamespace() |
|
my_namespace.__dict__.update(kwargs) |
|
my_namespace.__dict__.setdefault("result", "The result variable is not assigned") |
|
|
|
try: |
|
exec(pycode, my_namespace.__dict__) |
|
except Exception as e: |
|
my_namespace.result = f"Error in python code: {e}" |
|
|
|
return (my_namespace.result,) |
|
|
|
else: |
|
IDEs_DICT[unique_id].js_complete = False |
|
IDEs_DICT[unique_id].js_result = None |
|
|
|
PromptServer.instance.send_sync( |
|
"alekpet_js_result", |
|
{"unique_id": unique_id, "vars": json.dumps(kwargs)}, |
|
) |
|
if not run(wait_js_complete(unique_id)): |
|
print(f"IDENode_{unique_id}: Failed to get data!") |
|
else: |
|
print(f"IDENode_{unique_id}: Data received successful!") |
|
|
|
print(IDEs_DICT[unique_id].js_result) |
|
return (IDEs_DICT[unique_id].js_result,) |
|
|
