1
|
|
|
"""General utility functions.""" |
2
|
|
|
|
3
|
|
|
from typing import Tuple |
4
|
|
|
import sys |
5
|
|
|
import time |
6
|
|
|
import datetime |
7
|
|
|
|
8
|
|
|
import numpy as np |
9
|
|
|
import numba |
10
|
|
|
from scipy.spatial.distance import squareform |
11
|
|
|
|
12
|
|
|
|
13
|
|
|
def diameter(cell): |
14
|
|
|
"""Diameter of a unit cell (as a square matrix in Cartesian/Orthogonal form) |
15
|
|
|
in 3 or fewer dimensions.""" |
16
|
|
|
|
17
|
|
|
dims = cell.shape[0] |
18
|
|
|
if dims == 1: |
19
|
|
|
return cell[0][0] |
20
|
|
|
if dims == 2: |
21
|
|
|
d = np.amax(np.linalg.norm(np.array([cell[0] + cell[1], cell[0] - cell[1]]), axis=-1)) |
22
|
|
|
elif dims == 3: |
23
|
|
|
diams = np.array([ |
24
|
|
|
cell[0] + cell[1] + cell[2], |
|
|
|
|
25
|
|
|
cell[0] + cell[1] - cell[2], |
|
|
|
|
26
|
|
|
cell[0] - cell[1] + cell[2], |
|
|
|
|
27
|
|
|
- cell[0] + cell[1] + cell[2] |
28
|
|
|
]) |
29
|
|
|
d = np.amax(np.linalg.norm(diams, axis=-1)) |
|
|
|
|
30
|
|
|
else: |
31
|
|
|
raise ValueError(f'diameter only implimented for dimensions <= 3.') |
|
|
|
|
32
|
|
|
return d |
33
|
|
|
|
34
|
|
|
|
35
|
|
|
@numba.njit() |
36
|
|
|
def cellpar_to_cell(a, b, c, alpha, beta, gamma): |
|
|
|
|
37
|
|
|
"""Simplified version of function from :mod:`ase.geometry` of the same name. |
38
|
|
|
3D unit cell parameters a,b,c,α,β,γ --> cell as 3x3 NumPy array. |
39
|
|
|
""" |
40
|
|
|
|
41
|
|
|
eps = 2 * np.spacing(90.0) # ~1.4e-14 |
42
|
|
|
|
43
|
|
|
cos_alpha = 0. if abs(abs(alpha) - 90.) < eps else np.cos(alpha * np.pi / 180.) |
44
|
|
|
cos_beta = 0. if abs(abs(beta) - 90.) < eps else np.cos(beta * np.pi / 180.) |
45
|
|
|
cos_gamma = 0. if abs(abs(gamma) - 90.) < eps else np.cos(gamma * np.pi / 180.) |
46
|
|
|
|
47
|
|
|
if abs(gamma - 90) < eps: |
48
|
|
|
sin_gamma = 1. |
49
|
|
|
elif abs(gamma + 90) < eps: |
50
|
|
|
sin_gamma = -1. |
51
|
|
|
else: |
52
|
|
|
sin_gamma = np.sin(gamma * np.pi / 180.) |
53
|
|
|
|
54
|
|
|
cy = (cos_alpha - cos_beta * cos_gamma) / sin_gamma |
55
|
|
|
cz_sqr = 1. - cos_beta ** 2 - cy ** 2 |
56
|
|
|
if cz_sqr < 0: |
57
|
|
|
raise RuntimeError('Could not create unit cell from given parameters.') |
58
|
|
|
|
59
|
|
|
cell = np.zeros((3, 3)) |
60
|
|
|
cell[0, 0] = a |
61
|
|
|
cell[1, 0] = b * cos_gamma |
62
|
|
|
cell[1, 1] = b * sin_gamma |
63
|
|
|
cell[2, 0] = c * cos_beta |
64
|
|
|
cell[2, 1] = c * cy |
65
|
|
|
cell[2, 2] = c * np.sqrt(cz_sqr) |
66
|
|
|
|
67
|
|
|
return cell |
68
|
|
|
|
69
|
|
|
|
70
|
|
|
@numba.njit() |
71
|
|
|
def cellpar_to_cell_2D(a, b, alpha): |
72
|
|
|
"""2D unit cell parameters a,b,α --> cell as 2x2 ndarray.""" |
73
|
|
|
|
74
|
|
|
cell = np.zeros((2, 2)) |
75
|
|
|
cell[0, 0] = a |
76
|
|
|
cell[1, 0] = b * np.cos(alpha * np.pi / 180.) |
77
|
|
|
cell[1, 1] = b * np.sin(alpha * np.pi / 180.) |
78
|
|
|
|
79
|
|
|
return cell |
80
|
|
|
|
81
|
|
|
|
82
|
|
|
def cell_to_cellpar(cell): |
83
|
|
|
"""Unit cell as a 3x3 NumPy array -> list of 6 lengths + angles.""" |
84
|
|
|
lengths = np.linalg.norm(cell, axis=-1) |
85
|
|
|
angles = [] |
86
|
|
|
for i, j in [(1, 2), (0, 2), (0, 1)]: |
87
|
|
|
ang_rad = np.arccos(np.dot(cell[i], cell[j]) / (lengths[i] * lengths[j])) |
88
|
|
|
angles.append(np.rad2deg(ang_rad)) |
89
|
|
|
return np.concatenate((lengths, np.array(angles))) |
90
|
|
|
|
91
|
|
|
|
92
|
|
|
def cell_to_cellpar_2D(cell): |
93
|
|
|
"""Unit cell as a 2x2 NumPy array -> list of 2 lengths and an angle.""" |
94
|
|
|
cellpar = np.zeros((3, )) |
95
|
|
|
lengths = np.linalg.norm(cell, axis=-1) |
96
|
|
|
ang_rad = np.arccos(np.dot(cell[0], cell[1]) / (lengths[0] * lengths[1])) |
97
|
|
|
cellpar[0] = lengths[0] |
98
|
|
|
cellpar[1] = lengths[1] |
99
|
|
|
cellpar[2] = np.rad2deg(ang_rad) |
100
|
|
|
return cellpar |
101
|
|
|
|
102
|
|
|
|
103
|
|
|
def neighbours_from_distance_matrix( |
104
|
|
|
n: int, |
105
|
|
|
dm: np.ndarray |
106
|
|
|
) -> Tuple[np.ndarray, np.ndarray]: |
107
|
|
|
"""Given a distance matrix, find the n nearest neighbours of each item. |
108
|
|
|
|
109
|
|
|
Parameters |
110
|
|
|
---------- |
111
|
|
|
n : int |
112
|
|
|
Number of nearest neighbours to find for each item. |
113
|
|
|
dm : :class:`numpy.ndarray` |
114
|
|
|
2D distance matrix or 1D condensed distance matrix. |
115
|
|
|
|
116
|
|
|
Returns |
117
|
|
|
------- |
118
|
|
|
(nn_dm, inds) : Tuple[:class:`numpy.ndarray`, :class:`numpy.ndarray`] |
|
|
|
|
119
|
|
|
``nn_dm[i][j]`` is the distance from item :math:`i` to its :math:`j+1` st |
120
|
|
|
nearest neighbour, and ``inds[i][j]`` is the index of this neighbour |
|
|
|
|
121
|
|
|
(:math:`j+1` since index 0 is the first nearest neighbour). |
122
|
|
|
""" |
123
|
|
|
|
124
|
|
|
inds = None |
125
|
|
|
|
126
|
|
|
# 2D distance matrix |
127
|
|
|
if len(dm.shape) == 2: |
128
|
|
|
inds = np.array([np.argpartition(row, n)[:n] for row in dm]) |
129
|
|
|
|
130
|
|
|
# 1D condensed distance vector |
131
|
|
|
elif len(dm.shape) == 1: |
132
|
|
|
dm = squareform(dm) |
133
|
|
|
inds = [] |
134
|
|
|
for i, row in enumerate(dm): |
135
|
|
|
inds_row = np.argpartition(row, n+1)[:n+1] |
136
|
|
|
inds_row = inds_row[inds_row != i][:n] |
137
|
|
|
inds.append(inds_row) |
138
|
|
|
inds = np.array(inds) |
139
|
|
|
|
140
|
|
|
else: |
141
|
|
|
ValueError( |
142
|
|
|
'Input must be an ndarray, either a 2D distance matrix ' |
143
|
|
|
'or a condensed distance matrix (returned by pdist).') |
144
|
|
|
|
145
|
|
|
# inds are the indexes of nns: inds[i,j] is the j-th nn to point i |
146
|
|
|
nn_dm = np.take_along_axis(dm, inds, axis=-1) |
147
|
|
|
sorted_inds = np.argsort(nn_dm, axis=-1) |
148
|
|
|
inds = np.take_along_axis(inds, sorted_inds, axis=-1) |
149
|
|
|
nn_dm = np.take_along_axis(nn_dm, sorted_inds, axis=-1) |
150
|
|
|
return nn_dm, inds |
151
|
|
|
|
152
|
|
|
|
153
|
|
|
def random_cell(length_bounds=(1, 2), angle_bounds=(60, 120), dims=3): |
154
|
|
|
"""Dimensions 2 and 3 only. Random unit cell with uniformally chosen length and |
|
|
|
|
155
|
|
|
angle parameters between bounds.""" |
156
|
|
|
|
157
|
|
|
if dims == 3: |
158
|
|
|
while True: |
159
|
|
|
lengths = [np.random.uniform(low=length_bounds[0], |
160
|
|
|
high=length_bounds[1]) |
161
|
|
|
for _ in range(dims)] |
162
|
|
|
angles = [np.random.uniform(low=angle_bounds[0], |
163
|
|
|
high=length_bounds[1]) |
164
|
|
|
for _ in range(dims)] |
165
|
|
|
|
166
|
|
|
try: |
167
|
|
|
cell = cellpar_to_cell(*lengths, *angles) |
168
|
|
|
break |
169
|
|
|
except RuntimeError: |
170
|
|
|
continue |
171
|
|
|
|
172
|
|
|
elif dims == 2: |
173
|
|
|
lengths = [np.random.uniform(low=length_bounds[0], |
174
|
|
|
high=length_bounds[1]) |
175
|
|
|
for _ in range(dims)] |
|
|
|
|
176
|
|
|
alpha = np.random.uniform(low=angle_bounds[0], |
177
|
|
|
high=length_bounds[1]) |
178
|
|
|
cell = cellpar_to_cell_2D(*lengths, alpha) |
179
|
|
|
|
180
|
|
|
else: |
181
|
|
|
raise ValueError(f'random_cell only implimented for dimensions 2 and 3 (passed {dims})') |
182
|
|
|
|
183
|
|
|
return cell |
184
|
|
|
|
185
|
|
|
|
186
|
|
|
class _ETA: |
187
|
|
|
"""Pass total amount to do on construction, |
188
|
|
|
then call .update() on every loop.""" |
189
|
|
|
|
190
|
|
|
# epochtime_{n+1} = factor * epochtime + (1-factor) * epochtime_{n} |
191
|
|
|
_moving_av_factor = 0.3 |
192
|
|
|
|
193
|
|
|
def __init__(self, to_do, update_rate=1000): |
194
|
|
|
self.to_do = to_do |
195
|
|
|
self.update_rate = update_rate |
196
|
|
|
self.counter = 0 |
197
|
|
|
self.start_time = time.perf_counter() |
198
|
|
|
self.tic = self.start_time |
199
|
|
|
self.time_per_epoch = None |
200
|
|
|
self.done = False |
201
|
|
|
|
202
|
|
|
def update(self): |
|
|
|
|
203
|
|
|
|
204
|
|
|
self.counter += 1 |
205
|
|
|
|
206
|
|
|
if self.counter == self.to_do: |
|
|
|
|
207
|
|
|
sys.stdout.write(self.finished()) |
208
|
|
|
self.done = True |
209
|
|
|
return |
210
|
|
|
|
211
|
|
|
elif self.counter > self.to_do: |
212
|
|
|
return |
213
|
|
|
|
214
|
|
|
if not self.counter % self.update_rate: |
215
|
|
|
sys.stdout.write(self._end_epoch()) |
216
|
|
|
|
217
|
|
|
def finished(self): |
|
|
|
|
218
|
|
|
total = time.time() - self.start_time |
219
|
|
|
msg = f'Total time: {round(total, 2)}s, ' \ |
220
|
|
|
f'n passes: {self.counter} ' \ |
221
|
|
|
f'({round(self.to_do/total, 2)} items/s)\r\n' |
222
|
|
|
return msg |
223
|
|
|
|
224
|
|
|
def _end_epoch(self): |
225
|
|
|
toc = time.perf_counter() |
226
|
|
|
epoch_time = toc - self.tic |
227
|
|
|
if self.time_per_epoch is None: |
228
|
|
|
self.time_per_epoch = epoch_time |
229
|
|
|
else: |
230
|
|
|
self.time_per_epoch = _ETA._moving_av_factor * epoch_time + \ |
231
|
|
|
(1 - _ETA._moving_av_factor) * self.time_per_epoch |
232
|
|
|
|
|
|
|
|
233
|
|
|
percent = round(100 * self.counter / self.to_do, 2) |
234
|
|
|
remaining = int(((self.to_do - self.counter) / self.update_rate) * self.time_per_epoch) |
235
|
|
|
eta = str(datetime.timedelta(seconds=remaining)) |
236
|
|
|
self.tic = toc |
237
|
|
|
return f'{percent}%, ETA {eta}' + ' ' * 20 + '\r' |
238
|
|
|
|