自个开发的图像处理,调用了百度的API

首先是图片处理主程序:

1
2
3
4
5
6
7
8
9
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
from imageprocess import *
import base64
import xlrd
import os


# 列出文件,不包含文件夹,返回列表
print('你好,欢迎来到图像处理应用!')
if os.path.isfile("user_id.xls")==True:
user_id=xlrd.open_workbook('user_id.xls')
table = user_id.sheet_by_name(u'user_id')
else:
print('您当前未录入信息,请录入。')
print('请在文件目录下找到录入信息的软件,录入后请重新打开此软件')
exit=input('')
""" 你的 APPID AK SK """
APP_ID = table.cell(0,0).value
API_KEY = table.cell(0,1).value
SECRET_KEY = table.cell(0,2).value

client = AipImageProcess(APP_ID, API_KEY, SECRET_KEY)
def dehaze():
""" 读取图片 """
def get_file_content(filePath):
with open(filePath, 'rb') as fp:
return fp.read()
image_lj=input('请输入你的图片路径:')
image = get_file_content(image_lj)

""" 调用图像去雾 """
base64_image_dict=client.dehaze(image);

base64_image=base64_image_dict['image']
imgdata = base64.b64decode(base64_image)
# 将二进制数据装换为图片

with open("./OK.png", "wb") as f2:
f2.write(imgdata)
exit=input('你好,已处理完成,图片放置在当前目录下,名字为OK.png.请输入回车退出')
'''
---------------一条优美的分割线-------------------------------
'''
def clourful():
""" 读取图片 """
def get_file_content(filePath):
with open(filePath, 'rb') as fp:
return fp.read()
image_lj=input('请输入你的图片路径:')
image = get_file_content(image_lj)

base64_image_dict=client.contrastEnhance(image);

base64_image=base64_image_dict['image']
imgdata = base64.b64decode(base64_image)
# 将二进制数据装换为图片


with open("./OK.png", "wb") as f2:
f2.write(imgdata)
exit=input('你好,已处理完成,图片放置在当前目录下,名字为OK.png.请输入回车退出')
'''
---------------一条优美的分割线-------------------------------
'''
def imageQualityEnhance():
""" 读取图片 """
def get_file_content(filePath):
with open(filePath, 'rb') as fp:
return fp.read()
image_lj=input('请输入你的图片路径:')
image = get_file_content(image_lj)

base64_image_dict=client.imageQualityEnhance(image);

base64_image=base64_image_dict['image']
imgdata = base64.b64decode(base64_image)
# 将二进制数据装换为图片


with open("./OK.png", "wb") as f2:
f2.write(imgdata)
exit=input('你好,已处理完成,图片放置在当前目录下,名字为OK.png.请输入回车退出')
choose=str(input('你要用哪个服务?(1.图像去雾 2.图像增彩 3.图像放大 ;输入数字)'))
if choose=='1':
dehaze()
if choose=='2':
clourful()
if choose=='3':
imageQualityEnhance()

然后是百度的SDK:

imageprocess.py:

1
2
3
4
5
6
7
8
9
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
# -*- coding: utf-8 -*-

"""
图像处理
"""

import re
import sys
import math
import time
from base import AipBase
from base import base64
from base import json
from base import urlencode
from base import quote

class AipImageProcess(AipBase):

"""
图像处理
"""

__imageQualityEnhanceUrl = 'https://aip.baidubce.com/rest/2.0/image-process/v1/image_quality_enhance'

__dehazeUrl = 'https://aip.baidubce.com/rest/2.0/image-process/v1/dehaze'

__contrastEnhanceUrl = 'https://aip.baidubce.com/rest/2.0/image-process/v1/contrast_enhance'


def imageQualityEnhance(self, image, options=None):
"""
图像无损放大
"""
options = options or {}

data = {}
data['image'] = base64.b64encode(image).decode()

data.update(options)

return self._request(self.__imageQualityEnhanceUrl, data)

def dehaze(self, image, options=None):
"""
图像去雾
"""
options = options or {}

data = {}
data['image'] = base64.b64encode(image).decode()

data.update(options)

return self._request(self.__dehazeUrl, data)

def contrastEnhance(self, image, options=None):
"""
图像对比度增强
"""
options = options or {}

data = {}
data['image'] = base64.b64encode(image).decode()

data.update(options)

return self._request(self.__contrastEnhanceUrl, data)

BASE64:

1
2
3
4
5
6
7
8
9
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
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
# -*- coding: utf-8 -*-

"""
AipBase
"""
import hmac
import json
import hashlib
import datetime
import base64
import time
import sys
import requests
requests.packages.urllib3.disable_warnings()


if sys.version_info.major == 2:
from urllib import urlencode
from urllib import quote
from urlparse import urlparse
else:
from urllib.parse import urlencode
from urllib.parse import quote
from urllib.parse import urlparse

class AipBase(object):
"""
AipBase
"""

__accessTokenUrl = 'https://aip.baidubce.com/oauth/2.0/token'

__reportUrl = 'https://aip.baidubce.com/rpc/2.0/feedback/v1/report'

__scope = 'brain_all_scope'

def __init__(self, appId, apiKey, secretKey):
"""
AipBase(appId, apiKey, secretKey)
"""

self._appId = appId.strip()
self._apiKey = apiKey.strip()
self._secretKey = secretKey.strip()
self._authObj = {}
self._isCloudUser = None
self.__client = requests
self.__connectTimeout = 60.0
self.__socketTimeout = 60.0
self._proxies = {}
self.__version = '2_2_17'

def getVersion(self):
"""
version
"""
return self.__version

def setConnectionTimeoutInMillis(self, ms):
"""
setConnectionTimeoutInMillis
"""

