Skip to content

I
integrated-scheduling-v3
Commit 62c4edc6 authored by 张晓彤's avatar 张晓彤
Browse files
Options

优化穿越装载区调度，优化二次调度触发机制

parent 97c4b861
Show whitespace changes
Inline Side-by-side
Showing with 766 additions and 331 deletions
app.py
View file @ 62c4edc6
......@@ -14,6 +14,7 @@ from core.group import GroupDispatcher
from core.group import group_direct2redis
import uuid
from core.util import POST
from util.area_analysis import *
config = {
"DEBUG": True, # some Flask specific configs
......@@ -333,8 +334,16 @@ def redispatch_request():
logger.info("truck_locates_dict")
logger.info(truck_locates_dict)
closer_area_id = '7ff73575-2134-afd4-1065-d5d60e8751c9'
further_area_id = '79584290-1134-8b85-f4cc-1dcf64fc3456'
with open(json_file, encoding='UTF-8') as f:
load_value = json.load(f)
postgre_config = load_value["postgresql"]
closer_area_name = postgre_config["closer_area_name"]
closer_area_id = session_postgre.query(DiggingWorkArea).filter_by(Name=closer_area_name).first().Id
further_area_name = postgre_config["further_area_name"]
further_area_id = session_postgre.query(DiggingWorkArea).filter_by(Name=further_area_name).first().Id
logger.info("近端装载区id")
logger.info(closer_area_id)
......@@ -431,302 +440,5 @@ def redispatch_request():
return jsonify(msg="success", code=0)
def area_choose(excavators_id, closer_area_id, further_area_id, lane_set,
logger, truck, truck_locates_dict):
"""
两装载区均不空闲，执行二次调度
:param excavators_id: 挖机集合
:param closer_area_id: 近端装载区id
:param further_area_id: 远端装载区id
:param lane_set: 近端及远端装载区间路段集合
:param logger: 日志器
:param truck: 车辆对象
:param truck_locates_dict: 车辆位置
:return: target_excavator
"""
logger.info("excavator_hold_truck_list")
logger.info(truck.excavator_hold_truck_list)
logger.info(list(excavators_id)[0])
logger.info(list(excavators_id)[1])
logger.info(truck.excavator_hold_truck_list[list(excavators_id)[0]])
logger.info(truck.excavator_hold_truck_list[list(excavators_id)[1]])
# 统计驶往两装载区的车辆
arrival_truck_set = truck.excavator_hold_truck_list[list(excavators_id)[0]] \
+ truck.excavator_hold_truck_list[list(excavators_id)[1]]
# 统计车辆抵达时间
arrival_truck_reach_time = [truck.cur_truck_reach_excavator[truck.truck_uuid_to_index_dict[truck_id]] for \
truck_id in arrival_truck_set]
# arrival_truck_set = ['309705a0-5ddf-4559-b6c4-ee17a57677ad', '899705a0-5ddf-4559-b6c4-ee17a57677ad']
#
# arrival_truck_reach_time = [8.04, 6.05]
logger.info("arrival_truck_reach_time")
logger.info(arrival_truck_reach_time)
logger.info("arrival_truck_set")
logger.info(arrival_truck_set)
arrival_truck_list = list(zip(np.array(arrival_truck_set), np.array(arrival_truck_reach_time)))
arrival_truck_list = sorted(arrival_truck_list, key=lambda item: item[1])
logger.info("arrival_truck_list")
logger.info(arrival_truck_list)
logger.info("arrival_truck_list")
logger.info(arrival_truck_list)
# 统计不同状态车辆数量
goto_closer_area_num = 0
goto_further_area_num = 0
for truck_id, reach_time in arrival_truck_list:
if truck_id in truck_locates_dict:
truck_lane_id = truck_locates_dict[truck_id]
# 车辆已经经过近端装载区
if truck_lane_id in lane_set:
# 前往远端装载区车辆数加1
goto_further_area_num += 1
# 车辆未经过近端装载区
else:
# 前往近端或近端装载区车辆数加1
goto_closer_area_num += 1
else:
continue
# goto_further_area_num -= 1
logger.info("goto_further_area_num-goto_closer_area_num")
logger.info(goto_further_area_num)
logger.info(goto_closer_area_num)
# 默认当前请调车辆与近端装载点前没有车辆，因此前往近端装载区的车辆仅其本身
goto_closer_area_num = 1
# 在远处排队等待的车辆更少
if goto_closer_area_num > goto_further_area_num:
logger.info("远端挖机排队时间短, 调度车辆前往")
target_excavator = DispatchInfo.load_excavator_dict[further_area_id]
# truck_dispatch_to_redis(request_truck_id, group_id, DispatchInfo.load_excavator_dict[further_area_id])
else:
logger.info("近端挖机排队时间短, 调度车辆前往")
target_excavator = DispatchInfo.load_excavator_dict[closer_area_id]
# truck_dispatch_to_redis(request_truck_id, group_id, DispatchInfo.load_excavator_dict[closer_area_id])
return target_excavator
def area_analysis(load_area_uuid):
"""
Analysis which area is closer.
:param load_area_uuid:
:return: closer_area_uuid, further_area_uuid
"""
try:
excavator_uuid_to_load_area_uuid_dict = get_value("excavator_uuid_to_load_area_uuid_dict")
load_area_uuid = list(load_area_uuid)
load_area_uuid[0] = excavator_uuid_to_load_area_uuid_dict[load_area_uuid[0]]
load_area_uuid[1] = excavator_uuid_to_load_area_uuid_dict[load_area_uuid[1]]
distance_a = session_postgre.query(WalkTimePark)\
.filter_by(load_area_id=load_area_uuid[0]).first().park_load_distance
distance_b = session_postgre.query(WalkTimePark)\
.filter_by(load_area_id=load_area_uuid[1]).first().park_load_distance
if distance_a > distance_b:
return load_area_uuid[1], load_area_uuid[0]
else:
return load_area_uuid[0], load_area_uuid[1]
except Exception as es:
logger.error("装载区距离分析异常")
logger.error(es)
return load_area_uuid[0], load_area_uuid[1]
def truck_pass_first_area(truck_id, lane_id, closer_entrance_node_id, further_entrance_node_id):
"""
Truck has gone through the first area.
:param truck_id:
:param lane_id:
:param closer_entrance_node_id:
:param further_entrance_node_id:
:return:
"""
# try:
def backtracking(root_node):
from collections import deque
que = deque([root_node])
while que:
size = len(que)
for _ in range(size):
cur_node = que.popleft()
if cur_node is None:
continue
logger.info(cur_node)
if cur_node == closer_entrance_node_id:
logger.info("closer_entrance_node")
return 0
if cur_node == further_entrance_node_id:
logger.info("further_entrance_node")
return 1
for item in session_postgre.query(Lane).filter_by(StartNodeId=cur_node).all():
if item:
que.append(item.EndNodeId)
node_id = session_postgre.query(Lane).filter_by(Id=lane_id).first().EndNodeId
if backtracking(root_node=node_id) == 0:
return True
else:
return False
# except Exception as es:
# logger.error("车辆行驶位置判断异常")
# logger.error(es)
# return True
def get_trucks_locate():
"""
get trucks locates.
:return: truck_locate_dict
"""
try:
truck_name_to_uuid_dict = get_value("truck_name_to_uuid_dict")
truck_locate_dict = {}
device_name_set = redis2.keys()
for item in device_name_set:
item = item.decode(encoding='utf-8')
key_value_dict = redis2.hgetall(item)
if str_to_byte('type') in key_value_dict:
device_type = key_value_dict[str_to_byte('type')]
is_online = key_value_dict[str_to_byte('online')]
key_set = key_value_dict.keys()
if (device_type == str_to_byte("1")) \
and (str_to_byte('online') in key_set) \
and (bytes.decode(is_online) in ["true" or "True"]) \
and (str_to_byte('laneId') in key_set):
truck_locate = key_value_dict[str_to_byte('laneId')]
# logger.error(item)
# logger.error(eval(truck_locate))
if eval(truck_locate) is not '':
truck_locate_dict[truck_name_to_uuid_dict[item]] = eval(truck_locate)
logger.error(truck_locate_dict)
else:
continue
return truck_locate_dict
except Exception as es:
logger.error("车辆所在路段读取异常")
logger.error(es)
return {}
def truck_dispatch_to_redis(truck_id, group_id, excavator_id):
"""
write truck dispatch to redis.
:param truck_id:
:param group_id:
:param excavator_id:
:return:
"""
# 查询车辆相关派车计划
record = {}
try:
# dump_id = DispatchInfo.unload_area_dump_dict[unload_area_id]
item = (session_mysql.query(DispatchSetting).filter_by(exactor_id=excavator_id, group_id=group_id,
isdeleted=0, ).first())
if item is None:
raise Exception("调度计划配置异常")
except Exception as es:
item = (session_mysql.query(DispatchSetting).filter_by(group_id=group_id, isdeleted=0, ).first())
logger.error(es)
# 其余调度信息写入
try:
# record["dispatchId"] = item.id
record["dispatchId"] = str(uuid.uuid1())
record["exactorId"] = item.exactor_id
record["loadAreaId"] = item.load_area_id
record["dumpId"] = item.dump_id
record["unloadAreaId"] = item.unload_area_id
record["groupId"] = group_id
record["isdeleted"] = False
record["isTemp"] = False
record["haulFlag"] = -1
record["groupName"] = DispatchInfo.group_name[group_id]
logger.info(f'{truck_id} redis 注入 {record}')
except Exception as es:
logger.error("调度结果写入异常-矿卡空载")
logger.error(es)
finally:
redis5.set(truck_id, str(json.dumps(record)))
def get_lanes_between_entrances(closer_node_id, further_node_id):
"""
get lanes between two entrance nodes.
:param closer_node_id:
:param further_node_id:
:return: lane set
"""
try:
max_find_it = 100
next_node_id = closer_node_id
lane_set = []
while max_find_it > 0 and next_node_id != further_node_id:
item = session_postgre.query(Lane).filter_by(StartNodeId=next_node_id, Type=2).first()
if item:
next_lane_id = item.Id
next_node_id = item.EndNodeId
lane_set.append(next_lane_id)
max_find_it -= 1
max_find_it -= 1
return lane_set
except Exception as es:
logger.error("获取装载区间路段异常")
logger.error(es)
return []
def get_excavator_state(excavator_id):
"""
get group_excavators state.
:param excavator_id:
:return: state
"""
try:
logger.error(excavator_id)
device_name = session_mysql.query(Equipment).filter_by(id=excavator_id, device_type=2).first().device_name
key_value_dict = redis2.hgetall(device_name)
if str_to_byte('online') in key_value_dict:
is_online = key_value_dict[str_to_byte('online')]
else:
logger.warning(f'挖机 {device_name} 不在线')
return 0
key_set = key_value_dict.keys()
state = 100
if (str_to_byte('online') in key_set) and (bytes.decode(is_online) in ["true" or "True"]):
if str_to_byte('workState') in key_set:
state = key_value_dict[str_to_byte('workState')]
else:
logger.warning(f'挖机 {device_name} 状态未知')
return 0
return int(float(byte_to_str(state)))
except Exception as es:
logger.error("挖机状态读取异常")
logger.error(es)
return 0
if __name__ == "__main__":
app.run(host='0.0.0.0', port=80)
\ No newline at end of file
config.json
View file @ 62c4edc6
......@@ -6,25 +6,26 @@
"dump_target_mass": 5000,
"excavator_target_mass": 5000
},
"mysql": {
"host": "172.16.0.103",
"host": "172.16.0.51",
"port": "3306",
"user": "root",
"password": "Huituo@123",
"database": "ht_zhunneng"
},
"postgresql": {
"host": "172.16.0.103",
"host": "172.16.0.51",
"port": "5432",
"user": "postgres",
"password": "Huituo@123",
"database": "gis_zhunneng"
},
"redis": {
"host": "172.16.0.103",
"host": "172.16.0.51",
"password": "Huituo@123"
},
"gothrough": {
"closer_area_name": "哈1010平盘",
"further_area_name": "哈1026平盘"
}
}
\ No newline at end of file
core/group.py
View file @ 62c4edc6
......@@ -18,6 +18,7 @@ from util import CoreException
from equipment import TruckInfo, ExcavatorInfo, DumpInfo
from core.schedule import PreSchedule
from graph.topo_graph import Topo
from util.area_analysis import *
def direct2redis():
......@@ -350,6 +351,7 @@ class GroupDispatcher:
self.group = group
if self.group.topo is not None:
self.redispatcher = ReDispatcher(self.group)
self.gothroghdispatcher = GoThroughDispatcher(self.group)
def group_dispatch(self, Solver) -> Mapping[str, List[str]]:
"""
......@@ -526,7 +528,8 @@ class GroupDispatcher:
truck_info: CurrentTruck, truck_task: int, truck_trip: List[int]):
# 车辆停止或者车辆位于卸载区内, 调度车辆前往装载区
if truck_task in [-2, 3, 4, 5]:
if truck_task == 3 and self.group.truck.truck_current_state[i] == 2:
# if truck_task == 3 and self.group.truck.truck_current_state[i] == 2:
if truck_task == 3:
if self.group.topo is not None:
try:
truck_locate = self.group.truck.get_truck_locate_dict()[i]
......@@ -545,9 +548,7 @@ class GroupDispatcher:
self.logger.info(f'车辆状态 {self.group.truck.truck_current_state[i]}')
self.logger.info(f'车辆临时 {truck_is_temp} {type(truck_is_temp)}')
# 车辆当前位于交叉路口前，且排队等待
if (truck_locate in self.group.topo.cross_bf_lanes) and (
self.group.truck.truck_current_state[i] == 2) \
and (not truck_is_temp):
if (truck_locate in self.group.topo.cross_bf_lanes) and (not truck_is_temp):
# self.redispatch_to_dump(i, truck_dispatch, truck_locate, truck_trip)
self.redispatcher.redispatch_to_dump(i, truck_dispatch, truck_locate, truck_trip)
else:
......@@ -584,7 +585,8 @@ class GroupDispatcher:
self.logger.error(es)
# 车辆位于装载区内, 调度车辆前往卸载区
elif truck_task in [0, 1, 2]:
if truck_task == 0 and self.group.truck.truck_current_state[i] == 2:
# if truck_task == 0 and self.group.truck.truck_current_state[i] == 2:
if truck_task == 0:
if self.group.topo is not None:
if i in self.group.truck.get_truck_locate_dict():
truck_locate = self.group.truck.get_truck_locate_dict()[i]
......@@ -611,12 +613,17 @@ class GroupDispatcher:
self.logger.info(truck_is_temp)
# if (truck_locate in self.topo.cross_bf_lanes) and (self.truck.truck_current_state[truck_id] == 2) \
# and (not truck_is_temp):
if (truck_locate in self.group.topo.cross_bf_lanes) and (
self.group.truck.truck_current_state[i] == 2):
# if (truck_locate in self.group.topo.cross_bf_lanes) and (
# self.group.truck.truck_current_state[i] == 2):
if (truck_locate in self.group.topo.cross_bf_lanes) and (not truck_is_temp):
self.logger.info("触发二次调度")
# self.redispatch_to_excavator(i, truck_dispatch, truck_locate)
self.redispatcher.redispatch_to_excavator(i, truck_dispatch, truck_locate)
# 穿越装载区判断
self.gothroghdispatcher.update_lanes_info()
self.gothroghdispatcher.redispatch_request(request_truck_id=i, truck=self.group.truck)
else:
pass
else:
......@@ -1000,3 +1007,358 @@ class ReDispatcher:
print("congestion_lane_list")
print(list(set(congestion_lane_list)))
return list(set(congestion_lane_list))
class GoThroughDispatcher:
"""
go through load areas controller
"""
def __init__(self, group: Group):
self.logger = get_logger("zxt.GoThroughDispatcher")
self.group = group
def update_lanes_info(self):
try:
with open(json_file, encoding='UTF-8') as f:
load_value = json.load(f)
gothrough_config = load_value["gothrough"]
self.closer_area_name = gothrough_config["closer_area_name"]
self.closer_area_id = str(session_postgre.query(DiggingWorkArea).filter_by(Name=self.closer_area_name).first().Id)
self.further_area_name =gothrough_config["further_area_name"]
self.further_area_id = str(session_postgre.query(DiggingWorkArea).filter_by(Name=self.further_area_name).first().Id)
self.logger.info("近端装载区id")
self.logger.info(self.closer_area_id)
self.logger.info("远端装载区id")
self.logger.info(self.further_area_id)
self.logger.info("DispatchInfo.load_excavator_dict")
self.logger.info(DispatchInfo.load_excavator_dict)
# 读取两个挖机id
self.logger.info(self.closer_area_id in DispatchInfo.load_excavator_dict)
self.logger.info(self.further_area_id in DispatchInfo.load_excavator_dict)
if (self.closer_area_id in DispatchInfo.load_excavator_dict) and \
(self.further_area_id in DispatchInfo.load_excavator_dict):
self.closer_excavator_id, self.further_excavator_id = DispatchInfo.load_excavator_dict[self.closer_area_id], \
DispatchInfo.load_excavator_dict[self.further_area_id]
else:
self.logger.error("装载点信息错误")
return
# 读取挖机状态
self.closer_excavator_state, self.further_excavator_state = get_excavator_state(self.closer_excavator_id), \
get_excavator_state(self.further_excavator_id)
# 读取两个装载区入场点id
self.closer_entrance_node_id = session_postgre.query(DiggingWorkArea).filter_by(
Id=self.closer_area_id).first().EntranceNodeId
self.further_entrance_node_id = session_postgre.query(DiggingWorkArea).filter_by(
Id=self.further_area_id).first().EntranceNodeId
self.logger.info("近端装载区入场点")
self.logger.info(self.closer_entrance_node_id)
self.logger.info("远端装载区入场点")
self.logger.info(self.further_entrance_node_id)
except Exception as es:
self.logger.error("读取装载区及车辆信息异常")
self.logger.error(es)
return
'''
2. 寻找车辆穿越装载区自动判定区域
'''
try:
further_area_backtrack_node_id = self.further_entrance_node_id
self.further_area_backtrack_lanes = []
for i in range(5):
# 查询回溯路段对象
item = session_postgre.query(Lane).filter_by(EndNodeId=further_area_backtrack_node_id).first()
if item is None:
break
# 将路段Id加入回溯列表
self.further_area_backtrack_lanes.append(str(item.Id))
# 更新当前回溯节点
further_area_backtrack_node_id = item.StartNodeId
except Exception as es:
self.logger.error("回溯近端装载路段异常")
self.logger.error(es)
try:
lowest_common_ancestor = None
closer_area_backtrack_node_id = self.closer_entrance_node_id
self.closer_area_backtrack_lanes = []
for i in range(5):
# 查询回溯路段对象
item = session_postgre.query(Lane).filter_by(EndNodeId=closer_area_backtrack_node_id).first()
if item is None:
break
# 若查询到共同祖先
if str(item.Id) in self.further_area_backtrack_lanes:
lowest_common_ancestor = item
break
# 将路段Id加入回溯列表
self.closer_area_backtrack_lanes.append(str(item.Id))
# 更新当前回溯节点
closer_area_backtrack_node_id = item.StartNodeId
# 交叉口路段
cross_lane = str(lowest_common_ancestor.Id)
except Exception as es:
self.logger.error("回溯远端装载路段异常")
self.logger.error(es)
return
self.logger.info("cross_lane")
self.logger.info(cross_lane)
# 截取远端入场点到交叉口的路段
self.further_area_backtrack_lanes = self.further_area_backtrack_lanes[0:self.further_area_backtrack_lanes.index(cross_lane)]
# 穿越装载区检查路段
self.gothrough_check_lanes = [cross_lane]
try:
# 多回溯一个路段
if lowest_common_ancestor.Length < 200:
item = session_postgre.query(Lane).filter_by(EndNodeId=lowest_common_ancestor.StartNodeId).first()
self.gothrough_check_lanes.append(item.Id)
self.logger.info("gothrough_check_lanes")
self.logger.info(self.gothrough_check_lanes)
except Exception as es:
self.logger.error("穿越装载区交叉口判断异常")
self.logger.error(es)
def redispatch_request(self, request_truck_id: str, truck: TruckInfo):
# 调度开始时间
rtd_start_time = datetime.now()
# 更新周期参数
self.logger.info("#####################################请求调度更新开始#####################################")
'''
1. 读取二次调度所需信息（车辆位置、分组、挖机等）
'''
try:
truck_uuid_to_name_dict = get_value("truck_uuid_to_name_dict")
# 获取请调车辆名
if request_truck_id not in truck_uuid_to_name_dict:
raise CoreException(102, f'truck_uuid_to_name_dict 缺失车辆 {request_truck_id} 信息')
request_truck_name = truck_uuid_to_name_dict[request_truck_id]
except Exception as es:
self.logger.error("数据库异常, 车辆编号未知")
self.logger.error(es)
return
try:
# 读取请调车辆所属分组
group_id = DispatchInfo.truck_group_dict[request_truck_id]
# 读取分组挖机集合
excavators_id = DispatchInfo.get_excavator(group_id)
# 读取车辆位置信息
truck_locates_dict = get_trucks_locate()
self.logger.info("truck_locates_dict")
self.logger.info(truck_locates_dict)
except Exception as es:
self.logger.error("车辆位置信息读取异常")
self.logger.error(es)
# try:
#
# with open(json_file, encoding='UTF-8') as f:
# load_value = json.load(f)
# gothrough_config = load_value["gothrough"]
#
# closer_area_name = gothrough_config["closer_area_name"]
# closer_area_id = str(session_postgre.query(DiggingWorkArea).filter_by(Name=closer_area_name).first().Id)
#
# further_area_name =gothrough_config["further_area_name"]
# further_area_id = str(session_postgre.query(DiggingWorkArea).filter_by(Name=further_area_name).first().Id)
#
# self.logger.info("近端装载区id")
# self.logger.info(closer_area_id)
# self.logger.info("远端装载区id")
# self.logger.info(further_area_id)
#
# self.logger.info("DispatchInfo.load_excavator_dict")
# self.logger.info(DispatchInfo.load_excavator_dict)
#
# # 读取两个挖机id
# self.logger.info(closer_area_id in DispatchInfo.load_excavator_dict)
# self.logger.info(further_area_id in DispatchInfo.load_excavator_dict)
# if (closer_area_id in DispatchInfo.load_excavator_dict) and \
# (further_area_id in DispatchInfo.load_excavator_dict):
# closer_excavator_id, further_excavator_id = DispatchInfo.load_excavator_dict[closer_area_id], \
# DispatchInfo.load_excavator_dict[further_area_id]
# else:
# self.logger.error("装载点信息错误")
# return
#
# # 读取挖机状态
# closer_excavator_state, further_excavator_state = get_excavator_state(closer_excavator_id), \
# get_excavator_state(further_excavator_id)
#
# # 读取两个装载区入场点id
# closer_entrance_node_id = session_postgre.query(DiggingWorkArea).filter_by(
# Id=closer_area_id).first().EntranceNodeId
#
# further_entrance_node_id = session_postgre.query(DiggingWorkArea).filter_by(
# Id=further_area_id).first().EntranceNodeId
#
# self.logger.info("近端装载区入场点")
# self.logger.info(closer_entrance_node_id)
# self.logger.info("远端装载区入场点")
# self.logger.info(further_entrance_node_id)
#
# except Exception as es:
# self.logger.error("读取装载区及车辆信息异常")
# self.logger.error(es)
# return
try:
# 读取请调车辆所在路段信息
request_truck_lane_id = truck_locates_dict[request_truck_id]
self.logger.info("request_truck_lane_id:")
self.logger.info(request_truck_lane_id)
self.logger.info(truck_locates_dict)
except Exception as es:
self.logger.error(f'车辆 {request_truck_name} 位置信息不可用')
self.logger.error(es)
return
# '''
# 2. 寻找车辆穿越装载区自动判定区域
# '''
# try:
# further_area_backtrack_node_id = further_entrance_node_id
# further_area_backtrack_lanes = []
# for i in range(5):
# # 查询回溯路段对象
# item = session_postgre.query(Lane).filter_by(EndNodeId=further_area_backtrack_node_id).first()
# if item is None:
# break
# # 将路段Id加入回溯列表
# further_area_backtrack_lanes.append(str(item.Id))
# # 更新当前回溯节点
# further_area_backtrack_node_id = item.StartNodeId
#
# except Exception as es:
# self.logger.error("回溯近端装载路段异常")
# self.logger.error(es)
#
# try:
# lowest_common_ancestor = None
# closer_area_backtrack_node_id = closer_entrance_node_id
# closer_area_backtrack_lanes = []
# for i in range(5):
# # 查询回溯路段对象
# item = session_postgre.query(Lane).filter_by(EndNodeId=closer_area_backtrack_node_id).first()
# if item is None:
# break
# # 若查询到共同祖先
# if str(item.Id) in further_area_backtrack_lanes:
# lowest_common_ancestor = item
# break
# # 将路段Id加入回溯列表
# closer_area_backtrack_lanes.append(str(item.Id))
# # 更新当前回溯节点
# closer_area_backtrack_node_id = item.StartNodeId
#
# # 交叉口路段
# cross_lane = str(lowest_common_ancestor.Id)
#
# except Exception as es:
# self.logger.error("回溯远端装载路段异常")
# self.logger.error(es)
# return
#
#
# self.logger.info("cross_lane")
# self.logger.info(cross_lane)
#
# # 截取远端入场点到交叉口的路段
# further_area_backtrack_lanes = further_area_backtrack_lanes[0:further_area_backtrack_lanes.index(cross_lane)]
#
# # 穿越装载区检查路段
# gothrough_check_lanes = [cross_lane]
#
# try:
# # 多回溯一个路段
# if lowest_common_ancestor.Length < 200:
# item = session_postgre.query(Lane).filter_by(EndNodeId=lowest_common_ancestor.StartNodeId).first()
# gothrough_check_lanes.append(item.Id)
#
# self.logger.info("gothrough_check_lanes")
# self.logger.info(gothrough_check_lanes)
#
# except Exception as es:
# self.logger.error("穿越装载区交叉口判断异常")
# self.logger.error(es)
'''
3. 调度判断逻辑
'''
if request_truck_id not in self.group.truck_info_list:
raise CoreException(102, f'truck_info_list 缺失车辆 {request_truck_id} 信息')
truck_trip = self.group.truck_info_list[request_truck_id].get_trip()
excavator_index = int(truck_trip[1])
if excavator_index == -1:
raise CoreException(110, f'驶往挖机车辆 {request_truck_id} 行程计划为-1')
if excavator_index not in self.group.excavator_info.truck_index_to_uuid_dict:
raise CoreException(111, f'truck_index_to_uuid_dict 缺失 {excavator_index} 号挖机信息')
current_truck_goto_excavator_id = self.group.excavator_info.truck_index_to_uuid_dict[excavator_index]
if request_truck_lane_id in self.gothrough_check_lanes:
# 选择合适装载区
target_excavator = None
try:
target_excavator = area_choose(excavators_id, self.closer_area_id, self.further_area_id,
self.further_area_backtrack_lanes, self.closer_area_backtrack_lanes, self.logger,
truck, truck_locates_dict, self.closer_excavator_state, self.further_excavator_state)
except Exception as es:
logger.error("寻找最优装载区异常")
logger.error(es)
if (target_excavator is not None) and (current_truck_goto_excavator_id != target_excavator):
self.logger.info(f'更新车辆 {request_truck_name} 派车计划前往 {target_excavator}')
# 派车计划写入redis
truck_dispatch_to_redis(request_truck_id, group_id, target_excavator)
POST(request_truck_id)
else:
# 派车计划未更改
self.logger.info("派车计划未变动")
self.logger.info("#####################################请求调度更新结束#####################################")
# 调度结束时间
rtd_end_time = datetime.now()
print(f'调度时耗 {rtd_end_time - rtd_start_time}')
\ No newline at end of file
core/util.py
View file @ 62c4edc6
......@@ -8,7 +8,7 @@
import requests
from tables import *
with open(json_file) as f:
with open(json_file, encoding='UTF-8') as f:
mysql_config = json.load(f)["mysql"]
......@@ -50,11 +50,17 @@ with open(json_file) as f:
def get_cross_next_lanes(truck_locate):
"""
获取当前车辆所在路段（交叉口），的子孙（后续两个）节点
:param truck_locate:
:return: lane_id list
"""
next_lane_list = []
startNodeId = session_postgre.query(Lane).filter_by(Id=truck_locate).first().EndNodeId
for item in session_postgre.query(Lane).filter_by(StartNodeId=startNodeId).all():
next_lane_list.append(str(item.Id))
next_lane_end_node = item.EndNodeId
if item.Length < 100:
for next_item in session_postgre.query(Lane).filter_by(StartNodeId=next_lane_end_node).all():
next_lane_list.append(str(next_item.Id))
print("next_lane_list")
......
equipment/truck.py
View file @ 62c4edc6
......@@ -8,6 +8,7 @@
from data.para_config import *
from typing import List
from util.core_exception import CoreException
# 矿卡设备类
......@@ -136,6 +137,7 @@ class TruckInfo(WalkManage):
################################################ short term update ################################################
# TODO: 应该一次读取形成一个对象，然后分配各属性
# 更新矿卡当前任务
def update_truck_current_task(self):
self.truck_current_task = {}
......@@ -145,7 +147,7 @@ class TruckInfo(WalkManage):
truck_name_to_uuid_dict = get_value("truck_name_to_uuid_dict")
for item in device_name_set:
# try:
try:
item = item.decode(encoding="utf-8")
key_value_dict = redis2.hgetall(item) # reids str可以自动转为bytes
if str_to_byte("type") in key_value_dict:
......@@ -153,6 +155,9 @@ class TruckInfo(WalkManage):
else:
continue
if device_type == 1:
if item not in truck_name_to_uuid_dict:
raise CoreException(107, f'车辆{item}不存在于truck_name_to_uuid_dict')
if truck_name_to_uuid_dict[item] in self.dynamic_truck_set:
# currentTask = int(key_value_dict[str_to_byte("currentTask")])
currentTask = int(byte_to_str(key_value_dict[str_to_byte("currentTask")]))
......@@ -163,9 +168,9 @@ class TruckInfo(WalkManage):
self.truck_current_state[
truck_name_to_uuid_dict[item]
] = currentState
# except Exception as es:
# self.logger.error("读取矿卡任务异常-reids读取异常")
# self.logger.error(es)
except Exception as es:
self.logger.warning("读取矿卡任务异常-reids读取异常")
self.logger.warning(es)
self.logger.info("矿卡当前任务：")
self.logger.info(self.truck_current_task)
......@@ -208,10 +213,16 @@ class TruckInfo(WalkManage):
if item not in truck_name_to_uuid_dict:
continue
key_value_dict = byte_to_str(redis5.get(truck_name_to_uuid_dict[item])) # reids str可以自动转为bytes
if item not in truck_name_to_uuid_dict:
raise CoreException(107, f'车辆{item}不存在于truck_name_to_uuid_dict')
key_value_dict = redis5.get(truck_name_to_uuid_dict[item]) # reids str可以自动转为bytes
if key_value_dict is None:
raise CoreException(108, f'车辆{item}不存在于redis5')
key_value_dict = json.loads(byte_to_str(key_value_dict))
print(key_value_dict)
key_value_dict = json.loads(key_value_dict)
if truck_name_to_uuid_dict[item] in self.dynamic_truck_set:
try:
isTemp = key_value_dict["isTemp"]
......
realtime_dispatch_test.py
View file @ 62c4edc6
......@@ -188,4 +188,4 @@ if __name__ == "__main__":
logger.info(" ")
logger.info("调度系统启动")
main(30, dispatcher)
main(20, dispatcher)
settings.py
View file @ 62c4edc6
......@@ -23,16 +23,16 @@ global redispatch_active
json_file = "config.json"
with open(json_file) as f:
with open(json_file, encoding='UTF-8') as f:
para_config = json.load(f)["para"]
with open(json_file) as f:
with open(json_file, encoding='UTF-8') as f:
mysql_config = json.load(f)["mysql"]
with open(json_file) as f:
with open(json_file, encoding='UTF-8') as f:
postgre_config = json.load(f)["postgresql"]
with open(json_file) as f:
with open(json_file, encoding='UTF-8') as f:
redis_config = json.load(f)["redis"]
# 全局参数
......
start.sh
View file @ 62c4edc6
......@@ -3,3 +3,4 @@ ps -ef|grep realtime_dispatch.py|grep -v grep |awk {'print $2'}
kill -9 ps
python3 topo_update_server.py
nohup python3 realtime_dispatch.py --active >/dev/null 2>&1 &
nohup flask run --host=0.0.0.0 >/dev/null 2>&1 &
\ No newline at end of file
tables.py
View file @ 62c4edc6
......@@ -21,8 +21,8 @@ relative_path = os.path.dirname(os.path.abspath(__file__)) # 获取当前文件
json_file = os.path.join(relative_path, "config.json")
# json_file = "config.json"
with open(json_file) as f:
load_vale = json.load(f)
with open(json_file, encoding='UTF-8') as f:
load_vale = json.load(f, encoding='UTF-8')
postgre_config = load_vale["postgresql"]
......
util/area_analysis.py 0 → 100644
View file @ 62c4edc6
#!E:\Pycharm Projects\Waytous
# -*- coding: utf-8 -*-
# @Time : 2023/3/23 10:13
# @Author : Opfer
# @Site :
# @File : area_analysis.py
# @Software: PyCharm
from data.para_config import *
from data.dispatchInfo import DispatchInfo
import uuid
def area_choose(excavators_id, closer_area_id, further_area_id, further_lane_set, closer_lane_set,
logger, truck, truck_locates_dict, closer_excavator_state, further_excavator_state):
"""
两装载区均不空闲，执行二次调度
:param excavators_id: 挖机集合
:param closer_area_id: 近端装载区id
:param further_area_id: 远端装载区id
:param lane_set: 近端及远端装载区间路段集合
:param logger: 日志器
:param truck: 车辆对象
:param truck_locates_dict: 车辆位置
:return: target_excavator
"""
logger.info("excavator_hold_truck_list")
logger.info(truck.excavator_hold_truck_list)
logger.info(list(excavators_id)[0])
logger.info(list(excavators_id)[1])
logger.info(truck.excavator_hold_truck_list[list(excavators_id)[0]])
logger.info(truck.excavator_hold_truck_list[list(excavators_id)[1]])
# 统计驶往两装载区的车辆
arrival_truck_set = truck.excavator_hold_truck_list[list(excavators_id)[0]] \
+ truck.excavator_hold_truck_list[list(excavators_id)[1]]
# 统计车辆抵达时间
arrival_truck_reach_time = [truck.cur_truck_reach_excavator[truck.truck_uuid_to_index_dict[truck_id]] for \
truck_id in arrival_truck_set]
# arrival_truck_set = ['309705a0-5ddf-4559-b6c4-ee17a57677ad', '899705a0-5ddf-4559-b6c4-ee17a57677ad']
#
# arrival_truck_reach_time = [8.04, 6.05]
logger.info("arrival_truck_reach_time")
logger.info(arrival_truck_reach_time)
logger.info("arrival_truck_set")
logger.info(arrival_truck_set)
arrival_truck_list = list(zip(np.array(arrival_truck_set), np.array(arrival_truck_reach_time)))
arrival_truck_list = sorted(arrival_truck_list, key=lambda item: item[1])
logger.info("arrival_truck_list")
logger.info(arrival_truck_list)
logger.info("arrival_truck_list")
logger.info(arrival_truck_list)
# 统计不同状态车辆数量
goto_closer_area_num = 0
goto_further_area_num = 0
before_cross_num = 0
for truck_id, reach_time in arrival_truck_list:
if truck_id in truck_locates_dict:
truck_lane_id = truck_locates_dict[truck_id]
# 车辆已经经过近端装载区
if truck_lane_id in further_lane_set:
# 前往远端装载区车辆数加1
goto_further_area_num += 1
# 车辆未经过近端装载区
elif truck_lane_id in closer_lane_set:
# 前往近端或近端装载区车辆数加1
goto_closer_area_num += 1
else:
before_cross_num += 1
else:
continue
logger.info("closer_excavator_state")
logger.info(closer_excavator_state)
logger.info(goto_closer_area_num)
logger.info("further_excavator_state")
logger.info(further_excavator_state)
logger.info(goto_further_area_num)
logger.info(before_cross_num)
# 近端挖机空闲
if closer_excavator_state == 0 and goto_closer_area_num == 0:
logger.info("近端挖机空闲, 调度车辆前往近端装载区")
target_excavator = DispatchInfo.load_excavator_dict[closer_area_id]
# truck_dispatch_to_redis(request_truck_id, group_id, DispatchInfo.load_excavator_dict[closer_area_id])
# 远端挖机空闲
elif further_excavator_state == 0 and goto_further_area_num == 0:
logger.info("远端挖机空闲, 调度车辆前往远端装载区")
target_excavator = DispatchInfo.load_excavator_dict[further_area_id]
# truck_dispatch_to_redis(request_truck_id, group_id, DispatchInfo.load_excavator_dict[further_area_id])
# 两挖机均不空闲
else:
# goto_further_area_num -= 1
logger.info("goto_further_area_num-goto_closer_area_num")
logger.info(goto_further_area_num)
logger.info(goto_closer_area_num)
# 默认当前请调车辆与近端装载点前没有车辆，因此前往近端装载区的车辆仅其本身
# goto_closer_area_num = 1
# 在远处排队等待的车辆更少
if goto_closer_area_num + before_cross_num > goto_further_area_num:
logger.info("远端挖机排队时间短, 调度车辆前往")
target_excavator = DispatchInfo.load_excavator_dict[further_area_id]
# truck_dispatch_to_redis(request_truck_id, group_id, DispatchInfo.load_excavator_dict[further_area_id])
else:
logger.info("近端挖机排队时间短, 调度车辆前往")
target_excavator = DispatchInfo.load_excavator_dict[closer_area_id]
# truck_dispatch_to_redis(request_truck_id, group_id, DispatchInfo.load_excavator_dict[closer_area_id])
return target_excavator
def area_analysis(load_area_uuid):
"""
Analysis which area is closer.
:param load_area_uuid:
:return: closer_area_uuid, further_area_uuid
"""
try:
excavator_uuid_to_load_area_uuid_dict = get_value("excavator_uuid_to_load_area_uuid_dict")
load_area_uuid = list(load_area_uuid)
load_area_uuid[0] = excavator_uuid_to_load_area_uuid_dict[load_area_uuid[0]]
load_area_uuid[1] = excavator_uuid_to_load_area_uuid_dict[load_area_uuid[1]]
distance_a = session_postgre.query(WalkTimePark)\
.filter_by(load_area_id=load_area_uuid[0]).first().park_load_distance
distance_b = session_postgre.query(WalkTimePark)\
.filter_by(load_area_id=load_area_uuid[1]).first().park_load_distance
if distance_a > distance_b:
return load_area_uuid[1], load_area_uuid[0]
else:
return load_area_uuid[0], load_area_uuid[1]
except Exception as es:
logger.error("装载区距离分析异常")
logger.error(es)
return load_area_uuid[0], load_area_uuid[1]
def truck_pass_first_area(truck_id, lane_id, closer_entrance_node_id, further_entrance_node_id):
"""
Truck has gone through the first area.
:param truck_id:
:param lane_id:
:param closer_entrance_node_id:
:param further_entrance_node_id:
:return:
"""
# try:
def backtracking(root_node):
from collections import deque
que = deque([root_node])
while que:
size = len(que)
for _ in range(size):
cur_node = que.popleft()
if cur_node is None:
continue
logger.info(cur_node)
if cur_node == closer_entrance_node_id:
logger.info("closer_entrance_node")
return 0
if cur_node == further_entrance_node_id:
logger.info("further_entrance_node")
return 1
for item in session_postgre.query(Lane).filter_by(StartNodeId=cur_node).all():
if item:
que.append(item.EndNodeId)
node_id = session_postgre.query(Lane).filter_by(Id=lane_id).first().EndNodeId
if backtracking(root_node=node_id) == 0:
return True
else:
return False
# except Exception as es:
# logger.error("车辆行驶位置判断异常")
# logger.error(es)
# return True
def get_trucks_locate():
"""
get trucks locates.
:return: truck_locate_dict
"""
try:
truck_name_to_uuid_dict = get_value("truck_name_to_uuid_dict")
truck_locate_dict = {}
device_name_set = redis2.keys()
for item in device_name_set:
item = item.decode(encoding='utf-8')
key_value_dict = redis2.hgetall(item)
if str_to_byte('type') in key_value_dict:
device_type = key_value_dict[str_to_byte('type')]
is_online = key_value_dict[str_to_byte('online')]
key_set = key_value_dict.keys()
if (device_type == str_to_byte("1")) \
and (str_to_byte('online') in key_set) \
and (bytes.decode(is_online) in ["true" or "True"]) \
and (str_to_byte('laneId') in key_set):
truck_locate = key_value_dict[str_to_byte('laneId')]
# logger.error(item)
# logger.error(eval(truck_locate))
if eval(truck_locate) is not '':
truck_locate_dict[truck_name_to_uuid_dict[item]] = eval(truck_locate)
logger.info(truck_locate_dict)
else:
continue
return truck_locate_dict
except Exception as es:
logger.error("车辆所在路段读取异常")
logger.error(es)
return {}
def truck_dispatch_to_redis(truck_id, group_id, excavator_id):
"""
write truck dispatch to redis.
:param truck_id:
:param group_id:
:param excavator_id:
:return:
"""
# 查询车辆相关派车计划
record = {}
try:
# dump_id = DispatchInfo.unload_area_dump_dict[unload_area_id]
item = (session_mysql.query(DispatchSetting).filter_by(exactor_id=excavator_id, group_id=group_id,
isdeleted=0, ).first())
if item is None:
raise Exception("调度计划配置异常")
except Exception as es:
item = (session_mysql.query(DispatchSetting).filter_by(group_id=group_id, isdeleted=0, ).first())
logger.error(es)
# 其余调度信息写入
try:
# record["dispatchId"] = item.id
record["dispatchId"] = str(uuid.uuid1())
record["exactorId"] = item.exactor_id
record["loadAreaId"] = item.load_area_id
record["dumpId"] = item.dump_id
record["unloadAreaId"] = item.unload_area_id
record["groupId"] = group_id
record["isdeleted"] = False
record["isTemp"] = False
record["haulFlag"] = -1
record["groupName"] = DispatchInfo.group_name[group_id]
logger.info(f'{truck_id} redis 注入 {record}')
except Exception as es:
logger.error("调度结果写入异常-矿卡空载")
logger.error(es)
finally:
redis5.set(truck_id, str(json.dumps(record)))
def get_lanes_between_entrances(closer_node_id, further_node_id):
"""
get lanes between two entrance nodes.
:param closer_node_id:
:param further_node_id:
:return: lane set
"""
try:
max_find_it = 100
next_node_id = closer_node_id
lane_set = []
while max_find_it > 0 and next_node_id != further_node_id:
item = session_postgre.query(Lane).filter_by(StartNodeId=next_node_id, Type=2).first()
if item:
next_lane_id = item.Id
next_node_id = item.EndNodeId
lane_set.append(next_lane_id)
max_find_it -= 1
max_find_it -= 1
return lane_set
except Exception as es:
logger.error("获取装载区间路段异常")
logger.error(es)
return []
def get_excavator_state(excavator_id):
"""
get group_excavators state.
:param excavator_id:
:return: state
"""
try:
logger.error(excavator_id)
device_name = session_mysql.query(Equipment).filter_by(id=excavator_id, device_type=2).first().device_name
key_value_dict = redis2.hgetall(device_name)
if str_to_byte('online') in key_value_dict:
is_online = key_value_dict[str_to_byte('online')]
else:
logger.warning(f'挖机 {device_name} 不在线')
return 0
key_set = key_value_dict.keys()
state = 100
if (str_to_byte('online') in key_set) and (bytes.decode(is_online) in ["true" or "True"]):
if str_to_byte('workState') in key_set:
state = key_value_dict[str_to_byte('workState')]
else:
logger.warning(f'挖机 {device_name} 状态未知')
return 0
return int(float(byte_to_str(state)))
except Exception as es:
logger.error("挖机状态读取异常")
logger.error(es)
return 0
\ No newline at end of file
util/core_exception.py
View file @ 62c4edc6
......@@ -13,20 +13,25 @@ code_msg = {101: "调度失败",
103: "数组越界异常",
104: "数组维度不一致",
105: "数组越界",
106: "挖机不存在或信息缺失"}
106: "挖机不存在或信息缺失",
107: "redis2设备信息与数据库不一致",
108: "redis5调度信息为空",
109: "配置信息不完整",
110: "车辆行程信息异常",
111: "挖机不存在或信息确实"}
class CoreException(Exception):
""" class for my exception."""
def __init__(self, Code: int = None, ErrorInfo: str = None):
def __init__(self, Code: int = 101, ErrorInfo: str = None):
super().__init__(self, ErrorInfo)
self.__error_info = ErrorInfo
self.code = Code
self.__traceback = ""
def __str__(self) -> str:
return '{0:}-{1:} \n{2:}'.format(code_msg[self.code], self.__error_info, self.__traceback)
return '{0:}-{1:} {2:}'.format(code_msg[self.code], self.__error_info, self.__traceback)
# def with_traceback(self, tb) -> BaseException:
# self.__traceback = tb
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment