forked from AmbaPant/mantid
-
Notifications
You must be signed in to change notification settings - Fork 1
/
CreateCacheFilename.py
142 lines (120 loc) · 4.99 KB
/
CreateCacheFilename.py
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
# Mantid Repository : https://github.com/mantidproject/mantid
#
# Copyright © 2018 ISIS Rutherford Appleton Laboratory UKRI,
# NScD Oak Ridge National Laboratory, European Spallation Source,
# Institut Laue - Langevin & CSNS, Institute of High Energy Physics, CAS
# SPDX - License - Identifier: GPL - 3.0 +
#pylint: disable=no-init,invalid-name,bare-except,too-many-arguments
from mantid.api import *
from mantid.kernel import *
import mantid
import os
# See ticket #14716
class CreateCacheFilename(PythonAlgorithm):
""" Create cache filename
"""
def category(self):
"""
"""
return "Workflow\\DataHandling"
def name(self):
"""
"""
return "CreateCacheFilename"
def summary(self):
""" Return summary
"""
return """Create cache filename"""
def PyInit(self):
""" Declare properties
"""
# this is the requirement of using this plugin
# is there a place to register that?
self.declareProperty("PropertyManager", "", "Name of a property manager from which properties are extracted from")
self.declareProperty(
StringArrayProperty("Properties", Direction.Input),
"A list of property names to be included")
self.declareProperty(
StringArrayProperty("OtherProperties", Direction.Input),
"A list of key=value strings for other properties not in the property manager")
self.declareProperty("Prefix", "", "prefix for the output file name")
self.declareProperty(
FileProperty(name='CacheDir', defaultValue='', action=FileAction.OptionalDirectory),
doc='Directory storing cache files for reuse, in-lieu of repetitive, time-consuming calculations')
self.declareProperty("OutputFilename", "", "Full path of output file name", Direction.Output)
self.declareProperty("OutputSignature", "", "sha1 string, 40 characters long", Direction.Output)
return
def validateInputs(self):
issues = dict()
manager = self.getPropertyValue('PropertyManager').strip()
if len(manager) > 0 and manager not in mantid.PropertyManagerDataService:
issues['PropertyManager'] = 'Does not exist'
if len(manager) <= 0 and not self.getProperty('OtherProperties').value:
message = "Either PropertyManager or OtherProperties should be supplied"
issues['PropertyManager'] = message
issues['OtherProperties'] = message
return issues
def PyExec(self):
""" Main Execution Body
"""
# Inputs
prop_manager = self.getPropertyValue("PropertyManager").strip()
if prop_manager in mantid.PropertyManagerDataService:
prop_manager = mantid.PropertyManagerDataService[prop_manager]
else:
prop_manager = None
other_props = self.getProperty("OtherProperties").value
if not prop_manager and not other_props:
raise ValueError("Either PropertyManager or OtherProperties should be supplied")
# default to all properties in the manager
props = self.getProperty("Properties").value
if not props and prop_manager:
props = list(prop_manager.keys())
# output settings
prefix = self.getPropertyValue("Prefix")
cache_dir = self.getPropertyValue("CacheDir")
if not cache_dir:
cache_dir = os.path.join(ConfigService.getUserPropertiesDir(), "cache")
# calculate
file_name, sha1_hash = self._calculate(prop_manager, props, other_props, prefix, cache_dir)
self.setProperty("OutputFilename", file_name)
self.setProperty("OutputSignature", sha1_hash)
return
def _get_signature(self, prop_manager, props, other_props):
# get matched properties
if prop_manager:
props = matched(list(prop_manager.keys()), props)
# create the list of key=value strings
kvpairs = [
'%s=%s' % (prop, prop_manager.getPropertyValue(prop))
for prop in props
]
else:
kvpairs = []
kvpairs += other_props
# sort
kvpairs.sort()
# one string out of the list
s = ','.join(kvpairs)
return s
def _calculate(self, prop_manager, props, other_props, prefix, cache_dir):
s = self._get_signature(prop_manager, props, other_props)
h = _hash(s) # sha1 hash
if prefix:
fn = "%s_%s.nxs" % (prefix, h) # filename
else:
fn = "%s.nxs" % h
return os.path.join(cache_dir, fn), h
def _hash(s):
import hashlib
return hashlib.sha1(str(s).encode('utf-8')).hexdigest()
def matched(keys, patterns):
"return keys that match any of the given patterns"
import fnmatch
filtered = []
for pat in patterns:
filtered += fnmatch.filter(keys, pat)
continue
return set(filtered)
# Register algorithm with Mantid
AlgorithmFactory.subscribe(CreateCacheFilename)