Animation with motivation as color

Author: chraibi

anim.py

@@ -22,10 +22,14 @@
 DUMMY_SPEED = -1000
 
 
-def _speed_to_color(speed: float, min_speed: float, max_speed: float) -> str:
-    """Map a speed value to a color using a colormap."""
-    normalized_speed = (speed - min_speed) / (max_speed - min_speed)
-    r, g, b = plt.cm.jet_r(normalized_speed)[:3]  # type: ignore
+def _value_to_color(value: float, min_value: float, max_value: float) -> str:
+    """Map a scalar value to a color using a colormap."""
+    if max_value <= min_value:
+        normalized_value = 0.5
+    else:
+        normalized_value = (value - min_value) / (max_value - min_value)
+    normalized_value = min(max(normalized_value, 0.0), 1.0)
+    r, g, b = plt.cm.jet_r(normalized_value)[:3]  # type: ignore
     return f"rgba({r*255:.0f}, {g*255:.0f}, {b*255:.0f}, 0.5)"
 
 
@@ -85,7 +89,10 @@ def _get_geometry_traces(area: Polygon) -> Scatter:
 
 
 def _get_colormap(
-    frame_data: pd.DataFrame, max_speed: float, color_mode: str
+    frame_data: pd.DataFrame,
+    min_value: float,
+    max_value: float,
+    color_mode: str,
 ) -> List[Scatter]:
     """Utilize scatter plots with varying colors for each agent instead of individual shapes.
 
@@ -97,8 +104,17 @@ def _get_colormap(
             "color": frame_data["speed"],
             "colorscale": "Jet_r",
             "colorbar": {"title": "Speed [m/s]"},
-            "cmin": 0,
-            "cmax": max_speed,
+            "cmin": min_value,
+            "cmax": max_value,
+        }
+    elif color_mode == "Motivation":
+        marker_dict = {
+            "size": frame_data["radius"] * 2,
+            "color": frame_data["motivation"],
+            "colorscale": "Jet_r",
+            "colorbar": {"title": "Motivation"},
+            "cmin": min_value,
+            "cmax": max_value,
         }
     else:
         colors = frame_data["gender"].map({2: "blue", 1: "green"})
@@ -113,7 +129,13 @@ def _get_colormap(
         y=frame_data["y"],
         mode="markers",
         marker=marker_dict,
-        text=frame_data["speed"] if color_mode == "speed" else frame_data["gender"],
+        text=(
+            frame_data["speed"]
+            if color_mode == "Speed"
+            else frame_data["motivation"]
+            if color_mode == "Motivation"
+            else frame_data["gender"]
+        ),
         showlegend=False,
         hoverinfo="none",
     )
@@ -122,7 +144,10 @@ def _get_colormap(
 
 
 def _get_shapes_for_frame(
-    frame_data: pd.DataFrame, min_speed: float, max_speed: float, color_mode: str
+    frame_data: pd.DataFrame,
+    min_value: float,
+    max_value: float,
+    color_mode: str,
 ) -> Tuple[Shape, Scatter, Shape]:
     """Construct circles as Shapes for agents, Hover and Directions."""
 
@@ -162,7 +187,9 @@ def create_shape(row: pd.DataFrame) -> Shape:
                 _create_orientation_line(row, color="rgba(255,255,255,0)"),
             )
         if color_mode == "Speed":
-            color = _speed_to_color(row["speed"], min_speed, max_speed)
+            color = _value_to_color(row["speed"], min_value, max_value)
+        elif color_mode == "Motivation":
+            color = _value_to_color(row["motivation"], min_value, max_value)
         else:
             gender_colors = {
                 1: "blue",  # Assuming 1 is for female
@@ -301,7 +328,14 @@ def _get_processed_frame_data(
     """Process frame data and ensure it matches the maximum agent count."""
     frame_data = data_df[data_df["frame"] == frame_num]
     agent_count = len(frame_data)
-    dummy_agent_data = {"x": 0, "y": 0, "radius": 0, "speed": DUMMY_SPEED}
+    dummy_agent_data = {
+        "x": 0,
+        "y": 0,
+        "radius": 0,
+        "speed": DUMMY_SPEED,
+        "motivation": 0.0,
+        "gender": 0,
+    }
     while len(frame_data) < max_agents:
         dummy_df = pd.DataFrame([dummy_agent_data])
         frame_data = pd.concat([frame_data, dummy_df], ignore_index=True)
@@ -326,8 +360,15 @@ def animate(
     """Animate a trajectory."""
     data_df["radius"] = radius
     frames = data_df["frame"].unique()
-    min_speed = data_df["speed"].min()
-    max_speed = data_df["speed"].max()
+    if color_mode == "Speed":
+        min_value = data_df["speed"].min()
+        max_value = data_df["speed"].max()
+    elif color_mode == "Motivation":
+        min_value = data_df["motivation"].min()
+        max_value = data_df["motivation"].max()
+    else:
+        min_value = 0.0
+        max_value = 1.0
     max_agents = data_df.groupby("frame").size().max()
     frames = []
     steps = []
@@ -340,14 +381,16 @@ def animate(
         initial_shapes,
         initial_hover_trace,
         initial_arrows,
-    ) = _get_shapes_for_frame(initial_frame_data, min_speed, max_speed, color_mode)
-    color_map_trace = _get_colormap(initial_frame_data, max_speed, color_mode)
+    ) = _get_shapes_for_frame(initial_frame_data, min_value, max_value, color_mode)
+    color_map_trace = _get_colormap(
+        initial_frame_data, min_value, max_value, color_mode
+    )
     for frame_num in selected_frames:
         frame_data, agent_count = _get_processed_frame_data(
             data_df, frame_num, max_agents
         )
         shapes, hover_traces, arrows = _get_shapes_for_frame(
-            frame_data, min_speed, max_speed, color_mode
+            frame_data, min_value, max_value, color_mode
         )
         # title = f"<b>{title_note + '  |  ' if title_note else ''}N: {agent_count}</b>"
         title = f"<b>{title_note + '  |  ' if title_note else ''}Number  of Agents: {initial_agent_count}</b>"

app.py

@@ -5,17 +5,23 @@
 Date: August 11, 2023
 """
 
