Allow for sudden increase in the heat pump awareness

simulation/__main__.py

@@ -172,6 +172,18 @@ def check_string_is_isoformat_datetime(string) -> str:
         type=convert_to_datetime,
     )
 
+    parser.add_argument(
+        "--heat-pump-awareness-campaign-date",
+        default=datetime.datetime(2028, 1, 1),
+        type=convert_to_datetime,
+    )
+
+    parser.add_argument(
+        "--heat-pump-awareness-due-to-campaign",
+        default=0.8,
+        type=float_between_0_and_1,
+    )
+
     # SOURCE: Default values from https://energysavingtrust.org.uk/about-us/our-data/ (England, Scotland and Wales)
     # These fuel prices were last updated in November 2021, based on predicted fuel prices for 2022
     parser.add_argument("--price-gbp-per-kwh-gas", type=float, default=0.062)
@@ -226,6 +238,8 @@ def validate_args(args):
             args.heat_pump_installer_count,
             args.heat_pump_installer_annual_growth_rate,
             ENGLAND_WALES_ANNUAL_NEW_BUILDS if args.include_new_builds else None,
+            args.heat_pump_awareness_campaign_date,
+            args.heat_pump_awareness_due_to_campaign,
         )
 
         with smart_open.open(args.history_file, "w") as file:

simulation/agents.py

@@ -123,6 +123,7 @@ def __init__(
         roof_energy_efficiency: int,
         is_heat_pump_suitable_archetype: bool,
         is_heat_pump_aware: bool,
+        is_heat_pump_aware_after_campaign: bool,
     ):
         self.id = id
         # Property / tenure attributes
