Commit f7501d60 authored by BO ZHANG's avatar BO ZHANG 🏀
Browse files

add csst_dag.cli.csst_cpic_l1

parent 88da0a3e

csst_dag/cli/cpic.py

deleted100644 → 0
+0 −76
Original line number Original line Diff line number Diff line 
"""

Aim

---

Process an CPIC dataset, given a set of parameters.



Example

-------

python -m csst_dag.cli.cpic -h



python -m csst_dag.cli.cpic \

    --dataset=csst-cpic-c11-hip71681-v1 \

    --obs-group=hip71681 \

    --batch-id=csci-test-20250507 \

    --priority=1 \

    --initial-prc-status=-1024 \

    --final-prc-status=-2 \

    --demo

"""



from csst_dag.dag import CsstDAG

import argparse



parser = argparse.ArgumentParser(

    description="Scheduler for CPIC L1 pipeline.",

    formatter_class=argparse.ArgumentDefaultsHelpFormatter,

)



parser.add_argument("--dataset", type=str, help="Dataset name")

# parser.add_argument("--instrument", type=str, help="Instrument name", default="CPIC")

parser.add_argument("--obs-group", type=str, help="Observation group", default="none")

parser.add_argument("--obs-type", type=str, help="Observation type", default="")

parser.add_argument("--batch-id", type=str, help="Batch ID", default="default_batch")

parser.add_argument("--priority", type=str, help="Task priority", default=1)

parser.add_argument(

    "--initial-prc-status", type=int, help="Initial processing status", default=-1024

)

parser.add_argument(

    "--final-prc-status", type=int, help="Final processing status", default=-2

)

parser.add_argument(

    "--demo", action="store_true", help="Enable demo mode", default=False

)



args = parser.parse_args()

print("CLI parameters: ", args)



DAG_LOOP_MAP = {

    "SCI": ["csst-cpic-l1"],

    "DSF": ["csst-cpic-l1"],

    "CALS": ["csst-cpic-l1"],

    "BIAS": ["csst-cpic-l1-qc0"],

    "DARK": ["csst-cpic-l1-qc0"],

    "FLAT": ["csst-cpic-l1-qc0"],

    "BKG": ["csst-cpic-l1-qc0"],

    "LASER": ["csst-cpic-l1-qc0"],

}

if args.obs_type:

    assert args.obs_type in DAG_LOOP_MAP.keys(), f"Unknown obs_type: {args.obs_type}"

    DAG_LOOP_MAP = {args.obs_type: DAG_LOOP_MAP[args.obs_type]}



for obs_type, dag_ids in DAG_LOOP_MAP.items():

    print(f"* Processing {obs_type}")

    for dag_id in dag_ids:

        print(f"   - Scheduling `{dag_id}` -> ", end="")

        dag = CsstDAG.get_dag(dag_id=dag_id)

        msgs = dag.schedule(

            dataset=args.dataset,

            obs_type=obs_type,

            obs_group=args.obs_group,

            batch_id=args.batch_id,

            initial_prc_status=args.initial_prc_status,

            final_prc_status=args.final_prc_status,

            demo=True,

            priority=args.priority,

        )

        print(f"{len(msgs)} tasks.")

csst_dag/cli/csst_cpic_l1.py

0 → 100644
+126 −0
Original line number Original line Diff line number Diff line 
"""

Aim

---

Process an MSC dataset, given a set of parameters.



Example

-------

python -m csst_dag.cli.csst_msc_l1 -h



python -m csst_dag.cli.csst_msc_l1 \

    --dataset=csst-msc-c9-25sqdeg-v3 \

    --instrument=MSC \

    --obs-type=WIDE \

    --obs-group=W2 \

    --obs-id=10100232366 \

    --detector=09 \

    --prc-status=-1024 \

    --batch-id=test-b1 \

    --priority=1 \

    --pmapname=csst_000070.pmap \

    --ref-cat=trilegal_093

"""



import argparse

import os



import joblib



from csst_dag import CSST_DAGS, Dispatcher, BaseDAG, dfs



parser = argparse.ArgumentParser(

    description="Scheduler for CSST MSC L1 pipeline.",

    formatter_class=argparse.ArgumentDefaultsHelpFormatter,

)



# data parameters

parser.add_argument("--dataset", type=str, help="Dataset name")

parser.add_argument("--instrument", type=str, help="Instrument name", default=None)

parser.add_argument("--obs-type", type=str, help="Observation type", default=None)

parser.add_argument("--obs-group", type=str, help="Observation group", default=None)

parser.add_argument("--obs-id", type=str, help="Observation ID", default=None)

parser.add_argument("--detector", type=str, help="Detector name", default=None)

parser.add_argument(

    "--prc-status", type=int, help="Initial processing status", default=None

)

# task parameters

parser.add_argument("--batch-id", type=str, help="Batch ID", default="test-batch")

parser.add_argument("--priority", type=str, help="Task priority", default=1)

# DAG parameters

parser.add_argument("--pmapname", type=str, help="CCDS pmapname", default="")

parser.add_argument(

    "--ref-cat", type=str, help="Reference catalog", default="trilegal_093"

)

# submit

parser.add_argument("--submit", action="store_true", help="Push results", default=False)

# post-processing parameters

parser.add_argument(

    "--final-prc-status", type=int, help="Final processing status", default=-2

)



args = parser.parse_args()

print("CLI parameters: ", args)





plan_basis, data_basis = Dispatcher.find_plan_level0_basis(

    dataset=args.dataset,

    instrument=args.instrument,

    obs_type=args.obs_type,

    obs_group=args.obs_group,

    obs_id=args.obs_id,

    detector=args.detector,

    prc_status=args.prc_status,

)

