Kapsam 1, 2 ve 3: ÇSY Raporlama Yazılımı için Veri Modelleme
Il Sera Gazı Protokolü (Sera Gazı Protokolü) fiili evrensel çerçeve haline geldi kurumsal sera gazı emisyonlarını ölçmek ve raporlamak. Dünya Kaynakları Enstitüsü (WRI) tarafından geliştirildi ve Dünya Sürdürülebilir Kalkınma İş Konseyi (WBCSD), emisyonların üç "kapsamını" tanımlamaktadır: bir kuruluşun değer zincirinin tamamını kapsarlar.
Yürürlüğe girmesiyle birlikte CSRD (Kurumsal Sürdürülebilirlik Raporlaması Direktifi) Birliğin Avrupa ve standart ESRS E1, şirketler dahil binlerce işletme yazılım ve teknoloji — artık kendi emisyon verilerini toplamaları, hesaplamaları ve yayınlamaları gerekiyor sıkı metodolojilere göre. 2024 yılı itibarıyla 500'den fazla çalışanı olan büyük şirketlerin 2025 yılında ilk raporlama zorunluluğuyla birlikte veri toplamaya başlanması; 250'den fazla çalışanı olan şirketler 2026'dan itibaren takip edecekler.
Ama neden ben geliştirici Kapsam 1, 2 ve 3'ü anlamaları gerekiyor mu? Neden bir sistem kurmalıyım? Güvenilir ÇSY yalnızca sürdürülebilirlik yöneticilerinin sorunu değildir: veri mühendisliği. Tutarlı bir veri modeli, doğru hesaplama hatları ve API'ler gerektirir. etkinlik koleksiyonu ve denetime hazır raporlar. Bu sistemleri kimin tasarlayıp uygulayacağını belirliyor Tüm ÇSY raporlamalarının kalitesi ve dolayısıyla güvenilirliği.
Bu yazıda bir inşa edeceğiz ÇSY raporlaması için veri modelleme sistemi, kapsam tanımlarından veritabanı tablolarına, Python hesaplamalarından FastAPI uç noktalarına kadar, 200 çalışanı olan bir yazılım KOBİ'si için tam bir örnek olay incelemesi.
Ne Öğreneceksiniz
- Sera Gazı Protokolünün yapısı: Kesin tanımlarla Kapsam 1, 2 ve 3
- Doğrudan, dolaylı ve değer zinciri emisyonları için varlık-ilişki modelleri
- Kapsam 2 için lokasyon bazlı ve pazar bazlı: farklılıklar ve veriler üzerindeki etkisi
- 15 Kapsam 3 kategorisi ve hangilerinin yazılım şirketleriyle ilgili olduğu
- Tam SQLAlchemy şeması: Organizasyon, Tesis, EmissionSource, EmissionFactor, ActivityData
- Etkinliğe dayalı tahmin ve harcamaya dayalı tahminle Python hesaplamaları
- Toplama, yıldan yıla karşılaştırma ve SBTi hedefleriyle uyum
- Görevleri göndermek, emisyonları hesaplamak ve raporlar oluşturmak için FastAPI içeren REST API
- Pytest ve emisyon faktörlerinin doğrulanması ile sera gazı hesaplamaları için birim testi
- Uçtan uca vaka çalışması: Yazılım KOBİ 200 çalışanı, Kapsam 1+2+3 tamamlandı
Yeşil Yazılım Serisi — 10 Makale
| # | Başlık | Durum |
|---|---|---|
| 1 | Yeşil Yazılım Mühendisliği ve SCI ilkeleri | Yayınlandı |
| 2 | CodeCarbon: Python Kod Emisyonlarını Ölçme | Yayınlandı |
| 3 | Carbon Aware SDK: İş Yükü Değiştirme ve Zaman Değiştirme | Yayınlandı |
| 4 | Climatiq API: Emisyon Faktörleri ve Karbon Hesaplaması | Yayınlandı |
| 5 | Kapsam 1, 2 ve 3: ÇSY Raporlaması için Veri Modelleme | Güncel makale |
| 6 | ESG ve CSRD: Teknoloji Şirketleri için Avrupa Yükümlülükleri | Yakında |
| 7 | Sürdürülebilir Yazılım Modelleri: Düşük Etkili Mimariler | Yakında |
| 8 | GreenOps: Sürdürülebilirlik için Bulut Optimizasyonu | Yakında |
| 9 | Kapsam 3 İşlem Hattı: Değer Zincirini Otomatikleştirin | Yakında |
| 10 | Yapay Zeka Karbon Ayak İzi: Yüksek Lisans, Eğitim ve Çıkarım | Yakında |
Sera Gazı Protokolü: Emisyonlar için Evrensel Çerçeve
2001 yılında yayınlanan ve 2004 yılında güncellenen Sera Gazı Protokolü Kurumsal Standardı, neredeyse tüm iklim raporlama çerçeveleri: CDP, TCFD, CSRD/ESRS E1, SBTi ve diğerleri doğrudan ona atıfta bulunurlar. Gücü kavramsal netlikte yatmaktadır: tüm emisyonlar Bir kuruluş birbirini dışlayan ancak genel olarak kapsamlı üç kategoriye ayrılır.
Çerçeve şu kavramı kullanır: organizasyonel sınırlar (organizasyon sınırı) Hesaplamaya hangi emisyonların dahil edileceğini belirleyin. İki yaklaşım vardır:özsermaye yaklaşımı (finansal katılım oranına göre) ve kontrol yaklaşımı (kontrol esasına göre operasyonel veya finansal). Çoğu şirket operasyonel kontrolü birincil kriter olarak benimser.
Üç Kapsam: Genel Bakış
| Süpürgeler | Tanım | Tipik örnekler | ESRS E1 zorunlu |
|---|---|---|---|
| Kapsam 1 | Kuruluşun sahip olduğu veya kontrol ettiği kaynaklardan kaynaklanan doğrudan emisyonlar | Doğalgaz kazanları, şirket filosu, dizel jeneratörler, endüstriyel prosesler | Zorunlu |
| Kapsam 2 | Enerji alımından kaynaklanan dolaylı emisyonlar (elektrik, buhar, ısı, soğuk) | Ofis elektriği, veri merkezi soğutması, bölgesel ısıtma | Zorunlu (her iki yöntem) |
| Kapsam 3 | Değer zincirindeki diğer tüm dolaylı emisyonlar (yukarı ve aşağı) | İş gezileri, işe gidip gelme, mal/hizmet satın alma, satılan ürünleri kullanma | Zorunlu (malzeme kategorileri) |
Kapsam 1: Doğrudan Emisyonlara İlişkin Veri Modeli
Kapsam 1 emisyonları doğrudan şirket tarafından kontrol edilen fiziksel kaynaklardan kaynaklanmaktadır. Onlar için yazılım şirketleri, bunlar genellikle küçüktür ancak ihmal edilemez: ısıtmayı içerirler ofislerin (doğal gaz), yedek jeneratörlerin ve şirket filosunun (şirket arabaları) veya donanım teslimat kamyonetleri).
Sera Gazı Protokolü Kapsam 1 emisyonlarının dört kategorisini tanımlar:
- Sabit yanma: Fosil yakıt yakan kazanlar, ısıtıcılar, jeneratörler
- Mobil yanma: Benzinli, dizel veya LPG'li şirket araçları
- Proses emisyonları: Kimyasal veya biyolojik reaksiyonlar (nadiren yazılımla ilgilidir)
- Kaçak emisyonlar: HVAC sistemlerinden soğutucu akışkan (R-410A, R-134a) sızıntıları
Kapsam 1 hesaplamasının temel formülü şöyledir:
Emisyonlar (tCO₂e) = Faaliyet verileri × Emisyon faktörü × GWP
Nerede:
- Verilen aktivite: tüketilen yakıt miktarı (litre, m³, kWh)
- Emisyon faktörü: birim yakıt başına kg CO₂ (kaynak: IPCC, DEFRA, IEA)
- GWP: Küresel Isınma Potansiyeli (CO₂=1, CH₄=28, N₂O=265, HFC değişiklik gösterir)
Kapsam 1'e ilişkin varlık-ilişki şeması aşağıdaki ana varlıkları içerir:
-- Entity-Relationship: Scope 1 Emissioni Dirette
ORGANIZATION (1) ---< FACILITY (>1)
|id, name, tax_id |id, org_id, name, type, country, area_m2
FACILITY (1) ---< EMISSION_SOURCE (>1)
|id, facility_id, source_type, description
| source_type: STATIONARY | MOBILE | FUGITIVE | PROCESS
EMISSION_SOURCE (1) ---< ACTIVITY_DATA (>1)
|id, source_id, period_start, period_end
|quantity, unit, data_quality, evidence_url
ACTIVITY_DATA (N) ---> EMISSION_FACTOR (1)
|id, gas, fuel_type, unit_from, unit_to
|factor_value, gwp_ar5, source, year, region
ACTIVITY_DATA (1) ---< GHG_CALCULATION (1)
|id, activity_id, scope, co2_kg, ch4_kg
|n2o_kg, hfc_kg, co2e_total_kg, method
|calculated_at, calculated_byKapsam 2: Lokasyon Tabanlı ve Pazar Tabanlı
Kapsam 2, teknoloji şirketleri için özellikle karmaşıktır çünkü tüketilen elektrik ofislerden ve veri merkezlerinden genellikle Dolaylı emisyonların ana kaynağı. Sera Gazı Protokolü Kapsam 2 Rehberi (2015), raporlama yükümlülüğünü getirmiştir. ikisi birden Yöntemler: konuma dayalı ve pazara dayalı.
Lokasyon Tabanlı ve Pazar Tabanlı: Temel Farklılıklar
| bekliyorum | Lokasyon Tabanlı | Piyasa Bazlı |
|---|---|---|
| Tanım | Tüketimin gerçekleştiği yerel elektrik şebekesinin ortalama yoğunluğu | Sözleşmeye bağlı olarak enerji satın alınan jeneratörlerden kaynaklanan emisyonlar |
| Emisyon faktörü | Ülkeye/bölgeye göre şebeke ortalama emisyon faktörü (gCO₂/kWh) | Tedarikçiye özgü faktör veya artık karışım faktörü |
| Sözleşmeye bağlı araçlar | Uygulanamaz | GO (Menşe Garantileri), REC (Yenilenebilir Enerji Sertifikaları), PPA |
| Örnek İtalya 2024 | ~310 gCO₂/kWh (AIB artık karışım verileri) | %100 doğrulanmış yenilenebilir MG’ler satın alırsanız 0 gCO₂/kWh |
| Sera Gazı Protokolü 2025 teklifi | Zaman hassasiyetine sahip yeni faktör hiyerarşisi | Yenilenebilir enerji kaynakları için saatlik eşleştirme talep edin |
| ESRS E1'de kullanın | Zorunlu açıklama (E1-6) | Zorunlu açıklama (E1-6) |
Kapsam 2'yi veritabanında doğru şekilde modellemek için enerji sertifikaları (GO/REC) "tüketilen" ayrı varlıklar olarak Şirketin elektrik tüketimine karşı piyasa bazlı faktörü sıfıra indirerek:
-- Modello Scope 2: Energia e Certificati
ELECTRICITY_CONSUMPTION
id UUID PRIMARY KEY
facility_id UUID REFERENCES FACILITY
period_start DATE
period_end DATE
kwh_consumed DECIMAL(15,3)
meter_id VARCHAR(50)
-- Location-based
grid_region VARCHAR(20) -- e.g. 'IT', 'DE-WEST', 'FR'
grid_ef_g_kwh DECIMAL(8,3) -- gCO2e/kWh from IEA/AIB
lb_co2e_kg DECIMAL(15,3) GENERATED ALWAYS AS
(kwh_consumed * grid_ef_g_kwh / 1000)
-- Market-based
supplier_ef_g_kwh DECIMAL(8,3) DEFAULT NULL
residual_mix_ef DECIMAL(8,3) DEFAULT NULL
mb_co2e_kg DECIMAL(15,3) -- calcolato dopo allocazione GO/REC
RENEWABLE_CERTIFICATE
id UUID PRIMARY KEY
cert_type VARCHAR(10) -- 'GO' | 'REC' | 'I-REC'
cert_id VARCHAR(100) UNIQUE
technology VARCHAR(30) -- 'WIND' | 'SOLAR' | 'HYDRO'
country VARCHAR(2)
production_date DATE
kwh_value DECIMAL(15,3)
issuer VARCHAR(100)
status VARCHAR(20) -- 'ACTIVE' | 'CANCELLED' | 'EXPIRED'
CERTIFICATE_ALLOCATION
id UUID PRIMARY KEY
cert_id UUID REFERENCES RENEWABLE_CERTIFICATE
consumption_id UUID REFERENCES ELECTRICITY_CONSUMPTION
kwh_allocated DECIMAL(15,3)
allocation_date DATE
-- Vincolo: kwh_allocated <= RENEWABLE_CERTIFICATE.kwh_valueKapsam 3: Değer Zincirinin 15 Kategorisi
Kapsam 3 tipik olarak ana kaynak yazılım şirketleri için emisyon oranları: toplamın %70-80'inden fazlasını temsil edebilir. Hesaplama için Sera Gazı Protokolü Teknik Kılavuzu Kapsam 3 Emisyonları, üretime yönelik (mal ve hizmet satın alımıyla bağlantılı) olarak bölünmüş 15 kategoriyi tanımlar. ve satış yönünde (satılan ürünlerin kullanımıyla bağlantılı).
Yazılım Şirketleri için 15 Kapsam 3 Kategorisi
| # | Kategori | Tip | Yazılım Uygunluğu | Hesaplama Yöntemi |
|---|---|---|---|---|
| 1 | Satın Alınan Mal ve Hizmetler | Yukarı akış | Yüksek (SaaS, bulut, donanım) | Harcamaya dayalı veya tedarikçiye özel |
| 2 | Sermaye Malları | Yukarı akış | Medya (dizüstü bilgisayarlar, sunucular, mobilyalar) | Harcamaya dayalı veya fiziksel birim |
| 3 | Yakıt ve Enerjiyle İlgili Faaliyetler | Yukarı akış | Ortalama (T&D kayıpları, yukarı akış yakıtı) | Etkinlik bazlı |
| 4 | Yukarı Akım Taşımacılığı ve Dağıtımı | Yukarı akış | Düşük (donanım teslimi) | Harcamaya dayalı veya mesafeye dayalı |
| 5 | Operasyonlarda Oluşan Atıklar | Yukarı akış | Düşük | Atık türüne özel |
| 6 | İş Seyahati | Yukarı akış | Yüksek (uçuşlar, oteller, trenler) | Mesafeye dayalı veya harcamaya dayalı |
| 7 | Çalışanların İşe Gidişi | Yukarı akış | Yüksek (işe gidip gelme, uzaktan çalışma) | Anket + mesafeye dayalı |
| 8 | Yukarı Yönde Kiralanan Varlıklar | Yukarı akış | Orta (kiralık ofisler) | Varlığa özgü |
| 9 | Mansap Taşımacılığı | Aşağı akış | Düşük | Mesafeye dayalı |
| 10 | Satılan Ürünlerin İşlenmesi | Aşağı akış | Uygulanamaz (yazılım) | Etkinlik bazlı |
| 11 | Satılan Ürünlerin Kullanımı | Aşağı akış | Yüksek (yazılımı çalıştırmak için enerji) | Ömür boyu enerji kullanımı |
| 12 | Yaşam Sonu Tedavisi | Aşağı akış | Düşük (kullanıcı cihazları) | Atık türüne özel |
| 13 | Alt Akış Kiralanan Varlıklar | Aşağı akış | Düşük | Varlığa özgü |
| 14 | Bayilikler | Aşağı akış | Uygulanamaz | Yok |
| 15 | Yatırımlar | Aşağı akış | Medya (VC/başlangıç portföyü) | Portföy bazlı |
Yazılım şirketleri için en alakalı kategoriler genellikle şunlardır: Kedi. 1 (satın alınan bulut bilişim), Kedi. 6 (iş seyahati), Kedi. 7 (çalışanların işe gidiş gelişleri) e Kedi. 11 (kullanıcılar tarafından tüketilen enerji Yazılımı çalıştırmak için). İkincisi genellikle hafife alınır ancak milyonlarca kurumsal SaaS için Kullanıcı sayısı çok büyük olabilir.
SQLAlchemy Şeması: Tam Model
Şimdi SQLAlchemy 2.0'da üç kapsamın tümünü destekleyen eksiksiz bir veritabanı şeması oluşturalım. Tasarım, her biri için ayrı tablolarla yüksek uyum ve düşük bağlantı ilkelerini takip etmektedir. Hakimiyet kavramı ve açık ilişkiler.
# models/base.py
from sqlalchemy import Column, String, DateTime, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import DeclarativeBase
import uuid
class Base(DeclarativeBase):
pass
class TimestampMixin:
"""Mixin per created_at e updated_at automatici."""
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())# models/organization.py
from sqlalchemy import Column, String, Integer, Boolean, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from .base import Base, TimestampMixin
class Organization(Base, TimestampMixin):
"""Azienda che produce il report ESG."""
__tablename__ = "organizations"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name = Column(String(200), nullable=False)
legal_name = Column(String(200))
tax_id = Column(String(50)) # Partita IVA o VAT number
country = Column(String(2), nullable=False) # ISO 3166-1 alpha-2
industry_code = Column(String(10)) # NACE Rev.2
employee_count = Column(Integer)
fiscal_year_end = Column(String(5), default="12-31") # MM-DD
reporting_currency = Column(String(3), default="EUR")
boundary_approach = Column(String(20), default="OPERATIONAL_CONTROL")
# OPERATIONAL_CONTROL | FINANCIAL_CONTROL | EQUITY_SHARE
active = Column(Boolean, default=True)
facilities = relationship("Facility", back_populates="organization")
reporting_periods = relationship("ReportingPeriod", back_populates="organization")
class Facility(Base, TimestampMixin):
"""Sede fisica (ufficio, data center, magazzino)."""
__tablename__ = "facilities"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
organization_id = Column(UUID(as_uuid=True), ForeignKey("organizations.id"), nullable=False)
name = Column(String(200), nullable=False)
facility_type = Column(String(30))
# OFFICE | DATA_CENTER | WAREHOUSE | LAB | REMOTE
address = Column(String(500))
city = Column(String(100))
country = Column(String(2), nullable=False)
grid_region = Column(String(30)) # Per fattore emissione location-based
area_sqm = Column(Integer) # Per calcoli per m2
employee_fte = Column(Integer) # Full-time equivalent
organization = relationship("Organization", back_populates="facilities")
emission_sources = relationship("EmissionSource", back_populates="facility")# models/emission_source.py
from sqlalchemy import Column, String, Enum, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import enum, uuid
from .base import Base, TimestampMixin
class SourceType(str, enum.Enum):
# Scope 1
STATIONARY_COMBUSTION = "STATIONARY_COMBUSTION"
MOBILE_COMBUSTION = "MOBILE_COMBUSTION"
FUGITIVE = "FUGITIVE"
PROCESS = "PROCESS"
# Scope 2
ELECTRICITY = "ELECTRICITY"
STEAM = "STEAM"
HEATING = "HEATING"
COOLING = "COOLING"
# Scope 3 (le 15 categorie)
S3_PURCHASED_GOODS = "S3_PURCHASED_GOODS" # Cat. 1
S3_CAPITAL_GOODS = "S3_CAPITAL_GOODS" # Cat. 2
S3_FUEL_ENERGY = "S3_FUEL_ENERGY" # Cat. 3
S3_UPSTREAM_TRANSPORT = "S3_UPSTREAM_TRANSPORT" # Cat. 4
S3_WASTE = "S3_WASTE" # Cat. 5
S3_BUSINESS_TRAVEL = "S3_BUSINESS_TRAVEL" # Cat. 6
S3_EMPLOYEE_COMMUTE = "S3_EMPLOYEE_COMMUTE" # Cat. 7
S3_UPSTREAM_LEASED = "S3_UPSTREAM_LEASED" # Cat. 8
S3_DOWNSTREAM_TRANSPORT = "S3_DOWNSTREAM_TRANSPORT" # Cat. 9
S3_USE_OF_PRODUCTS = "S3_USE_OF_PRODUCTS" # Cat. 11
S3_END_OF_LIFE = "S3_END_OF_LIFE" # Cat. 12
S3_INVESTMENTS = "S3_INVESTMENTS" # Cat. 15
class EmissionSource(Base, TimestampMixin):
"""Sorgente specifica di emissione (es. 'Caldaia edificio A')."""
__tablename__ = "emission_sources"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
facility_id = Column(UUID(as_uuid=True), ForeignKey("facilities.id"), nullable=True)
organization_id = Column(UUID(as_uuid=True), ForeignKey("organizations.id"), nullable=False)
source_type = Column(String(40), nullable=False) # SourceType enum value
scope = Column(Integer, nullable=False) # 1, 2, o 3
scope3_category = Column(Integer, nullable=True) # 1-15 se scope=3
description = Column(String(500))
unit_of_measure = Column(String(20)) # 'kWh', 'litri', 'km', 'EUR', 'kg'
active = Column(Boolean, default=True)
facility = relationship("Facility", back_populates="emission_sources")
activity_records = relationship("ActivityRecord", back_populates="source")# models/emission_factor.py
from sqlalchemy import Column, String, Numeric, Integer, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from .base import Base, TimestampMixin
class EmissionFactor(Base, TimestampMixin):
"""
Fattore di emissione da fonti ufficiali.
Converte un'attività (es. kWh) in kg CO2e.
"""
__tablename__ = "emission_factors"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
name = Column(String(200), nullable=False)
category = Column(String(50)) # 'electricity' | 'fuel' | 'transport' | ...
fuel_type = Column(String(50)) # 'natural_gas' | 'diesel' | 'petrol' | ...
technology = Column(String(50)) # 'gas_boiler' | 'bev' | ...
# Unità
activity_unit = Column(String(20), nullable=False) # 'kWh', 'litre', 'km', ...
emission_unit = Column(String(20), default="kg_co2e") # Output sempre in kg CO2e
# Valori per singolo gas (kg/unità)
co2_factor = Column(Numeric(18, 8))
ch4_factor = Column(Numeric(18, 8))
n2o_factor = Column(Numeric(18, 8))
hfc_factor = Column(Numeric(18, 8))
# GWP (AR5 100-year, IPCC 2014)
gwp_co2 = Column(Numeric(6, 2), default=1.0)
gwp_ch4 = Column(Numeric(6, 2), default=28.0)
gwp_n2o = Column(Numeric(6, 2), default=265.0)
# Metadati fonte
source = Column(String(100)) # 'DEFRA_2024' | 'IEA_2024' | 'IPCC_AR6'
year = Column(Integer)
region = Column(String(10)) # 'IT' | 'EU' | 'UK' | 'GLOBAL'
version = Column(String(20))
activity_records = relationship("ActivityRecord", back_populates="emission_factor")# models/activity_record.py
from sqlalchemy import Column, String, Numeric, Date, ForeignKey, Text
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
import uuid
from .base import Base, TimestampMixin
class ActivityRecord(Base, TimestampMixin):
"""
Dato di attività misurato o stimato per una fonte di emissione.
Es: 'Ufficio Milano ha consumato 15.432 kWh a novembre 2024'.
"""
__tablename__ = "activity_records"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
source_id = Column(UUID(as_uuid=True), ForeignKey("emission_sources.id"), nullable=False)
factor_id = Column(UUID(as_uuid=True), ForeignKey("emission_factors.id"), nullable=False)
reporting_period_id = Column(UUID(as_uuid=True), ForeignKey("reporting_periods.id"))
# Periodo
period_start = Column(Date, nullable=False)
period_end = Column(Date, nullable=False)
# Quantità
quantity = Column(Numeric(18, 4), nullable=False)
unit = Column(String(20), nullable=False)
# Qualità del dato
data_quality = Column(String(20), default="MEASURED")
# MEASURED | ESTIMATED | CALCULATED | MODELED
uncertainty_pct = Column(Numeric(5, 2)) # Es: 10.00 = +-10%
evidence_url = Column(Text) # Link a fattura, report, foglio elettrico
# Note
notes = Column(Text)
reviewer = Column(String(100))
source = relationship("EmissionSource", back_populates="activity_records")
emission_factor = relationship("EmissionFactor", back_populates="activity_records")
calculation = relationship("GHGCalculation", back_populates="activity_record", uselist=False)
class GHGCalculation(Base, TimestampMixin):
"""
Risultato del calcolo GHG per un ActivityRecord.
Separato per permettere ricalcoli senza perdere il dato sorgente.
"""
__tablename__ = "ghg_calculations"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
activity_record_id = Column(UUID(as_uuid=True),
ForeignKey("activity_records.id"), nullable=False, unique=True)
# Emissioni per gas (kg)
co2_kg = Column(Numeric(18, 4), default=0)
ch4_kg = Column(Numeric(18, 4), default=0)
n2o_kg = Column(Numeric(18, 4), default=0)
hfc_kg = Column(Numeric(18, 4), default=0)
# Totale in CO2e (kg e tonnellate)
co2e_kg = Column(Numeric(18, 4), nullable=False)
co2e_tonnes = Column(Numeric(18, 6), nullable=False)
# Metodo di calcolo
calculation_method = Column(String(50))
# ACTIVITY_BASED | SPEND_BASED | AVERAGE_DATA | HYBRID
# Scope 2 specifico
is_location_based = Column(Boolean)
is_market_based = Column(Boolean)
calculated_at = Column(DateTime(timezone=True), server_default=func.now())
calculator_version = Column(String(20), default="1.0.0")
activity_record = relationship("ActivityRecord", back_populates="calculation")Python'da sera gazı hesaplamaları: Faaliyet Tabanlı ve Harcama Tabanlı
Tanımlanan veri modeliyle hesaplamaları gerçekleştirebiliriz. Sera Gazı Protokolü şunları tanır: azalan doğruluk sırasına göre dört hesaplama yöntemi:
- Tedarikçiye özel yöntem: Doğrudan satıcıya ait veriler (en doğru)
- Hibrit yöntem: Birincil ve ikincil verilerin birleşimi
- Ortalama veri yöntemi: Fiziksel birim başına ortalama emisyon faktörleri
- Harcamaya dayalı yöntem: EUR/USD cinsinden harcama birimi başına faktörler (daha az doğru)
# calculators/ghg_calculator.py
from decimal import Decimal
from dataclasses import dataclass
from typing import Optional
from models import ActivityRecord, EmissionFactor, GHGCalculation
@dataclass
class CalculationResult:
co2_kg: Decimal
ch4_kg: Decimal
n2o_kg: Decimal
hfc_kg: Decimal
co2e_kg: Decimal
co2e_tonnes: Decimal
method: str
def activity_based_calculation(
quantity: Decimal,
factor: EmissionFactor
) -> CalculationResult:
"""
Calcolo activity-based: quantità * fattore emissione.
Formula GHG Protocol: E = AD * EF * GWP
"""
co2_kg = Decimal(str(quantity)) * Decimal(str(factor.co2_factor or 0))
ch4_kg = Decimal(str(quantity)) * Decimal(str(factor.ch4_factor or 0))
n2o_kg = Decimal(str(quantity)) * Decimal(str(factor.n2o_factor or 0))
hfc_kg = Decimal(str(quantity)) * Decimal(str(factor.hfc_factor or 0))
# Converti in CO2e usando GWP AR5
co2e_kg = (
co2_kg * Decimal(str(factor.gwp_co2))
+ ch4_kg * Decimal(str(factor.gwp_ch4))
+ n2o_kg * Decimal(str(factor.gwp_n2o))
+ hfc_kg * Decimal("1300") # GWP medio HFC-134a
)
return CalculationResult(
co2_kg=co2_kg,
ch4_kg=ch4_kg,
n2o_kg=n2o_kg,
hfc_kg=hfc_kg,
co2e_kg=co2e_kg,
co2e_tonnes=co2e_kg / Decimal("1000"),
method="ACTIVITY_BASED"
)
def spend_based_calculation(
spend_eur: Decimal,
spend_factor_kg_per_eur: Decimal
) -> CalculationResult:
"""
Calcolo spend-based per Cat. 1 Scope 3 quando non si hanno
dati fisici. Usa EXIOBASE o fattori DEFRA per settore NACE.
Meno accurato ma utile come stima iniziale.
"""
co2e_kg = spend_eur * spend_factor_kg_per_eur
return CalculationResult(
co2_kg=co2e_kg, # Semplificazione: tutto attribuito a CO2
ch4_kg=Decimal("0"),
n2o_kg=Decimal("0"),
hfc_kg=Decimal("0"),
co2e_kg=co2e_kg,
co2e_tonnes=co2e_kg / Decimal("1000"),
method="SPEND_BASED"
)
def business_travel_calculation(
distance_km: Decimal,
transport_mode: str,
cabin_class: Optional[str] = None
) -> CalculationResult:
"""
Calcolo Cat. 6 (Business Travel) con fattori DEFRA 2024.
Fattori in kgCO2e per km per passeggero.
"""
# Fattori DEFRA 2024 (kgCO2e/km/passeggero)
FACTORS = {
"SHORT_HAUL_ECONOMY": Decimal("0.15342"), # Volo <1000km economy
"SHORT_HAUL_BUSINESS": Decimal("0.23013"),
"LONG_HAUL_ECONOMY": Decimal("0.19085"), # Volo >1000km economy
"LONG_HAUL_BUSINESS": Decimal("0.57256"), # Business = 3x economy
"LONG_HAUL_FIRST": Decimal("0.76340"),
"RAIL_NATIONAL": Decimal("0.03549"), # Treno nazionale UK
"RAIL_EUROSTAR": Decimal("0.00416"), # Eurostar molto efficiente
"CAR_AVERAGE": Decimal("0.17100"), # Auto media
"CAR_ELECTRIC": Decimal("0.05280"),
"HOTEL_NIGHT": Decimal("20.600"), # Per notte, non per km
}
key = transport_mode
if cabin_class and f"{transport_mode}_{cabin_class}" in FACTORS:
key = f"{transport_mode}_{cabin_class}"
factor = FACTORS.get(key, Decimal("0"))
co2e_kg = distance_km * factor
return CalculationResult(
co2_kg=co2e_kg,
ch4_kg=Decimal("0"),
n2o_kg=Decimal("0"),
hfc_kg=Decimal("0"),
co2e_kg=co2e_kg,
co2e_tonnes=co2e_kg / Decimal("1000"),
method="ACTIVITY_BASED"
)# calculators/scope2_calculator.py
from decimal import Decimal
from dataclasses import dataclass
from typing import Optional
@dataclass
class Scope2Result:
location_based_co2e_kg: Decimal
market_based_co2e_kg: Decimal
renewable_kwh_matched: Decimal
grid_ef_g_kwh: Decimal
market_ef_g_kwh: Decimal
def scope2_dual_calculation(
kwh_consumed: Decimal,
grid_region: str,
go_certificates_kwh: Decimal = Decimal("0"),
supplier_ef_g_kwh: Optional[Decimal] = None
) -> Scope2Result:
"""
Calcolo Scope 2 con doppio metodo obbligatorio (GHG Protocol S2 Guidance).
location_based: usa grid average emission factor (IEA/AIB)
market_based: usa supplier factor o residual mix, dedotti GO/REC
"""
# Grid emission factors per regione (gCO2e/kWh) - IEA 2024
GRID_FACTORS = {
"IT": Decimal("310.0"), # Italia: mix gas + rinnovabili
"DE": Decimal("364.0"), # Germania: ancora carbone
"FR": Decimal("52.0"), # Francia: nucleare dominante
"ES": Decimal("198.0"),
"NL": Decimal("289.0"),
"EU_AVERAGE": Decimal("231.0"),
"UK": Decimal("209.0"),
"US_AVERAGE": Decimal("386.0"),
}
grid_ef = GRID_FACTORS.get(grid_region, GRID_FACTORS["EU_AVERAGE"])
# Location-based (sempre con grid average)
lb_co2e_kg = kwh_consumed * grid_ef / Decimal("1000")
# Market-based:
# 1. Kwh coperti da GO/REC = 0 emissioni
# 2. Kwh residui: usa supplier factor o residual mix
covered_by_go = min(go_certificates_kwh, kwh_consumed)
residual_kwh = kwh_consumed - covered_by_go
# Residual mix factor (tipicamente > grid average perché
# le energie rinnovabili sono "estratte" dal grid average)
# Fonte: AIB European Residual Mixes 2023
RESIDUAL_MIX = {
"IT": Decimal("414.0"), # Residual mix IT più alto del grid
"DE": Decimal("542.0"),
"FR": Decimal("73.0"),
"EU_AVERAGE": Decimal("395.0"),
}
market_ef = supplier_ef_g_kwh or RESIDUAL_MIX.get(grid_region, Decimal("400"))
mb_co2e_kg = residual_kwh * market_ef / Decimal("1000")
return Scope2Result(
location_based_co2e_kg=lb_co2e_kg,
market_based_co2e_kg=mb_co2e_kg,
renewable_kwh_matched=covered_by_go,
grid_ef_g_kwh=grid_ef,
market_ef_g_kwh=market_ef
)# calculators/scope3_commute.py
from decimal import Decimal
from dataclasses import dataclass
from typing import Dict
@dataclass
class CommuteData:
"""Dati raccolti tramite survey annuale dipendenti."""
employee_count: int
avg_working_days: int # Es: 220
remote_work_days_pct: Decimal # Es: 0.40 = 40% remote
transport_split: Dict[str, Decimal]
# {'car_solo': 0.45, 'car_pool': 0.10, 'public_transit': 0.30,
# 'bicycle': 0.10, 'walk': 0.05}
avg_commute_km_oneway: Decimal # km sola andata
def employee_commute_calculation(data: CommuteData) -> Decimal:
"""
Cat. 7 Scope 3: Pendolarismo dipendenti.
Formula: n_dipendenti * giorni_ufficio * km_totali * EF * split_modale
"""
# Fattori emissione (kgCO2e/km/passeggero) - DEFRA 2024
MODE_FACTORS = {
"car_solo": Decimal("0.171"), # Auto singola
"car_pool": Decimal("0.0855"), # Carpooling 2 persone
"public_transit": Decimal("0.089"), # Mix metropolitana/bus
"bicycle": Decimal("0.0"),
"walk": Decimal("0.0"),
"electric_car": Decimal("0.053"),
"motorcycle": Decimal("0.114"),
}
# Giorni effettivi in ufficio
office_days_pct = Decimal("1") - data.remote_work_days_pct
annual_office_days = Decimal(str(data.avg_working_days)) * office_days_pct
km_per_day = data.avg_commute_km_oneway * Decimal("2") # A/R
total_co2e_kg = Decimal("0")
for mode, share in data.transport_split.items():
ef = MODE_FACTORS.get(mode, Decimal("0.1"))
mode_co2e = (
Decimal(str(data.employee_count))
* annual_office_days
* km_per_day
* share
* ef
)
total_co2e_kg += mode_co2e
return total_co2e_kgYıllık Toplama ve Yıldan Yıla Karşılaştırma
Her Faaliyet Kaydı için emisyonlar hesaplandıktan sonra bunların dönemlere göre toplanması gerekir. raporlama yapın, bunları önceki yılla karşılaştırın ve SBTi hedefleriyle uyumu doğrulayın.
# services/reporting_service.py
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from decimal import Decimal
from typing import Dict
from models import GHGCalculation, ActivityRecord, EmissionSource, ReportingPeriod
@dataclass
class AnnualReport:
organization_id: str
year: int
scope1_co2e_tonnes: Decimal
scope2_lb_co2e_tonnes: Decimal # Location-based
scope2_mb_co2e_tonnes: Decimal # Market-based
scope3_co2e_tonnes: Decimal
scope3_by_category: Dict[int, Decimal]
total_co2e_tonnes: Decimal # S1 + S2(MB) + S3
intensity_per_employee: Decimal # tCO2e / FTE
intensity_per_revenue: Decimal # tCO2e / M EUR
def generate_annual_report(
session: Session,
organization_id: str,
year: int,
employee_fte: int,
revenue_million_eur: Decimal
) -> AnnualReport:
"""Aggrega tutti i calcoli GHG per anno fiscale."""
# Query aggregata per scope
stmt = (
select(
EmissionSource.scope,
EmissionSource.scope3_category,
GHGCalculation.is_location_based,
func.sum(GHGCalculation.co2e_tonnes).label("total_tonnes")
)
.join(ActivityRecord, GHGCalculation.activity_record_id == ActivityRecord.id)
.join(EmissionSource, ActivityRecord.source_id == EmissionSource.id)
.where(
EmissionSource.organization_id == organization_id,
func.extract("year", ActivityRecord.period_start) == year
)
.group_by(
EmissionSource.scope,
EmissionSource.scope3_category,
GHGCalculation.is_location_based
)
)
rows = session.execute(stmt).all()
scope1 = Decimal("0")
scope2_lb = Decimal("0")
scope2_mb = Decimal("0")
scope3_by_cat: Dict[int, Decimal] = {}
for row in rows:
tonnes = Decimal(str(row.total_tonnes or 0))
if row.scope == 1:
scope1 += tonnes
elif row.scope == 2:
if row.is_location_based:
scope2_lb += tonnes
else:
scope2_mb += tonnes
elif row.scope == 3:
cat = row.scope3_category or 0
scope3_by_cat[cat] = scope3_by_cat.get(cat, Decimal("0")) + tonnes
scope3_total = sum(scope3_by_cat.values())
total = scope1 + scope2_mb + scope3_total
return AnnualReport(
organization_id=organization_id,
year=year,
scope1_co2e_tonnes=scope1,
scope2_lb_co2e_tonnes=scope2_lb,
scope2_mb_co2e_tonnes=scope2_mb,
scope3_co2e_tonnes=scope3_total,
scope3_by_category=scope3_by_cat,
total_co2e_tonnes=total,
intensity_per_employee=total / Decimal(str(employee_fte)) if employee_fte else Decimal("0"),
intensity_per_revenue=total / revenue_million_eur if revenue_million_eur else Decimal("0")
)
def check_sbti_alignment(
report: AnnualReport,
baseline_year: int,
baseline_emissions: Decimal,
target_year: int = 2030,
reduction_target_pct: Decimal = Decimal("0.50") # SBTi: -50% by 2030
) -> Dict:
"""
Verifica allineamento con target SBTi.
Near-term: -50% entro 2030 (1.5°C pathway).
Net-zero: -90-95% entro 2050.
"""
current_year_progress = report.year - baseline_year
total_years = target_year - baseline_year
linear_target = baseline_emissions * (
1 - reduction_target_pct * Decimal(str(current_year_progress)) / Decimal(str(total_years))
)
actual_reduction_pct = (baseline_emissions - report.total_co2e_tonnes) / baseline_emissions * 100
return {
"baseline_emissions_tonnes": float(baseline_emissions),
"current_emissions_tonnes": float(report.total_co2e_tonnes),
"linear_target_tonnes": float(linear_target),
"actual_reduction_pct": float(actual_reduction_pct),
"on_track": report.total_co2e_tonnes <= linear_target,
"gap_tonnes": float(report.total_co2e_tonnes - linear_target),
"required_annual_reduction_pct": float(
reduction_target_pct / Decimal(str(total_years)) * 100
)
}ESG için REST API: FastAPI Uç Noktaları
Kurumsal bir ESG sistemi, çeşitli ekiplerin gönderim yapmasına olanak sağlamak için güçlü API'lere ihtiyaç duyar. aktivite verileri, tetikleme hesaplamaları ve erişim raporları. FastAPI seçimdir Tip güvenliği, otomatik dokümantasyon ve performansı açısından idealdir.
# api/routers/activities.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from pydantic import BaseModel, validator
from typing import Optional
from datetime import date
from decimal import Decimal
import uuid
router = APIRouter(prefix="/api/v1", tags=["activities"])
class ActivitySubmitRequest(BaseModel):
source_id: uuid.UUID
period_start: date
period_end: date
quantity: Decimal
unit: str
data_quality: str = "MEASURED"
evidence_url: Optional[str] = None
notes: Optional[str] = None
@validator("data_quality")
def validate_quality(cls, v: str) -> str:
allowed = ["MEASURED", "ESTIMATED", "CALCULATED", "MODELED"]
if v not in allowed:
raise ValueError(f"data_quality deve essere uno di: {allowed}")
return v
@validator("period_end")
def validate_dates(cls, v: date, values: dict) -> date:
if "period_start" in values and v < values["period_start"]:
raise ValueError("period_end deve essere >= period_start")
return v
class ActivitySubmitResponse(BaseModel):
activity_id: uuid.UUID
co2e_tonnes: Decimal
calculation_method: str
message: str
@router.post("/activities", response_model=ActivitySubmitResponse, status_code=201)
async def submit_activity(
req: ActivitySubmitRequest,
session: Session = Depends(get_db),
current_org: Organization = Depends(get_current_org)
) -> ActivitySubmitResponse:
"""
Invia un dato di attività e calcola le emissioni GHG.
Validazione automatica: unità coerente con la sorgente.
"""
source = session.get(EmissionSource, req.source_id)
if not source or source.organization_id != current_org.id:
raise HTTPException(status_code=404, detail="Sorgente emissione non trovata")
# Trova il fattore di emissione più recente per questa sorgente
factor = get_best_emission_factor(session, source, req.unit, req.period_start.year)
if not factor:
raise HTTPException(
status_code=422,
detail=f"Nessun fattore emissione trovato per {source.source_type} in {req.unit}"
)
# Calcola emissioni
result = activity_based_calculation(req.quantity, factor)
# Salva ActivityRecord e GHGCalculation
record = ActivityRecord(
source_id=req.source_id,
factor_id=factor.id,
period_start=req.period_start,
period_end=req.period_end,
quantity=req.quantity,
unit=req.unit,
data_quality=req.data_quality,
evidence_url=req.evidence_url,
notes=req.notes
)
session.add(record)
session.flush() # Get the ID
calc = GHGCalculation(
activity_record_id=record.id,
co2_kg=result.co2_kg,
ch4_kg=result.ch4_kg,
n2o_kg=result.n2o_kg,
co2e_kg=result.co2e_kg,
co2e_tonnes=result.co2e_tonnes,
calculation_method=result.method
)
session.add(calc)
session.commit()
return ActivitySubmitResponse(
activity_id=record.id,
co2e_tonnes=result.co2e_tonnes,
calculation_method=result.method,
message=f"Emissioni calcolate: {result.co2e_tonnes:.4f} tCO2e"
)# api/routers/reports.py
from fastapi import APIRouter, Query
from pydantic import BaseModel
from typing import Dict
from decimal import Decimal
router = APIRouter(prefix="/api/v1", tags=["reports"])
class ESGReportResponse(BaseModel):
organization_id: str
year: int
scope1_co2e_tonnes: float
scope2_location_based_tonnes: float
scope2_market_based_tonnes: float
scope3_co2e_tonnes: float
scope3_breakdown: Dict[int, float]
total_co2e_tonnes: float
intensity_per_employee: float
intensity_per_revenue_m_eur: float
sbti_alignment: Dict
class YoYComparisonResponse(BaseModel):
year_current: int
year_previous: int
total_change_pct: float
scope1_change_pct: float
scope2_change_pct: float
scope3_change_pct: float
top_increasing_categories: list
top_decreasing_categories: list
@router.get("/reports/annual/{year}", response_model=ESGReportResponse)
async def get_annual_report(
year: int,
session: Session = Depends(get_db),
current_org: Organization = Depends(get_current_org)
) -> ESGReportResponse:
"""
Genera report ESG annuale con tutte le metriche GHG Protocol.
Include verifica allineamento SBTi.
"""
employee_fte = get_employee_fte(session, current_org.id, year)
revenue = get_annual_revenue(session, current_org.id, year)
report = generate_annual_report(
session, str(current_org.id), year, employee_fte, revenue
)
baseline = get_baseline_emissions(session, str(current_org.id))
sbti = check_sbti_alignment(report, baseline["year"], baseline["tonnes"])
return ESGReportResponse(
organization_id=str(current_org.id),
year=year,
scope1_co2e_tonnes=float(report.scope1_co2e_tonnes),
scope2_location_based_tonnes=float(report.scope2_lb_co2e_tonnes),
scope2_market_based_tonnes=float(report.scope2_mb_co2e_tonnes),
scope3_co2e_tonnes=float(report.scope3_co2e_tonnes),
scope3_breakdown={k: float(v) for k, v in report.scope3_by_category.items()},
total_co2e_tonnes=float(report.total_co2e_tonnes),
intensity_per_employee=float(report.intensity_per_employee),
intensity_per_revenue_m_eur=float(report.intensity_per_revenue),
sbti_alignment=sbti
)
@router.get("/reports/yoy/{year}", response_model=YoYComparisonResponse)
async def year_over_year(year: int, ...) -> YoYComparisonResponse:
"""Confronto Year-over-Year con identificazione trend per categoria."""
curr = generate_annual_report(session, org_id, year, fte, rev)
prev = generate_annual_report(session, org_id, year - 1, fte_prev, rev_prev)
def pct_change(curr_val: Decimal, prev_val: Decimal) -> float:
if prev_val == 0:
return 0.0
return float((curr_val - prev_val) / prev_val * 100)
return YoYComparisonResponse(
year_current=year,
year_previous=year - 1,
total_change_pct=pct_change(curr.total_co2e_tonnes, prev.total_co2e_tonnes),
scope1_change_pct=pct_change(curr.scope1_co2e_tonnes, prev.scope1_co2e_tonnes),
scope2_change_pct=pct_change(curr.scope2_mb_co2e_tonnes, prev.scope2_mb_co2e_tonnes),
scope3_change_pct=pct_change(curr.scope3_co2e_tonnes, prev.scope3_co2e_tonnes),
top_increasing_categories=get_top_changes(curr, prev, ascending=False),
top_decreasing_categories=get_top_changes(curr, prev, ascending=True)
)Test: Sera Gazı Hesaplamaları için Birim Testi
Sera gazı hesaplamaları titizlikle test edilmelidir: emisyon faktörlerindeki hatalar veya formüller, potansiyel yasal sonuçları olan (yeşil yıkama) hatalı ÇSY raporlamasına yol açar ve CSRD kapsamındaki yaptırımlar. Test isteğe bağlı değildir.
# tests/test_ghg_calculator.py
import pytest
from decimal import Decimal
from calculators.ghg_calculator import (
activity_based_calculation,
spend_based_calculation,
business_travel_calculation
)
from calculators.scope2_calculator import scope2_dual_calculation
from calculators.scope3_commute import employee_commute_calculation, CommuteData
from models import EmissionFactor
@pytest.fixture
def natural_gas_factor() -> EmissionFactor:
"""Fattore gas naturale DEFRA 2024: 2.04264 kgCO2e/m3."""
factor = EmissionFactor()
factor.co2_factor = Decimal("1.88900") # kgCO2/m3
factor.ch4_factor = Decimal("0.00011") # kgCH4/m3
factor.n2o_factor = Decimal("0.00003") # kgN2O/m3
factor.hfc_factor = Decimal("0")
factor.gwp_co2 = Decimal("1.0")
factor.gwp_ch4 = Decimal("28.0")
factor.gwp_n2o = Decimal("265.0")
return factor
class TestActivityBasedCalculation:
def test_natural_gas_calculation(self, natural_gas_factor):
"""Test: 1000 m3 gas naturale = ~2.04 tCO2e."""
result = activity_based_calculation(Decimal("1000"), natural_gas_factor)
assert result.method == "ACTIVITY_BASED"
# CO2: 1000 * 1.889 = 1889 kg
assert abs(result.co2_kg - Decimal("1889.00")) < Decimal("0.01")
# CH4: 1000 * 0.00011 * 28 = 3.08 kgCO2e
expected_co2e = (
Decimal("1889.00") * Decimal("1.0") # CO2
+ Decimal("0.11") * Decimal("28.0") # CH4
+ Decimal("0.03") * Decimal("265.0") # N2O
)
assert abs(result.co2e_kg - expected_co2e) < Decimal("1.0")
assert result.co2e_tonnes == result.co2e_kg / Decimal("1000")
def test_zero_consumption(self, natural_gas_factor):
"""Zero consumo = zero emissioni."""
result = activity_based_calculation(Decimal("0"), natural_gas_factor)
assert result.co2e_kg == Decimal("0")
def test_precision_decimal(self, natural_gas_factor):
"""Verifica che i calcoli usino Decimal per evitare floating point errors."""
result = activity_based_calculation(Decimal("333.333"), natural_gas_factor)
# Non deve avere floating point drift
assert isinstance(result.co2e_kg, Decimal)
class TestScope2DualMethod:
def test_italy_location_based(self):
"""Test location-based Italia: 10.000 kWh * 310 gCO2e/kWh = 3100 kg."""
result = scope2_dual_calculation(
kwh_consumed=Decimal("10000"),
grid_region="IT"
)
assert abs(result.location_based_co2e_kg - Decimal("3100.0")) < Decimal("1.0")
def test_full_go_coverage_zeroes_market_based(self):
"""Con GO = consumo totale, market-based deve essere 0."""
result = scope2_dual_calculation(
kwh_consumed=Decimal("10000"),
grid_region="IT",
go_certificates_kwh=Decimal("10000")
)
assert result.market_based_co2e_kg == Decimal("0")
assert result.renewable_kwh_matched == Decimal("10000")
def test_partial_go_coverage(self):
"""Con GO parziali, market-based usa residual mix sul rimanente."""
result = scope2_dual_calculation(
kwh_consumed=Decimal("10000"),
grid_region="IT",
go_certificates_kwh=Decimal("6000")
)
# 4000 kWh residui * 414 gCO2e/kWh (residual mix IT) / 1000
expected = Decimal("4000") * Decimal("414") / Decimal("1000")
assert abs(result.market_based_co2e_kg - expected) < Decimal("0.01")
class TestEmployeeCommute:
def test_standard_commute(self):
"""Test pendolarismo con mix modale realistico."""
data = CommuteData(
employee_count=200,
avg_working_days=220,
remote_work_days_pct=Decimal("0.40"),
transport_split={
"car_solo": Decimal("0.45"),
"public_transit": Decimal("0.40"),
"bicycle": Decimal("0.10"),
"walk": Decimal("0.05")
},
avg_commute_km_oneway=Decimal("15")
)
result = employee_commute_calculation(data)
assert result > Decimal("0")
# Ordine di grandezza: ~50-150 tCO2e per 200 dipendenti
assert Decimal("40000") < result < Decimal("200000") # in kg
def test_full_remote_work_zero(self):
"""100% remote work = 0 emissioni pendolarismo."""
data = CommuteData(
employee_count=200,
avg_working_days=220,
remote_work_days_pct=Decimal("1.0"),
transport_split={"car_solo": Decimal("1.0")},
avg_commute_km_oneway=Decimal("20")
)
result = employee_commute_calculation(data)
assert result == Decimal("0")Dikkat: Sorun ve Sürüm Oluşturma Faktörleri
Emisyon faktörleri her yıl güncellenmektedir (DEFRA, IEA, EPA). Bir ESG sistemi doğru olmalıdır:
- Veritabanındaki faktörleri açık yıl ve kaynakla birlikte sürümlendirin
- Raporlama yılı faktörünü kullanın (en güncel olanı değil)
- Faktörler düzeltilirse geçmişin yeniden hesaplamalarına izin ver
- Denetim izini sürdürün: hesaplama sırasında hangi faktör etkindi
- Yerinde bir sera gazı hesaplamasını asla güncellemeyin: her zaman yeni bir sürüm oluşturun
Örnek Olay: 200 Çalışanlı KOBİ Yazılımı
Somut bir uçtan uca örnek görelim. TechFlow S.r.l. bir KOBİ'dir 200 çalışanlı İtalyan yazılımı, Milano'daki ana ofis (kiralık), ikincil ofis Roma'da kendi veri merkezi yok (tümü AWS eu-west-1 ve Azure westeurope'da).
Şirket Profili TechFlow S.r.l.
| Parametre | Değer |
|---|---|
| Çalışanlar | 200 FTE (%70 Milano, %20 Roma, %10 tam uzaktan) |
| Satış | 15 milyon avro |
| Ofisler | 2 adet kiralık lokasyon (toplam 2.400 m²) |
| BT altyapısı | %100 bulut (AWS + Azure), sıfır fiziksel sunucu |
| Filo | 3 dizel şirket arabası, 1 dizel kamyonet |
| İş seyahati | ~120 uçuş/yıl, 40 gece otel, 200 tren güzergahı |
# Esempio calcolo completo TechFlow S.r.l. - Anno 2024
from decimal import Decimal
from calculators.ghg_calculator import activity_based_calculation, business_travel_calculation
from calculators.scope2_calculator import scope2_dual_calculation
from calculators.scope3_commute import employee_commute_calculation, CommuteData
# ---- SCOPE 1 ----
# Gas naturale: riscaldamento uffici (caldaie condominiali, incluse nel canone)
# TechFlow ha controllo operativo: include le caldaie
scope1_gas_heating = activity_based_calculation(
Decimal("12500"), # m3 gas naturale annui (entrambe le sedi)
NATURAL_GAS_FACTOR_IT_2024
)
# Risultato atteso: ~25.5 tCO2e
# Flotta aziendale: 4 veicoli diesel
scope1_fleet_km = activity_based_calculation(
Decimal("85000"), # km totali flotta annui
DIESEL_MOBILE_FACTOR_2024 # 0.16844 kgCO2e/km
)
# Risultato atteso: ~14.3 tCO2e
# Refrigeranti HVAC: perdita stimata 2% del carico R-410A
scope1_refrigerants = activity_based_calculation(
Decimal("1.5"), # kg R-410A persi
R410A_FACTOR # GWP=2088 kgCO2e/kg
)
# Risultato atteso: ~3.1 tCO2e
scope1_total = (
scope1_gas_heating.co2e_tonnes
+ scope1_fleet_km.co2e_tonnes
+ scope1_refrigerants.co2e_tonnes
)
print(f"Scope 1 totale: {scope1_total:.2f} tCO2e")
# Output: Scope 1 totale: 42.90 tCO2e
# ---- SCOPE 2 ----
# Elettricità Milano: 320.000 kWh, GO acquistate: 200.000 kWh
scope2_milano = scope2_dual_calculation(
kwh_consumed=Decimal("320000"),
grid_region="IT",
go_certificates_kwh=Decimal("200000")
)
# Elettricità Roma: 95.000 kWh, nessuna GO
scope2_roma = scope2_dual_calculation(
kwh_consumed=Decimal("95000"),
grid_region="IT",
go_certificates_kwh=Decimal("0")
)
scope2_lb_total = scope2_milano.location_based_co2e_kg + scope2_roma.location_based_co2e_kg
scope2_mb_total = scope2_milano.market_based_co2e_kg + scope2_roma.market_based_co2e_kg
print(f"Scope 2 Location-Based: {scope2_lb_total/1000:.2f} tCO2e")
# Output: Scope 2 Location-Based: 128.45 tCO2e
print(f"Scope 2 Market-Based: {scope2_mb_total/1000:.2f} tCO2e")
# Output: Scope 2 Market-Based: 55.12 tCO2e (ridotto grazie alle GO)
# ---- SCOPE 3 ----
# Cat. 1: Cloud computing (AWS+Azure) - spend-based
# Spesa totale cloud: EUR 480.000/anno
# Fattore EXIOBASE settore "Computer Services" IT: ~0.062 kgCO2e/EUR
scope3_cat1_cloud = spend_based_calculation(
spend_eur=Decimal("480000"),
spend_factor_kg_per_eur=Decimal("0.062")
)
# Risultato atteso: ~29.8 tCO2e
# NOTA: Provider-specific è più accurato se disponibile
# AWS Carbon Footprint: ~18 tCO2e (più basso per energia rinnovabile AWS)
# Cat. 2: Capital goods - laptop/monitor (life cycle amortized)
# 40 laptop nuovi a EUR 1.200 cad, ammortizzati su 4 anni
scope3_cat2_laptops = spend_based_calculation(
spend_eur=Decimal("40") * Decimal("1200") / Decimal("4"),
spend_factor_kg_per_eur=Decimal("0.35") # Electronic equipment
)
# Risultato atteso: ~4.2 tCO2e
# Cat. 6: Business travel
scope3_cat6_flights = business_travel_calculation(
distance_km=Decimal("95000"), # Totale km voli
transport_mode="SHORT_HAUL_ECONOMY" # Mix short/medium haul
)
scope3_cat6_rail = business_travel_calculation(
distance_km=Decimal("18000"), # Totale km treno
transport_mode="RAIL_NATIONAL"
)
scope3_cat6_hotel = business_travel_calculation(
distance_km=Decimal("40"), # 40 notti hotel
transport_mode="HOTEL_NIGHT"
)
scope3_cat6_total = (
scope3_cat6_flights.co2e_tonnes
+ scope3_cat6_rail.co2e_tonnes
+ scope3_cat6_hotel.co2e_tonnes
)
# Risultato atteso: ~16.3 tCO2e
# Cat. 7: Employee commuting
commute_data = CommuteData(
employee_count=180, # Esclude 20 full-remote
avg_working_days=220,
remote_work_days_pct=Decimal("0.35"), # Media 35% smart working
transport_split={
"car_solo": Decimal("0.38"),
"public_transit": Decimal("0.45"),
"bicycle": Decimal("0.08"),
"walk": Decimal("0.06"),
"car_pool": Decimal("0.03")
},
avg_commute_km_oneway=Decimal("12")
)
scope3_cat7_kg = employee_commute_calculation(commute_data)
# Risultato atteso: ~68.400 kg = 68.4 tCO2e
# ---- RIEPILOGO FINALE ----
scope3_total = (
scope3_cat1_cloud.co2e_tonnes
+ scope3_cat2_laptops.co2e_tonnes
+ scope3_cat6_total
+ scope3_cat7_kg / Decimal("1000")
# + Cat. 11 Use of sold products (SaaS): stimato ~15 tCO2e
# + Cat. 3 T&D losses: ~8.2 tCO2e
)
grand_total = scope1_total + scope2_mb_total / Decimal("1000") + scope3_total
print("\n=== TECHFLOW S.R.L. - GHG INVENTORY 2024 ===")
print(f"Scope 1: {float(scope1_total):8.1f} tCO2e ( {float(scope1_total/grand_total)*100:4.1f}%)")
print(f"Scope 2: {float(scope2_mb_total/1000):8.1f} tCO2e ( {float(scope2_mb_total/1000/grand_total)*100:4.1f}%)")
print(f"Scope 3: {float(scope3_total):8.1f} tCO2e ( {float(scope3_total/grand_total)*100:4.1f}%)")
print(f"TOTALE: {float(grand_total):8.1f} tCO2e")
print(f"Intensità: {float(grand_total/200):5.2f} tCO2e/dipendente")
# === TECHFLOW S.R.L. - GHG INVENTORY 2024 ===
# Scope 1: 42.9 tCO2e ( 18.2%)
# Scope 2: 55.1 tCO2e ( 23.4%)
# Scope 3: 137.8 tCO2e ( 58.4%)
# TOTALE: 235.8 tCO2e
# Intensità: 1.18 tCO2e/dipendenteKarşılaştırma Ölçütü: Yazılım Şirketleri için Karbon Yoğunluğu (2024)
| Boyut | tCO₂e/bağımlı | % Kapsam 3 | Ana giriş S3 |
|---|---|---|---|
| Startup'lar (<50 çalışan) | 0,8 - 1,2 | %65-75 | İş seyahati |
| Yazılım KOBİ'leri (50-500) | 1,0 - 1,8 | %55-70 | İşe gidip gelme + bulut |
| Kurumsal teknoloji (500+) | 1,5 - 3,0 | %70-85 | Bulut + ürünlerin kullanımı |
| Büyük Teknoloji (aşırı ölçekleyiciler) | 3,0 - 8,0 | %75-90 | Donanım tedarik zinciri + veri merkezleri |
| TechFlow (vaka çalışması) | 1.18 | %58 | işe gidip gelme |
TechFlow için Azaltma Fırsatları
Sera Gazı Envanteri 2024'ün analizinden en etkili azaltma araçları ortaya çıkıyor:
- Akıllı çalışma %50'ye çıkarıldı: Kedi. 7 ~%35 azalma (-24 tCO2e). İşe gidip gelme en büyük tek öğedir.
- Her iki konum için de %100 GO: Kapsam 2 pazar bazında ~0 tCO2e'ye düşüyor (-55 tCO2e). GO'nun tahmini maliyeti: 3.000-5.000 Avro/yıl.
- Seyahat karbon bütçesi politikası: Kısa uçuşların (<3 saat) trenle değiştirilmesi. Kedi. 6 ~%40 azalma (-6,5 tCO2e).
- AWS Karbon Ayak İzi Aracı: Harcamaya dayalı tahminden belirli sağlayıcıya geçiş yapın Cat'i azaltır. 1 bulutu ~%40 oranında (-12 tCO2e) artırır ve veri kalitesini artırır.
TechFlow Karbonsuzlaştırma Yol Haritası — Hedef SBTi 1,5°C
| Yıl | Hedef (tCO₂e) | Azaltma ve 2024 | Ana eylemler |
|---|---|---|---|
| 2024 (temel) | 235.8 | - | İlk CSRD raporlaması |
| 2025 | 211.0 | -%10,5 | GO %100, akıllı çalışma %50 |
| 2027 | 165.0 | -30% | Elektrikli filo, seyahat politikası |
| 2030 | 118.0 | -%50 (SBTi yakın vadede) | Yeşil, uzaktan öncelikli bulut sağlayıcıları |
| 2050 | <24 | -90% (SBTi net-sıfır) | Nötrleştirilmiş kalıntı, tazminat yok |
Sonuçlar ve Sonraki Adımlar
ÇSY raporlamasına yönelik veri modelleme basit bir uyumluluk çalışması değildir: güvenilir ve doğrulanabilir bir karbondan arındırma stratejisinin oluşturulmasının temelini oluşturur. Bir şirketin değer zincirinin tamamını kapsayan bir sistemin nasıl yapılandırılacağını gördük Kalorifer kazanından (Kapsam 1) ticari uçak yolculuğuna (Kapsam 3 Kat. 6) kadar yazılım Kullanıcıların ürünü çalıştırmak için tükettiği enerjiye kadar (Kapsam 3 Kat. 11).
Yanınıza almanız gereken önemli noktalar:
- Kapsam 3 hakimdir: Yazılım şirketleri için emisyonların %55-75'i Kapsam 3'tür. Doğru şekilde modellememek, karbon ayak izinizin çarpık bir görüntüsüne sahip olmak anlamına gelir.
- İkili raporlama Kapsamı 2: ESRS E1 her iki yöntemi de gerektirir (konum bazlı ve pazara dayalı). Piyasa temelli, sertifikalı yenilenebilir enerjinin satın alınmasını teşvik eder.
- Veri kalitesi her şeydir: Etkinliğe dayalı > harcamaya dayalı hiyerarşi veri toplama seçimlerine rehberlik etmelidir. Harcamaya dayalı tahminlerle başlayın ve bunları her geçen yıl geliştirin.
- Hesaplamaların değişmezliği: Hiçbir zaman bir sera gazı hesaplamasını mevcut durumda güncellemeyin. Sürüm faktörleri ve ESG denetçi incelemesi için eksiksiz denetim takibini sürdürün.
- Kuzey yıldızı olarak SBTi: Bilime Dayalı hedef (2030'a kadar -%50, 2050'ye kadar -%90) minimum CSRD uyumluluğuna değil, azaltma önceliklerine yön vermelidir.
Serideki Sonraki Yazılar
Serinin bir sonraki makalesinde, "ESG ve CSRD: Teknoloji Şirketleri için Avrupa Yükümlülükleri", düzenleyici çerçevenin tamamını inceleyelim: CSRD'ye kimler tabidir, son tarihler KOBİ'ler için çift önemlilik değerlendirmesi kavramı ve veri modelleme sistemi gibi Bu makalede oluşturulan açıklamalar, Kurumdan talep edilen ESRS E1 açıklamalarına çevrilmektedir.
Ayrıca ilgili makalelere bakın:
- MLOps Serisi: Karbon izlemeyi makine öğrenimi ardışık düzenlerine nasıl entegre edebilirim? (MLflow modeli kaydındaki sürdürülebilirlik ölçümleri)
- Yapay Zeka Mühendisliği Serisi: Üretimdeki RAG ve LLM sistemlerinin karbon ayak izi
- Madde 9 (bu seri): Kapsam 3 İşlem Hattı — Otomatikleştirin tedarikçiler için API bağlayıcıları ile değer zincirinden veri toplama