from __future__ import print_function
from __future__ import absolute_import
import os.path as op
import shutil
import numpy as nm
from sfepy.base.base import get_default, Struct
from sfepy.homogenization.coefficients import Coefficients
from sfepy.homogenization.engine import HomogenizationEngine
from sfepy.applications import PDESolverApp
import sfepy.discrete.fem.periodic as per
import sfepy.linalg as la
import sfepy.base.multiproc as multi
import six
from six.moves import range
[docs]class HomogenizationApp(HomogenizationEngine):
[docs] @staticmethod
def process_options(options):
"""
Application options setup. Sets default values for missing
non-compulsory options.
"""
get = options.get
volume = get('volume', None)
volumes = get('volumes', None)
if volume is None and volumes is None:
raise ValueError('missing "volume" in options!')
return Struct(print_digits=get('print_digits', 3),
float_format=get('float_format', '%8.3e'),
coefs_filename=get('coefs_filename', 'coefs'),
tex_names=get('tex_names', None),
coefs=get('coefs', None, 'missing "coefs" in options!'),
requirements=get('requirements', None,
'missing "requirements" in options!'),
return_all=get('return_all', False),
mesh_update_variable=get('mesh_update_variable', None),
macro_data=get('macro_data', None),
micro_update=get('micro_update', {}),
n_micro=get('n_micro', None),
multiprocessing=get('multiprocessing', True),
use_mpi=get('use_mpi', False),
store_micro_idxs=get('store_micro_idxs', []),
volume=volume,
volumes=volumes)
def __init__(self, conf, options, output_prefix, **kwargs):
PDESolverApp.__init__(self, conf, options, output_prefix,
init_equations=False)
self.setup_options()
self.n_micro = kwargs.get('n_micro',
self.app_options.get('n_micro', None))
self.updating_corrs = None
self.micro_state_cache = {}
self.multiproc_mode = None
self.micro_states = None if self.n_micro is None else {}
# macroscopic data given in problem options dict.
macro_data = self.app_options.macro_data
if macro_data is not None:
self.n_micro = macro_data[list(macro_data.keys())[0]].shape[0]
self.setup_macro_data(macro_data)
if self.n_micro is not None:
for k in self.app_options.micro_update:
if not k == 'coors':
self.micro_states[k] = None
coors = self.problem.domain.get_mesh_coors()
c_sh = (self.n_micro,) + coors.shape
self.micro_states['coors'] = nm.empty(c_sh, dtype=nm.float64)
mac_ids = kwargs.get('mac_ids',
self.app_options.get('mac_ids', None))
self.micro_states['id'] = []
for im in range(self.n_micro):
self.micro_states['coors'][im] = coors
self.micro_states['id'].append(
mac_ids[im] if mac_ids is not None else im)
output_dir = self.problem.output_dir
if conf._filename is not None:
shutil.copyfile(conf._filename,
op.join(output_dir, op.basename(conf._filename)))
[docs] def setup_options(self):
PDESolverApp.setup_options(self)
po = HomogenizationApp.process_options
self.app_options += po(self.conf.options)
if hasattr(self, 'he'):
self.he.setup_options()
[docs] def setup_macro_data(self, data):
"""
Setup macroscopic deformation gradient.
"""
self.macro_data = data
self.problem.homogenization_macro_data = self.macro_data
[docs] def get_micro_cache_key(self, key, icoor, itime):
tt = '' if itime is None else '_t%03d' % itime
return '%s_%d%s' % (key, icoor, tt)
[docs] def update_micro_states(self):
"""
Update microstructures state according to the macroscopic data
and corrector functions.
"""
def calculate_local_update(state, corrs, var, macro_vals, mul=1):
for ic, corr in enumerate(corrs):
if state is None:
sh = corr.states[corr.components[0]][var].shape \
if hasattr(corr, 'states') else corr.state[var].shape
state = nm.zeros((len(corrs),) + sh, dtype=nm.float64)
else:
sh = state[ic].shape
if hasattr(corr, 'states'):
corr_arr = nm.array(
[corr.states[jj][var] for jj in corr.components]).T
mval = macro_vals[ic].reshape((corr_arr.shape[1], 1))
state[ic] += mul * nm.dot(corr_arr, mval).reshape(sh)
else:
if macro_vals is None:
state[ic] += mul * corr.state[var].reshape(sh)
else:
state[ic] += mul\
* (corr.state[var] * macro_vals[ic]).reshape(sh)
return state
micro_update = self.app_options.micro_update
for key, upd_obj in six.iteritems(micro_update):
if '_prev' in key:
continue
state = self.micro_states[key]
if key + '_prev' in micro_update and state is not None:
self.micro_states[key + '_prev'] = state.copy()
if key == 'coors':
if hasattr(upd_obj, '__call__'):
upd_obj(state, self.macro_data, self.problem)
else:
# macro strain - in the first sequence of the list
mtx_e = self.macro_data[upd_obj[0][2]]
state += la.dot_sequences(state, mtx_e, 'ABT')
if hasattr(upd_obj, '__call__'):
upd_obj(state, self.macro_data, self.problem)
else:
if self.updating_corrs is not None:
for v in upd_obj:
if len(v) == 4:
cname, vname, mname, mul = v
else:
cname, vname, mname = v
mul = 1
macro_data = None if mname is None \
else self.macro_data[mname]
if cname is not None:
state0 = calculate_local_update(
state, self.updating_corrs[cname],
vname, macro_data, mul)
else:
state += macro_data[..., 0]
if state0 is not state:
self.micro_states[key] = state0
state = state0
[docs] def call(self, verbose=False, ret_all=None, itime=None, iiter=None):
"""
Call the homogenization engine and compute the homogenized
coefficients.
Parameters
----------
verbose : bool
If True, print the computed coefficients.
ret_all : bool or None
If not None, it can be used to override the 'return_all' option.
If True, also the dependencies are returned.
time_tag: str
The time tag used in file names.
Returns
-------
coefs : Coefficients instance
The homogenized coefficients.
dependencies : dict
The dependencies, if `ret_all` is True.
"""
opts = self.app_options
ret_all = get_default(ret_all, opts.return_all)
force_init_he = hasattr(self.problem, 'force_init_he')\
and self.problem.force_init_he
if not hasattr(self, 'he') or force_init_he:
volumes = {}
if hasattr(opts, 'volumes') and (opts.volumes is not None):
volumes.update(opts.volumes)
elif hasattr(opts, 'volume') and (opts.volume is not None):
volumes['total'] = opts.volume
else:
volumes['total'] = 1.0
self.he = HomogenizationEngine(self.problem, self.options,
volumes=volumes)
if self.micro_states is not None:
self.update_micro_states()
self.he.set_micro_states(self.micro_states)
multiproc_mode = None
if opts.multiprocessing and multi.use_multiprocessing:
multiproc, multiproc_mode = multi.get_multiproc(mpi=opts.use_mpi)
if multiproc_mode is not None:
upd_var = self.app_options.mesh_update_variable
if upd_var is not None:
uvar = self.problem.create_variables([upd_var])[upd_var]
uvar.field.mappings0 = multiproc.get_dict('mappings0',
soft_set=True)
per.periodic_cache = multiproc.get_dict('periodic_cache',
soft_set=True)
time_tag = ('' if itime is None else '_t%03d' % itime)\
+ ('' if iiter is None else '_i%03d' % iiter)
aux = self.he(ret_all=ret_all, time_tag=time_tag)
if ret_all:
coefs, dependencies = aux
# store correctors for coors update
self.updating_corrs = {}
for upd_obj in six.itervalues(opts.micro_update):
if upd_obj is not None and not hasattr(upd_obj, '__call__'):
for v in upd_obj:
cr = v[0]
if cr is not None:
self.updating_corrs[cr] = dependencies[cr]
else:
coefs = aux
if coefs is not None:
coefs = Coefficients(**coefs.to_dict())
if verbose:
prec = nm.get_printoptions()['precision']
if hasattr(opts, 'print_digits'):
nm.set_printoptions(precision=opts.print_digits)
print(coefs)
nm.set_printoptions(precision=prec)
ms_cache = self.micro_state_cache
for ii in self.app_options.store_micro_idxs:
for k in self.micro_states.keys():
key = self.get_micro_cache_key(k, ii, itime)
ms_cache[key] = self.micro_states[k][ii]
coef_save_name = op.join(opts.output_dir, opts.coefs_filename)
coefs.to_file_hdf5(coef_save_name + '%s.h5' % time_tag)
coefs.to_file_txt(coef_save_name + '%s.txt' % time_tag,
opts.tex_names,
opts.float_format)
if ret_all:
return coefs, dependencies
else:
return coefs