-
Notifications
You must be signed in to change notification settings - Fork 122
/
IndirectCommon.py
478 lines (392 loc) · 17.5 KB
/
IndirectCommon.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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
from mantid.simpleapi import *
from mantid.api import TextAxis
from mantid import config, logger
from IndirectImport import import_mantidplot
import os.path, math, datetime, re
import numpy as np
import itertools
def StartTime(prog):
logger.notice('----------')
message = 'Program ' + prog +' started @ ' + str(datetime.datetime.now())
logger.notice(message)
def EndTime(prog):
message = 'Program ' + prog +' ended @ ' + str(datetime.datetime.now())
logger.notice(message)
logger.notice('----------')
def loadInst(instrument):
workspace = '__empty_' + instrument
if not mtd.doesExist(workspace):
idf_dir = config['instrumentDefinition.directory']
idf = idf_dir + instrument + '_Definition.xml'
LoadEmptyInstrument(Filename=idf, OutputWorkspace=workspace)
def loadNexus(filename):
'''
Loads a Nexus file into a workspace with the name based on the
filename. Convenience function for not having to play around with paths
in every function.
'''
name = os.path.splitext(os.path.split(filename)[1])[0]
LoadNexus(Filename=filename, OutputWorkspace=name)
return name
def getInstrRun(ws_name):
'''
Get the instrument name and run number from a workspace.
@param ws_name - name of the workspace
@return tuple of form (instrument, run number)
'''
workspace = mtd[ws_name]
run_number = str(workspace.getRunNumber())
if run_number == '0':
#attempt to parse run number off of name
match = re.match(r'([a-zA-Z]+)([0-9]+)', ws_name)
if match:
run_number = match.group(2)
else:
raise RuntimeError("Could not find run number associated with workspace.")
instrument = workspace.getInstrument().getName()
facility = config.getFacility()
instrument = facility.instrument(instrument).filePrefix(int(run_number))
instrument = instrument.lower()
return instrument, run_number
def getWSprefix(wsname):
'''
Returns a string of the form '<ins><run>_<analyser><refl>_' on which
all of our other naming conventions are built. The workspace is used to get the
instrument parameters.
'''
if wsname == '':
return ''
workspace = mtd[wsname]
facility = config['default.facility']
ws_run = workspace.getRun()
if 'facility' in ws_run:
facility = ws_run.getLogData('facility').value
(instrument, run_number) = getInstrRun(wsname)
if facility == 'ILL':
run_name = instrument + '_'+ run_number
else:
run_name = instrument + run_number
try:
analyser = workspace.getInstrument().getStringParameter('analyser')[0]
reflection = workspace.getInstrument().getStringParameter('reflection')[0]
except IndexError:
analyser = ''
reflection = ''
prefix = run_name + '_' + analyser + reflection
if len(analyser + reflection) > 0:
prefix += '_'
return prefix
def getEfixed(workspace, det_index=0):
inst = mtd[workspace].getInstrument()
return inst.getNumberParameter("efixed-val")[0]
def checkUnitIs(ws, unit_id, axis_index=0):
"""
Check that the workspace has the correct units by comparing
against the UnitID.
"""
axis = mtd[ws].getAxis(axis_index)
unit = axis.getUnit()
return unit.unitID() == unit_id
# Get the default save directory and check it's valid
def getDefaultWorkingDirectory():
workdir = config['defaultsave.directory']
if not os.path.isdir(workdir):
raise IOError("Default save directory is not a valid path!")
return workdir
def createQaxis(inputWS):
result = []
workspace = mtd[inputWS]
n_hist = workspace.getNumberHistograms()
if workspace.getAxis(1).isSpectra():
inst = workspace.getInstrument()
sample_pos = inst.getSample().getPos()
beam_pos = sample_pos - inst.getSource().getPos()
for i in range(0, n_hist):
efixed = getEfixed(inputWS, i)
detector = workspace.getDetector(i)
theta = detector.getTwoTheta(sample_pos, beam_pos) / 2
lamda = math.sqrt(81.787/efixed)
q_value = 4 * math.pi * math.sin(theta) / lamda
result.append(q_value)
else:
axis = workspace.getAxis(1)
msg = 'Creating Axis based on Detector Q value: '
if not axis.isNumeric():
msg += 'Input workspace must have either spectra or numeric axis.'
raise ValueError(msg)
if axis.getUnit().unitID() != 'MomentumTransfer':
msg += 'Input must have axis values of Q'
raise ValueError(msg)
for i in range(0, n_hist):
result.append(float(axis.label(i)))
return result
def GetWSangles(in_ws):
nhist = mtd[in_ws].getNumberHistograms() # get no. of histograms/groups
source_pos = mtd[in_ws].getInstrument().getSource().getPos()
sample_pos = mtd[in_ws].getInstrument().getSample().getPos()
beam_pos = sample_pos - source_pos
angles = [] # will be list of angles
for index in range(0, nhist):
detector = mtd[in_ws].getDetector(index) # get index
two_theta = detector.getTwoTheta(sample_pos, beam_pos)*180.0/math.pi # calc angle
angles.append(two_theta) # add angle
return angles
def GetThetaQ(workspace):
efixed = getEfixed(workspace)
wavelas = math.sqrt(81.787/efixed) # elastic wavelength
k0_val = 4.0*math.pi/wavelas
theta = np.array(GetWSangles(workspace))
q_val = k0_val * np.sin(0.5 * np.radians(theta))
return theta, q_val
def ExtractFloat(data_string):
""" Extract float values from an ASCII string"""
values = data_string.split()
values = map(float, values)
return values
def ExtractInt(data_string):
""" Extract int values from an ASCII string"""
values = data_string.split()
values = map(int, values)
return values
def PadArray(inarray, nfixed): # pad a list to specified size
npt = len(inarray)
padding = nfixed-npt
outarray = []
outarray.extend(inarray)
outarray += [0]*padding
return outarray
def CheckAnalysers(in1_ws, in2_ws, verbose):
ws1 = mtd[in1_ws]
analyser1 = ws1.getInstrument().getStringParameter('analyser')[0]
reflection1 = ws1.getInstrument().getStringParameter('reflection')[0]
ws2 = mtd[in2_ws]
analyser2 = ws2.getInstrument().getStringParameter('analyser')[0]
reflection2 = ws2.getInstrument().getStringParameter('reflection')[0]
if analyser1 != analyser2:
raise ValueError('Workspace '+in1_ws+' and '+in2_ws+' have different analysers')
elif reflection1 != reflection2:
raise ValueError('Workspace '+in1_ws+' and '+in2_ws+' have different reflections')
else:
if verbose:
logger.notice('Analyser is '+analyser1+reflection1)
def CheckHistZero(in_ws):
nhist = mtd[in_ws].getNumberHistograms() # no. of hist/groups in WS
if nhist == 0:
raise ValueError('Workspace '+in_ws+' has NO histograms')
x_in = mtd[in_ws].readX(0)
ntc = len(x_in) - 1 # no. points from length of x array
if ntc == 0:
raise ValueError('Workspace '+in_ws+' has NO points')
return nhist, ntc
def CheckHistSame(in1_ws, name1, in2_ws, name2):
num_hist1 = mtd[in1_ws].getNumberHistograms() # no. of hist/groups in WS1
x_range1 = mtd[in1_ws].readX(0)
x_len1 = len(x_range1)
num_hist2 = mtd[in2_ws].getNumberHistograms() # no. of hist/groups in WS2
x_range2 = mtd[in2_ws].readX(0)
x_len2 = len(x_range2)
if num_hist1 != num_hist2: # check that no. groups are the same
error_str = '{0} ({1}) histograms ({2}) not = {3}({4}) histograms ({5})'.format(
name1, in1_ws, num_hist1, name2, in2_ws, num_hist2)
raise ValueError(error_str)
elif x_len1 != x_len2:
error_str = '{0} ({1}) array length ({2}) not = {3}({4}) array length ({5})'.format(
name1, in1_ws, x_len1, name2, in2_ws, x_len2)
raise ValueError(error_str)
def CheckXrange(x_range, range_type):
if not ((len(x_range) == 2) or (len(x_range) == 4)):
raise ValueError(range_type + ' - Range must contain either 2 or 4 numbers')
for lower, upper in zip(x_range[::2], x_range[1::2]):
if math.fabs(lower) < 1e-5:
raise ValueError(range_type+' - input minimum ('+str(lower)+') is Zero')
if math.fabs(upper) < 1e-5:
raise ValueError(range_type+' - input maximum ('+str(upper)+') is Zero')
if upper < lower:
raise ValueError(range_type+' - input max ('+str(upper)+') < min ('+str(lower)+')')
def CheckElimits(e_range, x_in):
x_len = len(x_in) - 1
if math.fabs(e_range[0]) < 1e-5:
raise ValueError('Elimits - input emin ( '+str(e_range[0])+' ) is Zero')
if e_range[0] < x_in[0]:
raise ValueError('Elimits - input emin ( '+str(e_range[0])+' ) < data emin ( '+str(x_in[0])+' )')
if math.fabs(e_range[1]) < 1e-5:
raise ValueError('Elimits - input emax ( '+str(e_range[1])+' ) is Zero')
if e_range[1] > x_in[x_len]:
raise ValueError('Elimits - input emax ( '+str(e_range[1])+' ) > data emax ( '+str(x_in[x_len])+' )')
if e_range[1] < e_range[0]:
raise ValueError('Elimits - input emax ( '+str(e_range[1])+' ) < emin ( '+str(e_range[0])+' )')
def getInstrumentParameter(ws, param_name):
"""Get an named instrument parameter from a workspace.
@param ws The workspace to get the instrument from.
@param param_name The name of the parameter to look up.
"""
inst = mtd[ws].getInstrument()
# create a map of type parameters to functions. This is so we avoid writing lots of
# if statements becuase there's no way to dynamically get the type.
func_map = {'double': inst.getNumberParameter, 'string': inst.getStringParameter,
'int': inst.getIntParameter, 'bool': inst.getBoolParameter}
if inst.hasParameter(param_name):
param_type = inst.getParameterType(param_name)
if param_type != '':
param = func_map[param_type](param_name)[0]
else:
raise ValueError('Unable to retrieve %s from Instrument Parameter file.' % param_name)
else:
raise ValueError('Unable to retrieve %s from Instrument Parameter file.' % param_name)
return param
def plotSpectra(ws, y_axis_title, indicies=[]):
"""
Plot a selection of spectra given a list of indicies
@param ws - the workspace to plot
@param y_axis_title - label for the y axis
@param indicies - list of spectrum indicies to plot
"""
if len(indicies) == 0:
num_spectra = mtd[ws].getNumberHistograms()
indicies = range(num_spectra)
try:
mtd_plot = import_mantidplot()
plot = mtd_plot.plotSpectrum(ws, indicies, True)
layer = plot.activeLayer()
layer.setAxisTitle(mtd_plot.Layer.Left, y_axis_title)
except RuntimeError:
# User clicked cancel on plot so don't do anything
return
def plotParameters(ws, *param_names):
"""
Plot a number of spectra given a list of parameter names
This searchs for relevent spectra using the text axis label.
@param ws - the workspace to plot from
@param param_names - list of names to search for
"""
axis = mtd[ws].getAxis(1)
if axis.isText() and len(param_names) > 0:
num_spectra = mtd[ws].getNumberHistograms()
for name in param_names:
indicies = [i for i in range(num_spectra) if name in axis.label(i)]
if len(indicies) > 0:
plotSpectra(ws, name, indicies)
def convertToElasticQ(input_ws, output_ws=None):
"""
Helper function to convert the spectrum axis of a sample to ElasticQ.
@param input_ws - the name of the workspace to convert from
@param output_ws - the name to call the converted workspace
"""
if output_ws is None:
output_ws = input_ws
axis = mtd[input_ws].getAxis(1)
if axis.isSpectra():
e_fixed = getEfixed(input_ws)
ConvertSpectrumAxis(input_ws, Target='ElasticQ', EMode='Indirect', EFixed=e_fixed, OutputWorkspace=output_ws)
elif axis.isNumeric():
# check that units are Momentum Transfer
if axis.getUnit().unitID() != 'MomentumTransfer':
raise RuntimeError('Input must have axis values of Q')
CloneWorkspace(input_ws, OutputWorkspace=output_ws)
else:
raise RuntimeError('Input workspace must have either spectra or numeric axis.')
def transposeFitParametersTable(params_table, output_table=None):
"""
Transpose the parameter table created from a multi domain Fit.
This function will make the output consistent with PlotPeakByLogValue.
@param params_table - the parameter table output from Fit.
@param output_table - name to call the transposed table. If omitted,
the output_table will be the same as the params_table
"""
params_table = mtd[params_table]
table_ws = '__tmp_table_ws'
table_ws = CreateEmptyTableWorkspace(OutputWorkspace=table_ws)
param_names = params_table.column(0)[:-1] #-1 to remove cost function
param_values = params_table.column(1)[:-1]
param_errors = params_table.column(2)[:-1]
# find the number of parameters per function
func_index = param_names[0].split('.')[0]
num_params = 0
for i, name in enumerate(param_names):
if name.split('.')[0] != func_index:
num_params = i
break
# create columns with parameter names for headers
column_names = ['.'.join(name.split('.')[1:]) for name in param_names[:num_params]]
column_error_names = [name + '_Err' for name in column_names]
column_names = zip(column_names, column_error_names)
table_ws.addColumn('double', 'axis-1')
for name, error_name in column_names:
table_ws.addColumn('double', name)
table_ws.addColumn('double', error_name)
# output parameter values to table row
for i in xrange(0, params_table.rowCount()-1, num_params):
row_values = param_values[i:i+num_params]
row_errors = param_errors[i:i+num_params]
row = [value for pair in zip(row_values, row_errors) for value in pair]
row = [i/num_params] + row
table_ws.addRow(row)
if output_table is None:
output_table = params_table.name()
RenameWorkspace(table_ws.name(), OutputWorkspace=output_table)
def search_for_fit_params(suffix, table_ws):
"""
Find all fit parameters in a table workspace with the given suffix.
@param suffix - the name of the parameter to find.
@param table_ws - the name of the table workspace to search.
"""
return [name for name in mtd[table_ws].getColumnNames() if name.endswith(suffix)]
def convertParametersToWorkspace(params_table, x_column, param_names, output_name):
"""
Convert a parameter table output by PlotPeakByLogValue to a MatrixWorkspace.
This will make a spectrum for each parameter name using the x_column vairable as the
x values for the spectrum.
@param params_table - the table workspace to convert to a MatrixWorkspace.
@param x_column - the column in the table to use for the x values.
@param parameter_names - list of parameter names to add to the workspace
@param output_name - name to call the output workspace.
"""
# search for any parameters in the table with the given parameter names,
# ignoring their function index and output them to a workspace
workspace_names = []
for param_name in param_names:
column_names = search_for_fit_params(param_name, params_table)
column_error_names = search_for_fit_params(param_name+'_Err', params_table)
param_workspaces = []
for name, error_name in zip(column_names, column_error_names):
ConvertTableToMatrixWorkspace(params_table, x_column, name, error_name, OutputWorkspace=name)
param_workspaces.append(name)
workspace_names.append(param_workspaces)
# transpose list of workspaces, ignoring unequal length of lists
# this handles the case where a parameter occurs only once in the whole workspace
workspace_names = map(list, itertools.izip_longest(*workspace_names))
workspace_names = [filter(None, sublist) for sublist in workspace_names]
# join all the parameters for each peak into a single workspace per peak
temp_workspaces = []
for peak_params in workspace_names:
temp_peak_ws = peak_params[0]
for param_ws in peak_params[1:]:
ConjoinWorkspaces(temp_peak_ws, param_ws, False)
temp_workspaces.append(temp_peak_ws)
# join all peaks into a single workspace
##TODO: I'm not sure exactly what this is supposed to do
temp_workspace = temp_workspaces[0]
for temp_ws in temp_workspaces[1:]:
ConjoinWorkspaces(temp_workspace, temp_peak_ws, False)
RenameWorkspace(temp_workspace, OutputWorkspace=output_name)
# replace axis on workspaces with text axis
axis = TextAxis.create(mtd[output_name].getNumberHistograms())
workspace_names = [name for sublist in workspace_names for name in sublist]
for i, name in enumerate(workspace_names):
axis.setLabel(i, name)
mtd[output_name].replaceAxis(1, axis)
def addSampleLogs(ws, sample_logs):
"""
Add a dictionary of logs to a workspace.
The type of the log is inferred by the type of the value passed to the log.
@param ws - workspace to add logs too.
@param sample_logs - dictionary of logs to append to the workspace.
"""
for key, value in sample_logs.iteritems():
if isinstance(value, bool):
log_type = 'String'
elif isinstance(value, (int, long, float)):
log_type = 'Number'
else:
log_type = 'String'
AddSampleLog(Workspace=ws, LogName=key, LogType=log_type, LogText=str(value))