Skip to content

lstautorta.crab_cleanup

Scan DL1 HDF5 files, estimate telescope pointing vs Crab in AltAz, and decide KEEP/DELETE per obs_id. Also collects associated logs and exports deletion/keep lists + CSV summary.

Log file is timestamped to avoid overwriting between runs.

Example : "python3 crab_cleanup.py --base-day-dir /fefs/onsite/pipeline/rta/data/2025/12/ --out-dir ./crab_cleanup_out --max-sep-deg 5.0 --tel-group tel_001 --step 200"

Functions:

Name Description
crab_separation_stats

Compute separation stats (deg) between telescope pointing and Crab in AltAz.
find_logs_for_dl1

Look for logs in // matching:
find_obs_root_from_dl1

Try to locate the directory path that corresponds to "...//" within dl1_path.
open_dl1_get_pointing_and_time

Extract a subsample of (time, alt_tel, az_tel) from DL1.
scan_dl1_files

Recursively scan all DL1 *.h5 under day_dir.

crab_separation_stats 

crab_separation_stats(dl1_path, site, crab, tel_group, step, max_sep_deg)

Compute separation stats (deg) between telescope pointing and Crab in AltAz. Default decision: median separation < max_sep_deg -> KEEP else DELETE.

Source code in src/lstautorta/crab_cleanup.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def crab_separation_stats(
    dl1_path: Path, site: EarthLocation, crab: SkyCoord, tel_group: str, step: int, max_sep_deg: float
):
    """
    Compute separation stats (deg) between telescope pointing and Crab in AltAz.
    Default decision: median separation < max_sep_deg -> KEEP else DELETE.
    """
    times_unix, alt_tel, az_tel = open_dl1_get_pointing_and_time(dl1_path, tel_group=tel_group, step=step)

    if len(times_unix) == 0:
        return {
            "sep_min_deg": np.nan,
            "sep_med_deg": np.nan,
            "sep_mean_deg": np.nan,
            "sep_max_deg": np.nan,
            "decision": "ERROR",
            "error": "no events",
        }

    t = Time(times_unix, format="unix", scale="utc")

    crab_altaz = crab.transform_to(AltAz(obstime=t, location=site))
    tel_altaz = SkyCoord(az=az_tel * u.rad, alt=alt_tel * u.rad, frame=AltAz(obstime=t, location=site))

    sep_deg = tel_altaz.separation(crab_altaz).to_value(u.deg)

    stats = {
        "sep_min_deg": float(np.min(sep_deg)),
        "sep_med_deg": float(np.median(sep_deg)),
        "sep_mean_deg": float(np.mean(sep_deg)),
        "sep_max_deg": float(np.max(sep_deg)),
        "error": "",
    }

    keep = stats["sep_med_deg"] < max_sep_deg
    stats["decision"] = "KEEP" if keep else "DELETE"
    return stats

find_logs_for_dl1 

find_logs_for_dl1(dl1_path, log_subdir)

Look for logs in // matching: .o- and .e-

Source code in src/lstautorta/crab_cleanup.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def find_logs_for_dl1(dl1_path: Path, log_subdir: Path):
    """
    Look for logs in <obs_root>/<log_subdir>/ matching:
      <dl1_prefix>.o-* and <dl1_prefix>.e-*
    """
    obs = extract_obs_id_from_name(dl1_path.name)
    if obs is None:
        return []

    obs_root = find_obs_root_from_dl1(dl1_path, obs)
    if obs_root is None:
        return []

    log_dir = obs_root / log_subdir
    if not log_dir.exists():
        return []

    prefix = dl1_path.stem
    logs = []
    logs.extend(sorted(log_dir.glob(f"{prefix}.o-*")))
    logs.extend(sorted(log_dir.glob(f"{prefix}.e-*")))
    return [p for p in logs]

find_obs_root_from_dl1 

find_obs_root_from_dl1(dl1_path, obs_id)

Try to locate the directory path that corresponds to "...//" within dl1_path.

Source code in src/lstautorta/crab_cleanup.py 
135
136
137
138
139
140
141
142
143
144
def find_obs_root_from_dl1(dl1_path: Path, obs_id: int) -> Path | None:
    """
    Try to locate the directory path that corresponds to ".../<obs_id>/" within dl1_path.
    """
    parts = dl1_path.parts
    obs_str = str(obs_id)
    if obs_str not in parts:
        return None
    i = parts.index(obs_str)
    return Path(*parts[: i + 1])

open_dl1_get_pointing_and_time 

open_dl1_get_pointing_and_time(dl1_path, tel_group, step)

Extract a subsample of (time, alt_tel, az_tel) from DL1.

Assumptions: - dl1/event/subarray/trigger/time = unix seconds (UTC) - dl1/event/telescope/parameters//alt_tel, az_tel = radians

Source code in src/lstautorta/crab_cleanup.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def open_dl1_get_pointing_and_time(dl1_path: Path, tel_group: str, step: int):
    """
    Extract a subsample of (time, alt_tel, az_tel) from DL1.

    Assumptions:
      - dl1/event/subarray/trigger/time = unix seconds (UTC)
      - dl1/event/telescope/parameters/<tel_group>/alt_tel, az_tel = radians
    """
    with h5py.File(dl1_path, "r") as f:
        trig = f["dl1/event/subarray/trigger"]
        pars = f[f"dl1/event/telescope/parameters/{tel_group}"]

        n = len(trig)
        if n == 0:
            return np.array([]), np.array([]), np.array([])

        idx = np.arange(0, n, step, dtype=int)

        times_unix = trig["time"][idx].astype(float)
        alt_rad = pars["alt_tel"][idx].astype(float)
        az_rad = pars["az_tel"][idx].astype(float)

    return times_unix, alt_rad, az_rad

scan_dl1_files 

scan_dl1_files(day_dir)

Recursively scan all DL1 *.h5 under day_dir. Returns dict: obs_id -> list[Path]

Source code in src/lstautorta/crab_cleanup.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def scan_dl1_files(day_dir: Path):
    """
    Recursively scan all DL1 *.h5 under day_dir.
    Returns dict: obs_id -> list[Path]
    """
    by_obs = {}
    for p in day_dir.rglob("*.h5"):
        if not is_dl1_file(p):
            continue
        obs = extract_obs_id_from_name(p.name)
        if obs is None:
            continue
        by_obs.setdefault(obs, []).append(p)

    for obs in by_obs:
        by_obs[obs] = sorted(by_obs[obs])
    return by_obs