Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ee7ba94a30 | ||
|
|
3810320f6e | ||
|
|
c87c24924e | ||
|
|
e9092b877c | ||
|
|
a18e5e659c | ||
|
|
93f0102897 |
@@ -3,9 +3,9 @@
|
||||
The Prompt PostProcessor (PPP), formerly known as "sd-webui-sendtonegative", is an extension designed to process the prompt, possibly after other extensions have modified it. This extension is compatible with:
|
||||
|
||||
* [AUTOMATIC1111 Stable Diffusion WebUI](https://github.com/AUTOMATIC1111/stable-diffusion-webui)
|
||||
* [SD.Next](https://github.com/vladmandic/automatic).
|
||||
* [Forge](https://github.com/lllyasviel/stable-diffusion-webui-forge)
|
||||
* [reForge](https://github.com/Panchovix/stable-diffusion-webui-reForge)
|
||||
* [SD.Next](https://github.com/vladmandic/automatic).
|
||||
* ...and probably other forks
|
||||
* [ComfyUI](https://github.com/comfyanonymous/ComfyUI)
|
||||
|
||||
@@ -35,6 +35,8 @@ Notes:
|
||||
In SD.Next that means only the *A1111* or *Full* parsers. It will warn you if you use the *Compel* parser.
|
||||
|
||||
Does not recognize tokenizer separators like "TE2:" and "TE3:", so sending to negative prompt from those sections of the prompt will not add them in the corresponding section of the negative prompt.
|
||||
|
||||
ComfyUI only supports natively the attention using parenthesis, so the ones with the braces will be converted. The other constructs are not natively supported but some custom nodes implement them.
|
||||
2. It recognizes wildcards in the *\_\_wildcard\_\_* and *{choice|choice}* formats (and almost everything that [Dynamic Prompts](https://github.com/adieyal/sd-dynamic-prompts) supports).
|
||||
3. It does not create *AND/BREAK* constructs when moving content to the negative prompt.
|
||||
|
||||
@@ -47,6 +49,8 @@ On A1111 compatible webuis:
|
||||
3. Click the Install button
|
||||
4. Restart the webui
|
||||
|
||||
On SD.Next I recommend you disable the native wildcard processing.
|
||||
|
||||
On ComfyUI:
|
||||
|
||||
1. Go to Manager > Custom Nodes Manager
|
||||
@@ -126,9 +130,9 @@ __parameters$$wildcard'filter'(var=value)__
|
||||
|
||||
The parameters, the filter, and the setting of a variable are optional. The parameters follow the same format as for the choices.
|
||||
|
||||
The wildcard identifier can contain globbing formatting, to read multiple wildcards and merge their choices. Note that if there are no parameters specified, the globbing will use the ones from the first wildcard that matches and have parameters (sorted by keys), so if you don't want that you might want to specify them.
|
||||
The wildcard identifier can contain globbing formatting, to read multiple wildcards and merge their choices. Note that if there are no parameters specified, the globbing will use the ones from the first wildcard that matches and have parameters (sorted by keys), so if you don't want that you might want to specify them. Also note that, unlike with Dynamic Prompts, the wildcard name has to be specified with its full path (unless you use globbing).
|
||||
|
||||
The filter can be used to filter specific choices from the wildcard. The filtering works before applying the choice conditions (if any). The surrounding quotes can be single or double. The filter is a comma separated list of an integer (positional choice index) or choice label. You can also compound them with "+". That is, the comma separated items act as an OR and the "+" inside them as an AND. Using labels can simplify the definitions of complex wildcards where you want to have direct access to specific choices on occasion (you don't need to create wildcards for each individual choice). There are some additional formats when using filters. You can specify "^wildcard" as a filter to use the filter of a previous wildcard in the chain. You can start the filter (regular or inherited) with "#" and it will not be applied to the current wildcard choices, but the filter will remain in memory to use by other descendant wildcards. You use "#" and "^" when you want to pass a filter to inner wildcards (see the test files).
|
||||
The filter can be used to filter specific choices from the wildcard. The filtering works before applying the choice conditions (if any). The surrounding quotes can be single or double. The filter is a comma separated list of an integer (positional choice index; zero-based) or choice label. You can also compound them with "+". That is, the comma separated items act as an OR and the "+" inside them as an AND. Using labels can simplify the definitions of complex wildcards where you want to have direct access to specific choices on occasion (you don't need to create wildcards for each individual choice). There are some additional formats when using filters. You can specify "^wildcard" as a filter to use the filter of a previous wildcard in the chain. You can start the filter (regular or inherited) with "#" and it will not be applied to the current wildcard choices, but the filter will remain in memory to use by other descendant wildcards. You use "#" and "^" when you want to pass a filter to inner wildcards (see the test files).
|
||||
|
||||
The variable value only applies during the evaluation of the selected choices and is discarded afterward (the variable keeps its original value if there was one).
|
||||
|
||||
@@ -151,15 +155,34 @@ __path/wildcard(var=value)__ # select 1 choice using the specified variabl
|
||||
|
||||
A wildcard definition can be:
|
||||
|
||||
* A txt file. The wildcard name will be the relative path of the file, without the extension. Each line will be a choice. Lines starting with "#" or empty are ignored.
|
||||
* An array or string inside a json or yaml file. The wildcard name includes the relative folder path of the file (without the name or extension) but also the path of the object inside the file.
|
||||
* A txt file. The wildcard name will be the relative path of the file, without the extension. Each line will be a choice. Lines starting with "#" or empty are ignored. Doesn't support nesting.
|
||||
* An array or scalar value inside a json or yaml file. The wildcard name includes the relative folder path of the file, without the extension, but also the path of the value inside the file (if there is one). If the file contains a dictionary, the filename part is not used for the wildcard name. Supports nesting by having dictionaries inside dictionaries.
|
||||
|
||||
The best format is a yaml file with a dictionary of wildcards inside. An editor supporting yaml syntax is recommended.
|
||||
|
||||
In a choice, the content after a "#" is ignored.
|
||||
|
||||
If the first choice follows the format of wildcard parameters, it will be used as default parameters for that wildcard (see examples in the tests folder). The choices of the wildcard follow the same format as in the choices construct, or the object format of **Dynamic Prompts** (only in structured files). If using the object format for a choice you can use a new "if" property for the condition in addition to the standard "weight" and "text"/"content".
|
||||
If the first choice follows the format of wildcard parameters, it will be used as default parameters for that wildcard (see examples in the tests folder). The choices of the wildcard follow the same format as in the choices construct, or the object format of **Dynamic Prompts** (only in structured files). If using the object format for a choice you can use a new "if" property for the condition, and the "labels" property (an array of strings) in addition to the standard "weight" and "text"/"content".
|
||||
|
||||
```yaml
|
||||
{ labels: ["some_label"], weight: 2, if: "_is_pony", content: "the text" } # "text" property can be used instead of "content"
|
||||
```
|
||||
|
||||
Wildcard parameters in a json/yaml file can also be in object format, and support two additional properties, prefix and suffix:
|
||||
|
||||
```yaml
|
||||
{ sampler: "~", repeating: false, count: 2, prefix: "prefix-", suffix: "-suffix", separator: "/" }
|
||||
{ sampler: "~", repeating: false, from: 2, to: 3, prefix: "prefix-", suffix: "-suffix", separator: "/" }
|
||||
```
|
||||
|
||||
The prefix and suffix are added to the result along with the selected choices and separators. They can contain other constructs, but the separator can't.
|
||||
|
||||
It is recommended to use the object format for the wildcard parameters and for choices with complex options.
|
||||
|
||||
Wildcards can contain just one choice. In json and yaml formats this allows the use of a string value for the keys, rather than an array.
|
||||
|
||||
A choice inside a wildcard can also be a list or a dictionary of one element containing a list. These are considered anonymous wildcards. With a list it will be an anonymous wildcard with no choice options, and with a dictionary the key will be the options for the choice containing the anonymous wildcard and the value the choices of the anonymous wildcard. Anonymous wildcards can help formatting complex choice values that are used in only one place and thus creating a regular wildcard is not necessary. See test.yaml for examples.
|
||||
|
||||
#### Detection of remaining wildcards
|
||||
|
||||
This extension should run after any other wildcard extensions, so if you don't use the internal wildcards processing, any remaining wildcards present in the prompt or negative_prompt at this point must be invalid. Usually you might not notice this problem until you check the image metadata, so this option gives you some ways to detect and treat the problem.
|
||||
@@ -336,6 +359,13 @@ This should still work as intended, and the only negative point i see is the unn
|
||||
|
||||
## Configuration
|
||||
|
||||
### A1111 (and compatible UIs) UI options
|
||||
|
||||
* **Force equal seeds**: Changes the image seeds and variation seeds to be equal to the first of the batch. This allows using the same values for all the images in a batch.
|
||||
* **Unlink seed**: Uses the specified seed for the prompt generation instead of the one from the image.
|
||||
* **Seed**: The seed to use for the prompt generation. If -1 a random one will be used for each image in the batch. This seed is only used for wildcards and choices.
|
||||
* **Variable seed**: If the seed is not -1 you can use this to increase it for the other images in the batch.
|
||||
|
||||
### ComfyUI specific inputs
|
||||
|
||||
* **model**: Connect here the MODEL or a string with the model class name used by ComfyUI. Needed for the model kind system variables.
|
||||
@@ -380,6 +410,8 @@ This should still work as intended, and the only negative point i see is the unn
|
||||
* **Merge attention modifiers (weights) when possible**: it merges attention modifiers when possible (merges into one, multiplying their values). Only merges individually nested modifiers.
|
||||
* **Remove extra spaces**: removes other unnecessary spaces.
|
||||
|
||||
Please note that ComfyUI does not support the BREAK and AND constructs, but the related settings are kept in that UI.
|
||||
|
||||
### Content removal settings
|
||||
|
||||
* **Remove extra network tags**: removes all extra network tags.
|
||||
|
||||
-10
@@ -13,16 +13,6 @@ sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||||
from .ppp_comfyui import PromptPostProcessorComfyUINode
|
||||
|
||||
NODE_CLASS_MAPPINGS = {"ACBPromptPostProcessor": PromptPostProcessorComfyUINode}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {"ACBPromptPostProcessor": "ACB Prompt Post Processor"}
|
||||
|
||||
MANIFEST = {
|
||||
"name": "ACB Prompt Post Processor",
|
||||
"version": PromptPostProcessorComfyUINode.VERSION,
|
||||
"author": "ACB",
|
||||
"project": "https://github.com/acorderob/sd-webui-prompt-postprocessor",
|
||||
"description": "Node for processing prompts",
|
||||
"license": "MIT",
|
||||
}
|
||||
|
||||
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]
|
||||
|
||||
+10
-2
@@ -1,5 +1,13 @@
|
||||
import os
|
||||
import launch
|
||||
|
||||
requirements_filename = os.path.join(os.path.dirname(os.path.realpath(__file__)), "requirements.txt")
|
||||
launch.run_pip(f'install -r "{requirements_filename}"', "requirements for Prompt Post-Processor")
|
||||
|
||||
try:
|
||||
from modules.launch_utils import requirements_met, run_pip # A1111
|
||||
|
||||
if not requirements_met(requirements_filename):
|
||||
run_pip(f'install -r "{requirements_filename}"', "requirements for Prompt Post-Processor")
|
||||
except ImportError:
|
||||
import launch
|
||||
|
||||
launch.run_pip(f'install -r "{requirements_filename}"', "requirements for Prompt Post-Processor")
|
||||
|
||||
@@ -7,14 +7,14 @@ import textwrap
|
||||
import time
|
||||
from collections import namedtuple
|
||||
from enum import Enum
|
||||
from typing import Callable, Optional
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
import lark
|
||||
import lark.parsers
|
||||
import numpy as np
|
||||
|
||||
from ppp_logging import DEBUG_LEVEL
|
||||
from ppp_wildcards import PPPWildcards
|
||||
from ppp_wildcards import PPPWildcard, PPPWildcards
|
||||
|
||||
|
||||
class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-instance-attributes
|
||||
@@ -22,8 +22,28 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
The PromptPostProcessor class is responsible for processing and manipulating prompt strings.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def get_version_from_pyproject() -> str:
|
||||
"""
|
||||
Reads the version from the pyproject.toml file.
|
||||
|
||||
Returns:
|
||||
str: The version string.
|
||||
"""
|
||||
version_str = "0.0.0"
|
||||
try:
|
||||
pyproject_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "pyproject.toml")
|
||||
with open(pyproject_path, "r", encoding="utf-8") as file:
|
||||
for line in file:
|
||||
if line.startswith("version = "):
|
||||
version_str = line.split("=")[1].strip().strip('"')
|
||||
break
|
||||
except Exception as e: # pylint: disable=broad-exception-caught
|
||||
logging.getLogger().exception(e)
|
||||
return version_str
|
||||
|
||||
NAME = "Prompt Post-Processor"
|
||||
VERSION = (2, 7, 0)
|
||||
VERSION = get_version_from_pyproject()
|
||||
|
||||
class IFWILDCARDS_CHOICES(Enum):
|
||||
ignore = "ignore"
|
||||
@@ -42,8 +62,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
self,
|
||||
logger: logging.Logger,
|
||||
interrupt: Optional[Callable],
|
||||
env_info: dict[str, any],
|
||||
options: Optional[dict[str, any]] = None,
|
||||
env_info: dict[str, Any],
|
||||
options: Optional[dict[str, Any]] = None,
|
||||
grammar_content: Optional[str] = None,
|
||||
wildcards_obj: PPPWildcards = None,
|
||||
):
|
||||
@@ -60,7 +80,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
"""
|
||||
self.logger = logger
|
||||
self.rng = np.random.default_rng() # gets seeded on each process prompt call
|
||||
self.the_interrupt = interrupt
|
||||
self.interrupt_callback = interrupt
|
||||
self.options = options
|
||||
self.env_info = env_info
|
||||
self.wildcard_obj = wildcards_obj
|
||||
@@ -116,14 +136,24 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
propagate_positions=True,
|
||||
start="choicesoptions",
|
||||
)
|
||||
self.parser_condition = lark.Lark(
|
||||
grammar_content,
|
||||
propagate_positions=True,
|
||||
start="condition",
|
||||
)
|
||||
self.parser_choicevalue = lark.Lark(
|
||||
grammar_content,
|
||||
propagate_positions=True,
|
||||
start="choicevalue",
|
||||
)
|
||||
self.__init_sysvars()
|
||||
self.user_variables = {}
|
||||
|
||||
def interrupt(self):
|
||||
if self.the_interrupt is not None:
|
||||
self.the_interrupt()
|
||||
if self.interrupt_callback is not None:
|
||||
self.interrupt_callback()
|
||||
|
||||
def formatOutput(self, text: str) -> str:
|
||||
def format_output(self, text: str) -> str:
|
||||
"""
|
||||
Formats the output text by encoding it using unicode_escape and decoding it using utf-8.
|
||||
|
||||
@@ -135,7 +165,19 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
"""
|
||||
return text.encode("unicode_escape").decode("utf-8")
|
||||
|
||||
def is_comfy_ui(self) -> bool:
|
||||
"""
|
||||
Checks if the current environment is ComfyUI.
|
||||
|
||||
Returns:
|
||||
bool: True if the environment is ComfyUI, False otherwise.
|
||||
"""
|
||||
return self.env_info.get("app", "") == "comfyui"
|
||||
|
||||
def __init_sysvars(self):
|
||||
"""
|
||||
Initializes the system variables.
|
||||
"""
|
||||
self.system_variables = {}
|
||||
sdchecks = {
|
||||
"sd1": self.env_info.get("is_sd1", False),
|
||||
@@ -345,6 +387,16 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
return text
|
||||
|
||||
def __processprompts(self, prompt, negative_prompt):
|
||||
"""
|
||||
Process the prompt and negative prompt.
|
||||
|
||||
Args:
|
||||
prompt (str): The prompt.
|
||||
negative_prompt (str): The negative prompt.
|
||||
|
||||
Returns:
|
||||
tuple: A tuple containing the processed prompt and negative prompt.
|
||||
"""
|
||||
self.user_variables = {}
|
||||
|
||||
# Process prompt
|
||||
@@ -359,8 +411,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
|
||||
# Insertions in the negative prompt
|
||||
if self.debug_level == DEBUG_LEVEL.full:
|
||||
self.logger.debug(self.formatOutput(f"New negative additions: {p_processor.add_at}"))
|
||||
self.logger.debug(self.formatOutput(f"New negative indexes: {n_processor.insertion_at}"))
|
||||
self.logger.debug(self.format_output(f"New negative additions: {p_processor.add_at}"))
|
||||
self.logger.debug(self.format_output(f"New negative indexes: {n_processor.insertion_at}"))
|
||||
negative_prompt = self.__add_to_insertion_points(
|
||||
negative_prompt, p_processor.add_at["insertion_point"], n_processor.insertion_at
|
||||
)
|
||||
@@ -384,9 +436,9 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
ppwl = ", ".join(p_processor.detectedWildcards)
|
||||
npwl = ", ".join(n_processor.detectedWildcards)
|
||||
if foundP:
|
||||
self.logger.error(self.formatOutput(f"In the positive prompt: {ppwl}"))
|
||||
self.logger.error(self.format_output(f"In the positive prompt: {ppwl}"))
|
||||
if foundNP:
|
||||
self.logger.error(self.formatOutput(f"In the negative prompt: {npwl}"))
|
||||
self.logger.error(self.format_output(f"In the negative prompt: {npwl}"))
|
||||
if self.wil_ifwildcards == self.IFWILDCARDS_CHOICES.warn:
|
||||
prompt = self.WILDCARD_WARNING + prompt
|
||||
elif self.wil_ifwildcards == self.IFWILDCARDS_CHOICES.stop:
|
||||
@@ -396,6 +448,13 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
if foundNP:
|
||||
negative_prompt = self.WILDCARD_STOP.format(npwl) + negative_prompt
|
||||
self.interrupt()
|
||||
# Check for special character sequences that should not be in the result
|
||||
compound_prompt = prompt + "\n" + negative_prompt
|
||||
found_sequences = re.findall(r"::|\$\$|\$\{|[{|}]", compound_prompt)
|
||||
if len(found_sequences) > 0:
|
||||
self.logger.warning(
|
||||
f"""Found probably invalid character sequences on the result ({', '.join(map(lambda x: '"' + x + '"', set(found_sequences)))}). Something might be wrong!"""
|
||||
)
|
||||
return prompt, negative_prompt
|
||||
|
||||
def process_prompt(
|
||||
@@ -405,7 +464,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
seed: int = 0,
|
||||
):
|
||||
"""
|
||||
Process the prompt and negative prompt by moving content to the negative prompt, and cleaning up.
|
||||
Initializes the random number generator and processes the prompt and negative prompt.
|
||||
|
||||
Args:
|
||||
original_prompt (str): The original prompt.
|
||||
@@ -417,7 +476,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
"""
|
||||
try:
|
||||
if seed == -1:
|
||||
seed = np.random.randint(0, 2**32)
|
||||
seed = np.random.randint(0, 2**32, dtype=np.int64)
|
||||
self.rng = np.random.default_rng(seed & 0xFFFFFFFF)
|
||||
prompt = original_prompt
|
||||
negative_prompt = original_negative_prompt
|
||||
@@ -425,14 +484,14 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
if self.debug_level != DEBUG_LEVEL.none:
|
||||
self.logger.info(f"System variables: {self.system_variables}")
|
||||
self.logger.info(f"Input seed: {seed}")
|
||||
self.logger.info(self.formatOutput(f"Input prompt: {prompt}"))
|
||||
self.logger.info(self.formatOutput(f"Input negative_prompt: {negative_prompt}"))
|
||||
self.logger.info(self.format_output(f"Input prompt: {prompt}"))
|
||||
self.logger.info(self.format_output(f"Input negative_prompt: {negative_prompt}"))
|
||||
t1 = time.time()
|
||||
prompt, negative_prompt = self.__processprompts(prompt, negative_prompt)
|
||||
t2 = time.time()
|
||||
if self.debug_level != DEBUG_LEVEL.none:
|
||||
self.logger.info(self.formatOutput(f"Result prompt: {prompt}"))
|
||||
self.logger.info(self.formatOutput(f"Result negative_prompt: {negative_prompt}"))
|
||||
self.logger.info(self.format_output(f"Result prompt: {prompt}"))
|
||||
self.logger.info(self.format_output(f"Result negative_prompt: {negative_prompt}"))
|
||||
self.logger.info(f"Process prompt pair time: {t2 - t1:.3f} seconds")
|
||||
|
||||
# Check for constructs not processed due to parsing problems
|
||||
@@ -447,10 +506,22 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
return original_prompt, original_negative_prompt
|
||||
|
||||
def parse_prompt(self, prompt_description: str, prompt: str, parser: lark.Lark, raise_parsing_error: bool = False):
|
||||
"""
|
||||
Parses a prompt using the specified parser.
|
||||
|
||||
Args:
|
||||
prompt_description (str): The description of the prompt.
|
||||
prompt (str): The prompt to be parsed.
|
||||
parser (lark.Lark): The parser to be used.
|
||||
raise_parsing_error (bool): Whether to raise a parsing error.
|
||||
|
||||
Returns:
|
||||
Tree: The parsed prompt.
|
||||
"""
|
||||
t1 = time.time()
|
||||
try:
|
||||
if self.debug_level == DEBUG_LEVEL.full:
|
||||
self.logger.debug(self.formatOutput(f"Parsing {prompt_description}: '{prompt}'"))
|
||||
self.logger.debug(self.format_output(f"Parsing {prompt_description}: '{prompt}'"))
|
||||
parsed_prompt = parser.parse(prompt)
|
||||
# we store the contents so we can use them later even if the meta position is not valid anymore
|
||||
for n in parsed_prompt.iter_subtrees():
|
||||
@@ -462,7 +533,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
except lark.exceptions.UnexpectedInput:
|
||||
if raise_parsing_error:
|
||||
raise
|
||||
self.logger.exception(self.formatOutput(f"Parsing failed on prompt!: {prompt}"))
|
||||
self.logger.exception(self.format_output(f"Parsing failed on prompt!: {prompt}"))
|
||||
t2 = time.time()
|
||||
if self.debug_level == DEBUG_LEVEL.full:
|
||||
self.logger.debug("Tree:\n" + textwrap.indent(re.sub(r"\n$", "", parsed_prompt.pretty()), " "))
|
||||
@@ -488,8 +559,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
self.__ppp = ppp
|
||||
self.AccumulatedShell = namedtuple("AccumulatedShell", ["type", "data"])
|
||||
self.NegTag = namedtuple("NegTag", ["start", "end", "content", "parameters", "shell"])
|
||||
self.__shell: list[self.AccumulatedShell] = []
|
||||
self.__negtags: list[self.NegTag] = []
|
||||
self.__shell: list[self.AccumulatedShell] = [] # type: ignore
|
||||
self.__negtags: list[self.NegTag] = [] # type: ignore
|
||||
self.__already_processed: list[str] = []
|
||||
self.__is_negative = False
|
||||
self.__wildcard_filters = {}
|
||||
@@ -566,42 +637,84 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
return added_result
|
||||
|
||||
def __get_original_node_content(self, node: lark.Tree | lark.Token, default=None) -> str:
|
||||
"""
|
||||
Get the original content of a node.
|
||||
|
||||
Args:
|
||||
node (Tree|Token): The node to get the content from.
|
||||
default: The default value to return if the content is not found.
|
||||
|
||||
Returns:
|
||||
str: The original content of the node.
|
||||
"""
|
||||
return (
|
||||
node.meta.content
|
||||
if hasattr(node, "meta") and node.meta is not None and not node.meta.empty
|
||||
else default
|
||||
)
|
||||
|
||||
def __get_user_variable_value(self, name: str, default="", evaluate=True) -> str:
|
||||
if evaluate:
|
||||
v = self.__ppp.user_variables.get(name, default)
|
||||
def __get_user_variable_value(self, name: str, evaluate=True, visit=False) -> str:
|
||||
"""
|
||||
Get the value of a user variable.
|
||||
|
||||
Args:
|
||||
name (str): The name of the user variable.
|
||||
evaluate (bool): Whether to evaluate the variable.
|
||||
visit (bool): Whether to also visit the variable (add to result).
|
||||
|
||||
Returns:
|
||||
str: The value of the user variable.
|
||||
"""
|
||||
v = self.__ppp.user_variables.get(name, None)
|
||||
if v is not None:
|
||||
visited = False
|
||||
if isinstance(v, lark.Tree):
|
||||
v = self.__visit(v, True)
|
||||
else:
|
||||
v = (
|
||||
self.__ppp.user_variables[name]
|
||||
if isinstance(self.__ppp.user_variables[name], str)
|
||||
else self.__get_original_node_content(
|
||||
self.__ppp.user_variables[name], default or "not evaluated yet"
|
||||
)
|
||||
)
|
||||
if evaluate:
|
||||
v = self.__visit(v, not visit)
|
||||
visited = visit
|
||||
else:
|
||||
v = self.__get_original_node_content(v, "not evaluated yet")
|
||||
if visit and not visited:
|
||||
self.result += v
|
||||
return v
|
||||
|
||||
def __set_user_variable_value(self, name: str, value: str):
|
||||
"""
|
||||
Set the value of a user variable.
|
||||
|
||||
Args:
|
||||
name (str): The name of the user variable.
|
||||
value (str): The value to be set.
|
||||
"""
|
||||
self.__ppp.user_variables[name] = value
|
||||
|
||||
def __remove_user_variable(self, name: str):
|
||||
"""
|
||||
Remove a user variable.
|
||||
|
||||
Args:
|
||||
name (str): The name of the user variable.
|
||||
"""
|
||||
if name in self.__ppp.user_variables:
|
||||
del self.__ppp.user_variables[name]
|
||||
|
||||
def __debug_end(self, construct: str, start_result: str, duration: float, info=None):
|
||||
"""
|
||||
Log the end of a construct processing.
|
||||
|
||||
Args:
|
||||
construct (str): The name of the construct.
|
||||
start_result (str): The initial result.
|
||||
duration (float): The duration of the processing.
|
||||
info: Additional information to log.
|
||||
"""
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
info = f"({info}) " if info is not None and info != "" else ""
|
||||
output = self.result[len(start_result) :]
|
||||
if output != "":
|
||||
output = f" >> '{output}'"
|
||||
self.__ppp.logger.debug(
|
||||
self.__ppp.formatOutput(f"TreeProcessor.{construct} {info}({duration:.3f} seconds){output}")
|
||||
self.__ppp.format_output(f"TreeProcessor.{construct} {info}({duration:.3f} seconds){output}")
|
||||
)
|
||||
|
||||
def __eval_condition(self, cond_var: str, cond_comp: str, cond_value: str | list[str]) -> bool:
|
||||
@@ -616,7 +729,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
Returns:
|
||||
bool: The result of the condition evaluation.
|
||||
"""
|
||||
var_value = self.__ppp.system_variables.get(cond_var, self.__get_user_variable_value(cond_var, None))
|
||||
var_value = self.__ppp.system_variables.get(cond_var, self.__get_user_variable_value(cond_var))
|
||||
if var_value is None:
|
||||
var_value = ""
|
||||
self.__ppp.logger.warning(f"Unknown variable {cond_var}")
|
||||
@@ -717,6 +830,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
"""
|
||||
Process a prompt composition construct in the tree.
|
||||
"""
|
||||
# if self.__ppp.is_comfy_ui():
|
||||
# self.__ppp.logger.warning("Prompt composition is not supported in ComfyUI.")
|
||||
start_result = self.result
|
||||
t1 = time.time()
|
||||
self.__visit(tree.children[0])
|
||||
@@ -744,6 +859,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
"""
|
||||
Process a scheduling construct in the tree and add it to the accumulated shell.
|
||||
"""
|
||||
# if self.__ppp.is_comfy_ui():
|
||||
# self.__ppp.logger.warning("Prompt scheduling is not supported in ComfyUI.")
|
||||
start_result = self.result
|
||||
t1 = time.time()
|
||||
before = tree.children[0]
|
||||
@@ -778,6 +895,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
"""
|
||||
Process an alternation construct in the tree and add it to the accumulated shell.
|
||||
"""
|
||||
# if self.__ppp.is_comfy_ui():
|
||||
# self.__ppp.logger.warning("Prompt alternation is not supported in ComfyUI.")
|
||||
start_result = self.result
|
||||
t1 = time.time()
|
||||
# self.__shell.append(self.AccumulatedShell("al", len(tree.children)))
|
||||
@@ -832,7 +951,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
weight = math.floor(weight * 100) / 100 # we round to 2 decimals
|
||||
weight_str = f"{weight:.2f}".rstrip("0").rstrip(".")
|
||||
self.__shell.append(self.AccumulatedShell("at", weight))
|
||||
if weight == 0.9:
|
||||
if weight == 0.9 and not self.__ppp.is_comfy_ui():
|
||||
starttag = "["
|
||||
self.result += starttag
|
||||
self.__visit(current_tree)
|
||||
@@ -947,7 +1066,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
else:
|
||||
info += " = "
|
||||
self.__set_user_variable_value(variable, newvalue)
|
||||
currentvalue = self.__get_user_variable_value(variable, None, False)
|
||||
currentvalue = self.__get_user_variable_value(variable, False)
|
||||
if currentvalue is None:
|
||||
info += "not evaluated yet"
|
||||
else:
|
||||
@@ -973,16 +1092,14 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
"""
|
||||
t1 = time.time()
|
||||
start_result = self.result
|
||||
value = self.__get_user_variable_value(variable, None)
|
||||
if default is not None:
|
||||
default_value = self.__visit(default, True) # for log
|
||||
value = self.__get_user_variable_value(variable, True, True)
|
||||
if value is None:
|
||||
if default is not None:
|
||||
value = self.__visit(default, False, True)
|
||||
self.result += self.__visit(default, False, True)
|
||||
else:
|
||||
value = ""
|
||||
self.__ppp.logger.warning(f"Unknown variable {variable}")
|
||||
self.result += value
|
||||
t2 = time.time()
|
||||
info = variable
|
||||
if default is not None:
|
||||
@@ -1039,8 +1156,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
|
||||
def __get_choices(
|
||||
self,
|
||||
options: lark.Tree | None,
|
||||
choice_values: list[lark.Tree],
|
||||
options: dict | None,
|
||||
choice_values: list[dict],
|
||||
filter_specifier: Optional[list[list[str]]] = None,
|
||||
wildcard_key: str = None,
|
||||
) -> str:
|
||||
@@ -1048,40 +1165,25 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
Select choices based on the options.
|
||||
|
||||
Args:
|
||||
options (Tree): The tree object representing the options construct.
|
||||
choice_values (list[Tree]): A list of choice tree objects.
|
||||
options (dict): The object representing the options construct.
|
||||
choice_values (list[dict]): A list of choice objects.
|
||||
filter_specifier (list[list[str]]): The filter specifier.
|
||||
wildcard_key (str): The wildcard key if it is a wildcard.
|
||||
|
||||
Returns:
|
||||
str: The selected choice.
|
||||
"""
|
||||
sampler: str = "~"
|
||||
repeating: bool = False
|
||||
from_value: int = 1
|
||||
to_value: int = 1
|
||||
separator: str = self.__ppp.wil_choice_separator
|
||||
if options is not None:
|
||||
if len(options.children) == 1:
|
||||
sampler = options.children[0] if options.children[0] is not None else "~"
|
||||
else:
|
||||
sampler = options.children[0].children[0] if options.children[0] is not None else "~"
|
||||
repeating = options.children[1].children[0] == "r" if options.children[1] is not None else False
|
||||
if len(options.children) == 4:
|
||||
ifrom = 2
|
||||
ito = 2
|
||||
isep = 3
|
||||
else: # 6
|
||||
ifrom = 2
|
||||
ito = 3
|
||||
isep = 4
|
||||
from_value = int(options.children[ifrom].children[0]) if options.children[ifrom] is not None else 1
|
||||
to_value = int(options.children[ito].children[0]) if options.children[ito] is not None else 1
|
||||
separator = (
|
||||
self.__visit(options.children[isep], False, True)
|
||||
if options.children[isep] is not None
|
||||
else self.__ppp.wil_choice_separator
|
||||
)
|
||||
if options is None:
|
||||
options = {}
|
||||
sampler: str = options.get("sampler", "~")
|
||||
repeating: bool = options.get("repeating", False)
|
||||
if "count" in options:
|
||||
from_value = options["count"]
|
||||
to_value = from_value
|
||||
else:
|
||||
from_value: int = options.get("from", 1)
|
||||
to_value: int = options.get("to", 1)
|
||||
separator: str = options.get("separator", self.__ppp.wil_choice_separator)
|
||||
if sampler != "~":
|
||||
msg = f"wildcard '{wildcard_key}'" if wildcard_key else "choices"
|
||||
self.__ppp.logger.warning(f"Unsupported sampler '{sampler}' in {msg} options!")
|
||||
@@ -1090,12 +1192,6 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
if filter_specifier is not None:
|
||||
filtered_choice_values = []
|
||||
for i, c in enumerate(choice_values):
|
||||
c_label_obj = c.children[0]
|
||||
choice_labels = (
|
||||
[x.value.lower() for x in c_label_obj.children[1:-1]] # should be a token
|
||||
if c_label_obj is not None
|
||||
else []
|
||||
)
|
||||
passes = False
|
||||
for o in filter_specifier:
|
||||
tmp_pass = True
|
||||
@@ -1104,7 +1200,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
if int(a) != i:
|
||||
tmp_pass = False
|
||||
break
|
||||
elif a.lower() not in choice_labels:
|
||||
elif a.lower() not in c.get("labels", []):
|
||||
tmp_pass = False
|
||||
break
|
||||
if tmp_pass:
|
||||
@@ -1138,61 +1234,250 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
repeating = False
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
self.__ppp.logger.debug(
|
||||
self.__ppp.formatOutput(
|
||||
self.__ppp.format_output(
|
||||
f"Selecting {'repeating ' if repeating else ''}{num_choices} choice"
|
||||
+ (f"s and separating with '{separator}'" if num_choices > 1 else "")
|
||||
)
|
||||
)
|
||||
if num_choices > 0:
|
||||
available_choices: list[lark.Tree] = []
|
||||
available_choices: list[dict] = []
|
||||
weights = []
|
||||
included_choices = 0
|
||||
excluded_choices = 0
|
||||
excluded_weights_sum = 0
|
||||
for i, c in enumerate(filtered_choice_values):
|
||||
c.choice_index = i # we index them to later sort the results
|
||||
w = float(c.children[1].children[0]) if c.children[1] is not None else 1.0
|
||||
if w > 0 and (c.children[2] is None or self.__evaluate_if(c.children[2].children[0])):
|
||||
c["choice_index"] = i # we index them to later sort the results
|
||||
weight = float(c.get("weight", 1.0))
|
||||
condition = c.get("if", None)
|
||||
if weight > 0 and (condition is None or self.__evaluate_if(condition)):
|
||||
available_choices.append(c)
|
||||
weights.append(w)
|
||||
weights.append(weight)
|
||||
included_choices += 1
|
||||
else:
|
||||
weights.append(-1)
|
||||
excluded_choices += 1
|
||||
excluded_weights_sum += w
|
||||
excluded_weights_sum += weight
|
||||
if excluded_choices > 0: # we need to redistribute the excluded weights
|
||||
weights = [w + excluded_weights_sum / included_choices for w in weights if w >= 0]
|
||||
weights = [weight + excluded_weights_sum / included_choices for weight in weights if weight >= 0]
|
||||
weights = np.array(weights)
|
||||
weights /= weights.sum() # normalize weights
|
||||
selected_choices: list[lark.Tree] = list(
|
||||
selected_choices: list[dict] = list(
|
||||
self.__ppp.rng.choice(available_choices, size=num_choices, p=weights, replace=repeating)
|
||||
)
|
||||
if self.__ppp.wil_keep_choices_order:
|
||||
selected_choices = sorted(selected_choices, key=lambda x: x.choice_index)
|
||||
selected_choices = sorted(selected_choices, key=lambda x: x["choice_index"])
|
||||
selected_choices_text = []
|
||||
prefix: str = (
|
||||
self.__visit(options.get("prefix", None), False, True)
|
||||
if options.get("prefix", None) is not None
|
||||
else ""
|
||||
)
|
||||
if prefix != "" and re.match(r"\w", prefix[-1]):
|
||||
prefix += " "
|
||||
for i, c in enumerate(selected_choices):
|
||||
t1 = time.time()
|
||||
choice_content_obj = c.children[3]
|
||||
choice_content = self.__visit(choice_content_obj, False, True)
|
||||
choice_content_obj = c.get("content", c.get("text", None))
|
||||
if isinstance(choice_content_obj, str):
|
||||
choice_content = choice_content_obj
|
||||
else:
|
||||
choice_content = self.__visit(choice_content_obj, False, True)
|
||||
t2 = time.time()
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
self.__ppp.logger.debug(
|
||||
f"Adding choice {i+1} ({t2-t1:.3f} seconds):\n"
|
||||
+ textwrap.indent(re.sub(r"\n$", "", c.pretty()), " ")
|
||||
+ textwrap.indent(re.sub(r"\n$", "", choice_content), " ")
|
||||
)
|
||||
selected_choices_text.append(choice_content)
|
||||
suffix: str = (
|
||||
self.__visit(options.get("suffix", None), False, True)
|
||||
if options.get("suffix", None) is not None
|
||||
else ""
|
||||
)
|
||||
if suffix != "" and re.match(r"\w", suffix[0]):
|
||||
suffix = " " + suffix
|
||||
# remove comments
|
||||
results = [re.sub(r"\s*#[^\n]*(?:\n|$)", "", r, flags=re.DOTALL) for r in selected_choices_text]
|
||||
return separator.join(results)
|
||||
return prefix + separator.join(results) + suffix
|
||||
return ""
|
||||
|
||||
def __convert_choices_options(self, options: Optional[lark.Tree]) -> dict:
|
||||
"""
|
||||
Convert the choices options to a dictionary.
|
||||
|
||||
Args:
|
||||
options (Tree): The choices options tree.
|
||||
|
||||
Returns:
|
||||
dict: The converted choices options.
|
||||
"""
|
||||
if options is None:
|
||||
return None
|
||||
options_dict = {}
|
||||
if len(options.children) == 1:
|
||||
options_dict["sampler"] = options.children[0] if options.children[0] is not None else "~"
|
||||
else:
|
||||
options_dict["sampler"] = options.children[0].children[0] if options.children[0] is not None else "~"
|
||||
options_dict["repeating"] = (
|
||||
options.children[1].children[0] == "r" if options.children[1] is not None else False
|
||||
)
|
||||
if len(options.children) == 4:
|
||||
ifrom = 2
|
||||
ito = 2
|
||||
isep = 3
|
||||
else: # 6
|
||||
ifrom = 2
|
||||
ito = 3
|
||||
isep = 4
|
||||
options_dict["from"] = (
|
||||
int(options.children[ifrom].children[0]) if options.children[ifrom] is not None else 1
|
||||
)
|
||||
options_dict["to"] = int(options.children[ito].children[0]) if options.children[ito] is not None else 1
|
||||
options_dict["separator"] = (
|
||||
self.__visit(options.children[isep], False, True)
|
||||
if options.children[isep] is not None
|
||||
else self.__ppp.wil_choice_separator
|
||||
)
|
||||
return options_dict
|
||||
|
||||
def __convert_choice(self, choice: lark.Tree) -> dict:
|
||||
"""
|
||||
Convert the choice to a dictionary.
|
||||
|
||||
Args:
|
||||
choice (Tree): The choice tree.
|
||||
|
||||
Returns:
|
||||
dict: The converted choice.
|
||||
"""
|
||||
choice_dict = {}
|
||||
c_label_obj = choice.children[0]
|
||||
choice_dict["labels"] = (
|
||||
[x.value.lower() for x in c_label_obj.children[1:-1]] # should be a token
|
||||
if c_label_obj is not None
|
||||
else []
|
||||
)
|
||||
choice_dict["weight"] = float(choice.children[1].children[0]) if choice.children[1] is not None else 1.0
|
||||
choice_dict["if"] = choice.children[2].children[0] if choice.children[2] is not None else None
|
||||
choice_dict["content"] = choice.children[3]
|
||||
return choice_dict
|
||||
|
||||
def __check_wildcard_initialization(self, wildcard: PPPWildcard):
|
||||
"""
|
||||
Initializes a wildcard if it hasn't been yet.
|
||||
|
||||
Args:
|
||||
wildcard (PPPWildcard): The wildcard to check.
|
||||
"""
|
||||
choice_values = wildcard.choices
|
||||
options = wildcard.options
|
||||
if choice_values is None:
|
||||
t1 = time.time()
|
||||
choice_values = []
|
||||
n = 0
|
||||
# we check the first choice to see if it is actually options
|
||||
if isinstance(wildcard.unprocessed_choices[0], dict):
|
||||
if self.__ppp.wildcard_obj.is_dict_choices_options(wildcard.unprocessed_choices[0]):
|
||||
options = wildcard.unprocessed_choices[0]
|
||||
prefix = options.get("prefix", None)
|
||||
if prefix is not None and isinstance(prefix, str):
|
||||
try:
|
||||
options["prefix"] = self.__ppp.parse_prompt(
|
||||
"choicevalue", prefix, self.__ppp.parser_choicevalue, True
|
||||
)
|
||||
except lark.exceptions.UnexpectedInput as e:
|
||||
self.__ppp.logger.warning(
|
||||
f"Error parsing choice prefix '{prefix}' in wildcard '{wildcard.key}'! : {e.__class__.__name__}"
|
||||
)
|
||||
del options["prefix"]
|
||||
suffix = options.get("suffix", None)
|
||||
if suffix is not None and isinstance(suffix, str):
|
||||
try:
|
||||
options["suffix"] = self.__ppp.parse_prompt(
|
||||
"choicevalue", suffix, self.__ppp.parser_choicevalue, True
|
||||
)
|
||||
except lark.exceptions.UnexpectedInput as e:
|
||||
self.__ppp.logger.warning(
|
||||
f"Error parsing choice suffix '{suffix}' in wildcard '{wildcard.key}'! : {e.__class__.__name__}"
|
||||
)
|
||||
del options["suffix"]
|
||||
n = 1
|
||||
else:
|
||||
try:
|
||||
options = self.__convert_choices_options(
|
||||
self.__ppp.parse_prompt(
|
||||
"as choices options",
|
||||
wildcard.unprocessed_choices[0],
|
||||
self.__ppp.parser_choicesoptions,
|
||||
True,
|
||||
)
|
||||
)
|
||||
n = 1
|
||||
except lark.exceptions.UnexpectedInput:
|
||||
options = None
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
self.__ppp.logger.debug("Does not have options")
|
||||
wildcard.options = options
|
||||
# we process the choices
|
||||
for cv in wildcard.unprocessed_choices[n:]:
|
||||
if isinstance(cv, dict):
|
||||
if self.__ppp.wildcard_obj.is_dict_choice_options(cv):
|
||||
condition = cv.get("if", None)
|
||||
if condition is not None and isinstance(condition, str):
|
||||
try:
|
||||
cv["if"] = self.__ppp.parse_prompt(
|
||||
"condition", condition, self.__ppp.parser_condition, True
|
||||
)
|
||||
except lark.exceptions.UnexpectedInput as e:
|
||||
self.__ppp.logger.warning(
|
||||
f"Error parsing condition '{condition}' in wildcard '{wildcard.key}'! : {e.__class__.__name__}"
|
||||
)
|
||||
cv["if"] = None
|
||||
content = cv.get("content", cv.get("text", None))
|
||||
cv["content"] = content
|
||||
if "text" in cv:
|
||||
del cv["text"]
|
||||
if content is not None and isinstance(content, str):
|
||||
try:
|
||||
cv["content"] = self.__ppp.parse_prompt(
|
||||
"choicevalue", content, self.__ppp.parser_choicevalue, True
|
||||
)
|
||||
except lark.exceptions.UnexpectedInput as e:
|
||||
self.__ppp.logger.warning(
|
||||
f"Error parsing choice content '{content}' in wildcard '{wildcard.key}'! : {e.__class__.__name__}"
|
||||
)
|
||||
cv["content"] = None
|
||||
if cv["content"] is not None:
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
self.__ppp.logger.debug(f"Processed choice {cv}")
|
||||
choice_values.append(cv)
|
||||
else:
|
||||
self.__ppp.logger.warning(f"Invalid choice {cv} in wildcard '{wildcard.key}'!")
|
||||
else:
|
||||
self.__ppp.logger.warning(f"Invalid choice '{cv}' in wildcard '{wildcard.key}'!")
|
||||
else:
|
||||
try:
|
||||
choice_values.append(
|
||||
self.__convert_choice(
|
||||
self.__ppp.parse_prompt("choice", cv, self.__ppp.parser_choice, True)
|
||||
)
|
||||
)
|
||||
except lark.exceptions.UnexpectedInput as e:
|
||||
self.__ppp.logger.warning(
|
||||
f"Error parsing choice '{cv}' in wildcard '{wildcard.key}'! : {e.__class__.__name__}"
|
||||
)
|
||||
wildcard.choices = choice_values
|
||||
t2 = time.time()
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
self.__ppp.logger.debug(f"Processed choices for wildcard '{wildcard.key}' ({t2-t1:.3f} seconds)")
|
||||
return (options, choice_values)
|
||||
|
||||
def wildcard(self, tree: lark.Tree):
|
||||
"""
|
||||
Process a wildcard construct in the tree.
|
||||
"""
|
||||
t1 = time.time()
|
||||
start_result = self.result
|
||||
options = tree.children[0]
|
||||
applied_options = self.__convert_choices_options(tree.children[0])
|
||||
wildcard_key: str = tree.children[1].value # should be a token
|
||||
wc = self.__get_original_node_content(tree, f"?__{wildcard_key}__")
|
||||
if self.__ppp.wil_process_wildcards:
|
||||
@@ -1205,7 +1490,6 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
t2 = time.time()
|
||||
self.__debug_end("wildcard", start_result, t2 - t1, wc)
|
||||
return
|
||||
variablename = None
|
||||
filter_specifier = None
|
||||
filter_object = tree.children[2]
|
||||
if filter_object is not None:
|
||||
@@ -1237,13 +1521,15 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
f"Using a globbing wildcard '{wildcard_key}' with positional index filters is not recommended!"
|
||||
)
|
||||
var_object = tree.children[3]
|
||||
variablename = None
|
||||
variablebackup = None
|
||||
if var_object is not None:
|
||||
variablename = var_object.children[0] # should be a token
|
||||
variablevalue = self.__visit(var_object.children[1], False, True)
|
||||
variablebackup = self.__ppp.user_variables.get(variablename, None)
|
||||
self.__remove_user_variable(variablename)
|
||||
self.__set_user_variable_value(variablename, variablevalue)
|
||||
choice_values_obj_all = []
|
||||
choice_values_all = []
|
||||
for wildcard in selected_wildcards:
|
||||
if wildcard is None:
|
||||
self.detectedWildcards.append(wc)
|
||||
@@ -1251,45 +1537,15 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
t2 = time.time()
|
||||
self.__debug_end("wildcard", start_result, t2 - t1, wc)
|
||||
return
|
||||
choice_values_obj = wildcard.choices_obj
|
||||
options_obj = wildcard.options_obj
|
||||
if choice_values_obj is None:
|
||||
t1 = time.time()
|
||||
choice_values_obj = []
|
||||
try:
|
||||
options_obj = self.__ppp.parse_prompt(
|
||||
"as choices options", wildcard.choices[0], self.__ppp.parser_choicesoptions, True
|
||||
)
|
||||
n = 1
|
||||
except lark.exceptions.UnexpectedInput:
|
||||
options_obj = None
|
||||
n = 0
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
self.__ppp.logger.debug("Does not have options")
|
||||
wildcard.options_obj = options_obj
|
||||
for cv in wildcard.choices[n:]:
|
||||
try:
|
||||
choice_values_obj.append(
|
||||
self.__ppp.parse_prompt("choice", cv, self.__ppp.parser_choice, True)
|
||||
)
|
||||
except lark.exceptions.UnexpectedInput as e:
|
||||
self.__ppp.logger.warning(
|
||||
f"Error parsing choice '{cv}' in wildcard '{wildcard.key}'! : {e.__class__.__name__}"
|
||||
)
|
||||
wildcard.choices_obj = choice_values_obj
|
||||
t2 = time.time()
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
self.__ppp.logger.debug(
|
||||
f"Processed choices for wildcard '{wildcard.key}' ({t2-t1:.3f} seconds)"
|
||||
)
|
||||
if options_obj is not None:
|
||||
if options is None:
|
||||
options = options_obj
|
||||
(options, choice_values) = self.__check_wildcard_initialization(wildcard)
|
||||
if options is not None:
|
||||
if applied_options is None:
|
||||
applied_options = options
|
||||
else:
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
self.__ppp.logger.debug(f"Options for wildcard '{wildcard.key}' are ignored!")
|
||||
choice_values_obj_all += choice_values_obj
|
||||
self.result += self.__get_choices(options, choice_values_obj_all, filter_specifier, wildcard_key)
|
||||
choice_values_all += choice_values
|
||||
self.result += self.__get_choices(applied_options, choice_values_all, filter_specifier, wildcard_key)
|
||||
if wildcard_key in self.__wildcard_filters:
|
||||
del self.__wildcard_filters[wildcard_key]
|
||||
if variablename is not None:
|
||||
@@ -1308,8 +1564,8 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
"""
|
||||
t1 = time.time()
|
||||
start_result = self.result
|
||||
options = tree.children[0]
|
||||
choice_values = tree.children[1::]
|
||||
options = self.__convert_choices_options(tree.children[0])
|
||||
choice_values = [self.__convert_choice(c) for c in tree.children[1::]]
|
||||
ch = self.__get_original_node_content(tree, "?{...}")
|
||||
if self.__ppp.wil_process_wildcards:
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
@@ -1383,7 +1639,7 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
self.__already_processed.append(content)
|
||||
if self.__ppp.debug_level == DEBUG_LEVEL.full:
|
||||
self.__ppp.logger.debug(
|
||||
self.__ppp.formatOutput(f"Adding content at position {position}: {content}")
|
||||
self.__ppp.format_output(f"Adding content at position {position}: {content}")
|
||||
)
|
||||
if position == "e":
|
||||
self.add_at["end"].append(content)
|
||||
@@ -1393,6 +1649,6 @@ class PromptPostProcessor: # pylint: disable=too-few-public-methods,too-many-in
|
||||
else: # position == "s" or invalid
|
||||
self.add_at["start"].append(content)
|
||||
else:
|
||||
self.__ppp.logger.warning(self.__ppp.formatOutput(f"Ignoring repeated content: {content}"))
|
||||
self.__ppp.logger.warning(self.__ppp.format_output(f"Ignoring repeated content: {content}"))
|
||||
t2 = time.time()
|
||||
self.__debug_end("start", "", t2 - t1)
|
||||
|
||||
+2
-2
@@ -4,8 +4,8 @@ from typing import Tuple
|
||||
|
||||
class PPPLRUCache:
|
||||
|
||||
ProcessInput = Tuple[int, str, str]
|
||||
ProcessResult = Tuple[str, str]
|
||||
ProcessInput = Tuple[int, int, str, str] # (seed, wildcards_hash, positive_prompt, negative_prompt)
|
||||
ProcessResult = Tuple[str, str] # (positive_prompt, negative_prompt)
|
||||
|
||||
def __init__(self, capacity: int):
|
||||
self.cache = OrderedDict()
|
||||
|
||||
+5
-4
@@ -16,8 +16,6 @@ if __name__ == "__main__":
|
||||
|
||||
class PromptPostProcessorComfyUINode:
|
||||
|
||||
VERSION = PromptPostProcessor.VERSION
|
||||
|
||||
logger = None
|
||||
|
||||
def __init__(self):
|
||||
@@ -27,6 +25,7 @@ class PromptPostProcessorComfyUINode:
|
||||
with open(grammar_filename, "r", encoding="utf-8") as file:
|
||||
self.grammar_content = file.read()
|
||||
self.wildcards_obj = PPPWildcards(lf.log)
|
||||
self.logger.info(f"{PromptPostProcessor.NAME} {PromptPostProcessor.VERSION} initialized")
|
||||
|
||||
class SmartType(str):
|
||||
def __ne__(self, other):
|
||||
@@ -222,7 +221,7 @@ class PromptPostProcessorComfyUINode:
|
||||
"cleanup_breaks": (
|
||||
"BOOLEAN",
|
||||
{
|
||||
"default": True,
|
||||
"default": False,
|
||||
"tooltip": "Cleanup around BREAKs",
|
||||
"label_on": "Yes",
|
||||
"label_off": "No",
|
||||
@@ -244,7 +243,7 @@ class PromptPostProcessorComfyUINode:
|
||||
"cleanup_ands": (
|
||||
"BOOLEAN",
|
||||
{
|
||||
"default": True,
|
||||
"default": False,
|
||||
"tooltip": "Cleanup around ANDs",
|
||||
"label_on": "Yes",
|
||||
"label_off": "No",
|
||||
@@ -354,6 +353,8 @@ class PromptPostProcessorComfyUINode:
|
||||
cleanup_merge_attention,
|
||||
remove_extranetwork_tags,
|
||||
):
|
||||
if wc_process_wildcards:
|
||||
return float("NaN") # since we can't detect changes in wildcards we assume they are always changed when enabled
|
||||
new_run = { # everything except debug_level
|
||||
"model": model,
|
||||
"modelname": modelname,
|
||||
|
||||
+256
-83
@@ -2,59 +2,112 @@ import fnmatch
|
||||
import os
|
||||
import json
|
||||
from typing import Optional
|
||||
import logging
|
||||
import yaml
|
||||
|
||||
from ppp_logging import DEBUG_LEVEL
|
||||
|
||||
|
||||
def deep_freeze(obj):
|
||||
"""
|
||||
Deep freeze an object.
|
||||
|
||||
Args:
|
||||
obj (object): The object to freeze.
|
||||
|
||||
Returns:
|
||||
object: The frozen object.
|
||||
"""
|
||||
if isinstance(obj, dict):
|
||||
return tuple((k, deep_freeze(v)) for k, v in sorted(obj.items()))
|
||||
elif isinstance(obj, list):
|
||||
return tuple(deep_freeze(i) for i in obj)
|
||||
elif isinstance(obj, set):
|
||||
return tuple(deep_freeze(i) for i in sorted(obj))
|
||||
else:
|
||||
return obj
|
||||
|
||||
|
||||
class PPPWildcard:
|
||||
"""
|
||||
A wildcard object.
|
||||
|
||||
Attributes:
|
||||
key (str): The key of the wildcard.
|
||||
file (str): The path to the file where the wildcard is defined.
|
||||
unprocessed_choices (list[str]): The unprocessed choices of the wildcard.
|
||||
choices (list[dict]): The processed choices of the wildcard.
|
||||
options (dict): The options of the wildcard.
|
||||
"""
|
||||
|
||||
def __init__(self, fullpath: str, key: str, choices: list[str]):
|
||||
self.key: str = key
|
||||
self.file: str = fullpath
|
||||
self.choices: list[str] = choices
|
||||
self.choices_obj: list[object] = None
|
||||
self.options_obj: object = None
|
||||
self.unprocessed_choices: list[str] = choices
|
||||
self.choices: list[dict] = None
|
||||
self.options: dict = None
|
||||
|
||||
def __hash__(self) -> int:
|
||||
t = (self.key, deep_freeze(self.unprocessed_choices))
|
||||
return hash(t)
|
||||
|
||||
|
||||
class PPPWildcards:
|
||||
"""
|
||||
A class to manage wildcards.
|
||||
|
||||
Attributes:
|
||||
wildcards (dict[str, PPPWildcard]): The wildcards.
|
||||
"""
|
||||
|
||||
DEFAULT_WILDCARDS_FOLDER = "wildcards"
|
||||
|
||||
def __init__(self, logger):
|
||||
self.logger = logger
|
||||
self.debug_level = DEBUG_LEVEL.none
|
||||
self.wildcards_folders = []
|
||||
self.__logger: logging.Logger = logger
|
||||
self.__debug_level = DEBUG_LEVEL.none
|
||||
self.__wildcards_folders = []
|
||||
self.__wildcard_files = {}
|
||||
self.wildcards: dict[str, PPPWildcard] = {}
|
||||
self.wildcard_files = {}
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(deep_freeze(self.wildcards))
|
||||
|
||||
def refresh_wildcards(self, debug_level: DEBUG_LEVEL, wildcards_folders: Optional[list[str]]):
|
||||
"""
|
||||
Initialize the wildcards.
|
||||
"""
|
||||
self.debug_level = debug_level
|
||||
self.wildcards_folders = wildcards_folders
|
||||
self.__debug_level = debug_level
|
||||
self.__wildcards_folders = wildcards_folders
|
||||
if wildcards_folders is not None:
|
||||
# if self.debug_level != DEBUG_LEVEL.none:
|
||||
# self.logger.info("Initializing wildcards...")
|
||||
# t1 = time.time()
|
||||
for fullpath in list(self.wildcard_files.keys()):
|
||||
for fullpath in list(self.__wildcard_files.keys()):
|
||||
path = os.path.dirname(fullpath)
|
||||
if not os.path.exists(fullpath) or not any(
|
||||
os.path.commonpath([path, folder]) == folder for folder in self.wildcards_folders
|
||||
os.path.commonpath([path, folder]) == folder for folder in self.__wildcards_folders
|
||||
):
|
||||
self.__remove_wildcards_from_file(fullpath)
|
||||
for f in self.wildcards_folders:
|
||||
for f in self.__wildcards_folders:
|
||||
self.__get_wildcards_in_directory(f, f)
|
||||
# t2 = time.time()
|
||||
# if self.debug_level != DEBUG_LEVEL.none:
|
||||
# self.logger.info(f"Wildcards init time: {t2 - t1:.3f} seconds")
|
||||
else:
|
||||
self.wildcards_folders = []
|
||||
self.__wildcards_folders = []
|
||||
self.wildcards = {}
|
||||
self.wildcard_files = {}
|
||||
self.__wildcard_files = {}
|
||||
|
||||
def get_wildcards(self, key: str) -> list[PPPWildcard]:
|
||||
"""
|
||||
Get all wildcards that match a key.
|
||||
|
||||
Args:
|
||||
key (str): The key to match.
|
||||
|
||||
Returns:
|
||||
list: A list of all wildcards that match the key.
|
||||
"""
|
||||
keys = sorted(fnmatch.filter(self.wildcards.keys(), key))
|
||||
return [self.wildcards[k] for k in keys]
|
||||
|
||||
@@ -104,11 +157,11 @@ class PPPWildcards:
|
||||
full_path (str): The path to the file.
|
||||
debug (bool): Whether to print debug messages or not.
|
||||
"""
|
||||
last_modified_cached = self.wildcard_files.get(full_path, None)
|
||||
if debug and last_modified_cached is not None and self.debug_level != DEBUG_LEVEL.none:
|
||||
self.logger.debug(f"Removing wildcards from file: {full_path}")
|
||||
if full_path in self.wildcard_files.keys():
|
||||
del self.wildcard_files[full_path]
|
||||
last_modified_cached = self.__wildcard_files.get(full_path, None)
|
||||
if debug and last_modified_cached is not None and self.__debug_level != DEBUG_LEVEL.none:
|
||||
self.__logger.debug(f"Removing wildcards from file: {full_path}")
|
||||
if full_path in self.__wildcard_files.keys():
|
||||
del self.__wildcard_files[full_path]
|
||||
for key in list(self.wildcards.keys()):
|
||||
if self.wildcards[key].file == full_path:
|
||||
del self.wildcards[key]
|
||||
@@ -122,87 +175,207 @@ class PPPWildcards:
|
||||
full_path (str): The path to the file.
|
||||
"""
|
||||
last_modified = os.path.getmtime(full_path)
|
||||
last_modified_cached = self.wildcard_files.get(full_path, None)
|
||||
if last_modified_cached is not None and last_modified == self.wildcard_files[full_path]:
|
||||
last_modified_cached = self.__wildcard_files.get(full_path, None)
|
||||
if last_modified_cached is not None and last_modified == self.__wildcard_files[full_path]:
|
||||
return
|
||||
filename = os.path.basename(full_path)
|
||||
name, extension = os.path.splitext(filename)
|
||||
_, extension = os.path.splitext(filename)
|
||||
if extension not in (".txt", ".json", ".yaml", ".yml"):
|
||||
return
|
||||
self.__remove_wildcards_from_file(full_path, False)
|
||||
if last_modified_cached is not None and self.debug_level != DEBUG_LEVEL.none:
|
||||
self.logger.debug(f"Updating wildcards from file: {full_path}")
|
||||
relfolders = os.path.relpath(os.path.dirname(full_path), base)
|
||||
if relfolders == ".":
|
||||
relfolders = ""
|
||||
elif relfolders != "":
|
||||
relfolders += "/"
|
||||
if last_modified_cached is not None and self.__debug_level != DEBUG_LEVEL.none:
|
||||
self.__logger.debug(f"Updating wildcards from file: {full_path}")
|
||||
if extension == ".txt":
|
||||
self.__get_wildcards_in_text_file(full_path, relfolders, name)
|
||||
self.__get_wildcards_in_text_file(full_path, base)
|
||||
elif extension in (".json", ".yaml", ".yml"):
|
||||
self.__get_wildcards_in_structured_file(full_path, relfolders, extension)
|
||||
self.wildcard_files[full_path] = last_modified
|
||||
self.__get_wildcards_in_structured_file(full_path, base)
|
||||
self.__wildcard_files[full_path] = last_modified
|
||||
|
||||
def __get_wildcards_in_structured_file(self, full_path, relfolders, extension):
|
||||
def is_dict_choices_options(self, d: dict) -> bool:
|
||||
"""
|
||||
Check if a dictionary is a valid choices options dictionary.
|
||||
|
||||
Args:
|
||||
d (dict): The dictionary to check.
|
||||
|
||||
Returns:
|
||||
bool: Whether the dictionary is a valid choices options dictionary or not.
|
||||
"""
|
||||
return all(
|
||||
k in ["sampler", "repeating", "count", "from", "to", "prefix", "suffix", "separator"] for k in d.keys()
|
||||
)
|
||||
|
||||
def is_dict_choice_options(self, d: dict) -> bool:
|
||||
"""
|
||||
Check if a dictionary is a valid choice options dictionary.
|
||||
|
||||
Args:
|
||||
d (dict): The dictionary to check.
|
||||
|
||||
Returns:
|
||||
bool: Whether the dictionary is a valid choice options dictionary or not.
|
||||
"""
|
||||
return all(k in ["labels", "weight", "if", "content", "text"] for k in d.keys())
|
||||
|
||||
def __get_choices(self, obj: object, full_path: str, key_parts: list[str]) -> list:
|
||||
"""
|
||||
We process the choices in the object and return them as a list.
|
||||
|
||||
Args:
|
||||
obj (object): the value of a wildcard
|
||||
full_path (str): path to the file where the wildcard is defined
|
||||
key_parts (list[str]): parts of the key for the wildcard
|
||||
|
||||
Returns:
|
||||
list: list of choices
|
||||
"""
|
||||
choices = None
|
||||
if obj is not None:
|
||||
if isinstance(obj, (str, dict)):
|
||||
choices = [obj]
|
||||
elif isinstance(obj, (int, float, bool)):
|
||||
choices = [str(obj)]
|
||||
elif isinstance(obj, list) and len(obj) > 0:
|
||||
choices = []
|
||||
for i, c in enumerate(obj):
|
||||
invalid_choice = False
|
||||
if isinstance(c, str):
|
||||
choice = c
|
||||
elif isinstance(c, (int, float, bool)):
|
||||
choice = str(c)
|
||||
elif isinstance(c, list):
|
||||
# we create an anonymous wildcard
|
||||
choice = self.__create_anonymous_wildcard(full_path, key_parts, i, c)
|
||||
elif isinstance(c, dict):
|
||||
if self.is_dict_choices_options(c) or self.is_dict_choice_options(c):
|
||||
# we assume it is a choice or wildcard parameters in object format
|
||||
choice = c
|
||||
choice_content = choice.get("content", choice.get("text", None))
|
||||
if choice_content is not None and isinstance(choice_content, list):
|
||||
# we create an anonymous wildcard
|
||||
choice["content"] = self.__create_anonymous_wildcard(
|
||||
full_path, key_parts, i, choice_content
|
||||
)
|
||||
if "text" in choice:
|
||||
del choice["text"]
|
||||
elif len(c) == 1:
|
||||
# we assume it is an anonymous wildcard with options
|
||||
firstkey = list(c.keys())[0]
|
||||
choice = self.__create_anonymous_wildcard(full_path, key_parts, i, c[firstkey], firstkey)
|
||||
else:
|
||||
invalid_choice = True
|
||||
else:
|
||||
invalid_choice = True
|
||||
if invalid_choice:
|
||||
self.__logger.warning(
|
||||
f"Invalid choice {i+1} in wildcard '{'/'.join(key_parts)}' in file '{full_path}'!"
|
||||
)
|
||||
else:
|
||||
choices.append(choice)
|
||||
return choices
|
||||
|
||||
def __create_anonymous_wildcard(self, full_path, key_parts, i, content, options=None):
|
||||
"""
|
||||
Create an anonymous wildcard.
|
||||
|
||||
Args:
|
||||
full_path (str): The path to the file that contains it.
|
||||
key_parts (list[str]): The parts of the key.
|
||||
i (int): The index of the wildcard.
|
||||
content (object): The content of the wildcard.
|
||||
options (str): The options for the choice where the wildcard is defined.
|
||||
|
||||
Returns:
|
||||
str: The resulting value for the choice.
|
||||
"""
|
||||
new_parts = key_parts + [f"#ANON_{i}"]
|
||||
self.__add_wildcard(content, full_path, new_parts)
|
||||
value = f"__{'/'.join(new_parts)}__"
|
||||
if options is not None:
|
||||
value = f"{options}::{value}"
|
||||
return value
|
||||
|
||||
def __add_wildcard(self, content: object, full_path: str, external_key_parts: list[str]):
|
||||
"""
|
||||
Add a wildcard to the wildcards dictionary.
|
||||
|
||||
Args:
|
||||
content (object): The content of the wildcard.
|
||||
full_path (str): The path to the file that contains it.
|
||||
external_key_parts (list[str]): The parts of the key.
|
||||
"""
|
||||
key_parts = external_key_parts.copy()
|
||||
if isinstance(content, dict):
|
||||
key_parts.pop()
|
||||
keys = self.__get_keys_in_dict(content)
|
||||
for key in keys:
|
||||
tmp_key_parts = key_parts.copy()
|
||||
tmp_key_parts.extend(key.split("/"))
|
||||
fullkey = "/".join(tmp_key_parts)
|
||||
if self.wildcards.get(fullkey, None) is not None:
|
||||
self.__logger.warning(
|
||||
f"Duplicate wildcard '{fullkey}' in file '{full_path}' and '{self.wildcards[fullkey].file}'!"
|
||||
)
|
||||
else:
|
||||
obj = self.__get_nested(content, key)
|
||||
choices = self.__get_choices(obj, full_path, tmp_key_parts)
|
||||
if choices is None:
|
||||
self.__logger.warning(f"Invalid wildcard '{fullkey}' in file '{full_path}'!")
|
||||
else:
|
||||
self.wildcards[fullkey] = PPPWildcard(full_path, fullkey, choices)
|
||||
return
|
||||
if isinstance(content, str):
|
||||
content = [content]
|
||||
elif isinstance(content, (int, float, bool)):
|
||||
content = [str(content)]
|
||||
if not isinstance(content, list):
|
||||
self.__logger.warning(f"Invalid wildcard in file '{full_path}'!")
|
||||
return
|
||||
fullkey = "/".join(key_parts)
|
||||
if self.wildcards.get(fullkey, None) is not None:
|
||||
self.__logger.warning(
|
||||
f"Duplicate wildcard '{fullkey}' in file '{full_path}' and '{self.wildcards[fullkey].file}'!"
|
||||
)
|
||||
else:
|
||||
choices = self.__get_choices(content, full_path, key_parts)
|
||||
if choices is None:
|
||||
self.__logger.warning(f"Invalid wildcard '{fullkey}' in file '{full_path}'!")
|
||||
else:
|
||||
self.wildcards[fullkey] = PPPWildcard(full_path, fullkey, choices)
|
||||
|
||||
def __get_wildcards_in_structured_file(self, full_path, base):
|
||||
"""
|
||||
Get all wildcards in a structured file.
|
||||
|
||||
Args:
|
||||
full_path (str): The path to the file.
|
||||
base (str): The base path for the wildcards.
|
||||
"""
|
||||
external_key: str = os.path.relpath(os.path.splitext(full_path)[0], base)
|
||||
external_key_parts = external_key.split(os.sep)
|
||||
_, extension = os.path.splitext(full_path)
|
||||
with open(full_path, "r", encoding="utf-8") as file:
|
||||
if extension == ".json":
|
||||
content = json.loads(file.read())
|
||||
else:
|
||||
content = yaml.safe_load(file)
|
||||
keys = self.__get_keys_in_dict(content)
|
||||
for key in keys:
|
||||
fullkey = f"{relfolders}{key}"
|
||||
if self.wildcards.get(fullkey, None) is not None:
|
||||
self.logger.warning(
|
||||
f"Duplicate wildcard '{fullkey}' in file '{full_path}' and '{self.wildcards[fullkey].file}'!"
|
||||
)
|
||||
else:
|
||||
obj = self.__get_nested(content, key)
|
||||
if obj is not None:
|
||||
if isinstance(obj, str):
|
||||
choices = [obj]
|
||||
elif isinstance(obj, (int, float, bool)):
|
||||
choices = [str(obj)]
|
||||
elif isinstance(obj, list) and len(obj) > 0:
|
||||
choices = []
|
||||
for c in obj:
|
||||
if isinstance(c, str):
|
||||
choices.append(c)
|
||||
elif isinstance(c, dict): # we convert the dict to a string
|
||||
d = ""
|
||||
if "weight" in c.keys():
|
||||
d += str(c["weight"])
|
||||
if "if" in c.keys():
|
||||
d += f" if {c['if']}"
|
||||
if d != "":
|
||||
d += "::"
|
||||
if "text" in c.keys():
|
||||
d += c["text"]
|
||||
elif "content" in c.keys():
|
||||
d += c["content"]
|
||||
choices.append(d)
|
||||
else:
|
||||
obj = None
|
||||
if obj is None:
|
||||
self.logger.warning(f"Invalid wildcard '{fullkey}' in file '{full_path}'!")
|
||||
else:
|
||||
self.wildcards[fullkey] = PPPWildcard(full_path, fullkey, choices)
|
||||
self.__add_wildcard(content, full_path, external_key_parts)
|
||||
|
||||
def __get_wildcards_in_text_file(self, full_path, relfolders, name):
|
||||
def __get_wildcards_in_text_file(self, full_path, base):
|
||||
"""
|
||||
Get all wildcards in a text file.
|
||||
|
||||
Args:
|
||||
full_path (str): The path to the file.
|
||||
base (str): The base path for the wildcards.
|
||||
"""
|
||||
external_key: str = os.path.relpath(os.path.splitext(full_path)[0], base)
|
||||
external_key_parts = external_key.split(os.sep)
|
||||
with open(full_path, "r", encoding="utf-8") as file:
|
||||
text_content = map(lambda x: x.strip("\n\r"), file.readlines())
|
||||
text_content = list(filter(lambda x: x.strip() != "" and not x.strip().startswith("#"), text_content))
|
||||
text_content = [x.split("#")[0].rstrip() if len(x.split("#")) > 1 else x for x in text_content]
|
||||
fullkey = f"{relfolders}{name}"
|
||||
if self.wildcards.get(fullkey, None) is not None:
|
||||
self.logger.warning(
|
||||
f"Duplicate wildcard '{fullkey}' in file '{full_path}' and '{self.wildcards[fullkey].file}'!"
|
||||
)
|
||||
else:
|
||||
if len(text_content) == 0:
|
||||
self.logger.warning(f"Invalid wildcard in file '{full_path}'!")
|
||||
else:
|
||||
self.wildcards[fullkey] = PPPWildcard(full_path, fullkey, text_content)
|
||||
self.__add_wildcard(text_content, full_path, external_key_parts)
|
||||
|
||||
def __get_wildcards_in_directory(self, base: str, directory: str):
|
||||
"""
|
||||
@@ -213,7 +386,7 @@ class PPPWildcards:
|
||||
directory (str): The path to the directory.
|
||||
"""
|
||||
if not os.path.exists(directory):
|
||||
self.logger.warning(f"Wildcard directory '{directory}' does not exist!")
|
||||
self.__logger.warning(f"Wildcard directory '{directory}' does not exist!")
|
||||
return
|
||||
for filename in os.listdir(directory):
|
||||
full_path = os.path.abspath(os.path.join(directory, filename))
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
[project]
|
||||
name = "sd-webui-prompt-postprocessor"
|
||||
description = "Stable Diffusion WebUI & ComfyUI extension to post-process the prompt, including sending content from the prompt to the negative prompt and wildcards."
|
||||
version = "2.7.0"
|
||||
version = "2.8.1"
|
||||
license = {file = "LICENSE.txt"}
|
||||
dependencies = ["lark"]
|
||||
|
||||
|
||||
@@ -1 +1,3 @@
|
||||
lark
|
||||
numpy
|
||||
pyyaml
|
||||
+88
-25
@@ -4,6 +4,7 @@ if __name__ == "__main__":
|
||||
import sys
|
||||
import os
|
||||
import time
|
||||
import numpy as np
|
||||
|
||||
sys.path.append(os.path.join(sys.path[0], ".."))
|
||||
|
||||
@@ -57,6 +58,7 @@ class PromptPostProcessorA1111Script(scripts.Script):
|
||||
with open(grammar_filename, "r", encoding="utf-8") as file:
|
||||
self.grammar_content = file.read()
|
||||
self.wildcards_obj = PPPWildcards(lf.log)
|
||||
self.ppp_logger.info(f"{PromptPostProcessor.NAME} {PromptPostProcessor.VERSION} initialized")
|
||||
|
||||
def title(self):
|
||||
"""
|
||||
@@ -79,7 +81,48 @@ class PromptPostProcessorA1111Script(scripts.Script):
|
||||
"""
|
||||
return scripts.AlwaysVisible
|
||||
|
||||
def process(self, p: StableDiffusionProcessing, *args, **kwargs): # pylint: disable=unused-argument
|
||||
def ui(self, is_img2img):
|
||||
with gr.Accordion(PromptPostProcessor.NAME, open=False):
|
||||
force_equal_seeds = gr.Checkbox(
|
||||
label="Force equal seeds",
|
||||
info="Force all image seeds and variation seeds to be equal to the first one, disabling the default autoincrease.",
|
||||
default=False,
|
||||
# show_label=True,
|
||||
elem_id="ppp_force_equal_seeds",
|
||||
)
|
||||
gr.HTML(
|
||||
"""<br><div>Unlink the seed to use the specified one for the prompts instead of the image seed.
|
||||
This seed will only change for each image in the batch if the value is -1 or 'variable seed' is checked.</div>
|
||||
<div>Seeds are only used for the wildcards and choice constructs.</div>"""
|
||||
)
|
||||
unlink_seed = gr.Checkbox(
|
||||
label="Unlink seed",
|
||||
default=False,
|
||||
# show_label=True,
|
||||
elem_id="ppp_unlink_seed",
|
||||
)
|
||||
seed = gr.Number(
|
||||
label="Seed",
|
||||
default=-1,
|
||||
precision=0,
|
||||
# minimum=-1,
|
||||
# maximum=2**32 - 1,
|
||||
# step=1,
|
||||
# show_label=True,
|
||||
min_width=100,
|
||||
elem_id="ppp_seed",
|
||||
)
|
||||
variable_seed = gr.Checkbox(
|
||||
label="Variable seed",
|
||||
default=False,
|
||||
# show_label=True,
|
||||
elem_id="ppp_variable_seed",
|
||||
)
|
||||
return [force_equal_seeds, unlink_seed, seed, variable_seed]
|
||||
|
||||
def process(
|
||||
self, p: StableDiffusionProcessing, input_force_equal_seeds, input_unlink_seed, input_seed, input_variable_seed
|
||||
): # pylint: disable=arguments-differ
|
||||
"""
|
||||
Processes the prompts and applies post-processing operations.
|
||||
|
||||
@@ -92,7 +135,8 @@ class PromptPostProcessorA1111Script(scripts.Script):
|
||||
t1 = time.time()
|
||||
if getattr(opts, "prompt_attention", "") == "Compel parser":
|
||||
self.ppp_logger.warning("Compel parser is not supported!")
|
||||
is_i2i = getattr(p, "init_images", [None])[0] is not None
|
||||
init_images = getattr(p, "init_images", [None]) or [None]
|
||||
is_i2i = bool(init_images[0])
|
||||
self.ppp_debug_level = DEBUG_LEVEL(getattr(opts, "ppp_gen_debug_level", DEBUG_LEVEL.none.value))
|
||||
do_i2i = getattr(opts, "ppp_gen_doi2i", False)
|
||||
if is_i2i and not do_i2i:
|
||||
@@ -138,7 +182,7 @@ class PromptPostProcessorA1111Script(scripts.Script):
|
||||
env_info["is_ssd"] = False # ?
|
||||
env_info["is_sd3"] = getattr(p.sd_model, "is_sd3", False)
|
||||
env_info["is_flux"] = p.sd_model.model_config.__class__.__name__ == "Flux"
|
||||
env_info["is_auraflow"] = False # p.sd_model.model_config.__class__.__name__ == "AuraFlow"
|
||||
env_info["is_auraflow"] = False # p.sd_model.model_config.__class__.__name__ == "AuraFlow"
|
||||
else: # assume A1111 compatible (p.sd_model.__class__.__name__=="DiffusionEngine")
|
||||
env_info["model_class"] = p.sd_model.__class__.__name__
|
||||
env_info["is_sd1"] = getattr(p.sd_model, "is_sd1", False)
|
||||
@@ -185,19 +229,38 @@ class PromptPostProcessorA1111Script(scripts.Script):
|
||||
)
|
||||
prompts_list = []
|
||||
|
||||
seeds = getattr(p, "all_seeds", [])
|
||||
subseeds = getattr(p, "all_subseeds", [])
|
||||
subseed_strength = getattr(p, "subseed_strength", 0.0)
|
||||
if subseed_strength > 0:
|
||||
calculated_seeds = [
|
||||
int(subseed * subseed_strength + seed * (1 - subseed_strength))
|
||||
for seed, subseed in zip(seeds, subseeds)
|
||||
]
|
||||
if input_force_equal_seeds:
|
||||
if self.ppp_debug_level != DEBUG_LEVEL.none:
|
||||
self.ppp_logger.info("Forcing equal seeds")
|
||||
seeds = getattr(p, "all_seeds", [])
|
||||
subseeds = getattr(p, "all_subseeds", [])
|
||||
p.all_seeds = [seeds[0] for _ in seeds]
|
||||
p.all_subseeds = [subseeds[0] for _ in subseeds]
|
||||
|
||||
if input_unlink_seed:
|
||||
if self.ppp_debug_level != DEBUG_LEVEL.none:
|
||||
self.ppp_logger.info("Using unlinked seed")
|
||||
num_seeds = len(getattr(p, "all_seeds", []))
|
||||
if input_seed == -1:
|
||||
calculated_seeds = np.random.randint(0, 2**32, size=num_seeds, dtype=np.int64)
|
||||
elif input_variable_seed:
|
||||
calculated_seeds = [input_seed + i for i in range(num_seeds)]
|
||||
else:
|
||||
calculated_seeds = [input_seed for _ in range(num_seeds)]
|
||||
else:
|
||||
calculated_seeds = seeds
|
||||
if len(set(calculated_seeds)) < len(calculated_seeds):
|
||||
self.ppp_logger.info("Adjusting seeds because some are equal.")
|
||||
calculated_seeds = [seed + i for i, seed in enumerate(calculated_seeds)]
|
||||
seeds = getattr(p, "all_seeds", [])
|
||||
subseeds = getattr(p, "all_subseeds", [])
|
||||
subseed_strength = getattr(p, "subseed_strength", 0.0)
|
||||
if subseed_strength > 0:
|
||||
calculated_seeds = [
|
||||
int(subseed * subseed_strength + seed * (1 - subseed_strength))
|
||||
for seed, subseed in zip(seeds, subseeds)
|
||||
]
|
||||
# if len(set(calculated_seeds)) < len(calculated_seeds):
|
||||
# self.ppp_logger.info("Adjusting seeds because some are equal.")
|
||||
# calculated_seeds = [seed + i for i, seed in enumerate(calculated_seeds)]
|
||||
else:
|
||||
calculated_seeds = seeds
|
||||
|
||||
# adds regular prompts
|
||||
rpr = getattr(p, "all_prompts", None)
|
||||
@@ -222,24 +285,24 @@ class PromptPostProcessorA1111Script(scripts.Script):
|
||||
for i, (prompttype, seed, prompt, negative_prompt) in enumerate(prompts_list):
|
||||
if self.ppp_debug_level != DEBUG_LEVEL.none:
|
||||
self.ppp_logger.info(f"processing prompts[{i+1}] ({prompttype})")
|
||||
if self.lru_cache.get((seed, prompt, negative_prompt)) is None:
|
||||
pp, np = ppp.process_prompt(prompt, negative_prompt, seed)
|
||||
self.lru_cache.put((seed, prompt, negative_prompt), (pp, np))
|
||||
if self.lru_cache.get((seed, hash(self.wildcards_obj), prompt, negative_prompt)) is None:
|
||||
posp, negp = ppp.process_prompt(prompt, negative_prompt, seed)
|
||||
self.lru_cache.put((seed, hash(self.wildcards_obj), prompt, negative_prompt), (posp, negp))
|
||||
# adds also the result so i2i doesn't process it unnecessarily
|
||||
self.lru_cache.put((seed, pp, np), (pp, np))
|
||||
self.lru_cache.put((seed, hash(self.wildcards_obj), posp, negp), (posp, negp))
|
||||
elif self.ppp_debug_level != DEBUG_LEVEL.none:
|
||||
self.ppp_logger.info("result already in cache")
|
||||
|
||||
# updates the prompts
|
||||
if rpr is not None and rnr is not None:
|
||||
for i, (seed, prompt, negative_prompt) in enumerate(zip(calculated_seeds, rpr, rnr)):
|
||||
found = self.lru_cache.get((seed, prompt, negative_prompt))
|
||||
found = self.lru_cache.get((seed, hash(self.wildcards_obj), prompt, negative_prompt))
|
||||
if found is not None:
|
||||
rpr[i] = found[0]
|
||||
rnr[i] = found[1]
|
||||
if rph is not None and rnh is not None:
|
||||
for i, (seed, prompt, negative_prompt) in enumerate(zip(calculated_seeds, rph, rnh)):
|
||||
found = self.lru_cache.get((seed, prompt, negative_prompt))
|
||||
found = self.lru_cache.get((seed, hash(self.wildcards_obj), prompt, negative_prompt))
|
||||
if found is not None:
|
||||
rph[i] = found[0]
|
||||
rnh[i] = found[1]
|
||||
@@ -336,7 +399,7 @@ def on_ui_settings():
|
||||
# wildcard settings
|
||||
shared.opts.add_option(
|
||||
key="ppp_wil_sep",
|
||||
info=new_html_title('<br><h2>Wildcard settings</h2>'),
|
||||
info=new_html_title("<br><h2>Wildcard settings</h2>"),
|
||||
)
|
||||
shared.opts.add_option(
|
||||
key="ppp_wil_processwildcards",
|
||||
@@ -395,7 +458,7 @@ def on_ui_settings():
|
||||
# content removal settings
|
||||
shared.opts.add_option(
|
||||
key="ppp_rem_sep",
|
||||
info=new_html_title('<br><h2>Content removal settings</h2>'),
|
||||
info=new_html_title("<br><h2>Content removal settings</h2>"),
|
||||
)
|
||||
shared.opts.add_option(
|
||||
key="ppp_rem_removeextranetworktags",
|
||||
@@ -409,7 +472,7 @@ def on_ui_settings():
|
||||
# send to negative settings
|
||||
shared.opts.add_option(
|
||||
key="ppp_stn_sep",
|
||||
info=new_html_title('<br><h2>Send to Negative settings</h2>'),
|
||||
info=new_html_title("<br><h2>Send to Negative settings</h2>"),
|
||||
)
|
||||
shared.opts.add_option(
|
||||
key="ppp_stn_separator",
|
||||
@@ -430,7 +493,7 @@ def on_ui_settings():
|
||||
# clean-up settings
|
||||
shared.opts.add_option(
|
||||
key="ppp_cup_sep",
|
||||
info=new_html_title('<br><h2>Clean-up settings</h2>'),
|
||||
info=new_html_title("<br><h2>Clean-up settings</h2>"),
|
||||
)
|
||||
shared.opts.add_option(
|
||||
key="ppp_cup_emptyconstructs",
|
||||
|
||||
+56
-2
@@ -101,6 +101,18 @@ class TestPromptPostProcessor(unittest.TestCase):
|
||||
self.__grammar_content,
|
||||
self.__wildcards_obj,
|
||||
)
|
||||
self.__comfyuippp = PromptPostProcessor(
|
||||
self.__ppp_logger,
|
||||
self.__interrupt,
|
||||
{
|
||||
**self.__def_env_info,
|
||||
"app": "comfyui",
|
||||
"model_class": "SDXL",
|
||||
},
|
||||
self.__defopts,
|
||||
self.__grammar_content,
|
||||
self.__wildcards_obj,
|
||||
)
|
||||
|
||||
def __interrupt(self):
|
||||
self.__interrupted = True
|
||||
@@ -575,8 +587,8 @@ class TestPromptPostProcessor(unittest.TestCase):
|
||||
self.__process(
|
||||
PromptPair("__bad_wildcard__", "{option1|option2}"),
|
||||
PromptPair(
|
||||
PromptPostProcessor.WILDCARD_STOP + "__bad_wildcard__",
|
||||
PromptPostProcessor.WILDCARD_STOP + "{option1|option2}",
|
||||
PromptPostProcessor.WILDCARD_STOP.format("__bad_wildcard__") + "__bad_wildcard__",
|
||||
"{option1|option2}",
|
||||
),
|
||||
ppp=PromptPostProcessor(
|
||||
self.__ppp_logger,
|
||||
@@ -593,6 +605,24 @@ class TestPromptPostProcessor(unittest.TestCase):
|
||||
interrupted=True,
|
||||
)
|
||||
|
||||
def test_wcinvar_warn(self): # wildcards in var with warn option
|
||||
self.__process(
|
||||
PromptPair("${v=__bad_wildcard__}${v}", ""),
|
||||
PromptPair(PromptPostProcessor.WILDCARD_WARNING + "__bad_wildcard__", ""),
|
||||
ppp=PromptPostProcessor(
|
||||
self.__ppp_logger,
|
||||
self.__interrupt,
|
||||
self.__def_env_info,
|
||||
{
|
||||
**self.__defopts,
|
||||
"process_wildcards": False,
|
||||
"if_wildcards": PromptPostProcessor.IFWILDCARDS_CHOICES.warn.value,
|
||||
},
|
||||
self.__grammar_content,
|
||||
self.__wildcards_obj,
|
||||
),
|
||||
)
|
||||
|
||||
def test_wc_wildcard1a_text(self): # simple text wildcard
|
||||
self.__process(
|
||||
PromptPair("the choices are: __text/wildcard1__", ""),
|
||||
@@ -780,6 +810,30 @@ class TestPromptPostProcessor(unittest.TestCase):
|
||||
ppp=self.__nocupppp,
|
||||
)
|
||||
|
||||
def test_wc_wildcardPS_yaml(self): # yaml wildcard with object formatted choices and options and prefix and suffix
|
||||
self.__process(
|
||||
PromptPair("the choices are: __yaml/wildcardPS__", ""),
|
||||
PromptPair("the choices are: prefix-choice2/choice3-suffix", ""),
|
||||
ppp=self.__nocupppp,
|
||||
)
|
||||
|
||||
def test_wc_anonymouswildcard_yaml(self): # yaml anonymous wildcard
|
||||
self.__process(
|
||||
PromptPair("the choices are: __yaml/anonwildcards__", ""),
|
||||
PromptPair("the choices are: six", ""),
|
||||
ppp=self.__nocupppp,
|
||||
)
|
||||
|
||||
# ComfyUI tests
|
||||
|
||||
def test_comfyui_attention(self): # attention conversion
|
||||
self.__process(
|
||||
PromptPair("(test1) (test2:1.5) [test3] [(test4)]", ""),
|
||||
PromptPair("(test1) (test2:1.5) (test3:0.9) (test4:0.99)", ""),
|
||||
ppp=self.__comfyuippp,
|
||||
)
|
||||
|
||||
|
||||
# def test_mix(self):
|
||||
# self.__process(
|
||||
# PromptPair(
|
||||
|
||||
@@ -8,7 +8,7 @@ yaml:
|
||||
- ~r2-3$$-
|
||||
- "'label1,label2'4::choice1"
|
||||
- "3:: choice2 "
|
||||
- "'label1,label3'2::choice3"
|
||||
- { labels: ["label1", "label3"], weight: 2, content: choice3 }
|
||||
- 5 if _is_sd1::choice4
|
||||
|
||||
wildcard2bis:
|
||||
@@ -28,9 +28,42 @@ yaml:
|
||||
- { weight: 2, text: choice1 }
|
||||
- { weight: 3, content: choice2 }
|
||||
- { text: choice3 }
|
||||
- { weight: 4, if: "_is_ssd", text: choice4 }
|
||||
- { weight: 4, if: _is_ssd, text: choice4 }
|
||||
|
||||
wildcard7:
|
||||
- <lora:test1:1>
|
||||
- <lora:test2:1>
|
||||
- <lora:test3:1>
|
||||
- <lora:test3:1>
|
||||
|
||||
wildcardPS:
|
||||
- {
|
||||
sampler: "~",
|
||||
repeating: false,
|
||||
count: 2,
|
||||
prefix: "prefix-",
|
||||
suffix: "-suffix",
|
||||
separator: "/",
|
||||
}
|
||||
- { weight: 3, text: choice1 }
|
||||
- { weight: 2, text: choice2 }
|
||||
- { weight: 1, text: choice3 }
|
||||
|
||||
more_nested:
|
||||
even_more_nested: # this would be __yaml/more_nested/even_more_nested__
|
||||
- one
|
||||
- two
|
||||
|
||||
anonwildcards:
|
||||
- one
|
||||
- two
|
||||
- # choice without options and anonymous wildcard
|
||||
- three
|
||||
- four
|
||||
- 3 if _is_sdxl: # choice with options and anonymous wildcard
|
||||
- five
|
||||
- six
|
||||
- { weight: 1, text: [seven, eight] } # anonymous wildcard used in a choice in object format
|
||||
- # choice without options and anonymous wildcard with parameters
|
||||
- { count: 2, prefix: "#" }
|
||||
- nine
|
||||
- ten
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
- one
|
||||
- 2
|
||||
- three
|
||||
@@ -0,0 +1 @@
|
||||
one choice
|
||||
Reference in New Issue
Block a user