diff --git a/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/collectors.py b/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/collectors.py
index fa14fc19d2cecade6c9cc20d2d928b26a58b104b..a9545581ca316e0e9ddbbe64a877a34af5db3dce 100644
--- a/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/collectors.py
+++ b/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/collectors.py
@@ -144,7 +144,7 @@ class SECICollector(CollectorIF):
:return:
"""
self.logger.info("Collecting SECI products for staging...")
- cache_path = self.parameters["seciCachePath"]
+ cache_path = self.parameters["sourcePath"]
workflow_dir = self.parameters["workflowDir"]
staging_dir = self.parameters["staging_area"]
tar_name = self.create_artifacts_name()
@@ -159,3 +159,31 @@ class SECICollector(CollectorIF):
exit(1)
aux_file_to_staging(self.staging_source_dir)
+
+
+class ObservationCollector(CollectorIF):
+ """Collect Observation Products for Ingestion"""
+
+ def __init__(self, parameters):
+ self.logger = logging.getLogger("ingest_envoy")
+ self.parameters = parameters
+ self.staging_source_dir = self.parameters["staging_area"] + "/" + self.parameters["workflowDir"]
+
+ def collect_products(self):
+ """
+ Run observation product collection script
+ """
+ self.logger.info("Collecting observation products for staging...")
+ source_path = self.parameters["sourcePath"]
+ workflow_dir = self.parameters["workflowDir"]
+ staging_dir = self.parameters["staging_area"]
+
+ # run script
+ collector = subprocess.run(
+ ["./observation-product-collector.sh", source_path, workflow_dir, staging_dir],
+ stdout=sys.stdout,
+ stderr=sys.stderr,
+ )
+ if collector.returncode != 0:
+ self.logger.error("ERROR: Observation product collection failed!")
+ exit(1)
diff --git a/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/ingest.py b/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/ingest.py
index ef5d8f307792b3983e1cf59e2804e384973a83c7..2eb12035e82f443d81cb7cad42b1e48797400343 100644
--- a/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/ingest.py
+++ b/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/ingest.py
@@ -65,12 +65,9 @@ def _get_settings(
parameters["calSpl"] = cal_spl
if source_dir is not None:
parameters["workflowDir"] = pathlib.Path(source_dir).name
- parameters["seciCachePath"] = pathlib.Path(source_dir).parent
-
- # Use the default staging area if it wasn't set by the solicitor
- if not parameters["staging_area"]:
- parameters["staging_area"] = ingestion_settings.stagingDirectory
+ parameters["sourcePath"] = pathlib.Path(source_dir).parent
+ parameters["staging_area"] = ingestion_settings.stagingDirectory
parameters["storage_area"] = ingestion_settings.storageDirectory
parameters["useIngest"] = strtobool(ingestion_settings.useIngest)
parameters["workflowUrl"] = workflow_url
diff --git a/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/launchers.py b/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/launchers.py
index 1f1c2482ddb34790ecdd0928dc44b9fb915ac58d..b871bf64eb59c22e8596afb624528fe452a7093e 100644
--- a/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/launchers.py
+++ b/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/launchers.py
@@ -24,6 +24,7 @@ from typing import Union
from ingest_envoy.collectors import (
ImageCollector,
SECICollector,
+ ObservationCollector,
collect_image_metadata,
)
from ingest_envoy.ingestion_manifest import IngestionManifestBuilder
@@ -216,6 +217,7 @@ class IngestObservationLauncher(IngestLauncherIF):
self.sci_product_type = "execution_block"
self.parameters = parameters
self.staging_source_dir = self.parameters["staging_area"] + "/" + self.parameters["workflowDir"]
+ self.collector = ObservationCollector(self.parameters)
def launch_ingestion(self) -> int:
"""
@@ -236,12 +238,19 @@ class IngestObservationLauncher(IngestLauncherIF):
:return:
"""
self.logger.info("Preparing for ingest...")
- # 1. run collection script to create calibration tarfile
- # self.run_collection_script()
+ # 1. run collection script to move observation data to the staging area
+ self.run_collector()
# 2. create ingestion manifest
self.create_manifest()
+ def run_collector(self):
+ """
+ Run ObservationCollector
+ """
+ # 1. collect products to staging area
+ self.collector.collect_products()
+
def create_manifest(self, additional_file=None):
"""
Create the observation ingestion manifest
diff --git a/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/solicitor.py b/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/solicitor.py
index f1877a7d292f57cb3d7ecd26dd2ef0bb780b756c..e3d9ddb4c84f83b8298426caa74eaf0f2cc35220 100644
--- a/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/solicitor.py
+++ b/apps/cli/executables/pexable/ingest_envoy/ingest_envoy/solicitor.py
@@ -25,8 +25,7 @@ import pathlib
from typing import List, Union
import requests
-from ingest_envoy.utilities import IngestType, Telescope, VLASSIngestType
-from pycapo import CapoConfig
+from ingest_envoy.utilities import IngestType, VLASSIngestType
INVALID_INITIAL_VERSION = "Initial version not valid for ingest"
@@ -218,14 +217,6 @@ class Solicitor:
"project": self.metadata["projectMetadata"]["projectCode"], # needed for post ingestion messaging
}
- # VLBA and GMVA share a non-default staging directory
- if obs["telescope"].upper() in [Telescope.VLBA.value, Telescope.GMVA.value]:
- try:
- obs["staging_area"] = \
- CapoConfig().settings("edu.nrao.workspaces.IngestionSettings").vlbiStagingDirectory
- except KeyError:
- self.logger.info("Couldn't retrieve VLBI staging area from capo, using default staging directory")
-
return {**obs}
def solicit_seci_params(self) -> dict:
diff --git a/apps/cli/executables/wf_framework/ingest_requirements/observation-product-collector.sh b/apps/cli/executables/wf_framework/ingest_requirements/observation-product-collector.sh
new file mode 100644
index 0000000000000000000000000000000000000000..59407a43ac70926723550b30f3bf4af5e0a2c0b4
--- /dev/null
+++ b/apps/cli/executables/wf_framework/ingest_requirements/observation-product-collector.sh
@@ -0,0 +1,47 @@
+#!/usr/bin/env bash
+#
+# Copyright (C) 2021 Associated Universities, Inc. Washington DC, USA.
+#
+# This file is part of NRAO Workspaces.
+#
+# Workspaces is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Workspaces is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Workspaces. If not, see <https://www.gnu.org/licenses/>.
+
+#
+# A file to copy all files from a provided source directory to
+# the Workspaces staging directory for ingestion.
+#
+# Arguments:
+# 1: The path to the source directory
+# 2: The source directory name
+# 3: The staging directory root path
+#
+set -o errexit -o nounset -o xtrace
+
+SRC_PATH=$1;shift
+SRC_NAME=$1;shift
+STAGE_DIR=$1;shift
+
+SRC_DIR=${SRC_PATH}/${SRC_NAME}
+
+# Create the staging directory carefully:
+mkdir -p ${STAGE_DIR}/${SRC_NAME}
+
+# Link all the files over to the staging directory (Create hard link, and be insistent)
+#
+# We just want the file name, so we'll work in the source directory
+pushd ${SRC_DIR}
+# link the files
+for srcFile in $(ls);do cp ${SRC_DIR}/${srcFile} ${STAGE_DIR}/${SRC_NAME};done
+# move back to the working directory
+popd