# SPDX-License-Identifier: AGPL-3.0-or-later
import copy
import re
from os import PathLike
from io import TextIOBase
from pathlib import Path
from typing import Iterable, Mapping, Sequence, Union, TextIO, BinaryIO, Any, cast
import numpy as np
import pandas as pd
from . import util
from ._typehints import FileHandleText, FileHandleBinary
[docs]
class Table:
"""Manipulate multi-dimensional spreadsheet-like data."""
def __init__(self,
shapes: Mapping[str,Union[Union[int,np.integer],tuple[Union[int,np.integer],...]]] = {},
data: np.ndarray | pd.DataFrame | None = None,
comments: str | Iterable[str] | None = None,
dtypes: Mapping[str,str | np.dtype] | None = None):
"""
New spreadsheet.
Parameters
----------
shapes : dict with str:tuple pairs, optional
Shapes of the data columns. Mandatory if 'data' is given.
For instance, 'F':(3,3) for a deformation gradient, or 'r':(1,) for a scalar.
data : numpy.ndarray or pandas.DataFrame, optional
Data. Existing column labels of a pandas.DataFrame will be replaced.
comments : (iterable of) str, optional
Additional, human-readable information.
dtypes : dict, optional
Data type of each column.
If given, overwrites existing data types inferred from numpy.ndarray or pandas.DataFrame.
"""
self.comments = [] if comments is None else \
[comments] if isinstance(comments,str) else \
[str(c) for c in comments]
self.shapes = { k:(v,) if isinstance(v,(np.integer,int)) else v for k,v in shapes.items() }
self.data = pd.DataFrame(data=data)
self._relabel('uniform')
if dtypes is not None:
self.data = self.data.astype({k:v for k,v in dtypes.items()
if k in set(self.shapes.keys())})
def __repr__(self) -> str:
"""
Return repr(self).
Give short, human-readable summary.
"""
self._relabel('shapes')
data_repr = self.data.__repr__()
self._relabel('uniform')
return '\n'.join(['# '+c for c in self.comments])+'\n'+data_repr
def __eq__(self,
other: object) -> bool:
"""
Return self==other.
Test equality of other.
Parameters
----------
other : Table
Table to check for equality.
Returns
-------
equal : bool
Whether both arguments are equal.
"""
return NotImplemented if not isinstance(other,Table) else \
self.shapes == other.shapes and self.data.equals(other.data)
def __getitem__(self,
item) -> 'Table':
"""
Return self[item].
Return slice according to item.
Parameters
----------
item : row and/or column indexer
Slice to select from Table.
Returns
-------
slice : damask.Table
Sliced part of the Table.
Examples
--------
>>> import damask
>>> import numpy as np
>>> tbl = damask.Table(shapes=dict(colA=(1,),colB=(1,),colC=(1,)),
... data=np.arange(12).reshape((4,3)))
>>> tbl['colA','colB']
colA colB
0 0 1
1 3 4
2 6 7
3 9 10
>>> tbl[3]
colA colB colC
3 9 10 11
>>> tbl[::2,['colB','colA']]
colB colA
0 1 0
2 7 6
>>> tbl[[True,False,False,True],'colB']
colB
0 1
3 10
"""
item_ = (item,slice(None,None,None)) if isinstance(item,(slice,np.ndarray)) else \
(np.array(item),slice(None,None,None)) if isinstance(item,list) and np.array(item).dtype == np.bool_ else \
(slice(int(item),int(item)),slice(None,None,None)) if isinstance(item,(int,np.integer)) else \
(np.array(item[0]),item[1]) if isinstance(item[0],list) else \
(item[0], item[1]) if isinstance(item[0],(slice,np.ndarray)) else \
(slice(None,None,None),item)
sliced = self.data.loc[item_]
cols = np.array(sliced.columns if isinstance(sliced,pd.DataFrame) else [item_[1]])
_,idx = np.unique(cols,return_index=True)
return self.__class__(shapes={k:self.shapes[k] for k in cols[np.sort(idx)]},
data=sliced,
comments=self.comments)
def __len__(self) -> int:
"""
Return len(self).
Number of rows.
"""
return len(self.data)
def __copy__(self) -> 'Table':
"""
Return deepcopy(self).
Create deep copy.
"""
return copy.deepcopy(self)
copy = __copy__
def _label(self,
what: str | Sequence[str],
how: str) -> list[str]:
"""
Expand labels according to data shape.
Parameters
----------
what : (sequence of) str
Labels to expand.
how : {'uniform, 'shapes', 'linear'}
Mode of labeling.
'uniform' ==> v v v
'shapes' ==> 3:v v v
'linear' ==> 1_v 2_v 3_v
'export' ==> 3x3:1_t 3x3:2_t 3x3:3_t 3x3:4_t 3x3:5_t 3x3:6_t 3x3:7_t 3x3:8_t 3x3:9_t
"""
what = [what] if isinstance(what,str) else what
labels = []
for label in what:
shape = self.shapes[label]
size = np.prod(shape,dtype=np.int64) # type: ignore
if how == 'uniform':
labels += [label] * size
elif how == 'shapes':
labels += [('' if size == 1 or i>0 else f'{util.srepr(shape,"x")}:')+label for i in range(size)]
elif how == 'linear':
labels += [('' if size == 1 else f'{i+1}_')+label for i in range(size)]
elif how == 'export':
labels += [ ('' if len(shape)<2 else f'{util.srepr(shape,"x")}:')
+('' if size == 1 else f'{i+1}_')
+label for i in range(size)]
else:
raise KeyError
return labels
def _relabel(self,
how: str):
"""
Modify labeling of data in-place.
Parameters
----------
how : {'uniform, 'shapes', 'linear'}
Mode of labeling.
'uniform' ==> v v v
'shapes' ==> 3:v v v
'linear' ==> 1_v 2_v 3_v
'export' ==> 3x3:1_t 3x3:2_t 3x3:3_t 3x3:4_t 3x3:5_t 3x3:6_t 3x3:7_t 3x3:8_t 3x3:9_t
"""
self.data.columns = self._label(self.shapes,how) # type: ignore
[docs]
def isclose(self,
other: 'Table',
rtol: float = 1e-5,
atol: float = 1e-8,
equal_nan: bool = True) -> np.ndarray:
"""
Report where values are approximately equal to corresponding ones of other Table.
Parameters
----------
other : damask.Table
Table to compare against.
rtol : float, optional
Relative tolerance of equality.
atol : float, optional
Absolute tolerance of equality.
equal_nan : bool, optional
Consider matching NaN values as equal. Defaults to True.
Returns
-------
mask : numpy.ndarray of bool
Mask indicating where corresponding table values are close.
"""
return np.isclose( self.data.to_numpy(),
other.data.to_numpy(),
rtol=rtol,
atol=atol,
equal_nan=equal_nan)
[docs]
def allclose(self,
other: 'Table',
rtol: float = 1e-5,
atol: float = 1e-8,
equal_nan: bool = True) -> bool:
"""
Test whether all values are approximately equal to corresponding ones of other Table.
Parameters
----------
other : damask.Table
Table to compare against.
rtol : float, optional
Relative tolerance of equality.
atol : float, optional
Absolute tolerance of equality.
equal_nan : bool, optional
Consider matching NaN values as equal. Defaults to True.
Returns
-------
answer : bool
Whether corresponding values are close between both tables.
"""
return np.allclose( self.data.to_numpy(),
other.data.to_numpy(),
rtol=rtol,
atol=atol,
equal_nan=equal_nan)
[docs]
@staticmethod
def load(fname: FileHandleText) -> 'Table':
"""
Load a table from text table or NumPy archive format.
For path-like inputs, a case-insensitive ``.npz`` suffix selects
NumPy archive format; all other suffixes select text table format.
Open binary streams are interpreted as NumPy archives, while open
text streams are interpreted as text tables.
Text tables contain whitespace-separated fields. Leading comment
lines begin with ``#``, and the first non-comment line contains the
column labels.
Component labels encode the shape of vector and tensor data:
- Vector components are labeled ``1_v, 2_v, ..., n_v``.
- Tensor components are labeled ``3x3:1_T, 3x3:2_T, ..., 3x3:9_T``.
Parameters
----------
fname : str, path-like, or file-like object
Input path or open stream. Text streams must contain a text
table, and binary streams must contain a NumPy archive.
Returns
-------
table : damask.Table
Loaded table.
"""
def infer_shapes(labels: list[str]) -> dict[str, tuple[int, ...]]:
shapes: dict[str, tuple[int, ...]] = {}
for label in labels:
if (tensor_column := re.search(r'[0-9,x]*?:[0-9]*?_', label)) is not None:
shape = tensor_column.group().split(':', 1)[0].split('x')
shapes[label.split('_', 1)[1]] = tuple(int(d) for d in shape)
elif re.match(r'[0-9]*?_', label) is not None:
shapes[label.split('_',1)[1]] = (int(label.split('_',1)[0]),)
else:
shapes[label] = (1,)
return shapes
def load_npz(source: FileHandleBinary) -> 'Table':
if not isinstance(source, (str, PathLike)):
source.seek(0)
if not isinstance(loaded := np.load(source), np.lib.npyio.NpzFile):
raise ValueError('expected a NumPy .npz archive, not a .npy array')
with loaded as content:
return Table(
shapes=infer_shapes(content.files),
data=np.column_stack([content[name] for name in content.files]) if content.files else np.empty((0, 0)),
)
def load_text(stream: TextIO) -> 'Table':
stream.seek(0)
comments: list[str] = []
while (line := stream.readline().strip()).startswith('#'):
comments.append(line.removeprefix('#').strip())
labels = line.split()
return Table(
shapes=infer_shapes(labels),
data=pd.read_csv(stream,names=list(range(len(labels))),sep=r'\s+'),
comments=comments,
)
if isinstance(fname, (str, PathLike)):
if Path(fname).suffix.casefold() == '.npz':
return load_npz(fname)
with util.open_text(fname) as stream:
return load_text(stream)
if isinstance(fname, TextIOBase):
return load_text(cast(TextIO,fname))
return load_npz(cast(BinaryIO,fname))
[docs]
@staticmethod
def load_ang(fname: FileHandleText,
shapes = {'eu':3,
'pos':2,
'IQ':1,
'CI':1,
'ID':1,
'intensity':1,
'fit':1}) -> 'Table':
"""
Load from ANG file.
Regular ANG files feature the following columns:
- Euler angles (Bunge notation) in radians, 3 floats, label 'eu'.
- Spatial position in meters, 2 floats, label 'pos'.
- Image quality, 1 float, label 'IQ'.
- Confidence index, 1 float, label 'CI'.
- Phase ID, 1 int, label 'ID'.
- SEM signal, 1 float, label 'intensity'.
- Fit, 1 float, label 'fit'.
Parameters
----------
fname : file, str, or pathlib.Path
Filename or file to read.
shapes : dict with str:int pairs, optional
Column labels and their width.
Defaults to standard TSL ANG format.
Returns
-------
loaded : damask.Table
Table data from file.
"""
with util.open_text(fname) as f:
f.seek(0)
content = f.readlines()
comments = [util.execution_stamp('Table','from_ang')]
for i,line in enumerate(content):
if line.startswith('#'):
comments.append(line.split('#',1)[1].strip())
else:
break
data = np.loadtxt(content[i:])
if (remainder := data.shape[1]-sum(shapes.values())) > 0:
shapes['unknown'] = remainder
return Table(shapes,data,comments)
@property
def labels(self) -> list[str]:
return list(self.shapes)
[docs]
def get(self,
label: str) -> np.ndarray:
"""
Get column data.
Parameters
----------
label : str
Column label.
Returns
-------
data : numpy.ndarray
Array of column data.
"""
data = self.data[label].to_numpy().reshape((-1,)+self.shapes[label])
return data.astype(type(data.flatten()[0]))
[docs]
def set(self,
label: str,
data: np.ndarray,
info: str | None = None) -> 'Table':
"""
Add new or replace existing column data.
Parameters
----------
label : str
Column label.
data : numpy.ndarray
Column data. First dimension needs to match number of rows.
info : str, optional
Human-readable information about the data.
Returns
-------
updated : damask.Table
Updated table.
"""
def add_comment(label: str, shape: tuple[int, ...],info: str) -> list[str]:
specific = f'{label}{" "+str(shape) if np.prod(shape,dtype=np.int64) > 1 else ""}: {info}'
general = util.execution_stamp('Table')
return [f'{specific} / {general}']
dup = self.copy()
if info is not None: dup.comments += add_comment(label,data.shape[1:],info)
key = m.group(1) if (m := re.match(r'(.*)\[((\d+,)*(\d+))\]',label)) else label
if key in dup.shapes:
if m:
idx = np.ravel_multi_index(tuple(map(int,m.group(2).split(","))),
self.shapes[key])
iloc = dup.data.columns.get_loc(key).tolist().index(True) + idx # type: ignore
dup.data.iloc[:,iloc] = data
else:
dup.data[label] = data.reshape(dup.data[label].shape)
else:
dup.shapes[label] = data.shape[1:] if data.ndim > 1 else (1,)
size = np.prod(data.shape[1:],dtype=np.int64)
new = pd.DataFrame(data=data.reshape((-1,size)),
columns=[label]*size,
)
if not dup.data.index.empty: new.index = dup.data.index
dup.data = pd.concat([dup.data,new],axis=1)
return dup
[docs]
def delete(self,
label: str) -> 'Table':
"""
Delete column data.
Parameters
----------
label : str
Column label.
Returns
-------
updated : damask.Table
Updated table.
"""
dup = self.copy()
dup.data.drop(columns=label,inplace=True)
del dup.shapes[label]
return dup
[docs]
def rename(self,
old: str | Iterable[str],
new: str | Iterable[str],
info: str | None = None) -> 'Table':
"""
Rename column data.
Parameters
----------
old : (iterable of) str
Old column labels.
new : (iterable of) str
New column labels.
Returns
-------
updated : damask.Table
Updated table.
"""
dup = self.copy()
columns = dict(zip([old] if isinstance(old,str) else old,
[new] if isinstance(new,str) else new))
dup.data.rename(columns=columns,inplace=True)
dup.comments.append(f'{old} => {new}'+('' if info is None else f': {info}'))
dup.shapes = {(label if label not in columns else columns[label]):dup.shapes[label] for label in dup.shapes}
return dup
[docs]
def sort_by(self,
labels: str | list[str],
ascending: bool | list[bool] = True) -> 'Table':
"""
Sort table by data of given columns.
Parameters
----------
labels : (list of) str
Column labels for sorting.
ascending : (list of) bool, optional
Set sort order. Defaults to True.
Returns
-------
updated : damask.Table
Updated table.
"""
labels_ = util.to_list(labels)
for i,l in enumerate(labels_):
if m := re.match(r'(.*)\[((\d+,)*(\d+))\]',l):
idx = np.ravel_multi_index(tuple(map(int,m.group(2).split(','))),
self.shapes[m.group(1)])
labels_[i] = f'{1+idx}_{m.group(1)}'
dup = self.copy()
dup._relabel('linear')
dup.data.sort_values(labels_,axis=0,inplace=True,ascending=ascending)
dup._relabel('uniform')
dup.comments.append(f'sorted {"ascending" if ascending else "descending"} by {labels}')
return dup
[docs]
def unique(self,
return_index: bool = False,
return_inverse: bool = False,
return_counts: bool = False,
):
"""
Return the unique elements across the table in order of appearance.
There are three optional outputs in addition to the unique elements:
* the indices of the input table that give the unique values
* the indices of the unique table that reconstruct the input table
* the number of times each unique value comes up in the input table
Parameters
----------
return_index : bool, optional
If True, also return the indices that result in the unique table.
Defaults to False.
return_inverse : bool, optional
If True, also return the indices of the unique table
that can be used to reconstruct the original table.
Defaults to False.
return_counts : bool, optional
If True, also return the number of times each unique value combination appears.
Defaults to False.
Returns
-------
unique : damask.Table
Table containing the sorted unique value combinations of the selected columns.
unique_indices : numpy.ndarray of int, optional
The indices of the first occurrence of the unique values in the input table.
Only provided if return_index is True.
unique_inverse : numpy.ndarray of int, optional
The indices to reconstruct the input table from the unique table.
Only provided if return_inverse is True.
unique_counts : numpy.ndarray of int, optional
The number of times each unique value comes up in the input Table.
Only provided if return_counts is True.
"""
_,idx,inv,cnt = np.unique(
np.hstack([self.get(l) for l in self.labels]),
return_index=True,
return_inverse=True,
return_counts=True,
axis=0,
)
s = np.sort(idx)
m = np.argsort(idx)
M = np.argsort(m)
dup = self[s]
dup.data.reset_index(drop=True,inplace=True)
dup.comments.append('selecting unique combinations')
if return_index or return_inverse or return_counts:
return (dup,) \
+ ((s,) if return_index else ()) \
+ ((M[inv.reshape(-1)],) if return_inverse else ()) \
+ ((cnt[m],) if return_counts else ())
else:
return dup
[docs]
def append(self,
other: 'Table') -> 'Table':
"""
Append other table vertically (similar to numpy.vstack).
Requires matching labels/shapes and order.
Parameters
----------
other : damask.Table
Table to append.
Returns
-------
updated : damask.Table
Updated table.
"""
if self.shapes != other.shapes or not self.data.columns.equals(other.data.columns):
raise KeyError('mismatch of shapes or labels or their order')
dup = self.copy()
dup.data = pd.concat([dup.data,other.data],ignore_index=True)
return dup
[docs]
def join(self,
other: 'Table') -> 'Table':
"""
Append other table horizontally (similar to numpy.hstack).
Requires matching number of rows and no common labels.
Parameters
----------
other : damask.Table
Table to join.
Returns
-------
updated : damask.Table
Updated table.
"""
if set(self.shapes) & set(other.shapes) or self.data.shape[0] != other.data.shape[0]:
raise KeyError('duplicated keys or row count mismatch')
dup = self.copy()
dup.data = dup.data.join(other.data)
for key in other.shapes:
dup.shapes[key] = other.shapes[key]
return dup
[docs]
def save(self,
fname: FileHandleText | FileHandleBinary) -> None:
"""
Save the table in text table or NumPy archive format.
For path-like outputs, a case-insensitive ``.npz`` suffix selects
NumPy archive format; all other suffixes select text table format.
Open binary streams produce NumPy archives, while open text streams
produce text tables.
NumPy archives are written in compressed ``.npz`` format.
Text tables contain leading comments marked by ``#``, a column-label line,
and whitespace-separated fields.
Parameters
----------
fname : str, path-like, or file-like object
Output path or open stream. Text streams must be opened in text
mode, and NumPy archive streams must be opened in binary mode.
"""
labels = self._label(self.labels, "export")
data = self.data.to_numpy()
def save_npz(target: FileHandleBinary) -> None:
arrays = {
label: column
for label, column in zip(labels, data.T)
}
# These names collide with parameters of savez_compressed().
reserved = arrays.keys() & {'file', 'allow_pickle'}
if reserved:
names = ', '.join(sorted(reserved))
raise ValueError(
f'column labels reserved by '
f'np.savez_compressed: {names}'
)
# NumPy uses arbitrary keyword arguments as archive member names.
np.savez_compressed(
target,
**cast(dict[str, Any], arrays),
)
def save_text(stream: TextIO) -> None:
header = '\n'.join(
[f'# {comment}' for comment in self.comments]
+ [' '.join(labels)]
)
stream.write(header + '\n')
csv_options = {
'sep': ' ',
'na_rep': 'nan',
'index': False,
'header': False,
}
try:
# pandas >= 1.5
self.data.to_csv(
stream,
**csv_options,
lineterminator='\n',
) # type: ignore[call-overload]
except TypeError:
# Backward compatibility with pandas < 1.5.
self.data.to_csv(
stream,
**csv_options,
line_terminator='\n',
) # type: ignore[call-overload]
if isinstance(fname, (str, PathLike)):
if Path(fname).suffix.casefold() == '.npz':
save_npz(fname)
else:
with util.open_text(fname, 'w') as stream:
save_text(stream)
elif isinstance(fname, TextIOBase):
save_text(cast(TextIO, fname))
else:
save_npz(cast(FileHandleBinary, fname))
