Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from times tau where p_store(tau) > 0 **and** p_dispatch(tau) > 0, which
is allowed (even though not economic). Therefor p_store is necessarily
equal to negative entries of p, vice versa for p_dispatch.
"""
sus = n.storage_units
sus_i = sus.index
if sus_i.empty: return
sns = n.snapshots
c = 'StorageUnit'
pnl = n.pnl(c)
description = {}
eh = expand_series(n.snapshot_weightings, sus_i)
stand_eff = expand_series(1-n.df(c).standing_loss, sns).T.pow(eh)
dispatch_eff = expand_series(n.df(c).efficiency_dispatch, sns).T
store_eff = expand_series(n.df(c).efficiency_store, sns).T
inflow = get_as_dense(n, c, 'inflow') * eh
spill = eh[pnl.spill.columns] * pnl.spill
description['Spillage Limit'] = pd.Series({'min':
(inflow[spill.columns] - spill).min().min()})
if 'p_store' in pnl:
soc = pnl.state_of_charge
store = store_eff * eh * pnl.p_store#.clip(upper=0)
dispatch = 1/dispatch_eff * eh * pnl.p_dispatch#(lower=0)
start = soc.iloc[-1].where(sus.cyclic_state_of_charge,
sus.state_of_charge_initial)
previous_soc = stand_eff * soc.shift().fillna(start)
def describe_store_contraints(n):
"""
Checks whether all stores are balanced over time.
"""
stores = n.stores
stores_i = stores.index
if stores_i.empty: return
sns = n.snapshots
c = 'Store'
pnl = n.pnl(c)
eh = expand_series(n.snapshot_weightings, stores_i)
stand_eff = expand_series(1-n.df(c).standing_loss, sns).T.pow(eh)
start = pnl.e.iloc[-1].where(stores.e_cyclic, stores.e_initial)
previous_e = stand_eff * pnl.e.shift().fillna(start)
return (previous_e - pnl.p - pnl.e).unstack().describe()\
.to_frame('SOC Balance Store')
"""
Defines state of charge (soc) constraints for storage units. In principal
the constraints states:
previous_soc + p_store - p_dispatch + inflow - spill == soc
"""
sus_i = n.storage_units.index
if sus_i.empty: return
c = 'StorageUnit'
# spillage
upper = get_as_dense(n, c, 'inflow', sns).loc[:, lambda df: df.max() > 0]
spill = write_bound(n, 0, upper)
set_varref(n, spill, 'StorageUnit', 'spill')
eh = expand_series(n.snapshot_weightings[sns], sus_i) #elapsed hours
eff_stand = expand_series(1-n.df(c).standing_loss, sns).T.pow(eh)
eff_dispatch = expand_series(n.df(c).efficiency_dispatch, sns).T
eff_store = expand_series(n.df(c).efficiency_store, sns).T
soc = get_var(n, c, 'state_of_charge')
cyclic_i = n.df(c).query('cyclic_state_of_charge').index
noncyclic_i = n.df(c).query('~cyclic_state_of_charge').index
prev_soc_cyclic = soc.shift().fillna(soc.loc[sns[-1]])
coeff_var = [(-1, soc),
(-1/eff_dispatch * eh, get_var(n, c, 'p_dispatch')),
(eff_store * eh, get_var(n, c, 'p_store'))]
lhs, *axes = linexpr(*coeff_var, return_axes=True)
previous_soc + p_store - p_dispatch + inflow - spill == soc
"""
sus_i = n.storage_units.index
if sus_i.empty: return
c = 'StorageUnit'
# spillage
upper = get_as_dense(n, c, 'inflow', sns).loc[:, lambda df: df.max() > 0]
spill = write_bound(n, 0, upper)
set_varref(n, spill, 'StorageUnit', 'spill')
eh = expand_series(n.snapshot_weightings[sns], sus_i) #elapsed hours
eff_stand = expand_series(1-n.df(c).standing_loss, sns).T.pow(eh)
eff_dispatch = expand_series(n.df(c).efficiency_dispatch, sns).T
eff_store = expand_series(n.df(c).efficiency_store, sns).T
soc = get_var(n, c, 'state_of_charge')
cyclic_i = n.df(c).query('cyclic_state_of_charge').index
noncyclic_i = n.df(c).query('~cyclic_state_of_charge').index
prev_soc_cyclic = soc.shift().fillna(soc.loc[sns[-1]])
coeff_var = [(-1, soc),
(-1/eff_dispatch * eh, get_var(n, c, 'p_dispatch')),
(eff_store * eh, get_var(n, c, 'p_store'))]
lhs, *axes = linexpr(*coeff_var, return_axes=True)
def masked_term(coeff, var, cols):
return linexpr((coeff[cols], var[cols]))\
.reindex(index=axes[0], columns=axes[1], fill_value='').values
is allowed (even though not economic). Therefor p_store is necessarily
equal to negative entries of p, vice versa for p_dispatch.
"""
sus = n.storage_units
sus_i = sus.index
if sus_i.empty: return
sns = n.snapshots
c = 'StorageUnit'
pnl = n.pnl(c)
description = {}
eh = expand_series(n.snapshot_weightings, sus_i)
stand_eff = expand_series(1-n.df(c).standing_loss, sns).T.pow(eh)
dispatch_eff = expand_series(n.df(c).efficiency_dispatch, sns).T
store_eff = expand_series(n.df(c).efficiency_store, sns).T
inflow = get_as_dense(n, c, 'inflow') * eh
spill = eh[pnl.spill.columns] * pnl.spill
description['Spillage Limit'] = pd.Series({'min':
(inflow[spill.columns] - spill).min().min()})
if 'p_store' in pnl:
soc = pnl.state_of_charge
store = store_eff * eh * pnl.p_store#.clip(upper=0)
dispatch = 1/dispatch_eff * eh * pnl.p_dispatch#(lower=0)
start = soc.iloc[-1].where(sus.cyclic_state_of_charge,
sus.state_of_charge_initial)
previous_soc = stand_eff * soc.shift().fillna(start)
def define_store_constraints(n, sns):
"""
Defines energy balance constraints for stores. In principal this states:
previous_e - p == e
"""
stores_i = n.stores.index
if stores_i.empty: return
c = 'Store'
variables = write_bound(n, -np.inf, np.inf, axes=[sns, stores_i])
set_varref(n, variables, c, 'p')
eh = expand_series(n.snapshot_weightings[sns], stores_i) #elapsed hours
eff_stand = expand_series(1-n.df(c).standing_loss, sns).T.pow(eh)
e = get_var(n, c, 'e')
cyclic_i = n.df(c).query('e_cyclic').index
noncyclic_i = n.df(c).query('~e_cyclic').index
previous_e_cyclic = e.shift().fillna(e.loc[sns[-1]])
coeff_var = [(-eh, get_var(n, c, 'p')), (-1, e)]
lhs, *axes = linexpr(*coeff_var, return_axes=True)
def masked_term(coeff, var, cols):
return linexpr((coeff[cols], var[cols]))\
.reindex(index=axes[0], columns=axes[1], fill_value='').values
requires the network to contain the separate variables p_store and
p_dispatch, since they cannot be reconstructed from p. The latter results
from times tau where p_store(tau) > 0 **and** p_dispatch(tau) > 0, which
is allowed (even though not economic). Therefor p_store is necessarily
equal to negative entries of p, vice versa for p_dispatch.
"""
sus = n.storage_units
sus_i = sus.index
if sus_i.empty: return
sns = n.snapshots
c = 'StorageUnit'
pnl = n.pnl(c)
description = {}
eh = expand_series(n.snapshot_weightings, sus_i)
stand_eff = expand_series(1-n.df(c).standing_loss, sns).T.pow(eh)
dispatch_eff = expand_series(n.df(c).efficiency_dispatch, sns).T
store_eff = expand_series(n.df(c).efficiency_store, sns).T
inflow = get_as_dense(n, c, 'inflow') * eh
spill = eh[pnl.spill.columns] * pnl.spill
description['Spillage Limit'] = pd.Series({'min':
(inflow[spill.columns] - spill).min().min()})
if 'p_store' in pnl:
soc = pnl.state_of_charge
store = store_eff * eh * pnl.p_store#.clip(upper=0)
dispatch = 1/dispatch_eff * eh * pnl.p_dispatch#(lower=0)
start = soc.iloc[-1].where(sus.cyclic_state_of_charge,
sus.state_of_charge_initial)