RealLight/utils.py at main · camuslab/RealLight · GitHub

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
from algos import FixedTime, RealLight
from copy import copy

def generator_agents(args, env):
    agents = []
    for index in range(args['n_agent']):
        args_ = copy(args)
        args_['name'] = env.intersection_name[index]
        args_['in_dim'] = env.observation_space[index]

        if 'daejeon_seogu' == args['env_name']:
            n_action = env.action_space[index] - 1
        else:
            n_action = 4

        args_['out_dim'] = n_action * 5
        args_['phase_dim'] = n_action
        agents.append(RealLight(args_))
    return agents

def generator_fixed_agents(args, env):
    agents = []
    for index in range(args['n_agent']):
        args_ = copy(args)
        args_['name'] = env.intersection_name[index]
        agents.append(FixedTime(args_))
    return agents

## not use yellow phase
def get_actions(agents, state_lst):
    phase_lst = []

    for index, agent in enumerate(agents):
        if agent.count == 0:
            agent.state = state_lst[index]
            action, prob, phase, timing = agent.get_action(agent.state)
            agent.action = action
            agent.prob = prob
            agent.phase = phase
            agent.timing = timing
            agent.count = timing
        else:
            phase = agent.phase
        agent.count -= 1

        phase_lst.append({
            f'{agent.args["name"]}': phase + 1,
            'yellow_status': agent.yellow['status']
        })
    return phase_lst

## use yellow phase
def get_actions_yellow(agents, state_lst):
    phase_lst = []

    for index, agent in enumerate(agents):
        if agent.yellow['status'] == None or (agent.count == 0 and agent.yellow['time'] == 0):
            agent.state = state_lst[index]
            action, prob, phase, timing = agent.get_action(agent.state)
            agent.action = action
            agent.prob = prob
            agent.phase = phase
            agent.timing = timing
            agent.count = timing
            agent.count -= 1
            agent.yellow['status'] = False

        elif agent.count == 0 and agent.yellow['status']:
            phase = agent.phase
            agent.yellow['time'] -= 1

        else:
            phase = agent.phase
            agent.count -= 1

        phase_lst.append({
            f'{agent.args["name"]}': phase + 1,
            'yellow_status': agent.yellow['status']
        })

        if agent.count == 0 and agent.yellow['status'] == False:
            agent.yellow['status'] = True
            agent.yellow['time'] = 3
        elif agent.yellow['time'] == 0 and agent.yellow['status'] == True:
            agent.yellow['status'] = None
    return phase_lst

def get_fixed_agent_actions(agents):
    phase_lst = []

    for _, agent in enumerate(agents):
        phase = agent.get_action()
        phase_lst.append({
            f'{agent.args["name"]}': phase + 1,
            'yellow_status': agent.yellow['status']
        })
    return phase_lst

def get_fixed_agent_actions_yellow(agents):
    phase_lst = []

    for _, agent in enumerate(agents):
        phase = agent.get_action_yellow()
        phase_lst.append({
            f'{agent.args["name"]}': phase + 1,
            'yellow_status': agent.yellow['status']
        })
    return phase_lst