msc.py

"""
Aim
---
Process an MSC dataset, given a set of parameters.

Example
-------
python -m csst_dag.cli.msc -h

python -m csst_dag.cli.msc \
    --batch-id=csci-test-20250507 \
    --dataset=csst-msc-c9-25sqdeg-v3 \
    --obs-group=W1 \
    --priority=1 \
    --initial-prc-status=-1024 \
    --final-prc-status=-2 \
    --demo
"""

from csst_dag.dag import CsstDAGs
import argparse

parser = argparse.ArgumentParser(
    description="Scheduler for MSC L1 pipeline.",
    formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)

# task related parameters
parser.add_argument("--batch-id", type=str, help="Batch ID", default="default_batch")
parser.add_argument("--priority", type=str, help="Task priority", default=1)
# data related parameters
parser.add_argument("--dataset", type=str, help="Dataset name")
# parser.add_argument("--instrument", type=str, help="Instrument name", default="MSC")
parser.add_argument("--obs-group", type=str, help="Observation group", default="none")
parser.add_argument("--obs-type", type=str, help="Observation type", default="")
# status related parameters
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)

# define DAP LIST in this group
DAG_MAP = {
    "WIDE": ["csst-msc-l1-mbi", "csst-msc-l1-sls", "csst-msc-l1-ast"],
    "DEEP": ["csst-msc-l1-mbi", "csst-msc-l1-sls", "csst-msc-l1-ast"],
    "BIAS": ["csst-msc-l1-qc0"],
    "DARK": ["csst-msc-l1-qc0"],
    "FLAT": ["csst-msc-l1-qc0"],
}

# if obs_type is set
if args.obs_type:
    assert args.obs_type in DAG_MAP.keys(), f"Unknown obs_type: {args.obs_type}"
    DAG_MAP = {args.obs_type: DAG_MAP[args.obs_type]}

for obs_type, dags in DAG_MAP.items():
    print(f"* Processing `{obs_type}`")
    for this_dag in dags:
        print(f"  - Scheduling `{this_dag}` -> ", end="")
        dag = CsstDAGs().get(this_dag)
        msgs = dag.schedule(
            batch_id=args.batch_id,
            priority=args.priority,
            dataset=args.dataset,
            obs_type=obs_type,
            obs_group=args.obs_group,
            initial_prc_status=args.initial_prc_status,
            final_prc_status=args.final_prc_status,
            demo=args.demo,
        )
        print(f"{len(msgs)} tasks.")