Skip to content
Snippets Groups Projects

Implemented manifest generation for EB full curation in ingest_envoy

Merged Sam Kagan requested to merge teach-ingest_envoy-full-curation into 2.8.2.1-DEVELOMENT
Files
8
@@ -21,13 +21,15 @@ import logging
import sys
import tarfile
from pathlib import Path
from pycapo import CapoConfig
# pylint: disable=C0301, E0401, R0903, R1721
from typing import Tuple, List
from typing import List, Optional, Tuple
import arrow
from ingest_envoy.manifest_components import (
CURATOR_MANIFEST_FILENAME,
INGESTION_ARTIFACTS_NAME,
INIT_WEBLOG_FILENAME,
JSON,
@@ -43,20 +45,19 @@ from ingest_envoy.manifest_components import (
OutputGroup,
OutputScienceProduct,
ReingestGroup,
CURATOR_MANIFEST_FILENAME,
)
from ingest_envoy.schema import AbstractTextFile
from ingest_envoy.std_img_manifest_utils import ImageIngestionProductsFinder
from ingest_envoy.std_obs_manifest_utils import ObservationIngestionProductsFinder
from ingest_envoy.utilities import (
AncillaryProductType,
CuratorType,
IngestionManifestException,
NoScienceProductException,
ScienceProductType,
Telescope,
find_output_tars,
find_weblogs,
CuratorType,
)
logger = logging.getLogger(__name__)
@@ -189,8 +190,10 @@ class IngestionManifest(ManifestComponentIF):
:return:
"""
me_dict = self.to_dict()
if self.reingest is not None:
if self.reingest is not None and self.reingest.targets is not None:
output_path = Path.cwd() / CURATOR_MANIFEST_FILENAME
elif self.reingest is not None:
output_path = self.staging_source_dir / CURATOR_MANIFEST_FILENAME
else:
output_path = self.staging_source_dir / MANIFEST_FILENAME
@@ -216,7 +219,7 @@ class IngestionManifest(ManifestComponentIF):
if IngestionManifestKey.REINGEST.value in me_dict and me_dict[IngestionManifestKey.REINGEST.value] is not None:
to_return[IngestionManifestKey.REINGEST.value] = me_dict[IngestionManifestKey.REINGEST.value].to_dict()
# curator manifests have no output groups
# partial-curation manifests have no output groups
if (
IngestionManifestKey.OUTPUT_GROUP.value in me_dict
and me_dict[IngestionManifestKey.OUTPUT_GROUP.value] is not None
@@ -250,6 +253,8 @@ class IngestionManifestBuilder:
additional_metadata: AbstractTextFile = None,
filename: str = None,
curate: (CuratorType, str, List[str]) = None,
# Curation doesn't need a curation_source, but it does need a place to stick the manifest
manifest_destination_dir: Optional[Path] = None,
):
# get the telescope
self.telescope = Telescope(telescope)
@@ -278,17 +283,32 @@ class IngestionManifestBuilder:
raise NotImplementedError(f"Don't know yet how to build a {self.sp_type.value} manifest")
if self.curation_type is not None:
if self.curation_type not in [CuratorType.PARTIAL]:
if self.curation_type not in [CuratorType.PARTIAL, CuratorType.FULL]:
raise NotImplementedError(
f"Don't know how to build a {self.curation_type.value} curation {self.sp_type} manifest"
)
self.curation_source = Path(curate[1]) if curate and curate[1] else None
self.target_list = curate[2]
if self.curation_type == CuratorType.FULL and self.target_list not in [None, []]:
raise IngestionManifestException(
f"Expected targets=None for full curation, got targets={self.target_list}"
)
if locator is not None:
# we are not running observation ingestion, use a locator
self.locator = locator
# directory to put the manifest in
self.manifest_destination_dir = self.staging_source_dir
if not self.manifest_destination_dir:
self.manifest_destination_dir = self.curation_source
if not self.manifest_destination_dir:
self.manifest_destination_dir = manifest_destination_dir
if self.manifest_destination_dir is None:
raise IngestionManifestException(
f"IngestionManifestBuilder: Expected a directory to house the manifest, found instead staging_source_dir={self.staging_source_dir}, curation_source={self.curation_source}, manifest_dir={self.manifest_destination_dir}"
)
if staging_source_dir is not None:
# we are not running curation, look for files
self.files_found = [file for file in staging_source_dir.iterdir()]
@@ -306,18 +326,18 @@ class IngestionManifestBuilder:
:return: the ingestion manifest and the file containing its JSON
"""
if self.curation_type == CuratorType.PARTIAL:
return self._build_curation_manifest()
return self._build_partial_curation_manifest()
if self.sp_type == ScienceProductType.EVLA_CAL:
return self._build_evla_cal_manifest()
elif self.sp_type == ScienceProductType.EXEC_BLOCK:
return self._build_observation_manifest()
return self._build_observation_manifest(self.curation_type == CuratorType.FULL)
return self._build_image_manifest()
def _build_curation_manifest(self) -> (IngestionManifest, Path):
def _build_partial_curation_manifest(self) -> (IngestionManifest, Path):
"""
Build a manifest for curator
Build a manifest for partial curator
Partial curation is simple, only parameters and reingest groups are needed
:return: the manifest file object and path
@@ -351,8 +371,7 @@ class IngestionManifestBuilder:
# We can't create the ingestion artifacts tar quite yet,
# because it will contain the yet-to-be-written manifest itself
# (required for ingestion, evidently)
artifacts_filename = self._build_artifacts_filename()
artifacts_ap = AncillaryProduct(AncillaryProductType.INGESTION_ARTIFACTS, filename=artifacts_filename)
artifacts_ap = self._build_artifacts_product()
if artifacts_ap not in manifest.output_group.ancillary_products:
manifest.output_group.ancillary_products.append(artifacts_ap)
@@ -363,35 +382,34 @@ class IngestionManifestBuilder:
manifest.output_group.ancillary_products.append(weblog_ap)
manifest_file = manifest.write()
artifacts_file = self.staging_source_dir / artifacts_filename
self.write_ingestion_artifacts_tar(artifacts_file)
self.write_ingestion_artifacts_tar(self.manifest_destination_dir / artifacts_ap.filename)
return manifest, manifest_file
def _build_observation_manifest(self) -> (IngestionManifest, Path):
def _build_observation_manifest(self, is_full_curation: bool = False) -> tuple[IngestionManifest, Path]:
reingest_group = None
if is_full_curation:
reingest_group = self._build_reingest_group()
# create the manifest
manifest = IngestionManifest(
telescope=self.telescope,
locator=None,
sp_type=self.sp_type,
staging_source_dir=self.staging_source_dir,
reingest_group=None,
staging_source_dir=self.manifest_source_dir,
reingest_group=reingest_group,
input_group=InputGroup([]),
output_group=self._build_observation_output_group(),
filename=self.filename,
)
artifacts_filename = self._build_artifacts_filename()
artifacts_ap = AncillaryProduct(AncillaryProductType.INGESTION_ARTIFACTS, filename=artifacts_filename)
if artifacts_ap not in manifest.output_group.ancillary_products:
manifest.output_group.ancillary_products.append(artifacts_ap)
if not manifest.output_group.ancillary_products:
manifest.output_group.ancillary_products = []
artifacts_ap = self._build_artifacts_product()
if artifacts_ap not in manifest.output_group.ancillary_products:
manifest.output_group.ancillary_products.append(artifacts_ap)
manifest_file = manifest.write()
artifacts_file = self.staging_source_dir / artifacts_filename
self.write_ingestion_artifacts_tar(artifacts_file)
self.write_ingestion_artifacts_tar(self.manifest_destination_dir / artifacts_ap.filename)
return manifest, manifest_file
@@ -426,12 +444,11 @@ class IngestionManifestBuilder:
output_group=self._build_imaging_output_group(),
)
artifacts_file = self.staging_source_dir / self._build_artifacts_filename()
artifacts_ap = AncillaryProduct(type=AncillaryProductType.INGESTION_ARTIFACTS, filename=artifacts_file.name)
artifacts_ap = self._build_artifacts_product()
if artifacts_ap not in manifest.output_group.ancillary_products:
manifest.output_group.ancillary_products.append(artifacts_ap)
manifest_file = manifest.write()
self.write_ingestion_artifacts_tar(artifacts_file)
self.write_ingestion_artifacts_tar(self.manifest_destination_dir / artifacts_ap.filename)
return manifest, manifest_file
@@ -511,7 +528,7 @@ class IngestionManifestBuilder:
:return:
"""
products_finder = ObservationIngestionProductsFinder(self.staging_source_dir, self.sp_type)
products_finder = ObservationIngestionProductsFinder(self.manifest_source_dir, self.sp_type)
science_products = products_finder.output_science_products
ancillary_products = products_finder.ancillary_products
@@ -528,20 +545,27 @@ class IngestionManifestBuilder:
timestamp = format_timestamp(current_time)
return f"{INGESTION_ARTIFACTS_NAME}{timestamp}{TARFILE_EXT}"
@staticmethod
def _build_artifacts_product() -> AncillaryProduct:
return AncillaryProduct(
AncillaryProductType.INGESTION_ARTIFACTS, IngestionManifestBuilder._build_artifacts_filename()
)
def write_ingestion_artifacts_tar(self, artifacts_path: Path) -> tarfile.TarFile:
"""
Take the list of files and build a tar for inclusion into the archive.
Take the list of files and write a tar file for inclusion in the archive.
This happens in the staging area for ingestion.
The EVLA CAL tar will contain just the manifest.
:return: a .tar archive of the ingestion artifacts
:param artifacts_path: Path to create the resulting tar file at
:return: tar file of the ingestion artifacts
"""
addl_md_file = None
if self.additional_metadata:
# find the additional metadata
addl_md_filename = self.additional_metadata.filename
addl_md_file = self.staging_source_dir / addl_md_filename
addl_md_file = self.manifest_source_dir / addl_md_filename
with tarfile.open(artifacts_path, "w") as ingestion_artifacts_tar:
if addl_md_file:
@@ -549,11 +573,21 @@ class IngestionManifestBuilder:
# The manifest file itself is considered an ingestion artifact.
# (It's turtles all the way down.)
manifest_file = self.staging_source_dir / MANIFEST_FILENAME
manifest_file = self.manifest_destination_dir / MANIFEST_FILENAME
if not manifest_file.exists():
manifest_file = self.manifest_destination_dir / CURATOR_MANIFEST_FILENAME
if not manifest_file.exists():
raise FileNotFoundError(
f"No manifest (i.e. {MANIFEST_FILENAME} or {CURATOR_MANIFEST_FILENAME}) found in {self.manifest_destination_dir}"
)
ingestion_artifacts_tar.add(manifest_file)
return ingestion_artifacts_tar
@property
def manifest_source_dir(self) -> Path | None:
return self.staging_source_dir if self.staging_source_dir is not None else self.curation_source
def format_timestamp(datetime: arrow.Arrow) -> str:
"""
Loading