"""Noise models are used to simulate the noise in bandflux measurements for a given set
of observations in an ObsTable. They extract information about the instrument and observing
conditions from the ObsTable and then apply noise to the input bandflux measurements.
Each noise model uses a given set of parameters with given units from the ObsTable.
Users adding new observation parameters should derive the necessary parameters in the
necessary units in the ObsTable's `_derive_noise_columns` method.
Alternatively users can create new noise models that use different input parameter/unit
combinations.
"""
from abc import ABC, abstractmethod
import numpy as np
from lightcurvelynx.astro_utils.mag_flux import mag2flux
from lightcurvelynx.noise_models.noise_utils import poisson_bandflux_std
[docs]
class FluxNoiseModel(ABC):
"""An abstract baseclass noise model for simulating bandflux measurements."""
# A list of column names that must be present in the ObsTable for this noise model to work.
_required_values = []
@property
[docs]
def required_values(self):
"""List of column names that must be present in the ObsTable for this noise model to work."""
return self._required_values
@abstractmethod
[docs]
def apply_noise(
self,
bandflux,
*,
obs_table=None,
indices=None,
rng=None,
**kwargs,
):
"""Compute the noise parameters for given observations in
an ObsTable and apply noise to the input bandflux.
Parameters
----------
bandflux : array_like of float
Source bandflux in energy units, e.g. nJy.
obs_table : ObsTable, optional
Table containing the observation parameters, including all
parameters needed to compute the noise.
indices : array_like of int, optional
Indices of the observations in the ObsTable to which noise should
be applied.
rng : np.random.Generator, optional
The random number generator to use for applying noise. If None,
a default generator will be used.
**kwargs
Additional parameters for the noise model.
Returns
-------
flux : array_like
The updated flux measurements after applying noise, in the same
units as the input bandflux.
flux_err : array_like
The bandflux measurement error used for applying noise, in the
same units as the input bandflux.
"""
raise NotImplementedError("Subclasses must implement this method.")
[docs]
def check_compatibility(self, obs_table, fail_on_incompatible=False):
"""Check if the noise model is compatible with the given ObsTable.
Parameters
----------
obs_table : ObsTable
The observation table to check for compatibility.
fail_on_incompatible : bool, optional
If True, raise a ValueError if the noise model is not compatible with the ObsTable.
If False, simply return False in that case. Default is False.
Returns
-------
bool
True if the noise model is compatible with the ObsTable, False otherwise.
"""
missing_columns = [col for col in self._required_values if col not in obs_table]
if missing_columns:
if fail_on_incompatible:
raise ValueError(
f"Noise model {self.__class__.__name__} is not compatible with the given ObsTable. "
f"Missing required columns: {missing_columns}"
)
return False
# Check if the required columns have valid data for each row.
for col in self._required_values:
values = obs_table.get_value_per_row(col)
if np.issubdtype(values.dtype, np.number) and not np.isfinite(values).all():
if fail_on_incompatible:
raise ValueError(f"Found invalid values in column '{col}'")
return False
return True
[docs]
class ConstantFluxNoiseModel(FluxNoiseModel):
"""A noise model that simulates photon noise for bandflux measurements
sampled from a normal distribution with a constant standard deviation.
This class is primarily meant for testing purposes.
Attributes
----------
noise_level : float
The (constant) standard deviation of the noise to apply to the bandflux
measurements, in the same units as the input bandflux.
"""
def __init__(self, noise_level):
if noise_level < 0:
raise ValueError("Noise level must be non-negative.")
[docs]
self.noise_level = noise_level
[docs]
def apply_noise(
self,
bandflux,
*,
rng=None,
**kwargs,
):
"""Compute the noise parameters for given observations in
an ObsTable and apply noise to the input bandflux.
Parameters
----------
bandflux : array_like of float
Source bandflux in energy units, e.g. nJy.
rng : np.random.Generator, optional
The random number generator to use for applying noise. If None,
a default generator will be used.
**kwargs
Additional parameters for the noise model.
Returns
-------
flux : array_like
The updated flux measurements after applying noise, in the same
units as the input bandflux.
flux_err : array_like
The bandflux measurement error used for applying noise, in the
same units as the input bandflux.
"""
if rng is None:
rng = np.random.default_rng()
noisy_bandflux = rng.normal(loc=bandflux, scale=self.noise_level)
return noisy_bandflux, np.full_like(bandflux, self.noise_level, dtype=float)
[docs]
class PoissonFluxNoiseModel(FluxNoiseModel):
"""A noise model that simulates photon noise for bandflux measurements
with a Poisson noise level that are extracted from values in an ObsTable.
This noise model uses the following values from the ObsTable:
- dark_current: Mean dark current (electrons per pixel per second).
- exptime: The total exposure time for the observation (seconds).
- nexposure: The number of exposures (optional, default is 1).
- psf_footprint: Point spread function effective area (pixel^2).
- read_noise: Standard deviation of the readout electrons per pixel per exposure.
- sky_bg_e: Sky background (electrons / pixel^2).
- zp: The photometric zero point (nJy / electron).
- zp_err_mag: The uncertainty in the photometric zero point in magnitudes (optional, default is 0.0).
Users should ensure that the necessary parameters (in the correct units) are derived in
the ObsTable's `_derive_noise_columns` method.
"""
# Note that both nexposure and zp_err_mag can fall back to default values.
_required_values = ["exptime", "sky_bg_e", "psf_footprint", "zp", "read_noise", "dark_current"]
def __init__(self):
pass
[docs]
def compute_flux_error(self, bandflux, obs_table, indices):
"""Compute the flux error for the given bandflux and observation parameters.
Parameters
----------
bandflux : array_like of float
Source bandflux in nJy.
obs_table : ObsTable
Table containing the observation parameters needed to compute the noise.
indices : array_like of int
Indices of the observations in the ObsTable for which to compute the noise.
Returns
-------
flux_err : array_like
The standard deviation of the bandflux measurement error (in nJy)
"""
# Extract the features needed to compute the noise from the ObsTable.
total_exposure_time = obs_table.get_value_per_row("exptime", indices=indices)
exposure_count = obs_table.get_value_per_row("nexposure", indices=indices, default=1)
sky = obs_table.get_value_per_row("sky_bg_e", indices=indices)
psf_footprint = obs_table.get_value_per_row("psf_footprint", indices=indices)
zp = obs_table.get_value_per_row("zp", indices=indices)
readout_noise = obs_table.get_value_per_row("read_noise", indices=indices)
dark_current = obs_table.get_value_per_row("dark_current", indices=indices)
zp_err_mag = obs_table.get_value_per_row("zp_err_mag", indices=indices, default=0.0)
# Compute the flux error standard deviation.
return poisson_bandflux_std(
bandflux,
total_exposure_time=total_exposure_time,
exposure_count=exposure_count,
psf_footprint=psf_footprint,
sky=sky,
zp=zp,
readout_noise=readout_noise,
dark_current=dark_current,
zp_err_mag=zp_err_mag,
)
[docs]
def apply_noise(
self,
bandflux,
*,
obs_table=None,
indices=None,
rng=None,
**kwargs,
):
"""Compute the noise parameters for given observations in
an ObsTable and apply noise to the input bandflux.
Parameters
----------
bandflux : array_like of float
Source bandflux in energy units, e.g. nJy.
obs_table : ObsTable, optional
Table containing the observation parameters, including all
parameters needed to compute the noise.
indices : array_like of int, optional
Indices of the observations in the ObsTable to which noise should
be applied.
rng : np.random.Generator, optional
The random number generator to use for applying noise. If None,
a default generator will be used.
**kwargs
Additional parameters for the noise model.
Returns
-------
flux : array_like
The updated flux measurements after applying noise, in the same
units as the input bandflux.
flux_err : array_like
The bandflux measurement error used for applying noise, in the
same units as the input bandflux.
"""
if obs_table is None:
raise ValueError("ObsTable must be provided for PoissonFluxNoiseModel.")
if indices is None:
raise ValueError("Indices must be provided for PoissonFluxNoiseModel.")
if len(indices) != len(bandflux):
raise ValueError("Length of indices must match length of bandflux.")
flux_err = self.compute_flux_error(
bandflux,
obs_table=obs_table,
indices=indices,
)
# Make sure the array is a numpy array.
flux_err = np.asarray(flux_err)
# Generate the actual noisy bandflux measurements.
rng = np.random.default_rng(rng)
noisy_bandflux = rng.normal(loc=bandflux, scale=flux_err)
return noisy_bandflux, flux_err
[docs]
class GivenNoiseModel(FluxNoiseModel):
"""A noise model that simulates photon noise for bandflux measurements with a
given (per-row) noise level.
This noise model uses the following values from the ObsTable:
- bandflux_error: The standard deviation of the noise to apply to the bandflux measurements (nJy).
"""
_required_values = ["bandflux_error"]
def __init__(self):
pass
[docs]
def apply_noise(
self,
bandflux,
*,
obs_table=None,
indices=None,
rng=None,
**kwargs,
):
"""Compute the noise parameters for given observations in
an ObsTable and apply noise to the input bandflux.
Parameters
----------
bandflux : array_like of float
Source bandflux in energy units, e.g. nJy.
obs_table : ObsTable, optional
Table containing the observation parameters, including all
parameters needed to compute the noise.
indices : array_like of int, optional
Indices of the observations in the ObsTable to which noise should
be applied.
rng : np.random.Generator, optional
The random number generator to use for applying noise. If None,
a default generator will be used.
**kwargs
Additional parameters for the noise model.
Returns
-------
flux : array_like
The updated flux measurements after applying noise, in the same
units as the input bandflux.
flux_err : array_like
The bandflux measurement error used for applying noise, in the
same units as the input bandflux.
"""
if obs_table is None:
raise ValueError("ObsTable must be provided for GivenNoiseModel.")
if indices is None:
raise ValueError("Indices must be provided for GivenNoiseModel.")
if len(indices) != len(bandflux):
raise ValueError("Length of indices must match length of bandflux.")
flux_err = obs_table.get_value_per_row("bandflux_error", indices=indices)
flux_err = np.asarray(flux_err, dtype=float)
# Generate the actual noisy bandflux measurements.
rng = np.random.default_rng(rng)
noisy_bandflux = rng.normal(loc=bandflux, scale=flux_err)
return noisy_bandflux, flux_err
[docs]
class FiveSigmaDepthNoiseModel(FluxNoiseModel):
"""A noise model that simulates photon noise from only the five-sigma depth information.
This noise model uses the following values from the ObsTable:
- five_sigma_depth: The five-sigma depth in AB magnitudes.
Note
----
This noise model is not as accurate as the PoissonFluxNoiseModel, but it can be used when
only the five-sigma depth is available in the ObsTable.
"""
_required_values = ["five_sigma_depth"]
def __init__(self):
pass
[docs]
def apply_noise(
self,
bandflux,
*,
obs_table=None,
indices=None,
rng=None,
**kwargs,
):
"""Compute the noise parameters for given observations in
an ObsTable and apply noise to the input bandflux.
Parameters
----------
bandflux : array_like of float
Source bandflux in energy units, e.g. nJy.
obs_table : ObsTable, optional
Table containing the observation parameters, including all
parameters needed to compute the noise.
indices : array_like of int, optional
Indices of the observations in the ObsTable to which noise should
be applied.
rng : np.random.Generator, optional
The random number generator to use for applying noise. If None,
a default generator will be used.
**kwargs
Additional parameters for the noise model.
Returns
-------
flux : array_like
The updated flux measurements after applying noise, in the same
units as the input bandflux.
flux_err : array_like
The bandflux measurement error used for applying noise, in the
same units as the input bandflux.
"""
if obs_table is None:
raise ValueError("ObsTable must be provided for FiveSigmaDepthNoiseModel.")
if indices is None:
raise ValueError("Indices must be provided for FiveSigmaDepthNoiseModel.")
if len(indices) != len(bandflux):
raise ValueError("Length of indices must match length of bandflux.")
# Compute the standard deviation of the noise from the five-sigma depth information.
# This uses five_sigma_depth in AB magnitudes, so we convert it to bandflux in nJy first.
five_sigma_depth = obs_table.get_value_per_row("five_sigma_depth", indices=indices)
flux_err = mag2flux(five_sigma_depth) / 5.0
# Generate the actual noisy bandflux measurements.
rng = np.random.default_rng(rng)
noisy_bandflux = rng.normal(loc=bandflux, scale=flux_err)
return noisy_bandflux, flux_err