input_damages

Calculate damages from the projection system using VSL

dscim.preprocessing.input_damages.calculate_energy_impacts

calculate_energy_impacts(input_path, file_prefix, variable)

Calculate impacts for labor results for individual modeling unit.

Read in individual damages files from the labor projection system output and re-index to add region dimension. This is needed to adjust the projection file outcomes that do not have a region dimension

Paramemters

input_path str Path to model/gcm/iam/rcp/ folder, usually from the _parse_projection_filesys function. file_prefix str Prefix of the MC output filenames variable str Variable to use within xr.Dataset

Returns:

Type	Description
xr.Dataset object with per-capita monetary damages

Source code in src/dscim/preprocessing/input_damages.py

def calculate_energy_impacts(input_path, file_prefix, variable):
    """Calculate impacts for labor results for individual modeling unit.

    Read in individual damages files from the labor projection system output
    and re-index to add region dimension. This is needed to adjust the
    projection file outcomes that do not have a region dimension

    Paramemters
    ----------
    input_path str
        Path to model/gcm/iam/rcp/ folder, usually from the
        `_parse_projection_filesys` function.
    file_prefix str
        Prefix of the MC output filenames
    variable str
        Variable to use within `xr.Dataset`

    Returns
    -------
        xr.Dataset object with per-capita monetary damages
    """

    damages_delta = xr.open_dataset(f"{input_path}/{file_prefix}_delta.nc4").sel(
        year=slice(2010, 2099)
    )
    damages_histclim = xr.open_dataset(f"{input_path}/{file_prefix}_histclim.nc4").sel(
        year=slice(2010, 2099)
    )

    # generate the histclim variable
    impacts = damages_histclim[variable].to_dataset(name=f"histclim_{variable}")
    # generate the delta variable
    impacts[f"delta_{variable}"] = damages_delta[variable]

    return impacts

dscim.preprocessing.input_damages.calculate_labor_impacts

calculate_labor_impacts(input_path, file_prefix, variable, val_type)

Calculate impacts for labor results.

Paramemters

input_path str Path to model/gcm/iam/rcp/ folder, usually from the _parse_projection_filesys function. file_prefix str Prefix of the MC output filenames variable str Variable to use within xr.Dataset val_type str Valuation type.

Returns:

Type	Description
xr.Dataset object with per-capita monetary damages

Source code in src/dscim/preprocessing/input_damages.py

def calculate_labor_impacts(input_path, file_prefix, variable, val_type):
    """Calculate impacts for labor results.

    Paramemters
    ----------
    input_path str
        Path to model/gcm/iam/rcp/ folder, usually from the
        `_parse_projection_filesys` function.
    file_prefix str
        Prefix of the MC output filenames
    variable str
        Variable to use within `xr.Dataset`
    val_type str
        Valuation type.

    Returns
    -------
        xr.Dataset object with per-capita monetary damages
    """

    damages_val = xr.open_dataset(f"{input_path}/{file_prefix}-{val_type}.nc4").sel(
        year=slice(2010, 2099)
    )
    damages_histclim = xr.open_dataset(
        f"{input_path}/{file_prefix}-histclim-{val_type}.nc4"
    ).sel(year=slice(2010, 2099))

    # labour needs indexing assigned
    damages_val_index = damages_val.assign_coords({"region": damages_val.regions})

    damages_histclim_index = damages_histclim.assign_coords(
        {"region": damages_histclim.regions}
    )

    # calculate the delta output
    impacts_delta = damages_val_index[variable] - damages_histclim_index[variable]

    # generate the histclim variable
    impacts = damages_histclim_index[variable].to_dataset(name=f"histclim_{variable}")
    # generate the delta variable
    impacts[f"delta_{variable}"] = impacts_delta

    return impacts

dscim.preprocessing.input_damages.compute_ag_damages

compute_ag_damages(input_path, save_path, pop, varname, query='exists==True', topcode=None, scalar=None, integration=False, batches=range(0, 15), num_cpus=15, file='/disaggregated_damages.nc4', vars=None, min_year=2010, max_year=2099)

Reshapes ag estimate runs for use in integration system, then converts to negative per capita damages.

Parameters:

Name	Type	Description	Default
input_path		Path to NetCDF4 files to be reshaped.	required
file		Name of files to be globbed.	'/disaggregated_damages.nc4'
integration		If True, will format files to be integrated with other sectors.	False
pop		Population data to convert ag damages into per capita terms.	required
save_path		Path where files should be saved	required

