"""
This module provides the ClusterExpansion class.
"""
import os
import pandas as pd
import numpy as np
import pickle
import tempfile
import tarfile
from copy import deepcopy
from typing import Any
from icet import ClusterSpace
from icet.core.cluster_space import SUMMARY_LABEL_WIDTH
from icet.input_output.repr_tools import (MAXIMUM_TABLE_ROWS,
MINIMUM_TABLE_ROWS,
bounded_html_table,
html_representation,
text_rows)
from ase import Atoms
[docs]
class ClusterExpansion:
"""Cluster expansions are obtained by combining a cluster space with a set
of parameters, where the latter is commonly obtained by optimization.
Instances of this class allow one to predict the property of interest for
a given structure.
Note
----
Each element of the parameter vector corresponds to an effective cluster
interaction (ECI) multiplied by the multiplicity of the underlying orbit.
Attributes
----------
cluster_space
Cluster space that was used for constructing the cluster expansion.
parameters
Parameter vector.
metadata
Metadata dictionary, user-defined metadata to be stored together with
cluster expansion.
Will be pickled when CE is written to file.
By default contains icet version, username, hostname and date.
Raises
------
ValueError
If :attr:`cluster_space` and :attr:`parameters` differ in length.
Example
-------
The following snippet illustrates the initialization and usage of a
:class:`ClusterExpansion` object.
Here, the parameters are taken to be a list of ones.
Usually, they would be obtained by training with respect to a set of
reference data::
>>> from ase.build import bulk
>>> from icet import ClusterSpace, ClusterExpansion
>>> # create cluster expansion with fake parameters
>>> prim = bulk('Au')
>>> cs = ClusterSpace(prim, cutoffs=[7.0, 5.0],
... chemical_symbols=[['Au', 'Pd']])
>>> parameters = len(cs) * [1.0]
>>> ce = ClusterExpansion(cs, parameters)
>>> # make prediction for supercell
>>> sc = prim.repeat(3)
>>> for k in [1, 4, 7]:
>>> sc[k].symbol = 'Pd'
>>> print(ce.predict(sc))
"""
def __init__(self, cluster_space: ClusterSpace, parameters: np.array,
metadata: dict | None = None) -> None:
if len(cluster_space) != len(parameters):
raise ValueError('cluster_space ({}) and parameters ({}) must have'
' the same length'.format(len(cluster_space), len(parameters)))
self._cluster_space = cluster_space.copy()
if isinstance(parameters, list):
parameters = np.array(parameters)
self._parameters = parameters
# add metadata
if metadata is None:
metadata = dict()
self._metadata = metadata
self._add_default_metadata()
[docs]
def predict(self, structure: Atoms) -> float:
"""
Returns the property value predicted by the cluster expansion.
Parameters
----------
structure
Atomic configuration.
"""
cluster_vector = self._cluster_space.get_cluster_vector(structure)
prop = np.dot(cluster_vector, self.parameters)
return prop
[docs]
def copy(self) -> 'ClusterExpansion':
"""
Returns an independent copy of this cluster expansion.
Nothing is shared with this cluster expansion except the orbit list,
which cannot be modified and therefore makes the copy cheap.
Pruning or editing either cluster expansion leaves the other as it was,
including through a value nested in the metadata.
The copy carries the metadata of this cluster expansion instead of
metadata describing when the copy was made.
A value in the metadata that cannot be copied makes this method raise.
Writing a cluster expansion pickles its metadata, so such a value
already prevents :func:`write` from working.
The :mod:`copy` module of the standard library goes through this
method, so :func:`copy.copy` and :func:`copy.deepcopy` both return an
equally independent cluster expansion.
"""
cluster_expansion = ClusterExpansion.__new__(ClusterExpansion)
cluster_expansion._cluster_space = self._cluster_space.copy()
cluster_expansion._parameters = np.array(self._parameters)
cluster_expansion._metadata = deepcopy(self._metadata)
return cluster_expansion
def __copy__(self) -> 'ClusterExpansion':
return self.copy()
def __deepcopy__(self, memo: dict) -> 'ClusterExpansion':
return self.copy()
[docs]
def get_cluster_space_copy(self) -> ClusterSpace:
"""
Returns copy of cluster space on which cluster expansion is based.
The copy shares the immutable orbit list with this cluster expansion
and is therefore cheap.
Pruning or merging it replaces its orbit list and leaves the cluster
expansion untouched, which is what keeps the parameters of the
expansion consistent with the cluster space they were fitted against.
"""
return self._cluster_space.copy()
[docs]
def to_dataframe(self) -> pd.DataFrame:
"""Returns a representation of the cluster expansion in the form of a
DataFrame including effective cluster interactions (ECIs)."""
rows = self._cluster_space.as_list
for row, param in zip(rows, self.parameters):
row['parameter'] = param
row['eci'] = param / row['multiplicity']
df = pd.DataFrame(rows)
del df['index']
return df
@property
def chemical_symbols(self) -> list[list[str]]:
""" Species identified by their chemical symbols (copy). """
return self._cluster_space.chemical_symbols.copy()
@property
def cutoffs(self) -> list[float]:
"""
Cutoffs for different n-body clusters (copy).
The cutoff radius (in Ångstroms) defines the largest interatomic
distance in a cluster.
"""
return self._cluster_space.cutoffs.copy()
@property
def orders(self) -> list[int]:
""" Orders included in cluster expansion. """
return list(range(len(self._cluster_space.cutoffs) + 2))
@property
def parameters(self) -> list[float]:
"""Parameter vector.
Each element of the parameter vector corresponds to an effective
cluster interaction (ECI) multiplied by the multiplicity of the
respective orbit."""
return self._parameters
@property
def metadata(self) -> dict:
""" Metadata associated with the cluster expansion. """
return self._metadata
@property
def symprec(self) -> float:
""" Tolerance imposed when analyzing the symmetry using spglib
(inherited from the underlying cluster space). """
return self._cluster_space.symprec
@property
def position_tolerance(self) -> float:
""" Tolerance applied when comparing positions in Cartesian coordinates
(inherited from the underlying cluster space). """
return self._cluster_space.position_tolerance
@property
def fractional_position_tolerance(self) -> float:
""" Tolerance applied when comparing positions in fractional coordinates
(inherited from the underlying cluster space). """
return self._cluster_space.fractional_position_tolerance
@property
def primitive_structure(self) -> Atoms:
""" Primitive structure on which cluster expansion is based. """
return self._cluster_space.primitive_structure.copy()
def __len__(self) -> int:
return len(self._parameters)
def _get_string_representation(self,
print_threshold: int | None = None,
print_minimum: int = MINIMUM_TABLE_ROWS) -> str:
"""
String representation of the cluster expansion, which is the table of
orbits of the underlying cluster space with the parameters and the
effective cluster interactions appended to it.
Parameters
----------
print_threshold
if the number of parameters exceeds this number print dots
print_minimum
number of lines printed from the top and the bottom of the table of
parameters if `print_threshold` is exceeded
"""
cluster_space = self._cluster_space
orbit_header, orbit_body = cluster_space._get_orbit_table(print_threshold,
print_minimum)
multiplicities = [orbit['multiplicity'] for orbit in cluster_space.as_list]
column_width = max(len('{:9.3g}'.format(max(self._parameters, key=abs))), len('ECI'))
def append_columns(line: str, *cells: str) -> str:
return ' | '.join([line] + ['{s:^{n}}'.format(s=cell, n=column_width)
for cell in cells])
summary = text_rows(self._get_rows(), label_width=SUMMARY_LABEL_WIDTH)
header = append_columns(orbit_header, 'parameter', 'ECI')
width = max([len(header)] + [len(line) for line in summary])
body = []
for index, line in orbit_body:
if index is None:
body += [line]
continue
parameter = self._parameters[index]
eci = parameter / multiplicities[index]
body += [append_columns(line, f'{parameter:9.3g}', f'{eci:9.3g}')]
s = [' Cluster Expansion '.center(width, '=')]
s += summary
s += [''.center(width, '-')]
s += [header]
s += [''.center(width, '-')]
s += body
s += [''.center(width, '=')]
return '\n'.join(s)
def __str__(self) -> str:
""" String representation. """
return self._get_string_representation(print_threshold=MAXIMUM_TABLE_ROWS)
def _get_rows(self) -> list[tuple[str, Any]]:
"""Returns the label and value pairs that summarize this cluster
expansion, which head both the string and the HTML representation.
The number of nonzero parameters is counted by order, since an order
that has no orbits at all is missing from the count of orbits by order
and pairing the two sequences by position would then misplace every
count that follows it.
"""
cluster_space = self._cluster_space
number_of_orbits_by_order = cluster_space.number_of_orbits_by_order
nonzero_by_order = dict.fromkeys(number_of_orbits_by_order, 0)
for orbit, parameter in zip(cluster_space.as_list, self._parameters):
if parameter != 0:
nonzero_by_order[orbit['order']] += 1
rows = [('space group', cluster_space.space_group)]
for sublattice in cluster_space.get_sublattices(
self.primitive_structure).active_sublattices:
rows += [(f'chemical species (sublattice {sublattice.symbol})',
list(sublattice.chemical_symbols))]
rows += [('cutoffs', ' '.join('{:.4f}'.format(c) for c in cluster_space.cutoffs))]
rows += [('total number of parameters (nonzero)',
f'{len(self)} ({sum(nonzero_by_order.values())})')]
for order, count in number_of_orbits_by_order.items():
rows += [(f'number of parameters of order {order} (nonzero)',
f'{count} ({nonzero_by_order[order]})')]
rows += [('fractional_position_tolerance',
cluster_space.fractional_position_tolerance),
('position_tolerance', cluster_space.position_tolerance),
('symprec', cluster_space.symprec)]
return rows
def _repr_html_(self) -> str:
"""HTML representation. Used, e.g., in jupyter notebooks.
The summary is followed by the table of parameters, so that the
representation carries the same information as the string
representation.
"""
s = [html_representation(title='Cluster Expansion', rows=self._get_rows())]
s += [bounded_html_table(self.to_dataframe())]
return ''.join(s)
def __repr__(self) -> str:
""" Representation. """
s = type(self).__name__ + '('
s += f'cluster_space={self._cluster_space.__repr__()}'
s += f', parameters={list(self._parameters).__repr__()}'
s += ')'
return s
[docs]
def prune(self, indices: list[int] | None = None, tol: float = 0) -> None:
"""Removes orbits from the cluster expansion, for which the absolute
values of the corresponding parameters are zero or close to zero.
This commonly reduces the computational cost for evaluating the cluster
expansion.
It is therefore recommended to apply this method prior to using the
cluster expansion in production.
If the method is called without arguments only orbits will be pruned,
for which the ECIs are strictly zero.
Less restrictive pruning can be achieved by setting the :attr:`tol`
keyword.
Parameters
----------
indices
Indices of parameters to remove from the cluster expansion.
tol
All orbits will be pruned for which the absolute parameter value(s)
is/are within this tolerance.
By default only orbits whose parameters vanish exactly are pruned.
"""
# find orbit indices to be removed
if indices is None:
indices = [i for i, param in enumerate(
self.parameters) if np.abs(param) <= tol and i > 0]
df = self.to_dataframe()
indices = list(set(indices))
if 0 in indices:
raise ValueError('Orbit index cannot be 0 since the zerolet may not be pruned.')
orbit_candidates_for_removal = df.orbit_index[np.array(indices)].tolist()
safe_to_remove_orbits, safe_to_remove_params = [], []
for oi in set(orbit_candidates_for_removal):
if oi == -1:
continue
orbit_count = df.orbit_index.tolist().count(oi)
oi_remove_count = orbit_candidates_for_removal.count(oi)
if orbit_count <= oi_remove_count:
safe_to_remove_orbits.append(oi)
safe_to_remove_params += df.index[df['orbit_index'] == oi].tolist()
# prune cluster space
self._cluster_space.prune_orbit_list(indices=safe_to_remove_orbits)
self._parameters = self._parameters[np.setdiff1d(
np.arange(len(self._parameters)), safe_to_remove_params)]
assert len(self._parameters) == len(self._cluster_space)
[docs]
def write(self, filename: str) -> None:
"""
Writes ClusterExpansion object to file.
Parameters
----------
filename
name of file to which to write
"""
items = dict()
items['parameters'] = self.parameters
items['metadata'] = self._metadata
with tarfile.open(name=filename, mode='w') as tar_file:
cs_file = tempfile.NamedTemporaryFile(delete=False)
cs_file.close()
self._cluster_space.write(cs_file.name)
tar_file.add(cs_file.name, arcname='cluster_space')
# write items
temp_file = tempfile.TemporaryFile()
pickle.dump(items, temp_file)
temp_file.seek(0)
tar_info = tar_file.gettarinfo(arcname='items', fileobj=temp_file)
tar_file.addfile(tar_info, temp_file)
os.remove(cs_file.name)
temp_file.close()
[docs]
@staticmethod
def read(filename: str) -> 'ClusterExpansion':
"""
Reads :class:`ClusterExpansion` object from file.
Parameters
----------
filename
File from which to read.
"""
with tarfile.open(name=filename, mode='r') as tar_file:
cs_file = tempfile.NamedTemporaryFile(delete=False)
cs_file.write(tar_file.extractfile('cluster_space').read())
cs_file.close()
cs = ClusterSpace.read(cs_file.name)
items = pickle.load(tar_file.extractfile('items'))
os.remove(cs_file.name)
ce = ClusterExpansion.__new__(ClusterExpansion)
ce._cluster_space = cs
ce._parameters = items['parameters']
ce._metadata = items.get('metadata', {})
assert list(items['parameters']) == list(ce.parameters)
return ce
def _add_default_metadata(self):
""" Adds default metadata to metadata dict. """
import getpass
import socket
from datetime import datetime
from icet import __version__ as icet_version
self._metadata['date_created'] = datetime.now().strftime('%Y-%m-%dT%H:%M:%S')
self._metadata['username'] = getpass.getuser()
self._metadata['hostname'] = socket.gethostname()
self._metadata['icet_version'] = icet_version
