Spaces:
Sleeping
Sleeping
File size: 3,596 Bytes
e788d0c | 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 | """
Executor: THE ONLY CODE THAT CHANGES A TRAFFIC LIGHT.
A1 asks the executor for a new green. The executor then runs the safe
transition, one second at a time:
GREEN --> CLEARANCE (walk signal off, vehicles keep green, until the
crossing is empty: the safety mask decides)
--> YELLOW (3 s for every vehicle movement losing green)
--> new GREEN
scripts/check_architecture.py fails if any other file writes a light.
The masters call it; nothing else may.
"""
from ..core import config
from .backend import traci
YELLOW = config.agents()["safety"]["yellow"]
class Executor:
def __init__(self, models, mask, log):
self.models, self.mask, self.log = models, mask, log
self.st = {}
self._written = {}
def start(self, now):
self.now = now
for tls, m in self.models.items():
self.st[tls] = {"green": m.green_phases[0], "since": now, "phase": "green"}
self._write(tls, m.green_phases[0])
# ---------------------------------------------------------------- for A1
def status(self, tls):
s = self.st[tls]
return {"green": s["green"], "busy": s["phase"] != "green",
"time_in_green": self.now - s["since"] if s["phase"] == "green" else 0}
def request(self, tls, target, now, emergency=False, extra_yellow=0, extra_clearance=0):
s = self.st[tls]
if s["phase"] != "green" or target == s["green"]:
return False
if not self.mask.min_green_ok(now - s["since"], emergency):
return False
crossings = self.mask.crossings_needing_clearance(tls, s["green"], target)
s.update(target=target, t0=now, crossings=crossings,
ctx={"extra_clearance": extra_clearance},
yellow=YELLOW + extra_yellow, phase="clearance" if crossings else "yellow")
return True
# ---------------------------------------------------------------- every second
def step(self, now):
self.now = now
for tls, s in self.st.items():
m = self.models[tls]
if s["phase"] == "clearance":
state = list(s["green"])
for c in s["crossings"]:
for i in m.crossings[c]["indices"]:
state[i] = "r" # "don't walk"; vehicles keep green
self._write(tls, "".join(state))
if self.mask.clearance_done(tls, s["crossings"], now - s["t0"], now, s["ctx"]):
s.update(phase="yellow", t0=now)
if s["phase"] == "yellow":
if now - s["t0"] < s["yellow"]:
self._write(tls, self._yellow(m, s["green"], s["target"]))
else:
s.update(phase="green", green=s["target"], since=now)
self._write(tls, s["green"])
@staticmethod
def _yellow(m, cur, target):
out = []
for i, (c, t) in enumerate(zip(cur, target)):
link = m.links[i]
if c in "Gg" and t not in "Gg":
out.append("r" if link and link.is_crossing else "y")
elif link and link.is_crossing and t not in "Gg":
out.append("r")
else:
out.append(c if c in "Gg" else "r")
return "".join(out)
def _write(self, tls, state):
if self._written.get(tls) != state:
traci.trafficlight.setRedYellowGreenState(tls, state)
self._written[tls] = state
|