tests: COUPLE mask shortcut

Tests for step count
test: Verify that basic paren escapes don't get stripped
2026-02-02 19:33:34 +02:00 · 2026-02-02 19:26:13 +02:00 · 2026-02-02 19:26:13 +02:00 · 2026-02-02 19:26:13 +02:00 · 2026-02-02 19:26:13 +02:00 · 2026-02-02 19:26:13 +02:00
34 changed files with 2879 additions and 3446 deletions
@@ -19,5 +19,5 @@ jobs:
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
-      - run: pip install pytest typing-extensions
+      - run: pip install pytest typing-extensions -r requirements.txt
      - run: PYTHONPATH=ComfyUI pytest tests/test_parser.py
@@ -31,7 +31,7 @@ jobs:
      - name: install-torch
        run: pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
      - name: install ComfyUI
-        run: pip install pytest typing-extensions -r ComfyUI/requirements.txt
+        run: pip install pytest typing-extensions -r requirements.txt -r ComfyUI/requirements.txt
      - name: Download clip_l.safetensors
        run: curl -LO https://huggingface.co/comfyanonymous/flux_text_encoders/resolve/main/clip_l.safetensors
      - name: Force Comfy to use the CPU
@@ -12,7 +12,7 @@ format:
 	ruff format

 test:
-	PYTHONPATH=../../ pytest tests/test_parser.py tests/test_cutout.py tests/test_macros.py $(ARGS)
+	PYTHONPATH=../../ pytest tests/test_parser.py $(ARGS)

 test_graph:
 	PYTHONPATH=../../  pytest tests/test_graph.py $(ARGS)
@@ -20,9 +20,6 @@ test_graph:
 test_encode:
 	PYTHONPATH=../../  pytest tests/test_encode.py $(ARGS)

-test_workflow:
-	PYTHONPATH=../../  pytest tests/test_workflow.py $(ARGS)
-
 test_encode_both:
 	TEST_TE="clip_l t5" PYTHONPATH=../../  pytest tests/test_encode.py $(ARGS)

@@ -1,32 +1,24 @@
 # ComfyUI prompt control

-Control LoRA and prompt scheduling, advanced text encoding, regional prompting, and much more, through your text prompt. Prompt Control generates dynamic graphs that are literally identical to handcrafted noodle soup, condensing complicated workflows with dozens of nodes into simple text prompts.
+Control LoRA and prompt scheduling, advanced text encoding, regional prompting, and much more, through your text prompt. Generates dynamic graphs that are literally identical to handcrafted noodle soup.

 Prompt Control comes with `PCTextEncode`, which provides advanced text encoding with many additional features compared to ComfyUI's base `CLIPTextEncode`.

 A `Basic Text to Image` template is included with the extension, and can be loaded from ComfyUI's template library.

-> [!NOTE]
-> v3.0.0 is backwards compatible with existing workflows, but requires at least ComfyUI v0.8.0
-> The parser was rewritten using parsy. It is intended to have the same behaviour as the old parser, but is **significantly** faster.
-> Please report any bugs or incompatibilities you find.
-
-## Notable changes
-
- `PC: Schedule Prompt` now strips surrounding whitespace by default, which may change some prompts. Add `NOSTRIP()` to your prompt to restore previous behaviour.
-
 ## What can it do?

+You can use text prompts to control the following:
+
 - A1111-style prompt scheduling and filtering without noodle soup.
- LoRA loading and [scheduling](/doc/schedules.md) using ComfyUI's built-in hook system.
+- LoRA loading and [scheduling](/doc/schedules.md) via the prompt, using ComfyUI's hook system
 - Masking, composition and area control ([regional prompting](/doc/regional_prompts.md)) with an implementation of [Attention Couple](/doc/attention_couple.md), also fully schedulable.
 - [Advanced prompt encoding](/doc/basic.md)
-  - Per-encoder prompts for models with multiple text encoders, such as SDXL and Flux.
+  - Per-encoder prompts for models with multiple text encoders, such as SDXL and Flux
  - Prompt combinators like `BREAK`, as well as `CAT`, `AVG()` and `AND` corresponding to ComfyUI's `ConditioningConcat`, `ConditioningAverage` and `ConditioningCombine` nodes.
  - Different weight interpretation types (ComfyUI, A1111, compel, etc.)
