Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# @date 2010-02-20
from __future__ import absolute_import
from __future__ import print_function
import os
import subprocess
import sys
import time
from multiprocessing import Process, freeze_support
sys.path.append(os.path.join(os.environ["SUMO_HOME"], "tools"))
import sumolib # noqa
import traci # noqa
PORT = sumolib.miscutils.getFreeSocketPort()
DELTA_T = 1000
sumoBinary = sumolib.checkBinary(sys.argv[1])
def traciLoop(port, traciEndTime, index, orderOdd):
orderTime = 0.25
time.sleep(orderTime * index) # assure ordering of outputs
print("Starting process %s" % (index))
sys.stdout.flush()
time.sleep(orderTime * index) # assure ordering of outputs
step = 1
try:
traci.init(port)
if orderOdd and index % 2 == 1:
traci.setOrder(index)
sumoStop = False
# @author Michael Behrisch
# @author Jakob Erdmann
# @date 2008-05-12
from __future__ import absolute_import
from __future__ import print_function
import os
import subprocess
import sys
import time
THIS_DIR = os.path.abspath(os.path.dirname(__file__))
sys.path.append(os.path.join(THIS_DIR, '..', "tools"))
import sumolib # noqa
PORT = str(sumolib.miscutils.getFreeSocketPort())
server_args = sys.argv[1:] + ["--remote-port", PORT]
binaryDir, server = os.path.split(server_args[0])
client = "TraCITestClient"
if server.endswith("D") or server.endswith("D.exe"):
client += "D"
client_args = [os.path.join(binaryDir, client), "-def",
"testclient.prog", "-o", "testclient_out.txt", "-p", PORT]
# start sumo as server
serverProcess = subprocess.Popen(
server_args, stdout=sys.stdout, stderr=sys.stderr)
success = False
for retry in range(7):
time.sleep(retry)
clientProcess = subprocess.Popen(
import subprocess
import sys
import shutil
sys.path.append(os.path.join(os.environ["SUMO_HOME"], "tools"))
import sumolib # noqa
import traci # noqa
sumoBinary = sumolib.checkBinary(sys.argv[1])
if sys.argv[1] == "sumo":
addOption = []
secondConfig = "sumo.sumocfg"
else:
addOption = ["-S", "-Q"]
secondConfig = "sumo_log.sumocfg"
PORT = sumolib.miscutils.getFreeSocketPort()
sumoProc = subprocess.Popen(
[sumoBinary, "-c", "sumo.sumocfg", "--remote-port", str(PORT)] + addOption)
traci.init(PORT)
subprocess.call([sumoBinary, "-c", secondConfig,
"--remote-port", str(PORT)] + addOption)
step = 0
while not step > 100:
traci.simulationStep()
vehs = traci.vehicle.getIDList()
if vehs.index("horiz") < 0 or len(vehs) > 1:
print("Something is false")
step += 1
traci.close()
sumoProc.wait()
sys.stdout.flush()
from __future__ import absolute_import
import os
import subprocess
import sys
sys.path.append(os.path.join(os.environ["SUMO_HOME"], "tools"))
import sumolib # noqa
import traci # noqa
sumoBinary = sumolib.checkBinary(sys.argv[1])
if sys.argv[1] == "sumo":
addOption = ""
else:
addOption = "-S -Q"
PORT = sumolib.miscutils.getFreeSocketPort()
def run():
"""execute the TraCI control loop"""
traci.init(PORT)
step = 0
while traci.simulation.getMinExpectedNumber() > 0 and step < 100:
traci.simulationStep()
step += 1
if step == 4:
traci.trafficlight.setProgram("center", "0")
traci.close()
sys.stdout.flush()
sumoProcess = subprocess.Popen([sumoBinary,
flow.utils.exceptions.FatalFlowError
if the render mode is not set to a valid value
"""
self.env_params = env_params
if scenario is not None:
deprecated_attribute(self, "scenario", "network")
self.network = scenario if scenario is not None else network
self.net_params = self.network.net_params
self.initial_config = self.network.initial_config
self.sim_params = sim_params
time_stamp = ''.join(str(time.time()).split('.'))
if os.environ.get("TEST_FLAG", 0):
# 1.0 works with stress_test_start 10k times
time.sleep(1.0 * int(time_stamp[-6:]) / 1e6)
# FIXME: this is sumo-specific
self.sim_params.port = sumolib.miscutils.getFreeSocketPort()
# time_counter: number of steps taken since the start of a rollout
self.time_counter = 0
# step_counter: number of total steps taken
self.step_counter = 0
# initial_state:
self.initial_state = {}
self.state = None
self.obs_var_labels = []
# simulation step size
self.sim_step = sim_params.sim_step
# the simulator used by this environment
self.simulator = simulator
# create the Flow kernel