<gh_stars>0
"""
Implement the numeric model
"""


from collections import OrderedDict
from numba import jitclass, float64, int64
import numpy as np

from .costs import calc_insurance_premium, calc_construction_cost
from .exposure import ExposureModel
from .flood import FloodModel
from .sealevel import SeaLevelModel


@jitclass(
    OrderedDict(
        # PARAMETERS
        initial_height_cm=float64,
        construction_quadratic=float64,
        construction_linear=float64,
        construction_fixed=float64,
        # VALUES
        construction_cost_euro=float64[:],
        exposed_value_euro=float64[:],
        insurance_premium_euro=float64[:],
        levee_height_cm=float64[:],
        mean_sea_level_cm=float64[:],
        storm_surge_cm=float64[:],
        year=float64[:],
        index=int64,
        # MODULES
        exposure_model=ExposureModel.class_type.instance_type,  # type: ignore
        flood_model=FloodModel.class_type.instance_type,  # type: ignore
        sea_level_model=SeaLevelModel.class_type.instance_type,  # type: ignore
        # HISTORICAL DATA
        historical_length=int64,
        historical_exposure=float64[:],
        historical_sea_level=float64[:],
        historical_surge=float64[:],
    )
)
class SimulationModel:
    """
    All the number crunching of the model goes here
    
    You won't want to build this by itself, as it takes in as arguments
    a bunch of models that are messy to build. Instead, 
    """

    def __init__(
        self,
        start_year: int,
        end_year: int,
        construction_quadratic: float,
        construction_linear: float,
        construction_fixed: float,
        exposure_model: ExposureModel,
        flood_model: FloodModel,
        sea_level_model: SeaLevelModel,
        historical_sea_level: np.ndarray,
        historical_surge: np.ndarray,
        historical_exposure: np.ndarray,
        initial_height_cm: float,
    ) -> None:
        """
        Build the model
        """

        # get the historical data
        self.historical_length = historical_exposure.size
        assert historical_sea_level.size == self.historical_length
        assert historical_surge.size == self.historical_length
        self.historical_surge = historical_surge.astype(np.float64)
        self.historical_sea_level = historical_sea_level.astype(np.float64)
        self.historical_exposure = historical_exposure.astype(np.float64)

        # initialize values
        self.index = -1
        self.year = np.arange(start_year, end_year + 1).astype(np.float64)
        self.construction_cost_euro = (np.nan * np.ones_like(self.year)).astype(
            np.float64
        )
        self.exposed_value_euro = (np.nan * np.ones_like(self.year)).astype(np.float64)
        self.insurance_premium_euro = (np.nan * np.ones_like(self.year)).astype(
            np.float64
        )
        self.levee_height_cm = (np.nan * np.ones_like(self.year)).astype(np.float64)
        self.mean_sea_level_cm = (np.nan * np.ones_like(self.year)).astype(np.float64)
        self.storm_surge_cm = (np.nan * np.ones_like(self.year)).astype(np.float64)

        # set up the three models of exogenous variables
        self.exposure_model = exposure_model
        self.flood_model = flood_model
        self.sea_level_model = sea_level_model

        # update the exposure model to use historical exposure
        self.exposure_model.exposure_euro = self.historical_exposure[-1]

        # store parameters
        self.construction_quadratic = construction_quadratic
        self.construction_linear = construction_linear
        self.construction_fixed = construction_fixed
        self.initial_height_cm = initial_height_cm

    def step(self, height_increase_cm: float) -> float:
        """
        Take the action, update the environment, and get the reward
        
        IMPORTANT:
        This *will* throw an error unless you manually update some of the values,
        as done in the LeveeEnv
        """

        self.index += 1
        i = self.index  # for shorthand
        year = self.year[i]  # shorthand

        # if historical, we don't need to step the exogenous models forward
        if i < self.historical_length:
            self.levee_height_cm[i] = self.initial_height_cm
            self.mean_sea_level_cm[i] = self.historical_sea_level[i]
            self.storm_surge_cm[i] = self.historical_surge[i]
            self.exposed_value_euro[i] = self.historical_exposure[i]

        # if it's not historical, we need to step the exogenous models forward
        else:
            self.levee_height_cm[i] = self.levee_height_cm[i - 1] + height_increase_cm
            self.mean_sea_level_cm[i] = self.sea_level_model.simulate_msl(year=year)
            self.storm_surge_cm[i] = self.flood_model.simulate_flood(year=year)
            self.exposed_value_euro[i] = self.exposure_model.simulate_exposure()

        # calculate the dependent variables
        flood_prob = self.flood_model.calc_exceedance_prob(
            year=year, height=self.levee_height_cm[i] - self.mean_sea_level_cm[i]
        )
        self.insurance_premium_euro[i] = calc_insurance_premium(
            exposed_value_euro=self.exposed_value_euro[i], flood_prob=flood_prob
        )
        self.construction_cost_euro[i] = calc_construction_cost(
            levee_height_cm=self.levee_height_cm[i],
            height_increase_cm=height_increase_cm,
            construction_quadratic=self.construction_quadratic,
            construction_linear=self.construction_linear,
            construction_fixed=self.construction_fixed,
        )

        # done, calculate reward
        total_cost = self.insurance_premium_euro[i] + self.construction_cost_euro[i]
        reward = -total_cost

        return reward

"""
Implement the numeric model
"""


from collections import OrderedDict
from numba import jitclass, float64, int64
import numpy as np

from .costs import calc_insurance_premium, calc_construction_cost
from .exposure import ExposureModel
from .flood import FloodModel
from .sealevel import SeaLevelModel


