544 lines
25 KiB
Python
544 lines
25 KiB
Python
# -*- coding: UTF-8 no BOM -*-
|
|
|
|
# $Id$
|
|
|
|
import os,sys
|
|
import numpy as np
|
|
|
|
class ASCIItable():
|
|
"""Read and write to ASCII tables"""
|
|
|
|
__slots__ = ['__IO__',
|
|
'info',
|
|
'labeled',
|
|
'data',
|
|
]
|
|
|
|
tmpext = '_tmp' # filename extension for in-place access
|
|
|
|
# ------------------------------------------------------------------
|
|
def __init__(self,
|
|
name = None,
|
|
outname = None,
|
|
buffered = False, # flush writes
|
|
labeled = True, # assume table has labels
|
|
readonly = False, # no reading from file
|
|
):
|
|
self.__IO__ = {'output': [],
|
|
'buffered': buffered,
|
|
'labeled': labeled, # header contains labels
|
|
'labels': [], # labels according to file info
|
|
'readBuffer': [], # buffer to hold non-advancing reads
|
|
'dataStart': 0,
|
|
}
|
|
|
|
self.__IO__['inPlace'] = not outname and name and not readonly
|
|
if self.__IO__['inPlace']: outname = name + self.tmpext # transparently create tmp file
|
|
try:
|
|
self.__IO__['in'] = (open( name,'r') if os.access( name, os.R_OK) else None) if name else sys.stdin
|
|
except TypeError:
|
|
self.__IO__['in'] = name
|
|
|
|
try:
|
|
self.__IO__['out'] = (open(outname,'w') if (not os.path.isfile(outname) or
|
|
os.access( outname, os.W_OK)
|
|
) and
|
|
(not self.__IO__['inPlace'] or
|
|
not os.path.isfile(name) or
|
|
os.access( name, os.W_OK)
|
|
) else None) if outname else sys.stdout
|
|
except TypeError:
|
|
self.__IO__['out'] = outname
|
|
|
|
self.info = []
|
|
self.labels = []
|
|
self.data = []
|
|
self.line = ''
|
|
|
|
if self.__IO__['in'] is None \
|
|
or self.__IO__['out'] is None: raise IOError # complain if any required file access not possible
|
|
|
|
# ------------------------------------------------------------------
|
|
def _transliterateToFloat(self,
|
|
x):
|
|
try:
|
|
return float(x)
|
|
except:
|
|
return 0.0
|
|
|
|
# ------------------------------------------------------------------
|
|
def close(self,
|
|
dismiss = False):
|
|
self.input_close()
|
|
self.output_flush()
|
|
self.output_close(dismiss)
|
|
|
|
# ------------------------------------------------------------------
|
|
def input_close(self):
|
|
try:
|
|
if self.__IO__['in'] != sys.stdin: self.__IO__['in'].close()
|
|
except:
|
|
pass
|
|
|
|
# ------------------------------------------------------------------
|
|
def output_write(self,
|
|
what):
|
|
"""aggregate a single row (string) or list of (possibly containing further lists of) rows into output"""
|
|
if not isinstance(what, (str, unicode)):
|
|
try:
|
|
for item in what: self.output_write(item)
|
|
except:
|
|
self.__IO__['output'] += [str(what)]
|
|
else:
|
|
self.__IO__['output'] += [what]
|
|
|
|
return self.__IO__['buffered'] or self.output_flush()
|
|
|
|
# ------------------------------------------------------------------
|
|
def output_flush(self,
|
|
clear = True):
|
|
try:
|
|
self.__IO__['output'] == [] or self.__IO__['out'].write('\n'.join(self.__IO__['output']) + '\n')
|
|
except IOError:
|
|
return False
|
|
if clear: self.output_clear()
|
|
return True
|
|
|
|
# ------------------------------------------------------------------
|
|
def output_clear(self):
|
|
self.__IO__['output'] = []
|
|
|
|
# ------------------------------------------------------------------
|
|
def output_close(self,
|
|
dismiss = False):
|
|
try:
|
|
if self.__IO__['out'] != sys.stdout: self.__IO__['out'].close()
|
|
except:
|
|
pass
|
|
if dismiss and os.path.isfile(self.__IO__['out'].name):
|
|
os.remove(self.__IO__['out'].name)
|
|
elif self.__IO__['inPlace']:
|
|
os.rename(self.__IO__['out'].name, self.__IO__['out'].name[:-len(self.tmpext)])
|
|
|
|
# ------------------------------------------------------------------
|
|
def head_read(self):
|
|
"""
|
|
get column labels by either reading
|
|
|
|
the first row or, if keyword "head[*]" is present,
|
|
the last line of the header
|
|
"""
|
|
import re
|
|
|
|
try:
|
|
self.__IO__['in'].seek(0)
|
|
except:
|
|
pass
|
|
|
|
firstline = self.__IO__['in'].readline().strip()
|
|
m = re.search('(\d+)\s+head', firstline.lower()) # search for "head" keyword
|
|
|
|
if m: # proper ASCIItable format
|
|
|
|
if self.__IO__['labeled']: # table features labels
|
|
|
|
self.info = [self.__IO__['in'].readline().strip() for i in xrange(1,int(m.group(1)))]
|
|
self.labels = self.__IO__['in'].readline().split() # store labels found in last line
|
|
|
|
else:
|
|
|
|
self.info = [self.__IO__['in'].readline().strip() for i in xrange(0,int(m.group(1)))] # all header is info ...
|
|
|
|
else: # other table format
|
|
try:
|
|
self.__IO__['in'].seek(0) # try to rewind
|
|
except:
|
|
self.__IO__['readBuffer'] = [firstline] # or at least save data in buffer
|
|
|
|
while self.data_read(advance = False, respectLabels = False):
|
|
if self.line[0] in ['#','!','%','/','|','*','$']: # "typical" comment indicators
|
|
self.info_append(self.line) # store comment as info
|
|
self.data_read() # wind forward one line
|
|
else: break # last line of comments
|
|
|
|
if self.__IO__['labeled']: # table features labels
|
|
self.labels = self.data # get labels from last line in "header"...
|
|
self.data_read() # ...and remove from buffer
|
|
|
|
if self.__IO__['labeled']: # table features labels
|
|
self.__IO__['labels'] = list(self.labels) # backup labels (make COPY, not link)
|
|
|
|
try:
|
|
self.__IO__['dataStart'] = self.__IO__['in'].tell() # current file position is at start of data
|
|
except IOError:
|
|
pass
|
|
|
|
# ------------------------------------------------------------------
|
|
def head_write(self,
|
|
header = True):
|
|
"""write current header information (info + labels)"""
|
|
head = ['{}\theader'.format(len(self.info)+self.__IO__['labeled'])] if header else []
|
|
head.append(self.info)
|
|
if self.__IO__['labeled']: head.append('\t'.join(self.labels))
|
|
|
|
return self.output_write(head)
|
|
|
|
# ------------------------------------------------------------------
|
|
def head_getGeom(self):
|
|
"""interpret geom header"""
|
|
identifiers = {
|
|
'grid': ['a','b','c'],
|
|
'size': ['x','y','z'],
|
|
'origin': ['x','y','z'],
|
|
}
|
|
mappings = {
|
|
'grid': lambda x: int(x),
|
|
'size': lambda x: float(x),
|
|
'origin': lambda x: float(x),
|
|
'homogenization': lambda x: int(x),
|
|
'microstructures': lambda x: int(x),
|
|
}
|
|
info = {
|
|
'grid': np.zeros(3,'i'),
|
|
'size': np.zeros(3,'d'),
|
|
'origin': np.zeros(3,'d'),
|
|
'homogenization': 0,
|
|
'microstructures': 0,
|
|
}
|
|
extra_header = []
|
|
|
|
for header in self.info:
|
|
headitems = map(str.lower,header.split())
|
|
if len(headitems) == 0: continue # skip blank lines
|
|
if headitems[0] in mappings.keys():
|
|
if headitems[0] in identifiers.keys():
|
|
for i in xrange(len(identifiers[headitems[0]])):
|
|
info[headitems[0]][i] = \
|
|
mappings[headitems[0]](headitems[headitems.index(identifiers[headitems[0]][i])+1])
|
|
else:
|
|
info[headitems[0]] = mappings[headitems[0]](headitems[1])
|
|
else:
|
|
extra_header.append(header)
|
|
|
|
return info,extra_header
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
def head_putGeom(self,info):
|
|
"""translate geometry description to header"""
|
|
self.info_append([
|
|
"grid\ta {}\tb {}\tc {}".format(*info['grid']),
|
|
"size\tx {}\ty {}\tz {}".format(*info['size']),
|
|
"origin\tx {}\ty {}\tz {}".format(*info['origin']),
|
|
"homogenization\t{}".format(info['homogenization']),
|
|
"microstructures\t{}".format(info['microstructures']),
|
|
])
|
|
|
|
# ------------------------------------------------------------------
|
|
def labels_append(self,
|
|
what,
|
|
reset = False):
|
|
"""add item or list to existing set of labels (and switch on labeling)"""
|
|
if not isinstance(what, (str, unicode)):
|
|
try:
|
|
for item in what: self.labels_append(item)
|
|
except:
|
|
self.labels += [str(what)]
|
|
else:
|
|
self.labels += [what]
|
|
|
|
self.__IO__['labeled'] = True # switch on processing (in particular writing) of labels
|
|
if reset: self.__IO__['labels'] = list(self.labels) # subsequent data_read uses current labels as data size
|
|
|
|
# ------------------------------------------------------------------
|
|
def labels_clear(self):
|
|
"""delete existing labels and switch to no labeling"""
|
|
self.labels = []
|
|
self.__IO__['labeled'] = False
|
|
|
|
# ------------------------------------------------------------------
|
|
def label_index(self,
|
|
labels):
|
|
"""
|
|
tell index of column label(s).
|
|
|
|
return numpy array if asked for list of labels.
|
|
transparently deals with label positions implicitly given as numbers or their headings given as strings.
|
|
"""
|
|
from collections import Iterable
|
|
|
|
if isinstance(labels, Iterable) and not isinstance(labels, str): # check whether list of labels is requested
|
|
idx = []
|
|
for label in labels:
|
|
if label is not None:
|
|
try:
|
|
idx.append(int(label)-1) # column given as integer number?
|
|
except ValueError:
|
|
try:
|
|
idx.append(self.labels.index(label)) # locate string in label list
|
|
except ValueError:
|
|
try:
|
|
idx.append(self.labels.index('1_'+label)) # locate '1_'+string in label list
|
|
except ValueError:
|
|
idx.append(-1) # not found...
|
|
else:
|
|
try:
|
|
idx = int(labels)-1 # offset for python array indexing
|
|
except ValueError:
|
|
try:
|
|
idx = self.labels.index(labels)
|
|
except ValueError:
|
|
try:
|
|
idx = self.labels.index('1_'+labels) # locate '1_'+string in label list
|
|
except ValueError:
|
|
idx = None if labels is None else -1
|
|
|
|
return np.array(idx) if isinstance(idx,Iterable) else idx
|
|
|
|
# ------------------------------------------------------------------
|
|
def label_dimension(self,
|
|
labels):
|
|
"""
|
|
tell dimension (length) of column label(s).
|
|
|
|
return numpy array if asked for list of labels.
|
|
transparently deals with label positions implicitly given as numbers or their headings given as strings.
|
|
"""
|
|
from collections import Iterable
|
|
|
|
if isinstance(labels, Iterable) and not isinstance(labels, str): # check whether list of labels is requested
|
|
dim = []
|
|
for label in labels:
|
|
if label is not None:
|
|
myDim = -1
|
|
try: # column given as number?
|
|
idx = int(label)-1
|
|
myDim = 1 # if found has at least dimension 1
|
|
if self.labels[idx].startswith('1_'): # column has multidim indicator?
|
|
while idx+myDim < len(self.labels) and self.labels[idx+myDim].startswith("%i_"%(myDim+1)):
|
|
myDim += 1 # add while found
|
|
except ValueError: # column has string label
|
|
if label in self.labels: # can be directly found?
|
|
myDim = 1 # scalar by definition
|
|
elif '1_'+label in self.labels: # look for first entry of possible multidim object
|
|
idx = self.labels.index('1_'+label) # get starting column
|
|
myDim = 1 # (at least) one-dimensional
|
|
while idx+myDim < len(self.labels) and self.labels[idx+myDim].startswith("%i_"%(myDim+1)):
|
|
myDim += 1 # keep adding while going through object
|
|
|
|
dim.append(myDim)
|
|
else:
|
|
dim = -1 # assume invalid label
|
|
idx = -1
|
|
try: # column given as number?
|
|
idx = int(labels)-1
|
|
dim = 1 # if found has at least dimension 1
|
|
if self.labels[idx].startswith('1_'): # column has multidim indicator?
|
|
while idx+dim < len(self.labels) and self.labels[idx+dim].startswith("%i_"%(dim+1)):
|
|
dim += 1 # add as long as found
|
|
except ValueError: # column has string label
|
|
if labels in self.labels: # can be directly found?
|
|
dim = 1 # scalar by definition
|
|
elif '1_'+labels in self.labels: # look for first entry of possible multidim object
|
|
idx = self.labels.index('1_'+labels) # get starting column
|
|
dim = 1 # is (at least) one-dimensional
|
|
while idx+dim < len(self.labels) and self.labels[idx+dim].startswith("%i_"%(dim+1)):
|
|
dim += 1 # keep adding while going through object
|
|
|
|
return np.array(dim) if isinstance(dim,Iterable) else dim
|
|
|
|
# ------------------------------------------------------------------
|
|
def label_indexrange(self,
|
|
labels):
|
|
"""
|
|
tell index range for given label(s).
|
|
|
|
return numpy array if asked for list of labels.
|
|
transparently deals with label positions implicitly given as numbers or their headings given as strings.
|
|
"""
|
|
from collections import Iterable
|
|
|
|
start = self.label_index(labels)
|
|
dim = self.label_dimension(labels)
|
|
|
|
return map(lambda a,b: xrange(a,a+b), zip(start,dim)) if isinstance(labels, Iterable) and not isinstance(labels, str) \
|
|
else xrange(start,start+dim)
|
|
|
|
# ------------------------------------------------------------------
|
|
def info_append(self,
|
|
what):
|
|
"""add item or list to existing set of infos"""
|
|
if not isinstance(what, (str, unicode)):
|
|
try:
|
|
for item in what: self.info_append(item)
|
|
except:
|
|
self.info += [str(what)]
|
|
else:
|
|
self.info += [what]
|
|
|
|
# ------------------------------------------------------------------
|
|
def info_clear(self):
|
|
"""delete any info block"""
|
|
self.info = []
|
|
|
|
# ------------------------------------------------------------------
|
|
def data_rewind(self):
|
|
self.__IO__['in'].seek(self.__IO__['dataStart']) # position file to start of data section
|
|
self.__IO__['readBuffer'] = [] # delete any non-advancing data reads
|
|
self.labels = list(self.__IO__['labels']) # restore label info found in header (as COPY, not link)
|
|
self.__IO__['labeled'] = len(self.labels) > 0
|
|
|
|
# ------------------------------------------------------------------
|
|
def data_skipLines(self,
|
|
count):
|
|
"""wind forward by count number of lines"""
|
|
for i in xrange(count):
|
|
alive = self.data_read()
|
|
|
|
return alive
|
|
|
|
# ------------------------------------------------------------------
|
|
def data_read(self,
|
|
advance = True,
|
|
respectLabels = True):
|
|
"""read next line (possibly buffered) and parse it into data array"""
|
|
self.line = self.__IO__['readBuffer'].pop(0) if len(self.__IO__['readBuffer']) > 0 \
|
|
else self.__IO__['in'].readline().strip() # take buffered content or get next data row from file
|
|
|
|
if not advance:
|
|
self.__IO__['readBuffer'].append(self.line) # keep line just read in buffer
|
|
|
|
self.line = self.line.rstrip('\n')
|
|
|
|
if self.__IO__['labeled'] and respectLabels: # if table has labels
|
|
items = self.line.split()[:len(self.__IO__['labels'])] # use up to label count (from original file info)
|
|
self.data = items if len(items) == len(self.__IO__['labels']) else [] # take entries if label count matches
|
|
else:
|
|
self.data = self.line.split() # otherwise take all
|
|
|
|
return self.data != []
|
|
|
|
# ------------------------------------------------------------------
|
|
def data_readArray(self,
|
|
labels = []):
|
|
"""read whole data of all (given) labels as numpy array"""
|
|
from collections import Iterable
|
|
|
|
try:
|
|
self.data_rewind() # try to wind back to start of data
|
|
except:
|
|
pass # assume/hope we are at data start already...
|
|
|
|
if labels is None or labels == []:
|
|
use = None # use all columns (and keep labels intact)
|
|
labels_missing = []
|
|
else:
|
|
if isinstance(labels, str) or not isinstance(labels, Iterable): # check whether labels are a list or single item
|
|
labels = [labels]
|
|
indices = self.label_index(labels) # check requested labels ...
|
|
dimensions = self.label_dimension(labels) # ... and remember their dimension
|
|
present = np.where(indices >= 0)[0] # positions in request list of labels that are present ...
|
|
missing = np.where(indices < 0)[0] # ... and missing in table
|
|
labels_missing = np.array(labels)[missing] # labels of missing data
|
|
|
|
columns = []
|
|
for i,(c,d) in enumerate(zip(indices[present],dimensions[present])): # for all valid labels ...
|
|
# ... transparently add all components unless column referenced by number or with explicit dimension
|
|
columns += range(c,c + \
|
|
(d if str(c) != str(labels[present[i]]) else \
|
|
1))
|
|
use = np.array(columns)
|
|
|
|
self.labels = list(np.array(self.labels)[use]) # update labels with valid subset
|
|
|
|
self.data = np.loadtxt(self.__IO__['in'],usecols=use,ndmin=2)
|
|
|
|
return labels_missing
|
|
|
|
# ------------------------------------------------------------------
|
|
def data_write(self,
|
|
delimiter = '\t'):
|
|
"""write current data array and report alive output back"""
|
|
if len(self.data) == 0: return True
|
|
|
|
if isinstance(self.data[0],list):
|
|
return self.output_write([delimiter.join(map(str,items)) for items in self.data])
|
|
else:
|
|
return self.output_write(delimiter.join(map(str,self.data)))
|
|
|
|
# ------------------------------------------------------------------
|
|
def data_writeArray(self,
|
|
fmt = None,
|
|
delimiter = '\t'):
|
|
"""write whole numpy array data"""
|
|
for row in self.data:
|
|
try:
|
|
output = [fmt % value for value in row] if fmt else map(repr,row)
|
|
except:
|
|
output = [fmt % row] if fmt else [repr(row)]
|
|
|
|
self.__IO__['out'].write(delimiter.join(output) + '\n')
|
|
|
|
# ------------------------------------------------------------------
|
|
def data_append(self,
|
|
what):
|
|
if not isinstance(what, (str, unicode)):
|
|
try:
|
|
for item in what: self.data_append(item)
|
|
except:
|
|
self.data += [str(what)]
|
|
else:
|
|
self.data += [what]
|
|
|
|
# ------------------------------------------------------------------
|
|
def data_set(self,
|
|
what, where):
|
|
"""update data entry in column "where". grows data array if needed."""
|
|
idx = -1
|
|
try:
|
|
idx = self.label_index(where)
|
|
if len(self.data) <= idx:
|
|
self.data_append(['n/a' for i in xrange(idx+1-len(self.data))]) # grow data if too short
|
|
self.data[idx] = str(what)
|
|
except(ValueError):
|
|
pass
|
|
|
|
return idx
|
|
|
|
# ------------------------------------------------------------------
|
|
def data_clear(self):
|
|
self.data = []
|
|
|
|
# ------------------------------------------------------------------
|
|
def data_asFloat(self):
|
|
return map(self._transliterateToFloat,self.data)
|
|
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
def microstructure_read(self,
|
|
grid,
|
|
type = 'i',
|
|
strict = False):
|
|
"""read microstructure data (from .geom format)"""
|
|
def datatype(item):
|
|
return int(item) if type.lower() == 'i' else float(item)
|
|
|
|
N = grid.prod() # expected number of microstructure indices in data
|
|
microstructure = np.zeros(N,type) # initialize as flat array
|
|
|
|
i = 0
|
|
while i < N and self.data_read():
|
|
items = self.data
|
|
if len(items) > 2:
|
|
if items[1].lower() == 'of': items = np.ones(datatype(items[0]))*datatype(items[2])
|
|
elif items[1].lower() == 'to': items = np.arange(datatype(items[0]),1+datatype(items[2]))
|
|
else: items = map(datatype,items)
|
|
else: items = map(datatype,items)
|
|
|
|
s = min(len(items), N-i) # prevent overflow of microstructure array
|
|
microstructure[i:i+s] = items[:s]
|
|
i += len(items)
|
|
|
|
return (microstructure, i == N and not self.data_read()) if strict else microstructure # check for proper point count and end of file
|