sumolib.vehicletype

  1# Eclipse SUMO, Simulation of Urban MObility; see https://eclipse.dev/sumo
  2# Copyright (C) 2008-2026 German Aerospace Center (DLR) and others.
  3# This program and the accompanying materials are made available under the
  4# terms of the Eclipse Public License 2.0 which is available at
  5# https://www.eclipse.org/legal/epl-2.0/
  6# This Source Code may also be made available under the following Secondary
  7# Licenses when the conditions for such availability set forth in the Eclipse
  8# Public License 2.0 are satisfied: GNU General Public License, version 2
  9# or later which is available at
 10# https://www.gnu.org/licenses/old-licenses/gpl-2.0-standalone.html
 11# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
 12
 13# @file    vehicletype.py
 14# @author  Mirko Barthauer (Technische Universitaet Braunschweig, Institut fuer Verkehr und Stadtbauwesen)
 15# @author  Jakob Erdmann
 16# @author  Michael Behrisch
 17# @author  Maxwell Schrader
 18# @date    2021-07-19
 19
 20# This file is a port of createVehTypeDistribution.py to sumolib
 21
 22from __future__ import absolute_import
 23from __future__ import print_function
 24import os
 25import json
 26import re
 27import random
 28from typing import Union, Any  # noqa
 29
 30import sumolib
 31
 32
 33class _FixDistribution(object):
 34
 35    def __init__(self, params, isNumeric=True):
 36        if isNumeric:
 37            self._params = tuple([float(p) for p in params])
 38        else:
 39            self._params = params
 40        self._limits = (0, None)
 41        self._isNumeric = isNumeric
 42        self._maxSampleAttempts = 10
 43
 44    def setMaxSamplingAttempts(self, n):
 45        if n is not None:
 46            self._maxSampleAttempts = n
 47
 48    def setLimits(self, limits):
 49        self._limits = limits
 50
 51    def sampleValue(self):
 52        if self._isNumeric:
 53            value = None
 54            nrSampleAttempts = 0
 55            # Sample until value falls into limits
 56            while nrSampleAttempts < self._maxSampleAttempts \
 57                    and (value is None or (self._limits[1] is not None and value > self._limits[1]) or
 58                         (self._limits[0] is not None and value < self._limits[0])):
 59                value = self._sampleValue()
 60                nrSampleAttempts += 1
 61            # Eventually apply fallback cutting value to limits
 62            if self._limits[0] is not None and value < self._limits[0]:
 63                value = self._limits[0]
 64            elif self._limits[1] is not None and value > self._limits[1]:
 65                value = self._limits[1]
 66        else:
 67            value = self._sampleValue()
 68        return value
 69
 70    def sampleValueString(self, decimalPlaces):
 71        if self._isNumeric:
 72            decimalPattern = "%." + str(decimalPlaces) + "f"
 73            return decimalPattern % self.sampleValue()
 74        return self.sampleValue()
 75
 76    def _sampleValue(self):
 77        return self._params[0]
 78
 79
 80class _NormalDistribution(_FixDistribution):
 81
 82    def __init__(self, mu, sd):
 83        _FixDistribution.__init__(self, (mu, sd))
 84
 85    def _sampleValue(self):
 86        return random.normalvariate(self._params[0], self._params[1])
 87
 88
 89class _LogNormalDistribution(_FixDistribution):
 90
 91    def __init__(self, mu, sd):
 92        _FixDistribution.__init__(self, (mu, sd))
 93
 94    def _sampleValue(self):
 95        return random.lognormvariate(self._params[0], self._params[1])
 96
 97
 98class _NormalCappedDistribution(_FixDistribution):
 99
