add generate_meta

csst_common/io.py

@@ -12,6 +12,7 @@ Modified-History:
 """
 
 import os
+import json
 import bisect
 import warnings
 from copy import deepcopy
@@ -20,6 +21,69 @@ from typing import Optional
 from astropy.io import fits
 
 
+REQUIRED_KEYS_IN_META = dict(
+    # 观测数据相关的ID参数
+    dataset="__instrument__",
+    instrument="__instrument__",
+    obs_type="__obs_type__",
+    obs_group="__obs_group__",
+    obs_id="__obs_id__",
+    detector="__detector__",
+    custom_id="__custom_id__",  # 替代brick_id
+    # unique_id="",  # 替代level0/1 ID 比如obs_id+detector
+    object="",
+    # 数据处理相关的参数
+    data_model="__data_model__",  # 数据产品类型，可用于判断入库处理方式；"csst-msc-l1-mbi-cat" -> Table/DataFrame,
+    batch_id="__batch_id__",
+    qc_status=-1024,
+    dag="__dag__",
+    dag_run="__dag_run__",
+    image="__image__",
+    build="__build__",
+    pmapname="__pmapname__",
+    # 额外的观测筛选参数
+    obs_date="1970-01-01T00:00:00",
+    create_date="1970-01-01T00:00:00",
+    filter="__filter__",
+    ra=-3.141592653589793,
+    dec=-3.141592653589793,
+)
+
+
+def generate_meta(**kwargs) -> fits.Header:
+    """
+    Generate ingest meta.
+
+    Parameters
+    ----------
+    kwargs : dict
+        Meta data.
+
+    Returns
+    -------
+    fits.Header
+        Meta data header.
+    """
+    meta = deepcopy(REQUIRED_KEYS_IN_META)
+    for k in kwargs.keys():
+        if k in meta.keys():
+            # 类型必须兼容
+            assert isinstance(
+                kwargs[k], type(meta[k])
+            ), f"类型不兼容: kwargs['{k}'] 应为 {type(meta[k]).__name__} 或其子类"
+            meta[k] = kwargs[k]
+            # 赋值
+            kwargs[k] = str(kwargs[k])
+        else:
+            raise KeyError(f"未知的meta参数: {k}")
+    meta_card = fits.Card(
+        keyword="META",
+        value=json.dumps(kwargs, separators=(",", ":"), ensure_ascii=False),
+    )
+    meta_header = fits.Header(cards=[meta_card])
+    return meta_header
+
+
 def verify_checksum(file_path) -> bool:
     """
     Verify a .fits file via checksum.