inspect.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 \
    --obs-id=10100232366 \
    --detector=09 \
    --prc-status=0
"""

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")
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)
# 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)
# reset prc_status
parser.add_argument("--reset", type=int, help="Reset prc_status", default=None)

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...")
    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,
    )
else:
    print("Inspecting level1 data...")
    plan_basis, data_basis = Dispatcher.find_plan_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,
        data_model=args.data_model,
        batch_id=args.batch_id,
    )
print(f"{len(plan_basis)} plan basis, {len(data_basis)} data basis found")
data_basis.remove_columns(["file_name", "_id"])

# zero / non-zero prc_status
u_prc_status, c_prc_status = np.unique(data_basis["prc_status"].data, return_counts=True)
t_prc_status = table.Table([u_prc_status, c_prc_status], names=["prc_status", "count"])
print("Prc status statistics:")
t_prc_status.pprint_all()

for stats_type, stats_keys in dict(
        obs_type=["dataset", "instrument", "obs_type", "prc_status"],
        obs_id=["dataset", "instrument", "obs_type", "obs_id", "prc_status"],
        detector=["dataset", "instrument", "obs_type", "detector", "prc_status"],
).items():
    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,)