EOS/tests/test_geneticsimulation.py

import numpy as np
import pytest

from akkudoktoreos.config.configabc import TimeWindow, TimeWindowSequence
from akkudoktoreos.devices.genetic.battery import (
    Battery,
    ElectricVehicleParameters,
    SolarPanelBatteryParameters,
)
from akkudoktoreos.devices.genetic.homeappliance import (
    HomeAppliance,
    HomeApplianceParameters,
)
from akkudoktoreos.devices.genetic.inverter import (
    Inverter,
    InverterParameters,
)
from akkudoktoreos.optimization.genetic.genetic import (
    GeneticEnergyManagementParameters,
    GeneticSimulation,
    GeneticSimulationResult,
)
from akkudoktoreos.utils.datetimeutil import to_duration, to_time

start_hour = 1


@pytest.mark.parametrize("efficiency_curve", [None, [(0.0, 0.8), (1.0, 0.97)]])
def test_below_min_soc_requires_real_charging_energy(efficiency_curve):
    """Discharge attempts before and after partial charging must not create energy."""
    battery = Battery(
        SolarPanelBatteryParameters(
            device_id="battery1",
            capacity_wh=26_400,
            initial_soc_percentage=0,
            min_soc_percentage=6,
            max_charge_power_w=7_000,
            charging_efficiency=0.95,
            discharging_efficiency=0.95,
        ),
        prediction_hours=4,
        slot_duration_h=0.25,
    )
    inverter = Inverter(
        InverterParameters(
            device_id="inverter1",
            battery_id="battery1",
            max_power_wh=10_000,
            max_ac_charge_power_w=6_000,
            ac_to_dc_efficiency=0.97,
            dc_to_ac_efficiency=0.97,
            dc_to_ac_efficiency_curve=efficiency_curve,
        ),
        battery=battery,
        slot_duration_h=0.25,
    )
    simulation = GeneticSimulation()
    simulation.prepare(
        GeneticEnergyManagementParameters(
            pv_forecast_wh=[0.0] * 4,
            total_load=[280.0] * 4,
            electricity_price_per_wh=[0.00037, 0.00037, 0.00050, 0.00063],
            feed_in_tariff_per_wh=0.0,
            price_per_wh_battery=0.0,
        ),
        optimization_hours=4,
        prediction_hours=4,
        inverter=inverter,
    )
    assert simulation.ac_charge_hours is not None
    assert simulation.bat_discharge_hours is not None
    simulation.ac_charge_hours[:] = [0.0, 1.0, 0.0, 0.0]
    simulation.bat_discharge_hours[:] = [1.0, 0.0, 1.0, 1.0]

    result = simulation.simulate(0)

    stored_wh = 1_500 * 0.97 * 0.95
    assert result["akku_soc_pro_stunde"] == pytest.approx(
        [0.0, 0.0, stored_wh / 26_400 * 100, stored_wh / 26_400 * 100]
    )
    assert result["Netzbezug_Wh_pro_Stunde"] == pytest.approx([280, 1780, 280, 280])
    assert battery.soc_wh == pytest.approx(stored_wh)
    assert battery.soc_wh < battery.min_soc_wh
    assert result["Gesamt_Verluste"] == pytest.approx(1_500 - stored_wh)


