#!/usr/bin/env python3
# pylint: disable=too-many-lines
"""Handle data processing DB interactions."""
import logging
from math import inf
from sqlalchemy import sql, select, update, and_, or_
from astropy.coordinates import SkyCoord
from astropy import units as astropy_units
from autowisp.multiprocessing_util import setup_process
from autowisp.database.processing import ProcessingManager
from autowisp.database.interface import start_db_session, get_project_home
from autowisp.exceptions import Component, MasterSelectionError, PipelineError
from autowisp.error_context import capture_errors, error_context
from autowisp import processing_steps
from autowisp.database.user_interface import get_processing_sequence
from autowisp.data_reduction.data_reduction_file import DataReductionFile
from autowisp.evaluator import Evaluator
# False positive due to unusual importing
# pylint: disable=no-name-in-module
from autowisp.astrometry import Transformation
from autowisp.database.data_model import (
StepDependencies,
ImageProcessingProgress,
ProcessedImages,
Step,
Image,
ImageDiagnostics,
PhotometryDiagnostics,
DiagnosticType,
ObservingSession,
MasterType,
MasterFile,
InputMasterTypes,
Condition,
ConditionExpression,
ImageMasterSelection,
)
from autowisp.database.data_model.provenance import (
Camera,
CameraChannel,
CameraType,
)
# pylint: enable=no-name-in-module
[docs]
class NoMasterError(MasterSelectionError):
"""Raised when no suitable master can be found for a batch of frames."""
# Intended to be used as simple callable
# pylint: disable=too-few-public-methods
[docs]
class ExpressionMatcher:
"""
Compare condition expressions for an image/channel to a target.
Usually check if matched expressions and master expression values are
identical, but also handles special case of calibrate step.
"""
[docs]
def _get_master_values(self, image_id, channel):
"""Return ready to compare masster expression values."""
if channel is None:
return tuple(
self._get_master_values(image_id, channel)
for channel in sorted(
filter(None, self._evaluated_expressions[image_id].keys())
)
)
self._logger.debug(
"Getting master expression values for expression ids %s, "
"image %d, channel %s",
repr(self._master_expression_ids),
image_id,
channel,
)
return tuple(
self._evaluated_expressions[image_id][channel]["values"][
expression_id
]
for expression_id in self._master_expression_ids
)
[docs]
def __init__( # pylint: disable=too-many-arguments
self,
evaluated_expressions,
ref_image_id,
ref_channel,
master_expression_ids,
*,
masters_only=False,
):
"""
Set up comparison to the given evaluated expressions.
"""
self._logger = logging.getLogger(__name__)
self._evaluated_expressions = evaluated_expressions
self._master_expression_ids = master_expression_ids
reference_evaluated = evaluated_expressions[ref_image_id][ref_channel]
self._ref_matched = reference_evaluated["matched"]
self.ref_master_values = self._get_master_values(
ref_image_id, ref_channel
)
self._masters_only = masters_only
self._logger.debug(
"Finding images matching expressions %s and values %s",
repr(self._ref_matched),
repr(self.ref_master_values),
)
[docs]
def __call__(self, image_id, channel):
"""True iff the expressions for the given image/channel match."""
image_evaluated = self._evaluated_expressions[image_id][channel]
image_master_values = self._get_master_values(image_id, channel)
self._logger.debug(
"Comparing %s to %s and %s to %s",
repr(image_evaluated["matched"]),
repr(self._ref_matched),
repr(image_master_values),
repr(self.ref_master_values),
)
return (
self._masters_only
or image_evaluated["matched"] == self._ref_matched
) and image_master_values == self.ref_master_values
# pylint: enable=too-few-public-methods
[docs]
def get_master_expression_ids(step_id, image_type_id, db_session):
"""
List all condition expression IDs determining input or output masters.
Args:
step_id(int): The ID of the step for which to return the master
expression IDs.
image_type_id(int): The type of images being processed by the step
for which to return the master expression IDs.
Returns:
[int]:
The combined expression IDs reqired to determine which required
masters can be used for the given step or which masters will be
created by it.
"""
return sorted(
set(
db_session.scalars(
select(ConditionExpression.id)
.select_from(InputMasterTypes)
.join(MasterType)
.join(
Condition,
# False positive
# pylint: disable=no-member
MasterType.condition_id == Condition.id,
# pylint: enable=no-member
)
.join(ConditionExpression)
.where(InputMasterTypes.step_id == step_id)
.where(InputMasterTypes.image_type_id == image_type_id)
.group_by(ConditionExpression.id)
).all()
+ db_session.scalars(
select(ConditionExpression.id)
.select_from(MasterType)
.join(
Condition,
or_(
# False positive
# pylint: disable=no-member
MasterType.condition_id == Condition.id,
(
MasterType.maker_image_split_condition_id
== Condition.id
),
# pylint: enable-no-member
),
)
.join(ConditionExpression)
.where(MasterType.maker_step_id == step_id)
.where(MasterType.maker_image_type_id == image_type_id)
).all()
)
)
[docs]
def remove_failed_prerequisite(
pending, pending_image_type_id, prereq_step_id, db_session
):
"""Remove from pending any entries that failed the prerequisite step."""
prereq_statuses = [
db_session.execute(
select(ProcessedImages.status)
.outerjoin(ImageProcessingProgress)
.where(
(ProcessedImages.image_id == image.id),
ProcessedImages.channel == channel,
ImageProcessingProgress.step_id == prereq_step_id,
(
ImageProcessingProgress.image_type_id
== pending_image_type_id
),
)
).scalar_one_or_none()
for image, channel, _ in pending
]
dropped = []
for i in range(len(pending) - 1, -1, -1):
if prereq_statuses[i] and prereq_statuses[i] < 0:
dropped.append(pending.pop(i))
return dropped
# pylint: disable=too-many-instance-attributes
[docs]
class ImageProcessingManager(ProcessingManager):
"""
Read configuration and record processing progress in the database.
Attrs:
See `ProcessingManager`.
pending(dict): Indexed by step ID, and image type ID list of
(Image, channel name, status) tuples listing all the images of the
given type that have not been processed by the currently selected
version of the step in the key and their status if previous
processing by that step was interrupted or None if not.
_failed_dependencies(dict): Dictionary with keys (step, image_type)
that contains the list of images and channels that failed the given
step.
"""
[docs]
def _set_calibration_config(self, config, first_image):
"""Retrun the specially formatted argument for the calibration step."""
config["split_channels"] = self._get_split_channels(first_image)
config["extra_header"] = self._get_extra_header(first_image)
result = {
(
"split_channels",
"".join(
repr(c)
for c in first_image.observing_session.camera.channels
),
),
("observing_session", config["extra_header"]["OBSSSNID"]),
}
self._logger.debug(
"Calibration step configuration:\n%s",
"\n\t".join((f"{k}: {v!r}" for k, v in config.items())),
)
return result
[docs]
def _split_by_master(self, batch, input_master_type):
"""Split the given list of images by the best master of given type."""
result = {}
for image, channel, status in batch:
if channel is None:
best_master = tuple(
(
channel,
self._evaluated_expressions[image.id][channel][
"masters"
][input_master_type.master_type.name],
)
for channel in sorted(
filter(
None, self._evaluated_expressions[image.id].keys()
)
)
)
else:
best_master = self._evaluated_expressions[image.id][channel][
"masters"
][input_master_type.master_type.name]
if best_master in result:
result[best_master].append((image, channel, status))
else:
result[best_master] = [(image, channel, status)]
return result
# Could not find good way to simplify
# pylint: disable=too-many-locals
[docs]
def _get_batch_config(
self, batch, master_expression_values, step, db_session
):
"""
Split given batch of images by configuration for given step.
The batch must already be split by all relevant condition expressions.
Only splits batches by the best master for each image.
Args:
batch([Image, channel, status]): List of database image instances
and for channels which to find the configuration(s). The channel
should be ``None`` for the ``calibrate`` step
master_expression_values(tuple): The values the expressions
required to select input masters or to guarantee a unique output
master. Should be provided in consistent order for all batches
processed by the same step.
step(Step): The database step instance to configure.
db_session: Database session to use for queries.
Returns:
dict:
keys: guaranteed to match iff configuration, output master
conditions, and all best input master(s) match. In other
words, if this function is called separately on multiple
batches, it is safe to combine and process together those
that end up with the same key.
values:
dict: The configuration to use for the given (sub-)batch.
[Image]: The (sub-)batch of images to process with given
configuration.
"""
self._logger.debug("Finding configuration for batch: %s", repr(batch))
first_image_expressions = self._evaluated_expressions[batch[0][0].id]
config, config_key = self.get_config(
first_image_expressions[batch[0][1]]["matched"],
db_session,
db_step=step,
)
config_key |= {master_expression_values}
if step.name == "calibrate":
config_key |= self._set_calibration_config(config, batch[0][0])
config["processing_step"] = step.name
config["image_type"] = batch[0][0].image_type.name
result = {config_key: (config, batch)}
for input_master_type in db_session.scalars(
select(InputMasterTypes).filter_by(
step_id=step.id, image_type_id=batch[0][0].image_type_id
)
).all():
for config_key, (config, sub_batch) in list(result.items()):
del result[config_key]
splits = self._split_by_master(sub_batch, input_master_type)
for best_master, sub_batch in splits.items():
if best_master is None:
if input_master_type.optional:
assert config_key not in result, (
"Two groups of images ended up sharing "
f"configuration {config_key} when split by "
f"{input_master_type.master_type.name} "
"master!"
)
result[config_key] = (config, sub_batch)
else:
result[None] = (
"No master "
+ input_master_type.master_type.name
+ " found!",
sub_batch,
)
else:
new_config = dict(config)
new_config[
input_master_type.config_name.replace("-", "_")
] = (
best_master
if isinstance(best_master, str)
else dict(best_master)
)
key_extra = {
(input_master_type.config_name, best_master)
}
result[config_key | key_extra] = (new_config, sub_batch)
return result
# pylint: enable=too-many-locals
[docs]
def _clean_pending_per_dependencies(
self, db_session, from_step_id=None, from_image_type_id=None
):
"""Remove pending images from steps if they failed a required step."""
dropped = {}
for (step_id, image_type_id), pending in self.pending.items():
if (
from_image_type_id is not None
and image_type_id != from_image_type_id
):
continue
for prereq_step_id in db_session.scalars(
select(StepDependencies.blocking_step_id).where(
StepDependencies.blocked_step_id == step_id,
StepDependencies.blocked_image_type_id == image_type_id,
StepDependencies.blocking_image_type_id == image_type_id,
StepDependencies.allow_pending.is_not(True),
)
):
if from_step_id is not None and prereq_step_id != from_step_id:
continue
if (step_id, image_type_id) not in dropped:
dropped[(step_id, image_type_id)] = []
failed_prereq = remove_failed_prerequisite(
pending, image_type_id, prereq_step_id, db_session
)
self.pending[(step_id, image_type_id)] = pending
dropped[(step_id, image_type_id)].extend(failed_prereq)
self._logger.info(
"The following image/channel combinations failed %s. "
"Excluding from %s:\n\t%s",
db_session.scalar(
select(Step.name).filter_by(id=prereq_step_id)
),
db_session.scalar(select(Step.name).filter_by(id=step_id)),
"\n\t".join(
image.raw_fname + ":" + channel
for image, channel in failed_prereq
),
)
return dropped
[docs]
def _require_blocking_complete(self, query, step, image_type, db_session):
"""
Restrict query to images where allow_pending blocking steps are done.
For same-type dependencies marked allow_pending=True, inner-join the
query so only images for which the blocking step has successfully
completed are included. Images where the blocking step is pending or
in-progress are silently deferred; images where it failed are handled
separately by the failed_prereq subquery in set_pending.
Args:
query: The base SQLAlchemy select to restrict.
step(Step): The blocked step being prepared.
image_type(ImageType): The blocked image type.
db_session: Active database session.
Returns:
The query with zero or more inner joins added.
"""
# pylint: disable=singleton-comparison
for (
blocking_step_id,
blocking_image_type_id,
) in db_session.execute(
select(
StepDependencies.blocking_step_id,
StepDependencies.blocking_image_type_id,
)
.where(StepDependencies.blocked_step_id == step.id)
.where(StepDependencies.blocked_image_type_id == image_type.id)
.where(StepDependencies.blocking_image_type_id == image_type.id)
.where(StepDependencies.allow_pending == True)
).all():
blocking_complete_sq = (
select(
ProcessedImages.image_id,
ProcessedImages.channel,
)
.join(ImageProcessingProgress)
.where(ImageProcessingProgress.step_id == blocking_step_id)
.where(
ImageProcessingProgress.image_type_id
== blocking_image_type_id
)
.where(ProcessedImages.final == True)
.where(ProcessedImages.status > 0)
.subquery()
)
query = query.join(
blocking_complete_sq,
and_(
Image.id # pylint: disable=no-member
== blocking_complete_sq.c.image_id,
CameraChannel.name == blocking_complete_sq.c.channel,
),
)
# pylint: enable=singleton-comparison
return query
[docs]
def _check_ready(self, step, image_type, db_session):
"""
Check if the given type of images is ready to process with given step.
Args:
step(Step): The step to check for readiness.
image_type(ImageType): The type of images to check for readiness.
db_session(Session): The database session to use.
Returns:
bool: Whether all requirements for the specified processing are
satisfied.
"""
for requirement in db_session.execute(
select(
StepDependencies.blocking_step_id,
StepDependencies.blocking_image_type_id,
)
.where(StepDependencies.blocked_step_id == step.id)
.where(StepDependencies.blocked_image_type_id == image_type.id)
.where(
or_(
StepDependencies.allow_pending.is_not(True),
StepDependencies.blocking_image_type_id != image_type.id,
)
)
).all():
if self.pending[requirement]:
self._logger.debug(
"Not ready for %s of %d %s frames because of %d pending %s "
"type ID images for step ID %s:\n\t%s",
step.name,
len(self.pending[(step.id, image_type.id)]),
image_type.name,
len(self.pending[requirement]),
requirement[1],
requirement[0],
"\n\t".join(
f"{e[0]!r}: {e[1]!r}" for e in self.pending[requirement]
),
)
return False
return True
[docs]
def _get_interrupted(self, need_cleanup, db_session):
"""Return list of interrupted files and configuration for cleanup."""
self.current_step = need_cleanup[0][2]
self._current_processing = db_session.scalar(
select(ImageProcessingProgress).where(
ImageProcessingProgress.id == need_cleanup[0][1].progress_id
)
)
input_type = getattr(
processing_steps, self.current_step.name
).input_type
for entry in need_cleanup:
assert entry[2] == self.current_step, (
f"Interrupted processing of step {entry[2].name} turned up "
f"while cleaning up after {self.current_step.name}!"
)
pending = [
(
image,
None if input_type == "raw" else processed.channel,
processed.status,
)
for image, processed, _ in need_cleanup
]
for image, _, __ in need_cleanup:
if image.id not in self._evaluated_expressions:
self.evaluate_expressions_image(image, db_session)
cleanup_batches = self._get_config_batches(
pending, input_type, db_session
)
result = {}
for (config_key, status), (config, batch) in cleanup_batches.items():
if config_key not in result:
result[config_key] = (config, [])
result[config_key][1].extend([(fname, status) for fname in batch])
return list(result.values())
[docs]
def _cleanup_interrupted(self, db_session):
"""Cleanup previously interrupted processing for the current step."""
need_cleanup = db_session.execute(
select(Image, ProcessedImages, Step)
.join(ProcessedImages)
.join(ImageProcessingProgress)
.join(Step)
.where(~ProcessedImages.final)
.order_by(Step.name)
).all()
if not need_cleanup:
return
step_module = getattr(processing_steps, need_cleanup[0][2].name)
for config, interrupted in self._get_interrupted(
need_cleanup, db_session
):
self._logger.warning(
"Cleaning up interrupted %s processing of %d images:\n"
"%s\n"
"config: %s",
need_cleanup[0][2],
len(interrupted),
repr(interrupted),
repr(config),
)
self.check_interrupted_statuses(
step_module, self.current_step.name, interrupted
)
new_status = step_module.cleanup_interrupted(interrupted, config)
# Whatever cleanup leaves behind is what the step will be
# started from next time, so it has to be a status the step can
# actually start from. -1 deletes the record entirely, leaving
# the step with no previous processing at all.
self.check_start_status(
step_module,
self.current_step.name,
None if new_status == -1 else new_status,
)
for _, processed, _ in need_cleanup:
assert new_status <= processed.status, (
f"Cleaning up interrupted {self.current_step.name} "
f"reported status {new_status}, further along than the "
f"{processed.status} reached before the interruption!"
)
if new_status == -1:
db_session.delete(processed)
else:
processed.status = new_status
[docs]
def _init_processed_ids(self, image, channels, step_input_type):
"""Prepare to record processing of the given image by current step."""
if channels == [None]:
channels = self._evaluated_expressions[image.id].keys()
for channel_name in channels:
if channel_name is None:
continue
step_input_fname = self.get_step_input(
image, channel_name, step_input_type
)
if step_input_fname not in self._processed_ids:
self._processed_ids[step_input_fname] = []
self._processed_ids[step_input_fname].append(
{"image_id": image.id, "channel": channel_name}
)
[docs]
def _start_step(self, step, image_type, db_session):
"""
Record the start of a processing step and return the images to process.
Args:
step(Step): The database step to start.
image_type(ImageType): The database type of image to start
processing.
db_session: Active session for database queries.
Returns:
[(Image, str)]:
The list of images and channels to process.
str:
The type of input expected by the current step.
"""
self._create_current_processing(
step, ("image_type", image_type.id), db_session
)
pending_images = self.pending[(step.id, image_type.id)].copy()
for image, channel, status in self._failed_dependencies.get(
(step.id, image_type.id), []
):
self._logger.info(
"Prerequisite failed for %s of %s", step.name, image
)
db_session.add(
ProcessedImages(
image_id=image.id,
channel=channel,
progress_id=self._current_processing.id,
status=-1,
final=True,
)
)
self._some_failed = True
self._processed_ids = {}
step_input_type = getattr(processing_steps, step.name).input_type
if step_input_type == "raw":
added = set()
new_pending = []
for image, _, status in pending_images:
if image.id not in added:
added.add(image.id)
new_pending.append((image, None, status))
pending_images = new_pending
for image, channel_name, _ in pending_images:
self.evaluate_expressions_image(image, db_session)
self._init_processed_ids(image, [channel_name], step_input_type)
self._logger.info(
"Starting %s step for %d %s images",
self.current_step.name,
len(pending_images),
image_type.name,
)
return pending_images, step_input_type
[docs]
def _process_batch( # pylint: disable=too-many-arguments
self, batch, *, start_status, config, step_name, image_type_name
):
"""Run the current step for a batch of images given configuration."""
# ``error_context`` (outer) scopes the step name and its resolved
# config so both are still active when ``_run_step``'s
# ``capture_errors`` stamps the exception on the way out -- the
# context manager's reset fires only after the inner ``except`` has
# run. Scoping config here (uniformly for every step) is what lets a
# parent-side error carry the failing step's config.
with error_context(step_name=step_name, config=config):
new_masters = self._run_step(batch, start_status, config, step_name)
if new_masters:
self.add_masters(new_masters, step_name, image_type_name)
[docs]
@capture_errors(component=Component.STEP)
def _run_step(self, batch, start_status, config, step_name):
"""Invoke the step's entry function for a batch of images."""
step_module = getattr(processing_steps, step_name)
self.check_start_status(step_module, step_name, start_status)
return getattr(step_module, step_name)(
batch,
start_status,
config,
self._start_processing,
self._end_processing,
)
[docs]
def _start_processing(self, input_fname, status=0):
"""
Mark in the database that processing the given file has begun.
Args:
input_fname: The filename of the input (DR or FITS) that is about
to begin processing.
Returns:
None
"""
assert (
self.current_step is not None
), f"Marking {input_fname} as started outside of any processing step!"
assert self._current_processing is not None, (
f"Marking {input_fname} as started before the "
f"{self.current_step.name} progress record was created!"
)
self._logger.debug(
"Starting processing IDs: %s",
repr(self._processed_ids[input_fname]),
)
with start_db_session() as db_session:
for starting_id in self._processed_ids[input_fname]:
db_session.add(
ProcessedImages(
**starting_id,
progress_id=self._current_processing.id,
status=status,
final=False,
)
)
[docs]
def _save_diagnostics(self, finished_id, diagnostics, db_session):
"""
Save diagnostic values for a single image/channel.
Args:
finished_id(dict): Must contain ``image_id`` and ``channel``
keys identifying the processed image and channel.
diagnostics: A list of ``(name, value)`` tuples where *name*
must match a :class:`DiagnosticType` row.
db_session: Active database session.
Raises:
ValueError: If a diagnostic name is not found in the
``diagnostic_type`` table.
"""
for diag_name, diag_value in diagnostics:
diag_type_id = db_session.scalar(
select(DiagnosticType.id).where(
DiagnosticType.name == diag_name
)
)
if diag_type_id is None:
if diag_name.startswith("pixel_q"):
quantile_digits = diag_name[len("pixel_q") :]
new_type = DiagnosticType(
name=diag_name,
description=(
f"The 0.{quantile_digits} quantile of "
"calibrated pixel values"
),
)
db_session.add(new_type)
db_session.flush()
diag_type_id = new_type.id
else:
raise PipelineError(
f"Unknown diagnostic type {diag_name!r}"
)
db_session.add(
ImageDiagnostics(
image_id=finished_id["image_id"],
channel=finished_id["channel"],
diagnostic_id=diag_type_id,
value=float(diag_value),
)
)
[docs]
def _save_photometry_diagnostics(
self, finished_id, photometry_diagnostics, db_session
):
"""
Save per-photometry diagnostic values for a single image/channel.
If a row already exists for the same image, channel, photometry,
and diagnostic type, its value is updated (this supports iterative
steps like magnitude fitting).
Args:
finished_id(dict): Must contain ``image_id`` and ``channel``
keys identifying the processed image and channel.
photometry_diagnostics: A list of
``(name, value, photometry_id)`` tuples where *name* must
match a :class:`DiagnosticType` row.
db_session: Active database session.
Raises:
ValueError: If a diagnostic name is not found in the
``diagnostic_type`` table.
"""
for diag_name, diag_value, phot_id in photometry_diagnostics:
diag_type_id = db_session.scalar(
select(DiagnosticType.id).where(
DiagnosticType.name == diag_name
)
)
if diag_type_id is None:
raise PipelineError(f"Unknown diagnostic type {diag_name!r}")
existing_id = db_session.scalar(
select(PhotometryDiagnostics.id).where(
PhotometryDiagnostics.image_id == finished_id["image_id"],
PhotometryDiagnostics.channel == finished_id["channel"],
PhotometryDiagnostics.photometry_id == phot_id,
PhotometryDiagnostics.diagnostic_id == diag_type_id,
)
)
if existing_id is not None:
db_session.execute(
update(PhotometryDiagnostics)
.where(PhotometryDiagnostics.id == existing_id)
.values(value=float(diag_value))
)
else:
db_session.add(
PhotometryDiagnostics(
image_id=finished_id["image_id"],
channel=finished_id["channel"],
photometry_id=phot_id,
diagnostic_id=diag_type_id,
value=float(diag_value),
)
)
[docs]
def _end_processing(
self,
input_fname,
status=1,
final=True,
diagnostics=None,
photometry_diagnostics=None,
):
"""
Record that the current step has finished processing the given file.
Args:
input_fname: The filename of the input (DR or FITS) that was
processed.
status: The status code to record.
final: Whether this is the final status for this processing.
diagnostics: An optional list of ``(name, value)`` tuples to
record in the ``image_diagnostics`` table for each
image/channel processed from *input_fname*, or a dict
mapping channel names to such lists for per-channel
diagnostics.
photometry_diagnostics: An optional list of
``(name, value, photometry_id)`` tuples to record in the
``photometry_diagnostics`` table for each image/channel
processed from *input_fname*.
Returns:
None
"""
assert (
self.current_step is not None
), f"Marking {input_fname} as finished outside of any processing step!"
assert self._current_processing is not None, (
f"Marking {input_fname} as finished before the "
f"{self.current_step.name} progress record was created!"
)
assert status != -1, (
f"Status -1 is reserved for {input_fname} being skipped because "
"a prerequisite step failed, so a step may not report it!"
)
if status < 0:
self._some_failed = True
self._logger.debug(
"Finished processing %s", repr(self._processed_ids[input_fname])
)
with start_db_session() as db_session:
for finished_id in self._processed_ids[input_fname]:
if diagnostics:
if isinstance(diagnostics, dict):
channel_diags = diagnostics.get(finished_id["channel"])
else:
channel_diags = diagnostics
if channel_diags:
self._save_diagnostics(
finished_id, channel_diags, db_session
)
if photometry_diagnostics:
self._save_photometry_diagnostics(
finished_id, photometry_diagnostics, db_session
)
if self.current_step.name == "find_stars" and status >= 0:
dr_fname = self._evaluated_expressions[
finished_id["image_id"]
][finished_id["channel"]]["dr"]
image = db_session.get(Image, finished_id["image_id"])
if (
image is not None
and image.observing_session is not None
):
with DataReductionFile(dr_fname, mode="r+") as dr_file:
dr_file.add_provenance(image.observing_session)
db_session.execute(
update(ProcessedImages)
.where(ProcessedImages.image_id == finished_id["image_id"])
.where(ProcessedImages.channel == finished_id["channel"])
.where(
ProcessedImages.progress_id
== self._current_processing.id
)
.values(status=status, final=final)
)
# No good way to simplify
# pylint: disable=too-many-locals
[docs]
def _get_config_batches(self, pending_images, step_input_type, db_session):
"""Return the batches of images to process with identical config."""
result = {}
check_image_type_id = pending_images[0][0].image_type_id
for (
by_condition,
master_expression_values,
) in self.group_pending_by_conditions(
pending_images,
db_session,
match_observing_session=self.current_step.name == "calibrate",
):
for config_key, (config, batch) in self._get_batch_config(
by_condition,
master_expression_values,
self.current_step,
db_session,
).items():
if config_key is None:
self._logger.warning(
"Excluding the following images from %s:\n\t%s",
config,
"\n\t".join(
[
self.get_step_input(
image, channel, step_input_type
)
for image, channel, _ in batch
]
),
)
continue
for image, channel, status in batch:
assert image.image_type_id == check_image_type_id, (
f"{image.raw_fname} is of image type "
f"{image.image_type_id} in a batch collected for "
f"image type {check_image_type_id}!"
)
if (config_key, status) not in result:
result[config_key, status] = (config, [])
result[config_key, status][1].append(
self.get_step_input(image, channel, step_input_type)
)
return result
# pylint: enable=too-many-locals
[docs]
def _get_photrefs_for_condition(
self, photref_type, expressions, db_session
):
"""Return {expr_values: [MasterFile]} for all enabled single_photrefs.
Evaluates the condition expressions against each photref's filename
(same logic as _get_master) to group them by their expression values.
"""
photrefs_by_expr = {}
for master_file in db_session.scalars(
select(MasterFile).filter_by(type_id=photref_type.id, enabled=True)
).all():
pf_eval = Evaluator(master_file.filename)
expr_values = tuple(pf_eval(expr) for _, expr in expressions)
photrefs_by_expr.setdefault(expr_values, []).append(master_file)
return photrefs_by_expr
[docs]
def _get_photref_diagnostics(self, photrefs_by_expr, db_session):
"""Return {pf.id: {name: value}} of astrometry diagnostics per photref.
Each photref DR file's FITS header contains RAWFNAME and CLRCHNL which
identify the source Image row. ra_center, dec_center, and diagonal_fov
are then read from ImageDiagnostics for that image/channel.
Photrefs whose source image or diagnostics cannot be found are logged
and omitted from the result.
"""
result = {}
for master_files in photrefs_by_expr.values():
for pf in master_files:
if pf.id in result:
continue
try:
with DataReductionFile(pf.filename, "r") as dr_file:
header = dr_file.get_frame_header()
image_id = db_session.scalar(
select(Image.id).where( # pylint: disable=no-member
Image.raw_fname.like( # pylint: disable=no-member
f"%/{header['RAWFNAME']}.%"
)
)
)
if image_id is None:
self._logger.warning(
"Cannot find source image for photref %s"
" (RAWFNAME=%s).",
pf.filename,
header["RAWFNAME"],
)
continue
diags = dict(
db_session.execute(
select(DiagnosticType.name, ImageDiagnostics.value)
.join(
DiagnosticType,
ImageDiagnostics.diagnostic_id
== DiagnosticType.id,
)
.where(
ImageDiagnostics.image_id == image_id,
ImageDiagnostics.channel == header["CLRCHNL"],
DiagnosticType.name.in_(
["ra_center", "dec_center", "diagonal_fov"]
),
)
).all()
)
if all(
k in diags
for k in ["ra_center", "dec_center", "diagonal_fov"]
):
result[pf.id] = diags
else:
self._logger.warning(
"Incomplete astrometry diagnostics for photref %s.",
pf.filename,
)
except Exception: # pylint: disable=broad-except
self._logger.warning(
"Could not retrieve diagnostics for photref %s.",
pf.filename,
exc_info=True,
)
return result
[docs]
def _get_photref_binding_counts(self, photref_ids, db_session):
"""
Return the number of images bound to each single photometric reference
Return:
dict:
MasterFile.id: count of existing ImageMasterSelection rows per
photometric reference.
"""
return {
pf_id: (
db_session.scalar(
select(sql.func.count()) # pylint: disable=not-callable
.select_from(ImageMasterSelection)
.where(ImageMasterSelection.master_file_id == pf_id)
)
or 0
)
for pf_id in photref_ids
}
[docs]
def _select_photref_for_image(
self,
image,
channel,
photref_type,
expressions,
photrefs_by_expr,
photref_diagnostics,
binding_counts,
max_sep,
db_session,
):
"""Pick and record the best photref binding for one image/channel.
Returns True if bound (either pre-existing or newly written), False if
no suitable photref is found.
The distance threshold is max_sep * diagonal_fov of the photref.
"""
if (
db_session.scalar(
select(ImageMasterSelection.master_file_id).where(
ImageMasterSelection.image_id == image.id,
ImageMasterSelection.channel == channel,
ImageMasterSelection.master_type_id == photref_type.id,
)
)
is not None
):
return True
image_diags = dict(
db_session.execute(
select(DiagnosticType.name, ImageDiagnostics.value)
.join(
DiagnosticType,
ImageDiagnostics.diagnostic_id == DiagnosticType.id,
)
.where(
ImageDiagnostics.image_id == image.id,
ImageDiagnostics.channel == channel,
DiagnosticType.name.in_(["ra_center", "dec_center"]),
)
).all()
)
if not all(k in image_diags for k in ["ra_center", "dec_center"]):
self._logger.warning(
"Astrometry diagnostics missing for image %d channel %s;"
" cannot bind to photref.",
image.id,
channel,
)
return False
image_coord = SkyCoord(
ra=image_diags["ra_center"] * astropy_units.deg,
dec=image_diags["dec_center"] * astropy_units.deg,
frame="icrs",
)
image_expr_values = tuple(
self._evaluated_expressions[image.id][channel]["values"][expr_id]
for expr_id, _ in expressions
)
candidates = photrefs_by_expr.get(image_expr_values, [])
best_pf = None
best_count = -1
for pf in candidates:
if pf.id not in photref_diagnostics:
continue
pf_diags = photref_diagnostics[pf.id]
pf_coord = SkyCoord(
ra=pf_diags["ra_center"] * astropy_units.deg,
dec=pf_diags["dec_center"] * astropy_units.deg,
frame="icrs",
)
sep = image_coord.separation(pf_coord).to_value(astropy_units.deg)
if (
sep <= max_sep * pf_diags["diagonal_fov"]
and binding_counts[pf.id] > best_count
):
best_count = binding_counts[pf.id]
best_pf = pf
if best_pf is None:
self._logger.info(
"No suitable photref for image %d channel %s; leaving pending.",
image.id,
channel,
)
return False
db_session.merge(
ImageMasterSelection(
image_id=image.id,
channel=channel,
master_type_id=photref_type.id,
master_file_id=best_pf.id,
)
)
binding_counts[best_pf.id] += 1
return True
[docs]
def _bind_photref_for_pending(self, pending_images, step, db_session):
"""Bind unbound fit_magnitudes pending images to photrefs.
For each pending image/channel without an ImageMasterSelection entry,
assigns it to the best registered single_photref (condition expressions
match + image center within max_photref_separation * photref
diagonal_fov + most existing bindings). Returns only the images that
have a valid binding; the rest are silently left pending until a
suitable photref is registered via the BUI.
"""
if not pending_images:
return pending_images
first_img, first_ch, _ = pending_images[0]
config = self.get_config(
self._evaluated_expressions[first_img.id][first_ch]["matched"],
db_session,
db_step=step,
)[0]
max_sep = config.get("max_photref_separation", 0.2)
if max_sep is None or max_sep == inf:
return pending_images
photref_type = db_session.scalar(
select(MasterType).filter_by(name="single_photref")
)
if photref_type is None:
return []
expressions = db_session.execute(
select(ConditionExpression.id, ConditionExpression.expression)
.join_from(
Condition,
ConditionExpression,
Condition.expression_id # pylint: disable=no-member
== ConditionExpression.id,
)
.where(
Condition.id # pylint: disable=no-member
== photref_type.condition_id
)
.order_by(ConditionExpression.id)
).all()
photrefs_by_expr = self._get_photrefs_for_condition(
photref_type, expressions, db_session
)
if not photrefs_by_expr:
return []
photref_diagnostics = self._get_photref_diagnostics(
photrefs_by_expr, db_session
)
binding_counts = self._get_photref_binding_counts(
list(photref_diagnostics), db_session
)
result = []
for image, channel, status in pending_images:
if self._select_photref_for_image(
image,
channel,
photref_type,
expressions,
photrefs_by_expr,
photref_diagnostics,
binding_counts,
max_sep,
db_session,
):
result.append((image, channel, status))
db_session.flush()
return result
[docs]
def _prepare_processing(self, step, image_type, limit_to_steps):
"""Prepare for processing images of given type by a calibration step."""
with start_db_session() as db_session:
setup_process(
task="main",
parent_pid="",
processing_step=step.name,
image_type=image_type.name,
**self._processing_config,
)
step = db_session.merge(step)
image_type = db_session.merge(image_type)
self.set_pending(db_session, [(step, image_type)])
if limit_to_steps is not None and step.name not in limit_to_steps:
self._logger.debug(
"Skipping disabled %s for %s frames",
step.name,
image_type.name,
)
return step.name, image_type.name, None
if not self._check_ready(step, image_type, db_session):
return step.name, image_type.name, None
pending_images, step_input_type = self._start_step(
step, image_type, db_session
)
if not pending_images:
return step.name, image_type.name, None
if step.name == "fit_magnitudes":
pending_images = self._bind_photref_for_pending(
pending_images, step, db_session
)
if not pending_images:
return step.name, image_type.name, None
return (
step.name,
image_type.name,
self._get_config_batches(
pending_images, step_input_type, db_session
),
)
[docs]
def _finalize_processing(self):
"""Update database and instance after processing."""
with start_db_session() as db_session:
self._current_processing = db_session.merge(
self._current_processing
)
self._current_processing.finished = (
# False positive
# pylint: disable=not-callable
sql.func.now()
# pylint: enable=not-callable
)
pending = self.pending[
(
self._current_processing.step_id,
self._current_processing.image_type_id,
)
]
self._logger.info(
"Removing from pending all successful images for "
"progress: %s",
self._current_processing,
)
for finished_image_id, finished_channel in db_session.execute(
select(ProcessedImages.image_id, ProcessedImages.channel)
.where(
ProcessedImages.progress_id == self._current_processing.id
)
.where(
# pylint: disable=singleton-comparison
ProcessedImages.final
== True
# pylint: enable=singleton-comparison
)
.where(
or_(ProcessedImages.status > 0, ProcessedImages.status < -1)
)
).all():
found = False
for i, (image, channel, _) in enumerate(pending):
if (
image.id == finished_image_id
and channel == finished_channel
):
assert not found, (
f"Image {finished_image_id} channel "
f"{finished_channel} is listed more than once "
"among the images still to be processed!"
)
del pending[i]
found = True
break
if not found:
self._logger.error(
"Completed image ID %d, channel %s not found in "
"pending for step ID %d, image type ID %d:\n\t%s",
finished_image_id,
finished_channel,
self._current_processing.step_id,
self._current_processing.image_type_id,
"\n\t".join(f"{e[0]!r}: {e[1]!r}" for e in pending),
)
raise PipelineError("Finished non-pending image!")
self.pending[
(
self._current_processing.step_id,
self._current_processing.image_type_id,
)
] = pending
# if self._some_failed:
# dropped = self._clean_pending_per_dependencies(
# db_session,
# self._current_processing.step_id,
# self._current_processing.image_type_id
# )
# for step_imtype, dropped_images in dropped.items():
# if step_imtype in self._failed_dependencies:
# self._failed_dependencies[
# step_imtype
# ].extend(
# dropped_images
# )
# else:
# self._failed_dependencies[step_imtype] = (
# dropped_images
# )
[docs]
def __init__(self, *args, **kwargs):
"""Initialize self._failed_dependencies in addition to normali init."""
self._failed_dependencies = {}
super().__init__(*args, **kwargs)
[docs]
def set_pending(self, db_session, steps_imtypes=None, invert=False):
"""
Set the unprocessed images and channels split by step and image type.
Set the self.pending attribute to a dictionary with format ``{(step.id,
image_type.id): (Image, str)}``, containing the images and channels of
the specified type for which the specified step has not applied with the
current configuration.
Args:
db_session(Session): The database session to use.
steps_imtypes(Step, ImageType): The step image type combinations
to determine pending images for. If unspecified, the full
processing sequence defined in the database is used.
invert(bool): If True, returns successfully completed (not
failed) instead of pending.
Returns:
None
"""
status_select = (
select(
ProcessedImages.image_id,
ProcessedImages.channel,
sql.func.max(ProcessedImages.status).label("status"),
)
.join(ImageProcessingProgress)
.where(ProcessedImages.status > 0)
.where(ProcessedImages.final == 0)
.group_by(ProcessedImages.image_id, ProcessedImages.channel)
)
for step, image_type in steps_imtypes or get_processing_sequence(
db_session, True
):
failed_prereq_subquery = (
select(ProcessedImages.image_id, ProcessedImages.channel)
.select_from(StepDependencies)
.join(
ImageProcessingProgress,
and_(
StepDependencies.blocking_step_id
== ImageProcessingProgress.step_id,
StepDependencies.blocking_image_type_id
== ImageProcessingProgress.image_type_id,
),
)
.join(ProcessedImages)
.where(StepDependencies.blocked_step_id == step.id)
.where(StepDependencies.blocked_image_type_id == image_type.id)
.where(ProcessedImages.status < 0)
.group_by(ProcessedImages.image_id, ProcessedImages.channel)
.subquery()
)
processed_subquery = (
select(ProcessedImages.image_id, ProcessedImages.channel)
.join(ImageProcessingProgress)
.where(ImageProcessingProgress.step_id == step.id)
.where(ImageProcessingProgress.image_type_id == image_type.id)
.where(
ImageProcessingProgress.configuration_version
== self.step_version[step.name]
)
.where(ProcessedImages.final)
)
status_subquery = (
status_select.where(ImageProcessingProgress.step_id == step.id)
.where(ImageProcessingProgress.image_type_id == image_type.id)
.where(
ImageProcessingProgress.configuration_version
== self.step_version[step.name]
)
.subquery()
)
if invert:
processed_subquery = processed_subquery.where(
ProcessedImages.status > 0
)
processed_subquery = processed_subquery.subquery()
query = (
select(Image, CameraChannel.name, status_subquery.c.status)
.join(
ObservingSession,
)
.join(Camera)
.join(CameraType)
.join(CameraChannel)
.outerjoin(
processed_subquery,
# False positive
# pylint: disable=no-member
and_(
Image.id == processed_subquery.c.image_id,
CameraChannel.name == processed_subquery.c.channel,
),
# pylint: enable=no-member
)
.outerjoin(
failed_prereq_subquery,
and_(
Image.id # pylint: disable=no-member
== failed_prereq_subquery.c.image_id,
CameraChannel.name == failed_prereq_subquery.c.channel,
),
)
.outerjoin(
status_subquery,
and_(
Image.id # pylint: disable=no-member
== status_subquery.c.image_id,
CameraChannel.name == status_subquery.c.channel,
),
)
.where(
Image.image_type_id # pylint: disable=no-member
== image_type.id
)
)
# This is how NULL comparison is done in SQLAlchemy
# pylint: disable=singleton-comparison
if invert:
query = query.where(processed_subquery.c.image_id != None)
else:
query = query.where(processed_subquery.c.image_id == None)
pending_query = self._require_blocking_complete(
query, step, image_type, db_session
)
self.pending[(step.id, image_type.id)] = db_session.execute(
pending_query.where(failed_prereq_subquery.c.image_id == None)
).all()
self._failed_dependencies[(step.id, image_type.id)] = (
db_session.execute(
query.where(failed_prereq_subquery.c.image_id != None)
).all()
)
# pylint: enable=singleton-comparison
self._logger.debug(
"%s is pending for %d and failed dependencies for %d %s images",
step.name,
len(self.pending[(step.id, image_type.id)]),
len(self._failed_dependencies[(step.id, image_type.id)]),
image_type.name,
)
self._logger.debug("Pending: %s", repr(self.pending))
[docs]
def group_pending_by_conditions( # pylint: disable=too-many-arguments
self,
pending_images,
db_session,
*,
match_observing_session=False,
step_id=None,
masters_only=False,
):
"""
Group pendig_images by condition expression values.
Args:
pending_images([Image, str]): A list of the images (instance of
Image DB class) and channels to group.
db_session: Database session to use for querries.
match_observing_session: Whether each group of images needs to
be from the same observing session.
step_id(int): The ID of the step for which to group the pending
images. If not specified, defaults to the current step.
masters_only: If True, grouping is done only by the values
expressions required to determine the input or output masters
for the current step.
Returns:
[([Image, str], tuple)]:
Each entry is contains a list of the image/channel combinations
matching a unique set of conditions and the second entry is the
master expression values for all images in the list.
"""
image_type_id = pending_images[0][0].image_type_id
result = []
master_expression_ids = get_master_expression_ids(
step_id or self.current_step.id, image_type_id, db_session
)
while pending_images:
self._logger.debug(
"Finding images matching the same expressions as image id %d, "
"channel %s",
pending_images[-1][0].id,
pending_images[-1][1],
)
batch = []
match_expressions = ExpressionMatcher(
self._evaluated_expressions,
pending_images[-1][0].id,
pending_images[-1][1],
master_expression_ids,
masters_only=masters_only,
)
observing_session_id = pending_images[-1][0].observing_session_id
for i in range(len(pending_images) - 1, -1, -1):
if (
not match_observing_session
or pending_images[i][0].observing_session_id
== observing_session_id
) and match_expressions(
pending_images[i][0].id, pending_images[i][1]
):
batch.append(pending_images.pop(i))
else:
self._logger.debug("Not a match")
self._logger.debug(
"Image batch:\n\t%s",
"\n\t".join(
f"{image.raw_fname}: {channel} status {status}"
for image, channel, status in batch
),
)
result.append((batch, match_expressions.ref_master_values))
return result
#: Image processing records progress here; drives the shared
#: ``find_processing_outputs`` in the base class.
_progress_model = ImageProcessingProgress
[docs]
def _progress_image_type(self, processing_progress, db_session):
"""The image type is carried directly on an image progress row."""
return processing_progress.image_type.name
[docs]
def __call__(self, limit_to_steps=None, step_imtype_filter=None):
"""Perform all the processing for the given steps (all if None)."""
with start_db_session() as db_session:
processing_sequence = get_processing_sequence(db_session, True)
DataReductionFile.get_file_structure()
if step_imtype_filter:
self._logger.info(
"Applying step-image-type filter: %s",
step_imtype_filter,
)
for step, image_type in processing_sequence:
# If (step, image_type) combo is filtered out, skip it.
if (
step_imtype_filter
and step.name in step_imtype_filter
and image_type.name not in step_imtype_filter[step.name]
):
self._logger.info(
"User skipped %s for %s - skipping step",
step.name,
image_type.name,
)
continue
(step_name, image_type_name, processing_batches) = (
self._prepare_processing(step, image_type, limit_to_steps)
)
self._logger.debug(
"At start of %s step for %s images, "
"project home %s pending:\n\t%s",
step_name,
image_type_name,
get_project_home(),
"\n\t".join(
f"{key!r}: {len(val)}" for key, val in self.pending.items()
),
)
# If filtered or not ready, stop processing here
if processing_batches is None:
continue
self._finalize_processing()
for (_, start_status), (
config,
batch,
) in processing_batches.items():
with start_db_session() as db_session:
self._create_current_processing(
step, ("image_type", image_type.id), db_session
)
self._logger.debug(
"Starting %s for a batch of %d %s images from status %s "
"with config:\n%s",
step_name,
len(batch),
image_type_name,
start_status,
repr(config),
)
self._process_batch(
batch,
start_status=start_status,
config=config,
step_name=step_name,
image_type_name=image_type_name,
)
self._logger.debug(
"Processed %s batch of %d images.", step_name, len(batch)
)
self._finalize_processing()
self._logger.debug(
"After processing batch, pending:\n\t%s",
"\n\t".join(
f"{key!r}: {len(val)}"
for key, val in self.pending.items()
),
)
self._some_failed = False
[docs]
def add_raw_images(self, image_collection):
"""Add the given RAW images to the database for processing."""
with start_db_session() as db_session:
default_expression_id = db_session.scalar(
select(ConditionExpression.id).where(
ConditionExpression.notes == "Default expression"
)
)
configuration = self.get_config(
{default_expression_id},
db_session,
step_name="add_images_to_db",
)[0]
processing_steps.add_images_to_db.add_images_to_db(
image_collection, configuration
)
# pylint: enable=too-many-instance-attributes