Skip to content

API Reference

This page contains the technical API documentation for the GHGPy package.

Activities Module

Energy Sector

GHG inventory for Energy sector Base on the IPCC 2006 w 2019 refinement (c) Bui Khac Tu (bkt92)

Combustion

Bases: ADCombussion

Stationary combustion (SC) name: Activity name desc: More description fuels_list: Fuel list efs_list: List of emission factor corresponding to fuels results: Emission evaluation result

Source code in src\ghgpy\activities\energy.py
class Combustion(ADCombussion):
    """
    Stationary combustion (SC)
    `name`: Activity name
    `desc`: More description
    `fuels_list`: Fuel list
    `efs_list`: List of emission factor corresponding to fuels
    `results`: Emission evaluation result <not input>
    """
    #def __init__(self, **kwargs):
    #    super().__init__(**kwargs)
    def connection(self, fuel_db, ef_db, ghg_db):
        self.fuel_db = fuel_db
        self.ef_db = ef_db
        self.ghg_db = ghg_db

    def eval(self):
        """
        Evaluate the activity for ghg emission
        - Current GHG: CO2, CH4 and N2O
        - Required GHG database
        - Return a list of GHG emission from this activity
        """
        if len(self.fuels_list) == 0:
            return "No fuel in this activity, please add some fuels"
        else:
            co2v, ch4v, n2ov = 0, 0, 0
            for key, fuel in self.fuels_list.items():
                energy = fuel.cal_energy()
                co2v += energy*self.efs_list[key]['co2'].to_ufnum()
                ch4v += energy*self.efs_list[key]['ch4'].to_ufnum()
                n2ov += energy*self.efs_list[key]['n2o'].to_ufnum()
            co2 = CO2(ufnum(co2v).to_unum(), 'kg', self.ghg_db)
            ch4 = CH4(ufnum(ch4v).to_unum(), 'kg', self.ghg_db)
            n2o = N2O(ufnum(n2ov).to_unum(), 'kg', self.ghg_db)
        self.results = [co2, ch4, n2o]
        return self.results

    def emission(self, unit='t'):
        """
        Total emission in unit of CO2e (`Default in metric tonnes`)
        """
        self.eval()
        return weigh_units.convert(sum([x.emission() for x in self.results]), 't', unit)

    def add(self, name, amount, unit, fuel_db=None, efdb=None, force=False):
        """
        Add a fuel to fuels list 
        - Required database for emission factors
        - Required fuel class (datamodel)
        - `force` = True: Overwrite existing fuel data

        """
        if efdb == None:
            if self.ef_db == None:
                logging.info("No db connection")
                return "No db connection"
            else:
                efdb = self.ef_db
        if fuel_db == None:
            if self.fuel_db == None:
                logging.info("No db connection")
            else:
                fuel_db = self.fuel_db

        if not force:
            if name in self.fuels_list.keys():
                return f'{name} exist in list, using `force = True` to overwrite!'
        if not efdb.check_exist(name):
            raise ValueError(f'{name} not found in database.')
        efco2 = UNumber(**efdb.get(name)["co2"])
        efch4 = UNumber(**efdb.get(name)["ch4"])
        efn2o = UNumber(**efdb.get(name)["n2o"])
        self.fuels_list[name] = DefaultFuel(name, amount, unit, fuel_db)
        self.efs_list[name] = {'co2': efco2, 'ch4': efch4, 'n2o': efn2o}
        return f'Added {name} to {self.name}'

    def add_custom(self, fuel_data, ef_data, force=False):
        """
        Add a custom fuel to fuels list with custom data
        fuel_data = {
            'name': 'Diesel_Oil',
            'amount': {'value': 1000.0, 'uncertainty': 0},
            'unit': 'l',
            'properties': {
                'desc': 'Gas/Diesel Oil',
                'ncv': {'value': 43.0, 'uncertainty': 0.0},
                'ccf': {'value': 20.2, 'uncertainty': 0.4},
                'density': {'value': 844.0, 'uncertainty': 0.0}}}

        ef_data = {
            "co2": {"value": 54600,
                    "uncertainty": 0
                    },
            "ch4": {"value": 1,
                    "uncertainty": 0
                    },
            "n2o": {"value": 0.1,
                    "uncertainty": 0}
        """
        if not force:
            if fuel_data["name"] in self.fuels_list.keys():
                return f'{fuel_data["name"]} exist in list, using `force = True` to overwrite!'

        efco2 = UNumber(**ef_data["co2"])
        efch4 = UNumber(**ef_data["ch4"])
        efn2o = UNumber(**ef_data["n2o"])
        self.fuels_list[fuel_data["name"]] = BaseFuel(**fuel_data)
        self.efs_list[fuel_data["name"]] = {'co2': efco2, 'ch4': efch4, 'n2o': efn2o}
        return f'Added {fuel_data["name"]} to {self.name}'

    def remove(self, name: str):
        """
        Remove a fuel from the list\n
        Use list_fuels() method to check the list
        """
        if name in self.fuels_list.keys():
            self.fuels_list.pop(name)
            self.efs_list.pop(name)
            return f'Removed {name} from {self.name}'
        else:
            return f'{name} not in the list'

    def list_fuels(self):
        """
        List all fuels in the combustion class
        """
        return self.fuels_list.keys()

    @classmethod
    def load_data(cls, data):
        """
        Load data to the class
        """
        return cls(**data)

add(name, amount, unit, fuel_db=None, efdb=None, force=False)

Add a fuel to fuels list - Required database for emission factors - Required fuel class (datamodel) - force = True: Overwrite existing fuel data

