USER
import numpy as np
from PIL import Image
import cv2
from scipy.fftpack import dct, idct
import matplotlib.pyplot as plt
class DCTSteganography:
def __init__(self, alpha=0.1):
self.alpha = alpha
self.block_size = 8
self.num_coefficients = 1
def _split_into_blocks(self, image):
height, width = image.shape
blocks = []
for i in range(0, height, self.block_size):
for j in range(0, width, self.block_size):
block = image[i:i+self.block_size, j:j+self.block_size]
if block.shape[0] == self.block_size and block.shape[1] == self.block_size:
blocks.append(block)
return blocks
def _merge_blocks(self, blocks, height, width):
image = np.zeros((height, width))
block_idx = 0
for i in range(0, height, self.block_size):
for j in range(0, width, self.block_size):
if block_idx < len(blocks):
image[i:i+self.block_size, j:j+self.block_size] = blocks[block_idx]
block_idx += 1
return image
def _get_largest_coefficients(self, dct_block):
ac_block = dct_block.copy()
ac_block[0,0] = 0
flat_index = np.argmax(np.abs(ac_block.ravel()))
return np.unravel_index([flat_index], dct_block.shape)
def embed(self, image_path, message):
message = message + '\0'
message_bytes = message.encode('utf-8')
message_bits = ''.join(format(byte, '08b') for byte in message_bytes)
message_length = len(message_bits)
length_bits = format(message_length, '032b')
full_message_bits = length_bits + message_bits
img_color = cv2.imread(image_path)
if img_color is None:
raise ValueError("Не удалось прочитать изображение")
red_channel = img_color[:,:,2].copy()
height, width = red_channel.shape
blocks = self._split_into_blocks(red_channel)
embedded_bits = 0
bit_idx = 0
for block_idx, block in enumerate(blocks):
if bit_idx >= len(full_message_bits):
break
dct_block = dct(dct(block.T, norm='ortho').T, norm='ortho')
rows, cols = self._get_largest_coefficients(dct_block)
for i in range(self.num_coefficients):
if bit_idx >= len(full_message_bits):
break
bit = int(full_message_bits[bit_idx])
c_i = dct_block[rows[i], cols[i]]
if bit == 1:
dct_block[rows[i], cols[i]] = abs(c_i) * (1 + self.alpha)
else:
dct_block[rows[i], cols[i]] = -abs(c_i) * (1 + self.alpha)
bit_idx += 1
embedded_bits += 1
restored_block = idct(idct(dct_block.T, norm='ortho').T, norm='ortho')
blocks[block_idx] = np.clip(restored_block, 0, 255)
red_channel_modified = self._merge_blocks(blocks, height, width)
stego_image = img_color.copy()
stego_image[:,:,2] = red_channel_modified
psnr = self.calculate_psnr(img_color, stego_image)
return stego_image.astype(np.uint8), embedded_bits, psnr
def extract(self, stego_image_path):
img_color = cv2.imread(stego_image_path)
if img_color is None:
return "Ошибка: не удалось прочитать изображение"
red_channel = img_color[:,:,2]
blocks = self._split_into_blocks(red_channel)
extracted_bits = ""
length_bits = ""
bit_count = 0
block_idx = 0
while bit_count < 32 and block_idx < len(blocks):
dct_block = dct(dct(blocks[block_idx].T, norm='ortho').T, norm='ortho')
rows, cols = self._get_largest_coefficients(dct_block)
for i in range(self.num_coefficients):
if bit_count >= 32:
break
coef = dct_block[rows[i], cols[i]]
bit = '1' if coef >= 0 else '0'
length_bits += bit
bit_count += 1
block_idx += 1
try:
message_length = int(length_bits, 2)
while len(extracted_bits) < message_length and block_idx < len(blocks):
dct_block = dct(dct(blocks[block_idx].T, norm='ortho').T, norm='ortho')
rows, cols = self._get_largest_coefficients(dct_block)
for i in range(self.num_coefficients):
if len(extracted_bits) >= message_length:
break
coef = dct_block[rows[i], cols[i]]
bit = '1' if coef >= 0 else '0'
extracted_bits += bit
block_idx += 1
bytes_data = bytearray()
for i in range(0, len(extracted_bits) - (len(extracted_bits) % 8), 8):
byte = int(extracted_bits[i:i+8], 2)
bytes_data.append(byte)
message = bytes_data.decode('utf-8', errors='ignore')
if '\0' in message:
message = message[:message.index('\0')]
clean_message = ''.join(char for char in message
if (32 <= ord(char) <= 126) or (1040 <= ord(char) <= 1103))
if len(clean_message) > 0:
return clean_message
return "Сообщение не найдено"
except Exception as e:
return f"Ошибка извлечения: {str(e)}"
def calculate_errors(self, original_message, extracted_message):
if not extracted_message or extracted_message == "Сообщение не найдено":
return len(original_message), 100.0
extracted = extracted_message[:len(original_message)]
extracted = extracted.ljust(len(original_message))
errors = sum(1 for a, b in zip(original_message, extracted) if a != b)
error_rate = (errors / len(original_message)) * 100
return errors, error_rate
def calculate_bit_errors(self, original_message, extracted_message):
if not extracted_message or extracted_message == "Сообщение не найдено":
orig_bits = ''.join(format(ord(c), '08b') for c in original_message)
return len(orig_bits), 100.0
orig_bits = ''.join(format(ord(c), '08b') for c in original_message)
extracted = extracted_message[:len(original_message)]
extracted = extracted.ljust(len(original_message))
extracted_bits = ''.join(format(ord(c), '08b') for c in extracted)
bit_errors = sum(1 for a, b in zip(orig_bits, extracted_bits) if a != b)
bit_error_rate = (bit_errors / len(orig_bits)) * 100
return bit_errors, bit_error_rate
def calculate_psnr(self, original_image, modified_image):
if original_image.shape != modified_image.shape:
return 0
mse = np.mean((original_image.astype(float) - modified_image.astype(float)) ** 2)
if mse == 0:
return float('inf')
max_pixel = 255.0
psnr = 20 * np.log10(max_pixel / np.sqrt(mse))
return psnr
# Пример использования
if __name__ == "__main__":
stego = DCTSteganography(alpha=0.1)
# Читаем оригинальное изображение
original_image = cv2.imread("input.bmp")
if original_image is None:
print("Ошибка: не удалось прочитать входное изображение")
exit(1)
message = "Hello World!"
stego_img, embedded_bits, embed_psnr = stego.embed("input.bmp", message)
cv2.imwrite("stego.bmp", stego_img)
print(f"Embedded bits count: {embedded_bits - 40} + 8 (/0)")
print(f"Secret length: {len(message)}")
# Проверка извлечения из оригинального стего-изображения
extracted_message = stego.extract("stego.bmp")
print(f"Extracted message: {extracted_message}")
errors, error_rate = stego.calculate_errors(message, extracted_message)
print(f"Errors: {errors}/{len(message)}")
print(f"PSNR: {embed_psnr:.2f} dB")
# Проверка устойчивости к JPEG-сжатию
print("\nCompressed JPEG:")
quality = 50
jpeg_path = "compressed.jpg"
cv2.imwrite(jpeg_path, stego_img, [cv2.IMWRITE_JPEG_QUALITY, quality])
compressed_image = cv2.imread(jpeg_path)
psnr_compressed = stego.calculate_psnr(stego_img, compressed_image)
compressed_message = stego.extract(jpeg_path)
print(f"Extracted message: {compressed_message}")
errors, error_rate = stego.calculate_errors(message, compressed_message)
print(f"Errors: {errors}/{len(message)}")
bit_errors, bit_error_rate = stego.calculate_bit_errors(message, compressed_message)
print(f"Errors (bits): {bit_errors}/{len(message)*8}")
print(f"PSNR: {psnr_compressed:.2f} dB")
# Добавляем построение графиков
alpha_values = np.linspace(0.1, 1.0, 50) # 20 точек от 0.1 до 1.0
psnr_values = []
error_rates = []
for alpha in alpha_values:
stego = DCTSteganography(alpha=alpha)
stego_img, _, psnr = stego.embed("input.bmp", message)
cv2.imwrite("temp_stego.bmp", stego_img)
# Получаем значения для графиков
extracted = stego.extract("temp_stego.bmp")
errors, error_rate = stego.calculate_errors(message, extracted)
psnr_values.append(psnr)
error_rates.append(error_rate)
# Создаем графики
plt.figure(figsize=(12, 5))
# График PSNR
plt.subplot(1, 2, 1)
plt.plot(alpha_values, psnr_values, 'b-')
plt.xlabel('Alpha')
plt.ylabel('PSNR (dB)')
plt.title('Зависимость PSNR от alpha')
plt.grid(True)
# График ошибок
plt.subplot(1, 2, 2)
plt.plot(alpha_values, error_rates, 'r-')
plt.xlabel('Alpha')
plt.ylabel('Процент ошибок (%)')
plt.title('Зависимость ошибок извлечения от alpha')
plt.grid(True)
plt.tight_layout()
plt.savefig('analysis_plots.png')
plt.close()
сделай так, чтобы пользователь сам вводил сообщениеASSISTANT
Для того чтобы пользователь мог вводить сообщение, можем использовать встроенную функцию `input()` в Python. Я внесу изменения в основной блок кода, чтобы пользователь мог вводить сообщение перед его встраиванием в изображение. Вот как это можно сделать:
```python
if __name__ == "__main__":
stego = DCTSteganography(alpha=0.1)
# Читаем оригинальное изображение
original_image = cv2.imread("input.bmp")
if original_image is None:
print("Ошибка: не удалось прочитать входное изображение")
exit(1)
# Получаем сообщение от пользователя
message = input("Введите сообщение для встраивания: ")
# Встраиваем сообщение в изображение
stego_img, embedded_bits, embed_psnr = stego.embed("input.bmp", message)
cv2.imwrite("stego.bmp", stego_img)
print(f"Embedded bits count: {embedded_bits - 40} + 8 (/0)")
print(f"Secret length: {len(message)}")
# Проверка извлечения из оригинального стего-изображения
extracted_message = stego.extract("stego.bmp")
print(f"Extracted message: {extracted_message}")
errors, error_rate = stego.calculate_errors(message, extracted_message)
print(f"Errors: {errors}/{len(message)}")
print(f"PSNR: {embed_psnr:.2f} dB")
# Проверка устойчивости к JPEG-сжатию
print("\nCompressed JPEG:")
quality = 50
jpeg_path = "compressed.jpg"
cv2.imwrite(jpeg_path, stego_img, [cv2.IMWRITE_JPEG_QUALITY, quality])
compressed_image = cv2.imread(jpeg_path)
psnr_compressed = stego.calculate_psnr(stego_img, compressed_image)
compressed_message = stego.extract(jpeg_path)
print(f"Extracted message: {compressed_message}")
errors, error_rate = stego.calculate_errors(message, compressed_message)
print(f"Errors: {errors}/{len(message)}")
bit_errors, bit_error_rate = stego.calculate_bit_errors(message, compressed_message)
print(f"Errors (bits): {bit_errors}/{len(message)*8}")
print(f"PSNR: {psnr_compressed:.2f} dB")
# Добавляем построение графиков
alpha_values = np.linspace(0.1, 1.0, 50) # 20 точек от 0.1 до 1.0
psnr_values = []
error_rates = []
for alpha in alpha_values:
stego = DCTSteganography(alpha=alpha)
stego_img, _, psnr = stego.embed("input.bmp", message)
cv2.imwrite("temp_stego.bmp", stego_img)
# Получаем значения для графиков
extracted = stego.extract("temp_stego.bmp")
errors, error_rate = stego.calculate_errors(message, extracted)
psnr_values.append(psnr)
error_rates.append(error_rate)
# Создаем графики
plt.figure(figsize=(12, 5))
# График PSNR
plt.subplot(1, 2, 1)
plt.plot(alpha_values, psnr_values, 'b-')
plt.xlabel('Alpha')
plt.ylabel('PSNR (dB)')
plt.title('Зависимость PSNR от alpha')
plt.grid(True)
# График ошибок
plt.subplot(1, 2, 2)
plt.plot(alpha_values, error_rates, 'r-')
plt.xlabel('Alpha')
plt.ylabel('Процент ошибок (%)')
plt.title('Зависимость ошибок извлечения от alpha')
plt.grid(True)
plt.tight_layout()
plt.savefig('analysis_plots.png')
plt.close()
```
Теперь, когда вы запустите данный код, программа запросит у вас ввести сообщение, которое вы хотите встроить в изображение. После ввода сообщения, оно будет обработано и встроено, как и прежде.