'''
Specify the core interventions available in FPsim. Other interventions can be
defined by the user by inheriting from these classes.
'''
import numpy as np
import sciris as sc
import starsim as ss
from . import utils as fpu
from . import methods as fpm
_make_repr = fpu.make_recreatable_intervention_repr
#>> Generic intervention classes
__all__ = ['change_par', 'update_methods', 'add_method', 'change_people_state',
'change_initiation_prob', 'change_initiation', 'method_switching']
##%% Helper functions for JSON serialization
[docs]
class change_par(ss.Intervention):
'''
Change a parameter at a specified point in time.
Args:
par (str): the parameter to change
years (float/arr): the year(s) at which to apply the change
vals (any): a value or list of values to change to (if a list, must have the same length as years); or a dict of year:value entries
If any value is ``'reset'``, reset to the original value.
**Example**::
ec0 = fp.change_par(par='exposure_factor', years=[2000, 2010], vals=[0.0, 2.0]) # Reduce exposure factor
ec0 = fp.change_par(par='exposure_factor', vals={2000:0.0, 2010:2.0}) # Equivalent way of writing
sim = fp.Sim(interventions=ec0).run()
'''
def __init__(self, par, years=None, vals=None, verbose=False, **kwargs):
super().__init__(**kwargs)
self.par = par
self.verbose = verbose
if isinstance(years, dict): # Swap if years is supplied as a dict, so can be supplied first
vals = years
if vals is None:
errormsg = 'Values must be supplied'
raise ValueError(errormsg)
if isinstance(vals, dict):
years = sc.dcp(list(vals.keys()))
vals = sc.dcp(list(vals.values()))
else:
if years is None:
errormsg = 'If vals is not supplied as a dict, then year(s) must be supplied'
raise ValueError(errormsg)
else:
years = sc.toarray(sc.dcp(years))
vals = sc.dcp(vals)
if sc.isnumber(vals):
vals = sc.tolist(vals) # We want to be careful not to take something that might already be an array and interpret different values as years
n_years = len(years)
n_vals = len(vals)
if n_years != n_vals:
errormsg = f'Number of years ({n_years}) does not match number of values ({n_vals})'
raise ValueError(errormsg)
self.years = years
self.vals = vals
return
def init_pre(self, sim):
super().init_pre(sim)
# Validate parameter name
if self.par not in sim.pars.fp:
errormsg = f'Parameter "{self.par}" is not a valid sim parameter'
raise ValueError(errormsg)
# Validate years and values
years = self.years
min_year = min(years)
max_year = max(years)
if min_year < sim.pars.start:
errormsg = f'Intervention start {min_year} is before the start of the simulation'
raise ValueError(errormsg)
if max_year > sim.pars.stop:
errormsg = f'Intervention end {max_year} is after the end of the simulation'
raise ValueError(errormsg)
if years != sorted(years):
errormsg = f'Years {years} should be monotonic increasing'
raise ValueError(errormsg)
# Convert intervention years to sim timesteps
self.counter = 0
self.inds = sc.autolist()
for y in years:
self.inds += sc.findnearest(sim.t.yearvec, y)
# Store original value
self.orig_val = sc.dcp(sim.pars.fp[self.par])
return
def step(self):
sim = self.sim
if len(self.inds) > self.counter:
ind = self.inds[self.counter] # Find the current index
if sim.ti == ind: # Check if the current timestep matches
curr_val = sc.dcp(sim.pars.fp[self.par])
val = self.vals[self.counter]
if val == 'reset':
val = self.orig_val
sim.pars.fp[self.par] = val # Update the parameter value -- that's it!
if self.verbose:
label = f'Sim "{sim.label}": ' if sim.label else ''
print(f'{label}On {sim.t.year}, change {self.counter+1}/{len(self.inds)} applied: "{self.par}" from {curr_val} to {sim.pars.fp[self.par]}')
self.counter += 1
return
def finalize(self):
# Check that all changes were applied
n_counter = self.counter
n_vals = len(self.vals)
if n_counter != n_vals:
errormsg = f'Not all values were applied ({n_vals} ≠ {n_counter})'
raise RuntimeError(errormsg)
super().finalize()
return
[docs]
class add_method(ss.Intervention):
"""
Intervention to add a new contraceptive method to the simulation at a specified time.
Args:
year (float): The year at which to activate the new method
method (Method, optional): A Method object defining the new contraceptive method.
If None, the method will be copied from the source method (specified by ``copy_from``).
method_pars (dict, optional): Dictionary of parameters to update the method attributes.
If provided, these values will override corresponding attributes in the method object
(whether it was provided directly or copied from source). If None, defaults to empty dict.
copy_from (str): Name of the existing method to copy switching probabilities from.
Also used as the source method when ``method=None``.
split_shares (float, optional): If provided, % who would have chosen the 'copy_from' method
and now choose the new method
verbose (bool): Whether to print messages when method is activated (default True)
**Examples**::
# Using a Method object directly
new_method = fp.Method(name='new_impl', label='New Implant', efficacy=0.999,
dur_use=ss.lognorm_ex(ss.years(3), ss.years(0.5)), modern=True)
intv = fp.add_method(year=2010, method=new_method, copy_from='impl')
# Copying from source method (method=None, method_pars=None)
# Creates a copy of 'impl' with name 'impl_copy'
intv = fp.add_method(year=2010, copy_from='impl')
# Copying from source and overriding properties
intv = fp.add_method(year=2010, method_pars={'name': 'new_inj', 'efficacy': 0.995},
copy_from='inj')
# Using method object and overriding properties with method_pars
base_method = fp.Method(name='new_method', efficacy=0.90)
intv = fp.add_method(year=2010, method=base_method,
method_pars={'efficacy': 0.998}, copy_from='impl')
"""
def __init__(self, year=None, method=None, method_pars=None, copy_from=None, split_shares=None, verbose=True, **kwargs):
super().__init__(**kwargs)
# Validate inputs
if year is None:
raise ValueError('Year must be specified for add_method intervention')
if copy_from is None:
raise ValueError('copy_from must specify an existing method name to copy switching behavior from')
if method_pars is not None and not isinstance(method_pars, dict):
raise TypeError(f'method_pars must be a dict, not {type(method_pars)}')
if split_shares is not None:
if not sc.isnumber(split_shares):
raise TypeError(f'split_shares must be a number, not {type(split_shares)}')
if not (0.0 <= split_shares <= 1.0):
raise ValueError(f'split_shares must be between 0 and 1, got {split_shares}')
self.year = year
self.method = method
self.method_pars = method_pars if method_pars is not None else {}
self.copy_from = copy_from
self.verbose = verbose
self.split_shares = split_shares
self.activated = False
self._method_idx = None # Will be set after method is added
return
[docs]
def init_pre(self, sim):
"""
Initialize the intervention before the simulation starts.
This registers the new method but does not activate it yet.
"""
super().init_pre(sim)
# Validate year is within simulation range
if not (sim.pars.start <= self.year <= sim.pars.stop):
raise ValueError(f'Intervention year {self.year} must be between {sim.pars.start} and {sim.pars.stop}')
cm = sim.connectors.contraception
source_method = cm.get_method(self.copy_from)
self._copy_from_name = source_method.name # Store resolved name for use in step()
if self.method is None:
# If the user didn't provide a Method object, clone the source method
# (e.g., copy_from='inj') so the new method inherits all defaults
# such as efficacy, modern/traditional flag, and duration-of-use distribution.
self.method = sc.dcp(source_method)
if 'name' not in self.method_pars:
# Ensure the copied method gets a unique internal name unless the
# user overrides it via method_pars.
self.method.name += '_copy'
for mp, mpar in self.method_pars.items():
# Apply user-specified overrides (e.g., rename, label, rel_dur_use, dur_use, etc.)
setattr(self.method, mp, mpar)
# Extra logic to handle copying of distributions, which isn't currently well-supported
self.method._source_dur = source_method.name
# Register the method in the contraception connector so it exists in `cm.methods`
# and can be referenced by name/index from other interventions.
cm.add_method(self.method)
# Cache the method index for later use (e.g., reporting and activation).
self._method_idx = cm.methods[self.method.name].idx
fp_mod = sim.connectors.fp
old_mix = fp_mod.method_mix
# The FP module tracks method mix as an array of shape (n_methods, n_timepoints).
# Adding a method increases the number of options, so we need to resize this array
# while preserving historical values.
new_mix = np.zeros((cm.n_options, sim.t.npts))
new_mix[:old_mix.shape[0], :] = old_mix
fp_mod.method_mix = new_mix
if self.verbose:
print(f'Registered new method "{self.method.name}" (idx={self._method_idx}), will activate in year {self.year}')
return
[docs]
def step(self):
"""
At the specified year, copy switching probabilities to make the method available.
"""
sim = self.sim
# Check if we've reached the activation year and haven't already activated
if not self.activated and sim.t.year >= self.year:
self.activated = True
if self.verbose:
print(f'Activating new contraceptive method "{self.method.name}" in year {sim.t.year:.1f}')
# Copy switching probabilities
cm = sim.connectors.contraception
cm.copy_switching_to_method(
source_to_method=self._copy_from_name,
dest_to_method=self.method.name,
split_shares=self.split_shares
)
cm.copy_switching_from_method(
source_from_method=self._copy_from_name,
dest_from_method=self.method.name,
)
# Renormalize all probabilities
cm.renormalize_method_choice_pars()
return
[docs]
def finalize(self):
"""
Report summary statistics about the new method usage.
"""
super().finalize()
if self.verbose:
sim = self.sim
fp_mod = sim.connectors.fp
# Get final method usage for the new method
final_usage = fp_mod.method_mix[self._method_idx, -1] if self._method_idx < fp_mod.method_mix.shape[0] else 0
# Count current users
n_users = np.sum(fp_mod.method == self._method_idx)
print(f'add_method finalized: "{self.method.name}" has {n_users} users ({final_usage*100:.2f}% of method mix)')
return
[docs]
class change_people_state(ss.Intervention):
"""
Intervention to modify values of a People's boolean state at one specific
point in time.
Args:
state_name (string): name of the People's state that will be modified
new_val (bool, float): the new state value eligible people will have
years (list, float): The year we want to start the intervention.
if years is None, uses start and end years of sim as defaults
if years is a number or a list with a single element, eg, 2000.5, or [2000.5],
this is interpreted as the start year of the intervention, and the
end year of intervention will be the end of the simulation
eligibility (inds/callable): indices OR callable that returns inds
prop (float): a value between 0 and 1 indicating the x% of eligible people
who will have the new state value
annual (bool): whether the increase, prop, represents a "per year" increase, or per time step
"""
def __init__(self, state_name, new_val, years=None, eligibility=None, prop=1.0, annual=False, **kwargs):
super().__init__(**kwargs)
self.define_pars(
state_name=state_name,
new_val=new_val,
years=years,
eligibility=eligibility,
prop=prop,
annual=annual
)
self.module_name = None
if '.' in state_name:
self.module_name, self.pars.state_name = state_name.split('.')
self.annual_perc = None
return
def init_pre(self, sim):
super().init_pre(sim)
self._validate_pars()
# Lastly, adjust the probability by the sim's timestep, if it's an annual probability
if self.pars.annual:
# per timestep/monthly growth rate or perc of eligible women who will be made to choose contraception
self.annual_perc = self.pars.prop
self.pars.prop = ((1 + self.annual_perc) ** sim.dt)-1
# Validate years and values
if self.pars.years is None:
# f'Intervention start and end years not provided. Will use sim start an end years'
self.pars.years = [sim.pars['start'], sim.pars['stop']]
if sc.isnumber(self.pars.years) or len(self.pars.years) == 1:
self.pars.years = sc.promotetolist(self.pars.years)
# Assumes that start year has been specified, append end of the simulation as end year of the intervention
self.pars.years.append(sim.pars['stop'])
min_year = min(self.pars.years)
max_year = max(self.pars.years)
if min_year < sim.pars['start']:
errormsg = f'Intervention start {min_year} is before the start of the simulation.'
raise ValueError(errormsg)
if max_year > sim.pars['stop']:
errormsg = f'Intervention end {max_year} is after the end of the simulation.'
raise ValueError(errormsg)
if self.pars.years != sorted(self.pars.years):
errormsg = f'Years {self.pars.years} should be monotonically increasing'
raise ValueError(errormsg)
return
def _validate_pars(self):
# Validation
if self.pars.state_name is None:
errormsg = 'A state name must be supplied.'
raise ValueError(errormsg)
if self.pars.new_val is None:
errormsg = 'A new value must be supplied.'
raise ValueError(errormsg)
if self.pars.eligibility is None:
errormsg = 'Eligibility needs to be provided'
raise ValueError(errormsg)
return
[docs]
def check_eligibility(self):
"""
Return an array of uids of agents eligible
"""
if callable(self.pars.eligibility):
eligible_uids = self.pars.eligibility(self.sim)
elif sc.isarray(self.pars.eligibility):
eligible_uids = self.pars.eligibility
else:
errormsg = 'Eligibility must be a function or an array of uids'
raise ValueError(errormsg)
return ss.uids(eligible_uids)
def step(self):
if self.pars.years[0] <= self.sim.t.year <= self.pars.years[1]: # Inclusive range
eligible_uids = self.check_eligibility()
if self.module_name is not None:
self.sim.people[self.module_name][self.pars.state_name][eligible_uids] = self.pars.new_val
else:
self.sim.people[self.pars.state_name][eligible_uids] = self.pars.new_val
return
[docs]
class update_methods(ss.Intervention):
"""
Intervention to modify method efficacy and/or switching matrix.
Args:
year (float): The year we want to change the method.
eff (dict):
An optional key for changing efficacy; its value is a dictionary with the following schema:
{method: efficacy}
Where method is the name of the contraceptive method to be changed,
and efficacy is a number with the efficacy
dur_use (dict):
Optional key for changing the duration of use; its value is a dictionary with the following schema:
{method: dur_use}
Where method is the method to be changed, and dur_use is a dict representing a distribution, e.g.
dur_use = {'Injectables: dict(dist='lognormal', par1=a, par2=b)}
p_use (float): probability of using any form of contraception
method_mix (list/arr): probabilities of selecting each form of contraception
"""
def __init__(self, year, eff=None, dur_use=None, p_use=None, method_mix=None,
method_choice_pars=None, verbose=False, **kwargs):
super().__init__(**kwargs)
self.define_pars(
year=year,
eff=eff,
dur_use=dur_use,
p_use=p_use,
method_mix=method_mix,
method_choice_pars=method_choice_pars,
verbose=verbose
)
self.applied = False
return
def init_pre(self, sim):
super().init_pre(sim)
self._validate()
par_name = None
if self.pars.p_use is not None and isinstance(sim.connectors.contraception, fpm.SimpleChoice):
par_name = 'p_use'
if self.pars.method_mix is not None and isinstance(sim.connectors.contraception, fpm.SimpleChoice):
par_name = 'method_mix'
if par_name is not None:
errormsg = (
f"Contraceptive module {type(sim.connectors.contraception)} does not have `{par_name}` parameter. "
f"For this type of module, the probability of contraceptive use depends on people attributes and can't be reset using this intervention.")
print(errormsg)
return
def _validate(self):
# Validation
if self.pars.year is None:
errormsg = 'A year must be supplied'
raise ValueError(errormsg)
has_input = any([
self.pars.eff is not None,
self.pars.dur_use is not None,
self.pars.p_use is not None,
self.pars.method_mix is not None,
self.pars.method_choice_pars is not None,
])
if not has_input:
errormsg = 'At least one of eff, dur_use, p_use, method_mix, or method_choice_pars must be supplied'
raise ValueError(errormsg)
return
[docs]
def step(self):
"""
Applies the efficacy or contraceptive uptake changes if it is the specified year
based on scenario specifications.
"""
sim = self.sim
cm = sim.connectors.contraception
if not self.applied and sim.t.year >= self.pars.year:
self.applied = True # Ensure we don't apply this more than once
# Implement efficacy
if self.pars.eff is not None:
for k, rawval in self.pars.eff.items():
cm.update_efficacy(method_label=k, new_efficacy=rawval)
# Implement changes in duration of use
if self.pars.dur_use is not None:
for k, rawval in self.pars.dur_use.items():
cm.update_duration(method_label=k, new_duration=rawval)
# Change in probability of use
if self.pars.p_use is not None:
cm.pars['p_use'].set(self.pars.p_use)
# Change in method mix
if self.pars.method_mix is not None:
this_mix = self.pars.method_mix / np.sum(self.pars.method_mix) # Renormalise in case they are not adding up to 1
cm.pars['method_mix'] = this_mix
# Change in switching matrix
if self.pars.method_choice_pars is not None:
print(f'Changed contraceptive switching matrix in year {sim.t.year}')
cm.method_choice_pars = self.pars.method_choice_pars
return
[docs]
class change_initiation_prob(ss.Intervention):
"""
Intervention to change the probabilty of contraception use trend parameter in
contraceptive choice modules that have a logistic regression model.
Args:
year (float): The year in which this intervention will be applied
prob_use_intercept (float): A number that changes the intercept in the logistic regression model
p_use = 1 / (1 + np.exp(-rhs + p_use_time_trend + p_use_intercept))
"""
def __init__(self, year=None, prob_use_intercept=0.0, verbose=False, **kwargs):
super().__init__(**kwargs)
self.year = year
self.prob_use_intercept = prob_use_intercept
self.verbose = verbose
self.applied = False
self.par_name = None
return
def init_pre(self, sim=None):
super().init_pre(sim)
# self._validate()
if isinstance(sim.people.contraception_module, (fpm.SimpleChoice)):
self.par_name = 'prob_use_intercept'
if self.par_name is None:
errormsg = (
f"Contraceptive module {type(sim.people.contraception_module)} does not have `{self.par_name}` parameter.")
raise ValueError(errormsg)
return
[docs]
def step(self):
"""
Applies the changes to efficacy or contraceptive uptake changes if it is the specified year
based on scenario specifications.
"""
sim = self.sim
if not self.applied and sim.t.year >= self.year:
self.applied = True # Ensure we don't apply this more than once
sim.people.contraception_module.pars[self.par_name] = self.prob_use_intercept
return
[docs]
class change_initiation(ss.Intervention):
"""
Intervention that modifies the outcomes of whether women are on contraception or not
Select a proportion of women and sets them on a contraception method.
Args:
years (list, float): The year we want to start the intervention.
if years is None, uses start and end years of sim as defaults
if years is a number or a list with a single element, eg, 2000.5, or [2000.5],
this is interpreted as the start year of the intervention, and the
end year of intervention will be the end of the simulation
eligibility (callable): callable that returns a filtered version of
people eligible to receive the intervention
perc (float): a value between 0 and 1 indicating the x% extra of women
who will be made to select a contraception method.
The proportion or % is with respect to the number of
women who were on contraception:
- the previous year (12 months earlier)?
- at the beginning of the intervention.
annual (bool): whether the increase, perc, represents a "per year"
increase.
age_range (tuple, optional): (min_age, max_age) to restrict eligibility.
If provided, only women in this age range are eligible. Use with perc_of_eligible=True.
perc_of_eligible (bool): If True, perc is percentage of eligible women (not of current users).
Use this with age_range for age-targeted scale-up scenarios.
target_method (str/int, optional): Method name or index for new users. If None, uses
natural method distribution.
final_perc (float, optional): If provided, perc will scale linearly from initial perc to
final_perc over the intervention period. Enables scale-up scenarios.
"""
def __init__(self, years=None, eligibility=None, perc=0.0, annual=True, force_theoretical=False,
age_range=None, perc_of_eligible=False, target_method=None, final_perc=None,
verbose=False, **kwargs):
super().__init__(**kwargs)
self.years = years
self.eligibility = eligibility
self.perc = perc
self.annual = annual
self.annual_perc = None
self.force_theoretical = force_theoretical
self.current_women_oncontra = None
self.verbose = verbose
# New parameters for age-restricted, targeted initiation
self.age_range = age_range
self.perc_of_eligible = perc_of_eligible
self.target_method = target_method
self.target_method_idx = None
self.final_perc = final_perc
self.init_perc = perc # Store initial perc for linear interpolation
# Initial value of women on contra at the start of the intervention. Tracked for validation.
self.init_women_oncontra = None
# Theoretical number of women on contraception we should have by the end of the intervention period, if
# nothing else affected the dynamics of the contraception. Tracked for validation.
self.expected_women_oncontra = None
# Bernoulli distribution for selecting women to initiate on contraception
self._p_selected = ss.bernoulli(p=0.0)
return
def init_pre(self, sim=None):
super().init_pre(sim)
# Validate years and values
if self.years is None:
# f'Intervention start and end years not provided. Will use sim start an end years'
self.years = [sim.pars['start'], sim.pars['stop']]
if sc.isnumber(self.years) or len(self.years) == 1:
self.years = sc.promotetolist(self.years)
# Assumes that start year has been specified, append end of the simulation as end year of the intervention
self.years.append(sim.pars['stop'])
min_year = min(self.years)
max_year = max(self.years)
if min_year < sim.pars['start']:
errormsg = f'Intervention start {min_year} is before the start of the simulation.'
raise ValueError(errormsg)
if max_year > sim.pars['stop']:
errormsg = f'Intervention end {max_year} is after the end of the simulation.'
raise ValueError(errormsg)
if self.years != sorted(self.years):
errormsg = f'Years {self.years} should be monotonically increasing'
raise ValueError(errormsg)
# Resolve target method if specified
if self.target_method is not None:
cm = sim.connectors.contraception
if isinstance(self.target_method, str):
if self.target_method not in cm.methods:
errormsg = f'Target method "{self.target_method}" not found'
raise ValueError(errormsg)
self.target_method_idx = cm.methods[self.target_method].idx
else:
self.target_method_idx = self.target_method
# Validate age_range if provided
if self.age_range is not None:
if len(self.age_range) != 2:
errormsg = 'age_range must be a tuple of (min_age, max_age)'
raise ValueError(errormsg)
# Convert final_perc if needed
if self.final_perc is not None and self.annual:
self.final_perc = ((1 + self.final_perc) ** float(sim.dt)) - 1
# Adjust the probability by the sim's timestep, if it's an annual probability
if self.annual:
# per timestep/monthly growth rate or perc of eligible women who will be made to choose contraception
self.annual_perc = self.perc
self.perc = ((1 + self.annual_perc) ** float(sim.dt))-1
return
[docs]
def check_eligibility(self):
"""
Select eligible who is eligible
"""
sim = self.sim
contra_choosers = []
if self.eligibility is None:
contra_choosers = self._default_contra_choosers()
return contra_choosers
def _default_contra_choosers(self):
# TODO: do we care whether women people have ti_contra > 0? For instance postpartum women could be made to choose earlier?
# Though it is trickier because we need to reset many postpartum-related attributes
ppl = self.sim.people
# Base eligibility criteria
eligible = (ppl.female & ppl.alive & # living women
(ppl.age < self.sim.pars.fp['age_limit_fecundity']) & # who are fecund
ppl.fp.sexual_debut & # who already had their sexual debut
(~ppl.fp.pregnant) & # who are not currently pregnant
(~ppl.fp.postpartum) & # who are not in postpartum
(~ppl.fp.on_contra) # who are not already on contra
)
# Add age restriction if specified
if self.age_range is not None:
min_age, max_age = self.age_range
eligible = eligible & (ppl.age >= min_age) & (ppl.age < max_age)
return eligible.uids
def _get_current_perc(self, year):
"""Calculate current perc based on linear interpolation if final_perc is specified."""
if self.final_perc is None:
return self.perc
start_year, end_year = self.years[0], self.years[1]
if year <= start_year:
return self.perc
elif year >= end_year:
return self.final_perc
else:
# Linear interpolation
frac = (year - start_year) / (end_year - start_year)
return self.perc + frac * (self.final_perc - self.perc)
def step(self):
sim = self.sim
ti = sim.ti
year = sim.t.year
# Calculate current perc (may vary over time)
current_perc = self._get_current_perc(year)
# Save theoretical number based on the value of women on contraception at start of intervention
if self.years[0] == year:
self.expected_women_oncontra = (sim.people.alive & sim.people.fp.on_contra).sum()
self.init_women_oncontra = self.expected_women_oncontra
# Apply intervention within this time range
if self.years[0] <= year <= self.years[1]: # Inclusive range
self.current_women_oncontra = (sim.people.alive & sim.people.fp.on_contra).sum()
# Eligible population
can_choose_contra_uids = self.check_eligibility()
n_eligible = len(can_choose_contra_uids)
if n_eligible == 0:
return
# Calculate number of new users
if self.perc_of_eligible:
# Percentage of eligible women (for age-targeted scenarios)
new_on_contra = current_perc * n_eligible
else:
# Traditional behavior: percentage based on current users
nnew_on_contra = current_perc * self.expected_women_oncontra
# NOTE: TEMPORARY: force specified increase
# how many more women should be added per time step
# However, if the current number of women on contraception is >> than the expected value, this
# intervention does nothing. The forcing ocurrs in one particular direction, making it incomplete.
# If the forcing had to be fully function, when there are more women than the expected value
# this intervention should additionaly 'reset' the contraceptive state and related attributes (ie, like the duration on the method)
if self.force_theoretical:
additional_women_on_contra = self.expected_women_oncontra - self.current_women_oncontra
if additional_women_on_contra < 0:
additional_women_on_contra = 0
new_on_contra = nnew_on_contra + additional_women_on_contra
else:
new_on_contra = current_perc * self.current_women_oncontra
self.expected_women_oncontra += nnew_on_contra
if new_on_contra == 0:
return
if n_eligible < new_on_contra:
if not self.perc_of_eligible: # Only warn for non-eligible mode
print(f"There are fewer eligible women ({n_eligible}) than "
f"the number of women who should be initiated on contraception ({new_on_contra}).")
new_on_contra = n_eligible
# Of eligible women, select who will be asked to choose contraception
p_selected = new_on_contra * np.ones(n_eligible) / n_eligible
selected = np.random.random(n_eligible) < p_selected
sim.people.fp.on_contra[can_choose_contra_uids[selected]] = True
new_users_uids = can_choose_contra_uids[selected]
# Assign method
if self.target_method_idx is not None:
# Use specified target method
sim.people.fp.method[new_users_uids] = self.target_method_idx
else:
# Use natural method distribution
sim.people.fp.method[new_users_uids] = sim.connectors.contraception.init_method_dist(new_users_uids)
sim.people.fp.ever_used_contra[new_users_uids] = 1
method_dur = sim.connectors.contraception.set_dur_method(new_users_uids)
sim.people.fp.ti_contra[new_users_uids] = ti + method_dur
return
# Note: age_restricted_initiation functionality has been integrated into change_initiation
# Use change_initiation with age_range, perc_of_eligible, target_method, and final_perc parameters
[docs]
class method_switching(ss.Intervention):
"""
Intervention that causes women to switch from specific contraceptive methods to a target method.
This intervention modifies the switching probabilities in the contraception module's switching
matrix to induce transitions from one or more source methods to a destination method.
Args:
year (float): Year when switching intervention begins
from_methods (str/list): Method name(s) or index(es) to switch from
to_method (str/int): Method name or index to switch to
switch_prob (float/dict): Probability of switching. Can be:
- float: Same probability for all source methods
- dict: {method_name: probability} for different rates per source
age_groups (list): Age group names to apply switching. If None, applies to all age groups.
Use None for all, or list like ['<18', '18-20', '20-25', '25-35', '>35']
postpartum (int/list): Postpartum states to modify (0, 1, 6). If None, modifies all states.
annual (bool): If True, switch_prob is annual and converted to per-timestep
verbose (bool): Print detailed information
**Example**::
# 10% of injectable users switch to DMPA-SC starting in 2025
intv1 = fp.method_switching(
year=2025,
from_methods='inj',
to_method='dmpasc',
switch_prob=0.10,
annual=False
)
# Different rates for different methods
intv2 = fp.method_switching(
year=2030,
from_methods=['dmpasc3', 'withdrawal', 'other_trad'],
to_method='dmpasc6',
switch_prob={'dmpasc3': 0.26, 'withdrawal': 0.20, 'other_trad': 0.20},
annual=False
)
"""
def __init__(self, year=None, from_methods=None, to_method=None, switch_prob=None,
age_groups=None, postpartum=None, annual=False, verbose=False, **kwargs):
super().__init__(**kwargs)
self.year = year
self.from_methods = sc.promotetolist(from_methods)
self.to_method = to_method
self.switch_prob = switch_prob
self.age_groups = age_groups
self.postpartum = postpartum if postpartum is not None else [0, 1, 6]
self.postpartum = sc.promotetolist(self.postpartum)
self.annual = annual
self.verbose = verbose
self.applied = False
return
def init_pre(self, sim=None):
super().init_pre(sim)
# Validate year
if self.year is None:
errormsg = 'year must be specified'
raise ValueError(errormsg)
if self.year < sim.pars['start'] or self.year > sim.pars['stop']:
errormsg = f'Intervention year {self.year} must be within simulation [{sim.pars["start"]}, {sim.pars["stop"]}]'
raise ValueError(errormsg)
# Validate methods
if self.from_methods is None or self.to_method is None:
errormsg = 'Both from_methods and to_method must be specified'
raise ValueError(errormsg)
if self.switch_prob is None:
errormsg = 'switch_prob must be specified'
raise ValueError(errormsg)
# Resolve method names to indices
cm = sim.connectors.contraception
self.from_method_names = []
self.from_method_indices = []
for method in self.from_methods:
if isinstance(method, str):
if method not in cm.methods:
errormsg = f'Source method "{method}" not found'
raise ValueError(errormsg)
self.from_method_names.append(method)
self.from_method_indices.append(cm.methods[method].idx)
else:
self.from_method_indices.append(method)
# Find name from index
for name, m in cm.methods.items():
if m.idx == method:
self.from_method_names.append(name)
break
if isinstance(self.to_method, str):
if self.to_method not in cm.methods:
errormsg = f'Target method "{self.to_method}" not found'
raise ValueError(errormsg)
self.to_method_idx = cm.methods[self.to_method].idx
self.to_method_name = self.to_method
else:
self.to_method_idx = self.to_method
for name, m in cm.methods.items():
if m.idx == self.to_method:
self.to_method_name = name
break
# Convert switch_prob to dict format
if not isinstance(self.switch_prob, dict):
# Same probability for all source methods
self.switch_prob_dict = {name: self.switch_prob for name in self.from_method_names}
else:
self.switch_prob_dict = self.switch_prob
# Convert annual to per-timestep if needed
if self.annual:
for key in self.switch_prob_dict:
annual_prob = self.switch_prob_dict[key]
self.switch_prob_dict[key] = ((1 + annual_prob) ** float(sim.dt)) - 1
# Determine age groups to modify
if self.age_groups is None:
# Get all age groups from method_choice_pars
sample_pp = self.postpartum[0]
if sample_pp in cm.pars.method_choice_pars:
self.age_groups = [k for k in cm.pars.method_choice_pars[sample_pp].keys() if k != 'method_idx']
if self.verbose:
print(f'Method switching intervention initialized:')
print(f' Year: {self.year}')
print(f' From methods: {self.from_method_names}')
print(f' To method: {self.to_method_name}')
print(f' Switch probabilities: {self.switch_prob_dict}')
print(f' Age groups: {self.age_groups}')
print(f' Postpartum states: {self.postpartum}')
return
def step(self):
sim = self.sim
# Apply once when year is reached
if self.applied or sim.t.year < self.year:
return
self.applied = True
cm = sim.connectors.contraception
# Use the existing set_switching_prob method for each source method
for from_name in self.from_method_names:
switch_prob = self.switch_prob_dict.get(from_name, 0)
if switch_prob == 0:
continue
for pp in self.postpartum:
for age_grp in self.age_groups:
# Get current probability of switching from source to target
try:
from_method_key = 'birth' if pp == 1 else from_name
current_prob = cm.get_switching_prob(from_method_key, self.to_method_name, pp, age_grp)
# Increase the switching probability
# This represents: "X% of people who would have stayed on from_method now switch to to_method"
new_prob = current_prob + (1.0 - current_prob) * switch_prob
# Set the new probability (with renormalization)
cm.set_switching_prob(from_method_key, self.to_method_name, new_prob,
postpartum=pp, age_grp=age_grp, renormalize=True)
except (ValueError, KeyError) as e:
# Skip combinations that don't exist in the data
if self.verbose:
print(f' Skipping {from_name}->{self.to_method_name} for pp={pp}, age={age_grp}: {e}')
continue
if self.verbose:
print(f'Year {sim.t.year}: Applied method switching intervention')
return
