Skip to content
Snippets Groups Projects

WS-543: Addressed issues found in calibration ingestion testing

Merged Janet Goldstein requested to merge WS-543-fix-manifest-issues-2021-07-22 into main
4 files
+ 11
67
Compare changes
  • Side-by-side
  • Inline
Files
4
@@ -69,15 +69,6 @@ class ManifestIF(ManifestComponentIF):
self.files_found = [file for file in self.staging_source_dir.iterdir()]
@abc.abstractmethod
def create(self):
"""
Build and write the manifest, which includes gathering various items in
ingestion_path to get info for the manifest.
:return:
"""
@abc.abstractmethod
def write(self):
"""
@@ -111,13 +102,13 @@ class IngestionManifestBuilder:
def __init__(
self,
staging_source_dir: Path,
sp_type: ScienceProductType,
sp_type: str,
locator: str,
telescope: Telescope,
):
self.telescope = telescope
self.staging_source_dir = staging_source_dir
self.sp_type = sp_type
self.sp_type = ScienceProductType(sp_type)
self.locator = locator
self.files_found = [file for file in staging_source_dir.iterdir()]
if len(self.files_found) == 0:
@@ -260,7 +251,7 @@ class IngestionManifestBuilder:
class IngestionManifest(ManifestIF):
"""needed for ingestion-launching interface"""
"""write ingestion manifest to file"""
def build_ingest_parameters(self):
"""
@@ -279,34 +270,7 @@ class IngestionManifest(ManifestIF):
staging_source_dir=self.staging_source_dir,
)
def _build_input_group(self):
"""
Create the input group using the parameters.
:return:
"""
# N.B. this is sufficient for most types of ingestion,
# but ALMA CALs will have multiple EB SPs, identified only by locator,
# and VLBAs have no input group at all.
sp_in = InputScienceProduct(locator=self.locator)
return InputGroup([sp_in])
def _build_output_group(self) -> OutputGroup:
"""
Create the output group using the parameters.
:return:
"""
sp_tar = self._find_science_product_tar()
find_output_science_products(self.files_found, self.staging_source_dir)
sps_out = [OutputScienceProduct(self.sp_type, sp_tar.name)]
# find ancillary products, if any
ancillary_products = self._find_ancillary_products()
weblog = Path(self.ingestion_path / WEBLOG_FILENAME)
if weblog.exists():
ancillary_products.append(AncillaryProduct(type=WEBLOG_FILENAME, filename=str(weblog)))
return OutputGroup(sps_out)
# @property
def ingestion_path(self) -> Path:
return self.parameters.ingestion_path
@@ -326,27 +290,6 @@ class IngestionManifest(ManifestIF):
return output_path
def create(self):
"""
Create the ingestion manifest in this directory for a product of this type,
identified by this locator.
:return:
"""
if self.sp_type != ScienceProductType.EVLA_CAL:
raise NotImplementedError(
f"Don't yet know how to handle {self.sp_type.value} ingestion"
)
builder = IngestionManifestBuilder(
staging_source_dir=Path(self.staging_source_dir),
sp_type=self.sp_type,
locator=self.locator,
telescope=self.telescope,
)
builder.build()
def to_json(self) -> JSON:
"""
Turn this object into a JSON string suitable for writing to a file
Loading