Merge branch 'develop' into documentation

examples/Demo_Drive.py

@@ -21,8 +21,9 @@
 parser.add_argument("--location", type=str, default="iai:ubc_roundabout")
 args = parser.parse_args()
 
-iai.add_apikey("")
+iai.add_apikey(args.api_key)
 
+# response = iai.available_locations("carla", "roundabout")
 response = iai.location_info(location=args.location)
 
 file_name = args.location.replace(":", "_")
@@ -35,29 +36,24 @@
     rendered_map = np.array(response.birdview_image, dtype=np.uint8)
     image = cv2.imdecode(rendered_map, cv2.IMREAD_COLOR)
     cv2.imwrite(file_path, image)
-response = iai.initialize(
+simulation = iai.Simulation(
     location=args.location,
     agent_count=10,
+    monitor_infractions=True,
+    render_birdview=True,
 )
-agent_attributes = response.agent_attributes
 frames = []
 pbar = tqdm(range(50))
 for i in pbar:
-    response = iai.drive(
-        agent_attributes=agent_attributes,
-        agent_states=response.agent_states,
-        recurrent_states=response.recurrent_states,
-        get_birdviews=True,
-        location=args.location,
-        get_infractions=True,
-    )
+    simulation.step(current_ego_agent_states=[])
+    collision, offroad, wrong_way = simulation.infractions
     pbar.set_description(
-        f"Collision rate: {100*np.array([inf.collisions for inf in response.infractions]).mean():.2f}% | "
-        + f"Off-road rate: {100*np.array([inf.offroad for inf in response.infractions]).mean():.2f}% | "
-        + f"Wrong-way rate: {100*np.array([inf.wrong_way for inf in response.infractions]).mean():.2f}%"
+        f"Collision rate: {100*np.array(collision).mean():.2f}% | "
+        + f"Off-road rate: {100*np.array(offroad).mean():.2f}% | "
+        + f"Wrong-way rate: {100*np.array(wrong_way).mean():.2f}%"
     )
 
-    birdview = np.array(response.bird_view, dtype=np.uint8)
+    birdview = np.array(simulation.birdview, dtype=np.uint8)
     image = cv2.imdecode(birdview, cv2.IMREAD_COLOR)
     frames.append(image)
     im = PImage.fromarray(image)

examples/Demo_Drive_REST.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env ipython
+import os
+import sys
+from PIL import Image as PImage
+import imageio
+import numpy as np
+import cv2
+from tqdm import tqdm
+import argparse
+from dotenv import load_dotenv
+
+load_dotenv()
+if os.environ.get("DEV", False):
+    sys.path.append("../")
+import invertedai as iai
+
+# logger.setLevel(10)
+
+parser = argparse.ArgumentParser(description="Simulation Parameters.")
+parser.add_argument("--api_key", type=str, default="")
+parser.add_argument("--location", type=str, default="iai:ubc_roundabout")
+args = parser.parse_args()
+
+iai.add_apikey("")
+
+response = iai.location_info(location=args.location)
+
+file_name = args.location.replace(":", "_")
+if response.osm_map is not None:
+    file_path = f"{file_name}.osm"
+    with open(file_path, "w") as f:
+        f.write(response.osm_map[0])
+if response.birdview_image is not None:
+    file_path = f"{file_name}.jpg"
+    rendered_map = np.array(response.birdview_image, dtype=np.uint8)
+    image = cv2.imdecode(rendered_map, cv2.IMREAD_COLOR)
+    cv2.imwrite(file_path, image)
+response = iai.initialize(
+    location=args.location,
+    agent_count=10,
+)
+agent_attributes = response.agent_attributes
+frames = []
+pbar = tqdm(range(50))
+for i in pbar:
+    response = iai.drive(
+        agent_attributes=agent_attributes,
+        agent_states=response.agent_states,
+        recurrent_states=response.recurrent_states,
+        get_birdviews=True,
+        location=args.location,
+        get_infractions=True,
+    )
+    pbar.set_description(
+        f"Collision rate: {100*np.array([inf.collisions for inf in response.infractions]).mean():.2f}% | "
+        + f"Off-road rate: {100*np.array([inf.offroad for inf in response.infractions]).mean():.2f}% | "
+        + f"Wrong-way rate: {100*np.array([inf.wrong_way for inf in response.infractions]).mean():.2f}%"
+    )
+
+    birdview = np.array(response.bird_view, dtype=np.uint8)
+    image = cv2.imdecode(birdview, cv2.IMREAD_COLOR)
+    frames.append(image)
+    im = PImage.fromarray(image)
+imageio.mimsave("iai-drive.gif", np.array(frames), format="GIF-PIL")

examples/Drive-Demo.ipynb

@@ -119,24 +119,15 @@
     "rendered_map = cv2.imdecode(np.array(rendered_map, dtype=np.uint8), cv2.IMREAD_COLOR)\n",
     "renderer.add_frame(rendered_map)\n",
     "\n",
-    "response = iai.initialize(\n",
+    "simulation = iai.Simulation(\n",
     "    location=location,\n",
     "    agent_count=10,\n",
+    "    render_birdview=True,\n",
     ")\n",
-    "agent_sizes = response.agent_sizes\n",
     "frames = []\n",
     "for t in range(simulation_length):\n",
-    "    response = iai.drive(\n",
-    "        agent_sizes=agent_sizes,\n",
-    "        agent_states=response.agent_states,\n",
-    "        recurrent_states=response.recurrent_states,\n",
-    "        get_birdviews=True,\n",
-    "        location=location,\n",
-    "        steps=1,\n",
-    "        exclude_ego_agent=1,\n",
-    "        traffic_states_id=response.traffic_states_id,\n",
-    "    )\n",
-    "    birdview = cv2.imdecode(np.array(response.bird_view, dtype=np.uint8), cv2.IMREAD_COLOR)\n",
+    "    simulation.step(current_ego_agent_states=[])\n",
+    "    birdview = cv2.imdecode(np.array(simulation.birdview, dtype=np.uint8), cv2.IMREAD_COLOR)\n",
     "    renderer.add_frame(birdview)"
    ]
   },