self.__connectTimeout = ms / 1000.0

def setSocketTimeoutInMillis(self, ms):
"""
setSocketTimeoutInMillis
"""

self.__socketTimeout = ms / 1000.0

def setProxies(self, proxies):
"""
proxies
"""

self._proxies = proxies

def _request(self, url, data, headers=None):
"""
self._request('', {})
"""
try:
result = self._validate(url, data)
if result != True:
return result

authObj = self._auth()
params = self._getParams(authObj)

data = self._proccessRequest(url, params, data, headers)
headers = self._getAuthHeaders('POST', url, params, headers)
response = self.__client.post(url, data=data, params=params,
headers=headers, verify=False, timeout=(
self.__connectTimeout,
self.__socketTimeout,
), proxies=self._proxies
)
obj = self._proccessResult(response.content)

if not self._isCloudUser and obj.get('error_code', '') == 110:
authObj = self._auth(True)
params = self._getParams(authObj)
response = self.__client.post(url, data=data, params=params,
headers=headers, verify=False, timeout=(
self.__connectTimeout,
self.__socketTimeout,
), proxies=self._proxies
)
obj = self._proccessResult(response.content)
except (requests.exceptions.ReadTimeout, requests.exceptions.ConnectTimeout) as e:
return {
'error_code': 'SDK108',
'error_msg': 'connection or read data timeout',
}

return obj

def _validate(self, url, data):
"""
validate
"""

return True

def _proccessRequest(self, url, params, data, headers):
"""
参数处理
"""

params['aipSdk'] = 'python'
params['aipVersion'] = self.__version

return data

def _proccessResult(self, content):
"""
formate result
"""

if sys.version_info.major == 2:
return json.loads(content) or {}
else:
return json.loads(content.decode()) or {}

def _auth(self, refresh=False):
"""
api access auth
"""

#未过期
if not refresh:
tm = self._authObj.get('time', 0) + int(self._authObj.get('expires_in', 0)) - 30
if tm > int(time.time()):
return self._authObj

obj = self.__client.get(self.__accessTokenUrl, verify=False, params={
'grant_type': 'client_credentials',
'client_id': self._apiKey,
'client_secret': self._secretKey,
}, timeout=(
self.__connectTimeout,
self.__socketTimeout,
), proxies=self._proxies).json()

self._isCloudUser = not self._isPermission(obj)
obj['time'] = int(time.time())
self._authObj = obj

return obj

def _isPermission(self, authObj):
"""
check whether permission
"""

scopes = authObj.get('scope', '')

return self.__scope in scopes.split(' ')

def _getParams(self, authObj):
"""
api request http url params
"""

params = {}

if self._isCloudUser == False:
params['access_token'] = authObj['access_token']

return params

def _getAuthHeaders(self, method, url, params=None, headers=None):
"""
api request http headers
"""

headers = headers or {}
params = params or {}

if self._isCloudUser == False:
return headers

urlResult = urlparse(url)
for kv in urlResult.query.strip().split('&'):
if kv:
k, v = kv.split('=')
params[k] = v

# UTC timestamp
timestamp = datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')
headers['Host'] = urlResult.hostname
headers['x-bce-date'] = timestamp
version, expire = '1', '1800'

# 1 Generate SigningKey
val = "bce-auth-v%s/%s/%s/%s" % (version, self._apiKey, timestamp, expire)
signingKey = hmac.new(self._secretKey.encode('utf-8'), val.encode('utf-8'),
hashlib.sha256
).hexdigest()

# 2 Generate CanonicalRequest
# 2.1 Genrate CanonicalURI
canonicalUri = quote(urlResult.path)
# 2.2 Generate CanonicalURI: not used here
# 2.3 Generate CanonicalHeaders: only include host here

canonicalHeaders = []
for header, val in headers.items():
canonicalHeaders.append(
'%s:%s' % (
quote(header.strip(), '').lower(),
quote(val.strip(), '')
)
)
canonicalHeaders = '\n'.join(sorted(canonicalHeaders))

# 2.4 Generate CanonicalRequest
canonicalRequest = '%s\n%s\n%s\n%s' % (
method.upper(),
canonicalUri,
'&'.join(sorted(urlencode(params).split('&'))),
canonicalHeaders
)

# 3 Generate Final Signature
signature = hmac.new(signingKey.encode('utf-8'), canonicalRequest.encode('utf-8'),
hashlib.sha256
).hexdigest()

headers['authorization'] = 'bce-auth-v%s/%s/%s/%s/%s/%s' % (
version,
self._apiKey,
timestamp,
expire,
';'.join(headers.keys()).lower(),
signature
)

return headers

def report(self, feedback):
"""
数据反馈
"""

data = {}
data['feedback'] = feedback

return self._request(self.__reportUrl, data)

def post(self, url, data, headers=None):
"""
self.post('', {})
"""
return self._request(url, data, headers)

当然,我会重点讲我的主程序,为什么呢?因为百度那些我都不懂

这是流程图

1
2
3
4
5
6
7
8
A[用户输入] -->|数据存到变量| B(选择)
B --> C{判断条件}
C -->|假如选一| D[去雾]
C -->|假如选二| E[图片增彩]
C -->|假如选三| F[无损放大]
D --> G[输出结果]
E --> G[输出结果]
F --> G[输出结果]

看懂了撒?

看不懂没关系,文字版的来了:

  1. 用户输入选择
  2. 储存到变量
  3. 选择
  4. 判断
  5. 选择对应函数
  6. 输出处理好的文件

这个是成品:

成品

效果放送

这是未处理的:
test.jpg
这是处理过的:
OK.png
(我放的那张测试图片好像雾大了点,所以才是黑黑的)