Source code in src/dscim/preprocessing/input_damages.py

def compute_ag_damages(
    input_path,
    save_path,
    pop,
    varname,
    query="exists==True",
    topcode=None,
    scalar=None,
    integration=False,
    batches=range(0, 15),
    num_cpus=15,
    file="/disaggregated_damages.nc4",
    vars=None,
    min_year=2010,
    max_year=2099,
):
    """
    Reshapes ag estimate runs for use in integration system,
    then converts to negative per capita damages.

    Parameters
    ----------
    input_path str or list
        Path to NetCDF4 files to be reshaped.
    file str
        Name of files to be globbed.
    integration bool
        If True, will format files to be integrated with other sectors.
    pop xr.DataArray
        Population data to convert ag damages into per capita terms.
    save_path str
        Path where files should be saved
    """
    if vars is None:
        vars = ["wc_no_reallocation"]

    if integration:
        assert (
            topcode is not None
        ), "Data is being processed for integration. Please pass a topcode."

    if isinstance(input_path, str):
        input_path = [input_path]

    paths = []
    for f in input_path:
        p = _parse_projection_filesys(input_path=f, query=query)
        paths.append(p)

    paths = pd.concat(paths)
    paths["path"] = paths.path + file

    def prep(ds):
        ds.coords["model"] = ds.model.str.replace("high", "OECD Env-Growth")
        ds.coords["model"] = ds.model.str.replace("low", "IIASA GDP")
        return ds

    def process_batch(g):
        i = g.batch.values[0]
        if i in [f"batch{j}" for j in batches]:
            print(f"Processing damages in {i}")
            ds = xr.open_mfdataset(g.path, preprocess=prep, parallel=True)[vars]
            attrs = ds.attrs
            ds = ds.sel({"year": slice(min_year, max_year)})
            # ag has missing 2099 damages so we fill these in with the 2098 damages
            ds = ds.reindex(year=range(min_year, max_year + 1), method="ffill")

            # squeeze ag-specific dimensions out so that
            # it can be stacked with other sectors
            if integration:
                ds = ds.sel(demand_topcode=topcode)
                print(f"Selecting topcode {topcode}.")
                for var in [
                    "continent",
                    "iam",
                    "Es_Ed",
                    "market_level",
                    "demand_topcode",
                ]:
                    if var in ds.coords:
                        attrs[var] = ds[var].values
                        ds = ds.drop(var)

            # get in per capita 2019 PPP-adjusted USD damages
            ds = (ds / pop.load()) * -1 * 1.273526

            # replace infinite values with missings
            for var in ds.keys():
                ds[var] = xr.where(np.isinf(ds[var]), np.nan, ds[var])

            if scalar is not None:
                print("Scaling for reallocation.")
                ds["wc_reallocation"] = ds["wc_no_reallocation"] * scalar

            ds.attrs = attrs
            return ds
        else:
            print(f"Skipped {i}.")

    batches = p_map(
        process_batch, [g for i, g in paths.groupby("batch")], num_cpus=num_cpus
    )
    batches = [ds for ds in batches if ds is not None]
    chunkies = {
        "rcp": 1,
        "region": -1,
        "gcm": 1,
        "year": 10,
        "model": 1,
        "ssp": 1,
        "batch": 15,
    }
    batches = (
        xr.concat(batches, "batch", combine_attrs="override")
        .chunk(chunkies)
        .drop("variable")
        .squeeze()
    )
    batches = xr.where(np.isinf(batches), np.nan, batches)
    batches = batches.astype(np.float32)

    batches.rename({"wc_reallocation": varname})[varname].to_dataset().to_zarr(
        store=save_path, mode="a", consolidated=True
    )

dscim.preprocessing.input_damages.concatenate_damage_output

concatenate_damage_output(damage_dir, basename, save_path)

Concatenate labor/energy damage output across batches.

Parameters:

Name	Type	Description	Default
damage_dir		Directory containing separate labor/energy damage output files by batches.	required
basename		Prefix of the damage output filenames (ex. {basename}_batch0.zarr)	required
save_path		Path to save concatenated file in .zarr format	required

Source code in src/dscim/preprocessing/input_damages.py