Source code in src\ghgpy\activities\energy.py
def add(self, name, amount, unit, fuel_db=None, efdb=None, force=False):
    """
    Add a fuel to fuels list 
    - Required database for emission factors
    - Required fuel class (datamodel)
    - `force` = True: Overwrite existing fuel data

    """
    if efdb == None:
        if self.ef_db == None:
            logging.info("No db connection")
            return "No db connection"
        else:
            efdb = self.ef_db
    if fuel_db == None:
        if self.fuel_db == None:
            logging.info("No db connection")
        else:
            fuel_db = self.fuel_db

    if not force:
        if name in self.fuels_list.keys():
            return f'{name} exist in list, using `force = True` to overwrite!'
    if not efdb.check_exist(name):
        raise ValueError(f'{name} not found in database.')
    efco2 = UNumber(**efdb.get(name)["co2"])
    efch4 = UNumber(**efdb.get(name)["ch4"])
    efn2o = UNumber(**efdb.get(name)["n2o"])
    self.fuels_list[name] = DefaultFuel(name, amount, unit, fuel_db)
    self.efs_list[name] = {'co2': efco2, 'ch4': efch4, 'n2o': efn2o}
    return f'Added {name} to {self.name}'

add_custom(fuel_data, ef_data, force=False)

Add a custom fuel to fuels list with custom data fuel_data = { 'name': 'Diesel_Oil', 'amount': {'value': 1000.0, 'uncertainty': 0}, 'unit': 'l', 'properties': { 'desc': 'Gas/Diesel Oil', 'ncv': {'value': 43.0, 'uncertainty': 0.0}, 'ccf': {'value': 20.2, 'uncertainty': 0.4}, 'density': {'value': 844.0, 'uncertainty': 0.0}}}

