astrophotography/pipeline/mo_shiftstack.py
laurence c6299f41ab Move the processing code under pipeline/
Preparing to merge this repository into a combined astrophotography
repo. session-scripts/ becomes pipeline/ because the scripts import
layout.py from their own directory and must stay together, and because
'pipeline' says what it is rather than how it came about. observing/
stays at the top level: observing plans are not processing code.
2026-07-21 17:13:54 +01:00

211 lines
9 KiB
Python

"""Where the one known minor planet was, and how deep digital tracking goes.
MPChecker and JPL's sb_ident agree that exactly one catalogued minor planet lay
within 30 arcmin of the field centre on this night: (427494) 2002 BK26, at
RA 13 27 00.6, Dec -43 11 26 at 09:22 UTC, V = 21.7, moving 53.2 arcsec/hr
toward position angle 71.7 deg.
The first thing this script does is check whether it was actually inside the
frame, which is not obvious from the offsets alone. The field is 42.9' x 28.6'
at position angle -89 deg - the long axis runs very nearly along declination,
not right ascension - so the RA half-width is only about 14.6 arcmin. The
asteroid is 17.0 arcmin east of centre. It was outside the frame, by roughly
2.4 arcmin, and there is therefore nothing known to recover.
The second thing is to measure how deep the search could have gone at that
rate, by shift-and-stack ("digital tracking"): each sub is shifted by the
target's own motion before combining, so a source moving at exactly that rate
adds coherently while the stars trail. Because the field position no longer
matters for the depth measurement, this is done on an empty patch of sky, and
a synthetic source of V = 21.7 is injected to show directly whether such an
object would have been recovered had it been in frame.
Output: _mo_shiftstack.npz
"""
import os
import numpy as np
import sep
from astropy import units as u
from astropy.coordinates import SkyCoord
from astropy.io import fits
from astropy.stats import sigma_clipped_stats
from astropy.wcs import WCS
from scipy.ndimage import shift as ndshift
from scipy.spatial import cKDTree
import mo_common as C
NAME = "(427494) 2002 BK26"
EPH_RA = 201.752708 # deg, 13 27 00.65, at 09:22 UTC
EPH_DEC = -43.190667 # deg, -43 11 26.4
EPH_UTC_HOURS = 9.0 + 22.0 / 60.0
DRA_COSDEC = 50.5 # arcsec/hr
DDEC = 16.7 # arcsec/hr
RATE = np.hypot(DRA_COSDEC, DDEC) # 53.2 arcsec/hr
VMAG = 21.7
HALF = 60 # px half-size of the extracted stamp
NX, NY = 4788, 3194
def utc_hours(path):
d = fits.getheader(path)["DATE-OBS"]
hh, mm, ss = d.split("T")[1].split(":")
return int(hh) + int(mm) / 60.0 + float(ss) / 3600.0 + 150.0 / 3600.0
def footprint_check(wcs):
tgt = SkyCoord(EPH_RA * u.deg, EPH_DEC * u.deg)
px, py = wcs.world_to_pixel(tgt)
cen = wcs.pixel_to_world(NX / 2, NY / 2)
dra, ddec = cen.spherical_offsets_to(tgt)
corners = [wcs.pixel_to_world(x, y) for x, y in
((0, 0), (NX - 1, 0), (0, NY - 1), (NX - 1, NY - 1))]
ras = [c.ra.deg for c in corners]
decs = [c.dec.deg for c in corners]
inside = (0 <= px < NX) and (0 <= py < NY)
print(f"{NAME}: V={VMAG}, {RATE:.1f}\"/hr at PA "
f"{np.degrees(np.arctan2(DRA_COSDEC, DDEC)):.1f} deg")
print(f" offset from field centre: {dra.to_value(u.arcmin):+.2f}' in RA, "
f"{ddec.to_value(u.arcmin):+.2f}' in Dec")
print(f" frame RA span {min(ras):.4f} to {max(ras):.4f} deg "
f"({(max(ras) - min(ras)) * np.cos(np.radians(EPH_DEC)) * 60:.1f}' "
f"on sky)")
print(f" frame Dec span {min(decs):.4f} to {max(decs):.4f} deg "
f"({(max(decs) - min(decs)) * 60:.1f}')")
print(f" predicted pixel ({px:.0f}, {py:.0f}) in a "
f"{NX} x {NY} frame -> "
f"{'INSIDE' if inside else 'OUTSIDE the frame'}")
return inside, float(px), float(py)
def blank_patch(mobjs, times, wcs, galx=2394.0, galy=1597.0, galrad=1000.0):
"""A patch of sky with no source near the whole track, off the galaxy.
Keeping clear of NGC 5128 itself matters more than it looks: the galaxy's
smooth light has a steep gradient that swamps the sky noise and would make
any depth measured on top of it meaningless.
"""
tree = cKDTree(np.column_stack([mobjs["x"], mobjs["y"]]))
# Track length in reference pixels over the sequence.
vx = (DRA_COSDEC / C.SCALE)
vy = (DDEC / C.SCALE)
span = times[-1]
rng = np.random.default_rng(7)
for _ in range(20000):
x0 = rng.uniform(400, NX - 400)
y0 = rng.uniform(400, NY - 400)
track = [(x0 + vx * t, y0 + vy * t) for t in times]
if not all(300 < x < NX - 300 and 300 < y < NY - 300
for x, y in track):
continue
if any(np.hypot(x - galx, y - galy) < galrad for x, y in track):
continue
d, _ = tree.query(np.array(track), distance_upper_bound=22.0)
if np.all(~np.isfinite(d)):
return x0, y0, vx, vy, span
raise RuntimeError("no empty patch found")
def stamp(img, x, y, half=HALF):
xi, yi = int(round(x)), int(round(y))
if not (half < xi < img.shape[1] - half and
half < yi < img.shape[0] - half):
return None
cut = img[yi - half:yi + half + 1, xi - half:xi + half + 1].astype(float)
return ndshift(cut, (yi - y, xi - x), order=3, mode="nearest")
def main():
hdr = fits.getheader(os.path.join(C.OUT, "master-Luminance.fit"))
wcs = WCS(hdr)
inside, epx, epy = footprint_check(wcs)
times = C.lum_times()
tforms = C.frame_transforms()
mobjs, map_, _, _, mimg = C.master_sources()
del mimg
x0, y0, vx, vy, span = blank_patch(mobjs, times, wcs)
print(f"\ndigital-tracking test patch: reference pixel "
f"({x0:.0f}, {y0:.0f}), track {np.hypot(vx, vy) * span:.0f} px "
f"over {span * 60:.0f} min")
tracked, fixed, zps = [], [], []
for i, (key, path) in enumerate(C.lum_frames()):
xr, yr = x0 + vx * times[i], y0 + vy * times[i]
inv = tforms[key].inverse
nx_, ny_ = inv(np.array([[xr, yr]]))[0]
fx_, fy_ = inv(np.array([[x0, y0]]))[0]
with fits.open(path, memmap=False) as hd:
img = hd[0].data.astype(np.float32)
objs, ap, _ = C.detect(img)
zp, _ = C.frame_zeropoint(objs["x"], objs["y"], ap, mobjs, map_,
tforms[key])
zps.append(zp)
# A source at V=21.7 moving at the asteroid's rate, injected into the
# raw sub, trailing across the 300 s exposure exactly as it would.
lin = np.linalg.inv(tforms[key].params[:2, :2])
dxy = lin @ np.array([vx * 300.0 / 3600.0, vy * 300.0 / 3600.0])
img2 = img.copy()
C.add_source(img2, nx_, ny_,
10 ** ((zp - VMAG) / 2.5) / C.APFRAC, dxy[0], dxy[1])
a = stamp(img2, nx_, ny_)
b = stamp(img, nx_, ny_) # same patch, no synthetic
del img, img2
tracked.append(a - np.median(a))
fixed.append(b - np.median(b))
print(f" {key}: zp {zp:.3f}")
def combine(cube):
m, _, _ = sigma_clipped_stats(np.array(cube), sigma=3.0, maxiters=2,
axis=0)
return np.asarray(m, dtype=np.float32)
def flatten(stack):
"""Remove any residual sky gradient before measuring noise."""
bkg = sep.Background(stack, bw=24, bh=24, fw=3, fh=3)
return stack - bkg.back()
with_src = flatten(combine(tracked))
empty = flatten(combine(fixed))
zp_eff = float(np.mean(zps))
_, _, sd = sigma_clipped_stats(empty, sigma=3.0)
noise_ap = sd * np.sqrt(np.pi * C.APRAD ** 2)
lim5 = -2.5 * np.log10(5.0 * noise_ap) + zp_eff
c = float(HALF)
f_src, _, _ = sep.sum_circle(with_src, np.array([c]), np.array([c]),
C.APRAD, err=float(sd), gain=1.0)
f_emp, _, _ = sep.sum_circle(empty, np.array([c]), np.array([c]),
C.APRAD, err=float(sd), gain=1.0)
snr_src = float(f_src[0]) / noise_ap
print(f"\nstacked 12 x 300 s along the track")
print(f" effective zero point {zp_eff:.3f}, pixel sigma {sd:.3f}")
print(f" 5 sigma point-source limit of the tracked stack: "
f"G = {lim5:.2f}")
# The untrailed limit above is optimistic: at 53"/hr the object smears
# 8.2 px within each 300 s sub, and a 5 px aperture cannot hold a streak
# that long. Scaling the injected source's recovered significance to
# 5 sigma gives the limit that actually applies at this rate.
lim5_trail = VMAG - 2.5 * np.log10(5.0 / max(snr_src, 1e-3))
print(f" injected V = {VMAG} source recovered at "
f"{snr_src:.2f} sigma "
f"({'DETECTED' if snr_src > 5 else 'NOT detectable'})")
print(f" -> 5 sigma limit for a source trailing at {RATE:.0f}\"/hr: "
f"G = {lim5_trail:.2f} (vs {lim5:.2f} for an untrailed source)")
print(f" same aperture on the un-injected stack: "
f"{float(f_emp[0]) / noise_ap:.2f} sigma (blank, as expected)")
print(f"\n{NAME} at V={VMAG} is {VMAG - lim5:+.2f} mag relative to that "
f"limit.")
np.savez(os.path.join(C.OUT, "_mo_shiftstack.npz"),
with_src=with_src, empty=empty, zp_eff=zp_eff, lim5=lim5,
sd=sd, snr_src=snr_src, lim5_trail=lim5_trail, vmag=VMAG, rate=RATE, name=NAME,
inside=inside, eph_px=epx, eph_py=epy, x0=x0, y0=y0,
vx=vx, vy=vy, times=times)
print("\nsaved _mo_shiftstack.npz")
if __name__ == "__main__":
main()