def concatenate_damage_output(damage_dir, basename, save_path):
    """Concatenate labor/energy damage output across batches.

    Parameters
    ----------
    damage_dir str
        Directory containing separate labor/energy damage output files by batches.
    basename str
        Prefix of the damage output filenames (ex. {basename}_batch0.zarr)
    save_path str
        Path to save concatenated file in .zarr format
    """
    paths = [
        f"{damage_dir}/{basename}_{b}.zarr"
        for b in ["batch" + str(i) for i in range(0, 15)]
    ]
    data = xr.open_mfdataset(paths=paths, engine="zarr")

    for v in data:
        del data[v].encoding["chunks"]

    chunkies = {
        "batch": 15,
        "rcp": 1,
        "gcm": 1,
        "model": 1,
        "ssp": 1,
        "region": -1,
        "year": 10,
    }

    data = data.chunk(chunkies)

    for v in list(data.coords.keys()):
        if data.coords[v].dtype == object:
            data.coords[v] = data.coords[v].astype("unicode")
    data.coords["batch"] = data.coords["batch"].astype("unicode")
    for v in list(data.variables.keys()):
        if data[v].dtype == object:
            data[v] = data[v].astype("unicode")

    data.to_zarr(save_path, mode="w")

dscim.preprocessing.input_damages.concatenate_energy_damages

concatenate_energy_damages(input_path, save_path, ec_cls, file_prefix='TINV_clim_integration_total_energy', variable='rebased', format_file='netcdf', **kwargs)

Concatenate damages across batches and create a lazy array for future calculations.

Using the value_mortality_damages function this function lazily loads all damages for SCC calculations and scale damage to per-capital damages and also scale labor inpacts to indicate increasing damages as positive, and gains from warming as negative.

Parameters:

Name	Type	Description	Default
input_path		Directory containing all raw projection output.	required
ec_cls		EconVars class with population and GDP data to rescale damages	required
save_path		Path to save concatenated file in .zarr or .nc4 format	required
file_prefix		Prefix of the MC output filenames	'TINV_clim_integration_total_energy'
variables		Variable names to extract from calculated damages	required
format_file		Format to save file. Options are 'netcdf' or 'zarr'	'netcdf'
**kwargs		Other options passed to the value_mortality_damages function and the _parse_projection_filesys: query="rcp=='rcp45'"	{}

Source code in src/dscim/preprocessing/input_damages.py

def concatenate_energy_damages(
    input_path,
    save_path,
    ec_cls,
    file_prefix="TINV_clim_integration_total_energy",
    variable="rebased",
    format_file="netcdf",
    **kwargs,
):
    """Concatenate damages across batches and create a lazy array for future
    calculations.

    Using the `value_mortality_damages` function this function lazily loads all
    damages for SCC calculations and scale damage to per-capital damages and
    also scale labor inpacts to indicate increasing damages as positive, and
    gains from warming as negative.

    Parameters
    ----------
    input_path str
        Directory containing all raw projection output.
    ec_cls dscim.simple_storage.EconVars
        EconVars class with population and GDP data to rescale damages
    save_path str
        Path to save concatenated file in .zarr or .nc4 format
    file_prefix str
        Prefix of the MC output filenames
    variables list
        Variable names to extract from calculated damages
    format_file str
        Format to save file. Options are 'netcdf' or 'zarr'
    **kwargs
        Other options passed to the `value_mortality_damages` function
        and the `_parse_projection_filesys`: query="rcp=='rcp45'"
    """

    # Load df with paths
    df = _parse_projection_filesys(
        input_path=input_path, query=kwargs.get("query", "exists==True")
    )

    # Process files by batches and save as .zarr files
    for i, g in df.groupby("batch"):
        logger.info(f"Processing damages in batch: {i}")
        list_damages_batch = []
        for idx, row in g.iterrows():
            try:
                ds = calculate_energy_impacts(
                    input_path=row.path,
                    file_prefix=file_prefix,
                    variable=variable,
                )
                list_damages_batch.append(ds)

            except Exception as e:
                logger.error(f"Error in batch{i}: {e}")
                pass

        # Concatenate file within batch
        conversion_value = 1.273526
        concat_ds = xr.combine_by_coords(list_damages_batch)
        for v in [f"histclim_{variable}", f"delta_{variable}"]:
            concat_ds[v] = (
                concat_ds[v] / ec_cls.econ_vars.pop.load()
            ) * conversion_value

        # Save file
        file_name = f"{variable}_{i}"
        path_to_file = os.path.join(save_path, file_name)

        # convert to float32
        concat_ds = concat_ds.astype(np.float32)
        logger.info(f"Concatenating and processing {i}")

        if format_file == "zarr":
            to_store = concat_ds.copy()
            for var in to_store.variables:
                to_store[var].encoding.clear()

            to_store.to_zarr(f"{path_to_file}.zarr", mode="w", consolidated=True)
        elif format_file == "netcdf":
            to_store = concat_ds.copy()
            for var in to_store.variables:
                to_store[var].encoding.clear()

            to_store.to_netcdf(f"{path_to_file}.nc4")

    return concat_ds