# Example initialization of necessary components
@pytest.fixture
def genetic_simulation(config_eos) -> GeneticSimulation:
    """Fixture to create an EnergyManagement instance with given test parameters."""
    # Assure configuration holds the correct values
    config_eos.merge_settings_from_dict(
        {"prediction": {"hours": 48}, "optimization": {"hours": 24}}
    )
    assert config_eos.prediction.hours == 48
    assert config_eos.optimization.genetic.horizon_hours == 24

    # Initialize the battery and the inverter
    akku = Battery(
        SolarPanelBatteryParameters(
            device_id="battery1",
            capacity_wh=5000,
            initial_soc_percentage=80,
            min_soc_percentage=10,
        ),
        prediction_hours = config_eos.prediction.hours,
    )
    akku.reset()

    inverter = Inverter(
        InverterParameters(device_id="inverter1", max_power_wh=10000, battery_id=akku.parameters.device_id),
        battery = akku,
    )

    # Household device (currently not used, set to None)
    home_appliance = HomeAppliance(
        HomeApplianceParameters(
            device_id="dishwasher1",
            consumption_wh=2000,
            duration_h=2,
            time_windows=None,
        ),
        optimization_hours = config_eos.optimization.genetic.horizon_hours,
        prediction_hours = config_eos.prediction.hours,
    )

    # Example initialization of electric car battery
    eauto = Battery(
        ElectricVehicleParameters(
            device_id="ev1", capacity_wh=26400, initial_soc_percentage=10, min_soc_percentage=10
        ),
        prediction_hours = config_eos.prediction.hours,
    )
    eauto.set_charge_per_hour(np.full(config_eos.prediction.hours, 1))

    # Parameters based on previous example data
    pv_prognose_wh = [
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        8.05,
        352.91,
        728.51,
        930.28,
        1043.25,
        1106.74,
        1161.69,
        6018.82,
        5519.07,
        3969.88,
        3017.96,
        1943.07,
        1007.17,
        319.67,
        7.88,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        5.04,
        335.59,
        705.32,
        1121.12,
        1604.79,
        2157.38,
        1433.25,
        5718.49,
        4553.96,
        3027.55,
        2574.46,
        1720.4,
        963.4,
        383.3,
        0,
        0,
        0,
    ]

    strompreis_euro_pro_wh = [
        0.0003384,
        0.0003318,
        0.0003284,
        0.0003283,
        0.0003289,
        0.0003334,
        0.0003290,
        0.0003302,
        0.0003042,
        0.0002430,
        0.0002280,
        0.0002212,
        0.0002093,
        0.0001879,
        0.0001838,
        0.0002004,
        0.0002198,
        0.0002270,
        0.0002997,
        0.0003195,
        0.0003081,
        0.0002969,
        0.0002921,
        0.0002780,
        0.0003384,
        0.0003318,
        0.0003284,
        0.0003283,
        0.0003289,
        0.0003334,
        0.0003290,
        0.0003302,
        0.0003042,
        0.0002430,
        0.0002280,
        0.0002212,
        0.0002093,
        0.0001879,
        0.0001838,
        0.0002004,
        0.0002198,
        0.0002270,
        0.0002997,
        0.0003195,
        0.0003081,
        0.0002969,
        0.0002921,
        0.0002780,
    ]

    einspeiseverguetung_euro_pro_wh = 0.00007
    preis_euro_pro_wh_akku = 0.0001

    gesamtlast = [
        676.71,
        876.19,
        527.13,
        468.88,
        531.38,
        517.95,
        483.15,
        472.28,
        1011.68,
        995.00,
        1053.07,
        1063.91,
        1320.56,
        1132.03,
        1163.67,
        1176.82,
        1216.22,
        1103.78,
        1129.12,
        1178.71,
        1050.98,
        988.56,
        912.38,
        704.61,
        516.37,
        868.05,
        694.34,
        608.79,
        556.31,
        488.89,
        506.91,
        804.89,
        1141.98,
        1056.97,
        992.46,
        1155.99,
        827.01,
        1257.98,
        1232.67,
        871.26,
        860.88,
        1158.03,
        1222.72,
        1221.04,
        949.99,
        987.01,
        733.99,
        592.97,
    ]

    # Initialize the energy management system with the respective parameters
    simulation = GeneticSimulation()
    simulation.prepare(
        GeneticEnergyManagementParameters.model_validate(dict(
            pv_prognose_wh=pv_prognose_wh,
            strompreis_euro_pro_wh=strompreis_euro_pro_wh,
            einspeiseverguetung_euro_pro_wh=einspeiseverguetung_euro_pro_wh,
            preis_euro_pro_wh_akku=preis_euro_pro_wh_akku,
            gesamtlast=gesamtlast,
        )),
        optimization_hours = config_eos.optimization.genetic.horizon_hours,
        prediction_hours = config_eos.prediction.hours,
        inverter=inverter,
        ev=eauto,
        home_appliance=home_appliance,
    )

    # Init for test
    assert simulation.ac_charge_hours is not None
    assert simulation.dc_charge_hours is not None
    assert simulation.bat_discharge_hours is not None
    assert simulation.ev_charge_hours is not None
    simulation.ac_charge_hours[start_hour] = 1.0
    simulation.dc_charge_hours[start_hour] = 1.0
    simulation.bat_discharge_hours[start_hour] = 1.0
    simulation.ev_charge_hours[start_hour] = 1.0
    simulation.home_appliance_start_hour = 2

    return simulation


