Repository navigation
Expand file tree
/
Copy pathfile_copying.py
More file actions
106 lines (85 loc) · 3.02 KB
/
Copy pathfile_copying.py
File metadata and controls
106 lines (85 loc) · 3.02 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
"""Tools to facilitate copying files.
This is mainly aimed at cases where a file is being copied with some sort of
modification, or where CVs need to be disk-cached.
"""
import click
from tqdm.auto import tqdm
from paths_cli.param_core import (
Option, Argument, StorageLoader, OPSStorageLoadNames
)
from paths_cli.parameters import HELP_MULTIPLE
INPUT_APPEND_FILE = StorageLoader(
param=Argument('append_file',
type=click.Path(writable=True, readable=True)),
mode='a'
)
class PrecomputeLoadNames(OPSStorageLoadNames):
def get(self, storage, name):
if len(name) == 0:
return list(getattr(storage, self.store))
elif len(name) == 1 and name[0] == '--':
return []
return super(PrecomputeLoadNames, self).get(storage, name)
PRECOMPUTE_CVS = PrecomputeLoadNames(
param=Option('--cv', type=str, multiple=True,
help=('name of CV to precompute; if not specified all will'
+ ' be used' + HELP_MULTIPLE
+ ' (use `--cv --` to disable precomputing)')),
store='cvs'
)
def make_blocks(listlike, blocksize):
"""Make blocks out of a listlike object.
Parameters
----------
listlike : Iterable
must be an iterable that supports slicing
blocksize : int
number of objects per block
Returns
-------
List[List[Any]] :
the input iterable chunked into blocks
"""
n_objs = len(listlike)
partial_block = 1 if n_objs % blocksize else 0
n_blocks = (n_objs // blocksize) + partial_block
minval = lambda i: i * blocksize
maxval = lambda i: min((i + 1) * blocksize, n_objs)
blocks = [listlike[minval(i):maxval(i)] for i in range(n_blocks)]
return blocks
def precompute_cvs(cvs, block):
"""Calculate a CV for a a given block.
Parameters
----------
cvs : List[:class:`openpathsampling.CollectiveVariable`]
CVs to precompute
block : List[Any]
b
"""
for cv in cvs:
cv.enable_diskcache()
_ = cv(block)
def precompute_cvs_func_and_inputs(input_storage, cvs, blocksize):
"""
Parameters
----------
input_storage : :class:`openpathsampling.Storage`
storage file to read from
cvs : List[:class:`openpathsampling.CollectiveVariable`]
list of CVs to precompute; if None, use all CVs in ``input_storage``
blocksize : int
number of snapshots per block to precompute
"""
if cvs is None:
cvs = list(input_storage.cvs)
precompute_func = lambda inps: precompute_cvs(cvs, inps)
snapshot_proxies = input_storage.snapshots.all().as_proxies()
snapshot_blocks = make_blocks(snapshot_proxies, blocksize)
return precompute_func, snapshot_blocks
def rewrite_file(stage_names, stage_mapping):
stages = tqdm(stage_names, desc="All stages")
for stage in stages:
store_func, inputs = stage_mapping[stage]
desc = "This stage: {}".format(stage)
for obj in tqdm(inputs, desc=desc, leave=False):
store_func(obj)