ef_data = { "co2": {"value": 54600, "uncertainty": 0 }, "ch4": {"value": 1, "uncertainty": 0 }, "n2o": {"value": 0.1, "uncertainty": 0}

Source code in src\ghgpy\activities\energy.py
def add_custom(self, fuel_data, ef_data, force=False):
    """
    Add a custom fuel to fuels list with custom data
    fuel_data = {
        'name': 'Diesel_Oil',
        'amount': {'value': 1000.0, 'uncertainty': 0},
        'unit': 'l',
        'properties': {
            'desc': 'Gas/Diesel Oil',
            'ncv': {'value': 43.0, 'uncertainty': 0.0},
            'ccf': {'value': 20.2, 'uncertainty': 0.4},
            'density': {'value': 844.0, 'uncertainty': 0.0}}}

    ef_data = {
        "co2": {"value": 54600,
                "uncertainty": 0
                },
        "ch4": {"value": 1,
                "uncertainty": 0
                },
        "n2o": {"value": 0.1,
                "uncertainty": 0}
    """
    if not force:
        if fuel_data["name"] in self.fuels_list.keys():
            return f'{fuel_data["name"]} exist in list, using `force = True` to overwrite!'

    efco2 = UNumber(**ef_data["co2"])
    efch4 = UNumber(**ef_data["ch4"])
    efn2o = UNumber(**ef_data["n2o"])
    self.fuels_list[fuel_data["name"]] = BaseFuel(**fuel_data)
    self.efs_list[fuel_data["name"]] = {'co2': efco2, 'ch4': efch4, 'n2o': efn2o}
    return f'Added {fuel_data["name"]} to {self.name}'

emission(unit='t')

Total emission in unit of CO2e (Default in metric tonnes)

Source code in src\ghgpy\activities\energy.py
def emission(self, unit='t'):
    """
    Total emission in unit of CO2e (`Default in metric tonnes`)
    """
    self.eval()
    return weigh_units.convert(sum([x.emission() for x in self.results]), 't', unit)

eval()

Evaluate the activity for ghg emission - Current GHG: CO2, CH4 and N2O - Required GHG database - Return a list of GHG emission from this activity

Source code in src\ghgpy\activities\energy.py
def eval(self):
    """
    Evaluate the activity for ghg emission
    - Current GHG: CO2, CH4 and N2O
    - Required GHG database
    - Return a list of GHG emission from this activity
    """
    if len(self.fuels_list) == 0:
        return "No fuel in this activity, please add some fuels"
    else:
        co2v, ch4v, n2ov = 0, 0, 0
        for key, fuel in self.fuels_list.items():
            energy = fuel.cal_energy()
            co2v += energy*self.efs_list[key]['co2'].to_ufnum()
            ch4v += energy*self.efs_list[key]['ch4'].to_ufnum()
            n2ov += energy*self.efs_list[key]['n2o'].to_ufnum()
        co2 = CO2(ufnum(co2v).to_unum(), 'kg', self.ghg_db)
        ch4 = CH4(ufnum(ch4v).to_unum(), 'kg', self.ghg_db)
        n2o = N2O(ufnum(n2ov).to_unum(), 'kg', self.ghg_db)
    self.results = [co2, ch4, n2o]
    return self.results

list_fuels()

List all fuels in the combustion class

Source code in src\ghgpy\activities\energy.py
def list_fuels(self):
    """
    List all fuels in the combustion class
    """
    return self.fuels_list.keys()

load_data(data) classmethod

Load data to the class

Source code in src\ghgpy\activities\energy.py
@classmethod
def load_data(cls, data):
    """
    Load data to the class
    """
    return cls(**data)

remove(name)

Remove a fuel from the list

Use list_fuels() method to check the list

Source code in src\ghgpy\activities\energy.py
def remove(self, name: str):
    """
    Remove a fuel from the list\n
    Use list_fuels() method to check the list
    """
    if name in self.fuels_list.keys():
        self.fuels_list.pop(name)
        self.efs_list.pop(name)
        return f'Removed {name} from {self.name}'
    else:
        return f'{name} not in the list'

m_combustion

Mobile combustion

Calculate GHG emissions from mobile sources (vehicles).

Source code in src\ghgpy\activities\energy.py
class m_combustion:
    """
    Mobile combustion

    Calculate GHG emissions from mobile sources (vehicles).
    """
    def __init__(self, name='', vehicles=None) -> None:
        self.name = name
        self.vehicles = vehicles if vehicles is not None else []

Data Model Module

Fuel Classes

iClimate - ghgpy

Fuel objects Represent a amount of a specific fuel (C) Bui Khac Tu (bkt92) (C) iClimate

BaseFuel

Bases: FuelData

Base Fuel Object (use for all kind of fuel)

Representative by TJ

Allow "+, -" operators with the same kind of fuel

Allow convert to different property (Weight, Volume, Carbon Content)

Source code in src\ghgpy\datamodel\fuel.py
class BaseFuel(FuelData):
    """
    Base Fuel Object (use for all kind of fuel) \n
    Representative by TJ \n
    Allow "+, -" operators with the same kind of fuel \n
    Allow convert to different property (Weight, Volume, Carbon Content)
    """
    # Init attributes of fuel
    #def __init__(self, **kwargs):
    #    super().__init__(**kwargs)

    # Validate unit
    def _is_valid_unit(self, unit):
        if not ((unit in weigh_units.units) or (unit in volume_units.units)):
            raise ValueError("Invalid Unit.")
        return unit

    # Convert to TJ
    def cal_energy(self, unit='tj'):
        '''
        Energy of Fuel in unit \n
        Default unit `Tj`
        '''
        if any(x == None for x in [self.amount, self.properties.ncv]):
            return None
        amount = self.amount.to_ufnum()
        ncv = self.properties.ncv.to_ufnum()
        if self.unit in weigh_units.units:
            energy = weigh_units.convert(amount, self.unit, 'Gg')*ncv
            return energy_units.convert(energy, 'tj', unit)
        if self.unit in volume_units.units:
            if self.properties.density == None:
                return None
            else:
                density = self.properties.density.to_ufnum()
                energy = weigh_units.convert(volume_units.convert(amount, self.unit, 'm3')\
                                           *density, 'kg', 'Gg')*ncv
                return energy_units.convert(energy, 'tj', unit)

    # Convert to kg
    def cal_weight(self, unit='kg'):
        '''
        Weight of Fuel in unit \n
        Unit default `kg`
        '''
        amount = self.amount.to_ufnum()
        if self.unit in weigh_units.units:
            return weigh_units.convert(amount, self.unit, unit)
        if self.unit in volume_units.units:
            if self.properties.density == None:
                return None
            else:
                density = self.properties.density.to_ufnum()
                weight = volume_units.convert(amount, self.unit, 'm3')*density
                return weigh_units.convert(weight, 'kg', unit)

    # Convert carbon content (kg)
    def calc_cc(self, unit='kg'):
        '''
        Carbon content of Fuel in unit \n
        Default unit `kg`
        '''
        if self.properties.ccf == None:
            return None
        else:
            ccf = self.properties.ccf.to_ufnum()
            return weigh_units.convert(1000*self.cal_energy()*ccf, 'kg', unit)

    # Convert to volume (litre)
    def cal_volume(self, unit='l'):
        '''
        Volume of Fuel in unit \n
        Default unit `litre`
        '''
        amount = self.amount.to_ufnum()
        if self.unit in weigh_units.units:
            if self.properties.density == None:
                return None
            else:
                density = self.properties.density.to_ufnum()
                volume = weigh_units.convert(amount, self.unit, 'kg')/density
                return volume_units.convert(volume, 'm3', unit)
        if self.unit in volume_units.units:
            return volume_units.convert(amount, self.unit, unit)

    # Check two object have enough fuel data
    def _check_fuel(self, other):
        if not isinstance(other, BaseFuel):
            return NotImplemented
        if (self.name, self.properties.desc, self.properties.ncv, self.properties.density) == \
            (other.name, other.properties.desc, other.properties.ncv, other.properties.density):
            return True
        else:
            return False

    # Convert data to dict
    def to_dict(self, properties = True):
        if properties:
            return self.dict()
        else:
            data = self.dict()
            data.pop("properties")
            return data

    # Return to tj (energy of fuel)
    def __repr__(self) -> str:
        return (
            'fuel('
            f'name={self.name!r}, amount={self.cal_energy()!r} Tj)'
        )

    def __hash__(self) -> int:
        return hash((self.name, self.properties.desc, self.cal_energy(), self.properties.ncv))

    def __eq__(self, other) -> bool:
        if not isinstance(other, BaseFuel):
            return NotImplemented
        return (
            (self.name, self.cal_energy(), self.properties.ncv) == 
            (other.name, other.cal_energy(), other.properties.ncv))

    def __add__(self, other) -> float:
        if not isinstance(other, BaseFuel):
            return NotImplemented
        if self._check_fuel(other):
            sum = self.cal_weight()+other.cal_weight()
            amount = ufnum(sum.value, sum.uncertainty)
            return BaseFuel(self.name, self.properties.desc, self.properties.ncv.to_ufnum(), \
                            self.properties.ccf.to_ufnum(), self.properties.density.to_ufnum(), amount, "kg")    
        else:
            return NotImplemented

    def __sub__(self, other) -> float:
        if not isinstance(other, BaseFuel):
            return NotImplemented
        if self._check_fuel(other):
            sub = self.cal_weight()-other.cal_weight()
            amount = ufnum(sub.value, sub.uncertainty)
            return BaseFuel(self.name, self.properties.desc, self.properties.ncv.to_ufnum(), \
                            self.properties.ccf.to_ufnum(), self.properties.density.to_ufnum(), amount, "kg")    
        else:
            return NotImplemented

    # Load fuel from json file
    @classmethod
    def from_dict(cls, data):
        '''
        Load json file to fuel class
        schema:
        { 
        name: Name/code of fuel,
        amount: Amount of fuel (unit),
        unit: Unit of amount fuel,
        properties:
            {
            desc: More information about fuel
            ncv: Net calorific value (Tj/Gg)
            ccf: Carbon content of fuel (kg/GJ)
            density: Density of fuel (kg/m3)
            }
        }
        '''
        return cls(**data)                

cal_energy(unit='tj')

Energy of Fuel in unit

Default unit Tj

Source code in src\ghgpy\datamodel\fuel.py
def cal_energy(self, unit='tj'):
    '''
    Energy of Fuel in unit \n
    Default unit `Tj`
    '''
    if any(x == None for x in [self.amount, self.properties.ncv]):
        return None
    amount = self.amount.to_ufnum()
    ncv = self.properties.ncv.to_ufnum()
    if self.unit in weigh_units.units:
        energy = weigh_units.convert(amount, self.unit, 'Gg')*ncv
        return energy_units.convert(energy, 'tj', unit)
    if self.unit in volume_units.units:
        if self.properties.density == None:
            return None
        else:
            density = self.properties.density.to_ufnum()
            energy = weigh_units.convert(volume_units.convert(amount, self.unit, 'm3')\
                                       *density, 'kg', 'Gg')*ncv
            return energy_units.convert(energy, 'tj', unit)

cal_volume(unit='l')

Volume of Fuel in unit

Default unit litre

Source code in src\ghgpy\datamodel\fuel.py
def cal_volume(self, unit='l'):
    '''
    Volume of Fuel in unit \n
    Default unit `litre`
    '''
    amount = self.amount.to_ufnum()
    if self.unit in weigh_units.units:
        if self.properties.density == None:
            return None
        else:
            density = self.properties.density.to_ufnum()
            volume = weigh_units.convert(amount, self.unit, 'kg')/density
            return volume_units.convert(volume, 'm3', unit)
    if self.unit in volume_units.units:
        return volume_units.convert(amount, self.unit, unit)

cal_weight(unit='kg')

Weight of Fuel in unit

Unit default kg

Source code in src\ghgpy\datamodel\fuel.py
def cal_weight(self, unit='kg'):
    '''
    Weight of Fuel in unit \n
    Unit default `kg`
    '''
    amount = self.amount.to_ufnum()
    if self.unit in weigh_units.units:
        return weigh_units.convert(amount, self.unit, unit)
    if self.unit in volume_units.units:
        if self.properties.density == None:
            return None
        else:
            density = self.properties.density.to_ufnum()
            weight = volume_units.convert(amount, self.unit, 'm3')*density
            return weigh_units.convert(weight, 'kg', unit)

calc_cc(unit='kg')

Carbon content of Fuel in unit

Default unit kg

Source code in src\ghgpy\datamodel\fuel.py
def calc_cc(self, unit='kg'):
    '''
    Carbon content of Fuel in unit \n
    Default unit `kg`
    '''
    if self.properties.ccf == None:
        return None
    else:
        ccf = self.properties.ccf.to_ufnum()
        return weigh_units.convert(1000*self.cal_energy()*ccf, 'kg', unit)

from_dict(data) classmethod

Load json file to fuel class schema: { name: Name/code of fuel, amount: Amount of fuel (unit), unit: Unit of amount fuel, properties: { desc: More information about fuel ncv: Net calorific value (Tj/Gg) ccf: Carbon content of fuel (kg/GJ) density: Density of fuel (kg/m3) } }

Source code in src\ghgpy\datamodel\fuel.py
@classmethod
def from_dict(cls, data):
    '''
    Load json file to fuel class
    schema:
    { 
    name: Name/code of fuel,
    amount: Amount of fuel (unit),
    unit: Unit of amount fuel,
    properties:
        {
        desc: More information about fuel
        ncv: Net calorific value (Tj/Gg)
        ccf: Carbon content of fuel (kg/GJ)
        density: Density of fuel (kg/m3)
        }
    }
    '''
    return cls(**data)                

DefaultFuel

Bases: BaseFuel

amount: amount of fuel

unit: unit of fuel


Data Sources:

Calorific value: 2006 IPCC Guidelines for National Greenhouse Gas Inventories V2_Ch1 - TABLE 1.2: https://www.ipcc-nggip.iges.or.jp/public/2006gl/pdf/2_Volume2/V2_1_Ch1_Introduction.pdf

Density: IEA Database documentation: https://wds.iea.org/wds/pdf/oil_documentation.pdf

Source code in src\ghgpy\datamodel\fuel.py
class DefaultFuel(BaseFuel):
    '''
    `amount: amount of fuel`\n
    `unit: unit of fuel`

    -------------
    Data Sources: \n
    Calorific value: 2006 IPCC Guidelines for National Greenhouse Gas Inventories V2_Ch1 - TABLE 1.2:
    https://www.ipcc-nggip.iges.or.jp/public/2006gl/pdf/2_Volume2/V2_1_Ch1_Introduction.pdf \n
    Density: IEA Database documentation:
    https://wds.iea.org/wds/pdf/oil_documentation.pdf \n

    '''
    def __init__(self, fuel: str, amount: UNumber, unit: str, database :FuelDataHandle):
        if not database.check_exist(fuel):
            raise ValueError("Fuel not found in database.")

        properties = FuelProperties(**database.get(fuel))
        super().__init__(**{"name": fuel, "amount": amount, "unit": unit, "properties": properties})

    # Load fuel from json file
    @classmethod
    def from_json(cls, database, data):
        '''
        Load json file to fuel class
        Schema:
        { 
        name: Name/code of fuel,
        amount: amount of fuel (unit),
        unit: unit of amount fuel
        }
        '''
        if not database.check_exist(data["name"]):
            raise ValueError("Fuel not found in database.")
        return cls(database, data["name"], data["amount"],  data["unit"])

from_json(database, data) classmethod

Load json file to fuel class Schema: { name: Name/code of fuel, amount: amount of fuel (unit), unit: unit of amount fuel }

Source code in src\ghgpy\datamodel\fuel.py
@classmethod
def from_json(cls, database, data):
    '''
    Load json file to fuel class
    Schema:
    { 
    name: Name/code of fuel,
    amount: amount of fuel (unit),
    unit: unit of amount fuel
    }
    '''
    if not database.check_exist(data["name"]):
        raise ValueError("Fuel not found in database.")
    return cls(database, data["name"], data["amount"],  data["unit"])

energy_units

Convert between difference energy units

Source code in src\ghgpy\datamodel\unit_converters.py
class energy_units:
    """
    Convert between difference energy units
    """
    units = ['j', 'kj', 'mj', 'tj', 'gj', 'Tj', 'kWh', 'Wh','MWh', 'GWh', 'TWh']
    @staticmethod
    def convert(val: float, unit_in: str, unit_out: str) -> float:
        SI = {'j': 1.0, 'kj': 1000.0, 'mj': 1.0*10**6, 'gj': 1.0*10**9, 'tj': 1.0*10**12, \
              'Wh': 3600 , 'kWh': 3600*10**3, 'MWh': 3600*10**6, 'GWh': 3600*10**9, 'TWh': 3600*10**12}
        return val * SI[unit_in] / SI[unit_out]

volume_units

Convert between difference volume units

Source code in src\ghgpy\datamodel\unit_converters.py
class volume_units:
    """
    Convert between difference volume units
    """
    units = ['litre', 'l', 'm3']
    @staticmethod
    def convert(val: float, unit_in: str, unit_out: str) -> float:
        SI = {'litre': 1.0, 'l': 1.0, 'm3': 1000.0}
        return val * SI[unit_in] / SI[unit_out]

weigh_units

Convert between difference weigh units

Source code in src\ghgpy\datamodel\unit_converters.py
class weigh_units:
    """
    Convert between difference weigh units
    """
    units = ['g', 'kg', 'tonne', 'Gg']
    @staticmethod
    def convert(val: float, unit_in: str, unit_out: str) -> float:
        SI = {'g': 1.0, 'kg': 1000.0, 'tonne': 1.0*10**6, 't': 1.0*10**6,'tonnes': 1.0*10**6, 'Gg': 1.0*10**9}
        return val * SI[unit_in] / SI[unit_out]

dict_to_fuel(data, database)

Schema: Class FuelData

Load json file to fuel class schema: { name: Name/code of fuel, amount: Amount of fuel (unit), unit: Unit of amount fuel, properties: { desc: More information about fuel ncv: Net calorific value (Tj/Gg) ccf: Carbon content of fuel (kg/GJ) density: Density of fuel (kg/m3) } }

Source code in src\ghgpy\datamodel\fuel.py
def dict_to_fuel(data, database):
    '''
    Schema: Class FuelData

    Load json file to fuel class
    schema:
    { 
    name: Name/code of fuel,
    amount: Amount of fuel (unit),
    unit: Unit of amount fuel,
    properties:
        {
        desc: More information about fuel
        ncv: Net calorific value (Tj/Gg)
        ccf: Carbon content of fuel (kg/GJ)
        density: Density of fuel (kg/m3)
        }
    }
    '''
    _required_data = ['name', 'amount', 'unit']
    _attributes_list = set(['desc', 'ncv', 'ccf', 'density'])

    if not all(x in data.keys() for x in _required_data):
        raise ValueError("Invalid data, missing fields!")

    elif not "properties" in data.keys():
        if not data["name"] in database.keys():
            raise ValueError("Invalid data, fuel name not found in default fuel list!")
        else:
            fuel = data["name"]
            return DefaultFuel(database, data["name"], ufnum.from_dict(data["amount"]),  data["unit"])
    else:
        if all(x in data["properties"].keys() for x in _attributes_list):
            return BaseFuel(data["name"], data["properties"]['desc'], ufnum.from_dict(data["properties"]['ncv']), \
                        ufnum.from_dict(data["properties"]['ccf']), ufnum.from_dict(data["properties"]['density']), \
                            ufnum.from_dict(data["amount"]), data["unit"])                
        elif not data["name"] in database.keys():
            raise ValueError("Invalid data, missing atribute which can not get from default fuel list!")
        else:
            fuel = data["name"]
            input = {}
            for i in _attributes_list.difference(data["properties"].keys()):
                input[i] = database.get(fuel)[i]
            for i in data["properties"].keys():
                input[i] = data["properties"][i]
            return BaseFuel(data["name"], input['desc'], ufnum.from_dict(input['ncv']), \
                        ufnum.from_dict(input['ccf']), ufnum.from_dict(input['density']), \
                            ufnum.from_dict(data["amount"]),  data["unit"])

GHG Gas Classes

iClimate - ghgpy

GHG Gas objects Represent a amount of a specific ghg gas (C) Bui Khac Tu (bkt92) (C) iClimate

GHGGas

Bases: GHGDATA

GHG Gas Object (use for all kind of GHG gas)

Representative by tCO2e

Allow "+, -" operators with the same kind of gas

Source code in src\ghgpy\datamodel\ghg.py
class GHGGas(GHGDATA):
    """
    GHG Gas Object (use for all kind of GHG gas) \n
    Representative by tCO2e \n
    Allow "+, -" operators with the same kind of gas
    """
    # Convert to tonnes
    def cal_weight(self, unit='tonne'):
        '''
        Convert to metric unit /n
        Default unit metric tonnes
        '''
        amount = self.amount.to_ufnum()
        if self.unit in weigh_units.units:
            return weigh_units.convert(amount, self.unit, unit)
        if self.unit in volume_units.units:
            if self.density == None:
                return None
            else:
                weight = volume_units.convert(amount, self.unit, 'm3')*self.density
                return weigh_units.convert(weight, 'kg', unit)

    def emission(self, unit='t'):
        return weigh_units.convert(self.cal_weight()*self.gwp, 't', unit)

    def __repr__(self) -> str:
        return (f'name={self.name!r}, amount={self.emission()!r} tCO2e')

    def __hash__(self) -> int:
        return hash((self.amount, self.name, self.gwp))

    def __eq__(self, other) -> bool:
        if not isinstance(other, GHGGas):
            return NotImplemented
        return (
            (self.amount, self.name, self.gwp) == 
            (other.amount, other.name, other.gwp))

    def __add__(self, other) -> float:
        if not isinstance(other, GHGGas):
            return NotImplemented
        if (self.name, self.gwp) == (other.name, other.gwp):
            amount  = ufnum(self.cal_weight() + other.cal_weight())
            return GHGGas(self.name, amount, "tonne", \
                          self.desc, self.gwp, self.density)
        else:
            return NotImplemented

    def __sub__(self, other) -> float:
        if not isinstance(other, GHGGas):
            return NotImplemented
        if (self.name, self.gwp) == (other.name, other.gwp):
            amount  = ufnum(self.cal_weight() - other.cal_weight())
            return GHGGas(self.name, amount, "tonne", \
                          self.desc, self.gwp, self.density)
        else:
            return NotImplemented

    @classmethod
    def from_dict(cls, data):
        '''
        {'name': {'title': 'Name', 'type': 'string'},
        'amount': {'title': 'Amount', 'type': 'number'},
        'unit': {'title': 'Unit', 'type': 'string'},
        'desc': {'title': 'Desc', 'type': 'string'},
        'gwp': {'title': 'Gwp', 'type': 'number'},
        'density': {'title': 'Density', 'type': 'number'}}
        '''
        return cls(**data)

cal_weight(unit='tonne')

Convert to metric unit /n Default unit metric tonnes

Source code in src\ghgpy\datamodel\ghg.py
def cal_weight(self, unit='tonne'):
    '''
    Convert to metric unit /n
    Default unit metric tonnes
    '''
    amount = self.amount.to_ufnum()
    if self.unit in weigh_units.units:
        return weigh_units.convert(amount, self.unit, unit)
    if self.unit in volume_units.units:
        if self.density == None:
            return None
        else:
            weight = volume_units.convert(amount, self.unit, 'm3')*self.density
            return weigh_units.convert(weight, 'kg', unit)

from_dict(data) classmethod

{'name': {'title': 'Name', 'type': 'string'}, 'amount': {'title': 'Amount', 'type': 'number'}, 'unit': {'title': 'Unit', 'type': 'string'}, 'desc': {'title': 'Desc', 'type': 'string'}, 'gwp': {'title': 'Gwp', 'type': 'number'}, 'density': {'title': 'Density', 'type': 'number'}}

Source code in src\ghgpy\datamodel\ghg.py
@classmethod
def from_dict(cls, data):
    '''
    {'name': {'title': 'Name', 'type': 'string'},
    'amount': {'title': 'Amount', 'type': 'number'},
    'unit': {'title': 'Unit', 'type': 'string'},
    'desc': {'title': 'Desc', 'type': 'string'},
    'gwp': {'title': 'Gwp', 'type': 'number'},
    'density': {'title': 'Density', 'type': 'number'}}
    '''
    return cls(**data)

energy_units

Convert between difference energy units

Source code in src\ghgpy\datamodel\unit_converters.py
class energy_units:
    """
    Convert between difference energy units
    """
    units = ['j', 'kj', 'mj', 'tj', 'gj', 'Tj', 'kWh', 'Wh','MWh', 'GWh', 'TWh']
    @staticmethod
    def convert(val: float, unit_in: str, unit_out: str) -> float:
        SI = {'j': 1.0, 'kj': 1000.0, 'mj': 1.0*10**6, 'gj': 1.0*10**9, 'tj': 1.0*10**12, \
              'Wh': 3600 , 'kWh': 3600*10**3, 'MWh': 3600*10**6, 'GWh': 3600*10**9, 'TWh': 3600*10**12}
        return val * SI[unit_in] / SI[unit_out]

volume_units

Convert between difference volume units

Source code in src\ghgpy\datamodel\unit_converters.py
class volume_units:
    """
    Convert between difference volume units
    """
    units = ['litre', 'l', 'm3']
    @staticmethod
    def convert(val: float, unit_in: str, unit_out: str) -> float:
        SI = {'litre': 1.0, 'l': 1.0, 'm3': 1000.0}
        return val * SI[unit_in] / SI[unit_out]

weigh_units

Convert between difference weigh units

Source code in src\ghgpy\datamodel\unit_converters.py
class weigh_units:
    """
    Convert between difference weigh units
    """
    units = ['g', 'kg', 'tonne', 'Gg']
    @staticmethod
    def convert(val: float, unit_in: str, unit_out: str) -> float:
        SI = {'g': 1.0, 'kg': 1000.0, 'tonne': 1.0*10**6, 't': 1.0*10**6,'tonnes': 1.0*10**6, 'Gg': 1.0*10**9}
        return val * SI[unit_in] / SI[unit_out]

Unit Converters

Unit Converters for GHG inventory (c) Bui Khac Tu (bkt92)

energy_units

Convert between difference energy units

Source code in src\ghgpy\datamodel\unit_converters.py
class energy_units:
    """
    Convert between difference energy units
    """
    units = ['j', 'kj', 'mj', 'tj', 'gj', 'Tj', 'kWh', 'Wh','MWh', 'GWh', 'TWh']
    @staticmethod
    def convert(val: float, unit_in: str, unit_out: str) -> float:
        SI = {'j': 1.0, 'kj': 1000.0, 'mj': 1.0*10**6, 'gj': 1.0*10**9, 'tj': 1.0*10**12, \
              'Wh': 3600 , 'kWh': 3600*10**3, 'MWh': 3600*10**6, 'GWh': 3600*10**9, 'TWh': 3600*10**12}
        return val * SI[unit_in] / SI[unit_out]

volume_units

Convert between difference volume units

Source code in src\ghgpy\datamodel\unit_converters.py
class volume_units:
    """
    Convert between difference volume units
    """
    units = ['litre', 'l', 'm3']
    @staticmethod
    def convert(val: float, unit_in: str, unit_out: str) -> float:
        SI = {'litre': 1.0, 'l': 1.0, 'm3': 1000.0}
        return val * SI[unit_in] / SI[unit_out]

weigh_units

Convert between difference weigh units

Source code in src\ghgpy\datamodel\unit_converters.py
class weigh_units:
    """
    Convert between difference weigh units
    """
    units = ['g', 'kg', 'tonne', 'Gg']
    @staticmethod
    def convert(val: float, unit_in: str, unit_out: str) -> float:
        SI = {'g': 1.0, 'kg': 1000.0, 'tonne': 1.0*10**6, 't': 1.0*10**6,'tonnes': 1.0*10**6, 'Gg': 1.0*10**9}
        return val * SI[unit_in] / SI[unit_out]

Database Handlers

iClimate - ghgpy

Data handle

Middleware between app and database

(C) Bui Khac Tu (bkt92) (C) iClimate

FuelDataHandle

Data handle for fuel data in ghgpy

Source code in src\ghgpy\datamodel\db.py
class FuelDataHandle:
    """
    Data handle for fuel data in ghgpy
    """
    def __init__(self, database):
        self.data = database
    def check_exist(self, fuel):
        return fuel in self.data.keys()
    def get(self, fuel):
        return self.data[fuel]

fuels_database

The database class for the default and custom data

Default data includes
  • The NCVs are taken from IPCC 2006 w 2019 refinements
  • Densitys are taken from
  • EFs taken from IPCC 2006 w 2019 refinements
Source code in src\ghgpy\datamodel\db.py
class fuels_database:
    """
    The database class for the default and custom data

    Default data includes:
        - The NCVs are taken from IPCC 2006 w 2019 refinements
        - Densitys are taken from 
        - EFs taken from IPCC 2006 w 2019 refinements
    """
    def __init__(self, db='xlsx', *args, **kwargs):
        self.db = db
        if db == 'sqlite':
            engine = create_engine(f"sqlite:///{root_path}/database/database.db", echo=True)
        if db == 'mysql':
            pass
        if db == 'xlsx':
            self.data = pd.read_excel(os.path.join(os.path.dirname(__file__), 'database.xlsx'))

    def connect(self):
        """
        Connect to database engineer \n
        Not applicable for import data from csv or xlsx
        """
        connection = self.engine.connect()

    def update(self, bylist=False):
        """
        Change default values to new values
        - bylist=False (Default): change only one value
        - bylist=True: input a dictionary of the values
        """
        pass

    def reset_to_default(self):
        """
        Reverse back to the default values
        """

    def datalist(self):
        """
        Print the list of default data on the database.
        """

connect()

Connect to database engineer

Not applicable for import data from csv or xlsx

Source code in src\ghgpy\datamodel\db.py
def connect(self):
    """
    Connect to database engineer \n
    Not applicable for import data from csv or xlsx
    """
    connection = self.engine.connect()

datalist()

Print the list of default data on the database.

Source code in src\ghgpy\datamodel\db.py
def datalist(self):
    """
    Print the list of default data on the database.
    """

reset_to_default()

Reverse back to the default values

Source code in src\ghgpy\datamodel\db.py
def reset_to_default(self):
    """
    Reverse back to the default values
    """

update(bylist=False)

Change default values to new values - bylist=False (Default): change only one value - bylist=True: input a dictionary of the values

Source code in src\ghgpy\datamodel\db.py
def update(self, bylist=False):
    """
    Change default values to new values
    - bylist=False (Default): change only one value
    - bylist=True: input a dictionary of the values
    """
    pass

Factory Module

iClimate - ghgpy

Factory objects Represent a factory which doing ghg inventory

(C) Bui Khac Tu (bkt92) (C) iClimate

FactoryGeneral

Bases: BaseModel

Source code in src\ghgpy\factory.py
class FactoryGeneral(BaseModel, extra=Extra.allow):
    name: Optional[str] = None # name of factory
    desc: Optional[str] = None # factory description
    # Scope 1 emission
    combustion: Dict[str, Combustion] = {} # Stationary combustion
    refrigerantuse: Dict[str, RefrigerantUse] = {} # Product use emission
    # Scope 2 emission
    elecuse: Dict[str, ElecUse] = {} # Indirect emission by using electricity
    # Scope 3 emission not implement yet
    fuel_db: Optional[object] = Field(exclude=True)
    ghg_db: Optional[object] = Field(exclude=True)
    ef_db: Optional[object] = Field(exclude=True)

    def connection(self, fuel_db, ef_db, ghg_db):
        self.fuel_db = fuel_db
        self.ef_db = ef_db
        self.ghg_db = ghg_db

    def add_combustion(self, name, desc: str = None, force=False):
        """
        Add a combussion process
        - `force` = True: Overwrite existing fuel data
        """
        if not force:
            if name in self.combustion.keys():
                return "The process already exist in the list, using `force = True` to overwrite!" 
        self.combustion[name] = Combustion(name=name, desc=desc, fuel_db=self.fuel_db, \
                                           ef_db=self.ef_db, ghg_db=self.ghg_db)
        return f'Added {name} to {self.name}'

    def remove_combustion(self, name: str):
        """
        Remove a combussion process\n
        """
        if name in self.combustion.keys():
            self.combustion.pop(name)
            return f'Removed {name} from {self.name}'
        else:
            f'{name} not in the list'

    def add_refrigerantuse(self, name, desc=None, process=None, force=False):
        """
        Add a emission from refrigerant
        - `force` = True: Overwrite existing fuel data
        """
        if not force:
            if name in self.refrigerantuse.keys():
                return "Already exist in the list, using `force = True` to overwrite!" 
        self.refrigerantuse[name] = RefrigerantUse(name=name, desc=desc, process=process, ghg_db=self.ghg_db)
        return f'Added {name} to {self.name}'

    def remove_refrigerantuse(self, name: str):
        """
        Remove emission from refrigerant\n
        """
        if name in self.refrigerantuse.keys():
            self.refrigerantuse.pop(name)
            return f'Removed {name} from {self.name}'
        else:
            return f'{name} not in the list'

    def add_elecuse(self, name, desc: str = None, process=None, force=False):
        """
        Add a electriciy end-users
        - `force` = True: Overwrite existing fuel data
        """
        if not force:
            if name in self.elecuse.keys():
                return "The end-users already exist in the list, using `force = True` to overwrite!" 
        self.elecuse[name] = ElecUse(name=name, desc=desc, process=process)
        return f'Added {name} to {self.name}'

    def remove_elecuse(self, name: str):
        """
        Remove a electriciy end-users
        """
        if name in self.elecuse.keys():
            self.elecuse.pop(name)
            return f'Removed {name} from {self.name}'
        else:
            f'{name} not in the list'

    def emission(self, scope=1):
        if scope == 1:
            emission = 0
            for process in self.combustion.values():
                emission += process.emission()
            for ref in self.refrigerantuse.values():
                emission += ref.emission()
            return emission
        if scope == 2:
            emission = 0
            for e in self.elecuse.values():
                emission += e.emission()
            return emission

    @classmethod
    def load_data(cls, data):
        """
        Load data to the class
        """
        return cls(**data) 

add_combustion(name, desc=None, force=False)

Add a combussion process - force = True: Overwrite existing fuel data

Source code in src\ghgpy\factory.py
def add_combustion(self, name, desc: str = None, force=False):
    """
    Add a combussion process
    - `force` = True: Overwrite existing fuel data
    """
    if not force:
        if name in self.combustion.keys():
            return "The process already exist in the list, using `force = True` to overwrite!" 
    self.combustion[name] = Combustion(name=name, desc=desc, fuel_db=self.fuel_db, \
                                       ef_db=self.ef_db, ghg_db=self.ghg_db)
    return f'Added {name} to {self.name}'

add_elecuse(name, desc=None, process=None, force=False)

Add a electriciy end-users - force = True: Overwrite existing fuel data

Source code in src\ghgpy\factory.py
def add_elecuse(self, name, desc: str = None, process=None, force=False):
    """
    Add a electriciy end-users
    - `force` = True: Overwrite existing fuel data
    """
    if not force:
        if name in self.elecuse.keys():
            return "The end-users already exist in the list, using `force = True` to overwrite!" 
    self.elecuse[name] = ElecUse(name=name, desc=desc, process=process)
    return f'Added {name} to {self.name}'

add_refrigerantuse(name, desc=None, process=None, force=False)

Add a emission from refrigerant - force = True: Overwrite existing fuel data

Source code in src\ghgpy\factory.py
def add_refrigerantuse(self, name, desc=None, process=None, force=False):
    """
    Add a emission from refrigerant
    - `force` = True: Overwrite existing fuel data
    """
    if not force:
        if name in self.refrigerantuse.keys():
            return "Already exist in the list, using `force = True` to overwrite!" 
    self.refrigerantuse[name] = RefrigerantUse(name=name, desc=desc, process=process, ghg_db=self.ghg_db)
    return f'Added {name} to {self.name}'

load_data(data) classmethod

Load data to the class

Source code in src\ghgpy\factory.py
@classmethod
def load_data(cls, data):
    """
    Load data to the class
    """
    return cls(**data) 

remove_combustion(name)

Remove a combussion process

Source code in src\ghgpy\factory.py
def remove_combustion(self, name: str):
    """
    Remove a combussion process\n
    """
    if name in self.combustion.keys():
        self.combustion.pop(name)
        return f'Removed {name} from {self.name}'
    else:
        f'{name} not in the list'

remove_elecuse(name)

Remove a electriciy end-users

Source code in src\ghgpy\factory.py
def remove_elecuse(self, name: str):
    """
    Remove a electriciy end-users
    """
    if name in self.elecuse.keys():
        self.elecuse.pop(name)
        return f'Removed {name} from {self.name}'
    else:
        f'{name} not in the list'

remove_refrigerantuse(name)

Remove emission from refrigerant

Source code in src\ghgpy\factory.py
def remove_refrigerantuse(self, name: str):
    """
    Remove emission from refrigerant\n
    """
    if name in self.refrigerantuse.keys():
        self.refrigerantuse.pop(name)
        return f'Removed {name} from {self.name}'
    else:
        return f'{name} not in the list'