# -*- coding: UTF-8 no BOM -*- import h5py import re import numpy as np from queue import Queue from . import util # ------------------------------------------------------------------ class DADF5(): """Read and write to DADF5 files""" # ------------------------------------------------------------------ def __init__(self, filename, mode = 'r', ): if mode not in ['a','r']: print('Invalid file access mode') else: with h5py.File(filename,mode): pass with h5py.File(filename,'r') as f: if f.attrs['DADF5-major'] != 0 or f.attrs['DADF5-minor'] != 2: raise TypeError('Unsupported DADF5 version {} '.format(f.attrs['DADF5-version'])) self.structured = 'grid' in f['geometry'].attrs.keys() if self.structured: self.grid = f['geometry'].attrs['grid'] self.size = f['geometry'].attrs['size'] r=re.compile('inc[0-9]+') self.increments = [{'inc': int(u[3:]), 'time': round(f[u].attrs['time/s'],12), } for u in f.keys() if r.match(u)] self.constituents = np.unique(f['mapping/cellResults/constituent']['Name']).tolist() # ToDo: I am not to happy with the name self.constituents = [c.decode() for c in self.constituents] self.materialpoints = np.unique(f['mapping/cellResults/materialpoint']['Name']).tolist() # ToDo: I am not to happy with the name self.materialpoints = [m.decode() for m in self.materialpoints] self.Nconstituents = [i for i in range(np.shape(f['mapping/cellResults/constituent'])[1])] self.Nmaterialpoints = np.shape(f['mapping/cellResults/constituent'])[0] self.c_output_types = [] for c in self.constituents: for o in f['inc{:05}/constituent/{}'.format(self.increments[0]['inc'],c)].keys(): self.c_output_types.append(o) self.c_output_types = list(set(self.c_output_types)) # make unique self.m_output_types = [] for m in self.materialpoints: for o in f['inc{:05}/materialpoint/{}'.format(self.increments[0]['inc'],m)].keys(): self.m_output_types.append(o) self.m_output_types = list(set(self.m_output_types)) # make unique self.active= {'increments': self.increments, 'constituents': self.constituents, 'materialpoints': self.materialpoints, 'constituent': self.Nconstituents, 'c_output_types': self.c_output_types, 'm_output_types': self.m_output_types} self.filename = filename self.mode = mode def get_candidates(self,l): groups = [] if type(l) is not list: print('mist') with h5py.File(self.filename,'r') as f: for g in self.get_active_groups(): if set(l).issubset(f[g].keys()): groups.append(g) return groups def get_active_groups(self): groups = [] for i,x in enumerate(self.active['increments']): group_inc = 'inc{:05}'.format(self.active['increments'][i]['inc']) for c in self.active['constituents']: group_constituent = group_inc+'/constituent/'+c for t in self.active['c_output_types']: group_output_types = group_constituent+'/'+t groups.append(group_output_types) for m in self.active['materialpoints']: group_materialpoint = group_inc+'/materialpoint/'+m for t in self.active['m_output_types']: group_output_types = group_materialpoint+'/'+t groups.append(group_output_types) return groups def list_data(self): """Shows information on all datasets in the file""" with h5py.File(self.filename,'r') as f: group_inc = 'inc{:05}'.format(self.active['increments'][0]['inc']) for c in self.active['constituents']: print('\n'+c) group_constituent = group_inc+'/constituent/'+c for t in self.active['c_output_types']: print(' {}'.format(t)) group_output_types = group_constituent+'/'+t try: for x in f[group_output_types].keys(): print(' {} ({})'.format(x,f[group_output_types+'/'+x].attrs['Description'].decode())) except: pass for m in self.active['materialpoints']: group_materialpoint = group_inc+'/materialpoint/'+m for t in self.active['m_output_types']: print(' {}'.format(t)) group_output_types = group_materialpoint+'/'+t try: for x in f[group_output_types].keys(): print(' {} ({})'.format(x,f[group_output_types+'/'+x].attrs['Description'].decode())) except: pass def get_dataset_location(self,label): """Returns the location of all active datasets with given label""" path = [] with h5py.File(self.filename,'r') as f: for i in self.active['increments']: group_inc = 'inc{:05}'.format(i['inc']) for c in self.active['constituents']: group_constituent = group_inc+'/constituent/'+c for t in self.active['c_output_types']: try: f[group_constituent+'/'+t+'/'+label] path.append(group_constituent+'/'+t+'/'+label) except Exception as e: print('unable to locate constituents dataset: '+ str(e)) for m in self.active['materialpoints']: group_materialpoint = group_inc+'/materialpoint/'+m for t in self.active['m_output_types']: try: f[group_materialpoint+'/'+t+'/'+label] path.append(group_materialpoint+'/'+t+'/'+label) except Exception as e: print('unable to locate materialpoints dataset: '+ str(e)) return path def read_dataset(self,path,c): """ Dataset for all points/cells If more than one path is given, the dataset is composed of the individual contributions """ with h5py.File(self.filename,'r') as f: shape = (self.Nmaterialpoints,) + np.shape(f[path[0]])[1:] if len(shape) == 1: shape = shape +(1,) dataset = np.full(shape,np.nan) for pa in path: label = pa.split('/')[2] try: p = np.where(f['mapping/cellResults/constituent'][:,c]['Name'] == str.encode(label))[0] u = (f['mapping/cellResults/constituent'][p,c]['Position']) a = np.array(f[pa]) if len(a.shape) == 1: a=a.reshape([a.shape[0],1]) dataset[p,:] = a[u,:] except Exception as e: print('unable to read constituent: '+ str(e)) try: p = np.where(f['mapping/cellResults/materialpoint']['Name'] == str.encode(label))[0] u = (f['mapping/cellResults/materialpoint'][p.tolist()]['Position']) a = np.array(f[pa]) if len(a.shape) == 1: a=a.reshape([a.shape[0],1]) dataset[p,:] = a[u,:] except Exception as e: print('unable to read materialpoint: '+ str(e)) return dataset def add_Cauchy(self,P='P',F='F'): """Adds Cauchy stress calculated from 1st Piola-Kirchhoff stress and deformation gradient""" def Cauchy(F,P): return 1.0/np.linalg.det(F)*np.dot(P,F.T) args = [{'label':F,'shape':[3,3],'unit':'-'}, {'label':P,'shape':[3,3],'unit':'Pa'} ] result = {'label':'sigma', 'unit':'Pa', 'Description': 'Cauchy stress calculated from 1st Piola-Kirchhoff stress and deformation gradient'} self.add_generic_pointwise(Cauchy,args,result) def add_Mises_stress(self,stress='sigma'): """Adds equivalent von Mises stress""" def Mises_stress(stress): dev = stress - np.trace(stress)/3.0*np.eye(3) symdev = 0.5*(dev+dev.T) return np.sqrt(np.sum(symdev*symdev.T)*3.0/2.0) args = [{'label':stress,'shape':[3,3],'unit':'Pa'}] result = {'label':'Mises({})'.format(stress), 'unit':'Pa', 'Description': 'Equivalent Mises stress'} self.add_generic_pointwise(Mises_stress,args,result) def add_norm(self,x,ord=None): """Adds norm of vector or tensor or magnitude of a scalar.""" # ToDo: The output unit should be the input unit args = [{'label':x,'shape':None,'unit':None}] result = {'label':'norm_{}({})'.format(str(ord),x), 'unit':'n/a', 'Description': 'Norm of vector or tensor or magnitude of a scalar. See numpy.linalg.norm manual for details'} self.add_generic_pointwise(np.linalg.norm,args,result) def add_determinant(self,a): """Adds the determinant of a tensor""" # ToDo: The output unit should be the input unit args = [{'label':a,'shape':[3,3],'unit':None}] result = {'label':'det({})'.format(a), 'unit':'n/a', 'Description': 'Determinant of a tensor'} self.add_generic_pointwise(np.linalg.det,args,result) def add_strain_tensors(self,defgrad='F'): """Adds a strain definition""" def strain(defgrad): (U,S,Vh) = np.linalg.svd(defgrad) # singular value decomposition R_inv = np.linalg.inv(np.dot(U,Vh)) # inverse rotation of polar decomposition U = np.dot(R_inv,defgrad) # F = RU U = np.where(abs(U) < 1e-12, 0, U) # kill nasty noisy data (D,V) = np.linalg.eig(U) # eigen decomposition (of symmetric matrix) neg = np.where(D < 0.0) # find negative eigenvalues ... D[neg] *= -1. # ... flip value ... V[:,neg] *= -1. # ... and vector for i,eigval in enumerate(D): if np.dot(V[:,i],V[:,(i+1)%3]) != 0.0: # check each vector for orthogonality V[:,(i+1)%3] = np.cross(V[:,(i+2)%3],V[:,i]) # correct next vector V[:,(i+1)%3] /= np.sqrt(np.dot(V[:,(i+1)%3],V[:,(i+1)%3].conj())) # and renormalize (hyperphobic?) d = np.log(D) # operate on eigenvalues of U o r V return np.dot(V,np.dot(np.diag(d),V.T)).real # build tensor back from eigenvalue/vector basis # ToDo: The output unit should be the input unit args = [{'label':defgrad,'shape':[3,3],'unit':None}] result = {'label':'strain({})'.format(defgrad), 'unit':'-', 'Description': 'strain (ln(V)) of a deformation gradient'} self.add_generic_pointwise(strain,args,result) def get_fitting(self,data): groups = [] if type(data) is not list: print('mist') with h5py.File(self.filename,'r') as f: for g in self.get_candidates([l['label'] for l in data]): print(g) fits = True for d in data: # ToDo: check for unit if d['shape'] is not None: fits = fits and np.all(np.array(f[g+'/'+d['label']].shape[1:]) == np.array(d['shape'])) if fits: groups.append(g) return groups def add_generic_pointwise(self,func,args,result): """ Ggeneral function to add pointwise data function 'func' first needs to have data arguments before other arguments """ groups = self.get_fitting(args) def job(args): out = args['out'] datasets_in = args['dat'] func = args['fun'] # calling the function per point might be performance-wise not optimal # could be worth to investigate the performance for vectorized add_XXX functions that do the # loops internally for i in range(out.shape[0]): arg = tuple([d[i,] for d in datasets_in]) out[i,] = func(*arg) args['results'].put({'out':out,'group':args['group']}) Nthreads = 4 # ToDo: should be a parameter results = Queue(Nthreads+1) todo = [] for g in groups: with h5py.File(self.filename,'r') as f: datasets_in = [f[g+'/'+u['label']][()] for u in args] # figure out dimension of results testArg = tuple([d[0,] for d in datasets_in]) # to call function with first point out = np.empty([datasets_in[0].shape[0]] + list(func(*testArg).shape)) # shape is Npoints x shape of the results for one point todo.append({'dat':datasets_in,'fun':func,'out':out,'group':g,'results':results}) # Instantiate a thread pool with worker threads pool = util.ThreadPool(Nthreads) missingResults = len(todo) # Add the jobs in bulk to the thread pool. Alternatively you could use # `pool.add_task` to add single jobs. The code will block here, which # makes it possible to cancel the thread pool with an exception when # the currently running batch of workers is finishnumpy.linalg.normed. pool.map(job, todo[:Nthreads+1]) i = 0 while missingResults > 0: r=results.get() # noqa print(r['group']) with h5py.File(self.filename,'r+') as f: dataset_out = f[r['group']].create_dataset(result['label'],data=r['out']) dataset_out.attrs['Unit'] = result['unit'] dataset_out.attrs['Description'] = result['Description'] dataset_out.attrs['Creator'] = 'dadf5.py v{}'.format('n/a') missingResults-=1 try: pool.add_task(job,todo[Nthreads+1+i]) except: pass i+=1 pool.wait_completion()