Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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
lhs += masked_term(eff_stand, previous_e_cyclic, cyclic_i)
lhs += masked_term(eff_stand.loc[sns[1:]], e.shift().loc[sns[1:]], noncyclic_i)
rhs = pd.DataFrame(0, sns, stores_i)
rhs.loc[sns[0], noncyclic_i] -= n.df(c)['e_initial'][noncyclic_i]
define_constraints(n, lhs, '==', rhs, c, 'mu_state_of_charge')
status = get_var(n, c, 'status').loc[sns[1:], gens_i]
status_prev = get_var(n, c, 'status').shift(1).loc[sns[1:], gens_i]
lhs = linexpr((1, p[gens_i]), (-1, p_prev[gens_i]),
(limit_start - limit_up, status_prev), (- limit_start, status))
define_constraints(n, lhs, '<=', 0, c, 'mu_ramp_limit_up', spec='com.')
# fix down
gens_i = rdown_i & fix_i
lhs = linexpr((1, p[gens_i]), (-1, p_prev[gens_i]))
rhs = n.df(c).loc[gens_i].eval('-1 * ramp_limit_down * p_nom')
define_constraints(n, lhs, '>=', rhs, c, 'mu_ramp_limit_down', spec='nonext.')
# ext down
gens_i = rdown_i & ext_i
limit_pu = n.df(c)['ramp_limit_down'][gens_i]
p_nom = get_var(n, c, 'p_nom')[gens_i]
lhs = linexpr((1, p[gens_i]), (-1, p_prev[gens_i]), (limit_pu, p_nom))
define_constraints(n, lhs, '>=', 0, c, 'mu_ramp_limit_down', spec='ext.')
# com down
gens_i = rdown_i & com_i
if not gens_i.empty:
limit_shut = n.df(c).loc[gens_i].eval('ramp_limit_shut_down * p_nom')
limit_down = n.df(c).loc[gens_i].eval('ramp_limit_down * p_nom')
status = get_var(n, c, 'status').loc[sns[1:], gens_i]
status_prev = get_var(n, c, 'status').shift(1).loc[sns[1:], gens_i]
lhs = linexpr((1, p[gens_i]), (-1, p_prev[gens_i]),
(limit_down - limit_shut, status), (limit_shut, status_prev))
define_constraints(n, lhs, '>=', 0, c, 'mu_ramp_limit_down', spec='com.')
def bus_injection(c, attr, groupcol='bus', sign=1):
# additional sign only necessary for branches in reverse direction
if 'sign' in n.df(c):
sign = sign * n.df(c).sign
expr = linexpr((sign, get_var(n, c, attr))).rename(columns=n.df(c)[groupcol])
# drop empty bus2, bus3 if multiline link
if c == 'Link':
expr.drop(columns='', errors='ignore', inplace=True)
return expr
def map_solution(c, attr):
variables = get_var(n, c, attr, pop=pop)
predefined = True
if (c, attr) not in lookup.index:
predefined = False
n.sols[c] = n.sols[c] if c in n.sols else Dict(df=pd.DataFrame(), pnl={})
n.solutions.at[(c, attr), 'in_comp'] = predefined
if isinstance(variables, pd.DataFrame):
# case that variables are timedependent
n.solutions.at[(c, attr), 'pnl'] = True
pnl = n.pnl(c) if predefined else n.sols[c].pnl
values = variables.stack().map(variables_sol).unstack()
if c in n.passive_branch_components:
set_from_frame(pnl, 'p0', values)
set_from_frame(pnl, 'p1', - values)
elif c == 'Link':
set_from_frame(pnl, 'p0', values)
for i in ['1'] + additional_linkports(n):
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
if ('StorageUnit', 'spill') in n.variables.index:
lhs += masked_term(-eh, get_var(n, c, 'spill'), spill.columns)
lhs += masked_term(eff_stand, prev_soc_cyclic, cyclic_i)
lhs += masked_term(eff_stand.loc[sns[1:]], soc.shift().loc[sns[1:]], noncyclic_i)
rhs = -get_as_dense(n, c, 'inflow', sns).mul(eh)
rhs.loc[sns[0], noncyclic_i] -= n.df(c).state_of_charge_initial[noncyclic_i]
define_constraints(n, lhs, '==', rhs, c, 'mu_state_of_charge')
def define_committable_generator_constraints(n, snapshots):
c, attr = 'Generator', 'status'
com_i = n.df(c).query('committable and not p_nom_extendable').index
if com_i.empty: return
nominal = n.df(c)[nominal_attrs[c]][com_i]
min_pu, max_pu = get_bounds_pu(n, c, snapshots, com_i, 'p')
lower = min_pu.mul(nominal)
upper = max_pu.mul(nominal)
status = get_var(n, c, attr)
p = get_var(n, c, 'p')[com_i]
lhs = linexpr((lower, status), (-1, p))
define_constraints(n, lhs, '<=', 0, 'Generators', 'committable_lb')
lhs = linexpr((upper, status), (-1, p))
define_constraints(n, lhs, '>=', 0, 'Generators', 'committable_ub')