Commit 1f63caeb authored by BO ZHANG's avatar BO ZHANG 🏀
Browse files

refactor(dag): 重构Brick DAG的healpix查询逻辑，按data_model层级分流

parent 5db366f9

README.md

+5 −4
Original line number Diff line number Diff line
@@ -109,9 +109,10 @@ DagRunGroup.trigger( 
Brick / healpix 查询补充说明：



- 对 `FileBrickDAG` 来说，`data_model` 只是上游数据的业务筛选条件，不要求名字体现 `l1` / `l2`。

- `FileBrickDAG` 在生成 DagRun 前查询唯一 `healpix` 时，始终固定走 `level1.find_unique_healpix()`。

- 对 `CatalogBrickDAG` 也是同样逻辑，`data_model` 不要求体现层级，但唯一 `healpix` 查询始终固定走 `level2.find_unique_healpix()`。

- 也就是说，Brick DAG 的 unique healpix 搜索后端由 DAG 类型决定，而不是由 `data_model` 名称或注册层级决定。

- `CatalogBrickDAG` 也是同样逻辑，`data_model` 不要求体现层级。

- 在当前 v2 实现里，唯一 `healpix` 查询后端由 `data_model.toml` 中声明的存储层级决定。

- `level1` 统一使用 `find_unique_healpix()`。

- `level2` 统一使用 `find_unique_healpix_in_catalog()`，禁止继续使用 `find_unique_healpix()`。



#### `proc` 支持的参数
@@ -142,7 +143,7 @@ match: 


- `match` 不应再包含 `dag_type`。

- `dag_type` 不是输入匹配键，而是某个 DAG 的固有属性。

- 在当前 v2 实现里，DAG 类型由注册时使用的 Python 类决定，例如 `FileDAG` 或 `BrickDAG`。

- 在当前 v2 实现里，DAG 类型由注册时使用的 Python 类决定，例如 `FileDAG`、`FileBrickDAG` 或 `CatalogBrickDAG`。

- 因此新增 DAG 时，不需要在 YAML 的 `match` 中再写 `dag_type`。



#### DAG YAML 中的 `submission` 配置

csst_dag/v2/config_loader.py

+67 −8
Original line number Diff line number Diff line
@@ -7,12 +7,25 @@ DAG_CONFIG_DIR_V2 = os.path.join( 
)





def _normalize_field_group(field_group: dict | None) -> tuple[dict, dict]:

    """将字段中心的 submission 配置展开为 defaults/fixed 两张表。"""

def _normalize_option_values(raw_value) -> list[str]:

    """将配置值规整为可用于下拉框的字符串列表。"""

    if raw_value is None:

        return []

    if isinstance(raw_value, (list, tuple, set)):

        values = [str(item).strip() for item in raw_value if str(item).strip()]

        return list(dict.fromkeys(values))



    value = str(raw_value).strip()

    return [value] if value else []





def _normalize_field_group(field_group: dict | None) -> tuple[dict, dict, dict]:

    """将字段中心的 submission 配置展开为 defaults/fixed/options 三张表。"""

    defaults: dict = {}

    fixed: dict = {}

    options: dict = {}

    if not isinstance(field_group, dict):

        return defaults, fixed

        return defaults, fixed, options



    for field_name, field_cfg in field_group.items():

        if isinstance(field_cfg, dict):
@@ -20,10 +33,44 @@ def _normalize_field_group(field_group: dict | None) -> tuple[dict, dict]: 
                defaults[field_name] = field_cfg.get("value")

            if bool(field_cfg.get("fixed")) and "value" in field_cfg:

                fixed[field_name] = field_cfg.get("value")

            option_values = _normalize_option_values(field_cfg.get("options"))

            if option_values:

                options[field_name] = option_values

            continue

        if field_cfg is not None:

            defaults[field_name] = field_cfg

    return defaults, fixed

    return defaults, fixed, options





def _normalize_match_options(match_cfg: dict | None) -> dict:

    """提取 DAG match 段中可直接作为 Submission 下拉候选项的字段。"""

    options: dict = {}

    if not isinstance(match_cfg, dict):

        return options



    for field_name, field_value in match_cfg.items():

        option_values = _normalize_option_values(field_value)

        if option_values:

            options[field_name] = option_values

    return options





def _normalize_multiplier_options(dag_cfg: dict | None) -> dict:

    """提取 DAG 乘子字段，供 Submission 页面生成下拉选项。"""

    options: dict = {}

    if not isinstance(dag_cfg, dict):

        return options



    for item in dag_cfg.get("brick_multipliers") or []:

        if not isinstance(item, dict):

            continue

        field_name = str(item.get("field") or "").strip()

        if not field_name:

            continue

        option_values = _normalize_option_values(item.get("values"))

        if option_values:

            options[field_name] = option_values

    return options





def _normalize_submission_section(submission_cfg: dict | None) -> dict:
@@ -31,20 +78,24 @@ def _normalize_submission_section(submission_cfg: dict | None) -> dict: 
    normalized = {

        "data_defaults": {},

        "data_fixed": {},

        "data_options": {},

        "proc_defaults": {},

        "proc_fixed": {},

        "proc_options": {},

    }

    if not isinstance(submission_cfg, dict):

        return normalized



    # 新结构：submission.data.<field>.value/fixed 与 submission.proc.<field>.value/fixed

    data_defaults, data_fixed = _normalize_field_group(submission_cfg.get("data"))

    proc_defaults, proc_fixed = _normalize_field_group(submission_cfg.get("proc"))

    if data_defaults or data_fixed or proc_defaults or proc_fixed:

    data_defaults, data_fixed, data_options = _normalize_field_group(submission_cfg.get("data"))

    proc_defaults, proc_fixed, proc_options = _normalize_field_group(submission_cfg.get("proc"))

    if data_defaults or data_fixed or data_options or proc_defaults or proc_fixed or proc_options:

        normalized["data_defaults"] = data_defaults

        normalized["data_fixed"] = data_fixed

        normalized["data_options"] = data_options

        normalized["proc_defaults"] = proc_defaults

        normalized["proc_fixed"] = proc_fixed

        normalized["proc_options"] = proc_options

        return normalized



    # 兼容旧结构：data_defaults/data_fixed/proc_defaults/proc_fixed
@@ -90,4 +141,12 @@ def load_dag_config(dag_name: str) -> dict: 
def load_submission_config(dag_name: str) -> dict:

    """读取 DAG YAML 中的 submission 表单默认值/固定值配置。"""

    dag_cfg = load_dag_config(dag_name)

    return _normalize_submission_section(dag_cfg.get("submission"))
 
    normalized = _normalize_submission_section(dag_cfg.get("submission"))



    for field_name, option_values in _normalize_match_options(dag_cfg.get("match")).items():

        normalized["data_options"].setdefault(field_name, option_values)



    for field_name, option_values in _normalize_multiplier_options(dag_cfg).items():

        normalized["data_options"].setdefault(field_name, option_values)



    return normalized

csst_dag/v2/dag/__init__.py

+5 −2
Original line number Diff line number Diff line 
from .base import BaseDAG, DependencyBundle

from .brick import BrickDAG, CatalogBrickDAG, FileBrickDAG

from .brick import CatalogBrickDAG, FileBrickDAG

from .dispatcher import Dispatcher

from .file import FileDAG
@@ -26,6 +26,10 @@ CSST_DAGS = { 
        "csst-msc-l1-sls",

        dispatcher=Dispatcher.dispatch_file,

    ),

    "csst-msc-l1-sls-1d": FileDAG(

        "csst-msc-l1-sls-1d",

        dispatcher=Dispatcher.dispatch_file,

    ),

    "csst-msc-l1-ooc": FileDAG(

        "csst-msc-l1-ooc",

        dispatcher=Dispatcher.dispatch_obsgroup_detector,
@@ -70,7 +74,6 @@ CSST_DAGS = { 


__all__ = [

    "BaseDAG",

    "BrickDAG",

    "CatalogBrickDAG",

    "CSST_DAGS",

    "DependencyBundle",

csst_dag/v2/dag/brick.py

+14 −23
Original line number Diff line number Diff line
@@ -21,8 +21,9 @@ class BaseBrickDAG(BaseDAG): 
    调用 DFS 的 `find_unique_healpix()` 检索唯一 healpix 列表。

    随后以 healpix 为基本分组键生成 DagRun；若 DAG 配置中声明了

    `brick_multipliers`，则进一步生成 `healpix × multiplier` 的排列组合。

    这里选择哪个 DFS 后端由 DAG 类型决定，而不是由 `data_model`

    的字符串形式或注册层级决定。

    这里选择哪个 DFS 后端由 `data_model.toml` 中声明的存储层级决定：

    - level1 使用 `find_unique_healpix`

    - level2 使用 `find_unique_healpix_in_catalog`

    """



    UNIQUE_QUERY_KEYS = (
@@ -55,8 +56,6 @@ class BaseBrickDAG(BaseDAG): 
        data_model = str(data_query.get("data_model") or "").strip()

        if not data_model or data_model.lower() == "raw":

            raise ValueError(f"{self.__class__.__name__} 需要在 data 中显式提供非 raw 的 data_model。")

        # 对 Brick DAG 来说，data_model 只是上游数据筛选条件，

        # 不能再用它反推 unique_healpix 应该访问 level1 还是 level2。

        return plan_query, data_query, upstream_query



    @staticmethod
@@ -78,9 +77,13 @@ class BaseBrickDAG(BaseDAG): 
        }



    def resolve_unique_healpix(self, data_query: dict[str, Any]) -> list[str]:

        """按 DAG 类型选择固定的 unique_healpix 搜索后端。"""

        """按 data_model 的注册层级选择 unique_healpix 搜索后端。"""

        return self.find_unique_healpix(**data_query)



    def find_unique_healpix(self, **data_query) -> list[str]:

        """统一通过 data_model 注册层级分流唯一 healpix 查询。"""

        return dfs_data.find_unique_healpix(**data_query)



    def resolve_data_dependency(self, data_query: dict[str, Any]) -> Table:

        unique_healpix = self.resolve_unique_healpix(self._build_unique_healpix_query(data_query))

        if not unique_healpix:
@@ -208,33 +211,21 @@ class BaseBrickDAG(BaseDAG): 


class FileBrickDAG(BaseBrickDAG):

    """

    基于 `level1.find_unique_healpix()` 的文件型 Brick DAG。

    文件型 Brick DAG。



    Notes

    -----

    `data_model` 可以是任意业务数据模型名，不要求名字体现 `l1`/`l2`。

    FileBrickDAG 的 unique healpix 搜索后端始终固定为 level1。

    该类型仅表达 DAG 的业务语义；唯一 healpix 查询后端由 `data_model`

    的存储层级决定，不再由 DAG 类型固定。

    """



    def find_unique_healpix(self, **data_query) -> list[str]:

        """文件型 Brick 固定通过 level1 检索唯一 healpix。"""

        return dfs_data.find_level1_unique_healpix(**data_query)





class CatalogBrickDAG(BaseBrickDAG):

    """

    基于 `level2.find_unique_healpix()` 的星表型 Brick DAG。

    星表型 Brick DAG。



    Notes

    -----

    `data_model` 可以是任意业务数据模型名，不要求名字体现 `l1`/`l2`。

    CatalogBrickDAG 的 unique healpix 搜索后端始终固定为 level2。

    该类型仅表达 DAG 的业务语义；唯一 healpix 查询后端由 `data_model`

    的存储层级决定，不再由 DAG 类型固定。

    """
 


    def find_unique_healpix(self, **data_query) -> list[str]:

        """星表型 Brick 固定通过 level2 检索唯一 healpix。"""

        return dfs_data.find_level2_unique_healpix(**data_query)





class BrickDAG(FileBrickDAG):

    """兼容旧导入路径的别名，默认仍表示 level1 文件型 Brick。"""

csst_dag/v2/dag_config/csst-msc-l1-mbi.yml

+1 −1
Original line number Diff line number Diff line
@@ -10,7 +10,7 @@ tags: ["csst", "msc", "l1", "mbi"] 
# DAG 匹配关键字（用于 dispatcher 根据观测/数据特征选择 DAG）

match:

  instrument: "MSC"

  obs_type: ["WIDE", "DEEP", "CALF"]

  obs_type: ["WIDE", "DEEP", "CALF", "CALSP", "CALSS"]

  detector_group: "MBI"



# 默认输入参数（JSON 字符串；DAG 的输入统一从该字符串解析）