Skip to content

TwoColorInGray

stegobox.codec.TwoColorInGray

Bases: BaseCodec

This steganography method is named: "TwoColorInGray"

In TwoColorInGray, carrier-->1 channels gray image * 1, payload-->3 channels color image * 2, the watermark is universal for carrier(gray) 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/gray_watermarking.py
class TwoColorInGray(BaseCodec):
    """
    This steganography method is named: "TwoColorInGray"

    In TwoColorInGray, carrier-->1 channels gray image * 1,
                    payload-->3 channels color image * 2,
                    the watermark is universal for carrier(gray) 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/gray_watermarking.pth.tar"
        # preprocessing for color_image
        self.transform_color = transforms.Compose(
            [transforms.Resize([self.imagesize, self.imagesize]), transforms.ToTensor()]
        )
        self.transform_gray = transforms.Compose(
            [
                transforms.Grayscale(num_output_channels=1),
                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 two color images in one gray image with GPU.")
            else:
                print("Running two color images in one gray image with CPU.")

    def _transform_color(self, image: Image.Image) -> torch.Tensor:
        return self.transform_color(image)  # type: ignore

    def _transform_gray(self, image: Image.Image) -> torch.Tensor:
        return self.transform_gray(image)  # type: ignore

    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 * 2,
            output_nc=1 * 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=1 * 1,
            output_nc=3 * 2,
            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.DataParallel(hnet).cuda()
        rnet = torch.nn.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, _, __) -> None:
        raise NotImplementedError("Use encode_multiple() instead.")

    def encode_multiple(
        self, payload1: Image.Image, payload2: Image.Image, carrier: Image.Image
    ) -> torch.Tensor:
        """Encode two color images with format png into a gray image with format png.

        Args:
            carrier: cover(gray) image
            payload1: Payload secret color image1
            payload2: Payload secret color image2

        Returns:
            Encoded steganographic image with format torch.tensor
        """
        payload1_t = self._transform_color(payload1)
        payload2_t = self._transform_color(payload2)
        carrier_t = self._transform_gray(carrier)
        payload1_t = payload1_t.unsqueeze(dim=0)
        payload2_t = payload2_t.unsqueeze(dim=0)
        carrier_t = carrier_t.unsqueeze(dim=0)
        payload1_t = payload1_t.to(self.device)
        payload2_t = payload2_t.to(self.device)
        carrier_t = carrier_t.to(self.device)
        # payload.size()-->torch.Size([1, 6, 128, 128])
        payload = torch.cat((payload1_t, payload2_t), dim=1)
        # secret_p.size()-->torch.Size([1, 1, 128, 128])
        secret_p = self.Hnet(payload)
        container = secret_p + carrier_t
        return container

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

        Args:
            container: Encoded carrier(gray) image.

        Returns:
            Two decoded color images if decode is successful.
        """

        # container.size()-->torch.Size([1, 128, 128])
        container_t = self._transform_gray(container)
        container_t = container_t.unsqueeze(dim=0)
        container_t = container_t.to(self.device)
        # reveal_payload.size()-->torch.Size([1, 6, 128, 128])
        reveal_payload = self.Rnet(container_t)
        # reveal_payload1.size()=reveal_payload2.size()-->torch.Size([1, 3, 128, 128])
        reveal_payload1 = reveal_payload.narrow(1, 0, 3)
        reveal_payload2 = reveal_payload.narrow(1, 3, 3)
        return reveal_payload1, reveal_payload2

encode_multiple(payload1, payload2, carrier)

Encode two color images with format png into a gray image with format png.

Parameters:

Name Type Description Default
carrier Image

cover(gray) image

required
payload1 Image

Payload secret color image1

required
payload2 Image

Payload secret color image2

required

Returns:

Type Description
Tensor

Encoded steganographic image with format torch.tensor

Source code in stegobox/codec/udh_nips2020/gray_watermarking.py
def encode_multiple(
    self, payload1: Image.Image, payload2: Image.Image, carrier: Image.Image
) -> torch.Tensor:
    """Encode two color images with format png into a gray image with format png.

    Args:
        carrier: cover(gray) image
        payload1: Payload secret color image1
        payload2: Payload secret color image2

    Returns:
        Encoded steganographic image with format torch.tensor
    """
    payload1_t = self._transform_color(payload1)
    payload2_t = self._transform_color(payload2)
    carrier_t = self._transform_gray(carrier)
    payload1_t = payload1_t.unsqueeze(dim=0)
    payload2_t = payload2_t.unsqueeze(dim=0)
    carrier_t = carrier_t.unsqueeze(dim=0)
    payload1_t = payload1_t.to(self.device)
    payload2_t = payload2_t.to(self.device)
    carrier_t = carrier_t.to(self.device)
    # payload.size()-->torch.Size([1, 6, 128, 128])
    payload = torch.cat((payload1_t, payload2_t), dim=1)
    # secret_p.size()-->torch.Size([1, 1, 128, 128])
    secret_p = self.Hnet(payload)
    container = secret_p + carrier_t
    return container

decode(container)

Decode secret image from encoded steganographic image.

Parameters:

Name Type Description Default
container Image

Encoded carrier(gray) image.

required

Returns:

Type Description
tuple[Tensor, Tensor]

Two decoded color images if decode is successful.

Source code in stegobox/codec/udh_nips2020/gray_watermarking.py
def decode(self, container: Image.Image) -> tuple[torch.Tensor, torch.Tensor]:
    """Decode secret image from encoded steganographic image.

    Args:
        container: Encoded carrier(gray) image.

    Returns:
        Two decoded color images if decode is successful.
    """

    # container.size()-->torch.Size([1, 128, 128])
    container_t = self._transform_gray(container)
    container_t = container_t.unsqueeze(dim=0)
    container_t = container_t.to(self.device)
    # reveal_payload.size()-->torch.Size([1, 6, 128, 128])
    reveal_payload = self.Rnet(container_t)
    # reveal_payload1.size()=reveal_payload2.size()-->torch.Size([1, 3, 128, 128])
    reveal_payload1 = reveal_payload.narrow(1, 0, 3)
    reveal_payload2 = reveal_payload.narrow(1, 3, 3)
    return reveal_payload1, reveal_payload2