PR-7502 Implemented CNM creation

Author: benbart

.github/workflows/lint.yml

@@ -25,5 +25,5 @@ jobs:
       - uses: actions/checkout@v4
       - uses: astral-sh/ruff-action@v3
         with:
-          version: "~=0.13.3"
+          version: "~=0.15.13"
           args: format --check --diff --output-format=github

ctorm/ctorm.cfg.example

@@ -1,22 +1,26 @@
 [ctorm]
 
+calc_md5 = true
+granule_goal = 200
 granules_sqs_queue_url = "https://sqs.us-west-2.amazonaws.com/123456789012/ctorm-dev-granules"
 
 [[ ctorm.pipelines ]]
 bucketname = "nisar-bucket-0"
 ingest_keypair_name = ""
 ingest_queue_name = "sds-n-cumulus-test-nisar-workflow-queue"
 ingest_sqs_queue_url = ""
+prepare_keypair_name = "NISAR_P"
+provider = "FOO"
 share = 50
 ummg_prefix = "UMMG/FOO_COLLECTION/"
-prepare_keypair_name = "NISAR_P"
 
 [[ctorm.pipelines]]
 bucketname = "nisar-bucket-2"
 ingest_keypair_name = ""
 ingest_queue_name = "sds-n-cumulus-test-nisar-workflow-queue"
 ingest_sqs_queue_url = ""
 prepare_keypair_name = "NISAR"
+provider = "FOO"
 share = 45
 ummg_prefix = "UMMG/BAR_COLLECTION/"
 
@@ -26,5 +30,6 @@ ingest_keypair_name = ""
 ingest_queue_name = "asf-cumulus-test-opera-workflow-queue"
 ingest_sqs_queue_url = ""
 prepare_keypair_name = "OPERA"
+provider = "FOO"
 share = 5
 ummg_prefix = "UMMG/BAZ_COLLECTION/"

ctorm/ctorm/cnm.py

@@ -0,0 +1,131 @@
+import re
+import uuid
+from collections.abc import Callable
+from datetime import datetime, timezone
+from logging import getLogger
+from pathlib import Path
+
+from ctorm.config import (
+    MD5_CHECKSUM_PATTERN,
+    CtormPipeline,
+    CtormPreparedFile,
+    CtormPreparedGranule,
+)
+
+log = getLogger(__name__)
+
+DATA_TYPE_MAP = {
+    ".context.json": "metadata",
+    ".dataset.json": "metadata",
+    ".h5": "data",
+    ".iso.xml": "metadata",
+    ".jpg": "browse",
+    ".log": "metadata",
+    ".md5": "metadata",
+    ".met.json": "metadata",
+    ".nc": "data",
+    ".pdf": "qa",
+    ".png": "browse",
+    ".qa.h5": "qa",
+    ".rc.yaml": "metadata",
+    ".tif": "data",
+    ".xml": "data",
+}
+
+type CtormCnmSGeneratorType = Callable[[CtormPreparedGranule, str], dict]
+
+
+class CtormCnmSGenerator:
+    """input dict like:
+    ```
+    {
+      "bm": {
+        "sds-n-cumulus-prod-nisar-products": "B1",
+        "sds-n-cumulus-prod-nisar-jpl-private-data": "B2"
+      },
+      "g": "NISAR_L0_RRST_VC08_20250821T101036_20250821T101041_P00408_J_001",
+      "c": "NISAR_L0A_RRST_BETA_V1",
+      "cv": "1",
+      "f": [
+        {
+          "f": "s3://$B1/NISAR_L0A_RRST_BETA_V1/$G/$G.bin",
+          "m": "e989430f4c5bfa04eeb26e56f4cd8375",
+          "s": 1073829760
+        },
+        {
+          "f": "s3://$B1/NISAR_L0A_RRST_BETA_V1/$G/$G.rc.yaml",
+          "s": 120597,
+          "m": "11bcaec780f0d246f3996b3f08680410"
+        },
+        {
+          "f": "s3://$B1/NISAR_L0A_RRST_BETA_V1/$G/$G.bin.qa",
+          "s": 1134,
+        },
+      ]
+    },
+    ```
+    """
+
+    def _assemble_file_dict(
+        self, file_dict: CtormPreparedFile, g_name: str, bucket_map: dict
+    ):
+        file_s3uri = file_dict["f"].replace("$G", g_name)
+        for bucket, replacetoken in bucket_map.items():
+            file_s3uri = file_s3uri.replace(f"${replacetoken}", bucket)
+        filename = Path(file_s3uri).name
+        return {
+            "name": filename,
+            "type": self._get_type(filename),
+            "uri": file_s3uri,
+            "size": file_dict["s"],
+            "checksum": self._get_checksum(file_dict),
+            "checksumType": "md5",
+        }
+
+    def __call__(
+        self,
+        granule_dict: CtormPreparedGranule,
+        provider: str,
+    ) -> dict:
+        cnm_s = {
+            "identifier": str(uuid.uuid4()),
+            "collection": granule_dict["c"],
+            "version": "1.3",
+            "submissionTime": datetime.now(tz=timezone.utc).strftime(
+                "%Y-%m-%dT%H:%M:%S.%fZ"
+            ),
+            "product": {
+                "name": granule_dict["g"],
+                "dataVersion": granule_dict["cv"],
+                "files": [
+                    self._assemble_file_dict(
+                        file, granule_dict["g"], granule_dict["bm"]
+                    )
+                    for file in granule_dict["f"]
+                ],
+            },
+            "provider": provider,
+        }
+
+        return cnm_s
+
+    def _get_type(self, filename: str):
+
+        suffixes = Path(filename).suffixes
+        while suffixes:
+            data_type = DATA_TYPE_MAP.get("".join(suffixes))
+
+            if data_type:
+                return data_type
+
+            suffixes = suffixes[1:]
+
+        log.debug("suffix from %s is not in map. Using 'data'", filename)
+        return "data"
+
+    def _get_checksum(self, file_dict: CtormPreparedFile):
+        m = MD5_CHECKSUM_PATTERN.match(file_dict.get("m", "not md5"))
+        if m:
+            return m.group(1)
+
+        return "00000000000000000000000000000000"

