traci._lane

  1# -*- coding: utf-8 -*-
  2# Eclipse SUMO, Simulation of Urban MObility; see https://eclipse.dev/sumo
  3# Copyright (C) 2011-2026 German Aerospace Center (DLR) and others.
  4# This program and the accompanying materials are made available under the
  5# terms of the Eclipse Public License 2.0 which is available at
  6# https://www.eclipse.org/legal/epl-2.0/
  7# This Source Code may also be made available under the following Secondary
  8# Licenses when the conditions for such availability set forth in the Eclipse
  9# Public License 2.0 are satisfied: GNU General Public License, version 2
 10# or later which is available at
 11# https://www.gnu.org/licenses/old-licenses/gpl-2.0-standalone.html
 12# SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
 13
 14# @file    _lane.py
 15# @author  Michael Behrisch
 16# @author  Daniel Krajzewicz
 17# @author  Laura Bieker
 18# @author  Jakob Erdmann
 19# @date    2011-03-17
 20
 21from __future__ import absolute_import
 22from .domain import Domain
 23from . import constants as tc
 24
 25
 26def _readLinks(result):
 27    result.read("!Bi")  # Type Compound, Length
 28    nbLinks = result.readInt()
 29    links = []
 30    for _ in range(nbLinks):
 31        result.read("!B")                           # Type String
 32        approachedLane = result.readString()
 33        result.read("!B")                           # Type String
 34        approachedInternal = result.readString()
 35        result.read("!B")                           # Type Byte
 36        hasPrio = bool(result.read("!B")[0])
 37        result.read("!B")                           # Type Byte
 38        isOpen = bool(result.read("!B")[0])
 39        result.read("!B")                           # Type Byte
 40        hasFoe = bool(result.read("!B")[0])
 41        result.read("!B")                           # Type String
 42        state = result.readString()
 43        result.read("!B")                           # Type String
 44        direction = result.readString()
 45        result.read("!B")                           # Type Float
 46        length = result.readDouble()
 47        links.append((approachedLane, hasPrio, isOpen, hasFoe,
 48                      approachedInternal, state, direction, length))
 49    return tuple(links)
 50
 51
 52_RETURN_VALUE_FUNC = {tc.LANE_LINKS: _readLinks}
 53
 54
 55class LaneDomain(Domain):
 56
 57    def __init__(self):
 58        Domain.__init__(self, "lane", tc.CMD_GET_LANE_VARIABLE, tc.CMD_SET_LANE_VARIABLE,
 59                        tc.CMD_SUBSCRIBE_LANE_VARIABLE, tc.RESPONSE_SUBSCRIBE_LANE_VARIABLE,
 60                        tc.CMD_SUBSCRIBE_LANE_CONTEXT, tc.RESPONSE_SUBSCRIBE_LANE_CONTEXT,
 61                        _RETURN_VALUE_FUNC, subscriptionDefault=(tc.LAST_STEP_VEHICLE_NUMBER,))
 62
 63    def getLength(self, laneID):
 64        """getLength(string) -> double
 65
 66        Returns the length in m.
 67        """
 68        return self._getUniversal(tc.VAR_LENGTH, laneID)
 69
 70    def getMaxSpeed(self, laneID):
 71        """getMaxSpeed(string) -> double
 72
 73        Returns the maximum allowed speed on the lane in m/s.
 74        """
 75        return self._getUniversal(tc.VAR_MAXSPEED, laneID)
 76
 77    def getFriction(self, laneID):
 78        """getFriction(string) -> double
 79
 80        Returns the friction on the lane.
 81        """
 82        return self._getUniversal(tc.VAR_FRICTION, laneID)
 83
 84    def getWidth(self, laneID):
 85        """getWidth(string) -> double
 86
 87        Returns the width of the lane in m.
 88        """
 89        return self._getUniversal(tc.VAR_WIDTH, laneID)
 90
 91    def getAllowed(self, laneID):
 92        """getAllowed(string) -> tuple(string)
 93
 94        Returns a tuple of allowed vehicle classes. An empty tuple means all vehicles are allowed.
 95        """
 96        return self._getUniversal(tc.LANE_ALLOWED, laneID)
 97
 98    def getDisallowed(self, laneID):
 99        """getDisallowed(string) -> tuple(string)
100
101        Returns a tuple of disallowed vehicle classes.
102        """
103        return self._getUniversal(tc.LANE_DISALLOWED, laneID)
104
105    def getChangePermissions(self, laneID, direction):
106        """getChangePermissions(string, int) -> tuple(string)
107
108        Returns a tuple of vehicle classesa allowed to change to the neighbor lane indicated by the direction
109        (left=0, right=1).
110        """
111        return self._getUniversal(tc.LANE_CHANGES, laneID, "b", direction)
112
113    def getLinkNumber(self, laneID):
114        """getLinkNumber(string) -> integer
115
116        Returns the number of connections to successive lanes.
117        """
118        return self._getUniversal(tc.LANE_LINK_NUMBER, laneID)
119
120    def getLinks(self, laneID, extended=True):
121        """getLinks(string) -> tuple((string, bool, bool, bool))
122        A tuple containing id of successor lane together with priority, open and foe
123        for each link.
124        if extended=True, each result tuple contains
125        (string approachedLane, bool hasPrio, bool isOpen, bool hasFoe,
126        string approachedInternal, string state, string direction, float length)
127
128        isOpen: whether a vehicle driving at the speed limit (minimum auf
129                incoming and outgoing lane) could safely pass the junction with
130                regard to approaching foes if it were to enter it in this step
131                (false for red traffic light).
132                Foe vehicles that are already on the junction are ignored!
133        hasPrio: whether the link is the main road at a priority junction or
134                 currently has green light ('G')
135        hasFoe: whether any foe vehicles are approaching the junction or on the
136                junction that would interfere with passing it in the current time step
137        """
138        complete_data = self._getUniversal(tc.LANE_LINKS, laneID)
139        if extended:
140            return complete_data
141        else:
142            # for downward compatibility
143            return [tuple(d[:4]) for d in complete_data]
144
145    def getShape(self, laneID):
146        """getShape(string) -> tuple((double, double))
147
148        Tuple of 2D positions (cartesian) describing the geometry.
149        """
150        return self._getUniversal(tc.VAR_SHAPE, laneID)
151
152    def getEdgeID(self, laneID):
153        """getEdgeID(string) -> string
154
155        Returns the id of the edge the lane belongs to.
156        """
157        return self._getUniversal(tc.LANE_EDGE_ID, laneID)
158
159    def getCO2Emission(self, laneID):
160        """getCO2Emission(string) -> double
161
162        Returns the CO2 emission in mg/s for the last time step on the given lane.
163        Multiply by the step length to get the value for one step.
164        """
165        return self._getUniversal(tc.VAR_CO2EMISSION, laneID)
166
167    def getCOEmission(self, laneID):
168        """getCOEmission(string) -> double
169
170        Returns the CO emission in mg/s for the last time step on the given lane.
171        Multiply by the step length to get the value for one step.
172        """
173        return self._getUniversal(tc.VAR_COEMISSION, laneID)
174
175    def getHCEmission(self, laneID):
176        """getHCEmission(string) -> double
177
178        Returns the HC emission in mg/s for the last time step on the given lane.
179        Multiply by the step length to get the value for one step.
180        """
181        return self._getUniversal(tc.VAR_HCEMISSION, laneID)
182
183    def getPMxEmission(self, laneID):
184        """getPMxEmission(string) -> double
185
186        Returns the particular matter emission in mg/s for the last time step on the given lane.
187        Multiply by the step length to get the value for one step.
188        """
189        return self._getUniversal(tc.VAR_PMXEMISSION, laneID)
190
191    def getNOxEmission(self, laneID):
192        """getNOxEmission(string) -> double
193
194        Returns the NOx emission in mg/s for the last time step on the given lane.
195        Multiply by the step length to get the value for one step.
196        """
197        return self._getUniversal(tc.VAR_NOXEMISSION, laneID)
198
199    def getFuelConsumption(self, laneID):
200        """getFuelConsumption(string) -> double
201
202        Returns the fuel consumption in mg/s for the last time step on the given lane.
203        Multiply by the step length to get the value for one step.
204        """
205        return self._getUniversal(tc.VAR_FUELCONSUMPTION, laneID)
206
207    def getNoiseEmission(self, laneID):
208        """getNoiseEmission(string) -> double
209
210        Returns the noise emission in db for the last time step on the given lane.
211        """
212        return self._getUniversal(tc.VAR_NOISEEMISSION, laneID)
213
214    def getElectricityConsumption(self, laneID):
215        """getElectricityConsumption(string) -> double
216
217        Returns the electricity consumption in Wh/s for the last time step.
218        Multiply by the step length to get the value for one step.
219        """
220        return self._getUniversal(tc.VAR_ELECTRICITYCONSUMPTION, laneID)
221
222    def getLastStepMeanSpeed(self, laneID):
223        """getLastStepMeanSpeed(string) -> double
224
225        Returns the average speed in m/s for the last time step on the given lane.
226        """
227        return self._getUniversal(tc.LAST_STEP_MEAN_SPEED, laneID)
228
229    def getLastStepOccupancy(self, laneID):
230        """getLastStepOccupancy(string) -> double
231
232        Returns the occupancy in % for the last time step on the given lane.
233        """
234        return self._getUniversal(tc.LAST_STEP_OCCUPANCY, laneID)
235
236    def getLastStepLength(self, laneID):
237        """getLastStepLength(string) -> double
238
239        Returns the mean vehicle length in m for the last time step on the given lane.
240        """
241        return self._getUniversal(tc.LAST_STEP_LENGTH, laneID)
242
243    def getWaitingTime(self, laneID):
244        """getWaitingTime() -> double
245
246        .
247        """
248        return self._getUniversal(tc.VAR_WAITING_TIME, laneID)
249
250    def getTraveltime(self, laneID):
251        """getTraveltime(string) -> double
252
253        Returns the estimated travel time in s for the last time step on the given lane.
254        """
255        return self._getUniversal(tc.VAR_CURRENT_TRAVELTIME, laneID)
256
257    def getLastStepVehicleNumber(self, laneID):
258        """getLastStepVehicleNumber(string) -> integer
259
260        Returns the total number of vehicles for the last time step on the given lane.
261        """
262        return self._getUniversal(tc.LAST_STEP_VEHICLE_NUMBER, laneID)
263
264    def getLastStepHaltingNumber(self, laneID):
265        """getLastStepHaltingNumber(string) -> integer
266
267        Returns the total number of halting vehicles for the last time step on the given lane.
268        A speed of less than 0.1 m/s is considered a halt.
269        """
270        return self._getUniversal(tc.LAST_STEP_VEHICLE_HALTING_NUMBER, laneID)
271
272    def getLastStepVehicleIDs(self, laneID):
273        """getLastStepVehicleIDs(string) -> tuple(string)
274
275        Returns the ids of the vehicles for the last time step on the given lane.
276        """
277        return self._getUniversal(tc.LAST_STEP_VEHICLE_ID_LIST, laneID)
278
279    def getFoes(self, laneID, toLaneID):
280        """getFoes(string, string) -> tuple(string)
281        Returns the ids of incoming lanes that have right of way over the connection from laneID to toLaneID.
282        """
283        return self._getUniversal(tc.VAR_FOES, laneID, "s", toLaneID)
284
285    def getInternalFoes(self, laneID):
286        """getFoes(string) -> tuple(string)
287        Returns the ids of internal lanes that are in conflict with the given internal lane id.
288        """
289        return self.getFoes(laneID, "")
290
291    def getPendingVehicles(self, laneID):
292        """getPendingVehicles(string) -> tuple(string)
293        Returns a tuple of all vehicle ids waiting for insertion on this lane (with depart delay).
294        """
295        return self._getUniversal(tc.VAR_PENDING_VEHICLES, laneID)
296
297    def getAngle(self, laneID, relativePosition=tc.INVALID_DOUBLE_VALUE):
298        """getAngle(string, double) -> double
299        Returns the heading of the straight line segment formed by the lane at the given position.
300        If the given position equals TraCI constant INVALID_DOUBLE_VALUE, it returns the total angle
301        formed by the lane, from its start point to its end point.
302        """
303        return self._getUniversal(tc.VAR_ANGLE, laneID, "d", relativePosition)
304
305    def getBidiLane(self, laneID):
306        """getBidiLane(string) -> string
307
308        Returns the id of the bidi lane or ""
309        """
310        return self._getUniversal(tc.VAR_BIDI, laneID)
311
312    def setAllowed(self, laneID, allowedClasses):
313        """setAllowed(string, list) -> None
314
315        Sets a list of allowed vehicle classes. Setting an empty list means all vehicles are allowed.
316        """
317        if isinstance(allowedClasses, str):
318            allowedClasses = [allowedClasses]
319        self._setCmd(tc.LANE_ALLOWED, laneID, "l", allowedClasses)
320
321    def setDisallowed(self, laneID, disallowedClasses):
322        """setDisallowed(string, list) -> None
323
324        Sets a list of disallowed vehicle classes.
325        """
326        if isinstance(disallowedClasses, str):
327            disallowedClasses = [disallowedClasses]
328        self._setCmd(tc.LANE_DISALLOWED, laneID, "l", disallowedClasses)
329
330    def setChangePermissions(self, laneID, allowedClasses, direction):
331        """setChangePermissions(string, list, int) -> None
332
333        Sets a list of vehicle classes allowed to change to the neighbor lane indicated by direction
334        (left=1, right=-1).
335        """
336        self._setCmd(tc.LANE_CHANGES, laneID, "tlb", 2, allowedClasses, direction)
337
338    def setMaxSpeed(self, laneID, speed):
339        """setMaxSpeed(string, double) -> None
340
341        Sets a new maximum allowed speed on the lane in m/s.
342        """
343        self._setCmd(tc.VAR_MAXSPEED, laneID, "d", speed)
344
345    def setFriction(self, laneID, friction):
346        """setFriction(string, double) -> None
347
348        Sets the friction of the lane.
349        """
350        self._setCmd(tc.VAR_FRICTION, laneID, "d", friction)
351
352    def setLength(self, laneID, length):
353        """setLength(string, double) -> None
354
355        Sets the length of the lane in m.
356        """
357        self._setCmd(tc.VAR_LENGTH, laneID, "d", length)