100    def __init__(self, mu, sd, min, max):
101        _FixDistribution.__init__(self, (mu, sd, min, max))
102        if mu < min or mu > max:
103            raise ValueError("mean %s is outside cutoff bounds [%s, %s]" % (mu, min, max))
104
105    def _sampleValue(self):
106        while True:
107            cand = random.normalvariate(self._params[0], self._params[1])
108            if cand >= self._params[2] and cand <= self._params[3]:
109                return cand
110
111
112class _UniformDistribution(_FixDistribution):
113
114    def __init__(self, a, b):
115        _FixDistribution.__init__(self, (a, b))
116
117    def _sampleValue(self):
118        return random.uniform(self._params[0], self._params[1])
119
120
121class _GammaDistribution(_FixDistribution):
122
123    def __init__(self, alpha, beta):
124        _FixDistribution.__init__(self, (alpha, 1.0 / beta))
125
126    def _sampleValue(self):
127        return random.gammavariate(self._params[0], self._params[1])
128
129
130_DIST_DICT = {
131    'normal': _NormalDistribution,
132    'lognormal': _LogNormalDistribution,
133    'normalCapped': _NormalCappedDistribution,
134    'uniform': _UniformDistribution,
135    'gamma': _GammaDistribution
136}
137
138
139class VehAttribute:
140
141    def __init__(self, name, is_param=False, distribution=None, distribution_params=None,
142                 bounds=None, attribute_value=None):
143        # type: (str, bool, str, Union[dict, Any], tuple, str) -> None
144        """
145        This emmulates one line of example config.txt in
146            https://sumo.dlr.de/docs/Tools/Misc.html#createvehtypedistributionpy
147        Either distribution or attribute_value should be populated
148        Args:
149            name (str): the name of the attribute. Examples: "tau", "sigma", "length"
150            is_param (bool, optional): is the attribute a parameter that should be added as a child element.
151            distribution (str, optional): the name of the distribution to use ()
152            distribution_params (Union[dict, Any], optional): the parameters corresponding to the distribution
153            bounds (tuple, optional): the bounds of the distribution.
154            attribute_value (str, optional): if no distribution is given, the fixed value for the attribute
155        """
156        self.is_param = is_param
157        self.name = name
158        self.distribution = distribution
159        self.distribution_params = distribution_params
160        self.bounds = bounds
161        self.attribute_value = attribute_value
162        if self.attribute_value and self.distribution:
163            raise ValueError("Only one of distribution or attribute value should be defined, not both")
164        self.d_obj = self._dist_helper(distribution, distribution_params, bounds)
165
166    def _dist_helper(self, distribution, dist_params, dist_bounds):
167        # type: (...) -> Union[None, _FixDistribution]
168        if distribution:
169            try:
170                d = _DIST_DICT[distribution](**dist_params)
171                d.setLimits(dist_bounds) if dist_bounds else d.setLimits((0, None))
172            except KeyError:
173                raise ValueError("The distribution %s is not known. Please select one of: \n%s " %
174                                 (distribution, "\n".join(_DIST_DICT.keys())))
175        else:
176            isNumeric = False if self.name == "emissionClass" else len(
177                re.findall(r'^(-?[0-9]+(\.[0-9]+)?)$', self.attribute_value)) > 0
178            d = _FixDistribution((self.attribute_value, ), isNumeric)
179        return d
180
181    def add_sampling_attempts(self, attempts):
182        if self.d_obj:
183            self.d_obj.setMaxSamplingAttempts(attempts)
184
185
186class CreateVehTypeDistribution:
187
188    def __init__(self, seed=None, size=100, name='vehDist', resampling=100, decimal_places=3):
189        # type: (int, int, str, int, int) -> None
190        """
191        Creates a VehicleType Distribution.
192        See https://sumo.dlr.de/docs/Definition_of_Vehicles,_Vehicle_Types,_and_Routes.html#vehicle_type_distributions
193
194        Args:
195            seed (int, optional): random seed.
196            size (int, optional): number of vTypes in the distribution.
197            name (str, optional): alphanumerical ID used for the created vehicle type distribution.
198            resampling (int, optional): number of attempts to resample a value until it lies in the specified bounds.
199            decimal_places (int, optional): number of decimal places.
200        """
201        if seed:
202            random.seed(seed)
203
204        self.size = size
205        self.name = name
206        self.resampling = resampling
207        self.decimal_places = decimal_places
208        self.attributes = []  # type: list[VehAttribute]
209
210    def add_attribute(self, attribute):
211        # type: (Union[VehAttribute, dict]) -> None
212        """
213        Add an instance of the attribute class to the Parameters. Pass the sampling attempts "global" parameter
214        Args:
215            attribute (VehAttribute or dict): An instance of VehAttribute or
216             a dictionary of parameters to be passed to the VehAttribute constructor
217        """
218        attribute = attribute if isinstance(attribute, VehAttribute) else VehAttribute(**attribute)
219        attribute.add_sampling_attempts(self.resampling)
220        self.attributes.append(attribute)
221
222    def create_veh_dist(self, xml_dom):
223        # type: (sumolib.xml.CompoundObject) -> sumolib.xml.CompoundObject
224        # create the vehicleDist node
225        vtype_dist_node = xml_dom.addChild("vTypeDistribution", {"id": self.name})
226
227        # create the vehicle types
228        for i in range(self.size):
229            veh_type_node = vtype_dist_node.addChild("vType", {"id": self.name + str(i)})
230            for attr in self.attributes:
231                if attr.is_param:
232                    veh_type_node.addChild("param", {"key": attr.name,
233                                                     "value": attr.d_obj.sampleValueString(self.decimal_places)})
234                else:
235                    veh_type_node.setAttribute(attr.name, attr.d_obj.sampleValueString(self.decimal_places))
236
237        return vtype_dist_node
238
239    def to_xml(self, file_path):
240        # type: (str) -> None
241        xml_dom, existing_file = self._check_existing(file_path)
242        if existing_file:
243            self._handle_existing(xml_dom)
244        self.create_veh_dist(xml_dom)
245        with sumolib.openz(file_path, 'w') as f:
246            sumolib.xml.writeHeader(f)
247            f.write(xml_dom.toXML())
248        print("Output written to %s" % file_path)
249
250    def _handle_existing(self, xmlTree):
251        # type: (sumolib.xml.CompoundObject, sumolib.xml.CompoundObject) -> None
252        if xmlTree.vTypeDistribution:
253            for existingDistNode in xmlTree.vTypeDistribution:
254                if existingDistNode.hasAttribute("id") and existingDistNode.id == self.name:
255                    xmlTree.removeChild(existingDistNode)
256                    break
257
258    @staticmethod
259    def _check_existing(file_path):
260        # type: (str) -> tuple[sumolib.xml.CompoundObject, bool]
261        if os.path.exists(file_path):
262            try:
263                xmlTree = list(sumolib.xml.parse(file_path, outputLevel=0))[0]
264                return xmlTree, True
265            except Exception as e:
266                raise ValueError("Cannot parse existing %s. Error: %s" % (file_path, e))
267        else:
268            return sumolib.xml.create_document("additional"), False
269
270    def save_myself(self, file_path):
271        # type: (str) -> None
272        """
273        This function saves the class to a json format. Used for logging simulation inputs
274
275        Args:
276            file_path (str): path to save json to
277        """
278        with sumolib.openz(file_path, "w") as f:
279            f.write(
280                json.dumps(
281                    self,
282                    default=lambda o: {
283                        key: param for key, param in o.__dict__.items() if '_' not in key[0]},
284                    sort_keys=True,
285                    indent=4,
286                )
287            )
288
289
290class CreateMultiVehTypeDistributions(CreateVehTypeDistribution):
291
292    def __init__(self):
293        # type: () -> None
294        self.distributions = []  # type: list[CreateVehTypeDistribution]
295
296    def register_veh_type_distribution(self, veh_type_dist, veh_attributes):
297        # type: (Union[dict, CreateVehTypeDistribution], list[Union[dict, VehAttribute]]) -> None
298        veh_type_dist = veh_type_dist if isinstance(
299            veh_type_dist, CreateVehTypeDistribution) else CreateVehTypeDistribution(**veh_type_dist)
300
301        for attr in veh_attributes:
302            veh_type_dist.add_attribute(attr if isinstance(attr, VehAttribute) else VehAttribute(**attr))
303
304        self.distributions.append(veh_type_dist)
305
306    def write_xml(self, file_path):
307        # type: (str) -> None
308        """
309        This function will overwrite existing files
310
311        Args:
312            file_path (str): Path to the file to write to
313        """
314        xml_dom, _ = self._check_existing(file_path)
315        with sumolib.openz(file_path, 'w') as f:
316            sumolib.xml.writeHeader(f, root="additional")
317            for dist in self.distributions:
318                f.write(dist.create_veh_dist(xml_dom=xml_dom).toXML())
319            print("</additional>", file=f)