Skip to content

Offset

Offset module.

PerSensor dataclass

Bases: Component

Offset implementation which assumes an additive offset between sensors.

The offset is which is constant in space and time and accounts for calibration differences between sensors. To maintain parameter identifiability, the offset for the first sensor (with index 0) is assumed to be 0, and other sensor offsets are defined relative to this beam.

Attributes:

Name Type Description
n_sensor int

number of sensors in the sensor object used for analysis.
offset ndarray

array of sampled offset values, populated in self.from_mcmc() after the MCMC run is completed.
precision_scalar ndarray

array of sampled offset precision values, populated in self.from_mcmc() after the MCMC run is completed. Only populated if update_precision is True.
indicator_basis csc_matrix

[nof_observations x (nof_sensors - 1)] sparse matrix which assigns the offset parameters to the correct observations.
update_precision bool

logical indicating whether the offset prior precision parameter should be updated as part of the analysis.
mean_offset float

prior mean parameter for the offsets, assumed to be the same for each beam. Default is 0.
prior_precision_shape float

shape parameter for the prior gamma distribution for the scalar precision parameter. Default is 1e-3.
prior_precision_rate float

rate parameter for the prior gamma distribution for the scalar precision parameter(s). Default is 1e-3.
initial_precision float

initial value for the scalar precision parameter. Default is 1.0.
Source code in src/pyelq/component/offset.py 
 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
@dataclass
class PerSensor(Component):
    """Offset implementation which assumes an additive offset between sensors.

    The offset is which is constant in space and time and accounts for calibration differences between sensors.
    To maintain parameter identifiability, the offset for the first sensor (with index 0) is assumed to be 0, and other
    sensor offsets are defined relative to this beam.

    Attributes:
        n_sensor (int): number of sensors in the sensor object used for analysis.
        offset (np.ndarray): array of sampled offset values, populated in self.from_mcmc() after the MCMC run is
            completed.
        precision_scalar (np.ndarray): array of sampled offset precision values, populated in self.from_mcmc() after
            the MCMC run is completed. Only populated if update_precision is True.
        indicator_basis (sparse.csc_matrix): [nof_observations x (nof_sensors - 1)] sparse matrix which assigns the
            offset parameters to the correct observations.
        update_precision (bool): logical indicating whether the offset prior precision parameter should be updated as
            part of the analysis.
        mean_offset (float): prior mean parameter for the offsets, assumed to be the same for each beam. Default is 0.
        prior_precision_shape (float): shape parameter for the prior gamma distribution for the scalar precision
            parameter. Default is 1e-3.
        prior_precision_rate (float): rate parameter for the prior gamma distribution for the scalar precision
            parameter(s). Default is 1e-3.
        initial_precision (float): initial value for the scalar precision parameter. Default is 1.0.

    """

    n_sensor: int = field(init=False)
    offset: np.ndarray = field(init=False)
    precision_scalar: np.ndarray = field(init=False)
    indicator_basis: sparse.csc_matrix = field(init=False)
    update_precision: bool = False
    mean_offset: float = 0.0
    prior_precision_shape: float = 1e-3
    prior_precision_rate: float = 1e-3
    initial_precision: float = 1.0

    def initialise(self, sensor_object: SensorGroup, meteorology: Meteorology, gas_species: GasSpecies):
        """Take data inputs and extract relevant properties.

        Args:
            sensor_object (SensorGroup): sensor data
            meteorology (MeteorologyGroup): meteorology data wind data
            gas_species (GasSpecies): gas species information

        """
        self.n_sensor = len(sensor_object)
        self.indicator_basis = sparse.csc_matrix(
            np.equal(sensor_object.sensor_index[:, np.newaxis], np.array(range(1, self.n_sensor)))
        )

    def make_model(self, model: list = None) -> list:
        """Take model list and append new elements from current model component.

        Args:
            model (list, optional): Current list of model elements. Defaults to [].

        Returns:
            list: model output list.

        """
        if model is None:
            model = []
        off_precision_predictor = parameter.ScaledMatrix(matrix="P_d", scalar="lambda_d")
        model.append(Normal("d", mean="mu_d", precision=off_precision_predictor))
        if self.update_precision:
            model.append(Gamma("lambda_d", shape="a_lam_d", rate="b_lam_d"))
        return model

    def make_sampler(self, model: Model, sampler_list: list = None) -> list:
        """Take sampler list and append new elements from current model component.

        Args:
            model (Model): Full model list of distributions.
            sampler_list (list, optional): Current list of samplers. Defaults to [].

        Returns:
            list: sampler output list.

        """
        if sampler_list is None:
            sampler_list = []
        sampler_list.append(NormalNormal("d", model))
        if self.update_precision:
            sampler_list.append(NormalGamma("lambda_d", model))
        return sampler_list

    def make_state(self, state: dict = None) -> dict:
        """Take state dictionary and append initial values from model component.

        Args:
            state (dict, optional): current state vector. Defaults to {}.

        Returns:
            dict: current state vector with components added.

        """
        if state is None:
            state = {}
        state["mu_d"] = np.ones((self.n_sensor - 1, 1)) * self.mean_offset
        state["d"] = np.zeros((self.n_sensor - 1, 1))
        state["B_d"] = self.indicator_basis
        state["P_d"] = sparse.eye(self.n_sensor - 1, format="csc")
        state["lambda_d"] = self.initial_precision
        if self.update_precision:
            state["a_lam_d"] = self.prior_precision_shape
            state["b_lam_d"] = self.prior_precision_rate
        return state

    def from_mcmc(self, store: dict):
        """Extract results of mcmc from mcmc.store and attach to components.

        Args:
            store (dict): mcmc result dictionary.

        """
        self.offset = store["d"]
        if self.update_precision:
            self.precision_scalar = store["lambda_d"]

    def plot_iterations(self, plot: "Plot", sensor_object: Union[SensorGroup, Sensor], burn_in_value: int) -> "Plot":
        """Plots the offset values for every sensor with respect to the MCMC iterations.

        Args:
            sensor_object (Union[SensorGroup, Sensor]): Sensor object associated with the offset_model
            burn_in_value (int): Burn in value to show in plot.
            plot (Plot): Plot object to which this figure will be added in the figure dictionary

        Returns:
            plot (Plot): Plot object to which this figure is added in the figure dictionary with
                key 'offset_iterations'

        """
        plot.plot_trace_per_sensor(
            object_to_plot=self, sensor_object=sensor_object, plot_type="line", burn_in=burn_in_value
        )

        return plot

    def plot_distributions(self, plot: "Plot", sensor_object: Union[SensorGroup, Sensor], burn_in_value: int) -> "Plot":
        """Plots the distribution of the offset values after the burn in for every sensor.

        Args:
            sensor_object (Union[SensorGroup, Sensor]): Sensor object associated with the offset_model
            burn_in_value (int): Burn in value to use for plot.
            plot (Plot): Plot object to which this figure will be added in the figure dictionary

        Returns:
            plot (Plot): Plot object to which this figure is added in the figure dictionary with
                key 'offset_distributions'

        """
        plot.plot_trace_per_sensor(
            object_to_plot=self, sensor_object=sensor_object, plot_type="box", burn_in=burn_in_value
        )

        return plot