def test_simulation(genetic_simulation):
    """Test the EnergyManagement simulation method."""
    simulation = genetic_simulation

    # Simulate starting from hour 1 (this value can be adjusted)

    result = simulation.simulate(start_hour=start_hour)

    # visualisiere_ergebnisse(
    #     simulation.gesamtlast,
    #     simulation.pv_prognose_wh,
    #     simulation.strompreis_euro_pro_wh,
    #     result,
    #     simulation.akku.discharge_array+simulation.akku.charge_array,
    #     None,
    #     simulation.pv_prognose_wh,
    #     start_hour,
    #     48,
    #     np.full(48, 0.0),
    #     filename="visualization_results.pdf",
    #     extra_data=None,
    # )

    # Assertions to validate results
    assert result is not None, "Result should not be None"
    assert isinstance(result, dict), "Result should be a dictionary"
    assert "Last_Wh_pro_Stunde" in result, "Result should contain 'Last_Wh_pro_Stunde'"

    """
    Check the result of the simulation based on expected values.
    """
    # Example result returned from the simulation (used for assertions)
    assert result is not None, "Result should not be None."

    # Check that the result is a dictionary
    assert isinstance(result, dict), "Result should be a dictionary."
    assert GeneticSimulationResult(**result) is not None

    # Check the length of the main arrays
    assert len(result["Last_Wh_pro_Stunde"]) == 47, (
        "The length of 'Last_Wh_pro_Stunde' should be 48."
    )
    assert len(result["Netzeinspeisung_Wh_pro_Stunde"]) == 47, (
        "The length of 'Netzeinspeisung_Wh_pro_Stunde' should be 48."
    )
    assert len(result["Netzbezug_Wh_pro_Stunde"]) == 47, (
        "The length of 'Netzbezug_Wh_pro_Stunde' should be 48."
    )
    assert len(result["Kosten_Euro_pro_Stunde"]) == 47, (
        "The length of 'Kosten_Euro_pro_Stunde' should be 48."
    )
    assert len(result["akku_soc_pro_stunde"]) == 47, (
        "The length of 'akku_soc_pro_stunde' should be 48."
    )

    # Verify specific values in the 'Last_Wh_pro_Stunde' array
    assert result["Last_Wh_pro_Stunde"][1] == 1527.13, (
        "The value at index 1 of 'Last_Wh_pro_Stunde' should be 1527.13."
    )
    assert result["Last_Wh_pro_Stunde"][2] == 1468.88, (
        "The value at index 2 of 'Last_Wh_pro_Stunde' should be 1468.88."
    )
    assert result["Last_Wh_pro_Stunde"][12] == 1132.03, (
        "The value at index 12 of 'Last_Wh_pro_Stunde' should be 1132.03."
    )

    # Verify that the value at index 0 is 'None'
    # Check that 'Netzeinspeisung_Wh_pro_Stunde' and 'Netzbezug_Wh_pro_Stunde' are consistent
    assert result["Netzbezug_Wh_pro_Stunde"][1] == 1527.13, (
        "The value at index 1 of 'Netzbezug_Wh_pro_Stunde' should be 1527.13."
    )

    # Reprice the physical grid flows independently. The new direct-use
    # probability model changes the old aggregate monetary golden values.
    costs = np.dot(result["Netzbezug_Wh_pro_Stunde"], simulation.elect_price_hourly[start_hour:])
    revenues = np.dot(result["Netzeinspeisung_Wh_pro_Stunde"], simulation.elect_revenue_per_hour_arr[start_hour:])
    assert result["Gesamtkosten_Euro"] == pytest.approx(costs)
    assert result["Gesamteinnahmen_Euro"] == pytest.approx(revenues)
    assert result["Gesamtbilanz_Euro"] == pytest.approx(costs - revenues)

    # Check the losses
    assert abs(result["Gesamt_Verluste"] - 1620.0) < 1e-5, (
        "Total losses should be 1620.0 ."
    )

    # Check the values in 'akku_soc_pro_stunde'
    assert result["akku_soc_pro_stunde"][-1] == 98.0, (
        "The value at index -1 of 'akku_soc_pro_stunde' should be 98.0."
    )
    assert result["akku_soc_pro_stunde"][1] == 98.0, (
        "The value at index 1 of 'akku_soc_pro_stunde' should be 98.0."
    )

    # Check home appliances
    assert sum(simulation.home_appliance.get_load_curve()) == 2000, (
        "The sum of 'simulation.home_appliance.get_load_curve()' should be 2000."
    )

    assert (
        np.nansum(
            np.where(
                result["Home_appliance_wh_per_hour"] is None,
                np.nan,
                np.array(result["Home_appliance_wh_per_hour"]),
            )
        )
        == 2000
    ), "The sum of 'Home_appliance_wh_per_hour' should be 2000."

    print("All tests passed successfully.")


