import logging
from collections import OrderedDict
from time import strftime
from andes import __version__ as version
from andes.io.txt import dump_data
from andes.shared import np
from andes.utils.misc import elapsed
logger = logging.getLogger(__name__)
all_formats = {}
[docs]def report_info(system):
info = list()
info.append('ANDES' + ' ' + version + '\n')
info.append('Copyright (C) 2015-2022 Hantao Cui\n\n')
info.append('ANDES comes with ABSOLUTELY NO WARRANTY\n')
info.append('Case file: ' + str(system.files.case) + '\n')
info.append('Report time: ' + strftime("%m/%d/%Y %I:%M:%S %p") + '\n\n')
if system.PFlow.converged is True:
info.append(f'Power flow converged in {system.PFlow.niter + 1} iterations.\n')
info.append('Flat-start: ' +
('Yes' if system.Bus.config.flat_start else 'No') + '\n')
return info
[docs]class Report:
"""
Report class to store system static analysis reports
"""
[docs] def __init__(self, system):
self.system = system
self.basic = OrderedDict()
self.extended = OrderedDict()
@property
def info(self):
return report_info(self.system)
[docs] def update(self):
"""
Update values based on the requested content
"""
system = self.system
self.basic.update({
'Buses': system.Bus.n,
'Generators': system.PV.n + system.Slack.n,
'Committed Gens': int(sum(system.PV.u.v) + sum(system.Slack.u.v)),
'Loads': system.PQ.n,
'Shunts': system.Shunt.n,
'Lines': system.Line.n,
'Transformers': np.count_nonzero(system.Line.trans.v == 1),
'Areas': system.Area.n,
})
if self.system.PFlow.converged is False:
logger.warning('Cannot update extended summary. Power flow not solved.')
return
self.extended.update({
'Pg': sum(system.PV.u.v * system.PV.p.v) + sum(system.Slack.u.v * system.Slack.p.v),
'Qg': sum(system.PV.u.v * system.PV.q.v) + sum(system.Slack.u.v * system.Slack.q.v),
'Pl': round(float(sum(system.PQ.p0.v)), 6),
'Ql': round(float(sum(system.PQ.q0.v)), 6),
'Ptot': sum(system.PV.pmax.v) + sum(system.Slack.pmax.v),
'Pon': sum(system.PV.u.v * system.PV.pmax.v),
'Qtot_min': sum(system.PV.qmin.v) + sum(system.Slack.qmin.v),
'Qtot_max': sum(system.PV.qmax.v) + sum(system.Slack.qmax.v),
'Qon_min': sum(system.PV.u.v * system.PV.qmin.v),
'Qon_max': sum(system.PV.u.v * system.PV.qmax.v),
})
[docs] def write(self):
"""
Write report to file.
"""
system = self.system
if system.files.no_output is True:
return
text = list()
header = list()
row_name = list()
data = list()
self.update()
t, _ = elapsed()
# ----------------------------------------
# info section
text.append(self.info)
header.append(None)
row_name.append(None)
data.append(None)
# ----------------------------------------
# ----------------------------------------
# summary section
text.append(['Statistics:\n'])
header.append(None)
row_name.append(self.basic.keys())
data.append(list(self.basic.values()))
# ----------------------------------------
if len(self.extended):
text.append(['EXTENDED SUMMARY:\n'])
header.append(['P (pu)', 'Q (pu)'])
row_name.append(
['Generation', 'Load'])
Pcol = [
self.extended['Pg'],
self.extended['Pl'],
]
Qcol = [
self.extended['Qg'],
self.extended['Ql'],
]
data.append([Pcol, Qcol])
if system.PFlow.converged:
# ----------------------------------------
# Bus data
angle_unit = 'deg.' if system.PFlow.config.degree else 'rad.'
angles = np.rad2deg(system.Bus.a.v) \
if system.PFlow.config.degree == 1 else system.Bus.a.v
text.append(['BUS DATA:\n'])
header.append(['Bus Name', 'Vm(pu)', f'Va({angle_unit})'])
row_name.append(system.Bus.idx.v)
data.append([system.Bus.name.v, system.Bus.v.v, angles])
# ----------------------------------------
# ----------------------------------------
# Node data
if hasattr(system, 'Node') and system.Node.n:
text.append(['NODE DATA:\n'])
header.append(['Node Name', 'V(pu)'])
row_name.append(system.Node.idx.v)
data.append([system.Node.name.v, system.Node.v.v])
# ----------------------------------------
# ----------------------------------------
# Line data
text.append(['LINE DATA:\n'])
header.append([
'Line Name', 'Fr. Bus (idx)', 'To Bus (idx)', 'P From (pu)', 'Q From (pu)',
'P To (pu)', 'Q To (pu)'
])
row_name.append(system.Line.idx.v)
data.append([system.Line.name.v,
system.Line.bus1.v,
system.Line.bus2.v,
system.Line.a1.e,
system.Line.v1.e,
system.Line.a2.e,
system.Line.v2.e
])
# ----------------------------------------
# ----------------------------------------
# Additional Algebraic data
text.append(['OTHER ALGEBRAIC VARIABLES:\n'])
header.append([''])
row_name.append(system.dae.y_name[2 * system.Bus.n:system.dae.m])
data.append([round(i, 6) for i in system.dae.y[2 * system.Bus.n:]])
# Additional State variable data
if system.dae.n:
text.append(['OTHER STATE VARIABLES:\n'])
header.append([''])
row_name.append(system.dae.x_name[:])
data.append([round(i, 6) for i in system.dae.x[:]])
dump_data(text, header, row_name, data, system.files.txt)
_, s = elapsed(t)
logger.info(f'Report saved to "{system.files.txt}" in {s}.')