invertedai/__init__.py

@@ -5,6 +5,7 @@
     initialize,
     location_info,
 )
+from invertedai.simulation import Simulation
 from invertedai.utils import Jupyter_Render, IAILogger, Session
 
 load_dotenv()
@@ -26,6 +27,7 @@
     "location_info": ("get", "/location_info"),
 }
 __all__ = [
+    "Simulation",
     "drive",
     "initialize",
     "location_info",

invertedai/simulation.py

@@ -0,0 +1,115 @@
+from typing import Optional, List
+
+from invertedai.api_resources import initialize, drive
+
+
+class Simulation:
+    """
+    Stateful wrapper around Inverted AI API to simplify co-simulation.
+    Typically, it's sufficient to call `ego_agent_states` and `step`.
+    """
+
+    def __init__(
+        self,
+        # all arguments to initialize are also given to this constructor
+        location: str = "CARLA:Town03:Roundabout",
+        agent_count: int = 1,
+        # some further configuration options
+        monitor_infractions: bool = False,
+        render_birdview: bool = False,
+        ego_agent_mask: Optional[List[bool]] = None,
+    ):
+        self._location = location
+        response = initialize(location=location, agent_count=agent_count)
+        self._agent_count = len(
+            response.agent_attributes
+        )  # initialize may produce different agent count
+        self._agent_attributes = response.agent_attributes
+        self._agent_states = response.agent_states
+        self._recurrent_states = response.recurrent_states
+        self._monitor_infractions = monitor_infractions
+        self._infractions = None
+        self._render_birdview = render_birdview
+        self._birdview = None
+        if ego_agent_mask is None:
+            self._ego_agent_mask = [False] * self.agent_count
+        else:
+            self._ego_agent_mask = ego_agent_mask
+        self._time_step = 0
+
+    @property
+    def location(self):
+        return self._location
+
+    @property
+    def agent_count(self):
+        return self._agent_count
+
+    @property
+    def agent_states(self):
+        return self._agent_states
+
+    @property
+    def ego_agent_mask(self):
+        return self._ego_agent_mask
+
+    @ego_agent_mask.setter
+    def ego_agent_mask(self, value):
+        self.ego_agent_mask = value
+
+    @property
+    def infractions(self):
+        return self._infractions
+
+    @property
+    def birdview(self):
+        return self._birdview
+
+    def npc_states(self):
+        """
+        Returns the predicted states of NPCs (non-ego agents) only in order.
+        """
+        npc_states = []
+        for (i, s) in self._agent_states:
+            if not self._ego_agent_mask[i]:
+                npc_states.append(s)
+        return npc_states
+
+    def step(self, current_ego_agent_states):
+        """
+        Call the API to advance the simulation by one time step.
+        :param current_ego_agent_states:  States of ego agents before the step.
+        :return: None - call `npc_states` to retrieve predictions.
+        """
+        self._update_ego_states(current_ego_agent_states)
+        response = drive(
+            location=self.location,
+            agent_attributes=self._agent_attributes,
+            agent_states=self.agent_states,
+            recurrent_states=self._recurrent_states,
+            get_infractions=self._monitor_infractions,
+            get_birdviews=self._render_birdview,
+        )
+        self._agent_states = response.agent_states
+        self._recurrent_states = response.recurrent_states
+        if self._monitor_infractions and (response.infractions is not None):
+            # TODO: package infractions in dataclass
+            self._infractions = (
+                [inf.collisions for inf in response.infractions],
+                [inf.offroad for inf in response.infractions],
+                [inf.wrong_way for inf in response.infractions],
+            )
+        if self._render_birdview:
+            self._birdview = response.bird_view
+        self._time_step += 1
+
+    def _update_ego_states(self, ego_agent_states):
+        new_states = []
+        ego_idx = 0
+        for (i, s) in enumerate(self.agent_states):
+            if self.ego_agent_mask[i]:
+                new_states.append(ego_agent_states[ego_idx])
+                ego_idx += 1
+            else:
+                new_states.append(self.agent_states[i])
+        self._agent_states = new_states