feat(emission factors): emission factor model by year, zone, and mode (#8496)

seljaks · web-flow · commit 45bef8884b32 · 2025-11-17T16:41:01.000+01:00
* feat(emission factors): emission factor model by year, zone, and mode

* fix

* rename to fix name collision

* renaming dropped the file

* add top level function that's a switch
diff --git a/electricitymap/contrib/config/emission_factors_lookup.py b/electricitymap/contrib/config/emission_factors_lookup.py
@@ -0,0 +1,269 @@
+from datetime import datetime, timezone
+from operator import itemgetter
+from typing import Any
+
+from electricitymap.contrib.config import (
+    CO2EQ_PARAMETERS_DIRECT,
+    CO2EQ_PARAMETERS_LIFECYCLE,
+    ZONE_PARENT,
+    ZONES_CONFIG,
+)
+from electricitymap.contrib.config.constants import ENERGIES
+from electricitymap.contrib.config.model import (
+    EmissionFactorVariant,
+    YearZoneModeEmissionFactor,
+)
+from electricitymap.contrib.lib.types import ZoneKey
+
+
+def get_zone_specific_co2eq_parameter(
+    co2eq_parameters: dict,
+    zone_key: str,
+    key: str,
+    sub_key: str,
+    dt: datetime,
+    metadata: bool = False,
+) -> dict[str, float]:
+    if metadata:
+        return _get_zone_specific_co2eq_parameter_with_metadata(
+            co2eq_parameters=co2eq_parameters,
+            zone_key=zone_key,
+            key=key,
+            sub_key=sub_key,
+            dt=dt,
+        )
+    else:
+        return _get_zone_specific_co2eq_parameter_no_metadata(
+            co2eq_parameters=co2eq_parameters,
+            zone_key=zone_key,
+            key=key,
+            sub_key=sub_key,
+            dt=dt,
+        )
+
+
+def _get_zone_specific_co2eq_parameter_no_metadata(
+    co2eq_parameters: dict, zone_key: str, key: str, sub_key: str, dt: datetime
+) -> dict[str, float]:  # TODO: actually this returns Union[Dict, bool]
+    """Accessor for co2eq_parameters.
+    If Available, it will return the zoneOverride value. If not, it will return the default value.
+
+    Args:
+        co2eq_parameters (dict): The dictionary to read from.
+        zone_key (str): The zone_key to try and find a zoneOverride for.
+        key (str): The key of the parameter to find.
+        sub_key (str): The specific sub key inside the parameter that you want to access
+        dt (datetime): Will return the most recent co2eq for that dt. For the latest co2eq, pass `datetime.max`
+
+    Raises:
+        ValueError: Raised when both the zoneOverride and default value is unavailable
+    """
+    # TODO this doesn't raise a ValueError at the moment. Which gives typing errors because it can return None.
+    params = co2eq_parameters[key]
+    zone_key = ZoneKey(zone_key)
+
+    defaults = params["defaults"][sub_key]
+    zone_override = params["zoneOverrides"].get(zone_key, {}).get(sub_key, None)
+    # If no entry was found, use the parent if it exists
+    if ZONE_PARENT.get(zone_key) and not zone_override:
+        zone_override = (
+            params["zoneOverrides"].get(ZONE_PARENT[zone_key], {}).get(sub_key, None)
+        )
+
+    res = None
+    res = zone_override if zone_override is not None else defaults
+
+    if isinstance(res, list):
+        # `n` dates are sorted in ascending order (d1, d2, ..., dn)
+        # d1 is valid from from (epoch to d2)
+        # d2 is valid from (d2 to d3)
+        # dn is valid from (dn to end_of_time)
+
+        if len(res) == 0:
+            raise ValueError(
+                f"Error in given co2eq_parameters. List is empty for [{zone_key}, {key}, {sub_key}]"
+            )
+
+        res.sort(key=itemgetter("datetime"))
+        dt = dt.replace(tzinfo=timezone.utc)
+        for co2eq in reversed(res):
+            co2eq_dt = datetime.fromisoformat(co2eq["datetime"]).replace(
+                tzinfo=timezone.utc
+            )
+            if co2eq_dt <= dt:
+                return co2eq  # type: ignore[no-any-return]
+        breakpoint()
+        return res[0]  # type: ignore[no-any-return]
+
+    else:
+        return res  # type: ignore[no-any-return]
+
+
+def _get_zone_specific_co2eq_parameter_with_metadata(
+    co2eq_parameters: dict,
+    zone_key: str,
+    key: str,
+    sub_key: str,
+    dt: datetime,
+) -> dict[str, Any]:
+    """
+    Lookup logic identical to get_zone_specific_co2eq_parameter.
+    Adds a 'variant' field that provides context about where an emission factor comes from.
+    Variant selection logic:
+
+    GLOBAL/ZONE -> defined in config/defaults.yaml or config/zones/<zone_key>.yaml
+
+    *_EXACT_TIMELESS -> no "datetime"
+    *_EXACT_TIMELY -> has "datetime" and "datetime".year == dt.year
+    *_FALLBACK_LATEST -> has "datetime" and max("datetime").year < dt.year
+    *_FALLBACK_OLDER -> has "datetime" and max("datetime").year < dt.year < min("datetime").year and is not EXACT
+    *_FALLBACK_OLDEST -> has "datetime" and dt.year < min("datetime").year
+    """
+    params = co2eq_parameters[key]
+    zone_key = ZoneKey(zone_key)
+
+    defaults = params["defaults"][sub_key]
+    zone_override = params["zoneOverrides"].get(zone_key, {}).get(sub_key, None)
+    # If no entry was found, use the parent if it exists
+    if ZONE_PARENT.get(zone_key) and not zone_override:
+        zone_override = (
+            params["zoneOverrides"].get(ZONE_PARENT[zone_key], {}).get(sub_key, None)
+        )
+
+    res = None
+    res = zone_override if zone_override is not None else defaults
+
+    if isinstance(res, list):
+        # `n` dates are sorted in ascending order (d1, d2, ..., dn)
+        # d1 is valid from from (epoch to d2)
+        # d2 is valid from (d2 to d3)
+        # dn is valid from (dn to end_of_time)
+
+        if len(res) == 0:
+            raise ValueError(
+                f"Error in given co2eq_parameters. List is empty for [{zone_key}, {key}, {sub_key}]"
+            )
+
+        res.sort(key=itemgetter("datetime"))
+        dt = dt.replace(tzinfo=timezone.utc)
+        for i, co2eq in enumerate(reversed(res)):
+            co2eq_dt = datetime.fromisoformat(co2eq["datetime"]).replace(
+                tzinfo=timezone.utc
+            )
+            if co2eq_dt <= dt:
+                ret = co2eq
+
+                ef_year = co2eq_dt.year
+                dt_year = dt.year
+                if ef_year == dt_year:
+                    if zone_override:
+                        variant = EmissionFactorVariant.ZONE_EXACT_TIMELY
+                    else:
+                        variant = EmissionFactorVariant.GLOBAL_EXACT_TIMELY
+                elif ef_year < dt_year:
+                    if i == 0:
+                        if zone_override:
+                            variant = EmissionFactorVariant.ZONE_FALLBACK_LATEST
+                        else:
+                            variant = EmissionFactorVariant.GLOBAL_FALLBACK_LATEST
+                    else:
+                        if zone_override:
+                            variant = EmissionFactorVariant.ZONE_FALLBACK_OLDER
+                        else:
+                            variant = EmissionFactorVariant.GLOBAL_FALLBACK_OLDER
+
+                ret["variant"] = variant.value
+                return ret
+
+        # res is sorted in ascending order, if we get here
+        # it means dt < res[0]["datetime"]
+        ret = res[0]
+        if zone_override:
+            variant = EmissionFactorVariant.ZONE_FALLBACK_OLDEST
+        else:
+            variant = EmissionFactorVariant.GLOBAL_FALLBACK_OLDEST
+        ret["variant"] = variant.value
+        return ret
+
+    else:
+        # res is not a list, which implies it's a dict
+        # there are two kinds of dicts
+        # if it has "datetime" then it's timely
+        # else it's timeless
+        ret = res
+        ret_dt = res.get("datetime")
+        if ret_dt is None:
+            if zone_override:
+                variant = EmissionFactorVariant.ZONE_EXACT_TIMELESS
+            else:
+                variant = EmissionFactorVariant.GLOBAL_EXACT_TIMELESS
+        else:
+            ret_dt_year = (
+                datetime.fromisoformat(ret_dt).replace(tzinfo=timezone.utc).year
+            )
+            dt_year = dt.year
+            if ret_dt_year == dt_year:
+                if zone_override:
+                    variant = EmissionFactorVariant.ZONE_EXACT_TIMELY
+                else:
+                    variant = EmissionFactorVariant.GLOBAL_EXACT_TIMELY
+            elif ret_dt_year < dt_year:
+                if zone_override:
+                    variant = EmissionFactorVariant.ZONE_FALLBACK_LATEST
+                else:
+                    variant = EmissionFactorVariant.GLOBAL_FALLBACK_LATEST
+            else:
+                if zone_override:
+                    variant = EmissionFactorVariant.ZONE_FALLBACK_OLDEST
+                else:
+                    variant = EmissionFactorVariant.GLOBAL_FALLBACK_OLDEST
+
+        ret["variant"] = variant.value
+        return ret
+
+
+def _get_emission_factor_lifecycle_and_direct(
+    zone_key, dt, mode
+) -> YearZoneModeEmissionFactor:
+    item = {}
+    item["dt"] = dt
+    item["zone_key"] = zone_key
+    item["mode"] = mode
+    for description, data in [
+        ("lifecycle", CO2EQ_PARAMETERS_LIFECYCLE),
+        ("direct", CO2EQ_PARAMETERS_DIRECT),
+    ]:
+        result = get_zone_specific_co2eq_parameter(
+            co2eq_parameters=data,
+            zone_key=zone_key,
+            key="emissionFactors",
+            sub_key=mode,
+            dt=dt,
+            metadata=True,
+        )
+        d = {
+            f"{description}_{k}": v
+            for k, v in result.items()
+            if k in ("datetime", "source", "value", "variant")
+        }
+        item = {**item, **d}
+    print(item)
+    model_obj = YearZoneModeEmissionFactor(**item)
+    return model_obj
+
+
+def get_emission_factors_with_metadata_all_years() -> list[YearZoneModeEmissionFactor]:
+    start = 2015
+    end = datetime.now(tz=timezone.utc).year
+
+    acc = []
+    for zone_key in ZONES_CONFIG:
+        for i in range(start, end):
+            dt = datetime(year=i, month=1, day=1, tzinfo=timezone.utc)
+            for mode in ENERGIES:
+                model_obj = _get_emission_factor_lifecycle_and_direct(
+                    zone_key, dt, mode
+                )
+                acc.append(model_obj)
+
+    return acc
diff --git a/electricitymap/contrib/config/model.py b/electricitymap/contrib/config/model.py
@@ -1,5 +1,6 @@
 from collections.abc import Callable
-from datetime import date, datetime
+from datetime import date, datetime, timezone
+from enum import Enum
 
 from pydantic import (
     BaseModel,
@@ -402,6 +403,77 @@ def ids_match_configs(cls, v):
         return v
 
 
+class EmissionFactorVariant(Enum):
+    """
+    Describes where an emission factor (EF) comes from.
+    See electricitymap/contrib/config/emission_factors.py::_get_zone_specific_co2eq_parameter_with_metadata
+    """
+
+    GLOBAL_EXACT_TIMELESS = "global_exact_timeless"
+    GLOBAL_EXACT_TIMELY = "global_exact_timely"
+    GLOBAL_FALLBACK_LATEST = "global_fallback_latest"
+    GLOBAL_FALLBACK_OLDER = "global_fallback_older"
+    GLOBAL_FALLBACK_OLDEST = "global_fallback_oldest"
+
+    ZONE_EXACT_TIMELESS = "zone_exact_timeless"
+    ZONE_EXACT_TIMELY = "zone_exact_timely"
+    ZONE_FALLBACK_LATEST = "zone_fallback_latest"
+    ZONE_FALLBACK_OLDER = "zone_fallback_older"
+    ZONE_FALLBACK_OLDEST = "zone_fallback_oldest"
+
+
+class EmissionFactorMode(Enum):
+    BIOMASS = "biomass"
+    COAL = "coal"
+    GAS = "gas"
+    GEOTHERMAL = "geothermal"
+    HYDRO = "hydro"
+    NUCLEAR = "nuclear"
+    OIL = "oil"
+    SOLAR = "solar"
+    UNKNOWN = "unknown"
+    WIND = "wind"
+    BATTERY_DISCHARGE = "battery discharge"
+    HYDRO_DISCHARGE = "hydro discharge"
+
+
+class YearZoneModeEmissionFactor(StrictBaseModelWithAlias):
+    dt: datetime = Field(..., alias="datetime")
+    zone_key: ZoneKey
+    mode: EmissionFactorMode
+    lifecycle_value: NonNegativeFloat
+    lifecycle_source: str
+    lifecycle_variant: EmissionFactorVariant
+    lifecycle_datetime: datetime | None
+    direct_value: NonNegativeFloat
+    direct_source: str
+    direct_variant: EmissionFactorVariant
+    direct_datetime: datetime | None
+
+    @validator("dt", "lifecycle_datetime", "direct_datetime", pre=True)
+    def validate_timezone_aware(cls, v: datetime | None) -> datetime | None:
+        if v is None:
+            return v
+
+        if isinstance(v, str):
+            v = datetime.fromisoformat(v).replace(tzinfo=timezone.utc)
+
+        if v.tzinfo is None or v.tzinfo.utcoffset(v) is None:
+            raise ValueError("Datetime must be timezone-aware")
+
+        truncated_to_year = v.replace(
+            month=1,
+            day=1,
+            hour=0,
+            minute=0,
+            second=0,
+            microsecond=0,
+        )
+        if v != truncated_to_year:
+            raise ValueError("Datetime must be truncated to year.")
+        return v
+
+
 DATA_CENTERS_CONFIG_MODEL = DataCenters(data_centers=DATA_CENTERS_CONFIG)
 
 
diff --git a/tests/config/test_emission_factors.py b/tests/config/test_emission_factors.py
@@ -0,0 +1,39 @@
+from datetime import datetime, timezone
+
+from electricitymap.contrib.config import CO2EQ_PARAMETERS_DIRECT
+from electricitymap.contrib.config.emission_factors_lookup import (
+    get_emission_factors_with_metadata_all_years,
+    get_zone_specific_co2eq_parameter,
+)
+
+
+def test_all_emission_factors():
+    efs = get_emission_factors_with_metadata_all_years()
+    expected_len = 46560
+    assert len(efs) == expected_len
+
+
+def test_get_zone_specific_co2eq_parameter_identical_return():
+    zone_key = "AT"
+    year = 2024
+    dt = datetime(year=year, month=1, day=1, tzinfo=timezone.utc)
+    mode = "coal"
+
+    res1 = get_zone_specific_co2eq_parameter(
+        co2eq_parameters=CO2EQ_PARAMETERS_DIRECT,
+        zone_key=zone_key,
+        key="emissionFactors",
+        sub_key=mode,
+        dt=dt,
+    )
+
+    res2 = get_zone_specific_co2eq_parameter(
+        co2eq_parameters=CO2EQ_PARAMETERS_DIRECT,
+        zone_key=zone_key,
+        key="emissionFactors",
+        sub_key=mode,
+        dt=dt,
+        metadata=True,
+    )
+
+    assert res1["value"] == res2["value"]
