" in description:
#description = markdown.markdown(description)
# description = html.escape(description)
description = description.replace('\n', '
').replace(f"('", '').replace(f"')", '')
html_content = template_node.substitute(
title=html.escape(name),
name=html.escape(name),
root1=root1,
category=html.escape(json_dict['category']),
documentation=description,
root2=root2,
boop=html.escape(boop),
output_node=json_dict['output_node'],
input_content=''.join(input_sections),
output_content=''.join(output_rows)
)
return html_content
def get_node_info(node_data: dict) -> Dict[str, Any]:
"""Transform node object_info route result into .html."""
input_parameters = {}
for k, node_param_meta in node_data.get('input', {}).items():
if not k in ["required", "optional"]:
continue
input_parameters[k] = {}
for param_key, param_meta in node_param_meta.items():
lst = None
typ = param_meta[0]
if isinstance(typ, list):
typ = match_combo(typ)
lst = param_meta
input_parameters[k][param_key] = {
"type": typ
}
try:
meta = param_meta[1]
if lst is not None:
if (choice_list := meta.get('choice', None)) is None:
data = [x.replace('_', ' ') for x in lst[0]][:JOV_LIST_MAX]
input_parameters[k][param_key]["choice"] = data
meta.update(lst[1])
else:
input_parameters[k][param_key]["choice"] = [choice_list][:JOV_LIST_MAX]
meta['default'] = 'dynamic'
elif (default_top := meta.get('default_top', None)) is not None:
meta['default'] = default_top
# only stuff that makes sense...
junk = ['default', 'min', 'max']
meta = node_param_meta[param_key][1]
if (tip := meta.get("tooltip", None)) is None:
if (tip := Lexicon._tooltipsDB.get(param_key, None)) is None:
# logger.warning(f"no tooltip for {node_class}[{k}]::{param_key}")
junk.append("tooltip")
tip = "Unknown Explanation!"
input_parameters[k][param_key]["tooltip"] = tip
for scrape in junk:
if (val := meta.get(scrape, None)) is not None and val != "":
input_parameters[k][param_key][scrape] = val
except IndexError:
pass
return_types = [
match_combo(x) if isinstance(x, list) or isinstance(x, tuple) else x for x in node_data.get('output', [])
]
output_parameters = {}
tooltips = node_data.get('output_tooltips', [])
return_names = [t.lower() for t in node_data.get('output_name', [])]
for name, typ, tip in zip(return_names, return_types, tooltips):
if tip == "":
tip = Lexicon._tooltipsDB.get(name, "")
output_parameters[name] = '$'.join([typ, tip])
data = {
"class": node_data['name'],
"input_parameters": collapse_repeating_parameters(input_parameters),
"output_parameters": output_parameters,
"name": node_data['name'],
"output_node": node_data['output_node'],
"category": node_data['category'].strip('\n').strip(),
"description": node_data['description']
}
data[".html"] = json2html(data)
if markdownify:
md = markdownify(data[".html"], keep_inline_images_in=True)
md = md.split('\n')[8:]
md = '\n'.join([m for m in md if m != ''])
data[".md"] = md
return data
def deep_merge(d1: dict, d2: dict) -> Dict[str, str]:
"""
Deep merge multiple dictionaries recursively.
Args:
*dicts: Variable number of dictionaries to be merged.
Returns:
dict: Merged dictionary.
"""
for key in d2:
if key in d1:
if isinstance(d1[key], dict) and isinstance(d2[key], dict):
deep_merge(d1[key], d2[key])
else:
d1[key] = d2[key]
else:
d1[key] = d2[key]
return d1
def update_nested_dict(d, path, value) -> None:
keys = path.split('.')
current = d
for key in keys[:-1]:
current = current.setdefault(key, {})
last_key = keys[-1]
# Check if the key already exists
if last_key in current and isinstance(current[last_key], dict):
current[last_key].update(value)
else:
current[last_key] = value
# ==============================================================================
# === API RESPONSE ===
# ==============================================================================
class TimedOutException(Exception): pass
class ComfyAPIMessage:
# STASH = {}
MESSAGE = {}
#@classmethod
#def send(cls, ident, message) -> None:
#cls.MESSAGE[str(ident)] = message
@classmethod
def poll(cls, ident, period=0.01, timeout=3) -> Any:
_t = time.perf_counter()
if isinstance(ident, (set, list, tuple, )):
ident = ident[0]
sid = str(ident)
logger.debug(f'sid {sid} -- {cls.MESSAGE}')
while not (sid in cls.MESSAGE) and time.perf_counter() - _t < timeout:
time.sleep(period)
if not (sid in cls.MESSAGE):
# logger.warning(f"message failed {sid}")
raise TimedOutException
dat = cls.MESSAGE.pop(sid)
return dat
def comfy_api_post(route:str, ident:str, data:dict) -> None:
data['id'] = ident
PromptServer.instance.send_sync(route, data)
@PromptServer.instance.routes.get("/jovimetrix")
async def jovimetrix_home(req) -> Any:
data = template_load('home.html')
return web.Response(text=data.template, content_type='text/html')
@PromptServer.instance.routes.get("/jovimetrix/message")
async def jovimetrix_message(req) -> Any:
return web.json_response(ComfyAPIMessage.MESSAGE)
@PromptServer.instance.routes.post("/jovimetrix/message")
async def jovimetrix_message_post(req) -> Any:
json_data = await req.json()
logger.info(json_data)
if (did := json_data.get("id")) is not None:
ComfyAPIMessage.MESSAGE[str(did)] = json_data
return web.json_response(json_data)
return web.json_response({})
@PromptServer.instance.routes.get("/jovimetrix/config")
async def jovimetrix_config(req) -> Any:
global JOV_CONFIG, JOV_CONFIG_FILE
if len(JOV_CONFIG) == 0:
JOV_CONFIG = configLoad(JOV_CONFIG_FILE)
return web.json_response(JOV_CONFIG)
async def object_info(node_class: str, scheme:str, host: str) -> Any:
global COMFYUI_OBJ_DATA
if (info := COMFYUI_OBJ_DATA.get(node_class, None)) is None:
# look up via the route...
url = f"{scheme}://{host}/object_info/{node_class}"
# Make an asynchronous HTTP request using aiohttp.ClientSession
async with ClientSession() as session:
try:
async with session.get(url) as response:
if response.status == 200:
info = await response.json()
if (data := info.get(node_class, None)) is not None:
info = get_node_info(data)
else:
info = {'.html': f"No data for {node_class}"}
COMFYUI_OBJ_DATA[node_class] = info
else:
info = {'.html': f"Failed to get docs {node_class}, status: {response.status}"}
logger.error(info)
except Exception as e:
logger.error(f"Failed to get docs {node_class}")
logger.exception(e)
info = {'.html': f"Failed to get docs {node_class}\n{e}"}
return info
@PromptServer.instance.routes.get("/jovimetrix/doc")
async def jovimetrix_doc(req) -> Any:
for node_class in NODE_CLASS_MAPPINGS.keys():
if COMFYUI_OBJ_DATA.get(node_class, None) is None:
COMFYUI_OBJ_DATA[node_class] = await object_info(node_class, req.scheme, req.host)
node = NODE_DISPLAY_NAME_MAPPINGS[node_class]
fname = node.split(" (JOV)")[0]
path = Path(JOV_INTERNAL_DOC.replace("{name}", fname))
path.mkdir(parents=True, exist_ok=True)
if JOV_INTERNAL:
if (md := COMFYUI_OBJ_DATA[node_class].get('.md', None)) is not None:
with open(str(path / f"{fname}.md"), "w", encoding='utf-8') as f:
f.write(md)
with open(str(path / f"{fname}.html"), "w", encoding='utf-8') as f:
f.write(COMFYUI_OBJ_DATA[node_class]['.html'])
return web.json_response(COMFYUI_OBJ_DATA)
@PromptServer.instance.routes.get("/jovimetrix/doc/{node}")
async def jovimetrix_doc_node_comfy(req) -> Any:
node_class = req.match_info.get('node')
if COMFYUI_OBJ_DATA.get(node_class, None) is None:
COMFYUI_OBJ_DATA[node_class] = await object_info(node_class, req.scheme, req.host)
return web.Response(text=COMFYUI_OBJ_DATA[node_class]['.html'], content_type='text/html')
# ==============================================================================
# === SUPPORT ===
# ==============================================================================
def parse_reset(ident:str) -> int:
try:
data = ComfyAPIMessage.poll(ident, timeout=0.05)
ret = data.get('cmd', None)
return ret == 'reset'
except TimedOutException as e:
return -1
except Exception as e:
logger.error(str(e))
def configLoad(fname:Path, as_json:bool=True) -> Any | list[str] | None:
try:
with open(fname, 'r', encoding='utf-8') as fn:
if as_json:
return json.load(fn)
return fn.read().splitlines()
except (IOError, FileNotFoundError) as e:
pass
except Exception as e:
logger.error(e)
return []
def load_module(name: str) -> None|ModuleType:
module = inspect.getmodule(inspect.stack()[0][0]).__name__
try:
route = str(name).replace("\\", "/")
route = route.split(f"{module}/core/")[1]
route = route.split('.')[0].replace('/', '.')
except Exception as e:
logger.warning(f"module failed {name}")
logger.warning(str(e))
return
try:
module = f"{module}.core.{route}"
module = importlib.import_module(module)
except Exception as e:
logger.warning(f"module failed {module}")
logger.warning(str(e))
return
return module
def loader():
global JOV_CONFIG, JOV_IGNORE_NODE, NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS
NODE_LIST_MAP = {}
if JOV_CONFIG_FILE.exists():
JOV_CONFIG = configLoad(JOV_CONFIG_FILE)
if JOV_IGNORE_NODE.exists():
JOV_IGNORE_NODE = configLoad(JOV_IGNORE_NODE, False)
else:
JOV_IGNORE_NODE = []
for fname in ROOT.glob('core/**/*.py'):
if fname.stem.startswith('_'):
continue
if fname.stem in JOV_IGNORE_NODE or fname.stem+'.py' in JOV_IGNORE_NODE:
logger.warning(f"💀 [IGNORED] .core.{fname.stem}")
continue
if (module := load_module(fname)) is None:
continue
# check if there is a dynamic register function....
try:
for class_name, class_def in module.import_dynamic():
setattr(module, class_name, class_def)
except Exception as e:
pass
classes = inspect.getmembers(module, inspect.isclass)
for class_name, class_object in classes:
# assume both attrs are good enough....
if not class_name.endswith('BaseNode') and hasattr(class_object, 'NAME') and hasattr(class_object, 'CATEGORY'):
if (name := class_object.NAME) in JOV_IGNORE_NODE:
logger.warning(f"😥 {name}")
continue
name = class_object.NAME
NODE_DISPLAY_NAME_MAPPINGS[name] = name
NODE_CLASS_MAPPINGS[name] = class_object
if not name.endswith(Lexicon.GLSL_CUSTOM):
desc = class_object.DESCRIPTION if hasattr(class_object, 'DESCRIPTION') else name
NODE_LIST_MAP[name] = desc.split('.')[0].strip('\n')
else:
logger.debug(f"customs {name}")
NODE_CLASS_MAPPINGS = {x[0] : x[1] for x in sorted(NODE_CLASS_MAPPINGS.items(),
key=lambda item: getattr(item[1], 'SORT', 0))}
keys = NODE_CLASS_MAPPINGS.keys()
#for name in keys:
# logger.debug(f"✅ {name}")
logger.info(f"{len(keys)} nodes loaded")
# only do the list on local runs...
if JOV_INTERNAL:
with open(str(ROOT) + "/node_list.json", "w", encoding="utf-8") as f:
json.dump(NODE_LIST_MAP, f, sort_keys=True, indent=4 )
# ==============================================================================
# === BOOTSTRAP ===
# ==============================================================================
loader()