src/WattFieldsCommon/simu_utils.py

# Utils to create simulation - from params to config, fron config to simu, from simu to config, ...

import os
import random
import string
import urllib
from datetime import date, datetime, timedelta
from typing import Literal

from awswrangler import s3

from WattFieldsCommon.constants import PATH_S3_FOLDER


def create_simu_config(simulation=None, **kwargs):
    """Creates a simulation configuration fron either the simu itself or the
    kwarg parameters.

    Args:
        simulation: A simulation object containing the data.
        **kwargs: Individual parameters for the configuration if the simulation object is not provided.

    Returns:
        A dictionary representing the simulation configuration.
    """
    config = {}

    # WeatherData
    config["WeatherData"] = {
        "lat": simulation.weather_data.lat if simulation else kwargs.get("lat"),
        "lon": simulation.weather_data.lon if simulation else kwargs.get("lon"),
        "alt": simulation.weather_data.alt if simulation else kwargs.get("alt"),
        "planting_date": (
            simulation.weather_data.planting_date if simulation else kwargs.get("planting_date")
        ),
        "use_tmy": (simulation.weather_data.use_tmy if simulation else kwargs.get("use_tmy")),
        "max_simulation_duration": (
            simulation.weather_data.max_simulation_duration
            if simulation
            else kwargs.get("max_simulation_duration")
        ),
    }

    # SolarFarmBifacial
    config["SolarFarmBifacial"] = {
        "rotation_axis_height": (
            simulation.solar_farm.rotation_axis_height
            if simulation
            else kwargs.get("rotation_axis_height")
        ),
        "panel_layout": (
            simulation.solar_farm.panel_layout if simulation else kwargs.get("panel_layout")
        ),
        "max_angle": (simulation.solar_farm.max_angle if simulation else kwargs.get("max_angle")),
        "distance_between_panels": (
            simulation.solar_farm.distance_between_panels
            if simulation
            else kwargs.get("distance_between_panels")
        ),
        "axis_azimuth": (
            simulation.solar_farm.axis_azimuth if simulation else kwargs.get("axis_azimuth")
        ),
        "pv_module_config_name": (
            simulation.solar_farm.pv_module_config_name
            if simulation
            else kwargs.get("pv_module_config_name")
        ),
        "pv_inverter_config_name": (
            simulation.solar_farm.pv_inverter_config_name
            if simulation
            else kwargs.get("pv_inverter_config_name")
        ),
        "post_inverter_elec_efficiency": (
            simulation.solar_farm.post_inverter_elec_efficiency
            if simulation
            else kwargs.get("post_inverter_elec_efficiency")
        ),
        "is_bifacial": (
            simulation.solar_farm.is_bifacial if simulation else kwargs.get("is_bifacial")
        ),
        # "panel_peak_power": simulation.solar_farm.panel_peak_power if simulation else kwargs.get("panel_peak_power"),
        # "panel_surface": simulation.solar_farm.panel_surface if simulation else kwargs.get("panel_surface")
    }

    # AgridRunner
    config["AgridRunner"] = {
        "soil_type": (
            simulation.agrid_runner.soil_type if simulation else kwargs.get("soil_type")
        ),
        "soil_depth": (
            simulation.agrid_runner.soil_depth if simulation else kwargs.get("soil_depth")
        ),
        "soil_RU": (
            simulation.agrid_runner.soil_RU if simulation else kwargs.get("soil_RU")
        ),
        "soil_percent_available_water_start_simu": (
            simulation.agrid_runner.soil_percent_available_water_start_simu if simulation else kwargs.get("soil_percent_available_water_start_simu")
        ),
        "crop_specie": (
            simulation.agrid_runner.crop_specie if simulation else kwargs.get("crop_specie")
        ),
        "mean_crop_height": (
            simulation.solar_farm.mean_crop_height if simulation else kwargs.get("mean_crop_height")
        ),
        "crop_cultivar": (
            simulation.agrid_runner.crop_cultivar if simulation else kwargs.get("crop_cultivar")
        ),
        "nb_rows_per_dssat_simu": (
            simulation.agrid_runner.nb_rows_per_dssat_simu
            if simulation
            else kwargs.get("nb_rows_per_dssat_simu")
        ),
        "agri_row_spacing": (
            simulation.agrid_runner.agri_row_spacing
            if simulation
            else kwargs.get("agri_row_spacing")
        ),
        "agri_row_spacing_intra_row": (
            simulation.agrid_runner.agri_row_spacing_intra_row
            if simulation
            else kwargs.get("agri_row_spacing_intra_row")
        ),
        "symetric_hypothesis": (
            simulation.agrid_runner.symetric_hypothesis
            if simulation
            else kwargs.get("symetric_hypothesis")
        ),
        "distance_between_panels": (
            simulation.agrid_runner.distance_between_panels
            if simulation
            else kwargs.get("distance_between_panels")
        ),
        "upper_bound_mid_band": (
            simulation.agrid_runner.upper_bound_mid_band
            if simulation
            else kwargs.get("upper_bound_mid_band")
        ),
        "location": (
            simulation.agrid_runner.location
            if simulation
            else (kwargs.get("lat"), kwargs.get("lon"), kwargs.get("alt"))
        ),
    }

    # Simulation
    config["Simulation"] = {
        "simu_id": simulation.id if simulation else kwargs.get("simu_id"),
        "simu_creation_date": (
            simulation.simu_creation_date if simulation else kwargs.get("simu_creation_date")
        ),
        "nb_max_generations_in_optim": (
            simulation.nb_max_generations_in_optim
            if simulation
            else kwargs.get("nb_max_generations_in_optim")
        ),
        "max_agri_loss": (simulation.max_agri_loss if simulation else kwargs.get("max_agri_loss")),
        "info": simulation.info if simulation else kwargs.get("info"),
        "nb_days_per_block": (
            simulation.nb_days_per_block
            if simulation
            else kwargs.get("nb_days_per_block")
        ),
    }

    config["General"] = {
        "site": simulation.site if simulation else kwargs.get("site"),
        "company_id": simulation.company_id if simulation else kwargs.get("company_id"),
        "user_name": simulation.user_name if simulation else kwargs.get("user_name"),
    }

    return config


