traci.connection

  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    connection.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 socket
 25import struct
 26import sys
 27import warnings
 28import threading
 29
 30from . import constants as tc
 31from .exceptions import TraCIException, FatalTraCIError
 32from .domain import DOMAINS
 33from .storage import Storage
 34from .step import StepManager
 35
 36_DEBUG = False
 37_RESULTS = {0x00: "OK", 0x01: "Not implemented", 0xFF: "Error"}
 38
 39_connections = {}
 40_connectHook = None
 41
 42
 43def check():
 44    if "" not in _connections:
 45        raise FatalTraCIError("Not connected.")
 46    return _connections[""]
 47
 48
 49def has(label):
 50    return label in _connections
 51
 52
 53def get(label="default"):
 54    if label not in _connections:
 55        raise TraCIException("Connection '%s' is not known." % label)
 56    return _connections[label]
 57
 58
 59def switch(label):
 60    con = get(label)
 61    _connections[""] = con
 62    for domain in DOMAINS:
 63        domain._setConnection(con)
 64
 65
 66class Connection(StepManager):
 67
 68    """Contains the socket, the composed message string
 69    together with a list of TraCI commands which are inside.
 70    """
 71
 72    def __init__(self, host, port, process, traceFile, traceGetters, label=None):
 73        StepManager.__init__(self)
 74        if label in _connections:
 75            raise TraCIException("Connection '%s' is already active." % label)
 76        if sys.platform.startswith('java'):
 77            # working around jython 2.7.0 bug #2273
 78            self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)
 79        else:
 80            self._socket = socket.socket()
 81        self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 82        try:
 83            self._socket.connect((host, port))
 84        except socket.error:
 85            self._socket.close()
 86            raise
 87        self._process = process
 88        self._string = bytes()
 89        self._queue = []
 90        self._subscriptionMapping = {}
 91        self._lock = threading.Lock()
 92        if traceFile is not None:
 93            self.startTracing(traceFile, traceGetters, DOMAINS)
 94        for domain in DOMAINS:
 95            domain._register(self, self._subscriptionMapping)
 96        self._label = label
 97        if _connectHook is not None:
 98            _connectHook(self)
 99        if label is not None:
