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