dscim.preprocessing.input_damages.concatenate_labor_damages

concatenate_labor_damages(input_path, save_path, ec_cls, file_prefix='uninteracted_main_model', variable='rebased', val_type='wage-levels', format_file='netcdf', **kwargs)

Concatenate damages across batches.

Parameters:

Name	Type	Description	Default
input_path		Directory containing all raw projection output.	required
ec_cls		EconVars class with population and GDP data to rescale damages	required
save_path		Path to save concatenated file in .zarr or .nc4 format	required
file_prefix		Prefix of the MC output filenames	'uninteracted_main_model'
variables		Variable names to extract from calculated damages	required
format_file		Format to save file. Options are 'netcdf' or 'zarr'	'netcdf'
**kwargs		Other options passed to the _parse_projection_filesys: query="rcp=='rcp45'"	{}

Source code in src/dscim/preprocessing/input_damages.py

def concatenate_labor_damages(
    input_path,
    save_path,
    ec_cls,
    file_prefix="uninteracted_main_model",
    variable="rebased",
    val_type="wage-levels",
    format_file="netcdf",
    **kwargs,
):
    """Concatenate damages across batches.

    Parameters
    ----------
    input_path str
        Directory containing all raw projection output.
    ec_cls dscim.simple_storage.EconVars
        EconVars class with population and GDP data to rescale damages
    save_path str
        Path to save concatenated file in .zarr or .nc4 format
    file_prefix str
        Prefix of the MC output filenames
    variables list
        Variable names to extract from calculated damages
    format_file str
        Format to save file. Options are 'netcdf' or 'zarr'
    **kwargs
        Other options passed to the `_parse_projection_filesys`: query="rcp=='rcp45'"
    """

    # Load df with paths
    df = _parse_projection_filesys(
        input_path=input_path, query=kwargs.get("query", "exists==True")
    )

    # Process files by batches and save as .zarr files
    for i, g in df.groupby("batch"):
        logger.info(f"Processing damages in batch: {i}")
        list_damages_batch = []
        for idx, row in g.iterrows():
            try:
                ds = calculate_labor_impacts(
                    input_path=row.path,
                    file_prefix=file_prefix,
                    variable=variable,
                    val_type=val_type,
                )
                ds = ds.assign_coords(
                    {
                        "ssp": row.ssp,
                        "rcp": row.rcp,
                        "gcm": row.gcm,
                        "model": row.iam,
                        "batch": row.batch,
                    }
                )

                ds_exp = ds.expand_dims(["ssp", "rcp", "model", "gcm", "batch"])
                list_damages_batch.append(ds_exp)

            except Exception as e:
                logger.error(f"Error in batch{i}: {e}")
                pass

        # Concatenate file within batch
        conversion_value = 1.273526
        concat_ds = xr.combine_by_coords(list_damages_batch)
        for v in [f"histclim_{variable}", f"delta_{variable}"]:
            concat_ds[v] = (
                (concat_ds[v] / ec_cls.econ_vars.pop.load()) * -1 * conversion_value
            )

        # Save file
        file_name = f"{variable}_{val_type}_{i}"
        path_to_file = os.path.join(save_path, file_name)

        # convert to float32
        concat_ds = concat_ds.astype(np.float32)
        logger.info(f"Concatenating and processing {i}")

        # save out
        if format_file == "zarr":
            to_store = concat_ds.copy()
            for var in to_store.variables:
                to_store[var].encoding.clear()

            to_store.to_zarr(f"{path_to_file}.zarr", mode="w", consolidated=True)
        elif format_file == "netcdf":
            concat_ds.to_netcdf(f"{path_to_file}.nc4")

    return concat_ds

dscim.preprocessing.input_damages.read_energy_files

