export_cdc.py

import string
import unicodedata

PROD_HEADER = [
    "Horodatage",
    "Production (W)",
    "Production (kWh)",
    "Surplus (kWh)",
    "Production autoconsommée (kWh)",
]
CONS_HEADER = [
    "Horodatage",
    "Consommation (W)",
    "Consommation (kWh)",
    "Alloconsommation (kWh)",
    "Autoconsommation (kWh)",
]


def make_header_lines(operation, partner_id, prm_id, data_type):
    """
    Données Elocoop de production de l'opération [nom opé] - [contact ou PRM]

    :param operation: id de l operation
    :param start_date: date de debut
    :param end_date: date de fin
    :param partner_id: id du partner
    :param prm_id: id du prm
    :param data_type: type de données (consomation ou production)
    :return:
    """
    header = []
    if prm_id:
        desc = f" - {get_prm_name_from_id(operation, prm_id)}"
    elif partner_id:
        desc = f" - {get_partner_name_from_id(operation, partner_id)}"
    else:
        desc = ""

    if data_type in ["cons"]:
        header.append(
            f"Données Elocoop de consommation de l'opération {operation.name}{desc}"
        )
        header.append(";".join(CONS_HEADER))
    elif data_type in ["prod"]:
        header.append(
            f"Données Elocoop de production de l'opération {operation.name}{desc}"
        )
        header.append(";".join(PROD_HEADER))

    return header


def make_filename(
    operation, start_date, end_date, partner_id, prm_id, data_type, file_type="csv"
):
    """
    Genere le nom du fichier exporte sous la forme
    Elocoop_[nom opé]_[date_debut]_[date_fin]_[production ou consommation]_[contact ou PRM].csv
    :param operation: id de l operation
    :param start_date: date de debut
    :param end_date: date de fin
    :param partner_id: id du partner
    :param prm_id: id du prm
    :param data_type: type de données (consomation ou production)
    :param file_type: type de fichier par defaut csv
    :return:
    """
    if prm_id:
        desc = f"_{clean_for_title(get_prm_name_from_id(operation, prm_id))}"
    elif partner_id:
        desc = f"_{clean_for_title(get_partner_name_from_id(operation, partner_id))}"
    else:
        desc = ""

    filename = (
        f"Elocoop_{clean_for_title(operation.name)}_"
        f"{start_date.replace('/', '')}_{end_date.replace('/', '')}_"
        f"{'consommation' if data_type == 'cons' else 'production'}{desc}.{file_type}"
    )
    return filename


def make_cons_data(raw_data):
    """
    make data file with cons hearder
    horodatage cons w cons kwh allocons autocans
    :param raw_data:
    :return:
    """

    sum_value = {"cons": 0, "allo_cons": 0, "auto_cons": 0}

    data_file_lines = []
    rounding = 3
    for row in raw_data:
        divider = get_divider_from_row(row)
        data_file_lines.append(
            ";".join(
                [
                    # horodatage
                    row[0].strftime("%d/%m/%Y %H:%M"),
                    # consommation en watt
                    str(round((row[1] * 2000)/divider, rounding)),
                    # consommation en kwh
                    str(round(row[1]/divider, rounding)),
                    # allocons
                    str(round(row[3]/divider, rounding)),
                    # autocons
                    str(round(row[2]/divider, rounding)),
                ]
            )
        )

        sum_value["cons"] += round(row[1]/divider, rounding)
        sum_value["allo_cons"] += round(row[3]/divider, rounding)
        sum_value["auto_cons"] += round(row[2]/divider, rounding)

    tot = ";".join(
        [
            "TOTAL",
            "",
            str(round(sum_value.get("cons"), rounding)),
            str(round(sum_value.get("allo_cons"), rounding)),
            str(round(sum_value.get("auto_cons"), rounding)),
        ]
    )
    data_file_lines.insert(0, tot)
    return data_file_lines


def make_prod_data(raw_data):
    """
    make data file with prod hearder
    horodatage prod w prod kwh surplus autocons
    :param raw_data:
    :return:
    """
    data_file_lines = []
    rounding = 3
    sum_value = {"prod": 0, "surplus": 0, "auto_cons": 0}
    for row in raw_data:
        divider = get_divider_from_row(row)
        data_file_lines.append(
            ";".join(
                [
                    # horodatage
                    row[0].strftime("%d/%m/%Y %H:%M"),
                    # production en watt
                    str(round((row[3] * 2000)/divider, rounding)),
                    # production en kwh
                    str(round(row[3]/divider, rounding)),
                    # surplus
                    str(round(row[2]/divider, rounding)),
                    # autocons
                    str(round(row[1]/divider, rounding)),
                ]
            )
        )
        sum_value["prod"] += round(row[3]/divider, rounding)
        sum_value["surplus"] += round(row[2]/divider, rounding)
        sum_value["auto_cons"] += round(row[1]/divider, rounding)

    tot = ";".join(
        [
            "TOTAL",
            "",
            str(round(sum_value.get("prod"), rounding)),
            str(round(sum_value.get("surplus"), rounding)),
            str(round(sum_value.get("auto_cons"), rounding)),
        ]
    )
    data_file_lines.insert(0, tot)

    return data_file_lines


def clean_for_title(word):
    """
    remplace les espace et caracteres speciaux par '_'
    retire les accents

    :param word: chaine de caractere d entrée
    :return: chaine de caractere "nettoyée"
    """
    if isinstance(word, str):
        char = list(string.punctuation)
        char.append(" ")

        for c in char:
            word = word.replace(c, "_")

    return "".join(
        c for c in unicodedata.normalize("NFD", word) if unicodedata.category(c) != "Mn"
    )


def get_prm_name_from_id(operation, prm_id):
    """
    :param operation: operation
    :param prm_id: id du prm
    :return: chaine avec le nom du prm
    """

    return operation.env["acc.counter"].browse(int(prm_id)).name


def get_partner_name_from_id(operation, partner_id):
    """
    :param operation: operation
    :param partner_id: id du partner
    :return: chaine avec le nom du partner
    """

    return operation.env["res.partner"].browse(int(partner_id)).name


def get_divider_from_row(row):
    """
    Get divider from a data row
    divider depends of the timestep
    by default return 2000 (timestep of 30 min)
    """

    if row[-1] == 15:
        return 4000
    return 2000