print(f"{len(plan_basis)} plan basis, {len(data_basis)} data basis found")



# generate DAG group run

dag_group_run = BaseDAG.gen_dag_group_run(

    dag_group="csst-msc-l1",

    batch_id=args.batch_id,

    priority=args.priority,

)



# generate DAG run list

dag_run_list = []

for dag in [

    "csst-cpic-l1",

    "csst-cpic-l1-qc0",

]:

    this_dag_run_list = CSST_DAGS["csst-msc-l1-mbi"].schedule(

        dag_group_run=dag_group_run,

        plan_basis=plan_basis,

        data_basis=data_basis,

        pmapname=args.pmapname,

        ref_cat=args.ref_cat,

    )

    print(f"- [{dag}] : {len(this_dag_run_list)} dag_runs")

    dag_run_list.extend(this_dag_run_list)



# print dag_group_run and dag_run_list

print(">>> dag_group_run:")

for k, v in dag_group_run.items():

    print(f"\t- {k}: {v}")

print(f">>> dag_run_list[{len(dag_run_list)}]:")

if len(dag_run_list) > 0:

    for dag_run in dag_run_list:

        print(f"\t- {dag_run}")



# dump dag_group_run

joblib.dump(

    dict(

        dag_group_run=dag_group_run,

        dag_run_list=dag_run_list,

    ),

    os.path.join(

        os.getenv("HOME"),

        "csst_dag",

        f"{dag_group_run['dag_group_run']}.joblib",

    ),

)

# submit DAG group run

if args.submit:

    res = dfs.dag.new_dag_group_run(

        dag_group_run=dag_group_run,

        dag_run_list=dag_run_list,

    )

    print(res)

csst_dag/cli/csst_msc_l1.py

+13 −11
Original line number Original line Diff line number Diff line
@@ -20,14 +20,13 @@ python -m csst_dag.cli.csst_msc_l1 \ 
    --pmapname=csst_000070.pmap \

    --pmapname=csst_000070.pmap \

    --ref-cat=trilegal_093

    --ref-cat=trilegal_093





# 25平方度宽场

python -m csst_dag.cli.csst_msc_l1 \

python -m csst_dag.cli.csst_msc_l1 \

    --dataset=csst-msc-c9-25sqdeg-v3 \

    --dataset=csst-msc-c9-25sqdeg-v3 \

    --instrument=MSC \

    --instrument=MSC \

    --obs-type=WIDE \

    --obs-type=WIDE \

    --obs-group=W2 \

    --prc-status=-2 \

    --obs-id=10100232366 \

    --batch-id=25sqdeg-test-b2 \

    --detector=09 \

    --batch-id=test-b1 \

    --priority=1 \

    --priority=1 \

    --pmapname=csst_000070.pmap \

    --pmapname=csst_000070.pmap \

    --ref-cat=trilegal_093 \

    --ref-cat=trilegal_093 \
@@ -37,18 +36,19 @@ python -m csst_dag.cli.csst_msc_l1 \ 
python -m csst_dag.cli.csst_msc_l1 \

python -m csst_dag.cli.csst_msc_l1 \

    --dataset=csst-msc-c11-1000sqdeg-wide-test-v1 \

    --dataset=csst-msc-c11-1000sqdeg-wide-test-v1 \

    --instrument=MSC \

    --instrument=MSC \

    --prc-status=-1024 \

    --batch-id=1000sqdeg-test-b1 \

    --batch-id=1000sqdeg-test-b1 \

    --ref-cat=trilegal_1000_w1 \

    --ref-cat=trilegal_1000_w1 \

    --submit

    --submit

"""

"""





import argparse

import os

import os





from csst_dag.dag import CSST_DAGS, Dispatcher, BaseDAG

from csst_dag import dfs

import argparse

import joblib

import joblib





from csst_dag import CSST_DAGS, Dispatcher, BaseDAG, dfs



parser = argparse.ArgumentParser(

parser = argparse.ArgumentParser(

    description="Scheduler for CSST MSC L1 pipeline.",

    description="Scheduler for CSST MSC L1 pipeline.",

    formatter_class=argparse.ArgumentDefaultsHelpFormatter,

    formatter_class=argparse.ArgumentDefaultsHelpFormatter,
@@ -134,16 +134,18 @@ for dag in [ 
        pmapname=args.pmapname,

        pmapname=args.pmapname,

        ref_cat=args.ref_cat,

        ref_cat=args.ref_cat,

    )

    )

    print(f"- [{dag}] : {len(this_dag_run_list)} dag_runs")

    dag_run_list.extend(this_dag_run_list)

    dag_run_list.extend(this_dag_run_list)





# print dag_group_run and dag_run_list

# print dag_group_run and dag_run_list

print("`dag_group_run`:")

print(">>> dag_group_run:")

for k, v in dag_group_run.items():

for k, v in dag_group_run.items():

    print(f"\t- {k}: {v}")

    print(f"\t- {k}: {v}")

print(f"`dag_run_list`[{len(dag_run_list)}]:")

print(f">>> dag_run_list[{len(dag_run_list)}]:")

if len(dag_run_list) > 0:

if len(dag_run_list) > 0:

    for k, v in dag_run_list[0].items():

    for dag_run in dag_run_list:

        print(f"\t- {k}: {v}")

        print(f"\t- {dag_run}")



# dump dag_group_run

# dump dag_group_run

joblib.dump(

joblib.dump(

    dict(

    dict(