Merge branch 'tickets/DM-43599'

python/lsst/ap/association/diaPipe.py

@@ -569,7 +569,7 @@ def run(self,
 
  # Store DiaSources, updated DiaObjects, and DiaForcedSources in the
  # Apdb.
- self.log.info(f"Updating {len(diaForcedSources)} diaForcedSources from the APDB")
+ self.log.info(f"Updating {len(diaForcedSources)} diaForcedSources in the APDB")
  diaForcedSources = convertTableToSdmSchema(self.schema, diaForcedSources,
  tableName="DiaForcedSource",
  )
@@ -582,6 +582,7 @@ def run(self,
  diaObjectStore,
  diaSourceStore,
  diaForcedSourceStore)
+ self.log.info("APDB updated.")
 
  if self.config.doPackageAlerts:
  if len(loaderResult.diaForcedSources) > 1:

python/lsst/ap/association/loadDiaCatalogs.py

@@ -138,6 +138,8 @@ def loadDiaObjects(self, region, apdb, schema):
  DiaObjects loaded from the Apdb that are within the area defined
  by ``pixelRanges``.
  """
+ self.log.info("Loading DiaObjects")
+
  if region is None:
  # If no area is specified return an empty DataFrame with the
  # the column used for indexing later in AssociationTask.
@@ -182,6 +184,8 @@ def loadDiaSources(self, diaObjects, region, dateTime, apdb, schema):
  DiaSources loaded from the Apdb that are within the area defined
  by ``pixelRange`` and associated with ``diaObjects``.
  """
+ self.log.info("Loading DiaSources")
+
  if region is None:
  # If no area is specified return an empty DataFrame with the
  # the column used for indexing later in AssociationTask.
@@ -230,6 +234,8 @@ def loadDiaForcedSources(self, diaObjects, region, dateTime, apdb, schema):
  DiaObjects loaded from the Apdb that are within the area defined
  by ``pixelRanges``.
  """
+ self.log.info("Loading DiaForcedSources")
+
  if len(diaObjects) == 0:
  # If no diaObjects are available return an empty DataFrame with
  # the the column used for indexing later in AssociationTask.

python/lsst/ap/association/packageAlerts.py

@@ -48,6 +48,7 @@
 import lsst.pex.config as pexConfig
 from lsst.pex.exceptions import InvalidParameterError
 import lsst.pipe.base as pipeBase
+import lsst.utils.logging
 from lsst.utils.timer import timeMethod
 
 
@@ -247,7 +248,14 @@ def run(self,
  diffImPhotoCalib = diffIm.getPhotoCalib()
  calexpPhotoCalib = calexp.getPhotoCalib()
  templatePhotoCalib = template.getPhotoCalib()
+
+ n_sources = len(diaSourceCat)
+ self.log.info("Packaging alerts for %d DiaSources.", n_sources)
+ # Log every 10 seconds as proof of liveness.
+ loop_logger = lsst.utils.logging.PeriodicLogger(self.log, interval=10.0)
+
  for srcIndex, diaSource in diaSourceCat.iterrows():
+ loop_logger.log("%s/%s sources have been packaged.", len(alerts), n_sources)
  # Get all diaSources for the associated diaObject.
  # TODO: DM-31992 skip DiaSources associated with Solar System
  # Objects for now.
@@ -305,10 +313,13 @@ def run(self,
  templateCutout))
 
  if self.config.doProduceAlerts:
+ self.log.info("Producing alerts to %s.", self.kafkaTopic)
  self.produceAlerts(alerts, visit, detector)
 
  if self.config.doWriteAlerts:
- with open(os.path.join(self.config.alertWriteLocation, f"{visit}_{detector}.avro"), "wb") as f:
+ avro_path = os.path.join(self.config.alertWriteLocation, f"{visit}_{detector}.avro")
+ self.log.info("Writing alerts to %s.", avro_path)
+ with open(avro_path, "wb") as f:
  self.alertSchema.store_alerts(f, alerts)
 
  def _patchDiaSources(self, diaSources):