Automatic mission routes

docs/conf.py

@@ -124,6 +124,7 @@
 }
 nitpick_ignore_regex = {
     (r"py:.*", r"av2\..*"),
+    (r"py:.*", r"configparser\..*"),
     (r"py:.*", r"google\.protobuf\..*"),
     (r"py:.*", r"grpc\..*"),
     (r"py:.*", r"gym\..*"),
@@ -142,9 +143,9 @@
     (r"py:.*", r"tornado\..*"),
     (r"py:.*", r"traci\..*"),
     (r"py:.*", r"typing(_extensions)?\..*"),
-    (r"py:.*", r"configparser\..*"),
-    (r"py:class", r".*\.?T"),
     (r"py:class", r".*\.?S"),
+    (r"py:class", r".*\.?T"),
+    (r"py:class", r".*typing.Literal\[<SmartsLiteral\..*>\].*"),
 }
 
 # -- Options for spelling ----------------------------------------------------

envision/tests/test_data_formatter.py

@@ -233,26 +233,25 @@ def complex_data():
 
 
 def test_covered_data_format(covered_data):
-    for item in covered_data:
+    for unformatted, formatted in covered_data:
         es = EnvisionDataFormatter(EnvisionDataFormatterArgs(None))
-        vt = item[0]
-        _formatter_map[type(vt)](vt, es)
+        _formatter_map[type(unformatted)](unformatted, es)
 
         data = es.resolve()
 
-        assert data == item[1]
+        assert data == formatted
         assert data == unpack(data)
 
 
 def test_primitive_data_format(primitive_data):
-    for item in primitive_data:
-        vt = item[0]
+    for unformatted, formatted in primitive_data:
+
         es = EnvisionDataFormatter(EnvisionDataFormatterArgs(None))
-        es.add_any(vt)
+        es.add_any(unformatted)
 
         data = es.resolve()
 
-        assert data == item[1]
+        assert data == formatted
         assert data == unpack(data)
 
 
@@ -276,14 +275,13 @@ def test_layer():
 
 
 def test_complex_data(complex_data):
-    for item in complex_data:
-        vt = item[0]
+    for unformatted, formatted in complex_data:
         es = EnvisionDataFormatter(EnvisionDataFormatterArgs(None))
-        es.add_any(vt)
+        es.add_any(unformatted)
 
         data = es.resolve()
 
-        assert data == item[1]
+        assert data == formatted
         assert data == unpack(data)
 
 

smarts/core/plan.py

@@ -24,29 +24,36 @@
 import math
 import random
 import sys
-import warnings
 from dataclasses import dataclass, field
-from typing import List, Optional, Tuple
+from typing import List, Literal, Optional, Tuple, Union
 
 import numpy as np
 
 from smarts.core.coordinates import Dimensions, Heading, Point, Pose, RefLinePoint
 from smarts.core.road_map import RoadMap
 from smarts.core.utils.math import min_angles_difference_signed, vec_to_radians
+from smarts.primatives.constants import MISSING
 from smarts.sstudio.types import EntryTactic, TrapEntryTactic
 
-MISSING = sys.maxsize
-
 
 class PlanningError(Exception):
     """Raised in cases when map related planning fails."""
 
     pass
 
 
+@dataclass(frozen=True)
+class StartBase:
+    """The base type for Start objects."""
+
+    def resolve(self, scenario, vehicle) -> "Start":
+        """Converts an abstract start into a concrete one."""
+        raise NotImplementedError()
+
+
 # XXX: consider using smarts.core.coordinates.Pose for this
 @dataclass(frozen=True)
-class Start:
+class Start(StartBase):
     """A starting state for a route or mission."""
 
     position: np.ndarray
@@ -68,6 +75,11 @@ def from_pose(cls, pose: Pose):
         )
 
 
+@dataclass(frozen=True)
+class InheritedStart(StartBase):
+    """A starting state that inherits from the original vehicle."""
+
+
 @dataclass(frozen=True, unsafe_hash=True)
 class Goal:
     """Describes an expected end state for a route or mission."""
