import logging
import inspect
import warnings
from collections import OrderedDict
import numpy as np
from andes.core.service import BackRef
from andes.shared import pd
from andes.utils.func import list_flatten, validate_keys_values
logger = logging.getLogger(__name__)
[docs]class GroupBase:
"""
Base class for groups.
"""
[docs] def __init__(self):
self.common_params = ['u', 'name']
self.common_vars = []
self.models = OrderedDict() # model name: model instance
self._idx2model = OrderedDict() # element idx: model instance
self.uid = {} # idx - group internal 0-indexed uid
self.services_ref = OrderedDict() # BackRef
def __setattr__(self, key, value):
if hasattr(value, 'owner'):
if value.owner is None:
value.owner = self
if hasattr(value, 'name'):
if value.name is None:
value.name = key
if isinstance(value, BackRef):
self.services_ref[key] = value
super().__setattr__(key, value)
@property
def class_name(self):
return self.__class__.__name__
@property
def n(self):
"""
Total number of devices.
"""
return len(self._idx2model)
[docs] def add_model(self, name: str, instance):
"""
Add a Model instance to group.
Parameters
----------
name : str
Model name
instance : Model
Model instance
Returns
-------
None
"""
if name not in self.models:
self.models[name] = instance
else:
raise KeyError(f"{self.class_name}: Duplicate model registration of {name}")
[docs] def add(self, idx, model):
"""
Register an idx from model_name to the group
Parameters
----------
idx: Union[str, float, int]
Register an element to a model
model: Model
instance of the model
Returns
-------
"""
if idx in self._idx2model:
raise KeyError(f'Group <{self.class_name}> already contains <{repr(idx)}> from '
f'<{self._idx2model[idx].class_name}>')
self.uid[idx] = self.n
self._idx2model[idx] = model
[docs] def idx2model(self, idx, allow_none=False):
"""
Find model name for the given idx.
Parameters
----------
idx : float, int, str, array-like
idx or idx-es of devices.
allow_none : bool
If True, return `None` at the positions where idx is not found.
Returns
-------
If `idx` is a list, return a list of model instances.
If `idx` is a single element, return a model instance.
"""
ret = []
idx, single = self._1d_vectorize(idx)
for i in idx:
try:
if i is None and allow_none:
ret.append(None)
else:
ret.append(self._idx2model[i])
except KeyError:
raise KeyError(f'Group <{self.class_name}> does not contain device with idx={i}')
if single:
ret = ret[0]
return ret
[docs] def idx2uid(self, idx):
"""
Convert idx to the 0-indexed unique index.
Parameters
----------
idx : array-like, numbers, or str
idx of devices
Returns
-------
list
A list containing the unique indices of the devices
"""
vec_idx, single = self._1d_vectorize(idx)
out = [self.uid[i] if i is not None else None for i in vec_idx]
if single:
out = out[0]
return out
[docs] def get(self, src: str, idx, attr: str = 'v', allow_none=False, default=0.0):
"""
Based on the indexer, get the `attr` field of the `src` parameter or variable.
Parameters
----------
src : str
param or var name
idx : array-like
device idx
attr
The attribute of the param or var to retrieve
allow_none : bool
True to allow None values in the indexer
default : float
If `allow_none` is true, the default value to use for None indexer.
Returns
-------
The requested param or variable attribute. If `idx` is a list, return a list of values.
If `idx` is a single element, return a single value.
"""
self._check_src(src)
self._check_idx(idx)
idx, single = self._1d_vectorize(idx)
n = len(idx)
if n == 0:
return np.zeros(0)
ret = [''] * n
_type_set = False
models = self.idx2model(idx, allow_none=allow_none)
for i, idx in enumerate(idx):
if models[i] is not None:
uid = models[i].idx2uid(idx)
instance = models[i].__dict__[src]
val = instance.__dict__[attr][uid]
else:
val = default
# deduce the type for ret
if not _type_set:
if isinstance(val, str):
ret = [''] * n
else:
ret = np.zeros(n)
_type_set = True
ret[i] = val
if single:
ret = ret[0]
return ret
# Sentinel for distinguishing "not passed" from None/0/False
_SENTINEL = object()
[docs] def set(self, src: str, idx, *args, value=_SENTINEL, attr='v', base=None):
"""
Set the value of a group property, dispatching to the correct model.
Delegates to each model's :meth:`Model.set`. See its docstring
for details on ``base`` and ``attr``.
Parameters
----------
src : str
Name of property.
idx : str, int, float, array-like
Indices of devices.
value : float or array-like
New values to be set. Can be passed as the third positional
argument or as a keyword.
attr : str, optional
Attribute to write (default ``'v'``). Ignored when
``base='device'``.
base : ``None`` or ``'device'``, optional
``None`` (default): system-base direct write.
``'device'``: device/input-base with per-unit conversion.
Returns
-------
bool
True when successful.
"""
# ---- Parse arguments (backward compat + new style) ----
_VALID_ATTRS = {'v', 'a', 'e', 'vin'}
_S = self._SENTINEL
if len(args) == 2 and isinstance(args[0], str) and args[0] in _VALID_ATTRS:
# Old-style positional: set('M', 1, 'v', 10.0)
warnings.warn(
"set(src, idx, attr, value) is deprecated. "
f"Use set('{src}', {idx!r}, {args[1]!r}, attr='{args[0]}') instead.",
FutureWarning, stacklevel=2
)
attr = args[0]
actual_value = args[1]
elif len(args) == 1 and value is _S:
actual_value = args[0]
elif len(args) == 0 and value is not _S:
actual_value = value
elif len(args) == 0 and value is _S:
raise TypeError(
f"set() missing 'value'. "
f"Usage: ss.<Group>.set('{src}', {idx!r}, <value>)"
)
else:
raise TypeError(
"set() got unexpected arguments. "
"Usage: set(src, idx, value, *, attr='v', base=None)"
)
self._check_src(src)
self._check_idx(idx)
idx, _ = self._1d_vectorize(idx)
models = self.idx2model(idx)
if isinstance(actual_value, (str, int, float, np.integer, np.floating)):
actual_value = [actual_value] * len(idx)
for mdl, ii, val in zip(models, idx, actual_value):
mdl.set(src, ii, val, attr=attr, base=base)
return True
[docs] def set_status(self, idx, value):
"""
Set the online status of a device and propagate to dependents.
Delegates to the concrete model's :meth:`set_status` which calls
:meth:`System.set_status` for propagation.
Parameters
----------
idx : str, int, float
Device idx.
value : int or float
New status value (0 or 1).
"""
mdl = self.idx2model(idx)
return mdl.set_status(idx, value)
[docs] def get_status(self, idx):
"""
Get the effective online status of a device.
Parameters
----------
idx : str, int, float
Device idx.
Returns
-------
float
Effective status value (0 or 1).
"""
mdl = self.idx2model(idx)
return mdl.get_status(idx)
[docs] def alter(self, src, idx, value, attr='v'):
"""
Alter values of input parameters or constant service for a group of models.
.. deprecated::
Use :meth:`set` with ``base='device'`` instead.
Parameters
----------
src : str
The parameter name to alter
idx : str, float, int
The unique identifier for the device to alter
value : float
The desired value
attr : str, optional
The attribute to alter. Default is 'v'.
"""
warnings.warn(
"alter() is deprecated. Use set() with base='device' instead.",
FutureWarning, stacklevel=2
)
self._check_src(src)
self._check_idx(idx)
idx, _ = self._1d_vectorize(idx)
models = self.idx2model(idx)
if isinstance(value, (str, int, float, np.integer, np.floating)):
value = [value] * len(idx)
for mdl, ii, val in zip(models, idx, value):
# Call Model.alter directly (which handles the attr='vin' legacy)
mdl.alter(src, ii, val, attr=attr)
return True
[docs] def find_idx(self, keys, values, allow_none=False, default=None, allow_all=False):
"""
Find indices of devices that satisfy the given `key=value` condition.
This method iterates over all models in this group.
Parameters
----------
keys : str, array-like, Sized
A string or an array-like of strings containing the names of parameters for the search criteria.
values : array, array of arrays, Sized
Values for the corresponding key to search for. If keys is a str, values should be an array of
elements. If keys is a list, values should be an array of arrays, each corresponding to the key.
allow_none : bool, optional
Allow key, value to be not found. Used by groups. Default is False.
default : bool, optional
Default idx to return if not found (missing). Default is None.
allow_all : bool, optional
Return all matches if set to True. Default is False.
Returns
-------
list
Indices of devices.
"""
keys, values = validate_keys_values(keys, values)
n_mdl, n_pair = len(self.models), len(values[0])
indices_found = []
# `indices_found` contains found indices returned from all models of this group
for model in self.models.values():
indices_found.append(model.find_idx(keys, values, allow_none=True, default=default, allow_all=True))
# --- find missing pairs ---
i_val_miss = []
for i in range(n_pair):
idx_cross_mdls = [indices_found[j][i] for j in range(n_mdl)]
if all(item == [default] for item in idx_cross_mdls):
i_val_miss.append(i)
if (not allow_none) and i_val_miss:
miss_pairs = []
for i in i_val_miss:
miss_pairs.append([values[j][i] for j in range(len(keys))])
raise IndexError(f'No {self.class_name} device found matching {list(keys)} = {miss_pairs}')
# --- output ---
out_pre = []
for i in range(n_pair):
idx_cross_mdls = [indices_found[j][i] for j in range(n_mdl)]
if all(item == [default] for item in idx_cross_mdls):
out_pre.append([default])
continue
for item in idx_cross_mdls:
if item != [default]:
out_pre.append(item)
break
if allow_all:
out = out_pre
else:
out = [item[0] for item in out_pre]
return out
def _check_src(self, src: str):
"""
Helper function for checking if ``src`` is a shared field.
The requirement is not strictly enforced and is only for debugging purposed.
"""
if src not in self.common_vars + self.common_params:
logger.debug(f'Group <{self.class_name}> does not share property <{src}>.')
def _check_idx(self, idx):
"""
Helper function for checking if ``idx`` is None.
Raises IndexError if idx is None.
"""
if idx is None:
raise IndexError(f'{self.class_name}: idx cannot be None')
def _1d_vectorize(self, idx):
"""
Helper function to convert a single element, list, or nested lists
into a list.
If the input is a nested list, flatten it into a 1-dimensional
list.
Returns
-------
idx : list
List of indices.
single : bool
True if the input is a single element.
"""
single = False
list_alike = (list, tuple, np.ndarray)
if not isinstance(idx, list_alike):
idx = [idx]
single = True
elif len(idx) > 0 and isinstance(idx[0], list_alike):
idx = list_flatten(idx)
return idx, single
[docs] def get_field(self, src: str, idx, field: str):
"""
Helper function for retrieving an attribute of a member variable shared
by models in this group.
Returns
-------
list
A list with the length equal to ``len(idx)``.
"""
self._check_src(src)
self._check_idx(idx)
idx, _ = self._1d_vectorize(idx)
models = self.idx2model(idx, allow_none=True)
ret = [None] * len(models)
for ii, model in enumerate(models):
if model is not None:
ret[ii] = getattr(model.__dict__[src], field)
return ret
def _find_controller(self, system, idx, controller_group, param_name):
"""
Find the controller device in ``controller_group`` whose ``param_name``
points to ``idx`` in this group.
Uses a lazily-built reverse lookup cache for O(1) repeated calls.
Parameters
----------
system : System
The system instance.
idx : str, int, float
Device idx in this group.
controller_group : str
Name of the controller group (e.g., ``'TurbineGov'``).
param_name : str
Name of the IdxParam on the controller that references this group
(e.g., ``'syn'``).
Returns
-------
tuple
``(model_instance, controller_idx)`` if found, ``(None, None)``
otherwise.
"""
if not hasattr(self, '_ctrl_cache'):
self._ctrl_cache = {}
cache_key = (controller_group, param_name)
if cache_key not in self._ctrl_cache:
lookup = {}
grp = system.groups.get(controller_group)
if grp is not None and grp.n > 0:
for mdl in grp.models.values():
param = mdl.__dict__.get(param_name)
if param is None:
continue
for uid, ctrl_idx in enumerate(mdl.idx.v):
target_idx = param.v[uid]
lookup[target_idx] = ctrl_idx
self._ctrl_cache[cache_key] = lookup
ctrl_idx = self._ctrl_cache[cache_key].get(idx)
if ctrl_idx is None:
return None, None
grp = system.groups[controller_group]
ctrl_mdl = grp.idx2model(ctrl_idx)
return ctrl_mdl, ctrl_idx
def _resolve_setpoint(self, system, idx, name):
"""
Resolve the controller chain for a setpoint and return the
target ``(model, device_idx)``.
Searches ``_setpoint_priority`` (defined on subclasses) for a
connected controller that declares ``name`` in its
``_setpoints``. Falls back to the device itself.
``_setpoint_priority`` is a class-level dict on group subclasses
that maps setpoint names to a list of ``(controller_group,
param_name)`` tuples to search, in order. ``param_name`` is the
IdxParam on the controller that references this group (e.g.,
``'syn'`` for TurbineGov/Exciter, ``'reg'`` for RenExciter).
These param names are enforced by each controller group's
``common_params``.
Example on SynGen::
_setpoint_priority = {
'pref': [('TurbineGov', 'syn')],
'vref': [('Exciter', 'syn')],
}
Parameters
----------
system : System
The system instance.
idx : str, int, float
Device idx in this group.
name : str
Setpoint key (e.g., ``'pref'``, ``'vref'``).
Returns
-------
tuple
``(model_instance, device_idx)``
Raises
------
KeyError
If neither a controller nor the device itself declares
the requested setpoint.
"""
priority = getattr(self, '_setpoint_priority', {})
for ctrl_group, ctrl_param in priority.get(name, []):
ctrl_mdl, ctrl_idx = self._find_controller(
system, idx, ctrl_group, ctrl_param
)
if ctrl_mdl is not None:
sp_map = getattr(ctrl_mdl, '_setpoints', {})
if name in sp_map:
return ctrl_mdl, ctrl_idx
# Fallback: the device itself
mdl = self.idx2model(idx)
sp_map = getattr(mdl, '_setpoints', {})
if name not in sp_map:
raise KeyError(
f"No controller found for {self.class_name} idx={idx!r} "
f"setpoint '{name}', and <{mdl.class_name}> does not "
f"declare '{name}' in _setpoints."
)
if priority.get(name):
logger.warning(
"No %s controller found for %s idx=%r. "
"Writing '%s' directly to <%s>.",
priority[name][0][0], self.class_name, idx,
name, mdl.class_name,
)
return mdl, idx
[docs] def set_setpoint(self, system, idx, name, value):
"""
Set a setpoint value by resolving the controller chain.
Searches for a connected controller (e.g., TurbineGov for
``'pref'``, Exciter for ``'vref'``) and writes there.
Falls back to the device itself with a warning.
.. note::
The physical meaning of a setpoint depends on the specific
controller model that owns it. For example, ``'vref'``
writes to the exciter's voltage reference input, which may
include compensation terms (e.g., EXAC4 initializes
``vref0 = v + vf0/KA``). It is **not** necessarily the
exact terminal voltage. Similarly, ``'pref'`` is the
governor's power reference in system-base per-unit. Use
:meth:`get_setpoint` to read the current value before
applying incremental changes.
Parameters
----------
system : System
The system instance.
idx : str, int, float
Device idx in this group.
name : str
Setpoint key (e.g., ``'pref'``, ``'vref'``, ``'qref'``).
value : float
Value to write (absolute, in system-base per-unit).
"""
mdl, dev_idx = self._resolve_setpoint(system, idx, name)
attr_name = mdl._setpoints[name]
uid = mdl.idx2uid(dev_idx)
mdl.__dict__[attr_name].v[uid] = value
[docs] def get_setpoint(self, system, idx, name):
"""
Get the current setpoint value by resolving the controller chain.
See :meth:`set_setpoint` for notes on the physical meaning of
setpoint values.
Parameters
----------
system : System
The system instance.
idx : str, int, float
Device idx in this group.
name : str
Setpoint key (e.g., ``'pref'``, ``'vref'``, ``'qref'``).
Returns
-------
float
Current value of the setpoint (system-base per-unit).
"""
mdl, dev_idx = self._resolve_setpoint(system, idx, name)
attr_name = mdl._setpoints[name]
uid = mdl.idx2uid(dev_idx)
return mdl.__dict__[attr_name].v[uid]
[docs] def set_backref(self, name, from_idx, to_idx):
"""
Set idxes to ``BackRef``, and set them to models.
"""
uid = self.idx2uid(to_idx)
self.services_ref[name].v[uid].append(from_idx)
model = self.idx2model(to_idx)
model.set_backref(name, from_idx, to_idx)
[docs] def get_next_idx(self, idx=None, model_name=None):
"""
Get a no-conflict idx for a new device.
Use the provided ``idx`` if no conflict.
Generate a new one otherwise.
Parameters
----------
idx : str or None
Proposed idx. If None, assign a new one.
model_name : str or None
Model name. If not, prepend the group name.
Returns
-------
str
New device name.
"""
if model_name is None:
model_name = self.class_name
need_new = False
if idx is not None:
if idx not in self._idx2model:
# name is good
pass
else:
logger.warning("Group <%s>: idx=%s is used by %s. Data may be inconsistent.",
self.class_name, idx, self.idx2model(idx).class_name)
need_new = True
else:
need_new = True
if need_new is True:
count = self.n
while True:
# IMPORTANT: automatically assigned index is 1-indexed. Namely, `GENCLS_1` is the first generator.
# This is because when we say, for example, `GENCLS_10`, people usually assume it starts at 1.
idx = model_name + '_' + str(count + 1)
if idx not in self._idx2model:
break
else:
count += 1
return idx
[docs] def get_all_idxes(self):
"""
Return all the devices idx in this group.
.. note::
New in version 1.9.3.
Returns
-------
list
List of indices.
Notes
-----
The default models sequence depends on the order of the models in the group,
which comes from OrderedDict `file_classes` in `models.__init__.py`.
Examples
--------
>>> ss = andes.load(andes.get_case('ieee14/ieee14_pvd1.xlsx'))
>>> ss.DG.get_all_idxes()
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
>>> ss.StaticGen.get_all_idxes()
[2, 3, 4, 5, 6, 1]
"""
return list(self._idx2model.keys())
[docs] def as_dict(self, vin=False):
"""
Export group common parameters as a dictionary.
.. note::
New in version 1.9.3.
This method returns a dictionary where the keys are the `ModelData` parameter names
and the values are array-like structures containing the data in the order they were added.
Unlike `ModelData.as_dict()`, this dictionary does not include the `uid` field.
Parameters
----------
vin : bool, optional
If True, includes the `vin` attribute in the dictionary. Default is False.
Returns
-------
dict
A dictionary of common parameters.
"""
out_all = []
out_params = self.common_params.copy()
out_params.insert(2, 'idx')
for mdl in self.models.values():
if mdl.n <= 0:
continue
mdl_data = mdl.as_df(vin=True) if vin else mdl.as_dict()
mdl_dict = {k: mdl_data.get(k) for k in out_params if k in mdl_data}
out_all.append(mdl_dict)
if not out_all:
return {}
out = {key: np.concatenate([item[key] for item in out_all]) for key in out_all[0].keys()}
return out
[docs] def as_df(self, vin=False):
"""
Export group common parameters as a `pandas.DataFrame` object.
.. note::
New in version 1.9.3.
Parameters
----------
vin : bool
If True, export all parameters from original input (``vin``).
Returns
-------
DataFrame
A dataframe containing all model data. An `uid` column is added.
"""
return pd.DataFrame(self.as_dict(vin=vin))
[docs] def doc(self, export='plain'):
"""
Return the documentation of the group in a string.
"""
out = ''
if export == 'rest':
out += f'.. _{self.class_name}:\n\n'
group_header = '=' * 80 + '\n'
else:
group_header = ''
if export == 'rest':
out += group_header + f'{self.class_name}\n' + group_header
else:
out += group_header + f'Group <{self.class_name}>\n' + group_header
if self.__doc__ is not None:
out += inspect.cleandoc(self.__doc__) + '\n\n'
if len(self.common_params):
out += 'Common Parameters: ' + ', '.join(self.common_params)
out += '\n\n'
if len(self.common_vars):
out += 'Common Variables: ' + ', '.join(self.common_vars)
out += '\n\n'
if len(self.models):
out += 'Available models:\n'
model_name_list = list(self.models.keys())
if export == 'rest':
def add_reference(name_list):
return [f'{item}_' for item in name_list]
model_name_list = add_reference(model_name_list)
out += ',\n'.join(model_name_list) + '\n'
return out
[docs] def doc_all(self, export='plain'):
"""
Return documentation of the group and its models.
Parameters
----------
export : 'plain' or 'rest'
Export format, plain-text or RestructuredText
Returns
-------
str
"""
out = self.doc(export=export)
out += '\n'
for instance in self.models.values():
out += instance.doc(export=export)
out += '\n'
return out
class Undefined(GroupBase):
"""
The undefined group. Holds models with no ``group``.
"""
pass
class ACNode(GroupBase):
def __init__(self):
super().__init__()
self.common_vars.extend(('a', 'v'))
class DCTopology(GroupBase):
def __init__(self):
super().__init__()
self.common_vars.extend(('v',))
class Collection(GroupBase):
"""Collection of topology models"""
pass
class Calculation(GroupBase):
"""Group of classes that calculates based on other models."""
pass
class StaticGen(GroupBase):
"""
Static generator group.
Static generators include PV and Slack, which are used to impose algebraic
equations. Static generators are used primarily for power flow.
Static generators do not have the modeling details for stability
simulation. Although some of them can stay for time-domain simulation, most
of them should be substituted by dynamic generators, including synchronous
generators and inverter-based resources upon TDS initialization.
The substitution is done by setting the ``gen`` field of a dynamic generator
to refer to the static generator. To replace one StaticGen by multiple
dynamic generators, see the notes in :ref:`SynGen`. Generators connected to
the same bus need to have their `gammap` and `gammaq`, respectively, summed
up to be exactly 1.0.
TDS initialization will ensure that the dynamic generators impose the same
amount of power as the static generator. At the end of initialization,
`StaticGen`'s that have been substituted will have their connectivity status
``u`` changed to ``0``.
"""
def __init__(self):
super().__init__()
self.common_params.extend(('bus', 'Sn', 'Vn', 'p0', 'q0', 'ra', 'xs', 'subidx'))
self.common_vars.extend(('q', 'a', 'v'))
self.SynGen = BackRef()
class ACLine(GroupBase):
def __init__(self):
super(ACLine, self).__init__()
self.common_params.extend(('bus1', 'bus2', 'r', 'x'))
self.common_vars.extend(('v1', 'v2', 'a1', 'a2'))
class ACShort(GroupBase):
def __init__(self):
super(ACShort, self).__init__()
self.common_params.extend(('bus1', 'bus2'))
self.common_vars.extend(('v1', 'v2', 'a1', 'a2'))
class StaticLoad(GroupBase):
"""
Static load group.
"""
pass
class StaticShunt(GroupBase):
"""
Static shunt compensator group.
"""
pass
class DynLoad(GroupBase):
"""
Dynamic load group.
"""
pass
class SynGen(GroupBase):
"""
Synchronous generator group.
SynGen replaces StaticGen upon the initialization of dynamic studies. SynGen
and inverter-based resources contain parameters ``gammap`` and ``gammaq``
for splitting the initial power of a StaticGen into multiple dynamic ones.
``gammap``, for example, is the active power ratio of the dynamic generator
to the static one. If a StaticGen is supposed to be replaced by one SynGen,
the ``gammap`` and ``gammaq`` should both be ``1``.
It is critical to ensure that ``gammap`` and ``gammaq``, respectively, of
all dynamic power sources sum up to 1.0. Otherwise, the initial power
injections imposed by dynamic sources will differ from the static ones. The
initialization will then fail with mismatches power injection equations
corresponding to bus ``a`` and ``v``.
"""
_setpoint_priority = {
'pref': [('TurbineGov', 'syn')],
'vref': [('Exciter', 'syn')],
'paux': [('TurbineGov', 'syn')],
}
def __init__(self):
super().__init__()
self.common_params.extend(('bus', 'gen', 'Sn', 'Vn', 'fn', 'M', 'D', 'subidx'))
self.common_vars.extend(('omega', 'delta', ))
self.idx_island = []
self.uid_island = []
self.delta_addr = []
def set_pref(self, system, idx, value):
"""
Set active power reference for a synchronous generator.
Routes to the turbine governor's ``pref0`` if one is connected;
otherwise falls back to the generator's ``tm0`` with a warning.
The value is in system-base per-unit.
Use :meth:`get_pref` to read the current value before applying
incremental changes.
"""
self.set_setpoint(system, idx, 'pref', value)
def get_pref(self, system, idx):
"""
Get active power reference. See :meth:`set_pref` for semantics.
"""
return self.get_setpoint(system, idx, 'pref')
def set_vref(self, system, idx, value):
"""
Set voltage reference for a synchronous generator.
Routes to the exciter's ``vref0`` if one is connected;
otherwise falls back to the generator's ``vf0`` with a warning.
.. warning::
The exciter voltage reference is **not** necessarily the
terminal voltage. Its meaning depends on the exciter model.
Most exciters initialize ``vref0 = vref`` (the pre-fault
voltage error input), but some (e.g., EXAC4) use a
model-specific formula. Use :meth:`get_vref` to read the
current value before applying incremental changes.
"""
self.set_setpoint(system, idx, 'vref', value)
def get_vref(self, system, idx):
"""
Get voltage reference. See :meth:`set_vref` for semantics.
"""
return self.get_setpoint(system, idx, 'vref')
def set_paux(self, system, idx, value):
"""
Set auxiliary power input for a synchronous generator.
Routes to the turbine governor's ``paux0``. Raises
``KeyError`` if no governor is connected (generators do not
have an auxiliary power input on their own).
The auxiliary signal is additive to the power reference inside
the governor equations. The value is in system-base per-unit.
Commonly used for AGC signals or reinforcement-learning inputs.
"""
self.set_setpoint(system, idx, 'paux', value)
def get_paux(self, system, idx):
"""
Get auxiliary power input. See :meth:`set_paux` for semantics.
"""
return self.get_setpoint(system, idx, 'paux')
def store_idx_island(self, bus_idx):
"""
Get ``idx`` of generators in the given islanded. Also store the
addresses of the ``delta`` variable in the largest island.
This function can only be called after initializing dynamic devices.
Parameters
----------
bus_idx : list
A list of bus idx in the largest island
"""
idx_gen = list(self.uid.keys())
bus_gen = self.get(src='bus', idx=idx_gen, attr='v')
def intersect(lst1, lst2):
return list(set(lst1) & set(lst2))
bus_gen_island = intersect(bus_idx, bus_gen)
self.idx_island = self.find_idx(keys='bus',
values=bus_gen_island)
self.uid_island = self.idx2uid(self.idx_island)
self.delta_addr = self.get('delta', self.idx_island, 'a').astype(int)
class RenGen(GroupBase):
"""
Renewable generator (converter) group.
See :ref:`SynGen` for the notes on replacing StaticGen and setting the power
ratio parameters.
Attention is needed for the power base ``Sn``. When replacing a synchronous
generator, the renewable generator should have the same or larger ``Sn``.
Improper ``Sn`` will cause the initial values to exceed typical per-unit
ranges and cause the initialization to fail.
"""
_setpoint_priority = {
'pref': [('RenExciter', 'reg')],
'qref': [('RenExciter', 'reg')],
}
def __init__(self):
super().__init__()
self.common_params.extend(('bus', 'gen', 'Sn'))
self.common_vars.extend(('Pe', 'Qe'))
def set_pref(self, system, idx, value):
"""
Set active power reference for a renewable generator.
Routes to the renewable exciter's ``Pref0`` (e.g., REECA1,
REECB1) if one is connected; otherwise falls back to the
converter model with a warning.
The value is in system-base per-unit.
For REECA1 with ``PFLAG=1``, the internal equation normalizes
by generator speed (``wg``), so the setpoint represents
electrical power, not mechanical power.
Use :meth:`get_pref` to read the current value before applying
incremental changes.
"""
self.set_setpoint(system, idx, 'pref', value)
def get_pref(self, system, idx):
"""
Get active power reference. See :meth:`set_pref` for semantics.
"""
return self.get_setpoint(system, idx, 'pref')
def set_qref(self, system, idx, value):
"""
Set reactive power reference for a renewable generator.
Routes to the renewable exciter's ``qref0`` (e.g., REECA1,
REECB1) if one is connected; otherwise falls back to the
converter model with a warning.
The value is in system-base per-unit.
The interpretation depends on the exciter's reactive power
control mode (voltage regulation vs. constant Q).
Use :meth:`get_qref` to read the current value before applying
incremental changes.
"""
self.set_setpoint(system, idx, 'qref', value)
def get_qref(self, system, idx):
"""
Get reactive power reference. See :meth:`set_qref` for semantics.
"""
return self.get_setpoint(system, idx, 'qref')
class RenExciter(GroupBase):
"""
Renewable electrical control (exciter) group.
"""
def __init__(self):
super().__init__()
self.common_params.extend(('reg',))
self.common_vars.extend(('Pref', 'Qref', 'wg', 'Pord'))
class RenPlant(GroupBase):
"""
Renewable plant control group.
"""
def __init__(self):
super().__init__()
self.common_params.extend(('ree',))
class RenGovernor(GroupBase):
"""
Renewable turbine governor group.
"""
def __init__(self):
super().__init__()
self.common_params.extend(('ree', 'w0', 'Sn', 'Pe0'))
self.common_vars.extend(('Pm', 'wr0', 'wt', 'wg', 's3_y'))
class RenAerodynamics(GroupBase):
"""
Renewable aerodynamics group.
"""
def __init__(self):
super().__init__()
self.common_params.extend(('rego',))
self.common_vars.extend(('theta',))
class RenPitch(GroupBase):
"""
Renewable generator pitch controller group.
"""
def __init__(self):
super().__init__()
self.common_params.extend(('rea',))
class RenTorque(GroupBase):
"""
Renewable torque (Pref) controller.
"""
def __init__(self):
super().__init__()
class DG(GroupBase):
"""
Distributed generation (small-scale).
See :ref:`SynGen` for the notes on replacing StaticGen and setting the power
ratio parameters.
"""
# No external controller chain; setpoints live on the device itself.
_setpoint_priority = {}
def __init__(self):
super().__init__()
self.common_params.extend(('bus', 'fn'))
def set_pref(self, system, idx, value):
"""
Set active power reference for a distributed generator.
Writes to the device's ``pref0`` directly.
The value is in system-base per-unit.
Use :meth:`get_pref` to read the current value before applying
incremental changes.
"""
self.set_setpoint(system, idx, 'pref', value)
def get_pref(self, system, idx):
"""
Get active power reference. See :meth:`set_pref` for semantics.
"""
return self.get_setpoint(system, idx, 'pref')
def set_qref(self, system, idx, value):
"""
Set reactive power reference for a distributed generator.
Writes to the device's ``qref0`` directly.
The value is in system-base per-unit.
Use :meth:`get_qref` to read the current value before applying
incremental changes.
"""
self.set_setpoint(system, idx, 'qref', value)
def get_qref(self, system, idx):
"""
Get reactive power reference. See :meth:`set_qref` for semantics.
"""
return self.get_setpoint(system, idx, 'qref')
def set_paux(self, system, idx, value):
"""
Set auxiliary power input for a distributed generator.
Writes to the device's ``Pext0`` (external power signal).
The auxiliary signal is additive to the power reference inside
the model equations. Commonly used for AGC signals.
The value is in system-base per-unit.
"""
self.set_setpoint(system, idx, 'paux', value)
def get_paux(self, system, idx):
"""
Get auxiliary power input. See :meth:`set_paux` for semantics.
"""
return self.get_setpoint(system, idx, 'paux')
class DGProtection(GroupBase):
"""
Protection model for DG.
"""
def __init__(self):
super().__init__()
class TurbineGov(GroupBase):
"""
Turbine governor group for synchronous generator.
"""
def __init__(self):
super().__init__()
self.common_params.extend(('syn',))
self.common_vars.extend(('pout',))
class DynShaft(GroupBase):
"""
Dynamic shaft model group for multi-mass torsional models.
"""
def __init__(self):
super().__init__()
self.common_params.extend(('syn',))
class Exciter(GroupBase):
"""
Exciter group for synchronous generators.
"""
def __init__(self):
super().__init__()
self.common_params.extend(('syn',))
self.common_vars.extend(('vout', 'vi',))
class VoltComp(GroupBase):
"""
Voltage compensator group for synchronous generators.
"""
def __init__(self):
super().__init__()
self.common_params.extend(('avr', 'rc', 'xc',))
self.common_vars.extend(('vcomp',))
class PSS(GroupBase):
"""Power system stabilizer group."""
def __init__(self):
super().__init__()
self.common_params.extend(('avr',))
self.common_vars.extend(('vsout',))
class Experimental(GroupBase):
"""Experimental group"""
pass
class DCLink(GroupBase):
"""Basic DC links"""
pass
class StaticACDC(GroupBase):
"""AC DC device for power flow"""
pass
class TimedEvent(GroupBase):
"""Timed event group"""
pass
class FreqMeasurement(GroupBase):
"""Frequency measurements."""
def __init__(self):
super().__init__()
self.common_params.extend(('bus',))
self.common_vars.extend(('f',))
class PhasorMeasurement(GroupBase):
"""Phasor measurements"""
def __init__(self):
super().__init__()
self.common_params.extend(('bus',))
self.common_vars.extend(('am', 'vm'))
class PLL(GroupBase):
"""Phase-locked loop models."""
def __init__(self):
super().__init__()
self.common_params.extend(('bus',))
self.common_vars.extend(('am',))
class Motor(GroupBase):
"""Induction Motor group
"""
def __init__(self):
super().__init__()
self.common_params.extend(('bus',))
class Information(GroupBase):
"""
Group for information container models.
"""
def __init__(self):
GroupBase.__init__(self)
self.common_params = []
class OutputSelect(GroupBase):
"""
Group for selecting outputs.
"""
def __init__(self):
super().__init__()
self.common_params = []
class Interface(GroupBase):
"""
Group for interface models.
"""
def __init__(self):
super().__init__()
self.common_params = []
class DataSeries(GroupBase):
"""
Group for TimeSeries models.
"""
def __init__(self):
super().__init__()
self.common_params = []
