Skip to content
Snippets Groups Projects

WS-601: Rewrite/refactor ingestion manifest builder to pass disabled tests, part one

Merged WS-601: Rewrite/refactor ingestion manifest builder to pass disabled tests, part one
WS-601-ingestion-manifest-resolve into main
Merged Janet Goldstein requested to merge WS-601-ingestion-manifest-resolve into main
Compare
and
2 files
+ 102
− 255
Compare changes
  • Side-by-side
  • Inline
Files
2
apps/cli/executables/pexable/ingest_envoy/ingest_envoy/ingestion_manifest.py
+ 42
− 199
@@ -99,6 +99,7 @@ class ManifestIF(ManifestComponentIF):
staging_source_dir=self.staging_source_dir,
)
self.additional_metadata = additional_metadata
return params
@abc.abstractmethod
@@ -138,26 +139,28 @@ class IngestionManifestBuilder:
def __init__(
self,
staging_source_dir: Path,
sp_type: ScienceProductType,
sp_type: str,
locator: str,
telescope: Telescope,
telescope: str,
additional_metadata: AbstractTextFile = None,
):
self.telescope = telescope
self.telescope = Telescope(telescope)
self.staging_source_dir = staging_source_dir
if sp_type not in [ScienceProductType.EVLA_CAL, ScienceProductType.IMAGE]:
raise NotImplementedError(f"Don't know yet how to build a {sp_type.value} manifest")
self.sp_type = ScienceProductType(sp_type)
if self.sp_type not in [ScienceProductType.EVLA_CAL, ScienceProductType.IMAGE]:
raise NotImplementedError(
f"Don't know yet how to build a {self.sp_type.value} manifest"
)
self.locator = locator
# (for EVLA CAL, this will be None)
self.additional_metadata = additional_metadata
# self.files_found = [file for file in staging_source_dir.iterdir()]
# if len(self.files_found) == 0:
# raise IngestionManifestException(f"No ingestion files found at {staging_source_dir}")
self.files_found = [file for file in staging_source_dir.iterdir()]
if len(self.files_found) == 0:
raise IngestionManifestException(f"No ingestion files found at {staging_source_dir}")
def build(self) -> Tuple[ManifestIF, Path]:
"""
@@ -175,6 +178,7 @@ class IngestionManifestBuilder:
return self._build_image_manifest()
def _build_evla_cal_manifest(self):
# create the manifest
manifest = IngestionManifest(
telescope=self.telescope,
@@ -182,39 +186,38 @@ class IngestionManifestBuilder:
sp_type=self.sp_type,
staging_source_dir=self.staging_source_dir,
input_group=self._build_input_group(),
output_group=self._build_output_group(),
output_group=self._build_evla_cal_output_group(),
)
artifacts_tar = self.write_ingestion_artifacts_tar()
manifest.output_group.ancillary_products.append(
AncillaryProduct(AncillaryProductType.INGESTION_ARTIFACTS, filename=str(artifacts_tar))
)
if self.additional_metadata:
manifest.parameters.additional_metadata = self.additional_metadata
manifest_file = manifest.write()
self.write_ingestion_artifacts_tar()
return manifest, manifest_file
def _build_evla_cal_manifest(self):
def _build_image_manifest(self):
"""
Image manifest has additional_metadata, and output group is way more complicated
:return:
"""
# create the manifest
manifest = IngestionManifest(
telescope=self.telescope,
locator=self.locator,
additional_metadata=self.additional_metadata,
sp_type=self.sp_type,
staging_source_dir=self.staging_source_dir,
input_group=self._build_input_group(),
output_group=self._build_output_group(),
output_group=self._build_imaging_output_group(),
)
manifest_file = manifest.write()
self.write_ingestion_artifacts_tar()
return manifest, manifest_file
def _build_image_manifest(self):
"""
Image manifest has additional_metadata, and output group is way more complicated
:return:
"""
# TODO:
def _build_input_group(self):
"""
Create the input group using the parameters.
@@ -229,72 +232,25 @@ class IngestionManifestBuilder:
return InputGroup([sp_in])
def _define_output_science_products(self) -> List[OutputScienceProduct]:
def _build_evla_cal_output_group(self):
"""
Find in the staging dir a science product and, if applicable, its ancillary products.
Create imaging manifest output group using the parameters
and the contents of the staging dir.
HEADS UP! ASSUMPTION: only one science product in the staging dir.
This works for EVLA CAL and image ingestion but may need an overhaul
for future ingestion types. We return a list because that's what there is
in our example image ingestion manifest. YMMV, void where prohibited,
professional driver on closed course, not FDIC insured.
:return: output science products found
:return:
"""
# find science product (we expect just one for this SP type)
if self.sp_type == ScienceProductType.EVLA_CAL:
tars_found = find_output_tars(self.files_found, self.staging_source_dir)
for file in tars_found:
sci_prod = OutputScienceProduct(type=self.sp_type, filename=str(file))
return [sci_prod]
elif self.sp_type == ScienceProductType.IMAGE:
products_finder = ImageIngestionProductsFinder(self.staging_source_dir)
science_products = products_finder.science_products
ancillary_products = products_finder.ancillary_products
# image_products = self._find_image_products()
# sps = []
# for ip in image_products:
# if ip.type == AncillaryProductType.QUICKLOOK_IMAGE:
# # this is the science product, a quicklook image
# sp_itself = ip
# elif ip.type == AncillaryProductType.QUICKLOOK_RMS_IMAGE:
# sps.append(ip)
# elif ip.type == AncillaryProductType.PIPELINE_WEBLOG_TYPE:
# sps.append(ip)
# # elif ip.type == AncillaryProductType.
#
# aips = []
# for file in tars_found:
# ap = self._build_ancillary_image_science_product(file)
# if ap:
# aips.append(ap)
#
# maybe_weblogs = [file for file in self.staging_source_dir.glob(WEBLOG_FILENAME)]
# if len(maybe_weblogs) > 0:
# weblog = Path(maybe_weblogs[0])
# weblog_ap = AncillaryProduct(
# type=AncillaryProductType.PIPELINE_WEBLOG_TYPE, filename=str(weblog)
# )
# aips.append(weblog_ap)
#
# sci_prod = OutputScienceProduct(
# type=sp_itself.type, filename=sp_itself.filename, ancillary_products=aips
# )
# return [sci_prod]
# else:
# raise ValueError(f"Don't know yet how to handle a {self.sp_type.vaue}")
# def _build_evla_cal_output_group(self) -> OutputGroup:
# # find science products
# science_products = self._define_output_science_products()
#
# return OutputGroup(science_products=science_products)
tars_found = find_output_tars(self.files_found, self.staging_source_dir)
for file in tars_found:
sci_prod = OutputScienceProduct(type=self.sp_type, filename=str(file))
return OutputGroup(science_products=[sci_prod])
def _build_imaging_output_group(self) -> OutputGroup:
"""
Create the output group using the parameters.
Create imaging manifest output group using the parameters
and the contents of the staging dir.
:return:
"""
@@ -305,54 +261,6 @@ class IngestionManifestBuilder:
return OutputGroup(science_products=science_products, ancillary_products=ancillary_products)
# def _build_ancillary_image_science_product(self, file: Path):
# """
# Image science products will have ancillary products of their very own,
# distinct from other ancillary products that might be in the staging dir.
#
# :param file: a possible ancillary image product
# :return: the corresponding AncillaryProduct, if applicable
# """
# filename = str(file)
# if "image" in file.name:
# if file.name.endswith(".png"):
# return AncillaryProduct(type=AncillaryProductType.THUMBNAIL_IMG, filename=filename)
# elif file.name.endswith(".fits"):
# if "rms" in file.name:
# return AncillaryProduct(
# type=AncillaryProductType.QUICKLOOK_RMS_IMAGE, filename=filename
# )
# else:
# return AncillaryProduct(
# type=AncillaryProductType.QUICKLOOK_IMAGE, filename=filename
# )
#
# def _build_ancillary_product(self, file: Path) -> AncillaryProduct:
# """
# If this file is required for ingestion manifest creation,
# create an ancillary product from it.
#
# :param file: file found in staging dir
# :return: ancillary product represented by this file, if any
# """
# if file.name == WEBLOG_FILENAME:
# return AncillaryProduct(
# type=AncillaryProductType.PIPELINE_WEBLOG_TYPE, filename=str(file)
# )
#
# if AncillaryProductType.PIPELINE_ARTIFACTS.value in file.name:
# return AncillaryProduct(
# type=AncillaryProductType.PIPELINE_ARTIFACTS, filename=str(file)
# )
#
# if AncillaryProductType.INGESTION_ARTIFACTS.value in file.name:
# return AncillaryProduct(
# type=AncillaryProductType.INGESTION_ARTIFACTS, filename=str(file)
# )
#
# # this is not an ancillary product
# return None
@staticmethod
def build_artifacts_filename() -> str:
"""
@@ -369,88 +277,23 @@ class IngestionManifestBuilder:
Take the list of files and build a tar for inclusion into the archive.
This happens in the staging area for ingestion.
The EVLA CAL tar will contain just the manifest.
For image ingestion, this must be done -after- the pipeline artifacts tar is built.
:return: a .tar archive of the ingestion artifacts
"""
ingestion_files = [file for file in self.staging_source_dir.iterdir() if file.is_file]
manifest_file = find_manifest(self.staging_source_dir)
ing_tar = self.staging_source_dir / self.build_artifacts_filename()
with tarfile.open(ing_tar, "w") as ingestion_artifacts_tar:
for file in ingestion_files:
for file in self.files_found:
ingestion_artifacts_tar.add(file)
# include the manifest
if manifest_file not in ingestion_files:
if manifest_file not in self.files_found:
ingestion_artifacts_tar.add(manifest_file)
return ing_tar
# def _find_image_products(self) -> List[AncillaryProduct]:
# """
# Get ancillary products that belong to science products
#
# :return:
# """
# ingestion_files = self._find_ingestion_files()
# aips = []
# for file in ingestion_files:
# aip = self._build_ancillary_image_science_product(file)
# if aip:
# aips.append(aip)
#
# return aips
# def _find_ancillary_products_for_img_ingest(self) -> List[AncillaryProduct]:
# """
# Round up any ancillary files found in image ingestion staging dir
#
# :return: ancillary product(s) found
# """
#
# ancillary_products = []
#
# # TODO: START HERE THU 2021-07-29: the image files are science product ancillaries; the tars and the weblog are
# # ancillary products. in test_creates_expected_manifest(), we expect 1 sci prod w/2 sp ancillaries,
# # and 3 ancillary prods.
#
# ingestion_files = self._find_ingestion_files()
# for file in ingestion_files:
# maybe_ap = self._build_ancillary_product(file)
# if maybe_ap and maybe_ap not in ancillary_products:
# ancillary_products.append(maybe_ap)
#
# # There should be a weblog in here; grab it
# maybe_weblogs = [file for file in self.files_found if file.name == WEBLOG_FILENAME]
# if len(maybe_weblogs) > 0:
# file = self.staging_source_dir / WEBLOG_FILENAME
# weblog_ap = AncillaryProduct(
# type=AncillaryProductType.PIPELINE_WEBLOG_TYPE, filename=str(file)
# )
# if weblog_ap not in ancillary_products:
# ancillary_products.append(weblog_ap)
# else:
# raise FileNotFoundError(f"No weblog found in {self.staging_source_dir}")
#
# return ancillary_products
def _find_ingestion_files(self) -> List[Path]:
"""
Round up any other necessary ingestion file(s)
:return: additional relevant files found in ingestion path, if any
"""
if self.sp_type == ScienceProductType.EVLA_CAL:
# there won't be any others
return []
if self.sp_type == ScienceProductType.IMAGE:
return [file for file in self.staging_source_dir.iterdir()]
# TODO when the time comes: we'll have extra information for other ingestion types;
# see archive-metaproject
raise NotImplementedError
class IngestionManifest(ManifestIF):
"""write ingestion manifest to file"""