@@ -81,6 +93,20 @@ def is_reached(self, vehicle_state) -> bool:
         return False
 
 
+@dataclass(frozen=True)
+class InheritedGoal(Goal):
+    """Describes a goal that is inherited from the vehicle (or original dataset)."""
+
+    pass
+
+
+@dataclass(frozen=True, unsafe_hash=True)
+class AutomaticGoal(Goal):
+    """A goal that determines an end result from pre-existing vehicle and mission values."""
+
+    pass
+
+
 @dataclass(frozen=True, unsafe_hash=True)
 class EndlessGoal(Goal):
     """A goal that can never be completed."""
@@ -213,7 +239,7 @@ class Mission:
     # An optional list of road IDs between the start and end goal that we want to
     # ensure the mission includes
     route_vias: Tuple[str, ...] = field(default_factory=tuple)
-    start_time: float = MISSING
+    start_time: Union[float, Literal[MISSING]] = MISSING
     entry_tactic: Optional[EntryTactic] = None
     via: Tuple[Via, ...] = ()
     # if specified, will use vehicle_spec to build the vehicle (for histories)

smarts/core/trap_manager.py

@@ -32,6 +32,7 @@
 from smarts.core.utils.file import replace
 from smarts.core.utils.math import clip, squared_dist
 from smarts.core.vehicle import Vehicle
+from smarts.primatives.constants import AUTO
 from smarts.sstudio.types import MapZone, PositionalZone, TrapEntryTactic
 
 
