#000: Shubham: Implement common interface for mpi and kafka

common/Cargo.toml

@@ -12,3 +12,4 @@ serde_derive = "1.0.103"
 serde_json = "1.0.85"
 serde_yaml = "0.9.13"
 validator = { version = "0.16.0", features = ["derive"] }
+log = "0.4.20"

common/config/test/travel_plan.json

@@ -0,0 +1,163 @@
+{
+  "engine_configs": [
+    {
+      "engine_id": "engine1",
+      "config": {
+        "sim_id": "sim-timestamp",
+        "population": {
+          "Auto": {
+            "number_of_agents": 10000,
+            "public_transport_percentage": 0.2,
+            "working_percentage": 0.7
+          }
+        },
+        "disease": {
+          "regular_transmission_start_day": 5,
+          "high_transmission_start_day": 20,
+          "last_day": 40,
+          "asymptomatic_last_day": 9,
+          "mild_infected_last_day": 12,
+          "regular_transmission_rate": 0.025,
+          "high_transmission_rate": 0.25,
+          "death_rate": 0.035,
+          "percentage_asymptomatic_population": 0.3,
+          "percentage_severe_infected_population": 0.3,
+          "exposed_duration": 48,
+          "pre_symptomatic_duration": 48
+        },
+        "geography_parameters": {
+          "grid_size": 250,
+          "hospital_beds_percentage": 0.003
+        },
+        "hours": 10000,
+        "interventions": [
+          {
+            "Vaccinate": {
+              "at_hour": 5000,
+              "percent": 0.2
+            }
+          },
+          {
+            "Lockdown": {
+              "at_number_of_infections": 100,
+              "essential_workers_population": 0.1
+            }
+          }
+        ]
+      }
+    },
+    {
+      "engine_id": "engine2",
+      "config": {
+        "sim_id": "sim-timestamp",
+        "population": {
+          "Auto": {
+            "number_of_agents": 10000,
+            "public_transport_percentage": 0.2,
+            "working_percentage": 0.7
+          }
+        },
+        "disease": {
+          "regular_transmission_start_day": 5,
+          "high_transmission_start_day": 20,
+          "last_day": 40,
+          "asymptomatic_last_day": 9,
+          "mild_infected_last_day": 12,
+          "regular_transmission_rate": 0.025,
+          "high_transmission_rate": 0.25,
+          "death_rate": 0.035,
+          "percentage_asymptomatic_population": 0.3,
+          "percentage_severe_infected_population": 0.3,
+          "exposed_duration": 48,
+          "pre_symptomatic_duration": 48
+        },
+        "geography_parameters": {
+          "grid_size": 250,
+          "hospital_beds_percentage": 0.003
+        },
+        "hours": 10000,
+        "interventions": [
+          {
+            "Vaccinate": {
+              "at_hour": 5000,
+              "percent": 0.2
+            }
+          },
+          {
+            "Lockdown": {
+              "at_number_of_infections": 100,
+              "essential_workers_population": 0.1
+            }
+          }
+        ]
+      }
+    },
+    {
+      "engine_id": "engine3",
+      "config": {
+        "sim_id": "sim-timestamp",
+        "population": {
+          "Auto": {
+            "number_of_agents": 10000,
+            "public_transport_percentage": 0.2,
+            "working_percentage": 0.7
+          }
+        },
+        "disease": {
+          "regular_transmission_start_day": 5,
+          "high_transmission_start_day": 20,
+          "last_day": 40,
+          "asymptomatic_last_day": 9,
+          "mild_infected_last_day": 12,
+          "regular_transmission_rate": 0.025,
+          "high_transmission_rate": 0.25,
+          "death_rate": 0.035,
+          "percentage_asymptomatic_population": 0.3,
+          "percentage_severe_infected_population": 0.3,
+          "exposed_duration": 48,
+          "pre_symptomatic_duration": 48
+        },
+        "geography_parameters": {
+          "grid_size": 250,
+          "hospital_beds_percentage": 0.003
+        },
+        "hours": 10000,
+        "interventions": [
+          {
+            "Vaccinate": {
+              "at_hour": 5000,
+              "percent": 0.2
+            }
+          },
+          {
+            "Lockdown": {
+              "at_number_of_infections": 100,
+              "essential_workers_population": 0.1
+            }
+          }
+        ]
+      }
+    }
+  ],
+  "travel_plan": {
+    "regions": ["engine1", "engine2", "engine3"],
+    "migration": {
+      "enabled": true,
+      "matrix": [
+        [0, 156, 10],
+        [108, 0, 290],
+        [90, 75, 0]
+      ],
+      "start_migration_hour": 48,
+      "end_migration_hour": 336
+    },
+    "commute": {
+      "enabled": true,
+      "matrix": [
+        [0, 15, 2],
+        [10, 0, 29],
+        [9, 7, 0]
+      ]
+    }
+  }
+}