initialise(sensor_object, meteorology, gas_species)

Take data inputs and extract relevant properties.

Parameters:

Name Type Description Default
sensor_object SensorGroup

sensor data

 required
meteorology MeteorologyGroup

meteorology data wind data

 required
gas_species GasSpecies

gas species information

 required
Source code in src/pyelq/component/offset.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
def initialise(self, sensor_object: SensorGroup, meteorology: Meteorology, gas_species: GasSpecies):
    """Take data inputs and extract relevant properties.

    Args:
        sensor_object (SensorGroup): sensor data
        meteorology (MeteorologyGroup): meteorology data wind data
        gas_species (GasSpecies): gas species information

    """
    self.n_sensor = len(sensor_object)
    self.indicator_basis = sparse.csc_matrix(
        np.equal(sensor_object.sensor_index[:, np.newaxis], np.array(range(1, self.n_sensor)))
    )

make_model(model=None)

Take model list and append new elements from current model component.

Parameters:

Name Type Description Default
model list

Current list of model elements. Defaults to [].

 None

Returns:

Name Type Description
list list

model output list.
Source code in src/pyelq/component/offset.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def make_model(self, model: list = None) -> list:
    """Take model list and append new elements from current model component.

    Args:
        model (list, optional): Current list of model elements. Defaults to [].

    Returns:
        list: model output list.

    """
    if model is None:
        model = []
    off_precision_predictor = parameter.ScaledMatrix(matrix="P_d", scalar="lambda_d")
    model.append(Normal("d", mean="mu_d", precision=off_precision_predictor))
    if self.update_precision:
        model.append(Gamma("lambda_d", shape="a_lam_d", rate="b_lam_d"))
    return model

make_sampler(model, sampler_list=None)

Take sampler list and append new elements from current model component.

Parameters:

Name Type Description Default
model Model

Full model list of distributions.

 required
sampler_list list

Current list of samplers. Defaults to [].

 None

Returns:

Name Type Description
list list

