#!/usr/bin/env python
# -*-coding:utf-8 -*-
import copy
import datetime
import threading
import sched
import time
import data_api
import pandas as pd
from agv_routiong_process_fusion import AGVRoutingProcess
import numpy as np
import logging
import json
import mqtt_server as mq_server

scheduler = sched.scheduler(time.time, time.sleep)



class NpEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, np.integer):
            return int(obj)
        elif isinstance(obj, np.floating):
            return float(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, datetime.datetime):
            return obj.strftime('%Y-%m-%d %H:%M:%S')
        else:
            return super(NpEncoder, self).default(obj)

#模拟器动态重排
def re_routing(stop_dt,stop_agv_pos={}):
    write_data = json.load(open("data/random_task.json", encoding='utf-8'))
    json_data = json.load(open("data/random_task.json", encoding='utf-8'))['data']
    random_task_df=pd.DataFrame(json_data['unDispatchedTasks'])
    agv_list = json_data['agvs']
    win_time_df = pd.read_excel("data/task_dispatch_time.xlsx", index_col=False)
    win_start_time = pd.to_datetime(win_time_df["startTime"].min(), "%Y-%m-%d %H:%M:%S")
    win_end_time = pd.to_datetime(win_time_df["endTime"].max(), "%Y-%m-%d %H:%M:%S")
    win_sum_time = (win_end_time - win_start_time).total_seconds()
    task_list = list(win_time_df['taskId'])
    task_count = win_time_df['taskId'].value_counts()
    move_start_time = win_start_time + datetime.timedelta(seconds=stop_dt)
    rest_task_detail_df = win_time_df[win_time_df['endTime'] > move_start_time]
    finish_task_df = win_time_df[win_time_df['endTime'] <=move_start_time]
    rest_task_count =rest_task_detail_df['taskId'].value_counts()
    rest_task_df = pd.concat([task_count,rest_task_count], axis=1, keys='index')
    rest_task_df.reset_index(inplace=True)
    rest_task_df.sort_index()
    rest_task_df = rest_task_df.fillna(value=0)
    dis_task_json_model=json.load(open("data/full_xicai_json_map.json", encoding='utf-8'))['data']['dispatchedTasks'][0]
    sections_json_model=dis_task_json_model['sections'][0]
    dis_task_json_list=[]
    un_task_json_list=[]
    for k in range(len(json_data['unDispatchedTasks'])):
        task_detail=json_data['unDispatchedTasks'][k]
        task_id=task_detail['id']
        if task_id in list(win_time_df['taskId']):
            dis_task_json=copy.deepcopy(dis_task_json_model)
            dis_task_detail_df=win_time_df[win_time_df['taskId']==task_id]
            dis_task_json["status"]='EXECUTING' if task_id in list(finish_task_df['taskId']) else 'READY'
            dis_task_json["taskId"]=task_id
            dis_task_json['id']=task_id
            dis_task_json["positionIds"]=task_detail['positionIds']
            dis_task_json["agvId"]=dis_task_detail_df['taskAgv'].values[0]
            sections=[]
            for  index ,col in dis_task_detail_df.iterrows():
                sections_json_model=copy.deepcopy(sections_json_model)
                sections_json_model['taskId']=col['taskId']
                sections_json_model['startPositionId']=col['startPositionId']
                sections_json_model['endPositionId'] =col['endPositionId']
                sections_json_model['agvId'] = col['taskAgv']
                sections_json_model['executeTime'] = col['startTime']
                end_time=col['endTime']
                status='READY'
                if dis_task_json["status"]=='EXECUTING':
                    if end_time<=move_start_time:
                        status='FINISHED'
                    else:
                        status = 'READY'
                sections_json_model['status'] = status
                sections.append(sections_json_model)
            dis_task_json['sections'] = sections
            dis_task_json_list.append(dis_task_json)
        else:
            un_task_json_list.append(task_detail)
    update_agv_list=[]
    for agv in agv_list:
        agv_id=agv['id']
        if agv_id in stop_agv_pos.keys():
            finish_task_agv_df=finish_task_df[finish_task_df['taskAgv'] == agv_id]
            if finish_task_agv_df.shape[0]>0:
                positionId=finish_task_agv_df.sort_values(by = ['endTime'],ascending=True)['endPositionId'].values[-1]
                agv['realTimeVO']['positionId']=positionId
                agv['realTimeVO']["x"]=stop_agv_pos[agv_id][0]
                agv['realTimeVO']["y"]=stop_agv_pos[agv_id][1]
        update_agv_list.append(agv)
    json_data['unDispatchedTasks']=un_task_json_list
    json_data['dispatchedTasks'] = dis_task_json_list
    json_data['agvs'] = update_agv_list
    write_data['data'] = json_data

    with open('data/random_task_monitor.json', 'w', encoding="utf-8") as f:
        json.dump(write_data, f, sort_keys=True, ensure_ascii=False, indent=2,cls=NpEncoder)
    print('generate new task json  finish !')
    logging.info(datetime.datetime.now())
    data = open("data/random_task.json", encoding='utf-8')
    # data = open("data/random_task_monitor.json", encoding='utf-8')
    json_data = write_data
    agv_routing_cls=AGVRoutingProcess()
    agv_routing_cls.init_map_data(json_data['data'])
    agv_routing_cls.available_crash_edges = agv_routing_cls._map.get_vailable_crash_edges()
    agv_routing_cls.entire_process(json_data['data'],1)
    logging.info(datetime.datetime.now())

#定时获取资源池任务执行任务调度
def monitor_execution():
    # 执行需要定时执行的任务
    print("这是5秒钟执行一次的任务")
    json_data=data_api.receive()
    # json_data =json.load( open("data/random_task.json", encoding='utf-8'))
    b = json.dumps(json_data)
    f2 = open('data/monitor_json.json', 'w')
    f2.write(b)
    f2.close()
    routing_basic_data = json_data['data']
    print(json.dumps(routing_basic_data))
    agv_routing_cls = AGVRoutingProcess()
    agv_routing_cls.init_map_data(routing_basic_data)
    agv_routing_cls.available_crash_edges = agv_routing_cls._map.get_vailable_crash_edges()
    agv_routing_cls.entire_process(routing_basic_data, '1')
    logging.info(datetime.datetime.now())
    # 每次执行完任务后，计算下一次任务的时间
    scheduler.enter(5, 1, monitor_execution, ())

def run_mq_server():
    print("run mq server")
    mq_server.run()

def run_monitor_func():
    print("run monitor func")
    # # 初次启动任务调度器
    scheduler.enter(5, 1, monitor_execution, ())
    scheduler.run()

def get_task_threading():
    threads = []  # 定义一个线程池
    t1 = threading.Thread(target=run_mq_server)
    threads.append(t1)  # 把t1线程装到线程池里
    t2 = threading.Thread(target=run_monitor_func)
    threads.append(t2)  # 把t2线程装到线程池里
    for t in threads:
        t.start()


if __name__ == '__main__':

    # re_routing(  100.4,{"agvId0":(19,20)})
    get_task_threading()
解析一下这段代码