"""Plotting components and utilities for TopFarm optimizations."""
import os
from openmdao.api import ExplicitComponent
import matplotlib.pyplot as plt
import numpy as np
import topfarm
import sys
from IPython.display import display
import uuid
[docs]
def in_notebook():
try:
from IPython import get_ipython
return get_ipython() is not None
except BaseException:
return False
[docs]
class NoPlot():
"""Plotting component for no plotting"""
# fmt:off
BASE_COLORS = [
'b', 'r', 'm', 'c', 'g', 'y', 'magenta',
'orange', 'indigo', 'grey', 'purple',
'violet', 'firebrick', 'coral', 'pink', 'lime'
]
# fmt:on
[docs]
class XYPlotComp(ExplicitComponent):
"""Plotting component for turbine locations"""
colors = BASE_COLORS * (2000 // len(BASE_COLORS) + 1) # ~2000 colors delay tech-debt
[docs]
def __init__(self, memory=10, delay=0.05, plot_initial=True, plot_improvements_only=False, ax=None, legendloc=1, save_plot_per_iteration=False, folder_name='Figures', file_prefix='iteration', callback=None):
"""Initialize component for plotting turbine locations
Parameters
----------
memory : int, optional
Number of previous iterations to remember
delay : float, optional
Time delay in seconds between plotting updates
plot_initial : bool, optional
Flag to plot the initial turbine locations
plot_improvements_only : bool, optional
Flag to plot only improvements in cost
ax : matplotlib axes, optional
Axes into which to make the plot
legendloc : int
Location of the legend in the plot
folder_name : str
name of the folder to save plots in
callback : None or function
optional function that accepts ax as an argument
file_prefix : str
prefix of saved plot file names
"""
ExplicitComponent.__init__(self)
self.delay = delay
self.plot_improvements_only = plot_improvements_only
self._ax = ax
self.memory = memory
self.delay = max([delay, 1e-6])
self.plot_initial = plot_initial
self.history = []
self.counter = 0
self.by_pass = False
self.legendloc = legendloc
self.callback = callback
self.folder_name = folder_name
self.file_prefix = file_prefix
self.save_plot_per_iteration = save_plot_per_iteration
self._use_notebook = in_notebook()
if ax is None:
self._fig, self._ax = plt.subplots()
if self._use_notebook:
plt.close(self._fig)
else:
self._ax = ax
self._fig = ax.figure
@property
def ax(self):
return self._ax
[docs]
def show(self):
plt.show()
[docs]
def setup(self):
if topfarm.x_key in self.problem.design_vars:
units_x = self.problem.design_vars[topfarm.x_key][-1]
else:
units_x = None
if topfarm.y_key in self.problem.design_vars:
units_y = self.problem.design_vars[topfarm.y_key][-1]
else:
units_y = None
self.add_input(topfarm.x_key, np.zeros(self.n_wt), units=units_x)
self.add_input(topfarm.y_key, np.zeros(self.n_wt), units=units_y)
if hasattr(self.problem, 'xy_boundary'):
self.xy_boundary = self.problem.xy_boundary
if hasattr(self.problem.cost_comp, 'output_key'):
self.cost_key = self.problem.cost_comp.output_key
self.cost_unit = self.problem.cost_comp.output_unit
else:
self.cost_key = "Cost"
self.cost_unit = ""
self.add_input(self.cost_key, 0.)
self.add_output('plot_counter')
[docs]
def init_plot(self, limits):
self.ax.cla()
self.ax.axis('equal')
# mi = limits.min(0)
# ma = limits.max(0)
# ra = ma - mi + 1
# ext = .1
# xlim, ylim = np.array([mi - ext * ra, ma + ext * ra]).T
# self.ax.set_xlim(xlim)
# self.ax.set_ylim(ylim)
# def plot_boundary(self):
# b = np.r_[self.xy_boundary[:], self.xy_boundary[:1]]
# plt.plot(b[:, 0], b[:, 1], 'k')
[docs]
def plot_constraints(self):
for constr in self.problem.model.constraint_components:
constr.plot(self.ax)
[docs]
def plot_history(self, x, y):
rec = self.problem.recorder
if rec.num_cases > 0:
def get(xy, xy_key, pw):
rec_xy = pw[xy_key][-self.memory:]
if len(rec_xy.shape) == 1:
rec_xy = rec_xy[:, np.newaxis]
return np.r_[rec_xy, [xy]]
pw = self.problem.get_vars_from_recorder()
x = get(x, topfarm.x_key, pw)
y = get(y, topfarm.y_key, pw)
for c, x_, y_ in zip(self.colors, x.T, y.T):
self.ax.plot(x_, y_, '--', color=c)
[docs]
def plot_initial2current(self, x0, y0, x, y):
rec = self.problem.recorder
if rec.num_cases > 0:
pw = self.problem.get_vars_from_recorder()
x0 = np.atleast_1d(pw['x0'])
y0 = np.atleast_1d(pw['y0'])
for c, x0_, y0_, x_, y_ in zip(self.colors, x0, y0, x, y):
self.ax.plot(x0_, y0_, '>', markerfacecolor=c, markeredgecolor='k')
self.ax.plot((x0_, x_), (y0_, y_), '-', color=c)
self.ax.plot([], [], '>k', markerfacecolor="#00000000", markeredgecolor='k', label='Initial position')
[docs]
def plot_current_position(self, x, y):
for c, x_, y_ in zip(self.colors, x, y):
self.ax.plot(x_, y_, 'o', color=c, ms=5)
self.ax.plot(x_, y_, 'xk', ms=4)
self.ax.plot([], [], 'xk', label='Current position')
[docs]
def set_title(self, cost0, cost):
rec = self.problem.recorder
if hasattr(self.problem.cost_comp, 'inc_or_exp'):
inc_or_exp = self.problem.cost_comp.inc_or_exp
else:
inc_or_exp = 1.0
title = "%d) %s %f %s" % (rec.num_cases, self.cost_key, cost * inc_or_exp, self.cost_unit)
if cost0 != 0:
title += " (%+.2f%%)" % ((cost - cost0) / cost0 * 100)
self.ax.set_title(title)
[docs]
def get_initial(self):
rec = self.problem.recorder
if rec.num_cases > 0:
pw = self.problem.get_vars_from_recorder()
cost0 = self.problem.recorder[self.cost_key][0]
# cost0 = pw['cost0']
return pw['x0'], pw['y0'], cost0
[docs]
def compute(self, inputs, outputs):
if self.by_pass is False:
rec = self.problem.recorder
cost = inputs[self.cost_key][0]
if (self.plot_improvements_only and
'cost' in rec.driver_iteration_dict and
len(rec['cost']) and
cost > rec['cost'].min()):
return
# detect new optimization run
if self._use_notebook:
if not hasattr(self, "_last_num_cases"):
self._last_num_cases = -1
if rec.num_cases < self._last_num_cases:
# new run detected → reset display
if hasattr(self, "_display_id"):
del self._display_id
self._last_num_cases = rec.num_cases
# find limits
def get_lim(key):
if (key in self.problem.design_vars and
isinstance(self.problem.design_vars[key], tuple) and
len(self.problem.design_vars[key]) == 4):
return np.min(self.problem.design_vars[key][1]), np.max(np.min(self.problem.design_vars[key][2]))
else:
return min(inputs[key]), max(inputs[key])
min_x, max_x = get_lim(topfarm.x_key)
min_y, max_y = get_lim(topfarm.y_key)
self.init_plot(np.array([[min_x, min_y], [max_x, max_y]]))
self.plot_constraints()
initial = self.get_initial()
x = inputs[topfarm.x_key]
y = inputs[topfarm.y_key]
if initial is not None:
x0, y0, cost0 = initial
if self.plot_initial:
self.plot_initial2current(x0, y0, x, y)
if self.memory > 0:
self.plot_history(x, y)
else:
cost0 = cost
self.plot_current_position(x, y)
self.set_title(cost0, cost)
self.ax.legend(loc=self.legendloc)
if self._use_notebook:
if not hasattr(self, "_display_id"):
self._display_id = str(uuid.uuid4())
display(self._fig, display_id=self._display_id)
else:
from IPython.display import update_display
update_display(self._fig, display_id=self._display_id)
if plt.fignum_exists(self._fig.number):
plt.close(self._fig)
import time
time.sleep(self.delay)
else:
plt.draw()
plt.pause(self.delay)
self.counter += 1
outputs['plot_counter'] = np.sum(inputs[topfarm.x_key]) * 0.0 + self.counter
if self.callback is not None:
self.callback(self.ax)
if self.save_plot_per_iteration:
if not os.path.exists(self.folder_name):
os.makedirs(self.folder_name)
plt.savefig(f'{self.folder_name}/{self.file_prefix}_%s.png' % self.counter)
[docs]
class PlotComp(XYPlotComp):
def __init__(self, memory=10, delay=0.05, plot_initial=True, plot_improvements_only=False, ax=None):
XYPlotComp.__init__(self, memory=memory, delay=delay, plot_initial=plot_initial,
plot_improvements_only=plot_improvements_only, ax=ax)
sys.stderr.write("%s is deprecated. Use XYPlotComp instead\n" % self.__class__.__name__)
[docs]
class TurbineTypePlotComponent(XYPlotComp):
"""Plotting component for turbine types"""
colors = np.array(BASE_COLORS * 10)
markers = np.array(list("123v^<>.o48spP*hH+xXDd|_"))
[docs]
def __init__(self, turbine_type_names, **kwargs):
"""Initialize component for plotting turbine types
Parameters
----------
turbine_type_names : list of str
Names of turbine types for legend
**kwargs : keyword arguments, optional
Keyword arguments that can be passed into XYPlotComp
"""
self.turbine_type_names = turbine_type_names
XYPlotComp.__init__(self, **kwargs)
[docs]
def setup(self):
XYPlotComp.setup(self)
self.add_input(topfarm.type_key, np.zeros(self.n_wt, dtype=int))
[docs]
def compute(self, inputs, outputs):
self.types = np.asarray(inputs[topfarm.type_key], dtype=int)
XYPlotComp.compute(self, inputs, outputs)
[docs]
def init_plot(self, limits):
XYPlotComp.init_plot(self, limits)
for m, n, c in zip(self.markers, self.turbine_type_names, self.colors):
self.ax.plot([], [], m + c, label=n)
self.ax.legend()
[docs]
def plot_initial2current(self, x0, y0, x, y):
rec = self.problem.recorder
if rec.num_cases > 0:
pw = self.problem.get_vars_from_recorder()
x0 = np.atleast_1d(pw['x0'])
y0 = np.atleast_1d(pw['y0'])
for c, x0_, y0_, x_, y_ in zip(self.colors[self.types], x0, y0, x, y):
self.ax.plot(x0_, y0_, '>', markerfacecolor=c, markeredgecolor='k')
self.ax.plot((x0_, x_), (y0_, y_), '-', color=c)
self.ax.plot([], [], '>k', markerfacecolor="#00000000", markeredgecolor='k', label='Initial position')
[docs]
def plot_history(self, x, y):
rec = self.problem.recorder
if rec.num_cases > 0:
def get(xy, xy_key, pw):
rec_xy = pw[xy_key][-self.memory:]
if len(rec_xy.shape) == 1:
rec_xy = rec_xy[:, np.newaxis]
return np.r_[rec_xy, [xy]]
pw = self.problem.get_vars_from_recorder()
x = get(x, topfarm.x_key, pw)
y = get(y, topfarm.y_key, pw)
for c, x_, y_ in zip(self.colors[self.types], x.T, y.T):
self.ax.plot(x_, y_, '--', color=c)
[docs]
def plot_current_position(self, x, y):
for m, c, x_, y_ in zip(self.markers[self.types], self.colors[self.types], x, y):
# self.ax.plot(x_, y_, 'o', color=c, ms=5)
self.ax.plot(x_, y_, m + 'k', markeredgecolor=c, markeredgewidth=1, ms=20)