# -*- coding: utf-8 -*-
# Copyright 2021 IRT Saint Exupéry, https://www.irt-saintexupery.com
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License version 3 as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
# Contributors:
# INITIAL AUTHORS - initial API and implementation and/or
# initial documentation
# :author: Francois Gallard
# OTHER AUTHORS - MACROSCOPIC CHANGES
"""
Generate a N2 and XDSM from an Excel description of the MDO problem
*******************************************************************
"""
from __future__ import absolute_import, division, unicode_literals
from ast import literal_eval
from future import standard_library
from pandas import read_excel
from six import string_types
from gemseo import LOGGER
from gemseo.api import (
create_design_space,
create_scenario,
generate_n2_plot,
get_available_formulations,
)
from gemseo.core.coupling_structure import MDOCouplingStructure
from gemseo.core.discipline import MDODiscipline
standard_library.install_aliases()
[docs]class XLSStudyParser(object):
"""
Parse the input excel files that describe the GEMSEO study.
The Excel file must contain one sheet per discipline.
The name of the sheet is the discipline name.
The sheet has at least two columns, one for inputs and one for outputs,
with the following format:
+--------+---------+----------------+--------------------+----------------+
| Inputs | Outputs |Design variables| Objective function | Constraints |
+========+=========+================+====================+================+
| in1 | out1 | in1 | out1 | out2 |
+--------+---------+----------------+--------------------+----------------+
| in2 | out2 | | | |
+--------+---------+----------------+--------------------+----------------+
Empty lines are ignored.
All Objective functions and constraints must be outputs of a discipline,
not necessarily the one of the current sheet
All Design variables must be inputs of a discipline, not necessarily
the one of the current sheet
"""
SCENARIO_PREFIX = "Scenario"
DISCIPLINE = "Discipline"
DISCIPLINES = "Disciplines"
OBJECTIVE_FUNCTION = "Objective function"
CONSTRAINTS = "Constraints"
DESIGN_VARIABLES = "Design variables"
FORMULATION = "Formulation"
OPTIONS = "Options"
OPTIONS_VALUES = "Options values"
def __init__(self, xls_study_path):
"""
Initializes the study from the excel specification
:param xls_study_path: path to the excel file describing the study
"""
self.xls_study_path = xls_study_path
try:
self.frames = read_excel(xls_study_path, sheet_name=None, engine="openpyxl")
except IOError:
LOGGER.error("Failed to open study file !")
raise
LOGGER.info("Detected the following disciplines: %s", list(self.frames.keys()))
self.disciplines = {}
self.scenarios = {}
self.inputs = []
self.outputs = []
self._init_disciplines()
self._get_opt_pb_descr()
if not self.scenarios:
raise ValueError("Found no scenario in the xls file !")
def _init_disciplines(self):
""" Initializes disciplines. """
for disc_name, frame in self.frames.items():
if disc_name.startswith(self.SCENARIO_PREFIX):
continue
LOGGER.info("Parsing discipline %s", disc_name)
try:
inputs = self._get_frame_series_values(frame, "Inputs")
except ValueError:
raise ValueError(
"Discipline "
+ str(disc_name)
+ "'s sheet must have an Inputs column !"
)
self.inputs += inputs
try:
outputs = self._get_frame_series_values(frame, "Outputs")
except ValueError:
raise ValueError(
"Discipline "
+ str(disc_name)
+ "'s sheet must have an Outputs column !"
)
if not outputs:
raise ValueError("Discipline " + str(disc_name) + " has no Outputs")
self.outputs += outputs
disc = MDODiscipline(disc_name)
disc.input_grammar.initialize_from_data_names(inputs)
disc.output_grammar.initialize_from_data_names(outputs)
LOGGER.info("Inputs : %s", inputs)
LOGGER.info("Outputs : %s", outputs)
self.disciplines[disc_name] = disc
self.inputs = set(self.inputs)
self.outputs = set(self.outputs)
@staticmethod
def _get_frame_series_values(frame, series_name, return_none=False):
"""
Gets the data list of a named column
Removes empty data
:param frame: the pandas frame of the sheet
:param series_name: name of the series
:param return_none: if the series does not exists, returns None
instead of raising a ValueError
"""
series = frame.get(series_name)
if series is None:
if return_none:
return None
raise ValueError("The sheet has no series named " + str(series_name))
# Remove empty data
# pylint: disable=comparison-with-itself
return list([val for val in series.tolist() if val == val])
def _get_opt_pb_descr(self):
"""
Initilalize the objective function, constraints and design_variables
"""
self.scenarios = {}
for frame_name, frame in self.frames.items():
if not frame_name.startswith(self.SCENARIO_PREFIX):
continue
LOGGER.info("Detected scenario in sheet: %s", frame_name)
try:
disciplines = self._get_frame_series_values(frame, self.DISCIPLINES)
except ValueError:
raise ValueError(
"Scenario "
+ str(frame_name)
+ " has no "
+ self.DISCIPLINES
+ " column !"
)
try:
design_variables = self._get_frame_series_values(
frame, self.DESIGN_VARIABLES
)
except ValueError:
raise ValueError(
"Scenario "
+ str(frame_name)
+ " has no "
+ self.DESIGN_VARIABLES
+ " column !"
)
try:
objectives = self._get_frame_series_values(
frame, self.OBJECTIVE_FUNCTION
)
except ValueError:
raise ValueError(
"Scenario "
+ str(frame_name)
+ " has no "
+ self.OBJECTIVE_FUNCTION
+ " column !"
)
try:
constraints = self._get_frame_series_values(frame, self.CONSTRAINTS)
except ValueError:
raise ValueError(
"Scenario "
+ str(frame_name)
+ " has no "
+ self.CONSTRAINTS
+ " column !"
)
try:
formulation = self._get_frame_series_values(frame, self.FORMULATION)
except ValueError:
raise ValueError(
"Scenario "
+ str(frame_name)
+ " has no "
+ self.FORMULATION
+ " column !"
)
options = self._get_frame_series_values(frame, self.OPTIONS, True)
options_values = self._get_frame_series_values(
frame, self.OPTIONS_VALUES, True
)
if len(formulation) != 1:
raise ValueError(
"Scenario "
+ str(frame_name)
+ " must have 1 "
+ self.FORMULATION
+ " value !"
)
if options is not None:
if len(options) != len(options_values):
raise ValueError(
"Options "
+ str(options)
+ " and Options values "
+ str(options_values)
+ " must have the same length!"
)
formulation = formulation[0]
scn = {}
scn[self.DISCIPLINES] = disciplines
scn[self.OBJECTIVE_FUNCTION] = objectives
scn[self.CONSTRAINTS] = constraints
scn[self.DESIGN_VARIABLES] = design_variables
scn[self.FORMULATION] = formulation
scn[self.OPTIONS] = options
scn[self.OPTIONS_VALUES] = options_values
self.scenarios[frame_name] = scn
for name, desc in self.scenarios.items():
self._check_opt_pb(
desc[self.OBJECTIVE_FUNCTION],
desc[self.CONSTRAINTS],
desc[self.DISCIPLINES],
desc[self.DESIGN_VARIABLES],
desc[self.FORMULATION],
name,
)
def _check_opt_pb(
self,
objectives,
constraints,
disciplines,
design_variables,
formulation,
scn_name,
):
"""
Checks the optimization problem consistency.
Raises errors if needed
:param objectives: list of objectives
:param constraints: list of constraints
:param disciplines: list of MDODisciplines
:param design_variables: list of design varaibles
:param formulation : mdo formulation name
:param scn_name: name of the scenario
"""
LOGGER.info("New scenario: %s", scn_name)
LOGGER.info("Objectives: %s", objectives)
LOGGER.info("Disciplines: %s", disciplines)
LOGGER.info("Constraints: %s", constraints)
LOGGER.info("Design variables: %s", design_variables)
LOGGER.info("Formulation: %s", formulation)
missing = set(design_variables) - self.inputs
if missing:
raise ValueError(
scn_name + " : some design variables are "
"not the inputs of any discipline :" + str(list(missing))
)
missing = (
set(disciplines)
- set(list(self.disciplines.keys()))
- set(list(self.scenarios))
)
if missing:
raise ValueError(
scn_name + " : some disciplines dont exist :" + str(list(missing))
)
missing = set(constraints) - self.outputs
if missing:
raise ValueError(
scn_name + " : some constraints are not "
"the outputs of any discipline :" + str(list(missing))
)
missing = set(objectives) - self.outputs
if missing:
raise ValueError(
scn_name + " : some objectives are not "
"the outputs of any discipline :" + str(list(missing))
)
if not objectives:
raise ValueError(scn_name + " : no objectives are defined!")
if formulation not in get_available_formulations():
raise ValueError(
"Unknown formulation "
+ str(formulation)
+ " Use one of "
+ str(get_available_formulations())
)
[docs]class StudyAnalysis(object):
"""
Generate a N2 (equivalent to the Design Structure Matrix) diagram,
showing the couplings between discipline and
XDSM, (Extended Design Structure Matrix), showing the MDO process,
from a Excel specification of the inputs, outputs, design variables,
objectives and constraints.
The input excel files contains one sheet per discipline.
The name of the sheet is the discipline name.
The sheet has at least two columns, one for inputs and one for outputs,
with the following format:
+--------+---------+
| Inputs | Outputs |
+========+=========+
| in1 | out1 |
+--------+---------+
| in2 | out2 |
+--------+---------+
[Disc1]
Empty lines are ignored.
The scenarios (at least one, or multiple for distributed formulations)
must appear in a Excel sheet name starting by "Scenario".
The sheet has the following columns, with some constraints :
All of them are mandatory, even if empty for the Constraints
The order may be any
1 and only 1 formulation must be declared
At least 1 objective must be provided, and 1 design variable
+----------------+--------------------+----------------+----------------+----------------+----------------+----------------+
|Design variables| Objective function | Constraints | Disciplines | Formulation | Options | Options values |
+================+====================+================+================+================+================+================+
| in1 | out1 | out2 | Disc1 | MDF | tolerance | 0.1 |
+----------------+--------------------+----------------+----------------+----------------+----------------+----------------+
| | | | Disc2 | | | |
+----------------+--------------------+----------------+----------------+----------------+----------------+----------------+
[Scenario1]
All Objective functions and constraints must be outputs of a discipline,
not necessarily the one of the current sheet.
All Design variables must be inputs of a discipline, not necessarily
the one of the current sheet.
The Options and Options values columns are used to pass
the formulation options
To use multi level MDO formulations, create multiple scenarios,
and add the name of the sub scenarios
in the list of disciplines of the main (system) scenario.
An arbitrary number of levels can be generated this way
(three, four levels etc formulations).
"""
AVAILABLE_DISTRIBUTED_FORMULATIONS = ("BiLevel", "BLISS98B")
def __init__(self, xls_study_path):
"""
Initializes the study from the excel specification
:param xls_study_path: path to the excel file describing the study
"""
self.xls_study_path = xls_study_path
self.study = XLSStudyParser(self.xls_study_path)
self.disciplines_descr = self.study.disciplines
self.scenarios_descr = self.study.scenarios
self.disciplines = {}
self.scenarios = {}
self.main_scenario = None
self._create_scenarios()
[docs] def generate_n2(
self,
file_path="n2.pdf",
show_data_names=True,
save=True,
show=False,
figsize=(15, 10),
):
"""
Generate a N2 plot for the disciplines list.
:param file_path: File path of the figure.
:type file_path: str
:param show_data_names: If true, the names of the
coupling data is shown
otherwise, circles are drawn, which size depend on the
number of coupling names.
:type show_data_names: bool
:param save: If True, saved the figure to file_path.
:type save: bool
:param show: If True, shows the plot.
:type show: bool
:param figsize: Size of the figure.
:type figsize: tuple(float)
"""
generate_n2_plot(
list(self.disciplines.values()),
file_path,
show_data_names,
save,
show,
figsize,
)
@staticmethod
def _create_scenario(disciplines, scenario_descr):
"""
Create a MDO scenario
:param disciplines: list of MDODisciplines
:param scenario_descr: description dict of the scenario
:returns: the MDOScenario
"""
coupl_struct = MDOCouplingStructure(disciplines)
couplings = coupl_struct.get_all_couplings()
design_space = create_design_space()
scn_dv = scenario_descr[XLSStudyParser.DESIGN_VARIABLES]
for var in set(scn_dv) | set(couplings):
design_space.add_variable(var, size=1)
options = scenario_descr[XLSStudyParser.OPTIONS]
options_dict = {}
if options is not None:
options_values = scenario_descr[XLSStudyParser.OPTIONS_VALUES]
for opt, val in zip(options, options_values):
if isinstance(val, string_types):
try:
val = literal_eval(val)
except ValueError as err:
LOGGER.error(err)
raise ValueError(
"Failed to parse option " + str(opt) + " value :" + str(val)
)
else:
pass
options_dict[opt] = val
scenario = create_scenario(
disciplines,
scenario_descr[XLSStudyParser.FORMULATION],
scenario_descr[XLSStudyParser.OBJECTIVE_FUNCTION],
design_space,
**options_dict
)
for cstr in scenario_descr[XLSStudyParser.CONSTRAINTS]:
scenario.add_constraint(cstr)
return scenario
def _get_disciplines_instances(self, scn):
"""
Returns instances of the disciplines of the scenario,
or None if not all available
"""
discs = []
for disc_name in scn[XLSStudyParser.DISCIPLINES]:
disc_inst = self.disciplines_descr.get(disc_name)
if disc_inst is None: # not a discipline, so maybe a scenario
disc_inst = self.scenarios.get(disc_name)
if disc_inst is None:
return None
discs.append(disc_inst)
return discs
def _create_scenarios(self):
"""
Create the main scenario, eventually including sub scenarios
"""
n_scn = len(self.scenarios_descr)
i = 0
while len(self.scenarios) != n_scn and i <= n_scn:
i += 1
for name, scn in self.scenarios_descr.items():
discs = self._get_disciplines_instances(scn)
if discs is not None: # All depdendencies resolved
for disc in discs:
if not disc.is_scenario():
self.disciplines[disc.name] = disc
scenario = self._create_scenario(discs, scn)
self.scenarios[name] = scenario
# The last scenario created is the one
# with the most dependencies
# so the main one
self.main_scenario = scenario
# At each while iteration at least 1 scenario must be resolved
# otherwise this means there is a cross dependency between
# scenarios
if len(self.scenarios) != n_scn:
raise ValueError(
"Scenarios dependencies cannot be resolved,"
" check for cycling dependencies "
"between scenarios!"
)
[docs] def generate_xdsm(self, output_dir, latex_output=False, open_browser=False):
"""
Creates an xdsm.json file from the current scenario.
:param output_dir: the directory where XDSM html files are generated
:param latex_output: build .tex, .tikz and .pdf file
:returns: the MDOScenario, that contains the DesignSpace, the
formulation, but the disciplines have only correct
input and output grammars but no _run methods so that can't be executed
"""
LOGGER.info("Generated the following Scenario:")
self.main_scenario.log_me()
self.main_scenario.formulation.opt_problem.log_me()
self.main_scenario.xdsmize(
outdir=output_dir, latex_output=latex_output, open_browser=open_browser
)
return self.main_scenario
