Commit deeb7d7e authored by Theo Steininger's avatar Theo Steininger

Added magnetic_field class and factories.

Modified Pipeline accordingly.
parent 2a40a142
......@@ -2,6 +2,7 @@
from .version import __version__
from likelihoods import *
from magnetic_fields import *
from observers import *
from priors import *
......
# -*- coding: utf-8 -*-
from magnetic_field_factory import MagneticFieldFactory
from magnetic_field import MagneticField
from jf12_factory import JF12Factory
# -*- coding: utf-8 -*-
from magnetic_field_factory import MagneticFieldFactory
class JF12Factory(MagneticFieldFactory):
@property
def descriptor(self):
return 'JF12'
def _initialize_parameter_defaults(self):
self._parameter_defaults = {'b51_ran_b1': 10.8,
'b51_ran_b2': 6.96,
'b51_ran_b3': 9.59,
'b51_ran_b4': 6.96,
'b51_ran_b5': 1.96,
'b51_ran_b6': 16.34,
'b51_ran_b7': 37.29,
'b51_ran_b8': 10.35,
'b51_coh_b1': 0.1,
'b51_coh_b2': 3.0,
'b51_coh_b3': -0.9,
'b51_coh_b4': -0.8,
'b51_coh_b5': -2.0,
'b51_coh_b6': -4.2,
'b51_coh_b7': 0.0,
'b51_z0_spiral': 0.61,
'b51_z0_smooth': 2.84,
'b51_r0_smooth': 10.97,
'b51_b0_smooth': 4.68,
'b51_b0_x': 4.6,
'b51_Xtheta': 49.,
'b51_r0_x': 2.9,
'b51_h_disk': 0.4,
'b51_Bn': 1.4,
'b51_Bs': -1.1,
'b51_z0_halo': 5.3,
'b51_b_ring': 0.1,
'b51_b0_interior': 7.63,
'b51_reg_b0': 0,
'b51_shift': 0}
def _initialize_variable_to_parameter_mappings(self, n=3):
self._variable_to_parameter_mappings = {
'b51_ran_b1': self._interval(10.8, 2.33, n),
'b51_ran_b2': self._interval(6.96, 1.58, n),
'b51_ran_b3': self._interval(9.59, 1.10, n),
'b51_ran_b4': self._interval(6.96, 0.87, n),
'b51_ran_b5': self._interval(1.96, 1.32, n),
'b51_ran_b6': self._interval(16.34, 2.53, n),
'b51_ran_b7': self._interval(37.29, 2.39, n),
'b51_ran_b8': self._interval(10.35, 4.43, n),
'b51_coh_b1': self._interval(0.1, 1.8, n),
'b51_coh_b2': self._interval(3.0, 0.6, n),
'b51_coh_b3': self._interval(-0.9, 0.8, n),
'b51_coh_b4': self._interval(-0.8, 0.3, n),
'b51_coh_b5': self._interval(-2.0, 0.1, n),
'b51_coh_b6': self._interval(-4.2, 0.5, n),
'b51_coh_b7': self._interval(0.0, 1.8, n),
'b51_z0_spiral': self._positive_interval(0.61, 0.04, n),
'b51_z0_smooth': self._positive_interval(2.84, 1.30, n),
'b51_r0_smooth': self._positive_interval(10.97, 3.80, n),
'b51_b0_smooth': self._interval(4.68, 1.39, n),
'b51_b0_x': self._interval(4.6, 0.3, n),
'b51_Xtheta': self._positive_interval(49., 1., n),
'b51_r0_x': self._positive_interval(2.9, 0.1, n),
'b51_h_disk': self._positive_interval(0.4, 0.03, n),
'b51_Bn': self._interval(1.4, 0.1, n),
'b51_Bs': self._interval(-1.1, 0.1, n),
'b51_z0_halo': self._positive_interval(5.3, 1.6, n),
'b51_b_ring': self._interval(0.1, 0.1, n),
'b51_b0_interior': self._interval(7.63, 1.39, n),
'b51_reg_b0': [0, 10, 1],
'b51_shift': [0, 10, 1],
}
# -*- coding: utf-8 -*-
import numpy as np
from keepers import Loggable
class MagneticField(Loggable, object):
def __init__(self, box_dimensions, resolution, descriptor,
parameters={}):
self._box_dimensions = box_dimensions
self._resolution = resolution
self._descriptor = str(descriptor)
self._parameters = {}
for (key, value) in parameters.items():
self._parameters[str(key)] = np.float(value)
self._array = None
@property
def parameters(self):
return self._parameters
@property
def box_dimensions(self):
return self._box_dimensions
@property
def resolution(self):
return self._resolution
@property
def descriptor(self):
return self._descriptor
@property
def array(self):
if self._array is None:
self._array = self._create_array()
return self._array
def _create_array(self):
raise NotImplementedError
# -*- coding: utf-8 -*-
import abc
import numpy as np
from keepers import Loggable
from imagine.carrier_mapper import carrier_mapper
from magnetic_field import MagneticField
class MagneticFieldFactory(Loggable, object):
def __init__(self, box_dimensions, resolution):
self.box_dimensions = box_dimensions
self.resolution = resolution
self._initialize_parameter_defaults()
self._initialize_variable_to_parameter_mappings()
@abc.abstractproperty
def descriptor(self):
raise NotImplementedError
def _initialize_parameter_defaults(self):
self._parameter_defaults = {}
def _initialize_variable_to_parameter_mappings(self):
self._variable_to_parameter_mapping = {}
@staticmethod
def _interval(mean, sigma, n):
return [mean-n*sigma, mean, mean+n*sigma]
@staticmethod
def _positive_interval(mean, sigma, n):
return [max(0, mean-n*sigma), mean, mean+n*sigma]
@property
def parameter_defaults(self):
return self._parameter_defaults
@parameter_defaults.setter
def parameter_defaults(self, new_defaults):
self._parameter_defaults.update((str(k), np.float(v))
for k, v in new_defaults.items()
if k in self._parameter_defaults)
@property
def variable_to_parameter_mappings(self):
return self._variable_to_parameter_mappings
@variable_to_parameter_mappings.setter
def variable_to_parameter_mappings(self, new_mapping):
"""
The parameter-mapping must be a dictionary with
key: parameter-name
value: [min, mean, max]
"""
for k, v in new_mapping.items():
if k in self._variable_to_parameter_mappings:
key = str(k)
value = [np.float(v[1]), np.float(v[2]), np.float(v[3])]
self._variable_to_parameter_mappings.update((key, value))
self.logger.debug("Updated variable_to_parameter_mapping %s "
"to %s" % (key, str(value)))
def _map_variables_to_parameters(self, variables):
parameter_dict = {}
for variable_name in variables:
if variable_name in self.variable_to_parameter_mappings:
mapping = self.variable_to_parameter_mappings[variable_name]
mapped_variable = carrier_mapper(variables[variable_name],
a=mapping[0],
m=mapping[1],
b=mapping[2])
else:
mapped_variable = np.float(variables[variable_name])
parameter_dict[variable_name] = mapped_variable
return parameter_dict
def generate(self, variables):
mapped_variables = self._map_variables_to_parameters(variables)
work_parameters = self.parameter_defaults.copy()
work_parameters.update(mapped_variables)
result_magnetic_field = MagneticField(
box_dimensions=self.box_dimensions,
resolution=self.resolution,
descriptor=self.descriptor,
parameters=work_parameters)
return result_magnetic_field
......@@ -3,3 +3,4 @@
class Observer(object):
pass
# -*- coding: utf-8 -*-
import numpy as np
from keepers import Loggable
from likelihoods import Likelihood
from magnetic_fields import MagneticFieldFactory
from observers import Observer
from priors import Prior
from carrier_mapper import carrier_mapper
class Pipeline(Loggable, object):
def __init__(self, observer, likelihood, prior, ensemble_size=1,
active_parameters=[], parameter_mapping={}):
"""
The pipeline
- posses all the building blocks: magnetic_field, observer,
likelihood and prior.
- if multiple log-likelihoods and log-priors are given: sum the result
- coordinates the repeated observation in order to compute an ensemble
- controls which parameters of the magnetic field are tested
(active parameters)
"""
def __init__(self, magnetic_field_factory, observer, likelihood, prior,
active_variables=[], ensemble_size=1):
self.logger.debug("Setting up pipeline.")
self.magnetic_field_factory = magnetic_field_factory
self.observer = observer
self.likelihood = likelihood
self.prior = prior
self.active_parameters = active_parameters
self.parameter_mapping = parameter_mapping
self.active_variables = active_variables
self.ensemble_size = ensemble_size
@property
......@@ -41,7 +49,10 @@ class Pipeline(Loggable, object):
def likelihood(self, likelihood):
if not isinstance(likelihood, Likelihood):
raise TypeError(
"likelihood must be an instance of likelihood-class.")
"likelihood must be an instance of Likelihood-class.")
self.logger.debug("Setting likelihood.")
self._likelihood = likelihood
@property
def prior(self):
......@@ -51,27 +62,38 @@ class Pipeline(Loggable, object):
def prior(self, prior):
if not isinstance(prior, Prior):
raise TypeError(
"prior must be an instance of prior-class.")
"prior must be an instance of Prior-class.")
self.logger.debug("Setting prior.")
self._prior = prior
@property
def parameter_mapping(self):
return self._parameter_mapping
@parameter_mapping.setter
def parameter_mapping(self, parameters):
"""
The parameter-mapping must be a dictionary with
key: parameter-name
value: [min, mean, max]
"""
new_mapping = {}
for p in parameters:
new_key = str(p[0])
new_value = [p[1], p[2], p[3]]
new_mapping[new_key] = new_value
self.logger.debug("Setting parameter_mapping %s to %s." %
(new_key, new_mapping[new_key]))
self._parameter_mapping = new_mapping
def magnetic_field_factory(self):
return self._magnetic_field_factory
@magnetic_field_factory.setter
def magnetic_field_factory(self, magnetic_field_factory):
if not isinstance(magnetic_field_factory, MagneticFieldFactory):
raise TypeError(
"magnetic_field_factory must be an instance of the "
"MagneticFieldFactory-class.")
self.logger.debug("Setting magnetic_field_factory.")
self._magnetic_field_factory = magnetic_field_factory
@property
def active_variables(self):
return self._active_variables
@active_variables.setter
def active_variables(self, active_variables):
if not isinstance(active_variables, list):
raise TypeError(
"active_variables must be a list.")
self.logger.debug("Resetting active_variables to %s" %
str(active_variables))
new_active = []
for av in active_variables:
new_active += [str(av)]
self._active_variables = new_active
@property
def ensemble_size(self):
......@@ -86,19 +108,5 @@ class Pipeline(Loggable, object):
self.logger.debug("Setting ensemble size to %i." % ensemble_size)
self._ensemble_size = ensemble_size
def _map_parameters(self, parameter_list):
parameter_dict = {}
for i, name in enumerate(self.active_parameters):
if name in self.parameter_mapping:
mapping = self.parameter_mapping[name]
mapped_parameter = carrier_mapper(parameter_list[i],
a=mapping[1],
m=mapping[2],
b=mapping[3])
else:
mapped_parameter = np.float(parameter_list[i])
parameter_dict[name] = mapped_parameter
return parameter_dict
def __call__(self, parameter_list):
mapped_parameters = self._map_parameters(parameter_list)
def __call__(self, variables):
pass
\ No newline at end of file
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment