Merge pull request #78 from unity-sds/develop

Releast Attempt 1

Author: wphyojpl

CHANGELOG.md

@@ -7,6 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ===================================================================
 
+## [1.1.0.dev000004] - 2024-12-09
+### Fixed
+- [#76](https://github.com/unity-sds/mdps-ds-lib/pull/76) chore: update jupyter notebok
+
+## [1.1.0.dev000003] - 2024-12-09
+### Fixed
+- [#72](https://github.com/unity-sds/mdps-ds-lib/pull/72) fix: remove duplicated collection id prefix
+
+## [1.1.0.dev000002] - 2024-12-09
+### Fixed
+- [#71](https://github.com/unity-sds/mdps-ds-lib/pull/71) fix: no stage out file to s3 if empty
+
+## [1.1.0.dev000001] - 2024-12-09
+### Fixed
+- [#70](https://github.com/unity-sds/mdps-ds-lib/pull/70) fix: catalog links should not have colon
+
 ## [1.1.0] - 2024-12-03
 
 - [#67](https://github.com/unity-sds/mdps-ds-lib/pull/67) release 1.1.0

examples/stage-in-out-via-jupyter.ipynb

@@ -168,9 +168,10 @@ def download(self, **kwargs) -> str:
             error_list_list.append(error_list.get())
 
         for each_local_stac_item in local_items_list:
-            FileUtils.write_json(os.path.join(self._download_dir, f'{each_local_stac_item.id}.stac.json'),
+            local_stac_item_id_updated = each_local_stac_item.id.replace(':', '-').replace('/', '-').replace('\\', '-')
+            FileUtils.write_json(os.path.join(self._download_dir, f'{local_stac_item_id_updated}.stac.json'),
                                  each_local_stac_item.to_dict(False, False), overwrite=False, prettify=True)
-            catalog.add_link(Link('item', f'{each_local_stac_item.id}.stac.json', 'application/json'))
+            catalog.add_link(Link('item', f'{local_stac_item_id_updated}.stac.json', 'application/json'))
         self._granules_json.items = local_items_list
         LOGGER.debug(f'writing features collection json to downloading directory')
         granules_json_dict = self._granules_json.to_dict(transform_hrefs=False)

mdps_ds_lib/stage_in_out/download_granules_abstract.py

@@ -42,8 +42,10 @@ def generate_sample_stac(self, filepath: str):
         file_checksum = FileUtils.get_checksum(filepath, True)
         # https://github.com/stac-extensions/file
         # https://github.com/stac-extensions/file/blob/main/examples/item.json
+        granule_id = os.path.splitext(filename)[0]
+        granule_id = granule_id if granule_id.startswith(self.__collection_id) else f'{self.__collection_id}:{granule_id}'
         sample_stac_item = Item(
-                         id=f'{self.__collection_id}:{os.path.splitext(filename)[0]}',
+                         id=granule_id,
                          stac_extensions=["https://stac-extensions.github.io/file/v2.1.0/schema.json"],
                          geometry={
                              "type": "Point",
@@ -78,7 +80,8 @@ def execute_job(self, job_obj, lock) -> bool:
         updating_assets = {}
         try:
             LOGGER.audit(f'uploading auxiliary file: {job_obj}')
-            s3_url = self.__s3.upload(job_obj, self.__staging_bucket, f'{self.__collection_id}/{self.__collection_id}:{sample_stac_item.id}', self.__delete_files)
+            # NOTE: sample_stac_item.id is guaranteed to start with colleciton id.
+            s3_url = self.__s3.upload(job_obj, self.__staging_bucket, f'{self.__collection_id}/{sample_stac_item.id}', self.__delete_files)
             updating_assets[os.path.basename(s3_url)] = s3_url
             uploading_current_granule_stac = f'{s3_url}.stac.json'
             self.__s3.set_s3_url(uploading_current_granule_stac)

mdps_ds_lib/stage_in_out/upload_arbitrary_files_as_granules.py

@@ -50,8 +50,22 @@ def validate_job(self, job_obj):
     #     return
 
     def __exec_actual_job(self, each_child, lock) -> bool:
-        current_granule_stac: Item = self.__gc.get_granules_item(each_child)
-        current_collection_id = current_granule_stac.collection_id.strip()
+        try:
+            current_granule_stac: Item = self.__gc.get_granules_item(each_child)
+            current_collection_id = current_granule_stac.collection_id.strip()
+        except Exception as e:
+            LOGGER.exception(f'error while processing: {each_child}')
+            error_item = Item(id='unknown',
+                              properties={'message': 'unknown error', 'granule': each_child, 'details': str(e)},
+                              geometry={
+                                  "type": "Point",
+                                  "coordinates": [0.0, 0.0]
+                              },
+                              bbox=[0.0, 0.0, 0.0, 0.0],
+                              datetime=TimeUtils().parse_from_unix(0, True).get_datetime_obj(),
+                              collection='unknown')
+            self.__error_list.put(error_item.to_dict(False, False))
+            return True
         try:
             current_collection_id = GranulesCatalog.get_unity_formatted_collection_id(current_collection_id, self.__project_venue_set)
             LOGGER.debug(f'reformatted current_collection_id: {current_collection_id}')
@@ -211,14 +225,19 @@ def __actual_upload(self):
         FileUtils.write_json(failed_features_file, failed_item_collections.to_dict(False))
         if len(failed_item_collections.items) > 0:
             LOGGER.fatal(f'One or more Failures: {failed_item_collections.to_dict(False)}')
+
+        LOGGER.debug(f'creating response catalog')
+        catalog_json = GranulesCatalog().update_catalog(catalog_file_path, [successful_features_file, failed_features_file])
+        LOGGER.debug(f'catalog_json: {catalog_json}')
+        if len(successful_item_collections) < 1:  # TODO check this.
+            LOGGER.debug(f'No successful items in Upload: Not uploading successful_features_ to s3://{self._staging_bucket}/{self._result_path_prefix}')
+            return json.dumps(catalog_json)
+
         s3_url = self.__s3.upload(successful_features_file, self._staging_bucket,
                                   self._result_path_prefix,
                                   s3_name=f'successful_features_{TimeUtils.get_current_time()}.json',
                                   delete_files=self._delete_files)
         LOGGER.debug(f'uploaded successful features to S3: {s3_url}')
-        LOGGER.debug(f'creating response catalog')
-        catalog_json = GranulesCatalog().update_catalog(catalog_file_path, [successful_features_file, failed_features_file])
-        LOGGER.debug(f'catalog_json: {catalog_json}')
         return json.dumps(catalog_json)
 
     def __exec_dry_run(self):

mdps_ds_lib/stage_in_out/upload_granules_by_complete_catalog_s3.py

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "mdps-ds-lib"
-version="1.1.0"
+version="1.1.0.dev000004"
 description = "Mission Data System - MDPS (Unity) Data Service Core library"
 authors = ["Wai Phyo <wai.phyo@jpl.nasa.gov>"]
 license = "Apache 2.0"