@jitclass(
    OrderedDict(
        # PARAMETERS
        initial_height_cm=float64,
        construction_quadratic=float64,
        construction_linear=float64,
        construction_fixed=float64,
        # VALUES
        construction_cost_euro=float64[:],
        exposed_value_euro=float64[:],
        insurance_premium_euro=float64[:],
        levee_height_cm=float64[:],
        mean_sea_level_cm=float64[:],
        storm_surge_cm=float64[:],
        year=float64[:],
        index=int64,
        # MODULES
        exposure_model=ExposureModel.class_type.instance_type,  # type: ignore
        flood_model=FloodModel.class_type.instance_type,  # type: ignore
        sea_level_model=SeaLevelModel.class_type.instance_type,  # type: ignore
        # HISTORICAL DATA
        historical_length=int64,
        historical_exposure=float64[:],
        historical_sea_level=float64[:],
        historical_surge=float64[:],
    )
)
class SimulationModel:
    """
    All the number crunching of the model goes here
    
    You won't want to build this by itself, as it takes in as arguments
    a bunch of models that are messy to build. Instead, 
    """

    def __init__(
        self,
        start_year: int,
        end_year: int,
        construction_quadratic: float,
        construction_linear: float,
        construction_fixed: float,
        exposure_model: ExposureModel,
        flood_model: FloodModel,
        sea_level_model: SeaLevelModel,
        historical_sea_level: np.ndarray,
        historical_surge: np.ndarray,
        historical_exposure: np.ndarray,
        initial_height_cm: float,
    ) -> None:
        """
        Build the model
        """

        # get the historical data
        self.historical_length = historical_exposure.size
        assert historical_sea_level.size == self.historical_length
        assert historical_surge.size == self.historical_length
        self.historical_surge = historical_surge.astype(np.float64)
        self.historical_sea_level = historical_sea_level.astype(np.float64)
        self.historical_exposure = historical_exposure.astype(np.float64)

        # initialize values
        self.index = -1
        self.year = np.arange(start_year, end_year + 1).astype(np.float64)
        self.construction_cost_euro = (np.nan * np.ones_like(self.year)).astype(
            np.float64
        )
        self.exposed_value_euro = (np.nan * np.ones_like(self.year)).astype(np.float64)
        self.insurance_premium_euro = (np.nan * np.ones_like(self.year)).astype(
            np.float64
        )
        self.levee_height_cm = (np.nan * np.ones_like(self.year)).astype(np.float64)
        self.mean_sea_level_cm = (np.nan * np.ones_like(self.year)).astype(np.float64)
        self.storm_surge_cm = (np.nan * np.ones_like(self.year)).astype(np.float64)

        # set up the three models of exogenous variables
        self.exposure_model = exposure_model
        self.flood_model = flood_model
        self.sea_level_model = sea_level_model

        # update the exposure model to use historical exposure
        self.exposure_model.exposure_euro = self.historical_exposure[-1]

        # store parameters
        self.construction_quadratic = construction_quadratic
        self.construction_linear = construction_linear
        self.construction_fixed = construction_fixed
        self.initial_height_cm = initial_height_cm

    def step(self, height_increase_cm: float) -> float:
        """
        Take the action, update the environment, and get the reward
        
        IMPORTANT:
        This *will* throw an error unless you manually update some of the values,
        as done in the LeveeEnv
        """

        self.index += 1
        i = self.index  # for shorthand
        year = self.year[i]  # shorthand

        # if historical, we don't need to step the exogenous models forward
        if i < self.historical_length:
            self.levee_height_cm[i] = self.initial_height_cm
            self.mean_sea_level_cm[i] = self.historical_sea_level[i]
            self.storm_surge_cm[i] = self.historical_surge[i]
            self.exposed_value_euro[i] = self.historical_exposure[i]

        # if it's not historical, we need to step the exogenous models forward
        else:
            self.levee_height_cm[i] = self.levee_height_cm[i - 1] + height_increase_cm
            self.mean_sea_level_cm[i] = self.sea_level_model.simulate_msl(year=year)
            self.storm_surge_cm[i] = self.flood_model.simulate_flood(year=year)
            self.exposed_value_euro[i] = self.exposure_model.simulate_exposure()

        # calculate the dependent variables
        flood_prob = self.flood_model.calc_exceedance_prob(
            year=year, height=self.levee_height_cm[i] - self.mean_sea_level_cm[i]
        )
        self.insurance_premium_euro[i] = calc_insurance_premium(
            exposed_value_euro=self.exposed_value_euro[i], flood_prob=flood_prob
        )
        self.construction_cost_euro[i] = calc_construction_cost(
            levee_height_cm=self.levee_height_cm[i],
            height_increase_cm=height_increase_cm,
            construction_quadratic=self.construction_quadratic,
            construction_linear=self.construction_linear,
            construction_fixed=self.construction_fixed,
        )

        # done, calculate reward
        total_cost = self.insurance_premium_euro[i] + self.construction_cost_euro[i]
        reward = -total_cost

        return reward
Implement the numeric model # PARAMETERS # VALUES # MODULES # type: ignore # type: ignore # type: ignore # HISTORICAL DATA All the number crunching of the model goes here          You won't want to build this by itself, as it takes in as arguments     a bunch of models that are messy to build. Instead, Build the model # get the historical data # initialize values # set up the three models of exogenous variables # update the exposure model to use historical exposure # store parameters Take the action, update the environment, and get the reward                  IMPORTANT:         This *will* throw an error unless you manually update some of the values,         as done in the LeveeEnv # for shorthand # shorthand # if historical, we don't need to step the exogenous models forward # if it's not historical, we need to step the exogenous models forward # calculate the dependent variables # done, calculate reward