+from pathlib import Path
+
 import jupedsim as jps
-import streamlit as st
+import pandas as pd
 import pedpy
+import streamlit as st
+from anim import animate
+from jupedsim.internal.notebook_utils import read_sqlite_file
 from src import docs
 import glob
 from src.logger_config import init_logger
 from src.simutilities import (
+    call_simulation,
     extract_motivation_parameters,
     plot_motivation_model,
 )
-from src.ui import init_sidebar, simulation_tab
+from src.ui import init_sidebar, simulation_tab, ui_simulation_controls
 
 if __name__ == "__main__":
     init_logger()
@@ -34,7 +40,58 @@
         case "Documentation":
             docs.main()
         case "Simulation":
+            c1, c2, _c3 = st.columns(3)
             data = simulation_tab()
+            config_file, output_file, fps = ui_simulation_controls(data)
+
+            if c1.button("Run Simulation"):
+                call_simulation(config_file, output_file, data)
+
+            if c2.button("Visualization"):
+                output_path = Path(output_file)
+                if output_path.exists():
+                    trajectory_data, walkable_area = read_sqlite_file(output_file)
+                    motivation_path = output_path.with_name(
+                        output_path.stem + "_motivation.csv"
+                    )
+                    if not motivation_path.exists():
+                        st.warning(f"Motivation file not found: {motivation_path}")
+                    else:
+                        trajectory_df = trajectory_data.data.copy()
+                        motivation_df = pd.read_csv(motivation_path)
+                        data_with_motivation = trajectory_df.merge(
+                            motivation_df[["frame", "id", "motivation"]],
+                            on=["frame", "id"],
+                            how="left",
+                        )
+                        data_with_motivation["motivation"] = data_with_motivation[
+                            "motivation"
+                        ].fillna(1.0)
+                        data_with_motivation["gender"] = 1
+                        data_with_motivation["speed"] = 0.0
+                        width = data["motivation_parameters"]["width"]
+                        vertices = data["motivation_parameters"]["motivation_doors"][
+                            0
+                        ]["vertices"]
+                        x0 = 0.5 * (vertices[0][0] + vertices[1][0]) - width
+                        y0 = 0.5 * (vertices[0][1] + vertices[1][1]) - width
+                        x1 = 0.5 * (vertices[0][0] + vertices[1][0]) + width
+                        y1 = 0.5 * (vertices[1][1] + vertices[1][1]) + width
+
+                        animation = animate(
+                            data_with_motivation,
+                            walkable_area,
+                            every_nth_frame=int(fps),
+                            color_mode="Motivation",
+                            radius=0.1,
+                            x0=x0,
+                            y0=y0,
+                            x1=x1,
+                            y1=y1,
+                        )
+                        st.plotly_chart(animation)
+                else:
+                    st.warning(f"Trajectory file not found: {output_file}")
 
             params = extract_motivation_parameters(data)
             mapping = params.get("mapping_block", {})

src/simutilities.py

@@ -1,17 +1,20 @@
 """Utilities for simulation."""
 
+import logging
 from pathlib import Path
 from typing import Any, Dict, List, Tuple
 
 import streamlit as st
 from src import motivation_model as mm
 from src import motivation_mapping as mmap
 from src.inifile_parser import (
+    parse_fps,
     parse_motivation_doors,
     parse_motivation_strategy,
     parse_normal_time_gap,
     parse_normal_v_0,
     parse_number_agents,
+    parse_simulation_time,
     parse_time_step,
     parse_velocity_init_parameters,
 )
@@ -182,6 +185,43 @@ def create_motivation_strategy(params: Dict[str, Any]) -> mm.MotivationStrategy:
         raise ValueError(f"Unknown strategy: {strategy}")
 
 
+def call_simulation(config_file: str, output_file: str, data: Dict[str, Any]) -> None:
+    """Run the simulation based on the provided configuration and data."""
+    from simulation import init_and_run_simulation
+
+    msg = st.empty()
+    output_path = Path(output_file)
+    if output_path.exists():
+        output_path.unlink()
+
+    number_agents = parse_number_agents(data)
+    fps = parse_fps(data)
+    time_step = parse_time_step(data)
+    simulation_time = parse_simulation_time(data)
+    open_door_time = float(data["simulation_parameters"].get("open_door_time", 0.0))
+    strategy = parse_motivation_strategy(data)
+
+    logging.info(f"Call simulation {number_agents = }")
+    msg.code(f"Running simulation with {number_agents}. Strategy: <{strategy}>...")
+
+    with st.spinner("Simulating..."):
+        try:
+            evac_time, _ = init_and_run_simulation(
+                fps,
+                time_step,
+                simulation_time,
+                open_door_time,
+                data,
+                output_path,
+                msg,
+            )
+        except ValueError as exc:
+            st.error(f"Logistic configuration error: {exc}")
+            return
+
+    msg.code(f"Finished simulation. Evac time {evac_time:.2f} s")
+
+
 def plot_motivation_model(params: Dict[str, Any]) -> None:
     """Plot the motivation model based on the given parameters.