sampler output list.
Source code in src/pyelq/component/offset.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def make_sampler(self, model: Model, sampler_list: list = None) -> list:
    """Take sampler list and append new elements from current model component.

    Args:
        model (Model): Full model list of distributions.
        sampler_list (list, optional): Current list of samplers. Defaults to [].

    Returns:
        list: sampler output list.

    """
    if sampler_list is None:
        sampler_list = []
    sampler_list.append(NormalNormal("d", model))
    if self.update_precision:
        sampler_list.append(NormalGamma("lambda_d", model))
    return sampler_list

make_state(state=None)

Take state dictionary and append initial values from model component.

Parameters:

Name Type Description Default
state dict

current state vector. Defaults to {}.

 None

Returns:

Name Type Description
dict dict

current state vector with components added.
Source code in src/pyelq/component/offset.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def make_state(self, state: dict = None) -> dict:
    """Take state dictionary and append initial values from model component.

    Args:
        state (dict, optional): current state vector. Defaults to {}.

    Returns:
        dict: current state vector with components added.

    """
    if state is None:
        state = {}
    state["mu_d"] = np.ones((self.n_sensor - 1, 1)) * self.mean_offset
    state["d"] = np.zeros((self.n_sensor - 1, 1))
    state["B_d"] = self.indicator_basis
    state["P_d"] = sparse.eye(self.n_sensor - 1, format="csc")
    state["lambda_d"] = self.initial_precision
    if self.update_precision:
        state["a_lam_d"] = self.prior_precision_shape
        state["b_lam_d"] = self.prior_precision_rate
    return state

from_mcmc(store)

Extract results of mcmc from mcmc.store and attach to components.

Parameters:

Name Type Description Default
store dict

mcmc result dictionary.

 required
Source code in src/pyelq/component/offset.py 
136
137
138
139
140
141
142
143
144
145
def from_mcmc(self, store: dict):
    """Extract results of mcmc from mcmc.store and attach to components.

    Args:
        store (dict): mcmc result dictionary.

    """
    self.offset = store["d"]
    if self.update_precision:
        self.precision_scalar = store["lambda_d"]

plot_iterations(plot, sensor_object, burn_in_value)

Plots the offset values for every sensor with respect to the MCMC iterations.

Parameters:

Name Type Description Default
sensor_object Union[SensorGroup, Sensor]

Sensor object associated with the offset_model

 required
burn_in_value int

Burn in value to show in plot.

 required
plot Plot

Plot object to which this figure will be added in the figure dictionary

 required

Returns:

Name Type Description
plot Plot

Plot object to which this figure is added in the figure dictionary with key 'offset_iterations'
Source code in src/pyelq/component/offset.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def plot_iterations(self, plot: "Plot", sensor_object: Union[SensorGroup, Sensor], burn_in_value: int) -> "Plot":
    """Plots the offset values for every sensor with respect to the MCMC iterations.

    Args:
        sensor_object (Union[SensorGroup, Sensor]): Sensor object associated with the offset_model
        burn_in_value (int): Burn in value to show in plot.
        plot (Plot): Plot object to which this figure will be added in the figure dictionary

    Returns:
        plot (Plot): Plot object to which this figure is added in the figure dictionary with
            key 'offset_iterations'

    """
    plot.plot_trace_per_sensor(
        object_to_plot=self, sensor_object=sensor_object, plot_type="line", burn_in=burn_in_value
    )

    return plot

plot_distributions(plot, sensor_object, burn_in_value)

Plots the distribution of the offset values after the burn in for every sensor.

Parameters:

Name Type Description Default
sensor_object Union[SensorGroup, Sensor]

Sensor object associated with the offset_model

 required
burn_in_value int

Burn in value to use for plot.

 required
plot Plot

Plot object to which this figure will be added in the figure dictionary

 required

Returns:

Name Type Description
plot Plot

Plot object to which this figure is added in the figure dictionary with key 'offset_distributions'
Source code in src/pyelq/component/offset.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def plot_distributions(self, plot: "Plot", sensor_object: Union[SensorGroup, Sensor], burn_in_value: int) -> "Plot":
    """Plots the distribution of the offset values after the burn in for every sensor.

    Args:
        sensor_object (Union[SensorGroup, Sensor]): Sensor object associated with the offset_model
        burn_in_value (int): Burn in value to use for plot.
        plot (Plot): Plot object to which this figure will be added in the figure dictionary

    Returns:
        plot (Plot): Plot object to which this figure is added in the figure dictionary with
            key 'offset_distributions'

    """
    plot.plot_trace_per_sensor(
        object_to_plot=self, sensor_object=sensor_object, plot_type="box", burn_in=burn_in_value
    )

    return plot