Skip to content

ColorWatermarking

stegobox.codec.ColorWatermarking

Bases: BaseCodec

This steganography method is named: "ColorWatermarking"

In ColorWatermarking, carrier-->3 channels color image * 1, payload-->3 channels color image * 1, the watermark is universal for carrier(color) image.

  • Created by: Ruinan Ma
  • Created time: 2022/11/4

This is a pytorch implementation of image steganography via deep learning, which is released in paper - UDH: Universal Deep Hiding for Steganography, Watermarking, and Light Field Messaging https://papers.nips.cc/paper/2020/file/73d02e4344f71a0b0d51a925246990e7-Paper.pdf

Source code in stegobox/codec/udh_nips2020/color_watermarking.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class ColorWatermarking(BaseCodec):
    """
    This steganography method is named: "ColorWatermarking"

    In ColorWatermarking, carrier-->3 channels color image * 1,
                    payload-->3 channels color image * 1,
                    the watermark is universal for carrier(color) image.

    * Created by: Ruinan Ma
    * Created time: 2022/11/4

    This is a pytorch implementation of image steganography via deep learning,
    which is released in paper - UDH: Universal Deep Hiding for Steganography,
    Watermarking, and Light Field Messaging
    https://papers.nips.cc/paper/2020/file/73d02e4344f71a0b0d51a925246990e7-Paper.pdf
    """

    def __init__(self, verbose: bool = False, norm: str = "batch") -> None:
        super().__init__()
        self.verbose = verbose
        self.norm = norm
        self.num_downs = 5
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.imagesize = 128
        self.checkpoint = "ckpt/udh_nips2020/color_watermarking.pth.tar"
        # preprocessing for color_image
        self.transform_color = transforms.Compose(
            [transforms.Resize([self.imagesize, self.imagesize]), transforms.ToTensor()]
        )
        # select the type of norm operation
        if self.norm == "instance":
            self.norm_layer = nn.InstanceNorm2d
        elif self.norm == "batch":
            self.norm_layer = nn.BatchNorm2d  # type: ignore
        else:
            self.norm_layer = None  # type: ignore

        self.Hnet, self.Rnet = self.net_init()

        if verbose:
            if torch.cuda.is_available():
                print("Running color universal watermaking with GPU.")
            else:
                print("Running color universal watermaking with CPU.")

    def weights_init(self, m):
        classname = m.__class__.__name__
        if classname.find("Conv") != -1:
            nn.init.kaiming_normal_(m.weight.data, a=0, mode="fan_out")
        elif classname.find("BatchNorm") != -1:
            m.weight.data.fill_(1.0)
            m.bias.data.fill_(0)

    def net_init(self):
        # input_nc = channel_secret * num_secret
        # output_nc = channel_cover * num_cover
        hnet = UnetGenerator(
            input_nc=3 * 1,
            output_nc=3 * 1,
            num_downs=self.num_downs,
            norm_layer=self.norm_layer,
            output_function=nn.Tanh,  # type: ignore
        )
        # input_nc = channel_cover * num_cover
        # output_nc = channel_secret * num_secret
        rnet = RevealNet(
            input_nc=3 * 1,
            output_nc=3 * 1,
            nhf=64,
            norm_layer=self.norm_layer,
            output_function=nn.Sigmoid,
        )
        # Using kaiming normalization to initialize model.
        hnet.apply(self.weights_init)
        rnet.apply(self.weights_init)
        hnet = torch.nn.parallel.DataParallel(hnet).cuda()
        rnet = torch.nn.parallel.DataParallel(rnet).cuda()
        checkpoint = torch.load(self.checkpoint)
        hnet.load_state_dict(checkpoint["H_state_dict"])
        rnet.load_state_dict(checkpoint["R_state_dict"])

        return hnet, rnet

    def encode(self, carrier: Image.Image, payload: Image.Image) -> torch.Tensor:
        """Encode a color image with format png into a color image with format png.

        Args:
            carrier: cover image
            payload: Payload secret color image.

        Returns:
            Encoded steganographic image with format torch.tensor
        """
        payload = self.transform_color(payload)
        carrier = self.transform_color(carrier)
        payload = payload.unsqueeze(dim=0)  # type: ignore
        carrier = carrier.unsqueeze(dim=0)  # type: ignore
        payload = payload.to(self.device)  # type: ignore
        carrier = carrier.to(self.device)  # type: ignore

        # secret_p(universal watermarking)-->torch.Size([1, 3, 128, 128])
        secret_p = self.Hnet(payload)
        # you can save secret_p(universal watermarking)
        # torchvision.utils.save_image(secret_p, "./result-pic/secret_p.png")
        container = carrier + secret_p
        return container

    def decode(self, container: Image.Image) -> torch.Tensor:
        """Decode secret image from encoded steganographic image.

        Args:
            container: Encoded carrier(color) image.

        Returns:
            One decoded color image if decode is successful.
        """
        container = self.transform_color(container)
        container = container.unsqueeze(dim=0)  # type: ignore
        container = container.to(self.device)  # type: ignore
        reveal_payload = self.Rnet(container)
        return reveal_payload

encode(carrier, payload)

Encode a color image with format png into a color image with format png.

Parameters:

Name Type Description Default
carrier Image

cover image

 required
payload Image

Payload secret color image.

 required

Returns:

Type Description
Tensor

Encoded steganographic image with format torch.tensor
Source code in stegobox/codec/udh_nips2020/color_watermarking.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def encode(self, carrier: Image.Image, payload: Image.Image) -> torch.Tensor:
    """Encode a color image with format png into a color image with format png.

    Args:
        carrier: cover image
        payload: Payload secret color image.

    Returns:
        Encoded steganographic image with format torch.tensor
    """
    payload = self.transform_color(payload)
    carrier = self.transform_color(carrier)
    payload = payload.unsqueeze(dim=0)  # type: ignore
    carrier = carrier.unsqueeze(dim=0)  # type: ignore
    payload = payload.to(self.device)  # type: ignore
    carrier = carrier.to(self.device)  # type: ignore

    # secret_p(universal watermarking)-->torch.Size([1, 3, 128, 128])
    secret_p = self.Hnet(payload)
    # you can save secret_p(universal watermarking)
    # torchvision.utils.save_image(secret_p, "./result-pic/secret_p.png")
    container = carrier + secret_p
    return container

decode(container)

Decode secret image from encoded steganographic image.

Parameters:

Name Type Description Default
container Image

Encoded carrier(color) image.

 required

Returns:

Type Description
Tensor

One decoded color image if decode is successful.
Source code in stegobox/codec/udh_nips2020/color_watermarking.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def decode(self, container: Image.Image) -> torch.Tensor:
    """Decode secret image from encoded steganographic image.

    Args:
        container: Encoded carrier(color) image.

    Returns:
        One decoded color image if decode is successful.
    """
    container = self.transform_color(container)
    container = container.unsqueeze(dim=0)  # type: ignore
    container = container.to(self.device)  # type: ignore
    reveal_payload = self.Rnet(container)
    return reveal_payload