orchestrator/src/config.rs → common/src/config/configuration.rs

@@ -17,13 +17,14 @@
  *
  */
 
+use crate::config::Population::Auto;
+use log::debug;
 use std::error::Error;
 use std::fs::File;
 
-use common::config::Population::Auto;
-use common::config::{Config, TravelPlanConfig};
-use common::models::custom_types::Percentage;
-use common::models::travel_plan::TravelPlan;
+use crate::config::{Config, TravelPlanConfig};
+use crate::models::custom_types::Percentage;
+use crate::models::travel_plan::TravelPlan;
 
 pub const TRANSPORT_AREA_RELATIVE_SIZE: Percentage = 0.2;
 
@@ -42,6 +43,10 @@ impl Configuration {
         self.engine_configs.iter().map(|s| s.engine_id.clone()).collect()
     }
 
+    pub fn get_engine_configs(&self) -> &Vec<EngineConfig> {
+        &self.engine_configs
+    }
+
     pub fn read(filename: &str) -> Result<Configuration, Box<dyn Error>> {
         let reader = File::open(filename)?;
         let config: Configuration = serde_json::from_reader(reader)?;
@@ -114,16 +119,15 @@ impl Configuration {
 
 // just a struct for easier parsing
 #[derive(Deserialize, Serialize)]
-struct EngineConfig {
-    engine_id: String,
-    config: Config,
+pub struct EngineConfig {
+    pub engine_id: String,
+    pub config: Config,
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::get_hours;
-    use crate::utils::read_simulation_conf;
+    use crate::utils::{get_hours, read_simulation_conf};
 
     #[test]
     fn should_read_config() {

common/src/config/mod.rs

@@ -22,6 +22,7 @@ mod population;
 mod starting_infections;
 mod travel_plan_config;
 
+mod configuration;
 pub mod intervention_config;
 pub mod request;
 
@@ -34,6 +35,7 @@ use validator::Validate;
 pub use crate::config::geography_parameters::GeographyParameters;
 pub use crate::config::population::*;
 pub use crate::config::starting_infections::StartingInfections;
+pub use configuration::{Configuration, EngineConfig};
 
 use crate::disease::{Disease, DiseaseOverride};
 use crate::models::custom_types::{Hour, Size};

common/src/utils/mod.rs

@@ -20,3 +20,20 @@
 mod random_wrapper;
 
 pub use random_wrapper::RandomWrapper;
+use serde_json::Value;
+use std::fs::File;
+
+pub fn read_simulation_conf(filename: &str) -> String {
+    let reader = File::open(filename).unwrap();
+    let config: Value = serde_json::from_reader(reader).unwrap();
+    let sim = config.as_object().unwrap();
+    serde_json::to_string(sim).unwrap()
+}
+
+pub fn get_hours(filename: &str) -> i64 {
+    let reader = File::open(filename).unwrap();
+    let config: Value = serde_json::from_reader(reader).unwrap();
+    let sim = config.get("engine_configs").unwrap().as_array().unwrap();
+    let hours = sim[0].get("config").unwrap().get("hours");
+    hours.unwrap().as_i64().unwrap()
+}

engine-app/Cargo.toml

@@ -13,3 +13,4 @@ clap = { version = "4.0.32", features = ["derive"] }
 env_logger = "0.10.0"
 opentelemetry = { version = "0.18.0", features = ["rt-tokio"] }
 opentelemetry-jaeger = { version = "0.17.0", features = ["rt-tokio"] }
+mpi = { version = "0.7.0", features = ["user-operations", "derive"] }