diff --git a/rllib/contrib/sumo/connector.py b/rllib/contrib/sumo/connector.py new file mode 100644 index 000000000..6b1d3d1d4 --- /dev/null +++ b/rllib/contrib/sumo/connector.py @@ -0,0 +1,253 @@ +""" RLLIB SUMO Utils - SUMO Connector + + Author: Lara CODECA lara.codeca@gmail.com + + See: + https://github.com/lcodeca/rllibsumoutils + https://github.com/lcodeca/rllibsumodocker + for further details. +""" + +import logging +import os +import sys + +# Attach $SUMO_HOME/tools to the path to import SUMO libraries +if "SUMO_HOME" in os.environ: + sys.path.append(os.path.join(os.environ["SUMO_HOME"], "tools")) +else: + raise Exception("Please declare environment variable 'SUMO_HOME'") + +############################################################################### + +logging.basicConfig() +logger = logging.getLogger(__name__) + +############################################################################### + +DEFAULT_CONFIG = { + # SUMO Connector. Default: "libsumo". + # Possible strings: "libsumo" or "traci" + "sumo_connector": "libsumo", + # Enable the GUI, works only with traci + "sumo_gui": False, + # SUMO configuration file. Required. String. + "sumo_cfg": None, + # Overides . + # Required when using multiple environments at the same time. String. + "sumo_output": "", + # Additional parameter for the SUMO command line. + # It cannot contain --output-prefix. List of strings. + "sumo_params": None, + # Enable TraCI trace file. Boolean. + "trace_file": False, + # SUMO Simulation ending time, in seconds. Float. + "end_of_sim": None, + # SUMO update frequency in number of traci.simulationStep() calls. Integer. + "update_freq": 1, + # SUMO tripinfo file as defined in the sumo configuration file in + # . + # Required for gathering metrics only. String. + "tripinfo_keyword": None, + # SUMO tripinfo XML Schema file. + # Required for gathering metrics only. String. + "tripinfo_xml_schema": None, + # Logging legel. Should be one of DEBUG, INFO, WARN, or ERROR. + "log_level": "WARN", + # Anything. User defined. + "misc": None, +} + +############################################################################### + + +class SUMOConnector(object): + """ Handler of a SUMO simulation. """ + + def __init__(self, config): + """ + Initialize SUMO and sets the beginning of the simulation. + + Param: + config: Dict. See DEFAULT_CONFIG. + """ + self._config = config + + # logging + level = logging.getLevelName(config["log_level"]) + logger.setLevel(level) + + # libsumo vs TraCI selection + if config["sumo_connector"] == "libsumo": + import libsumo as traci + elif config["sumo_connector"] == "traci": + import traci + else: + raise Exception( + "ERROR: '{}' is not a valid option for 'sumo_connector'. " + "The possible connectors are 'traci' or 'libsumo'.".format( + config["sumo_connector"])) + + # TraCI Handler and SUMO simulation + logger.debug("Starting SUMOConnector in process %d.", os.getpid()) + self._sumo_label = "{}".format(os.getpid()) + self._sumo_output_prefix = "{}{}".format(config["sumo_output"], + self._sumo_label) + self._sumo_parameters = ["sumo", "-c", config["sumo_cfg"]] + if config["sumo_gui"] and config["sumo_connector"] == "traci": + self._sumo_parameters[0] = "sumo-gui" + self._sumo_parameters.extend(["--start", "--quit-on-end"]) + if config["sumo_params"] is not None: + self._sumo_parameters.extend(config["sumo_params"]) + self._sumo_parameters.extend( + ["--output-prefix", self._sumo_output_prefix]) + logger.debug("SUMO command line: %s", str(self._sumo_parameters)) + if config["trace_file"]: + traci.start( + self._sumo_parameters, + traceFile="{}.tracefile.log".format(self._sumo_output_prefix)) + else: + traci.start(self._sumo_parameters) + self.traci_handler = traci + # From now on, the call must always be to self.traci_handler + + self._is_ongoing = True + self._start_time = self.traci_handler.simulation.getTime() + self._sumo_steps = 0 + self._manually_stopped = False + + # Initialize simulation + self._initialize_simulation() + + # Initialize metrics + self._initialize_metrics() + + def __del__(self): + logger.debug("Deleting SUMOConnector in process %d.", os.getpid()) + try: + self.end_simulation() + except KeyError: + logger.warning("Simulation %s already closed.", self._sumo_label) + + ########################################################################### + + def _initialize_simulation(self): + """ Specific simulation initialization. """ + raise NotImplementedError + + def _initialize_metrics(self): + """ Specific metrics initialization """ + raise NotImplementedError + + def _default_step_action(self, agents): + """ Specific code to be executed in every simulation step """ + raise NotImplementedError + + ########################################################################### + + def _stopping_condition(self, current_step_counter, until_end): + """ Computes the stopping condition. """ + if self._manually_stopped: + return True + if self.traci_handler.simulation.getMinExpectedNumber() <= 0: + # No entities left in the simulation. + return True + if self._config["end_of_sim"] is not None: + if self.traci_handler.simulation.getTime( + ) > self._config["end_of_sim"]: + # the simulatio reach the predefined (from parameters) end + return True + if (current_step_counter == self._config["update_freq"] + and not until_end): + return True + return False + + def step(self, until_end=False, agents=set()): + """ + Runs a "learning" step and returns if the simulation has finished. + This function in meant to be called by the RLLIB Environment. + + Params: + until_end: Bool. If True, run the sumo simulation + until the end. + agents: Set(String). It passes the agents to the + _default_step_action function. + + Return: + Bool. True iff the simulation is still ongoing. + """ + # Execute SUMO steps until the learning needs to happen + current_step_counter = 0 + logger.debug( + "=================================================================" + ) + while not self._stopping_condition(current_step_counter, until_end): + logger.debug("[%s] Current step counter: %d, Update frequency: %d", + str(until_end), current_step_counter, + self._config["update_freq"]) + self.traci_handler.simulationStep() + self._sumo_steps += 1 + current_step_counter += 1 + self._default_step_action(agents) + logger.debug( + "=================================================================" + ) + + # If the simulation has finished + if self.is_ongoing_sim(): + return True + logger.debug("The SUMO simulation is done.") + return False + + def fast_forward(self, time): + """ + Move the simulation forward (without doing anything else) until the + given time. + Param: + time: Float, simulation time in seconds. + """ + logger.debug("Fast-forward from time %.2f", + self.traci_handler.simulation.getTime()) + self.traci_handler.simulationStep(float(time)) + logger.debug("Fast-forward to time %.2f", + self.traci_handler.simulation.getTime()) + + ########################################################################### + + def get_sumo_steps(self): + """ Returns the total number of traci.simulationStep() calls.""" + return self._sumo_steps + + def is_ongoing_sim(self): + """ Return True iff the SUMO simulation is still ongoing. """ + if self._manually_stopped: + return False + if self.traci_handler.simulation.getMinExpectedNumber() <= 0: + # No entities left in the simulation. + return False + if self._config["end_of_sim"] is not None: + if self.traci_handler.simulation.getTime( + ) > self._config["end_of_sim"]: + # the simulatio reach the predefined (from parameters) end + return False + return True + + def get_current_time(self): + """ + Returns the current simulation time, or None if the simulation is + not ongoing. + """ + if self.is_ongoing_sim(): + return self.traci_handler.simulation.getTime() + return None + + def end_simulation(self): + """ Forces the simulation to stop. """ + if self.is_ongoing_sim(): + logger.info("Closing TraCI %s", self._sumo_label) + self._manually_stopped = True + self.traci_handler.close() + else: + logger.warning("TraCI %s is already closed.", self._sumo_label) + + ########################################################################### diff --git a/rllib/contrib/sumo/utils.py b/rllib/contrib/sumo/utils.py new file mode 100644 index 000000000..a689a65c5 --- /dev/null +++ b/rllib/contrib/sumo/utils.py @@ -0,0 +1,324 @@ +""" RLLIB SUMO Utils - SUMO Connector Wrapper + + Author: Lara CODECA lara.codeca@gmail.com + + See: + https://github.com/lcodeca/rllibsumoutils + https://github.com/lcodeca/rllibsumodocker + for further details. +""" + +import collections +from copy import deepcopy +import logging +import os +from pprint import pformat +import sys + +from lxml import etree + +from ray.rllib.contrib.sumo.connector import SUMOConnector, DEFAULT_CONFIG + +# """ Import SUMO library """ +if "SUMO_HOME" in os.environ: + sys.path.append(os.path.join(os.environ["SUMO_HOME"], "tools")) + # from traci.exceptions import TraCIException + import traci.constants as tc +else: + sys.exit("please declare environment variable 'SUMO_HOME'") + +############################################################################### + +logging.basicConfig() +logger = logging.getLogger(__name__) + +############################################################################### + + +def sumo_default_config(): + """ Return the default configuration for the SUMO Connector. """ + return deepcopy(DEFAULT_CONFIG) + + +############################################################################### + + +class SUMOUtils(SUMOConnector): + """ + A wrapper for the interaction with the SUMO simulation that adds + functionalities. + """ + + def _initialize_metrics(self): + """ Specific metrics initialization """ + # Default TripInfo file metrics + self.tripinfo = collections.defaultdict(dict) + self.personinfo = collections.defaultdict(dict) + + ########################################################################### + # TRIPINFO FILE + + def process_tripinfo_file(self): + """ + Closes the TraCI connections, then reads and process the tripinfo + data. It requires "tripinfo_xml_file" and "tripinfo_xml_schema" + configuration parametes set. + """ + + if "tripinfo_keyword" not in self._config: + raise Exception( + "Function process_tripinfo_file requires the parameter " + "'tripinfo_keyword' set.", self._config) + + if "tripinfo_xml_schema" not in self._config: + raise Exception( + "Function process_tripinfo_file requires the parameter " + "'tripinfo_xml_schema' set.", self._config) + + # Make sure that the simulation is finished and the tripinfo file is + # written. + self.end_simulation() + + # Reset the data structures. + self.tripinfo = collections.defaultdict(dict) + self.personinfo = collections.defaultdict(dict) + + schema = etree.XMLSchema(file=self._config["tripinfo_xml_schema"]) + parser = etree.XMLParser(schema=schema) + tripinfo_file = "{}{}".format(self._sumo_output_prefix, + self._config["tripinfo_keyword"]) + tree = etree.parse(tripinfo_file, parser) + + logger.info("Processing %s tripinfo file.", tripinfo_file) + for element in tree.getroot(): + if element.tag == "tripinfo": + self.tripinfo[element.attrib["id"]] = dict(element.attrib) + elif element.tag == "personinfo": + self.personinfo[element.attrib["id"]] = dict(element.attrib) + stages = [] + for stage in element: + stages.append([stage.tag, dict(stage.attrib)]) + self.personinfo[element.attrib["id"]]["stages"] = stages + else: + raise Exception("Unrecognized element in the tripinfo file.") + logger.debug("TRIPINFO: \n%s", pformat(self.tripinfo)) + logger.debug("PERSONINFO: \n%s", pformat(self.personinfo)) + + def get_timeloss(self, entity, default=float("NaN")): + """ Returns the timeLoss computed by SUMO for the given entity. """ + + if entity in self.tripinfo: + logger.debug("TRIPINFO for %s", entity) + if "timeLoss" in self.tripinfo[entity]: + logger.debug("timeLoss %s", self.tripinfo[entity]["timeLoss"]) + return float(self.tripinfo[entity]["timeLoss"]) + logger.debug("timeLoss not found.") + return default + elif entity in self.personinfo: + logger.debug("PERSONINFO for %s", entity) + logger.debug("%s", pformat(self.personinfo[entity])) + time_loss, ts_found = 0.0, False + for _, stage in self.personinfo[entity]["stages"]: + if "timeLoss" in stage: + logger.debug("timeLoss %s", stage["timeLoss"]) + time_loss += float(stage["timeLoss"]) + ts_found = True + if not ts_found: + logger.debug("timeLoss not found.") + return default + if time_loss <= 0: + logger.debug("ERROR: timeLoss is %.2f", time_loss) + return default + logger.debug("total timeLoss %.2f", time_loss) + return time_loss + else: + logger.debug("Entity %s not found.", entity) + return default + + def get_depart(self, entity, default=float("NaN")): + """ + Returns the departure recorded by SUMO for the given entity. + + The functions process_tripinfo_file() needs to be called in advance + to initialize the data structures required. + + If the entity does not exist or does not have the value, it returns + the default value. + """ + if entity in self.tripinfo: + logger.debug("TRIPINFO for %s", entity) + if "depart" in self.tripinfo[entity]: + logger.debug("depart %s", self.tripinfo[entity]["depart"]) + return float(self.tripinfo[entity]["depart"]) + logger.debug("depart not found.") + elif entity in self.personinfo: + logger.debug("PERSONINFO for %s", entity) + logger.debug("%s", pformat(self.personinfo[entity])) + if "depart" in self.personinfo[entity]: + logger.debug("depart %s", self.personinfo[entity]["depart"]) + return float(self.personinfo[entity]["depart"]) + logger.debug("depart not found.") + else: + logger.debug("Entity %s not found.", entity) + return default + + def get_duration(self, entity, default=float("NaN")): + """ + Returns the duration computed by SUMO for the given entity. + + The functions process_tripinfo_file() needs to be called in advance + to initialize the data structures required. + + If the entity does not exist or does not have the value, it returns + the default value. + """ + if entity in self.tripinfo: + logger.debug("TRIPINFO for %s", entity) + if "duration" in self.tripinfo[entity]: + logger.debug("duration %s", self.tripinfo[entity]["duration"]) + return float(self.tripinfo[entity]["duration"]) + logger.debug("duration not found.") + elif entity in self.personinfo: + logger.debug("PERSONINFO for %s", entity) + logger.debug("%s", pformat(self.personinfo[entity])) + if "depart" in self.personinfo[entity]: + depart = float(self.personinfo[entity]["depart"]) + arrival = depart + for _, stage in self.personinfo[entity]["stages"]: + if "arrival" in stage: + arrival = float(stage["arrival"]) + duration = arrival - depart + if duration > 0: + logger.debug("duration %d", duration) + return duration + logger.debug("duration impossible to compute.") + else: + logger.debug("Entity %s not found.", entity) + return default + + def get_arrival(self, entity, default=float("NaN")): + """ + Returns the arrival computed by SUMO for the given entity. + + The functions process_tripinfo_file() needs to be called in advance + to initialize the data structures required. + + If the entity does not exist or does not have the value, it returns + the default value. + """ + if entity in self.tripinfo: + logger.debug("TRIPINFO for %s", entity) + if "arrival" in self.tripinfo[entity]: + logger.debug("arrival %s", self.tripinfo[entity]["arrival"]) + return float(self.tripinfo[entity]["arrival"]) + logger.debug("arrival not found.") + return default + elif entity in self.personinfo: + logger.debug("PERSONINFO for %s", entity) + arrival, arrival_found = 0.0, False + for _, stage in self.personinfo[entity]["stages"]: + if "arrival" in stage: + logger.debug("arrival %s", stage["arrival"]) + arrival = float(stage["arrival"]) + arrival_found = True + if not arrival_found: + logger.debug("arrival not found.") + return default + if arrival <= 0: + logger.debug("ERROR: arrival is %.2f", arrival) + return default + logger.debug("total arrival %.2f", arrival) + return arrival + else: + logger.debug("Entity %s not found.", entity) + return default + + def get_global_travel_time(self): + """ + Returns the global travel time computed from SUMO tripinfo data. + + The functions process_tripinfo_file() needs to be called in advance + to initialize the data structures required. + """ + gtt = 0 + for entity in self.tripinfo: + gtt += self.get_duration(entity, default=0.0) + for entity in self.personinfo: + gtt += self.get_duration(entity, default=0.0) + return gtt + + ########################################################################### + # ROUTING + + @staticmethod + def get_mode_parameters(mode): + """ + Return the correst TraCI parameters for the requested mode. + See: https://sumo.dlr.de/docs/TraCI/Simulation_Value_Retrieval.html + #command_0x87_find_intermodal_route + + Param: mode, String. + Returns: _mode, _ptype, _vtype + """ + if mode == "public": + return "public", "", "" + if mode == "bicycle": + return "bicycle", "", "bicycle" + if mode == "walk": + return "", "pedestrian", "" + return "car", "", mode # (but car is not always necessary, and it may + # creates unusable alternatives) + + def is_valid_route(self, mode, route): + """ + Handle findRoute and findIntermodalRoute results. + + Params: + mode, String. + route, return value of findRoute or findIntermodalRoute. + """ + if route is None: + # traci failed + return False + _mode, _ptype, _vtype = self.get_mode_parameters(mode) + if not isinstance(route, (list, tuple)): + # only for findRoute + if len(route.edges) >= 2: + return True + elif _mode == "public": + for stage in route: + if stage.line: + return True + elif _mode in ("car", "bicycle"): + for stage in route: + if stage.type == tc.STAGE_DRIVING and len(stage.edges) >= 2: + return True + else: + for stage in route: + if len(stage.edges) >= 2: + return True + return False + + @staticmethod + def cost_from_route(route): + """ + Compute the route cost. + Params: + route, return value of findRoute or findIntermodalRoute. + """ + cost = 0.0 + for stage in route: + cost += stage.cost + return cost + + @staticmethod + def travel_time_from_route(route): + """ + Compute the route estimated travel time. + Params: + route, return value of findRoute or findIntermodalRoute. + """ + ett = 0.0 + for stage in route: + ett += stage.estimatedTime + return ett diff --git a/rllib/examples/simulators/sumo/marlenvironment.py b/rllib/examples/simulators/sumo/marlenvironment.py new file mode 100644 index 000000000..150ed3d98 --- /dev/null +++ b/rllib/examples/simulators/sumo/marlenvironment.py @@ -0,0 +1,404 @@ +""" Example MARL Environment for RLLIB SUMO Utlis + + Author: Lara CODECA lara.codeca@gmail.com + + See: + https://github.com/lcodeca/rllibsumoutils + https://github.com/lcodeca/rllibsumodocker + for further details. +""" + +import collections +import logging +import os +import sys +from pprint import pformat + +from numpy.random import RandomState + +import gym +from ray.rllib.env import MultiAgentEnv + +from ray.rllib.contrib.sumo.utils import SUMOUtils, sumo_default_config + +# """ Import SUMO library """ +if "SUMO_HOME" in os.environ: + sys.path.append(os.path.join(os.environ["SUMO_HOME"], "tools")) + # from traci.exceptions import TraCIException + import traci.constants as tc +else: + sys.exit("please declare environment variable 'SUMO_HOME'") + +############################################################################### + +logger = logging.getLogger(__name__) + +############################################################################### + + +def env_creator(config): + """ Environment creator used in the environment registration. """ + logger.info("Environment creation: SUMOTestMultiAgentEnv") + return SUMOTestMultiAgentEnv(config) + + +############################################################################### + +MS_TO_KMH = 3.6 + + +class SUMOSimulationWrapper(SUMOUtils): + """ A wrapper for the interaction with the SUMO simulation """ + + def _initialize_simulation(self): + """ Specific simulation initialization. """ + try: + super()._initialize_simulation() + except NotImplementedError: + pass + + def _initialize_metrics(self): + """ Specific metrics initialization """ + try: + super()._initialize_metrics() + except NotImplementedError: + pass + self.veh_subscriptions = dict() + self.collisions = collections.defaultdict(int) + + def _default_step_action(self, agents): + """ Specific code to be executed in every simulation step """ + try: + super()._default_step_action(agents) + except NotImplementedError: + pass + # get collisions + collisions = self.traci_handler.simulation.getCollidingVehiclesIDList() + logger.debug("Collisions: %s", pformat(collisions)) + for veh in collisions: + self.collisions[veh] += 1 + # get subscriptions + self.veh_subscriptions = \ + self.traci_handler.vehicle.getAllSubscriptionResults() + for veh, vals in self.veh_subscriptions.items(): + logger.debug("Subs: %s, %s", pformat(veh), pformat(vals)) + running = set() + for agent in agents: + if agent in self.veh_subscriptions: + running.add(agent) + if len(running) == 0: + logger.info("All the agent left the simulation..") + self.end_simulation() + return True + + +############################################################################### + + +class SUMOAgent: + """ Agent implementation. """ + + def __init__(self, agent, config): + self.agent_id = agent + self.config = config + self.action_to_meaning = dict() + for pos, action in enumerate(config["actions"]): + self.action_to_meaning[pos] = config["actions"][action] + logger.debug("Agent '%s' configuration \n %s", self.agent_id, + pformat(self.config)) + + def step(self, action, sumo_handler): + """ Implements the logic of each specific action passed as input. """ + logger.debug("Agent %s: action %d", self.agent_id, action) + # Subscriptions EXAMPLE: + # {"agent_0": {64: 14.603468282230542, 104: None}, + # "agent_1": {64: 12.922797055918513, + # 104: ("veh.19", 27.239870121802596)}} + logger.debug("Subscriptions: %s", + pformat(sumo_handler.veh_subscriptions[self.agent_id])) + previous_speed = sumo_handler.veh_subscriptions[self.agent_id][ + tc.VAR_SPEED] + new_speed = previous_speed + self.action_to_meaning[action] + logger.debug("Before %.2f", previous_speed) + sumo_handler.traci_handler.vehicle.setSpeed(self.agent_id, new_speed) + logger.debug("After %.2f", new_speed) + return + + def reset(self, sumo_handler): + """ Resets the agent and return the observation. """ + route = "{}_rou".format(self.agent_id) + # https://sumo.dlr.de/pydoc/traci._route.html#RouteDomain-add + sumo_handler.traci_handler.route.add(route, ["road"]) + # insert the agent in the simulation + # traci.vehicle.add(self, vehID, routeID, typeID="DEFAULT_VEHTYPE", + # depart=None, departLane="first", departPos="base", departSpeed="0", + # arrivalLane="current", arrivalPos="max", arrivalSpeed="current", + # fromTaz="", toTaz="", line="", personCapacity=0, personNumber=0) + sumo_handler.traci_handler.vehicle.add( + self.agent_id, route, departLane="best", departSpeed="max") + sumo_handler.traci_handler.vehicle.subscribeLeader(self.agent_id) + sumo_handler.traci_handler.vehicle.subscribe( + self.agent_id, varIDs=[tc.VAR_SPEED]) + logger.info("Agent %s reset done.", self.agent_id) + return self.agent_id, self.config["start"] + + +############################################################################### + +DEFAULT_SCENARIO_CONFING = { + "sumo_config": sumo_default_config(), + "agent_rnd_order": True, + "log_level": "WARN", + "seed": 42, + "misc": { + "max_distance": 5000, # [m] + } +} + +DEFAULT_AGENT_CONFING = { + "origin": "road", + "destination": "road", + "start": 0, + "actions": { # increase/decrease the speed of: + "acc": 1.0, # [m/s] + "none": 0.0, # [m/s] + "dec": -1.0, # [m/s] + }, + "max_speed": 130, # km/h +} + + +class SUMOTestMultiAgentEnv(MultiAgentEnv): + """ + A RLLIB environment for testing MARL environments with SUMO simulations. + """ + + def __init__(self, config): + """ Initialize the environment. """ + super(SUMOTestMultiAgentEnv, self).__init__() + + self._config = config + + # logging + level = logging.getLevelName(config["scenario_config"]["log_level"]) + logger.setLevel(level) + + # SUMO Connector + self.simulation = None + + # Random number generator + self.rndgen = RandomState(config["scenario_config"]["seed"]) + + # Agent initialization + self.agents_init_list = dict() + self.agents = dict() + for agent, agent_config in self._config["agent_init"].items(): + self.agents[agent] = SUMOAgent(agent, agent_config) + + # Environment initialization + self.resetted = True + self.episodes = 0 + self.steps = 0 + + def seed(self, seed): + """ Set the seed of a possible random number generator. """ + self.rndgen = RandomState(seed) + + def get_agents(self): + """ Returns a list of the agents. """ + return self.agents.keys() + + def __del__(self): + logger.info("Environment destruction: SUMOTestMultiAgentEnv") + if self.simulation: + del self.simulation + + ########################################################################### + # OBSERVATIONS + + def get_observation(self, agent): + """ + Returns the observation of a given agent. + See http://sumo.sourceforge.net/pydoc/traci._simulation.html + """ + speed = 0 + distance = self._config["scenario_config"]["misc"]["max_distance"] + if agent in self.simulation.veh_subscriptions: + speed = round( + self.simulation.veh_subscriptions[agent][tc.VAR_SPEED] * + MS_TO_KMH) + leader = self.simulation.veh_subscriptions[agent][tc.VAR_LEADER] + if leader: # compatible with traci + veh, dist = leader + if veh: + # compatible with libsumo + distance = round(dist) + ret = [speed, distance] + logger.debug("Agent %s --> Obs: %s", agent, pformat(ret)) + return ret + + def compute_observations(self, agents): + """ For each agent in the list, return the observation. """ + obs = dict() + for agent in agents: + obs[agent] = self.get_observation(agent) + return obs + + ########################################################################### + # REWARDS + + def get_reward(self, agent): + """ Return the reward for a given agent. """ + speed = self.agents[agent].config[ + "max_speed"] # if the agent is not in the subscriptions + # and this function is called, the agent has + # reached the end of the road + if agent in self.simulation.veh_subscriptions: + speed = round( + self.simulation.veh_subscriptions[agent][tc.VAR_SPEED] * + MS_TO_KMH) + logger.debug("Agent %s --> Reward %d", agent, speed) + return speed + + def compute_rewards(self, agents): + """ For each agent in the list, return the rewards. """ + rew = dict() + for agent in agents: + rew[agent] = self.get_reward(agent) + return rew + + ########################################################################### + # REST & LEARNING STEP + + def reset(self): + """ Resets the env and returns observations from ready agents. """ + self.resetted = True + self.episodes += 1 + self.steps = 0 + + # Reset the SUMO simulation + if self.simulation: + del self.simulation + + self.simulation = SUMOSimulationWrapper( + self._config["scenario_config"]["sumo_config"]) + + # Reset the agents + waiting_agents = list() + for agent in self.agents.values(): + agent_id, start = agent.reset(self.simulation) + waiting_agents.append((start, agent_id)) + waiting_agents.sort() + + # Move the simulation forward + starting_time = waiting_agents[0][0] + self.simulation.fast_forward(starting_time) + self.simulation._default_step_action( + self.agents.keys()) # hack to retrieve the subs + + # Observations + initial_obs = self.compute_observations(self.agents.keys()) + + return initial_obs + + def step(self, action_dict): + """ + Returns observations from ready agents. + + The returns are dicts mapping from agent_id strings to values. The + number of agents in the env can vary over time. + + Returns + ------- + obs (dict): New observations for each ready agent. + rewards (dict): Reward values for each ready agent. If the + episode is just started, the value will be None. + dones (dict): Done values for each ready agent. The special key + "__all__" (required) is used to indicate env termination. + infos (dict): Optional info values for each agent id. + """ + self.resetted = False + self.steps += 1 + logger.debug( + "====> [SUMOTestMultiAgentEnv:step] Episode: %d - Step: %d <====", + self.episodes, self.steps) + dones = {} + dones["__all__"] = False + + shuffled_agents = sorted( + action_dict.keys()) # it may seem not smar to sort something that + # may need to be shuffled afterwards, but it + # is a matter of consistency instead of using + # whatever insertion order was used in the dict + if self._config["scenario_config"]["agent_rnd_order"]: + # randomize the agent order to minimize SUMO's + # insertion queues impact + logger.debug("Shuffling the order of the agents.") + self.rndgen.shuffle(shuffled_agents) # in-place shuffle + + # Take action + for agent in shuffled_agents: + self.agents[agent].step(action_dict[agent], self.simulation) + + logger.debug("Before SUMO") + ongoing_simulation = self.simulation.step( + until_end=False, agents=set(action_dict.keys())) + logger.debug("After SUMO") + + # end of the episode + if not ongoing_simulation: + logger.info("Reached the end of the SUMO simulation.") + dones["__all__"] = True + + obs, rewards, infos = {}, {}, {} + + for agent in action_dict: + # check for collisions + if self.simulation.collisions[agent] > 0: + # punish the agent and remove it from the simulation + dones[agent] = True + obs[agent] = [0, 0] + rewards[agent] = -self.agents[agent].config["max_speed"] + # infos[agent] = "Collision" + self.simulation.traci_handler.remove( + agent, reason=tc.REMOVE_VAPORIZED) + else: + dones[agent] = agent not in self.simulation.veh_subscriptions + obs[agent] = self.get_observation(agent) + rewards[agent] = self.get_reward(agent) + # infos[agent] = "" + + logger.debug("Observations: %s", pformat(obs)) + logger.debug("Rewards: %s", pformat(rewards)) + logger.debug("Dones: %s", pformat(dones)) + logger.debug("Info: %s", pformat(infos)) + logger.debug( + "========================================================") + return obs, rewards, dones, infos + + ########################################################################### + # ACTIONS & OBSERATIONS SPACE + + def get_action_space_size(self, agent): + """ Returns the size of the action space. """ + return len(self.agents[agent].config["actions"]) + + def get_action_space(self, agent): + """ Returns the action space. """ + return gym.spaces.Discrete(self.get_action_space_size(agent)) + + def get_set_of_actions(self, agent): + """ Returns the set of possible actions for an agent. """ + return set(range(self.get_action_space_size(agent))) + + def get_obs_space_size(self, agent): + """ Returns the size of the observation space. """ + return ((self.agents[agent].config["max_speed"] + 1) * + (self._config["scenario_config"]["misc"]["max_distance"] + 1)) + + def get_obs_space(self, agent): + """ Returns the observation space. """ + return gym.spaces.MultiDiscrete([ + self.agents[agent].config["max_speed"] + 1, + self._config["scenario_config"]["misc"]["max_distance"] + 1 + ]) diff --git a/rllib/examples/simulators/sumo/scenario/edges.xml b/rllib/examples/simulators/sumo/scenario/edges.xml new file mode 100644 index 000000000..42e3ac370 --- /dev/null +++ b/rllib/examples/simulators/sumo/scenario/edges.xml @@ -0,0 +1,4 @@ + + + + \ No newline at end of file diff --git a/rllib/examples/simulators/sumo/scenario/flows.xml b/rllib/examples/simulators/sumo/scenario/flows.xml new file mode 100644 index 000000000..38aca3da5 --- /dev/null +++ b/rllib/examples/simulators/sumo/scenario/flows.xml @@ -0,0 +1,4 @@ + + + + diff --git a/rllib/examples/simulators/sumo/scenario/nodes.xml b/rllib/examples/simulators/sumo/scenario/nodes.xml new file mode 100644 index 000000000..c62de5813 --- /dev/null +++ b/rllib/examples/simulators/sumo/scenario/nodes.xml @@ -0,0 +1,5 @@ + + + + + \ No newline at end of file diff --git a/rllib/examples/simulators/sumo/scenario/road.net.xml b/rllib/examples/simulators/sumo/scenario/road.net.xml new file mode 100644 index 000000000..2b00d43c4 --- /dev/null +++ b/rllib/examples/simulators/sumo/scenario/road.net.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + + + + + diff --git a/rllib/examples/simulators/sumo/scenario/sumo.cfg.xml b/rllib/examples/simulators/sumo/scenario/sumo.cfg.xml new file mode 100644 index 000000000..b60dd70dd --- /dev/null +++ b/rllib/examples/simulators/sumo/scenario/sumo.cfg.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + + + + + + + + diff --git a/rllib/examples/sumo_env_local.py b/rllib/examples/sumo_env_local.py new file mode 100644 index 000000000..baa96e452 --- /dev/null +++ b/rllib/examples/sumo_env_local.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +""" Example Trainer for RLLIB + SUMO Utlis + + Author: Lara CODECA lara.codeca@gmail.com + + See: + https://github.com/lcodeca/rllibsumoutils + https://github.com/lcodeca/rllibsumodocker + for further details. +""" + +import argparse +from copy import deepcopy +import logging +import os +import pathlib +from pprint import pformat + +import ray +from ray import tune + +from ray.rllib.agents.ppo import ppo +from ray.rllib.examples.simulators.sumo import marlenvironment +from ray.rllib.utils.test_utils import check_learning_achieved + +logging.basicConfig(level=logging.WARN) +logger = logging.getLogger("ppotrain") + +parser = argparse.ArgumentParser() +parser.add_argument( + "--sumo-connect-lib", + type=str, + default="libsumo", + choices=["libsumo", "traci"], + help="The SUMO connector to import. " + "Requires the env variable SUMO_HOME set.") +parser.add_argument( + "--sumo-gui", + action="store_true", + help="Enables the SUMO GUI. Possible only with TraCI connector.") +parser.add_argument( + "--sumo-config-file", + type=str, + default=None, + help="The SUMO configuration file for the scenario.") +parser.add_argument( + "--from-checkpoint", + type=str, + default=None, + help="Full path to a checkpoint file for restoring a previously saved " + "Trainer state.") +parser.add_argument("--num-workers", type=int, default=0) +parser.add_argument("--as-test", action="store_true") +parser.add_argument("--stop-iters", type=int, default=10) +parser.add_argument("--stop-reward", type=float, default=30000.0) +parser.add_argument("--stop-timesteps", type=int, default=10000000) + +if __name__ == "__main__": + args = parser.parse_args() + ray.init() + tune.register_env("sumo_test_env", marlenvironment.env_creator) + + # Algorithm. + policy_class = ppo.PPOTFPolicy + config = ppo.DEFAULT_CONFIG + config["framework"] = "tf" + config["gamma"] = 0.99 + config["lambda"] = 0.95 + config["log_level"] = "WARN" + config["lr"] = 0.001 + config["min_iter_time_s"] = 5 + config["num_gpus"] = int(os.environ.get("RLLIB_NUM_GPUS", "0")) + config["num_workers"] = args.num_workers + config["rollout_fragment_length"] = 200 + config["sgd_minibatch_size"] = 256 + config["simple_optimizer"] = True + config["train_batch_size"] = 4000 + + config["batch_mode"] = "complete_episodes" + config["no_done_at_end"] = True + + # Load default Scenario configuration for the LEARNING ENVIRONMENT + scenario_config = deepcopy(marlenvironment.DEFAULT_SCENARIO_CONFING) + scenario_config["seed"] = 42 + scenario_config["log_level"] = "INFO" + scenario_config["sumo_config"]["sumo_connector"] = args.sumo_connect_lib + scenario_config["sumo_config"]["sumo_gui"] = args.sumo_gui + if args.sumo_config_file is not None: + scenario_config["sumo_config"]["sumo_cfg"] = args.sumo_config_file + else: + filename = "{}/simulators/sumo/scenario/sumo.cfg.xml".format( + pathlib.Path(__file__).parent.absolute()) + scenario_config["sumo_config"]["sumo_cfg"] = filename + + scenario_config["sumo_config"]["sumo_params"] = [ + "--collision.action", "warn" + ] + scenario_config["sumo_config"]["trace_file"] = True + scenario_config["sumo_config"]["end_of_sim"] = 3600 # [s] + scenario_config["sumo_config"][ + "update_freq"] = 10 # number of traci.simulationStep() + # for each learning step. + scenario_config["sumo_config"]["log_level"] = "INFO" + logger.info("Scenario Configuration: \n %s", pformat(scenario_config)) + + # Associate the agents with their configuration. + agent_init = { + "agent_0": deepcopy(marlenvironment.DEFAULT_AGENT_CONFING), + "agent_1": deepcopy(marlenvironment.DEFAULT_AGENT_CONFING), + } + logger.info("Agents Configuration: \n %s", pformat(agent_init)) + + # MARL Environment Init + env_config = { + "agent_init": agent_init, + "scenario_config": scenario_config, + } + marl_env = marlenvironment.SUMOTestMultiAgentEnv(env_config) + + # Config for the PPO trainer from the MARLEnv + policies = {} + for agent in marl_env.get_agents(): + agent_policy_params = {} + policies[agent] = (policy_class, marl_env.get_obs_space(agent), + marl_env.get_action_space(agent), + agent_policy_params) + config["multiagent"]["policies"] = policies + config["multiagent"]["policy_mapping_fn"] = lambda agent_id: agent_id + config["multiagent"]["policies_to_train"] = ["ppo_policy"] + + config["env"] = "sumo_test_env" + config["env_config"] = env_config + + logger.info("PPO Configuration: \n %s", pformat(config)) + + stop = { + "training_iteration": args.stop_iters, + "timesteps_total": args.stop_timesteps, + "episode_reward_mean": args.stop_reward, + } + + # Run the experiment. + results = tune.run( + "PPO", + config=config, + stop=stop, + verbose=1, + checkpoint_freq=10, + restore=args.from_checkpoint) + + # And check the results. + if args.as_test: + check_learning_achieved(results, args.stop_reward) + + ray.shutdown()