ctorm/ctorm/cnm_sender.py

@@ -1,6 +1,25 @@
+from logging import getLogger
+from typing import List
+
+from ctorm.cnm import CtormCnmSGenerator
+from ctorm.config import CtormConfig, CtormPreparedGranule
+
+log = getLogger(__name__)
+
+
 class CnmSender:
-    def __init__(self, cfg):
+    def __init__(
+        self,
+        cfg: CtormConfig,
+        granules: List[CtormPreparedGranule],
+        provider: str = "TODO: PROVIDER",
+    ):
         self.cfg = cfg
+        self.granules = granules
+        self.provider = provider
+        self.cnm_s_generator = CtormCnmSGenerator()
 
-    def send(self):
-        pass
+    def send_all(self):
+        for granule in self.granules:
+            cnms = self.cnm_s_generator(granule, self.provider)
+            log.debug("Sending %s", cnms)

ctorm/ctorm/config.py

@@ -1,6 +1,8 @@
+import re
+import tomllib
 from dataclasses import dataclass, fields
+from typing import NotRequired, TypedDict
 
-import tomllib
 from boto3.session import Session
 from mypy_boto3_s3 import S3Client
 
@@ -11,6 +13,21 @@
 
 DEFAULT = "default"
 AWS_REGION = "us-west-2"  # We will never not want us-west-2
+MD5_CHECKSUM_PATTERN = re.compile(r"^([\da-f]{32})$")
+
+
+class CtormPreparedFile(TypedDict):
+    f: str
+    m: NotRequired[str]
+    s: int
+
+
+class CtormPreparedGranule(TypedDict):
+    bm: dict[str, str]
+    g: str
+    c: str
+    cv: str
+    f: list[CtormPreparedFile]
 
 
 @dataclass
@@ -33,13 +50,16 @@ class CtormPipeline:
     s3_client: S3Client = None
     session: Session = None
     ummg_prefix: str = "UMMG/"
+    provider: str = "TODO"
 
 
 @dataclass
 class CtormConfig:
     pipelines: list
     granules_sqs_queue_url: str
 
+    calc_md5: bool = True
+
     # This is the number of granules that will be prepared for test.
     granule_goal: int = 2000000
 

ctorm/ctorm/lambda_run.py

@@ -3,9 +3,21 @@
 """
 
 import json
+import logging
 
 from ctorm.load_tester import lambda_handler
 
+log = logging.getLogger()
+
+base_fmt_str = "%(levelname)s: %(message)s (%(filename)s line %(lineno)d/)"
+screen_fmt = logging.Formatter(
+    "%(asctime)s.%(msecs)d " + base_fmt_str, "%Y-%m-%dT%H:%M:%S"
+)
+screenlog = logging.StreamHandler()
+screenlog.setFormatter(screen_fmt)
+log.addHandler(screenlog)
+
+
 # mock event data
 with open("../../data/lambda_event.json", "r") as f:
     mock_event = json.load(f)

ctorm/ctorm/load_tester.py

@@ -1,11 +1,27 @@
+import json
 import logging
 import os
+from typing import List
 
+from aws_lambda_typing import context as context_
+from aws_lambda_typing import events
 from cnm_sender import CnmSender
 
+from ctorm.config import (
+    CtormConfig,
+    CtormPreparedGranule,
+)
+
 log = logging.getLogger(__name__)
 
 
+def configure_cfg():
+    cfg = CtormConfig.from_file(
+        cfg_file=os.getenv("CFG_FILE", "./ctorm.cfg"),
+    )
+    return cfg
+
+
 def configure_logging() -> None:
     level_name = os.environ.get("LOG_LEVEL", "INFO").upper()
     level = getattr(logging, level_name, logging.INFO)
@@ -21,17 +37,20 @@ def configure_logging() -> None:
     for handler in root_logger.handlers:
         handler.setLevel(level)
         handler.setFormatter(
-            logging.Formatter("%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)d - %(message)s")
+            logging.Formatter(
+                "%(asctime)s %(levelname)s %(name)s %(filename)s:%(lineno)d - %(message)s"
+            )
         )
 
 
-def load_test():
-    c_sender = CnmSender({})
-    c_sender.send()
+def load_test(cfg: CtormConfig, granule_list: List[CtormPreparedGranule]):
+    c_sender = CnmSender(cfg, granule_list)
+    c_sender.send_all()
 
 
-def lambda_handler(event, context):
+def lambda_handler(event: events.EventBridgeEvent, context: context_.Context):
     configure_logging()
+    cfg = configure_cfg()
 
     log.info(
         "Starting CNM sender invocation",
@@ -42,8 +61,9 @@ def lambda_handler(event, context):
 
     try:
         log.debug("Received event: %s", event)
-
-        load_test()
+        g_list = event["Records"].pop().get("body")
+        g_list = json.loads(g_list).get("granules")
+        load_test(cfg, g_list)
 
         log.info("CNM sender invocation completed")
         return {"ok": True}
@@ -54,5 +74,4 @@ def lambda_handler(event, context):
 
 
 if __name__ == "__main__":
-    configure_logging()
     lambda_handler({}, {})