You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
271 lines
9.2 KiB
271 lines
9.2 KiB
import mysql.connector
|
|
from mysql.connector import Error
|
|
import json
|
|
import os
|
|
from PyQt5.QtGui import QColor
|
|
|
|
def load_db_config():
|
|
# 获取当前文件的目录
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
# 构建配置文件的完整路径
|
|
config_path = os.path.join(current_dir, 'db_config.json')
|
|
with open(config_path, 'r') as config_file:
|
|
config = json.load(config_file)
|
|
return config
|
|
|
|
def create_connection():
|
|
"""进行数据库连接"""
|
|
connection = None
|
|
try:
|
|
config = load_db_config()
|
|
connection = mysql.connector.connect(
|
|
host=config['host'], # 数据库主机地址
|
|
user=config['user'], # 数据库用户名
|
|
password=config['password'], # 数据库密码
|
|
database=config['database'] # 数据库名称
|
|
)
|
|
if connection.is_connected():
|
|
print("Connected to MySQL database")
|
|
except Error as e:
|
|
print(f"Error: '{e}' occurred")
|
|
return connection
|
|
|
|
def close_connection(connection):
|
|
"""关闭连接"""
|
|
if connection.is_connected():
|
|
connection.close()
|
|
print("MySQL connection is closed")
|
|
|
|
def query_user(username, password):
|
|
"""查询用户信息"""
|
|
connection = create_connection()
|
|
user_exists = False
|
|
try:
|
|
if connection.is_connected():
|
|
cursor = connection.cursor(dictionary=True)
|
|
query = "SELECT * FROM users WHERE username = %s AND password = %s"
|
|
cursor.execute(query, (username, password))
|
|
result = cursor.fetchone()
|
|
if result:
|
|
user_exists = True
|
|
cursor.close()
|
|
except Error as e:
|
|
print(f"Error: '{e}' occurred")
|
|
finally:
|
|
if connection.is_connected():
|
|
close_connection(connection)
|
|
return user_exists
|
|
|
|
def register_user(username, password):
|
|
"""注册新用户"""
|
|
connection = create_connection()
|
|
success = False
|
|
try:
|
|
if connection.is_connected():
|
|
cursor = connection.cursor()
|
|
query = "INSERT INTO users (username, password) VALUES (%s, %s)"
|
|
cursor.execute(query, (username, password))
|
|
connection.commit()
|
|
success = True
|
|
cursor.close()
|
|
except Error as e:
|
|
print(f"Error: '{e}' occurred")
|
|
finally:
|
|
if connection.is_connected():
|
|
close_connection(connection)
|
|
return success
|
|
|
|
|
|
def save_danger_zone(area):
|
|
"""保存危险区域到数据库，避免重复"""
|
|
if danger_zone_exists(area):
|
|
print("危险区域已存在，跳过存储。")
|
|
return False
|
|
connection = create_connection()
|
|
success = False
|
|
try:
|
|
if connection.is_connected():
|
|
cursor = connection.cursor(dictionary=True)
|
|
|
|
# 标准化坐标
|
|
if area['type'] == 'circle':
|
|
coordinates = json.dumps({
|
|
'center': [round(c, 6) for c in area.get('center', (0, 0))],
|
|
'radius': round(area.get('radius', 50), 2)
|
|
})
|
|
elif area['type'] == 'rectangle':
|
|
coordinates = json.dumps({
|
|
'rect': [round(r, 6) for r in area.get('rect', (0, 0, 100, 100))]
|
|
})
|
|
elif area['type'] == 'polygon':
|
|
coordinates = json.dumps({
|
|
'points': [[round(p[0], 6), round(p[1], 6)] for p in area.get('points', [])]
|
|
})
|
|
else:
|
|
return False
|
|
|
|
# 检查是否已存在
|
|
check_query = """
|
|
SELECT id FROM danger_zones WHERE type = %s AND coordinates = %s
|
|
"""
|
|
cursor.execute(check_query, (area['type'], coordinates))
|
|
result = cursor.fetchone()
|
|
if result:
|
|
print("Danger zone already exists in the database.")
|
|
return False
|
|
|
|
# 插入新记录
|
|
insert_query = """
|
|
INSERT INTO danger_zones (type, coordinates, color)
|
|
VALUES (%s, %s, %s)
|
|
"""
|
|
color = area['color'].name() # 假设QColor对象
|
|
cursor.execute(insert_query, (area['type'], coordinates, color))
|
|
connection.commit()
|
|
success = True
|
|
cursor.close()
|
|
except Error as e:
|
|
print(f"Error: '{e}' occurred")
|
|
finally:
|
|
if connection.is_connected():
|
|
close_connection(connection)
|
|
return success
|
|
|
|
def load_danger_zones():
|
|
"""从数据库加载危险区域信息"""
|
|
connection = create_connection()
|
|
danger_zones = []
|
|
try:
|
|
if connection.is_connected():
|
|
cursor = connection.cursor(dictionary=True)
|
|
query = "SELECT type, coordinates, color FROM danger_zones"
|
|
cursor.execute(query)
|
|
results = cursor.fetchall()
|
|
for row in results:
|
|
coordinates = json.loads(row['coordinates'])
|
|
if row['type'] == 'circle':
|
|
zone = {
|
|
'type': row['type'],
|
|
'center': coordinates['center'],
|
|
'radius': coordinates['radius'],
|
|
'color': QColor(row['color'])
|
|
}
|
|
elif row['type'] == 'rectangle':
|
|
zone = {
|
|
'type': row['type'],
|
|
'rect': coordinates['rect'],
|
|
'color': QColor(row['color'])
|
|
}
|
|
elif row['type'] == 'polygon':
|
|
zone = {
|
|
'type': row['type'],
|
|
'points': coordinates['points'],
|
|
'color': QColor(row['color'])
|
|
}
|
|
danger_zones.append(zone)
|
|
cursor.close()
|
|
except Error as e:
|
|
print(f"Error: '{e}' occurred")
|
|
finally:
|
|
if connection.is_connected():
|
|
close_connection(connection)
|
|
return danger_zones
|
|
|
|
def clear_danger_zones():
|
|
"""删除数据库中的所有危险区域数据"""
|
|
connection = create_connection()
|
|
print("clear_danger_zones called")
|
|
try:
|
|
if connection.is_connected():
|
|
cursor = connection.cursor()
|
|
query = "DELETE FROM danger_zones"
|
|
cursor.execute(query)
|
|
connection.commit()
|
|
cursor.close()
|
|
except Error as e:
|
|
print(f"Error: '{e}' occurred")
|
|
finally:
|
|
if connection.is_connected():
|
|
close_connection(connection)
|
|
|
|
def save_path(drone_id, path):
|
|
"""保存路径信息到数据库"""
|
|
connection = create_connection()
|
|
success = False
|
|
print("save_path called")
|
|
try:
|
|
if connection.is_connected():
|
|
cursor = connection.cursor()
|
|
query = """
|
|
INSERT INTO paths (drone_id, path)
|
|
VALUES (%s, %s)
|
|
"""
|
|
path_str = json.dumps(path) # 将路径转换为JSON字符串
|
|
cursor.execute(query, (drone_id, path_str))
|
|
connection.commit()
|
|
success = True
|
|
cursor.close()
|
|
except Error as e:
|
|
print(f"Error: '{e}' occurred")
|
|
finally:
|
|
if connection.is_connected():
|
|
close_connection(connection)
|
|
return success
|
|
|
|
def clear_paths(drone_id):
|
|
"""删除数据库中与特定无人机ID相关的路径数据"""
|
|
connection = create_connection()
|
|
try:
|
|
if connection.is_connected():
|
|
cursor = connection.cursor()
|
|
query = "DELETE FROM paths WHERE drone_id = %s"
|
|
cursor.execute(query, (drone_id,))
|
|
connection.commit()
|
|
cursor.close()
|
|
except Error as e:
|
|
print(f"Error: '{e}' occurred")
|
|
finally:
|
|
if connection.is_connected():
|
|
close_connection(connection)
|
|
|
|
def danger_zone_exists(area):
|
|
"""检查数据库中是否存在相同的危险区域"""
|
|
connection = create_connection()
|
|
exists = False
|
|
try:
|
|
if connection.is_connected():
|
|
cursor = connection.cursor(dictionary=True)
|
|
|
|
# 标准化坐标
|
|
if area['type'] == 'circle':
|
|
coordinates = json.dumps({
|
|
'center': [round(c, 6) for c in area.get('center', (0, 0))],
|
|
'radius': round(area.get('radius', 50), 2)
|
|
})
|
|
elif area['type'] == 'rectangle':
|
|
coordinates = json.dumps({
|
|
'rect': [round(r, 6) for r in area.get('rect', (0, 0, 100, 100))]
|
|
})
|
|
elif area['type'] == 'polygon':
|
|
coordinates = json.dumps({
|
|
'points': [[round(p[0], 6), round(p[1], 6)] for p in area.get('points', [])]
|
|
})
|
|
else:
|
|
return False
|
|
|
|
# 检查是否已存在
|
|
check_query = """
|
|
SELECT id FROM danger_zones WHERE type = %s AND coordinates = %s
|
|
"""
|
|
cursor.execute(check_query, (area['type'], coordinates))
|
|
result = cursor.fetchone()
|
|
print(result)
|
|
exists = result is not None
|
|
print(exists)
|
|
cursor.close()
|
|
except Error as e:
|
|
print(f"Error: '{e}' occurred")
|
|
finally:
|
|
if connection.is_connected():
|
|
close_connection(connection)
|
|
return exists |