|
lsst.pipe.tasks g44ede4a5c5+f0c59e294f
|
Classes | |
| class | PrettyMosaicConfig |
| class | PrettyMosaicConnections |
| class | PrettyMosaicTask |
| class | PrettyPictureBackgroundFixerConnections |
| class | PrettyPictureConnections |
| class | PrettyPictureStarFixerConfig |
| class | PrettyPictureStarFixerConnections |
| class | PrettyPictureStarFixerTask |
Functions | |
| Struct | run (self, Mapping[str, Exposure] images, Projection[Any]|None image_wcs=None, Box|None image_box=None) |
| None | runQuantum (self, QuantumContext butlerQC, InputQuantizedConnection inputRefs, OutputQuantizedConnection outputRefs) |
| dict[str, Exposure] | makeInputsFromRefs (self, Iterable[DatasetRef] refs, Butler|QuantumContext butler) |
| dict[str, DeferredDatasetHandle] | makeInputsFromArrays (self, **kwargs) |
| dict[int, DeferredDatasetHandle] | makeInputsFromExposures (self, **kwargs) |
| findBackgroundPixels (self, image, threshold_pair=None) | |
| _findControlPoints (self, Exposure exposure, Point2I origin, bool use_detection_mask=False, tuple[float, float]|None threshhold_pair=None) | |
| fixBackground (self, image, list[float] yloc, list[float] xloc, list[float] values) | |
| run (self, Exposure inputCoadd, list[Exposure]|None neighbors=None) | |
Variables | |
| logger = logging.getLogger(__name__) | |
| doWrite | |
| r | |
| g | |
| b | |
| array : `numpy.array` | |
| mean : `float` | |
| sigma : `float` | |
| tuple | lower_pos = (array - mu) < sigma |
| sigma_ratio = min_sig / sigma | |
| arr : `numyp.ndarray` | |
| maxLikely = np.median(image[image < max_search_flux], axis=None) | |
| mask = image < maxLikely | |
| tuple | initial_std = (image[mask] - maxLikely).std() |
| sub_image = image[image < max(maxLikely + 3 * initial_std, max_search_flux)] | |
| center = mode(np.round(sub_image, 2)).mode | |
| _ | |
| sigma_hat | |
| floc | |
| mu_hat = center | |
| new_cut = image[image < (mu_hat + 3 * sigma_hat)] | |
| positivity_ratio = np.sum(new_cut > mu_hat) / np.sum(new_cut < mu_hat) | |
| threshold_pos = min(mu_hat + pos_sigma_mult * sigma_hat, max_search_flux) | |
| threshold_neg = mu_hat - pos_sigma_mult * sigma_hat | |
| lsst.pipe.tasks.prettyPictureMaker._task.findBackgroundPixels | ( | self, | |
| image, | |||
| threshold_pair = None ) |
Find pixels that are likely to be background based on image statistics.
This method estimates background pixels by analyzing the distribution of
pixel values in the image. It uses the median as an estimate of the background
level and fits a half-normal distribution to values below the median to
determine the background sigma. Pixels below a threshold (mean + sigma) are
classified as background.
Parameters
----------
image : `numpy.ndarray`
Input image array for which to find background pixels.
threshold_pair : `tuple` of `float`, `float` or None
A tuple representing the bottom and top values for which all pixels inside
these bounds should be considered background. If `None` theses bounds are
determined from the image statistics.
Returns
-------
result : `numpy.ndarray`
Boolean mask array where True indicates background pixels.
Notes
-----
This method works best for images with relatively uniform background. It may
not perform well in fields with high density or diffuse flux, as noted in
the implementation comments.
| lsst.pipe.tasks.prettyPictureMaker._task.fixBackground | ( | self, | |
| image, | |||
| list[float] | yloc, | ||
| list[float] | xloc, | ||
| list[float] | values ) |
Estimate and subtract the background from an image.
This function estimates the background level in an image using supplied control
values using Gaussian fitting and radial basis function interpolation.
It aims to provide a more accurate background estimation than a simple median
filter, especially in images with varying background levels.
Parameters
----------
image : `numpy.ndarray`
The input image as a NumPy array.
yloc : `list` of `float`
The list of y control points
xloc : `list` of `float`
The list of x control points
values : `list` of `float`
The list of the values at the control points
Returns
-------
numpy.ndarray
An array representing the estimated background level across the image.
| dict[str, DeferredDatasetHandle] lsst.pipe.tasks.prettyPictureMaker._task.makeInputsFromArrays | ( | self, | |
| ** | kwargs ) |
Make valid inputs for the run method from numpy arrays.
Parameters
----------
kwargs : `numpy.ndarray`
This is standard python kwargs where the left side of the equals
is the data band, and the right side is the corresponding `numpy.ndarray`
array.
Returns
-------
sortedImages : `dict` of `str` to \
`~lsst.daf.butler.DeferredDatasetHandle`
A dictionary of `~lsst.daf.butlger.DeferredDatasetHandle`\ s keyed
by the band they correspond to.
| dict[int, DeferredDatasetHandle] lsst.pipe.tasks.prettyPictureMaker._task.makeInputsFromExposures | ( | self, | |
| ** | kwargs ) |
Make valid inputs for the run method from `Exposure` objects.
Parameters
----------
kwargs : `Exposure`
This is standard python kwargs where the left side of the equals
is the data band, and the right side is the corresponding
`Exposure`.
Returns
-------
sortedImages : `dict` of `int` to \
`~lsst.daf.butler.DeferredDatasetHandle`
A dictionary of `~lsst.daf.butler.DeferredDatasetHandle`\ s keyed
by the band they correspond to.
| dict[str, Exposure] lsst.pipe.tasks.prettyPictureMaker._task.makeInputsFromRefs | ( | self, | |
| Iterable[DatasetRef] | refs, | ||
| Butler | QuantumContext | butler ) |
Make valid inputs for the run method from butler references.
Parameters
----------
refs : `Iterable` of `DatasetRef`
Some `Iterable` container of `Butler` `DatasetRef`\ s
butler : `Butler` or `QuantumContext`
This is the object that fetches the input data.
Returns
-------
sortedImages : `dict` of `str` to `Exposure`
A dictionary of `Exposure`\ s keyed by the band they
correspond to.
| lsst.pipe.tasks.prettyPictureMaker._task.run | ( | self, | |
| Exposure | inputCoadd, | ||
| list[Exposure] | None | neighbors = None ) |
Estimate a background for an input Exposure and remove it.
Parameters
----------
inputCoadd : `Exposure`
The exposure the background will be removed from.
neighbors : `list` of `Exposure`
Neighboring `Exposure` objects that can be used to constrain
backgrounds across boundaries.
Returns
-------
result : `Struct`
A `Struct` that contains the exposure with the background removed.
This `Struct` will have an attribute named ``outputCoadd``.
| Struct lsst.pipe.tasks.prettyPictureMaker._task.run | ( | self, | |
| Mapping[str, Exposure] | images, | ||
| Projection[Any] | None | image_wcs = None, | ||
| Box | None | image_box = None ) |
Turns the input arguments in arguments into an RGB array.
Parameters
----------
images : `Mapping` of `str` to `Exposure`
A mapping of input images and the band they correspond to.
image_wcs : `~lsst.images.Projection`, optional
A projection describing the sky coordinate of each pixel.
image_box : `~lsst.images.Box`, optional
A box that defines this image as part of a larger region.
Returns
-------
result : `Struct`
A struct with the corresponding RGB image, and mask used in
RGB image construction. The struct will have the attributes
outputRGB and outputRGBMask. Each of the outputs will
be a `~lsst.images.ColorImage` object.
Notes
-----
Construction of input images are made easier by use of the
makeInputsFrom* methods.
| None lsst.pipe.tasks.prettyPictureMaker._task.runQuantum | ( | self, | |
| QuantumContext | butlerQC, | ||
| InputQuantizedConnection | inputRefs, | ||
| OutputQuantizedConnection | outputRefs ) |
| lsst.pipe.tasks.prettyPictureMaker._task.arr : `numyp.ndarray` |
| lsst.pipe.tasks.prettyPictureMaker._task.array : `numpy.array` |
r = Field[float](doc="The amount of red contained in this channel")
g = Field[float](doc="The amount of green contained in this channel")
b = Field[float](doc="The amount of blue contained in this channel")
class PrettyPictureConfig(PipelineTaskConfig, pipelineConnections=PrettyPictureConnections):
channelConfig = ConfigDictField(
doc="A dictionary that maps band names to their rgb channel configurations",
keytype=str,
itemtype=ChannelRGBConfig,
default={},
)
cieWhitePoint = ListField[float](
doc="The white point of the input arrays in ciexz coordinates", maxLength=2, default=[0.28, 0.28]
)
arrayType = ChoiceField[str](
doc="The dataset type for the output image array",
default="uint8",
allowed={
"uint8": "Use 8 bit arrays, 255 max",
"uint16": "Use 16 bit arrays, 65535 max",
"half": "Use 16 bit float arrays, 1 max",
"float": "Use 32 bit float arrays, 1 max",
},
)
recenterNoise = Field[float](
doc="Recenter the noise away from zero. Supplied value is in units of sigma",
optional=True,
default=None,
)
noiseSearchThreshold = Field[float](
doc=(
"Flux threshold below which most flux will be considered noise, used to estimate noise properties"
),
default=2,
)
maxNoiseImbalance = Field[float](
doc=(
"When recentering noise, if the ratio of counts of positive pixels, to negative pixels passes "
"this threshold, consider there to be extended low flux and only estimate noise below zero."
),
default=1.5,
)
doPsfDeconvolve = Field[bool](
doc="Use the PSF in a Richardson-Lucy deconvolution on the luminance channel.", default=False
)
doPSFDeconcovlve = Field[bool](
doc="Use the PSF in a Richardson-Lucy deconvolution on the luminance channel.",
default=False,
deprecated="This field will be removed in v32. Use doPsfDeconvolve instead.",
optional=True,
)
doRemapGamut = Field[bool](
doc="Apply a color correction to unrepresentable colors; if False, clip them.", default=True
)
doExposureBrackets = Field[bool](
doc="Apply exposure bracketing to aid in dynamic range compression", default=True
)
doLocalContrast = Field[bool](doc="Apply local contrast optimizations to luminance.", default=True)
imageRemappingConfig = ConfigurableActionField[BoundsRemapper](
doc="Action controlling normalization process"
)
luminanceConfig = ConfigurableActionField[LumCompressor](
doc="Action controlling luminance scaling when making an RGB image"
)
localContrastConfig = ConfigurableActionField[LocalContrastEnhancer](
doc="Action controlling the local contrast correction in RGB image production"
)
colorConfig = ConfigurableActionField[ColorScaler](
doc="Action to control the color scaling process in RGB image production"
)
exposureBracketerConfig = ConfigurableActionField[ExposureBracketer](
doc=(
"Exposure scaling action used in creating multiple exposures with different scalings which will "
"then be fused into a final image"
),
)
gamutMapperConfig = ConfigurableActionField[GamutFixer](
doc="Action to fix pixels which lay outside RGB color gamut"
)
exposureBrackets = ListField[float](
doc=(
"Exposure scaling factors used in creating multiple exposures with different scalings which will "
"then be fused into a final image"
),
optional=True,
default=[1.25, 1, 0.75],
deprecated=(
"This field will stop working in v31 and be removed in v32, "
"please set exposureBracketerConfig.exposureBrackets"
),
)
gamutMethod = ChoiceField[str](
doc="If doRemapGamut is True this determines the method",
default="inpaint",
allowed={
"mapping": "Use a mapping function",
"inpaint": "Use surrounding pixels to determine likely value",
},
deprecated="This field will stop working in v31 and be removed in v32, please set gamutMapperConfig",
)
def setDefaults(self):
self.channelConfig["i"] = ChannelRGBConfig(r=1, g=0, b=0)
self.channelConfig["r"] = ChannelRGBConfig(r=0, g=1, b=0)
self.channelConfig["g"] = ChannelRGBConfig(r=0, g=0, b=1)
return super().setDefaults()
def _handle_deprecated(self):
# check if gamutMethod is set
if len(self._history["gamutMethod"]) > 1:
# This has been set in config, update it in the new location
self.gamutMapperConfig.gamutMethod = self.gamutMethod
if len(self._history["exposureBrackets"]) > 1:
self.exposureBracketerConfig.exposureBrackets = self.exposureBrackets
if self.exposureBrackets is None:
self.doExposureBrackets = False
if len(self.localContrastConfig._history["doLocalContrast"]) > 1:
self.doLocalContrast = self.localContrastConfig.doLocalContrast
# Handle doPsfDeconcovlve typo fix
if len(self._history["doPSFDeconcovlve"]) > 1:
self.doPsfDeconvolve = self.doPSFDeconcovlve
def freeze(self):
# ensure this is not already frozen
if self._frozen is not True:
self._handle_deprecated()
super().freeze()
class PrettyPictureTask(PipelineTask):
_DefaultName = "prettyPicture" ConfigClass = PrettyPictureConfig config: ConfigClass def _find_normal_stats(self, array):
| lsst.pipe.tasks.prettyPictureMaker._task.center = mode(np.round(sub_image, 2)).mode |
| lsst.pipe.tasks.prettyPictureMaker._task.logger = logging.getLogger(__name__) |
# Extract negative values efficiently
values_noise = array[array < self.config.noiseSearchThreshold]
# find the mode
center = mode(np.round(values_noise, 2)).mode
# extract the negative values
values_neg = array[array < center]
# Return infinity if no negative values found
if values_neg.size == 0:
return 0, np.inf
try:
# Fit half-normal distribution to absolute negative values
_, sigma = halfnorm.fit(np.abs(values_neg - center), floc=0)
mu = center
except (ValueError, RuntimeError):
# Handle fitting failures (e.g., constant data, optimization issues)
return 0, np.inf
# examine for excess positive flux, this means there is contaminating signal
new_cut = array[array < (mu + 3 * sigma)]
positivity_ratio = np.sum(new_cut > mu) / np.sum(new_cut < mu)
if positivity_ratio > self.config.maxNoiseImbalance:
# This means there is an excess flux, possibly diffuse source,
# only estimate around zero.
mu, sigma = halfnorm.fit(np.abs(values_noise[values_noise < 0]), floc=0)
return mu, sigma
def _match_sigmas_and_recenter(self, *arrays, factor=1):
| lsst.pipe.tasks.prettyPictureMaker._task.mask = image < maxLikely |
| lsst.pipe.tasks.prettyPictureMaker._task.maxLikely = np.median(image[image < max_search_flux], axis=None) |
| lsst.pipe.tasks.prettyPictureMaker._task.sigma_ratio = min_sig / sigma |
| lsst.pipe.tasks.prettyPictureMaker._task.sub_image = image[image < max(maxLikely + 3 * initial_std, max_search_flux)] |