# -*- coding: utf-8 -*-
#
# This file is subject to the terms and conditions defined in
# file 'LICENSE', which is part of this source code package.
#
#
import os
try:
from mpi4py import MPI
except ImportError:# pragma: no cover
print("MPI capabilities not available")
from BasicTools.IO import XdmfWriter as XW
from genericROM.FE import FETools as FT
from Mordicus.IO import StateIO as SIO
import numpy as np
[docs]def WriteSolution(mesh, solution, outputName):
"""
Functional API
Writes a solution on disk satisfying the corresponding format
Parameters
----------
mesh : BasicToolsUnstructuredMesh
high-dimensional mesh
solution : Solution
solution containing reducedCoordinates
outputName : str
name of the file on disk where the solution is written
"""
writer = XDMFWriter(outputName)
writer.WriteSolution(mesh, solution)
[docs]def WriteProblemDataSolutions(mesh, problemData, solutionNameRef, outputName):
"""
Functional API
Writes on disk the solutions defined in a ProblemData, satisfying the
corresponding format
Parameters
----------
mesh : BasicToolsUnstructuredMesh
high-dimensional mesh
problemData : ProblemData
ProblemData containing the solutions to be written
solutionNameRef : str
name of the solution whose timeSequence is chosen for the output
outputName : str
name of the file on disk where the solution is written
"""
writer = XDMFWriter(outputName)
writer.WriteProblemDataSolutions(mesh, problemData, solutionNameRef)
[docs]def WriteReducedOrderBases(mesh, problemData, reducedOrderBases, outputName):
"""
Functional API
Writes reducedOrderBases on disk satisfying the corresponding format
Parameters
----------
mesh : BasicToolsUnstructuredMesh
high-dimensional mesh
reducedCoordinates : dict
dictionary with time indices as keys and a np.ndarray of size
(numberOfModes,) containing the coefficients of the reduced solution
reducedOrderBasis : np.ndarray
of size (numberOfModes, numberOfDOFs)
outputName : str
name of the file on disk where the solution is written
"""
writer = XDMFWriter(outputName)
writer.WriteReducedOrderBases(mesh, problemData, reducedOrderBases)
[docs]class XDMFWriter(object):
"""
Class containing the PXDMF writer
"""
def __init__(self, outputName):
self.outputName = outputName
[docs] def WriteSolution(self, mesh, solution):
"""
Writes a solution on disk satisfying the corresponding format
Parameters
----------
mesh : BasicToolsUnstructuredMesh
high-dimensional mesh
solution : Solution
solution containing reducedCoordinates
"""
if MPI.COMM_WORLD.Get_size() > 1: # pragma: no cover
#ATTENTION: BasicTools Xdmf writer not supported in parallel
SIO.SaveState(self.outputName, solution)
else:
unstructuredMesh = FT.ConvertMeshToUnstructuredMesh(mesh)
writer = XW.XdmfWriter(self.outputName+'.xmf')
writer.SetTemporal(True)
writer.SetBinary(True)
writer.Open()
timeSequence = solution.GetTimeSequenceFromSnapshots()
for time in timeSequence:
pointFields = []
pointFieldsNames = []
cellFields = []
cellFieldsNames = []
data = solution.GetSnapshotAtTime(time).reshape(solution.GetNbeOfComponents(),-1)
name = solution.GetSolutionName()
if solution.GetPrimality() == True:
pointFields.append(data.T)
pointFieldsNames.append(name)
else:
dataAtCells = FT.IntegrationPointsToCellData(mesh, data)
cellFields.append(np.array(dataAtCells))
cellFieldsNames.append(name)
writer.Write(unstructuredMesh,PointFields=pointFields, PointFieldsNames=pointFieldsNames,
CellFields=cellFields, CellFieldsNames=cellFieldsNames, Time=time)
writer.Close()
[docs] def WriteProblemDataSolutions(self, mesh, problemData, solutionNameRef):
"""
Writes on disk the solutions defined in a ProblemData, satisfying the
corresponding format
Parameters
----------
mesh : BasicToolsUnstructuredMesh
high-dimensional mesh
problemData : ProblemData
ProblemData containing the solutions to be written
solutionNameRef : str
name of the solution whose timeSequence is chosen for the output
"""
if MPI.COMM_WORLD.Get_size() > 1: # pragma: no cover
#ATTENTION: BasicTools Xdmf writer not supported in parallel
SIO.SaveState("problemData_"+self.outputName, problemData)
else:
unstructuredMesh = FT.ConvertMeshToUnstructuredMesh(mesh)
writer = XW.XdmfWriter(self.outputName+'.xmf')
writer.SetTemporal(True)
writer.SetBinary(True)
writer.Open()
timeSequence = problemData.GetSolution(solutionNameRef).GetTimeSequenceFromSnapshots()
for time in timeSequence:
pointFields = []
pointFieldsNames = []
cellFields = []
cellFieldsNames = []
for solution in problemData.GetSolutions().values():
data = solution.GetSnapshotAtTime(time).reshape(solution.GetNbeOfComponents(),-1)
if solution.GetPrimality() == True:
pointFields.append(data.T)
pointFieldsNames.append(solution.GetSolutionName())
else:
dataAtCells = FT.IntegrationPointsToCellData(mesh, data)
cellFields.append(np.array(dataAtCells))
cellFieldsNames.append(solution.GetSolutionName())
writer.Write(unstructuredMesh,PointFields=pointFields, PointFieldsNames=pointFieldsNames,
CellFields=cellFields, CellFieldsNames=cellFieldsNames, Time=time)
writer.Close()
[docs] def WriteReducedOrderBases(self, mesh, problemData, reducedOrderBases):
"""
Writes reducedOrderBases on disk satisfying the corresponding format
Parameters
----------
mesh : BasicToolsUnstructuredMesh
high-dimensional mesh
reducedCoordinates : dict
dictionary with time indices as keys and a np.ndarray of size
(numberOfModes) containing the coefficients of the reduced solution
reducedOrderBasis : np.ndarray
of size (numberOfModes, numberOfDOFs)
"""
if MPI.COMM_WORLD.Get_size() > 1: # pragma: no cover
#ATTENTION: BasicTools Xdmf writer not supported in parallel
SIO.SaveState(self.outputName, reducedOrderBases)
else:
unstructuredMesh = FT.ConvertMeshToUnstructuredMesh(mesh)
writer = XW.XdmfWriter(self.outputName+'.xmf')
writer.SetTemporal(False)
writer.SetBinary(True)
writer.Open()
pointFields = []
pointFieldsNames = []
cellFields = []
cellFieldsNames = []
for solution in problemData.GetSolutions().values():
ROB = reducedOrderBases[solution.GetSolutionName()]
if solution.GetPrimality() == True:
for i in range(ROB.shape[0]):
vectROB = ROB[i].reshape(solution.GetNbeOfComponents(),-1).T
pointFields.append(vectROB)
pointFieldsNames.append("ROB_"+solution.GetSolutionName()+"_"+str(i))
else:
ROBAtCells = FT.IntegrationPointsToCellData(mesh, ROB)
for i in range(ROB.shape[0]):
cellFields.append(ROBAtCells[i])
cellFieldsNames.append("ROB_"+solution.GetSolutionName()+"_"+str(i))
writer.Write(unstructuredMesh,PointFields=pointFields, PointFieldsNames=pointFieldsNames,
CellFields=cellFields, CellFieldsNames=cellFieldsNames)
writer.Close()
if __name__ == "__main__":# pragma: no cover
from genericROM import RunTestFile
RunTestFile(__file__)