Spaces:
Running
Running
Update utils.py
Browse files
utils.py
CHANGED
@@ -1,87 +1,119 @@
|
|
1 |
-
import cv2
|
2 |
import numpy as np
|
3 |
from PIL import Image, PngImagePlugin
|
4 |
|
5 |
-
def png_encode(im_name,extra):
|
6 |
-
|
7 |
-
|
8 |
im = Image.open(im_name)
|
9 |
-
|
10 |
info = PngImagePlugin.PngInfo()
|
11 |
-
|
12 |
-
|
13 |
-
|
14 |
-
|
15 |
-
|
16 |
-
|
17 |
|
18 |
def to_bin(data):
|
19 |
-
"""Convert
|
20 |
if isinstance(data, str):
|
21 |
-
return ''.join([
|
22 |
elif isinstance(data, bytes):
|
23 |
-
return ''.join([
|
24 |
elif isinstance(data, np.ndarray):
|
25 |
-
return [
|
26 |
elif isinstance(data, int) or isinstance(data, np.uint8):
|
27 |
return format(data, "08b")
|
28 |
else:
|
29 |
raise TypeError("Type not supported.")
|
30 |
-
|
|
|
|
|
31 |
try:
|
32 |
-
|
33 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
34 |
binary_data = ""
|
|
|
35 |
for row in image:
|
36 |
for pixel in row:
|
37 |
r, g, b = to_bin(pixel)
|
38 |
binary_data += r[-1]
|
39 |
binary_data += g[-1]
|
40 |
binary_data += b[-1]
|
41 |
-
|
|
|
42 |
decoded_data = ""
|
43 |
-
for
|
44 |
-
|
45 |
-
if
|
|
|
|
|
|
|
|
|
|
|
|
|
46 |
break
|
47 |
-
|
48 |
-
|
|
|
|
|
|
|
|
|
49 |
except Exception as e:
|
50 |
-
|
51 |
-
return this
|
52 |
|
53 |
-
def encode(image_name, secret_data,txt=None):
|
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 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
1 |
+
import cv2
|
2 |
import numpy as np
|
3 |
from PIL import Image, PngImagePlugin
|
4 |
|
5 |
+
def png_encode(im_name, extra):
|
6 |
+
"""Encode watermark using PNG metadata"""
|
7 |
+
try:
|
8 |
im = Image.open(im_name)
|
|
|
9 |
info = PngImagePlugin.PngInfo()
|
10 |
+
# Encode the text as UTF-8 before adding to metadata
|
11 |
+
info.add_text("TXT", extra.encode('utf-8').decode('utf-8'))
|
12 |
+
im.save("test.png", pnginfo=info)
|
13 |
+
return "test.png", "Watermark added successfully"
|
14 |
+
except Exception as e:
|
15 |
+
return im_name, f"Error adding watermark: {str(e)}"
|
16 |
|
17 |
def to_bin(data):
|
18 |
+
"""Convert data to binary format as string"""
|
19 |
if isinstance(data, str):
|
20 |
+
return ''.join([format(ord(i), "08b") for i in data])
|
21 |
elif isinstance(data, bytes):
|
22 |
+
return ''.join([format(i, "08b") for i in data])
|
23 |
elif isinstance(data, np.ndarray):
|
24 |
+
return [format(i, "08b") for i in data]
|
25 |
elif isinstance(data, int) or isinstance(data, np.uint8):
|
26 |
return format(data, "08b")
|
27 |
else:
|
28 |
raise TypeError("Type not supported.")
|
29 |
+
|
30 |
+
def decode(image_name, txt=None):
|
31 |
+
"""Decode watermark from image"""
|
32 |
try:
|
33 |
+
# First try PNG metadata method
|
34 |
+
try:
|
35 |
+
im = Image.open(image_name)
|
36 |
+
if "TXT" in im.info:
|
37 |
+
return im.info["TXT"]
|
38 |
+
except:
|
39 |
+
pass
|
40 |
+
|
41 |
+
# If PNG metadata fails, try steganography method
|
42 |
+
image = cv2.imread(image_name)
|
43 |
+
if image is None:
|
44 |
+
raise ValueError("Could not read image file")
|
45 |
+
|
46 |
+
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
|
47 |
binary_data = ""
|
48 |
+
|
49 |
for row in image:
|
50 |
for pixel in row:
|
51 |
r, g, b = to_bin(pixel)
|
52 |
binary_data += r[-1]
|
53 |
binary_data += g[-1]
|
54 |
binary_data += b[-1]
|
55 |
+
|
56 |
+
# Process binary data in chunks of 8 bits
|
57 |
decoded_data = ""
|
58 |
+
for i in range(0, len(binary_data), 8):
|
59 |
+
byte = binary_data[i:i+8]
|
60 |
+
if len(byte) == 8: # Ensure we have a complete byte
|
61 |
+
try:
|
62 |
+
decoded_data += chr(int(byte, 2))
|
63 |
+
except ValueError:
|
64 |
+
continue
|
65 |
+
|
66 |
+
if decoded_data.endswith("====="):
|
67 |
break
|
68 |
+
|
69 |
+
if "#####" in decoded_data:
|
70 |
+
decoded_data = decoded_data.split("#####")[0]
|
71 |
+
|
72 |
+
return decoded_data.strip().rstrip("=")
|
73 |
+
|
74 |
except Exception as e:
|
75 |
+
return f"Error detecting watermark: {str(e)}"
|
|
|
76 |
|
77 |
+
def encode(image_name, secret_data, txt=None):
|
78 |
+
"""Encode watermark using steganography"""
|
79 |
+
try:
|
80 |
+
image = cv2.imread(image_name)
|
81 |
+
if image is None:
|
82 |
+
raise ValueError("Could not read image file")
|
83 |
+
|
84 |
+
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
|
85 |
+
|
86 |
+
# Calculate maximum bytes that can be encoded
|
87 |
+
n_bytes = image.shape[0] * image.shape[1] * 3 // 8
|
88 |
+
|
89 |
+
# Prepare the secret data
|
90 |
+
secret_data = str(secret_data).encode('utf-8').decode('utf-8') # Ensure proper encoding
|
91 |
+
secret_data_with_delimiter = secret_data + "#####"
|
92 |
+
|
93 |
+
if len(secret_data_with_delimiter) + 5 >= n_bytes:
|
94 |
+
return image_name, "Watermark is too large for Image Size"
|
95 |
+
|
96 |
+
secret_data_with_delimiter += "====="
|
97 |
+
|
98 |
+
# Convert secret data to binary
|
99 |
+
binary_secret_data = to_bin(secret_data_with_delimiter)
|
100 |
+
data_len = len(binary_secret_data)
|
101 |
+
data_index = 0
|
102 |
+
|
103 |
+
# Embed the secret data
|
104 |
+
for i in range(image.shape[0]):
|
105 |
+
for j in range(image.shape[1]):
|
106 |
+
if data_index < data_len:
|
107 |
+
# Modify the least significant bit of each color channel
|
108 |
+
for k in range(3): # For each color channel (RGB)
|
109 |
+
if data_index < data_len:
|
110 |
+
image[i, j, k] = int(to_bin(image[i, j, k])[:-1] + binary_secret_data[data_index], 2)
|
111 |
+
data_index += 1
|
112 |
+
|
113 |
+
# Save the result
|
114 |
+
output_path = "watermarked_image.png"
|
115 |
+
cv2.imwrite(output_path, cv2.cvtColor(image, cv2.COLOR_RGB2BGR))
|
116 |
+
return output_path, "Watermark embedded successfully"
|
117 |
+
|
118 |
+
except Exception as e:
|
119 |
+
return image_name, f"Error encoding watermark: {str(e)}"
|