inspect_data.py

"""
Aim
---
Inspect a dataset.

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

python -m csst_dag.cli.inspect \
    --dataset=csst-msc-c9-25sqdeg-v3 \
    --instrument=MSC \
    --obs-type=WIDE \
    --obs-group=W2

25平方度宽场
python -m csst_dag.cli.inspect \
    --level=0 \
    --dataset=csst-msc-c9-25sqdeg-v3 \
    --instrument=MSC \
    --obs-type=WIDE

python -m csst_dag.cli.inspect \
    --level=0 \
    --dataset=csst-msc-c11-1000sqdeg-wide-test-v1 \
    --instrument=MSC \
    --obs-type=WIDE

python -m csst_dag.cli.inspect \
    --level=1 \
    --dataset=csst-msc-c11-1000sqdeg-wide-test-v1 \
    --instrument=MSC \
    --obs-type=WIDE \
    --data-model=csst-msc-l1-mbi \
    --batch-id=1000sqdeg-test-b1

CPIC:
python -m csst_dag.cli.inspect \
    --level=0 \
    --dataset=csst-cpic-c11-hip71681-v1 \
    --instrument=CPIC \
    --obs-type=SCI \
    --obs-group=hip71681

"""

import argparse

import numpy as np
from astropy import table

from csst_dag import Dispatcher, dfs

parser = argparse.ArgumentParser(
    description="Inspector for CSST datasets.",
    formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)

# data level
parser.add_argument("--level", type=int, help="Data level", default=0)
# level0 data parameters
parser.add_argument("--dataset", type=str, help="Dataset name", default="csst-msc-c9-25sqdeg-v3")
parser.add_argument("--instrument", type=str, help="Instrument name", default="MSC")
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="Prc status", default=None)
# level1 data parameters
parser.add_argument("--data-model", type=str, help="Data model", default=None)
parser.add_argument("--batch-id", type=str, help="Batch ID", default=None)
parser.add_argument("--build", type=str, help="Build number", default=None)
parser.add_argument("--pmapname", type=str, help="CCDS pmap name", default=None)
# reset prc_status
parser.add_argument("--reset", type=int, help="Reset prc_status", default=None)
# additional options
parser.add_argument("--verbose", action="store_true", help="Verbose mode", default=False)
args = parser.parse_args()

# from csst_dag import DotDict
#
# args = DotDict(
#     dataset="csst-msc-c9-25sqdeg-v3",
#     instrument="MSC",
#     obs_type="WIDE",
#     obs_group="W2",
#     obs_id=None,
#     detector=None,
#     prc_status=None,
# )

print("CLI parameters: ", args)

if args.level == 0:
    print("Inspecting level0 data...")
    data_basis = Dispatcher.find_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,
    )
else:
    print("Inspecting level1 data...")
    data_basis = Dispatcher.find_level1_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,
        data_model=args.data_model, # level1 specific
        batch_id=args.batch_id,
        build=args.build,
        pmapname=args.pmapname,
    )
print(f">>> {len(data_basis)} data basis found")
# data_basis.remove_columns(["file_name", "_id"])

stats_config = dict(
    prc_status=["dataset", "prc_status", "qc_status"],
    obs_type=["dataset", "instrument", "obs_type", "prc_status", "qc_status"],
    obs_group=["dataset", "instrument", "obs_type", "obs_group", "prc_status", "qc_status"],
    detector=["dataset", "instrument", "obs_type", "obs_group", "detector", "prc_status", "qc_status"],
    obs_id=["dataset", "instrument", "obs_type", "obs_group", "obs_id", "prc_status", "qc_status"],
)
if args.level == 1:
    for k in stats_config.keys():
        stats_config[k].extend(["data_model", "batch_id", "build"])

for stats_type, stats_keys in stats_config.items():
    if args.verbose:
        data_basis.pprint_all()
    print("\n>>> STATS DIM: ", stats_type)
    u_data, c_data = np.unique(data_basis[stats_keys], return_counts=True)
    u_table = table.Table(u_data)
    u_table.add_column(table.Column(c_data, name="count"))
    u_table.pprint_all()

if args.reset is not None:
    print("Resetting prc_status to ", args.reset)
    dfs.update_prc_status_by_ids(
        list(data_basis["_id"]),
        args.reset,
    )