Compare commits
6
Commits
v3.0.0-beta.4
...
legacy
| Author | SHA1 | Date | |
|---|---|---|---|
 asagi4
|
1840bac168 | ||
|
asagi4
|
52fdc76c19 | ||
|
asagi4
|
8f21bc9227 | ||
|
asagi4
|
167c22388f | ||
|
asagi4
|
038cb1bcf3 | ||
|
asagi4
|
d70697842c |
@@ -1,21 +0,0 @@
|
||||
name: Publish to Comfy registry
|
||||
on:
|
||||
workflow_dispatch:
|
||||
push:
|
||||
branches:
|
||||
- master
|
||||
paths:
|
||||
- "pyproject.toml"
|
||||
|
||||
jobs:
|
||||
publish-node:
|
||||
name: Publish Custom Node to registry
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Check out code
|
||||
uses: actions/checkout@v4
|
||||
- name: Publish Custom Node
|
||||
uses: Comfy-Org/publish-node-action@main
|
||||
with:
|
||||
## Add your own personal access token to your Github Repository secrets and reference it here.
|
||||
personal_access_token: ${{ secrets.REGISTRY_ACCESS_TOKEN }}
|
||||
@@ -1,186 +1,8 @@
|
||||
# ComfyUI prompt control
|
||||
# ComfyUI prompt control (LEGACY VERSION)
|
||||
|
||||
Nodes for LoRA and prompt scheduling that make basic operations in ComfyUI completely prompt-controllable.
|
||||
Go to https://github.com/asagi4/comfyui-prompt-control for the revised version of prompt control.
|
||||
|
||||
LoRA and prompt scheduling should produce identical output to the equivalent ComfyUI workflow using multiple samplers or the various conditioning manipulation nodes. If you find situations where this is not the case, please report a bug.
|
||||
|
||||
## Note about stability
|
||||
|
||||
The nodes are currently undergoing some refactoring and rewriting.
|
||||
|
||||
Interfaces for nodes marked "v2" are subject to breaking change at any time, though at worst you'll likely just have to recreate a few nodes.
|
||||
|
||||
Nodes marked "experimental" are very likely to change or disappear completely, or might not even work completely.
|
||||
|
||||
## What can it do?
|
||||
|
||||
See [features](#features) below. Things you can control via the prompt:
|
||||
- Weight interpretation types (comfy, A1111, etc.)
|
||||
- Prompt editing and filtering without multiple samplers
|
||||
- LoRA loading and scheduling via ComfyUI's hook system
|
||||
- Masking, compositions and area control, combining prompts (`AND`), `BREAK`
|
||||
- Prompt masking with cutoff
|
||||
- SDXL parameters
|
||||
|
||||
If you find prompt scheduling inconvenient, `PCTextEncode` can be used as a drop-in replacement for `CLIPTextEncode` to get everything else.
|
||||
|
||||
[This example workflow](workflows/example.json?raw=1) implements a two-pass workflow illustrating most scheduling features.
|
||||
|
||||
The tools in this repository combine well with the macro and wildcard functionality in [comfyui-utility-nodes](https://github.com/asagi4/comfyui-utility-nodes)
|
||||
|
||||
## Requirements
|
||||
|
||||
For `PCEncodeSchedule` and `PCLoraHooksFromSchedule`, you'll need at least version 0.3.7 of ComfyUI (0.3.36 of ComfyUI desktop).
|
||||
|
||||
You need to have `lark` installed in your Python environment for parsing to work (If you reuse A1111's venv, it'll already be there).
|
||||
|
||||
If you use the portable version of ComfyUI on Windows with its embedded Python, you must open a terminal in the ComfyUI installation directory and run the command:
|
||||
```
|
||||
.\python_embeded\python.exe -m pip install lark
|
||||
```
|
||||
|
||||
Then restart ComfyUI afterwards.
|
||||
|
||||
## Notable changes
|
||||
|
||||
I try to avoid behavioural changes that break old prompts, but they may happen occasionally.
|
||||
|
||||
- 2024-12-03 ComfyUI merged support for model/conditioning hooks. There are two new nodes, `PCEncodeSchedule` and `PCLoraHooksFromSchedule` that can be used in combination with the hook nodes. Some functionality is still missing from them, but going forward, these nodes will be the only nodes supported; **I will not spend significant time fixing bugs in the old monkeypatched nodes anymore.**
|
||||
|
||||
## Note on how schedules work
|
||||
|
||||
ComfyUI does not use the step number to determine whether to apply conds; instead, it uses the sampler's timestep value which is affected by the scheduler you're using. This means that when the sampler scheduler isn't linear, the schedules generated by prompt control will not be either.
|
||||
|
||||
# Features
|
||||
## Scheduling and LoRA loading
|
||||
|
||||
See the [syntax documentation](doc/syntax.md)
|
||||
|
||||
## Advanced CLIP encoding
|
||||
|
||||
If you use `PCEncodeSchedule` or `PCTextEncode`. advanced encodings are available automatically. Thanks to BlenderNeko for the original code.
|
||||
|
||||
You can use the syntax `STYLE(weight_interpretation, normalization)` in a prompt to affect how prompts are interpreted.
|
||||
|
||||
The weight interpretations available are:
|
||||
- comfy (default)
|
||||
- comfy++
|
||||
- compel
|
||||
- down_weight
|
||||
- A1111
|
||||
- perp
|
||||
|
||||
Normalizations are:
|
||||
- none (default)
|
||||
- length
|
||||
- mean
|
||||
|
||||
The normalization calculations are independent operations and you can combine them with `+`, eg `STYLE(A1111, length+mean)` or `STYLE(comfy, mean+length)`, or even something silly like `STYLE(perp, mean+length+mean+length)`
|
||||
|
||||
For the legacy nodes, you need to have [Advanced CLIP Encoding nodes](https://github.com/BlenderNeko/ComfyUI_ADV_CLIP_emb/tree/master) cloned into your `custom_nodes` for more options will be available.
|
||||
|
||||
The style can be specified separately for each AND:ed prompt, but the first prompt is special; later prompts will "inherit" it as default. For example:
|
||||
|
||||
```
|
||||
STYLE(A1111) a (red:1.1) cat with (brown:0.9) spots and a long tail AND an (old:0.5) dog AND a (green:1.4) (balloon:1.1)
|
||||
```
|
||||
will interpret everything as A1111, but
|
||||
```
|
||||
a (red:1.1) cat with (brown:0.9) spots and a long tail AND STYLE(A1111) an (old:0.5) dog AND a (green:1.4) (balloon:1.1)
|
||||
```
|
||||
Will interpret the first one using the default ComfyUI behaviour, the second prompt with A1111 and the last prompt with the default again
|
||||
|
||||
For things (ie. the code imports) to work, the nodes must be cloned in a directory named exactly `ComfyUI_ADV_CLIP_emb`.
|
||||
|
||||
## Cutoff
|
||||
|
||||
`PCEncodeSchedule` reimplements cutoff from [ComfyUI Cutoff](https://github.com/BlenderNeko/ComfyUI_Cutoff). You do not need to have the nodes installed.
|
||||
|
||||
The syntax is
|
||||
```
|
||||
a group of animals, [CUT:white cat:white], [CUT:brown dog:brown:0.5:1.0:1.0:_]
|
||||
```
|
||||
You should read the prompt as `a group of animals, white cat, brown dog`, but CUT causes the tokens in `target_tokens` to be masked off from the base prompt in `region_text`, so that their effect can be isolated, and you're less likely to get brown cats or white dogs.
|
||||
|
||||
Target tokens are treated individually, separated by space, for example, `[CUT:green apple, red apple, green leaf:green apple]` will mask *both* greens and the apple, giving you `+ +, red +, + leaf`. To mask out just `green apple`, use `[CUT:green apple, red apple:green_apple]` which will result in a masked prompt of `+ +, red apple`. Escape `_` with a `\`.
|
||||
|
||||
the parameters in the `CUT` section are `region_text:target_tokens:weight;strict_mask:start_from_masked:padding_token` of which only the first two are required. The default values are `weight=1.0`, `strict_mask=1.0` `start_from_masked=1.0`, `padding_token=+`
|
||||
|
||||
If `strict_mask`, `start_from_masked` or `padding_token` are specified in more than one CUT, the *last* one becomes the default for any CUTs afterwards that do not explicitly set the parameters. For example, in:
|
||||
|
||||
`[CUT:white cat:white:0.5] and [CUT:black parrot, flying:black:1.0:0.5] and [CUT:green apple:green]`
|
||||
|
||||
`white cat` will a weight of 0.5, and 1.0 for all parameters, and `black parrot` and `green apple` will *both* have a `strict_mask` parameter of 0.5.
|
||||
|
||||
The parameters affect how the masked and unmasked prompts are combined to produce the final embedding. Just play around with them.
|
||||
|
||||
# Nodes
|
||||
|
||||
## PCLoraHooksFromSchedule
|
||||
|
||||
Creates a ComfyUI `HOOKS` object from a prompt schedule. Can be attached to a CLIP model to perform encoding and LoRA switching
|
||||
|
||||
The hooks can be a bit slow sometimes, especially in cases where the LoRA spec doesn't actually require reloading the patches every time you sample. Try `PCHooksFromScheduleWithOptimizationTest`. (Though that node is likely to go away eventually).
|
||||
|
||||
## PCEncodeSchedule
|
||||
|
||||
Encodes all prompts in a schedule. Pass in a `CLIP` object with hooks attached for LoRA scheduling, then use the resulting `CONDITIONING` normally
|
||||
|
||||
## PCTextEncode
|
||||
|
||||
Encodes a single prompt *without* scheduling or LoRA loading features (but including everything else).
|
||||
|
||||
Note: This does not currently ignore `<lora:...:1>` and will treat it as part of the prompt.
|
||||
|
||||
## PromptToSchedule
|
||||
Parses a schedule from a text prompt. A schedule is essentially an array of `(valid_until, prompt)` pairs that the other nodes can use.
|
||||
|
||||
## FilterSchedule
|
||||
Filters a schedule according to its parameters, removing any *changes* that do not occur within `[start, end)`.
|
||||
|
||||
The node also does tag filtering if any tags are specified.
|
||||
|
||||
Always returns at least the last prompt in the schedule if everything would otherwise be filtered.
|
||||
|
||||
`start=0, end=0` returns the prompt at the start and `start=1.0, end=1.0` returns the prompt at the end.
|
||||
|
||||
## PCScheduleSettings
|
||||
Returns an object representing **default values** for the `SDXL` function and allows configuring `MASK_SIZE` outside the prompt. You need to apply them to a schedule with `PCApplySettings`. Note that for the SDXL settings to apply, you still need to have `SDXL()` in the prompt.
|
||||
|
||||
The "steps" parameter currently does nothing; it's for future features.
|
||||
|
||||
## PCApplySettings
|
||||
Applies the give default values from `PCScheduleSettings` to a schedule
|
||||
|
||||
## PCPromptFromSchedule
|
||||
|
||||
Extracts a text prompt from a schedule; also logs it to the console.
|
||||
LoRAs are *not* included in the text prompt, though they are logged.
|
||||
|
||||
## PCScheduleAddMasks
|
||||
|
||||
Add masks to a schedule object, for use with IMASK (see [syntax documentation](doc/syntax.md))
|
||||
|
||||
# Experimental nodes
|
||||
|
||||
## PCLazyEncode
|
||||
`PCLazyEncode` is an experiment that uses ComfyUI's lazy graph execution mechanism to dynamically generate a graph of `PCTextEncode` and `SetConditioningTimestepRange` nodes from a prompt with schedules. This has the advantage that if a part of the schedule doesn't change, ComfyUI's caching mechanism allows you to avoid re-encoding the non-changed part.
|
||||
|
||||
for example, if you first encode `[cat:dog:0.1]` and later change that to `[cat:dog:0.5]`, no re-encoding takes place.
|
||||
|
||||
for added fun, put `NODE(NodeClassName, paramname)` in a prompt to encode the text using any other node (the default is to use `PCTextEncode` and `text`)
|
||||
|
||||
Note that the node can't have required parameters besides a single CLIP parameter (which must be named `clip`) and the text prompt, and it must return a `CONDITIONING` as its first return value.
|
||||
|
||||
## PCLazyLoraLoader
|
||||
|
||||
Another lazy node that will construct a graph of `LoraLoader`s and `CreateHookLora`s as necessary to provide the necessary LoRA scheduling.
|
||||
|
||||
If you have `apply_hooks` set to true, you **do not** need to apply the `HOOKS` output to a CLIP model separately.
|
||||
|
||||
# Legacy nodes
|
||||
|
||||
See [here](doc/legacy.md)
|
||||
These nodes exist only to reproduce old workflows. They are unmaintained
|
||||
|
||||
# Known issues
|
||||
|
||||
|
||||
+3
-28
@@ -17,38 +17,13 @@ from .prompt_control.legacy.node_aio import PromptControlSimple
|
||||
NODE_CLASS_MAPPINGS = {}
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {}
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
log = logging.getLogger("comfyui-prompt-control-legacy")
|
||||
log.propagate = False
|
||||
if not log.handlers:
|
||||
h = logging.StreamHandler(sys.stdout)
|
||||
h.setFormatter(logging.Formatter("[%(levelname)s] PromptControl: %(message)s"))
|
||||
h.setFormatter(logging.Formatter("[%(levelname)s] PromptControl (LEGACY VERSION): %(message)s"))
|
||||
log.addHandler(h)
|
||||
|
||||
if os.environ.get("COMFYUI_PC_DEBUG"):
|
||||
log.setLevel(logging.DEBUG)
|
||||
else:
|
||||
log.setLevel(logging.INFO)
|
||||
|
||||
from .prompt_control.nodes_lazy import NODE_CLASS_MAPPINGS as lazy_mappings, NODE_DISPLAY_NAME_MAPPINGS as lazy_display
|
||||
|
||||
NODE_CLASS_MAPPINGS.update(lazy_mappings)
|
||||
NODE_DISPLAY_NAME_MAPPINGS.update(lazy_display)
|
||||
|
||||
import importlib
|
||||
|
||||
if importlib.util.find_spec("comfy.hooks"):
|
||||
from .prompt_control.nodes_hooks import (
|
||||
NODE_CLASS_MAPPINGS as hook_mappings,
|
||||
NODE_DISPLAY_NAME_MAPPINGS as hook_display,
|
||||
)
|
||||
|
||||
NODE_CLASS_MAPPINGS.update(hook_mappings)
|
||||
NODE_DISPLAY_NAME_MAPPINGS.update(hook_display)
|
||||
else:
|
||||
log.warning(
|
||||
"Your ComfyUI version is too old, can't import comfy.hooks for PCEncodeSchedule and PCLoraHooksFromSchedule. Update your installation."
|
||||
)
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), "comfy"))
|
||||
|
||||
|
||||
@@ -57,10 +32,10 @@ NODE_CLASS_MAPPINGS.update(
|
||||
"PromptControlSimple": PromptControlSimple,
|
||||
"PromptToSchedule": PromptToSchedule,
|
||||
"PCSplitSampling": PCSplitSampling,
|
||||
"PCPromptFromSchedule": PCPromptFromSchedule,
|
||||
"PCScheduleSettings": PCScheduleSettings,
|
||||
"PCScheduleAddMasks": PCScheduleAddMasks,
|
||||
"PCApplySettings": PCApplySettings,
|
||||
"PCPromptFromSchedule": PCPromptFromSchedule,
|
||||
"PCWrapGuider": PCWrapGuider,
|
||||
"FilterSchedule": FilterSchedule,
|
||||
"ScheduleToCond": ScheduleToCond,
|
||||
|
||||
@@ -1,242 +0,0 @@
|
||||
import torch
|
||||
import numpy as np
|
||||
import itertools
|
||||
|
||||
|
||||
def _grouper(n, iterable):
|
||||
it = iter(iterable)
|
||||
while True:
|
||||
chunk = list(itertools.islice(it, n))
|
||||
if not chunk:
|
||||
return
|
||||
yield chunk
|
||||
|
||||
|
||||
def _norm_mag(w, n):
|
||||
d = w - 1
|
||||
return 1 + np.sign(d) * np.sqrt(np.abs(d) ** 2 / n)
|
||||
# return np.sign(w) * np.sqrt(np.abs(w)**2 / n)
|
||||
|
||||
|
||||
def weights_like(weights, emb):
|
||||
return torch.tensor(weights, dtype=emb.dtype, device=emb.device).reshape(1, -1, 1).expand(emb.shape)
|
||||
|
||||
|
||||
def divide_length(word_ids, weights):
|
||||
sums = dict(zip(*np.unique(word_ids, return_counts=True)))
|
||||
sums[0] = 1
|
||||
weights = [[_norm_mag(w, sums[id]) if id != 0 else 1.0 for w, id in zip(x, y)] for x, y in zip(weights, word_ids)]
|
||||
return weights
|
||||
|
||||
|
||||
def shift_mean_weight(word_ids, weights):
|
||||
delta = 1 - np.mean([w for x, y in zip(weights, word_ids) for w, id in zip(x, y) if id != 0])
|
||||
weights = [[w if id == 0 else w + delta for w, id in zip(x, y)] for x, y in zip(weights, word_ids)]
|
||||
return weights
|
||||
|
||||
|
||||
def scale_to_norm(weights, word_ids, w_max):
|
||||
top = np.max(weights)
|
||||
w_max = min(top, w_max)
|
||||
weights = [[w_max if id == 0 else (w / top) * w_max for w, id in zip(x, y)] for x, y in zip(weights, word_ids)]
|
||||
return weights
|
||||
|
||||
|
||||
def mask_word_id(tokens, word_ids, target_id, mask_token):
|
||||
new_tokens = [[mask_token if wid == target_id else t for t, wid in zip(x, y)] for x, y in zip(tokens, word_ids)]
|
||||
mask = np.array(word_ids) == target_id
|
||||
return (new_tokens, mask)
|
||||
|
||||
|
||||
def batched_clip_encode(tokens, length, encode_func, num_chunks):
|
||||
embs = []
|
||||
for e in _grouper(32, tokens):
|
||||
enc, pooled = encode_func(e)
|
||||
enc = enc.reshape((len(e), length, -1))
|
||||
embs.append(enc)
|
||||
embs = torch.cat(embs)
|
||||
embs = embs.reshape((len(tokens) // num_chunks, length * num_chunks, -1))
|
||||
return embs
|
||||
|
||||
|
||||
def from_masked(tokens, weights, word_ids, base_emb, length, encode_func, m_token=266):
|
||||
pooled_base = base_emb[0, length - 1 : length, :]
|
||||
wids, inds = np.unique(np.array(word_ids).reshape(-1), return_index=True)
|
||||
weight_dict = dict((id, w) for id, w in zip(wids, np.array(weights).reshape(-1)[inds]) if w != 1.0)
|
||||
|
||||
if len(weight_dict) == 0:
|
||||
return torch.zeros_like(base_emb), base_emb[0, length - 1 : length, :]
|
||||
|
||||
weight_tensor = weights_like(weights, base_emb)
|
||||
|
||||
# m_token = (clip.tokenizer.end_token, 1.0) if clip.tokenizer.pad_with_end else (0,1.0)
|
||||
# TODO: find most suitable masking token here
|
||||
m_token = (m_token, 1.0)
|
||||
|
||||
ws = []
|
||||
masked_tokens = []
|
||||
masks = []
|
||||
|
||||
# create prompts
|
||||
for id, w in weight_dict.items():
|
||||
masked, m = mask_word_id(tokens, word_ids, id, m_token)
|
||||
masked_tokens.extend(masked)
|
||||
masks.append(weights_like(m, base_emb))
|
||||
|
||||
ws.append(w)
|
||||
|
||||
# batch process prompts
|
||||
embs = batched_clip_encode(masked_tokens, length, encode_func, len(tokens))
|
||||
masks = torch.cat(masks)
|
||||
|
||||
embs = base_emb.expand(embs.shape) - embs
|
||||
pooled = embs[0, length - 1 : length, :]
|
||||
|
||||
embs *= masks
|
||||
embs = embs.sum(axis=0, keepdim=True)
|
||||
|
||||
pooled_start = pooled_base.expand(len(ws), -1)
|
||||
ws = torch.tensor(ws).reshape(-1, 1).expand(pooled_start.shape)
|
||||
pooled = (pooled - pooled_start) * (ws - 1)
|
||||
pooled = pooled.mean(axis=0, keepdim=True)
|
||||
|
||||
return ((weight_tensor - 1) * embs), pooled_base + pooled
|
||||
|
||||
|
||||
def mask_inds(tokens, inds, mask_token):
|
||||
clip_len = len(tokens[0])
|
||||
inds_set = set(inds)
|
||||
new_tokens = [
|
||||
[mask_token if i * clip_len + j in inds_set else t for j, t in enumerate(x)] for i, x in enumerate(tokens)
|
||||
]
|
||||
return new_tokens
|
||||
|
||||
|
||||
def down_weight(tokens, weights, word_ids, base_emb, length, encode_func, m_token=266):
|
||||
w, w_inv = np.unique(weights, return_inverse=True)
|
||||
|
||||
if np.sum(w < 1) == 0:
|
||||
return base_emb, tokens, base_emb[0, length - 1 : length, :]
|
||||
# m_token = (clip.tokenizer.end_token, 1.0) if clip.tokenizer.pad_with_end else (0,1.0)
|
||||
# using the comma token as a masking token seems to work better than aos tokens for SD 1.x
|
||||
m_token = (m_token, 1.0)
|
||||
|
||||
masked_tokens = []
|
||||
|
||||
masked_current = tokens
|
||||
for i in range(len(w)):
|
||||
if w[i] >= 1:
|
||||
continue
|
||||
masked_current = mask_inds(masked_current, np.where(w_inv == i)[0], m_token)
|
||||
masked_tokens.extend(masked_current)
|
||||
|
||||
embs = batched_clip_encode(masked_tokens, length, encode_func, len(tokens))
|
||||
embs = torch.cat([base_emb, embs])
|
||||
w = w[w <= 1.0]
|
||||
w_mix = np.diff([0] + w.tolist())
|
||||
w_mix = torch.tensor(w_mix, dtype=embs.dtype, device=embs.device).reshape((-1, 1, 1))
|
||||
|
||||
weighted_emb = (w_mix * embs).sum(axis=0, keepdim=True)
|
||||
return weighted_emb, masked_current, weighted_emb[0, length - 1 : length, :]
|
||||
|
||||
|
||||
def scale_emb_to_mag(base_emb, weighted_emb):
|
||||
norm_base = torch.linalg.norm(base_emb)
|
||||
norm_weighted = torch.linalg.norm(weighted_emb)
|
||||
embeddings_final = (norm_base / norm_weighted) * weighted_emb
|
||||
return embeddings_final
|
||||
|
||||
|
||||
def recover_dist(base_emb, weighted_emb):
|
||||
fixed_std = (base_emb.std() / weighted_emb.std()) * (weighted_emb - weighted_emb.mean())
|
||||
embeddings_final = fixed_std + (base_emb.mean() - fixed_std.mean())
|
||||
return embeddings_final
|
||||
|
||||
|
||||
def perp_weight(weights, unweighted_embs, empty_embs):
|
||||
unweighted, unweighted_pooled = unweighted_embs
|
||||
zero, zero_pooled = empty_embs
|
||||
|
||||
weights = weights_like(weights, unweighted)
|
||||
|
||||
if zero.shape != unweighted.shape:
|
||||
zero = zero.repeat(1, unweighted.shape[1] // zero.shape[1], 1)
|
||||
|
||||
perp = (
|
||||
torch.mul(zero, unweighted).sum(dim=-1, keepdim=True) / (unweighted.norm(dim=-1, keepdim=True) ** 2)
|
||||
) * unweighted
|
||||
|
||||
over1 = weights.abs() > 1.0
|
||||
result = unweighted + weights * perp
|
||||
result[~over1] = (unweighted - (1 - weights) * perp)[~over1]
|
||||
result[weights == 0.0] = zero[weights == 0.0]
|
||||
|
||||
return result, unweighted_pooled
|
||||
|
||||
|
||||
def advanced_encode_from_tokens(
|
||||
tokenized,
|
||||
token_normalization,
|
||||
weight_interpretation,
|
||||
encode_func,
|
||||
m_token=266,
|
||||
length=77,
|
||||
w_max=1.0,
|
||||
return_pooled=False,
|
||||
apply_to_pooled=False,
|
||||
**extra_args
|
||||
):
|
||||
tokens = [[t for t, _, _ in x] for x in tokenized]
|
||||
weights = [[w for _, w, _ in x] for x in tokenized]
|
||||
word_ids = [[wid for _, _, wid in x] for x in tokenized]
|
||||
|
||||
for op in token_normalization.split("+"):
|
||||
op = op.strip()
|
||||
if op == "length":
|
||||
# distribute down/up weights over word lengths
|
||||
weights = divide_length(word_ids, weights)
|
||||
if op == "mean":
|
||||
weights = shift_mean_weight(word_ids, weights)
|
||||
|
||||
pooled = None
|
||||
|
||||
if weight_interpretation == "comfy":
|
||||
weighted_tokens = [[(t, w) for t, w in zip(x, y)] for x, y in zip(tokens, weights)]
|
||||
weighted_emb, pooled_base = encode_func(weighted_tokens)
|
||||
pooled = pooled_base
|
||||
else:
|
||||
unweighted_tokens = [[(t, 1.0) for t, _, _ in x] for x in tokenized]
|
||||
base_emb, pooled_base = encode_func(unweighted_tokens)
|
||||
|
||||
if weight_interpretation == "A1111":
|
||||
weighted_emb = base_emb * weights_like(weights, base_emb) # from_zero
|
||||
weighted_emb = (base_emb.mean() / weighted_emb.mean()) * weighted_emb # renormalize
|
||||
pooled = pooled_base
|
||||
|
||||
if weight_interpretation == "compel":
|
||||
pos_tokens = [[(t, w) if w >= 1.0 else (t, 1.0) for t, w in zip(x, y)] for x, y in zip(tokens, weights)]
|
||||
weighted_emb, _ = encode_func(pos_tokens)
|
||||
weighted_emb, _, pooled = down_weight(pos_tokens, weights, word_ids, weighted_emb, length, encode_func)
|
||||
|
||||
if weight_interpretation == "comfy++":
|
||||
weighted_emb, tokens_down, _ = down_weight(unweighted_tokens, weights, word_ids, base_emb, length, encode_func)
|
||||
weights = [[w if w > 1.0 else 1.0 for w in x] for x in weights]
|
||||
# unweighted_tokens = [[(t,1.0) for t, _,_ in x] for x in tokens_down]
|
||||
embs, pooled = from_masked(unweighted_tokens, weights, word_ids, base_emb, length, encode_func)
|
||||
weighted_emb += embs
|
||||
|
||||
if weight_interpretation == "down_weight":
|
||||
weights = scale_to_norm(weights, word_ids, w_max)
|
||||
weighted_emb, _, pooled = down_weight(unweighted_tokens, weights, word_ids, base_emb, length, encode_func)
|
||||
|
||||
if weight_interpretation == "perp":
|
||||
weighted_emb, pooled = perp_weight(
|
||||
weights, (base_emb, pooled_base), encode_func(extra_args["tokenizer"].tokenize_with_weights(""))
|
||||
)
|
||||
|
||||
if return_pooled:
|
||||
if apply_to_pooled:
|
||||
return weighted_emb, pooled
|
||||
else:
|
||||
return weighted_emb, pooled_base
|
||||
return weighted_emb, None
|
||||
@@ -1,222 +0,0 @@
|
||||
import torch
|
||||
import copy
|
||||
import re
|
||||
|
||||
import numpy as np
|
||||
import logging
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
|
||||
|
||||
def replace_embeddings(max_token, prompt, replacements=None):
|
||||
"""Replaces embedding tensors in a token array and replaces them with increasing IDs past max_token"""
|
||||
|
||||
if replacements is None:
|
||||
emb_lookup = []
|
||||
else:
|
||||
emb_lookup = replacements.copy()
|
||||
max_token += len(emb_lookup)
|
||||
|
||||
def get_replacement(embedding):
|
||||
for e, n in emb_lookup:
|
||||
if torch.equal(embedding, e):
|
||||
return n
|
||||
return None
|
||||
|
||||
tokens = []
|
||||
for x in prompt:
|
||||
row = []
|
||||
for i in range(len(x)):
|
||||
emb = x[i][0]
|
||||
if not torch.is_tensor(emb):
|
||||
row.append(emb)
|
||||
else:
|
||||
n = get_replacement(emb)
|
||||
if n is not None:
|
||||
row.append(n)
|
||||
else:
|
||||
max_token += 1
|
||||
row.append(max_token)
|
||||
emb_lookup.append((emb, max_token))
|
||||
tokens.append(row)
|
||||
tokens = np.array(tokens)[:, 1:-1].reshape(-1)
|
||||
return (tokens, emb_lookup)
|
||||
|
||||
|
||||
def unpad_prompt(pad_token, prompt):
|
||||
res = np.trim_zeros(prompt, "b")
|
||||
return np.trim_zeros(res - pad_token, "b") + pad_token
|
||||
|
||||
|
||||
def get_sublists(super_list, sub_list):
|
||||
positions = []
|
||||
for candidate_ind in (i for i, e in enumerate(super_list) if e == sub_list[0]):
|
||||
if super_list[candidate_ind : candidate_ind + len(sub_list)] == sub_list:
|
||||
positions.append(candidate_ind)
|
||||
return positions
|
||||
|
||||
|
||||
def cutoff_add_region(
|
||||
clip_regions, tokenizer, region_text, target_text, weight, strict_mask, start_from_masked, mask_token
|
||||
):
|
||||
"""Adds a cut region to the clip_regions dictionary. It is modified in place"""
|
||||
base_tokens = clip_regions["base_tokens"]
|
||||
region_outputs = []
|
||||
target_outputs = []
|
||||
if strict_mask is not None:
|
||||
clip_regions["strict_mask"] = float(strict_mask)
|
||||
if start_from_masked is not None:
|
||||
clip_regions["start_from_masked"] = float(start_from_masked)
|
||||
if mask_token is not None:
|
||||
clip_regions["mask_token"] = tokenizer.tokenizer(mask_token)["input_ids"][1]
|
||||
if weight is None:
|
||||
weight = 1.0
|
||||
else:
|
||||
weight = float(weight)
|
||||
|
||||
region_text = region_text.strip()
|
||||
target_text = target_text.strip()
|
||||
|
||||
strict_mask = clip_regions["strict_mask"]
|
||||
start_from_masked = clip_regions["start_from_masked"]
|
||||
mask_token = clip_regions["mask_token"]
|
||||
log.info(f"CUT region {region_text=} {target_text=} {weight=} {strict_mask=} {start_from_masked=} {mask_token=}")
|
||||
|
||||
pad_token = tokenizer.end_token
|
||||
|
||||
prompt_tokens, emb_lookup = replace_embeddings(pad_token, base_tokens)
|
||||
|
||||
for rt in region_text.split("\n"):
|
||||
region_tokens = tokenizer.tokenize_with_weights(rt)
|
||||
region_tokens, _ = replace_embeddings(pad_token, region_tokens, emb_lookup)
|
||||
region_tokens = unpad_prompt(pad_token, region_tokens).tolist()
|
||||
|
||||
# calc region mask
|
||||
region_length = len(region_tokens)
|
||||
regions = get_sublists(list(prompt_tokens), region_tokens)
|
||||
|
||||
region_mask = np.zeros(len(prompt_tokens))
|
||||
for r in regions:
|
||||
region_mask[r : r + region_length] = 1
|
||||
region_mask = region_mask.reshape(-1, tokenizer.max_length - 2)
|
||||
region_mask = np.pad(region_mask, pad_width=((0, 0), (1, 1)), mode="constant", constant_values=0)
|
||||
region_mask = region_mask.reshape(1, -1)
|
||||
region_outputs.append(region_mask)
|
||||
|
||||
# calc target mask
|
||||
targets = []
|
||||
for target in target_text.split(" "):
|
||||
# deal with underscores
|
||||
target = re.sub(r"(?<!\\)_", " ", target)
|
||||
target = re.sub(r"\\_", "_", target)
|
||||
|
||||
target_tokens = tokenizer.tokenize_with_weights(target)
|
||||
target_tokens, _ = replace_embeddings(pad_token, target_tokens, emb_lookup)
|
||||
target_tokens = unpad_prompt(pad_token, target_tokens).tolist()
|
||||
|
||||
targets.extend([(x, len(target_tokens)) for x in get_sublists(region_tokens, target_tokens)])
|
||||
targets = [(t_start + r, t_start + t_end + r) for r in regions for t_start, t_end in targets]
|
||||
|
||||
targets_mask = np.zeros(len(prompt_tokens))
|
||||
for t_start, t_end in targets:
|
||||
targets_mask[t_start:t_end] = 1
|
||||
targets_mask = targets_mask.reshape(-1, tokenizer.max_length - 2)
|
||||
targets_mask = np.pad(targets_mask, pad_width=((0, 0), (1, 1)), mode="constant", constant_values=0)
|
||||
targets_mask = targets_mask.reshape(1, -1)
|
||||
target_outputs.append(targets_mask)
|
||||
|
||||
# prepare output
|
||||
region_mask_list = clip_regions["regions"].copy()
|
||||
region_mask_list.extend(region_outputs)
|
||||
target_mask_list = clip_regions["targets"].copy()
|
||||
target_mask_list.extend(target_outputs)
|
||||
weight_list = clip_regions["weights"].copy()
|
||||
weight_list.extend([weight] * len(region_outputs))
|
||||
|
||||
clip_regions["regions"] = region_mask_list
|
||||
clip_regions["targets"] = target_mask_list
|
||||
clip_regions["weights"] = weight_list
|
||||
|
||||
|
||||
def create_masked_prompt(weighted_tokens, mask, mask_token):
|
||||
mask_ids = list(zip(*np.nonzero(mask.reshape((len(weighted_tokens), -1)))))
|
||||
new_prompt = copy.deepcopy(weighted_tokens)
|
||||
for x, y in mask_ids:
|
||||
new_prompt[x][y] = (mask_token,) + new_prompt[x][y][1:]
|
||||
return new_prompt
|
||||
|
||||
|
||||
def process_cuts(encode, extra, tokens):
|
||||
if not extra.get("cuts"):
|
||||
return encode(tokens)
|
||||
|
||||
base = {
|
||||
"base_tokens": tokens,
|
||||
"regions": [],
|
||||
"targets": [],
|
||||
"weights": [],
|
||||
"strict_mask": 1.0,
|
||||
"start_from_masked": 1.0,
|
||||
"mask_token": extra["tokenizer"].tokenizer("+")["input_ids"][1],
|
||||
}
|
||||
|
||||
for cut in extra["cuts"]:
|
||||
cutoff_add_region(base, extra["tokenizer"], *cut)
|
||||
|
||||
return encode_regions(base, encode, extra["tokenizer"])
|
||||
|
||||
|
||||
def debug_tokens(label, prompt, tokenizer):
|
||||
log.debug("Tokens for %s", label)
|
||||
for tokens in prompt:
|
||||
tokens = (t for t in tokens if not torch.is_tensor(t[0]))
|
||||
log.debug(" ".join(f"{x[0][0]} {x[1]}" for x in tokenizer.untokenize(tokens) if x[0][0] != tokenizer.end_token))
|
||||
|
||||
|
||||
def encode_regions(clip_regions, encode, tokenizer):
|
||||
base_weighted_tokens = clip_regions["base_tokens"]
|
||||
start_from_masked = clip_regions["start_from_masked"]
|
||||
mask_token = clip_regions["mask_token"]
|
||||
strict_mask = clip_regions["strict_mask"]
|
||||
|
||||
# calc base embedding
|
||||
base_embedding_full, pool = encode(base_weighted_tokens)
|
||||
|
||||
# Avoid numpy value error and passthrough base embeddings if no regions are set.
|
||||
|
||||
# calc global target mask
|
||||
global_target_mask = np.any(np.stack(clip_regions["targets"]), axis=0).astype(int)
|
||||
|
||||
# calc global region mask
|
||||
global_region_mask = np.any(np.stack(clip_regions["regions"]), axis=0).astype(float)
|
||||
regions_sum = np.sum(np.stack(clip_regions["regions"]), axis=0)
|
||||
regions_normalized = np.divide(1, regions_sum, out=np.zeros_like(regions_sum), where=regions_sum != 0)
|
||||
|
||||
# mask base embeddings
|
||||
base_masked_prompt = create_masked_prompt(base_weighted_tokens, global_target_mask, mask_token)
|
||||
debug_tokens("base_masked", base_masked_prompt, tokenizer)
|
||||
base_embedding_masked, _ = encode(base_masked_prompt)
|
||||
base_embedding_start = base_embedding_full * (1 - start_from_masked) + base_embedding_masked * start_from_masked
|
||||
base_embedding_outer = base_embedding_full * (1 - strict_mask) + base_embedding_masked * strict_mask
|
||||
|
||||
region_embeddings = []
|
||||
for region, target, weight in zip(clip_regions["regions"], clip_regions["targets"], clip_regions["weights"]):
|
||||
region_masking = torch.tensor(
|
||||
regions_normalized * region * weight, dtype=base_embedding_full.dtype, device=base_embedding_full.device
|
||||
).unsqueeze(-1)
|
||||
|
||||
region_prompt = create_masked_prompt(base_weighted_tokens, global_target_mask - target, mask_token)
|
||||
debug_tokens("region", region_prompt, tokenizer)
|
||||
region_emb, _ = encode(region_prompt)
|
||||
region_emb -= base_embedding_start
|
||||
region_emb *= region_masking
|
||||
|
||||
region_embeddings.append(region_emb)
|
||||
region_embeddings = torch.stack(region_embeddings).sum(axis=0)
|
||||
|
||||
embeddings_final_mask = torch.tensor(
|
||||
global_region_mask, dtype=base_embedding_full.dtype, device=base_embedding_full.device
|
||||
).unsqueeze(-1)
|
||||
embeddings_final = base_embedding_start * embeddings_final_mask + base_embedding_outer * (1 - embeddings_final_mask)
|
||||
embeddings_final += region_embeddings
|
||||
return embeddings_final, pool
|
||||
@@ -6,7 +6,7 @@ import gc
|
||||
import comfy.model_management
|
||||
import os
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
log = logging.getLogger("comfyui-prompt-control-legacy")
|
||||
|
||||
|
||||
def has_hijack(obj):
|
||||
|
||||
@@ -20,6 +20,7 @@ class PromptControlSimple:
|
||||
},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("MODEL", "CONDITIONING", "CONDITIONING", "MODEL", "CONDITIONING", "CONDITIONING")
|
||||
RETURN_NAMES = ("model", "positive", "negative", "model_filtered", "pos_filtered", "neg_filtered")
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
|
||||
@@ -9,7 +9,7 @@ from comfy_extras.nodes_mask import FeatherMask, MaskComposite
|
||||
from node_helpers import conditioning_set_values
|
||||
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
log = logging.getLogger("comfyui-prompt-control-legacy")
|
||||
|
||||
try:
|
||||
from custom_nodes.ComfyUI_ADV_CLIP_emb.adv_encode import (
|
||||
@@ -142,6 +142,7 @@ class ScheduleToCond:
|
||||
"required": {"clip": ("CLIP",), "prompt_schedule": ("PROMPT_SCHEDULE",)},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("CONDITIONING",)
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "apply"
|
||||
@@ -163,8 +164,9 @@ class EditableCLIPEncode:
|
||||
"optional": {"filter_tags": ("STRING", {"default": ""})},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("CONDITIONING",)
|
||||
CATEGORY = "promptcontrol/_donotuse"
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "parse"
|
||||
|
||||
def parse(self, clip, text, filter_tags=""):
|
||||
|
||||
@@ -6,7 +6,7 @@ from ..parser import parse_prompt_schedules
|
||||
from .hijack import do_hijack
|
||||
from comfy.samplers import CFGGuider
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
log = logging.getLogger("comfyui-prompt-control-legacy")
|
||||
|
||||
|
||||
def apply_lora_for_step(schedules, step, total_steps, state, original_model, lora_cache, patch=True):
|
||||
@@ -141,6 +141,7 @@ class PCWrapGuider:
|
||||
},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "apply"
|
||||
RETURN_TYPES = ("GUIDER",)
|
||||
@@ -200,6 +201,7 @@ class ScheduleToModel:
|
||||
},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("MODEL",)
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "apply"
|
||||
@@ -218,6 +220,7 @@ class PCSplitSampling:
|
||||
},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("MODEL",)
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "apply"
|
||||
@@ -238,8 +241,9 @@ class LoRAScheduler:
|
||||
},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("MODEL",)
|
||||
CATEGORY = "promptcontrol/_donotuse"
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(self, model, text):
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import logging
|
||||
from ..parser import parse_prompt_schedules
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
log = logging.getLogger("comfyui-prompt-control-legacy")
|
||||
|
||||
|
||||
class FilterSchedule:
|
||||
@@ -16,8 +16,9 @@ class FilterSchedule:
|
||||
},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("PROMPT_SCHEDULE",)
|
||||
CATEGORY = "promptcontrol"
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(self, prompt_schedule, tags="", start=0.0, end=1.0):
|
||||
@@ -34,8 +35,9 @@ class PCApplySettings:
|
||||
def INPUT_TYPES(s):
|
||||
return {"required": {"prompt_schedule": ("PROMPT_SCHEDULE",), "settings": ("SCHEDULE_SETTINGS",)}}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("PROMPT_SCHEDULE",)
|
||||
CATEGORY = "promptcontrol"
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(self, prompt_schedule, settings):
|
||||
@@ -55,8 +57,9 @@ class PCScheduleAddMasks:
|
||||
},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("PROMPT_SCHEDULE",)
|
||||
CATEGORY = "promptcontrol"
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(self, prompt_schedule, mask1=None, mask2=None, mask3=None, mask4=None):
|
||||
@@ -83,8 +86,9 @@ class PCScheduleSettings:
|
||||
},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("SCHEDULE_SETTINGS",)
|
||||
CATEGORY = "promptcontrol"
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(
|
||||
@@ -124,8 +128,9 @@ class PCPromptFromSchedule:
|
||||
"optional": {"tags": ("STRING", {"default": ""})},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("STRING",)
|
||||
CATEGORY = "promptcontrol"
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(self, prompt_schedule, at, tags=""):
|
||||
@@ -144,8 +149,9 @@ class PromptToSchedule:
|
||||
},
|
||||
}
|
||||
|
||||
DEPRECATED = True
|
||||
RETURN_TYPES = ("PROMPT_SCHEDULE",)
|
||||
CATEGORY = "promptcontrol"
|
||||
CATEGORY = "promptcontrol/_legacy"
|
||||
FUNCTION = "parse"
|
||||
|
||||
def parse(self, text, settings=None):
|
||||
|
||||
@@ -11,7 +11,7 @@ import nodes
|
||||
|
||||
import comfy.model_management
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
log = logging.getLogger("comfyui-prompt-control-legacy")
|
||||
|
||||
FORCE_CPU_OFFLOAD = bool(environ.get("COMFYUI_PC_CPU_OFFLOAD"))
|
||||
|
||||
|
||||
@@ -1,177 +0,0 @@
|
||||
import logging
|
||||
import comfy.utils
|
||||
import comfy.hooks
|
||||
import folder_paths
|
||||
from .prompts import encode_prompt
|
||||
from .utils import lora_name_to_file
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
|
||||
|
||||
class PCLoraHooksFromSchedule:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {"prompt_schedule": ("PROMPT_SCHEDULE",)},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("HOOKS",)
|
||||
OUTPUT_TOOLTIPS = ("set of hooks created from the prompt schedule",)
|
||||
CATEGORY = "promptcontrol/v2"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(self, prompt_schedule):
|
||||
consolidated = consolidate_schedule(prompt_schedule)
|
||||
hooks = lora_hooks_from_schedule(consolidated, {})
|
||||
return (hooks,)
|
||||
|
||||
|
||||
class PCEncodeSchedule:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {"clip": ("CLIP",), "prompt_schedule": ("PROMPT_SCHEDULE",)},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("CONDITIONING",)
|
||||
CATEGORY = "promptcontrol/v2"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(self, clip, prompt_schedule):
|
||||
return (encode_schedule(clip, prompt_schedule),)
|
||||
|
||||
|
||||
class PCTextEncode:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {"clip": ("CLIP",), "text": ("STRING", {"multiline": True})},
|
||||
"optional": {"defaults": ("SCHEDULE_DEFAULTS",)},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("CONDITIONING",)
|
||||
CATEGORY = "promptcontrol/v2"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(self, clip, text, defaults=None):
|
||||
return (encode_prompt(clip, text, 0, 1.0, defaults or {}, None),)
|
||||
|
||||
|
||||
def consolidate_schedule(prompt_schedule):
|
||||
prev_loras = {}
|
||||
consolidated = []
|
||||
for end_pct, c in reversed(list(prompt_schedule)):
|
||||
loras = c["loras"]
|
||||
if loras != prev_loras:
|
||||
consolidated.append((end_pct, loras))
|
||||
prev_loras = loras
|
||||
return list(reversed(consolidated))
|
||||
|
||||
|
||||
def find_nonscheduled_loras(consolidated_schedule):
|
||||
consolidated_schedule = list(consolidated_schedule)
|
||||
if not consolidated_schedule:
|
||||
return {}
|
||||
last_end, candidate_loras = consolidated_schedule[0]
|
||||
print(candidate_loras)
|
||||
to_remove = set()
|
||||
for candidate, weights in candidate_loras.items():
|
||||
for end, loras in consolidated_schedule[1:]:
|
||||
last_end = end
|
||||
if loras.get(candidate) != weights:
|
||||
to_remove.add(candidate)
|
||||
# No candidates if the schedule does not span full time
|
||||
if last_end < 1.0:
|
||||
return {}
|
||||
return {k: v for (k, v) in candidate_loras.items() if k not in to_remove}
|
||||
|
||||
|
||||
def lora_hooks_from_schedule(schedules, non_scheduled):
|
||||
start_pct = 0.0
|
||||
lora_cache = {}
|
||||
all_hooks = []
|
||||
|
||||
def create_hook(loraspec, start_pct, end_pct, non_scheduled):
|
||||
nonlocal lora_cache
|
||||
hooks = []
|
||||
hook_kf = comfy.hooks.HookKeyframeGroup()
|
||||
for lora, info in loras.items():
|
||||
if non_scheduled.get(lora) == info:
|
||||
log.info("Skipping %s from hook, it's loaded directly on model", lora)
|
||||
continue
|
||||
path = lora_name_to_file(lora)
|
||||
if not path:
|
||||
continue
|
||||
if path not in lora_cache:
|
||||
lora_cache[path] = comfy.utils.load_torch_file(
|
||||
folder_paths.get_full_path("loras", path), safe_load=True
|
||||
)
|
||||
new_hook = comfy.hooks.create_hook_lora(
|
||||
lora_cache[path], strength_model=info["weight"], strength_clip=info["weight_clip"]
|
||||
)
|
||||
# Set hook_ref so that identical hooks compare equal
|
||||
new_hook.hooks[0].hook_ref = f"pc-{path}-{info['weight']}-{info['weight_clip']}"
|
||||
hooks.append(new_hook)
|
||||
if start_pct > 0.0:
|
||||
kf = comfy.hooks.HookKeyframe(strength=0.0, start_percent=0.0)
|
||||
hook_kf.add(kf)
|
||||
kf = comfy.hooks.HookKeyframe(strength=1.0, start_percent=start_pct)
|
||||
hook_kf.add(kf)
|
||||
if end_pct < 1.0:
|
||||
kf = comfy.hooks.HookKeyframe(strength=0.0, start_percent=end_pct)
|
||||
hook_kf.add(kf)
|
||||
hooks = comfy.hooks.HookGroup.combine_all_hooks(hooks)
|
||||
if hooks:
|
||||
hooks.set_keyframes_on_hooks(hook_kf=hook_kf)
|
||||
return hooks
|
||||
|
||||
for end_pct, loras in schedules:
|
||||
log.info("Creating LoRA hook from %s to %s: %s", start_pct, end_pct, loras)
|
||||
hook = create_hook(loras, start_pct, end_pct, non_scheduled)
|
||||
all_hooks.append(hook)
|
||||
start_pct = end_pct
|
||||
|
||||
del lora_cache
|
||||
|
||||
all_hooks = [x for x in all_hooks if x]
|
||||
|
||||
if all_hooks:
|
||||
hooks = comfy.hooks.HookGroup.combine_all_hooks(all_hooks)
|
||||
return hooks
|
||||
|
||||
|
||||
def debug_conds(conds):
|
||||
r = []
|
||||
for i, c in enumerate(conds):
|
||||
x = c[1].copy()
|
||||
if "pooled_output" in x:
|
||||
del x["pooled_output"]
|
||||
r.append((i, x))
|
||||
return r
|
||||
|
||||
|
||||
def encode_schedule(clip, schedules):
|
||||
start_pct = 0.0
|
||||
conds = []
|
||||
for end_pct, c in schedules:
|
||||
if start_pct < end_pct:
|
||||
prompt = c["prompt"]
|
||||
cond = encode_prompt(clip, prompt, start_pct, end_pct, schedules.defaults, schedules.masks)
|
||||
conds.extend(cond)
|
||||
start_pct = end_pct
|
||||
|
||||
log.debug("Final cond info: %s", debug_conds(conds))
|
||||
return conds
|
||||
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"PCLoraHooksFromSchedule": PCLoraHooksFromSchedule,
|
||||
"PCEncodeSchedule": PCEncodeSchedule,
|
||||
"PCTextEncode": PCTextEncode,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"PCLoraHooksFromSchedule": "PC Create LoRA Hooks",
|
||||
"PCEncodeSchedule": "PC Encode Schedule",
|
||||
"PCTextEncode": "PC Text Encode (no scheduling)",
|
||||
}
|
||||
@@ -1,193 +0,0 @@
|
||||
import logging
|
||||
from .parser import parse_prompt_schedules
|
||||
from comfy_execution.graph_utils import GraphBuilder
|
||||
|
||||
from .prompts import get_function
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
|
||||
from .nodes_hooks import consolidate_schedule, find_nonscheduled_loras
|
||||
from .utils import lora_name_to_file
|
||||
|
||||
|
||||
class PCLazyLoraLoader:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"text": ("STRING", {"multiline": True}),
|
||||
"model": ("MODEL",),
|
||||
"clip": ("CLIP",),
|
||||
"apply_hooks": ("BOOLEAN", {"default": True}),
|
||||
},
|
||||
"hidden": {"dynprompt": "DYNPROMPT", "unique_id": "UNIQUE_ID"},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("MODEL", "CLIP", "HOOKS")
|
||||
OUTPUT_TOOLTIPS = ("Returns a model and clip with LoRAs scheduled",)
|
||||
CATEGORY = "promptcontrol/_experimental"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(self, model, clip, text, apply_hooks, dynprompt, unique_id):
|
||||
schedule = parse_prompt_schedules(text)
|
||||
consolidated = consolidate_schedule(schedule)
|
||||
non_scheduled = find_nonscheduled_loras(consolidated)
|
||||
graph = GraphBuilder(f"PCLazyLoraLoader-{unique_id}")
|
||||
this_node = dynprompt.get_node(unique_id)
|
||||
|
||||
modelinput = this_node["inputs"]["model"]
|
||||
clipinput = this_node["inputs"]["clip"]
|
||||
for lora, info in non_scheduled.items():
|
||||
path = lora_name_to_file(lora)
|
||||
if path is None:
|
||||
log.info("Lazy expansion ignoring nonexistent LoRA %s", lora)
|
||||
continue
|
||||
loader = graph.node("LoraLoader")
|
||||
loader.set_input("model", modelinput)
|
||||
loader.set_input("clip", clipinput)
|
||||
loader.set_input("strength_model", info["weight"])
|
||||
loader.set_input("strength_clip", info["weight_clip"])
|
||||
loader.set_input("lora_name", path)
|
||||
modelinput = loader.out(0)
|
||||
clipinput = loader.out(1)
|
||||
|
||||
hook_nodes = {}
|
||||
start_pct = 0.0
|
||||
|
||||
def key(lora, info):
|
||||
return f"{lora}-{info['weight']}-{info['weight_clip']}"
|
||||
|
||||
for end_pct, loras in consolidated:
|
||||
for lora, info in loras.items():
|
||||
if non_scheduled.get(lora):
|
||||
continue
|
||||
path = lora_name_to_file(lora)
|
||||
if path is None:
|
||||
log.info("Lazy expansion ignoring nonexistent LoRA %s", lora)
|
||||
continue
|
||||
k = key(lora, info)
|
||||
existing_node = hook_nodes.get(key(lora, info))
|
||||
prev_keyframe = None
|
||||
if not existing_node:
|
||||
hook_node = graph.node("CreateHookLora")
|
||||
hook_node.set_input("lora_name", path)
|
||||
hook_node.set_input("strength_model", info["weight"])
|
||||
hook_node.set_input("strength_clip", info["weight_clip"])
|
||||
prev_hook_kf = None
|
||||
if start_pct > 0:
|
||||
prev_keyframe = graph.node("CreateHookKeyframe")
|
||||
prev_keyframe.set_input("strength_mult", 0.0)
|
||||
prev_keyframe.set_input("start_percent", 0.0)
|
||||
prev_hook_kf = prev_keyframe.out(0)
|
||||
else:
|
||||
hook_node, prev_keyframe = existing_node
|
||||
prev_hook_kf = prev_keyframe.out(0)
|
||||
|
||||
if (
|
||||
prev_keyframe
|
||||
and prev_keyframe.get_input("start_pct") == start_pct
|
||||
and prev_keyframe.get_input("strength_mult") == 0.0
|
||||
):
|
||||
next_keyframe = prev_keyframe
|
||||
else:
|
||||
next_keyframe = graph.node("CreateHookKeyframe")
|
||||
next_keyframe.set_input("start_percent", start_pct)
|
||||
next_keyframe.set_input("prev_hook_kf", prev_hook_kf)
|
||||
|
||||
next_keyframe.set_input("strength_mult", 1.0)
|
||||
prev_hook_kf = next_keyframe.out(0)
|
||||
if end_pct < 1.0:
|
||||
next_keyframe = graph.node("CreateHookKeyframe")
|
||||
next_keyframe.set_input("strength_mult", 1.0)
|
||||
next_keyframe.set_input("start_percent", end_pct)
|
||||
next_keyframe.set_input("prev_hook_kf", prev_hook_kf)
|
||||
|
||||
hook_nodes[k] = (hook_node, next_keyframe)
|
||||
start_pct = end_pct
|
||||
hooks = []
|
||||
for hook, kfs in hook_nodes.values():
|
||||
n = graph.node("SetHookKeyframes")
|
||||
n.set_input("hooks", hook.out(0))
|
||||
n.set_input("hook_kf", kfs.out(0))
|
||||
hooks.append(n)
|
||||
res = None
|
||||
if len(hooks) > 0:
|
||||
res = hooks[0]
|
||||
for h in hooks[:1]:
|
||||
n = graph.node("CombineHooks2")
|
||||
n.set_input("hooks_A", res.out(0))
|
||||
n.set_input("hooks_B", h.out(0))
|
||||
res = n
|
||||
res = res.out(0)
|
||||
if apply_hooks:
|
||||
n = graph.node("SetClipHooks")
|
||||
n.set_input("clip", clipinput)
|
||||
n.set_input("hooks", res)
|
||||
n.set_input("apply_to_conds", True)
|
||||
n.set_input("schedule_clip", True)
|
||||
clipinput = n.out(0)
|
||||
|
||||
r = graph.finalize()
|
||||
|
||||
return {"result": (modelinput, clipinput, res), "expand": r}
|
||||
|
||||
|
||||
class PCLazyTextEncode:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {"clip": ("CLIP",), "text": ("STRING", {"multiline": True})},
|
||||
# "optional": {"defaults": ("SCHEDULE_DEFAULTS",)},
|
||||
"hidden": {"dynprompt": "DYNPROMPT", "unique_id": "UNIQUE_ID"},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("CONDITIONING",)
|
||||
OUTPUT_TOOLTIPS = ("A fully encoded and scheduled conditioning",)
|
||||
CATEGORY = "promptcontrol/_experimental"
|
||||
FUNCTION = "apply"
|
||||
|
||||
def apply(self, clip, text, dynprompt, unique_id):
|
||||
schedules = parse_prompt_schedules(text)
|
||||
graph = GraphBuilder(f"PCEncodeLazy-{unique_id}")
|
||||
|
||||
this_node = dynprompt.get_node(unique_id)
|
||||
print("Lazy", this_node)
|
||||
|
||||
nodes = []
|
||||
start_pct = 0.0
|
||||
for end_pct, c in schedules:
|
||||
p = c["prompt"]
|
||||
p, classnames = get_function(p, "NODE", ["PCTextEncode", "text"])
|
||||
classname = "PCTextEncode"
|
||||
paramname = "text"
|
||||
if classnames:
|
||||
classname = classnames[0][0]
|
||||
paramname = classnames[0][1]
|
||||
node = graph.node(classname)
|
||||
timestep = graph.node("ConditioningSetTimestepRange")
|
||||
node.set_input("clip", this_node["inputs"]["clip"])
|
||||
node.set_input(paramname, p)
|
||||
timestep.set_input("conditioning", node.out(0))
|
||||
timestep.set_input("start", start_pct)
|
||||
timestep.set_input("end", end_pct)
|
||||
nodes.append(timestep)
|
||||
start_pct = end_pct
|
||||
node = nodes[0]
|
||||
for othernode in nodes[1:]:
|
||||
combiner = graph.node("ConditioningCombine")
|
||||
combiner.set_input("conditioning_1", node.out(0))
|
||||
combiner.set_input("conditioning_2", othernode.out(0))
|
||||
node = combiner
|
||||
|
||||
return {"result": (node.out(0),), "expand": graph.finalize()}
|
||||
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"PCLazyTextEncode": PCLazyTextEncode,
|
||||
"PCLazyLoraLoader": PCLazyLoraLoader,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"PCLazyTextEncode": "Encode prompt w/ scheduling (Lazy) (EXPERIMENTAL)",
|
||||
"PCLazyLoraLoader": "Load LoRAs from prompt w/ scheduling (Lazy) (EXPERIMENTAL)",
|
||||
}
|
||||
@@ -3,7 +3,7 @@ import logging
|
||||
from math import ceil
|
||||
|
||||
logging.basicConfig()
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
log = logging.getLogger("comfyui-prompt-control-legacy")
|
||||
|
||||
if lark.__version__ == "0.12.0":
|
||||
x = "Your lark package reports an ancient version (0.12.0) and will not work. If you have the 'lark-parser' package in your Python environment, remove that and *reinstall* lark!"
|
||||
|
||||
@@ -1,456 +0,0 @@
|
||||
import logging
|
||||
import re
|
||||
import torch
|
||||
from functools import partial
|
||||
from comfy_extras.nodes_mask import FeatherMask, MaskComposite
|
||||
|
||||
from .utils import safe_float, get_function, parse_floats
|
||||
from .adv_encode import advanced_encode_from_tokens
|
||||
from .cutoff import process_cuts
|
||||
from .parser import parse_cuts
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
|
||||
AVAILABLE_STYLES = ["comfy", "perp", "A1111", "compel", "comfy++", "down_weight"]
|
||||
AVAILABLE_NORMALIZATIONS = ["none", "mean", "length", "length+mean"]
|
||||
|
||||
SHUFFLE_GEN = torch.Generator(device="cpu")
|
||||
|
||||
|
||||
def get_sdxl(text, defaults):
|
||||
# Defaults fail to parse and get looked up from the defaults dict
|
||||
text, sdxl = get_function(text, "SDXL", ["none", "none", "none"])
|
||||
if not sdxl:
|
||||
return text, {}
|
||||
args = sdxl[0]
|
||||
d = defaults
|
||||
w, h = parse_floats(args[0], [d.get("sdxl_width", 1024), d.get("sdxl_height", 1024)], split_re="\\s+")
|
||||
tw, th = parse_floats(args[1], [d.get("sdxl_twidth", 1024), d.get("sdxl_theight", 1024)], split_re="\\s+")
|
||||
cropw, croph = parse_floats(args[2], [d.get("sdxl_cwidth", 0), d.get("sdxl_cheight", 0)], split_re="\\s+")
|
||||
|
||||
opts = {
|
||||
"width": int(w),
|
||||
"height": int(h),
|
||||
"target_width": int(tw),
|
||||
"target_height": int(th),
|
||||
"crop_w": int(cropw),
|
||||
"crop_h": int(croph),
|
||||
}
|
||||
return text, opts
|
||||
|
||||
|
||||
def get_clipweights(text, existing_spec=None):
|
||||
text, spec = get_function(text, "TE_WEIGHT", defaults=None)
|
||||
if not spec:
|
||||
return existing_spec or {}, text
|
||||
args = spec[0].strip()
|
||||
res = {}
|
||||
for arg in args.split(","):
|
||||
try:
|
||||
te, val = arg.strip().split("=")
|
||||
te, val = te.strip(), float(val.strip())
|
||||
res[te] = val
|
||||
except ValueError:
|
||||
log.warning("Invalid TE weight spec '%s', ignoring...", arg.strip())
|
||||
return res, text
|
||||
|
||||
|
||||
def get_style(text, default_style="comfy", default_normalization="none"):
|
||||
text, styles = get_function(text, "STYLE", [default_style, default_normalization])
|
||||
if not styles:
|
||||
return default_style, default_normalization, text
|
||||
style, normalization = styles[0]
|
||||
style = style.strip()
|
||||
normalization = normalization.strip()
|
||||
if style not in AVAILABLE_STYLES:
|
||||
log.warning("Unrecognized prompt style: %s. Using %s", style, default_style)
|
||||
style = default_style
|
||||
|
||||
if normalization not in AVAILABLE_NORMALIZATIONS:
|
||||
log.warning("Unrecognized prompt normalization: %s. Using %s", normalization, default_normalization)
|
||||
normalization = default_normalization
|
||||
|
||||
return style, normalization, text
|
||||
|
||||
|
||||
def shuffle_chunk(shuffle, c):
|
||||
func, shuffle = shuffle
|
||||
shuffle_count = int(safe_float(shuffle[0], 0))
|
||||
_, separator, joiner = shuffle
|
||||
if separator == "default":
|
||||
separator = ","
|
||||
|
||||
if not separator:
|
||||
separator = ","
|
||||
|
||||
joiner = {
|
||||
"default": ",",
|
||||
"separator": separator,
|
||||
}.get(joiner, joiner)
|
||||
|
||||
log.info("%s arg=%s sep=%s join=%s", func, shuffle_count, separator, joiner)
|
||||
separated = c.split(separator)
|
||||
if func == "SHIFT":
|
||||
shuffle_count = shuffle_count % len(separated)
|
||||
permutation = separated[shuffle_count:] + separated[:shuffle_count]
|
||||
elif func == "SHUFFLE":
|
||||
SHUFFLE_GEN.manual_seed(shuffle_count)
|
||||
permutation = [separated[i] for i in torch.randperm(len(separated), generator=SHUFFLE_GEN)]
|
||||
else:
|
||||
# ??? should never get here
|
||||
permutation = separated
|
||||
|
||||
permutation = [p for p in permutation if p.strip()]
|
||||
if permutation != separated:
|
||||
c = joiner.join(permutation)
|
||||
return c
|
||||
|
||||
|
||||
def fix_word_ids(tokens):
|
||||
"""Fix word indexes. Tokenizing separately (when BREAKs exist) causes the indexes to restart which causes problems with some weighting algorithms that rely on them"""
|
||||
for key in tokens:
|
||||
max_idx = 0
|
||||
for group in range(len(tokens[key])):
|
||||
for i, token in enumerate(tokens[key][group]):
|
||||
if len(token) < 3:
|
||||
# No need to fix ids when they don't exist
|
||||
return tokens
|
||||
# Ignore zeros, they represent the padding token
|
||||
if token[2] != 0 and token[2] < max_idx:
|
||||
tokens[key][group][i] = (token[0], token[1], token[2] + max_idx)
|
||||
max_idx = max(max_idx, max(x for _, _, x in tokens[key][group]))
|
||||
return tokens
|
||||
|
||||
|
||||
def encode_prompt_segment(
|
||||
clip,
|
||||
text,
|
||||
settings,
|
||||
default_style="comfy",
|
||||
default_normalization="none",
|
||||
clip_weights=None,
|
||||
) -> list[tuple[torch.Tensor, dict[str]]]:
|
||||
style, normalization, text = get_style(text, default_style, default_normalization)
|
||||
clip_weights, text = get_clipweights(text, clip_weights)
|
||||
text, cuts = parse_cuts(text)
|
||||
extra = {}
|
||||
if clip_weights:
|
||||
extra["clip_weights"] = clip_weights
|
||||
if cuts:
|
||||
extra["cuts"] = cuts
|
||||
|
||||
# defaults=None means there is no argument parsing at all
|
||||
text, l_prompts = get_function(text, "CLIP_L", defaults=None)
|
||||
chunks = re.split(r"\bBREAK\b", text)
|
||||
token_chunks = []
|
||||
need_word_ids = True
|
||||
for c in chunks:
|
||||
c, shuffles = get_function(c.strip(), "(SHIFT|SHUFFLE)", ["0", "default", "default"], return_func_name=True)
|
||||
r = c
|
||||
for s in shuffles:
|
||||
r = shuffle_chunk(s, r)
|
||||
if r != c:
|
||||
log.info("Shuffled prompt chunk to %s", r)
|
||||
c = r
|
||||
t = clip.tokenize(c, return_word_ids=need_word_ids)
|
||||
token_chunks.append(t)
|
||||
tokens = token_chunks[0]
|
||||
|
||||
for key in tokens:
|
||||
for c in token_chunks[1:]:
|
||||
tokens[key].extend(c[key])
|
||||
|
||||
# Non-SDXL has only "l"
|
||||
if "g" in tokens and l_prompts:
|
||||
text_l = " ".join(l_prompts)
|
||||
log.info("Encoded SDXL CLIP_L prompt: %s", text_l)
|
||||
tokens["l"] = clip.tokenize(text_l, return_word_ids=need_word_ids)["l"]
|
||||
|
||||
if "g" in tokens and "l" in tokens and len(tokens["l"]) != len(tokens["g"]):
|
||||
empty = clip.tokenize("", return_word_ids=need_word_ids)
|
||||
while len(tokens["l"]) < len(tokens["g"]):
|
||||
tokens["l"] += empty["l"]
|
||||
while len(tokens["l"]) > len(tokens["g"]):
|
||||
tokens["g"] += empty["g"]
|
||||
|
||||
tokens = fix_word_ids(tokens)
|
||||
|
||||
tes = []
|
||||
for k in tokens:
|
||||
if k in ["g", "l"]:
|
||||
tes.append(f"clip_{k}")
|
||||
else:
|
||||
tes.append(k)
|
||||
|
||||
clip = hook_te(clip, tes, style, normalization, extra)
|
||||
|
||||
return clip.encode_from_tokens_scheduled(tokens, add_dict=settings)
|
||||
|
||||
|
||||
def apply_weights(output, te_name, spec):
|
||||
"""Applies weights to TE outputs"""
|
||||
if not spec:
|
||||
return output
|
||||
|
||||
if te_name.startswith("clip_"):
|
||||
te_name = te_name[5:]
|
||||
|
||||
if isinstance(output, tuple):
|
||||
out, pooled = output
|
||||
if te_name in spec:
|
||||
log.info("Weighting %s output by %s", te_name, spec[te_name])
|
||||
out = out * spec[te_name]
|
||||
pkey = te_name + "_pooled"
|
||||
if pkey in spec:
|
||||
log.info("Weighting %s pooled output by %s", te_name, spec[pkey])
|
||||
pooled = pooled * spec[pkey]
|
||||
|
||||
return out, pooled
|
||||
else:
|
||||
if te_name in spec:
|
||||
log.info("Weighting %s output by %s", te_name, spec[te_name])
|
||||
output = output * spec[te_name]
|
||||
return output
|
||||
|
||||
|
||||
def make_patch(te_name, orig_fn, normalization, style, extra):
|
||||
def encode(t):
|
||||
r = advanced_encode_from_tokens(
|
||||
t, normalization, style, orig_fn, return_pooled=True, apply_to_pooled=False, **extra
|
||||
)
|
||||
return apply_weights(r, te_name, extra.get("clip_weights"))
|
||||
|
||||
if "cuts" in extra:
|
||||
return partial(process_cuts, encode, extra)
|
||||
return encode
|
||||
|
||||
|
||||
def hook_te(clip, te_names, style, normalization, extra):
|
||||
if style == "comfy" and normalization == "none" and not extra:
|
||||
return clip
|
||||
newclip = clip.clone()
|
||||
for te_name in te_names:
|
||||
if hasattr(clip.patcher.model, te_name):
|
||||
x = extra.copy()
|
||||
x["tokenizer"] = getattr(clip.tokenizer, te_name)
|
||||
log.debug("Hooked into %s with style=%s, normalization=%s", te_name, style, normalization)
|
||||
newclip.patcher.add_object_patch(
|
||||
f"{te_name}.encode_token_weights",
|
||||
make_patch(
|
||||
te_name,
|
||||
clip.patcher.get_model_object(f"{te_name}.encode_token_weights"),
|
||||
normalization,
|
||||
style,
|
||||
x,
|
||||
),
|
||||
)
|
||||
# 'g' and 'l' exist in these are clip_g and clip_l
|
||||
else:
|
||||
log.debug("Tokens contain items with key %s but no TE found on object with that name.", te_name)
|
||||
return newclip
|
||||
|
||||
|
||||
def get_area(text):
|
||||
text, areas = get_function(text, "AREA", ["0 1", "0 1", "1"])
|
||||
if not areas:
|
||||
return text, None
|
||||
|
||||
args = areas[0]
|
||||
x, w = parse_floats(args[0], [0.0, 1.0], split_re="\\s+")
|
||||
y, h = parse_floats(args[1], [0.0, 1.0], split_re="\\s+")
|
||||
weight = safe_float(args[2], 1.0)
|
||||
|
||||
def is_pct(f):
|
||||
return f >= 0.0 and f <= 1.0
|
||||
|
||||
def is_pixel(f):
|
||||
return f == 0 or f > 1
|
||||
|
||||
if all(is_pct(v) for v in [h, w, y, x]):
|
||||
area = ("percentage", h, w, y, x)
|
||||
elif all(is_pixel(v) for v in [h, w, y, x]):
|
||||
area = (int(h) // 8, int(w) // 8, int(y) // 8, int(x) // 8)
|
||||
else:
|
||||
raise Exception(
|
||||
f"AREA specified with invalid size {x} {w}, {h} {y}. They must either all be percentages between 0 and 1 or positive integer pixel values excluding 1"
|
||||
)
|
||||
|
||||
return text, (area, weight)
|
||||
|
||||
|
||||
def get_mask_size(text, defaults):
|
||||
text, sizes = get_function(text, "MASK_SIZE", ["512", "512"])
|
||||
if not sizes:
|
||||
return text, (defaults.get("mask_width", 512), defaults.get("mask_height", 512))
|
||||
w, h = sizes[0]
|
||||
return text, (int(w), int(h))
|
||||
|
||||
|
||||
def make_mask(args, size, weight):
|
||||
x1, x2 = parse_floats(args[0], [0.0, 1.0], split_re="\\s+")
|
||||
y1, y2 = parse_floats(args[1], [0.0, 1.0], split_re="\\s+")
|
||||
|
||||
def is_pct(f):
|
||||
return f >= 0.0 and f <= 1.0
|
||||
|
||||
def is_pixel(f):
|
||||
return f == 0 or f > 1
|
||||
|
||||
if all(is_pct(v) for v in [x1, x2, y1, y2]):
|
||||
w, h = size
|
||||
xs = int(w * x1), int(w * x2)
|
||||
ys = int(h * y1), int(h * y2)
|
||||
elif all(is_pixel(v) for v in [x1, x2, y1, y2]):
|
||||
w, h = size
|
||||
xs = int(x1), int(x2)
|
||||
ys = int(y1), int(y2)
|
||||
else:
|
||||
raise Exception(
|
||||
f"MASK specified with invalid size {x1} {x2}, {y1} {y2}. They must either all be percentages between 0 and 1 or positive integer pixel values excluding 1"
|
||||
)
|
||||
|
||||
mask = torch.full((h, w), 0, dtype=torch.float32, device="cpu")
|
||||
mask[ys[0] : ys[1], xs[0] : xs[1]] = weight
|
||||
mask = mask.unsqueeze(0)
|
||||
log.info("Mask xs=%s, ys=%s, shape=%s, weight=%s", xs, ys, mask.shape, weight)
|
||||
return mask
|
||||
|
||||
|
||||
def get_mask(text, size, input_masks):
|
||||
"""Parse MASK(x1 x2, y1 y2, weight), IMASK(i, weight) and FEATHER(left top right bottom)"""
|
||||
# TODO: combine multiple masks
|
||||
text, masks = get_function(text, "MASK", ["0 1", "0 1", "1", "multiply"])
|
||||
text, imasks = get_function(text, "IMASK", ["0", "1", "multiply"])
|
||||
text, feathers = get_function(text, "FEATHER", ["0 0 0 0"])
|
||||
text, maskw = get_function(text, "MASKW", ["1.0"])
|
||||
if not masks and not imasks:
|
||||
return text, None, None
|
||||
|
||||
def feather(f, mask):
|
||||
l, t, r, b, *_ = [int(x) for x in parse_floats(f[0], [0, 0, 0, 0], split_re="\\s+")]
|
||||
mask = FeatherMask().feather(mask, l, t, r, b)[0]
|
||||
log.info("FeatherMask l=%s, t=%s, r=%s, b=%s", l, t, r, b)
|
||||
return mask
|
||||
|
||||
mask = None
|
||||
totalweight = 1.0
|
||||
if maskw:
|
||||
totalweight = safe_float(maskw[0][0], 1.0)
|
||||
i = 0
|
||||
for m in masks:
|
||||
weight = safe_float(m[2], 1.0)
|
||||
op = m[3]
|
||||
nextmask = make_mask(m, size, weight)
|
||||
if i < len(feathers):
|
||||
nextmask = feather(feathers[i], nextmask)
|
||||
i += 1
|
||||
if mask is not None:
|
||||
log.info("MaskComposite op=%s", op)
|
||||
mask = MaskComposite().combine(mask, nextmask, 0, 0, op)[0]
|
||||
else:
|
||||
mask = nextmask
|
||||
|
||||
for idx, w, op in imasks:
|
||||
idx = int(safe_float(idx, 0.0))
|
||||
w = safe_float(w, 1.0)
|
||||
if len(input_masks) < idx + 1:
|
||||
log.warn("IMASK index %s not found, ignoring...", idx)
|
||||
continue
|
||||
nextmask = input_masks[idx] * w
|
||||
if i < len(feathers):
|
||||
nextmask = feather(feathers[i], nextmask)
|
||||
i += 1
|
||||
if mask is not None:
|
||||
mask = MaskComposite().combine(mask, nextmask, 0, 0, op)[0]
|
||||
else:
|
||||
mask = nextmask
|
||||
|
||||
# apply leftover FEATHER() specs to the whole
|
||||
for f in feathers[i:]:
|
||||
mask = feather(f, mask)
|
||||
|
||||
return text, mask, totalweight
|
||||
|
||||
|
||||
def get_noise(text):
|
||||
text, noises = get_function(
|
||||
text,
|
||||
"NOISE",
|
||||
["0.0", "none"],
|
||||
)
|
||||
if not noises:
|
||||
return text, None, None
|
||||
w = 0
|
||||
# Only take seed from first noise spec, for simplicity
|
||||
seed = safe_float(noises[0][1], "none")
|
||||
if seed == "none":
|
||||
gen = None
|
||||
else:
|
||||
gen = torch.Generator()
|
||||
gen.manual_seed(int(seed))
|
||||
for n in noises:
|
||||
w += safe_float(n[0], 0.0)
|
||||
return text, max(min(w, 1.0), 0.0), gen
|
||||
|
||||
|
||||
def apply_noise(cond, weight, gen):
|
||||
if cond is None or not weight:
|
||||
return cond
|
||||
|
||||
n = torch.randn(cond.size(), generator=gen).to(cond)
|
||||
|
||||
return cond * (1 - weight) + n * weight
|
||||
|
||||
|
||||
def encode_prompt(clip, text, start_pct, end_pct, defaults, masks):
|
||||
# First style modifier applies to ANDed prompts too unless overridden
|
||||
style, normalization, text = get_style(text)
|
||||
text, mask_size = get_mask_size(text, defaults)
|
||||
|
||||
prompts = [p.strip() for p in re.split(r"\bAND\b", text)]
|
||||
|
||||
p, sdxl_opts = get_sdxl(prompts[0], defaults)
|
||||
prompts[0] = p
|
||||
|
||||
def weight(t):
|
||||
opts = {}
|
||||
m = re.search(r":(-?\d\.?\d*)(![A-Za-z]+)?$", t)
|
||||
if not m:
|
||||
return (1.0, opts, t)
|
||||
w = float(m[1])
|
||||
tag = m[2]
|
||||
t = t[: m.span()[0]]
|
||||
if tag == "!noscale":
|
||||
opts["scale"] = 1
|
||||
|
||||
return w, opts, t
|
||||
|
||||
conds = []
|
||||
# TODO: is this still needed?
|
||||
# scale = sum(abs(weight(p)[0]) for p in prompts if not ("AREA(" in p or "MASK(" in p))
|
||||
for prompt in prompts:
|
||||
prompt, mask, mask_weight = get_mask(prompt, mask_size, masks)
|
||||
w, opts, prompt = weight(prompt)
|
||||
text, noise_w, generator = get_noise(text)
|
||||
if not w:
|
||||
continue
|
||||
prompt, area = get_area(prompt)
|
||||
prompt, local_sdxl_opts = get_sdxl(prompt, defaults)
|
||||
settings = {"prompt": prompt}
|
||||
settings["strength"] = w
|
||||
settings.update(sdxl_opts)
|
||||
settings.update(local_sdxl_opts)
|
||||
if area:
|
||||
settings["area"] = area[0]
|
||||
settings["strength"] = area[1]
|
||||
settings["set_area_to_bounds"] = False
|
||||
if mask is not None:
|
||||
settings["mask"] = mask
|
||||
settings["mask_strength"] = mask_weight
|
||||
|
||||
settings["start_percent"] = start_pct
|
||||
settings["end_percent"] = end_pct
|
||||
x = encode_prompt_segment(clip, prompt, settings, style, normalization)
|
||||
conds.extend(x)
|
||||
|
||||
return conds
|
||||
@@ -4,7 +4,7 @@ import logging
|
||||
|
||||
import folder_paths
|
||||
|
||||
log = logging.getLogger("comfyui-prompt-control")
|
||||
log = logging.getLogger("comfyui-prompt-control-legacy")
|
||||
|
||||
|
||||
def find_closing_paren(text, start):
|
||||
|
||||
+4
-4
@@ -1,16 +1,16 @@
|
||||
[project]
|
||||
name = "comfyui-prompt-control"
|
||||
description = "Nodes for convenient prompt editing, making many common operations prompt-controllable"
|
||||
name = "comfyui-prompt-control-legacy"
|
||||
description = "Legacy prompt control nodes. These exist only to allow old workflows to run"
|
||||
version = "1.2.1"
|
||||
license = { file = "LICENSE" }
|
||||
# some lark versions older than 1.1.9 apparently have a bug that breaks things, see https://github.com/asagi4/comfyui-prompt-control/issues/35
|
||||
dependencies = ["lark >= 1.1.9"]
|
||||
|
||||
[project.urls]
|
||||
Repository = "https://github.com/asagi4/comfyui-prompt-control"
|
||||
Repository = "https://github.com/asagi4/comfyui-prompt-control-legacy"
|
||||
# Used by Comfy Registry https://comfyregistry.org
|
||||
|
||||
[tool.comfy]
|
||||
PublisherId = "asagi4"
|
||||
DisplayName = "ComfyUI Prompt Control"
|
||||
DisplayName = "ComfyUI Prompt Control (LEGACY VERSION)"
|
||||
Icon = ""
|
||||
|
||||
Reference in New Issue
Block a user