@@ -150,6 +151,9 @@ def __init__(
         self.is_heat_pump_aware = (
             self.heating_system in HEAT_PUMPS or is_heat_pump_aware
         )
+        self.is_heat_pump_aware_after_campaign = (
+            self.heating_system in HEAT_PUMPS or is_heat_pump_aware_after_campaign
+        )
 
         # Household investment decision attributes
         self.is_renovating = False
@@ -416,6 +420,12 @@ def get_proba_rule_out_banned_heating_systems(self, model):
 
         return reverse_sigmoid(years_to_ban)
 
+    def update_heat_pump_awareness(self, model) -> None:
+        # If heat pump awareness campaign has been done, update awareness in response to campaign
+        if InterventionType.HEAT_PUMP_CAMPAIGN in model.interventions:
+            if model.current_datetime >= model.heat_pump_awareness_campaign_date:
+                self.is_heat_pump_aware = self.is_heat_pump_aware_after_campaign
+
     def get_heating_system_options(
         self, model: "DomesticHeatingABM", event_trigger: EventTrigger
     ) -> Set[HeatingSystem]:
@@ -440,8 +450,12 @@ def get_heating_system_options(
                     [HeatingSystem.BOILER_GAS, HeatingSystem.BOILER_OIL]
                 )
 
-        if not is_gas_oil_boiler_ban_announced:
-            # if a gas/boiler ban is announced, we assume all households are aware of heat pumps
+        is_gas_oil_boiler_ban_in_place = (
+            InterventionType.GAS_OIL_BOILER_BAN in model.interventions
+            and model.current_datetime >= model.gas_oil_boiler_ban_datetime
+        )
+        if not is_gas_oil_boiler_ban_in_place:
+            # if a gas/boiler ban is in place, we assume all households are aware of heat pumps
             if not self.is_heat_pump_aware:
                 heating_system_options -= HEAT_PUMPS
 
@@ -613,6 +627,7 @@ def compute_heat_pump_capacity_kw(self, heat_pump_type: HeatingSystem) -> int:
 
     def make_decisions(self, model):
 
+        self.update_heat_pump_awareness(model)
         self.update_heating_status(model)
         self.evaluate_renovation(model)
 
@@ -686,3 +701,6 @@ def make_decisions(self, model):
             self.heating_system_costs_subsidies = costs_subsidies
             self.heating_system_costs_insulation = costs_insulation
             self.insulation_element_upgrade_costs = chosen_insulation_costs
+
+        if self.is_heat_pump_aware:
+            model.households_heat_pump_aware_at_current_step += 1

simulation/collectors.py

@@ -97,6 +97,10 @@ def household_is_heat_pump_aware(household) -> bool:
     return household.is_heat_pump_aware
 
 
+def household_is_heat_pump_aware_after_campaign(household) -> bool:
+    return household.is_heat_pump_aware_after_campaign
+
+
 def household_is_renovating(household) -> bool:
     return household.is_renovating
 
@@ -277,6 +281,10 @@ def model_heat_pump_installations_at_current_step(model) -> int:
     return model.heat_pump_installations_at_current_step
 
 
+def model_heat_pump_awareness(model) -> int:
+    return model.heat_pump_awareness
+
+
 def is_first_timestep(model: "DomesticHeatingABM") -> bool:
     return model.current_datetime == model.start_datetime + model.step_interval
 
@@ -304,6 +312,9 @@ def get_agent_collectors(
         collect_when(model, is_first_timestep)(household_renovation_budget),
         collect_when(model, is_first_timestep)(household_is_heat_pump_suitable),
         collect_when(model, is_first_timestep)(household_is_heat_pump_aware),
+        collect_when(model, is_first_timestep)(
+            household_is_heat_pump_aware_after_campaign
+        ),
         household_heating_system,
         household_heating_system_previous,
         household_heating_functioning,
@@ -352,6 +363,7 @@ def get_model_collectors(
         model_heat_pump_installers,
         model_heat_pump_installation_capacity_per_step,
         model_heat_pump_installations_at_current_step,
+        model_heat_pump_awareness,
         collect_when(model, is_first_timestep)(model_price_gbp_per_kwh_gas),
         collect_when(model, is_first_timestep)(model_price_gbp_per_kwh_electricity),
         collect_when(model, is_first_timestep)(model_price_gbp_per_kwh_oil),

simulation/constants.py

@@ -206,6 +206,7 @@ class InterventionType(enum.Enum):
     BOILER_UPGRADE_SCHEME = 1
     GAS_OIL_BOILER_BAN = 2
     EXTENDED_BOILER_UPGRADE_SCHEME = 3
+    HEAT_PUMP_CAMPAIGN = 4
 
 
 # Source: https://www.ons.gov.uk/peoplepopulationandcommunity/birthsdeathsandmarriages/families/datasets/householdsbytypeofhouseholdandfamilyregionsofenglandandukconstituentcountries

simulation/model.py

@@ -46,6 +46,7 @@ def __init__(
         heat_pump_installer_count: int,
         heat_pump_installer_annual_growth_rate: float,
         annual_new_builds: Optional[Dict[int, int]],
+        heat_pump_awareness_campaign_date: datetime.datetime,
     ):
         self.start_datetime = start_datetime
         self.step_interval = step_interval
@@ -73,14 +74,20 @@ def __init__(
             heat_pump_installer_annual_growth_rate
         )
         self.heat_pump_installations_at_current_step = 0
+        self.households_heat_pump_aware_at_current_step = 0
         self.annual_new_builds = annual_new_builds
+        self.heat_pump_awareness_campaign_date = heat_pump_awareness_campaign_date
 
         super().__init__(UnorderedSpace())
 
     @property
     def household_count(self) -> int:
         return len(self.space.agents)
 
+    @property
+    def heat_pump_awareness(self) -> float:
+        return self.households_heat_pump_aware_at_current_step / self.household_count
+
     @property
     def heat_pump_installers(self) -> int:
 
@@ -199,15 +206,105 @@ def increment_timestep(self):
             self.boiler_upgrade_scheme_spend_gbp
         )
         self.heat_pump_installations_at_current_step = 0
+        self.households_heat_pump_aware_at_current_step = 0
+
+
+def population_heat_pump_awareness(
+    household_population: pd.DataFrame, heat_pump_awareness: float
+) -> List[bool]:
+    """Randomly assign heat pump awareness to each agent in the population
+      with a probability given by input param `heat_pump_awareness`
+
+    Args:
+        household_population (pd.DataFrame): household population
+        heat_pump_awareness (float): probability of household becoming heat
+          pump aware
+
+    Returns:
+        List[bool]: list of length equal to the number of households in the
+          population of True/False values depending on whether the household
+            is heat pump aware or not
+    """
+    return [
+        random.random() < heat_pump_awareness for _ in range(len(household_population))
+    ]
+
+
+def population_heat_pump_awareness_due_to_campaign(
+    agents_heat_pump_awareness: List[bool],
+    heat_pump_awareness_due_to_campaign: float,
+    heat_pump_awareness: float,
+) -> List[bool]:
+    """If heat pump campaign is set as intervention in input params to the ABM,
+      then some agents who are not initially heat pump aware will later become
+        heat pump aware at the time of the campaign.
+    This function randomly designates agents to become heat pump aware due to
+      the campaign.
+    The probability of an agent becoming heat pump aware due to the campaign
+      is given by the input param `heat_pump_awareness_due_to_campaign`.
+
+    Args:
+        agents_heat_pump_awareness (List[bool]): agent awareness of heat pumps
+          at the beginning of the simulation
+        heat_pump_awareness_due_to_campaign (float): probability of household
+          becoming heat pump aware due to campaign intervention
+        heat_pump_awareness (float): probability of a household being
+          initially heat pump aware irrespecictive the campaign intervention
+
+    Returns:
+        List[bool]: list of length equal to the number of households in the
+          population of True/False values depending on whether the household
+            will become heat pump aware or not due to the campaign intervention
+    """
+    # Ensure that the campaign target awareness is at least the initial awareness
+    # If a user sets a lower target awareness than the initial awareness this is the same as the campaign having no effect on heat pump awareness
+    heat_pump_awareness_due_to_campaign = max(
+        heat_pump_awareness_due_to_campaign, heat_pump_awareness
+    )
+
+    # A fraction of the population are always heat pump aware, as defined by `heat_pump_awareness`
+    # To reach the target awareness for the campaign, we need to increase the awareness of the remaning unaware population
+    # Here, the fraction of the remaining population which needs to become aware to reach the campaign target is defined
+    unaware_population_fraction = 1 - heat_pump_awareness
+    if unaware_population_fraction == 0.0:
+        # If all agents aware already, target awareness is zero
+        target_heat_pump_awareness_for_campaign = 0.0
+    else:
+        target_heat_pump_awareness_for_campaign = (
+            heat_pump_awareness_due_to_campaign - heat_pump_awareness
+        ) / unaware_population_fraction
+
+    agents_heat_pump_awareness_after_campaign = agents_heat_pump_awareness.copy()
+    # Randomly assign heat pump awareness to the initially non-aware agent population
+    for i in range(len(agents_heat_pump_awareness_after_campaign)):
+        if not agents_heat_pump_awareness_after_campaign[i]:
+            rand_gen = random.random()
+            if rand_gen < target_heat_pump_awareness_for_campaign:
+                agents_heat_pump_awareness_after_campaign[i] = True
+
+    return agents_heat_pump_awareness_after_campaign
 
 
 def create_household_agents(
     household_population: pd.DataFrame,
-    heat_pump_awareness: float,
     simulation_start_datetime: datetime.datetime,
     all_agents_heat_pump_suitable: bool,
+    heat_pump_awareness: float,
+    heat_pump_awareness_due_to_campaign: float,
 ) -> Iterator[Household]:
-    for household in household_population.itertuples():
+
+    agents_heat_pump_awareness = population_heat_pump_awareness(
+        household_population, heat_pump_awareness
+    )
+    agents_heat_pump_awareness_after_campaign = (
+        population_heat_pump_awareness_due_to_campaign(
+            agents_heat_pump_awareness,
+            heat_pump_awareness_due_to_campaign,
+            heat_pump_awareness,
+        )
+    )
+
+    for i, household in enumerate(household_population.itertuples()):
         yield Household(
             id=household.id,
             location=household.location,
@@ -236,7 +333,10 @@ def create_household_agents(
             is_heat_pump_suitable_archetype=True
             if all_agents_heat_pump_suitable
             else household.is_heat_pump_suitable_archetype,
-            is_heat_pump_aware=random.random() < heat_pump_awareness,
+            is_heat_pump_aware=agents_heat_pump_awareness[i],
+            is_heat_pump_aware_after_campaign=agents_heat_pump_awareness_after_campaign[
+                i
+            ],
         )
 
 
@@ -263,6 +363,8 @@ def create_and_run_simulation(
     heat_pump_installer_count: int,
     heat_pump_installer_annual_growth_rate: float,
     annual_new_builds: Dict[int, int],
+    heat_pump_awareness_campaign_date: datetime.datetime,
+    heat_pump_awareness_due_to_campaign: float,
 ):
 
     model = DomesticHeatingABM(
@@ -282,13 +384,15 @@ def create_and_run_simulation(
         heat_pump_installer_count=heat_pump_installer_count,
         heat_pump_installer_annual_growth_rate=heat_pump_installer_annual_growth_rate,
         annual_new_builds=annual_new_builds,
+        heat_pump_awareness_campaign_date=heat_pump_awareness_campaign_date,
     )
 
     households = create_household_agents(
         household_population,
-        heat_pump_awareness,
         model.start_datetime,
         all_agents_heat_pump_suitable,
+        heat_pump_awareness,
+        heat_pump_awareness_due_to_campaign,
     )
 
     model.add_agents(households)

simulation/tests/common.py

@@ -35,6 +35,7 @@ def household_factory(**agent_attributes):
         "roof_energy_efficiency": 3,
         "is_heat_pump_suitable_archetype": True,
         "is_heat_pump_aware": True,
+        "is_heat_pump_aware_after_campaign": True,
     }
     return Household(**{**default_values, **agent_attributes})
 
@@ -57,6 +58,7 @@ def model_factory(**model_attributes):
         "heat_pump_installer_count": 2_800,
         "heat_pump_installer_annual_growth_rate": 0,
         "annual_new_builds": None,
+        "heat_pump_awareness_campaign_date": datetime.datetime(2025, 2, 1),
     }
 
     return DomesticHeatingABM(**{**default_values, **model_attributes})

simulation/tests/test_agents.py

@@ -47,6 +47,7 @@ def test_create_household_factory(self) -> None:
             roof_energy_efficiency=2,
             is_heat_pump_suitable_archetype=True,
             is_heat_pump_aware=True,
+            is_heat_pump_aware_after_campaign=True,
         )
         assert household.id == 1
         assert household.location == "London"
@@ -68,6 +69,7 @@ def test_create_household_factory(self) -> None:
         assert household.is_heat_pump_suitable_archetype
         assert household.heating_fuel == HeatingFuel.ELECTRICITY
         assert household.is_heat_pump_aware
+        assert household.is_heat_pump_aware_after_campaign
         assert household.is_renovating is not None
 
     def test_household_renovation_budget_increases_with_property_value(self) -> None:

simulation/tests/test_main.py

@@ -158,13 +158,16 @@ def test_intervention_argument(self, mandatory_local_args):
                 "boiler_upgrade_scheme",
                 "--intervention",
                 "extended_boiler_upgrade_scheme",
+                "--intervention",
+                "heat_pump_campaign",
             ]
         )
 
         assert args.intervention == [
             InterventionType.RHI,
             InterventionType.BOILER_UPGRADE_SCHEME,
             InterventionType.EXTENDED_BOILER_UPGRADE_SCHEME,
+            InterventionType.HEAT_PUMP_CAMPAIGN,
         ]
 
     def test_gas_oil_boiler_ban_date_returns_datetime(self, mandatory_local_args):