#! usr/bin/env python3
# -*- coding: utf-8 -*-
"""
..
Copyright 2018 G2Elab / MAGE
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from omegalpes.general.optimisation.core import OptObject
from omegalpes.general.optimisation.elements import *
from omegalpes.energy.units.energy_units import EnergyUnit
__docformat__ = "restructuredtext en"
[docs]
class EconomicObject(OptObject):
def __init__(self, time, name, **units):
OptObject.__init__(self, name=name)
self.description = 'economic basis object'
self.costs_exp = '0'
self.incomes_exp = '0'
self.operation_maintenance_exp = '0'
for key, unit in units.items():
if not hasattr(unit, 'eco_balance_t'):
unit.eco_balance_t = Quantity(
name='eco_balance_t', unit='€',
vlen=time.LEN,
description='economical balance for each period',
parent=unit)
unit.calc_eco_balance_t = DefinitionDynamicConstraint(
name='calc_eco_balance_t',
exp_t='{0}_eco_balance_t[t] == {0'
'}_incomes_t[t] - {0}_costs_t[t]'.format(unit.name),
parent=unit)
if not hasattr(unit, 'eco_balance_tot'):
unit.eco_balance_tot = Quantity(name='eco_balance_tot',
unit='€',
vlen=1,
description='total economical '
'balance',
parent=unit)
unit.calc_eco_balance = DefinitionConstraint(
name='calc_eco_balance_tot',
exp='{0}_eco_balance_tot == {0'
'}_incomes_tot - {0}_costs_tot'.format(
unit.name),
parent=unit)
# Methods to deal with new costs quantities
[docs]
def add_cost_to_costs_exp(self, unit, cost_name):
self.costs_exp += '+ {0}_{1}[t] '.format(unit.name, cost_name)
[docs]
def calc_costs(self, unit):
unit.calc_costs = DefinitionDynamicConstraint(
name='calc_costs',
exp_t='{0}_costs_t[t] == ({1})'.format(
unit.name, self.costs_exp),
t_range='for t in time.I',
parent=unit)
unit.calc_costs_tot = DefinitionConstraint(
name='calc_costs_tot',
exp='{0}_costs_tot == lpSum({1} for t in time.I)'.format(
unit.name, self.costs_exp),
parent=unit)
# Methods to deal with new incomes quantities
[docs]
def add_income_to_incomes_exp(self, unit, income_name):
self.incomes_exp += '+ {0}_{1}[t] '.format(unit.name, income_name)
[docs]
def calc_incomes(self, unit):
unit.calc_incomes = DefinitionDynamicConstraint(
name='calc_incomes',
exp_t='{0}_incomes_t[t] == ({1})'.format(
unit.name, self.incomes_exp),
t_range='for t in time.I',
parent=unit)
unit.calc_incomes_tot = DefinitionConstraint(
name='calc_incomes_tot',
exp='{0}_incomes_tot == lpSum({1} for t in time.I)'.format(
unit.name, self.incomes_exp),
parent=unit)
# Methods to deal with operation and maintenance costs quantities
[docs]
def add_cost_to_operation_and_maintenance_exp(self, unit, cost_name):
self.operation_maintenance_exp += '+ {0}_{1}[t] '.format(unit.name,
cost_name)
[docs]
def calc_operation_and_maintenance_costs(self, unit):
unit.calc_operation_and_maintenance_costs = \
DefinitionDynamicConstraint(
name='calc_operation_and_maintenance_costs',
exp_t='{0}_operation_and_maintenance_costs_t[t] == ({'
'1})'.format(
unit.name, self.operation_maintenance_exp),
t_range='for t in time.I',
parent=unit)
unit.calc_operation_and_maintenance_costs_tot = DefinitionConstraint(
name='calc_operation_and_maintenance_costs_tot',
exp='{0}_operation_and_maintenance_costs_tot == lpSum({1} for t '
'in time.I)'.format(
unit.name, self.operation_maintenance_exp),
parent=unit)
# Generic methods
[docs]
def add_cashflow(self, time, name, description, cashflow_value, **units):
"""
Add a cashflow (cost, income or transaction) to the unit
**Parameters**
- cashflow_value: float: value of the cash flow to consider
"""
value = [0 for t in time.I]
value[0] = cashflow_value
for key, unit in units.items():
cashflow = Quantity(name=name, unit='€',
vlen=time.LEN,
description=description,
value=value, parent=unit)
setattr(unit, name, cashflow)
[docs]
def add_rate_cashflow(self, time, name, description, rate_value,
cashflow_value, **units):
"""
Add a cashflow (cost, income or transaction), based on another
cashflow value, to the unit
**Parameters**
- rate_value
- cashflow_value: float: value of the cash flow to consider
"""
value = [0 for t in time.I]
value[0] = rate_value * cashflow_value / 100
for key, unit in units.items():
cashflow = Quantity(name=name, unit='€',
vlen=time.LEN,
description=description,
value=value, parent=unit)
setattr(unit, name, cashflow)
[docs]
def add_spread_cashflow(self, time, name, description,
cashflow_value,
spread_duration, **units):
"""
**Description**
Spread CashFlow class to enable to spread a cashflow over a
duration
or over the whole time study (income, transaction or cost)
**Parameters**
- investment_value: float: value of the investment to consider
- spread_duration: int: duration over which the investment will be
spread, the whole duration of the study is considered by definition
"""
value = []
if spread_duration:
if spread_duration > time.LEN:
raise ValueError('the spread duration of the investment {} '
'should be lower than the project '
'duration'.format(name))
else:
value_by_step = cashflow_value / spread_duration
if spread_duration == time.LEN:
value = [value_by_step for t in time.I]
else:
for t in range(spread_duration):
value.append(value_by_step)
for t in range(spread_duration, time.LEN):
value.append(0)
else:
value_by_step = cashflow_value / time.LEN
value = [value_by_step for t in time.I]
for key, unit in units.items():
spread_cashflow = Quantity(name=name, unit='€', vlen=time.LEN,
description=description,
value=value, parent=unit)
setattr(unit, name, spread_cashflow)
[docs]
def add_quantity_dependant_cashflow(self, time, name, description, value,
quantity_name, quantity_unit, **units):
for key, unit in units.items():
qtt_dpt_cashflow = Quantity(name=name,
description=description,
lb=0, vlen=time.LEN,
parent=unit)
setattr(unit, name, qtt_dpt_cashflow)
if isinstance(value, (int, float)):
calc_value = DefinitionDynamicConstraint(
name='calc_{}'.format(name),
exp_t='{0}_{1}[t] == {2} * '
'{4}_{3}[t] * time.DT'.format(unit.name, name,
value, quantity_name,
quantity_unit.name),
t_range='for t in time.I', parent=unit)
elif isinstance(value, list):
if len(value) != time.LEN:
raise IndexError(
"Your operating cost should be the size of the "
"time "
"period.")
else:
calc_value = DefinitionDynamicConstraint(
name='calc_{}'.format(name),
exp_t='{0}_{1}[t] == {2}[t] * '
'{4}_{3}[t] * time.DT'.format(unit.name, name,
value,
quantity_name,
quantity_unit.name),
t_range='for t in time.I', parent=unit)
else:
raise TypeError("The cost linked to a "
"quantity_dependant_cashflow"
" should be an int, float or a"
"list and is {}".format(type(value)))
from omegalpes.economy.costs import EnergyUnitCosts
from omegalpes.economy.incomes import EnergyUnitIncomes
[docs]
class EnergyUnitCostsIncomes(EconomicObject):
"""
Class to define the costs and the incomes associated to an energy unit
**Parameters**
:param time: time of the model
:param energy_unit: energy unit on which the costs and incomes will be
added
:param investment_value : value of the investment
:param investment_spread_duration : duration on which the investment
is spread
"""
def __init__(self, *, time, energy_unit,
energy_cost: int or float or list = None,
operation_starting_cost: int or float or list = None,
operation_switch_off_cost: int or float or list = None,
operation_fixed_cost: int or float or list = None,
maintenance_fixed_cost: int or float or list = None,
maintenance_investment_rate_cost: int or float or list = None,
investment_cost: int or float = None,
investment_spread_duration: int = None,
energy_incomes=None,
grant_incomes=None,
grant_spread_duration=None
):
self.description = 'External costs and incomes linked to an energy ' \
'unit'
self.incomes = EnergyUnitIncomes(
time=time,
name='incomes',
energy_unit=energy_unit,
energy_incomes=energy_incomes,
grant_value=grant_incomes,
grant_spread_duration=
grant_spread_duration)
self.costs = EnergyUnitCosts(
time=time,
name='costs',
energy_unit=energy_unit,
energy_cost=energy_cost,
operation_starting_cost=operation_starting_cost,
operation_switch_off_cost=operation_switch_off_cost,
operation_fixed_cost=operation_fixed_cost,
maintenance_fixed_cost=maintenance_fixed_cost,
maintenance_investment_rate_cost=maintenance_investment_rate_cost,
investment_value=investment_cost,
investment_spread_duration=investment_spread_duration)
[docs]
class Transaction(EconomicObject):
def __init__(self, time, name, beneficiary, payer, fixed_cashflow,
spread_cashflow_duration,
energy_price, energy_unit):
EconomicObject.__init__(self, time=time, name=name,
unit1=beneficiary, unit2=payer)
self.description = 'Transactions'
self.name = name
if beneficiary == payer:
raise AttributeError("The payer {0} cannot be also the "
"beneficiary in the transaction {1}".format(
payer.name, name))
self.beneficiary = beneficiary
self.payer = payer
if energy_price is not None:
ep_name = '{}_energy_price'.format(
self.name)
self.add_quantity_dependant_cashflow(time=time, name=ep_name,
description='energy price',
value=energy_price,
quantity_name='p',
quantity_unit=energy_unit,
unit1=beneficiary,
unit2=payer)
self.add_income_to_incomes_exp(unit=beneficiary,
income_name=ep_name)
self.add_cost_to_costs_exp(unit=beneficiary,
cost_name=ep_name)
if fixed_cashflow is not None:
if spread_cashflow_duration is not None:
sc_name = '{}_spread_cashflow'.format(self.name)
self.add_spread_cashflow(time=time, name=name,
description='spread fixed cashflow',
cashflow_value=fixed_cashflow,
spread_duration=spread_cashflow_duration,
unit1=beneficiary, unit2=payer)
self.add_income_to_incomes_exp(unit=beneficiary,
income_name=sc_name)
self.add_cost_to_costs_exp(unit=beneficiary,
cost_name=sc_name)
else:
sfc_name = '{}_fixed_cashflow'.format(
self.name)
self.add_cashflow(time=time, name=sfc_name,
description='basic fixed cashflow',
cashflow_value=fixed_cashflow,
unit1=beneficiary, unit2=payer)
self.add_income_to_incomes_exp(unit=beneficiary,
income_name=sfc_name)
self.add_cost_to_costs_exp(unit=beneficiary,
cost_name=sfc_name)
[docs]
class EnergyUnitTransaction(Transaction, EnergyUnitCosts, EnergyUnitIncomes):
"""
**Description**
A transaction based on a subscription and on the price of the good
at each step
**Parameters**
"""
def __init__(self, time, name, beneficiary, payer,
fixed_cashflow=None, spread_cashflow_duration=None,
energy_price=None, energy_unit=None):
Transaction.__init__(self, time=time, name=name,
beneficiary=beneficiary, payer=payer,
fixed_cashflow=fixed_cashflow,
spread_cashflow_duration=spread_cashflow_duration,
energy_price=energy_price, energy_unit=energy_unit
)
if not isinstance(beneficiary, EnergyUnit):
raise TypeError('The buyer should be an EnergyUnit')
if not isinstance(payer, EnergyUnit):
raise TypeError('The seller should be an EnergyUnit')