From 6a319feb962c3ccfdcdb5510ca21b043560449e9 Mon Sep 17 00:00:00 2001 From: Charlotte Hausman <chausman@nrao.edu> Date: Fri, 26 Aug 2022 15:29:19 -0400 Subject: [PATCH] fix ingestion message catching for autocalibration --- docker.properties | 1 - .../system/services/archive_service.py | 16 +++++++++++----- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/docker.properties b/docker.properties index bd38ba4f8..788bc621c 100644 --- a/docker.properties +++ b/docker.properties @@ -26,7 +26,6 @@ edu.nrao.workspaces.ProcessingSettings.useCasa = false edu.nrao.workspaces.ProcessingSettings.rootDirectory = /lustre/aoc/cluster/pipeline/docker/workspaces/spool edu.nrao.workspaces.ProcessingSettings.scriptLocation = /lustre/aoc/cluster/pipeline/docker/workspaces/sbin edu.nrao.workspaces.ProcessingSettings.ramInGb = 0.2G -edu.nrao.workspaces.ProcessingSettings.autoGenerateStandardCals = False edu.nrao.workspaces.ProcessingSettings.CasaVersion.vlassSeci = /home/casa/packages/pipeline/casa-6.1.3-3-pipeline-2021.1.1.32 edu.nrao.archive.workflow.config.CasaVersions.homeForReprocessing = /home/casa/packages/pipeline/current diff --git a/shared/workspaces/workspaces/system/services/archive_service.py b/shared/workspaces/workspaces/system/services/archive_service.py index fd28aad1a..c97bf971f 100644 --- a/shared/workspaces/workspaces/system/services/archive_service.py +++ b/shared/workspaces/workspaces/system/services/archive_service.py @@ -185,7 +185,7 @@ class ArchiveService(ArchiveServiceIF): self.router = Router("archive") self.router.register(self) - @on_message(message="Ingestion complete") + @on_message(message="ingestion complete") def on_ingestion_complete(self, **message: Dict): """ Accept an Ingestion Complete message from the archive @@ -195,19 +195,25 @@ class ArchiveService(ArchiveServiceIF): log_data = message["logData"] + sdm_id = log_data["fileset_id"] + sb_type = log_data["schedblock_type"] + telescope = log_data["telescope"] + + if sb_type != "OBSERVER" or telescope != "EVLA": + # not a candidate for std_calibration + return None + # log message logger.info(f"Archive Ingestion Complete Event: {log_data}") - sdm_id = log_data["fileset_id"] if sdm_id.startswith("VLASS") or sdm_id.startswith("TSKY"): # Do not submit VLASS or TSKY execution blocks for calibration logger.info(f"Caught Ingestion Event was for VLASS or TSKY - ignoring.") return None - self.create_and_submit_from_sdm_id(message) + self.create_and_submit_from_sdm_id(log_data) - def create_and_submit_from_sdm_id(self, info: Dict): - log_data = info["logData"] + def create_and_submit_from_sdm_id(self, log_data: Dict): sdm_id = log_data["fileset_id"] execblock_id = log_data.get("execblock_id", "") -- GitLab