data.datasets.electricity_demand_timeseries.hh_profiles.houseprofiles_in_census_cells()

Complexity

Conditions

3

Size

Total Lines	115
Code Lines	46

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

"""
def houseprofiles_in_census_cells():
    """
    Allocate household electricity demand profiles for each census cell.

    Creates table `emand.egon_household_electricity_profile_in_census_cell` that maps
    household electricity demand profiles to census cells. Each row represents one cell
    and contains a list of profile IDs. This table is fundamental
    for creating subsequent data like demand profiles on MV grid level or for
    determining the peak load at load area level.

    Use :func:`get_houseprofiles_in_census_cells` to retrieve the data from
    the database as pandas.

    """

    def gen_profile_names(n):
        """Join from Format (str),(int) to (str)a000(int)"""
        a = f"{n[0]}a{int(n[1]):05d}"
        return a

    # Init random generators using global seed
    random.seed(RANDOM_SEED)
    np.random.seed(RANDOM_SEED)

    # Read demand profiles from egon-data-bundle
    df_iee_profiles = get_iee_hh_demand_profiles_raw()

    # Write raw profiles into db
    write_hh_profiles_to_db(df_iee_profiles)

    # Process profiles for further use
    df_iee_profiles = set_multiindex_to_profiles(df_iee_profiles)

    # Download zensus household NUTS-1 data with family type and age categories
    df_census_households_nuts1_raw = get_census_households_nuts1_raw()

    # Reduce age intervals and remove kids
    df_census_households_nuts1 = process_nuts1_census_data(
        df_census_households_nuts1_raw
    )

    # Regroup data to be compatible with categories from demand profile
    # generator.
    df_census_households_nuts1 = regroup_nuts1_census_data(
        df_census_households_nuts1
    )

    # Convert data from people living in households to households
    # Using a specified amount of inhabitants per household type
    df_census_households_nuts1 = inhabitants_to_households(
        df_census_households_nuts1
    )

    # Query census household grid data with family type
    df_census_households_grid = get_census_households_grid()

    # fill cells with missing household distribution values but population
    # by hh distribution value of random cell with same population value
    df_census_households_grid = impute_missing_hh_in_populated_cells(
        df_census_households_grid
    )

    # Refine census household grid data with additional NUTS-1 level attributes
    df_census_households_grid_refined = refine_census_data_at_cell_level(
        df_census_households_grid, df_census_households_nuts1
    )

    write_refinded_households_to_db(df_census_households_grid_refined)

    # Allocate profile ids to each cell by census data
    df_hh_profiles_in_census_cells = assign_hh_demand_profiles_to_cells(
        df_census_households_grid_refined, df_iee_profiles
    )

    # Annual household electricity demand on NUTS-3 level (demand regio)
    df_demand_regio = db.select_dataframe(
        sql="""
                SELECT year, nuts3, SUM (demand) as demand_mWha
                FROM demand.egon_demandregio_hh as egon_d
                GROUP BY nuts3, year
                ORDER BY year""",
        index_col=["year", "nuts3"],
    )

    # Scale profiles to meet demand regio annual demand projections
    df_hh_profiles_in_census_cells = adjust_to_demand_regio_nuts3_annual(
        df_hh_profiles_in_census_cells, df_iee_profiles, df_demand_regio
    )

    df_hh_profiles_in_census_cells = (
        df_hh_profiles_in_census_cells.reset_index(drop=False)
    )
    df_hh_profiles_in_census_cells["cell_id"] = df_hh_profiles_in_census_cells[
        "cell_id"
    ].astype(int)

    # Cast profile ids back to initial str format
    df_hh_profiles_in_census_cells[
        "cell_profile_ids"
    ] = df_hh_profiles_in_census_cells["cell_profile_ids"].apply(
        lambda x: list(map(gen_profile_names, x))
    )

    # Write allocation table into database
    HouseholdElectricityProfilesInCensusCells.__table__.drop(
        bind=engine, checkfirst=True
    )
    HouseholdElectricityProfilesInCensusCells.__table__.create(
        bind=engine, checkfirst=True
    )

    with db.session_scope() as session:
        session.bulk_insert_mappings(
            HouseholdElectricityProfilesInCensusCells,
            df_hh_profiles_in_census_cells.to_dict(orient="records"),
        )