traci.domain

  1# -*- coding: utf-8 -*-
  2# Eclipse SUMO, Simulation of Urban MObility; see https://eclipse.dev/sumo
  3# Copyright (C) 2008-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    domain.py
 15# @author  Michael Behrisch
 16# @author  Lena Kalleske
 17# @author  Mario Krumnow
 18# @author  Daniel Krajzewicz
 19# @author  Jakob Erdmann
 20# @date    2008-10-09
 21
 22from __future__ import print_function
 23from __future__ import absolute_import
 24import copy
 25import warnings
 26
 27from . import constants as tc
 28from .exceptions import FatalTraCIError, alias_param
 29
 30DOMAINS = []
 31DOMAIN_BY_ID = {}
 32
 33
 34def _readParameterWithKey(result):
 35    assert result.read("!i")[0] == 2  # compound size
 36    key = result.readTypedString()
 37    val = result.readTypedString()
 38    return key, val
 39
 40
 41def _parse(valueFunc, varID, data):
 42    varType = data.read("!B")[0]
 43    if varID in valueFunc:
 44        return valueFunc[varID](data)
 45    if varType in (tc.POSITION_2D, tc.POSITION_LON_LAT):
 46        return data.read("!dd")
 47    if varType in (tc.POSITION_3D, tc.POSITION_LON_LAT_ALT):
 48        return data.read("!ddd")
 49    if varType == tc.TYPE_POLYGON:
 50        return data.readShape()
 51    if varType == tc.TYPE_UBYTE:
 52        return data.read("!B")[0]
 53    if varType == tc.TYPE_BYTE:
 54        return data.read("!b")[0]
 55    if varType == tc.TYPE_INTEGER:
 56        return data.readInt()
 57    if varType == tc.TYPE_DOUBLE:
 58        return data.readDouble()
 59    if varType == tc.TYPE_STRING:
 60        return data.readString()
 61    if varType == tc.TYPE_STRINGLIST:
 62        return data.readStringList()
 63    if varType == tc.TYPE_DOUBLELIST:
 64        n = data.read("!i")[0]
 65        return tuple([data.readDouble() for i in range(n)])
 66    if varType == tc.TYPE_COLOR:
 67        return data.read("!BBBB")
 68    raise FatalTraCIError("Unknown variable %02x or invalid type %02x." % (varID, varType))
 69
 70
 71class SubscriptionResults:
 72
 73    def __init__(self, valueFunc):
 74        self._results = {}
 75        self._contextResults = {}
 76        self._valueFunc = valueFunc
 77
 78    def reset(self):
 79        self._results.clear()
 80        self._contextResults.clear()
 81
 82    def add(self, refID, varID, data):
 83        if refID not in self._results:
 84            self._results[refID] = {}
 85        self._results[refID][varID] = self.parse(varID, data)
 86
 87    def get(self, refID=None):
 88        if refID is None:
 89            return self._results
 90        return self._results.get(refID, {})
 91
 92    def addContext(self, refID, objectID=None, varID=None, value=None):
 93        if refID not in self._contextResults:
 94            self._contextResults[refID] = {}
 95        if objectID is not None and objectID not in self._contextResults[refID]:
 96            self._contextResults[refID][objectID] = {}
 97        if varID is not None and value is not None:
 98            self._contextResults[refID][objectID][varID] = value
 99
