"""Transient search, stage 2: vet the catalogue non-matches individually. Two questions have to be answered about every source that failed to match Gaia DR3 and SkyMapper DR2. Is it real? A source can be missing from the catalogues simply because it is not there - a cosmic ray pair that survived sigma clipping, a deblending artefact on the galaxy, a noise peak. The test used here is independence: demand that the source is detected on its own in at least six of the twelve 300 s subs. Nothing that is not on the sky can do that. Is it new? A transient is a source that is present tonight and absent from archival imagery. The catalogue non-match is weak evidence, because both Gaia and SkyMapper run out of depth around the magnitudes of interest here and neither is complete for extended or blended objects on a galaxy this bright. The strong evidence is a picture: this pulls a Digitized Sky Survey cutout at each position (DSS2 red, epoch ~1990s) through the CDS hips2fits service and puts it beside the master. Anything visible on a plate taken thirty years ago is not a transient. Writes mo-transient-vetted.csv, and caches the DSS cutouts for the figure. """ import csv import os import warnings import numpy as np from astropy import units as u from astropy.coordinates import SkyCoord from astropy.io import fits import mo_common as C NSUB_MIN = 6 CUT = 60 # px half-size of the postage stamps DSSDIR = os.path.join(C.OUT, "_dss") def load(): z = np.load(os.path.join(C.OUT, "_mo_transient.npz")) return z def dss_cutout(ra, dec, fov_arcmin=1.5, npix=120, survey="CDS/P/DSS2/red"): """Archival DSS2 red image of one position, cached to disk.""" tag = f"{ra:.5f}{dec:+.5f}_{survey.split('/')[-1]}.fits" path = os.path.join(DSSDIR, tag) os.makedirs(DSSDIR, exist_ok=True) if os.path.exists(path): with fits.open(path) as hd: return hd[0].data.astype(float) try: from astroquery.hips2fits import hips2fits with warnings.catch_warnings(): warnings.simplefilter("ignore") hdul = hips2fits.query( hips=survey, width=npix, height=npix, ra=ra * u.deg, dec=dec * u.deg, fov=(fov_arcmin / 60.0) * u.deg, projection="TAN", format="fits") data = np.asarray(hdul[0].data, dtype=float) fits.PrimaryHDU(data).writeto(path, overwrite=True) return data except Exception as exc: # noqa: BLE001 print(f" DSS fetch failed for {ra:.5f} {dec:+.5f}: " f"{type(exc).__name__}: {exc}") return None def nearest_catalogue(sc): """Nearest Gaia and SkyMapper source at ANY separation, plus SIMBAD/NED. The main search used fixed match radii. For a handful of finalists it is worth knowing what the nearest catalogued thing actually is and how far away it lies - a 4 arcsec offset from a SkyMapper source on a crowded galaxy usually means the same object, badly centroided. """ gz = np.load(os.path.join(C.OUT, "_gaia_deep.npz")) gcat = SkyCoord(gz["ra"] * u.deg, gz["dec"] * u.deg) sz = np.load(os.path.join(C.OUT, "_skymapper.npz")) scat = SkyCoord(sz["ra"] * u.deg, sz["dec"] * u.deg) gi, gd, _ = sc.match_to_catalog_sky(gcat) si, sd, _ = sc.match_to_catalog_sky(scat) return (gd.arcsec, gz["g"][gi], sd.arcsec, sz["g"][si], sz["r"][si], sz["cls"][si]) def ned_query(sc, radius_arcsec=15.0): """Anything NED knows about within a few arcsec of each position.""" out = [] try: from astroquery.ipac.ned import Ned except Exception: # noqa: BLE001 return ["NED unavailable"] * len(sc) for c in sc: try: with warnings.catch_warnings(): warnings.simplefilter("ignore") t = Ned.query_region(c, radius=radius_arcsec * u.arcsec) if t is None or len(t) == 0: out.append("") else: names = [f"{r['Object Name']}({r['Type']}," f"{r['Separation']:.1f}\")" for r in t[:3]] out.append("; ".join(names)) except Exception as exc: # noqa: BLE001 out.append(f"query failed: {type(exc).__name__}") return out def main(): z = load() ci = z["cand_idx"] nsub = z["nsub"] sel = np.where(nsub >= NSUB_MIN)[0] print(f"{len(ci)} sources matched neither Gaia DR3 nor SkyMapper DR2") print(f"{len(sel)} of them are independently detected in >= {NSUB_MIN} " f"of the 12 subs\n") from astropy.wcs import WCS hdr = fits.getheader(os.path.join(C.OUT, "master-Luminance.fit")) wcs = WCS(hdr) x = z["x"][ci][sel] y = z["y"][ci][sel] sc = wcs.pixel_to_world(x, y) gd, gg, sd, sg, sr, scls = nearest_catalogue(sc) ned = ned_query(sc) rows = [] for k, i in enumerate(sel): ra, dec = sc[k].ra.deg, sc[k].dec.deg dss = dss_cutout(ra, dec) # Is anything there on the archival plate? Compare the peak in the # central 6 arcsec against the frame-wide robust scatter. dss_sig = np.nan if dss is not None and np.isfinite(dss).any(): h = dss.shape[0] // 2 core = dss[h - 6:h + 7, h - 6:h + 7] med = np.nanmedian(dss) mad = np.nanmedian(np.abs(dss - med)) * 1.4826 if mad > 0: dss_sig = float((np.nanmax(core) - med) / mad) rows.append(dict( ra_deg=round(ra, 6), dec_deg=round(dec, 6), ra_hms=sc[k].ra.to_string(u.hour, sep=":", precision=2), dec_dms=sc[k].dec.to_string(u.deg, sep=":", precision=1, alwayssign=True), x=round(float(x[k]), 1), y=round(float(y[k]), 1), g_mag=round(float(z["mag"][ci][i]), 2), snr=round(float(z["snr"][ci][i]), 1), r50_over_psf=round(float(z["r50"][ci][i] / z["r50_star"]), 2), n_subs=int(nsub[i]), snr_R=round(float(z["snr_R"][i]), 1), snr_G=round(float(z["snr_G"][i]), 1), snr_B=round(float(z["snr_B"][i]), 1), dist_cenA_arcmin=round(float(z["dgal"][i]), 2), nearest_gaia_arcsec=round(float(gd[k]), 2), nearest_gaia_G=round(float(gg[k]), 2), nearest_smss_arcsec=round(float(sd[k]), 2), nearest_smss_g=round(float(sg[k]), 2), nearest_smss_classstar=round(float(scls[k]), 2), dss_peak_sigma=round(dss_sig, 1) if np.isfinite(dss_sig) else "", ned=ned[k])) r = rows[-1] print(f" {r['ra_hms']} {r['dec_dms']} G={r['g_mag']:5.2f} " f"SNR={r['snr']:6.1f} r50/psf={r['r50_over_psf']:.2f} " f"subs={r['n_subs']:2d}") print(f" nearest Gaia {r['nearest_gaia_arcsec']:6.2f}\" " f"(G={r['nearest_gaia_G']:.2f}), nearest SkyMapper " f"{r['nearest_smss_arcsec']:6.2f}\" (g={r['nearest_smss_g']:.2f}" f", ClassStar={r['nearest_smss_classstar']:.2f})") print(f" DSS2-red peak {r['dss_peak_sigma']} sigma, " f"{r['dist_cenA_arcmin']:.1f}' from Cen A centre") print(f" NED: {r['ned'] or '(nothing within 15\")'}") out = os.path.join(C.OUT, "mo-transient-vetted.csv") with open(out, "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=list(rows[0].keys())) w.writeheader() w.writerows(rows) print(f"\nwrote {out} ({len(rows)} rows)") np.save(os.path.join(C.OUT, "_mo_vetted.npy"), np.array(rows, dtype=object), allow_pickle=True) if __name__ == "__main__": main()