Views
No views yet

1conda create -n "watermark_anything" python=3.10.14
2conda activate watermark_anything
3conda install pytorch torchvision pytorch-cuda=12.4 -c pytorch -c nvidiapip install -r requirements.txtwget https://dl.fbaipublicfiles.com/watermark_anything/wam_mit.pth -P checkpoints/1from huggingface_hub import hf_hub_download
2ckpt_path = hf_hub_download(
3 repo_id="facebook/watermark-anything",
4 filename="checkpoint.pth"
5)params.json, notebooks/inference_utils.py). See notebooks/inference.ipynb for a notebook with the following scripts as well as visualizations.1import os
2import numpy as np
3from PIL import Image
4import torch
5import torch.nn.functional as F
6from torchvision.utils import save_image
7from huggingface_hub import hf_hub_download
8
9# Ensure these imports are available by cloning the official repository
10# and setting up your Python path, or copying the relevant files.
11from watermark_anything.data.metrics import msg_predict_inference
12from notebooks.inference_utils import (
13 load_model_from_checkpoint, default_transform, unnormalize_img,
14 create_random_mask, plot_outputs, msg2str
15)
16device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
17
18# Load the model from the specified checkpoint
19exp_dir = "checkpoints" # Assumes 'checkpoints' directory exists from cloning the repo
20json_path = os.path.join(exp_dir, "params.json")
21
22# Download the MIT licensed model weights
23ckpt_path = hf_hub_download(
24 repo_id="facebook/watermark-anything",
25 filename="wam_mit.pth"
26)
27# Ensure params.json is present from the cloned repository.
28# You might need to copy params.json from the cloned repo's checkpoints directory
29# if it's not automatically handled by your setup.
30if not os.path.exists(json_path):
31 print(f"Warning: params.json not found at {json_path}. Please ensure you have cloned the original repository and placed it correctly, or manually download it if available.")
32
33wam = load_model_from_checkpoint(json_path, ckpt_path).to(device).eval()
34
35# Define the directory containing the images to watermark
36img_dir = "assets/images" # Directory containing the original images
37output_dir = "outputs" # Directory to save the watermarked images
38os.makedirs(output_dir, exist_ok=True)[!TIP] You can specify thewam.scaling_wfactor, which controls the imperceptibility/robustness trade-off. Increasing it will lead to worse images but more robust watermarks, and vice versa. By default, it is set to 2.0, feel free to increase or decrease it to test how it influences the metrics.
1# Define a 32-bit message to be embedded into the images
2wm_msg = torch.randint(0, 2, (32,)).float().to(device)
3
4# Proportion of the image to be watermarked (0.5 means 50% of the image).
5# This is used here to show the watermark localization property. In practice, you may want to use a predifined mask or the entire image.
6proportion_masked = 0.5
7
8# Iterate over each image in the directory
9for img_ in os.listdir(img_dir):
10 # Load and preprocess the image
11 img_path = os.path.join(img_dir, img_)
12 img = Image.open(img_path).convert("RGB")
13 img_pt = default_transform(img).unsqueeze(0).to(device) # [1, 3, H, W]
14
15 # Embed the watermark message into the image
16 outputs = wam.embed(img_pt, wm_msg)
17
18 # Create a random mask to watermark only a part of the image
19 mask = create_random_mask(img_pt, num_masks=1,mask_percentage=proportion_masked) # [1, 1, H, W]
20 img_w = outputs['imgs_w'] * mask + img_pt * (1 - mask) # [1, 3, H, W]
21
22 # Detect the watermark in the watermarked image
23 preds = wam.detect(img_w)["preds"] # [1, 33, 256, 256]
24 mask_preds = F.sigmoid(preds[:, 0, :, :]) # [1, 256, 256], predicted mask
25 bit_preds = preds[:, 1:, :, :] # [1, 32, 256, 256], predicted bits
26
27 # Predict the embedded message and calculate bit accuracy
28 pred_message = msg_predict_inference(bit_preds, mask_preds).cpu().float() # [1, 32]
29 bit_acc = (pred_message == wm_msg).float().mean().item()
30
31 # Save the watermarked image and the detection mask
32 mask_preds_res = F.interpolate(mask_preds.unsqueeze(1), size=(img_pt.shape[-2], img_pt.shape[-1]), mode="bilinear", align_corners=False) # [1, 1, H, W]
33 save_image(unnormalize_img(img_w), f"{output_dir}/{img_}_wm.png")
34 save_image(mask_preds_res, f"{output_dir}/{img_}_pred.png")
35 save_image(mask, f"{output_dir}/{img_}_target.png")
36
37 # Print the predicted message and bit accuracy for each image
38 print(f"Predicted message for image {img_}: ", pred_message[0].numpy())
39 print(f"Bit accuracy for image {img_}: ", bit_acc)1from notebooks.inference_utils import multiwm_dbscan
2
3# DBSCAN parameters for detection
4epsilon = 1 # min distance between decoded messages in a cluster
5min_samples = 500 # min number of pixels in a 256x256 image to form a cluster
6
7# multiple 32 bit message to hide (could be more than 2; does not have to be 1 minus the other)
8wm_msgs = torch.randint(0, 2, (2, 32)).float().to(device)
9proportion_masked = 0.1 # max proportion per watermark, randomly placed
10
11for img_ in os.listdir(img_dir):
12 img = os.path.join(img_dir, img_)
13 img = Image.open(img, "r").convert("RGB")
14 img_pt = default_transform(img).unsqueeze(0).to(device)
15 # Mask to use. 1 values correspond to pixels where the watermark will be placed.
16 masks = create_random_mask(img_pt, num_masks=len(wm_msgs), mask_percentage=proportion_masked) # create one random mask per message
17 multi_wm_img = img_pt.clone()
18 for ii in range(len(wm_msgs)):
19 wm_msg, mask = wm_msgs[ii].unsqueeze(0), masks[ii]
20 outputs = wam.embed(img_pt, wm_msg)
21 multi_wm_img = outputs['imgs_w'] * mask + multi_wm_img * (1 - mask) # [1, 3, H, W]
22
23 # Detect the watermark in the multi-watermarked image
24 preds = wam.detect(multi_wm_img)["preds"] # [1, 33, 256, 256]
25 mask_preds = F.sigmoid(preds[:, 0, :, :]) # [1, 256, 256], predicted mask
26 bit_preds = preds[:, 1:, :, :] # [1, 32, 256, 256], predicted bits
27
28 # positions has the cluster number at each pixel. can be upsaled back to the original size.
29 centroids, positions = multiwm_dbscan(bit_preds, mask_preds, epsilon = epsilon, min_samples = min_samples)
30 centroids_pt = torch.stack(list(centroids.values()))
31
32 print(f"number messages found in image {img_}: {len(centroids)}")
33 for centroid in centroids_pt:
34 print(f"found centroid: {msg2str(centroid)}")
35 bit_acc = (centroid == wm_msgs).float().mean(dim=1)
36 # get message with maximum bit accuracy
37 bit_acc, idx = bit_acc.max(dim=0)
38 hamming = int(torch.sum(centroid != wm_msgs[idx]).item())
39 print(f"bit accuracy: {bit_acc.item()} - hamming distance: {hamming}/{len(wm_msgs[0])}")[!TIP] In the paper, the evaluated model was trained on the COCO dataset (with additional safety filters and where faces are blurred). For reproducibility purposes, we also release the weights (see above "Weights" subsection), but this model is under the CC-BY-NC License.
1@inproceedings{sander2025watermark,
2 title={Watermark Anything with Localized Messages},
3 author={Sander, Tom and Fernandez, Pierre and Durmus, Alain and Furon, Teddy and Douze, Matthijs},
4 booktitle={International Conference on Learning Representations (ICLR)},
5 year={2025}
6}