def _profile_simulation(
    battery_curves: dict, initial_soc_percentage: int = 50
) -> tuple[GeneticSimulation, Battery]:
    """48 slots with a battery that has the given profile curves, if any."""
    hours = 48
    battery = Battery(
        SolarPanelBatteryParameters(
            device_id="battery1",
            capacity_wh=8000,
            initial_soc_percentage=initial_soc_percentage,
            min_soc_percentage=10,
            max_charge_power_w=6000,
            max_discharge_power_w=4000,
            charging_efficiency=0.95,
            discharging_efficiency=0.9,
            levelized_cost_of_storage_kwh=0.05,
            **battery_curves,
        ),
        prediction_hours=hours,
    )
    inverter = Inverter(
        InverterParameters(device_id="inverter1", max_power_wh=10000, battery_id="battery1"),
        battery=battery,
    )
    rng = np.random.default_rng(7)
    simulation = GeneticSimulation()
    simulation.prepare(
        GeneticEnergyManagementParameters.model_validate(
            dict(
                pv_prognose_wh=(
                    np.clip(np.sin(np.linspace(0, 4 * np.pi, hours)), 0, None) * 6000
                ).tolist(),
                strompreis_euro_pro_wh=rng.uniform(0.0002, 0.0004, hours).tolist(),
                einspeiseverguetung_euro_pro_wh=0.00007,
                preis_euro_pro_wh_akku=0.0001,
                gesamtlast=rng.uniform(300, 2500, hours).tolist(),
            )
        ),
        optimization_hours=24,
        prediction_hours=hours,
        inverter=inverter,
    )
    return simulation, battery


def _run_plans(simulation: GeneticSimulation, battery: Battery) -> list[dict]:
    rng = np.random.default_rng(11)
    hours = battery.prediction_hours
    results = []
    for _ in range(5):
        ac = rng.choice([0.0, 0.5, 1.0], hours)
        dc = rng.choice([0.0, 1.0], hours)
        discharge = rng.choice([0, 1], hours)
        battery.reset()
        simulation.ac_charge_hours = ac
        simulation.dc_charge_hours = dc
        simulation.bat_discharge_hours = discharge
        battery.set_charge_per_hour(dc)
        battery.set_discharge_per_hour(discharge)
        results.append(simulation.simulate(start_hour=0))
    return results


def test_flat_profile_curves_reproduce_plain_simulation(config_eos):
    """Constant profile tables and curves give exactly the plain battery result.

    The battery with curves takes the generic discharge path of the inverter
    and its own energy bookkeeping (LCOS); both must match the inlined path.
    """
    config_eos.merge_settings_from_dict(
        {"prediction": {"hours": 48}, "optimization": {"hours": 24}}
    )
    plain = _run_plans(*_profile_simulation({}))
    flat = _run_plans(
        *_profile_simulation(
            dict(
                charge_power_limit_w_by_soc=[6000.0] * 101,
                discharge_power_limit_w_by_soc=[4000.0] * 101,
                charging_efficiency_curve=[(0.0, 0.95), (1.0, 0.95)],
                discharging_efficiency_curve=[(0.0, 0.9), (1.0, 0.9)],
            )
        )
    )
    for plain_result, flat_result in zip(plain, flat):
        for key in ("akku_soc_pro_stunde", "Netzbezug_Wh_pro_Stunde", "Kosten_Euro_pro_Stunde"):
            assert flat_result[key] == pytest.approx(plain_result[key]), key
        assert flat_result["Gesamt_Verluste"] == pytest.approx(plain_result["Gesamt_Verluste"])
        assert flat_result["Gesamtbilanz_Euro"] == pytest.approx(plain_result["Gesamtbilanz_Euro"])


def test_soc_taper_slows_charging_near_full(config_eos):
    """Above 80 % SoC the tapered charge limit applies, below it does not."""
    config_eos.merge_settings_from_dict(
        {"prediction": {"hours": 48}, "optimization": {"hours": 24}}
    )
    taper = [6000.0] * 81 + [500.0] * 20
    results = {}
    for name, table in (("flat", [6000.0] * 101), ("taper", taper)):
        simulation, battery = _profile_simulation(
            dict(charge_power_limit_w_by_soc=table), initial_soc_percentage=82
        )
        hours = battery.prediction_hours
        battery.reset()
        simulation.ac_charge_hours = np.ones(hours)
        simulation.dc_charge_hours = np.ones(hours)
        simulation.bat_discharge_hours = np.zeros(hours)
        battery.set_charge_per_hour(np.ones(hours))
        battery.set_discharge_per_hour(np.zeros(hours))
        results[name] = simulation.simulate(start_hour=0)["akku_soc_pro_stunde"]

    # The tapered battery takes at most 500 W x 1 h (before efficiency) per slot.
    assert results["taper"][1] < results["flat"][1]
    assert results["taper"][1] - results["taper"][0] <= 500.0 * 0.95 / 8000 * 100 + 1e-9