data.datasets.emobility.motorized_individual_travel.model_timeseries.write_model_data_to_db()

Complexity

Conditions

15

Size

Total Lines	348
Code Lines	225

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

"""
def write_model_data_to_db(
    static_params_dict: dict,
    load_time_series_df: pd.DataFrame,
    bus_id: int,
    scenario_name: str,
    run_config: pd.DataFrame,
    bat_cap: pd.DataFrame,
) -> None:
    """Write all results for grid district to database

    Parameters
    ----------
    static_params_dict : dict
        Static model params
    load_time_series_df : pd.DataFrame
        Load time series for grid district
    bus_id : int
        ID of grid district
    scenario_name : str
        Scenario name
    run_config : pd.DataFrame
        simBEV metadata: run config
    bat_cap : pd.DataFrame
        Battery capacities per EV type

    Returns
    -------
    None
    """

    def calc_initial_ev_soc(bus_id: int, scenario_name: str) -> pd.DataFrame:
        """Calculate an average initial state of charge for EVs in MV grid
        district.

        This is done by weighting the initial SoCs at timestep=0 with EV count
        and battery capacity for each EV type.
        """
        with db.session_scope() as session:
            query_ev_soc = (
                session.query(
                    EgonEvPool.type,
                    func.count(EgonEvTrip.egon_ev_pool_ev_id).label(
                        "ev_count"
                    ),
                    func.avg(EgonEvTrip.soc_start).label("ev_soc_start"),
                )
                .select_from(EgonEvTrip)
                .join(
                    EgonEvPool,
                    EgonEvPool.ev_id == EgonEvTrip.egon_ev_pool_ev_id,
                )
                .join(
                    EgonEvMvGridDistrict,
                    EgonEvMvGridDistrict.egon_ev_pool_ev_id
                    == EgonEvTrip.egon_ev_pool_ev_id,
                )
                .filter(
                    EgonEvTrip.scenario == scenario_name,
                    EgonEvPool.scenario == scenario_name,
                    EgonEvMvGridDistrict.scenario == scenario_name,
                    EgonEvMvGridDistrict.bus_id == bus_id,
                    EgonEvTrip.simbev_event_id == 0,
                )
                .group_by(EgonEvPool.type)
            )

        initial_soc_per_ev_type = pd.read_sql(
            query_ev_soc.statement, query_ev_soc.session.bind, index_col="type"
        )

        initial_soc_per_ev_type[
            "battery_capacity_sum"
        ] = initial_soc_per_ev_type.ev_count.multiply(bat_cap)
        initial_soc_per_ev_type[
            "ev_soc_start_abs"
        ] = initial_soc_per_ev_type.battery_capacity_sum.multiply(
            initial_soc_per_ev_type.ev_soc_start
        )

        return (
            initial_soc_per_ev_type.ev_soc_start_abs.sum()
            / initial_soc_per_ev_type.battery_capacity_sum.sum()
        )

    def write_to_db(write_lowflex_model: bool) -> None:
        """Write model data to eTraGo tables"""

        @db.check_db_unique_violation
        def write_bus(scenario_name: str) -> int:
            # eMob MIT bus
            emob_bus_id = db.next_etrago_id("bus")
            with db.session_scope() as session:
                session.add(
                    EgonPfHvBus(
                        scn_name=scenario_name,
                        bus_id=emob_bus_id,
                        v_nom=1,
                        carrier="Li_ion",
                        x=etrago_bus.x,
                        y=etrago_bus.y,
                        geom=etrago_bus.geom,
                    )
                )
            return emob_bus_id

        @db.check_db_unique_violation
        def write_link(scenario_name: str) -> None:
            # eMob MIT link [bus_el] -> [bus_ev]
            emob_link_id = db.next_etrago_id("link")
            with db.session_scope() as session:
                session.add(
                    EgonPfHvLink(
                        scn_name=scenario_name,
                        link_id=emob_link_id,
                        bus0=etrago_bus.bus_id,
                        bus1=emob_bus_id,
                        carrier="BEV_charger",
                        efficiency=float(run_config.eta_cp),
                        p_nom=(
                            load_time_series_df.simultaneous_plugged_in_charging_capacity.max()  # noqa: E501
                        ),
                        p_nom_extendable=False,
                        p_nom_min=0,
                        p_nom_max=np.Inf,
                        p_min_pu=0,
                        p_max_pu=1,
                        # p_set_fixed=0,
                        capital_cost=0,
                        marginal_cost=0,
                        length=0,
                        terrain_factor=1,
                    )
                )
            with db.session_scope() as session:
                session.add(
                    EgonPfHvLinkTimeseries(
                        scn_name=scenario_name,
                        link_id=emob_link_id,
                        temp_id=1,
                        p_min_pu=None,
                        p_max_pu=(
                            hourly_load_time_series_df.ev_availability.to_list()  # noqa: E501
                        ),
                    )
                )

        @db.check_db_unique_violation
        def write_store(scenario_name: str) -> None:
            # eMob MIT store
            emob_store_id = db.next_etrago_id("store")
            with db.session_scope() as session:
                session.add(
                    EgonPfHvStore(
                        scn_name=scenario_name,
                        store_id=emob_store_id,
                        bus=emob_bus_id,
                        carrier="battery_storage",
                        e_nom=static_params_dict["store_ev_battery.e_nom_MWh"],
                        e_nom_extendable=False,
                        e_nom_min=0,
                        e_nom_max=np.Inf,
                        e_min_pu=0,
                        e_max_pu=1,
                        e_initial=(
                            initial_soc_mean
                            * static_params_dict["store_ev_battery.e_nom_MWh"]
                        ),
                        e_cyclic=False,
                        sign=1,
                        standing_loss=0,
                    )
                )
            with db.session_scope() as session:
                session.add(
                    EgonPfHvStoreTimeseries(
                        scn_name=scenario_name,
                        store_id=emob_store_id,
                        temp_id=1,
                        e_min_pu=hourly_load_time_series_df.soc_min.to_list(),
                        e_max_pu=hourly_load_time_series_df.soc_max.to_list(),
                    )
                )

        @db.check_db_unique_violation
        def write_load(
            scenario_name: str, connection_bus_id: int, load_ts: list
        ) -> None:
            # eMob MIT load
            emob_load_id = db.next_etrago_id("load")
            with db.session_scope() as session:
                session.add(
                    EgonPfHvLoad(
                        scn_name=scenario_name,
                        load_id=emob_load_id,
                        bus=connection_bus_id,
                        carrier="land_transport_EV",
                        sign=-1,
                    )
                )
            with db.session_scope() as session:
                session.add(
                    EgonPfHvLoadTimeseries(
                        scn_name=scenario_name,
                        load_id=emob_load_id,
                        temp_id=1,
                        p_set=load_ts,
                    )
                )

        # Get eTraGo substation bus
        with db.session_scope() as session:
            query = session.query(
                EgonPfHvBus.scn_name,
                EgonPfHvBus.bus_id,
                EgonPfHvBus.x,
                EgonPfHvBus.y,
                EgonPfHvBus.geom,
            ).filter(
                EgonPfHvBus.scn_name == scenario_name,
                EgonPfHvBus.bus_id == bus_id,
                EgonPfHvBus.carrier == "AC",
            )
            etrago_bus = query.first()
            if etrago_bus is None:
                # TODO: raise exception here!
                print(
                    f"No AC bus found for scenario {scenario_name} "
                    f"with bus_id {bus_id} in table egon_etrago_bus!"
                )

        # Call DB writing functions for regular or lowflex scenario
        # * use corresponding scenario name as defined in datasets.yml
        # * no storage for lowflex scenario
        # * load timeseries:
        #   * regular (flex): use driving load
        #   * lowflex: use dumb charging load
        if write_lowflex_model is False:
            emob_bus_id = write_bus(scenario_name=scenario_name)
            write_link(scenario_name=scenario_name)
            write_store(scenario_name=scenario_name)
            write_load(
                scenario_name=scenario_name,
                connection_bus_id=emob_bus_id,
                load_ts=(
                    hourly_load_time_series_df.driving_load_time_series.to_list()  # noqa: E501
                ),
            )
        else:
            # Get lowflex scenario name
            lowflex_scenario_name = DATASET_CFG["scenario"]["lowflex"][
                "names"
            ][scenario_name]
            write_load(
                scenario_name=lowflex_scenario_name,
                connection_bus_id=etrago_bus.bus_id,
                load_ts=hourly_load_time_series_df.load_time_series.to_list(),
            )

    def write_to_file():
        """Write model data to file (for debugging purposes)"""
        results_dir = WORKING_DIR / Path("results", scenario_name, str(bus_id))
        results_dir.mkdir(exist_ok=True, parents=True)

        hourly_load_time_series_df[["load_time_series"]].to_csv(
            results_dir / "ev_load_time_series.csv"
        )
        hourly_load_time_series_df[["ev_availability"]].to_csv(
            results_dir / "ev_availability.csv"
        )
        hourly_load_time_series_df[["soc_min", "soc_max"]].to_csv(
            results_dir / "ev_dsm_profile.csv"
        )

        static_params_dict[
            "load_land_transport_ev.p_set_MW"
        ] = "ev_load_time_series.csv"
        static_params_dict["link_bev_charger.p_max_pu"] = "ev_availability.csv"
        static_params_dict["store_ev_battery.e_min_pu"] = "ev_dsm_profile.csv"
        static_params_dict["store_ev_battery.e_max_pu"] = "ev_dsm_profile.csv"

        file = results_dir / "ev_static_params.json"

        with open(file, "w") as f:
            json.dump(static_params_dict, f, indent=4)

The variable static_params_dict does not seem to be defined.

    print("  Writing model timeseries...")
    load_time_series_df = load_time_series_df.assign(
        ev_availability=(
            load_time_series_df.simultaneous_plugged_in_charging_capacity
            / static_params_dict["link_bev_charger.p_nom_MW"]
        )
    )

    # Resample to 1h
    hourly_load_time_series_df = load_time_series_df.resample("1H").agg(
        {
            "load_time_series": np.mean,
            "flex_time_series": np.mean,
            "simultaneous_plugged_in_charging_capacity": np.mean,
            "simultaneous_plugged_in_charging_capacity_flex": np.mean,
            "soc_min_absolute": np.min,
            "soc_max_absolute": np.max,
            "ev_availability": np.mean,
            "driving_load_time_series": np.sum,
        }
    )

    # Create relative SoC timeseries
    hourly_load_time_series_df = hourly_load_time_series_df.assign(
        soc_min=hourly_load_time_series_df.soc_min_absolute.div(
            static_params_dict["store_ev_battery.e_nom_MWh"]
        ),
        soc_max=hourly_load_time_series_df.soc_max_absolute.div(
            static_params_dict["store_ev_battery.e_nom_MWh"]
        ),
    )
    hourly_load_time_series_df = hourly_load_time_series_df.assign(
        soc_delta_absolute=(
            hourly_load_time_series_df.soc_max_absolute
            - hourly_load_time_series_df.soc_min_absolute
        ),
        soc_delta=(
            hourly_load_time_series_df.soc_max
            - hourly_load_time_series_df.soc_min
        ),
    )

    # Crop hourly TS if needed
    hourly_load_time_series_df = hourly_load_time_series_df[:8760]

    # Create lowflex scenario?
    write_lowflex_model = DATASET_CFG["scenario"]["lowflex"][
        "create_lowflex_scenario"
    ]

    # Get initial average storage SoC
    initial_soc_mean = calc_initial_ev_soc(bus_id, scenario_name)

    # Write to database: regular and lowflex scenario
    write_to_db(write_lowflex_model=False)
    print("    Writing flex scenario...")
    if write_lowflex_model is True:
        print("    Writing lowflex scenario...")
        write_to_db(write_lowflex_model=True)

    # Export to working dir if requested
    if DATASET_CFG["model_timeseries"]["export_results_to_csv"]:
        write_to_file()