refactoring to run simulation without subprocess

app.py

@@ -30,8 +30,14 @@
 from scipy import stats
 from shapely import Polygon
 from shapely.ops import unary_union
-
-
+from src.logger_config import init_logger, log_info
+import simulation
+from src.inifile_parser import (
+    parse_fps,
+    parse_time_step,
+    parse_number_agents,
+    parse_simulation_time,
+)
 from src.ui import (
     ui_motivation_parameters,
     ui_simulation_parameters,
@@ -63,6 +69,7 @@ def read_data(output_file: str) -> pd.DataFrame:
 
 
 if __name__ == "__main__":
+    init_logger()
     if "data" not in st.session_state:
         st.session_state.data = {}
 
@@ -123,20 +130,27 @@ def read_data(output_file: str) -> pd.DataFrame:
             if Path(OUTPUT_FILE).exists():
                 Path(OUTPUT_FILE).unlink()
             msg.empty()
-            msg.info("Running simulation ...")
-            command = f"python simulation.py {CONFIG_FILE} {OUTPUT_FILE}"
-            n_agents = st.session_state.data["simulation_parameters"]["number_agents"]
-            with st.spinner(f"Simulating with {n_agents}"):
-                with subprocess.Popen(
-                    command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
-                ) as process:
-                    stdout, stderr = process.communicate()
-                INFO_OUTPUT = stdout.decode().replace("\n", "  \n")
-                WARNINGS = stderr.decode().replace("\n", "  \n")
-                msg.code(INFO_OUTPUT)
-                if WARNINGS:
-                    msg.error(WARNINGS)
-
+            msg.code("Running simulation ...")
+            with open(CONFIG_FILE, "r", encoding="utf8") as f:
+                json_str = f.read()
+                data = json.loads(json_str)
+                fps = parse_fps(data)
+                time_step = parse_time_step(data)
+                number_agents = parse_number_agents(data)
+                simulation_time = parse_simulation_time(data)
+
+            with st.spinner(f"Simulating ..."):
+                if fps and time_step:
+                    simulation.main(
+                        number_agents,
+                        fps,
+                        time_step,
+                        simulation_time,
+                        data,
+                        Path(OUTPUT_FILE),
+                    )
+            msg.code("Finished simulation")
+            st.empty()
         output_path = Path(OUTPUT_FILE)
         if Path("values.txt").exists():
             print(output_path.name)

simulation.py

@@ -74,11 +74,11 @@ def init_simulation(
     :returns:
 
     """
-    width = parse_motivation_parameter(data, "width")
-    height = parse_motivation_parameter(data, "height")
-    seed = parse_motivation_parameter(data, "seed")
-    min_value = parse_motivation_parameter(data, "min_value")
-    max_value = parse_motivation_parameter(data, "max_value")
+    width = parse_motivation_parameter(_data, "width")
+    height = parse_motivation_parameter(_data, "height")
+    seed = parse_motivation_parameter(_data, "seed")
+    min_value = parse_motivation_parameter(_data, "min_value")
+    max_value = parse_motivation_parameter(_data, "max_value")
 
     accessible_areas = parse_accessible_areas(_data)
     geometry = build_geometry(accessible_areas)
@@ -103,7 +103,7 @@ def init_simulation(
     if not motivation_doors:
         log_error("json file does not contain any motivation door")
 
-    choose_motivation_strategy = parse_motivation_strategy(data)
+    choose_motivation_strategy = parse_motivation_strategy(_data)
 
     # =================
     if choose_motivation_strategy == "default":
@@ -257,7 +257,7 @@ def main(
     log_info(f"Running simulation for {len(ped_ids)} agents:")
     run_simulation(simulation, motivation_model, _simulation_time)
     log_info(f"Simulation completed after {simulation.iteration_count()} iterations")
-    log_info(f"simulation time: {simulation.iteration_count()*time_step} [s]")
+    log_info(f"simulation time: {simulation.iteration_count()*_time_step} [s]")
     # log_info(f"Trajectory: {_trajectory_path}")
 
 

src/logger_config.py

@@ -1,57 +1,68 @@
 """
 Setup logger
 """
-import logging
-import sys
 
+import logging
 import jupedsim as jps
 
-# Create a handler that writes INFO and DEBUG messages to stdout
-stdout_handler = logging.StreamHandler(sys.stdout)
-stdout_handler.setLevel(logging.DEBUG)
+# # Create a handler that writes INFO and DEBUG messages to stdout
+# stdout_handler = logging.StreamHandler(sys.stdout)
+# stdout_handler.setLevel(logging.DEBUG)
+
+# # Create a handler that writes WARNING, ERROR, and CRITICAL messages to stderr
+# stderr_handler = logging.StreamHandler(sys.stderr)
+# stderr_handler.setLevel(logging.WARNING)
+
+# # Create a formatter and set it for both handlers
+# formatter = logging.Formatter("%(levelname)s : %(message)s")
+# stdout_handler.setFormatter(formatter)
+# stderr_handler.setFormatter(formatter)
 
-# Create a handler that writes WARNING, ERROR, and CRITICAL messages to stderr
-stderr_handler = logging.StreamHandler(sys.stderr)
-stderr_handler.setLevel(logging.WARNING)
+# # Get the root logger
+# logger = logging.getLogger()
+# logger.setLevel(logging.DEBUG)  # Set the lowest level to log for the root logger
 
-# Create a formatter and set it for both handlers
-formatter = logging.Formatter("%(levelname)s : %(message)s")
-stdout_handler.setFormatter(formatter)
-stderr_handler.setFormatter(formatter)
+# # Add both handlers to the logger
+# logger.addHandler(stdout_handler)
+# logger.addHandler(stderr_handler)
 
-# Get the root logger
-logger = logging.getLogger()
-logger.setLevel(logging.DEBUG)  # Set the lowest level to log for the root logger
 
-# Add both handlers to the logger
-logger.addHandler(stdout_handler)
-logger.addHandler(stderr_handler)
+# Step 3: Configure logger
+
+
+def setup_logging() -> None:
+    """Define logging setup."""
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
 
 
 # Custom log functions
 def log_debug(msg: str) -> None:
     """debug messages"""
-    logger.debug(msg)
+    logging.debug(msg)
 
 
 def log_info(msg: str) -> None:
     """info messages"""
-    logger.info(msg)
+    logging.info(msg)
 
 
 def log_warning(msg: str) -> None:
     """warning messages"""
-    logger.warning(msg)
+    logging.warning(msg)
 
 
 def log_error(msg: str) -> None:
     """errror messages"""
-    logger.error(msg)
+    logging.error(msg)
 
 
 def init_logger() -> None:
     """init logger"""
-
+    setup_logging()
     # jps.set_debug_callback(log_debug)
     jps.set_info_callback(log_info)
     jps.set_warning_callback(log_warning)