@@ -355,11 +356,11 @@ def _mission2trap(self, road_map, mission: Mission, default_zone_dist: float = 6
         default_entry_speed = entry_tactic.default_entry_speed
         n_lane = None
 
-        if default_entry_speed is None:
+        if default_entry_speed is AUTO:
             n_lane = road_map.nearest_lane(mission.start.point)
             default_entry_speed = n_lane.speed_limit if n_lane is not None else 0
 
-        if zone is None:
+        if zone is AUTO:
             n_lane = n_lane or road_map.nearest_lane(mission.start.point)
             if n_lane is None:
                 zone = PositionalZone(mission.start.position[:2], size=(3, 3))

smarts/sstudio/types/constants.py → smarts/primatives/__init__.py

@@ -19,8 +19,3 @@
 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 # THE SOFTWARE.
-
-import sys
-
-MAX = sys.maxsize
-MISSING = sys.maxsize

smarts/primatives/constants.py

@@ -0,0 +1,43 @@
+# MIT License
+#
+# Copyright (C) 2023. Huawei Technologies Co., Ltd. All rights reserved.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+
+from enum import Enum
+from typing import Final
+
+
+class SmartsLiteral(Enum):
+    """Constants that SMARTS uses. This is intended to constant type the
+    values.
+    """
+
+    AUTO = "auto"
+    INHERIT = ...
+    MAX = 9223372036854775807
+    MISSING = MAX
+    NONE = None
+
+
+AUTO: Final = SmartsLiteral.AUTO
+INHERIT: Final = SmartsLiteral.INHERIT
+MAX: Final = SmartsLiteral.MAX
+MISSING = SmartsLiteral.MISSING
+NONE: Final = SmartsLiteral.NONE

smarts/sstudio/generators.py

@@ -28,7 +28,7 @@
 import random
 import subprocess
 import tempfile
-from typing import Optional
+from typing import Dict, Optional, Union
 
 from yattag import Doc, indent
 
@@ -262,7 +262,9 @@ def _writexml(
 
             # Make sure all routes are "resolved" (e.g. `RandomRoute` are converted to
             # `Route`) so that we can write them all to file.
-            resolved_routes = {}
+            resolved_routes: Dict[
+                Union[types.RandomRoute, types.Route], types.Route
+            ] = {}
             for route in {flow.route for flow in traffic.flows}:
                 resolved_routes[route] = self.resolve_route(route, fill_in_route_gaps)
 
@@ -277,12 +279,15 @@ def _writexml(
                 route = resolved_routes[flow.route]
                 for actor_idx, (actor, weight) in enumerate(flow.actors.items()):
                     vehs_per_hour = flow.rate * (weight / total_weight)
-                    rate_option = {}
+                    options = {}
                     if flow.randomly_spaced:
                         vehs_per_sec = vehs_per_hour * SECONDS_PER_HOUR_INV
-                        rate_option = dict(probability=vehs_per_sec)
+                        options["probability"] = vehs_per_sec
                     else:
-                        rate_option = dict(vehsPerHour=vehs_per_hour)
+                        options["vehsPerHour"] = vehs_per_hour
+
+                    if len(route.via):
+                        options["via"] = " ".join(route.via)
                     doc.stag(
                         "flow",
                         # have to encode the flow.repeat_route within the vehcile id b/c
@@ -303,26 +308,26 @@ def _writexml(
                         arrivalPos=route.end[2],
                         begin=flow.begin,
                         end=flow.end,
-                        **rate_option,
+                        **options,
                     )
             # write trip into xml format
             if traffic.trips:
                 self.write_trip_xml(traffic, doc, fill_in_route_gaps)
 
-        with open(route_path, "w") as f:
+        with open(route_path, "w", encoding="utf-8") as f:
             f.write(
                 indent(
                     doc.getvalue(), indentation="    ", newline="\r\n", indent_text=True
                 )
             )
 
-    def write_trip_xml(self, traffic, doc, fill_in_gaps):
+    def write_trip_xml(self, traffic: types.Traffic, doc: Doc, fill_in_gaps: bool):
         """Writes a trip spec into a route file. Typically this would be the source
         data to SUMO's DUAROUTER.
         """
         # Make sure all routes are "resolved" (e.g. `RandomRoute` are converted to
         # `Route`) so that we can write them all to file.
-        resolved_routes = {}
+        resolved_routes: Dict[Union[types.RandomRoute, types.Route], types.Route] = {}
         for route in {trip.route for trip in traffic.trips}:
             resolved_routes[route] = self.resolve_route(route, fill_in_gaps)
 
@@ -332,8 +337,11 @@ def write_trip_xml(self, traffic, doc, fill_in_gaps):
             # We don't de-dup flows since defining the same flow multiple times should
             # create multiple traffic flows. Since IDs can't be reused, we also unique
             # them here.
+        options: Dict[str, Union[str, int, float]] = {}
         for trip_idx, trip in enumerate(traffic.trips):
             route = resolved_routes[trip.route]
+            if len(route.via):
+                options["via"] = " ".join(route.via)
             actor = trip.actor
             doc.stag(
                 "vehicle",
@@ -346,6 +354,7 @@ def write_trip_xml(self, traffic, doc, fill_in_gaps):
                 departSpeed=actor.depart_speed,
                 arrivalLane=route.end[1],
                 arrivalPos=route.end[2],
+                **options,
             )
 
     def _cache_road_network(self):
@@ -383,7 +392,7 @@ def _map_for_route(self, route) -> RoadMap:
         road_map, _ = map_spec.builder_fn(map_spec)
         return road_map
 
-    def _fill_in_gaps(self, route: types.Route) -> types.Route:
+    def _fill_in_traffic_route_gaps(self, route: types.Route) -> types.Route:
         # TODO:  do this at runtime so each vehicle on the flow can take a different variation of the route ?
         # TODO:  or do it like SUMO and generate a huge *.rou.xml file instead ?
         road_map = self._map_for_route(route)
@@ -409,7 +418,7 @@ def resolve_route(self, route, fill_in_gaps: bool) -> types.Route:
             smarts.sstudio.types.route.Route: A complete route listing all road segments it passes through.
         """
         if not isinstance(route, types.RandomRoute):
-            return self._fill_in_gaps(route) if fill_in_gaps else route
+            return self._fill_in_traffic_route_gaps(route) if fill_in_gaps else route
 
         if not self._random_route_generator:
             road_map = self._map_for_route(route)