diff --git a/core/calc.py b/core/calc.py index 2203a0f..322e0b0 100644 --- a/core/calc.py +++ b/core/calc.py @@ -3,7 +3,6 @@ Jovimetrix - http://www.github.com/amorano/jovimetrix Calculation """ -import os import sys import math import random @@ -17,10 +16,9 @@ from scipy.special import gamma from loguru import logger from comfy.utils import ProgressBar -from nodes import interrupt_processing from Jovimetrix import JOV_TYPE_ANY, JOV_TYPE_FULL, JOV_TYPE_NUMBER, \ - JOV_TYPE_VECTOR, Lexicon, JOVBaseNode, ComfyAPIMessage, TimedOutException, \ + JOV_TYPE_VECTOR, Lexicon, JOVBaseNode, \ comfy_message, deep_merge, parse_reset from Jovimetrix.sup.util import EnumConvertType, EnumSwizzle, parse_dynamic, \ @@ -32,17 +30,6 @@ from Jovimetrix.sup.anim import EnumWave, EnumEase, ease_op, wave_op JOV_CATEGORY = "CALC" -# min amount of time before showing the cancel dialog -JOV_DELAY_MIN = 5 -try: JOV_DELAY_MIN = int(os.getenv("JOV_DELAY_MIN", JOV_DELAY_MIN)) -except: pass -JOV_DELAY_MIN = max(1, JOV_DELAY_MIN) - -# max 10 minutes to start -JOV_DELAY_MAX = 600 -try: JOV_DELAY_MAX = int(os.getenv("JOV_DELAY_MAX", JOV_DELAY_MAX)) -except: pass - # ============================================================================== # === LAMBDA === # ============================================================================== @@ -593,56 +580,6 @@ Evaluates two inputs (A and B) with a specified comparison operators and optiona outs = list(outs) return outs, *vals, -class DelayNode(JOVBaseNode): - NAME = "DELAY (JOV) βœ‹πŸ½" - CATEGORY = f"JOVIMETRIX πŸ”ΊπŸŸ©πŸ”΅/{JOV_CATEGORY}" - RETURN_TYPES = (JOV_TYPE_ANY,) - RETURN_NAMES = (Lexicon.PASS_OUT,) - SORT = 240 - DESCRIPTION = """ -Introduce pauses in the workflow that accept an optional input to pass through and a timer parameter to specify the duration of the delay. If no timer is provided, it defaults to a maximum delay. During the delay, it periodically checks for messages to interrupt the delay. Once the delay is completed, it returns the input passed to it. You can disable the screensaver with the `ENABLE` option -""" - - @classmethod - def INPUT_TYPES(cls) -> dict: - d = super().INPUT_TYPES() - d = deep_merge(d, { - "optional": { - Lexicon.PASS_IN: (JOV_TYPE_ANY, {"default": None}), - Lexicon.TIMER: ("INT", {"default" : 0, "mij": -1}), - Lexicon.ENABLE: ("BOOLEAN", {"default": True, "tooltips":"Enable or disable the screensaver"}) - }, - "outputs": { - 0: (Lexicon.PASS_OUT, {"tooltips":"Pass through data when the delay ends"}) - } - }) - return Lexicon._parse(d, cls) - - def run(self, ident, **kw) -> Tuple[Any]: - delay = parse_param(kw, Lexicon.TIMER, EnumConvertType.INT, -1, 0, JOV_DELAY_MAX)[0] - if delay < 0: - delay = JOV_DELAY_MAX - if delay > JOV_DELAY_MIN: - comfy_message(ident, "jovi-delay-user", {"id": ident, "timeout": delay}) - # enable = parse_param(kw, Lexicon.ENABLE, EnumConvertType.BOOLEAN, True) - - step = 1 - pbar = ProgressBar(delay) - while step <= delay: - try: - data = ComfyAPIMessage.poll(ident, timeout=1) - if data.get('id', None) == ident: - if data.get('cmd', False) == False: - interrupt_processing(True) - logger.warning(f"delay [cancelled] ({step}): {ident}") - break - except TimedOutException as _: - if step % 10 == 0: - logger.info(f"delay [continue] ({step}): {ident}") - pbar.update_absolute(step) - step += 1 - return kw[Lexicon.PASS_IN], - class LerpNode(JOVBaseNode): NAME = "LERP (JOV) πŸ”°" CATEGORY = f"JOVIMETRIX πŸ”ΊπŸŸ©πŸ”΅/{JOV_CATEGORY}" diff --git a/core/utility/info.py b/core/utility/info.py index 10e0736..cd7d6b0 100644 --- a/core/utility/info.py +++ b/core/utility/info.py @@ -14,7 +14,7 @@ import matplotlib.pyplot as plt from loguru import logger -from Jovimetrix import JOV_TYPE_ANY, JOV_TYPE_IMAGE, DynamicInputType, \ +from Jovimetrix import JOV_TYPE_ANY, JOV_TYPE_IMAGE, \ Lexicon, JOVBaseNode, deep_merge, parse_reset from Jovimetrix.sup.util import EnumConvertType, decode_tensor, parse_dynamic, \ @@ -233,64 +233,3 @@ Exports and Displays immediate information about images. count, height, width = image.shape cc = 1 return count, width, height, cc, (width, height), (width, height, cc) - -class Passthru(JOVBaseNode): - NAME = "PASSTHRU (JOV) 🚌" - CATEGORY = f"JOVIMETRIX πŸ”ΊπŸŸ©πŸ”΅/{JOV_CATEGORY}" - RETURN_TYPES = () - RETURN_NAMES = () - SORT = 860 - DESCRIPTION = """ -Passes the data into python so it can be probed. -""" - OUTPUT_NODE = True - - @classmethod - def INPUT_TYPES(cls) -> dict: - d = super().INPUT_TYPES() - d = deep_merge(d, { - "optional": { - Lexicon.UNKNOWN: (JOV_TYPE_ANY, {"default": None, "tooltips":"Pass through data."}), - } - }) - return Lexicon._parse(d, cls) - - def run(self, **kw) -> Tuple[Any, ...]: - inout = parse_param(kw, Lexicon.UNKNOWN, EnumConvertType.ANY, [None]) - for x in inout: - logger.info(f"{type(x)}") - # logger.info(dir(x)) - return () - -class RouteNode(JOVBaseNode): - NAME = "ROUTE (JOV) 🚌" - CATEGORY = f"JOVIMETRIX πŸ”ΊπŸŸ©πŸ”΅/{JOV_CATEGORY}" - RETURN_TYPES = ("BUS",) + (JOV_TYPE_ANY,) * 127 - RETURN_NAMES = (Lexicon.ROUTE,) - SORT = 850 - DESCRIPTION = """ -Routes the input data from the optional input ports to the output port, preserving the order of inputs. The `PASS_IN` optional input is directly passed through to the output, while other optional inputs are collected and returned as tuples, preserving the order of insertion. -""" - - @classmethod - def INPUT_TYPES(cls) -> dict: - d = super().INPUT_TYPES() - d = deep_merge(d, { - "optional": DynamicInputType(JOV_TYPE_ANY), - """ - "optional": { - Lexicon.ROUTE: ("BUS", {"default": None, "tooltips":"Pass through another route node to pre-populate the outputs."}), - }, - """ - "outputs": { - 0: (Lexicon.ROUTE, {"tooltips":"Pass through for Route node"}) - } - }) - return Lexicon._parse(d, cls) - - def run(self, **kw) -> Tuple[Any, ...]: - inout = parse_param(kw, Lexicon.ROUTE, EnumConvertType.ANY, [None]) - vars = kw.copy() - vars.pop(Lexicon.ROUTE, None) - vars.pop('ident', None) - return inout, *vars.values(), diff --git a/core/utility/io.py b/core/utility/io.py index dcb925c..f364d6f 100644 --- a/core/utility/io.py +++ b/core/utility/io.py @@ -7,7 +7,7 @@ import os import json from uuid import uuid4 from pathlib import Path -from typing import Any +from typing import Any, Tuple import torch import numpy as np @@ -18,8 +18,11 @@ from loguru import logger from comfy.utils import ProgressBar from folder_paths import get_output_directory +from nodes import interrupt_processing -from Jovimetrix import JOV_TYPE_IMAGE, Lexicon, JOVBaseNode, deep_merge +from Jovimetrix import JOV_TYPE_ANY, JOV_TYPE_IMAGE, Lexicon, JOVBaseNode, \ + ComfyAPIMessage, TimedOutException, DynamicInputType, \ + comfy_message, deep_merge from Jovimetrix.sup.util import EnumConvertType, path_next, parse_param, \ zip_longest_fill @@ -30,6 +33,17 @@ from Jovimetrix.sup.image import tensor2cv, tensor2pil JOV_CATEGORY = "UTILITY" +# min amount of time before showing the cancel dialog +JOV_DELAY_MIN = 5 +try: JOV_DELAY_MIN = int(os.getenv("JOV_DELAY_MIN", JOV_DELAY_MIN)) +except: pass +JOV_DELAY_MIN = max(1, JOV_DELAY_MIN) + +# max 10 minutes to start +JOV_DELAY_MAX = 600 +try: JOV_DELAY_MAX = int(os.getenv("JOV_DELAY_MAX", JOV_DELAY_MAX)) +except: pass + FORMATS = ["gif", "png", "jpg"] if (JOV_GIFSKI := os.getenv("JOV_GIFSKI", None)) is not None: if not os.path.isfile(JOV_GIFSKI): @@ -43,6 +57,56 @@ else: # ============================================================================== +class DelayNode(JOVBaseNode): + NAME = "DELAY (JOV) βœ‹πŸ½" + CATEGORY = f"JOVIMETRIX πŸ”ΊπŸŸ©πŸ”΅/{JOV_CATEGORY}" + RETURN_TYPES = (JOV_TYPE_ANY,) + RETURN_NAMES = (Lexicon.PASS_OUT,) + SORT = 240 + DESCRIPTION = """ +Introduce pauses in the workflow that accept an optional input to pass through and a timer parameter to specify the duration of the delay. If no timer is provided, it defaults to a maximum delay. During the delay, it periodically checks for messages to interrupt the delay. Once the delay is completed, it returns the input passed to it. You can disable the screensaver with the `ENABLE` option +""" + + @classmethod + def INPUT_TYPES(cls) -> dict: + d = super().INPUT_TYPES() + d = deep_merge(d, { + "optional": { + Lexicon.PASS_IN: (JOV_TYPE_ANY, {"default": None}), + Lexicon.TIMER: ("INT", {"default" : 0, "mij": -1}), + Lexicon.ENABLE: ("BOOLEAN", {"default": True, "tooltips":"Enable or disable the screensaver"}) + }, + "outputs": { + 0: (Lexicon.PASS_OUT, {"tooltips":"Pass through data when the delay ends"}) + } + }) + return Lexicon._parse(d, cls) + + def run(self, ident, **kw) -> Tuple[Any]: + delay = parse_param(kw, Lexicon.TIMER, EnumConvertType.INT, -1, 0, JOV_DELAY_MAX)[0] + if delay < 0: + delay = JOV_DELAY_MAX + if delay > JOV_DELAY_MIN: + comfy_message(ident, "jovi-delay-user", {"id": ident, "timeout": delay}) + # enable = parse_param(kw, Lexicon.ENABLE, EnumConvertType.BOOLEAN, True) + + step = 1 + pbar = ProgressBar(delay) + while step <= delay: + try: + data = ComfyAPIMessage.poll(ident, timeout=1) + if data.get('id', None) == ident: + if data.get('cmd', False) == False: + interrupt_processing(True) + logger.warning(f"delay [cancelled] ({step}): {ident}") + break + except TimedOutException as _: + if step % 10 == 0: + logger.info(f"delay [continue] ({step}): {ident}") + pbar.update_absolute(step) + step += 1 + return kw[Lexicon.PASS_IN], + class ExportNode(JOVBaseNode): NAME = "EXPORT (JOV) πŸ“½" CATEGORY = f"JOVIMETRIX πŸ”ΊπŸŸ©πŸ”΅/{JOV_CATEGORY}" @@ -140,6 +204,39 @@ Responsible for saving images or animations to disk. It supports various output img.save(output(format), optimize=optimize) return () +class RouteNode(JOVBaseNode): + NAME = "ROUTE (JOV) 🚌" + CATEGORY = f"JOVIMETRIX πŸ”ΊπŸŸ©πŸ”΅/{JOV_CATEGORY}" + RETURN_TYPES = ("BUS",) + (JOV_TYPE_ANY,) * 127 + RETURN_NAMES = (Lexicon.ROUTE,) + SORT = 850 + DESCRIPTION = """ +Routes the input data from the optional input ports to the output port, preserving the order of inputs. The `PASS_IN` optional input is directly passed through to the output, while other optional inputs are collected and returned as tuples, preserving the order of insertion. +""" + + @classmethod + def INPUT_TYPES(cls) -> dict: + d = super().INPUT_TYPES() + d = deep_merge(d, { + "optional": DynamicInputType(JOV_TYPE_ANY), + """ + "optional": { + Lexicon.ROUTE: ("BUS", {"default": None, "tooltips":"Pass through another route node to pre-populate the outputs."}), + }, + """ + "outputs": { + 0: (Lexicon.ROUTE, {"tooltips":"Pass through for Route node"}) + } + }) + return Lexicon._parse(d, cls) + + def run(self, **kw) -> Tuple[Any, ...]: + inout = parse_param(kw, Lexicon.ROUTE, EnumConvertType.ANY, [None]) + vars = kw.copy() + vars.pop(Lexicon.ROUTE, None) + vars.pop('ident', None) + return inout, *vars.values(), + class SaveOutput(JOVBaseNode): NAME = "SAVE OUTPUT (JOV) πŸ’Ύ" CATEGORY = f"JOVIMETRIX πŸ”ΊπŸŸ©πŸ”΅/{JOV_CATEGORY}" diff --git a/node_list.json b/node_list.json index ba78ce8..a43d4d2 100644 --- a/node_list.json +++ b/node_list.json @@ -39,7 +39,6 @@ "MIDI READER (JOV) \ud83c\udfb9": "Captures MIDI messages from an external MIDI device or controller", "OP BINARY (JOV) \ud83c\udf1f": "Execute binary operations like addition, subtraction, multiplication, division, and bitwise operations on input values, supporting various data types and vector sizes", "OP UNARY (JOV) \ud83c\udfb2": "Perform single function operations like absolute value, mean, median, mode, magnitude, normalization, maximum, or minimum on input values", - "PASSTHRU (JOV) \ud83d\ude8c": "Passes the data into python so it can be probed", "PIXEL MERGE (JOV) \ud83e\udec2": "Combines individual color channels (red, green, blue) along with an optional mask channel to create a composite image", "PIXEL SPLIT (JOV) \ud83d\udc94": "Takes an input image and splits it into its individual color channels (red, green, blue), along with a mask channel", "PIXEL SWAP (JOV) \ud83d\udd03": "Swap pixel values between two input images based on specified channel swizzle operations", diff --git a/sup/image/__init__.py b/sup/image/__init__.py index 51a7d19..0bb810e 100644 --- a/sup/image/__init__.py +++ b/sup/image/__init__.py @@ -11,7 +11,7 @@ from io import BytesIO import math import base64 from enum import Enum -from typing import List, Tuple +from typing import List, Tuple, Union import cv2 import torch @@ -45,9 +45,9 @@ TYPE_iRGBA = Tuple[int, int, int, int] TYPE_fRGB = Tuple[float, float, float] TYPE_fRGBA = Tuple[float, float, float, float] -TYPE_PIXEL = int | float | TYPE_iRGB | TYPE_iRGBA | TYPE_fRGB | TYPE_fRGBA -TYPE_IMAGE = np.ndarray | torch.Tensor -TYPE_VECTOR = TYPE_IMAGE | TYPE_PIXEL +TYPE_PIXEL = Union[int, float, TYPE_iRGB, TYPE_iRGBA, TYPE_fRGB, TYPE_fRGBA] +TYPE_IMAGE = Union[np.ndarray, torch.Tensor] +TYPE_VECTOR = Union[TYPE_IMAGE, TYPE_PIXEL] # ============================================================================== # === ENUMERATION ===