sumolib.net

   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    __init__.py
  14# @author  Daniel Krajzewicz
  15# @author  Laura Bieker
  16# @author  Karol Stosiek
  17# @author  Michael Behrisch
  18# @author  Jakob Erdmann
  19# @author  Robert Hilbrich
  20# @author  Mirko Barthauer
  21# @date    2008-03-27
  22
  23"""
  24This file contains a content handler for parsing sumo network xml files.
  25It uses other classes from this module to represent the road network.
  26"""
  27
  28from __future__ import print_function
  29from __future__ import absolute_import
  30import sys
  31import math
  32import heapq
  33import gzip
  34import warnings
  35import io
  36from xml.sax import handler, parse
  37from copy import copy
  38from collections import defaultdict
  39from itertools import chain
  40try:
  41    from functools import lru_cache
  42    HAVE_LRU_CACHE = True
  43except ImportError:
  44    HAVE_LRU_CACHE = False
  45
  46try:
  47    import lxml.etree
  48    import pathlib
  49    HAVE_LXML = True
  50except ImportError:
  51    HAVE_LXML = False
  52
  53try:
  54    import pyproj  # noqa
  55    HAVE_PYPROJ = True
  56except ImportError:
  57    HAVE_PYPROJ = False
  58
  59import sumolib
  60from . import lane, edge, netshiftadaptor, node, connection, roundabout  # noqa
  61from .connection import Connection
  62from sumolib.miscutils import intIfPossible, openz, PRACTICAL_INFINITY
  63
  64
  65class TLS:
  66
  67    """Traffic Light Signal for a sumo network"""
  68
  69    def __init__(self, id):
  70        self._id = id
  71        self._connections = []
  72        self._maxConnectionNo = -1
  73        self._programs = {}
  74
  75    def addConnection(self, inLane, outLane, linkNo):
  76        self._connections.append([inLane, outLane, linkNo])
  77        if linkNo > self._maxConnectionNo:
  78            self._maxConnectionNo = linkNo
  79
  80    def getConnections(self):
  81        return self._connections
  82
  83    def getID(self):
  84        return self._id
  85
  86    def getLinks(self):
  87        links = {}
  88        for the_connection in self._connections:
  89            if the_connection[2] not in links:
  90                links[the_connection[2]] = []
  91            links[the_connection[2]].append(the_connection)
  92        return links
  93
  94    def getEdges(self):
  95        edges = set()
  96        for c in self._connections:
  97            edges.add(c[0].getEdge())
  98        return edges
  99
 100    def addProgram(self, program):
 101        self._programs[program._id] = program
 102
 103    def removePrograms(self):
 104        self._programs.clear()
 105
 106    def toXML(self):
 107        ret = ""
 108        for p in self._programs:
 109            ret = ret + self._programs[p].toXML(self._id)
 110        return ret
 111
 112    def getPrograms(self):
 113        return self._programs
 114
 115
 116class Phase:
 117
 118    def __init__(self, duration, state, minDur=None, maxDur=None, next=tuple(), name="", earlyTarget=""):
 119        """
 120        Constructs a traffic light phase
 121        duration (float): the duration of the phase in seconds
 122        state (string): the state codes for each controlled link
 123        minDur (float): the minimum duration (ignored by static tls)
 124        maxDur (float): the maximum duration (ignored by static tls)
 125        next (intList): possible succesor phase (optional)
 126        name (string): the name of the phase
 127        earlyTarget (string): early switching to phase with the given index(es)
 128        """
 129        self.duration = duration
 130        self.state = state
 131        # minimum and maximum duration (only for actuated tls)
 132        self.minDur = minDur if minDur is not None else duration
 133        self.maxDur = maxDur if maxDur is not None else duration
 134        self.next = next
 135        self.name = name
 136        self.earlyTarget = earlyTarget
 137
 138    def __repr__(self):
 139        name = (", name='%s'" % self.name) if self.name else ""
 140        next = (", next='%s'" % str(self.next)) if self.next else ""
 141        earlyTarget = (", earlyTarget='%s'" % self.earlyTarget) if self.earlyTarget else ""
 142        return ("Phase(duration=%s, state='%s', minDur=%s, maxDur=%s%s%s%s)"
 143                % (self.duration, self.state, self.minDur, self.maxDur, name, next, earlyTarget))
 144
 145
 146class TLSProgram:
 147
 148    def __init__(self, id, offset, type):
 149        self._id = id
 150        self._type = type
 151        self._offset = offset
 152        self._phases = []
 153        self._params = {}
 154        self._conditions = {}
 155
 156    def addPhase(self, state, duration, minDur=-1, maxDur=-1, next=None, name="", earlyTarget=""):
 157        self._phases.append(Phase(duration, state, minDur, maxDur, next, name, earlyTarget))
 158
 159    def addCondition(self, id, value):
 160        self._conditions[id] = value
 161
 162    def toXML(self, tlsID):
 163        ret = '  <tlLogic id="%s" type="%s" programID="%s" offset="%s">\n' % (
 164            tlsID, self._type, self._id, self._offset)
 165        for p in self._phases:
 166            minDur = '' if p.minDur < 0 else ' minDur="%s"' % p.minDur
 167            maxDur = '' if p.maxDur < 0 else ' maxDur="%s"' % p.maxDur
 168            name = '' if p.name == '' else ' name="%s"' % p.name
 169            next = '' if len(p.next) == 0 else ' next="%s"' % ' '.join(map(str, p.next))
 170            earlyTarget = '' if p.earlyTarget == '' else ' earlyTarget="%s"' % p.earlyTarget
 171            ret += '    <phase duration="%s" state="%s"%s%s%s%s%s/>\n' % (
 172                p.duration, p.state, minDur, maxDur, name, next, earlyTarget)
 173        for k, v in self._params.items():
 174            ret += '    <param key="%s" value="%s"/>\n' % (k, v)
 175        for i, v in self._conditions.items():
 176            ret += '    <condition id="%s" value="%s"/>\n' % (i, v)
 177        ret += '  </tlLogic>\n'
 178        return ret
 179
 180    def getPhases(self):
 181        return self._phases
 182
 183    def getType(self):
 184        return self._type
 185
 186    def setParam(self, key, value):
 187        self._params[key] = value
 188
 189    def getParam(self, key, default=None):
 190        return self._params.get(key, default)
 191
 192    def getParams(self):
 193        return self._params
 194
 195    def getStages(self):
 196        stages = dict()
 197        for idx, phase in enumerate(self.getPhases()):
 198            if phase not in stages.values():
 199                if 'G' in phase.state and 'y' not in phase.state and phase.name:
 200                    stages[idx] = phase
 201        return stages
 202
 203    def getOffset(self):
 204        return self._offset
 205
 206
 207class EdgeType:
 208    def __init__(self, id, allow, disallow):
 209        self.id = id
 210        self.allow = allow
 211        self.disallow = disallow
 212
 213
 214class Net:
 215
 216    """The whole sumo network."""
 217
 218    def __init__(self):
 219        self._location = {}
 220        self._id2node = {}
 221        self._id2edge = {}
 222        self._crossings_and_walkingAreas = set()
 223        self._macroConnectors = set()
 224        self._id2tls = {}
 225        self._nodes = []
 226        self._edges = []
 227        self._tlss = []
 228        self._ranges = [[sys.float_info.max, -sys.float_info.max], [sys.float_info.max, -sys.float_info.max]]
 229        self._roundabouts = []
 230        self._rtreeEdges = {True: None, False: None}  # different rTrees depending on includeJunctions
 231        self._rtreeLanes = {True: None, False: None}  # different rTrees depending on includeJunctions
 232        self._allLanes = []
 233        self._origIdx = None
 234        self._proj = None
 235        self.hasInternal = False
 236        self.hasWalkingArea = False
 237        # store dijsktra heap for reuse if the same origin is used repeatedly
 238        self._shortestPathCache = None
 239        self._version = None
 240        self._edgeTypes = defaultdict(lambda: EdgeType("DEFAULT_EDGETYPE", "", ""))
 241        self._routingCache = None
 242        self._edgeSuccessorCache = None
 243        self._edgeCostCache = None
 244
 245    def initRoutingCache(self, maxsize=1000):
 246        self._routingCache = lru_cache(maxsize=maxsize)(lambda fromEdge, fastest, vClass, ignoreDirection, reversalPenalty, preferences: {})  # noqa
 247
 248    def getVersion(self):
 249        return self._version
 250
 251    def getEdgeType(self, typeID):
 252        return self._edgeTypes[typeID]
 253
 254    def setLocation(self, netOffset, convBoundary, origBoundary, projParameter):
 255        self._location["netOffset"] = netOffset
 256        self._location["convBoundary"] = convBoundary
 257        self._location["origBoundary"] = origBoundary
 258        self._location["projParameter"] = projParameter
 259
 260    def loadSelection(self, selectionFile):
 261        self.resetSelection()
 262        with io.open(selectionFile, "r", encoding="utf-8") as f:
 263            for line in f:
 264                line = line.strip()
 265                if line.startswith("edge:"):
 266                    edgeID = line[5:]
 267                    if edgeID in self._id2edge:
 268                        self.getEdge(edgeID).select()
 269                elif line.startswith("junction:"):
 270                    nodeID = line[9:]
 271                    if nodeID in self._id2node:
 272                        self.getNode(nodeID).select()
 273
 274    def resetSelection(self):
 275        for n in self._nodes:
 276            n.select(False)
 277        for e in self._edges:
 278            e.select(False)
 279
 280    def addNode(self, id, type=None, coord=None, incLanes=None, intLanes=None):
 281        if id is None:
 282            return None
 283        if id not in self._id2node:
 284            n = node.Node(id, type, coord, incLanes, intLanes)
 285            self._nodes.append(n)
 286            self._id2node[id] = n
 287        self.setAdditionalNodeInfo(
 288            self._id2node[id], type, coord, incLanes, intLanes)
 289        return self._id2node[id]
 290
 291    def setAdditionalNodeInfo(self, node, type, coord, incLanes, intLanes=None):
 292        if coord is not None and node._coord is None:
 293            node._coord = coord
 294            self._ranges[0][0] = min(self._ranges[0][0], coord[0])
 295            self._ranges[0][1] = max(self._ranges[0][1], coord[0])
 296            self._ranges[1][0] = min(self._ranges[1][0], coord[1])
 297            self._ranges[1][1] = max(self._ranges[1][1], coord[1])
 298        if incLanes is not None and node._incLanes is None:
 299            node._incLanes = incLanes
 300        if intLanes is not None and node._intLanes is None:
 301            node._intLanes = intLanes
 302        if type is not None and node._type is None:
 303            node._type = type
 304
 305    def addEdge(self, id, fromID, toID, prio, function, name, edgeType='', routingType=''):
 306        if id not in self._id2edge:
 307            fromN = self.addNode(fromID)
 308            toN = self.addNode(toID)
 309            e = edge.Edge(id, fromN, toN, prio, function, name, edgeType, routingType)
 310            self._edges.append(e)
 311            self._id2edge[id] = e
 312            if function:
 313                self.hasInternal = True
 314                if function == "walkingarea":
 315                    self.hasWalkingArea = True
 316        return self._id2edge[id]
 317
 318    def addLane(self, edge, speed, length, width, allow=None, disallow=None, acceleration=False):
 319        return lane.Lane(edge, speed, length, width, allow, disallow, acceleration)
 320
 321    def addRoundabout(self, nodes, edges=None):
 322        r = roundabout.Roundabout(nodes, edges)
 323        self._roundabouts.append(r)
 324        return r
 325
 326    def addConnection(self, fromEdge, toEdge, fromlane, tolane, direction,
 327                      tls, tllink, tllink2, allow, disallow, state, viaLaneID=None):
 328        conn = connection.Connection(
 329            fromEdge, toEdge, fromlane, tolane, direction,
 330            tls, tllink, tllink2, allow, disallow, state, viaLaneID)
 331        fromEdge.addOutgoing(conn)
 332        fromlane.addOutgoing(conn)
 333        toEdge._addIncoming(conn)
 334        if viaLaneID:
 335            try:
 336                # internal lanes are only available when building with option withInternal=True
 337                viaLane = self.getLane(viaLaneID)
 338                viaEdge = viaLane.getEdge()
 339                viaEdge._addIncoming(connection.Connection(
 340                    fromEdge, viaEdge, fromlane, viaLane, direction, tls,
 341                    tllink, tllink2, allow, disallow, state, ''))
 342            except Exception:
 343                pass
 344        return conn
 345
 346    def getEdges(self, withInternal=True):
 347        if not withInternal:
 348            return [e for e in self._edges if e.getFunction() == '']
 349        else:
 350            return self._edges
 351
 352    def getRoundabouts(self):
 353        return self._roundabouts
 354
 355    def hasEdge(self, id):
 356        return id in self._id2edge
 357
 358    def getEdge(self, id):
 359        return self._id2edge[id]
 360
 361    def getLane(self, laneID):
 362        edge_id, lane_index = laneID.rsplit("_", 1)
 363        return self.getEdge(edge_id).getLane(int(lane_index))
 364
 365    def _initRTree(self, shapeList, includeJunctions=True):
 366        import rtree  # noqa
 367        result = rtree.index.Index()
 368        result.interleaved = True
 369        for ri, shape in enumerate(shapeList):
 370            result.add(ri, shape.getBoundingBox(includeJunctions))
 371        return result
 372
 373    # Please be aware that the resulting list of edges is NOT sorted
 374    def getNeighboringEdges(self, x, y, r=0.1, includeJunctions=True, allowFallback=True):
 375        edges = []
 376        rtree = self._rtreeEdges[includeJunctions]
 377        try:
 378            if rtree is None:
 379                rtree = self._initRTree(self._edges, includeJunctions)
 380                self._rtreeEdges[includeJunctions] = rtree
 381            for i in rtree.intersection((x - r, y - r, x + r, y + r)):
 382                e = self._edges[i]
 383                d = sumolib.geomhelper.distancePointToPolygon(
 384                    (x, y), e.getShape(includeJunctions))
 385                if d < r:
 386                    edges.append((e, d))
 387        except ImportError:
 388            if not allowFallback:
 389                raise
 390            warnings.warn("Module 'rtree' not available. Using brute-force fallback.")
 391            for the_edge in self._edges:
 392                d = sumolib.geomhelper.distancePointToPolygon((x, y), the_edge.getShape(includeJunctions))
 393                if d < r:
 394                    edges.append((the_edge, d))
 395        return edges
 396
 397    def getNeighboringLanes(self, x, y, r=0.1, includeJunctions=True, allowFallback=True):
 398        lanes = []
 399        rtree = self._rtreeLanes[includeJunctions]
 400        try:
 401            if rtree is None:
 402                for the_edge in self._edges:
 403                    self._allLanes += the_edge.getLanes()
 404                rtree = self._initRTree(self._allLanes, includeJunctions)
 405                self._rtreeLanes[includeJunctions] = rtree
 406            for i in rtree.intersection((x - r, y - r, x + r, y + r)):
 407                the_lane = self._allLanes[i]
 408                d = sumolib.geomhelper.distancePointToPolygon((x, y), the_lane.getShape(includeJunctions))
 409                if d < r:
 410                    lanes.append((the_lane, d))
 411        except ImportError:
 412            if not allowFallback:
 413                raise
 414            warnings.warn("Module 'rtree' not available. Using brute-force fallback.")
 415            for the_edge in self._edges:
 416                for the_lane in the_edge.getLanes():
 417                    d = sumolib.geomhelper.distancePointToPolygon((x, y), the_lane.getShape(includeJunctions))
 418                    if d < r:
 419                        lanes.append((the_lane, d))
 420        return lanes
 421
 422    def getNearestLane(self, x, y, r=0.1, vClass=None, isGeo=False, allowFallback=True):
 423        """Return (lane, lanePos, dist) for the nearest lane at a distance less than r,
 424        or None if no lane matches. Both lanePos and dist are in meters.
 425
 426        If vClass is given, only lanes allowing that vehicle class are considered.
 427        If isGeo is True, x and y are longitude and latitude, respectively, and
 428        the network's geo-projection is used (requires pyproj).
 429        Distances are measured to the lane shape without junction extensions.
 430        Internal lanes are considered if the network was read withInternal=True.
 431        """
 432        if isGeo:
 433            x, y = self.convertLonLat2XY(x, y)
 434        candidates = [(lane, dist) for lane, dist in
 435                      self.getNeighboringLanes(x, y, r, includeJunctions=False, allowFallback=allowFallback)
 436                      if lane.allows(vClass)]
 437        if not candidates:
 438            return None
 439        lane, dist = min(candidates, key=lambda candidate: candidate[1])
 440        lanePos, dist = lane.getClosestLanePosAndDist((x, y))
 441        return lane, lanePos, dist
 442
 443    def hasNode(self, id):
 444        return id in self._id2node
 445
 446    def getNode(self, id):
 447        return self._id2node[id]
 448
 449    def getNodes(self):
 450        return self._nodes
 451
 452    def getTLS(self, tlid):
 453        return self._id2tls[tlid]
 454
 455    def getTLSSecure(self, tlid):
 456        if tlid in self._id2tls:
 457            tls = self._id2tls[tlid]
 458        else:
 459            tls = TLS(tlid)
 460            self._id2tls[tlid] = tls
 461            self._tlss.append(tls)
 462        return tls
 463
 464    def getTrafficLights(self):
 465        return self._tlss
 466
 467    def addTLS(self, tlid, inLane, outLane, linkNo):
 468        tls = self.getTLSSecure(tlid)
 469        tls.addConnection(inLane, outLane, linkNo)
 470        return tls
 471
 472    def addTLSProgram(self, tlid, programID, offset, type, removeOthers):
 473        tls = self.getTLSSecure(tlid)
 474        program = TLSProgram(programID, offset, type)
 475        if removeOthers:
 476            tls.removePrograms()
 477        tls.addProgram(program)
 478        return program
 479
 480    def setFoes(self, junctionID, index, foes, prohibits):
 481        self._id2node[junctionID].setFoes(index, foes, prohibits)
 482
 483    def forbids(self, possProhibitor, possProhibited):
 484        return possProhibitor.getFrom().getToNode().forbids(possProhibitor, possProhibited)
 485
 486    def getUpstreamEdges(self, edge, distance, stopOnTLS, stopOnTurnaround):
 487        """return a list of lists of the form
 488           [[firstEdge, pos, [edge_0, edge_1, ..., edge_k], aborted], ...]
 489           where
 490             firstEdge: is the upstream edge furthest away from the intersection,
 491             [edge_0, ..., edge_k]: is the list of edges from the intersection upstream to firstEdge
 492             pos: is the position on firstEdge with distance to the end of the input edge
 493             aborted: a flag indicating whether the downstream
 494                 search stopped at a TLS or a node without incoming edges before reaching the distance threshold
 495        """
 496        ret = []
 497        seen = set()
 498        toProc = []
 499        toProc.append([edge, 0, []])
 500        while not len(toProc) == 0:
 501            ie = toProc.pop()
 502            if ie[0] in seen:
 503                continue
 504            seen.add(ie[0])
 505            if ie[1] + ie[0].getLength() >= distance:
 506                ret.append(
 507                    [ie[0], ie[0].getLength() + ie[1] - distance, ie[2], False])
 508                continue
 509            if len(ie[0]._incoming) == 0:
 510                ret.append([ie[0], ie[0].getLength() + ie[1], ie[2], True])
 511                continue
 512            mn = []
 513            stop = False
 514            for ci in ie[0]._incoming:
 515                if ci not in seen:
 516                    prev = copy(ie[2])
 517                    if stopOnTLS and ci._tls and ci != edge and not stop:
 518                        ret.append([ie[0], ie[1], prev, True])
 519                        stop = True
 520                    elif (stopOnTurnaround and ie[0]._incoming[ci][0].getDirection() == Connection.LINKDIR_TURN and
 521                          not stop):
 522                        ret.append([ie[0], ie[1], prev, True])
 523                        stop = True
 524                    else:
 525                        prev.append(ie[0])
 526                        mn.append([ci, ie[0].getLength() + ie[1], prev])
 527            if not stop:
 528                toProc.extend(mn)
 529        return ret
 530
 531    def getEdgesByOrigID(self, origID):
 532        if self._origIdx is None:
 533            self._origIdx = defaultdict(set)
 534            for the_edge in self._edges:
 535                for the_lane in the_edge.getLanes():
 536                    for oID in the_lane.getParam("origId", "").split():
 537                        self._origIdx[oID].add(the_edge)
 538        return self._origIdx[origID]
 539
 540    def getGeometries(self, useLanes, includeJunctions=False):
 541        for e in self._edges:
 542            if useLanes:
 543                for the_lane in e.getLanes():
 544                    yield the_lane.getID(), the_lane.getShape(), the_lane.getWidth()
 545            else:
 546                yield e.getID(), e.getShape(includeJunctions), sum([the_lane.getWidth() for the_lane in e.getLanes()])
 547
 548    def getBBoxXY(self):
 549        """
 550        Get the bounding box (bottom left and top right coordinates) for a net;
 551        Coordinates are in X and Y (not Lat and Lon)
 552
 553        :return [(bottom_left_X, bottom_left_Y), (top_right_X, top_right_Y)]
 554        """
 555        return [(self._ranges[0][0], self._ranges[1][0]),
 556                (self._ranges[0][1], self._ranges[1][1])]
 557
 558    # the diagonal of the bounding box of all nodes
 559    def getBBoxDiameter(self):
 560        return math.sqrt(
 561            (self._ranges[0][0] - self._ranges[0][1]) ** 2 +
 562            (self._ranges[1][0] - self._ranges[1][1]) ** 2)
 563
 564    def hasGeoProj(self):
 565        projString = self._location["projParameter"]
 566        return projString != "!"
 567
 568    def getGeoProj(self):
 569        if not self.hasGeoProj() or not HAVE_PYPROJ:
 570            raise RuntimeError("Network does not provide geo-projection or pyproj not installed.")
 571        if self._proj is None:
 572            try:
 573                self._proj = pyproj.Proj(projparams=self._location["projParameter"])
 574            except RuntimeError:
 575                if hasattr(pyproj.datadir, 'set_data_dir'):
 576                    pyproj.datadir.set_data_dir('/usr/share/proj')
 577                    self._proj = pyproj.Proj(projparams=self._location["projParameter"])
 578                raise
 579        return self._proj
 580
 581    def getLocationOffset(self):
 582        """ offset to be added after converting from geo-coordinates to UTM"""
 583        return list(map(float, self._location["netOffset"].split(",")))
 584
 585    def getBoundary(self):
 586        """ return xmin,ymin,xmax,ymax network coordinates"""
 587        return list(map(float, self._location["convBoundary"].split(",")))
 588
 589    def convertLonLat2XY(self, lon, lat, rawUTM=False):
 590        x, y = self.getGeoProj()(lon, lat)
 591        if rawUTM:
 592            return x, y
 593        else:
 594            x_off, y_off = self.getLocationOffset()
 595            return x + x_off, y + y_off
 596
 597    def convertXY2LonLat(self, x, y, rawUTM=False):
 598        if not rawUTM:
 599            x_off, y_off = self.getLocationOffset()
 600            x -= x_off
 601            y -= y_off
 602        return self.getGeoProj()(x, y, inverse=True)
 603
 604    def move(self, dx, dy, dz=0):
 605        for n in self._nodes:
 606            n._coord = (n._coord[0] + dx, n._coord[1] + dy, n._coord[2] + dz)
 607        for e in self._edges:
 608            for _lane in e.getLanes():
 609                _lane.setShape([(p[0] + dx, p[1] + dy, p[2] + dz) for p in _lane.getShape3D()])
 610            e.rebuildShape()
 611
 612    def getInternalPath(self, conn, fastest=False):
 613        minInternalCost = PRACTICAL_INFINITY
 614        minPath = None
 615        for c in conn:
 616            if c.getViaLaneID() != "":
 617                viaCost = 0
 618                viaID = c.getViaLaneID()
 619                viaPath = []
 620                while viaID != "":
 621                    viaLane = self.getLane(viaID)
 622                    viaCost += viaLane.getLength() if not fastest else viaLane.getLength() / viaLane.getSpeed()
 623                    viaID = viaLane.getOutgoing()[0].getViaLaneID()
 624                    viaPath.append(viaLane.getEdge())
 625                if viaCost < minInternalCost:
 626                    minInternalCost = viaCost
 627                    minPath = viaPath
 628        return minPath, minInternalCost
 629
 630    def getOptimalPath(self, fromEdge, toEdge, fastest=False, maxCost=PRACTICAL_INFINITY, vClass=None,
 631                       reversalPenalty=0, includeFromToCost=True, withInternal=False, ignoreDirection=False,
 632                       fromPos=None, toPos=None, preferences={}):
 633        """
 634        Finds the optimal (shortest or fastest) path for vClass from fromEdge to toEdge
 635        by using using Dijkstra's algorithm.
 636        It returns a pair of a tuple of edges and the cost.
 637        If no path is found the first element is None.
 638        The cost for the returned path is equal to the sum of all edge costs in the path,
 639        including the internal connectors, if they are present in the network.
 640        The path itself does not include internal edges except for the case
 641        when the start or end edge are internal edges.
 642        The search may be limited using the given threshold.
 643        The preferences declare a mapping from the 'routingType' of each edge to divisor
 644        that is applied to the computed cost of that edge (i.e. a preference of 2 reduces cost by a factor of 0.5).
 645        """
 646
 647        if preferences:
 648            if fastest:
 649                def speedFunc(edge):
 650                    return preferences.get(edge.getRoutingType(), 1.0) * edge.getSpeed()
 651            else:
 652                def speedFunc(edge):
 653                    return preferences.get(edge.getRoutingType(), 1.0)
 654        elif fastest:
 655            def speedFunc(edge):
 656                return edge.getSpeed()
 657        else:
 658            def speedFunc(edge):
 659                return 1.0
 660
 661        def remainder(edge, pos):
 662            if pos < 0:
 663                return min(-pos, edge.getLength())
 664            return max(0., edge.getLength() - pos)
 665
 666        def getToNormalIncoming(edge):
 667            if edge.getFunction() == '':
 668                return [(e, None) for e in edge.getToNode().getIncoming() if e.getFunction() == '']
 669            else:
 670                return []
 671
 672        if self._edgeSuccessorCache != (vClass, ignoreDirection):
 673            self._edgeSuccessorCache = (vClass, ignoreDirection)
 674            for e1 in self.getEdges():
 675                e1._succCacche = tuple(chain(e1.getAllowedOutgoing(vClass).items(),
 676                                       e1.getIncoming().items() if ignoreDirection else [],
 677                                       getToNormalIncoming(e1) if ignoreDirection and not self.hasWalkingArea else []))
 678
 679        if self._edgeCostCache != (fastest, preferences):
 680            self._edgeCostCache = (fastest, preferences)
 681            for e1 in self.getEdges():
 682                e1._costCache = e1.getLength() / speedFunc(e1)
 683
 684        if self.hasInternal:
 685            appendix = []
 686            appendixCost = 0.
 687            while toEdge.getFunction() == "internal":
 688                appendix = [toEdge] + appendix
 689                appendixCost += toEdge._costCache
 690                toEdge = list(toEdge.getIncoming().keys())[0]
 691
 692        def finalizeCost(cost, path):
 693            if includeFromToCost:
 694                # add costs for (part of) the first edge, still needs to be fixed for wrong direction travel
 695                remainFrom = fromEdge.getLength() if fromPos is None else remainder(fromEdge, fromPos)
 696                cost += remainFrom / speedFunc(fromEdge)
 697                # remove costs for (part of) the last edge, still needs to be fixed for wrong direction travel
 698                removeTo = 0. if toPos is None else remainder(toEdge, toPos)
 699            else:
 700                removeTo = toEdge.getLength() if len(path) > 1 else 0.
 701            cost -= removeTo / speedFunc(toEdge)
 702            return cost
 703
 704        def constructPath(dist):
 705            # destination was already reached in a previous query
 706            cost, pred = dist[toEdge]
 707            path = [toEdge]
 708            while pred is not None:
 709                if self.hasInternal and withInternal:
 710                    viaPath, minInternalCost = self.getInternalPath(pred.getAllowedOutgoing(vClass).get(path[-1], []),
 711                                                                    fastest=fastest)
 712                    if viaPath is not None:
 713                        path += reversed(viaPath)
 714                path.append(pred)
 715                _, pred = dist[pred]
 716
 717            path.reverse()
 718            cost = finalizeCost(cost, path)
 719            assert cost >= 0
 720            if self.hasInternal:
 721                if appendix:
 722                    return tuple(path + appendix), cost + appendixCost
 723                elif ignoreDirection and self.hasWalkingArea and not withInternal:
 724                    return [e for e in path if e.getFunction() == ''], cost
 725            return tuple(path), cost
 726
 727        needLoop = (fromEdge == toEdge
 728                    and fromPos is not None
 729                    and toPos is not None
 730                    and fromPos > toPos
 731                    and not ignoreDirection)
 732
 733        seen = set()
 734        dist = {}
 735        q = []
 736        if self._routingCache is not None:
 737            if needLoop:
 738                # use cached results from all follower edges:
 739                bestCost = maxCost
 740                bestPath = None
 741                for e2, conn in fromEdge.getAllowedOutgoing(vClass).items():
 742                    path, cost = self.getOptimalPath(e2, toEdge, fastest=fastest, maxCost=maxCost, vClass=vClass,
 743                                                     reversalPenalty=reversalPenalty,
 744                                                     includeFromToCost=includeFromToCost,
 745                                                     withInternal=withInternal, fromPos=0, toPos=toPos,
 746                                                     preferences=preferences)
 747                    if path is not None and cost < bestCost:
 748                        bestPath = path
 749                        bestCost = cost
 750                if bestPath is not None:
 751                    path = [fromEdge]
 752                    if self.hasInternal and withInternal:
 753                        viaPath, minInternalCost = self.getInternalPath(
 754                            fromEdge.getAllowedOutgoing(vClass).get(path[0], []), fastest=fastest)
 755                        if viaPath is not None:
 756                            path += viaPath
 757                            bestCost += minInternalCost
 758                    path += list(bestPath)
 759                    if includeFromToCost:
 760                        bestCost += remainder(fromEdge, fromPos) / speedFunc(fromEdge)
 761                    return tuple(path), bestCost
 762                else:
 763                    return None, PRACTICAL_INFINITY
 764
 765            dist = self._routingCache(fromEdge, fastest, vClass, ignoreDirection, reversalPenalty,
 766                                      tuple(preferences.items()))
 767            if toEdge in dist:
 768                return constructPath(dist)
 769            else:
 770                # initialize heap from previous query
 771                q = []
 772                frontier = set(dist.keys())
 773                for cost, prev in dist.values():
 774                    frontier.discard(prev)
 775                for e in frontier:
 776                    cost, prev = dist[e]
 777                    heapq.heappush(q, (cost, e, prev))
 778        elif needLoop:
 779            # start search on successors of fromEdge
 780            for e2, conn in fromEdge.getAllowedOutgoing(vClass).items():
 781                q.append((e2._costCache, e2, fromEdge))
 782
 783        if len(dist) == 0:
 784            dist[fromEdge] = (0., None)
 785            if not needLoop:
 786                q.append((0., fromEdge, None))
 787
 788        while q:
 789            cost, e1, prev = heapq.heappop(q)
 790            if e1 in seen:
 791                continue
 792            seen.add(e1)
 793            if e1 == toEdge:
 794                return constructPath(dist)
 795            if cost > maxCost:
 796                return None, cost
 797
 798            for e2, conn in e1._succCacche:
 799                if e2 not in seen:
 800                    newCost = cost + e2._costCache
 801                    #  print(cost, newCost, e2.getID(), speedFunc(e2))
 802                    if e2 == e1.getBidi():
 803                        newCost += reversalPenalty
 804                    if self.hasInternal and conn is not None:
 805                        viaPath, minInternalCost = self.getInternalPath(conn, fastest=fastest)
 806                        if viaPath is not None:
 807                            newCost += minInternalCost
 808                    if e2 not in dist or newCost < dist[e2][0]:
 809                        dist[e2] = (newCost, e1)
 810                        heapq.heappush(q, (newCost, e2, e1))
 811        return None, PRACTICAL_INFINITY
 812
 813    def getShortestPath(self, fromEdge, toEdge, maxCost=PRACTICAL_INFINITY, vClass=None, reversalPenalty=0,
 814                        includeFromToCost=True, withInternal=False, ignoreDirection=False,
 815                        fromPos=None, toPos=None, preferences={}):
 816        """
 817        Finds the shortest path from fromEdge to toEdge respecting vClass, using Dijkstra's algorithm.
 818        It returns a pair of a tuple of edges and the cost. If no path is found the first element is None.
 819        The cost for the returned path is equal to the sum of all edge lengths in the path,
 820        including the internal connectors, if they are present in the network.
 821        The path itself does not include internal edges except for the case
 822        when the start or end edge are internal edges.
 823        The search may be limited using the given threshold.
 824        The preferences declare a mapping from the 'routingType' of each edge to divisor
 825        that is applied to the lenght of that edge
 826        (i.e. a preference of 2 reduces reduces effective length by a factor of 0.5).
 827        """
 828
 829        return self.getOptimalPath(fromEdge, toEdge, False, maxCost, vClass, reversalPenalty,
 830                                   includeFromToCost, withInternal, ignoreDirection, fromPos, toPos,
 831                                   preferences)
 832
 833    def getFastestPath(self, fromEdge, toEdge, maxCost=PRACTICAL_INFINITY, vClass=None, reversalPenalty=0,
 834                       includeFromToCost=True, withInternal=False, ignoreDirection=False,
 835                       fromPos=None, toPos=None, preferences={}):
 836        """
 837        Finds the fastest path from fromEdge to toEdge respecting vClass, using Dijkstra's algorithm.
 838        It returns a pair of a tuple of edges and the cost. If no path is found the first element is None.
 839        The cost for the returned path is equal to the sum of all edge costs in the path,
 840        including the internal connectors, if they are present in the network.
 841        The path itself does not include internal edges except for the case
 842        when the start or end edge are internal edges.
 843        The search may be limited using the given threshold.
 844        The preferences declare a mapping from the 'routingType' of each edge to divisor
 845        that is applied to the computed traveltime of that edge
 846        (i.e. a preference of 2 reduces effective traveltime by a factor of 0.5).
 847        """
 848
 849        return self.getOptimalPath(fromEdge, toEdge, True, maxCost, vClass, reversalPenalty,
 850                                   includeFromToCost, withInternal, ignoreDirection, fromPos, toPos,
 851                                   preferences)
 852
 853    def getReachable(self, source, vclass=None, useIncoming=False, cache=None):
 854        if vclass is not None and not source.allows(vclass):
 855            raise RuntimeError("'{}' does not allow {}".format(source.getID(), vclass))
 856        fringe = [source]
 857        found = set()
 858        found.add(source)
 859        while len(fringe) > 0:
 860            new_fringe = []
 861            for e in fringe:
 862                if vclass == "pedestrian":
 863                    cands = chain(chain(*e.getIncoming().values()), chain(*e.getOutgoing().values()))
 864                else:
 865                    cands = chain(*(e.getIncoming().values() if useIncoming else e.getOutgoing().values()))
 866                # print("\n".join(map(str, list(cands))))
 867                for conn in cands:
 868                    if vclass is None or (
 869                            conn.getFromLane().allows(vclass)
 870                            and conn.getToLane().allows(vclass)):
 871                        for reachable in [conn.getTo(), conn.getFrom()]:
 872                            if reachable not in found:
 873                                # print("added %s via %s" % (reachable, conn))
 874                                if cache and reachable in cache:
 875                                    found.update(cache[reachable])
 876                                else:
 877                                    found.add(reachable)
 878                                    new_fringe.append(reachable)
 879            fringe = new_fringe
 880        if cache is not None:
 881            cache[source] = tuple(found)
 882        return found
 883
 884
 885class NetReader(handler.ContentHandler):
 886
 887    """Reads a network, storing the edge geometries, lane numbers and max. speeds"""
 888
 889    def __init__(self, **others):
 890        self._net = others.get('net', Net())
 891        self._currentEdge = None
 892        self._currentNode = None
 893        self._currentConnection = None
 894        self._currentLane = None
 895        self._crossingID2edgeIDs = {}
 896        self._withPhases = others.get('withPrograms', False)
 897        self._latestProgram = others.get('withLatestPrograms', False)
 898        if self._latestProgram:
 899            self._withPhases = True
 900        self._withConnections = others.get('withConnections', True)
 901        self._withFoes = others.get('withFoes', True)
 902        self._withPedestrianConnections = others.get('withPedestrianConnections', False)
 903        self._withMacroConnectors = others.get('withMacroConnectors', False)
 904        self._withInternal = others.get('withInternal', self._withPedestrianConnections)
 905        if self._withPedestrianConnections and not self._withInternal:
 906            sys.stderr.write("Warning: Option withPedestrianConnections requires withInternal\n")
 907            self._withInternal = True
 908        self._bidiEdgeIDs = {}
 909
 910    def startElement(self, name, attrs):
 911        if name == 'net':
 912            parts = attrs["version"].split('.', 1)
 913            self._net._version = (int(parts[0]), float(parts[1]))
 914        elif name == 'location':
 915            self._net.setLocation(attrs["netOffset"], attrs["convBoundary"], attrs[
 916                                  "origBoundary"], attrs["projParameter"])
 917        elif name == 'type':
 918            self._net._edgeTypes[attrs['id']] = EdgeType(attrs['id'], attrs.get('allow'), attrs.get('disallow'))
 919        elif name == 'edge':
 920            function = attrs.get('function', '')
 921            if (function == ''
 922                    or (self._withInternal and function in ['internal', 'crossing', 'walkingarea'])
 923                    or (self._withMacroConnectors and function == 'connector')):
 924                prio = -1
 925                if 'priority' in attrs:
 926                    prio = int(attrs['priority'])
 927
 928                # get the  ids
 929                edgeID = attrs['id']
 930                fromNodeID = attrs.get('from', None)
 931                toNodeID = attrs.get('to', None)
 932
 933                # for internal junctions use the junction's id for from and to node
 934                if function == 'internal' or function == 'crossing' or function == 'walkingarea':
 935                    fromNodeID = toNodeID = edgeID[1:edgeID.rfind('_')]
 936
 937                # remember edges crossed by pedestrians to link them later to the crossing objects
 938                if function == 'crossing':
 939                    self._crossingID2edgeIDs[edgeID] = attrs.get('crossingEdges').split(' ')
 940
 941                self._currentEdge = self._net.addEdge(edgeID, fromNodeID, toNodeID, prio, function,
 942                                                      attrs.get('name', ''), attrs.get('type', ''),
 943                                                      attrs.get('routingType', ''))
 944
 945                self._currentEdge.setRawShape(convertShape(attrs.get('shape', '')))
 946
 947                bidi = attrs.get('bidi', '')
 948                if bidi:
 949                    self._bidiEdgeIDs[edgeID] = bidi
 950            else:
 951                if function in ['crossing', 'walkingarea']:
 952                    self._net._crossings_and_walkingAreas.add(attrs['id'])
 953                elif function == 'connector':
 954                    self._net._macroConnectors.add(attrs['id'])
 955                self._currentEdge = None
 956        elif name == 'lane' and self._currentEdge is not None:
 957            self._currentLane = self._net.addLane(
 958                self._currentEdge,
 959                float(attrs['speed']),
 960                float(attrs['length']),
 961                float(attrs.get('width', 3.2)),
 962                attrs.get('allow'),
 963                attrs.get('disallow'),
 964                attrs.get('acceleration') == "1")
 965            self._currentLane.setShape(convertShape(attrs.get('shape', '')))
 966        elif name == 'neigh' and self._currentLane is not None:
 967            self._currentLane.setNeigh(attrs['lane'])
 968        elif name == 'junction':
 969            if attrs['id'][0] != ':':
 970                intLanes = None
 971                if self._withInternal:
 972                    intLanes = attrs["intLanes"].split(" ")
 973                self._currentNode = self._net.addNode(attrs['id'], attrs['type'],
 974                                                      tuple(
 975                                                          map(float, [attrs['x'], attrs['y'],
 976                                                                      attrs['z'] if 'z' in attrs else '0'])),
 977                                                      attrs['incLanes'].split(" "), intLanes)
 978                self._currentNode.setShape(
 979                    convertShape(attrs.get('shape', '')))
 980                if 'fringe' in attrs:
 981                    self._currentNode._fringe = attrs['fringe']
 982
 983        elif name == 'succ' and self._withConnections:  # deprecated
 984            if attrs['edge'][0] != ':':
 985                self._currentEdge = self._net.getEdge(attrs['edge'])
 986                self._currentLane = attrs['lane']
 987                self._currentLane = int(
 988                    self._currentLane[self._currentLane.rfind('_') + 1:])
 989            else:
 990                self._currentEdge = None
 991        elif name == 'succlane' and self._withConnections:  # deprecated
 992            lid = attrs['lane']
 993            if lid[0] != ':' and lid != "SUMO_NO_DESTINATION" and self._currentEdge:
 994                connected = self._net.getEdge(lid[:lid.rfind('_')])
 995                tolane = int(lid[lid.rfind('_') + 1:])
 996                if 'tl' in attrs and attrs['tl'] != "":
 997                    tl = attrs['tl']
 998                    tllink = int(attrs['linkIdx'])
 999                    tlid = attrs['tl']
