Skip to main content

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:

Understanding Facility Emissions

Facility emissions typically include Scope 1 (on-site combustion) and Scope 2 (purchased electricity).

Emission Scopes

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:
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

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

# 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:
# 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"
}
Learn more about custom emission factors →

Step 3: Upload Natural Gas Invoices

Track on-site combustion (Scope 1):
# 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:
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'])}")
Common fuel types:
  • 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

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

# 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:
# 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:
# 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

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

# 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:
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:
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:
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:
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):
# 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

# 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

# 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

# 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

Next Steps