def check_if_valid_dssat_params(crop_specie: str, crop_cultivar: str):
    # TODO
    return True, None
    """# Random weather data DATES = pd.date_range('2008-01-01', '2010-12-31')
    N = len(DATES) df = pd.DataFrame( { 'tn': np.random.gamma(10, 1, N), 'rad':
    np.random.gamma(10, 1.5, N),

        'prec': [0.0]* N,
        'rh': 100 * np.random.beta(1.5, 1.15, N),
        },
        index=DATES,
    )
    df['TMAX'] = df.tn + np.random.gamma(5., .5, N)
    wth = Weather(
        df,
        {'tn': 'TMIN', 'TMAX': 'TMAX', 'prec': 'RAIN',
        'rad': 'SRAD', 'rh': 'RHUM'},
        4.3434237,-74.3606715, 1800
    )
    return True, None
    """


def check_if_valid_simu_params(
    rotation_axis_height: float,
    pitch: float,
    max_angle: float,
    mean_crop_height: float,
    max_simulation_duration: float,
    planting_date: datetime,
) -> (bool, str):
    """Check parameters validity before creating simu config.

    Args:
        rotation_axis_height (float): height of the rotation axis (from ground)
        pitch (float): distance between panels
        max_angle (float): max angle of rotation (comapred to horizontal, 30 means total deflection of 60 degrees)
        mean_crop_height (float): mean height of crop : will be used to caclulate the irradience
        max_simulation_duration (float): in days
        planting_date (datetime): planting date

    Returns:
        bool: True if valid params, False otherwise
        str: error message if invalid
    """
    # TODO : add additional parameter checks
    if rotation_axis_height < 0 or pitch < 0 or max_angle < 0 or mean_crop_height < 0:
        return (
            False,
            "Tous les champs hauteur, largeur, distance entre panneaux, angle doivent etre superieurs a 0.",
        )
    if max_simulation_duration < 30:
        return False, "La duree de la simulation doit au moins etre de 30 jours"
    elif mean_crop_height >= rotation_axis_height:
        return (
            False,
            "La hauteur des cultures ne peut pas etre superieure a la hauteur du panneau",
        )
    elif planting_date + timedelta(days=max_simulation_duration) > date(2024, 11, 1):
        return False, "La date de fin de simulation depasse le 01/11/2024"
    else:
        return True, None


