-
Notifications
You must be signed in to change notification settings - Fork 0
/
database.py
117 lines (108 loc) · 4.45 KB
/
database.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
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
import pymysql.cursors
import cv2
import numpy as np
import json
from model import *
class MyDatabase:
__table_image = 'image'
__col_id = 'id'
__col_name = 'name'
__col_path = 'path'
__col_trainKP = 'trainKP'
__col__trainDesc = 'trainDesc'
def __init__(self):
self.connection = DBUtils.getConnection()
self.connection.autocommit(True)
def open(self):
if(not self.connection.open):
self.connection = DBUtils.getConnection()
def getAllData(self):
sql = 'SELECT * FROM ' + self.__table_image
listImageData = []
try:
cursor = self.connection.cursor()
cursor.execute(sql)
rows = cursor.fetchall()
for row in rows:
imageData = ImageData()
imageData.id = row[self.__col_id]
imageData.name = row[self.__col_name]
imageData.path = row[self.__col_path]
imageData.image = cv2.imread(imageData.path, 0)
strTrainKP = row[self.__col_trainKP]
strTrainDesc = row[self.__col__trainDesc]
# convert Json to list
cvtDataKP = json.loads(strTrainKP)
cvtDataDesc = json.loads(strTrainDesc)
dataKP = []
for point in cvtDataKP:
temp = cv2.KeyPoint(x=point[0][0], y=point[0][1], _size=point[1],
_angle=point[2], _response=point[3], _octave=point[4], _class_id=point[5])
dataKP.append(temp)
imageData.trainKP = dataKP
imageData.trainDesc = np.asarray(cvtDataDesc, dtype=np.float32)
listImageData.append(imageData)
finally:
print("Getted")
return listImageData
def getDataByNamesIn(self, names):
sql = 'SELECT * FROM ' + self.__table_image + ' WHERE name IN(' + ','.join(names) + ')'
print(sql)
listImageData = []
try:
cursor = self.connection.cursor()
cursor.execute(sql)
print(cursor.description)
rows = cursor.fetchall()
for row in rows:
imageData = ImageData()
imageData.name = row[self.__col_name]
imageData.path = row[self.__col_path]
imageData.image = cv2.imread(imageData.path, 0)
strTrainKP = row[self.__col_trainKP]
strTrainDesc = row[self.__col__trainDesc]
# convert Json to list
cvtDataKP = json.loads(strTrainKP)
cvtDataDesc = json.loads(strTrainDesc)
dataKP = []
for point in cvtDataKP:
temp = cv2.KeyPoint(x=point[0][0], y=point[0][1], _size=point[1],
_angle=point[2], _response=point[3], _octave=point[4], _class_id=point[5])
dataKP.append(temp)
imageData.trainKP = dataKP
imageData.trainDesc = np.asarray(cvtDataDesc, dtype=np.float32)
listImageData.append(imageData)
finally:
print("Getted")
return listImageData
def insertData(self, imageData):
sql = 'INSERT INTO ' + self.__table_image + ' VALUE (DEFAULT, %s, %s, %s, %s) '
try:
cursor = self.connection.cursor()
name = imageData.name
path = imageData.path
cvtDataKP = []
for point in imageData.trainKP:
temp = (point.pt, point.size, point.angle,
point.response, point.octave, point.class_id)
cvtDataKP.append(temp)
strTrainKP = json.dumps(cvtDataKP)
strTrainDesc = json.dumps(imageData.trainDesc.tolist())
cursor.execute(sql, (name, path, strTrainKP, strTrainDesc))
self.connection.commit()
finally:
print("Inserted")
def close(self):
if(self.connection.open):
self.connection.close()
class DBUtils:
@staticmethod
def getConnection():
connection = pymysql.connect(host='XXXX',
port=3306,
user='XXXX',
password='XXXX',
db='XXXX',
charset='utf8',
cursorclass=pymysql.cursors.DictCursor)
return connection