import numpy as np
from .general import AgentBaseClass
[docs]
class Government(AgentBaseClass):
"""This class is used to simulate the government.
Args:
model: The GEB model.
agents: The class that includes all agent types (allowing easier communication between agents).
"""
def __init__(self, model, agents):
super().__init__(model)
self.agents = agents
self.config = (
self.model.config["agent_settings"]["government"]
if "government" in self.model.config["agent_settings"]
else {}
)
self.ratio_farmers_to_provide_subsidies_per_year = 0.05
@property
def name(self):
return "agents.government"
def spinup(self) -> None:
pass
def provide_subsidies(self) -> None:
if "subsidies" not in self.config:
return None
if self.model.current_timestep == 1:
for region in self.agents.crop_farmers.borewell_cost_1[1].keys():
self.agents.crop_farmers.borewell_cost_1[1][region] = [
0.5 * x for x in self.agents.crop_farmers.borewell_cost_1[1][region]
]
self.agents.crop_farmers.borewell_cost_2[1][region] = [
0.5 * x for x in self.agents.crop_farmers.borewell_cost_2[1][region]
]
return
def request_flood_cushions(self, reservoirIDs):
pass
def set_irrigation_limit(self) -> None:
if "irrigation_limit" not in self.config:
return None
irrigation_limit = self.config["irrigation_limit"]
if irrigation_limit["per"] == "capita":
self.agents.crop_farmers.var.irrigation_limit_m3[:] = (
self.agents.crop_farmers.var.household_size * irrigation_limit["limit"]
)
elif irrigation_limit["per"] == "area": # limit per m2 of field
self.agents.crop_farmers.var.irrigation_limit_m3[:] = (
self.agents.crop_farmers.field_size_per_farmer
* irrigation_limit["limit"]
)
elif irrigation_limit["per"] == "command_area":
farmer_command_area = self.agents.crop_farmers.command_area
farmers_per_command_area = np.bincount(
farmer_command_area[farmer_command_area != -1],
minlength=self.model.hydrology.lakes_reservoirs.n,
)
# get yearly usable release m3. We do not use the current year, as it
# may not be complete yet, and we only use up to the history fill index
yearly_usable_release_m3_per_command_area = np.full(
self.model.hydrology.lakes_reservoirs.n, np.nan, dtype=np.float32
)
yearly_usable_release_m3_per_command_area[
self.model.hydrology.lakes_reservoirs.is_reservoir
] = (self.agents.reservoir_operators.yearly_usuable_release_m3).mean(axis=1)
irritation_limit_per_command_area = (
yearly_usable_release_m3_per_command_area / farmers_per_command_area
)
# give all farmers there unique irrigation limit
# all farmers without a command area get no irrigation limit (nan)
irrigation_limit_per_farmer = irritation_limit_per_command_area[
farmer_command_area
]
irrigation_limit_per_farmer[farmer_command_area == -1] = np.nan
# make sure all farmers in a command area have an irrigation limit
assert not np.isnan(
irrigation_limit_per_farmer[farmer_command_area != -1]
).any()
self.agents.crop_farmers.var.irrigation_limit_m3[:] = (
irrigation_limit_per_farmer
)
else:
raise NotImplementedError(
"Only 'capita' and 'area' are implemented for irrigation limit"
)
if "min" in irrigation_limit:
self.agents.crop_farmers.irrigation_limit_m3[
self.agents.crop_farmers.irrigation_limit_m3 < irrigation_limit["min"]
] = irrigation_limit["min"]
[docs]
def step(self) -> None:
"""This function is run each timestep."""
self.set_irrigation_limit()
self.provide_subsidies()
self.report(self, locals())