import sys sys.path.append("..") import mytool.db_helper.mysqlhelper as mdb # 数据库操作 # 可自行添加操作函数 class DBOperation: def __init__(self,db_ip, db_port, db_name, db_user, db_password): self._db = mdb.DB(db_ip, db_port, db_name, db_user, db_password) # 查询 def query(self, sql): return self._db.query_all_sql(sql) # 更新 def update(self, sql): return self._db.execute_sql(sql) def query_parking_in_unit_tableid(self, unit,tableid): sql = "select * from space WHERE unit=%s and table_id=%s" return self._db.query_all_sql(sql, (unit,tableid)) def query_command_in_unit(self, unit): sql = "select * from command_queue WHERE unit=%s" return self._db.query_all_sql(sql, unit) def query_pick_command_in_unit(self, unit): sql = "select * from command_queue where unit=%s and type=2 order by queue_id ASC" return self._db.query_all_sql(sql, unit) def query_pick_command_in_unit_and_statu(self, unit, statu): sql = "select * from command_queue where unit=%s and statu=%s and type=2 order by queue_id ASC" return self._db.query_all_sql(sql, (unit, statu)) def query_sort_pick_command(self, unit): sql = "select * from command_queue where unit=%s and (statu=0 or statu=1) and type=2 order by queue_id ASC" return self._db.query_all_sql(sql, unit) def query_command_all_in_unit_and_sort(self, unit): sql = "select * from command_queue where unit=%s order by type,statu DESC,queue_id ASC" return self._db.query_all_sql(sql, unit) def query_command_all_in_unit_type_and_sort(self, unit, type): sql = "select * from command_queue where unit=%s and type=%s order by type,statu DESC,queue_id ASC" return self._db.query_all_sql(sql, (unit, type)) def query_command_all(self): sql = "select * from command_queue" return self._db.query_all_sql(sql) def query_command_in_car_number(self, car_number): sql = "select * from command_queue WHERE car_number=%s" return self._db.query_all_sql(sql, car_number) def query_command_in_primary_key(self, primary_key): sql = "select * from command_queue WHERE primary_key=%s" return self._db.query_all_sql(sql, primary_key) def query_command_in_height_unit_and_statu(self, height, unit): sql = "select * from command_queue WHERE height>=%s and unit=%s and statu=0 and type=1" # type=1 存车 return self._db.query_all_sql(sql, (height, unit)) def query_command_in_height_minmax_unit_and_statu(self, min,max, unit): sql = "select * from command_queue WHERE height>=%s and height<=%s and unit=%s and statu=0 and type=1" # type=1 存车 return self._db.query_all_sql(sql, ( min,max, unit)) def query_space_in_car_number(self, car_number): sql = "select * from space WHERE car_number=%s" return self._db.query_all_sql(sql, car_number) def query_space_in_primary_key(self, primary_key): sql = "select * from space WHERE primary_key=%s" return self._db.query_all_sql(sql, primary_key) def query_space_in_unit_and_table_id(self, unit, table_id): sql = "select * from space WHERE unit=%s and table_id=%s" return self._db.query_all_sql(sql, (unit, table_id)) def query_space_in_height(self, height): sql = "select * from space WHERE height>=%s" return self._db.query_all_sql(sql, height) def query_space_in_space_id(self, space_id): sql = "select * from space WHERE id=%s" return self._db.query_all_sql(sql, space_id) def query_space_in_height_and_empty(self, height): sql = "select * from space WHERE height>=%s and statu=0 and car_number IS NULL" return self._db.query_all_sql(sql, height) def query_space_in_height_unit_and_empty(self, height, unit): sql = "select * from space WHERE height>=%s and unit=%s and statu=0 and car_number IS NULL" return self._db.query_all_sql(sql, (height, unit)) def query_space_in_height_unit_and_empty_2(self, min,max, unit): sql = "select * from space WHERE height>=%s and height<=%s and unit=%s and statu=0 and car_number IS NULL" return self._db.query_all_sql(sql, (min, max, unit)) def query_space_in_unit(self, unit): sql = "select * from space WHERE unit=%s" return self._db.query_all_sql(sql, unit) def query_space_all(self): sql = "select * from space" return self._db.query_all_sql(sql) def query_vehicle_primary_key(self, car_number): sql = "select primary_key from vehicle where car_number=%s" return self._db.query_all_sql(sql, car_number) def update_space_status_in_space_id(self, space_id, statu): sql = "update space set statu=%s where id=%s" return self._db.execute_sql(sql, (statu, space_id)) def update_space_status_car_number_in_space_id(self, space_id,car_number,statu): sql = "update space set statu=%s,car_number=%s where id=%s" return self._db.execute_sql(sql, (statu, car_number,space_id)) def update_command_status_in_car_number(self,car_number,statu): sql = "update command_queue set statu=%s where car_number=%s" return self._db.execute_sql(sql, (statu, car_number)) def update_space_remark_and_statu_in_space_id(self,space_id,remark,statu): sql = "update space set remark=%s,statu=%s where id=%s" return self._db.execute_sql(sql, (remark, statu,space_id)) def clear_space_data(self, space_id): sql = "update space set car_number=NULL,primary_key=NULL,statu=0,space_info=NULL,measure_info=NULL,plate_color=NULL,plate_type=NULL,plate_confidence=NULL,recognition_time=NULL,plate_full_image=NULL,plate_clip_image=NULL,remark=NULL where id=%s" return self._db.execute_sql(sql, space_id) def delete_command(self, car_number): sql = "delete from command_queue where car_number=%s" return self._db.execute_sql(sql, car_number) # 查询一段时间内停车次数 def query_park_count_in_time(self, in_date, out_date): sql = "select * from record where in_time_start >= %s and in_time_start <= %s" return self._db.query_all_sql(sql, (in_date,out_date)) def query_pick_count_out_time(self, in_date, out_date): sql = "select * from record where out_time_start >= %s and out_time_start <= %s" return self._db.query_all_sql(sql, (in_date, out_date)) def query_park_count_in_time_and_unit(self, in_date, out_date,unit): sql = "select * from record where (in_time_start >= %s and in_time_start <= %s) and unit_id=%s" return self._db.query_all_sql(sql, (in_date,out_date,unit)) def query_pick_count_out_time_and_unit(self, in_date, out_date,unit): sql = "select * from record where (out_time_start >= %s and out_time_start <= %s) and unit_id=%s" return self._db.query_all_sql(sql, (in_date, out_date,unit)) # 查询一段时间内停车车数 # period_type = 1 :当天内 # 2 :当月内 # 3 :自定义起止时间 def query_carQuantity_inPeriod(self, period_type, date): sql = "select count(distinct(car_number)) from record where in_time_start like %s" if period_type == "day": date = date[0:10] elif period_type == "month": date = date[0:7] date = date + "%" # 模糊查询 result = self._db.query_all_sql(sql, date) # [{'count(distinct(car_number))': 4}] return result[0]["count(distinct(car_number))"]