data.datasets.electricity_demand_timeseries.cts_buildings.buildings_with_amenities()

Complexity

Conditions

5

Size

Total Lines	183
Code Lines	103

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

"""
def buildings_with_amenities():
    """
    Amenities which are assigned to buildings are determined
    and grouped per building and zensus cell. Buildings
    covering multiple cells therefore exists multiple times
    but in different zensus cells. This is necessary to cover
    all cells with a cts demand. If buildings exist in multiple
    substations, their amenities are summed and assigned and kept in
    one substation only. If as a result, a census cell is uncovered,
    a synthetic amenity is placed. The buildings are aggregated
    afterwards during the calculation of the profile_share.

    Returns
    -------
    df_buildings_with_amenities: gpd.GeoDataFrame
        Contains all buildings with amenities per zensus cell.
    df_lost_cells: gpd.GeoDataFrame
        Contains synthetic amenities in lost cells. Might be empty
    """

    from saio.openstreetmap import osm_amenities_in_buildings_filtered

    with db.session_scope() as session:
        cells_query = (
            session.query(
                osm_amenities_in_buildings_filtered,
                MapZensusGridDistricts.bus_id,
            )
            .filter(
                MapZensusGridDistricts.zensus_population_id
                == osm_amenities_in_buildings_filtered.zensus_population_id
            )
            .filter(
                EgonDemandRegioZensusElectricity.zensus_population_id
                == osm_amenities_in_buildings_filtered.zensus_population_id
            )
            .filter(
                EgonDemandRegioZensusElectricity.sector == "service",
                EgonDemandRegioZensusElectricity.scenario == "eGon2035",
            )
        )
    df_amenities_in_buildings = pd.read_sql(
        cells_query.statement, cells_query.session.bind, index_col=None
    )

    df_amenities_in_buildings["geom_building"] = df_amenities_in_buildings[
        "geom_building"
    ].apply(to_shape)
    df_amenities_in_buildings["geom_amenity"] = df_amenities_in_buildings[
        "geom_amenity"
    ].apply(to_shape)

    df_amenities_in_buildings["n_amenities_inside"] = 1

    # add identifier column for buildings in multiple substations
    df_amenities_in_buildings[
        "duplicate_identifier"
    ] = df_amenities_in_buildings.groupby(["id", "bus_id"])[
        "n_amenities_inside"
    ].transform(
        "cumsum"
    )
    df_amenities_in_buildings = df_amenities_in_buildings.sort_values(
        ["id", "duplicate_identifier"]
    )
    # sum amenities of buildings with multiple substations
    df_amenities_in_buildings[
        "n_amenities_inside"
    ] = df_amenities_in_buildings.groupby(["id", "duplicate_identifier"])[
        "n_amenities_inside"
    ].transform(
        "sum"
    )

    # create column to always go for bus_id with max amenities
    df_amenities_in_buildings[
        "max_amenities"
    ] = df_amenities_in_buildings.groupby(["id", "bus_id"])[
        "n_amenities_inside"
    ].transform(
        "sum"
    )
    # sort to go for
    df_amenities_in_buildings.sort_values(
        ["id", "max_amenities"], ascending=False, inplace=True
    )

    # identify lost zensus cells
    df_lost_cells = df_amenities_in_buildings.loc[
        df_amenities_in_buildings.duplicated(
            subset=["id", "duplicate_identifier"], keep="first"
        )
    ]
    df_lost_cells.drop_duplicates(
        subset=["zensus_population_id"], inplace=True
    )

    # drop buildings with multiple substation and lower max amenity
    df_amenities_in_buildings.drop_duplicates(
        subset=["id", "duplicate_identifier"], keep="first", inplace=True
    )

    # check if lost zensus cells are already covered
    if not df_lost_cells.empty:
        if not (
            df_amenities_in_buildings["zensus_population_id"]
            .isin(df_lost_cells["zensus_population_id"])
            .empty
        ):
            # query geom data for cell if not
            with db.session_scope() as session:
                cells_query = session.query(
                    DestatisZensusPopulationPerHa.id,
                    DestatisZensusPopulationPerHa.geom,
                ).filter(
                    DestatisZensusPopulationPerHa.id.in_(
                        df_lost_cells["zensus_population_id"]
                    )
                )

            df_lost_cells = gpd.read_postgis(
                cells_query.statement,
                cells_query.session.bind,
                geom_col="geom",
            )
            # TODO maybe adapt method
            # place random amenity in cell
            df_lost_cells["n_amenities_inside"] = 1
            df_lost_cells.rename(
                columns={
                    "id": "zensus_population_id",
                },
                inplace=True,
            )
            df_lost_cells = place_buildings_with_amenities(
                df_lost_cells, amenities=1
            )
            df_lost_cells.rename(
                columns={
                    # "id": "zensus_population_id",
                    "geom_point": "geom_amenity",
                },
                inplace=True,
            )
            df_lost_cells.drop(
                columns=["building_count", "n_amenities_inside"], inplace=True
            )
        else:
            df_lost_cells = None
    else:
        df_lost_cells = None

    # drop helper columns
    df_amenities_in_buildings.drop(
        columns=["duplicate_identifier"], inplace=True
    )

    # sum amenities per building and cell
    df_amenities_in_buildings[
        "n_amenities_inside"
    ] = df_amenities_in_buildings.groupby(["zensus_population_id", "id"])[
        "n_amenities_inside"
    ].transform(
        "sum"
    )
    # drop duplicated buildings
    df_buildings_with_amenities = df_amenities_in_buildings.drop_duplicates(
        ["id", "zensus_population_id"]
    )
    df_buildings_with_amenities.reset_index(inplace=True, drop=True)

    df_buildings_with_amenities = df_buildings_with_amenities[
        ["id", "zensus_population_id", "geom_building", "n_amenities_inside"]
    ]
    df_buildings_with_amenities.rename(
        columns={
            # "zensus_population_id": "cell_id",
            "egon_building_id": "id"
        },
        inplace=True,
    )

    return df_buildings_with_amenities, df_lost_cells