read_energy_files(df, seed='TINV_clim_price014_total_energy_fulladapt-histclim')

Read energy CSV files and trasnform them to Xarray objects

This function reads a dataframe with the filesystem metadata (from _parse_projection_filesys) to read all CSV files in it with the desired seed and transform to xarray object adding the directory metadata as new dimensions, this will be helpful for data concatenation.

This function is parallelized by read_energy_files_parallel

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame with projection system metadata by batch/RCP/IAM/GCM	required

Returns:

Type	Description
None	Saved data array with expanded damages from original CSV

Source code in src/dscim/preprocessing/input_damages.py

def read_energy_files(df, seed="TINV_clim_price014_total_energy_fulladapt-histclim"):
    """Read energy CSV files and trasnform them to Xarray objects

    This function reads a dataframe with the filesystem metadata (from
    ``_parse_projection_filesys``) to read all CSV files in it with the desired
    ``seed`` and transform to xarray object adding the directory metadata as
    new dimensions, this will be helpful for data concatenation.

    This function is parallelized by ``read_energy_files_parallel``

    Parameters
    ---------
    df : pd.DataFrame
        DataFrame with projection system metadata by batch/RCP/IAM/GCM

    Returns
    -------
    None
        Saved data array with expanded damages from original CSV
    """

    for idx, row in df.iterrows():
        path = os.path.join(row.path, f"{seed}.csv")
        try:
            damages = pd.read_csv(path)
            damages = damages[damages.year >= 2010]
            damages_arr = damages.set_index(["region", "year"]).to_xarray()

            # Add dims to array
            damages_arr = damages_arr.expand_dims(
                {
                    "batch": [row.batch],
                    "rcp": [row.rcp],
                    "gcm": [row.gcm],
                    "model": [row.iam],
                    "ssp": [row.ssp],
                }
            )

        except Exception as e:
            logger.error(f"Error in file {row.path}: {e}")
            pass

        damages_arr.to_netcdf(os.path.join(row.path, f"{seed}.nc4"))

    return None

dscim.preprocessing.input_damages.read_energy_files_parallel

read_energy_files_parallel(input_path, **kwargs)

Concatenate energy results from CSV to NetCDF by batches using multiprocessing

This function takes all CSV files per batch and maps the read_energy_files function to all the files within a batch. The files will be saved in the same path as the CSV files but in NetCDF format.

Once saved, the files will be read again, using a Dask Client and chunked to be finally saved as files per batch. Before saving, the function will calculate damages per capita using SSP populations for each scenario with a EconVars class and then calculate 2019 USD damages

Parameters:

Name	Type	Description	Default
input_path	str	Path to root folder organized by batch containing all projection system files	required
**kwargs		Other elements too the read_energy_files damages	{}

Returns:

Type	Description
None

Source code in src/dscim/preprocessing/input_damages.py

def read_energy_files_parallel(input_path, **kwargs):
    """Concatenate energy results from CSV to NetCDF by batches using
    multiprocessing

    This function takes all CSV files per batch and maps the
    ``read_energy_files`` function to all the files within a batch. The files
    will be saved in the same path as the CSV files but in NetCDF format.

    Once saved, the files will be read again, using a Dask ``Client`` and
    chunked to be finally saved as files per batch. Before saving, the function
    will calculate damages per capita using SSP populations for each scenario
    with a ``EconVars`` class and then calculate 2019 USD damages

    Parameters
    ----------
    input_path : str
        Path to root folder organized by batch containing all projection system
        files
    **kwargs
        Other elements too the ``read_energy_files`` damages

    Returns
    ------
        None
    """

    # Start reading and save to NetCDF
    for i in range(0, 15):
        logger.info(f"Processing damages in batch: {i}")
        # Read files available
        df = _parse_projection_filesys(
            input_path=input_path, query=f"exists==True&batch=='batch{i}'"
        )
        # Calculate the chunk size as an integer
        num_processes = multiprocessing.cpu_count()
        chunk_size = int(df.shape[0] / num_processes)
        logger.info(
            f"Starting multiprocessing in {num_processes} cores with {chunk_size} rows each"
        )
        chunks = [
            df.iloc[i : i + chunk_size, :] for i in range(0, df.shape[0], chunk_size)
        ]

        with multiprocessing.Pool(processes=num_processes) as p:
            result = p.map(partial(read_energy_files, **kwargs), chunks)

    return result