Facility Emissions Tutorial
Learn how to track emissions from facility operations, including electricity, heating, cooling, and on-site fuel combustion.Estimated time: 30 minutesWhat you’ll learn:
- Register facilities
- Upload utility invoices (electricity, gas, water)
- Track on-site fuel combustion
- Handle multi-facility splits
- Calculate Scope 1 & 2 emissions
Prerequisites
Before starting, ensure you have:- Dcycle API credentials (get them here)
- Basic knowledge of Python or JavaScript
- Utility bills and facility data
Using the Dcycle App?You can also manage facilities through our web interface:
- Create facilities (ES) - Register and manage facilities
- Archive facilities (ES) - Archive inactive facilities
- Connect Datadis (ES) - Automate electricity data in Spain
- Split invoices (ES) - Divide consumption across facilities
- Upload electricity bills (ES)
- Upload combustion invoices (ES)
- Upload water bills (ES)
Understanding Facility Emissions
Facility emissions typically include Scope 1 (on-site combustion) and Scope 2 (purchased electricity).Emission Scopes
Copy
Scope 1: Direct Emissions
├─ Natural gas heating
├─ Diesel generators
├─ LPG for forklifts
└─ Refrigerant leaks
Scope 2: Indirect Energy
├─ Grid electricity
├─ District heating
└─ District cooling
Data Flow
1
Register Facilities
Create facility records with location and type
2
Upload Utility Bills
Track electricity, gas, water consumption
3
Calculate Emissions
Dcycle applies location-based emission factors
4
Generate Reports
View emissions by facility, energy type, and period
Step 1: Register Facilities
Create facilities to organize your energy consumption:Copy
import requests
import os
headers = {
"Authorization": f"Bearer {os.getenv('DCYCLE_API_KEY')}",
"x-organization-id": os.getenv("DCYCLE_ORG_ID"),
"x-user-id": os.getenv("DCYCLE_USER_ID")
}
# Create a facility
facility_data = {
"name": "Madrid Headquarters",
"address": "Calle de Alcalá, 123",
"postal_code": "28009",
"city": "Madrid",
"country": "ES",
"facility_type": "office", # office, warehouse, factory, store, etc.
"surface_m2": 2500, # Optional
"employees": 120 # Optional
}
facility = requests.post(
"https://api.dcycle.io/api/v1/facilities",
headers=headers,
json=facility_data
).json()
facility_id = facility['id']
print(f"✅ Facility registered: {facility_id}")
print(f" Name: {facility['name']}")
print(f" Location: {facility['city']}, {facility['country']}")
List All Facilities
Copy
facilities = requests.get(
"https://api.dcycle.io/api/v1/facilities",
headers=headers,
params={"page": 1, "size": 50}
).json()
print("📋 Your Facilities:")
for f in facilities['items']:
print(f" - {f['name']} ({f['city']}, {f['country']})")
print(f" Type: {f['facility_type']}, Surface: {f.get('surface_m2', 'N/A')} m²")
Step 2: Upload Electricity Invoices
Track electricity consumption (Scope 2):Single Invoice
Copy
# Get unit IDs first
units = requests.get(
"https://api.dcycle.io/api/v1/units",
headers=headers,
params={"page": 1, "size": 100}
).json()
kwh_unit = next(u for u in units['items'] if u['abbreviation'] == 'kWh')
# Create electricity invoice
invoice_data = {
"facility_id": facility_id,
"invoice_type": "electricity", # electricity, natural_gas, water, etc.
"consumption": 15000, # kWh
"unit_id": kwh_unit['id'],
"start_date": "2024-01-01",
"end_date": "2024-01-31",
"invoice_number": "ELEC-2024-001",
"supplier": "Iberdrola",
"cost": 2450.00, # Optional
"currency": "EUR" # Optional
}
invoice = requests.post(
"https://api.dcycle.io/api/v1/invoices",
headers=headers,
json=invoice_data
).json()
print(f"✅ Invoice created")
print(f" Consumption: {invoice['consumption']} kWh")
print(f" CO2e (Scope 2): {invoice['co2e']:.2f} kg")
print(f" Emission factor: {invoice['emission_factor_used']}")
Location-Based vs Market-Based
Dcycle uses location-based emission factors by default (grid average for the country/region). For market-based (supplier-specific factors like renewable energy contracts), use custom emission factors:Copy
# If you have a renewable energy PPA
from guides.advanced.custom_emission_factors import create_ppa_factor
# Create custom factor for your PPA
ppa_factor_id = create_ppa_factor(
name="Wind Farm PPA 2024",
renewable_percentage=100.0,
co2e_per_kwh=0.0 # Zero emissions
)
# Use in invoice
invoice_data = {
"facility_id": facility_id,
"invoice_type": "electricity",
"consumption": 15000,
"unit_id": kwh_unit['id'],
"custom_emission_factor_id": ppa_factor_id, # Use PPA factor
"start_date": "2024-01-01",
"end_date": "2024-01-31"
}
Step 3: Upload Natural Gas Invoices
Track on-site combustion (Scope 1):Copy
# Get natural gas unit (kWh, m³, or therms)
m3_unit = next(u for u in units['items'] if u['abbreviation'] == 'm³')
# Create natural gas invoice
gas_invoice = {
"facility_id": facility_id,
"invoice_type": "natural_gas",
"consumption": 2500, # m³
"unit_id": m3_unit['id'],
"start_date": "2024-01-01",
"end_date": "2024-01-31",
"invoice_number": "GAS-2024-001",
"supplier": "Naturgy",
"cost": 1800.00,
"currency": "EUR"
}
gas = requests.post(
"https://api.dcycle.io/api/v1/invoices",
headers=headers,
json=gas_invoice
).json()
print(f"✅ Gas invoice created")
print(f" Consumption: {gas['consumption']} m³")
print(f" CO2e (Scope 1): {gas['co2e']:.2f} kg")
Supported Fuel Types
Get available facility fuel types:Copy
facility_fuels = requests.get(
"https://api.dcycle.io/api/v1/facility_fuels",
headers=headers
).json()
print("Available facility fuels:")
for fuel in facility_fuels['items']:
print(f" - {fuel['name']} ({fuel['category']})")
print(f" Units: {', '.join(fuel['compatible_units'])}")
electricity- Grid electricity (Scope 2)natural_gas- Natural gas heating (Scope 1)diesel- Diesel generators (Scope 1)lpg- Liquefied petroleum gas (Scope 1)district_heating- District heat (Scope 2)district_cooling- District cooling (Scope 2)
Step 4: Bulk Upload Invoices
For multiple facilities or monthly batches:CSV Format
Copy
facility_id,invoice_type,consumption,unit_id,start_date,end_date,invoice_number,supplier,cost,currency
facility-uuid-1,electricity,15000,kwh-uuid,2024-01-01,2024-01-31,ELEC-001,Iberdrola,2450.00,EUR
facility-uuid-1,natural_gas,2500,m3-uuid,2024-01-01,2024-01-31,GAS-001,Naturgy,1800.00,EUR
facility-uuid-2,electricity,8500,kwh-uuid,2024-01-01,2024-01-31,ELEC-002,Endesa,1420.00,EUR
facility-uuid-2,diesel,500,liter-uuid,2024-01-01,2024-01-31,DIE-001,Repsol,750.00,EUR
Upload Process
Copy
# Get presigned upload URL
upload_response = requests.post(
"https://api.dcycle.io/api/v1/invoices/bulk/csv",
headers=headers
).json()
# Upload CSV
with open('january_2024_invoices.csv', 'rb') as f:
requests.put(upload_response['upload_url'], data=f)
print("✅ CSV uploaded, processing...")
# Poll for completion
import time
while True:
status = requests.get(
upload_response['status_url'],
headers=headers
).json()
if status['status'] == 'completed':
print(f"✅ Processed {status['records_processed']} invoices")
print(f" Total CO2e: {status['total_co2e']:.2f} kg")
break
elif status['status'] == 'failed':
print(f"❌ Error: {status['error']}")
break
time.sleep(5)
Step 5: Handle Multi-Facility Scenarios
Scenario 1: Split Single Invoice Across Facilities
When one invoice covers multiple facilities:Copy
# Upload main invoice
main_invoice = {
"facility_id": facility_id, # Primary facility
"invoice_type": "electricity",
"consumption": 50000, # Total kWh
"unit_id": kwh_unit['id'],
"start_date": "2024-01-01",
"end_date": "2024-01-31",
"invoice_number": "ELEC-MULTI-001"
}
invoice = requests.post(url, headers=headers, json=main_invoice).json()
# Split consumption to other facilities
split_1 = {
"source_invoice_id": invoice['id'],
"target_facility_id": "facility-2-uuid",
"consumption": 15000, # Allocate 15,000 kWh to facility 2
"allocation_method": "manual" # or "surface_area", "employees"
}
requests.post(
"https://api.dcycle.io/api/v1/invoices/split",
headers=headers,
json=split_1
)
split_2 = {
"source_invoice_id": invoice['id'],
"target_facility_id": "facility-3-uuid",
"consumption": 10000, # Allocate 10,000 kWh to facility 3
"allocation_method": "manual"
}
requests.post(
"https://api.dcycle.io/api/v1/invoices/split",
headers=headers,
json=split_2
)
# Remaining 25,000 kWh stays with primary facility
Scenario 2: Proportional Allocation
Automatically split by surface area or employees:Copy
# Split by surface area
split_by_area = {
"source_invoice_id": invoice['id'],
"allocation_method": "surface_area",
"target_facilities": [
{"facility_id": "facility-1-uuid", "surface_m2": 1000},
{"facility_id": "facility-2-uuid", "surface_m2": 1500},
{"facility_id": "facility-3-uuid", "surface_m2": 500}
]
}
# Dcycle automatically calculates proportional consumption
# Facility 1: 33.3% (1000/3000)
# Facility 2: 50.0% (1500/3000)
# Facility 3: 16.7% (500/3000)
requests.post(
"https://api.dcycle.io/api/v1/invoices/split",
headers=headers,
json=split_by_area
)
Step 6: Query and Analyze
Get Facility Emissions Report
Copy
facility_id = "facility-uuid"
report = requests.get(
f"https://api.dcycle.io/api/v1/facilities/{facility_id}/emissions",
headers=headers,
params={
"start_date": "2024-01-01",
"end_date": "2024-12-31"
}
).json()
print(f"📊 Facility Emissions Report: {report['facility_name']}")
print(f" Period: {report['start_date']} to {report['end_date']}")
print(f"\n Scope 1 (Direct): {report['scope_1_co2e']:.2f} kg CO2e")
print(f" Scope 2 (Electricity): {report['scope_2_co2e']:.2f} kg CO2e")
print(f" Total: {report['total_co2e']:.2f} kg CO2e")
print(f"\n Breakdown by energy type:")
for energy in report['by_energy_type']:
print(f" - {energy['type']}: {energy['co2e']:.2f} kg CO2e")
Compare Facilities
Copy
# Get all facilities
facilities = requests.get(
"https://api.dcycle.io/api/v1/facilities",
headers=headers
).json()
comparison = []
for facility in facilities['items']:
report = requests.get(
f"https://api.dcycle.io/api/v1/facilities/{facility['id']}/emissions",
headers=headers,
params={"start_date": "2024-01-01", "end_date": "2024-12-31"}
).json()
comparison.append({
'name': facility['name'],
'total_co2e': report['total_co2e'],
'surface_m2': facility.get('surface_m2', 0),
'intensity': report['total_co2e'] / facility.get('surface_m2', 1) if facility.get('surface_m2') else 0
})
# Sort by total emissions
comparison.sort(key=lambda x: x['total_co2e'], reverse=True)
print("\n📊 Facility Comparison (2024)")
print(f"{'Facility':<30} {'Total CO2e':>15} {'Surface (m²)':>15} {'Intensity':>20}")
print("-" * 85)
for f in comparison:
print(f"{f['name']:<30} {f['total_co2e']:>15,.1f} {f['surface_m2']:>15,.0f} {f['intensity']:>20,.2f} kg/m²")
Real-World Example: Facility Management System
Complete workflow for managing facility emissions:Copy
import requests
import os
from datetime import date, datetime
from calendar import monthrange
class FacilityManager:
def __init__(self):
self.headers = {
"Authorization": f"Bearer {os.getenv('DCYCLE_API_KEY')}",
"x-organization-id": os.getenv("DCYCLE_ORG_ID"),
"x-user-id": os.getenv("DCYCLE_USER_ID")
}
self.base_url = "https://api.dcycle.io"
def create_facility(self, **kwargs):
"""Create new facility"""
response = requests.post(
f"{self.base_url}/api/v1/facilities",
headers=self.headers,
json=kwargs
)
return response.json()
def get_units(self):
"""Get available units and cache them"""
if not hasattr(self, '_units'):
response = requests.get(
f"{self.base_url}/api/v1/units",
headers=self.headers,
params={"page": 1, "size": 100}
)
self._units = {u['abbreviation']: u for u in response.json()['items']}
return self._units
def upload_invoice(self, **kwargs):
"""Upload single utility invoice"""
response = requests.post(
f"{self.base_url}/api/v1/invoices",
headers=self.headers,
json=kwargs
)
return response.json()
def bulk_upload_invoices(self, csv_file_path):
"""Bulk upload invoices from CSV"""
# Get presigned URL
upload_response = requests.post(
f"{self.base_url}/api/v1/invoices/bulk/csv",
headers=self.headers
).json()
# Upload
with open(csv_file_path, 'rb') as f:
requests.put(upload_response['upload_url'], data=f)
return upload_response['status_url']
def get_facility_report(self, facility_id, start_date, end_date):
"""Get emissions report for facility"""
response = requests.get(
f"{self.base_url}/api/v1/facilities/{facility_id}/emissions",
headers=self.headers,
params={
"start_date": start_date.isoformat(),
"end_date": end_date.isoformat()
}
)
return response.json()
def get_all_facilities(self):
"""Get all facilities"""
response = requests.get(
f"{self.base_url}/api/v1/facilities",
headers=self.headers,
params={"page": 1, "size": 100}
)
return response.json()['items']
def monthly_electricity_upload(self, facilities_data):
"""Upload monthly electricity for multiple facilities"""
units = self.get_units()
kwh_unit_id = units['kWh']['id']
results = []
for facility_data in facilities_data:
invoice = self.upload_invoice(
facility_id=facility_data['facility_id'],
invoice_type='electricity',
consumption=facility_data['kwh'],
unit_id=kwh_unit_id,
start_date=facility_data['start_date'],
end_date=facility_data['end_date'],
invoice_number=facility_data['invoice_number'],
supplier=facility_data['supplier'],
cost=facility_data.get('cost'),
currency=facility_data.get('currency', 'EUR')
)
results.append({
'facility_id': facility_data['facility_id'],
'consumption': invoice['consumption'],
'co2e': invoice['co2e']
})
return results
# Usage Example
manager = FacilityManager()
# 1. Create facilities
print("Creating facilities...")
hq = manager.create_facility(
name="Madrid HQ",
city="Madrid",
country="ES",
facility_type="office",
surface_m2=2500
)
warehouse = manager.create_facility(
name="Barcelona Warehouse",
city="Barcelona",
country="ES",
facility_type="warehouse",
surface_m2=5000
)
print(f"✅ Created HQ: {hq['id']}")
print(f"✅ Created Warehouse: {warehouse['id']}")
# 2. Upload monthly invoices
print("\nUploading January invoices...")
january_invoices = [
{
'facility_id': hq['id'],
'kwh': 15000,
'start_date': '2024-01-01',
'end_date': '2024-01-31',
'invoice_number': 'HQ-ELEC-JAN',
'supplier': 'Iberdrola',
'cost': 2450.00
},
{
'facility_id': warehouse['id'],
'kwh': 28000,
'start_date': '2024-01-01',
'end_date': '2024-01-31',
'invoice_number': 'WH-ELEC-JAN',
'supplier': 'Endesa',
'cost': 4200.00
}
]
results = manager.monthly_electricity_upload(january_invoices)
for result in results:
print(f"✅ Facility {result['facility_id'][:8]}...")
print(f" Consumption: {result['consumption']:,} kWh")
print(f" CO2e: {result['co2e']:.2f} kg")
# 3. Generate quarterly report
print("\nGenerating Q1 2024 report...")
facilities = manager.get_all_facilities()
for facility in facilities:
report = manager.get_facility_report(
facility['id'],
date(2024, 1, 1),
date(2024, 3, 31)
)
print(f"\n{facility['name']}:")
print(f" Scope 1: {report['scope_1_co2e']:,.2f} kg CO2e")
print(f" Scope 2: {report['scope_2_co2e']:,.2f} kg CO2e")
print(f" Total: {report['total_co2e']:,.2f} kg CO2e")
if facility.get('surface_m2'):
intensity = report['total_co2e'] / facility['surface_m2']
print(f" Intensity: {intensity:.2f} kg CO2e/m²")
Best Practices
1. Regular Data Collection
Set up monthly invoice upload routine:Copy
from datetime import date
from calendar import monthrange
def monthly_invoice_upload(year, month):
"""Run on 5th of each month for previous month"""
last_day = monthrange(year, month)[1]
start_date = date(year, month, 1)
end_date = date(year, month, last_day)
# Collect invoices for all facilities
# Upload to Dcycle
manager.bulk_upload_invoices(f"invoices_{year}_{month:02d}.csv")
2. Data Validation
Validate before upload:Copy
def validate_invoice_data(invoice):
"""Validate invoice data"""
# Check required fields
assert invoice.get('facility_id'), "facility_id required"
assert invoice.get('consumption', 0) > 0, "consumption must be positive"
assert invoice.get('invoice_type'), "invoice_type required"
# Sanity checks
if invoice['invoice_type'] == 'electricity':
if invoice['consumption'] > 100000: # kWh
print("⚠️ Very high electricity consumption")
# Date validation
start = date.fromisoformat(invoice['start_date'])
end = date.fromisoformat(invoice['end_date'])
assert end >= start, "end_date must be >= start_date"
# Period check (typically monthly)
days = (end - start).days
if days > 35:
print(f"⚠️ Long billing period: {days} days")
return True
3. Track Energy Intensity
Monitor performance over time:Copy
def calculate_intensity_metrics(facility_id, year):
"""Calculate energy intensity metrics"""
report = manager.get_facility_report(
facility_id,
date(year, 1, 1),
date(year, 12, 31)
)
facility = requests.get(
f"{manager.base_url}/api/v1/facilities/{facility_id}",
headers=manager.headers
).json()
metrics = {
'co2e_per_m2': report['total_co2e'] / facility['surface_m2'],
'co2e_per_employee': report['total_co2e'] / facility.get('employees', 1),
'kwh_per_m2': report['total_electricity_kwh'] / facility['surface_m2']
}
return metrics
4. Automate Data Collection
Integrate with utility providers’ APIs (where available):Copy
# Example: Datadis integration (Spain)
def sync_datadis_data(facility_id, cups_code, year, month):
"""
Sync electricity data from Datadis (Spain)
CUPS: Universal Supply Point Code
"""
# Connect to Datadis API
datadis_data = fetch_datadis_consumption(cups_code, year, month)
# Upload to Dcycle
invoice = manager.upload_invoice(
facility_id=facility_id,
invoice_type='electricity',
consumption=datadis_data['total_kwh'],
unit_id=kwh_unit_id,
start_date=datadis_data['start_date'],
end_date=datadis_data['end_date'],
invoice_number=f"DATADIS-{year}-{month:02d}",
supplier=datadis_data['supplier']
)
return invoice
Troubleshooting
Issue: Wrong Emission Factor Applied
Copy
# Check which factor was used
invoice = requests.get(
f"https://api.dcycle.io/api/v1/invoices/{invoice_id}",
headers=headers
).json()
print(f"Emission factor used: {invoice['emission_factor_used']}")
print(f"Country: {invoice['facility']['country']}")
print(f"Year: {invoice['year']}")
# Factors are location-based by default
# For custom factors (e.g., renewable energy), use custom_emission_factor_id
Issue: Invoice Split Not Adding Up
Copy
# Check split allocations
splits = requests.get(
f"https://api.dcycle.io/api/v1/invoices/{invoice_id}/splits",
headers=headers
).json()
total_original = invoice['consumption']
total_allocated = sum(s['consumption'] for s in splits['items'])
remaining = total_original - total_allocated
print(f"Original consumption: {total_original}")
print(f"Allocated: {total_allocated}")
print(f"Remaining in source facility: {remaining}")
if remaining < 0:
print("❌ Error: Over-allocated!")
Issue: Missing Unit ID
Copy
# List all available units
units = requests.get(
"https://api.dcycle.io/api/v1/units",
headers=headers,
params={"page": 1, "size": 100}
).json()
print("Available units:")
for unit in units['items']:
print(f" - {unit['name']} ({unit['abbreviation']}): {unit['id']}")
# Common energy units
# kWh, MWh, GJ, m³, liters, therms

