codecarbon/viz/data.py

from typing import Any, Dict, List, Tuple

import pandas as pd
import requests
from dash import dash_table as dt

from codecarbon.core.emissions import Emissions
from codecarbon.input import DataSource, DataSourceException


class Data:
    def __init__(self):
        self._data_source = DataSource()
        self._emissions = Emissions(self._data_source)

    @staticmethod
    def get_project_data(df: pd.DataFrame, project_name: str) -> dt.DataTable:
        project_df = df[df.project_name == project_name]
        project_df = project_df.sort_values(by="timestamp")
        project_data = project_df.to_dict("records")
        columns = [{"name": column, "id": column} for column in project_df.columns]
        return dt.DataTable(data=project_data, columns=columns)

    @staticmethod
    def get_project_summary(project_data: List[Dict]):
        last_run = project_data[-1]
        project_summary = {
            "last_run": {
                "timestamp": last_run["timestamp"],
                "duration": last_run["duration"],
                "emissions": round(last_run["emissions"], 1),
                "energy_consumed": round((last_run["energy_consumed"]), 1),
            },
            "total": {
                "duration": sum(
                    map(lambda experiment: experiment["duration"], project_data)
                ),
                "emissions": sum(
                    map(lambda experiment: experiment["emissions"], project_data)
                ),
                "energy_consumed": sum(
                    map(lambda experiment: experiment["energy_consumed"], project_data)
                ),
            },
            "country_name": last_run["country_name"],
            "country_iso_code": last_run["country_iso_code"],
            "region": last_run["region"],
            "on_cloud": last_run["on_cloud"],
            "cloud_provider": last_run["cloud_provider"],
            "cloud_region": last_run["cloud_region"],
        }
        return project_summary

    def get_car_miles(self, project_carbon_equivalent: float):
        """
        8.89 × 10-3 metric tons CO2/gallon gasoline ×
        1/22.0 miles per gallon car/truck average ×
        1 CO2, CH4, and N2O/0.988 CO2
        = 4.09 x 10-4 metric tons CO2E/mile
        = 0.409 kg CO2E/mile
        Source: EPA
        :param project_carbon_equivalent: total project emissions in kg CO2E
        :return: number of miles driven by avg car
        """
        return f"{project_carbon_equivalent / 0.409:.0f}"

    def get_tv_time(self, project_carbon_equivalent: float):
        """
        Gives the amount of time
        a 32-inch LCD flat screen TV will emit
        an equivalent amount of carbon
        Ratio is 0.097 kg CO2 / 1 hour tv
        :param project_carbon_equivalent: total project emissions in kg CO2E
        :return: equivalent TV time
        """
        time_in_minutes = project_carbon_equivalent * (1 / 0.097) * 60
        formated_value = f"{time_in_minutes:.0f} minutes"
        if time_in_minutes >= 60:
            time_in_hours = time_in_minutes / 60
            formated_value = f"{time_in_hours:.0f} hours"
            if time_in_hours >= 24:
                time_in_days = time_in_hours / 24
                formated_value = f"{time_in_days:.0f} days"
        return formated_value

    def get_household_fraction(self, project_carbon_equivalent: float):
        """
        Total CO2 emissions for energy use per home: 5.734 metric tons CO2 for electricity
        + 2.06 metric tons CO2 for natural gas + 0.26 metric tons CO2 for liquid petroleum gas
         + 0.30 metric tons CO2 for fuel oil  = 8.35 metric tons CO2 per home per year / 52 weeks
         = 160.58 kg CO2/week on average
        Source: EPA
        :param project_carbon_equivalent: total project emissions in kg CO2E
        :return: % of weekly emissions re: an average American household
        """
        return f"{project_carbon_equivalent / 160.58 * 100:.2f}"

    def get_global_emissions_choropleth_data(
        self, net_energy_consumed: float
    ) -> List[Dict]:
        global_energy_mix = self._data_source.get_global_energy_mix_data()
        choropleth_data = []
        for country_iso_code in global_energy_mix.keys():
            country_energy_mix = global_energy_mix[country_iso_code]
            country_name = country_energy_mix["country_name"]

            if country_iso_code not in ["_define", "ATA"]:
                from codecarbon.core.units import Energy

                energy_consumed = Energy.from_energy(kWh=net_energy_consumed)

                from codecarbon.external.geography import GeoMetadata

                country_emissions = self._emissions.get_country_emissions(
                    energy_consumed,
                    GeoMetadata(
                        country_name=country_name, country_iso_code=country_iso_code
                    ),
                )
                country_choropleth_data = self.get_country_choropleth_data(
                    country_energy_mix=country_energy_mix,
                    country_name=country_name,
                    country_iso_code=country_iso_code,
                    country_emissions=country_emissions,
                )
                choropleth_data.append(country_choropleth_data)
        return choropleth_data

    @staticmethod
    def get_country_choropleth_data(
        country_energy_mix: Dict,
        country_name: str,
        country_iso_code: str,
        country_emissions: float,
    ) -> Dict[str, Any]:
        def format_energy_percentage(energy_type: float, total: float) -> float:
            return float(f"{energy_type / total * 100:.1f}")

        total = country_energy_mix["total_TWh"]
        return {
            "iso_code": country_iso_code,
            "emissions": country_emissions,
            "country": country_name,
            "carbon_intensity": country_energy_mix["carbon_intensity"],
            "fossil": format_energy_percentage(country_energy_mix["fossil_TWh"], total),
            "hydroelectricity": format_energy_percentage(
                country_energy_mix["hydroelectricity_TWh"],
                total,
            ),
            "nuclear": format_energy_percentage(
                country_energy_mix["nuclear_TWh"], total
            ),
            "solar": format_energy_percentage(country_energy_mix["solar_TWh"], total),
            "wind": format_energy_percentage(country_energy_mix["wind_TWh"], total),
        }

    def get_regional_emissions_choropleth_data(
        self, net_energy_consumed: float, country_iso_code: str
    ) -> List[Dict]:
        # add country codes here to render for different countries
        if country_iso_code.upper() not in ["USA", "CAN"]:
            return [{"region_code": "", "region_name": "", "emissions": ""}]

        try:
            region_emissions = self._data_source.get_country_emissions_data(
                country_iso_code.lower()
            )
        except (
            DataSourceException
        ):  # This country has regional data at the energy mix level, not the emissions level
            country_energy_mix = self._data_source.get_country_energy_mix_data(
                country_iso_code.lower()
            )
            region_emissions = {
                region: {"regionCode": region}
                for region, energy_mix in country_energy_mix.items()
            }
        choropleth_data = []
        for region_name in region_emissions.keys():
            region_code = region_emissions[region_name]["regionCode"]
            if region_name not in ["_unit"]:
                from codecarbon.core.units import Energy

                energy_consumed = Energy.from_energy(kWh=net_energy_consumed)

                from codecarbon.external.geography import GeoMetadata

                emissions = self._emissions.get_region_emissions(
                    energy_consumed,
                    GeoMetadata(country_iso_code=country_iso_code, region=region_name),
                )

                choropleth_data.append(
                    {
                        "region_code": region_code,
                        "region_name": region_name.upper(),
                        "emissions": emissions,
                    }
                )
        return choropleth_data

    def get_cloud_emissions_barchart_data(
        self,
        net_energy_consumed: float,
        on_cloud: str,
        cloud_provider: str,
        cloud_region: str,
    ) -> Tuple[str, pd.DataFrame]:
        if on_cloud == "N":
            return (
                "",
                pd.DataFrame(data={"region": [], "emissions": [], "country_name": []}),
            )
        cloud_emissions = self._data_source.get_cloud_emissions_data()
        cloud_emissions = cloud_emissions[
            ["provider", "providerName", "region", "impact", "country_name"]
        ]

        from codecarbon.core.units import EmissionsPerKWh

        cloud_emissions["emissions"] = cloud_emissions.apply(
            lambda row: EmissionsPerKWh.from_g_per_kWh(row.impact).kgs_per_kWh
            * net_energy_consumed,
            axis=1,
        )

        cloud_emissions_project_region = cloud_emissions[
            cloud_emissions.region == cloud_region
        ]
        cloud_emissions = cloud_emissions[
            (cloud_emissions.provider == cloud_provider)
            & (cloud_emissions.region != cloud_region)
        ].sort_values(by="emissions")

        return (
            cloud_emissions_project_region.iloc[0, :].providerName,
            pd.concat([cloud_emissions_project_region, cloud_emissions]),
        )

    @staticmethod
    def get_data_from_api(host):
        transformed_projects = []
        project_list = Data.list_projects(host)
        for project in project_list:
            project_sum_by_experiments_url = (
                host + f"/experiments/{project['id']}/detailed_sums"
            )
            project_name = project["name"]
            sums = requests.get(project_sum_by_experiments_url).json()
            for experiment in sums:
                experiment["project_name"] = project_name
                # experiment["emission_rate"] = 0
                # if experiment["emissions_count"] > 0:
                #     experiment["emission_rate"] = (
                #         experiment["emissions_rate"] / experiment["emissions_count"]
                #     )
                transformed_projects.append(experiment)
        df_projects = pd.DataFrame(transformed_projects)
        return df_projects

    @staticmethod
    def list_projects(host):
        projects = []
        teams_url = host + "/teams"
        teams = requests.get(teams_url).json()
        for team in teams:
            projets_url = host + f"/projects/team/{team['id']}"
            team_projects = requests.get(projets_url).json()
            if team_projects:
                projects.append(
                    list(
                        map(
                            lambda x: {"id": x["id"], "name": x["name"]},
                            iter(team_projects),
                        )
                    )
                )
        project_list = sum(projects, [])
        return project_list