automatically set docker_image, build and created_time

csst_common/io.py

@@ -21,6 +21,8 @@ from typing import Optional, Any
 
 from astropy.io import fits
 
+from .time import now
+
 
 REQUIRED_KEYS_IN_META = dict(
     # 观测数据相关的ID参数
@@ -31,19 +33,19 @@ REQUIRED_KEYS_IN_META = dict(
     obs_id="__obs_id__",
     detector="__detector__",
     custom_id="__custom_id__",  # 替代brick_id
-    # unique_id="__unique_id__",  # 替代level0/1 ID 比如obs_id+detector
     object="__object__",
+    obs_date="1970-01-01T00:00:00.000",
+    # unique_id="__unique_id__",  # 替代level0/1 ID 比如obs_id+detector
     # 数据处理相关的参数
     data_model="__data_model__",  # 数据产品类型，可用于判断入库处理方式；"csst-msc-l1-mbi-cat" -> Table/DataFrame,
     batch_id="__batch_id__",
-    qc_status=-1024,
     dag="__dag__",
     dag_run="__dag_run__",
     pmapname="__pmapname__",
     ref_cat="__ref_cat__",
-    # 额外的观测筛选参数
-    obs_date="1970-01-01T00:00:00.000",
+    qc_status=-1024,
     created_date="1970-01-01T00:00:00.000",
+    # 额外的观测筛选参数
     filter="__filter__",
     ra=-3.141592653589793,
     dec=-3.141592653589793,
@@ -80,9 +82,10 @@ def generate_meta(**kwargs) -> fits.Header:
             meta[k] = kwargs[k]
         else:
             raise KeyError(f"未知的meta参数: {k}")
-    # automatically set docker_image and build
+    # automatically set docker_image, build, and created_date
     meta["docker_image"] = os.getenv("PIPELINE_ID", default="__docker_image__")
     meta["build"] = os.getenv("BUILD", default="__build__")
+    meta["created_date"] = now()
     # convert meta to fits header
     meta_card = fits.Card(
         keyword="META",
@@ -156,7 +159,9 @@ def verify_checksum(file_path) -> bool:
 
 
 def append_header(
-    h1: fits.Header, h2: fits.Header, duplicates: str = "delete"
+    h1: fits.Header,
+    h2: fits.Header,
+    duplicates: str = "delete",
 ) -> fits.Header:
     """
     Append h2 to h1.