1000                    toEdge = self._net.getEdge(lid[:lid.rfind('_')])
1001                    tolane2 = toEdge._lanes[tolane]
1002                    tls = self._net.addTLS(
1003                        tlid, self._currentEdge._lanes[self._currentLane], tolane2, tllink)
1004                    self._currentEdge.setTLS(tls)
1005                else:
1006                    tl = ""
1007                    tllink = -1
1008                toEdge = self._net.getEdge(lid[:lid.rfind('_')])
1009                tolane = toEdge._lanes[tolane]
1010                viaLaneID = attrs['via']
1011                self._net.addConnection(self._currentEdge, connected, self._currentEdge._lanes[
1012                                        self._currentLane], tolane,
1013                                        attrs['dir'], tl, tllink, -1,
1014                                        attrs.get('allow'), attrs.get('disallow'), attrs['state'], viaLaneID)
1015        elif name == 'connection' and self._withConnections and (attrs['from'][0] != ":" or self._withInternal):
1016            fromEdgeID = attrs['from']
1017            toEdgeID = attrs['to']
1018            if ((self._withPedestrianConnections or not (fromEdgeID in self._net._crossings_and_walkingAreas or
1019                                                         toEdgeID in self._net._crossings_and_walkingAreas))
1020                and (self._withMacroConnectors or not (fromEdgeID in self._net._macroConnectors or toEdgeID in
1021                                                       self._net._macroConnectors))):
1022                fromEdge = self._net.getEdge(fromEdgeID)
1023                toEdge = self._net.getEdge(toEdgeID)
1024                fromLane = fromEdge.getLane(int(attrs['fromLane']))
1025                toLane = toEdge.getLane(int(attrs['toLane']))
1026                if 'tl' in attrs and attrs['tl'] != "":
1027                    tl = attrs['tl']
1028                    tllink = int(attrs['linkIndex'])
1029                    tllink2 = int(attrs.get('linkIndex2', -1))
1030                    tls = self._net.addTLS(tl, fromLane, toLane, tllink)
1031                    fromEdge.setTLS(tls)
1032                else:
1033                    tl = ""
1034                    tllink = -1
1035                    tllink2 = -1
1036                try:
1037                    viaLaneID = attrs['via']
1038                except KeyError:
1039                    viaLaneID = ''
1040
1041                self._currentConnection = self._net.addConnection(
1042                    fromEdge, toEdge, fromLane, toLane, attrs['dir'], tl,
1043                    tllink, tllink2, attrs.get('allow'), attrs.get('disallow'), attrs['state'], viaLaneID)
1044
1045        # 'row-logic' is deprecated!!!
1046        elif self._withFoes and name == 'ROWLogic':
1047            self._currentNode = attrs['id']
1048        elif name == 'logicitem' and self._withFoes:  # deprecated
1049            self._net.setFoes(
1050                self._currentNode, int(attrs['request']), attrs["foes"], attrs["response"])
1051        elif name == 'request' and self._withFoes:
1052            self._currentNode.setFoes(
1053                int(attrs['index']), attrs["foes"], attrs["response"])
1054        # tl-logic is deprecated!!! NOTE: nevertheless, this is still used by
1055        # netconvert... (Leo)
1056        elif self._withPhases and name == 'tlLogic':
1057            self._currentProgram = self._net.addTLSProgram(
1058                attrs['id'], attrs['programID'],
1059                intIfPossible(float(attrs['offset'])), attrs['type'], self._latestProgram)
1060        elif self._withPhases and name == 'phase':
1061            self._currentProgram.addPhase(
1062                attrs['state'],
1063                intIfPossible(float(attrs['duration'])),
1064                intIfPossible(float(attrs['minDur'])) if 'minDur' in attrs else -1,
1065                intIfPossible(float(attrs['maxDur'])) if 'maxDur' in attrs else -1,
1066                list(map(int, attrs['next'].split())) if 'next' in attrs else [],
1067                attrs['name'] if 'name' in attrs else ""
1068            )
1069        elif name == 'roundabout':
1070            self._net.addRoundabout(
1071                attrs['nodes'].split(), attrs['edges'].split())
1072        elif name == 'param':
1073            if self._currentLane is not None:
1074                self._currentLane.setParam(attrs['key'], attrs['value'])
1075            elif self._currentEdge is not None:
1076                self._currentEdge.setParam(attrs['key'], attrs['value'])
1077            elif self._currentNode is not None:
1078                self._currentNode.setParam(attrs['key'], attrs['value'])
1079            elif self._currentConnection is not None:
1080                self._currentConnection.setParam(attrs['key'], attrs['value'])
1081            elif self._withPhases and self._currentProgram is not None:
1082                self._currentProgram.setParam(attrs['key'], attrs['value'])
1083
1084    def endElement(self, name):
1085        if name == 'lane':
1086            self._currentLane = None
1087        elif name == 'edge':
1088            self._currentEdge = None
1089        elif name == 'junction':
1090            self._currentNode = None
1091        elif name == 'connection':
1092            self._currentConnection = None
1093        # 'row-logic' is deprecated!!!
1094        elif name == 'ROWLogic' or name == 'row-logic':
1095            self._haveROWLogic = False
1096        # tl-logic is deprecated!!!
1097        elif self._withPhases and (name == 'tlLogic' or name == 'tl-logic'):
1098            self._currentProgram = None
1099        elif name == 'net':
1100            for edgeID, bidiID in self._bidiEdgeIDs.items():
1101                self._net.getEdge(edgeID)._bidi = self._net.getEdge(bidiID)
1102
1103    def endDocument(self):
1104        # set crossed edges of pedestrian crossings
1105        for crossingID, crossedEdgeIDs in self._crossingID2edgeIDs.items():
1106            pedCrossing = self._net.getEdge(crossingID)
1107            for crossedEdgeID in crossedEdgeIDs:
1108                pedCrossing._addCrossingEdge(self._net.getEdge(crossedEdgeID))
1109
1110    def getNet(self):
1111        return self._net
1112
1113
1114def convertShape(shapeString):
1115    """ Convert xml shape string into float tuples.
1116
1117    This method converts the 2d or 3d shape string from SUMO's xml file
1118    into a list containing 3d float-tuples. Non existent z coordinates default
1119    to zero. If shapeString is empty, an empty list will be returned.
1120    """
1121
1122    cshape = []
1123    for pointString in shapeString.split():
1124        p = [float(e) for e in pointString.split(",")]
1125        if len(p) == 2:
1126            cshape.append((p[0], p[1], 0.))
1127        elif len(p) == 3:
1128            cshape.append(tuple(p))
1129        else:
1130            raise ValueError(
1131                'Invalid shape point "%s", should be either 2d or 3d' % pointString)
1132    return cshape
1133
1134
1135def lane2edge(laneID):
1136    return laneID[:laneID.rfind("_")]
1137
1138
1139def lane2index(laneID):
1140    return int(laneID[laneID.rfind("_") + 1:])
1141
1142
1143def hasInternal(filename):
1144    with openz(filename) as f:
1145        for line in f:
1146            if '<edge id="' in line:
1147                if '<edge id=":' in line:
1148                    return True
1149                else:
1150                    return False
1151    return False
1152
1153
1154def readNet(filename, **others):
1155    """ load a .net.xml file
1156    The following named options are supported:
1157
1158        'net' : initialize data structures with an existing net object (default Net())
1159        'withPrograms' : import all traffic light programs (default False)
1160        'withLatestPrograms' : import only the last program for each traffic light.
1161                               This is the program that would be active in sumo by default.
1162                               (default False)
1163        'withConnections' : import all connections (default True)
1164        'withFoes' : import right-of-way information (default True)
1165        'withInternal' : import internal edges and lanes (default False)
1166        'withPedestrianConnections' : import connections between sidewalks, crossings (default False)
1167        'lxml' : set to False to use the xml.sax parser instead of the lxml parser
1168        'maxcache' : set maximum cache size (default 1000) or 0 to disable optimal route caching
1169    """
1170    netreader = NetReader(**others)
1171    try:
1172        source = gzip.open(filename)
1173        source.read(10)
1174        source.seek(0)
1175    except IOError:
1176        source = filename
1177    if HAVE_LXML and others.get("lxml", True):
1178        if isinstance(source, pathlib.Path):
1179            source = str(source)
1180        for event, v in lxml.etree.iterparse(source, events=("start", "end")):
1181            if event == "start":
1182                netreader.startElement(v.tag, v.attrib)
1183            elif event == "end":
1184                netreader.endElement(v.tag)
1185            v.clear()  # reduce memory footprint
1186    else:
1187        parse(source, netreader)
1188    net = netreader.getNet()
1189    maxcache = others.get('maxcache', 1000)
1190    if HAVE_LRU_CACHE and maxcache is not None and maxcache > 0:
1191        net.initRoutingCache(maxcache)
1192    return net