"""
Module for system.
"""
import configparser
import importlib
import inspect
import logging
import os
from collections import OrderedDict
from typing import Dict, Optional
import numpy as np
from ams.utils.fileman import FileMan
from ams.utils.misc import elapsed
from ams.utils.paths import _config_numpy, confirm_overwrite, load_config_rc
from ams.utils.tab import Tab
import ams
from ams.models.group import GroupBase
from ams.routines.typecls import TypeBase
from ams.models import file_classes
from ams.routines import all_routines
from ams.utils.paths import get_config_path
from ams.core import Config
from ams.core.matprocessor import MatProcessor
from ams.interface import to_andes
from ams.report import Report
from ams.shared import _load_andes_catalog
from ams.io.matpower import system2mpc
from ams.io.matpower import write as write_m
from ams.io.xlsx import write as write_xlsx
from ams.io.json import write as write_json
from ams.io.psse import write_raw
logger = logging.getLogger(__name__)
# Keys that were valid in older AMS releases but have since been removed
# from System.config. Purged from any loaded rc file so users' persisted
# ~/.ams/ams.rc don't trip `Config.check()` warnings on every load. Add
# a new entry here whenever a System.config key is retired.
_DEPRECATED_SYSTEM_CONFIG_KEYS: frozenset[str] = frozenset({
"save_stats", # retired in PR #213 (2026-04-21), runtime-stub retirement
})
[docs]
class System:
"""
Base system class, revised from `andes.system.System`.
This class encapsulates data, models, and routines for scheduling
modeling and analysis in power systems.
Parameters
----------
case : str, optional
The path to the case file.
name : str, optional
Name of the system instance.
config : dict, optional
Configuration options for the system. Overrides the default configuration if provided.
config_path : str, optional
The path to the configuration file.
default_config : bool, optional
If True, the default configuration file is loaded.
options : dict, optional
Additional configuration options for the system.
**kwargs :
Additional configuration options passed as keyword arguments.
Attributes
----------
name : str
Name of the system instance.
options : dict
A dictionary containing configuration options for the system.
models : OrderedDict
An ordered dictionary holding the model names and instances.
model_aliases : OrderedDict
An ordered dictionary holding model aliases and their corresponding instances.
groups : OrderedDict
An ordered dictionary holding group names and instances.
routines : OrderedDict
An ordered dictionary holding routine names and instances.
types : OrderedDict
An ordered dictionary holding type names and instances.
mats : MatrixProcessor, None
A matrix processor instance, initially set to None.
mat : OrderedDict
An ordered dictionary holding common matrices.
exit_code : int
Command-line exit code. 0 indicates normal execution, while other values indicate errors.
recent : RecentSolvedRoutines, None
An object storing recently solved routines, initially set to None.
dyn : ANDES System, None
linked dynamic system, initially set to None.
It is an instance of the ANDES system, which will be automatically
set when using ``System.to_andes()``.
files : FileMan
File path manager instance.
is_setup : bool
Internal flag indicating if the system has been set up.
Methods
-------
setup:
Set up the system.
to_andes:
Convert the system to an ANDES system.
"""
[docs]
def __init__(self,
case: Optional[str] = None,
name: Optional[str] = None,
config: Optional[Dict] = None,
config_path: Optional[str] = None,
default_config: Optional[bool] = False,
options: Optional[Dict] = None,
**kwargs
):
self.name = name
self.options = {}
if options is not None:
self.options.update(options)
if kwargs:
self.options.update(kwargs)
self.models = OrderedDict() # model names and instances
self.model_aliases = OrderedDict() # alias: model instance
self.groups = OrderedDict() # group names and instances
self.routines = OrderedDict() # routine names and instances
self.types = OrderedDict() # type names and instances
self.mats = MatProcessor(self) # matrix processor
# TODO: there should be an exit_code for each routine
self.exit_code = 0 # command-line exit code, 0 - normal, others - error.
self.recent = None # recent solved routines
self.dyn = None # ANDES system
# get and load default config file
self._config_path = get_config_path()
if config_path is not None:
self._config_path = config_path
if default_config is True:
self._config_path = None
self._config_object = load_config_rc(self._config_path)
self._update_config_object()
self.config = Config(self.__class__.__name__, dct=config)
self.config.load(self._config_object)
# Silently drop keys that were valid in older releases but have
# since been removed. A user's ~/.ams/ams.rc written against an
# older AMS will otherwise trip Config.check() warnings on every
# load. Purge both ``__dict__`` and Config's ``_dict`` backing
# store so ``as_dict()`` / ``save_config()`` can never re-persist
# retired keys. Add to ``_DEPRECATED_SYSTEM_CONFIG_KEYS`` whenever
# a System.config key is retired.
for _deprecated_key in _DEPRECATED_SYSTEM_CONFIG_KEYS:
self.config.__dict__.pop(_deprecated_key, None)
self.config._dict.pop(_deprecated_key, None)
# custom configuration for system goes after this line
self.config.add(OrderedDict((('freq', 60),
('mva', 100),
('seed', 'None'),
('np_divide', 'warn'),
('np_invalid', 'warn'),
)))
self.config.add_extra("_help",
freq='base frequency [Hz]',
mva='system base MVA',
seed='seed (or None) for random number generator',
np_divide='treatment for division by zero',
np_invalid='treatment for invalid floating-point ops.',
)
self.config.add_extra("_alt",
freq="float",
mva="float",
seed='int or None',
np_divide={'ignore', 'warn', 'raise', 'call', 'print', 'log'},
np_invalid={'ignore', 'warn', 'raise', 'call', 'print', 'log'},
)
self.config.check()
_config_numpy(seed=self.config.seed,
divide=self.config.np_divide,
invalid=self.config.np_invalid,
)
# TODO: revise the following attributes, it seems that these are not used in AMS
self._getters = dict(f=list(), g=list(), x=list(), y=list())
self._adders = dict(f=list(), g=list(), x=list(), y=list())
self._setters = dict(f=list(), g=list(), x=list(), y=list())
self.files = FileMan(case=case, **self.options) # file path manager
# internal flags
self.is_setup = False # if system has been setup
self.import_types()
self.import_groups()
self.import_models()
self.import_routines()
[docs]
def import_types(self):
"""
Import all types classes defined in ``routines/typecls.py``.
Types will be stored as instances with the name as class names.
All types will be stored to dictionary ``System.types``.
"""
module = importlib.import_module('ams.routines.typecls')
for m in inspect.getmembers(module, inspect.isclass):
name, cls = m
if name == 'TypeBase':
continue
elif not issubclass(cls, TypeBase):
# skip other imported classes such as `OrderedDict`
continue
self.__dict__[name] = cls()
self.types[name] = self.__dict__[name]
def _collect_group_data(self, items):
"""
Set the owner for routine attributes: `RParam`, `Var`, `ExpressionCalc`, `Expression`,
and `RBaseService`.
"""
for item_name, item in items.items():
if item.model is None:
continue
elif item.model in self.groups.keys():
item.is_group = True
item.owner = self.groups[item.model]
elif item.model in self.models.keys():
item.owner = self.models[item.model]
elif item.model == 'mats':
item.owner = self.mats
else:
logger.debug(f'item_name: {item_name}')
msg = f'Model indicator \'{item.model}\' of <{item.rtn.class_name}.{item_name}>'
msg += ' is not a model or group. Likely a modeling error.'
logger.warning(msg)
[docs]
def import_routines(self):
"""
Import routines as defined in ``routines/__init__.py``.
Routines will be stored as instances with the name as class names.
All routines will be stored to dictionary ``System.routines``.
Examples
--------
``System.PFlow`` is the power flow routine instance.
"""
for file, cls_list in all_routines.items():
for cls_name in cls_list:
routine = importlib.import_module('ams.routines.' + file)
the_class = getattr(routine, cls_name)
attr_name = cls_name
self.__dict__[attr_name] = the_class(system=self, config=self._config_object)
self.routines[attr_name] = self.__dict__[attr_name]
self.routines[attr_name].config.check()
# NOTE: the following code is not used in ANDES
for vname, rtn in self.routines.items():
# TODO: collect routiens into types
type_name = getattr(rtn, 'type')
type_instance = self.types[type_name]
type_instance.routines[vname] = rtn
# self.types[rtn.type].routines[vname] = rtn
# Collect RParams
rparams = getattr(rtn, 'rparams')
self._collect_group_data(rparams)
# Collect routine Vars
r_vars = getattr(rtn, 'vars')
self._collect_group_data(r_vars)
# Collect ExpressionCalcs
exprc = getattr(rtn, 'exprcs')
self._collect_group_data(exprc)
# Collect Expressions
expr = getattr(rtn, 'exprs')
self._collect_group_data(expr)
[docs]
def import_groups(self):
"""
Import all groups classes defined in ``models/group.py``.
Groups will be stored as instances with the name as class names.
All groups will be stored to dictionary ``System.groups``.
"""
module = importlib.import_module('ams.models.group')
for m in inspect.getmembers(module, inspect.isclass):
name, cls = m
if name == 'GroupBase':
continue
elif not issubclass(cls, GroupBase):
# skip other imported classes such as `OrderedDict`
continue
self.__dict__[name] = cls()
self.groups[name] = self.__dict__[name]
[docs]
def import_models(self):
"""
Import and instantiate models as System member attributes.
Models defined in ``models/__init__.py`` will be instantiated `sequentially` as attributes with the same
name as the class name.
In addition, all models will be stored in dictionary ``System.models`` with model names as
keys and the corresponding instances as values.
Examples
--------
``system.Bus`` stores the `Bus` object, and ``system.PV`` stores the PV generator object.
``system.models['Bus']`` points the same instance as ``system.Bus``.
"""
for fname, cls_list in file_classes:
for model_name in cls_list:
the_module = importlib.import_module('ams.models.' + fname)
the_class = getattr(the_module, model_name)
self.__dict__[model_name] = the_class(system=self, config=self._config_object)
self.models[model_name] = self.__dict__[model_name]
self.models[model_name].config.check()
# link to the group
group_name = self.__dict__[model_name].group
self.__dict__[group_name].add_model(model_name, self.__dict__[model_name])
# NOTE: model_aliases is not used in AMS currently
# for key, val in ams.models.model_aliases.items():
# self.model_aliases[key] = self.models[val]
# self.__dict__[key] = self.models[val]
[docs]
def collect_ref(self):
"""
Collect indices into `BackRef` for all models.
"""
models_and_groups = list(self.models.values()) + list(self.groups.values())
# create an empty list of lists for all `BackRef` instances
for model in models_and_groups:
for ref in model.services_ref.values():
ref.v = [list() for _ in range(model.n)]
# `model` is the model who stores `IdxParam`s to other models
# `BackRef` is declared at other models specified by the `model` parameter
# of `IdxParam`s.
for model in models_and_groups:
if model.n == 0:
continue
# skip: a group is not allowed to link to other groups
if not hasattr(model, "idx_params"):
continue
for idxp in model.idx_params.values():
if (idxp.model not in self.models) and (idxp.model not in self.groups):
continue
dest = self.__dict__[idxp.model]
if dest.n == 0:
continue
for name in (model.class_name, model.group):
# `BackRef` not requested by the linked models or groups
if name not in dest.services_ref:
continue
for model_idx, dest_idx in zip(model.idx.v, idxp.v):
if dest_idx not in dest.uid:
continue
dest.set_backref(name,
from_idx=model_idx,
to_idx=dest_idx)
[docs]
def reset(self, force=False):
"""
Reset to the state after reading data and setup.
"""
self.is_setup = False
self.setup()
[docs]
def add(self, model, param_dict=None, **kwargs):
"""
Add a device instance for an existing model.
Revised from ``andes.system.System.add``.
"""
if model not in self.models and (model not in self.model_aliases):
_, _, _, ad_dyn_models = _load_andes_catalog()
if model in ad_dyn_models:
logger.debug("ANDES dynamic model <%s> is skipped.", model)
else:
logger.warning("<%s> is not an existing model.", model)
return
if self.is_setup:
raise NotImplementedError("Adding devices are not allowed after setup.")
group_name = self.__dict__[model].group
group = self.groups[group_name]
if param_dict is None:
param_dict = {}
if kwargs is not None:
param_dict.update(kwargs)
# remove `uid` field
param_dict.pop('uid', None)
idx = param_dict.pop('idx', None)
if idx is not None and (not isinstance(idx, str) and np.isnan(idx)):
idx = None
idx = group.get_next_idx(idx=idx, model_name=model)
self.__dict__[model].add(idx=idx, **param_dict)
group.add(idx=idx, model=self.__dict__[model])
return idx
[docs]
def setup(self):
"""
Set up system for studies.
This function is to be called after adding all device data.
"""
ret = True
t0, _ = elapsed()
if self.is_setup:
logger.warning('System has been setup. Calling setup twice is not allowed.')
ret = False
return ret
self.collect_ref()
self._list2array() # `list2array` must come before `link_ext_param`
if not self.link_ext_param():
ret = False
# --- model parameters range check ---
# TODO: there might be other parameters check?
# NOTE: for Line and GCost parameters check, we use set rather than alter
# to respect the original value in the case file
adjusted_params = []
param_to_check = ['rate_a', 'rate_b', 'rate_c', 'amax', 'amin']
for pname in param_to_check:
param = self.Line.params[pname]
if np.any(param.v == 0):
adjusted_params.append(pname)
param.v[param.v == 0] = param.default
if adjusted_params:
adjusted_params_str = ', '.join(adjusted_params)
msg = f"Zero Line parameters detected, adjusted to default values: {adjusted_params_str}."
logger.info(msg)
# --- bus type correction ---
pq_bus = self.PQ.bus.v
pv_bus = self.PV.bus.v
slack_bus = self.Slack.bus.v
# TODO: how to include islanded buses?
if self.Bus.n > 0 and np.all(self.Bus.type.v == 1):
self.Bus.alter(src='type', idx=pq_bus, value=1)
self.Bus.alter(src='type', idx=pv_bus, value=2)
self.Bus.alter(src='type', idx=slack_bus, value=3)
logger.info("All bus type are PQ, adjusted given load and generator connection status.")
# === no device addition or removal after this point ===
self.calc_pu_coeff() # calculate parameters in system per units
if ret is True:
self.is_setup = True # set `is_setup` if no error occurred
else:
logger.error("System setup failed. Please resolve the reported issue(s).")
self.exit_code += 1
a0 = 0
for _, mdl in self.models.items():
for _, algeb in mdl.algebs.items():
algeb.v = np.zeros(algeb.owner.n)
algeb.a = np.arange(a0, a0 + algeb.owner.n)
a0 += algeb.owner.n
# NOTE: this is a temporary solution for building Y matrix
# consider refator this part if any other similar cases occur in the future
self.Line.a1a = self.Bus.get(src='a', attr='a', idx=self.Line.bus1.v)
self.Line.a2a = self.Bus.get(src='a', attr='a', idx=self.Line.bus2.v)
# assign bus type as placeholder; 1=PQ, 2=PV, 3=ref, 4=isolated
if self.Bus.type.v.sum() == self.Bus.n: # if all type are PQ
self.Bus.alter(src='type', idx=self.PV.bus.v,
value=np.ones(self.PV.n))
self.Bus.alter(src='type', idx=self.Slack.bus.v,
value=np.ones(self.Slack.n))
# --- assign column and row names ---
self.mats.Cft.col_names = self.Line.idx.v
self.mats.Cft.row_names = self.Bus.idx.v
self.mats.CftT.col_names = self.Bus.idx.v
self.mats.CftT.row_names = self.Line.idx.v
self.mats.Cg.col_names = self.StaticGen.get_all_idxes()
self.mats.Cg.row_names = self.Bus.idx.v
self.mats.Cl.col_names = self.PQ.idx.v
self.mats.Cl.row_names = self.Bus.idx.v
self.mats.Csh.col_names = self.Shunt.idx.v
self.mats.Csh.row_names = self.Bus.idx.v
self.mats.Bbus.col_names = self.Bus.idx.v
self.mats.Bbus.row_names = self.Bus.idx.v
self.mats.Bf.col_names = self.Bus.idx.v
self.mats.Bf.row_names = self.Line.idx.v
self.mats.PTDF.col_names = self.Bus.idx.v
self.mats.PTDF.row_names = self.Line.idx.v
self.mats.LODF.col_names = self.Line.idx.v
self.mats.LODF.row_names = self.Line.idx.v
_, s = elapsed(t0)
logger.info('System set up in %s.', s)
return ret
# ---------------------------------------------------------------------------
# Methods adapted from andes.system.System (GPL-3.0, Hantao Cui)
# ---------------------------------------------------------------------------
def _update_config_object(self):
"""
Apply command-line ``config_option`` overrides to the rc config object.
Notes
-----
Adapted from ``andes.system.System._update_config_object``.
Original author: Hantao Cui. License: GPL-3.0.
"""
config_option = self.options.get('config_option', None)
if config_option is None:
return
if len(config_option) == 0:
return
if self._config_object is None:
self._config_object = configparser.ConfigParser()
for item in config_option:
if item.count('=') != 1:
raise ValueError(f'config_option "{item}" must be an assignment expression')
field, value = item.split('=')
if field.count('.') != 1:
raise ValueError(
f'config_option left-hand side "{field}" must use format SECTION.FIELD'
)
section, key = field.split('.')
section, key, value = section.strip(), key.strip(), value.strip()
if not self._config_object.has_section(section):
self._config_object.add_section(section)
logger.debug("New config section added: %s", section)
self._config_object.set(section, key, value)
logger.debug("Config option set: %s.%s=%s", section, key, value)
[docs]
def call_models(self, method: str, models: OrderedDict, *args, **kwargs):
"""
Call a named method on each model in *models*.
Parameters
----------
method : str
Name of the model method to call.
models : OrderedDict
Mapping of model name → model instance.
Returns
-------
OrderedDict
Return values keyed by model name.
Notes
-----
Adapted from ``andes.system.System.call_models`` (stats tracking removed
as AMS does not use the ANDES call-stats machinery).
Original author: Hantao Cui. License: GPL-3.0.
"""
ret = OrderedDict()
for name, mdl in models.items():
ret[name] = getattr(mdl, method)(*args, **kwargs)
return ret
def _list2array(self):
"""
Call ``list2array`` on every model to pre-allocate NumPy arrays.
Notes
-----
Adapted from ``andes.system.System._list2array``.
Original author: Hantao Cui. License: GPL-3.0.
"""
self.call_models('list2array', self.models)
[docs]
def link_ext_param(self, model=None):
"""
Retrieve values for ``ExtParam`` instances across models.
Parameters
----------
model : None
Reserved for API compatibility; AMS always links all models.
Returns
-------
bool
True if all external parameters were linked successfully.
Notes
-----
Adapted from ``andes.system.System.link_ext_param``.
Original author: Hantao Cui. License: GPL-3.0.
"""
del model # API-compat noop — AMS always links all models
ret = True
for mdl in self.models.values():
for instance in mdl.params_ext.values():
ext_name = instance.model
ext_model = self.__dict__[ext_name]
try:
instance.link_external(ext_model)
except (IndexError, KeyError) as e:
logger.error(
'Error: <%s> cannot retrieve <%s> from <%s> using <%s>:\n %s',
mdl.class_name, instance.name, instance.model,
instance.indexer.name, repr(e),
)
ret = False
return ret
[docs]
def calc_pu_coeff(self):
"""
Perform per-unit value conversion for all model parameters.
Notes
-----
Adapted from ``andes.system.System.calc_pu_coeff``.
Original author: Hantao Cui. License: GPL-3.0.
"""
Sb = self.config.mva
for mdl in self.models.values():
Sn = mdl.Sn.v if 'Sn' in mdl.__dict__ else Sb
Vb, Vn = 1, 1
if 'bus' in mdl.__dict__:
Vb = self.Bus.get(src='Vn', idx=mdl.bus.v, attr='v')
Vn = mdl.Vn.v if 'Vn' in mdl.__dict__ else Vb
elif 'bus1' in mdl.__dict__:
Vb = self.Bus.get(src='Vn', idx=mdl.bus1.v, attr='v')
Vn = mdl.Vn1.v if 'Vn1' in mdl.__dict__ else Vb
Zn = Vn ** 2 / Sn
Zb = Vb ** 2 / Sb
Vdcb, Vdcn, Idcn = 1, 1, 1
if 'node' in mdl.__dict__:
Vdcb = self.Node.get(src='Vdcn', idx=mdl.node.v, attr='v')
Vdcn = mdl.Vdcn.v if 'Vdcn' in mdl.__dict__ else Vdcb
Idcn = mdl.Idcn.v if 'Idcn' in mdl.__dict__ else (Sb / Vdcb)
elif 'node1' in mdl.__dict__:
Vdcb = self.Node.get(src='Vdcn', idx=mdl.node1.v, attr='v')
Vdcn = mdl.Vdcn1.v if 'Vdcn1' in mdl.__dict__ else Vdcb
Idcn = mdl.Idcn.v if 'Idcn' in mdl.__dict__ else (Sb / Vdcb)
Idcb = Sb / Vdcb
Rb = Vdcb / Idcb
Rn = Vdcn / Idcn
coeffs = {
'voltage': Vn / Vb,
'power': Sn / Sb,
'ipower': Sb / Sn,
'current': (Sn / Vn) / (Sb / Vb),
'z': Zn / Zb,
'y': Zb / Zn,
'dc_voltage': Vdcn / Vdcb,
'dc_current': Idcn / Idcb,
'r': Rn / Rb,
'g': Rb / Rn,
}
for prop, coeff in coeffs.items():
for p in mdl.find_param(prop).values():
p.set_pu_coeff(coeff)
mdl.coeffs = coeffs
mdl.bases = {'Sn': Sn, 'Sb': Sb, 'Vn': Vn, 'Vb': Vb, 'Zn': Zn, 'Zb': Zb}
[docs]
def collect_config(self):
"""
Collect config data from system, routines, and models into a
:class:`configparser.ConfigParser` object.
Returns
-------
configparser.ConfigParser
Notes
-----
Adapted from ``andes.system.System.collect_config``.
Original author: Hantao Cui. License: GPL-3.0.
"""
config_dict = configparser.ConfigParser()
config_dict[self.__class__.__name__] = self.config.as_dict()
all_with_config = OrderedDict(
list(self.routines.items()) + list(self.models.items())
)
for name, instance in all_with_config.items():
cfg = instance.config.as_dict()
if len(cfg) > 0:
config_dict[name] = cfg
return config_dict
[docs]
def save_config(self, file_path=None, overwrite=False):
"""
Save all system, routine, and model configurations to an rc-formatted
file.
Parameters
----------
file_path : str or None, optional
Path to the configuration file. Defaults to ``~/.ams/ams.rc``.
overwrite : bool, optional
If ``True``, overwrite an existing file without prompting.
Returns
-------
str or None
Path to the written config file, or ``None`` if write was skipped.
Notes
-----
Adapted from ``andes.system.System.save_config``.
Original author: Hantao Cui. License: GPL-3.0.
"""
if file_path is None:
ams_path = os.path.join(os.path.expanduser('~'), '.ams')
os.makedirs(ams_path, exist_ok=True)
file_path = os.path.join(ams_path, 'ams.rc')
elif os.path.isfile(file_path):
if not confirm_overwrite(file_path, overwrite=overwrite):
return None
conf = self.collect_config()
with open(file_path, 'w', encoding='utf-8') as f:
conf.write(f)
logger.info('Config written to "%s"', file_path)
return file_path
[docs]
def supported_routines(self, export='plain'):
"""
Return the support type names and routine names in a table.
Returns
-------
str
A table-formatted string for the types and routines
"""
def rst_ref(name, export):
"""
Refer to the model in restructuredText mode so that
it renders as a hyperlink.
"""
if export == 'rest':
return ":ref:`" + name + '`'
else:
return name
pairs = list()
for g in self.types:
routines = list()
for m in self.types[g].routines:
routines.append(rst_ref(m, export))
if len(routines) > 0:
pairs.append((rst_ref(g, export), ', '.join(routines)))
tab = Tab(title='Supported Types and Routines',
header=['Type', 'Routines'],
data=pairs,
export=export,
)
return tab.draw()
[docs]
def supported_models(self, export='plain'):
"""
Return the supported group names and model names in a table.
Parameters
----------
export : str, optional
Export format: 'plain' (default) or 'rest' for reStructuredText.
Returns
-------
str
A table-formatted string listing groups and their models.
Notes
-----
Adapted from ``andes.system.System.supported_models``.
Original author: Hantao Cui. License: GPL-3.0.
"""
def rst_ref(name, export):
if export == 'rest':
return ":ref:`" + name + '`'
else:
return name
pairs = []
for g in self.groups:
models = []
for m in self.groups[g].models:
models.append(rst_ref(m, export))
if len(models) > 0:
pairs.append((rst_ref(g, export), ', '.join(models)))
tab = Tab(title='Supported Groups and Models',
header=['Group', 'Models'],
data=pairs,
export=export,
)
return tab.draw()
[docs]
def connectivity(self):
"""
Perform connectivity check for system.
"""
raise NotImplementedError
[docs]
def to_andes(self, addfile=None,
setup=False, no_output=False,
default_config=True,
verify=False, tol=1e-3,
**kwargs):
"""
Convert the AMS system to an ANDES system.
Wrapper method for `ams.interface.to_andes`.
A preferred dynamic system file to be added has following features:
1. The file contains both power flow and dynamic models.
2. The file can run in ANDES natively.
3. Power flow models are in the same shape as the AMS system.
4. Dynamic models, if any, are in the same shape as the AMS system.
This function is wrapped as the ``System`` class method ``to_andes()``.
Using the file conversion ``to_andes()`` will automatically
link the AMS system instance to the converted ANDES system instance
in the AMS system attribute ``dyn``.
It should be noted that detailed dynamic simualtion requires extra
dynamic models to be added to the ANDES system, which can be passed
through the ``addfile`` argument.
Parameters
----------
system : System
The AMS system to be converted to ANDES format.
addfile : str, optional
The additional file to be converted to ANDES dynamic mdoels.
setup : bool, optional
Whether to call `setup()` after the conversion. Default is True.
no_output : bool, optional
To ANDES system.
default_config : bool, optional
To ANDES system.
verify : bool
If True, the converted ANDES system will be verified with the source
AMS system using AC power flow.
tol : float
The tolerance of error.
Returns
-------
adsys : andes.system.System
The converted ANDES system.
Examples
--------
>>> import ams
>>> import andes
>>> sp = ams.load(ams.get_case('ieee14/ieee14_uced.xlsx'), setup=True)
>>> sa = sp.to_andes(addfile=andes.get_case('ieee14/ieee14_full.xlsx'),
... setup=False, overwrite=True, no_output=True)
Notes
-----
- Power flow models in the addfile will be skipped and only dynamic models
will be used.
- The addfile format is guessed based on the file extension.
- Index in the addfile is automatically adjusted when necessary.
"""
return to_andes(system=self, addfile=addfile,
setup=setup, no_output=no_output,
default_config=default_config,
verify=verify, tol=tol,
**kwargs)
[docs]
def summary(self):
"""
Print out system summary.
"""
# FIXME: add system connectivity check
# logger.info("-> System connectivity check results:")
rtn_check = OrderedDict((key, val._data_check()) for key, val in self.routines.items())
rtn_types = OrderedDict({tp: [] for tp in self.types.keys()})
for name, data_pass in rtn_check.items():
if data_pass:
r_type = self.routines[name].type
rtn_types[r_type].append(name)
nb = self.Bus.n
nl = self.Line.n
ng = self.StaticGen.n
pd = self.PQ.p0.v.sum()
qd = self.PQ.q0.v.sum()
out = list()
out.append("-> Systen size:")
out.append(f"Base: {self.config.mva} MVA; Frequency: {self.config.freq} Hz")
out.append(f"{nb} Buses; {nl} Lines; {ng} Static Generators")
out.append(f"Active load: {pd:,.2f} p.u.; Reactive load: {qd:,.2f} p.u.")
out.append("-> Data check results:")
for rtn_type, names in rtn_types.items():
if len(names) == 0:
continue
names = ", ".join(names)
out.append(f"{rtn_type}: {names}")
out_str = '\n'.join(out)
logger.info(out_str)
[docs]
def report(self, path=None):
"""
Write system routine reports to a plain-text file.
Parameters
----------
path : str, optional
Output file path or directory. When given, the report is
written to the resolved path and the system-level
``no_output`` flag is bypassed. When ``None``, the report
is written to ``self.files.txt`` only if
``self.files.no_output`` is ``False``.
Returns
-------
bool
``True`` if the report was written, ``False`` otherwise.
Notes
-----
Differences from ANDES:
- ANDES has no ``System.report()``; the equivalent is
per-routine (e.g. ``system.PFlow.report()``) and takes no
``path`` argument.
- In AMS, an explicit ``path`` overrides ``no_output=True`` so
users can produce a one-off report without flipping the
system flag.
.. versionchanged:: 1.3.0
Added ``path`` parameter.
"""
if path is None and self.files.no_output is True:
return False
Report(self).write(path=path)
return True
[docs]
def to_mpc(self):
"""
Export an AMS system to a MATPOWER dict.
Wrapper method for `ams.io.matpower.system2mpc`.
Returns
-------
dict
A dictionary representing the MATPOWER case.
Notes
-----
- In the `gen` section, slack generators are listed before PV generators.
- For uncontrolled generators (`ctrl.v == 0`), their max and min power
limits are set to their initial power (`p0.v`) in the converted MPC.
- In the converted MPC, the indices of area (`bus[:, 6]`) and zone (`bus[:, 10]`)
may differ from the original MPC. However, the mapping relationship is preserved.
For example, if the original MPC numbers areas starting from 1, the converted
MPC may number them starting from 0.
- The coefficients `c2` and `c1` in the generator cost data are scaled by
`baseMVA`.
- Unlike the XLSX and JSON converters, this implementation uses value providers
(`v`) instead of vin. As a result, any changes made through `model.set` will be
reflected in the generated MPC.
.. versionadded:: 1.0.10
"""
return system2mpc(self)
[docs]
def to_m(
self,
outfile: str,
overwrite: Optional[bool] = None,
skip_empty: bool = True,
to_andes: bool = False
) -> bool:
"""
Export an AMS system to a MATPOWER M-file.
Wrapper method for `ams.io.matpower.write`.
Parameters
----------
outfile : str
The output file name.
overwrite : bool, optional
If True, overwrite the existing file. Default is None.
Notes
-----
- In the `gen` section, slack generators are listed before PV generators.
- For uncontrolled generators (`ctrl.v == 0`), their max and min power
limits are set to their initial power (`p0.v`) in the converted MPC.
- In the converted MPC, the indices of area (`bus[:, 6]`) and zone (`bus[:, 10]`)
may differ from the original MPC. However, the mapping relationship is preserved.
For example, if the original MPC numbers areas starting from 1, the converted
MPC may number them starting from 0.
- The coefficients `c2` and `c1` in the generator cost data are scaled by
`baseMVA`.
- Unlike the XLSX and JSON converters, this implementation uses value providers
(`v`) instead of vin. As a result, any changes made through `model.set` will be
reflected in the generated MPC.
.. versionadded:: 1.0.10
"""
return write_m(self, outfile=outfile, overwrite=overwrite)
[docs]
def to_xlsx(
self,
outfile: str,
overwrite: Optional[bool] = None,
skip_empty: bool = True,
add_book: Optional[str] = None,
to_andes: bool = False
) -> bool:
"""
Export the AMS system to an Excel (XLSX) file.
Wrapper method for `ams.io.xlsx.write`.
Parameters
----------
outfile : str
The output file name.
overwrite : bool, optional
If True, overwrite the existing file. Default is None.
skip_empty : bool, optional
If True, skip output of empty models (n = 0). Default is True.
add_book : str, optional
An optional workbook to be added to the output spreadsheet.
If None, no additional workbook is added. Default is None.
to_andes : bool, optional
If True, write to an ANDES system, where non-ANDES models are skipped.
Returns
-------
bool
True if the file is written successfully, False otherwise.
Notes
-----
- Only models with n > 0 are exported if `skip_empty` is True.
- If `to_andes` is True, only ANDES-compatible models are exported.
- The `add_book` parameter allows merging another workbook into the output.
.. versionadded:: 1.0.10
"""
return write_xlsx(self, outfile=outfile,
skip_empty=skip_empty, overwrite=overwrite,
add_book=add_book, to_andes=to_andes)
[docs]
def to_json(
self,
outfile: str,
overwrite: Optional[bool] = None,
skip_empty: bool = True,
to_andes: bool = False
) -> bool:
"""
Export the AMS system to a JSON file.
Wrapper method for `ams.io.json.write`.
Parameters
----------
outfile : str
The output file name.
overwrite : bool, optional
If True, overwrite the existing file. Default is None.
skip_empty : bool, optional
If True, skip output of empty models (n = 0). Default is True.
to_andes : bool, optional
If True, write to an ANDES system, where non-ANDES models are skipped.
Returns
-------
bool
True if the file is written successfully, False otherwise.
Notes
-----
- Only models with n > 0 are exported if `skip_empty` is True.
- If `to_andes` is True, only ANDES-compatible models are exported.
.. versionadded:: 1.0.10
"""
return write_json(self, outfile=outfile,
skip_empty=skip_empty, overwrite=overwrite,
to_andes=to_andes)
[docs]
def to_raw(
self,
outfile: str,
overwrite: Optional[bool] = None,
skip_empty: bool = True,
to_andes: bool = False
) -> bool:
"""
Export the AMS system to a PSS/E RAW v33 file.
Wrapper method for `ams.io.psse.write_raw`.
Parameters
----------
outfile : str
The output file name.
overwrite : bool, optional
If True, overwrite the existing file. Default is None.
skip_empty : bool, optional
If True, skip output of empty models (n = 0). Default is True.
to_andes : bool, optional
If True, write only ANDES-compatible models.
Returns
-------
bool
True if the file is written successfully, False otherwise.
Notes
-----
- Only models with n > 0 are exported if `skip_empty` is True.
- If `to_andes` is True, only ANDES-compatible models are exported.
- This method has not been fully benchmarked yet.
.. versionadded:: 1.0.10
"""
return write_raw(self, outfile=outfile, overwrite=overwrite)
# --------------- Helper Functions ---------------
# NOTE: _config_numpy, load_config_rc are owned in ams.utils.paths
[docs]
def example(setup=True, no_output=True, **kwargs):
"""
Return an :py:class:`ams.system.System` object for the
``ieee14_uced.xlsx`` as an example.
This function is useful when a user wants to quickly get a
System object for testing.
Returns
-------
System
An example :py:class:`ams.system.System` object.
"""
return ams.load(ams.get_case('matpower/case14.m'),
setup=setup, no_output=no_output, **kwargs)