Update pyciemss-service intervention converters #119 (#120)

Updating the interventions structure to follow the new structure used in hmi-server/client

service/models/base.py

@@ -45,12 +45,16 @@ class InterventionSelection(BaseModel):
 class HMIStaticIntervention(BaseModel):
  timestep: float
  value: float
+ applied_to: str
+ type: str
 
 
 class HMIDynamicIntervention(BaseModel):
  parameter: str
  threshold: float
  value: float
+ applied_to: str
+ type: str
 
 
 class HMIInterventionPolicy(BaseModel):
@@ -60,8 +64,6 @@ class HMIInterventionPolicy(BaseModel):
 
 class HMIIntervention(BaseModel):
  name: str
- applied_to: str
- type: str
  static_interventions: Optional[list[HMIStaticIntervention]] = Field(default=None)
  dynamic_interventions: Optional[list[HMIDynamicIntervention]] = Field(default=None)
 

service/models/converters.py

@@ -15,8 +15,6 @@ def fetch_and_convert_static_interventions(policy_intervention_id, job_id):
  for inter in policy_intervention["interventions"]:
  intervention = HMIIntervention(
  name=inter["name"],
- applied_to=inter["applied_to"],
- type=inter["type"],
  static_interventions=inter["static_interventions"],
  dynamic_interventions=inter["dynamic_interventions"],
  )
@@ -32,8 +30,6 @@ def fetch_and_convert_dynamic_interventions(policy_intervention_id, job_id):
  for inter in policy_intervention["interventions"]:
  intervention = HMIIntervention(
  name=inter["name"],
- applied_to=inter["applied_to"],
- type=inter["type"],
  static_interventions=inter["static_interventions"],
  dynamic_interventions=inter["dynamic_interventions"],
  )
@@ -50,11 +46,11 @@ def convert_static_interventions(interventions: list[HMIIntervention]):
  for inter in interventions:
  for static_inter in inter.static_interventions:
  time = torch.tensor(float(static_inter.timestep))
- parameter_name = inter.applied_to
+ parameter_name = static_inter.applied_to
  value = torch.tensor(float(static_inter.value))
- if inter.type == "parameter":
+ if static_inter.type == "parameter":
  static_param_interventions[time][parameter_name] = value
- if inter.type == "state":
+ if static_inter.type == "state":
  static_state_interventions[time][parameter_name] = value
  return static_param_interventions, static_state_interventions
 
@@ -83,17 +79,17 @@ def convert_dynamic_interventions(interventions: list[HMIIntervention]):
  ] = defaultdict(dict)
  for inter in interventions:
  for dynamic_inter in inter.dynamic_interventions:
- parameter_name = inter.applied_to
+ parameter_name = dynamic_inter.applied_to
  threshold_value = torch.tensor(float(dynamic_inter.threshold))
  to_value = torch.tensor(float(dynamic_inter.value))
  threshold_func = make_var_threshold(
  dynamic_inter.parameter, threshold_value
  )
- if inter.type == "parameter":
+ if dynamic_inter.type == "parameter":
  dynamic_parameter_interventions[threshold_func].update(
  {parameter_name: to_value}
  )
- if inter.type == "state":
+ if dynamic_inter.type == "state":
  dynamic_state_interventions[threshold_func].update(
  {parameter_name: to_value}
  )

tests/examples/simulate/input/intervention.json

@@ -13,15 +13,16 @@
  "interventions":[
  {
  "name":"New_Intervention",
- "applied_to":"β",
  "static_interventions":[
  {
  "timestep":0,
- "value":0
+ "value":0,
+ "applied_to":"β"
  },
  {
  "timestep":1,
- "value":2
+ "value":2,
+ "applied_to":"β"
  }
  ],
  "dynamic_interventions":[