100    def parse(self, varID=None, data=None):
101        return _parse(self._valueFunc, varID, data)
102
103    def getContext(self, refID=None):
104        if refID is None:
105            return self._contextResults
106        return self._contextResults.get(refID, {})
107
108    def __repr__(self):
109        return "<%s, %s>" % (self._results, self._contextResults)
110
111
112class Domain:
113
114    def __init__(self, name, cmdGetID, cmdSetID,
115                 subscribeID, subscribeResponseID,
116                 contextID, contextResponseID,
117                 retValFunc=None, deprecatedFor=None,
118                 subscriptionDefault=(tc.TRACI_ID_LIST,)):
119        self._name = name
120        self._cmdGetID = cmdGetID
121        self._cmdSetID = cmdSetID
122        self._subscribeID = subscribeID
123        self._subscribeResponseID = subscribeResponseID
124        self._contextID = contextID
125        self._contextResponseID = contextResponseID
126        self._retValFunc = {tc.VAR_PARAMETER_WITH_KEY: _readParameterWithKey}
127        if retValFunc is not None:
128            self._retValFunc.update(retValFunc)
129        self._deprecatedFor = deprecatedFor
130        self._subscriptionDefault = subscriptionDefault
131        self._connection = None
132        DOMAINS.append(self)
133        DOMAIN_BY_ID[cmdGetID] = self
134        # alias
135        self.DOMAIN_ID = cmdGetID
136
137    def _register(self, connection, mapping):
138        dom = copy.copy(self)
139        dom._connection = connection
140        subscriptionResults = SubscriptionResults(self._retValFunc)
141        mapping[self._subscribeResponseID] = subscriptionResults
142        mapping[self._contextResponseID] = subscriptionResults
143        mapping[self._cmdGetID] = subscriptionResults
144        setattr(connection, self._name, dom)
145
146    def _setConnection(self, connection):
147        self._connection = connection
148
149    def _getUniversal(self, varID, objectID="", format="", *values):
150        if self._deprecatedFor:
151            warnings.warn("The domain %s is deprecated, use %s instead." % (self._name, self._deprecatedFor))
152        return _parse(self._retValFunc, varID, self._getCmd(varID, objectID, format, *values))
153
154    def _getCmd(self, varID, objID, format="", *values):
155        if self._connection is None:
156            raise FatalTraCIError("Not connected.")
157        r = self._connection._sendCmd(self._cmdGetID, varID, objID, format, *values)
158        r.readLength()
159        response, retVarID = r.read("!BB")
160        objectID = r.readString()
161        if response - self._cmdGetID != 16 or retVarID != varID or objectID != objID:
162            raise FatalTraCIError("Received answer %s,%s,%s for command %s,%s,%s."
163                                  % (response, retVarID, objectID, self._cmdGetID, varID, objID))
164        return r
165
166    def _setCmd(self, varID, objectID, format="", *values):
167        if self._connection is None:
168            raise FatalTraCIError("Not connected.")
169        self._connection._sendCmd(self._cmdSetID, varID, objectID, format, *values)
170
171    def getIDList(self):
172        """getIDList() -> tuple(string)
173
174        Returns a tuple of all objects in the network.
175        """
176        return self._getUniversal(tc.TRACI_ID_LIST)
177
178    def getIDCount(self):
179        """getIDCount() -> integer
180
181        Returns the number of currently loaded objects.
182        """
183        return self._getUniversal(tc.ID_COUNT)
184
185    def domainID(self):
186        return self._cmdGetID
187
188    def subscribe(self, objectID, varIDs=None, begin=tc.INVALID_DOUBLE_VALUE, end=tc.INVALID_DOUBLE_VALUE,
189                  parameters=None):
190        """subscribe(string, list(integer), double, double, map) -> None
191        The parameters map argument is needed when subscribing to functions that require additional parameters:
192        The map keys must be a subset of the varIDs
193        The map values may take any of the following forms:
194        - a single integer, float or string value (for methods that require only one additional argument of such type)
195        - a tuple where
196          - the first element is a string consisting of format specifiers (see method _pack in module connection)
197          - the remaining elements are the arguments (one per character in the format specifier string)
198        Subscribe to one or more object values for the given interval.
199        """
200        if varIDs is None:
201            varIDs = self._subscriptionDefault
202        self._connection._subscribe(self._subscribeID, begin, end, objectID, varIDs, parameters)
203
204    def unsubscribe(self, objectID):
205        """unsubscribe(string) -> None
206
207        Unsubscribe from receiving object values.
208        """
209        self._connection._subscribe(self._subscribeID, tc.INVALID_DOUBLE_VALUE, tc.INVALID_DOUBLE_VALUE,
210                                    objectID, [], None)
211
212    def getSubscriptionResults(self, objectID):
213        """getSubscriptionResults(string) -> dict(integer: <value_type>)
214
215        Returns the subscription results for the last time step and the given object.
216        If the object id is unknown or the subscription did for any reason return no data,
217        'None' is returned.
218        It is not possible to retrieve older subscription results than the ones
219        from the last time step.
220        """
221        return self._connection._getSubscriptionResults(self._subscribeResponseID).get(objectID)
222
223    def getAllSubscriptionResults(self):
224        """getAllSubscriptionResults() -> dict(string: dict(integer: <value_type>))
225
226        Returns the subscription results for the last time step and all objects of the domain.
227        It is not possible to retrieve older subscription results than the ones
228        from the last time step.
229        """
230        return self._connection._getSubscriptionResults(self._subscribeResponseID).get(None)
231
232    def subscribeContext(self, objectID, domain, dist, varIDs=None,
233                         begin=tc.INVALID_DOUBLE_VALUE, end=tc.INVALID_DOUBLE_VALUE, parameters=None):
234        """subscribeContext(string, int, double, list(integer), double, double) -> None
235
236        Subscribe to objects of the given domain (specified as domain=traci.constants.CMD_GET_<DOMAIN>_VARIABLE),
237        which are closer than dist to the object specified by objectID.
238        """
239        if varIDs is None:
240            varIDs = DOMAIN_BY_ID.get(domain, self)._subscriptionDefault
241        self._connection._subscribeContext(self._contextID, begin, end, objectID, domain, dist, varIDs, parameters)
242
243    def unsubscribeContext(self, objectID, domain, dist):
244        self._connection._subscribeContext(self._contextID, tc.INVALID_DOUBLE_VALUE, tc.INVALID_DOUBLE_VALUE,
245                                           objectID, domain, dist, [], None)
246
247    def getContextSubscriptionResults(self, objectID):
248        return self._connection._getSubscriptionResults(self._contextResponseID).getContext(objectID)
249
250    def getAllContextSubscriptionResults(self):
251        return self._connection._getSubscriptionResults(self._contextResponseID).getContext(None)
252
253    @alias_param(("objectID", "key"), ("objID", "param"))
254    def getParameter(self, objectID, key):
255        """getParameter(string, string) -> string
256
257        Returns the value of the given parameter for the given objectID
258        """
259        return self._getUniversal(tc.VAR_PARAMETER, objectID, "s", key)
260
261    @alias_param(("objectID", "key"), ("objID", "param"))
262    def getParameterWithKey(self, objectID, key):
263        """getParameterWithKey(string, string) -> (string, string)
264
265        Returns the (key, value) tuple of the given parameter for the given objectID
266        """
267        return self._getUniversal(tc.VAR_PARAMETER_WITH_KEY, objectID, "s", key)
268
269    @alias_param("objectID", "objID")
270    def subscribeParameterWithKey(self, objectID, key, begin=tc.INVALID_DOUBLE_VALUE, end=tc.INVALID_DOUBLE_VALUE):
271        """subscribeParameterWithKey(string, string) -> None
272
273        Subscribe for a generic parameter with the given key.
274        """
275        self.subscribe(objectID, (tc.VAR_PARAMETER_WITH_KEY,), begin, end, {tc.VAR_PARAMETER_WITH_KEY: ("s", key)})
276
277    @alias_param(("objectID", "key"), ("objID", "param"))
278    def setParameter(self, objectID, key, value):
279        """setParameter(string, string, string) -> None
280
281        Sets the value of the given parameter to value for the given objectID
282        """
283        self._setCmd(tc.VAR_PARAMETER, objectID, "tss", 2, key, value)