[RLlib] Integration with SUMO Simulator (#11710)

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 <output-prefix value=".."/>.
+    # 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
+    # <tripinfo-output value=".."/>.
+    # 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)
+
+    ###########################################################################

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

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
+        ])

rllib/examples/simulators/sumo/scenario/edges.xml

@@ -0,0 +1,4 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<edges>
+    <edge id="road" from="origin" to="destination" numLanes="3" speed="13.89"/>
+</edges>

rllib/examples/simulators/sumo/scenario/flows.xml

@@ -0,0 +1,4 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<routes>
+    <flow id="veh" from="road" to="road" period="1" begin="0.0" end="3600.0"/>
+</routes>

rllib/examples/simulators/sumo/scenario/nodes.xml

@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<nodes>
+    <node id="origin"       x="0.0"     y="0.0"     type="priority"/>
+    <node id="destination"  x="5000.0"  y="0.0"     type="priority"/>
+</nodes>

rllib/examples/simulators/sumo/scenario/road.net.xml

@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<net version="1.6" junctionCornerDetail="5" limitTurnSpeed="5.50" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://sumo.dlr.de/xsd/net_file.xsd">
+
+    <location netOffset="0.00,0.00" convBoundary="0.00,0.00,5000.00,0.00" origBoundary="0.00,0.00,5000.00,0.00" projParameter="!"/>
+
+    <edge id="road" from="origin" to="destination" priority="-1">
+        <lane id="road_0" index="0" speed="13.89" length="5000.00" shape="0.00,-8.00 5000.00,-8.00"/>
+        <lane id="road_1" index="1" speed="13.89" length="5000.00" shape="0.00,-4.80 5000.00,-4.80"/>
+        <lane id="road_2" index="2" speed="13.89" length="5000.00" shape="0.00,-1.60 5000.00,-1.60"/>
+    </edge>
+
+    <junction id="destination" type="dead_end" x="5000.00" y="0.00" incLanes="road_0 road_1 road_2" intLanes="" shape="5000.00,-9.60 5000.00,0.00"/>
+    <junction id="origin" type="dead_end" x="0.00" y="0.00" incLanes="" intLanes="" shape="0.00,0.00 0.00,-9.60"/>
+
+</net>

rllib/examples/simulators/sumo/scenario/sumo.cfg.xml

@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<configuration xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://sumo.dlr.de/xsd/sumoConfiguration.xsd">
+
+    <input>
+        <net-file value="road.net.xml"/>
+        <route-files value="flows.xml"/>
+    </input>
+
+    <time>
+        <begin value="0.0"/>
+        <step-length value="0.1"/>
+    </time>
+
+    <report>
+        <!-- Silent -->
+        <verbose value="false"/>
+        <no-step-log value="true"/>
+        <duration-log.statistics value="false"/>
+        <duration-log.disable value="true"/>
+        <no-warnings value="true"/>
+    </report>
+
+</configuration>

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()