Skip to content
Snippets Groups Projects
Commit 33da1fb8 authored by David Fairbrother's avatar David Fairbrother
Browse files

Re #18108 Implementation of InstrumentSettings Class

parent 06d19b05
No related branches found
No related tags found
No related merge requests found
Showing
with 182 additions and 172 deletions
from __future__ import (absolute_import, division, print_function)
import os
import mantid.simpleapi as mantid
from isis_powder.routines import common, yaml_parser
from isis_powder.routines import common, InstrumentSettings, yaml_parser
from isis_powder.routines.common_enums import InputBatchingEnum
from isis_powder.abstract_inst import AbstractInst
from isis_powder.pearl_routines import pearl_algs, pearl_output, pearl_spline, PearlRunSettings
from isis_powder.pearl_routines import pearl_algs, pearl_output, pearl_spline, pearl_advanced_config
class Pearl(AbstractInst):
def __init__(self, user_name, config_file=None, **kwargs):
expected_keys = ["calibration_directory", "output_directory", "attenuation_file_name",
def __init__(self, **kwargs):
expected_attr = ["user_name", "config_file", "calibration_dir", "output_dir", "attenuation_file_name",
"calibration_mapping_file"]
yaml_parser.set_kwargs_from_config_file(config_path=config_file, kwargs=kwargs, keys_to_find=expected_keys)
import pydevd
pydevd.settrace('localhost', port=51205, stdoutToServer=True, stderrToServer=True)
# Parse all possible locations that the parameters can be set from
basic_config_dict = yaml_parser.open_yaml_file_as_dictionary(kwargs.get("config_file", None))
self._inst_settings = InstrumentSettings.InstrumentSettings(
attr_mapping_dict=self.attr_mapping, adv_conf_dict=pearl_advanced_config.variables,
basic_conf_dict=basic_config_dict, kwargs=kwargs)
super(Pearl, self).__init__(user_name=user_name, calibration_dir=kwargs["calibration_directory"],
output_dir=kwargs["output_directory"])
self._inst_settings.check_expected_attributes_are_set(attr_mapping=self.attr_mapping,
expected_attr_names=expected_attr)
self._basic_config_file_path = config_file
self._calibration_mapping_path = kwargs["calibration_mapping_file"]
attenuation_file_name = kwargs["attenuation_file_name"] # "PRL112_DC25_10MM_FF.OUT"
self._attenuation_full_path = os.path.join(self._calibration_dir, attenuation_file_name)
super(Pearl, self).__init__(user_name=self._inst_settings.user_name,
calibration_dir=self._inst_settings.calibration_dir,
output_dir=self._inst_settings.output_dir)
self._run_settings = None
self._ads_workaround = 0
self._cached_run_details = None
self._cached_run_details_number = None
def focus(self, run_number, **kwargs):
self._run_settings = _get_settings_focus_kwargs(config_file_path=self._basic_config_file_path, kwargs=kwargs)
return self._focus(run_number=run_number, input_batching=InputBatchingEnum.Summed,
do_van_normalisation=self._run_settings.divide_by_vanadium)
def create_calibration_vanadium(self, run_in_range, **kwargs):
self._run_settings = _get_settings_van_calib_kwargs(config_file_path=self._basic_config_file_path,
kwargs=kwargs)
self._run_settings.number_of_splines = kwargs.get("num_of_splines", 60)
kwargs["tt_mode"] = "tt88"
self._inst_settings.update_attributes_from_kwargs(attr_mapping_dict=self.attr_mapping, kwargs=kwargs)
expected_attr = ["long_mode", "van_norm", "absorb_corrections"]
self._inst_settings.check_expected_attributes_are_set(attr_mapping=self.attr_mapping,
expected_attr_names=expected_attr)
run_details = self.get_run_details(run_number_string=int(run_in_range))
run_details.run_number = run_details.vanadium_run_numbers
return self._create_calibration_vanadium(vanadium_runs=run_details.vanadium_run_numbers,
empty_runs=run_details.empty_runs,
do_absorb_corrections=self._run_settings.absorption_corrections)
do_absorb_corrections=self._inst_settings.absorb_corrections)
# Params #
def get_default_group_names(self):
......@@ -59,14 +62,14 @@ class Pearl(AbstractInst):
if self._cached_run_details_number == first_run:
return self._cached_run_details
run_settings = self._run_settings
run_details = pearl_algs.get_run_details(absorb_on=run_settings.absorption_corrections,
long_mode_on=run_settings.divide_by_vanadium,
# TODO justt pass in inst settings instead
run_details = pearl_algs.get_run_details(absorb_on=self._inst_settings.absorb_corrections,
long_mode_on=self._inst_settings.long_mode,
run_number_string=run_number_string,
calibration_dir=self._calibration_dir,
mapping_file=self._calibration_mapping_path)
mapping_file=self._inst_settings.calibration_mapping_file)
run_details = pearl_algs.set_advanced_run_details(run_details=run_details, tt_mode=self._run_settings.tt_mode,
run_details = pearl_algs.set_advanced_run_details(run_details=run_details, tt_mode=self._inst_settings.tt_mode,
calibration_dir=self._calibration_dir)
self._cached_run_details_number = first_run
self._cached_run_details = run_details
......@@ -108,7 +111,7 @@ class Pearl(AbstractInst):
output_spectra = \
pearl_output.generate_and_save_focus_output(self, processed_spectra=processed_spectra,
run_details=run_details, focus_mode=output_mode,
perform_attenuation=self._run_settings.perform_attenuation)
perform_attenuation=self._inst_settings.perform_atten)
group_name = "PEARL" + str(run_details.run_number) + "-Results-D-Grp"
grouped_d_spacing = mantid.GroupWorkspaces(InputWorkspaces=output_spectra, OutputWorkspace=group_name)
return grouped_d_spacing
......@@ -140,35 +143,18 @@ class Pearl(AbstractInst):
return data_processed
# Implementation of instrument specific steps
def _get_settings_common_kwargs(config_file_path, kwargs):
expected_keys = ["long_mode"]
yaml_parser.set_kwargs_from_config_file(config_path=config_file_path, kwargs=kwargs, keys_to_find=expected_keys)
run_settings = PearlRunSettings.PearlRunSettings()
run_settings.long_mode = kwargs["long_mode"]
return run_settings
def _get_settings_focus_kwargs(config_file_path, kwargs):
run_settings = _get_settings_common_kwargs(config_file_path=config_file_path, kwargs=kwargs)
expected_keys = ["divide_vanadium", "tt_mode", "output_mode", "attenuate"]
yaml_parser.set_kwargs_from_config_file(config_path=config_file_path, kwargs=kwargs, keys_to_find=expected_keys)
run_settings.tt_mode = kwargs["tt_mode"]
run_settings.focus_mode = kwargs["output_mode"]
run_settings.perform_attenuation = kwargs["attenuate"]
run_settings.divide_by_vanadium = kwargs["divide_vanadium"]
return run_settings
def _get_settings_van_calib_kwargs(config_file_path, kwargs):
run_settings = _get_settings_common_kwargs(config_file_path=config_file_path, kwargs=kwargs)
expected_keys = ["absorption_corrections"]
yaml_parser.set_kwargs_from_config_file(config_path=config_file_path, kwargs=kwargs, keys_to_find=expected_keys)
run_settings.absorption_corrections = kwargs["absorption_corrections"]
run_settings.tt_mode = "tt88" # Use full range in vanadium mode
return run_settings
# Maps parameter/config name -> script names
attr_mapping = [("absorb_corrections", "absorb_corrections"),
("attenuation_file_name", "attenuation_file_name"),
("config_file", "config_file_name"),
("calibration_config_file", "calibration_mapping_file"),
("calibration_directory", "calibration_dir"),
("long_mode", "long_mode"),
("tt_mode", "tt_mode"),
("output_directory", "output_dir"),
("perform_atten", "perform_attenuation"),
("user_name", "user_name"),
("vanadium_normalisation", "van_norm")]
def _generate_file_name(run_number):
......
from __future__ import (absolute_import, division, print_function)
class PearlRunSettings(object):
absorption_corrections = None
divide_by_vanadium = None
focus_mode = None
long_mode = None
number_of_splines = None
tt_mode = None
perform_attenuation = None
......@@ -5,6 +5,28 @@ file_names = {
"tt35_grouping": "pearl_group_12_1_TT35.cal"
}
script_params = {
"b_spline_coefficient": 60,
}
tof_cropping_ranges = [
(1500, 19900), # Bank 1
(1500, 19900), # Bank 2
(1500, 19900), # Bank 3
(1500, 19900), # Bank 4
(1500, 19900), # Bank 5
(1500, 19900), # Bank 6
(1500, 19900), # Bank 7
(1500, 19900), # Bank 8
(1500, 19900), # Bank 9
(1500, 19900), # Bank 10
(1500, 19900), # Bank 11
(1500, 19900), # Bank 12
(1500, 19900), # Bank 13
(1500, 19900) # Bank 14
]
variable_help = {
"file_names": {
"vanadium_absorb_file_name": "Takes the name of the calculated vanadium absorption corrections. This file "
......@@ -21,4 +43,10 @@ variable_help = {
}
}
\ No newline at end of file
}
variables = {
"bank_tof_crop_values" : tof_cropping_ranges,
"file_names_dict": file_names,
"script_params_dict": script_params
}
......@@ -2,7 +2,7 @@ user_name : "Mantid_Tester"
calibration_directory : 'C:\Users\ieb35538\Documents\Repos\a_mantidBuild\ExternalData\Testing\Data\SystemTest\PEARL\Calibration'
output_directory : 'C:\Users\ieb35538\Documents\Repos\a_mantidBuild\ExternalData\Testing\Data\SystemTest\PEARL\data_out'
calibration_mapping_file : 'C:\Users\ieb35538\Documents\Repos\a_mantid\scripts\Diffraction\isis_powder\pearl_routines\pearl_calibration.yaml'
calibration_config_file : 'C:\Users\ieb35538\Documents\Repos\a_mantid\scripts\Diffraction\isis_powder\pearl_routines\pearl_calibration.yaml'
......@@ -7,12 +7,11 @@ import mantid.simpleapi as mantid
import isis_powder.routines.common as common
from isis_powder.routines import yaml_parser
from isis_powder.abstract_inst import AbstractInst
from isis_powder.polaris_routines import polaris_algs, polaris_output
from isis_powder.polaris_routines import polaris_advanced_config, polaris_algs, polaris_output
class Polaris(AbstractInst):
# Instrument specific properties
_masking_file_name = "VanaPeaks.dat"
def __init__(self, chopper_on, config_file=None, **kwargs):
......@@ -22,11 +21,8 @@ class Polaris(AbstractInst):
super(Polaris, self).__init__(user_name=kwargs["user_name"], calibration_dir=kwargs["calibration_directory"],
output_dir=kwargs["output_directory"])
self._chopper_on = chopper_on
self._apply_solid_angle = kwargs["apply_solid_angle"]
self._calibration_mapping_path = kwargs["calibration_mapping_file"]
self._spline_coeff = 100 # TODO move this out into advanced config
self._chopper_on = chopper_on
# Hold the last dictionary later to avoid us having to keep parsing the YAML
self._run_details_last_run_number = None
......@@ -56,7 +52,8 @@ class Polaris(AbstractInst):
if self._run_details_last_run_number == first_run:
return self._run_details_cached_obj
run_details = polaris_algs.get_run_details(chopper_on=self._chopper_on, sac_on=self._apply_solid_angle,
solid_angle_on = bool(polaris_advanced_config.standard_variables["apply_solid_angle_corrections"])
run_details = polaris_algs.get_run_details(chopper_on=self._chopper_on, sac_on=solid_angle_on,
run_number=first_run, calibration_dir=self._calibration_dir,
mapping_path=self._calibration_mapping_path)
......@@ -79,15 +76,12 @@ class Polaris(AbstractInst):
return normalised_ws
def apply_solid_angle_efficiency_corr(self, ws_to_correct, run_details):
if not self._apply_solid_angle:
return ws_to_correct
solid_angle_on = bool(polaris_advanced_config.standard_variables["apply_solid_angle_corrections"])
if not run_details or not os.path.isfile(run_details.solid_angle_corr):
corrections = \
polaris_algs.generate_solid_angle_corrections(run_details=run_details, instrument=self)
else:
corrections = mantid.Load(Filename=run_details.solid_angle_corr)
if not solid_angle_on:
return ws_to_correct
corrections = polaris_algs.generate_solid_angle_corrections(run_details=run_details, instrument=self)
corrected_ws = mantid.Divide(LHSWorkspace=ws_to_correct, RHSWorkspace=corrections)
common.remove_intermediate_workspace(corrections)
common.remove_intermediate_workspace(ws_to_correct)
......@@ -106,12 +100,12 @@ class Polaris(AbstractInst):
return spectra_name
def spline_vanadium_ws(self, focused_vanadium_spectra, instrument_version=''):
mode = "spline"
masking_file_path = os.path.join(self.calibration_dir, self._masking_file_name)
masking_file_name = polaris_advanced_config.file_names["bragg_peaks_masking"]
spline_coeff = polaris_advanced_config.standard_variables["b_spline_coefficient"]
masking_file_path = os.path.join(self.calibration_dir, masking_file_name)
output = polaris_algs.process_vanadium_for_focusing(bank_spectra=focused_vanadium_spectra,
spline_number=self._spline_coeff,
mode=mode, mask_path=masking_file_path)
spline_number=spline_coeff,
mask_path=masking_file_path)
return output
......
file_names = {
"bragg_peaks_masking": "VanaPeaks.dat"
}
standard_variables = {
"apply_solid_angle_corrections": False,
"b_spline_coefficient": 100
}
absorption_correction_params = {
"cylinder_sample_height": 4.0,
"cylinder_sample_radius": 0.4,
"attenuation_cross_section": 4.88350,
"scattering_cross_section": 5.15775,
"sample_number_density": 0.0718956,
"number_of_slices": 10,
"number_of_annuli": 10,
"number_of_wavelength_points": 100,
"exponential_method": "Normal"
}
......@@ -2,60 +2,26 @@ from __future__ import (absolute_import, division, print_function)
import mantid.simpleapi as mantid
import os
import isis_powder.routines.common as common
from isis_powder.routines import yaml_parser
from isis_powder.routines import common, yaml_parser
from isis_powder.routines.common_enums import InputBatchingEnum
from isis_powder.routines.RunDetails import RunDetails
def calculate_focus_binning_params(sample_ws):
# TODO remove this if they only want sane TOF values and not consistent binning
focus_bin_widths = [-0.0050, -0.0010, -0.0010, -0.0010, -0.00050]
focus_crop_start = 2 # These are used when calculating binning range
focus_crop_end = 0.95
calculated_binning_params = []
num_of_banks = sample_ws.getNumberHistograms()
for i in range(0, num_of_banks):
sample_data = sample_ws.readX(i)
starting_bin = sample_data[0] * (1 + focus_crop_start)
ending_bin = sample_data[-1] * focus_crop_end
bin_width = focus_bin_widths[i]
bank_binning_params = [str(starting_bin), str(bin_width), str(ending_bin)]
calculated_binning_params.append(bank_binning_params)
return calculated_binning_params
from isis_powder.polaris_routines import polaris_advanced_config
def generate_absorb_corrections(ws_to_match):
absorb_ws = mantid.CloneWorkspace(InputWorkspace=ws_to_match)
# TODO move all of this into defaults
cylinder_sample_height = str(4)
cylinder_sample_radius = str(0.4)
attenuation_cross_section = str(4.88350)
scattering_cross_section = str(5.15775)
sample_number_density = str(0.0718956)
number_of_slices = str(10)
number_of_annuli = str(10)
number_of_wavelength_points = str(100)
exp_method = "Normal"
# TODO move all of the above into defaults
absorb_dict = polaris_advanced_config.absorption_correction_params
absorb_ws = mantid.CylinderAbsorption(InputWorkspace=absorb_ws,
CylinderSampleHeight=cylinder_sample_height,
CylinderSampleRadius=cylinder_sample_radius,
AttenuationXSection=attenuation_cross_section,
ScatteringXSection=scattering_cross_section,
SampleNumberDensity=sample_number_density,
NumberOfSlices=number_of_slices,
NumberOfAnnuli=number_of_annuli,
NumberOfWavelengthPoints=number_of_wavelength_points,
ExpMethod=exp_method)
CylinderSampleHeight=absorb_dict["cylinder_sample_height"],
CylinderSampleRadius=absorb_dict["cylinder_sample_radius"],
AttenuationXSection=absorb_dict["attenuation_cross_section"],
ScatteringXSection=absorb_dict["scattering_cross_section"],
SampleNumberDensity=absorb_dict["sample_number_density"],
NumberOfSlices=absorb_dict["number_of_slices"],
NumberOfAnnuli=absorb_dict["number_of_annuli"],
NumberOfWavelengthPoints=absorb_dict["number_of_wavelength_points"],
ExpMethod=absorb_dict["exponential_method"])
return absorb_ws
......@@ -126,15 +92,10 @@ def split_into_tof_d_spacing_groups(processed_spectra):
return d_spacing_group, tof_group
def process_vanadium_for_focusing(bank_spectra, mode, mask_path, spline_number=None):
# TODO move spline number/mode out of params passed and instead get this to read it itself
if mode == "spline": # TODO support more modes
bragg_masking_list = _read_masking_file(mask_path)
output = _spline_vanadium_for_focusing(vanadium_spectra_list=bank_spectra,
spline_coefficient=spline_number, mask_list=bragg_masking_list)
else:
raise NotImplementedError("Other vanadium processing methods not yet implemented")
def process_vanadium_for_focusing(bank_spectra, mask_path, spline_number=None):
bragg_masking_list = _read_masking_file(mask_path)
output = _spline_vanadium_for_focusing(vanadium_spectra_list=bank_spectra,
spline_coefficient=spline_number, mask_list=bragg_masking_list)
return output
......
from __future__ import (absolute_import, division, print_function)
import warnings
# Have to patch warnings at runtime to not print the source code. This is even advertised as a 'feature' of
# the warnings library in the documentation: https://docs.python.org/3/library/warnings.html#warnings.showwarning
def warning_no_source(msg, *ignored):
return str(msg) + '\n'
warnings.formatwarning = warning_no_source
class InstrumentSettings(object):
# Holds instance variables updated at runtime
def __init__(self, attr_mapping_dict, adv_conf_dict=None, basic_conf_dict=None, kwargs=None):
self._parse_attributes(dict_to_parse=adv_conf_dict, attribute_mapping=attr_mapping_dict)
self._parse_attributes(dict_to_parse=basic_conf_dict, attribute_mapping=attr_mapping_dict)
self._parse_attributes(dict_to_parse=kwargs, attribute_mapping=attr_mapping_dict)
def check_expected_attributes_are_set(self, attr_mapping, expected_attr_names):
expected_params_dict = {}
# Filter down the full mapping list
found_tuple_list = [tuple_entry for tuple_entry in attr_mapping if tuple_entry[-1] in expected_attr_names]
expected_params_dict.update(dict(found_tuple_list))
self._check_attribute_is_set(expected_params_dict)
def update_attributes_from_kwargs(self, attr_mapping_dict, kwargs):
self._parse_attributes(dict_to_parse=kwargs, attribute_mapping=attr_mapping_dict)
def _check_attribute_is_set(self, expected_attributes_dict):
for config_name in expected_attributes_dict:
try:
getattr(self, expected_attributes_dict[config_name])
except AttributeError:
raise ValueError("Parameter '" + str(config_name) +
"' was not set in any of the config files or passed as a parameter.\n")
def _parse_attributes(self, dict_to_parse, attribute_mapping):
for config_key in dict_to_parse:
# Recurse down all dictionaries
if isinstance(dict_to_parse[config_key], dict):
self._parse_attributes(dict_to_parse[config_key], attribute_mapping)
continue # Skip so we don't accidentally re-add this dictionary
# Update attributes from said dictionary
found_attribute = next((attr_tuple for attr_tuple in attribute_mapping if config_key in attr_tuple[0]),
None)
if found_attribute:
# The first element of the attribute is the config name and the last element is the name scripts use
self._update_attribute(attr_name=found_attribute[-1], attr_val=dict_to_parse[found_attribute[0]])
else:
warnings.warn("Ignoring unknown configuration key: " + str(config_key))
continue
def _update_attribute(self, attr_name, attr_val):
setattr(self, attr_name, attr_val)
......@@ -7,7 +7,7 @@ class RunDetails(object):
"""
def __init__(self, run_number):
# Essential attributes
# Essential attribute
self.run_number = run_number
self.empty_runs = None
......@@ -20,4 +20,3 @@ class RunDetails(object):
self.vanadium_absorption_path = None
self.vanadium_run_numbers = None
self.solid_angle_corr = None # TODO move back into POLARIS
......@@ -9,8 +9,6 @@ from isis_powder.routines.common_enums import InputBatchingEnum
global g_ads_workaround
g_ads_workaround = {"read_ws": 0}
# --- Public API --- #
def create_calibration_by_names(calibration_runs, startup_objects, grouping_file_name, group_names):
_create_blank_cal_file(calibration_runs=calibration_runs, group_names=group_names,
......
......@@ -61,7 +61,6 @@ def _batched_run_focusing(input_batching, instrument, perform_vanadium_norm, run
for ws in read_ws_list:
output = _focus_one_ws(ws=ws, run_number=run_number, instrument=instrument,
perform_vanadium_norm=perform_vanadium_norm)
common.remove_intermediate_workspace(ws)
return output
......
......@@ -8,7 +8,7 @@ from isis_powder.routines import yaml_sanity
def get_run_dictionary(run_number, file_path):
config_file = _open_yaml_file_as_dictionary(file_path)
config_file = open_yaml_file_as_dictionary(file_path)
yaml_sanity.calibration_file_sanity_check(config_file)
run_key = _find_dictionary_key(dict_to_search=config_file, run_number=run_number)
......@@ -23,19 +23,7 @@ def is_run_range_key_unbounded(key):
return True if split_key[-1] == '' else False
def set_kwargs_from_config_file(config_path, kwargs, keys_to_find):
if config_path:
basic_config_dict = _open_yaml_file_as_dictionary(file_path=config_path)
else:
# Create an empty dictionary so we still get error checking below and nicer error messages
basic_config_dict = {}
# Set any unset properties:
for key in keys_to_find:
_get_kwarg_key_from_dict(config_dictionary=basic_config_dict, kwargs=kwargs, key=key)
def _open_yaml_file_as_dictionary(file_path):
def open_yaml_file_as_dictionary(file_path):
if not file_path or not os.path.isfile(file_path):
raise ValueError("Config file not found at path of:\n" + str(file_path) + '\n ')
......@@ -64,14 +52,3 @@ def _find_dictionary_key(dict_to_search, run_number):
return key
return None
def _get_kwarg_key_from_dict(config_dictionary, kwargs, key):
error_first = "Setting with name: '"
error_last = "' was not passed in the call or set in the basic config."
kwarg_value = kwargs.get(key, None)
if kwarg_value is None:
# Only try to parse it if it wasn't passed in
value = common.dictionary_key_helper(dictionary=config_dictionary, key=key, throws=True,
exception_msg=(error_first + key + error_last))
kwargs[key] = value
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment