Skip to content

MaxDct

stegobox.codec.MaxDct

Bases: BaseCodec

MaxDct - embedding algorithm that creates stealthy steganographic images and decodes them without relying on the original image.

Performance - MaxDct is fast and suitable for on-the-fly embedding.

Original implementation reports 300-350ms encoding time and 150-200ms decoding time for 1920x1080 images on CPU.

Originally implemented in ShieldMnt/invisible-watermark and is used inside CompVis/stable-diffusion.

Source code in stegobox/codec/invisible_watermark/max_dct.py 
 10
 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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
class MaxDct(BaseCodec):
    """MaxDct - embedding algorithm that creates stealthy steganographic images and
    decodes them without relying on the original image.

    Tip: Performance - MaxDct is fast and suitable for on-the-fly embedding.
        Original implementation reports 300-350ms encoding time and 150-200ms decoding
        time for 1920x1080 images on CPU.

    Originally implemented in
    [ShieldMnt/invisible-watermark](https://github.com/ShieldMnt/invisible-watermark)
    and is used inside
    [CompVis/stable-diffusion](https://github.com/CompVis/stable-diffusion).
    """

    def __init__(self, scales: list[int] = [0, 36, 36], block: int = 4) -> None:
        """Frequency embedding method - MaxDCT uses DWT and DCT transforms to embed
        payload bits into the max non-trivial coefficient of block DCT coefficients.

        Args:
            scales: A list of scaling factors, each one registered to apply to a color
                channel of an RGB image. Defaults to [0, 36, 36].
            block: Number of DWT blocks. Defaults to 4.
        """
        super().__init__()
        self.scales = scales
        self.block = block

    def encode(self, carrier: cv2.Mat, payload: str) -> tuple[cv2.Mat, int]:
        """Encodes payload string into an image with a DWT + DCT transform.

        Args:
            carrier: The carrier image, read with `io.image.read_cv2()`.
            payload: The payload string.

        Returns:
            The encoded image, write with `io.image.write_cv2()`.
            Length of payload in bits, you will need this when decoding.
        """
        bgr = carrier
        row, col, _ = bgr.shape
        yuv = cv2.cvtColor(bgr, cv2.COLOR_BGR2YUV)

        # Payload string encoded as bytes
        payload_seq = payload_in_bytes(payload)
        payload_len = len(payload_seq)
        print(f"Payload length: {payload_len} bits")

        for c in range(2):
            if self.scales[c] <= 0:
                continue

            c1, (h1, v2, d1) = pywt.dwt2(yuv[: row // 4 * 4, : col // 4 * 4, c], "haar")
            self._encode_frame(payload_seq, payload_len, c1, self.scales[c])

            yuv[: row // 4 * 4, : col // 4 * 4, c] = pywt.idwt2(
                (c1, (h1, v2, d1)), "haar"
            )

        bgr_encoded = cv2.cvtColor(yuv, cv2.COLOR_YUV2BGR)
        return bgr_encoded, payload_len

    def decode(self, _):
        raise NotImplementedError("This codec does not support decoding without length")

    def decode_with_length(self, carrier: cv2.Mat, payload_len: int) -> str:
        """Try to decode payload from an image with a DWT + DCT transform.

        Args:
            carrier: The encoded steganographic image, read with `io.image.read_cv2()`.
            payload_len: The length of your payload in bits.

        Returns:
            The extracted payload string if successful.
        """
        bgr = carrier
        row, col, _ = bgr.shape

        yuv = cv2.cvtColor(bgr, cv2.COLOR_BGR2YUV)
        scores: list[list[int]] = [[] for _ in range(payload_len)]

        for c in range(2):
            if self.scales[c] <= 0:
                continue

            c1, (_, _, _) = pywt.dwt2(yuv[: row // 4 * 4, : col // 4 * 4, c], "haar")
            scores = self._decode_frame(payload_len, c1, self.scales[c], scores)

        avg_scores = list(map(lambda x: np.array(x).mean(), scores))
        bits = np.array(avg_scores) * 255 > 127
        return reconstruct_bytes(bits, payload_len)

    def _decode_frame(
        self, wm_len: int, frame: np.ndarray, scale: int, scores: list[list[int]]
    ) -> list[list[int]]:
        (row, col) = frame.shape
        num = 0

        for i in range(row // self.block):
            for j in range(col // self.block):
                block = frame[
                    i * self.block : i * self.block + self.block,
                    j * self.block : j * self.block + self.block,
                ]

                score = self._infer_dct_matrix(block, scale)
                # score = self.infer_dct_svd(block, scale)
                wm_bit = num % wm_len
                scores[wm_bit].append(score)
                num = num + 1

        return scores

    def _diffuse_dct_svd(
        self, block: np.ndarray, wm_bit: np.uint8, scale: int
    ) -> np.ndarray:
        u, s, v = np.linalg.svd(cv2.dct(block))

        s[0] = (s[0] // scale + 0.25 + 0.5 * wm_bit) * scale
        return cv2.idct(np.dot(u, np.dot(np.diag(s), v)))

    def _infer_dct_svd(self, block: np.ndarray, scale: int) -> int:
        u, s, v = np.linalg.svd(cv2.dct(block))

        score = 0
        score = int((s[0] % scale) > scale * 0.5)
        return score
        # if score >= 0.5:
        #     return 1.0
        # else:
        #     return 0.0

    def _diffuse_dct_matrix(
        self, block: np.ndarray, wm_bit: np.uint8, scale: int
    ) -> np.ndarray:
        pos = np.argmax(abs(block.flatten()[1:])) + 1
        i, j = pos // self.block, pos % self.block
        val = block[i][j]
        if val >= 0.0:
            block[i][j] = (val // scale + 0.25 + 0.5 * wm_bit) * scale
        else:
            val = abs(val)
            block[i][j] = -1.0 * (val // scale + 0.25 + 0.5 * wm_bit) * scale
        return block

    def _infer_dct_matrix(self, block: np.ndarray, scale: int) -> int:
        pos = np.argmax(abs(block.flatten()[1:])) + 1
        i, j = pos // self.block, pos % self.block

        val = block[i][j]
        if val < 0:
            val = abs(val)

        if (val % scale) > 0.5 * scale:
            return 1
        else:
            return 0

    def _encode_frame(
        self, wm: list[np.uint8], wm_len: int, frame: np.ndarray, scale: int
    ) -> None:
        """
        frame is a numpy array of matrix (M, N)

        we get K (watermark bits size) blocks (self.block x self.block)

        For i-th block, we encode watermark[i] bit into it
        """
        (row, col) = frame.shape
        num = 0
        for i in range(row // self.block):
            for j in range(col // self.block):
                block = frame[
                    i * self.block : i * self.block + self.block,
                    j * self.block : j * self.block + self.block,
                ]
                wm_bit = wm[(num % wm_len)]

                diffused_block = self._diffuse_dct_matrix(block, wm_bit, scale)
                # diffusedBlock = self.diffuse_dct_svd(block, wmBit, scale)
                frame[
                    i * self.block : i * self.block + self.block,
                    j * self.block : j * self.block + self.block,
                ] = diffused_block

                num = num + 1

__init__(scales=[0, 36, 36], block=4)

Frequency embedding method - MaxDCT uses DWT and DCT transforms to embed payload bits into the max non-trivial coefficient of block DCT coefficients.

Parameters:

Name Type Description Default
scales list[int]

A list of scaling factors, each one registered to apply to a color channel of an RGB image. Defaults to [0, 36, 36].

 [0, 36, 36]
block int

Number of DWT blocks. Defaults to 4.

 4
Source code in stegobox/codec/invisible_watermark/max_dct.py 
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(self, scales: list[int] = [0, 36, 36], block: int = 4) -> None:
    """Frequency embedding method - MaxDCT uses DWT and DCT transforms to embed
    payload bits into the max non-trivial coefficient of block DCT coefficients.

    Args:
        scales: A list of scaling factors, each one registered to apply to a color
            channel of an RGB image. Defaults to [0, 36, 36].
        block: Number of DWT blocks. Defaults to 4.
    """
    super().__init__()
    self.scales = scales
    self.block = block

encode(carrier, payload)

Encodes payload string into an image with a DWT + DCT transform.

Parameters:

Name Type Description Default
carrier Mat

The carrier image, read with io.image.read_cv2().

 required
payload str

The payload string.

 required

Returns:

Type Description
Mat

The encoded image, write with io.image.write_cv2().
int

Length of payload in bits, you will need this when decoding.
Source code in stegobox/codec/invisible_watermark/max_dct.py 
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
def encode(self, carrier: cv2.Mat, payload: str) -> tuple[cv2.Mat, int]:
    """Encodes payload string into an image with a DWT + DCT transform.

    Args:
        carrier: The carrier image, read with `io.image.read_cv2()`.
        payload: The payload string.

    Returns:
        The encoded image, write with `io.image.write_cv2()`.
        Length of payload in bits, you will need this when decoding.
    """
    bgr = carrier
    row, col, _ = bgr.shape
    yuv = cv2.cvtColor(bgr, cv2.COLOR_BGR2YUV)

    # Payload string encoded as bytes
    payload_seq = payload_in_bytes(payload)
    payload_len = len(payload_seq)
    print(f"Payload length: {payload_len} bits")

    for c in range(2):
        if self.scales[c] <= 0:
            continue

        c1, (h1, v2, d1) = pywt.dwt2(yuv[: row // 4 * 4, : col // 4 * 4, c], "haar")
        self._encode_frame(payload_seq, payload_len, c1, self.scales[c])

        yuv[: row // 4 * 4, : col // 4 * 4, c] = pywt.idwt2(
            (c1, (h1, v2, d1)), "haar"
        )

    bgr_encoded = cv2.cvtColor(yuv, cv2.COLOR_YUV2BGR)
    return bgr_encoded, payload_len

decode_with_length(carrier, payload_len)

Try to decode payload from an image with a DWT + DCT transform.

Parameters:

Name Type Description Default
carrier Mat

The encoded steganographic image, read with io.image.read_cv2().

 required
payload_len int

The length of your payload in bits.

 required

Returns:

Type Description
str

The extracted payload string if successful.
Source code in stegobox/codec/invisible_watermark/max_dct.py 
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
def decode_with_length(self, carrier: cv2.Mat, payload_len: int) -> str:
    """Try to decode payload from an image with a DWT + DCT transform.

    Args:
        carrier: The encoded steganographic image, read with `io.image.read_cv2()`.
        payload_len: The length of your payload in bits.

    Returns:
        The extracted payload string if successful.
    """
    bgr = carrier
    row, col, _ = bgr.shape

    yuv = cv2.cvtColor(bgr, cv2.COLOR_BGR2YUV)
    scores: list[list[int]] = [[] for _ in range(payload_len)]

    for c in range(2):
        if self.scales[c] <= 0:
            continue

        c1, (_, _, _) = pywt.dwt2(yuv[: row // 4 * 4, : col // 4 * 4, c], "haar")
        scores = self._decode_frame(payload_len, c1, self.scales[c], scores)

    avg_scores = list(map(lambda x: np.array(x).mean(), scores))
    bits = np.array(avg_scores) * 255 > 127
    return reconstruct_bytes(bits, payload_len)