前言
在生产生活中，图像的合成实际上是一个比较常见的需求。而图像的合成往往也会在一些活动中，有着比较重要的作用和价值，例如在一个固定的图片模板上增加一些文字和一些验证码，形成一个新的图片，然后分享出去做一些营销活动；或者是在某些节假日，通过图像的合成，在头像上增加一些装饰物等。本文将会继续探索Serverless在图像处理中的应用，尤其是图像合成相关的应用。本文将会以阿里云函数计算以及华为云函数计算为例，为读者进行图像合成相关操作的探索。
为头像增加圣诞帽
为头像增加装饰是一个比较常见的小功能，本例子将会通过Serverless实现头像增加圣诞帽的功能。主要功能就是选择一个图片，然后函数部分进行图像的合成，这一部分主要是识别图片中的头像，然后在头像上增加圣诞帽的行为。
这一功能的主要做法就是，通过人工智能算法(此处是通过Dlib实现)，进行人脸的检测：
predictorPath = "shape_predictor_5_face_landmarks.dat"
predictor = dlib.shape_predictor(predictorPath)
detector = dlib.get_frontal_face_detector()
dets = detector(img, 1)
此处做法是，只检测一张脸，检测到即进行返回：
for d in dets:
x, y, w, h = d.left(), d.top(), d.right() - d.left(), d.bottom() - d.top()
# 关键点检测，5个关键点")
shape = predictor(img, d)
# 选取左右眼眼角的点")
point1 = shape.part(0)
point2 = shape.part(2)
# 求两点中心
eyes_center = ((point1.x + point2.x) // 2, (point1.y + point2.y) // 2)
# 根据人脸大小调整帽子大小
factor = 1.5
resizedHatH = int(round(rgbHat.shape[0] * w / rgbHat.shape[1] * factor))
resizedHatW = int(round(rgbHat.shape[1] * w / rgbHat.shape[1] * factor))
if resizedHatH > y:
resizedHatH = y - 1
# 根据人脸大小调整帽子大小
resizedHat = cv2.resize(rgbHat, (resizedHatW, resizedHatH))
# 用alpha通道作为mask
mask = cv2.resize(a, (resizedHatW, resizedHatH))
maskInv = cv2.bitwise_not(mask)
# 帽子相对与人脸框上线的偏移量
dh = 0
bgRoi = img[y + dh - resizedHatH:y + dh,
(eyes_center[0] - resizedHatW // 3):(eyes_center[0] + resizedHatW // 3 * 2)]
# 原图ROI中提取放帽子的区域
bgRoi = bgRoi.astype(float)
maskInv = cv2.merge((maskInv, maskInv, maskInv))
alpha = maskInv.astype(float) / 255
# 相乘之前保证两者大小一致（可能会由于四舍五入原因不一致）
alpha = cv2.resize(alpha, (bgRoi.shape[1], bgRoi.shape[0]))
bg = cv2.multiply(alpha, bgRoi)
bg = bg.astype('uint8')
# 提取帽子区域
hat = cv2.bitwise_and(resizedHat, cv2.bitwise_not(maskInv))
# 相加之前保证两者大小一致（可能会由于四舍五入原因不一致）")
hat = cv2.resize(hat, (bgRoi.shape[1], bgRoi.shape[0]))
# 两个ROI区域相加")
addHat = cv2.add(bg, hat)
# 把添加好帽子的区域放回原图
img[y + dh - resizedHatH:y + dh,
(eyes_center[0] - resizedHatW // 3):(eyes_center[0] + resizedHatW // 3 * 2)] = addHat
return img
整体代码通过Python Web框架进行封装：
# -*- coding: utf-8 -*-
import cv2
import dlib
import base64
import json
import uuid
import bottle
app = bottle.default_app()
predictorPath = "shape_predictor_5_face_landmarks.dat"
predictor = dlib.shape_predictor(predictorPath)
detector = dlib.get_frontal_face_detector()
return_msg = lambda error, msg: {
"uuid": str(uuid.uuid1()),
"error": error,
"message": msg
}
def addHat(img, hat_img):
# 分离rgba通道，合成rgb三通道帽子图，a通道后面做mask用
r, g, b, a = cv2.split(hat_img)
rgbHat = cv2.merge((r, g, b))
# dlib人脸关键点检测器,正脸检测
dets = detector(img, 1)
# 如果检测到人脸
if len(dets) > 0:
for d in dets:
x, y, w, h = d.left(), d.top(), d.right() - d.left(), d.bottom() - d.top()
# 关键点检测，5个关键点")
shape = predictor(img, d)
# 选取左右眼眼角的点")
point1 = shape.part(0)
point2 = shape.part(2)
# 求两点中心
eyes_center = ((point1.x + point2.x) // 2, (point1.y + point2.y) // 2)
# 根据人脸大小调整帽子大小
factor = 1.5
resizedHatH = int(round(rgbHat.shape[0] * w / rgbHat.shape[1] * factor))
resizedHatW = int(round(rgbHat.shape[1] * w / rgbHat.shape[1] * factor))
if resizedHatH > y:
resizedHatH = y - 1
# 根据人脸大小调整帽子大小
resizedHat = cv2.resize(rgbHat, (resizedHatW, resizedHatH))
# 用alpha通道作为mask
mask = cv2.resize(a, (resizedHatW, resizedHatH))
maskInv = cv2.bitwise_not(mask)
# 帽子相对与人脸框上线的偏移量
dh = 0
bgRoi = img[y + dh - resizedHatH:y + dh,
(eyes_center[0] - resizedHatW // 3):(eyes_center[0] + resizedHatW // 3 * 2)]
# 原图ROI中提取放帽子的区域
bgRoi = bgRoi.astype(float)
maskInv = cv2.merge((maskInv, maskInv, maskInv))
alpha = maskInv.astype(float) / 255
# 相乘之前保证两者大小一致（可能会由于四舍五入原因不一致）
alpha = cv2.resize(alpha, (bgRoi.shape[1], bgRoi.shape[0]))
bg = cv2.multiply(alpha, bgRoi)
bg = bg.astype('uint8')
# 提取帽子区域
hat = cv2.bitwise_and(resizedHat, cv2.bitwise_not(maskInv))
# 相加之前保证两者大小一致（可能会由于四舍五入原因不一致）")
hat = cv2.resize(hat, (bgRoi.shape[1], bgRoi.shape[0]))
# 两个ROI区域相加")
addHat = cv2.add(bg, hat)
# 把添加好帽子的区域放回原图
img[y + dh - resizedHatH:y + dh,
(eyes_center[0] - resizedHatW // 3):(eyes_center[0] + resizedHatW // 3 * 2)] = addHat
return img
@bottle.route('/add/hat', method='POST')
def addHatIndex():
try:
try:
# 将接收到的base64图像转为pic
postData = json.loads(bottle.request.body.read().decode("utf-8"))
imgData = base64.b64decode(postData.get("image", None))
with open('/tmp/picture.png', 'wb') as f:
f.write(imgData)
except Exception as e:
print(e)
return return_msg(True, "未能成功获取到头像，请检查pic参数是否为base64编码。")
try:
# 读取帽子素材以及用户头像
hatImg = cv2.imread("hat.png", -1)
userImg = cv2.imread("/tmp/picture.png")
output = addHat(userImg, hatImg)
cv2.imwrite("/tmp/output.jpg", output)
except Exception as e:
return return_msg(True, "图像添加圣诞帽失败，请检查图片中是否有圣诞帽或者图片是否可读。")
# 读取头像进行返回给用户，以Base64返回
with open("/tmp/output.jpg", "rb") as f:
base64Data = str(base64.b64encode(f.read()), encoding='utf-8')
return return_msg(False, {"picture": base64Data})
except Exception as e:
return return_msg(True, str(e))
同时声称依赖文件requirements.txt:
dlib==19.21.1
bottle==0.12.19
opencv-python==4.5.1.48
完成之后，我们需要进行依赖的安装，这里为了方便安装依赖，方便部署代码以及更新代码，可以使用Serverless devs开发者工具。首先进行Yaml的配置：
ServerlessBookChristmasHatDemo:
Component: fc
Provider: alibaba
Access: anycodes_release
Properties:
Region: cn-hongkong
Service:
Name: serverless-book-case
Description: Serverless 实践图书案例
Vpc:
SecurityGroupId: sg-j6c45wkv4vf4ghg104mc
VSwitchIds:
- vsw-j6c797ywau90y6y03ohbq
VpcId: vpc-j6c9lk4av0859r4e0tff7
Log: Auto
Nas: Auto
Function:
Name: ai-cv-image-christmas-hat
Description: 圣诞帽
CodeUri:
Src: ./src
Excludes:
- src/.fun
Handler: index.app
Environment:
- Key: PYTHONUSERBASE
Value: /mnt/auto/.fun/python
MemorySize: 3072
Runtime: python3
Timeout: 60
Triggers:
- Name: ImageAI
Type: HTTP
Parameters:
AuthType: ANONYMOUS
Methods:
- GET
- POST
- PUT
Domains:
- Domain: add-christmas-hat.cv.case.serverless.fun
Protocol:
- HTTP
Routes:
- Path: '/*'
配置完成，可以通过Serverless devs进行依赖安装：
s ServerlessBookChristmasHatDemo install docker
安装依赖之后，将安装好的依赖部署到NAS：
s ServerlessBookChristmasHatDemo nas sync ./src/.fun
最后将代码部署到线上：
s deploy
部署完成之后，可以通过Python语言进行代码的测试，测试脚本如下：
import base64
import urllib.request
import json
with open("test.png", 'rb') as f:
image = f.read()
image_base64 = str(base64.b64encode(image), encoding='utf-8')
url = "http://add-christmas-hat.cv.case.serverless.fun/add/hat"
response = urllib.request.urlopen(urllib.request.Request(url=url, data=json.dumps({
"image": image_base64
}).encode("utf-8"))).read().decode("utf-8")
responseAttr = json.loads(response)
if not responseAttr["error"]:
imgData = base64.b64decode(responseAttr['message']['picture'])
with open('./picture.png', 'wb') as f:
f.write(imgData)
else:
print(responseAttr['message'])
测试图片为：
执行之后的结果是：
至此，我们完成了一个基于阿里云函数计算以及AI的图像合成的服务。
为头像增加固定装饰
其实这个功能很简单，主要功能就是选择一个图片，上传自己的头像，然后函数部分进行图像的合成，这一部分并没有涉及到机器学习算法，仅仅是图像合成相关算法。
主要通过用户上传的图片，在指定位置增加预定图片作为装饰物进行添加，以华为云函数工作流为例，整个业务流程是：
- 将预定图片图片进行美化，此处仅是将其变成圆形：
def do_circle(base_pic):
icon_pic = Image.open(base_pic).convert("RGBA")
icon_pic = icon_pic.resize((500, 500), Image.ANTIALIAS)
icon_pic_x, icon_pic_y = icon_pic.size
temp_icon_pic = Image.new('RGBA', (icon_pic_x + 600, icon_pic_y + 600), (255, 255, 255))
temp_icon_pic.paste(icon_pic, (300, 300), icon_pic)
ima = temp_icon_pic.resize((200, 200), Image.ANTIALIAS)
size = ima.size
# 因为是要圆形，所以需要正方形的图片
r2 = min(size[0], size[1])
if size[0] != size[1]:
ima = ima.resize((r2, r2), Image.ANTIALIAS)
# 最后生成圆的半径
r3 = 60
imb = Image.new('RGBA', (r3 * 2, r3 * 2), (255, 255, 255, 0))
pima = ima.load() # 像素的访问对象
pimb = imb.load()
r = float(r2 / 2) # 圆心横坐标
for i in range(r2):
for j in range(r2):
lx = abs(i - r) # 到圆心距离的横坐标
ly = abs(j - r) # 到圆心距离的纵坐标
l = (pow(lx, 2) + pow(ly, 2)) ** 0.5 # 三角函数 半径
if l < r3:
pimb[i - (r - r3), j - (r - r3)] = pima[i, j]
return imb
- 添加该装饰到用户头像上：
def add_decorate():
try:
base_pic = "./code/decorate.png"
user_pic = Image.open("/tmp/picture.png").convert("RGBA")
temp_basee_user_pic = Image.new('RGBA', (440, 440), (255, 255, 255))
user_pic = user_pic.resize((400, 400), Image.ANTIALIAS)
temp_basee_user_pic.paste(user_pic, (20, 20))
temp_basee_user_pic.paste(do_circle(base_pic), (295, 295), do_circle(base_pic))
temp_basee_user_pic.save("/tmp/output.png")
return True
except Exception as e:
print(e)
return False
入口函数的业务逻辑为：
def handler(event, context):
jsonResponse = {
'statusCode': 200,
'isBase64Encoded': False,
'headers': {
"Content-type": "application/json"
},
}
# 将接收到的base64图像转为pic
imgData = base64.b64decode(json.loads(base64.b64decode(event["body"]))["image"])
with open('/tmp/picture.png', 'wb') as f:
f.write(imgData)
addResult = add_decorate()
if addResult:
with open("/tmp/output.png", "rb") as f:
base64Data = str(base64.b64encode(f.read()), encoding='utf-8')
jsonResponse['body'] = json.dumps({"picture": base64Data})
else:
jsonResponse['body'] = json.dumps({"error": True})
return jsonResponse
增加的装饰为（即在用户上传的头像右下角，增加下面的图片作为装饰）：
完成之后，安装相关依赖，然后打包上传到华为云的函数工作流：
新建函数，上传Zip代码包之后，还需要设置函数的超时时间和内存：
- 超时时间：15秒
- 内存：512MB
完成之后，可以创建触发器起：
创建触发器之后，在本地进行代码测试：
import base64
import urllib.request
import json
with open("test.png", 'rb') as f:
image = f.read()
image_base64 = str(base64.b64encode(image), encoding='utf-8')
url = "http://13535b378a964b96b8296808abda0bae.apig.cn-north-1.huaweicloudapis.com/photo_decorate"
response = urllib.request.urlopen(urllib.request.Request(url=url, data=json.dumps({
"image": image_base64
}).encode("utf-8"))).read().decode("utf-8")
responseAttr = json.loads(response)
imgData = base64.b64decode(responseAttr['picture'])
with open('./picture.png', 'wb') as f:
f.write(imgData)
所选择的测试图片为：
输出结果为：
至此，我们完成了在华为云创建一个图像合成的函数，可以为我们上传的头像在右下角增加一个小老鼠的装饰。
总结
Serverless架构毕竟是一个新的技术，或者说是一个比较新的Framework，如果刚开始就通过它来做一些很重的产品，可能会让学习者失去兴趣，但是前期可以通过Serverless架构不断的实现一些有趣的功能，小的应用，例如监控告警、图像识别、图像压缩、图像合成、文本摘要、关键词提取、简单的MapReduce等，通过这些小的应用，一方面可以让我们更加深入了解Serverless架构，另一方面也可以让我们对Serverless的实际应用和价值产生更大的信心。
传统情况下，我们如果要做这样的一个工具，可能需要一个服务器，哪怕没有人使用，也要有一台服务器苦苦支撑，那么仅仅就是一个Demo，也要无时无刻的支出成本，但是在Serverless架构下，通过Serverless弹性伸缩特点让我们不惧怕高并发，通过Serverless的按量付费模式，让我们不惧怕成本支出。
希望各位读者可以通过我的抛砖引玉，更加深入的了解Serverless架构。
