Climatiq API: Zintegruj obliczenia emisji gazów cieplarnianych z backendem
W 2025 roku wzrósł rynek rozwiązań do monitorowania emisji dwutlenku węgla aż 17,3 miliarda dolarów, napędzane coraz bardziej rygorystycznymi przepisami takie jak europejskie CSRD, ujawnienie informacji klimatycznych przez SEC w USA i ISO 14064. Firmy nie mogą koniec z poleganiem na arkuszach Excela i przybliżonych szacunkach: są przydatne automatyczne obliczenia, weryfikowalne i integrowane z systemami operacyjnymi.
Wyzwanie techniczne jest realne: obliczenie emisji gazów cieplarnianych wymaga dostępu do baz danych współczynniki emisji zaktualizowane, metodologie zatwierdzone przez Protokół GHG i konwersje wśród setek różnych jednostek miary. Wymaga to zbudowania i utrzymania tego wszystkiego we własnym zakresie miesiące specjalistycznej pracy. To tutaj się pojawia Klimat.
Climatiq to interfejs API REST, który zapewnia dostęp do over 190 000 współczynników emisji z ponad 40 zweryfikowanych źródeł (EPA, DEFRA/BEIS, IEA, ecoinvent), pokrycie Ponad 300 regionów geograficznychoraz obliczenia zgodne z Protokołem GHG dla Zakresu 1, 2 i 3. W tym artykule zbudujemy Kompletny backend FastAPI który integruje Climatiq do obliczeń emisji produkcyjnych, wraz z klientem TypeScript i kalkulatorem działającym w czasie rzeczywistym dla aplikacji SaaS.
Czego się nauczysz
- Architektura Climatiq API: punkty końcowe, uwierzytelnianie, limity szybkości i modele danych
- Baza danych współczynników emisji: jak wyszukiwać, filtrować i wybierać właściwe współczynniki
- Szacowanie oparte na aktywności: obliczenia oparte na konkretnych działaniach (kWh, km, kg)
- Szacowanie oparte na wydatkach: obliczenia wydatków walutowych w przypadku braku danych pierwotnych
- Protokół GHG Zakres 1, 2, 3: mapowanie kategorii i zgodne obliczenia
- Solidny klient Python z możliwością ponawiania prób, buforowaniem Redis i obsługą błędów
- Klient TypeScript/Node.js z Axios i typami do integracji z frontendem
- Kalkulator emisji dwutlenku węgla w czasie rzeczywistym dla SaaS z etykietami emisji dwutlenku węgla na produktach
- Testowanie z próbnym API dla środowisk CI/CD
- Alternatywy dla Climatiq i porównanie funkcjonalności
Seria o zielonej inżynierii oprogramowania
Ten artykuł jest częścią pełnej serii poświęconej inżynierii ekologicznego oprogramowania. Każdy przedmiot porusza konkretny aspekt zrównoważonego rozwoju cyfrowego:
| # | Przedmiot | Temat główny |
|---|---|---|
| 1 | Zasady inżynierii ekologicznego oprogramowania | GSF, specyfikacja SCI, 8 podstawowych zasad |
| 2 | CodeCarbon: Pomiar emisji kodu | Biblioteka Pythona, dashboard, integracja CI/CD |
| 3 | Climatiq API: Obliczenia GHG w backendie | REST API, zakres 1-3, integracja FastAPI + TypeScript |
| 4 | Pakiet SDK uwzględniający emisję dwutlenku węgla | Przesunięcie obciążenia, intensywność siatki, przesunięcie w czasie |
| 5 | Zakres 1, 2 i 3: Modelowanie danych na potrzeby raportowania ESG | Struktura danych, obliczenia, agregacja, raportowanie |
| 6 | GreenOps: infrastruktura świadoma emisji dwutlenku węgla | Planowanie Kubernetes, skalowanie w oparciu o problemy |
| 7 | Rurociąg emisji Zakres 3 Łańcuch wartości | Gromadzenie danych dostawcy, obliczenia, ścieżka audytu |
| 8 | API raportowania ESG: Integracja CSRD | Workflow CSRD, automatyzacja raportów, compliance |
| 9 | Zrównoważone wzorce architektoniczne | Pamięć masowa, inteligentne buforowanie, partia uwzględniająca węgiel |
| 10 | Sztuczna inteligencja i węgiel: ślad szkoleniowy ML | Emisje LLM, optymalizacja, Zielona AI |
1. Protokół GHG i potrzeba zautomatyzowanych obliczeń
Il Standard korporacyjny protokołu GHG jest to najbardziej przyjęta platforma na świecie do rozliczania emisji przedsiębiorstw. Klasyfikuje emisję w trzech obszarach:
- Zakres 1 (emisje bezpośrednie): Spalanie paliw w pojazdach służbowych, zakłady produkcyjne, ciepłownictwo. Znajdują się one pod bezpośrednią kontrolą firmy.
- Zakres 2 (Energia pośrednia): Zakupiona energia elektryczna, para, ciepło. Dzielą się na oparte na lokalizacji (skład sieci lokalnej) e oparte na rynku (certyfikaty energetyczne, PPA).
- Zakres 3 (łańcuch wartości): 15 kategorii obejmujących zakup towarów i usługi, transport typu upstream/downstream, wykorzystanie produktu, koniec życia, podróże służbowe, dojazdy pracowników i wiele więcej. Zazwyczaj reprezentują 70-90% całkowitej emisji firmy.
Do każdego obliczenia potrzebujesz a współczynnik emisji: współczynnik, który konwertuje czynność (litry oleju napędowego, kWh energii elektrycznej, euro zakupu) w kg CO₂e. Czynniki te różnią się w zależności od:
- Rok referencyjny: sieci elektroenergetyczne zmieniają skład co roku
- Region geograficzny: Włoskie kWh różnią się od niemieckich kWh
- Źródło danych: EPA dla USA, DEFRA dla Wielkiej Brytanii, ISPRA dla Włoch
- Obwód: w górę, w dół rzeki, od kołyski do bramy, od kołyski do grobu
Prowadzenie aktualnej bazy danych współczynników emisji wymaga dedykowanego zespołu. Climatiq robi to za nas, agregując dalej 40 zweryfikowanych źródeł z ciągłe aktualizacje.
2. Przegląd API Climatiq
Architektura i główne punkty końcowe
Climatiq API to interfejs API REST JSON oparty na adresie URL https://beta3.api.climatiq.io.
Uwierzytelnianie odbywa się poprzez Token okaziciela w nagłówku HTTP.
Dostępne plany obejmują:
| Podłoga | Połączenia/miesiąc | Funkcjonalność | Typowe zastosowanie |
|---|---|---|---|
| Wspólnota | 250 | Wszystkie punkty końcowe | Prototypowanie, testowanie |
| Rozrusznik | 5000 | + Czynniki prywatne | MŚP, MVP |
| Wzrost | 50 000 | + SLA, wsparcie | Rosnące firmy |
| Przedsiębiorstwo | Zwyczaj | + Ścieżka audytu, SSO | Duże organizacje |
Główne punkty końcowe to:
| Punkty końcowe | Metoda | Opis |
|---|---|---|
/search |
DOSTAWAĆ | Wyszukaj współczynniki emisji w bazie danych |
/estimate |
POST | Szacunkowa pojedyncza emisja z działalności |
/batch/estimate |
POST | Wiele szacunków (do 100 na żądanie) |
/travel/flights |
POST | Emisje linii lotniczych (Zakres 3.6) |
/freight |
POST | Emisje z multimodalnego transportu towarowego |
/procurement |
POST | Problemy z zakupami (Zakres 3.1, w oparciu o wydatki) |
/energy |
POST | Emisje związane ze zużyciem energii (Zakres 2) |
/compute |
POST | Emisje z przetwarzania w chmurze |
Struktura zapytania ofertowego
# Esempio di richiesta POST /estimate
{
"emission_factor": {
"activity_id": "electricity-supply_grid-source_residual_mix",
"data_version": "^21",
"region": "IT"
},
"parameters": {
"energy": 1000,
"energy_unit": "kWh"
}
}
# Risposta
{
"co2e": 0.415,
"co2e_unit": "kg",
"co2e_calculation_method": "ar5",
"co2e_calculation_origin": "source",
"emission_factor": {
"activity_id": "electricity-supply_grid-source_residual_mix",
"source": "IEA",
"year": 2022,
"region": "IT",
"category": "Electricity",
"lca_activity": "electricity_generation",
"data_quality_flags": []
},
"constituent_gases": {
"co2e_total": 0.415,
"co2e_other": null,
"co2": 0.415,
"ch4": null,
"n2o": null
}
}Wersja danych: Semantyka i najlepsze praktyki
Pole data_version kontroluje, której wersji bazy danych użyć.
Karetka (^21) użyj wersji 21 lub wyższej kompatybilnej, zapewniając
automatyczne aktualizacje współczynników emisji. W produkcji blok
dokładna wersja (np. "21") dla powtarzalności
obliczenia i ścieżki audytu. Celowo uaktualnij oprogramowanie do nowych wersji
z wyraźnym przeliczeniem wartości historycznych.
3. Model danych: Baza danych współczynników emisji
Struktura współczynnika emisji
Współczynnik emisji w Climatiq jest jednoznacznie identyfikowany przez
activity_id, source, region, year
e lca_activity. Zrozumienie tej struktury ma kluczowe znaczenie
aby wybrać właściwy współczynnik.
# Struttura di un Emission Factor nel database Climatiq
{
"activity_id": "fuel_type-diesel",
"uuid": "94de5038-8b06-4e24-8e8c-1b87e1e0",
"name": "Diesel",
"category": "Fuel",
"sector": "Transport",
"source": "DEFRA",
"source_link": "https://www.gov.uk/guidance/ghg-conversion-factors-for-company-reporting",
"source_dataset": "DEFRA 2023",
"year": 2023,
"year_released": 2023,
"region": "GB",
"region_name": "United Kingdom",
"description": "Diesel combustion emission factor",
"unit_type": ["Volume", "Weight"],
"supported_calculation_methods": ["ar5"],
"factor": 2.5179,
"factor_calculation_method": "ar5",
"factor_calculation_origin": "source",
"constituent_gases": {
"co2e_total": 2.5179,
"co2": 2.5148,
"ch4": 0.0009,
"n2o": 0.0022
}
}Główne źródła danych
| Źródło | Kraj/region | Aktualizacja | Objęte sektory |
|---|---|---|---|
| DEFRA/BEIS | UK | Coroczny | Energia, transport, zakupy, materiały |
| Agencja Ochrony Środowiska | USA | Coroczny | Energetyka, procesy przemysłowe, rolnictwo |
| MAE | Światowy | Coroczny | Energia elektryczna według kraju, energia pierwotna |
| ekowynalazek | Światowy | Półrocznie | Kompletna LCA, łańcuch dostaw, materiały |
| ADEME | Francja | Coroczny | Oparty na węglu, transport, energia FR |
| EOG | EU | Coroczny | Sieć elektryczna Europa, emisje sektorowe |
| ISPRA | Włochy | Coroczny | Krajowa inwentaryzacja gazów cieplarnianych, miks energetyczny IT |
4. Szacowanie oparte na działaniu: obliczenia dotyczące konkretnych działań
Podejście oparte na działaniu jest najdokładniejszy: opiera się na podstawowe dane dotyczące działalności, takie jak litry zużytego paliwa, kWh energii elektrycznej, przejechanych kilometrów czy ton zakupionego materiału. Wymaga gromadzenia danych operacyjnych szczegółowe, ale dają naukowo solidne wyniki.
Przykład: Obliczanie emisji z floty firmy (Zakres 1)
# activity_based_estimation.py
import httpx
from dataclasses import dataclass
from enum import Enum
class FuelType(str, Enum):
DIESEL = "fuel_type-diesel"
PETROL = "fuel_type-petrol"
HVO = "fuel_type-hvo_biodiesel"
LPG = "fuel_type-lpg"
CNG = "fuel_type-cng"
@dataclass
class VehicleTrip:
vehicle_id: str
fuel_type: FuelType
litres_consumed: float
region: str = "IT"
async def estimate_fleet_scope1(
api_key: str,
trips: list[VehicleTrip],
data_version: str = "^21"
) -> dict:
"""
Calcola Scope 1 per flotta aziendale con activity-based estimation.
Returns aggregato e dettaglio per veicolo.
"""
BASE_URL = "https://beta3.api.climatiq.io"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Costruisce le richieste batch (max 100 per chiamata)
batch_requests = [
{
"emission_factor": {
"activity_id": trip.fuel_type.value,
"data_version": data_version,
"region": trip.region
},
"parameters": {
"volume": trip.litres_consumed,
"volume_unit": "l"
}
}
for trip in trips
]
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{BASE_URL}/batch/estimate",
headers=headers,
json={"requests": batch_requests}
)
response.raise_for_status()
results = response.json()["results"]
# Aggrega risultati per veicolo
vehicle_emissions = {}
total_co2e_kg = 0.0
for trip, result in zip(trips, results):
if "error" in result:
print(f"Errore per veicolo {trip.vehicle_id}: {result['error']}")
continue
co2e_kg = result["co2e"]
total_co2e_kg += co2e_kg
if trip.vehicle_id not in vehicle_emissions:
vehicle_emissions[trip.vehicle_id] = {
"total_co2e_kg": 0.0,
"fuel_type": trip.fuel_type.value,
"trips": 0
}
vehicle_emissions[trip.vehicle_id]["total_co2e_kg"] += co2e_kg
vehicle_emissions[trip.vehicle_id]["trips"] += 1
return {
"scope": "scope_1",
"total_co2e_kg": total_co2e_kg,
"total_co2e_tco2e": total_co2e_kg / 1000,
"vehicles": vehicle_emissions,
"vehicle_count": len(vehicle_emissions)
}
# Utilizzo
import asyncio
async def main():
trips = [
VehicleTrip("VAN-001", FuelType.DIESEL, 120.5),
VehicleTrip("VAN-002", FuelType.DIESEL, 98.3),
VehicleTrip("TRUCK-001", FuelType.HVO, 245.0),
VehicleTrip("CAR-001", FuelType.PETROL, 42.1),
]
result = await estimate_fleet_scope1(
api_key="clq_live_your_key_here",
trips=trips
)
print(f"Scope 1 totale: {result['total_co2e_tco2e']:.2f} tCO2e")
for vid, data in result["vehicles"].items():
print(f" {vid}: {data['total_co2e_kg']:.1f} kg CO2e")
asyncio.run(main())Obliczanie emisji energii elektrycznej (zakres 2)
Zakres 2 wymaga obliczeń przy użyciu dwóch metod: opartej na lokalizacji (miks sieci) i rynkowe (resztkowy współczynnik mieszania lub PPA). Obie wartości należy zgłosić w pliku Protokół dotyczący ujawniania gazów cieplarnianych.
# scope2_electricity.py
from enum import Enum
class Scope2Method(str, Enum):
LOCATION_BASED = "location_based"
MARKET_BASED = "market_based"
async def estimate_scope2_electricity(
client: httpx.AsyncClient,
api_key: str,
kwh_consumed: float,
region: str,
method: Scope2Method,
renewable_percentage: float = 0.0,
data_version: str = "^21"
) -> dict:
"""
Scope 2 con metodo location-based o market-based.
- location_based: usa il mix elettrico della rete locale
- market_based: usa residual mix o certificati (IRECs, GOs)
"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Market-based: se 100% rinnovabile certificato, le emissioni sono zero
if method == Scope2Method.MARKET_BASED and renewable_percentage >= 100.0:
return {
"co2e_kg": 0.0,
"co2e_tco2e": 0.0,
"method": method.value,
"note": "100% energie rinnovabili certificate - emissioni zero"
}
# Activity ID diverso per location-based vs market-based
if method == Scope2Method.LOCATION_BASED:
activity_id = "electricity-supply_grid-source_residual_mix"
else:
# Market-based usa residual mix (esclude RES certificate)
activity_id = "electricity-supply_grid-source_residual_mix"
# Riduzione proporzionale per rinnovabili parziali (market-based)
effective_kwh = kwh_consumed
if method == Scope2Method.MARKET_BASED and renewable_percentage > 0:
effective_kwh = kwh_consumed * (1 - renewable_percentage / 100.0)
payload = {
"emission_factor": {
"activity_id": activity_id,
"data_version": data_version,
"region": region
},
"parameters": {
"energy": effective_kwh,
"energy_unit": "kWh"
}
}
response = await client.post(
"https://beta3.api.climatiq.io/estimate",
headers=headers,
json=payload
)
response.raise_for_status()
data = response.json()
return {
"co2e_kg": data["co2e"],
"co2e_tco2e": data["co2e"] / 1000,
"co2e_unit": data["co2e_unit"],
"method": method.value,
"kwh_consumed": kwh_consumed,
"effective_kwh": effective_kwh,
"renewable_percentage": renewable_percentage,
"emission_factor": data["emission_factor"],
"region": region
}5. Szacowanie oparte na wydatkach: obliczenia na podstawie danych dotyczących wydatków
Jeśli nie masz dokładnych danych dotyczących aktywności, podejście oparte na wydatkach wykorzystuje wydatki walutowe jako wskaźnik aktywności. Climatiq stosuje współczynniki emisji ekonomiczny (kg CO₂e na wydane euro/dolar) według kategorii zakupu, na podstawie danych wejście-wyjście z tabel OECD lub EXIOBASE. Jest to najczęstsza metoda dla Zakresu 3 kategoria 1 (Zakupione towary i usługi).
Dokładność oparta na wydatkach a dokładność na podstawie aktywności
Metody oparte na wydatkach pozwalają uzyskać szacunki z niepewnością wynoszącą: 30-100%, w porównaniu do 5-15% metod opartych na aktywności. Używaj ich tylko wtedy, gdy nie masz danych prawybory. Protokół GHG akceptuje je jako punkt wyjścia, ale wymaga postępu ulepszenia w zakresie danych dotyczących aktywności.
# scope3_spend_based.py
from typing import Optional
import httpx
# Mappa categorie NACE su activity_id Climatiq per spend-based
CATEGORY_TO_ACTIVITY_ID = {
# Settore IT e servizi digitali
"it_services": "professional_services-type_professional_services",
"cloud_hosting": "professional_services-type_professional_services",
"software_licenses": "professional_services-type_professional_services",
# Logistica e trasporti
"freight_road": "transport-type_freight_vehicle",
"freight_air": "transport-type_air_freight",
"courier_services": "transport-type_freight_vehicle",
# Produzione e manifattura
"raw_materials_steel": "steel-type_steel_products",
"raw_materials_plastic": "plastics-type_plastic_products",
"packaging": "paper-type_paper_products",
# Servizi professionali
"legal_services": "professional_services-type_professional_services",
"consulting": "professional_services-type_professional_services",
"marketing": "professional_services-type_professional_services",
# Utilities e energia
"electricity_bill": "electricity-supply_grid-source_residual_mix",
"gas_bill": "fuel_type-natural_gas",
}
async def calculate_scope3_spend_based(
api_key: str,
purchases: list[dict],
region: str = "IT",
currency: str = "EUR",
data_version: str = "^21"
) -> dict:
"""
Calcola Scope 3.1 con metodo spend-based.
purchases: [{"category": "it_services", "amount": 50000}, ...]
"""
BASE_URL = "https://beta3.api.climatiq.io"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
batch_requests = []
for purchase in purchases:
activity_id = CATEGORY_TO_ACTIVITY_ID.get(purchase["category"])
if not activity_id:
print(f"Categoria non mappata: {purchase['category']}")
continue
batch_requests.append({
"emission_factor": {
"activity_id": activity_id,
"data_version": data_version,
"region": region
},
"parameters": {
"money": purchase["amount"],
"money_unit": currency.lower()
}
})
# Chunking per batch limite 100
results_all = []
chunk_size = 100
async with httpx.AsyncClient(timeout=60.0) as client:
for i in range(0, len(batch_requests), chunk_size):
chunk = batch_requests[i:i + chunk_size]
response = await client.post(
f"{BASE_URL}/batch/estimate",
headers=headers,
json={"requests": chunk}
)
response.raise_for_status()
results_all.extend(response.json()["results"])
# Aggrega per categoria
total_co2e_kg = 0.0
category_breakdown = {}
for purchase, result in zip(purchases, results_all):
if "error" in result:
category_breakdown[purchase["category"]] = {
"error": result["error"],
"amount": purchase["amount"]
}
continue
co2e_kg = result.get("co2e", 0.0)
total_co2e_kg += co2e_kg
cat = purchase["category"]
if cat not in category_breakdown:
category_breakdown[cat] = {
"total_co2e_kg": 0.0,
"total_spend": 0.0,
"intensity_kg_per_eur": 0.0
}
category_breakdown[cat]["total_co2e_kg"] += co2e_kg
category_breakdown[cat]["total_spend"] += purchase["amount"]
# Calcola intensità per categoria
for cat_data in category_breakdown.values():
if "total_spend" in cat_data and cat_data["total_spend"] > 0:
cat_data["intensity_kg_per_eur"] = (
cat_data["total_co2e_kg"] / cat_data["total_spend"]
)
return {
"scope": "scope_3.1",
"method": "spend_based",
"total_co2e_kg": total_co2e_kg,
"total_co2e_tco2e": total_co2e_kg / 1000,
"currency": currency,
"categories": category_breakdown,
"uncertainty_note": "Incertezza stimata 30-100% (metodo spend-based)"
}6. Integracja protokołu GHG: zakres mapowania 1, 2, 3
Budowa kompletnego systemu rozliczania emisji dwutlenku węgla wymaga mapowania każdego Kategoria protokołu GHG do określonych punktów końcowych i identyfikator aktywności Climatiq. Ta sekcja zapewnia pełne mapowanie dla najbardziej odpowiednich kategorii.
| Zakres Protokół dotyczący gazów cieplarnianych | Kategoria | Punkt końcowy Climatiq | Metoda |
|---|---|---|---|
| Zakres 1 | Spalanie stacjonarne (ogrzewanie) | /estimate |
Aktywność (objętość) |
| Zakres 1 | Spalanie mobilne (flota) | /batch/estimate |
Aktywność (objętość) |
| Zakres 1 | Emisje niezorganizowane (czynniki chłodnicze) | /estimate |
Aktywność (waga) |
| Zakres 2 | Energia elektryczna (w zależności od lokalizacji) | /energy |
Aktywność (kWh) |
| Zakres 2 | Energia elektryczna (na rynku) | /energy |
Aktywność (kWh, mieszanka resztkowa) |
| Zakres 3.1 | Zakupione towary i usługi | /procurement |
Oparte na wydatkach (EUR) |
| Zakres 3.4 | Transport pod prąd | /freight |
Aktywność (ton-km) |
| Zakres 3.6 | Podróże służbowe (samoloty) | /travel/flights |
Aktywność (kod IATA) |
| Zakres 3.6 | Podróż służbowa (samochód/pociąg) | /estimate |
Aktywność (km) |
| Zakres 3.7 | Dojazdy pracowników | /batch/estimate |
Aktywność (km, połowa) |
| Zakres 3.11 | Korzystanie ze sprzedanych produktów | /estimate |
Aktywność/wydatki |
# ghg_protocol_calculator.py
# Sistema unificato per calcolo GHG Protocol completo
from dataclasses import dataclass, field
from typing import Optional
import asyncio
import httpx
@dataclass
class GHGProtocolReport:
"""Report GHG Protocol completo per anno fiscale."""
year: int
company_name: str
reporting_boundary: str = "operational_control"
# Scope 1
scope1_combustion_kg: float = 0.0
scope1_mobile_kg: float = 0.0
scope1_fugitive_kg: float = 0.0
# Scope 2
scope2_location_based_kg: float = 0.0
scope2_market_based_kg: float = 0.0
# Scope 3 (categorie principali)
scope3_cat1_purchased_goods_kg: float = 0.0
scope3_cat4_upstream_transport_kg: float = 0.0
scope3_cat6_business_travel_kg: float = 0.0
scope3_cat7_employee_commuting_kg: float = 0.0
# Metadata per audit trail
calculation_date: Optional[str] = None
data_version: str = ""
sources: list[str] = field(default_factory=list)
@property
def scope1_total_kg(self) -> float:
return (self.scope1_combustion_kg +
self.scope1_mobile_kg +
self.scope1_fugitive_kg)
@property
def scope2_selected_kg(self) -> float:
"""Market-based se disponibile, altrimenti location-based."""
return (self.scope2_market_based_kg
if self.scope2_market_based_kg > 0
else self.scope2_location_based_kg)
@property
def scope3_total_kg(self) -> float:
return (self.scope3_cat1_purchased_goods_kg +
self.scope3_cat4_upstream_transport_kg +
self.scope3_cat6_business_travel_kg +
self.scope3_cat7_employee_commuting_kg)
@property
def grand_total_tco2e(self) -> float:
return (self.scope1_total_kg +
self.scope2_selected_kg +
self.scope3_total_kg) / 1000
def to_csrd_dict(self) -> dict:
"""Output formato CSRD/ESRS E1-6."""
return {
"reporting_year": self.year,
"entity": self.company_name,
"ghg_emissions_location_based": {
"scope_1": self.scope1_total_kg / 1000,
"scope_2_location": self.scope2_location_based_kg / 1000,
"scope_3": self.scope3_total_kg / 1000,
"unit": "tCO2e"
},
"ghg_emissions_market_based": {
"scope_1": self.scope1_total_kg / 1000,
"scope_2_market": self.scope2_market_based_kg / 1000,
"scope_3": self.scope3_total_kg / 1000,
"unit": "tCO2e"
},
"data_version": self.data_version,
"methodology": "GHG Protocol Corporate Standard"
}7. Solidny klient Pythona: ponawianie prób, obsługa pamięci podręcznej i błędów
W środowisku produkcyjnym wywołania API muszą być odporne na przejściowe błędy, szacunek limity prędkości i minimalizować połączenia dzięki efektywnemu buforowaniu. Oto klient gotowy do produkcji.
# climatiq_client.py
import asyncio
import hashlib
import json
import logging
import time
from dataclasses import dataclass
from typing import Any, Optional
import httpx
import redis.asyncio as redis
logger = logging.getLogger(__name__)
class ClimatiqAPIError(Exception):
"""Errore API Climatiq con context."""
def __init__(self, message: str, status_code: int, response_body: dict):
super().__init__(message)
self.status_code = status_code
self.response_body = response_body
class RateLimitError(ClimatiqAPIError):
"""Rate limit superato - aspetta prima di riprovare."""
pass
class ClimatiqClient:
"""
Client asincrono per Climatiq API con:
- Retry automatico con exponential backoff
- Cache Redis per ridurre le chiamate API
- Logging strutturato per audit trail
- Gestione rate limit con rispetto dei retry-after header
"""
BASE_URL = "https://beta3.api.climatiq.io"
MAX_RETRIES = 3
BATCH_SIZE = 100 # Limite Climatiq
def __init__(
self,
api_key: str,
redis_client: Optional[redis.Redis] = None,
cache_ttl: int = 86400, # 24 ore - i fattori cambiano raramente
data_version: str = "^21"
):
self.api_key = api_key
self.redis = redis_client
self.cache_ttl = cache_ttl
self.data_version = data_version
self._http_client: Optional[httpx.AsyncClient] = None
async def __aenter__(self):
self._http_client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0, connect=10.0),
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self
async def __aexit__(self, *args):
if self._http_client:
await self._http_client.aclose()
def _cache_key(self, payload: dict) -> str:
"""Genera chiave cache deterministica da payload."""
payload_str = json.dumps(payload, sort_keys=True)
return f"climatiq:{hashlib.sha256(payload_str.encode()).hexdigest()[:16]}"
async def _get_cached(self, key: str) -> Optional[dict]:
"""Recupera risultato dalla cache Redis."""
if not self.redis:
return None
try:
cached = await self.redis.get(key)
if cached:
logger.debug(f"Cache HIT: {key}")
return json.loads(cached)
except Exception as e:
logger.warning(f"Errore cache GET: {e}")
return None
async def _set_cached(self, key: str, value: dict) -> None:
"""Salva risultato nella cache Redis."""
if not self.redis:
return
try:
await self.redis.setex(key, self.cache_ttl, json.dumps(value))
logger.debug(f"Cache SET: {key} (TTL: {self.cache_ttl}s)")
except Exception as e:
logger.warning(f"Errore cache SET: {e}")
async def _request_with_retry(
self, method: str, endpoint: str, payload: dict
) -> dict:
"""HTTP request con retry e exponential backoff."""
url = f"{self.BASE_URL}{endpoint}"
last_error = None
for attempt in range(self.MAX_RETRIES):
try:
response = await self._http_client.request(
method, url, json=payload
)
if response.status_code == 429: # Rate limit
retry_after = int(response.headers.get("Retry-After", 60))
logger.warning(
f"Rate limit superato. Aspetto {retry_after}s..."
)
await asyncio.sleep(retry_after)
continue
if response.status_code >= 400:
body = response.json() if response.content else {}
raise ClimatiqAPIError(
f"Errore API {response.status_code}: {body.get('error', 'Unknown')}",
status_code=response.status_code,
response_body=body
)
return response.json()
except httpx.NetworkError as e:
wait_time = 2 ** attempt
logger.warning(
f"Network error (attempt {attempt+1}/{self.MAX_RETRIES}), "
f"retry in {wait_time}s: {e}"
)
last_error = e
await asyncio.sleep(wait_time)
raise ConnectionError(
f"Falliti {self.MAX_RETRIES} tentativi: {last_error}"
)
async def estimate(
self,
activity_id: str,
parameters: dict,
region: Optional[str] = None
) -> dict:
"""
Stima singola con cache automatica.
activity_id: es. "electricity-supply_grid-source_residual_mix"
parameters: es. {"energy": 1000, "energy_unit": "kWh"}
"""
payload = {
"emission_factor": {
"activity_id": activity_id,
"data_version": self.data_version,
**({"region": region} if region else {})
},
"parameters": parameters
}
cache_key = self._cache_key(payload)
if cached := await self._get_cached(cache_key):
return cached
result = await self._request_with_retry("POST", "/estimate", payload)
await self._set_cached(cache_key, result)
logger.info(
"Estimate: activity=%(activity)s region=%(region)s "
"co2e=%(co2e).4f kg",
{
"activity": activity_id,
"region": region or "global",
"co2e": result.get("co2e", 0)
}
)
return result
async def batch_estimate(
self, requests: list[dict]
) -> list[dict]:
"""
Batch estimation con chunking automatico (100 per batch max).
"""
all_results = []
for i in range(0, len(requests), self.BATCH_SIZE):
chunk = requests[i:i + self.BATCH_SIZE]
payload = {"requests": chunk}
response = await self._request_with_retry(
"POST", "/batch/estimate", payload
)
all_results.extend(response.get("results", []))
# Piccola pausa tra chunk grandi per rispettare rate limit
if len(requests) > self.BATCH_SIZE:
await asyncio.sleep(0.5)
return all_results8. Klient TypeScript/Node.js z Axios
W przypadku aplikacji backendowych Node.js, TypeScript lub NestJS poniżej znajduje się klient z określonym typem który udostępnia tę samą funkcjonalność co klient Pythona z pełnym bezpieczeństwem typu.
// climatiq-client.ts
import axios, { AxiosInstance, AxiosError } from 'axios';
// Types per la Climatiq API
export interface EmissionFactor {
activity_id: string;
data_version: string;
region?: string;
}
export interface EstimateParameters {
energy?: number;
energy_unit?: 'kWh' | 'MWh' | 'GJ';
volume?: number;
volume_unit?: 'l' | 'gallon' | 'm3';
weight?: number;
weight_unit?: 'kg' | 't' | 'lb';
money?: number;
money_unit?: 'eur' | 'usd' | 'gbp';
distance?: number;
distance_unit?: 'km' | 'mi';
}
export interface EstimateRequest {
emission_factor: EmissionFactor;
parameters: EstimateParameters;
}
export interface ConstituentGases {
co2e_total: number;
co2?: number;
ch4?: number;
n2o?: number;
}
export interface EstimateResponse {
co2e: number;
co2e_unit: string;
co2e_calculation_method: string;
emission_factor: {
activity_id: string;
source: string;
year: number;
region: string;
category: string;
data_quality_flags: string[];
};
constituent_gases: ConstituentGases;
}
export interface BatchEstimateResult {
co2e?: number;
co2e_unit?: string;
error?: string;
constituent_gases?: ConstituentGases;
}
export class ClimatiqAPIError extends Error {
constructor(
message: string,
public readonly statusCode: number,
public readonly responseBody: unknown
) {
super(message);
this.name = 'ClimatiqAPIError';
}
}
export class ClimatiqClient {
private readonly http: AxiosInstance;
private readonly cache = new Map<string, { data: unknown; expiresAt: number }>();
private readonly cacheTtlMs: number;
constructor(
private readonly apiKey: string,
private readonly dataVersion = '^21',
cacheTtlSeconds = 3600
) {
this.cacheTtlMs = cacheTtlSeconds * 1000;
this.http = axios.create({
baseURL: 'https://beta3.api.climatiq.io',
headers: {
Authorization: `Bearer ${apiKey}`,
'Content-Type': 'application/json',
},
timeout: 30_000,
});
}
private cacheKey(payload: unknown): string {
return JSON.stringify(payload);
}
private getFromCache<T>(key: string): T | null {
const entry = this.cache.get(key);
if (!entry) return null;
if (Date.now() > entry.expiresAt) {
this.cache.delete(key);
return null;
}
return entry.data as T;
}
private setCache(key: string, data: unknown): void {
this.cache.set(key, {
data,
expiresAt: Date.now() + this.cacheTtlMs,
});
}
private handleAxiosError(error: AxiosError): never {
if (error.response) {
const status = error.response.status;
const body = error.response.data as Record<string, unknown>;
if (status === 429) {
throw new ClimatiqAPIError(
'Rate limit superato. Riprova tra qualche momento.',
429,
body
);
}
throw new ClimatiqAPIError(
`Climatiq API error ${status}: ${body['error'] ?? 'Unknown'}`,
status,
body
);
}
throw new ClimatiqAPIError(
`Network error: ${error.message}`,
0,
null
);
}
async estimate(
activityId: string,
parameters: EstimateParameters,
region?: string
): Promise<EstimateResponse> {
const payload: EstimateRequest = {
emission_factor: {
activity_id: activityId,
data_version: this.dataVersion,
...(region ? { region } : {}),
},
parameters,
};
const key = this.cacheKey(payload);
const cached = this.getFromCache<EstimateResponse>(key);
if (cached) return cached;
try {
const { data } = await this.http.post<EstimateResponse>(
'/estimate',
payload
);
this.setCache(key, data);
return data;
} catch (error) {
if (axios.isAxiosError(error)) this.handleAxiosError(error);
throw error;
}
}
async batchEstimate(
requests: EstimateRequest[]
): Promise<BatchEstimateResult[]> {
const CHUNK_SIZE = 100;
const allResults: BatchEstimateResult[] = [];
for (let i = 0; i < requests.length; i += CHUNK_SIZE) {
const chunk = requests.slice(i, i + CHUNK_SIZE);
try {
const { data } = await this.http.post<{ results: BatchEstimateResult[] }>(
'/batch/estimate',
{ requests: chunk }
);
allResults.push(...data.results);
} catch (error) {
if (axios.isAxiosError(error)) this.handleAxiosError(error);
throw error;
}
// Throttle tra chunk multipli
if (i + CHUNK_SIZE < requests.length) {
await new Promise(resolve => setTimeout(resolve, 200));
}
}
return allResults;
}
async estimateFlight(
originIata: string,
destinationIata: string,
passengers: number,
cabinClass: 'economy' | 'business' | 'first' = 'economy'
): Promise<EstimateResponse> {
try {
const { data } = await this.http.post<EstimateResponse>(
'/travel/flights',
{
legs: [{
from: originIata,
to: destinationIata,
passengers,
cabin_class: cabinClass,
}],
}
);
return data;
} catch (error) {
if (axios.isAxiosError(error)) this.handleAxiosError(error);
throw error;
}
}
}Korzystanie z klienta TypeScript
// usage-example.ts
import { ClimatiqClient } from './climatiq-client';
const client = new ClimatiqClient(
process.env['CLIMATIQ_API_KEY'] ?? '',
'^21',
3600
);
// Calcolo emissioni elettricità ufficio
async function calculateOfficeElectricity() {
const result = await client.estimate(
'electricity-supply_grid-source_residual_mix',
{ energy: 5000, energy_unit: 'kWh' },
'IT'
);
console.log(`Emissioni ufficio: ${result.co2e.toFixed(2)} kg CO2e`);
console.log(`Fonte: ${result.emission_factor.source} (${result.emission_factor.year})`);
return result;
}
// Calcolo batch per fleet management
async function calculateFleetEmissions(
vehicles: Array<{ id: string; litres: number; fuel: string }>
) {
const requests = vehicles.map(v => ({
emission_factor: {
activity_id: `fuel_type-${v.fuel}`,
data_version: '^21',
region: 'IT',
},
parameters: {
volume: v.litres,
volume_unit: 'l' as const,
},
}));
const results = await client.batchEstimate(requests);
return vehicles.map((v, i) => ({
vehicleId: v.id,
co2eKg: results[i].co2e ?? 0,
error: results[i].error,
}));
}
// Calcolo volo business travel
async function calculateBusinessFlight() {
const result = await client.estimateFlight('MXP', 'LHR', 2, 'economy');
console.log(`Volo MXP-LHR (2 pax): ${result.co2e.toFixed(1)} kg CO2e`);
}9. Kalkulator emisji dwutlenku węgla w czasie rzeczywistym dla SaaS
Potężnym przypadkiem użycia Climatiq jest dodanie pliku etykieta węglowa do produktów e-commerce lub SaaS: pokaż użytkownikowi wpływ emisji dwutlenku węgla wycena zlecenia lub działania przed jego zatwierdzeniem. Zwiększa to przejrzystość i wspiera świadome wybory konsumentów.
Architektura: API etykiet węglowych dla handlu elektronicznego
# carbon_label_api.py
# FastAPI endpoint per carbon label real-time su prodotti e-commerce
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from pydantic import BaseModel, Field
from typing import Optional
import asyncio
app = FastAPI(title="Carbon Label API", version="1.0.0")
class ProductCarbonRequest(BaseModel):
"""Richiesta calcolo carbon label per carrello."""
items: list[dict] = Field(
description="Lista prodotti con categoria e peso",
example=[
{"product_id": "SKU-123", "category": "electronics", "weight_kg": 0.5, "quantity": 1},
{"product_id": "SKU-456", "category": "clothing", "weight_kg": 0.3, "quantity": 2}
]
)
shipping_method: str = Field(
default="road",
description="Metodo spedizione: road, air, sea"
)
destination_region: str = Field(
default="IT",
description="Regione destinazione ISO 3166-2"
)
origin_region: str = Field(
default="CN",
description="Regione origine/produzione"
)
class CarbonLabelResponse(BaseModel):
"""Risposta con carbon label completa."""
total_co2e_kg: float
breakdown: dict
label: str # "A", "B", "C", "D", "E" come etichetta energetica EU
label_color: str
offset_cost_eur: float # Stima costo compensazione
trees_equivalent: float # Equivalente alberi annui
km_car_equivalent: float # Equivalente km auto
# Mappatura categoria prodotto su activity_id per produzione
PRODUCT_CATEGORY_PRODUCTION = {
"electronics": "electrical_equipment-type_small_electronics",
"clothing": "textiles-type_clothing",
"food": "food-type_mixed",
"furniture": "furniture-type_mixed",
"books": "paper-type_books",
"plastics": "plastics-type_plastic_products",
}
# Mappatura metodo spedizione su activity_id
SHIPPING_ACTIVITY_ID = {
"road": "transport-type_freight_vehicle-fuel_source_diesel-vehicle_type_hgv",
"air": "transport-type_air_freight",
"sea": "transport-type_sea_freight-route_type_container_ship",
}
def calculate_carbon_label(co2e_kg: float) -> tuple[str, str]:
"""
Calcola etichetta A-E basata su impatto carbonico.
Soglie ispirate alla proposta EU eco-label per e-commerce.
"""
if co2e_kg < 0.5:
return "A", "#2ecc71" # Verde - impatto molto basso
elif co2e_kg < 1.5:
return "B", "#27ae60" # Verde scuro - impatto basso
elif co2e_kg < 5.0:
return "C", "#f39c12" # Arancione - impatto medio
elif co2e_kg < 15.0:
return "D", "#e67e22" # Arancione scuro - impatto alto
else:
return "E", "#e74c3c" # Rosso - impatto molto alto
@app.post("/api/v1/carbon-label", response_model=CarbonLabelResponse)
async def get_carbon_label(
request: ProductCarbonRequest,
climatiq: ClimatiqClient = Depends(get_climatiq_client)
):
"""
Calcola carbon label real-time per un carrello e-commerce.
Considera produzione + imballaggio + spedizione.
"""
batch_requests = []
# 1. Emissioni produzione per ciascun prodotto
for item in request.items:
activity_id = PRODUCT_CATEGORY_PRODUCTION.get(
item["category"],
"manufactured_goods-type_mixed" # Fallback generico
)
# Calcolo per peso totale (qty * peso)
total_weight = item["weight_kg"] * item.get("quantity", 1)
batch_requests.append({
"emission_factor": {
"activity_id": activity_id,
"data_version": "^21",
"region": request.origin_region
},
"parameters": {
"weight": total_weight,
"weight_unit": "kg"
}
})
# 2. Emissioni spedizione (distanza stimata)
total_weight_kg = sum(
i["weight_kg"] * i.get("quantity", 1)
for i in request.items
)
# Stima distanza basata su regioni (semplificata)
estimated_distance_km = _estimate_distance(
request.origin_region,
request.destination_region
)
shipping_activity = SHIPPING_ACTIVITY_ID.get(
request.shipping_method,
SHIPPING_ACTIVITY_ID["road"]
)
# Freight: tonne * km = tonne-km
tonne_km = (total_weight_kg / 1000) * estimated_distance_km
batch_requests.append({
"emission_factor": {
"activity_id": shipping_activity,
"data_version": "^21"
},
"parameters": {
"weight": total_weight_kg,
"weight_unit": "kg",
"distance": estimated_distance_km,
"distance_unit": "km"
}
})
# Chiamata batch a Climatiq
try:
results = await climatiq.batch_estimate(batch_requests)
except ClimatiqAPIError as e:
raise HTTPException(
status_code=502,
detail=f"Errore calcolo emissioni: {str(e)}"
)
# Aggrega risultati
production_co2e = sum(
r.get("co2e", 0) for r in results[:-1] # Tutti tranne ultimo (spedizione)
)
shipping_co2e = results[-1].get("co2e", 0) if results else 0
total_co2e = production_co2e + shipping_co2e
label, color = calculate_carbon_label(total_co2e)
# Calcola equivalenze intuitive per l'utente
# Offset market rate ~15 EUR/tCO2e (mercato volontario 2025)
offset_cost = (total_co2e / 1000) * 15.0
# Un albero assorbe ~22 kg CO2/anno
trees_equivalent = total_co2e / 22.0
# Auto media emette ~0.21 kg CO2/km
km_car = total_co2e / 0.21
return CarbonLabelResponse(
total_co2e_kg=round(total_co2e, 3),
breakdown={
"production_kg": round(production_co2e, 3),
"shipping_kg": round(shipping_co2e, 3),
"shipping_method": request.shipping_method,
"distance_km": estimated_distance_km
},
label=label,
label_color=color,
offset_cost_eur=round(offset_cost, 2),
trees_equivalent=round(trees_equivalent, 1),
km_car_equivalent=round(km_car, 1)
)
def _estimate_distance(origin: str, destination: str) -> float:
"""Stima distanza tra regioni in km (lookup semplificato)."""
DISTANCES = {
("CN", "IT"): 9_000,
("DE", "IT"): 950,
("IT", "IT"): 300,
("US", "IT"): 8_500,
("IN", "IT"): 7_200,
}
key = (origin[:2].upper(), destination[:2].upper())
return DISTANCES.get(key, 5_000) # Default 5000km se sconosciutoWidget frontendowy pokazujący etykietę węgla
// carbon-label.component.ts (Angular/TypeScript)
import { Component, Input, OnChanges, SimpleChanges } from '@angular/core';
import { HttpClient } from '@angular/common/http';
interface CarbonLabel {
total_co2e_kg: number;
label: string;
label_color: string;
trees_equivalent: number;
km_car_equivalent: number;
offset_cost_eur: number;
}
@Component({
selector: 'app-carbon-label',
template: `
<div class="carbon-label" *ngIf="carbonData">
<div class="label-badge" [style.background]="carbonData.label_color">
{{ carbonData.label }}
</div>
<div class="carbon-info">
<span class="co2e">{{ carbonData.total_co2e_kg | number:'1.1-2' }} kg CO₂e</span>
<span class="equivalent">
= {{ carbonData.km_car_equivalent | number:'1.0-0' }} km in auto
</span>
<button (click)="offsetCarbon()" class="offset-btn">
Compensa per €{{ carbonData.offset_cost_eur | number:'1.2-2' }}
</button>
</div>
</div>
`
})
export class CarbonLabelComponent implements OnChanges {
@Input() cartItems: Array<{ product_id: string; category: string; weight_kg: number; quantity: number }> = [];
@Input() shippingMethod = 'road';
carbonData: CarbonLabel | null = null;
loading = false;
constructor(private http: HttpClient) {}
ngOnChanges(changes: SimpleChanges): void {
if (changes['cartItems'] || changes['shippingMethod']) {
this.loadCarbonLabel();
}
}
private loadCarbonLabel(): void {
if (!this.cartItems.length) return;
this.loading = true;
this.http.post<CarbonLabel>('/api/v1/carbon-label', {
items: this.cartItems,
shipping_method: this.shippingMethod,
destination_region: 'IT',
origin_region: 'CN',
}).subscribe({
next: (data) => {
this.carbonData = data;
this.loading = false;
},
error: (err) => {
console.error('Errore carbon label:', err);
this.loading = false;
},
});
}
offsetCarbon(): void {
// Integrazione con piattaforma di offsetting
window.open('https://example.com/offset', '_blank');
}
}10. Testowanie i próbne API
W CI/CD nie chcesz rzeczywistych wywołań API Climatiq. Oto jak to ustrukturyzować testy z solidnymi próbami, które symulują zarówno reakcje na sukces, jak i na niepowodzenie.
# tests/conftest.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from app.climatiq_client import ClimatiqClient
@pytest.fixture
def mock_climatiq_client():
"""
Mock del client Climatiq per test unitari.
Simula risposte realistiche senza chiamate API reali.
"""
client = AsyncMock(spec=ClimatiqClient)
# Risposta standard per stima elettricità italiana
electricity_response = {
"co2e": 415.0,
"co2e_unit": "kg",
"co2e_calculation_method": "ar5",
"emission_factor": {
"activity_id": "electricity-supply_grid-source_residual_mix",
"source": "IEA",
"year": 2022,
"region": "IT",
"category": "Electricity",
"data_quality_flags": []
},
"constituent_gases": {
"co2e_total": 415.0,
"co2": 415.0,
"ch4": None,
"n2o": None
}
}
# Risposta per diesel
diesel_response = {
"co2e": 302.15,
"co2e_unit": "kg",
"co2e_calculation_method": "ar5",
"emission_factor": {
"activity_id": "fuel_type-diesel",
"source": "DEFRA",
"year": 2023,
"region": "IT",
"category": "Fuel"
},
"constituent_gases": {
"co2e_total": 302.15,
"co2": 301.5,
"ch4": 0.02,
"n2o": 0.63
}
}
# Configura mock per rispondere in base all'activity_id
async def mock_estimate(activity_id, parameters, region=None):
if "electricity" in activity_id:
kwh = parameters.get("energy", 1000)
return {**electricity_response, "co2e": kwh * 0.415}
elif "diesel" in activity_id:
litres = parameters.get("volume", 100)
return {**diesel_response, "co2e": litres * 2.52}
return electricity_response
client.estimate = AsyncMock(side_effect=mock_estimate)
client.batch_estimate = AsyncMock(return_value=[electricity_response])
return client
# tests/test_activity_based.py
import pytest
from unittest.mock import AsyncMock
@pytest.mark.asyncio
async def test_fleet_scope1_calculates_correctly(mock_climatiq_client):
"""Test che il calcolo Scope 1 flotta produca risultati corretti."""
from app.services.fleet_calculator import calculate_fleet_scope1
result = await calculate_fleet_scope1(
client=mock_climatiq_client,
litres_diesel=100.0,
region="IT"
)
assert result["co2e_kg"] == pytest.approx(252.0, rel=0.01)
assert result["scope"] == "scope_1"
mock_climatiq_client.estimate.assert_called_once()
@pytest.mark.asyncio
async def test_rate_limit_error_is_handled(mock_climatiq_client):
"""Test che rate limit error non lasci il sistema in stato inconsistente."""
from app.climatiq_client import ClimatiqAPIError
mock_climatiq_client.estimate = AsyncMock(
side_effect=ClimatiqAPIError("Rate limit", 429, {"error": "rate_limit"})
)
with pytest.raises(ClimatiqAPIError) as exc_info:
await mock_climatiq_client.estimate(
"electricity-supply_grid-source_residual_mix",
{"energy": 1000, "energy_unit": "kWh"},
"IT"
)
assert exc_info.value.status_code == 429
@pytest.mark.asyncio
async def test_scope2_100_renewable_returns_zero(mock_climatiq_client):
"""Test che 100% rinnovabile dia emissioni zero (market-based)."""
from app.services.scope2_calculator import (
calculate_scope2_electricity, Scope2Method
)
result = await calculate_scope2_electricity(
client=mock_climatiq_client,
kwh_consumed=100_000,
region="IT",
method=Scope2Method.MARKET_BASED,
renewable_percentage=100.0
)
assert result["co2e_kg"] == 0.0
# Non deve chiamare l'API (nessun consumo da rete)
mock_climatiq_client.estimate.assert_not_called()11. Strategia szacowania wsadowego i buforowania dla Scali
Dla aplikacji przetwarzających tysiące rekordów (raporty miesięczne, analizy historyczne), Kombinacja wsadowych wywołań API i pamięci podręcznej Redis staje się niezbędna do ograniczenia zarówno czas przetwarzania, jak i koszty API.
# batch_processor.py
# Elaborazione batch per calcoli di emissioni su larga scala
import asyncio
from datetime import datetime
from typing import AsyncIterator
import redis.asyncio as redis
async def process_monthly_emissions_report(
api_key: str,
records: list[dict],
redis_url: str = "redis://localhost:6379"
) -> dict:
"""
Elabora report mensile emissioni per grandi dataset.
records: lista di attività (flotta, energia, acquisti, etc.)
Restituisce: aggregato mensile Scope 1, 2, 3.
"""
redis_client = await redis.from_url(redis_url)
async with ClimatiqClient(
api_key=api_key,
redis_client=redis_client,
cache_ttl=86400 * 7, # 7 giorni per fattori stabili
data_version="21" # Versione fissa per riproducibilità
) as client:
# Raggruppa per tipo di emissione
scope1_records = [r for r in records if r["scope"] == "scope1"]
scope2_records = [r for r in records if r["scope"] == "scope2"]
scope3_records = [r for r in records if r["scope"] == "scope3"]
# Elabora in parallelo i tre scope
scope1_result, scope2_result, scope3_result = await asyncio.gather(
_process_scope1_batch(client, scope1_records),
_process_scope2_batch(client, scope2_records),
_process_scope3_batch(client, scope3_records)
)
total = (
scope1_result["total_co2e_kg"] +
scope2_result["total_co2e_kg"] +
scope3_result["total_co2e_kg"]
)
await redis_client.aclose()
return {
"report_generated_at": datetime.utcnow().isoformat(),
"total_co2e_kg": total,
"total_co2e_tco2e": total / 1000,
"scope1": scope1_result,
"scope2": scope2_result,
"scope3": scope3_result,
"record_count": len(records)
}
async def _process_scope1_batch(
client: ClimatiqClient, records: list[dict]
) -> dict:
"""Elabora batch Scope 1 con chunking automatico."""
if not records:
return {"total_co2e_kg": 0.0, "record_count": 0}
batch_requests = [
{
"emission_factor": {
"activity_id": r["activity_id"],
"data_version": "21",
"region": r.get("region", "IT")
},
"parameters": {
r["param_key"]: r["param_value"],
f"{r['param_key']}_unit": r["param_unit"]
}
}
for r in records
]
results = await client.batch_estimate(batch_requests)
total = sum(r.get("co2e", 0) for r in results if "error" not in r)
errors = [r for r in results if "error" in r]
if errors:
import logging
logging.warning(f"{len(errors)} errori nel batch Scope 1")
return {
"total_co2e_kg": total,
"record_count": len(records),
"error_count": len(errors)
}12. Alternatywy dla Climatiq: porównanie
Climatiq nie jest jedyną opcją. Oto porównanie głównych alternatywy, które pomogą Ci wybrać właściwe rozwiązanie:
| Rozwiązanie | Główna siła | Limit | Idealny przypadek użycia | Darmowy plan |
|---|---|---|---|---|
| Klimat | Ponad 190 000 czynników, ponad 40 źródeł, solidne API | 250 połączeń miesięcznie w planie wspólnotowym | Przedsiębiorstwo, produkcja wielozakresowa | 250 połączeń miesięcznie |
| Interfejs węglowy | Proste API, świetne doświadczenie programistyczne | Mniej czynników, skoncentruj się na USA | Startupy, e-commerce, wysyłka | Tak (ograniczone) |
| Otwarte współczynniki emisji | Bezpłatny, open source, obsługiwany przez Climatiq | Tylko zestawy danych, brak bezpośredniego interfejsu API REST | Badania, dostęp do surowych danych | Tak (zbiór danych) |
| API przełomu | Gotowe do audytu funkcje korporacyjne | Tylko dla przedsiębiorstw, drogie | Duże firmy, raportowanie CSRD | No |
| Baza danych lotów EPA | Bezpłatnie, rząd USA | Tylko USA, nie REST API | Amerykańskie raporty, badania | Tak (zbiór danych) |
| Niestandardowa baza danych (DEFRA/IEA) | Pełna kontrola, brak kosztów API | Drogie utrzymanie wewnętrzne | Duże organizacje z dedykowanym zespołem | Tak (dane publiczne) |
Kiedy używać niestandardowej bazy danych
W niektórych przypadkach sensowne jest zbudowanie wewnętrznej bazy danych wskaźników emisji. Oto schemat PostgreSQL do obsługi tego problemu:
-- Schema PostgreSQL per database fattori di emissione personalizzato
CREATE TABLE IF NOT EXISTS emission_factors (
id SERIAL PRIMARY KEY,
activity_category VARCHAR(200) NOT NULL,
activity_name VARCHAR(500),
region_code VARCHAR(10) NOT NULL,
reference_year INTEGER NOT NULL,
unit_type VARCHAR(50) NOT NULL, -- 'litre', 'kWh', 'tonne', 'EUR'
factor_kg_co2e_per_unit DECIMAL(12, 6) NOT NULL,
source VARCHAR(100) NOT NULL, -- 'DEFRA', 'ISPRA', 'EPA'
source_version VARCHAR(50),
source_priority INTEGER DEFAULT 1,
constituent_co2 DECIMAL(12, 6),
constituent_ch4 DECIMAL(12, 6),
constituent_n2o DECIMAL(12, 6),
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_ef_lookup ON emission_factors
(activity_category, region_code, reference_year, unit_type);
-- Query di lookup con fallback regionale
SELECT factor_kg_co2e_per_unit, source, reference_year
FROM emission_factors
WHERE activity_category = $1
AND reference_year = $2
AND unit_type = $3
AND region_code IN ($4, 'EU', 'GLOBAL') -- Fallback a EU poi globale
ORDER BY
CASE region_code
WHEN $4 THEN 1 -- Regione specifica prima
WHEN 'EU' THEN 2 -- Poi EU
ELSE 3 -- Poi globale
END,
source_priority DESC,
reference_year DESC
LIMIT 1;Kiedy niestandardowa baza danych NIE jest właściwym wyborem
Budowa i utrzymywanie bazy danych wskaźników emisji wymaga: dedykowaną osobę na 2-3 miesiące do wstępnej budowy, obowiązkowe aktualizacje półroczne (czynniki zmieniają się co roku), zarządzanie wersjami ścieżki audytu i często zasięg geograficzny brak w przypadku regionów nienależących do OECD. Dla większości firm Climatiq oferuje doskonały zwrot z inwestycji nawet biorąc pod uwagę koszt API.
13. Studium przypadku: Platforma e-commerce z etykietą węglową
EcoShop Włochy to platforma e-commerce z 50 000 produktów i 200 000 zamówień miesięcznie. Wymóg: dodaj etykietę węglową do każdego produktu i pokaż szacunkową emisję przy kasie, z opcją offsetu.
Wdrożona architektura
- Wzbogacenie katalogu produktów: Nocna praca, która wstępnie oblicza emisje produkcyjne dla każdego SKU przy użyciu punktu końcowego partii. Wyniki zapisane w bazie produktów z 30-dniowym TTL.
- Zamówienie w czasie rzeczywistym: Przy kasie dynamiczne obliczanie emisji wysyłka na podstawie wagi koszyka, regionu docelowego i wybranej metody. Docelowe opóźnienie: poniżej 200 ms (z pamięcią podręczną Redis dla przesyłek standardowych).
- Sprzedawca deski rozdzielczej Carbon: Miesięczny raport dla sprzedawców z Zakres 3.4 (transport upstream) i szacunkowy Zakres 3.11 (wykorzystanie sprzedanych produktów).
# ecoshop_carbon_service.py
# Servizio di calcolo emissioni per EcoShop Italia
from dataclasses import dataclass
from typing import Optional
import asyncio
from datetime import datetime, timedelta
@dataclass
class ProductEmissionProfile:
"""Profilo emissioni di un prodotto nel catalogo."""
sku: str
production_co2e_kg: float
packaging_co2e_kg: float
category: str
origin_region: str
calculated_at: datetime
climatiq_data_version: str
@property
def total_product_co2e_kg(self) -> float:
return self.production_co2e_kg + self.packaging_co2e_kg
@property
def is_stale(self) -> bool:
"""Profilo è obsoleto se più vecchio di 30 giorni."""
return datetime.utcnow() - self.calculated_at > timedelta(days=30)
class EcoShopCarbonService:
"""
Servizio carbon per EcoShop: gestisce calcoli produzione,
spedizione e report venditori.
"""
def __init__(
self,
climatiq_client: ClimatiqClient,
db_pool, # asyncpg pool
):
self.climatiq = climatiq_client
self.db = db_pool
async def get_product_emission_profile(
self, sku: str
) -> Optional[ProductEmissionProfile]:
"""
Recupera profilo emissioni dal DB.
Se assente o obsoleto, ricalcola via Climatiq.
"""
# Controlla DB
profile = await self._load_from_db(sku)
if profile and not profile.is_stale:
return profile
# Ricalcola
product = await self._get_product_data(sku)
if not product:
return None
new_profile = await self._calculate_product_emissions(product)
await self._save_to_db(new_profile)
return new_profile
async def calculate_checkout_carbon(
self,
cart_items: list[dict],
destination_region: str,
shipping_method: str
) -> dict:
"""
Calcola carbon label per checkout in tempo reale.
Target: < 200ms con cache.
"""
# Recupera profili prodotti (con cache)
profiles = await asyncio.gather(*[
self.get_product_emission_profile(item["sku"])
for item in cart_items
])
# Calcola emissioni produzione
production_co2e = sum(
(profile.total_product_co2e_kg * item.get("quantity", 1))
for profile, item in zip(profiles, cart_items)
if profile is not None
)
# Calcola emissioni spedizione
total_weight_kg = sum(
item.get("weight_kg", 0.5) * item.get("quantity", 1)
for item in cart_items
)
shipping_co2e = await self._estimate_shipping(
total_weight_kg, destination_region, shipping_method
)
total = production_co2e + shipping_co2e
label, color = calculate_carbon_label(total)
return {
"total_co2e_kg": round(total, 3),
"production_co2e_kg": round(production_co2e, 3),
"shipping_co2e_kg": round(shipping_co2e, 3),
"label": label,
"label_color": color,
"offset_price_eur": round((total / 1000) * 15.0, 2),
"generated_at": datetime.utcnow().isoformat()
}
# Risultati dopo 6 mesi
ECOSHOP_METRICS = {
"prodotti_con_carbon_label": 47_832,
"ordini_con_label_mese": 180_000,
"percentuale_utenti_offset": 8.3, # % utenti che comprano offset
"revenue_offset_mensile_eur": 4_200,
"latenza_media_ms": 45, # ms (grazie a cache Redis)
"api_calls_risparmiate_cache": "92%",
"co2e_totale_calcolato_mese_tco2e": 1_240,
"nps_incremento": +12 # punti NPS grazie a trasparenza
}14. Wdrożenie i bezpieczna konfiguracja w produkcji
# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache
class Settings(BaseSettings):
"""Configurazione applicazione da variabili d'ambiente."""
# Climatiq
climatiq_api_key: str
climatiq_data_version: str = "^21"
# Cache
cache_ttl_seconds: int = 3600
redis_url: str = "redis://localhost:6379"
# API Security
api_secret_key: str
allowed_origins: list[str] = ["https://dashboard.tuaazienda.it"]
# Database (per audit trail)
database_url: str = "postgresql://user:pass@localhost/ghg_db"
model_config = SettingsConfigDict(env_file=".env", case_sensitive=False)
@lru_cache
def get_settings() -> Settings:
return Settings()
# .env (NON committare in git - usare secrets manager in produzione)
# CLIMATIQ_API_KEY=clq_live_xxxxxxxxxxxxxxxxxxxxxxx
# CLIMATIQ_DATA_VERSION=^21
# API_SECRET_KEY=generato-con-openssl-rand-hex-32
# CACHE_TTL_SECONDS=3600
# REDIS_URL=redis://redis:6379
# DATABASE_URL=postgresql://ghg_user:password@postgres:5432/ghg_dbNigdy nie ujawniaj klucza API Climatiq
Klucz API Climatiq nie może nigdy znajdować się w kodzie źródłowym, w logach, w odpowiedziach API lub zmiennych JavaScript po stronie klienta. Zawsze używaj: zmienne środowiskowe, menedżer sekretów (AWS Secrets Manager, HashiCorp Vault, Azure Key Vault) lub wpisy tajne Kubernetes z szyfrowaniem w stanie spoczynku. W przypadku przypadkowego odsłonięcia natychmiast przekręć kluczyk.
15. Najlepsze praktyki i antywzorce
Najlepsze praktyki
-
Zablokuj wersję_danych w produkcji: USA
"21"zamiast"^21"aby zapewnić powtarzalność obliczeń. Celowa aktualizacja z wyraźnym ponownym obliczeniem. - Agresywne buforowanie: Współczynniki emisji rzadko się zmieniają. 24-168-godzinna pamięć podręczna radykalnie ogranicza wywołania API, szczególnie do powtarzalnych obliczeń (tabor miesięczny, katalog produktów).
- Zawsze ścieżka audytu: Zapisz każde obliczenie za pomocą Activity_id, zastosowany współczynnik, wersja danych, znacznik czasu i źródło. Niezbędne dla CSRD.
- Jeśli to możliwe, używaj partii: Pojedyncze wywołanie wsadowe z 100 pozycji jest znacznie efektywniejsze niż 100 pojedynczych rozmów. Przestrzega limitów szybkości i zmniejsza całkowite opóźnienia.
- Sprawdź poprawność flag_jakości_danych: Climatiq informuje, kiedy a współczynnik ma niską jakość danych. Rozwiązuj te przypadki, stosując alternatywne czynniki lub notatki niepewności w raporcie.
- Podwójna sprawozdawczość dla Zakresu 2: Wymaga tego Protokół GHG zarówno metody oparte na lokalizacji, jak i metody rynkowe. Oblicz i zapisz oba.
- Pamięć podręczna w pamięci dla TypeScript: Zaimplementuj mapę z TTL w kliencie TypeScript, aby uniknąć zbędnych wywołań w aplikacjach Długowieczny Node.js bez Redis.
Anty-wzorce, których należy unikać
- Klucz API w interfejsie: Nie wystawiaj klucza Climatiq do środka JavaScript po stronie klienta. Zawsze przechodzi przez backend.
- Ignoruj gazy składowe: Aby zapewnić dokładne raportowanie CSRD, CO₂, CH₄ i N₂O należy zgłaszać oddzielnie oprócz całkowitego CO₂e.
- Nieprawidłowe jednostki miary: Climatiq akceptuje konwersje pomiędzy jednostek, ale przekazywanie litrów, gdy punkt końcowy oczekuje kWh, daje wyniki liczbowo wiarygodne, ale naukowo błędne. Zawsze aktualne.
- Partia > 100 szt.: Interfejs API zwraca błąd 422. Zawsze wdrażaj logikę fragmentacji dla dużych zbiorów danych.
-
Ignoruj dopasowanie regionu: Czynnik dla elektryczności
bez określania regionu użyj ustawień globalnych. W przypadku Włoch zawsze używaj
"IT". -
Blokowanie połączeń synchronicznych: Nigdy nie używaj synchronicznych bibliotek HTTP
w asynchronicznych punktach końcowych. Zawsze używaj
httpx.AsyncClientw Pythonie lubaxiosconasync/awaitw TypeScript.
Wnioski i dalsze kroki
Climatiq rozwiązuje najtrudniejszy problem automatycznego rozliczania emisji dwutlenku węgla: the baza danych współczynników emisji. Dzięki ponad 190 000 zweryfikowanych czynników, zasięg 300 regionów i ciągłe aktualizacje, pozwala budować gotowe do produkcji systemy obliczania emisji gazów cieplarnianych w ciągu dni zamiast miesięcy.
W tym artykule zbudowaliśmy:
- Un Solidny klient Pythona z ponowną próbą, pamięcią podręczną Redis i obsługą błędów wpisanych
- Un Klient TypeScript/Node.js z Axios i pełnym bezpieczeństwem typu dla integracji frontendowych
- Obliczenia dla Zakres 1 (flota diesla/HVO), Zakres 2 (metoda podwójna) i główne kategorie Zakres 3
- Un Carbon Label API w czasie rzeczywistym dla e-commerce z etykietą A-E i offsetami
- Testowanie z próbne API dla środowisk CI/CD bez rzeczywistych połączeń
- Strategie przetwarzanie wsadowe i buforowanie dla skali przedsiębiorstwa
Kontynuacja serii zielonego oprogramowania
- Poprzedni artykuł: CodeCarbon – Zmierz emisję kodu działa z biblioteką Pythona o otwartym kodzie źródłowym.
- Następny artykuł: Carbon Aware SDK — jak przenosić obciążenia w godzinach z najczystszą energią przy użyciu prognozowania intensywności sieci.
- Powiązany artykuł (seria MLOps): Optymalizuj szkolenie modeli ML w zakresie redukcji emisji gazów cieplarnianych.
- Powiązany artykuł (seria biznesowa dotycząca danych i sztucznej inteligencji): Zarządzanie danymi dla niezawodnej sztucznej inteligencji - Jak zintegrować wskaźniki zrównoważonego rozwoju z katalogiem danych.
Następnym praktycznym krokiem jest zapisanie się do planu wspólnotowego Climatiq (250 bezpłatnych połączeń miesięcznie), poznaj Eksplorator danych znaleźć czynniki istotne dla Twojej branży i wdrożyć je obliczenie zakresu 1 dla najprostszego przypadku użycia jako pierwszego kamienia milowego.
W związku z presją regulacyjną CSRD (obowiązkową dla dużych unijnych przedsiębiorstw od 2025 r. rozszerzone na MŚP od 2026 r.) oraz rosnąca uwaga inwestorów na wskaźniki ESG, posiadanie infrastruktury technicznej do automatycznego rozliczania emisji dwutlenku węgla już nie istnieje przewaga konkurencyjna: jest to a konieczność operacyjna.