Feature (frontend): Add invisible watermark decoder node. (#8967)

* Initial plan

* Add invisible watermark decoding node and utility method

Co-authored-by: lstein <111189+lstein@users.noreply.github.com>

* chore(frontend): typegen

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: lstein <111189+lstein@users.noreply.github.com>
This commit is contained in:
Lincoln Stein 2026-04-04 16:00:21 -04:00 committed by GitHub
parent 6963cd97ba
commit ed268b1cfc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 89 additions and 7 deletions

View File

@ -21,7 +21,7 @@ from invokeai.app.invocations.fields import (
WithBoard,
WithMetadata,
)
from invokeai.app.invocations.primitives import ImageOutput
from invokeai.app.invocations.primitives import ImageOutput, StringOutput
from invokeai.app.services.image_records.image_records_common import ImageCategory
from invokeai.app.services.shared.invocation_context import InvocationContext
from invokeai.app.util.misc import SEED_MAX
@ -581,6 +581,25 @@ class ImageWatermarkInvocation(BaseInvocation, WithMetadata, WithBoard):
return ImageOutput.build(image_dto)
@invocation(
"decode_watermark",
title="Decode Invisible Watermark",
tags=["image", "watermark"],
category="image",
version="1.0.0",
)
class DecodeInvisibleWatermarkInvocation(BaseInvocation):
"""Decode an invisible watermark from an image."""
image: ImageField = InputField(description="The image to decode the watermark from")
length: int = InputField(default=8, description="The expected watermark length in bytes")
def invoke(self, context: InvocationContext) -> StringOutput:
image = context.images.get_pil(self.image.image_name)
watermark = InvisibleWatermark.decode_watermark(image, self.length)
return StringOutput(value=watermark)
@invocation(
"mask_edge",
title="Mask Edge",

View File

@ -9,7 +9,7 @@ import numpy as np
from PIL import Image
import invokeai.backend.util.logging as logger
from invokeai.backend.image_util.imwatermark.vendor import WatermarkEncoder
from invokeai.backend.image_util.imwatermark.vendor import WatermarkDecoder, WatermarkEncoder
class InvisibleWatermark:
@ -25,3 +25,25 @@ class InvisibleWatermark:
encoder.set_watermark("bytes", watermark_text.encode("utf-8"))
bgr_encoded = encoder.encode(bgr, "dwtDct")
return Image.fromarray(cv2.cvtColor(bgr_encoded, cv2.COLOR_BGR2RGB)).convert("RGBA")
@classmethod
def decode_watermark(cls, image: Image.Image, length: int = 8) -> str:
"""Attempt to decode an invisible watermark from an image.
Args:
image: The PIL Image to decode the watermark from.
length: The expected watermark length in bytes. Must match the length used when encoding.
The WatermarkDecoder requires the length in bits; this value is multiplied by 8 internally.
Returns:
The decoded watermark text, or an empty string if no watermark is detected or decoding fails.
"""
logger.debug("Attempting to decode invisible watermark")
try:
bgr = cv2.cvtColor(np.array(image.convert("RGB")), cv2.COLOR_RGB2BGR)
decoder = WatermarkDecoder("bytes", length * 8)
watermark_bytes = decoder.decode(bgr, "dwtDct")
return watermark_bytes.decode("utf-8", errors="ignore").rstrip("\x00")
except Exception:
logger.debug("Failed to decode invisible watermark")
return ""

File diff suppressed because one or more lines are too long