def get_all_simulations_metadata_from_s3(
    company_id: str, site_id: str, status: Literal["ALL", "COMPLETED", "PENDING", 'FAILED']
):
    """Gathers all the simulations stored in s3 (both pending and completed)
    for a given company and site id Pretty nice as the name itself or the simu
    contains all the metadata ! Means that single s3 query gives you all the
    information instead of having to open some files...

    Args:
        company_id (str): company_id
        site_id (str): site_id
        status (str): status

    Returns:
        list: list of dict containing all the metadata associated to simulations
    """

    assert status in ["ALL", "COMPLETED", "PENDING", 'FAILED']

    list_simu_files = []
    if status == "COMPLETED" or status == "ALL" or status == "FAILED":
        s3_dir = os.path.join(PATH_S3_FOLDER, "simulations", f"{company_id}", f"{site_id}")
        try:
            list_objects = s3.list_objects(s3_dir)
            if status == 'COMPLETED':
                list_objects = [obj for obj in list_objects if 'COMPLETED' in obj]
            if status == 'FAILED':
                list_objects = [obj for obj in list_objects if 'FAILED' in obj]
            list_simu_files.extend(list_objects)
        except Exception as e:
            print(e)

    if status == "PENDING" or status == "ALL":
        s3_dir = os.path.join(PATH_S3_FOLDER, "simulations", "pending_simu_configs")
        try:
            list_simu_files.extend(s3.list_objects(s3_dir))
        except Exception as e:
            print(e)

    list_simus_metadata = [file_path_to_simu_metadata(file) for file in list_simu_files]
    list_simus_metadata = [
        simu_meta
        for simu_meta in list_simus_metadata
        if ((simu_meta["site_id"] == site_id) and (simu_meta["company_id"] == company_id))
    ]
    return list_simus_metadata


def decode_string(s):
    """Decode a previously encoded string."""
    return urllib.parse.unquote(s)


def encode_string(s):
    """Encode a string to make it safe for file paths."""
    return urllib.parse.quote(s, safe="")


def simu_to_file_name(
    company_id: str,
    site_id: str,
    simu_status: str,
    simu_id: str,
    user_name: str,
    simu_creation_date: str,
    simu_info: str,
) -> str:
    """Function to create the filename that will contained all the metadata
    information.

    Args:
        company_id (str): straightfoward
        site_id (str): straightfoward
        simu_status (str): status of the simulation (pending if in queue)
        simu_id (str): straightfoward
        user_name (str): straightfoward
        simu_creation_date (str): straightfoward
        simu_info (str): additional information about the simulation

    Returns:
        str: filename for both the input config and the output pkl file
            Extension could help discriminating these two
    """
    assert simu_status in ["PENDING", "FAILED", "COMPLETED"]
    if not isinstance(simu_creation_date, str):
        simu_creation_date = simu_creation_date = simu_creation_date.isoformat().replace("/", "-")
    encoded_site_id = encode_string(site_id)
    encoded_user_name = encode_string(user_name)
    encoded_info = encode_string(simu_info)
    return f"{company_id}_{encoded_site_id}_{simu_status}_{simu_id}_{encoded_user_name}_{simu_creation_date.replace('/','-')}_{encoded_info}"


def file_path_to_simu_metadata(file_path) -> dict:
    """Function to get metadata from filename.

    Args:
        file_path (_type_): ok

    Returns:
        dict: metadat dict
    """
    # file_path can be either a json file (config file) or a pickle file(output file)
    file_name = file_path.split("/")[-1]
    file_name = file_name.split(".")[0]
    parts = file_name.split("_")
    if len(parts) != 7:
        raise ValueError("Invalid file path format")
    return {
        "path": file_path,
        "company_id": parts[0],
        "site_id": parts[1],
        "simu_id": parts[3],
        "simu_status": parts[2],
        "user_name": decode_string(parts[4]),
        "simu_creation_date": parts[5],
        "file_name": file_name,
        "info": decode_string(parts[6]),
    }


def generate_random_simu_id(length=8):
    """Generate a random alphanumeric string of specified length.

    Args:
        length (int, optional): Length of the random string. Defaults to 8.
    """
    # Combine uppercase letters, lowercase letters, and digits
    characters = string.ascii_letters + string.digits
    # Use random.choices to generate a list of random characters of the specified length
    random_id = "".join(random.choices(characters, k=length))
    return random_id