# -*- coding: utf-8 -*- from io import BytesIO from traceback import format_exc import cv2 import requests from PIL import Image, ImageDraw, ImageFont import numpy as np from loguru import logger from enums.ExceptionEnum import ExceptionType from exception.CustomerException import ServiceException ''' 文字水印 ''' class TextWaterMark(): def __init__(self): self.color_dict = { # R G B # 网址查看：https://tool.oschina.net/commons?type=3 # 网址查看：http://tools.jb51.net/static/colorpicker/ 'white': (255, 255, 255, 255), 'black': (0, 0, 0, 255), 'gray': (205, 201, 201, 255), 'red': (255, 0, 0, 255), 'yellow': (255, 215, 0, 255), 'blue': (0, 0, 170, 255), 'purple': (205, 105, 201, 255), 'green': (0, 205, 0, 255) } self.position_list = [1, 2, 3, 4] """ 普通照片水印 params: image:图片 text:水印文字 position:水印位置 1：左上 2：右上 3：右下 4：左下 fontface: 字体 fontsize:字体大小 fontcolor:字体颜色 [white, black, gray, red, yellow, blue, purple, green] """ def common_water(self, image, text, position=1, fontface='../font/simsun.ttc', fontsize=20, fontcolor='black'): flag = False if isinstance(image, np.ndarray): # 判断是否OpenCV图片类型 flag = True image = Image.fromarray(cv2.cvtColor(image, cv2.COLOR_BGRA2RGBA)) if position not in self.position_list: position = 1 w, h = image.size[:2] keys = self.color_dict.keys() if fontcolor not in keys: fontcolor = 'black' color = self.color_dict[fontcolor] fnt = ImageFont.truetype(fontface, fontsize) im = image.convert('RGBA') mask = Image.new('RGBA', im.size, (0, 0, 0, 0)) d = ImageDraw.Draw(mask) size_w, size_h = d.textsize(text, font=fnt) if position == 1: weizhi = (w * 0.1, h * 0.1) elif position == 2: weizhi = (w * 0.9 - size_w, h * 0.1) elif position == 3: weizhi = (w * 0.9 - size_w, h * 0.9 - size_h) else: weizhi = (w * 0.1, h * 0.9 - size_h) # position 为左上角位置 d.text(weizhi, text, font=fnt, fill=color) out = Image.alpha_composite(im, mask) if flag: out = cv2.cvtColor(np.asarray(out), cv2.COLOR_BGRA2RGBA) return out """ 半透明水印，布满整张图，并且自动旋转45° params: image:图片 text:文字 fontsize:文字大小 """ def fill_water(self, image, text, fontsize): flag = False if isinstance(image, np.ndarray): # 判断是否OpenCV图片类型 flag = True image = Image.fromarray(cv2.cvtColor(image, cv2.COLOR_BGRA2RGBA)) font = ImageFont.truetype('../font/simsun.ttc', fontsize) # 添加背景 new_img = Image.new('RGBA', (image.size[0] * 3, image.size[1] * 3), (255, 255, 255, 255)) new_img.paste(image, image.size) # 添加水印 font_len = len(text) rgba_image = new_img.convert('RGBA') text_overlay = Image.new('RGBA', rgba_image.size, (0, 0, 0, 0)) image_draw = ImageDraw.Draw(text_overlay) for i in range(0, rgba_image.size[0], font_len * 40 + 100): for j in range(0, rgba_image.size[1], 200): # print(f'i:{i}, j:{j}, text:{text}, font:{font}') image_draw.text((i, j), text, font=font, fill=(0, 0, 0, 50)) text_overlay = text_overlay.rotate(-45) image_with_text = Image.alpha_composite(rgba_image, text_overlay) image_with_text = image_with_text.crop((image.size[0], image.size[1], image.size[0] * 2, image.size[1] * 2)) if flag: image_with_text = cv2.cvtColor(np.asarray(image_with_text), cv2.COLOR_BGRA2RGBA) return image_with_text class PictureWaterMark: __slots__ = ('logo', '__requestId') def __init__(self, logo=None, requestId=None): self.__requestId = requestId self.logo = logo if requestId is None: self.__requestId = '1' if logo is None: self.logo = cv2.imread("./image/logo.png", -1) # def common_water(self, image, logo): # width, height = image.shape[1], image.shape[0] # mark_width, mark_height = logo.shape[1], logo.shape[0] # rate = int(width * 0.2) / mark_width # logo_new = cv2.resize(logo, None, fx=rate, fy=rate, interpolation=cv2.INTER_NEAREST) # position = (int(width * 0.95 - logo_new.shape[1]), int(height * 0.95 - logo_new.shape[0])) # b = Image.new('RGBA', (width, height), (0, 0, 0, 0)) # 创建新图像：透明' # a = Image.fromarray(cv2.cvtColor(image, cv2.COLOR_BGR2RGB)) # watermark = Image.fromarray(cv2.cvtColor(logo_new, cv2.COLOR_BGRA2RGBA)) # # 图片旋转 # # watermark = watermark.rotate(45) # b.paste(a, (0, 0)) # b.paste(watermark, position, mask=watermark) # return cv2.cvtColor(np.asarray(b), cv2.COLOR_BGR2RGB) def common_water_1(self, image, logo, alpha=1): try: h, w = image.shape[0], image.shape[1] # if w >= h: rate = int(w * 0.1) / logo.shape[1] # else: # rate = int(h * 0.1) / logo.shape[0] mask = cv2.resize(logo, None, fx=rate, fy=rate, interpolation=cv2.INTER_NEAREST) mask_h, mask_w = mask.shape[0], mask.shape[1] mask_channels = cv2.split(mask) dst_channels = cv2.split(image) # b, g, r, a = cv2.split(mask) # 计算mask在图片的坐标 # if w >= h: ul_points = (int(h * 0.95) - mask_h, int(w - h * 0.05 - mask_w)) dr_points = (int(h * 0.95), int(w - h * 0.05)) # else: # ul_points = (int(h * 0.95) - mask_h, int(w - h * 0.05 - mask_w)) # dr_points = (int(h * 0.95), int(w - h * 0.05)) for i in range(3): dst_channels[i][ul_points[0]: dr_points[0], ul_points[1]: dr_points[1]] = dst_channels[i][ ul_points[0]: dr_points[0], ul_points[1]: dr_points[ 1]] * ( 255.0 - mask_channels[ 3] * alpha) / 255 dst_channels[i][ul_points[0]: dr_points[0], ul_points[1]: dr_points[1]] += np.array( mask_channels[i] * (mask_channels[3] * alpha / 255), dtype=np.uint8) dst_img = cv2.merge(dst_channels) return dst_img except Exception: logger.error("加水印异常:{}, requestId:{}", format_exc(), self.__requestId) return image def add_water_pic(image, logo, requestId, alpha=1): try: h, w = image.shape[0], image.shape[1] # if w >= h: rate = int(w * 0.1) / logo.shape[1] # else: # rate = int(h * 0.1) / logo.shape[0] mask = cv2.resize(logo, None, fx=rate, fy=rate, interpolation=cv2.INTER_NEAREST) mask_h, mask_w = mask.shape[0], mask.shape[1] mask_channels = cv2.split(mask) dst_channels = cv2.split(image) # b, g, r, a = cv2.split(mask) # 计算mask在图片的坐标 # if w >= h: ul_points = (int(h * 0.95) - mask_h, int(w - h * 0.05 - mask_w)) dr_points = (int(h * 0.95), int(w - h * 0.05)) # else: # ul_points = (int(h * 0.95) - mask_h, int(w - h * 0.05 - mask_w)) # dr_points = (int(h * 0.95), int(w - h * 0.05)) for i in range(3): dst_channels[i][ul_points[0]: dr_points[0], ul_points[1]: dr_points[1]] = dst_channels[i][ ul_points[0]: dr_points[0], ul_points[1]: dr_points[ 1]] * ( 255.0 - mask_channels[ 3] * alpha) / 255 dst_channels[i][ul_points[0]: dr_points[0], ul_points[1]: dr_points[1]] += np.array( mask_channels[i] * (mask_channels[3] * alpha / 255), dtype=np.uint8) dst_img = cv2.merge(dst_channels) return dst_img except Exception: logger.error("加水印异常:{}, requestId:{}", format_exc(), requestId) return image # 差值感知算法 def dHash(image): # 缩放9*8 image = cv2.resize(image, (9, 8), interpolation=cv2.INTER_CUBIC) # 转换灰度图 image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) # print(image.shape) hash = [] # 每行前一个像素大于后一个像素为1，相反为0，生成哈希 for i in range(8): for j in range(8): if image[i, j] > image[i, j + 1]: hash.append(1) else: hash.append(0) return hash # 计算汉明距离 def Hamming_distance(hash1, hash2): num = 0 for index in range(len(hash1)): if hash1[index] != hash2[index]: num += 1 return num def url2Array(url, enable_ex=True): try: response = requests.get(url) image = Image.open(BytesIO(response.content)) image1 = np.array(image) img_bgr = cv2.cvtColor(image1, cv2.COLOR_RGB2BGR) return img_bgr except Exception: logger.exception("url地址请求异常: {}", format_exc()) if enable_ex: raise ServiceException(ExceptionType.URL_ADDRESS_ACCESS_FAILED.value[0], ExceptionType.URL_ADDRESS_ACCESS_FAILED.value[1]) return None def url2Content(url): response = requests.get(url) return response.content def url2Image(url): response = requests.get(url) image = Image.open(BytesIO(response.content)) image1 = np.array(image) img_bgr = cv2.cvtColor(image1, cv2.COLOR_RGB2BGR) img = Image.fromarray(img_bgr) return img def url2Byte(url): response = requests.get(url) return BytesIO(response.content) def markRectangle(url, text, textCoordinate, imageLeftUpCoordinate, imageRightDownCoordinate, color): img = url2Array(url) # ( 蓝, 绿, 红) # 红色 (0, 0, 255) # 洋红色 (255, 0, 255) # 青色 (255, 255, 0) # 黑色 (0, 0, 0) # 蓝色 (255, 0, 0) # 绿色 (0, 255, 0) # 黄色 (0, 255, 255) # 不考虑 cv2.putText(img, text, textCoordinate, cv2.FONT_HERSHEY_SIMPLEX, 1.0, color, 1, cv2.LINE_AA) # rectangle 坐标的参数格式为左上角（x1, y1），右下角（x2, y2）, 颜色 , 粗细 cv2.rectangle(img, imageLeftUpCoordinate, imageRightDownCoordinate, color, 2) return img # def draw_painting_joint(img, xywh, score=0.5, color=None, # font={'line_thickness': None, 'boxLine_thickness': None, 'fontSize': None}): # imh, imw, imc = img.shape # tl = font['line_thickness'] or round(0.002 * (img.shape[0] + img.shape[1]) / 2) + 1 # box_tl = font['boxLine_thickness'] or round(0.002 * (img.shape[0] + img.shape[1]) / 2) + 1 # 根据图像的尺寸计算出适合用于绘制图像边框的线宽。 # c1, c2 = (int(xywh[0]), int(xywh[1])), (int(xywh[2]), int(xywh[3])) # cv2.rectangle(img, c1, c2, color, thickness=box_tl, lineType=cv2.LINE_AA) # # label = ' %.2f' % (score) # tf = max(tl, 1) # font thickness # fontScale = font['fontSize'] or tl * 0.33 # t_size = cv2.getTextSize(label, 0, fontScale=fontScale, thickness=tf)[0] # cv2.rectangle(img, (int(box[0]) + lw, int(box[1])), c2, color, -1, cv2.LINE_AA) # filled # cv2.putText(img, label, (c1[0] + lw, c1[1] - (lh - t_size[1]) // 2), 0, fontScale, [225, 255, 255], thickness=tf, # lineType=cv2.LINE_AA) # # print('#####line224 fontScale:',fontScale,' thickness:',tf,' line_thickness:',font['line_thickness'],' boxLine thickness:',box_tl) # return img def img_pad(img, size, pad_value=[114, 114, 114]): ###填充成固定尺寸 H, W, _ = img.shape r = max(H / size[0], W / size[1]) img_r = cv2.resize(img, (int(W / r), int(H / r))) tb = size[0] - img_r.shape[0] lr = size[1] - img_r.shape[1] top = int(tb / 2) bottom = tb - top left = int(lr / 2) right = lr - left pad_image = cv2.copyMakeBorder(img_r, top, bottom, left, right, cv2.BORDER_CONSTANT, value=pad_value) return pad_image, (top, left, r) def letterbox(img, new_shape=(640, 640), color=(114, 114, 114), auto=True, scaleFill=False, scaleup=True, stride=32): # 调整图像大小和填充图像，同时满足步幅多重约束 shape = img.shape[:2] # current shape [height, width] 当前形状 [高度、宽度] if isinstance(new_shape, int): new_shape = (new_shape, new_shape) # Scale ratio (new / old) 缩放比例（新/旧） r = min(new_shape[0] / shape[0], new_shape[1] / shape[1]) if not scaleup: # 仅缩减，不纵向扩展（为了更好的测试 mAP） r = min(r, 1.0) ratio = r, r # width, height ratios 宽度、高度比 new_unpad = int(round(shape[1] * r)), int(round(shape[0] * r)) dw, dh = new_shape[1] - new_unpad[0], new_shape[0] - new_unpad[1] # wh padding if auto: # 最小矩形 dw, dh = np.mod(dw, stride), np.mod(dh, stride) # wh padding elif scaleFill: # stretch dw, dh = 0.0, 0.0 new_unpad = (new_shape[1], new_shape[0]) ratio = new_shape[1] / shape[1], new_shape[0] / shape[0] # width, height ratios dw /= 2 # divide padding into 2 sides dh /= 2 if shape[::-1] != new_unpad: # resize img = cv2.resize(img, new_unpad, interpolation=cv2.INTER_LINEAR) top, bottom = int(round(dh - 0.1)), int(round(dh + 0.1)) left, right = int(round(dw - 0.1)), int(round(dw + 0.1)) img = cv2.copyMakeBorder(img, top, bottom, left, right, cv2.BORDER_CONSTANT, value=color) # add border return img, ratio, (dw, dh) # if __name__ == '__main__': # # img = cv2.imread("../test/a.jpg", -1) # # fontcolor = 'yellow' # # # # water = TextWaterMark() # # text = "hello world" # # # # # fill_img = water.common_water(img, text, position=4, fontface='../font/庞门正道标题体2.0增强版.ttf', fontsize=20, fontcolor='black') # # fill_img = water.fill_water(img, text, 20) # # # 一定要保存为png格式 # # cv2.imshow('result', fill_img) # # cv2.waitKey(111111110) # # print('finish') # pic = PictureWaterMark() # image = cv2.imread("a.jpg") # logo = cv2.imread("../image/logo.png", -1) # # print(image, logo) # start = time.time() # frame = pic.common_water(image, logo) # print(time.time() - start) # start1 = time.time() # frame1 = pic.common_water_1(image, logo) # # cv2.imwrite("watermarked.jpg", frame1) # print(time.time() - start1) # # cap = cv2.VideoCapture("../data/111111.mp4") # # logo = cv2.imread("../image/logo.png", -1) # # while True: # # is_opened, frame = cap.read() # # frame = pic.common_water(frame, logo) # # cv2.imshow('frame', frame) # # cv2.waitKey(1) # 等待输入任何按键 # # # 关闭 # # cap.release()