-  - Prompt masking with an implementation of [cutoff](https://github.com/BlenderNeko/ComfyUI_Cutoff).
- Organize complicated prompts with [segments and prompt macros](/doc/macros.md).
- [Schedule your own encoder nodes](/doc/node_function.md), allowing prompt control of eg. video or audio models with non-text inputs.
+  - Prompt masking with an implementation of [cutoff](https://github.com/BlenderNeko/ComfyUI_Cutoff)
+- Simple [prompt macros](/doc/macros.md) with `DEF`

 All features are fully schedulable unless otherwise stated. See the [scheduling syntax documentation](doc/schedules.md) to get started.

@@ -42,9 +34,16 @@ If you encounter issues as a user or if you're a node developer and Prompt Contr

 ## Requirements

-The v3 node schema uses features that require at least ComfyUI v0.8.0
+For LoRA scheduling to work, you'll need at least version 0.3.7 of ComfyUI (0.3.36 of ComfyUI desktop).

-If you run into problems, update ComfyUI first.
+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.

 # Core nodes

@@ -56,6 +55,10 @@ If you run into problems, update ComfyUI first.

 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, textinputname)` in a prompt to generate a graph using **any other node** that's compatible. 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. The "default" values are `PCTextEncode` and `text`.
+
+For example, if you for some reason do not want the advanced features of `PCTextEncode`, use `NODE(CLIPTextEncode)` in the prompt and you'll still get scheduling with ComfyUI's regular TE node.
+
 The advanced node enables filtering the prompt for multi-pass workflows.

 ## PCLazyLoraLoader and PCLazyLoraLoaderAdvanced
@@ -80,4 +83,6 @@ This node configures `PCTextEncode` default values for some functions by attachi

 # Known issues

+- ComfyUI's caching mechanism has an issue that makes it unnecessarily invalidate caches for certain inputs; you'll still get some benefit from the lazy nodes, but changing inputs that shouldn't affect downstream nodes (especially if using filtering) will still cause them to be recomputed because ComfyUI doesn't realize the inputs haven't changed.
+
 - Cutoff does not work with models that use non-CLIP text encoders, like Flux. This might be fixable, but it's uncertain if cutoff even makes sense for those models.
@@ -30,7 +30,7 @@ if "PYTEST_CURRENT_TEST" not in os.environ:
        h = logging.StreamHandler(sys.stdout)
        h.setFormatter(logging.Formatter("[PromptControl] %(levelname)s: %(message)s"))
        log.addHandler(h)
-    for node in ["base", "hooks", "tools", "lazy", "anima"]:
+    for node in ["base", "hooks", "tools", "lazy"]:
        mod = importlib.import_module(f".prompt_control.nodes_{node}", package=__name__)
        v3_modules.append(mod)

@@ -1,5 +1,7 @@
 # Attention Couple

+NOTE: This is still considered an experimental feature, so the syntax may change.
+
 Attention Couple is an attention-based implementation of regional prompting. it is faster and often more flexible than latent-based masking.

 The implementation is based on the one by [pamparamm](https://github.com/pamparamm/ComfyUI-ppm.git), modified to use ComfyUI's hook system. This enables it to work with prompt scheduling.
@@ -10,11 +12,6 @@ As a consequence of this, however, you can also use `COUPLE` in your negative pr

 To enable batching negative prompts, run your positive and negative prompt through the `PPCAttentionCoupleBatchNegative` node. This will make the outputs identical to pamparamm's implementation and will also improve performance. It will fall back to the default behaviour in cases where batching can't be done, so it should always be safe to use.

-## Anima
-
-There is a **very experimental** port of pamparamm's Anima support for Attention Couple in Prompt Control. Because ComfyUI lacks the built-in schedulable hooks required, you must first patch your model with `PC: Anima Attention Couple Model Patch` in addition to using `COUPLE` as usual.
-
-The code was hacked together with minimal thought, so expect bugs and misbehaviour. The port is also currently *not* compatible with NegPIP.

 ## Syntax

@@ -24,13 +21,6 @@ See also the [regional prompting documentation](/doc/regional_prompts.md) for in

 You can use `COUPLE` to attach attention-coupled prompts to a base prompt:

-For example:
-```
-dog FILL() COUPLE(0.5 1) cat
-```
-
-The full syntax looks as follows (to use `IMASK` you need to attach a custom mask)
-
 `base_prompt COUPLE MASK(0 0.5) coupled prompt 1 with mask COUPLE IMASK(0) coupled prompt 2 with custom mask`

 as a shortcut, `COUPLE(maskparams)` is expanded to `COUPLE MASK(maskparams)`, so the above prompt can also be written as:
@@ -38,17 +28,15 @@ as a shortcut, `COUPLE(maskparams)` is expanded to `COUPLE MASK(maskparams)`, so
 `base_prompt COUPLE(0 0.5) coupled prompt 1 with mask COUPLE IMASK(0) coupled prompt 2 with custom mask`

 Behaviour:
- If no mask is specified, an implicit `MASK()` is assumed, meaning that the prompt affects the entire image.
+- If no mask is specified, an implicit `MASK()` is assumed.

- For the base prompt, you can use `FILL()` to automatically mask all parts not masked by other coupled prompts
+- For the base prompt, you can also use `FILL()` to automatically mask all parts not masked by coupled prompts

- If the base prompt has weight set to zero (ie. ´:0` at the end), then the first coupled prompt with non-zero weight becomes the base prompt:
+- If the base prompt has weight set to zero (ie. ´:0` at the end), then the first coupled prompt with non-zero weight becomes the base prompt.

+For example:
 ```
-disabled prompt :0 COUPLE new base prompt COUPLE coupled prompt
+dog FILL() COUPLE(0.5 1) cat
 ```

-You can also schedule the weight normally: `prompt :[1:0:0.35]`
-
-> ![NOTE]
-> Note that because the generation still sees and diffuses the full latent, attention coupling is not guaranteed to perfectly limit the effect of your prompt to the masked area.
+Note that because the generation still sees and diffuses the full latent, attention coupling is not guaranteed to perfectly limit the effect of your prompt to the masked area.
@@ -31,7 +31,6 @@ Prompt operators are processed in the following order, meaning that all features

 - DEF macros are expanded
 - Scheduling is expanded, and for each scheduled prompt:
-  - SEGs are processed and the template is expanded
  - The prompt is split by AND, and for each:
    - Prompts are split by COUPLE. and for each:
      - Most functions (like MASK) and cutoffs are evaluated
@@ -58,42 +58,3 @@ a "$1" b "$2"
 a "" b "$2"
 a "A" b "$2"
 ```
-
-## SEG: Split your prompt into named segments
-
-Syntax: `SEG(segment_name)`
-
-To help with organizing prompts, you can use the `SEG` function. For example:
-```
-This is a comic
-Top panel: $CAT. $SEG3
-Bottom panel: $DOG
-
-SEG(DOG)
-A dog chasing its
-  tail in a living room.
-SEG(CAT)
-
-a sleeping cat
-
-SEG
-The cat has orange fur with white stripes
-```
-This produces:
-```
-This is a comic
-Top panel: a sleeping cat. The cat has orange fur with white stripes
-Bottom panel: A dog chasing its
-  tail in a living room.
-```
-
-> [!NOTE]
-> Unlike macros, SEGs are processed *after* scheduling syntax has been expanded, except in the LoRA loader (this may change later, but requires a bit of refactoring)
-
-In this case, the first section before any `SEG` becomes the *template* and any text after a `SEG` call becomes part of that segment. Whitespace is stripped from the start and end of segments and the template.
-
-In the template, you can refer to segments by either their index (starting from 1) or the given name, prefixed with a `$SEG`, so in this example, `$SEG1` is the same as `$PANEL1`
-
-Segments can also refer to each other. Recursion will terminate, but produces weird outputs.
-
-Naming segments is optional, in which case you will have to refer to it by its index.
@@ -1,41 +0,0 @@
-# The NODE function
-
-The `NODE` function allows you to use any other text encoding node within `PC: Schedule Prompt`, replacing the default `PCTextEncode` and allowing for example video model scheduling.
-
-> [!NOTE]
-> When using NODE, you lose access to *all* special syntax provided by `PCTextEncode`. Only SEGs, macros and scheduling will continue to work since those are processed at graph expansion time before the text prompt is passed into the node.
-
-## Basic usage
-
-Use `NODE(NodeClassName, textinputname)` in a prompt to generate a graph using any node that's compatible. The requirements are as follows:
- The node must have a CLIP parameter (which must be named `clip`)
- It must have a text field
- It must return a `CONDITIONING` as its first return value.
-
-For example, if you for some reason do not want the advanced features of `PCTextEncode`, use `NODE(CLIPTextEncode)` in the prompt and you'll still get scheduling with ComfyUI's regular TE node.
-
-The default parameters are `PCTextEncode` and `text`.
-
-## Advanced Usage with arbitrary parameters
-
-Advanced usage of `NODE` can be complicated. For an example, see [The H3 workflow](/example_workflows/Prompt%20Control%20with%20MiniMax%20H3.json?raw=1). You can also find it in the template library.
-
-The full synopsis of the function is `NODE(NodeClassName, textinputname, arg_spec)` where `arg_spec` is a semicolon-separated list of `parameter_name json_value` pairs. In raw form, it looks like this:
-```
-NODE(MiniMaxH3ImageToVideo, prompt, vae ["1", 0]; width 1024; height 1024; first_frame ["2", 0])
-```
-
-The names and inputs must match the ComfyUI API format which **may differ from frontend names**. You can export your workflow in API format and inspect it to see how inputs are passed in to nodes.
-
-The arrays are literal ComfyUI node links, meaning the `vae` parameter is taken from node ID "1" first output and `first_frame` from node ID "2" first output.
-
-The values are arbitrary JSON literals, meaning that you can also pass in constant values. To pass in literal strings for example, you need to use quotes `"like this"`.
-
-This is intended to be used with the helper node `PC: NODE Input Helper`, which can be used to pass arbitrary parameters (named `$a` to `$n`) to the encoder. The recommended pattern is to put something like the following:
-```
-SEG(node)
-NODE(MiniMaxH3ImageToVideo, prompt, vae $a; width $b; height $c; length $d; first_frame $e)
-```
-to the helper and then concatenate it at the end of your prompt (use whitespace as a separator). You can then trigger the node with `$node` in your prompt. (see [documentation](/doc/macros.md) for `SEG`)
-
-The helper will replace the parameters with the correct ComfyUI link values.
@@ -14,9 +14,6 @@ Besides the syntax documented below, the [basic syntax](/doc/basic.md) and [prom
 a [large::0.1] [cat|dog:0.05] [<lora:somelora:0.5:0.6>::0.5]
 [in a park:in space:0.4]
 ```
-## Note on whitespace
-`PC: Schedule Prompt` will strip leading and following whitespace from the prompt automatically. If you really want whitespace in your prompt, include `NOSTRIP()` in your prompt.
-
 ## Comments and escaping

 In schedules, any text on a line following a `#` is considered a comment and removed, including the `#` character.
@@ -1,167 +0,0 @@
-# Adapted from https://github.com/pamparamm/ComfyUI-ppm
-import itertools
-from collections.abc import Callable
-from functools import partial
-from math import lcm
-
-import torch
-import torch.nn.functional as F
-from comfy.ldm.anima.model import Anima as AnimaDIT
-from comfy.ldm.cosmos.predict2 import Attention as CosmosAttention
-from comfy.patcher_extension import WrapperExecutor
-from comfy.sampler_helpers import convert_cond
-from comfy.samplers import process_conds
-
-COND = 0
-UNCOND = 1
-
-
-def reshape_mask(mask: torch.Tensor, size: tuple[int, int], bs: int, num_tokens: int) -> torch.Tensor:
-    num_conds = mask.shape[0]
-
-    mask_downsample = F.interpolate(mask, size=size, mode="nearest")
-    mask_downsample_reshaped = mask_downsample.view(num_conds, num_tokens, 1).repeat_interleave(bs, dim=0)
-
-    return mask_downsample_reshaped
-
-
-def wrap_forwards(anima_model):
-    backups = {}
-    for block_name, b in (
-        (n, b) for n, b in anima_model.named_modules() if "cross_attn" in n and isinstance(b, CosmosAttention)
-    ):
-        backups[block_name] = b.forward
-        b.forward = partial(cosmos_attention_forward_couple, b.forward)
-    return backups
-
-
-def unwrap_forwards(anima_model, backups):
-    for block_name, b in (
-        (n, b) for n, b in anima_model.named_modules() if "cross_attn" in n and isinstance(b, CosmosAttention)
-    ):
-        b.forward = backups[block_name]
-
-
-def anima_sample_wrapper(executor, *args, **kwargs):
-    guider, _, extra_options, _, noise, latent_image, denoise_mask, *_ = args
-    seed = extra_options["seed"]
-    device = "cuda"  # TODO: fix
-
-    def pc_process_conds(pc_conds):
-        conds = [convert_cond([c])[0] for c in pc_conds]
-        conds = process_conds(
-            guider.inner_model,
-            noise,
-            {"positive": conds},
-            device,
-            latent_image,
-            denoise_mask,
-            seed,
-            latent_shapes=[latent_image.shape],
-        )
-        return [
-            c["model_conds"]["c_crossattn"].cond * pc_conds[i][1].get("strength", 1.0)
-            for i, c in enumerate(conds["positive"])
-        ]
-
-    extra_options["model_options"]["transformer_options"]["pc_process_conds"] = pc_process_conds
-    return executor(*args, **kwargs)
-
-
-def anima_forward_wrapper(executor: WrapperExecutor, *args, **kwargs):
-    """Model wrapper does something with activation shapes?"""
-    anima_model: AnimaDIT = executor.class_obj  # type: ignore
-
-    x: torch.Tensor = args[0]
-    transformer_options: dict = kwargs.get("transformer_options", {}).copy()
-    pc = transformer_options.get("pc_couple")
-    if pc and "processed_conds" not in pc:
-        pc["processed_conds"] = transformer_options["pc_process_conds"](pc["conds"])
-    patch_spatial = anima_model.patch_spatial
-
-    activations_shape = list(x.shape)
-    activations_shape[-2] = activations_shape[-2] // patch_spatial
-    activations_shape[-1] = activations_shape[-1] // patch_spatial
-
-    transformer_options["activations_shape"] = activations_shape
-    kwargs["transformer_options"] = transformer_options
-
-    b = {}
-    if pc:
-        b = wrap_forwards(anima_model)
-    r = executor(*args, **kwargs)
-    if pc:
-        unwrap_forwards(anima_model, b)
-    return r
-
-
-def cosmos_attention_forward_couple(_forward: Callable, x, context, rope_emb, transformer_options):
-    """attention block wrapper"""
-    if "pc_couple" not in transformer_options:
-        return _forward(x, context, rope_emb, transformer_options)
-    c: torch.Tensor = context
-    # FIXME: base cond weight
-    # c = args["processed_conds"][0]
-
-    args = transformer_options["pc_couple"]
-
-    mask = args["mask"]
-    conds = args["processed_conds"][1:]
-    num_conds = len(conds) + 1
-    num_tokens_c: list[int] = [c.shape[1] for c in conds]
-    cond_or_uncond = transformer_options["cond_or_uncond"]
-    cond_or_uncond_couple = []
-
-    num_chunks = len(cond_or_uncond)
-    bs = x.shape[0] // num_chunks
-
-    x_chunks = x.chunk(num_chunks, dim=0)
-    c_chunks = c.chunk(num_chunks, dim=0)
-    lcm_tokens_c = lcm(c.shape[1], *num_tokens_c)
-    conds_c_tensor = torch.cat(
-        [cond.repeat(bs, lcm_tokens_c // num_tokens_c[i], 1) for i, cond in enumerate(conds)],
-        dim=0,
-    )
-
-    xs, cs = [], []
-    for i, cond_type in enumerate(cond_or_uncond):
-        x_target = x_chunks[i]
-        c_target = c_chunks[i].repeat(1, lcm_tokens_c // c.shape[1], 1)
-        if cond_type == UNCOND:
-            xs.append(x_target)
-            cs.append(c_target)
-            cond_or_uncond_couple.append(UNCOND)
-        else:
-            xs.append(x_target.repeat(num_conds, 1, 1))
-            cs.append(torch.cat([c_target, conds_c_tensor], dim=0))
-            cond_or_uncond_couple.extend(itertools.repeat(COND, num_conds))
-
-    xs = torch.cat(xs, dim=0)
-    cs = torch.cat(cs, dim=0)
-
-    out = _forward(xs, cs, rope_emb, transformer_options)
-
-    size = tuple(transformer_options["activations_shape"][-2:])
-    num_tokens = out.shape[1]
-    mask_downsample = reshape_mask(mask, size, bs, num_tokens)
-
-    outputs = []
-    cond_outputs = []
-    i_cond = 0
-
-    for i, cond_type in enumerate(cond_or_uncond_couple):
-        pos, next_pos = i * bs, (i + 1) * bs
-
-        if cond_type == UNCOND:
-            outputs.append(out[pos:next_pos])
-        else:
-            pos_cond, next_pos_cond = i_cond * bs, (i_cond + 1) * bs
-            masked_output = out[pos:next_pos] * mask_downsample[pos_cond:next_pos_cond]
-            cond_outputs.append(masked_output)
-            i_cond += 1
-
-    if len(cond_outputs) > 0:
-        cond_output = torch.stack(cond_outputs).sum(0)
-        outputs.append(cond_output)
-
-    return torch.cat(outputs, dim=0)
@@ -52,7 +52,7 @@ class Proxy:
        return self

    def __call__(self, *args, **kwargs):
-        return self.function(*args, **kwargs)
+        return self.function(*args, *kwargs)


 class AttentionCoupleHook(TransformerOptionsHook):
@@ -63,23 +63,20 @@ class AttentionCoupleHook(TransformerOptionsHook):
    def __init__(self):
        super().__init__(hook_scope=EnumHookScope.HookedOnly)

-        self.transformers_dict: dict[str, Any] = {
+        self.transformers_dict = {
            "patches": {
                "attn2_output_patch": [Proxy(self.attn2_output_patch)],
                "attn2_patch": [Proxy(self.attn2_patch)],
-            },
-            "pc_couple": {},
+            }
        }
-
        self.has_negpip = False
-        # The list will be calculated later. All clones must refer to the same kv dict
-        self.kv: dict[str, list] = {"k": None, "v": None}  # type: ignore
+        # calculate later. All clones must refer to the same kv dict
+        self.kv = {"k": [], "v": []}

    def initialize_regions(self, base_cond, conds, fill):
        self.num_conds = len(conds) + 1
        self.base_strength = base_cond[1].get("strength", 1.0)
        self.strengths: list[float] = [cond[1].get("strength", 1.0) for cond in conds]
-        self.comfy_conds = [base_cond] + conds
        self.conds: list[torch.Tensor] = [base_cond[0]] + [cond[0] for cond in conds]
        base_mask = base_cond[1].get("mask", None)
        masks = [cond[1].get("mask") * cond[1].get("mask_strength") for cond in conds]
@@ -119,11 +116,6 @@ class AttentionCoupleHook(TransformerOptionsHook):
        self.mask = mask / mask.sum(dim=0, keepdim=True)

    def on_apply_hooks(self, model: ModelPatcher, transformer_options: dict[str, Any]):
-        self.transformers_dict["pc_couple"] = {
-            "conds": self.comfy_conds,
-            "num_conds": self.num_conds,
-            "mask": self.mask,
-        }
        if self.kv["k"] is None:
            self.has_negpip = model.model_options.get("ppm_negpip", False)
            log.debug("AttentionCouple has_negpip=%s", self.has_negpip)
@@ -1,25 +1,58 @@
-import re
+from typing import TypeAlias

-from .utils import parse_args
+import lark

-CUTOFF_RE = re.compile(r"\[CUT:((.*?):(.*?))\]")
+from .parser import flatten
+
+cut_parser = lark.Lark(
+    r"""
+!start: (cut | prompt | /[][:()]/+)*
+prompt: (PLAIN | WHITESPACE)+
+cut: "[CUT:" prompt ":" prompt [":" NUMBER  [ ":" NUMBER [":" NUMBER [ ":" PLAIN ] ] ] ]"]"
+WHITESPACE: /\s+/
+PLAIN: /([^\[\]:])+/
+%import common.SIGNED_NUMBER -> NUMBER
+"""
+)


-def noop(x):
-    return x
+class CutTransform(lark.Transformer):
+    def __default__(self, data, children, meta):
+        return children

+    def NUMBER(self, args):
+        return float(args)

-def parse_cuts(string):
-    text = CUTOFF_RE.sub(r"\2", string)
-    cutoffs = CUTOFF_RE.findall(string)
-    cs = []
-    for x, *_ in cutoffs:
-        p = x.split(":")
-        args = parse_args(
-            p, [(str, ""), (str, ""), (float, 0), (float, None), (float, None), (noop, None)], strip=False
+    def cut(self, args):
+        prompt, cutout, weight, strict_mask, start_from_masked, mask_token = args
+
+        # prompts and cutouts are always sequences of str
+        return (
+            "".join(prompt),
+            "".join(cutout),
+            weight,
+            strict_mask,
+            start_from_masked,
+            mask_token,
        )
-        args = tuple(args)
-        if not args[0] or not args[1] or (args[5] is not None and not args[5].strip()):
-            raise ValueError(f"Invalid CUT spec: [CUT:{x}]")
-        cs.append(args)
-    return text, cs
+
+    def start(self, args):
+        prompt = []
+        cuts = []
+        for a in flatten(args):
+            if isinstance(a, str):
+                prompt.append(a)
+            else:
+                prompt.append(a[0])
+                cuts.append(a)
+        return "".join(prompt), cuts
+
+    def PLAIN(self, args: str) -> str:
+        return str(args)
+
+
+CutResult: TypeAlias = tuple[str, str, float, float, float, str]
+
+
+def parse_cuts(text: str) -> tuple[str, CutResult]:
+    return CutTransform().transform(cut_parser.parse(text))
@@ -4,52 +4,12 @@ from __future__ import annotations
 import logging
 import re

-from .utils import find_closing_paren, get_function, split_by_function
+from .utils import find_closing_paren, get_function

+logging.basicConfig()
 log = logging.getLogger("comfyui-prompt-control")


-def substitute_template(template, segments, do_subs):
-    def _substitute(template, segments, stack):
-        name = ""
-        if "$" in template:
-            for name, value in sorted(segments):
-                value = substitute_var(value, name, "")
-                if name not in stack:
-                    stack.add(name)
-                    value = _substitute(value, segments, stack)
-                    stack.remove(name)
-                template = substitute_var(template, name, value)
-        if do_subs and name not in stack:
-            template = expand_subs(template)
-        return template
-
-    return _substitute(template, segments, set())
-
-
-def expand_segs(text, do_subs=True):
-    template, segments = split_by_function(text, "SEG", defaults=[""], require_args=True)
-    named_segs = [(f.args[0].strip() or f"SEG{i + 1}", c.strip()) for i, (c, f) in enumerate(segments)]
-
-    new_text = substitute_template(template, named_segs, do_subs).strip()
-    if new_text != text.strip():
-        log.debug("Template expanded to: %s", new_text)
-    return new_text
-
-
-def expand_subs(text):
-    text, subs = get_function(text, "SUB", defaults=None)
-    subs = [spec.strip() for f in subs for spec in f.args[0].split(";")]
-    for spec in subs:
-        if len(spec) <= 3 or spec[0] != "s":
-            log.warning("Invalid SUB spec ignored: '%s'", spec)
-            continue
-        splitchar = spec[1]
-        search, replace, *_ = spec[2:].split(splitchar)
-        text = re.sub(search, replace, text)
-    return text
-
-
 def parse_search(search):
    arg_start = search.find("(")
    args = ""
@@ -69,12 +29,8 @@ def parse_search(search):
    return name, args


-def expand_macros(text, defs=None):
-    silent = False
-    if defs is None:
-        text, defs = get_function(text, "DEF", defaults=None)
-    else:
-        silent = True
+def expand_macros(text):
+    text, defs = get_function(text, "DEF", defaults=None)
    res = text
    prevres = text
    replacements = []
@@ -92,6 +48,7 @@ def expand_macros(text, defs=None):
        iterations += 1
        if iterations > 10:
            raise ValueError("Unable to resolve DEFs, make sure there are no cycles!")
+            return text
        for search, replace in replacements:
            res = substitute_defcall(res, search, replace)
        if res == prevres:
@@ -99,33 +56,26 @@ def expand_macros(text, defs=None):
        prevres = res
    if res.strip() != text.strip():
        res = res.strip()
-        if not silent:
-            log.debug("DEFs expanded to: %s", res)
+        log.info("DEFs expanded to: %s", res)
    return res


-def substitute_var(text, name, replace, boundary=r"\b"):
-    if f"${name}" not in text:
-        return text
-    name = re.escape(str(name))
-    return re.sub(rf"\${name}{boundary}", replace, text)
-
-
 def substitute_defcall(text, search, replace):
    name, default_args = search
-
-    def run_macro(*parameters):
+    text, defns = get_function(text, name, defaults=None, placeholder=f"DEFNCALL{name}", require_args=False)
+    for i, d in enumerate(defns):
+        ph = d.placeholder
+        assert ph is not None, "This is a bug"
+        parameters = d.args
        paramvals = []
        if parameters:
            paramvals = [x.strip() for x in parameters[0].split(";")]
        r = replace
-        end_re = r"(?![0-9])"
        for i, v in enumerate(paramvals):
-            r = substitute_var(r, i + 1, v, boundary=end_re)
+            r = re.sub(rf"\${i + 1}\b", v, r)

        for i, v in enumerate(default_args):
-            r = substitute_var(r, i + 1, v, boundary=end_re)
-        return r
+            r = re.sub(rf"\${i + 1}\b", v, r)

-    text, _ = get_function(text, name, defaults=None, processor=run_macro, require_args=False)
+        text = text.replace(ph, r)
    return text
@@ -1,52 +0,0 @@
-# Adapted from ComfyUI-ppm into hook form
-
-
-import comfy.model_management
-import comfy.patcher_extension
-from comfy.model_base import Anima
-from comfy.model_patcher import ModelPatcher
-from comfy_api.latest import io
-
-from .anima_couple import (
-    anima_forward_wrapper,
-    anima_sample_wrapper,
-)
-
-
-class PCAnimaAttnCouplePatch(io.ComfyNode):
-    @classmethod
-    def define_schema(cls) -> io.Schema:
-        return io.Schema(
-            node_id="PCAnimaAttnCouplePatch",
-            display_name="PC: Anima attention Couple Model Patch",
-            category="promptcontrol/experimental",
-            inputs=[
-                io.Model.Input("model"),
-            ],
-            outputs=[
-                io.Model.Output(),
-            ],
-        )
-
-    @classmethod
-    def execute(cls, model: ModelPatcher) -> io.NodeOutput:
-        model_type = type(model.model)
-        m = model
-
-        if issubclass(model_type, Anima):
-            m = model.clone()
-            m.add_wrapper_with_key(
-                comfy.patcher_extension.WrappersMP.DIFFUSION_MODEL,
-                cls.__name__,
-                anima_forward_wrapper,
-            )
-            m.add_wrapper_with_key(
-                comfy.patcher_extension.WrappersMP.SAMPLER_SAMPLE,
-                cls.__name__,
-                anima_sample_wrapper,
-            )
-
-        return io.NodeOutput(m)
-
-
-NODES = [PCAnimaAttnCouplePatch]
@@ -2,8 +2,7 @@ import logging

 from comfy_api.latest import io

-from .macros import expand_segs
-from .prompts import encode_prompt, hook_te
+from .prompts import encode_prompt

 log = logging.getLogger("comfyui-prompt-control")

@@ -26,11 +25,10 @@ class PCTextEncodeWithRange(io.ComfyNode):
        )

    @classmethod
-    def execute(cls, clip, text, start=0.0, end=1.0) -> io.NodeOutput:
+    def execute(cls, clip, text, start=0.0, end=1.0) -> io.NodeOutput:  # ty: ignore[invalid-method-override]
        log.debug("PCTextEncode: Encoding '%s'", text)
        defaults = clip.patcher.model_options.get("x-promptcontrol.defaults", {})
        masks = clip.patcher.model_options.get("x-promptcontrol.masks", None)
-        text = expand_segs(text)
        out = encode_prompt(clip, text, start, end, defaults, masks)
        return io.NodeOutput(out)

@@ -51,36 +49,12 @@ class PCTextEncode(io.ComfyNode):
        )

    @classmethod
-    def execute(cls, clip, text) -> io.NodeOutput:
+    def execute(cls, clip, text) -> io.NodeOutput:  # ty: ignore[invalid-method-override]
        # Use the WithRange node for the range 0.0, 1.0
        return PCTextEncodeWithRange.execute(clip, text, 0.0, 1.0)


-class PCHookEncoderModsInternal(io.ComfyNode):
-    @classmethod
-    def define_schema(cls):
-        return io.Schema(
-            node_id="PCHookTextEncoderModsInternal",
-            display_name="PC: Apply Text Encoder Mods",
-            category="promptcontrol",
-            description="Apply TE modifications (internal)",
-            is_experimental=True,
-            is_dev_only=True,
-            inputs=[
-                io.Clip.Input("clip"),
-                io.String.Input("te_names"),
-                io.String.Input("style"),
-                io.String.Input("normalization"),
-                io.Custom("PC_EXTRA_DATA").Input("extra", optional=True),
-            ],
-            outputs=[io.Clip.Output()],
-        )
-
-    @classmethod
-    def execute(cls, clip, te_names, style, normalization, extra) -> io.NodeOutput:
-        te_names = [x.strip() for x in te_names.split(",")]
-        clip = hook_te(clip, te_names, style, normalization, extra)
-        return io.NodeOutput(clip)
-
-
-NODES = [PCTextEncodeWithRange, PCTextEncode, PCHookEncoderModsInternal]
+NODES = [
+    PCTextEncodeWithRange,
+    PCTextEncode,
+]
@@ -29,7 +29,7 @@ class PCLoraHooksFromText(io.ComfyNode):
        )

    @classmethod
-    def execute(cls, text) -> io.NodeOutput:
+    def execute(cls, text) -> io.NodeOutput:  # ty: ignore[invalid-method-override]
        prompt_schedule = parse_prompt_schedules(text)
        consolidated = consolidate_schedule(prompt_schedule)
        hooks = lora_hooks_from_schedule(consolidated, {})
@@ -104,7 +104,7 @@ class PCAttentionCoupleBatchNegative(io.ComfyNode):

    @classmethod
    @override
-    def execute(cls, positive, negative) -> io.NodeOutput:
+    def execute(cls, positive, negative) -> io.NodeOutput:  # ty: ignore[invalid-method-override]
        if len(negative) != 1:
            log.warning("Batching scheduled negatives is not supported yet")
            return io.NodeOutput(positive, negative)
@@ -3,16 +3,20 @@ from __future__ import annotations

 import json
 import logging
+import os

 from comfy_api.latest import io
 from comfy_execution.graph import ExecutionBlocker
 from comfy_execution.graph_utils import GraphBuilder

-from .macros import expand_macros, expand_segs
-from .parser import parse_prompt_schedules
-from .utils import consolidate_schedule, find_nonscheduled_loras, get_function, split_by_function
+from .utils import consolidate_schedule, find_nonscheduled_loras, get_function

 log = logging.getLogger("comfyui-prompt-control")
+if os.environ.get("PC_USE_NEW_PARSER", "0") == "1":
+    log.info("Using new parsy parser")
+    from .parser_parsy import parse_prompt_schedules as parse_prompt_schedules
+else:
+    from .parser import parse_prompt_schedules


 def create_lora_loader_nodes(graph, model, clip, loras):
@@ -159,7 +163,6 @@ class PCLazyLoraLoaderAdvanced(io.ComfyNode):

    @classmethod
    def execute(cls, model=None, clip=None, text="", apply_hooks=True, tags="", start=0.0, end=1.0, num_steps=0):
-        text = expand_segs(expand_macros(text))
        schedule = parse_prompt_schedules(text, filters=tags, start=start, end=end, num_steps=num_steps)
        graph = GraphBuilder()
        r = build_lora_schedule(graph, schedule, model, clip, apply_hooks=apply_hooks)
@@ -192,106 +195,25 @@ class PCLazyLoraLoader(io.ComfyNode):
        return io.NodeOutput(*no.args[:2], expand=no.expand)


-def parse_extra_inputs(args, defaults):
-    params = {}
-    if not args.strip():
-        return defaults + [{}]
-    defaults = defaults[:]
-    defaults.append("")
-    for i, v in enumerate(args.split(",", maxsplit=len(defaults) - 1)):
-        defaults[i] = v
-        # We should strip extra whitespace so that people don't have to worry about functions.
-    magic_spec = defaults[-1]
-    magic_spec.replace(r"\;", "__ESCAPED_SEMICOLON__")
-    extra_inputs = magic_spec.split(";") if magic_spec.strip() else []
-    for e in extra_inputs:
-        e = e.strip()
-        if not e:
-            continue
-        e = e.replace("__ESCAPED_SEMICOLON__", ";")
-        name, jsondata = e.split(maxsplit=1)
-        jsondata = jsondata.strip()
-        if not jsondata.strip():
-            continue
-        # From helper node:
-        if jsondata == "__EMPTY__":
-            continue
-        try:
-            params[name.strip()] = json.loads(jsondata.strip())
-        except ValueError as e:
-            raise ValueError(f"Invalid JSON input: '{jsondata}'") from e
-    return [x.strip() for x in defaults[:-1]] + [params]
-
-
-def make_node(graph, p, clip, strip):
-    p, classnames = get_function(p, "NODE", defaults=None)
-    p, filters = get_function(p, "FILTER", defaults=None)
-    args = ""
-    if len(classnames) > 1:
-        log.warning("You have more than one NODE call in your prompt. Only the first one will be used")
-    if classnames:
-        args = classnames[0].args[0]
-        if not args.strip():
-            raise ValueError("NODE can't be empty!")
-    classname, paramname, extras = parse_extra_inputs(args, ["PCTextEncode", "text"])
-    # We should strip extra whitespace so that people don't have to worry about functions.
-    node = graph.node(classname.strip())
-    node.set_input("clip", clip)
-    node.set_input(paramname.strip(), p.strip() if strip else p)
-    for e, v in extras.items():
-        node.set_input(e, v)
-
-    for f in filters:
-        classname, paramname, extras = parse_extra_inputs(f.args[0], ["", "conditioning"])
-        if not classname:
-            raise ValueError("FILTER requires a Node class name")
-        extras[paramname] = node.out(0)
-        node = graph.node(classname)
-        for e, v in extras.items():
-            node.set_input(e, v)
-
-    return node
-
-
-def build_prompt(graph, prompt, clip, start=None, end=None):
-    p = prompt
-    strip = "NOSTRIP()" not in p
-    p = p.replace("NOSTRIP()", "")
-    # Need to explicitly expand SEGs here *before* NODE is processed
-    p = expand_segs(p)
-    p, combines = split_by_function(p, "COMBINE")
-    current_cond = make_node(graph, p, clip, strip)
-    for text, f in combines:
-        classname, param1, param2, extra = parse_extra_inputs(f.args[0], ["", "conditioning_1", "conditioning_2"])
-        if classname.strip() == "":
-            raise ValueError("Can't use COMBINE without a class name")
-        combiner = graph.node(classname.strip())
-        c2 = make_node(graph, text, clip, strip)
-        extra[param1] = current_cond.out(0)
-        extra[param2] = c2.out(0)
-        for e, v in extra.items():
-            combiner.set_input(e, v)
-        current_cond = combiner
-
-    node = current_cond
-    if start is not None and end is not None:
-        node = graph.node("ConditioningSetTimestepRange")
-        node.set_input("conditioning", current_cond.out(0))
-        node.set_input("start", start)
-        node.set_input("end", end)
-
-    return node
-
-
 def build_scheduled_prompts(graph, schedules, clip):
    nodes = []
    start_pct = 0.0
    for end_pct, c in schedules:
        p = c["prompt"]
-        node = build_prompt(graph, p, clip, start_pct, end_pct)
-        nodes.append(node)
+        p, classnames = get_function(p, "NODE", ["PCTextEncode", "text"])
+        classname = "PCTextEncode"
+        paramname = "text"
+        if classnames:
+            classname, paramname = classnames[0].args
+        node = graph.node(classname)
+        node.set_input("clip", clip)
+        node.set_input(paramname, p)
+        timestep = graph.node("ConditioningSetTimestepRange")
+        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")
@@ -355,49 +277,9 @@ class PCLazyTextEncode(io.ComfyNode):
        return PCLazyTextEncodeAdvanced.execute(clip, text)


-predefined_macros = get_function(
-    """
-DEF(AND=COMBINE(ConditioningCombine, conditioning_1, conditioning_2))
-DEF(CAT=COMBINE(ConditioningConcat, conditioning_to, conditioning_from))
-DEF(AVG(0.5)=COMBINE(ConditioningAverage, conditioning_from, conditioning_to, conditioning_to_strength $1))
-""",
-    "DEF",
-    defaults=None,
-)
-
-
-class PCLazyTextEncodeSingle(io.ComfyNode):
-    @classmethod
-    def define_schema(cls):
-        return io.Schema(
-            node_id="PCLazyTextEncodeSingle",
-            display_name="PC: Prompt (without scheduling)",
-            is_experimental=True,
-            is_dev_only=True,
-            enable_expand=True,
-            category="promptcontrol",
-            inputs=[
-                io.Clip.Input("clip", raw_link=True),
-                io.String.Input("text", multiline=True, default=""),
-            ],
-            outputs=[
-                io.Conditioning.Output("conditioning"),
-            ],
-        )
-
-    @classmethod
-    def execute(cls, clip, text):
-        graph = GraphBuilder()
-        text = expand_macros(text, predefined_macros)
-        node = build_prompt(graph, text, clip)
-        g = graph.finalize()
-        return io.NodeOutput(node.out(0), expand=g)
-
-
 NODES = [
    PCLazyTextEncode,
    PCLazyTextEncodeAdvanced,
-    PCLazyTextEncodeSingle,
    PCLazyLoraLoader,
    PCLazyLoraLoaderAdvanced,
 ]
@@ -1,13 +1,8 @@
-import json
 import logging

 from comfy_api.latest import io

-from .macros import expand_macros as macroexpand
-from .macros import expand_segs as segexpand
-from .macros import expand_subs as subexpand
-from .macros import substitute_var
-from .parser import parse_prompt_schedules
+from .parser import expand_macros, parse_prompt_schedules

 log = logging.getLogger("comfyui-prompt-control")

@@ -136,32 +131,22 @@ class PCExtractScheduledPrompt(io.ComfyNode):
    def define_schema(cls):
        return io.Schema(
            node_id="PCExtractScheduledPrompt",
-            display_name="PC: Show Prompt",
+            display_name="PC: Extract Scheduled Prompt",
            category="promptcontrol/tools",
            description="Parses the input prompt and returns the prompt scheduled at the specified point",
            inputs=[
                io.String.Input("text", multiline=True),
                io.Float.Input("at", min=0.0, max=1.0, default=1.0, step=0.01),
                io.String.Input("tags", default="", optional=True),
-                io.Boolean.Input("expand_segs", default=False, optional=True),
-                io.Boolean.Input("expand_subs", default=False, optional=True),
-                io.Boolean.Input("expand_macros", default=False, optional=True),
            ],
            outputs=[io.String.Output()],
-            search_aliases=["extract scheduled prompt"],
        )

    @classmethod
-    def execute(cls, text, at, tags="", expand_segs=False, expand_subs=False, expand_macros=False) -> io.NodeOutput:
-        if expand_macros:
-            text = macroexpand(text)
+    def execute(cls, text, at, tags="") -> io.NodeOutput:
        schedule = parse_prompt_schedules(text, filters=tags)
-        _, entry = schedule.at_step(at)
+        _, entry = schedule.at_step(at, total_steps=1)
        prompt_text = entry.get("prompt", "")
-        if expand_segs:
-            prompt_text = segexpand(prompt_text, do_subs=expand_subs)
-        if expand_subs:
-            prompt_text = subexpand(prompt_text)
        return io.NodeOutput(prompt_text)


@@ -181,71 +166,7 @@ class PCMacroExpand(io.ComfyNode):

    @classmethod
    def execute(cls, text) -> io.NodeOutput:
-        return io.NodeOutput(macroexpand(text))
-
-
-class PCLinkHelper(io.ComfyNode):
-    # a-z
-    NAMES = [chr(97 + i) for i in range(26)]
-
-    @classmethod
-    def define_schema(cls):
-        t1 = io.Autogrow.TemplateNames(io.AnyType.Input("link", raw_link=True), min=0, names=cls.NAMES)
-        t2 = io.Autogrow.TemplateNames(
-            io.AnyType.Input("value", lazy=True), min=0, names=[f"var{i + 1}" for i in range(50)]
-        )
-        return io.Schema(
-            node_id="PCNODELinkHelper",
-            display_name="PC: Extra argument helper for NODE",
-            category="promptcontrol/tools",
-            description="Takes in arbitrary inputs and renders them as NODE-compatible values, replacing $a -> $z with JSON link values.",
-            is_experimental=True,
-            inputs=[
-                io.Autogrow.Input("links", template=t1),
-                io.Autogrow.Input(
-                    "vars",
-                    template=t2,
-                ),
-                io.String.Input(
-                    "template",
-                    tooltip="The variables $a to $z will be replaced in this text with their corresponding input's JSON link value",
-                    placeholder="In this text you can refer to the input links as $a, $b etc. and the var inputs as either $var1 or $json1 etc. (the latter will be rendered through Python's json.dumps function which will cause strings to be quoted)",
-                    multiline=True,
-                ),
-            ],
-            outputs=[io.String.Output()],
-        )
-
-    # This requires https://github.com/Comfy-Org/ComfyUI/pull/15103 to work properly
-    # Without that PR, all inputs will be evaluated non-lazily
-    @classmethod
-    def check_lazy_status(cls, template, links, vars):
-        r = []
-        for name, (v, input_name) in vars.items():
-            if v is None and f"${name}" in template or v is None and f"$json{name[3:]}" in template:
-                r.append(input_name)
-        return r
-
-    @classmethod
-    def execute(cls, template, links, vars) -> io.NodeOutput:
-        text = template
-        for k in cls.NAMES:
-            v = "__EMPTY__"
-            if k in links:
-                # Replace : with \: to avoid breaking scheduling syntax when linking subgraphs. Any function that consumes this should replace \: with :
-                v = json.dumps(links[k]).replace(":", r"\:")
-            text = substitute_var(text, k, v)
-        for i in range(50):
-            v = "__EMPTY__"
-            k = f"var{i + 1}"
-            if k in vars:
-                v = vars[k]
-            text = substitute_var(text, k, str(v))
-            if f"$json{i + 1}" in text:
-                v = v if v == "__EMPTY__" else json.dumps(v)
-                text = substitute_var(text, f"json{i + 1}", v)
-
-        return io.NodeOutput(text)
+        return io.NodeOutput(expand_macros(text))


 NODES = [
@@ -255,5 +176,4 @@ NODES = [
    PCSetLogLevel,
    PCExtractScheduledPrompt,
    PCMacroExpand,
-    PCLinkHelper,
 ]
@@ -1,397 +1,367 @@
+# vim: sw=4 ts=4
 from __future__ import annotations

-import itertools as it
-from dataclasses import dataclass
+import logging
+from functools import lru_cache
 from math import ceil
-from typing import Any, TypeAlias

-from typing_extensions import override
+import lark

 from .macros import expand_macros
-from .parsy import any_char, char_from, digit, eof, forward_declaration, generate, regex, seq, string, success

-FOREVER = float("inf")
+logging.basicConfig()
+log = logging.getLogger("comfyui-prompt-control")

-EvalResult: TypeAlias = tuple[float, str, list["LoRA"]]
+if lark.__version__ == "0.12.0":
+    from sys import executable
+
+    x = "\n".join(
+        [
+            "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!",
+            f"{executable} -m pip uninstall lark-parser lark",
+            f"{executable} -m pip install lark",
+        ]
+    )
+    log.error(x)
+    raise ImportError(x)


-def merge_until(i: EvalResult, minimum: float):
-    until, p, loras = i
-    until = min(until, minimum)
-    return until, p, loras
+ESCAPES = [
+    ("XxPCBackslashESCAPExX", "\\"),
+    ("XxPCColonESCAPExX", ":"),
+    ("XxPCCommentESCAPExX", "#"),
+]


-def batched(iterable, n, *, strict=False):
-    # batched('ABCDEFG', 2) → AB CD EF G
-    if n < 1:
-        raise ValueError("n must be at least one")
-    iterator = iter(iterable)
-    while batch := tuple(it.islice(iterator, n)):
-        if strict and len(batch) != n:
-            raise ValueError("batched(): incomplete batch")
-        yield batch
+def escape_specials(string: str) -> str:
+    for ph, c in ESCAPES:
+        string = string.replace(rf"\{c}", ph)
+    return string


-EvalResult: TypeAlias = tuple[float, str, list["LoRA"]]
+def restore_escaped(string: str) -> str:
+    for ph, c in ESCAPES:
+        string = string.replace(ph, c)
+    return string


-class Expression:
-    def eval(self, step: float, tags: list[str]) -> EvalResult:
-        return (FOREVER, "", [])
-
-    def required_steps(self, max_steps: float) -> set[float]:
-        return set()
+def remove_comments(string: str) -> str:
+    r = []
+    for line in string.split("\n"):
+        comment = line.find("#")
+        if comment >= 0:
+            r.append(line[:comment])
+        else:
+            r.append(line)
+    return "\n".join(r)


-@dataclass
-class Text(Expression):
-    string: str
-
-    @override
-    def eval(self, step: float, tags: list[str]) -> EvalResult:
-        assert isinstance(self.string, str)
-        return FOREVER, self.string, []
+prompt_parser = lark.Lark(
+    r"""
+!start: (prompt | /[][():|]/+)*
+prompt: (emphasized | embedding | scheduled | alternate | sequence | loraspec | PLAIN | | /\\:/ | /</ | />/ | WHITESPACE)+
+!emphasized: "(" prompt? ")"
+        | "(" prompt ":" prompt ")"
+        | "[" prompt "]"
+promptlist: ([prompt] ":")~1..3
+scheduled: "[" promptlist _WS? NUMBER ["," NUMBER] "]"
+        | "[" promptlist _WS? TAG "]"
+sequence.5:  "[SEQ" ":" [prompt] ":" NUMBER (":" [prompt] ":" NUMBER)* "]"
+alternate: "[" [prompt] ("|" [prompt])+ [":" NUMBER] "]"
+loraspec.99: "<lora:" FILENAME lora_weights [lora_block_weights] ">"
+lora_weights.1: (":" _WS? NUMBER)~1..2
+lora_block_weights.-1: ":" PLAIN
+embedding.100: "<emb:" FILENAME ">"
+WHITESPACE: /\s+/
+_WS: WHITESPACE
+PLAIN: /([^<>\\\[\]():|]|\\.)+/
+FILENAME: /[^<>:]+/
+TAG: /[A-Z_]+/
+%import common.SIGNED_NUMBER -> NUMBER
+""",
+    lexer="dynamic",
+)


-@dataclass
-class Alternate(Expression):
-    prompts: list[Expression]
-    step: float = 0.1
-
-    @override
-    def eval(self, step: float, tags: list[str]) -> EvalResult:
-        SCALE = 10_000
-        step = max(step, self.step)
-        position = (step * SCALE) / (self.step * SCALE)
-        idx = (ceil(position) - 1) % len(self.prompts)
-
-        r = self.prompts[max(0, idx)].eval(step, tags)
-        r = merge_until(r, max(self.step, ceil(position) * self.step))
-        return r
-
-    @override
-    def required_steps(self, max_steps: float):
-        r = set()
-        for x in self.prompts:
-            r.update(x.required_steps(max_steps))
-        r.update(set(x / 100 for x in range(0, int(max_steps * 100), int(self.step * 100))))
-        return r
+def flatten(x):
+    if type(x) in [str, tuple, int, type(None)] or isinstance(x, dict) and "type" in x:
+        yield x
+    else:
+        for g in x:
+            yield from flatten(g)


-@dataclass
-class Sequence(Expression):
-    prompts: list[tuple[Expression, float]]
+def clamp(a, b, c):
+    """clamp b between a and c"""
+    return min(max(a, b), c)

-    @override
-    def eval(self, step: float, tags: list[str]) -> EvalResult:
-        item = Text("")
-        found_step = FOREVER
-        for prompt, switch_step in self.prompts:
-            if step <= switch_step:
-                found_step = switch_step
-                item = prompt
+
+def get_steps(tree, num_steps):
+    res = [num_steps or 100]
+
+    def tostep(s):
+        steps = num_steps or 100
+        if "." in str(s) or not num_steps:
+            w = float(s)
+            value = w * steps
+        else:
+            w = int(s)
+            value = w
+
+        if w > 1 and not num_steps:
+            log.warning(
+                "You haven't configured the number of steps for Prompt Control to use, %s will be clipped to 1.0", w
+            )
+            value = steps
+
+        return int(clamp(0, value, steps))
+
+    class CollectSteps(lark.Visitor):
+        def scheduled(self, tree):
+            i = tree.children[-1]
+            if i and i.type == "TAG":
+                return
+            for i in [-1, -2]:
+                if tree.children[i] is not None:
+                    tree.children[i] = tostep(tree.children[i])
+                    res.append(tree.children[i])
+
+        def interp_steps(self, tree):
+            tree.children[-1] = tostep(tree.children[-1] or 0.1)
+            for i, _ in enumerate(tree.children[:-1]):
+                tree.children[i] = tostep(tree.children[i])
+
+            res.extend(tree.children[:-1])
+
+        def sequence(self, tree):
+            steps = tree.children[1::2]
+            for i, _ in enumerate(steps):
+                w = tostep(tree.children[i * 2 + 1])
+                tree.children[i * 2 + 1] = w
+                res.append(w)
+
+        def alternate(self, tree):
+            step_size = tostep(round(float(tree.children[-1] or 0.1), 2))
+            tree.children[-1] = step_size
+            res.extend([x for x in range(step_size, num_steps or 100, step_size)])
+
+    CollectSteps().visit(tree)
+
+    return sorted(set(res))
+
+
+def at_step(step, filters, tree):
+    class AtStep(lark.Transformer):
+        def scheduled(self, args):
+            before = None
+            during = None
+            after = None
+            when_end = None
+            pl, when, *rest = args
+            if rest:
+                when_end = rest[0]
+
+            pl = list(pl)
+            if len(pl) == 1:
+                (during,) = pl  # [after:0.5] == [::after:0.5,0.5]
+                if when_end is None:
+                    when_end = when
+                    after = during
+            elif len(pl) == 2:
+                during, after = pl  # [during:after:0.5] = [before::after:0.5,0.5]
+                if when_end is None:
+                    when_end = when
+                    before = during
+            else:
+                before, during, after = pl  # [before:during:after:0.5,0.8]
+
+            if isinstance(when, str):
+                return before or "" if when not in filters else after or ""
+
+            if when_end is None:
+                when_end = 1000_000
+
+            if step <= when:
+                return before or ""
+            if when < step <= when_end:
+                return during or ""
+            else:
+                return after or ""
+
+        def sequence(self, args):
+            previous_step = 0.0
+            prompts = args[::2]
+            steps = args[1::2]
+            for s, p in zip(steps, prompts, strict=False):
+                if s >= step and step >= previous_step:
+                    previous_step = step
+                    return p or ""
+                else:
+                    previous_step = s
+            return ""
+
+        def alternate(self, args):
+            step_size = args[-1]
+            idx = ceil(step / step_size)
+            return args[(idx - 1) % (len(args) - 1)] or ""
+
+        def start(self, args):
+            prompt = []
+            loraspecs = {}
+            args = flatten(args)
+            for a in args:
+                if isinstance(a, str):
+                    prompt.append(a)
+                elif isinstance(a, tuple):
+                    # sum identical specs together
+                    n = a[0]
+                    # if clip weight is not provided, use unet weight
+                    w, w_clip = a[1][0], a[1][1 % len(a[1])]
+                    e = loraspecs.get(n, {})
+                    loraspecs[n] = {
+                        "weight": round(e.get("weight", 0.0) + w, 2),
+                        "weight_clip": round(e.get("weight_clip", 0.0) + w_clip, 2),
+                    }
+                    lbw = a[2]
+                    if lbw:
+                        loraspecs[n]["lbw"] = lbw
+                    if loraspecs[n]["weight"] == 0 and loraspecs[n]["weight_clip"] == 0 and not lbw:
+                        del loraspecs[n]
+                else:
+                    pass
+            p = "".join(prompt)
+            return {"prompt": p, "loras": loraspecs}
+
+        def PLAIN(self, args):
+            return restore_escaped(args)
+
+        def FILENAME(self, value):
+            return str(value)
+
+        def embedding(self, args):
+            return "embedding:" + str(args[0])
+
+        def lora_weights(self, args):
+            return [float(str(a)) for a in args]
+
+        def lora_block_weights(self, args):
+            vals = args[0].split(";")
+            r = {}
+            for v in vals:
+                x = v.split("=", 2)
+                if len(x) != 2:
+                    continue
+                k, v = x[0].strip().upper(), x[1].strip()
+                r[k] = v
+            return r
+
+        def loraspec(self, args):
+            name = args[0]
+            params = args[1]
+            lbw = args[2]
+
+            return name, params, lbw
+
+        def __default__(self, data, children, meta):
+            return children
+
+    return AtStep().transform(tree)
+
+
+class PromptSchedule:
+    # 0 num_steps means unconfigured
+    def __init__(self, prompt, filters="", start=0.0, end=1.0, num_steps=0):
+        self.filters = filters
+        self.start = start
+        self.end = end
+        self.num_steps = num_steps
+        # placeholder is restored on parse
+        self.prompt = remove_comments(escape_specials(prompt.strip()))
+        self.defaults = {}
+        self.loaded_loras = {}
+
+        self.parsed_prompt = self._parse(num_steps)
+
+    def __iter__(self):
+        # Filter out zero, it's only useful for interpolation
+        return (x for x in self.parsed_prompt if x[0] != 0)
+
+    def _parse(self, num_steps):
+        filters = [x.strip() for x in self.filters.upper().split(",")]
+        try:
+            parsed = []
+            tree = prompt_parser.parse(self.prompt)
+            steps = get_steps(tree, num_steps=num_steps)
+
+            def f(x):
+                return round(x / (num_steps or 100), 2)
+
+            for t in steps:
+                p = at_step(t, filters, tree)
+                parsed.append([f(t), p])
+
+        except lark.exceptions.LarkError as e:
+            log.error("Prompt editing parse error: %s", e)
+            parsed = [[1.0, {"prompt": self.prompt, "loras": {}}]]
+            raise
+
+        # Tag filtering may return redundant prompts, so filter them out here
+        res = []
+        prev_end = -1
+
+        for end_at, p in parsed:
+            if end_at < self.start:
+                continue
+            elif end_at <= self.end:
+                res.append([end_at, p])
+                prev_end = end_at
+            elif end_at > self.end and prev_end < self.end:
+                res.append([end_at, p])
                break

-        return merge_until(item.eval(step, tags), found_step)
+        # Always use the last prompt if everything was filtered
+        if len(res) == 0:
+            res = [[1.0, parsed[-1][1]]]

-    @override
-    def required_steps(self, max_steps: float):
-        return set(step for _, step in self.prompts if step <= max_steps)
+        final = [res[0]]

-
-@dataclass
-class Schedule(Expression):
-    before: Prompt
-    during: Prompt
-    after: Prompt
-    start: float
-    end: float
-    tag: str | None
-
-    def tag_matches(self, tags: list[str]):
-        return self.tag in tags
-
-    @override
-    def eval(self, step: float, tags: list[str]) -> EvalResult:
-        if self.tag is not None and not self.tag_matches(tags):
-            return self.before.eval(step, tags)
-        if self.tag_matches(tags):
-            return self.during.eval(step, tags)
-
-        if step <= self.start:
-            return merge_until(self.before.eval(step, tags), self.start)
-        if self.start < step <= self.end:
-            return merge_until(self.during.eval(step, tags), self.end)
-        if step > self.end:
-            return self.after.eval(step, tags)
-        raise AssertionError("How are you here?")
-
-    @override
-    def required_steps(self, max_steps: float):
-        r = set()
-        if self.start < max_steps:
-            r.add(self.start)
-        if self.end < max_steps:
-            r.add(self.end)
-        r.update(self.before.required_steps(max_steps))
-        r.update(self.during.required_steps(max_steps))
-        r.update(self.after.required_steps(max_steps))
-        return r
-
-
-@dataclass
-class Prompt(Expression):
-    data: list[Expression]
-
-    @override
-    def eval(self, step: float, tags: list[str]) -> EvalResult:
-        evals = [x.eval(step, tags) for x in self.data]
-        text = "".join(x[1] for x in evals)
-        untils = [x[0] for x in evals]
-        loras = []
-        for x in evals:
-            loras.extend(x[2])
-        until = FOREVER if not untils else min(untils)
-        return until, text, loras
-
-    @override
-    def required_steps(self, max_steps):
-        r = set()
-        for x in self.data:
-            r.update(x.required_steps(max_steps))
-        return r
-
-
-@dataclass
-class LoRA(Expression):
-    filename: str
-    w_model: float = 1.0
-    w_te: float = 1.0
-
-    def eval(self, step: float, tags: list[str]) -> EvalResult:
-        return FOREVER, "", [self]
-
-
-def find_weight_at(weights: list[tuple[float, float]], step: float, until: float):
-    res_w = 0
-    for this, next in zip(weights, it.chain(weights[1:], [(0, FOREVER)]), strict=False):
-        w, start = this
-        _, next_start = next
-        if start > step or next_start < step:
-            until = min(until, start)
-            continue
-        res_w = w
-    return until, res_w
-
-
-@dataclass
-class LoRACTL(Expression):
-    filename: str
-    w_model: list[tuple[float, float]]
-    w_te: list[tuple[float, float]]
-
-    def eval(self, step: float, tags: list[str]) -> EvalResult:
-        until, w1 = find_weight_at(self.w_model, step, FOREVER)
-        until, w2 = find_weight_at(self.w_te, step, until)
-        lora = []
-        if w1 != 0 or w1 != 0:
-            lora = [LoRA(self.filename, w1, w2)]
-
-        return until, "", lora
-
-    def required_steps(self, max_steps):
-        r = set(x[1] for x in self.w_model)
-        r.update(set(x[1] for x in self.w_te))
-        return r
-
-
-def combine_arglist(prompts, start_end) -> Schedule:
-    a, b, c = prompts
-    start_or_tag, end = start_end
-    empty = Prompt([])
-    start = start_or_tag
-    # Handle [a:b:TAG]
-    if isinstance(start_or_tag, str):
-        if b is None:
-            before = empty
-            during = a  # [a:TAG] produces a when tag is active
-        else:
-            before, during = a, b  # [a:b:TAG] changes from a to b when tag is active
-        return Schedule(before, during, empty, start=0.0, end=FOREVER, tag=start_or_tag)
-    during = before = after = empty
-    if end is not None:
-        if b is None:  # [a:0,0.5] == [:a:0,0.5]
-            during = a
-            before = after = empty
-        elif c is None:  # [a:b:0,0.5]
-            before = empty
-            during = a
-            after = b
-        else:
-            before, during, after = a, b, c
-    else:
-        end = FOREVER
-        if b is None:  # [a:0.5] == [::a:0.5,0.5]
-            before = empty
-            during = a
-            after = a
-        else:
-            before = a
-            during = b
-            after = b
-            # c always gets ignored
-    start = float(start)  # for typechecking
-    return Schedule(before, during, after, start, end, tag=None)
-
-
-def token(s: str):
-    return string(s).map(Text)
-
-
-def combine_prompt(*prompts):
-    p = prompts
-    if len(p) == 1:
-        p = p[0]
-    if isinstance(p, Prompt):
-        p = p.data[0] if len(p.data) == 1 else combine_prompt(*p.data)
-    if isinstance(p, Expression):
-        return p
-    p = [combine_prompt(x) for x in p]
-    return Prompt(p)
-
-
-@dataclass
-class PromptSchedule:
-    parse_tree: Expression
-    filters: list[str]
-    start: float
-    end: float
-    num_steps: int
-
-    def at_step(self, step: float) -> tuple[float, dict[str, Any]]:
-        max_step = self.num_steps or 1.0
-        if max_step > 1 and step < 1:
-            step = step * max_step
-        until, p, lora_list = self.parse_tree.eval(step, self.filters)
-        loras = {}
-        for lora in lora_list:
-            d = loras.get(lora.filename, {})
-            d["weight"] = d.get("weight", 0) + lora.w_model
-            d["weight_clip"] = d.get("weight_clip", 0) + lora.w_te
-            loras[lora.filename] = d
-        if max_step > 0 and until > 1:
-            # TODO: better logic for this?
-            until = min(until / max_step, 1.0)
-        return (min(max_step, round(until, 2)), {"prompt": p, "loras": loras})
-
-    def with_filters(self, filters: str | None = None, start: float | None = None, end: float | None = None):
-        return PromptSchedule(
-            self.parse_tree,
-            self.filters if filters is None else parse_filters(filters),
-            self.start if start is None else start,
-            self.end if end is None else end,
-            self.num_steps,
-        )
+        # Clean up duplicates
+        for p in res[1:]:
+            if p[1] != final[-1][1]:
+                final.append(p)
+            else:
+                final[-1][0] = p[0]
+        return final

    def clone(self):
        return self.with_filters()

-    def __iter__(self):
-        return (x for x in self.parsed_prompt if x[0] != 0)
+    def with_filters(self, filters=None, start=None, end=None, defaults=None):
+        def ifspecified(x, defval):
+            return x if x is not None else defval

-    @property
-    def parsed_prompt(self):
-        max_step = self.num_steps or 1.0
-        required_steps = self.parse_tree.required_steps(max_step).union({max_step})
+        p = PromptSchedule(
+            self.prompt,
+            filters=ifspecified(filters, self.filters),
+            start=ifspecified(start, self.start),
+            end=ifspecified(end, self.end),
+            num_steps=self.num_steps,
+        )
+        return p

-        prompts = list(sorted((self.at_step(step) for step in required_steps), key=lambda x: x[0]))
-        res = []
-        prev_end = -1
-        for end_at, p in prompts:
-            if end_at < self.start:
-                continue
-            elif end_at < self.end and prev_end < end_at:
-                res.append([end_at, p])
-                prev_end = end_at
-            elif end_at >= self.end and prev_end < self.end:
-                res.append([end_at, p])
-                break
+    def at_step(self, step, total_steps=1):
+        _, x = self.at_step_idx(step, total_steps)
+        return x

-        if len(res) == 0:
-            res = [[1.0], prompts[-1][1]]
-
-        return res
+    def at_step_idx(self, step, total_steps=1):
+        for i, x in enumerate(self.parsed_prompt):
+            if x[0] * total_steps >= step:
+                return i, x
+        return len(self.parsed_prompt) - 1, self.parsed_prompt[-1]


-def lora_weights(p):
-    @generate
-    def parser():
-        w_model = yield col >> p
-        w_te = yield (col >> p).optional(w_model)
-        return [w_model, w_te]
-
-    return parser.desc("lora_weights")
-
-
-prompt = forward_declaration()
-empty = Text("")
-comma = token(",")
-col = token(":")
-lsq = token("[")
-rsq = token("]")
-lpar = token("(")
-rpar = token(")")
-tag = regex(r"[A-Z_]+")
-non_special = regex(r"[^:\[\]()|\\<>#]+").map(Text)
-filename = regex(r"[^:<>]+")
-
-comment = string("#") >> any_char.until(eof | char_from("\n")) >> success(empty)
-escape = (string("\\") >> char_from("\\[]:#") | string(r"\(") | string(r"\)")).map(Text)
-emphasis = seq(lpar, (prompt | col).at_least(0), rpar)
-sign = string("+") | string("-")
-number = (
-    (sign.optional("") + (digit.many() + string(".") * 1 + digit.many() | digit.at_least(1)).concat())
-    .concat()
-    .map(float)
-)
-
-opt_prompt = prompt.optional(empty)
-step_range = seq(number | tag, (comma >> number).optional())
-arglist = seq((opt_prompt << col).optional() * 3, step_range)
-schedule = lsq >> arglist.combine(combine_arglist) << rsq
-
-alternate = (lsq >> seq(prompt.sep_by(string("|"), min=1), (col >> number).optional(0.1)) << rsq).combine(Alternate)
-sequence = (lsq >> string("SEQ") >> seq(col >> opt_prompt << col, number).at_least(1) << rsq).map(Sequence)
-bracketed = seq(lsq, prompt.at_least(0), rsq) | sequence | schedule | alternate
-lora = (string("<lora:") >> filename * 1 + lora_weights(number) << string(">")).combine(LoRA)
-ctlweight = seq(number, (string("@") >> number).optional(0)).sep_by(comma, min=1)
-loractl = (string("<loractl:") >> filename * 1 + lora_weights(ctlweight) << string(">")).combine(LoRACTL)
-emb = (string("<emb:") >> filename << string(">")).map(lambda f: Text(f"embedding:{f}"))
-
-expr = (
-    escape
-    | comment
-    | non_special
-    | bracketed
-    | emphasis.combine(combine_prompt)
-    | lora
-    | loractl
-    | emb
-    | char_from("<>").map(Text)
-)
-prompt_ = expr.at_least(1).combine(combine_prompt)
-prompt.become(prompt_)
-# Treat any character that isn't valid prompt syntax as just text
-all = (prompt | any_char.map(Text)).at_least(0).combine(combine_prompt)
-
-
-def parse_filters(filters: str):
-    return [x.strip().upper() for x in filters.split(",") if x.strip()]
-
-
-def parse(text):
-    return combine_prompt(all.parse(text))
-
-
-def parse_prompt_schedules(text, filters="", start=0, end=1.0, num_steps=0):
-    return PromptSchedule(parse(expand_macros(text.strip())), parse_filters(filters), start, end, num_steps)
+@lru_cache
+def parse_prompt_schedules(prompt, **kwargs):
+    prompt = expand_macros(prompt)
+    return PromptSchedule(prompt, **kwargs)
@@ -0,0 +1,384 @@
+from __future__ import annotations
+
+import itertools as it
+from dataclasses import dataclass
+from math import ceil
+from typing import Any, TypeAlias
+
+from typing_extensions import override
+
+from .macros import expand_macros
+from .parsy import any_char, char_from, digit, eof, forward_declaration, generate, regex, seq, string, success
+
+FOREVER = float("inf")
+
+EvalResult: TypeAlias = tuple[float, str, list["LoRA"]]
+
+
+def merge_until(i: EvalResult, minimum: float):
+    until, p, loras = i
+    until = min(until, minimum)
+    return until, p, loras
+
+
+def batched(iterable, n, *, strict=False):
+    # batched('ABCDEFG', 2) → AB CD EF G
+    if n < 1:
+        raise ValueError("n must be at least one")
+    iterator = iter(iterable)
+    while batch := tuple(it.islice(iterator, n)):
+        if strict and len(batch) != n:
+            raise ValueError("batched(): incomplete batch")
+        yield batch
+
+
+EvalResult: TypeAlias = tuple[float, str, list["LoRA"]]
+
+
+class Expression:
+    def eval(self, step: float, tags: list[str]) -> EvalResult:
+        return (FOREVER, "", [])
+
+    def required_steps(self, max_steps: float) -> set[float]:
+        return set()
+
+
+@dataclass
+class Text(Expression):
+    string: str
+
+    @override
+    def eval(self, step: float, tags: list[str]) -> EvalResult:
+        assert isinstance(self.string, str)
+        return FOREVER, self.string, []
+
+
+@dataclass
+class Alternate(Expression):
+    prompts: list[Expression]
+    step: float = 0.1
+
+    @override
+    def eval(self, step: float, tags: list[str]) -> EvalResult:
+        SCALE = 10_000
+        step = max(step, self.step)
+        position = (step * SCALE) / (self.step * SCALE)
+        idx = (ceil(position) - 1) % len(self.prompts)
+
+        r = self.prompts[max(0, idx)].eval(step, tags)
+        r = merge_until(r, max(self.step, ceil(position) * self.step))
+        return r
+
+    @override
+    def required_steps(self, max_steps: float):
+        r = set()
+        for x in self.prompts:
+            r.update(x.required_steps(max_steps))
+        r.update(set(x / 100 for x in range(0, int(max_steps * 100), int(self.step * 100))))
+        return r
+
+
+@dataclass
+class Sequence(Expression):
+    prompts: list[tuple[Expression, float]]
+
+    @override
+    def eval(self, step: float, tags: list[str]) -> EvalResult:
+        item = Text("")
+        found_step = FOREVER
+        for prompt, switch_step in self.prompts:
+            if step <= switch_step:
+                found_step = switch_step
+                item = prompt
+                break
+
+        return merge_until(item.eval(step, tags), found_step)
+
+    @override
+    def required_steps(self, max_steps: float):
+        return set(step for _, step in self.prompts if step <= max_steps)
+
+
+@dataclass
+class Schedule(Expression):
+    before: Prompt
+    during: Prompt
+    after: Prompt
+    start: float
+    end: float
+    tag: str | None
+
+    def tag_matches(self, tags: list[str]):
+        return self.tag in tags
+
+    @override
+    def eval(self, step: float, tags: list[str]) -> EvalResult:
+        if self.tag is not None and not self.tag_matches(tags):
+            return self.before.eval(step, tags)
+        if self.tag_matches(tags):
+            return self.during.eval(step, tags)
+
+        if step <= self.start:
+            return merge_until(self.before.eval(step, tags), self.start)
+        if self.start < step <= self.end:
+            return merge_until(self.during.eval(step, tags), self.end)
+        if step > self.end:
+            return self.after.eval(step, tags)
+        raise AssertionError("How are you here?")
+
+    @override
+    def required_steps(self, max_steps: float):
+        r = set()
+        if self.tag is not None:
+            return r
+        if self.start < max_steps:
+            r.add(self.start)
+        if self.end < max_steps:
+            r.add(self.end)
+        r.update(self.before.required_steps(max_steps))
+        r.update(self.during.required_steps(max_steps))
+        r.update(self.after.required_steps(max_steps))
+        return r
+
+
+@dataclass
+class Prompt(Expression):
+    data: list[Expression]
+
+    @override
+    def eval(self, step: float, tags: list[str]) -> EvalResult:
+        evals = [x.eval(step, tags) for x in self.data]
+        text = "".join(x[1] for x in evals)
+        untils = [x[0] for x in evals]
+        loras = []
+        for x in evals:
+            loras.extend(x[2])
+        until = FOREVER if not untils else min(untils)
+        return until, text, loras
+
+    @override
+    def required_steps(self, max_steps):
+        r = set()
+        for x in self.data:
+            r.update(x.required_steps(max_steps))
+        return r
+
+
+@dataclass
+class LoRA(Expression):
+    filename: str
+    w_model: float = 1.0
+    w_te: float = 1.0
+
+    def eval(self, step: float, tags: list[str]) -> EvalResult:
+        return FOREVER, "", [self]
+
+
+def find_weight_at(weights: list[tuple[float, float]], step: float, until: float):
+    res_w = 0
+    for this, next in zip(weights, it.chain(weights[1:], [(0, FOREVER)]), strict=False):
+        w, start = this
+        _, next_start = next
+        if start > step or next_start < step:
+            until = min(until, start)
+            continue
+        res_w = w
+    return until, res_w
+
+
+@dataclass
+class LoRACTL(Expression):
+    filename: str
+    w_model: list[tuple[float, float]]
+    w_te: list[tuple[float, float]]
+
+    def eval(self, step: float, tags: list[str]) -> EvalResult:
+        until, w1 = find_weight_at(self.w_model, step, FOREVER)
+        until, w2 = find_weight_at(self.w_te, step, until)
+        lora = []
+        if w1 != 0 or w1 != 0:
+            lora = [LoRA(self.filename, w1, w2)]
+
+        return until, "", lora
+
+    def required_steps(self, max_steps):
+        r = set(x[1] for x in self.w_model)
+        r.update(set(x[1] for x in self.w_te))
+        return r
+
+
+def combine_arglist(prompts, start_end) -> Schedule:
+    a, b, c = prompts
+    start_or_tag, end = start_end
+    empty = Prompt([])
+    start = start_or_tag
+    # Handle [a:b:TAG]
+    if isinstance(start_or_tag, str):
+        if b is None:
+            before = empty
+            during = a  # [a:TAG] produces a when tag is active
+        else:
+            before, during = a, b  # [a:b:TAG] changes from a to b when tag is active
+        return Schedule(before, during, empty, start=0.0, end=FOREVER, tag=start_or_tag)
+    during = before = after = empty
+    if end is not None:
+        if b is None:  # [a:0,0.5] == [:a:0,0.5]
+            during = a
+            before = after = empty
+        elif c is None:  # [a:b:0,0.5]
+            before = empty
+            during = a
+            after = b
+        else:
+            before, during, after = a, b, c
+    else:
+        end = FOREVER
+        if b is None:  # [a:0.5] == [::a:0.5,0.5]
+            before = empty
+            during = a
+            after = a
+        else:
+            before = a
+            during = b
+            after = b
+            # c always gets ignored
+    start = float(start)  # for typechecking
+    return Schedule(before, during, after, start, end, tag=None)
+
+
+def token(s: str):
+    return string(s).map(Text)
+
+
+def combine_prompt(*prompts):
+    p = prompts
+    if len(p) == 1:
+        p = p[0]
+    if isinstance(p, Prompt):
+        p = p.data[0] if len(p.data) == 1 else combine_prompt(*p.data)
+    if isinstance(p, Expression):
+        return p
+    p = [combine_prompt(x) for x in p]
+    return Prompt(p)
+
+
+@dataclass
+class PromptSchedule:
+    parse_tree: Expression
+    filters: list[str]
+    start: float
+    end: float
+    num_steps: int
+
+    def at_step(self, step: float) -> tuple[float, dict[str, Any]]:
+        max_step = self.num_steps or 1.0
+        if max_step > 1 and step < 1:
+            step = step * max_step
+        until, p, lora_list = self.parse_tree.eval(step, self.filters)
+        loras = {}
+        for lora in lora_list:
+            d = loras.get(lora.filename, {})
+            d["weight"] = d.get("weight", 0) + lora.w_model
+            d["weight_clip"] = d.get("weight_clip", 0) + lora.w_te
+            loras[lora.filename] = d
+        if max_step > 0 and until > 1:
+            # TODO: better logic for this?
+            until = min(until / max_step, 1.0)
+        return (min(max_step, round(until, 2)), {"prompt": p, "loras": loras})
+
+    def with_filters(self, filters: str | None = None, start: float | None = None, end: float | None = None):
+        return PromptSchedule(
+            self.parse_tree,
+            self.filters if filters is None else parse_filters(filters),
+            self.start if start is None else start,
+            self.end if end is None else end,
+            self.num_steps,
+        )
+
+    def clone(self):
+        return self.with_filters()
+
+    def __iter__(self):
+        return (x for x in self.parsed_prompt if x[0] != 0)
+
+    @property
+    def parsed_prompt(self):
+        max_step = self.num_steps or 1.0
+        required_steps = self.parse_tree.required_steps(max_step).union({max_step})
+
+        prompts = list(sorted((self.at_step(step) for step in required_steps), key=lambda x: x[0]))
+        res = []
+        prev_end = -1
+        for end_at, p in prompts:
+            if end_at < self.start:
+                continue
+            elif end_at < self.end and prev_end < end_at:
+                res.append([end_at, p])
+                prev_end = end_at
+            elif end_at >= self.end and prev_end < self.end:
+                res.append([end_at, p])
+                break
+
+        if len(res) == 0:
+            res = [[1.0], prompts[-1][1]]
+
+        return res
+
+
+def lora_weights(p):
+    @generate
+    def parser():
+        w_model = yield col >> p
+        w_te = yield (col >> p).optional(w_model)
+        return [w_model, w_te]
+
+    return parser.desc("lora_weights")
+
+
+prompt = forward_declaration()
+empty = Text("")
+comma = token(",")
+col = token(":")
+lsq = token("[")
+rsq = token("]")
+lpar = token("(")
+rpar = token(")")
+tag = regex(r"[A-Z_]+")
+non_special = regex(r"[^:\[\]()|\\<>#]+").map(Text)
+filename = regex(r"[^:<>]+")
+
+comment = string("#") >> any_char.until(eof | char_from("\n")) >> success(empty)
+escape = (string("\\") >> char_from("\\[]:#")).map(Text)
+emphasis = seq(lpar, (prompt | col).at_least(0), rpar)
+number = (digit.at_least(1) + string(".") * 1 + digit.many() | digit.at_least(1)).concat().map(float)
+
+opt_prompt = prompt.optional(empty)
+step_range = seq(number | tag, (comma >> number).optional())
+arglist = seq((opt_prompt << col).optional() * 3, step_range)
+schedule = lsq >> arglist.combine(combine_arglist) << rsq
+
+alternate = (lsq >> seq(prompt.sep_by(string("|"), min=1), (col >> number).optional(0.1)) << rsq).combine(Alternate)
+sequence = (lsq >> string("SEQ") >> seq(col >> opt_prompt << col, number).at_least(1) << rsq).map(Sequence)
+bracketed = seq(lsq, prompt.at_least(0), rsq) | sequence | schedule | alternate
+lora = (string("<lora:") >> filename * 1 + lora_weights(number) << string(">")).combine(LoRA)
+ctlweight = seq(number, (string("@") >> number).optional(0)).sep_by(comma, min=1)
+loractl = (string("<loractl:") >> filename * 1 + lora_weights(ctlweight) << string(">")).combine(LoRACTL)
+emb = (string("<emb:") >> filename << string(">")).map(lambda f: Text(f"embedding:{f}"))
+
+expr = escape | comment | non_special | bracketed | emphasis.combine(combine_prompt) | lora | loractl | emb
+prompt_ = expr.at_least(1).combine(combine_prompt)
+prompt.become(prompt_)
+# Treat any character that isn't valid prompt syntax as just text
+all = (prompt | any_char.map(Text)).at_least(0).combine(combine_prompt)
+
+
+def parse_filters(filters: str):
+    return [x.strip().upper() for x in filters.split(",") if x.strip()]
+
+
+def parse(text):
+    return combine_prompt(all.parse(text))
+
+
+def parse_prompt_schedules(text, filters="", start=0, end=1.0, num_steps=0):
+    return PromptSchedule(parse(expand_macros(text.strip())), parse_filters(filters), start, end, num_steps)
@@ -241,7 +241,7 @@ def encode_prompt_segment(
    can_break = {}
    for k in empty:
        tokenizer = getattr(clip.tokenizer, f"clip_{k}", getattr(clip.tokenizer, k, None))
-        can_break[k] = tokenizer and getattr(tokenizer, "pad_to_max_length", False)
+        can_break[k] = tokenizer and tokenizer.pad_to_max_length

    clip = hook_te(clip, empty.keys(), style, normalization, extra)

@@ -604,7 +604,7 @@ def encode_prompt(clip, text, start_pct, end_pct, defaults, masks):
        return f"MASK({args[0]})"

    for prompt in prompts:
-        prompt, noise_w, generator = get_noise(prompt)
+        text, noise_w, generator = get_noise(text)
        base_prompt, attn_couple_prompts = split_by_function(prompt, "COUPLE", defaults=None, require_args=False)

        prompts = [base_prompt] + [couple_mask(f.args) + chunk for (chunk, f) in attn_couple_prompts]
@@ -3,7 +3,7 @@ from __future__ import annotations
 import copy
 import logging
 import re
-from collections.abc import Callable, Iterator
+from collections.abc import Iterator
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, TypeAlias, TypeVar
@@ -20,6 +20,7 @@ class FunctionSpec:
    name: str
    args: FunctionArgs
    position: int
+    placeholder: str | None


 # Allow testing
@@ -34,14 +35,6 @@ except ImportError:
 log = logging.getLogger("comfyui-prompt-control")


-def flatten(x):
-    if type(x) in [str, tuple, int, type(None)] or isinstance(x, dict) and "type" in x:
-        yield x
-    else:
-        for g in x:
-            yield from flatten(g)
-
-
 def call_node(cls, *args, **kwargs):
    if hasattr(cls, "execute"):
        # v3 node
@@ -141,10 +134,6 @@ def find_function_spans(
        if text[at_paren:after_first_paren] == "(":
            end = find_closing_paren(text, after_first_paren)
            if end < 0:
-                # Unclosed paren: skip past this match so the loop terminates
-                idx += match.end()
-                text = text[match.end() :]
-                match = rex.search(text)
                continue
            args = parse_strings(text[after_first_paren:end], defaults)
            end += 1
@@ -158,11 +147,7 @@ def find_function_spans(


 def get_function(
-    text: str,
-    func: str,
-    defaults: list[str] | None,
-    processor: Callable[..., str] | None = None,
-    require_args: bool = True,
+    text: str, func: str, defaults: list[str] | None, placeholder: str = "", require_args: bool = True
 ) -> tuple[str, list[FunctionSpec]]:
    spans = [x.span() for x in re.finditer(r'".+?"', text)]
    instances = []
@@ -171,13 +156,14 @@ def get_function(
    current = 0
    skipped = 0
    for start, end, funcname, args in find_function_spans(text, func, require_args, defaults):
+        ph = None
        if spans_include(spans, start, end):
            continue
-        instances.append(FunctionSpec(funcname, args, start - skipped))
+        if placeholder:
+            ph = f"\0{placeholder}{count}\0"
+        instances.append(FunctionSpec(funcname, args, start - skipped, ph))
        skipped += end - start
-        chunks.append(text[current:start])
-        if processor:
-            chunks.append(processor(*args))
+        chunks.append(text[current:start] + (ph or ""))
        current = end
        count += 1
    chunks.append(text[current:])
@@ -279,10 +265,6 @@ def lora_name_to_file(name: str) -> str | None:
    search = [f for f in filenames if all(p in f for p in parts)]
    if len(search) == 1:
        return search[0]
-    elif len(search) > 1:
-        if len(search) > 4:
-            search[4] = "..."
-        log.warning("Ignored LoRA search 's%'; matched more than one file: %s", name, ", ".join(search[:5]))

    return None

@@ -1,8 +1,10 @@
 [project]
 name = "comfyui-prompt-control"
-description = "Nodes for prompt editing and LoRA scheduling, advanced regional prompting (including attention masking) and advanced prompt encoding, all controlled through your text prompt. Feature keywords: comfyui-prompt-control, schedule, macros, attention couple, loractl, A1111"
-version = "3.0.0-beta.10"
+description = "Provides nodes for prompt editing and LoRA scheduling, advanced regional prompting (including attention masking) and more, all controlled through your text prompt"
+version = "2.1.2"
 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"]

 requires-python = ">= 3.10"

@@ -0,0 +1,2 @@
+# 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
+lark >= 1.1.9
@@ -73,16 +73,6 @@ def tensors_equal(t1, t2):
    npt.assert_equal(t1.detach().numpy(), t2.detach().numpy())


-def cond_neq(c1, c2, key=None, key_assert=None):
-    ok = False
-    try:
-        cond_equal(c1, c2, key=key, key_assert=key_assert)
-    except AssertionError:
-        ok = True
-    if not ok:
-        raise ValueError("Tensors should not be equal")
-
-
 def cond_equal(c1, c2, key=None, key_assert=None):
    assert len(c1) == len(c2)
    for i in range(len(c1)):
@@ -251,15 +241,3 @@ class TestPCTextEncode:
            (c2,) = run(pc_text_encode, clip, "test COUPLE MASK(0 0.2, 0.5) prompt1")
            cond_equal(c, c2)
            cond_equal(c, c2, "hooks", compare_hookgroup_mask)
-
-    def test_noise_weight0(self, text_encoder_clips, pc_text_encode, node_class_objs):
-        for _k, clip in text_encoder_clips:
-            (c1,) = run(pc_text_encode, clip, "test")
-            (c2,) = run(pc_text_encode, clip, "test NOISE(0, 0)")
-            cond_equal(c1, c2)
-
-    def test_noise(self, text_encoder_clips, pc_text_encode, node_class_objs):
-        for _k, clip in text_encoder_clips:
-            (c1,) = run(pc_text_encode, clip, "test")
-            (c2,) = run(pc_text_encode, clip, "test NOISE(1, 0)")
-            cond_neq(c1, c2)
@@ -460,102 +460,6 @@ def test_textencode_lora_with_schedule():
    }


-def test_textencode_custom():
-    r = te("NODE(CLIPTextEncode)simple [test:0.1,0.5] $p SEG(p) prompt")
-    assert r == {
-        "result": (["UID.0.0.8", 0],),
-        "expand": {
-            "UID.0.0.1": {
-                "class_type": "CLIPTextEncode",
-                "inputs": {"clip": [0, 0], "text": "simple  prompt"},
-            },
-            "UID.0.0.2": {
-                "class_type": "ConditioningSetTimestepRange",
-                "inputs": {"conditioning": ["UID.0.0.1", 0], "start": 0.0, "end": 0.1},
-            },
-            "UID.0.0.3": {
-                "class_type": "CLIPTextEncode",
-                "inputs": {"clip": [0, 0], "text": "simple test prompt"},
-            },
-            "UID.0.0.4": {
-                "class_type": "ConditioningSetTimestepRange",
-                "inputs": {"conditioning": ["UID.0.0.3", 0], "start": 0.1, "end": 0.5},
-            },
-            "UID.0.0.5": {
-                "class_type": "CLIPTextEncode",
-                "inputs": {"clip": [0, 0], "text": "simple  prompt"},
-            },
-            "UID.0.0.6": {
-                "class_type": "ConditioningSetTimestepRange",
-                "inputs": {"conditioning": ["UID.0.0.5", 0], "start": 0.5, "end": 1.0},
-            },
-            "UID.0.0.7": {
-                "class_type": "ConditioningCombine",
-                "inputs": {"conditioning_1": ["UID.0.0.2", 0], "conditioning_2": ["UID.0.0.4", 0]},
-            },
-            "UID.0.0.8": {
-                "class_type": "ConditioningCombine",
-                "inputs": {"conditioning_1": ["UID.0.0.7", 0], "conditioning_2": ["UID.0.0.6", 0]},
-            },
-        },
-    }
-
-
-def test_textencode_custom_extra():
-    r = te(
-        'NODE(CustomTextEncode, prompt, image ["1\:1", 0]; option "test"; float [10.0:__EMPTY__:0.5])simple [test:0.1,0.5] prompt'
-    )
-    assert r == {
-        "result": (["UID.0.0.8", 0],),
-        "expand": {
-            "UID.0.0.1": {
-                "class_type": "CustomTextEncode",
-                "inputs": {
-                    "clip": [0, 0],
-                    "prompt": "simple  prompt",
-                    "image": ["1:1", 0],
-                    "option": "test",
-                    "float": 10.0,
-                },
-            },
-            "UID.0.0.2": {
-                "class_type": "ConditioningSetTimestepRange",
-                "inputs": {"conditioning": ["UID.0.0.1", 0], "start": 0.0, "end": 0.1},
-            },
-            "UID.0.0.3": {
-                "class_type": "CustomTextEncode",
-                "inputs": {
-                    "clip": [0, 0],
-                    "prompt": "simple test prompt",
-                    "image": ["1:1", 0],
-                    "option": "test",
-                    "float": 10.0,
-                },
-            },
-            "UID.0.0.4": {
-                "class_type": "ConditioningSetTimestepRange",
-                "inputs": {"conditioning": ["UID.0.0.3", 0], "start": 0.1, "end": 0.5},
-            },
-            "UID.0.0.5": {
-                "class_type": "CustomTextEncode",
-                "inputs": {"clip": [0, 0], "prompt": "simple  prompt", "image": ["1:1", 0], "option": "test"},
-            },
-            "UID.0.0.6": {
-                "class_type": "ConditioningSetTimestepRange",
-                "inputs": {"conditioning": ["UID.0.0.5", 0], "start": 0.5, "end": 1.0},
-            },
-            "UID.0.0.7": {
-                "class_type": "ConditioningCombine",
-                "inputs": {"conditioning_1": ["UID.0.0.2", 0], "conditioning_2": ["UID.0.0.4", 0]},
-            },
-            "UID.0.0.8": {
-                "class_type": "ConditioningCombine",
-                "inputs": {"conditioning_1": ["UID.0.0.7", 0], "conditioning_2": ["UID.0.0.6", 0]},
-            },
-        },
-    }
-
-
 def test_loraloader_empty(monkeypatch, caplog):
    result = loraloader("prompt here <lora:nonexistent:1.0:0.5>")["expand"]
    result_adv = loraloader("prompt here <lora:nonexistent:1.0:0.5>", adv=True)["expand"]
@@ -678,9 +582,3 @@ def test_loraloader_adv_start():
 def test_loraloader_end_zero():
    result2 = loraloader("prompt [<lora:test:0.5>:0.5]", adv=True, end=0.5)["expand"]
    assert result2 == {}
-
-
-def test_loraloader_segs():
-    result = loraloader("prompt [<lora:test:0.5>:0.5]")["expand"]
-    result2 = loraloader("prompt [$lora:0.5]\nSEG(lora)<lora:test:0.5>\nSEG(lora2)<lora:ignored:1>")["expand"]
-    assert result == result2
@@ -1,77 +0,0 @@
-from textwrap import dedent
-
-import pytest
-
-from prompt_control.macros import expand_macros, expand_segs
-
-
-@pytest.mark.parametrize(
-    "text, result",
-    [
-        ("DEF(X(a;b)=$1 $2 $3 d)X(A) X(A;B;C)", "A b $3 d A B C d"),
-        (
-            "DEF(MACRO()=[empty:$1:$2])MACRO MACRO(;) MACRO(;0.5) MACRO(a;0.5)",
-            "[empty::$2] [empty::] [empty::0.5] [empty:a:0.5]",
-        ),
-        ("DEF(X=$1)DEF(Y()=$1)[X Y][X() Y()][X(1) Y(1)]", "[$1 ][ ][1 1]"),
-        (
-            "DEF(C_ANIMAL=cat)DEF(D_ANIMAL=dog)DEF(IT=It is a $1_ANIMAL $10_ANIMAL)IT(D) IT(C)",
-            "It is a dog $10_ANIMAL It is a cat $10_ANIMAL",
-        ),
-    ],
-)
-def test_basic_macro(text, result):
-    assert expand_macros(text) == result
-
-
-def test_macro_recursion():
-    with pytest.raises(ValueError) as c:
-        expand_macros("DEF(X=recurse Y) DEF(Y=recurse X) X")
-    assert "Unable to resolve DEFs" in str(c.value)
-
-
-def test_parsing_cornercase():
-    r = expand_macros("This should not get stuck DEF(")
-    assert r == "This should not get stuck DEF("
-
-
-@pytest.mark.parametrize(
-    "input, output",
-    [
-        (
-            """\
-      A red $b and
-      a blue $a
-      SEG(a)
-      cat
-      SEG(b)
-
-      dog
-      SEG(c)""",
-            "A red dog and\na blue cat",
-        ),
-        (
-            """\
-    $a and $b
-    SEG(a)
-    cat, $b
-    SEG(b)
-    dog, $c
-    SEG(c)
-    tiger
-    """,
-            "cat, dog, tiger and dog, tiger",
-        ),
-        (
-            """\
-    $a
-    SEG(a)
-    a $b
-    SEG(b)
-    b $a""",
-            "a b a b $a",
-        ),
-    ],
-)
-def test_segments(input, output):
-    assert expand_segs(dedent(input)) == output
@@ -1,6 +1,10 @@
+import os
+
 import pytest

-from prompt_control.parser import parse_prompt_schedules as parse
+from prompt_control.parser import expand_macros
+from prompt_control.parser import parse_prompt_schedules as old_parse  # noqa
+from prompt_control.parser_parsy import parse_prompt_schedules as new_parse  # noqa


 def lora_dict(*loras):
@@ -19,9 +23,14 @@ def assert_prompt(p, at, until, text, *loras):
    assert prompts_match(p.at_step(at), prompt(until, text, *loras))


-params = []
+parsers_to_test = os.environ.get("PC_PARSERS_TO_TEST", "new").split()

-params.append(parse)
+params = []
+if "old" in parsers_to_test:
+    params.append(old_parse)
+
+if "new" in parsers_to_test:
+    params.append(new_parse)


@pytest.fixture(scope="module", autouse=True, params=params)
@@ -102,9 +111,9 @@ def test_basic_ok(parse):

@pytest.mark.parametrize("step", [0, 0.5, 1])
 def test_lora(step, parse):
-    p = parse("This is a (lora:0.6) (prompt) with [no scheduling] features <lora:foo:0.5> <lora:bar:0.5:-1.0>")
+    p = parse("This is a (lora:0.6) (prompt) with [no scheduling] features <lora:foo:0.5> <lora:bar:0.5:1.0>")
    expected = prompt(
-        1.0, "This is a (lora:0.6) (prompt) with [no scheduling] features  ", ("foo", 0.5, 0.5), ("bar", 0.5, -1.0)
+        1.0, "This is a (lora:0.6) (prompt) with [no scheduling] features  ", ("foo", 0.5, 0.5), ("bar", 0.5, 1.0)
    )
    assert prompts_match(p.at_step(step), expected)

@@ -212,6 +221,23 @@ def test_def(parse):
    p2 = parse("[(test):(test:0.7):0.7] [(test):(test:0.5):0.5]")
    assert p.parsed_prompt == p2.parsed_prompt

+    p = expand_macros("DEF(X(a;b)=$1 $2 $3 d)X(A) X(A;B;C)")
+    assert p == "A b $3 d A B C d"
+
+    p = expand_macros("DEF(MACRO()=[empty:$1:$2])MACRO MACRO(;) MACRO(;0.5) MACRO(a;0.5)")
+    assert p == "[empty::$2] [empty::] [empty::0.5] [empty:a:0.5]"
+
+    p = expand_macros("DEF(X=$1)DEF(Y()=$1)[X Y][X() Y()][X(1) Y(1)]")
+    assert p == "[$1 ][ ][1 1]"
+
+    p = parse("DEF(test(1)=prompt $1)DEF(test2((a); (test))=[$1:$2:0.5])test test2")
+    p2 = parse("prompt 1 [(a):(prompt 1):0.5]")
+    assert p.parsed_prompt == p2.parsed_prompt
+
+    with pytest.raises(ValueError) as c:
+        expand_macros("DEF(X=recurse Y) DEF(Y=recurse X) X")
+    assert "Unable to resolve DEFs" in str(c.value)
+

@pytest.mark.parametrize(
    "text, cases",
@@ -223,7 +249,6 @@ def test_def(parse):
        ),
        (r"[a\:b\\:c:0.5]", [(0.0, 0.5, "a:b\\"), (0.55, 1, r"c")]),
        (r"[a:\#b:0.5]", [(0.0, 0.5, "a"), (0.55, 1, "#b")]),
-        (r"[a:b \(test\):0.2]", [(0, 0.2, r"a"), (0.25, 1, r"b \(test\)")]),
    ],
 )
 def test_escapes(text, cases, parse):
@@ -333,18 +358,6 @@ def test_cornercase_corrected(parse):
    assert p.parsed_prompt[1:] == p2.parsed_prompt


-def test_ltgt_in_schedule(parse):
-    p = parse("This should [<parse> correctly:be <Picture 1>:0.1]<lora:test:1>")
-    assert_prompt(p, 0.1, 0.1, "This should <parse> correctly", ("test", 1.0, 1.0))
-    assert_prompt(p, 0.15, 1.0, "This should be <Picture 1>", ("test", 1.0, 1.0))
-
-
-def test_floats(parse):
-    p = parse("[a:b:0.5] [c:d:e:0.2,0.7] <lora:test:-0.3>")
-    p2 = parse("[a:b:.5] [c:d:e:.2,.7] <lora:test:-.3>")
-    assert p.parsed_prompt == p2.parsed_prompt
-
-
 def test_alternating_lora(parse):
    p4 = parse("[cat|[dog:wolf<lora:canine:1>:0.5]:0.2]")
    for i, (text, *_loras) in enumerate(
@@ -361,9 +374,3 @@ def test_alternating_nested(parse):
    for i, x in enumerate(catdogtigers):
        step = round((i * 0.1) + 0.1, 2)
        assert_prompt(p3, step, step, x)
-
-
-def test_alternating_with_tags(parse):
-    p1 = parse("[[a|b]:HR]", filters="HR")
-    p2 = parse("[a|b]")
-    assert p1.parsed_prompt == p2.parsed_prompt
@@ -1,7 +0,0 @@
-from prompt_control import utils
-
-
-def test_smart_split():
-    assert utils.smarter_split(",", "foo,bar") == ["foo", "bar"]
-    assert utils.smarter_split(",", "(foo,bar),zonk") == ["(foo,bar)", "zonk"]
-    assert utils.smarter_split(",", r"\(foo,bar),zonk") == [r"\(foo", "bar)", "zonk"]
@@ -1,139 +0,0 @@
-{
-  "1": {
-    "inputs": {
-      "text": "positive prompt",
-      "clip": [
-        "4",
-        1
-      ]
-    },
-    "class_type": "PCLazyTextEncode",
-    "_meta": {
-      "title": "PC: Schedule prompt"
-    }
-  },
-  "2": {
-    "inputs": {
-      "ckpt_name": "$TEST_CHECKPOINT"
-    },
-    "class_type": "CheckpointLoaderSimple",
-    "_meta": {
-      "title": "Load Checkpoint"
-    }
-  },
-  "3": {
-    "inputs": {
-      "seed": 0,
-      "steps": 8,
-      "cfg": 3,
-      "sampler_name": "euler",
-      "scheduler": "simple",
-      "denoise": 1,
-      "model": [
-        "4",
-        0
-      ],
-      "positive": [
-        "9",
-        0
-      ],
-      "negative": [
-        "9",
-        1
-      ],
-      "latent_image": [
-        "5",
-        0
-      ]
-    },
-    "class_type": "KSampler",
-    "_meta": {
-      "title": "KSampler"
-    }
-  },
-  "4": {
-    "inputs": {
-      "text": "<lora:$TEST_LORA:1>",
-      "model": [
-        "2",
-        0
-      ],
-      "clip": [
-        "2",
-        1
-      ]
-    },
-    "class_type": "PCLazyLoraLoader",
-    "_meta": {
-      "title": "PC: Schedule LoRAs"
-    }
-  },
-  "5": {
-    "inputs": {
-      "width": 1024,
-      "height": 1024,
-      "batch_size": 1
-    },
-    "class_type": "EmptyLatentImage",
-    "_meta": {
-      "title": "Empty Latent Image"
-    }
-  },
-  "6": {
-    "inputs": {
-      "samples": [
-        "3",
-        0
-      ],
-      "vae": [
-        "2",
-        2
-      ]
-    },
-    "class_type": "VAEDecode",
-    "_meta": {
-      "title": "VAE Decode"
-    }
-  },
-  "7": {
-    "inputs": {
-      "images": [
-        "6",
-        0
-      ]
-    },
-    "class_type": "PreviewImage",
-    "_meta": {
-      "title": "Preview Image"
-    }
-  },
-  "8": {
-    "inputs": {
-      "text": "worst quality,",
-      "clip": [
-        "4",
-        1
-      ]
-    },
-    "class_type": "CLIPTextEncode",
-    "_meta": {
-      "title": "CLIP Text Encode (Prompt)"
-    }
-  },
-  "9": {
-    "inputs": {
-      "positive": [
-        "1",
-        0
-      ],
-      "negative": [
-        "8",
-        0
-      ]
-    },
-    "class_type": "PCAttentionCoupleBatchNegative",
-    "_meta": {
-      "title": "PC: Attention Couple (batch negative)"
-    }
-  }
-}
@@ -1,40 +0,0 @@
-import json
-import os
-import uuid
-from time import sleep
-
-import pytest
-import requests
-
-
-@pytest.fixture(scope="module", autouse=True)
-def workflow(request):
-    with open(str(request.path).replace(".py", ".json")) as f:
-        data = f.read()
-    data = data.replace("$TEST_CHECKPOINT", os.environ["PC_TEST_CHECKPOINT"])
-    data = data.replace("$TEST_LORA", os.environ["PC_TEST_LORA"])
-    return json.loads(data)
-
-
-def assert_prompt(url, p):
-    timeout = 60
-    r = requests.post(f"{url}/prompt", json={"prompt": p, "client_id": str(uuid.uuid4())}).json()
-    prompt_id = r["prompt_id"]
-    r = {"status": "pending"}
-    while r["status"] in ["pending", "in_progress"]:
-        sleep(1)
-        assert timeout > 0
-        timeout -= 1
-        r = requests.get(f"{url}/api/jobs/{prompt_id}").json()
-    assert r["status"] == "completed"
-
-
-@pytest.fixture
-def comfyui():
-    return os.environ.get("PC_TEST_COMFYUI", "http://localhost:8188")
-
-
-def test_workflow(workflow, comfyui):
-    prompt = "DEF(blue=green)a blue dog and a cat sitting [COUPLE(0 0.5, 0 1) red (cat,:1.3) COUPLE(0.5 1, 0 1) (blue:1.2) dog,:0.1]"
-    workflow["1"]["inputs"]["text"] = prompt
-    assert_prompt(comfyui, workflow)
Author	SHA1	Message	Date
asagi4	9857c662af	tests: COUPLE mask shortcut	2026-02-02 19:33:34 +02:00
asagi4	32f18ef338	Tests for step count	2026-02-02 19:26:13 +02:00
asagi4	d86d9f902c	test: Verify that basic paren escapes don't get stripped	2026-02-02 19:26:13 +02:00
asagi4	f600b1ef90	Fix type complaint	2026-02-02 19:26:13 +02:00
asagi4	72484d1034	test only new parser	2026-02-02 19:26:13 +02:00
asagi4	e7b444316f	Test multiple averages	2026-02-02 19:26:13 +02:00
asagi4	b9db1c7c4c	Make github tests work again	2026-02-02 19:26:13 +02:00
asagi4	3ee001a3fa	Disable method override complaint	2026-02-02 19:26:13 +02:00
asagi4	3fcd708dc6	Working importing for v3 nodes	2026-02-02 19:26:13 +02:00
asagi4	cb5ef8d97a	v3: nodes_lazy.py	2026-02-02 19:26:13 +02:00
asagi4	24fe09ba4e	v3: nodes_tools	2026-02-02 19:26:13 +02:00
asagi4	1c7149a811	v3: nodes_hooks.py	2026-02-02 19:26:13 +02:00
asagi4	b01bdc3d50	v3: nodes_base.py	2026-02-02 19:26:13 +02:00
asagi4	5ffd94d7d2	Initial v3 migration	2026-02-02 19:26:13 +02:00
asagi4	a433b246a7	handle setting steps	2026-02-02 19:26:13 +02:00
asagi4	bdd0c62665	parsy ruff fixes	2026-02-02 19:26:13 +02:00
asagi4	1c763f743c	parsy loractl	2026-02-02 19:23:55 +02:00
asagi4	c223603100	simplify LoRA weight parsing	2026-02-02 19:23:55 +02:00
asagi4	c9efa87291	Rewrite parser to use parsy	2026-02-02 19:23:50 +02:00
asagi4	e15d4d693c	Remove broken expand_graph.py	2026-02-02 19:19:21 +02:00
asagi4	26d5bfc961	Split cutoff parser to its own file	2026-02-02 19:19:21 +02:00
asagi4	cdb6c418eb	split macros out of parser	2026-02-02 19:19:21 +02:00
asagi4	dc08fa93b7	Fix NOISE doing nothing	2026-02-02 19:19:21 +02:00
asagi4	e29f3ca2d0	Fix github workflows	2026-02-02 19:19:21 +02:00
asagi4	ee186a73e5	Remove old tests	2026-02-02 19:19:17 +02:00
asagi4	7440b309fc	Fix LazyLoraLoader tests	2026-02-02 19:18:36 +02:00
asagi4	2ccc697311	Test for discovered corner case behaviour	2026-02-02 19:18:36 +02:00
asagi4	4559060535	Add a graph test for alternating	2026-02-02 19:18:36 +02:00
asagi4	60ceddbd73	Refactor tests for new parser	2026-02-02 19:18:36 +02:00
asagi4	6fd28cfca7	Use pytest tests	2026-02-02 19:18:36 +02:00
asagi4	11b1313ed4	Convert tests to pytest Not 100% sure these fully work yet	2026-02-02 19:18:36 +02:00
asagi4	9be7133e48	Ruff fixes etc.	2026-02-02 19:18:36 +02:00
asagi4	22e78f08eb	Typing fixes etc	2026-02-02 19:18:36 +02:00
asagi4	39befe2a34	Remove cache hack, it's broken anyway	2026-02-02 19:18:36 +02:00
asagi4	ea9017862f	Add ruff and ty	2026-02-02 19:18:36 +02:00
asagi4	6ae6cf0d65	Add pyright	2026-02-02 19:18:36 +02:00