100            _connections[label] = self
101
102    def getLabel(self):
103        return self._label
104
105    def _recvExact(self):
106        try:
107            result = bytes()
108            while len(result) < 4:
109                t = self._socket.recv(4 - len(result))
110                if not t:
111                    return None
112                result += t
113            length = struct.unpack("!i", result)[0] - 4
114            result = bytes()
115            while len(result) < length:
116                t = self._socket.recv(length - len(result))
117                if not t:
118                    return None
119                result += t
120            return Storage(result)
121        except socket.error:
122            return None
123
124    def _sendExact(self):
125        if self._socket is None:
126            raise FatalTraCIError("Connection already closed.")
127        length = struct.pack("!i", len(self._string) + 4)
128        if _DEBUG:
129            print("sending", Storage(length + self._string).getDebugString())
130        self._socket.send(length + self._string)
131        result = self._recvExact()
132        if _DEBUG:
133            print("receiving", result.getDebugString())
134        if not result:
135            self._socket.close()
136            self._socket = None
137            raise FatalTraCIError("Connection closed by SUMO.")
138        for command in self._queue:
139            prefix = result.read("!BBB")
140            err = result.readString()
141            if prefix[2] or err:
142                self._string = bytes()
143                self._queue = []
144                raise TraCIException(err, prefix[1], _RESULTS[prefix[2]])
145            elif prefix[1] != command:
146                raise FatalTraCIError("Received answer %s for command %s." % (prefix[1], command))
147            elif prefix[1] == tc.CMD_STOP:
148                length = result.read("!B")[0] - 1
149                result.read("!%sx" % length)
150        self._string = bytes()
151        self._queue = []
152        return result
153
154    def _pack(self, format, *values):
155        packed = bytes()
156        for f, v in zip(format, values):
157            if f == "i":
158                packed += struct.pack("!Bi", tc.TYPE_INTEGER, int(v))
159            elif f == "I":  # raw int for setOrder
160                packed += struct.pack("!i", int(v))
161            elif f == "d":
162                packed += struct.pack("!Bd", tc.TYPE_DOUBLE, float(v))
163            elif f == "D":  # raw double for some base commands like simstep
164                packed += struct.pack("!d", float(v))
165            elif f == "b":
166                packed += struct.pack("!Bb", tc.TYPE_BYTE, int(v))
167            elif f == "B":
168                packed += struct.pack("!BB", tc.TYPE_UBYTE, int(v))
169            elif f == "u":  # raw unsigned byte needed for distance command and subscribe
170                packed += struct.pack("!B", int(v))
171            elif f == "s":
172                if sys.version_info[0] > 2 or not isinstance(v, str):
173                    v = str(v).encode("utf8")
174                packed += struct.pack("!Bi", tc.TYPE_STRING, len(v)) + v
175            elif f == "p":  # polygon
176                if len(v) <= 255:
177                    packed += struct.pack("!BB", tc.TYPE_POLYGON, len(v))
178                else:
179                    packed += struct.pack("!BBi", tc.TYPE_POLYGON, 0, len(v))
180                for p in v:
181                    packed += struct.pack("!dd", *p)
182            elif f == "t":  # tuple aka compound
183                packed += struct.pack("!Bi", tc.TYPE_COMPOUND, v)
184            elif f == "c":  # color
185                packed += struct.pack("!BBBBB", tc.TYPE_COLOR, int(v[0]), int(v[1]), int(v[2]),
186                                      int(v[3]) if len(v) > 3 else 255)
187            elif f == "l":  # string list
188                packed += struct.pack("!Bi", tc.TYPE_STRINGLIST, len(v))
189                for s in v:
190                    s = str(s).encode("utf8")
191                    packed += struct.pack("!i", len(s)) + s
192            elif f == "f":  # float list
193                packed += struct.pack("!Bi", tc.TYPE_DOUBLELIST, len(v))
194                for x in v:
195                    packed += struct.pack("!d", x)
196            elif f == "o":
197                packed += struct.pack("!Bdd", tc.POSITION_2D, *v)
198            elif f == "O":
199                packed += struct.pack("!Bddd", tc.POSITION_3D, *v)
200            elif f == "g":
201                packed += struct.pack("!Bdd", tc.POSITION_LON_LAT, *v)
202            elif f == "G":
203                packed += struct.pack("!Bddd", tc.POSITION_LON_LAT_ALT, *v)
204            elif f == "r":
205                s = str(v[0]).encode("utf8")
206                packed += struct.pack("!Bi", tc.POSITION_ROADMAP, len(s)) + s
207                packed += struct.pack("!dB", v[1], v[2])
208        return packed
209
210    def _sendCmd(self, cmdID, varID, objID, format="", *values):
211        with self._lock:
212            self._queue.append(cmdID)
213            packed = self._pack(format, *values)
214            objID = str(objID).encode("utf8")
215            length = len(packed) + 1 + 1  # length and command
216            if varID is not None:
217                if isinstance(varID, tuple):  # begin and end of a subscription
218                    length += 8 + 8 + 4 + len(objID)
219                else:
220                    length += 1 + 4 + len(objID)
221            if length <= 255:
222                self._string += struct.pack("!BB", length, cmdID)
223            else:
224                self._string += struct.pack("!BiB", 0, length + 4, cmdID)
225            if varID is not None:
226                if isinstance(varID, tuple):
227                    self._string += struct.pack("!dd", *varID)
228                else:
229                    self._string += struct.pack("!B", varID)
230                self._string += struct.pack("!i", len(objID)) + objID
231            self._string += packed
232            return self._sendExact()
233
234    def _readSubscription(self, result):
235        if _DEBUG:
236            print("reading subscription", result.getDebugString())
237        result.readLength()
238        response = result.read("!B")[0]
239        isVariableSubscription = ((response >= tc.RESPONSE_SUBSCRIBE_INDUCTIONLOOP_VARIABLE and
240                                   response <= tc.RESPONSE_SUBSCRIBE_BUSSTOP_VARIABLE) or
241                                  (response >= tc.RESPONSE_SUBSCRIBE_PARKINGAREA_VARIABLE and
242                                   response <= tc.RESPONSE_SUBSCRIBE_OVERHEADWIRE_VARIABLE))
243        objectID = result.readString()
244        contextDomain = 0 if isVariableSubscription else result.read("!B")[0]
245        numVars = result.read("!B")[0]
246        if isVariableSubscription:
247            while numVars > 0:
248                varID, status = result.read("!BB")
249                if status:
250                    print("Error!", result.readTypedString())
251                elif response in self._subscriptionMapping:
252                    self._subscriptionMapping[response].add(objectID, varID, result)
253                else:
254                    raise FatalTraCIError(
255                        "Cannot handle subscription response %02x for %s." % (response, objectID))
256                numVars -= 1
257        else:
258            objectNo = result.read("!i")[0]
259            subsMap = self._subscriptionMapping[response]
260            subsMap.addContext(objectID)
261            for _ in range(objectNo):
262                oid = result.readString()
263                if numVars == 0:
264                    subsMap.addContext(objectID, oid)
265                for __ in range(numVars):
266                    varID, status = result.read("!BB")
267                    if status:
268                        print("Error!", result.readTypedString())
269                    elif contextDomain in self._subscriptionMapping:
270                        subsMap.addContext(objectID, oid, varID,
271                                           self._subscriptionMapping[contextDomain].parse(varID, result))
272                    else:
273                        raise FatalTraCIError(
274                            "Cannot handle subscription response %02x for %s." % (response, objectID))
275        return objectID, response
276
277    def _subscribe(self, cmdID, begin, end, objID, varIDs, parameters):
278        format = "u"
279        args = [len(varIDs)]
280        for v in varIDs:
281            format += "u"
282            args.append(v)
283            if parameters is not None and v in parameters:
284                if isinstance(parameters[v], tuple):
285                    format += parameters[v][0]
286                    for a in parameters[v][1:]:
287                        args.append(a)
288                elif isinstance(parameters[v], int):
289                    format += "i"
290                    args.append(parameters[v])
291                elif isinstance(parameters[v], float):
292                    format += "d"
293                    args.append(parameters[v])
294                else:
295                    format += "s"
296                    args.append(parameters[v])
297        result = self._sendCmd(cmdID, (begin, end), objID, format, *args)
298        if varIDs:
299            objectID, response = self._readSubscription(result)
300            if response - cmdID != 16 or objectID != objID:
301                raise FatalTraCIError("Received answer %02x,%s for subscription command %02x,%s." % (
302                    response, objectID, cmdID, objID))
303
304    def _getSubscriptionResults(self, cmdID):
305        return self._subscriptionMapping[cmdID]
306
307    def _subscribeContext(self, cmdID, begin, end, objID, domain, dist, varIDs, parameters=None):
308        result = self._sendCmd(cmdID, (begin, end), objID, "uDu" + (len(varIDs) * "u"),
309                               domain, dist, len(varIDs), *varIDs)
310        if varIDs:
311            objectID, response = self._readSubscription(result)
312            if response - cmdID != 16 or objectID != objID:
313                raise FatalTraCIError("Received answer %02x,%s for context subscription command %02x,%s." % (
314                    response, objectID, cmdID, objID))
315
316    def _addSubscriptionFilter(self, filterType, params=None):
317        if filterType in (tc.FILTER_TYPE_NONE, tc.FILTER_TYPE_NOOPPOSITE,
318                          tc.FILTER_TYPE_LEAD_FOLLOW):
319            # filter without parameter
320            assert params is None
321            self._sendCmd(tc.CMD_ADD_SUBSCRIPTION_FILTER, None, None, "u", filterType)
322        elif filterType in (tc.FILTER_TYPE_DOWNSTREAM_DIST, tc.FILTER_TYPE_UPSTREAM_DIST,
323                            tc.FILTER_TYPE_TURN, tc.FILTER_TYPE_FIELD_OF_VISION,
324                            tc.FILTER_TYPE_LATERAL_DIST):
325            # filter with float parameter
326            self._sendCmd(tc.CMD_ADD_SUBSCRIPTION_FILTER, None, None, "ud", filterType, params)
327        elif filterType in (tc.FILTER_TYPE_VCLASS, tc.FILTER_TYPE_VTYPE):
328            # filter with list(string) parameter
329            self._sendCmd(tc.CMD_ADD_SUBSCRIPTION_FILTER, None, None, "ul", filterType, params)
330        elif filterType == tc.FILTER_TYPE_LANES:
331            # filter with list(byte) parameter
332            # check uniqueness of given lanes in list
333            lanes = set()
334            for i in params:
335                lane = int(i)
336                if lane < 0:
337                    lane += 256
338                lanes.add(lane)
339            if len(lanes) < len(list(params)):
340                warnings.warn("Ignoring duplicate lane specification for subscription filter.")
341            self._sendCmd(tc.CMD_ADD_SUBSCRIPTION_FILTER, None, None,
342                          (len(lanes) + 2) * "u", filterType, len(lanes), *lanes)
343
344    def hasGUI(self):
345        try:
346            self.gui.getIDList()
347            return True
348        except TraCIException:
349            return False
350
351    def load(self, args):
352        """
353        Load a simulation from the given arguments.
354        """
355        if self._traceFile:
356            self._traceFile.write("traci.load(%s)\n" % repr(args))
357        self._sendCmd(tc.CMD_LOAD, None, None, "l", args)
358
359    def simulationStep(self, step=0.):
360        """
361        Make a simulation step and simulate up to the given second in sim time.
362        If the given value is 0 or absent, exactly one step is performed.
363        Values smaller than or equal to the current sim time result in no action.
364        """
365        if self._traceFile:
366            args = "" if step == 0 else str(step)
367            self._traceFile.write("traci.simulationStep(%s)\n" % args)
368        if type(step) is int and step >= 1000:
369            warnings.warn("API change now handles step as floating point seconds", stacklevel=2)
370        result = self._sendCmd(tc.CMD_SIMSTEP, None, None, "D", step)
371        for subscriptionResults in self._subscriptionMapping.values():
372            subscriptionResults.reset()
373        numSubs = result.readInt()
374        responses = []
375        while numSubs > 0:
376            responses.append(self._readSubscription(result))
377            numSubs -= 1
378        self.manageStepListeners(step)
379        return responses
380
381    def getVersion(self):
382        command = tc.CMD_GETVERSION
383        result = self._sendCmd(command, None, None)
384        result.readLength()
385        response = result.read("!B")[0]
386        if response != command:
387            raise FatalTraCIError("Received answer %s for command %s." % (response, command))
388        return result.readInt(), result.readString()
389
390    def setOrder(self, order):
391        self._sendCmd(tc.CMD_SETORDER, None, None, "I", order)
392
393    def close(self, wait=True):
394        StepManager.close(self, True)
395        for listenerID in list(self._stepListeners.keys()):
396            self.removeStepListener(listenerID)
397        if self._socket is not None:
398            self._sendCmd(tc.CMD_CLOSE, None, None)
399            self._socket.close()
400            self._socket = None
401        if wait and self._process is not None:
402            self._process.wait()
403        self.simulation._setConnection(None)
404        if self._label is not None:
405            if _connections.get("") == self:
406                del _connections[""]
407            del _connections[self._label]