# -*- coding: utf-8 -*-
"""
@author: Rochlitz.R
"""
import dolfin as df
import os
from custEM.misc import logger_print as lp
import json
[docs]
class PreProcessing:
"""
PreProcessing class called from MOD instance.
Methods
-------
- import_all_results()
import results of an existing model - all quantities
- import_selected_results()
import results of an existing model - selected quantities
- load()
basic import function for data files
- read_h5()
utility import function for **h5** data files
"""
def __init__(self, FS, MP, import_freq=None, field_selection='all',
self_mode=False, fs_type='None'):
"""
Initalize instance for importing existing model solutions.
Required arguments
------------------
- FS, type class
FunctionSpaces instance
- MP, type class
ModelParameters instance
- load, type bool or int
specify if data should be imported or not by choosing a boolean
value, if an integer is specified, the result to the corresponding
transmitter number is imported
Keyword arguments
-----------------
- field_selection = 'all', type str
flag controling which quantities are tried to be imported, possible
choices are a combination of **E_t**, **E_s**, **H_t**, **H_s**,
**A_t** or **A_s** appended in one continuous string
- self_mode = False, type bool
choose between distributed or non-distributed import
- fs_type = None, type str
data can be stored either on Nedelec or Lagrange
VectorFunctionSpaces, specify either by 'ned' or 'cg',
which data (on which function space) should be imported
"""
self.FS = FS
self.MP = MP
self.self_mode = self_mode
self.fs_type = fs_type
self.import_freq = import_freq
self.full_name = MP.out_dir + '/' + MP.mod_name
config_file = (self.MP.r_dir + '/' + self.MP.approach + '/' +
self.MP.mesh_name + '/' + self.MP.mod_name +
'_config.json')
with open(config_file, 'r') as cfile:
parameters = json.load(cfile)
self.MP.n_freqs = parameters['n_freqs']
self.MP.n_tx = parameters['n_tx']
# no idea where this super strange bug comes from, but it works if
# we inititalize the function spaces here as dummy
dummy1 = df.Function(self.FS.V)
dummy2 = df.Function(self.FS.V_cg)
if field_selection == 'None':
return
elif field_selection == 'all':
self.import_all_results(MP.file_format)
else:
self.import_selected_results(
field_selection, MP.file_format)
[docs]
def import_all_results(self, file_format):
"""
Load all existing quantities from already calculated solutions.
Required arguments
------------------
- file_format, type str
either 'h5' or 'xml', set within model paramers in **MOD** instance
"""
lp(self.MP.logger, 20,
'... importing results from "' + self.full_name + '" ...',
pre_dash=False)
if self.MP.n_freqs != 1:
if self.import_freq is None:
r_str = 'real'
i_str = 'imag'
else:
r_str = 'f{:d}'.format(self.import_freq) + '_real'
i_str = 'f{:d}'.format(self.import_freq) + '_imag'
else:
r_str = 'real'
i_str = 'imag'
if self.fs_type == 'None' or 'ned' in self.fs_type.lower():
self.E_t_r = self.load('E_t', file_format, 2, ri=r_str)
self.E_t_i = self.load('E_t', file_format, 2, ri=i_str)
self.E_s_r = self.load('E_s', file_format, 2, ri=r_str)
self.E_s_i = self.load('E_s', file_format, 2, ri=i_str)
self.H_t_r = self.load('H_t', file_format, 2, ri=r_str)
self.H_t_i = self.load('H_t', file_format, 2, ri=i_str)
self.H_s_r = self.load('H_s', file_format, 2, ri=r_str)
self.H_s_i = self.load('H_s', file_format, 2, ri=i_str)
self.A_t_r = self.load('A_t', file_format, 2, ri=r_str)
self.A_t_i = self.load('A_t', file_format, 2, ri=i_str)
self.A_s_r = self.load('A_s', file_format, 2, ri=r_str)
self.A_s_i = self.load('A_s', file_format, 2, ri=i_str)
if self.fs_type == 'None' or 'cg' in self.fs_type.lower():
self.E_t_r_cg = self.load('E_t_cg', file_format, 1, ri=r_str)
self.E_t_i_cg = self.load('E_t_cg', file_format, 1, ri=i_str)
self.E_s_r_cg = self.load('E_s_cg', file_format, 1, ri=r_str)
self.E_s_i_cg = self.load('E_s_cg', file_format, 1, ri=i_str)
self.H_t_r_cg = self.load('H_t_cg', file_format, 1, ri=r_str)
self.H_t_i_cg = self.load('H_t_cg', file_format, 1, ri=i_str)
self.H_s_r_cg = sel,f.load('H_s_cg', file_format, 1, ri=r_str)
self.H_s_i_cg = self.load('H_s_cg', file_format, 1, ri=i_str)
self.A_t_r_cg = self.load('A_t_cg', file_format, 1, ri=r_str)
self.A_t_i_cg = self.load('A_t_cg', file_format, 1, ri=i_str)
self.A_s_r_cg = self.load('A_s_cg', file_format, 1, ri=r_str)
self.A_s_i_cg = self.load('A_s_cg', file_format, 1, ri=i_str)
[docs]
def import_selected_results(self, selection, file_format):
"""
Load selected quantities from already calculated solutions.
Required arguments:
-------------------
- selection, type str
string that must contain a combination of **E_t**, **E_s**,
**H_t**, **H_s**, **A_t** or **A_s** added to a continuous string
- file_format, type str
either 'h5' or 'xml', set within model paramers in **MOD** instance
"""
if not self.MP.self_mode:
lp(self.MP.logger, 20,
'... importing ' + str(selection) +
' from model "' + self.MP.mod_name + '" ...', pre_dash=False)
else:
lp(self.MP.logger, 20,
'... importing ' + str(selection) +
' from model "' + self.MP.mod_name + '" ...', pre_dash=False,
barrier=False, root_only=False)
if self.import_freq is None:
r_str = 'real'
i_str = 'imag'
else:
r_str = 'f{:d}'.format(self.import_freq) + '_real'
i_str = 'f{:d}'.format(self.import_freq) + '_imag'
if 'E_dc' in selection:
if self.fs_type == 'None' or 'ned' in self.fs_type.lower():
self.E_dc = self.load('E_dc', file_format, 2)
if 'B_static' in selection:
if self.fs_type == 'None' or 'cg' in self.fs_type.lower():
self.B_static = self.load('B_static', file_format, 1)
if 'E_t' in selection:
if self.fs_type == 'None' or 'ned' in self.fs_type.lower():
self.E_t_r = self.load('E_t', file_format, 2, ri=r_str)
self.E_t_i = self.load('E_t', file_format, 2, ri=i_str)
if self.fs_type == 'None' or 'cg' in self.fs_type.lower():
self.E_t_r_cg = self.load('E_t_cg', file_format, 1, ri=r_str)
self.E_t_i_cg = self.load('E_t_cg', file_format, 1, ri=i_str)
if 'E_s' in selection:
if self.fs_type == 'None' or 'ned' in self.fs_type.lower():
self.E_s_r = self.load('E_s', file_format, 2, ri=r_str)
self.E_s_i = self.load('E_s', file_format, 2, ri=i_str)
if self.fs_type == 'None' or 'cg' in self.fs_type.lower():
self.E_s_r_cg = self.load('E_s_cg', file_format, 1, ri=r_str)
self.E_s_i_cg = self.load('E_s_cg', file_format, 1, ri=i_str)
if 'H_t' in selection:
if self.fs_type == 'None' or 'ned' in self.fs_type.lower():
self.H_t_r = self.load('H_t', file_format, 2, ri=r_str)
self.H_t_i = self.load('H_t', file_format, 2, ri=i_str)
if self.fs_type == 'None' or 'cg' in self.fs_type.lower():
self.H_t_r_cg = self.load('H_t_cg', file_format, 1, ri=r_str)
self.H_t_i_cg = self.load('H_t_cg', file_format, 1, ri=i_str)
if 'H_s' in selection:
if self.fs_type == 'None' or 'ned' in self.fs_type.lower():
self.H_s_r = self.load('H_s', file_format, 2, ri=r_str)
self.H_s_i = self.load('H_s', file_format, 2, ri=i_str)
if self.fs_type == 'None' or 'cg' in self.fs_type.lower():
self.H_s_r_cg = self.load('H_s_cg', file_format, 1, ri=r_str)
self.H_s_i_cg = self.load('H_s_cg', file_format, 1, ri=i_str)
if 'A_t' in selection:
if self.fs_type == 'None' or 'ned' in self.fs_type.lower():
self.A_t_r = self.load('A_t', file_format, 2, ri=r_str)
self.A_t_i = self.load('A_t', file_format, 2, ri=i_str)
if self.fs_type == 'None' or 'cg' in self.fs_type.lower():
self.A_t_r_cg = self.load('A_t_cg', file_format, 1, ri=r_str)
self.A_t_i_cg = self.load('A_t_cg', file_format, 1, ri=i_str)
if 'A_s' in selection:
if self.fs_type == 'None' or 'ned' in self.fs_type.lower():
self.A_s_r = self.load('A_s', file_format, 2, ri=r_str)
self.A_s_i = self.load('A_s', file_format, 2, ri=i_str)
if self.fs_type == 'None' or 'cg' in self.fs_type.lower():
self.A_s_r_cg = self.load('A_s_cg', file_format, 1, ri=r_str)
self.A_s_i_cg = self.load('A_s_cg', file_format, 1, ri=i_str)
[docs]
def load(self, quantity, file_format, switch, ri='data'):
"""
Import data in either *xml* or *h5* format.
Required arguments
------------------
- quantitiy, type str
quantitiy, e.g., **E_t_real_cg**, which should be imported
- file_format, type str
either 'h5' or 'xml', set within model paramers in **MOD** instance
- switch, type int
internally used switch, either 1 to use Lagrange
VectorFunctionSpace or 2 for Nedelec space
Keyword arguments
-----------------
-ri = 'data', type str
specify a function to be imported from the HDF5 data container, if
it contains multiple results; used internally to distinguish
between *real* and *imag* fields
"""
if self.import_freq is None:
return([])
if file_format == 'h5':
if os.path.isfile(self.full_name + '_' + quantity + '.h5'):
return(self.read_h5(self.full_name + '_' + quantity + '.h5',
switch, ri))
else:
if not self.MP.self_mode:
lp(self.MP.logger, 30,
'Warning, model "' + self.full_name + '_' + quantity +
'.h5" could not be found! Continuing ...',
pre_dash=False)
else:
lp(self.MP.logger, 30,
'Warning, model "' + self.full_name + '_' + quantity +
'.h5" could not be found! Continuing ...',
pre_dash=False, barrier=False, root_only=False)
elif file_format == 'xml':
tmp_full_name = self.full_name
tmp_full_name += '_f_{:d}'.format(self.import_freq)
target_f = []
for ti in range(self.MP.n_tx):
if self.MP.n_tx != 1:
full_name = tmp_full_name + '_tx_{:d}'.format(ti)
if os.path.isfile(full_name + '_' + quantity +
'_' + ri[-4:] + '.xml') or \
os.path.isfile(full_name + '_' + quantity +
'_' + ri[-4:] + '_cg.xml'):
if switch == 1:
target_f.append(
df.Function(self.FS.V_cg, full_name + '_' +
quantity + '_' + ri[-4:] + '_cg.xml'))
if switch == 2:
target_f.append(
df.Function(self.FS.V, full_name + '_' +
quantity + '_' + ri[-4:] + '.xml'))
else:
lp(self.MP.logger, 30,
'Warning, model "' + full_name + '_' + quantity +
'.xml' + '_..." could not be found. Continuing ...')
return(target_f)
[docs]
def read_h5(self, f_name, switch, ri):
"""
Read data files in 'h5' format.
Required arguments
------------------
- fname, type str
file name to import from, containing also the export path
- switch, type int
internally used switch, either 1 to use Lagrange
VectorFunctionSpace or 2 for Nedelec space
-ri = 'data', type str
specify a function to be imported from the HDF5 data container, if
it contains multiple results; used internally to distinguish
between *real* and *imag* fields
"""
# crashes since 0.99.92 if no dummy function spaces are intialized
# in the init, don't know where this bug comes from, super strange
if switch == 1:
target_f = [df.Function(self.FS.V_cg) for t in range(self.MP.n_tx)]
elif switch == 2:
target_f = [df.Function(self.FS.V) for t in range(self.MP.n_tx)]
if self.self_mode:
f = df.HDF5File(self.MP.mpi_cs, f_name, "r")
else:
f = df.HDF5File(self.MP.mpi_cw, f_name, "r")
for ti in range(self.MP.n_tx):
f.read(target_f[ti], ri + "/vector_%d" % ti)
f.close()
return(target_f)