Climatiq API: Sera Gazı Emisyonu Hesaplamalarını Arka Uca Entegre Edin
2025'te karbon emisyonu izleme çözümleri pazarı büyüdü kadar 17,3 milyar dolargiderek sıkılaşan düzenlemeler nedeniyle Avrupa CSRD, ABD'deki SEC iklim açıklaması ve ISO 14064 gibi. Artık Excel sayfalarına ve kaba tahminlere güvenmenize gerek yok: bunlar faydalıdır otomatik hesaplamalar, doğrulanabilir ve işletim sistemlerine entegre edilebilir.
Teknik zorluk gerçektir: Sera gazı (Sera Gazı) emisyonlarının hesaplanması veritabanlarına erişim gerektirir emisyon faktörleri güncellendi, sera gazı protokolü tarafından doğrulanan metodolojiler ve dönüşümler Yüzlerce farklı ölçü birimi arasında. Tüm bunların şirket içinde inşa edilmesi ve sürdürülmesi, aylar süren özel çalışma. İşte burada devreye giriyor İklim.
Climatiq, üzerinden erişim sağlayan bir REST API'sidir. 190.000 emisyon faktörü 40'tan fazla doğrulanmış kaynaktan (EPA, DEFRA/BEIS, IEA, ecoinvent), 300'den fazla coğrafi bölgeve Kapsam 1, 2 ve 3 için Sera Gazı Protokolüne uygun hesaplamalar. Bu yazıda bir inşa edeceğiz FastAPI arka ucunu tamamlayın Climatiq'i entegre eden TypeScript istemcisi ve gerçek zamanlı hesap makinesiyle birlikte üretim emisyon hesaplamaları için SaaS uygulamaları için.
Ne Öğreneceksiniz
- Climatiq API mimarisi: uç noktalar, kimlik doğrulama, hız sınırları ve veri modelleri
- Emisyon Faktörü Veritabanı: doğru faktörlerin aranması, filtrelenmesi ve seçilmesi
- Faaliyet bazlı tahmin: somut faaliyetlere dayalı hesaplamalar (kWh, km, kg)
- Harcamaya dayalı tahmin: birincil veriler eksik olduğunda para birimi harcama hesaplamaları
- Sera Gazı Protokolü Kapsamı 1, 2, 3: kategori eşleme ve uyumlu hesaplamalar
- Yeniden deneme, Redis önbelleğe alma ve hata işleme özelliklerine sahip sağlam Python istemcisi
- Axios'lu TypeScript/Node.js istemcisi ve ön uç entegrasyonu için türler
- Ürünlerde karbon etiketleri bulunan SaaS için Gerçek Zamanlı Karbon Hesaplayıcı
- CI/CD ortamları için sahte API ile test etme
- Climatiq'e alternatifler ve işlevsellik karşılaştırması
Yeşil Yazılım Mühendisliği Serisi
Bu makale Yeşil Yazılım Mühendisliği hakkındaki tüm serinin bir parçasıdır. Her öğe dijital sürdürülebilirliğin belirli bir yönünü ele alıyor:
| # | Öğe | Ana Konu |
|---|---|---|
| 1 | Yeşil Yazılım Mühendisliği Prensipleri | GSF, SCI spesifikasyonu, 8 temel prensip |
| 2 | CodeCarbon: Kod Emisyonlarını Ölçme | Python kitaplığı, kontrol paneli, CI/CD entegrasyonu |
| 3 | Climatiq API: Arka Uçta Sera Gazı Hesaplamaları | REST API, Kapsam 1-3, FastAPI + TypeScript entegrasyonu |
| 4 | Karbon Bilinçli SDK | İş yükü değişimi, şebeke yoğunluğu, zaman değişimi |
| 5 | Kapsam 1, 2 ve 3: ÇSY Raporlaması için Veri Modelleme | Veri yapısı, hesaplamalar, toplama, raporlama |
| 6 | GreenOps: Karbon Bilinçli Altyapı | Kubernetes planlama, sorunlara göre ölçeklendirme |
| 7 | Emisyon Boru Hattı Kapsam 3 Değer Zinciri | Tedarikçi verilerinin toplanması, hesaplanması, denetim takibi |
| 8 | ESG Raporlama API'si: CSRD Entegrasyonu | CSRD iş akışı, rapor otomasyonu, uyumluluk |
| 9 | Sürdürülebilir Mimari Desenler | Depolama, akıllı önbelleğe alma, karbona duyarlı toplu işlem |
| 10 | Yapay Zeka ve Karbon: ML Eğitim Ayak İzi | LLM emisyonları, optimizasyon, Yeşil Yapay Zeka |
1. Sera Gazı Protokolü ve Otomatik Hesaplama İhtiyacı
Il Sera Gazı Protokolü Kurumsal Standardı dünyada en çok benimsenen çerçevedir şirket emisyonlarının muhasebeleştirilmesi için. Emisyonları üç alana sınıflandırır:
- Kapsam 1 (Doğrudan Emisyonlar): Şirket araçlarında yakıtların yakılması, üretim tesisleri, ısıtma. Şirketin doğrudan kontrolü altındadırlar.
- Kapsam 2 (Dolaylı Enerji): Satın alınan elektrik, buhar, ısı. Bunlar bölünmüştür konuma dayalı (yerel ağ karışımı) e pazar bazlı (enerji sertifikaları, PPA).
- Kapsam 3 (Değer Zinciri): Mal satın almayı içeren 15 kategori ve hizmetler, yukarı/aşağı nakliye, ürün kullanımı, kullanım ömrünün sonu, iş seyahati, çalışanların işe gidiş gelişleri ve çok daha fazlası. Tipik olarak şunları temsil ederler: %70-90 toplam emisyonların bir şirketin.
Her hesaplama için bir emisyon faktörü: dönüştüren bir katsayı bir faaliyet (litre dizel, kWh elektrik, euro satın alma) kg CO₂e cinsinden. Bu faktörler şunlara göre değişir:
- Referans yılı: elektrik şebekeleri her yıl karışım değiştiriyor
- Coğrafi bölge: İtalyan kWh'ı Alman kWh'ından farklıdır
- Veri kaynağı: ABD için EPA, İngiltere için DEFRA, İtalya için ISPRA
- Çevre: yukarı akıntı, aşağı akıntı, beşikten kapıya, beşikten mezara
Güncellenmiş bir emisyon faktörü veri tabanının sürdürülmesi özel bir ekip gerektirir. Climatiq bunu bizim için yapıyor ve daha da fazlasını topluyor 40 doğrulanmış kaynak ile sürekli güncellemeler.
2. Climatiq API'ye Genel Bakış
Mimari ve Ana Uç Noktalar
Climatiq API, URL tabanlı bir JSON REST API'sidir https://beta3.api.climatiq.io.
Kimlik doğrulama şu şekilde yapılır: Taşıyıcı Jetonu HTTP başlığında.
Mevcut planlar şunları içerir:
| Zemin | Arama/ay | İşlevsellik | Tipik kullanım |
|---|---|---|---|
| Toplum | 250 | Tüm uç noktalar | Prototip oluşturma, test etme |
| Başlangıç | 5.000 | + Özel faktörler | KOBİ'ler, MVP'ler |
| Büyüme | 50.000 | + SLA, destek | Büyüyen şirketler |
| Girişim | Gelenek | + Denetim takibi, SSO | Büyük organizasyonlar |
Ana uç noktalar şunlardır:
| Uç noktalar | Yöntem | Tanım |
|---|---|---|
/search |
ELDE ETMEK | Veritabanında emisyon faktörlerini arayın |
/estimate |
POSTALAMAK | Faaliyetlerden kaynaklanan tahmini tek emisyon |
/batch/estimate |
POSTALAMAK | Çoklu tahminler (istek başına en fazla 100) |
/travel/flights |
POSTALAMAK | Havayolu emisyonları (Kapsam 3.6) |
/freight |
POSTALAMAK | Çok modlu yük taşımacılığı emisyonları |
/procurement |
POSTALAMAK | Satın alma sorunları (Kapsam 3.1, harcamaya dayalı) |
/energy |
POSTALAMAK | Enerji tüketimi emisyonları (Kapsam 2) |
/compute |
POSTALAMAK | Bulut bilişim emisyonları |
Tahmin Talebinin Yapısı
# 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
}
}Veri Sürümü: Anlambilim ve En İyi Uygulamalar
Alan data_version Veritabanının hangi versiyonunun kullanılacağını kontrol eder.
Düzeltme işareti (^21) sürüm 21 veya üzeri uyumlu kullanın;
Emisyon faktörlerinde otomatik güncellemeler. Üretimde, blok
tam sürüm (ör. "21") tekrarlanabilirliği için
hesaplamalar ve denetim izleri. Kasıtlı olarak yeni yazılım sürümlerine yükseltme
tarihsel değerlerin açıkça yeniden hesaplanmasıyla.
3. Veri Modeli: Emisyon Faktörleri Veritabanı
Emisyon Faktörünün Yapısı
Climatiq'teki bir emisyon faktörü benzersiz bir şekilde tanımlanır:
activity_id, source, region, year
e lca_activity. Bu yapıyı anlamak kritik önem taşıyor
Doğru faktörü seçmek için.
# 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
}
}Ana Veri Kaynakları
| Kaynak | Ülke/Bölge | Güncelleme | Kapsanan Sektörler |
|---|---|---|---|
| DEFRA/BEIS | UK | Yıllık | Enerji, ulaşım, satın almalar, malzemeler |
| EPA | Amerika | Yıllık | Enerji, endüstriyel süreçler, tarım |
| IEA | Küresel | Yıllık | Ülkeye göre elektrik, birincil enerji |
| eko icat | Küresel | Altı ayda bir | Eksiksiz LCA, tedarik zinciri, malzemeler |
| ADEME | Fransa | Yıllık | Kömür bazlı, ulaşım, FR enerji |
| AEA | EU | Yıllık | Şebeke elektriği Avrupa, sektörel emisyonlar |
| ISPRA | İtalya | Yıllık | Ulusal sera gazı envanteri, BT elektrik karışımı |
4. Faaliyet Bazlı Tahmin: Somut Faaliyetlere İlişkin Hesaplamalar
Yaklaşım aktivite bazlı en doğru olanıdır: dayanmaktadır Tüketilen litre yakıt, kWh elektrik gibi birincil faaliyet verileri, kat edilen kilometre veya satın alınan tonlarca malzeme. Operasyonel veri toplama gerektirir ayrıntılıdır ancak bilimsel olarak sağlam sonuçlar üretir.
Örnek: Şirket Filosu Emisyonlarının Hesaplanması (Kapsam 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())Elektrik Emisyonlarının Hesaplanması (Kapsam 2)
Kapsam 2, ikili yöntem hesaplaması gerektirir: konuma dayalı (ağ karışımı) ve pazara dayalı (artık karışım faktörü veya PPA). Her iki değer de rapor edilmelidir. Sera Gazı Protokolünün ifşa edilmesi.
# 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. Harcamaya Dayalı Tahmin: Harcama Verilerinden Yapılan Hesaplamalar
Kesin aktivite verileriniz olmadığında yaklaşım harcamaya dayalı Faaliyetin vekili olarak döviz harcamalarını kullanır. Climatiq emisyon faktörlerini uygular verilere dayalı olarak satın alma kategorisine göre ekonomik (harcanan euro/dolar başına kg CO₂e) OECD veya EXIOBASE tablolarının girdi-çıktıları. Kapsam 3 için en yaygın yöntemdir. kategori 1 (Satın Alınan Mal ve Hizmetler).
Harcamaya Dayalı ve Etkinliğe Dayalı Doğruluk
Harcamaya dayalı yöntemler belirsizliğe sahip tahminler üretir %30-100, aktiviteye dayalı yöntemlerin %5-15'i ile karşılaştırıldığında. Bunları yalnızca veriniz olmadığında kullanın ön seçimler. Sera Gazı Protokolü bunları bir başlangıç noktası olarak kabul ediyor ancak ilerleme gerektiriyor Faaliyet verilerine yönelik iyileştirme.
# 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. Sera Gazı Protokolü Entegrasyonu: Haritalama Kapsamı 1, 2, 3
Eksiksiz bir karbon muhasebe sistemi oluşturmak, her Belirli uç noktalara yönelik Sera Gazı Protokolü kategorisi ve Climatiq Activity_id. Bu bölüm en alakalı kategoriler için eksiksiz haritalama sağlar.
| Kapsam Sera Gazı Protokolü | Kategori | Uç Nokta İklimi | Yöntem |
|---|---|---|---|
| Kapsam 1 | Sabit yanma (ısıtma) | /estimate |
Etkinlik (hacim) |
| Kapsam 1 | Mobil yanma (filo) | /batch/estimate |
Etkinlik (hacim) |
| Kapsam 1 | Kaçak emisyonlar (soğutucu akışkanlar) | /estimate |
Aktivite (ağırlık) |
| Kapsam 2 | Elektrik (konum bazlı) | /energy |
Faaliyet (kWh) |
| Kapsam 2 | Elektrik (piyasa bazlı) | /energy |
Faaliyet (kWh, kalan karışım) |
| Kapsam 3.1 | Satın alınan mal ve hizmetler | /procurement |
Harcamaya dayalı (EUR) |
| Kapsam 3.4 | Yukarı yönde taşıma | /freight |
Faaliyet (ton-km) |
| Kapsam 3.6 | İş seyahati (uçak) | /travel/flights |
Faaliyet (IATA kodu) |
| Kapsam 3.6 | İş seyahati (araba/tren) | /estimate |
Faaliyet (km) |
| Kapsam 3.7 | Çalışanların işe gidiş gelişleri | /batch/estimate |
Faaliyet (km, yarım) |
| Kapsam 3.11 | Satılan ürünlerin kullanımı | /estimate |
Etkinlik/harcama |
# 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. Sağlam Python İstemcisi: Yeniden Deneme, Önbellek ve Hata İşleme
Üretimde API çağrıları geçici hatalara karşı dayanıklı olmalıdır. etkili önbelleğe alma ile hız sınırlarını belirleyin ve aramaları en aza indirin. İşte üretime hazır bir istemci.
# 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. Axios'lu TypeScript/Node.js istemcisi
Per applicazioni Node.js, TypeScript o backend NestJS, ecco un client tipizzato che espone le stesse funzionalità del client Python con piena type safety.
// 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;
}
}
}Utilizzo del Client 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. Real-Time Carbon Calculator per SaaS
Un caso d'uso potente per Climatiq è aggiungere una carbon label ai prodotti di un e-commerce o SaaS: mostrare all'utente l'impatto carbonico stimato di un ordine o di un'azione prima di confermarla. Questo aumenta la trasparenza e supporta le scelte consapevoli dei consumatori.
Architettura: Carbon Label API per E-commerce
# 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 Frontend per Mostrare la Carbon Label
// 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. Testing e Mock API
In CI/CD, non si vogliono chiamate reali all'API Climatiq. Ecco come strutturare i test con mock robusti che simulano sia risposte di successo che errori.
# 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. Batch Estimation e Caching Strategy per Scala
Per applicazioni che elaborano migliaia di record (report mensili, analisi storiche), la combinazione di batch API calls e cache Redis diventa fondamentale per ridurre sia i tempi di elaborazione sia i costi 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. Alternative a Climatiq: Confronto
Climatiq non è l'unica opzione. Ecco un confronto delle principali alternative per aiutarti a scegliere la soluzione giusta:
| Soluzione | Forza Principale | Limite | Caso d'uso Ideale | Piano Free |
|---|---|---|---|---|
| Climatiq | 190K+ fattori, 40+ fonti, API robusta | 250 calls/mese nel piano community | Produzione enterprise, multi-scope | 250 calls/mese |
| Carbon Interface | API semplice, ottima developer experience | Meno fattori, focus USA | Startup, e-commerce, shipping | Sì (limitato) |
| Open Emission Factors | Gratuito, open source, mantenuto da Climatiq | Solo dataset, no API REST diretta | Research, accesso dati grezzo | Sì (dataset) |
| Watershed API | Audit-ready, enterprise features | Solo enterprise, costoso | Grandi aziende, CSRD reporting | No |
| EPA FLIGHT Database | Gratuito, governativo USA | Solo USA, non API REST | Reporting USA, ricerca | Sì (dataset) |
| Custom DB (DEFRA/IEA) | Pieno controllo, nessun costo API | Manutenzione interna costosa | Grandi organizzazioni con team dedicato | Sì (dati pubblici) |
Quando Usare un Database Personalizzato
In alcuni casi, costruire un database interno di fattori di emissione ha senso. Ecco uno schema PostgreSQL per gestirlo:
-- 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;Özel Veritabanı Doğru Seçim OLMADIĞINDA
Emisyon faktörlerine ilişkin bir veri tabanı oluşturmak ve sürdürmek aşağıdakileri gerektirir: 2-3 ay boyunca kendini adamış bir kişi ilk inşaat için, altı aylık zorunlu güncellemeler (faktörler her yıl değişir), denetim takibi için sürüm yönetimi ve sıklıkla coğrafi kapsam OECD dışı bölgeler için eksik. Çoğu şirket için, Climatiq üstün yatırım getirisi sunuyor API maliyeti göz önüne alındığında bile.
13. Örnek Olay İncelemesi: Carbon Label'lı e-ticaret platformu
EcoShop İtalya 50.000 ürünün yer aldığı bir e-ticaret platformudur ve ayda 200.000 sipariş. Gereksinim: Her ürüne bir karbon etiketi ekleyin ve dengeleme seçeneğiyle ödeme sırasında tahmini emisyonları gösterin.
Uygulanan Mimari
- Ürün Kataloğu Zenginleştirme: Önceden hesaplayan gecelik iş Toplu uç noktayı kullanan her SKU için üretim emisyonları. Sonuçlar kaydedildi 30 günlük TTL ile ürün veritabanında.
- Gerçek Zamanlı Ödeme: Ödeme sırasında emisyonların dinamik hesaplanması nakliye sepeti ağırlığına, varış bölgesine ve seçilen yönteme göre yapılır. Hedef gecikme süresi: 200 ms'nin altında (standart gönderiler için Redis önbelleğiyle).
- Karbon Kontrol Paneli Satıcısı: Satıcılar için aylık rapor Kapsam 3.4 (yukarı yöndeki taşıma) ve tahmini Kapsam 3.11 (satılan ürünlerin kullanımı).
# 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. Üretimde Dağıtım ve Güvenli Yapılandırma
# 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_dbClimatiq API Anahtarını asla açığa çıkarmayın
Climatiq API anahtarı hiçbir zaman kaynak kodunda veya günlüklerde bulunmamalıdır. yanıt API'leri veya istemci tarafı JavaScript değişkenleri. Her zaman şunu kullanın: ortam değişkenleri, gizli dizi yöneticisi (AWS Secrets Manager, HashiCorp Vault, Azure Key Vault) veya kullanımda olmayan şifrelemeyle Kubernetes Secrets. Yanlışlıkla ortaya çıkarsa anahtarı hemen çevirin.
15. En İyi Uygulamalar ve Anti-Kalıplar
En İyi Uygulamalar
-
Üretimde data_version'ı kilitle: Amerika
"21"yerine"^21"Hesaplamaların tekrarlanabilirliğini sağlamak. Açıkça yeniden hesaplamayla kasıtlı olarak güncelleyin. - Agresif önbelleğe alma: Emisyon faktörleri nadiren değişir. 24-168 saatlik önbellek, özellikle API çağrılarını önemli ölçüde azaltır tekrarlanan hesaplamalar için (aylık filo, ürün kataloğu).
- Her zaman denetim takibi: Her hesaplamayı aktivite_id ile kaydedin, kullanılan faktör, veri sürümü, zaman damgası ve kaynak. CSRD için gereklidir.
- Mümkün olduğunda toplu işlemi kullanın: Tek bir toplu çağrı 100 öğe, 100 ayrı aramadan çok daha verimlidir. Hız sınırlarına saygı gösterir ve toplam gecikmeyi azaltır.
- data_quality_flags'i doğrulayın: Climatiq şunu bildirir: faktörün veri kalitesi düşüktür. Bu vakaları alternatif faktörlerle ele alın veya rapordaki belirsizlik notları.
- Kapsam 2 için ikili raporlama: Sera Gazı Protokolü gerektirir hem lokasyon bazlı hem de pazar bazlı yöntemler. Her ikisini de hesaplayın ve kaydedin.
- TypeScript için bellek içi önbellek: TTL ile Harita Uygulama uygulamalarda gereksiz çağrıları önlemek için TypeScript istemcisinde Redis olmadan uzun ömürlü Node.js.
Kaçınılması Gereken Anti-Desenler
- Ön uçtaki API anahtarı: Climatiq anahtarını açıkta bırakmayın. İstemci tarafı JavaScript. Her zaman arka uçtan geçer.
- Kurucu gazları dikkate almayın: Doğru CSRD raporlaması için, Toplam CO₂e'ye ek olarak CO₂, CH₄ ve N₂O ayrıca rapor edilmelidir.
- Yanlış ölçü birimleri: Climatiq aşağıdakiler arasındaki dönüşümleri kabul eder: birimler, ancak uç nokta kWh beklediğinde litrelerin geçmesi sonuç üretir sayısal olarak makul ama bilimsel olarak yanlış. Her zaman geçerlidir.
- Toplu > 100 ürün: API 422 numaralı hatayı döndürüyor. Büyük veri kümeleri için her zaman parçalama mantığını uygulayın.
-
Bölge eşleşmesini yoksay: Elektrik için bir faktör
bölge belirtmeden genel varsayılanı kullanın. İtalya için her zaman kullanın
"IT". -
Senkronize çağrıları engelleme: Hiçbir zaman senkronize HTTP kitaplıklarını kullanmayın
eşzamansız uç noktalarda. Her zaman kullan
httpx.AsyncClientPython'da veyaaxiosileasync/awaitTypeScript'te.
Sonuçlar ve Sonraki Adımlar
Climatiq, otomatik karbon muhasebesinin en zor sorununu çözüyor: the emisyon faktörleri veritabanı. 190.000'den fazla doğrulanmış faktörle, 300 bölgeyi kapsaması ve sürekli güncellemeler, oluşturmanıza olanak tanır Aylar yerine günler içinde üretime hazır sera gazı hesaplama sistemleri.
Bu yazıda şunları oluşturduk:
- Un Sağlam Python istemcisi yeniden deneme, Redis önbelleği ve yazılan hata işleme ile
- Un TypeScript/Node.js istemcisi Ön uç entegrasyonları için Axios ve tam tip güvenlik ile
- Hesaplamalar Kapsam 1 (dizel/HVO filosu), Kapsam 2 (ikili yöntem) ve ana kategoriler Kapsam 3
- Un Carbon Label API'si gerçek zamanlı A-E etiketli ve ofsetli e-ticaret için
- İle test etme sahte API gerçek çağrıların olmadığı CI/CD ortamları için
- Stratejileri toplu işleme ve önbelleğe alma kurumsal ölçek için
Yeşil Yazılım Serisi devam ediyor
- Önceki makale: CodeCarbon - Kod emisyonlarını ölçün açık kaynak Python kütüphanesiyle çalışıyor.
- Sonraki makale: Carbon Aware SDK - İş yükleri nasıl kaydırılır Şebeke yoğunluğu tahminini kullanarak en temiz enerjiye sahip saatlerde.
- İlgili makale (MLOps Serisi): Optimize et Karbon emisyonlarını azaltmak için makine öğrenimi modellerini eğitmek.
- İlgili makale (Veri ve Yapay Zeka İş Serisi): Veri Yönetişimi Güvenilir Yapay Zeka için - Sürdürülebilirlik ölçümleri veri kataloğuna nasıl entegre edilir?
Bir sonraki pratik adım Climatiq'in Topluluk planına kaydolmaktır (250 ücretsiz arama/ay), keşfedin Veri Gezgini Sektörünüzle ilgili faktörleri bulmak ve uygulamak İlk kilometre taşı olarak en basit kullanım durumunuz için Kapsam 1'in hesaplanması.
CSRD'nin düzenleyici baskılarıyla (2025'ten itibaren büyük AB şirketleri için zorunlu, 2026'dan itibaren KOBİ'leri kapsayacak şekilde genişletilmesi) ve yatırımcıların ÇSY ölçümlerine artan ilgisi, Otomatik karbon muhasebesi için teknik altyapıya sahip olmak artık rekabet avantajı: bu bir operasyonel gereklilik.