R3: an accumulating batch saver repeats the A1111 parameters record once per batch item, identical except "Batch index: N". The bounded PNG reader now keeps the last copy (Pillow's rule, which #165 relied on) instead of calling the file unreadable. Any other difference, including a missing or second batch index, is still a conflict, and decoding the copies draws from the duplicate role budget. R4: parameters text that starts with "Negative prompt:" has an empty positive (#165), instead of showing the delimiter line as the positive. R5: the settings footer may be followed by sd-dynamic-prompts' one-line "Template:" and "Negative Template:" records, and may be "Steps: N" alone when there is no full footer. R10: commas inside A1111's quoted settings values no longer split them, so an "ADetailer prompt" with a comma keeps the other facts known. The facts' A1111 reader mirrors R4, R5 and R10; the default (ingestion) profile keeps the historical split. Also from the Task 2.7 review: a guard test fails on any uncoded unresolved reason without a user line (M1), and the encoder bounds role lines to the client's 512 code units (M4).
112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
"""User-facing lines for non-resolved prompt outcomes, from the closed vocabulary.
|
|
|
|
The inspector shows one line per unresolved role in place of the generic
|
|
"Prompt could not be resolved." Ambiguous and absent roles keep their status
|
|
copy, so a candidate is never presented as a confirmed prompt. The wire
|
|
projection accepts exactly the closed ``reason_code`` vocabulary and names a
|
|
blocking node only beside a code that names one.
|
|
"""
|
|
|
|
from typing import Any, Optional
|
|
|
|
from .prompt_pair_resolver import (
|
|
EMPTY_POSITIVE_REASON,
|
|
MULTISTAGE_UNATTRIBUTED_REASON,
|
|
SEMANTICS_UNAVAILABLE_REASON,
|
|
)
|
|
|
|
REASON_CODES = frozenset(
|
|
{
|
|
"work-limit",
|
|
"depth-limit",
|
|
"text-limit",
|
|
"graph-ceiling",
|
|
"unsupported-node",
|
|
"runtime-value",
|
|
"composite",
|
|
"no-sampler",
|
|
"malformed",
|
|
"multistage-differs",
|
|
}
|
|
)
|
|
LIMIT_CODES = frozenset({"work-limit", "depth-limit", "text-limit", "graph-ceiling"})
|
|
# Codes whose outcome names the blocking node: ``node: {id, class_type}``.
|
|
NAMED_CODES = frozenset({"unsupported-node", "runtime-value", "composite"})
|
|
# Bound for a node ID or class name on the wire and in a line.
|
|
CLASS_NAME_LENGTH = 200
|
|
# Bound for a role's line on the wire, in UTF-16 code units (the client's count).
|
|
MESSAGE_LENGTH = 512
|
|
UNREADABLE_METADATA_REASON = "Media metadata could not be read"
|
|
_NODE_KEYS = frozenset({"id", "class_type"})
|
|
|
|
_NAMED_LINES = {
|
|
"unsupported-node": "Prompt uses a node the inspector does not support",
|
|
"runtime-value": "Prompt is chosen at run time",
|
|
"composite": "Prompt combines several texts",
|
|
}
|
|
_CODED_LINES = {
|
|
**dict.fromkeys(LIMIT_CODES, "Prompt exceeds inspection limits."),
|
|
"no-sampler": "No supported sampler was found.",
|
|
"malformed": "Prompt metadata in this file is malformed.",
|
|
"multistage-differs": "Sampler stages have different prompts.",
|
|
}
|
|
# Uncoded outcomes, keyed by the resolver's exact reason.
|
|
_UNCODED_LINES = {
|
|
EMPTY_POSITIVE_REASON: "Positive prompt is empty.",
|
|
MULTISTAGE_UNATTRIBUTED_REASON: (
|
|
"Several sampler stages were found; their prompts were not compared."
|
|
),
|
|
SEMANTICS_UNAVAILABLE_REASON: "The inspector's node table is unavailable.",
|
|
UNREADABLE_METADATA_REASON: "Metadata in this file could not be read.",
|
|
}
|
|
|
|
|
|
def _text(value: Any) -> Optional[str]:
|
|
return value if isinstance(value, str) else None
|
|
|
|
|
|
def _class_name(node: Any) -> Optional[str]:
|
|
if not isinstance(node, dict):
|
|
return None
|
|
name = node.get("class_type")
|
|
return name[:CLASS_NAME_LENGTH] if isinstance(name, str) and name else None
|
|
|
|
|
|
def outcome_message(resolution: Any) -> Optional[str]:
|
|
"""One line for an unresolved outcome, or ``None`` to keep the status copy."""
|
|
if not isinstance(resolution, dict) or resolution.get("state") != "unresolved":
|
|
return None
|
|
code = _text(resolution.get("reason_code"))
|
|
if code in _NAMED_LINES:
|
|
name = _class_name(resolution.get("node"))
|
|
return f"{_NAMED_LINES[code]} ({name})." if name else f"{_NAMED_LINES[code]}."
|
|
if code is None:
|
|
return _UNCODED_LINES.get(_text(resolution.get("reason")))
|
|
return _CODED_LINES.get(code)
|
|
|
|
|
|
def _node(code: Optional[str], node: Any) -> Optional[dict]:
|
|
if code not in NAMED_CODES or not isinstance(node, dict) or set(node) != _NODE_KEYS:
|
|
return None
|
|
node_id, name = node["id"], node["class_type"]
|
|
if not isinstance(node_id, str) or not node_id:
|
|
return None
|
|
if name is not None and not isinstance(name, str):
|
|
return None
|
|
return {
|
|
"id": node_id[:CLASS_NAME_LENGTH],
|
|
"class_type": name[:CLASS_NAME_LENGTH] if name else None,
|
|
}
|
|
|
|
|
|
def project_resolution(resolution: dict) -> dict:
|
|
"""Fail closed: an unknown or contradictory code and its node are dropped."""
|
|
code = _text(resolution.get("reason_code"))
|
|
if code not in REASON_CODES or resolution.get("state") in ("resolved", "absent"):
|
|
code = None
|
|
return {
|
|
**resolution,
|
|
"reason_code": code,
|
|
"node": _node(code, resolution.get("node")),
|
|
}
|