# -*- 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 - API and implementation and/or documentation
# :author: Matthias De Lozzo
# OTHER AUTHORS - MACROSCOPIC CHANGES
"""Provide routines for XDSM and tikz."""
from __future__ import division, unicode_literals
from pyxdsm.XDSM import XDSM
from gemseo.utils.py23_compat import Path, string_types
[docs]class XDSMToPDFConverter(object):
"""Convert an XDSM to a PDF file with tikz and latex."""
def __init__(self): # noqa: D107
self.__xdsm = XDSM()
[docs] def convert(self, xdsm_data, directory_path, filename_without_ext, scenario):
"""Convert a dictionary representation of a XDSM into a pdf.
Args:
xdsm_data: XDSM dictionary representation.
directory_path (str): output directory.
filename_without_ext (str): output file name, default is 'xdsm'.
scenario (str): scenario name, default is 'root'.
quiet (bool): set to True to suppress output from pdflatex.
"""
workflow = xdsm_data[scenario]["workflow"][1]
numbers = {}
self.__get_numbers(numbers, workflow)
self.__add_nodes(numbers, xdsm_data[scenario]["nodes"])
self.__add_edges(xdsm_data[scenario]["edges"])
self.__add_processes(workflow)
# workaround xdsm not well handling file path: latex expects posix path
file_path = str(directory_path / filename_without_ext).replace("\\", "/")
self.__xdsm.write(file_path)
if not Path(filename_without_ext).with_suffix(".pdf").exists():
raise RuntimeError(
"Something went wrong during the Latex compilation,"
" as xdsm.pdf has not been generated. Please have a look at the"
" Latex log file to investigate the root cause of the error."
)
# Build XDSM for sub-scenarios
if scenario == "root":
subscenarios = [key for key in xdsm_data.keys() if "scn-" in key]
for subscenario in subscenarios:
self.convert(
xdsm_data,
directory_path,
filename_without_ext=filename_without_ext + "_" + subscenario,
scenario=subscenario,
)
def __add_processes(self, workflow, prev=None):
"""Create the pyXDSM processes from a given workflow in a recursive way.
Args:
workflow (list): workflow component of the dictionary storing the XDSM.
prev (str): name of the previous node.
"""
systems = []
last_node = None
for idx, system in enumerate(workflow):
if isinstance(system, string_types):
# system is a node
systems.append(system)
last_node = system
elif isinstance(system, list):
# system is a group of nodes
if isinstance(system[0], string_types):
# system[0] is a node
if last_node is not None:
self.__xdsm.add_process([last_node, system[0]], arrow=True)
self.__add_processes(system, workflow[idx - 1])
elif isinstance(system[0], dict):
# system[0] is a parallel sequence
for sub_sys in system[0]["parallel"]:
# add a sub-process for each parallel element
sub_workflow = [last_node, sub_sys]
self.__add_processes(sub_workflow, last_node)
if prev is not None:
systems.append(prev)
self.__xdsm.add_process(systems, arrow=True)
def __get_numbers(self, numbers, nodes, current=0, end=0, following=1):
"""Give number to the different nodes in a recursive way.
Args:
numbers: dictionary.
nodes (list): workflow component of the dictionary storing the XDSM.
current (int): current step index.
end (int): end-loop step index.
following (int): following step index.
Returns:
The following step index.
"""
prev_node = "undefined"
for node in nodes:
if isinstance(node, string_types):
current = following
following = current + 1
end = current
current_l = [current]
if node in list(numbers.keys()):
current_l = numbers[node]["current"] + current_l
numbers[node] = {"current": current_l, "next": following, "end": end}
prev_node = node
elif isinstance(node, list):
following = self.__get_numbers(numbers, node, current, end, following)
numbers[prev_node]["end"] = following
following += 1
elif isinstance(node, dict):
for sub_node in node["parallel"]:
if not isinstance(sub_node, list):
following = self.__get_numbers(
numbers, [sub_node], current, end, following
)
else:
following = self.__get_numbers(
numbers, sub_node, current, end, following
)
return following
def __add_nodes(self, numbers, nodes):
"""Add the different nodes, called 'systems', in the XDSM."""
for node in nodes:
name = ",".join(
[str(current) for current in numbers[node["id"]]["current"]]
)
name_1 = name + ",{}-{}:".format(
str(numbers[node["id"]]["end"]), str(numbers[node["id"]]["next"])
)
name_2 = name + ":"
if node["type"] == "optimization":
node_type = "Optimization"
name = name_1
elif node["type"] == "lp_optimization":
node_type = "LP_Optimization"
name = name_1
if node["type"] == "doe":
node_type = "DOE"
name = name_1
elif node["type"] == "mda":
node_type = "MDA"
name = name_1
elif node["type"] == "mdo":
node_type = "MDO"
name = name_2
elif node["type"] == "analysis":
node_type = "Function"
name = name_2
elif node["type"] == "function":
node_type = "Function"
name = name_2
elif node["type"] == "metamodel":
node_type = "Metamodel"
name = name_2
node_replaced = node["name"]
escaped_characters = ["_", "$", "&", "{", "}", "%"]
for char in escaped_characters:
node_replaced = node_replaced.replace(char, r"\{}".format(char))
name = name + node_replaced
self.__xdsm.add_system(node["id"], node_type, r"\text{" + name + "}")
def __add_edges(self, edges):
"""Add the edges called connections, inputs, outputs to the XDSM."""
for edge in edges:
old_names = edge["name"].split(",")
names = []
for name in old_names:
sub = sup = None
s_name = name.split("^")
if len(s_name) == 2:
name = s_name[0]
suffix = s_name[1]
s_name = suffix.split("_")
if len(s_name) == 1:
s_name = name.split("_")
if len(s_name) > 1:
sup = suffix
sub = s_name[-1]
name = "".join(s_name[:-1])
else:
sub = s_name[0]
sup = s_name[1]
else:
s_name = name.split("_")
if len(s_name) > 1:
sub = s_name[-1]
name = "".join(s_name[:-1])
sup = "" if sup is None else "^{" + sup + "}"
sub = "" if sub is None else "_{" + sub + "}"
names.append(name + sub + sup)
names = ", ".join(names)
if edge["to"] == "_U_":
self.__xdsm.add_output(edge["from"], r"" + names + "", side="left")
elif edge["from"] != "_U_":
self.__xdsm.connect(edge["from"], edge["to"], r"" + names + "")
elif edge["from"] == "_U_":
self.__xdsm.add_input("Opt", r"" + names + "")
[docs]def xdsm_data_to_pdf(
xdsm_data, directory_path, filename_without_ext="xdsm", scenario="root"
):
"""Convert a dictionary representation of a XDSM to a pdf.
Args:
xdsm_data: XDSM dictionary representation.
directory_path (str): output directory.
filename_without_ext (str): output file name, default is 'xdsm'.
scenario (str): scenario name, default is 'root'.
"""
converter = XDSMToPDFConverter()
converter.convert(xdsm_data, directory_path, filename_without_ext, scenario)
