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
| # -*- coding: utf-8 -*-
|
| # !/usr/bin/python
| # -*- coding: utf-8 -*
|
| from mysqlhelper import MySqLHelper
| import json
|
| db = MySqLHelper()
|
|
| def bt_dt_save(data):
| try:
| obj = data[-1]
| sql2 = 'insert into dt_radar_boat_df_msg (object_id,lng,lat,height,minOfYear,second,direction,deviceTime,fusDevID,fps,cellid,msgCnt,nofobjects,dataSource,msgdata,isInside,serverTime,create_time) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,now(),now()) on duplicate key update msgdata=values(msgdata), lng=values(lng), lat=values(lat), direction=values(direction), height=values(height), isInside=values(isInside), update_time=now()'
| ret = db.insertone(sql2, (
| obj['object_id'], "{:.8f}".format(obj['lng']), "{:.8f}".format(obj['lat']), "{:.2f}".format(obj['height']),
| obj['minOfYear'], obj['second'], "{:.2f}".format(obj['direction']), obj['time'], obj['speed'], obj['fps'],
| obj['cellid'], obj['msgCnt'], obj['nofobjects'], obj['dataSource'], json.dumps(data), obj['isInside']))
| print(f"数据库保存成功 {ret}:！")
| except Exception as e:
| print(e)
| pass
|
|
| def bt_df_tracks_save(data):
| try:
| obj = data[-1]
| sql2 = 'insert into dt_radar_boat_tracks_df (object_id,date_time,lng,lat,height,speed,direction,create_time) VALUES (%s,%s,%s,%s,%s,%s,%s,now())'
| ret = db.insertone(sql2, (
| obj['object_id'], obj['time'], obj['lng'], obj['lat'], obj['height'], obj['speed'], obj['direction']))
| print(f"数据库dt_radar_boat_tracks_df保存成功 {ret}:！")
| except Exception as e:
| print(e)
| pass
|
|
| def bt_old_dt_clean():
| try:
| sql2 = 'update dt_radar_boat_df_msg set isInside = -1 where isInside =1'
| ret = db.update(sql2)
| print(f"数据库保存成功 {ret}:！")
| except Exception as e:
| print(e)
| pass
|
|
| def bt_cy_save(x):
| try:
| obj = x[-1] # tbname = tbnames[obj['region']] if obj['type'] == 'boat' else 'dt_radar_uav_tracks'
| if obj['region'] == 0:
| return 0
| tbnames = ['dt_radar_boat_df_msg', 'dt_radar_boat_cy_msg','dt_radar_boat_df_msg','dt_radar_boat_dy_msg']
| #if not obj['is_track']:
| sql2 = f'''insert into {tbnames[obj['region']]} (object_id,lng,lat,height,direction,deviceTime,fusDevID,msgdata,isInside,serverTime,create_time) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,now(),now()) on duplicate key update msgdata=values(msgdata), lng=values(lng), lat=values(lat), direction=values(direction), height=values(height), isInside=values(isInside), update_time=now()'''
| ret = db.insertone(sql2, (obj['object_id'], "{:.8f}".format(obj['lng']), "{:.8f}".format(obj['lat']), "{:.2f}".format(obj['height']),"{:.2f}".format(obj['direction']), obj['time'], obj['speed'], json.dumps(x), obj['isInside']))
| print(f"数据库dt_radar_boat_cy_msg保存成功：{ret}！")
| except Exception as e:
| print(e)
| pass
|
| def bt_tracks_save(x):
| try:
| if x['region'] == 0:
| return 0
| tbnames = ['dt_radar_uav_tracks', 'dt_radar_boat_tracks_cy','dt_radar_boat_tracks_df','dt_radar_boat_tracks_dy']
| sql = f'''INSERT INTO {tbnames[x["region"]]} (object_id, date_time, lng, lat, height, speed, direction) VALUES (%s,%s,%s,%s,%s,%s,%s)'''
| values = (
| x['object_id'],
| x['date_time'],
| round(x['lng'], 8),
| round(x['lat'], 8),
| round(x['height'], 2),
| round(x['speed'], 2),
| round(x['direction'], 2)
| )
| db.insertone(sql, values)
| except Exception as e:
| print(f"数据写入失败: {e}")
| pass
|
| def bt_tracks_batch_save(region, records):
| if not records:
| return
| try:
| tbnames = ['dt_radar_uav_tracks', 'dt_radar_boat_tracks_cy', 'dt_radar_boat_tracks_df', 'dt_radar_boat_tracks_dy']
| #tbname = tbnames[obj['region']] if obj['type'] == 'boat' else 'dt_radar_uav_tracks'
| table = tbnames[region]
| sql = f'''
| INSERT INTO {table}
| (object_id, date_time, lng, lat, height, speed, direction)
| VALUES
| (%s, %s, %s, %s, %s, %s, %s)
| '''
| values = []
| for x in records:
| values.append((
| x['object_id'],
| x['date_time'],
| round(x['lng'], 8),
| round(x['lat'], 8),
| round(x['height'], 2),
| round(x['speed'], 2),
| round(x['direction'], 2),
| ))
|
| # 使用连接池或全局连接对象
| db.insertmany(sql, values)
| except Exception as ex:
| print(f"[ERROR] 批量插入非预期异常: {ex}")
|
|
| def bt_cy_tracks_save(data):
| try:
| obj = data[-1]
| sql2 = 'insert into dt_radar_boat_tracks_cy (object_id,date_time,lng,lat,height,speed,direction,create_time) VALUES (%s,%s,%s,%s,%s,%s,%s,now())'
| ret = db.insertone(sql2, (
| obj['object_id'], obj['time'], obj['lng'], obj['lat'], obj['height'], obj['speed'], obj['direction']))
| except Exception as e:
| print(e)
| pass
|
|
| def bt_old_cy_clean():
| try:
| sql2 = 'update dt_radar_boat_cy_msg set isInside = -1 where isInside =1'
| ret = db.update(sql2)
| print(f"数据库保存成功 {ret}:！")
| except Exception as e:
| print(e)
| pass
|
|
| def uav_zte_dt_save(data):
| try:
| obj = data[-1]
| sql2 = 'insert into dt_radar_uav_zte_msg (object_id,lng,lat,height,minOfYear,second,direction,deviceTime,fusDevID,fps,cellid,msgCnt,nofobjects,dataSource,msgdata,isInside,serverTime,create_time) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,now(),now()) on duplicate key update msgdata=values(msgdata), lng=values(lng), lat=values(lat), direction=values(direction), height=values(height), isInside=values(isInside), update_time=now()'
| ret = db.insertone(sql2, (
| obj['object_id'], "{:.8f}".format(obj['lng']), "{:.8f}".format(obj['lat']), "{:.2f}".format(obj['height']),
| obj['minOfYear'], obj['second'], "{:.2f}".format(obj['direction']), obj['time'], obj['speed'], obj['fps'],
| obj['cellid'], obj['msgCnt'], obj['nofobjects'], obj['dataSource'], json.dumps(data), obj['isInside']))
| # print(sql2)
| # print(obj)
| print(f"数据库更新成功 {ret}:！")
| except Exception as e:
| print(e)
| pass
|
|
| def uav_zte_old_dt_clean():
| try:
| sql2 = 'update dt_radar_uav_zte_msg set isInside = -1 where isInside =1'
| ret = db.update(sql2)
| print(f"数据库保存成功 {ret}:！")
| except Exception as e:
| print(e)
| pass
|
|
| def uav_hw_dt_save(object_id, track):
| try:
| obj = track[-1]
| sql2 = 'insert into dt_radar_uav_hw_msg (object_id,lng,lat,isInside,msgCnt,height,fusDevID,direction,second,deviceTime,msgdata,serverTime,create_time) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,now(),now()) on duplicate key update msgdata=values(msgdata), lng=values(lng), lat=values(lat), height=values(height), fusDevID=values(fusDevID),direction=values(direction), msgCnt=values(msgCnt),update_time=now()'
| ret = db.insertone(sql2, (
| object_id, "{:.8f}".format(obj['lng']), "{:.8f}".format(obj['lat']), obj['isInside'], obj['batchSerial'],
| "{:.2f}".format(obj['height']), "{:.2f}".format(obj['velocityH']), obj['headH'], obj['second'],
| obj['deviceTime'], json.dumps(track)))
| print(f"数据库保存成功 {ret}:！")
| except Exception as e:
| print(e)
| pass
|
|
| def uav_hw_old_dt_clean():
| try:
| sql2 = 'update dt_radar_uav_hw_msg set isInside = -1 where isInside =1'
| ret = db.update(sql2)
| print(f"数据库保存成功 {ret}:！")
| except Exception as e:
| print(e)
| pass
|
|
| def main():
| print()
|
|
| if __name__ == '__main__':
| # print(time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(1712900732)))
| main()
|
|
