diff --git a/__init__.py b/__init__.py index e35278f..9cce517 100644 --- a/__init__.py +++ b/__init__.py @@ -1,13 +1,22 @@ -from .nodes.fetch_and_save_image import FetchAndSaveImage +from .nodes.download_image_from_url import DownloadImageFromURL from .nodes.generate_negative_prompt import GenerateNegativePrompt -from .nodes.groq_api_completion import GroqAPICompletion from .nodes.save_text_file import SaveTextFile +from .nodes.get_file_path import GetFilePath +from .nodes.groq_api_llm import GroqAPILLM +from .nodes.groq_api_vlm import GroqAPIVLM +from .nodes.groq_api_alm_transcribe import GroqAPIALMTranscribe +#from .nodes.groq_api_alm_translate import GroqAPIALMTranslate + NODE_CLASS_MAPPINGS = { + "📁 Get File Path": GetFilePath, "💾 Save Text File With Path": SaveTextFile, - "🖼️ Download Image from URL": FetchAndSaveImage, - "✨ Groq LLM API": GroqAPICompletion, + "🖼️ Download Image from URL": DownloadImageFromURL, + "✨💬 Groq LLM API": GroqAPILLM, + "✨📷 Groq VLM API": GroqAPIVLM, + "✨📝 Groq ALM API - Transcribe": GroqAPIALMTranscribe, + #"✨🌐 Groq ALM API - Translate [EN only]": GroqAPIALMTranslate, "⛔ Generate Negative Prompt": GenerateNegativePrompt, } -print("\033[34mMNeMiC Nodes: \033[92mLoaded\033[0m") +print("\033[34mMNeMiC Nodes: \033[92mLoaded\033[0m") \ No newline at end of file diff --git a/nodes/__init__.py b/nodes/__init__.py index 9d4d668..ddc919b 100644 --- a/nodes/__init__.py +++ b/nodes/__init__.py @@ -1,11 +1,19 @@ -from .fetch_and_save_image import FetchAndSaveImage +from .download_image_from_url import DownloadImageFromURL from .generate_negative_prompt import GenerateNegativePrompt -from .groq_api_completion import GroqAPICompletion from .save_text_file import SaveTextFile +from .get_file_path import GetFilePath +from .groq_api_llm import GroqAPILLM +from .groq_api_vlm import GroqAPIVLM +from .groq_api_alm_transcribe import GroqAPIALMTranscribe +#from .groq_api_alm_translate import GroqAPIALMTranslate __all__ = [ - "FetchAndSaveImage", - "GenerateNegativePrompt", - "GroqAPICompletion", + "DownloadImageFromURL", "SaveTextFile", + "GetFilePath", + "GroqAPILLM", + "GroqAPIVLM", + "GroqAPIALMTranscribe", + #"GroqAPIALMTranslate", + "GenerateNegativePrompt", ] diff --git a/nodes/fetch_and_save_image.py b/nodes/download_image_from_url.py similarity index 80% rename from nodes/fetch_and_save_image.py rename to nodes/download_image_from_url.py index 6c0db77..751a203 100644 --- a/nodes/fetch_and_save_image.py +++ b/nodes/download_image_from_url.py @@ -8,26 +8,28 @@ def pil2tensor(image): return torch.from_numpy(np.array(image).astype(np.float32) / 255.0).unsqueeze(0) -class FetchAndSaveImage: +class DownloadImageFromURL: OUTPUT_NODE = True RETURN_TYPES = ("IMAGE", "INT", "INT") # Image, Width, Height RETURN_NAMES = ("image", "width", "height") - FUNCTION = "FetchAndSaveImage" + OUTPUT_TOOLTIPS = ("The downloaded image", "The width of the image", "The height of the image") + FUNCTION = "DownloadImageFromURL" CATEGORY = "⚡ MNeMiC Nodes" + DESCRIPTION = "Downloads an image from a URL." @classmethod def INPUT_TYPES(cls): return { "required": { - "image_url": ("STRING", {"multiline": False, "default": ""}), + "image_url": ("STRING", {"multiline": False, "default": "", "tooltip": "URL of the image to download."}), }, "optional": { - "save_file_name_override": ("STRING", {"default": "", "multiline": False}), - "save_path": ("STRING", {"default": "", "multiline": False}) + "save_file_name_override": ("STRING", {"default": "", "multiline": False, "tooltip": "Optional override for the name of the saved image file."}), + "save_path": ("STRING", {"default": "", "multiline": False, "tooltip": "Optional path to save the image. Defaults to the current directory."}) } } - def FetchAndSaveImage(self, image_url, save_path='', save_file_name_override=''): + def DownloadImageFromURL(self, image_url, save_path='', save_file_name_override=''): if not image_url: print("Error: No image URL provided.") return None, None, None diff --git a/nodes/generate_negative_prompt.py b/nodes/generate_negative_prompt.py index 72528b0..a5f4032 100644 --- a/nodes/generate_negative_prompt.py +++ b/nodes/generate_negative_prompt.py @@ -11,21 +11,23 @@ def __init__(self): def INPUT_TYPES(cls): return { "required": { - "input_prompt": ("STRING", {"forceInput": True}), - "max_length": ("INT", {"default": 100, "min": 1, "max": 1024, "step": 1}), - "num_beams": ("INT", {"default": 1, "min": 1, "max": 10, "step": 1}), - "temperature": ("FLOAT", {"default": 1.0, "min": 0.1, "max": 2.0, "step": 0.1}), - "top_k": ("INT", {"default": 50, "min": 0, "max": 100, "step": 1}), - "top_p": ("FLOAT", {"default": 0.92, "min": 0.0, "max": 1.0, "step": 0.01}), - "blocked_words": ("STRING", {"default": "Blocked words, one per line, remove unwanted embeddings or words", "multiline": True}), + "input_prompt": ("STRING", {"forceInput": True, "tooltip": "The positive prompt you want to generate a negative prompt for."}), + "max_length": ("INT", {"default": 100, "min": 1, "max": 1024, "step": 1, "tooltip": "Maximum token length of the generated output."}), + "num_beams": ("INT", {"default": 1, "min": 1, "max": 10, "step": 1, "tooltip": "Number of beams for beam search. Higher values improve accuracy."}), + "temperature": ("FLOAT", {"default": 1.0, "min": 0.1, "max": 2.0, "step": 0.1, "tooltip": "Sampling temperature. Lower values make the output more deterministic."}), + "top_k": ("INT", {"default": 50, "min": 0, "max": 100, "step": 1, "tooltip": "Limits how many of the most likely words are considered for each choice.\n\nFor example, top_k=50 means the model picks from the top 50 most likely words.\n\nA lower value narrows the choices, making the output more predictable, while a higher value adds diversity."}), + "top_p": ("FLOAT", {"default": 0.92, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "Limits the pool of words the model can choose from based on their combined probability.\n\nSet it closer to 1 to allow more variety in output. Lowering this (e.g., 0.9) will restrict the output to the most likely words, making responses more focused."}), + "blocked_words": ("STRING", {"default": "Blocked words, one per line, remove unwanted embeddings or words", "multiline": True, "tooltip": "Words to exclude from the output."}), } } OUTPUT_NODE = True RETURN_TYPES = ("STRING",) RETURN_NAMES = ("negative_prompt",) + OUTPUT_TOOLTIPS = ("The generated negative prompt",) FUNCTION = "generate_negative_prompt" CATEGORY = "⚡ MNeMiC Nodes" + DESCRIPTION = "EXPERIMENTAL: Generates a negative prompt matching the input.\n\nThe model is quite weak and random though, so it doesn't work well. It mostly just generates random negative prompts trained on CivitAI negative prompts.\n\nNSFW words may appear." def generate_negative_prompt(self, input_prompt, max_length, num_beams, temperature, top_k, top_p, blocked_words): current_directory = os.path.dirname(os.path.realpath(__file__)) diff --git a/nodes/get_file_path.py b/nodes/get_file_path.py new file mode 100644 index 0000000..acfca92 --- /dev/null +++ b/nodes/get_file_path.py @@ -0,0 +1,78 @@ +import os +from pathlib import Path +from aiohttp import web +import folder_paths + +class GetFilePath: + OUTPUT_NODE = True + RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING") + RETURN_NAMES = ("full_file_path", "file_path_only", "file_name_only", "file_extension_only") + OUTPUT_TOOLTIPS = ("The full path to the file", "The path to the file", "The name of the file", "The extension of the file") + FUNCTION = "get_file_path" + CATEGORY = "⚡ MNeMiC Nodes" + DESCRIPTION = "Gets a file path and returns components of the file path." + DOCUMENTATION = "This is documentation" + + @classmethod + def INPUT_TYPES(cls): + input_dir = folder_paths.get_input_directory() + files = [f for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f))] + return { + "required": { + "file": (sorted(files), {"file_upload": True, "tooltip": "Place your files in the 'input'-folder inside ComfyUI.\n\nBrowsing functionality is not yet supported. Please send help!"}), + } + } + + def get_file_path(self, file): + try: + # Handle file upload within the node logic + uploaded_file_path = self.upload_file(file) + + # Resolve the full file path using folder_paths + full_file_path = Path(uploaded_file_path) + + # Check if the file exists + if not full_file_path.exists(): + print(f"Error: File does not exist: {full_file_path}") + return None, None, None, None + + # Extract file components + file_path_only = str(full_file_path.parent) + file_name_only = full_file_path.stem # File name without the extension + file_extension_only = full_file_path.suffix # File extension + + # Return all as strings + return ( + str(full_file_path), # Full file path + file_path_only, # Path only + file_name_only, # File name without extension + file_extension_only, # File extension + ) + + except Exception as e: + # Handle any unexpected errors + print(f"Error: Failed to process file path. Details: {str(e)}") + return None, None, None, None + + def upload_file(self, file): + try: + # Define where to save uploaded files (e.g., input directory) + input_dir = folder_paths.get_input_directory() + file_path = os.path.join(input_dir, file) + + # Check if file already exists in the directory + if os.path.exists(file_path): + print(f"File {file} already exists in {input_dir}. Skipping upload.") + return file_path + + # Mimic the upload logic + with open(file_path, "wb") as f: + # Here, you would write the file content to disk + f.write(file) # Assuming `file` contains the file data + + print(f"File uploaded successfully: {file_path}") + return file_path + + except Exception as e: + print(f"Error uploading file: {str(e)}") + return None diff --git a/nodes/groq/DefaultPrompts_ALM_Transcribe.json b/nodes/groq/DefaultPrompts_ALM_Transcribe.json new file mode 100644 index 0000000..1fcb4c7 --- /dev/null +++ b/nodes/groq/DefaultPrompts_ALM_Transcribe.json @@ -0,0 +1,10 @@ +[ + { + "name": "Transcribe the song lyrics", + "content": "" + }, + { + "name": "Transcribe meeting notes accurately", + "content": "Write [INAUDIBLE] when unclear." + } +] \ No newline at end of file diff --git a/nodes/groq/DefaultPrompts_ALM_Translate.json b/nodes/groq/DefaultPrompts_ALM_Translate.json new file mode 100644 index 0000000..659a716 --- /dev/null +++ b/nodes/groq/DefaultPrompts_ALM_Translate.json @@ -0,0 +1,6 @@ +[ + { + "name": "Translate the audio file using the style and guidance of [user_input]", + "content": "" + } +] \ No newline at end of file diff --git a/nodes/groq/DefaultPrompts_VLM.json b/nodes/groq/DefaultPrompts_VLM.json new file mode 100644 index 0000000..418a721 --- /dev/null +++ b/nodes/groq/DefaultPrompts_VLM.json @@ -0,0 +1,6 @@ +[ + { + "name": "Describe the attached image following the [user_input] instruction", + "content": "You are a vision-language model. Analyze the attached image and respond to the user request based on their query. If the query is empty, describe the image in a clear and descriptive manner." + } +] \ No newline at end of file diff --git a/nodes/groq/UserPrompts_ALM_Transcribe.json b/nodes/groq/UserPrompts_ALM_Transcribe.json new file mode 100644 index 0000000..f247d04 --- /dev/null +++ b/nodes/groq/UserPrompts_ALM_Transcribe.json @@ -0,0 +1,6 @@ +[ + { + "name": "Add your own presets in UserPrompts.json", + "content": "" + } +] diff --git a/nodes/groq/UserPrompts_ALM_Translate.json b/nodes/groq/UserPrompts_ALM_Translate.json new file mode 100644 index 0000000..f247d04 --- /dev/null +++ b/nodes/groq/UserPrompts_ALM_Translate.json @@ -0,0 +1,6 @@ +[ + { + "name": "Add your own presets in UserPrompts.json", + "content": "" + } +] diff --git a/nodes/groq/UserPrompts_VLM.json b/nodes/groq/UserPrompts_VLM.json new file mode 100644 index 0000000..f247d04 --- /dev/null +++ b/nodes/groq/UserPrompts_VLM.json @@ -0,0 +1,6 @@ +[ + { + "name": "Add your own presets in UserPrompts.json", + "content": "" + } +] diff --git a/nodes/groq_api_alm_transcribe.py b/nodes/groq_api_alm_transcribe.py new file mode 100644 index 0000000..dab8c84 --- /dev/null +++ b/nodes/groq_api_alm_transcribe.py @@ -0,0 +1,194 @@ +# nodes/groq_api_alm_transcribe.py + +import os +import json +import time +from configparser import ConfigParser +from colorama import init, Fore, Style +from groq import Groq +import requests + +from ..utils.api_utils import load_prompt_options, get_prompt_content + +init() # Initialize colorama + +class GroqAPIALMTranscribe: + DEFAULT_PROMPT = "Transcribe the song lyrics with the [user_input] as vocabulary" + + # Supported models for transcription + TRANSCRIPTION_MODELS = [ + "distil-whisper-large-v3-en", + "whisper-large-v3", + ] + + SUPPORTED_AUDIO_FORMATS = ['mp3', 'mp4', 'mpeg', 'mpga', 'm4a', 'wav', 'webm'] + + CLASS_TYPE = "text" # Necessary for node recognition + + def __init__(self): + current_directory = os.path.dirname(os.path.realpath(__file__)) + groq_directory = os.path.join(current_directory, 'groq') + config_path = os.path.join(groq_directory, 'GroqConfig.ini') + self.config = ConfigParser() + self.config.read(config_path) + self.api_key = self.config.get('API', 'key') + self.client = Groq(api_key=self.api_key) + + # Load prompt options + prompt_files = [ + os.path.join(groq_directory, 'DefaultPrompts_ALM_Transcribe.json'), + os.path.join(groq_directory, 'UserPrompts_ALM_Transcribe.json') + ] + self.prompt_options = load_prompt_options(prompt_files) + + @classmethod + def INPUT_TYPES(cls): + try: + current_directory = os.path.dirname(os.path.realpath(__file__)) + groq_directory = os.path.join(current_directory, 'groq') + prompt_files = [ + os.path.join(groq_directory, 'DefaultPrompts_ALM_Transcribe.json'), + os.path.join(groq_directory, 'UserPrompts_ALM_Transcribe.json') + ] + prompt_options = load_prompt_options(prompt_files) + except Exception as e: + print(Fore.RED + f"Failed to load prompt options: {e}" + Style.RESET_ALL) + prompt_options = {} + + return { + "required": { + "model": (cls.TRANSCRIPTION_MODELS, {"tooltip": "Select the transcription model to use."}), + "file_path": ("STRING", {"label": "Audio file path", "multiline": False, "default": "", "tooltip": "Path to the audio file to be transcribed."}), + "preset": ([cls.DEFAULT_PROMPT] + list(prompt_options.keys()), {"tooltip": "Select a preset for the transcription or custom prompts."}), + "user_input": ("STRING", {"label": "User Input (for prompt)", "multiline": True, "default": "", "tooltip": "Optional user input to guide the transcription."}), + "response_format": (["text", "text_with_linebreaks", "text_with_timestamps", "json", "verbose_json"], {"tooltip": "Format in which the transcription response is returned.\n\nText: Only the text, in one text chunk.\n\ntext_with_linebreaks: Only the text, with each line separated by a line break.\n\ntext_with_timestamps: Only the text, with each timestamp is separated by a line break.\n\njson: The JSON response from the API.\n\nverbose_json: The JSON response from the API with more details."}), + "temperature": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "Controls randomness in responses.\n\nA higher temperature makes the model take more risks, leading to more creative or varied answers.\n\nA lower temperature (closer to 0.1) makes the model more focused and predictable."}), + "language": ("STRING", {"label": "Language (ISO 639-1 code, e.g., 'en', 'fr')", "default": "en", "multiline": False, "tooltip": "Language of the audio file in ISO 639-1 code.\nhttps://www.wikiwand.com/en/articles/List_of_ISO_639_language_codes\n\nis tg uz zh ru tr hi la tk haw fr vi cs hu kk he cy bs sw ht mn gl si mg sa es ja pt lt mr fa sl kn uk ms ta hr bg pa yi fo th lv ln ca br sq jv sn gu ba te bn et sd tl ha de hy so oc nn az km yo ko pl da mi ml ka am tt su yue nl no ne mt my ur ps ar id fi el ro as en it sk be lo lb bo sv sr mk eu"}), + "max_retries": ("INT", {"default": 2, "min": 1, "max": 10, "step": 1, "tooltip": "Maximum number of retries in case of transcription failures."}), + } + } + + RETURN_TYPES = ("STRING", "BOOLEAN", "STRING") + RETURN_NAMES = ("transcription_result", "success", "status_code") + OUTPUT_TOOLTIPS = ("The API response. This is the transcription generated by the model", "Whether the request was successful", "The status code of the request") + + FUNCTION = "process_transcription_request" + CATEGORY = "⚡ MNeMiC Nodes" + DESCRIPTION = "Uses Groq API to transcribe audio." + OUTPUT_NODE = True + + def process_transcription_request(self, model, file_path, preset, user_input, response_format, temperature, language, max_retries): + # Validate file path + if not os.path.isfile(file_path): + print(Fore.RED + f"Error: File not found at path {file_path}" + Style.RESET_ALL) + return "File not found.", False, "400 Bad Request" + + # Validate file extension + file_extension = file_path.split('.')[-1].lower() + if file_extension not in self.SUPPORTED_AUDIO_FORMATS: + print(Fore.RED + f"Error: Unsupported audio format '{file_extension}'. Supported formats are: {', '.join(self.SUPPORTED_AUDIO_FORMATS)}" + Style.RESET_ALL) + return f"Unsupported audio format '{file_extension}'.", False, "400 Bad Request" + + # Load the audio file + try: + with open(file_path, 'rb') as audio_file: + audio_data = audio_file.read() + print("Audio file loaded successfully.") + except Exception as e: + print(Fore.RED + f"Error reading audio file: {str(e)}" + Style.RESET_ALL) + return "Error reading audio file.", False, "400 Bad Request" + + # Prepare the prompt + if preset == self.DEFAULT_PROMPT: + prompt = self.DEFAULT_PROMPT.replace('[user_input]', user_input.strip()) if user_input else '' + else: + prompt_template = get_prompt_content(self.prompt_options, preset) + prompt = prompt_template.replace('[user_input]', user_input.strip()) if user_input else prompt_template + + print(f"Using prompt: {prompt}") + + # Limit the prompt to 224 tokens + # if prompt: + # prompt = prompt[:1000] + + # Adjust api_response_format based on response_format + if response_format in ['json', 'verbose_json', 'text']: + api_response_format = response_format + elif response_format in ['text_with_timestamps', 'text_with_linebreaks']: + api_response_format = 'verbose_json' + else: + print(Fore.RED + "Unknown response format selected." + Style.RESET_ALL) + return "Unknown response format.", False, "400 Bad Request" + + url = 'https://api.groq.com/openai/v1/audio/transcriptions' + headers = {'Authorization': f'Bearer {self.api_key}'} + files = {'file': (os.path.basename(file_path), audio_data)} + data = { + 'model': model, + 'response_format': api_response_format, + 'temperature': str(temperature), + 'language': language or 'en' # Default to 'en' if not specified + } + if prompt: + data['prompt'] = prompt + + print(f"Sending request to {url} with data: {data} and headers: {headers}") + + # Send the request + for attempt in range(max_retries): + try: + print(f"Attempt {attempt + 1} of {max_retries}") + response = requests.post(url, headers=headers, data=data, files=files) + print(f"Response status: {response.status_code}") + if response.status_code == 200: + print("Request successful.") + if api_response_format == "text": + if response_format == "text": + # Return plain text as is + return response.text, True, "200 OK" + elif api_response_format in ["json", "verbose_json"]: + try: + response_json = json.loads(response.text) + except Exception as e: + print(Fore.RED + f"Error parsing JSON response: {str(e)}" + Style.RESET_ALL) + return "Error parsing JSON response.", False, "200 OK but failed to parse JSON" + if response_format == "json": + # Return JSON as formatted string + return json.dumps(response_json, indent=4), True, "200 OK" + elif response_format == "verbose_json": + # Return verbose JSON as formatted string + return json.dumps(response_json, indent=4), True, "200 OK" + elif response_format == "text_with_timestamps": + # Process segments to produce line-based timestamps + segments = response_json.get('segments', []) + transcription_text = "" + for segment in segments: + start_time = segment.get('start', 0) + # Convert start_time to minutes:seconds.milliseconds + minutes = int(start_time // 60) + seconds = int(start_time % 60) + milliseconds = int((start_time - int(start_time)) * 1000) + timestamp = f"[{minutes:02d}:{seconds:02d}.{milliseconds:03d}]" + text = segment.get('text', '').strip() + transcription_text += f"{timestamp}{text}\n" + return transcription_text.strip(), True, "200 OK" + elif response_format == "text_with_linebreaks": + # Extract text from each segment and concatenate with line breaks + segments = response_json.get('segments', []) + transcription_text = "" + for segment in segments: + text = segment.get('text', '').strip() + transcription_text += f"{text}\n" + return transcription_text.strip(), True, "200 OK" + else: + print(Fore.RED + "Unknown api_response_format." + Style.RESET_ALL) + return "Unknown api_response_format.", False, "400 Bad Request" + else: + print(Fore.RED + f"Error: {response.status_code} {response.reason}" + Style.RESET_ALL) + print(f"Response body: {response.text}") + return response.text, False, f"{response.status_code} {response.reason}" + except Exception as e: + print(Fore.RED + f"Request failed: {str(e)}" + Style.RESET_ALL) + time.sleep(2) + print(Fore.RED + "Failed after all retries." + Style.RESET_ALL) + return "Failed after all retries.", False, "Failed after all retries" \ No newline at end of file diff --git a/nodes/groq_api_alm_translate.py b/nodes/groq_api_alm_translate.py new file mode 100644 index 0000000..4aa485d --- /dev/null +++ b/nodes/groq_api_alm_translate.py @@ -0,0 +1,180 @@ +# nodes/groq_api_alm_translate.py +import os +import json +import time +from configparser import ConfigParser +from colorama import init, Fore, Style +from groq import Groq +import requests + +from ..utils.api_utils import load_prompt_options, get_prompt_content + +init() # Initialize colorama + +class GroqAPIALMTranslate: + DEFAULT_PROMPT = "Translate the audio file using the style and guidance of [user_input]" + + # Only whisper-large-v3 supports translation + TRANSLATION_MODELS = [ + "whisper-large-v3", + ] + + SUPPORTED_AUDIO_FORMATS = ['mp3', 'mp4', 'mpeg', 'mpga', 'm4a', 'wav', 'webm'] + + CLASS_TYPE = "text" # Added CLASS_TYPE property + + def __init__(self): + current_directory = os.path.dirname(os.path.realpath(__file__)) + groq_directory = os.path.join(current_directory, 'groq') + config_path = os.path.join(groq_directory, 'GroqConfig.ini') + self.config = ConfigParser() + self.config.read(config_path) + self.api_key = self.config.get('API', 'key') + self.client = Groq(api_key=self.api_key) + + # Load prompt options + prompt_files = [ + os.path.join(groq_directory, 'DefaultPrompts_ALM_Translate.json'), + os.path.join(groq_directory, 'UserPrompts_ALM_Translate.json') + ] + self.prompt_options = load_prompt_options(prompt_files) + + @classmethod + def INPUT_TYPES(cls): + try: + current_directory = os.path.dirname(os.path.realpath(__file__)) + groq_directory = os.path.join(current_directory, 'groq') + prompt_files = [ + os.path.join(groq_directory, 'DefaultPrompts_ALM_Translate.json'), + os.path.join(groq_directory, 'UserPrompts_ALM_Translate.json') + ] + prompt_options = load_prompt_options(prompt_files) + except Exception as e: + print(Fore.RED + f"Failed to load prompt options: {e}" + Style.RESET_ALL) + prompt_options = {} + + return { + "required": { + "model": (cls.TRANSLATION_MODELS, {"tooltip": "Select the translation model to use."}), + "file_path": ("STRING", {"label": "Audio file path", "multiline": False, "default": "", "tooltip": "Path to the audio file for translation."}), + "preset": ([cls.DEFAULT_PROMPT] + list(prompt_options.keys()), {"tooltip": "Select a preset or custom prompt for guiding the translation."}), + "user_input": ("STRING", {"label": "User Input (for prompt)", "multiline": True, "default": "", "tooltip": "Optional user input to guide the translation process."}), + "response_format": (["json", "verbose_json", "text", "text_with_timestamps", "text_with_linebreaks"], {"tooltip": "Choose the format in which the translation output will be returned."}), + "temperature": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.1, "tooltip": "Controls randomness in responses.\n\nA higher temperature makes the model take more risks, leading to more creative or varied answers.\n\nA lower temperature (closer to 0.1) makes the model more focused and predictable."}), + "max_retries": ("INT", {"default": 2, "min": 1, "max": 10, "step": 1, "tooltip": "Maximum number of retries in case of failures."}), + } + } + + RETURN_TYPES = ("STRING", "BOOLEAN", "STRING") + RETURN_NAMES = ("translation_result", "success", "status_code") + OUTPUT_TOOLTIPS = ("The API response. This is the translation generated by the model", "Whether the request was successful", "The status code of the request") + FUNCTION = "process_translation_request" + CATEGORY = "⚡ MNeMiC Nodes" + DESCRIPTION = "Uses Groq API to translate audio." + OUTPUT_NODE = True + + def process_translation_request(self, model, file_path, preset, user_input, response_format, temperature, max_retries): + # Validate file path + if not os.path.isfile(file_path): + print(Fore.RED + f"Error: File not found at path {file_path}" + Style.RESET_ALL) + return "File not found.", False, "400 Bad Request" + + # Validate file extension + file_extension = file_path.split('.')[-1].lower() + if file_extension not in self.SUPPORTED_AUDIO_FORMATS: + print(Fore.RED + f"Error: Unsupported audio format '{file_extension}'. Supported formats are: {', '.join(self.SUPPORTED_AUDIO_FORMATS)}" + Style.RESET_ALL) + return f"Unsupported audio format '{file_extension}'.", False, "400 Bad Request" + + # Load the audio file + try: + with open(file_path, 'rb') as audio_file: + audio_data = audio_file.read() + except Exception as e: + print(Fore.RED + f"Error reading audio file: {str(e)}" + Style.RESET_ALL) + return "Error reading audio file.", False, "400 Bad Request" + + # Prepare the prompt + if preset == self.DEFAULT_PROMPT: + prompt = self.DEFAULT_PROMPT.replace('[user_input]', user_input.strip()) if user_input else None + else: + prompt_template = get_prompt_content(self.prompt_options, preset) + prompt = prompt_template.replace('[user_input]', user_input.strip()) + + # Limit the prompt to 224 tokens + # if prompt: + # prompt = prompt[:1000] + + # Adjust api_response_format based on response_format + if response_format in ['json', 'verbose_json', 'text']: + api_response_format = response_format + elif response_format in ['text_with_timestamps', 'text_with_linebreaks']: + api_response_format = 'verbose_json' + else: + return "Unknown response format.", False, "400 Bad Request" + + url = 'https://api.groq.com/openai/v1/audio/translations' + headers = {'Authorization': f'Bearer {self.api_key}'} + files = {'file': (os.path.basename(file_path), audio_data)} + data = { + 'model': model, + 'response_format': api_response_format, + 'temperature': str(temperature), + } + if prompt: + data['prompt'] = prompt + + print(f"Sending request to {url} with data: {data} and headers: {headers}") + + # Send the request + for attempt in range(max_retries): + try: + response = requests.post(url, headers=headers, data=data, files=files) + print(f"Response status: {response.status_code}") + if response.status_code == 200: + if api_response_format == "text": + if response_format == "text": + # Return plain text as is + return response.text, True, "200 OK" + elif api_response_format in ["json", "verbose_json"]: + try: + response_json = json.loads(response.text) + except Exception as e: + print(Fore.RED + f"Error parsing JSON response: {str(e)}" + Style.RESET_ALL) + return "Error parsing JSON response.", False, "200 OK but failed to parse JSON" + if response_format == "json": + # Return JSON as formatted string + return json.dumps(response_json, indent=4), True, "200 OK" + elif response_format == "verbose_json": + # Return verbose JSON as formatted string + return json.dumps(response_json, indent=4), True, "200 OK" + elif response_format == "text_with_timestamps": + # Process segments to produce line-based timestamps + segments = response_json.get('segments', []) + translation_text = "" + for segment in segments: + start_time = segment.get('start', 0) + # Convert start_time to minutes:seconds.milliseconds + minutes = int(start_time // 60) + seconds = int(start_time % 60) + milliseconds = int((start_time - int(start_time)) * 1000) + timestamp = f"[{minutes:02d}:{seconds:02d}.{milliseconds:03d}]" + text = segment.get('text', '').strip() + translation_text += f"{timestamp}{text}\n" + return translation_text.strip(), True, "200 OK" + elif response_format == "text_with_linebreaks": + # Extract text from each segment and concatenate with line breaks + segments = response_json.get('segments', []) + translation_text = "" + for segment in segments: + text = segment.get('text', '').strip() + translation_text += f"{text}\n" + return translation_text.strip(), True, "200 OK" + else: + return "Unknown api_response_format.", False, "400 Bad Request" + else: + print(Fore.RED + f"Error: {response.status_code} {response.reason}" + Style.RESET_ALL) + return response.text, False, f"{response.status_code} {response.reason}" + except Exception as e: + print(Fore.RED + f"Request failed: {str(e)}" + Style.RESET_ALL) + time.sleep(2) + return "Failed after all retries.", False, "Failed after all retries" diff --git a/nodes/groq_api_completion.py b/nodes/groq_api_completion.py index ca6483e..90c8ef1 100644 --- a/nodes/groq_api_completion.py +++ b/nodes/groq_api_completion.py @@ -2,8 +2,13 @@ import time import json import requests +import random +import numpy as np +import torch from configparser import ConfigParser from groq import Groq +import base64 +from PIL import Image class GroqAPICompletion: DEFAULT_PROMPT = "Use [system_message] and [user_input]" @@ -20,6 +25,12 @@ def __init__(self): @classmethod def INPUT_TYPES(cls): + try: + prompt_options = cls.load_prompt_options() + except Exception as e: + print(f"Failed to load prompt options: {e}") + prompt_options = {} + return { "required": { "model": ([ @@ -34,18 +45,22 @@ def INPUT_TYPES(cls): "gemma-7b-it", "gemma2-9b-it", "whisper-large-v3", - "distil-whisper-large-v3-en" + "distil-whisper-large-v3-en", + "llava-v1.5-7b-4096-preview" ],), - "preset": ([cls.DEFAULT_PROMPT] + list(cls.load_prompt_options().keys()),), + "preset": ([cls.DEFAULT_PROMPT] + list(prompt_options.keys()),), "system_message": ("STRING", {"multiline": True, "default": ""}), "user_input": ("STRING", {"multiline": True, "default": ""}), - "temperature": ("FLOAT", {"default": 0.85, "min": 0.1, "max": 2.0, "step": 0.05}), # Max value is 2.0 + "temperature": ("FLOAT", {"default": 0.85, "min": 0.1, "max": 2.0, "step": 0.05}), "max_tokens": ("INT", {"default": 1024, "min": 1, "max": 131072, "step": 1}), - "top_p": ("FLOAT", {"default": 1.0, "min": 0.1, "max": 1.0, "step": 0.01}), # Max value is 1.0 - "seed": ("INT", {"default": 42, "min": 0, "max": 4294967295}), # Max value is 4294967295 + "top_p": ("FLOAT", {"default": 1.0, "min": 0.1, "max": 1.0, "step": 0.01}), + "seed": ("INT", {"default": 42, "min": 0, "max": 4294967295}), "max_retries": ("INT", {"default": 2, "min": 1, "max": 10, "step": 1}), "stop": ("STRING", {"default": ""}), - "json_mode": ("BOOLEAN", {"default": False}) + "json_mode": ("BOOLEAN", {"default": False}), + }, + "optional": { + "image": ("IMAGE", {"label": "Image (optional - for llava only)", "optional": True}), } } @@ -55,28 +70,59 @@ def INPUT_TYPES(cls): FUNCTION = "process_completion_request" CATEGORY = "⚡ MNeMiC Nodes" - def process_completion_request(self, model, preset, system_message, user_input, temperature, max_tokens, top_p, seed, max_retries, stop, json_mode): + def process_completion_request(self, model, preset, system_message, user_input, temperature, max_tokens, top_p, seed, max_retries, stop, json_mode, image=None): + # Set the seed for reproducibility + torch.manual_seed(seed) + np.random.seed(seed) + random.seed(seed) + system_message = system_message if preset == self.DEFAULT_PROMPT else self.get_prompt_content(preset) + url = 'https://api.groq.com/openai/v1/chat/completions' headers = {'Authorization': f'Bearer {self.api_key}'} + + messages = [ + {"role": "system", "content": system_message}, + {"role": "user", "content": user_input} + ] + + # If the selected model is llava-v1.5-7b-4096-preview, include the image + if model == "llava-v1.5-7b-4096-preview": + if image is not None and isinstance(image, torch.Tensor): + # Process the image only if it is provided + image_pil = self.tensor_to_pil(image) + base64_image = self.encode_image(image_pil) + + if base64_image: + combined_message = f"{system_message}\n{user_input}" + + # Send one single message containing both text and image + image_content = { + "role": "user", + "content": [ + {"type": "text", "text": combined_message}, + { + "type": "image_url", + "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"} + } + ] + } + messages = [image_content] + data = { 'model': model, - 'messages': [ - {"role": "system", "content": system_message}, - {"role": "user", "content": user_input} - ], + 'messages': messages, 'temperature': temperature, 'max_tokens': max_tokens, 'top_p': top_p, 'seed': seed } + if stop: # Only add stop if it's not empty data['stop'] = stop - if json_mode: # Only add response_format if JSON mode is True - data['response_format'] = {"type": "json_object"} - + print(f"Sending request to {url} with data: {json.dumps(data, indent=4)} and headers: {headers}") - + for attempt in range(max_retries): response = requests.post(url, headers=headers, json=data) print(f"Response status: {response.status_code}, Response body: {response.text}") @@ -96,7 +142,7 @@ def process_completion_request(self, model, preset, system_message, user_input, return "ERROR", False, f"{response.status_code} {response.reason}" time.sleep(2) - + return "Failed after all retries.", False, "Failed after all retries" @classmethod @@ -117,3 +163,38 @@ def load_prompt_options(cls): def get_prompt_content(self, prompt_name): return self.prompt_options.get(prompt_name, "No content found for selected prompt") + + # Function to encode image in base64 + def encode_image(self, image_path): + try: + with open(image_path, "rb") as image_file: + return base64.b64encode(image_file.read()).decode('utf-8') + except Exception as e: + print(f"Error encoding image: {e}") + return None + + def tensor_to_pil(self, image_tensor): + # Remove batch dimension if it exists (tensor shape [1, H, W, C]) + if image_tensor.ndim == 4 and image_tensor.shape[0] == 1: + image_tensor = image_tensor.squeeze(0) # Remove the batch dimension + + # Ensure the tensor is in the form [H, W, C] (height, width, channels) + if image_tensor.ndim == 3 and image_tensor.shape[2] == 3: # Expecting RGB image with 3 channels + image_array = image_tensor.cpu().numpy() + image_array = (image_array * 255).astype(np.uint8) # Convert from [0, 1] to [0, 255] + return Image.fromarray(image_array) + else: + raise TypeError(f"Unsupported image tensor shape: {image_tensor.shape}") + + # Save the image locally for debugging + def save_image(self, image_pil, filename="vlm_image_temp.png"): + try: + current_directory = os.path.dirname(os.path.realpath(__file__)) + groq_directory = os.path.join(current_directory, 'groq') + image_path = os.path.join(groq_directory, filename) + image_pil.save(image_path) + print(f"Image saved at {image_path}") + return image_path + except Exception as e: + print(f"Error saving image: {e}") + return None \ No newline at end of file diff --git a/nodes/groq_api_llm.py b/nodes/groq_api_llm.py new file mode 100644 index 0000000..c364878 --- /dev/null +++ b/nodes/groq_api_llm.py @@ -0,0 +1,118 @@ +import os +import json +import random +import numpy as np +import torch +from colorama import init, Fore, Style +from configparser import ConfigParser +from groq import Groq + +from ..utils.api_utils import make_api_request, load_prompt_options, get_prompt_content + +init() # Initialize colorama + +class GroqAPILLM: + DEFAULT_PROMPT = "Use [system_message] and [user_input]" + + LLM_MODELS = [ + "llama-3.1-8b-instant", + "llama-3.1-70b-versatile", + "llama3-8b-8192", + "llama3-70b-8192", + "llama-guard-3-8b", + "llama3-groq-8b-8192-tool-use-preview", + "llama3-groq-70b-8192-tool-use-preview", + "mixtral-8x7b-32768", + "gemma-7b-it", + "gemma2-9b-it", + ] + + def __init__(self): + current_directory = os.path.dirname(os.path.realpath(__file__)) + groq_directory = os.path.join(current_directory, 'groq') + config_path = os.path.join(groq_directory, 'GroqConfig.ini') + self.config = ConfigParser() + self.config.read(config_path) + self.api_key = self.config.get('API', 'key') + self.client = Groq(api_key=self.api_key) + + # Load prompt options + prompt_files = [ + os.path.join(groq_directory, 'DefaultPrompts.json'), + os.path.join(groq_directory, 'UserPrompts.json') + ] + self.prompt_options = load_prompt_options(prompt_files) + + @classmethod + def INPUT_TYPES(cls): + try: + current_directory = os.path.dirname(os.path.realpath(__file__)) + groq_directory = os.path.join(current_directory, 'groq') + prompt_files = [ + os.path.join(groq_directory, 'DefaultPrompts.json'), + os.path.join(groq_directory, 'UserPrompts.json') + ] + prompt_options = load_prompt_options(prompt_files) + except Exception as e: + print(Fore.RED + f"Failed to load prompt options: {e}" + Style.RESET_ALL) + prompt_options = {} + + return { + "required": { + "model": (cls.LLM_MODELS, {"tooltip": "Select the Large Language Model (LLM) to use."}), + "preset": ([cls.DEFAULT_PROMPT] + list(prompt_options.keys()), {"tooltip": "Select a preset or custom prompt for guiding the LLM."}), + "system_message": ("STRING", {"multiline": True, "default": "", "tooltip": "Optional system message to guide the LLM's behavior."}), + "user_input": ("STRING", {"multiline": True, "default": "", "tooltip": "User input or prompt to generate a response from the LLM."}), + "temperature": ("FLOAT", {"default": 0.85, "min": 0.1, "max": 2.0, "step": 0.05, "tooltip": "Controls randomness in responses.\n\nA higher temperature makes the model take more risks, leading to more creative or varied answers.\n\nA lower temperature (closer to 0.1) makes the model more focused and predictable."}), + "max_tokens": ("INT", {"default": 1024, "min": 1, "max": 131072, "step": 1, "tooltip": "Maximum number of tokens to generate in the response."}), + "top_p": ("FLOAT", {"default": 1.0, "min": 0.1, "max": 1.0, "step": 0.01, "tooltip": "Limits the pool of words the model can choose from based on their combined probability.\n\nSet it closer to 1 to allow more variety in output. Lowering this (e.g., 0.9) will restrict the output to the most likely words, making responses more focused."}), + "seed": ("INT", {"default": 42, "min": 0, "max": 4294967295, "tooltip": "Seed for random number generation, ensuring reproducibility."}), + "max_retries": ("INT", {"default": 2, "min": 1, "max": 10, "step": 1, "tooltip": "Maximum number of retries in case of request failure."}), + "stop": ("STRING", {"default": "", "tooltip": "Stop generation when the specified sequence is encountered."}), + "json_mode": ("BOOLEAN", {"default": False, "tooltip": "Enable JSON mode for structured output.\n\nIMPORTANT: Requires you to use the word 'JSON' in the prompt."}), + } + } + + OUTPUT_NODE = True + RETURN_TYPES = ("STRING", "BOOLEAN", "STRING") + RETURN_NAMES = ("api_response", "success", "status_code") + OUTPUT_TOOLTIPS = ("The API response. This is the text generated by the model", "Whether the request was successful", "The status code of the request") + FUNCTION = "process_completion_request" + CATEGORY = "⚡ MNeMiC Nodes" + DESCRIPTION = "Uses Groq API to generate text from language models." + + def process_completion_request(self, model, preset, system_message, user_input, temperature, max_tokens, top_p, seed, max_retries, stop, json_mode): + # Set the seed for reproducibility + torch.manual_seed(seed) + np.random.seed(seed) + random.seed(seed) + + if preset == self.DEFAULT_PROMPT: + system_message = system_message + else: + system_message = get_prompt_content(self.prompt_options, preset) + + url = 'https://api.groq.com/openai/v1/chat/completions' + headers = {'Authorization': f'Bearer {self.api_key}'} + + messages = [ + {"role": "system", "content": system_message}, + {"role": "user", "content": user_input} + ] + + data = { + 'model': model, + 'messages': messages, + 'temperature': temperature, + 'max_tokens': max_tokens, + 'top_p': top_p, + 'seed': seed + } + + if stop: # Only add stop if it's not empty + data['stop'] = stop + + print(f"Sending request to {url} with data: {json.dumps(data, indent=4)} and headers: {headers}") + + assistant_message, success, status_code = make_api_request(data, headers, url, max_retries) + return assistant_message, success, status_code \ No newline at end of file diff --git a/nodes/groq_api_vlm.py b/nodes/groq_api_vlm.py new file mode 100644 index 0000000..6008a9d --- /dev/null +++ b/nodes/groq_api_vlm.py @@ -0,0 +1,131 @@ +import os +import json +import random +import numpy as np +import torch +from colorama import init, Fore, Style +from configparser import ConfigParser +from groq import Groq + +from ..utils.api_utils import make_api_request, load_prompt_options, get_prompt_content +from ..utils.image_utils import encode_image, tensor_to_pil + +init() # Initialize colorama + +class GroqAPIVLM: + DEFAULT_PROMPT = "Use [system_message] and [user_input]" + + VLM_MODELS = [ + "llava-v1.5-7b-4096-preview", + ] + + def __init__(self): + current_directory = os.path.dirname(os.path.realpath(__file__)) + groq_directory = os.path.join(current_directory, 'groq') + config_path = os.path.join(groq_directory, 'GroqConfig.ini') + self.config = ConfigParser() + self.config.read(config_path) + self.api_key = self.config.get('API', 'key') + self.client = Groq(api_key=self.api_key) + + # Load prompt options + prompt_files = [ + os.path.join(groq_directory, 'DefaultPrompts_VLM.json'), + os.path.join(groq_directory, 'UserPrompts_VLM.json') + ] + self.prompt_options = load_prompt_options(prompt_files) + + @classmethod + def INPUT_TYPES(cls): + try: + current_directory = os.path.dirname(os.path.realpath(__file__)) + groq_directory = os.path.join(current_directory, 'groq') + prompt_files = [ + os.path.join(groq_directory, 'DefaultPrompts_VLM.json'), + os.path.join(groq_directory, 'UserPrompts_VLM.json') + ] + prompt_options = load_prompt_options(prompt_files) + except Exception as e: + print(Fore.RED + f"Failed to load prompt options: {e}" + Style.RESET_ALL) + prompt_options = {} + + return { + "required": { + "model": (cls.VLM_MODELS, {"tooltip": "Select the Vision-Language Model (VLM) to use."}), + "preset": ([cls.DEFAULT_PROMPT] + list(prompt_options.keys()), {"tooltip": "Select a preset prompt or use a custom prompt for the model."}), + "system_message": ("STRING", {"multiline": True, "default": "", "tooltip": "Optional system message to guide model behavior."}), + "user_input": ("STRING", {"multiline": True, "default": "", "tooltip": "User input or prompt for the model to generate a response."}), + "image": ("IMAGE", {"label": "Image (required for VLM models)", "tooltip": "Upload an image for processing by the VLM model."}), + "temperature": ("FLOAT", {"default": 0.85, "min": 0.1, "max": 2.0, "step": 0.05, "tooltip": "Controls randomness in responses.\n\nA higher temperature makes the model take more risks, leading to more creative or varied answers.\n\nA lower temperature (closer to 0.1) makes the model more focused and predictable."}), + "max_tokens": ("INT", {"default": 1024, "min": 1, "max": 131072, "step": 1, "tooltip": "Maximum number of tokens to generate in the output."}), + "top_p": ("FLOAT", {"default": 1.0, "min": 0.1, "max": 1.0, "step": 0.01, "tooltip": "Limits the pool of words the model can choose from based on their combined probability.\n\nSet it closer to 1 to allow more variety in output. Lowering this (e.g., 0.9) will restrict the output to the most likely words, making responses more focused."}), + "seed": ("INT", {"default": 42, "min": 0, "max": 4294967295, "tooltip": "Seed for random number generation, ensuring reproducibility."}), + "max_retries": ("INT", {"default": 2, "min": 1, "max": 10, "step": 1, "tooltip": "Maximum number of retries in case of failures."}), + "stop": ("STRING", {"default": "", "tooltip": "Stop generation when the specified sequence is encountered."}), + "json_mode": ("BOOLEAN", {"default": False, "tooltip": "Enable JSON mode for structured output.\n\nIMPORTANT: Requires you to use the word 'JSON' in the prompt."}), + } + } + + OUTPUT_NODE = True + RETURN_TYPES = ("STRING", "BOOLEAN", "STRING") + RETURN_NAMES = ("api_response", "success", "status_code") + OUTPUT_TOOLTIPS = ("The API response. This is the description of your input image generated by the model", "Whether the request was successful", "The status code of the request") + FUNCTION = "process_completion_request" + CATEGORY = "⚡ MNeMiC Nodes" + DESCRIPTION = "Uses Groq API for image processing." + + def process_completion_request(self, model, image, temperature, max_tokens, top_p, seed, max_retries, stop, json_mode, preset="", system_message="", user_input=""): + # Set the seed for reproducibility + torch.manual_seed(seed) + np.random.seed(seed) + random.seed(seed) + + if preset == self.DEFAULT_PROMPT: + system_message = system_message + else: + system_message = get_prompt_content(self.prompt_options, preset) + + url = 'https://api.groq.com/openai/v1/chat/completions' + headers = {'Authorization': f'Bearer {self.api_key}'} + + if image is not None and isinstance(image, torch.Tensor): + # Process the image + image_pil = tensor_to_pil(image) + base64_image = encode_image(image_pil) + if base64_image: + combined_message = f"{system_message}\n{user_input}" + # Send one single message containing both text and image + image_content = { + "role": "user", + "content": [ + {"type": "text", "text": combined_message}, + { + "type": "image_url", + "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"} + } + ] + } + messages = [image_content] + else: + print(Fore.RED + "Failed to encode image." + Style.RESET_ALL) + messages = [] + else: + print(Fore.RED + "Image is required for VLM models." + Style.RESET_ALL) + return "Image is required for VLM models.", False, "400 Bad Request" + + data = { + 'model': model, + 'messages': messages, + 'temperature': temperature, + 'max_tokens': max_tokens, + 'top_p': top_p, + 'seed': seed + } + + if stop: # Only add stop if it's not empty + data['stop'] = stop + + #print(f"Sending request to {url} with data: {json.dumps(data, indent=4)} and headers: {headers}") + + assistant_message, success, status_code = make_api_request(data, headers, url, max_retries) + return assistant_message, success, status_code \ No newline at end of file diff --git a/nodes/save_text_file.py b/nodes/save_text_file.py index 3fd6b3d..d321d9d 100644 --- a/nodes/save_text_file.py +++ b/nodes/save_text_file.py @@ -12,21 +12,23 @@ def __init__(self): def INPUT_TYPES(cls): return { "required": { - "file_text": ("STRING", {"forceInput": True}), - "path": ("STRING", {"default": '[time(%Y-%m-%d)]/', "multiline": False}), - "prefix": ("STRING", {"default": "[time(%Y-%m-%d - %H.%M.%S)]"}), - "counter_separator": ("STRING", {"default": "_"}), - "counter_length": ("INT", {"default": 3, "min": 0, "max": 24, "step": 1}), - "suffix": ("STRING", {"default": ""}), - "output_extension": ("STRING", {"default": "txt"}) + "file_text": ("STRING", {"forceInput": True, "tooltip": "The text to save."}), + "path": ("STRING", {"default": '[time(%Y-%m-%d)]/', "multiline": False, "tooltip": "The folder path to save the file to.\n\nThe following creates a date folder:\n[time(%Y-%m-%d)]"}), + "prefix": ("STRING", {"default": "[time(%Y-%m-%d - %H.%M.%S)]", "tooltip": "The prefix to add to the file name.\n\nThe following creates a file with a date and timestamp:\n[time(%Y-%m-%d - %H.%M.%S)]"}), + "counter_separator": ("STRING", {"default": "_", "tooltip": "The separator to use between the file name and the counter."}), + "counter_length": ("INT", {"default": 3, "min": 0, "max": 24, "step": 1, "tooltip": "The number of digits to use in the counter."}), + "suffix": ("STRING", {"default": "", "tooltip": "The suffix to add to the file name after the counter."}), + "output_extension": ("STRING", {"default": "txt", "tooltip": "The extension to use for the output file."}), } - } + } #{prefix}{separator}{counter_str}{separator}{suffix}{extension} OUTPUT_NODE = True RETURN_TYPES = ("STRING", "STRING") RETURN_NAMES = ("output_full_path", "output_name") + OUTPUT_TOOLTIPS = ("The full path to the saved file", "The name of the saved file") FUNCTION = "save_text_file" CATEGORY = "⚡ MNeMiC Nodes" + DESCRIPTION = "Saves text to a file with specified parameters." def save_text_file(self, file_text, path, prefix='[time(%Y-%m-%d %H.%M.%S)]', counter_separator='_', counter_length=3, suffix='', output_extension='txt'): path = replace_tokens(path)