feat: v0.2 — real FITS support, TAI/UTC fix, SkyBoT, two-pass bg

asteroidnet/candidate_classifier/classifier.py

@@ -0,0 +1,211 @@
+"""AsteroidNET Two-Stage Classifier (RF → CNN)."""
+from __future__ import annotations
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+from asteroidnet.tracklet_linker.linker import Tracklet
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class Classification:
+    tracklet: Tracklet
+    rf_score: float
+    cnn_score: float
+    is_asteroid: bool
+    priority: str    # 'ROUTINE' | 'HIGH' | 'HAZARDOUS'
+
+
+def classify_tracklets(
+    tracklets: list[Tracklet],
+    data_frames: list[np.ndarray],
+    config: Optional[dict] = None,
+) -> list[Classification]:
+    """
+    Two-stage classification: Random Forest (fast) → CNN (high precision).
+
+    RF stage uses kinematic features of the tracklet.
+    CNN stage uses 63×63 pixel cutouts from the detection positions.
+    """
+    cfg = (config or {}).get("classifier", {})
+    rf_thresh  = float(cfg.get("rf_threshold", 0.7))
+    cnn_thresh = float(cfg.get("cnn_threshold", 0.9))
+
+    rf_model  = _load_rf(cfg)
+    cnn_model = _load_cnn(cfg)
+
+    results: list[Classification] = []
+
+    for tracklet in tracklets:
+        # ── RF stage ─────────────────────────────────────────────────────
+        features = _extract_features(tracklet)
+        rf_score = _rf_predict(rf_model, features)
+        if rf_score < rf_thresh:
+            continue
+
+        # ── Satellite filter ──────────────────────────────────────────────
+        if _is_satellite(tracklet):
+            logger.debug("Satellite filter rejected tracklet (vel=%.3f, pa=%.1f)",
+                         tracklet.velocity_arcsec_s, tracklet.position_angle_deg)
+            continue
+
+        # ── CNN stage ─────────────────────────────────────────────────────
+        cutouts = _extract_cutouts(tracklet, data_frames)
+        cnn_score = _cnn_predict(cnn_model, cutouts)
+
+        is_asteroid = cnn_score >= cnn_thresh
+        if not is_asteroid:
+            continue
+
+        priority = _assign_priority(tracklet, cfg)
+        results.append(Classification(
+            tracklet=tracklet,
+            rf_score=rf_score,
+            cnn_score=cnn_score,
+            is_asteroid=True,
+            priority=priority,
+        ))
+
+    results.sort(key=lambda c: c.cnn_score, reverse=True)
+    logger.info("Classification: %d/%d tracklets confirmed", len(results), len(tracklets))
+    return results
+
+
+# ── Feature extraction ────────────────────────────────────────────────────────
+
+def _extract_features(t: Tracklet) -> np.ndarray:
+    """12-dimensional kinematic feature vector for RF classifier."""
+    snrs = [d["snr"] for d in t.detections]
+    mags = [d["mag"] for d in t.detections if d["mag"] < 90]
+    return np.array([
+        t.velocity_arcsec_s,
+        t.velocity_ra_arcsec_s,
+        t.velocity_dec_arcsec_s,
+        t.position_angle_deg / 360.0,
+        t.rms_residual_arcsec,
+        t.time_span_min,
+        len(t.detections),
+        float(np.mean(snrs)) if snrs else 0.0,
+        float(np.std(snrs))  if len(snrs) > 1 else 0.0,
+        float(np.mean(mags)) if mags else 25.0,
+        float(np.ptp(mags))  if len(mags) > 1 else 0.0,
+        len(set(t.frame_ids)),
+    ], dtype=np.float32)
+
+
+def _extract_cutouts(
+    t: Tracklet,
+    data_frames: list[np.ndarray],
+    size: int = 63,
+) -> Optional[np.ndarray]:
+    """Extract stacked cutouts from detection positions."""
+    if not data_frames:
+        return None
+    half = size // 2
+    cutouts = []
+    for det in t.detections:
+        fid = det.get("frame_id", 0)
+        if fid >= len(data_frames):
+            continue
+        data = data_frames[fid]
+        x, y = int(round(det.get("x", 0))), int(round(det.get("y", 0)))
+        h, w = data.shape
+        if x - half < 0 or y - half < 0 or x + half >= w or y + half >= h:
+            continue
+        cutout = data[y-half:y+half+1, x-half:x+half+1].copy()
+        if cutout.shape == (size, size):
+            finite = cutout[np.isfinite(cutout)]
+            if len(finite) > 0:
+                med = np.median(finite); mad = max(np.median(np.abs(finite-med)), 1e-10)
+                cutout = np.clip((cutout - med) / (3*mad), -3, 3)
+            cutouts.append(np.nan_to_num(cutout.astype(np.float32)))
+    return np.stack(cutouts) if cutouts else None
+
+
+# ── Model loading ─────────────────────────────────────────────────────────────
+
+def _load_rf(cfg: dict):
+    """Load RF model if available, else return None (heuristic fallback)."""
+    path = cfg.get("rf_model_path", "models/rf_classifier.pkl")
+    if Path(path).exists():
+        try:
+            import joblib
+            return joblib.load(path)
+        except Exception as exc:
+            logger.warning("Could not load RF model %s: %s", path, exc)
+    return None
+
+
+def _load_cnn(cfg: dict):
+    """Load CNN model if available, else return None (heuristic fallback)."""
+    path = cfg.get("cnn_model_path", "models/cnn_classifier.pth")
+    if Path(path).exists():
+        try:
+            import torch
+            model = torch.load(path, map_location="cpu")
+            model.eval()
+            return model
+        except Exception as exc:
+            logger.warning("Could not load CNN model %s: %s", path, exc)
+    return None
+
+
+def _rf_predict(model, features: np.ndarray) -> float:
+    """RF prediction — heuristic if model not trained yet."""
+    if model is not None:
+        try:
+            p = model.predict_proba(features.reshape(1, -1))[0, 1]
+            return float(p)
+        except Exception:
+            pass
+    # Heuristic: based on velocity, SNR, residual
+    vel  = features[0]
+    snr  = features[7]
+    rms  = features[4]
+    score = 0.3
+    if 0.01 <= vel <= 5.0:  score += 0.3
+    if snr >= 5.0:          score += 0.2
+    if rms <= 0.8:          score += 0.2
+    return min(score, 0.99)
+
+
+def _cnn_predict(model, cutouts: Optional[np.ndarray]) -> float:
+    """CNN prediction — heuristic if model not trained yet."""
+    if model is not None and cutouts is not None:
+        try:
+            import torch
+            x = torch.from_numpy(cutouts).unsqueeze(0).float()
+            with torch.no_grad():
+                out = model(x)
+                return float(torch.sigmoid(out).mean().item())
+        except Exception:
+            pass
+    # Heuristic: check if any cutout has a point source at center
+    if cutouts is not None and len(cutouts) > 0:
+        peaks = [float(np.max(c[28:35, 28:35])) if c.shape == (63, 63)
+                 else float(np.nanmax(c)) for c in cutouts]
+        return min(0.95, max(0.0, float(np.mean(peaks)) / 3.0 + 0.5))
+    return 0.5
+
+
+def _is_satellite(t: Tracklet) -> bool:
+    """Simple satellite/aircraft filter."""
+    if t.velocity_arcsec_s > 8.0:
+        return True
+    if t.rms_residual_arcsec < 0.01 and t.velocity_arcsec_s > 5.0:
+        return True
+    return False
+
+
+def _assign_priority(t: Tracklet, cfg: dict) -> str:
+    high_v = float(cfg.get("high_velocity_threshold", 1.0))
+    haz_v  = float(cfg.get("hazardous_velocity_threshold", 3.0))
+    if t.velocity_arcsec_s >= haz_v:
+        return "HAZARDOUS"
+    if t.velocity_arcsec_s >= high_v:
+        return "HIGH"
+    return "ROUTINE"

asteroidnet/catalog_matcher/matcher.py

@@ -0,0 +1,152 @@
+"""
+AsteroidNET Catalog Matcher (catalog_matcher.matcher)
+
+Removes known stars (Gaia DR3) and known asteroids (SkyBoT/JPL Horizons)
+from source catalogs, leaving only candidate moving objects.
+"""
+from __future__ import annotations
+import logging
+from typing import Optional
+
+import numpy as np
+from astropy.coordinates import SkyCoord, match_coordinates_sky
+from astropy.table import Table
+from astropy.time import Time
+import astropy.units as u
+
+logger = logging.getLogger(__name__)
+
+
+def remove_known_sources(
+    catalog: Table,
+    obs_time: Time,
+    config: Optional[dict] = None,
+) -> Table:
+    """
+    Remove known stars and known asteroids from a source catalog.
+
+    Steps:
+      1. Cross-match against Gaia DR3 with proper motion correction
+      2. Query SkyBoT for known solar system objects at this epoch
+      3. Return filtered catalog of candidate movers
+
+    Parameters
+    ----------
+    catalog : Table
+        Source catalog from extract_sources().
+    obs_time : Time
+        UTC mid-exposure time (used for PM correction and SkyBoT query).
+    config : dict, optional
+        Pipeline configuration.
+
+    Returns
+    -------
+    Table
+        Catalog with known sources removed; added column 'candidate_type'.
+    """
+    if len(catalog) == 0:
+        return catalog
+
+    cfg = (config or {}).get("matching", {})
+    star_r  = float(cfg.get("star_radius_arcsec", 2.0)) * u.arcsec
+    ast_r   = float(cfg.get("asteroid_radius_arcsec", 5.0)) * u.arcsec
+
+    coords = SkyCoord(ra=catalog["ra_deg"], dec=catalog["dec_deg"], unit=u.deg, frame="icrs")
+
+    # ── Step 1: Remove Gaia DR3 stars ───────────────────────────────────────
+    star_mask = _gaia_match_mask(coords, obs_time, star_r, cfg)
+
+    # ── Step 2: Remove known asteroids via SkyBoT ───────────────────────────
+    sso_mask = _skybot_match_mask(coords, obs_time, ast_r, config)
+
+    keep = ~(star_mask | sso_mask)
+    filtered = catalog[keep].copy()
+    filtered["candidate_type"] = "UNKNOWN_MOVER"
+
+    logger.info(
+        "Catalog: %d in → %d stars removed, %d SSOs removed → %d candidates",
+        len(catalog),
+        int(star_mask.sum()),
+        int(sso_mask.sum()),
+        len(filtered),
+    )
+    return filtered
+
+
+def _gaia_match_mask(
+    coords: SkyCoord,
+    obs_time: Time,
+    radius: u.Quantity,
+    cfg: dict,
+) -> np.ndarray:
+    """Return boolean mask: True = matched to a Gaia star."""
+    try:
+        from astroquery.gaia import Gaia
+        # Build bounding box for query
+        ra_ctr  = float(np.mean(coords.ra.deg))
+        dec_ctr = float(np.mean(coords.dec.deg))
+        ra_rng  = float(np.ptp(coords.ra.deg)) / 2 + 0.1
+        dec_rng = float(np.ptp(coords.dec.deg)) / 2 + 0.1
+
+        query = f"""
+        SELECT source_id, ra, dec, pmra, pmdec
+        FROM gaiadr3.gaia_source
+        WHERE ra BETWEEN {ra_ctr - ra_rng} AND {ra_ctr + ra_rng}
+          AND dec BETWEEN {dec_ctr - dec_rng} AND {dec_ctr + dec_rng}
+        """
+        job = Gaia.launch_job(query)
+        gaia = job.get_results()
+
+        if len(gaia) == 0:
+            return np.zeros(len(coords), dtype=bool)
+
+        # Apply proper motion correction to Gaia epoch
+        dt_yr = float((obs_time - Time(2016.0, format="jyear")).to("yr").value)
+        ra_corrected  = gaia["ra"]  + np.where(np.ma.is_masked(gaia["pmra"]),  0, gaia["pmra"])  / 3.6e6 * dt_yr
+        dec_corrected = gaia["dec"] + np.where(np.ma.is_masked(gaia["pmdec"]), 0, gaia["pmdec"]) / 3.6e6 * dt_yr
+
+        gaia_coords = SkyCoord(ra=ra_corrected, dec=dec_corrected, unit=u.deg, frame="icrs")
+        _, sep2d, _ = match_coordinates_sky(coords, gaia_coords)
+        return sep2d < radius
+
+    except Exception as exc:
+        logger.warning("Gaia match failed: %s — no star removal", exc)
+        return np.zeros(len(coords), dtype=bool)
+
+
+def _skybot_match_mask(
+    coords: SkyCoord,
+    obs_time: Time,
+    radius: u.Quantity,
+    config: Optional[dict],
+) -> np.ndarray:
+    """Return boolean mask: True = matched to a known SSO via SkyBoT."""
+    use_skybot = (config or {}).get("matching", {}).get("use_skybot", True)
+    if not use_skybot:
+        return np.zeros(len(coords), dtype=bool)
+
+    try:
+        from asteroidnet.data_access.skybot_client import (
+            query_skybot, skybot_table_to_skycoord
+        )
+
+        center = SkyCoord(
+            ra=float(np.mean(coords.ra.deg)),
+            dec=float(np.mean(coords.dec.deg)),
+            unit=u.deg, frame="icrs",
+        )
+        field_r = float(np.max(coords.separation(center).arcmin)) * u.arcmin + 5 * u.arcmin
+
+        observer = (config or {}).get("matching", {}).get("skybot_location", "500")
+        table = query_skybot(center, field_r, obs_time, observer=observer, config=config)
+        sso_coords = skybot_table_to_skycoord(table)
+
+        if sso_coords is None:
+            return np.zeros(len(coords), dtype=bool)
+
+        _, sep2d, _ = match_coordinates_sky(coords, sso_coords)
+        return sep2d < radius
+
+    except Exception as exc:
+        logger.warning("SkyBoT match failed: %s — no known-asteroid removal", exc)
+        return np.zeros(len(coords), dtype=bool)

asteroidnet/config/defaults.yaml

@@ -0,0 +1,129 @@
+# AsteroidNET v0.2 Configuration
+# All parameters overridable via ASTEROIDNET_SECTION__KEY env vars
+
+detection:
+  threshold_sigma: 3.0
+  bright_threshold_sigma: 5.0
+  fwhm_range: [2.0, 8.0]
+
+preprocessing:
+  background_box_size: 64
+  background_filter_size: 3
+  sigma_clip_sigma: 3.0
+  sigma_clip_maxiters: 10
+  source_mask_snr: 2.0        # SNR threshold for first-pass source masking
+  cosmic_ray_sigclip: 4.5
+  cosmic_ray_objlim: 5.0
+
+matching:
+  star_radius_arcsec: 2.0
+  star_2mass_radius_arcsec: 3.0
+  asteroid_radius_arcsec: 5.0
+  proper_motion_epoch_threshold: 1.0  # years
+  gaia_catalog: "gaiadr3"
+  skybot_location: "500"       # Default observer location for SkyBoT (geocenter)
+  use_skybot: true             # Use SkyBoT for known-object removal
+  use_local_cache: false
+
+tracking:
+  min_detections: 3
+  min_time_span_minutes: 30.0
+  velocity_range_arcsec_s: [0.01, 10.0]
+  max_acceleration_arcsec_s2: 0.001
+  position_tolerance_arcsec: 3.0
+  max_motion_residual_arcsec: 1.0
+
+classifier:
+  rf_threshold: 0.7
+  cnn_threshold: 0.9
+  rf_model_path: "models/rf_classifier.pkl"
+  cnn_model_path: "models/cnn_classifier.pth"
+  cnn_cutout_size: 64
+  use_gpu: true
+  satellite_rf_threshold: 0.8
+
+orbit:
+  method: "gauss"
+  min_arc_minutes: 30.0
+  observatory_location: null
+
+reporting:
+  observatory_code: null
+  observer_name: null
+  output_dir: "output"
+  alert_channels:
+    email: {enabled: false, smtp_host: "localhost", smtp_port: 587, recipients: []}
+    slack: {enabled: false, webhook_url: null}
+    voevent: {enabled: false, broker_url: null}
+    atel: {enabled: false}
+  priority_thresholds:
+    high_velocity_arcsec_s: 1.0
+    hazardous_delta_v_km_s: 10.0
+
+# ── NEW: Data access for real surveys ────────────────────────────────────────
+data_access:
+  ps1:
+    # Pan-STARRS Image Cutout Service (MAST/STScI)
+    filenames_url: "https://ps1images.stsci.edu/cgi-bin/ps1filenames.py"
+    cutout_url: "https://ps1images.stsci.edu/cgi-bin/fitscut.cgi"
+    default_size_pixels: 1200    # cutout size in pixels (1200px = 5' at 0.25"/px)
+    default_filters: "r"         # r-band warps most similar to IASC packages
+    image_type: "warp"           # "warp" = single epoch (required for moving objects)
+    timeout_s: 60
+    # CRITICAL: PS1 full skycell images missing TIMESYS=TAI and RADESYS=FK5
+    # The cutout service corrects headers — always use the cutout API
+    tai_utc_offset_s: 37.0       # leap seconds as of 2017 (PS1 DR2 era)
+
+  ztf:
+    # ZTF IRSA Image Browser Engine
+    ibe_search_url: "https://irsa.ipac.caltech.edu/ibe/search/ztf/products/sci"
+    ibe_data_url: "https://irsa.ipac.caltech.edu/ibe/data/ztf/products/sci"
+    diff_url: "https://irsa.ipac.caltech.edu/ibe/data/ztf/products/sci"
+    cutout_size_arcsec: 300.0    # field size for ZTF cutout requests
+    timeout_s: 90
+    # ZTF times are UTC; use OBSJD or OBSMJD + 0.5*EXPTIME for mid-exposure
+
+  mpc:
+    # MPC bulk data
+    mpcorb_url: "https://www.minorplanetcenter.net/iau/MPCORB/MPCORB.DAT.gz"
+    obs_url: "https://www.minorplanetcenter.net/iau/ECS/MPCAT-OBS/MPCAT-OBS.TXT.gz"
+
+  skybot:
+    # IMCCE SkyBoT cone-search service
+    url: "https://ssp.imcce.fr/webservices/skybot/api/conesearch.php"
+    default_field_radius_arcmin: 30.0
+    default_observer: "500"      # geocenter; use MPC obs code for topocentric
+
+  horizons:
+    # JPL Horizons via astroquery
+    default_quantities: "1,9,20,23,24"   # RA/Dec, V mag, rates
+
+# ── Training data builder ─────────────────────────────────────────────────────
+training:
+  output_dir: "training_data"
+  cutout_size_pixels: 63          # ALeRCE-compatible stamp size
+  negative_per_positive: 3        # negative examples per confirmed object
+  min_snr: 3.0
+  surveys: ["ps1", "ztf"]         # which archives to mine
+  date_range: ["2023-01-01", "2024-12-31"]
+
+database:
+  url: "postgresql://asteroidnet:asteroidnet@localhost:5432/asteroidnet"
+  pool_size: 10
+  echo: false
+
+celery:
+  broker_url: "redis://localhost:6379/0"
+  result_backend: "redis://localhost:6379/1"
+  task_serializer: "json"
+  worker_concurrency: 4
+
+api:
+  host: "0.0.0.0"
+  port: 8000
+  reload: false
+
+logging:
+  level: "INFO"
+  format: "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
+  file: null

asteroidnet/config/loader.py

@@ -0,0 +1,51 @@
+"""AsteroidNET configuration loader with env-var overrides."""
+from __future__ import annotations
+import hashlib, os, copy
+from pathlib import Path
+from typing import Any
+import yaml
+
+_DEFAULTS = Path(__file__).parent / "defaults.yaml"
+_ENV_PREFIX = "ASTEROIDNET_"
+
+
+def load_config(path: str | Path | None = None) -> dict:
+    with open(_DEFAULTS) as f:
+        cfg = yaml.safe_load(f)
+    if path:
+        with open(path) as f:
+            user = yaml.safe_load(f) or {}
+        cfg = _deep_merge(cfg, user)
+    cfg = _apply_env(cfg)
+    cfg["_hash"] = hashlib.sha256(
+        yaml.dump(cfg, sort_keys=True).encode()
+    ).hexdigest()[:12]
+    return cfg
+
+
+def _deep_merge(base: dict, override: dict) -> dict:
+    out = copy.deepcopy(base)
+    for k, v in override.items():
+        if k in out and isinstance(out[k], dict) and isinstance(v, dict):
+            out[k] = _deep_merge(out[k], v)
+        else:
+            out[k] = v
+    return out
+
+
+def _apply_env(cfg: dict, _prefix: str = _ENV_PREFIX) -> dict:
+    """Override config from ASTEROIDNET_SECTION__KEY=value env vars."""
+    out = copy.deepcopy(cfg)
+    for key, val in os.environ.items():
+        if not key.startswith(_prefix):
+            continue
+        parts = key[len(_prefix):].lower().split("__")
+        node = out
+        for part in parts[:-1]:
+            node = node.setdefault(part, {})
+        leaf = parts[-1]
+        try:
+            node[leaf] = yaml.safe_load(val)
+        except Exception:
+            node[leaf] = val
+    return out

asteroidnet/data_access/ps1_client.py

@@ -0,0 +1,287 @@
+"""
+AsteroidNET Pan-STARRS Data Access (data_access.ps1_client)
+
+Downloads single-epoch warp FITS images from Pan-STARRS DR2 via the
+PS1 Image Cutout Service at ps1images.stsci.edu.
+
+Key design decisions (from research):
+  - Always use WARP images (not stacks): stacks use asinh pseudo-luptitude
+    scaling — photometrically non-linear, unsuitable for source extraction
+  - Always use the cutout API: full skycell images are missing TIMESYS=TAI
+    and RADESYS=FK5; the fitscut.cgi service corrects these headers
+  - Always byte-swap after download: FITS big-endian arrays silently break
+    bottleneck-accelerated Background2D, producing wrong background estimates
+
+Reference:
+  https://outerspace.stsci.edu/display/PANSTARRS/PS1+Image+Cutout+Service
+  https://outerspace.stsci.edu/display/PANSTARRS/PS1+Warp+images
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+import requests
+from astropy.io import fits
+from astropy.table import Table
+from astropy.time import Time
+
+from asteroidnet.utils.time_utils import fix_ps1_header, parse_fits_time
+
+logger = logging.getLogger(__name__)
+
+# Default PS1 cutout service endpoints
+_FILENAMES_URL = "https://ps1images.stsci.edu/cgi-bin/ps1filenames.py"
+_CUTOUT_URL    = "https://ps1images.stsci.edu/cgi-bin/fitscut.cgi"
+
+# PS1 pixel scale: exactly 0.25 arcsec/pixel
+PS1_PIXEL_SCALE_ARCSEC = 0.25
+
+# IASC campaign packages: 4 frames per field, ~30 min cadence
+IASC_N_FRAMES = 4
+
+
+@dataclass
+class PS1WarpFrame:
+    """A single PS1 warp FITS frame, ready for the AsteroidNET pipeline."""
+    filename: str
+    data: np.ndarray         # float32, native byte-order, background NOT subtracted
+    header: fits.Header
+    obs_time: Time           # UTC, mid-exposure
+    exptime_s: float
+    filter_band: str
+    ra_center: float
+    dec_center: float
+    size_pixels: int
+    projection_id: str = ""  # PS1 skycell / projection cell ID
+
+
+def get_ps1_warp_list(
+    ra: float,
+    dec: float,
+    filter_band: str = "r",
+    config: Optional[dict] = None,
+) -> Table:
+    """
+    Query the PS1 filename service for all warp images at a sky position.
+
+    Pan-STARRS has approximately 12 warps per filter per position on average.
+    Each warp corresponds to one single-epoch exposure, making them suitable
+    for moving object detection (unlike stacked images).
+
+    Parameters
+    ----------
+    ra, dec : float
+        Sky position in decimal degrees (ICRS J2000).
+    filter_band : str
+        PS1 filter(s): 'g', 'r', 'i', 'z', 'y', or combination like 'gr'.
+    config : dict, optional
+        Pipeline configuration dict.
+
+    Returns
+    -------
+    Table
+        Astropy Table with columns: filename, type, projcell, subcell, filter, ...
+    """
+    url = (config or {}).get("data_access", {}).get("ps1", {}).get(
+        "filenames_url", _FILENAMES_URL)
+    timeout = (config or {}).get("data_access", {}).get("ps1", {}).get("timeout_s", 60)
+
+    params = {
+        "ra": f"{ra:.6f}",
+        "dec": f"{dec:.6f}",
+        "filters": filter_band,
+        "type": "warp",
+    }
+
+    logger.info("Querying PS1 warp list: RA=%.4f Dec=%.4f filter=%s", ra, dec, filter_band)
+    resp = requests.get(url, params=params, timeout=timeout)
+    resp.raise_for_status()
+
+    if not resp.text.strip():
+        logger.warning("No PS1 warps found at RA=%.4f Dec=%.4f", ra, dec)
+        return Table()
+
+    table = Table.read(io.StringIO(resp.text), format="ascii")
+    logger.info("Found %d PS1 warp images", len(table))
+    return table
+
+
+def download_ps1_cutout(
+    ra: float,
+    dec: float,
+    filename: str,
+    size_pixels: int = 1200,
+    config: Optional[dict] = None,
+) -> Optional[PS1WarpFrame]:
+    """
+    Download a FITS cutout from a specific PS1 warp image.
+
+    Parameters
+    ----------
+    ra, dec : float
+        Center of the cutout (decimal degrees).
+    filename : str
+        PS1 filename from get_ps1_warp_list() — e.g. 'rings.v3.skycell.1234.023.wrp.r.fits'
+    size_pixels : int
+        Cutout size in pixels. Default 1200 = 5 arcminutes at 0.25"/px.
+        IASC campaign packages cover ~20' fields; use 4800 for full coverage.
+    config : dict, optional
+        Pipeline configuration dict.
+
+    Returns
+    -------
+    PS1WarpFrame or None
+        Processed frame with corrected headers and native byte-order data.
+    """
+    url = (config or {}).get("data_access", {}).get("ps1", {}).get(
+        "cutout_url", _CUTOUT_URL)
+    timeout = (config or {}).get("data_access", {}).get("ps1", {}).get("timeout_s", 60)
+
+    params = {
+        "ra": f"{ra:.6f}",
+        "dec": f"{dec:.6f}",
+        "size": str(size_pixels),
+        "format": "fits",
+        "red": filename,
+    }
+
+    logger.info("Downloading PS1 cutout: %s (size=%dpx)", filename, size_pixels)
+    t0 = time.perf_counter()
+
+    try:
+        resp = requests.get(url, params=params, timeout=timeout, stream=True)
+        resp.raise_for_status()
+        fits_bytes = resp.content
+    except requests.exceptions.RequestException as exc:
+        logger.error("PS1 download failed for %s: %s", filename, exc)
+        return None
+
+    elapsed = time.perf_counter() - t0
+    logger.debug("Downloaded %.1f KB in %.2fs", len(fits_bytes) / 1024, elapsed)
+
+    try:
+        with fits.open(io.BytesIO(fits_bytes)) as hdul:
+            hdu = hdul[0]
+            header = hdu.header.copy()
+            raw_data = hdu.data
+
+        if raw_data is None:
+            logger.warning("PS1 cutout %s has no image data", filename)
+            return None
+
+        # ── CRITICAL FIX 1: byte-swap to native endianness ──────────────────
+        # FITS data is big-endian; bottleneck (used by Background2D) only
+        # accelerates native-endian arrays — non-native silently gives wrong results
+        data = np.array(raw_data, dtype=np.float32)  # float32, native byte-order
+
+        # ── CRITICAL FIX 2: add missing PS1 header keywords ────────────────
+        # The cutout service usually adds these, but apply defensively
+        fix_ps1_header(header)
+
+        # ── CRITICAL FIX 3: parse time with correct TAI scale ───────────────
+        obs_time = parse_fits_time(header, survey="ps1")
+
+        exptime_s = float(header.get("EXPTIME", 30.0))
+        filter_band = str(header.get("FILTER", "r")).strip()
+
+        frame = PS1WarpFrame(
+            filename=filename,
+            data=data,
+            header=header,
+            obs_time=obs_time,
+            exptime_s=exptime_s,
+            filter_band=filter_band,
+            ra_center=ra,
+            dec_center=dec,
+            size_pixels=size_pixels,
+        )
+        logger.info(
+            "PS1 frame OK: %s | %s UTC | filter=%s | shape=%s",
+            filename, obs_time.isot, filter_band, data.shape
+        )
+        return frame
+
+    except Exception as exc:
+        logger.error("Failed to parse PS1 FITS from %s: %s", filename, exc)
+        return None
+
+
+def get_iasc_style_sequence(
+    ra: float,
+    dec: float,
+    filter_band: str = "r",
+    n_frames: int = IASC_N_FRAMES,
+    size_pixels: int = 1200,
+    config: Optional[dict] = None,
+) -> list[PS1WarpFrame]:
+    """
+    Download a sequence of PS1 warp images mimicking an IASC campaign package.
+
+    Selects N frames from the available warps, preferring those with the
+    widest time coverage to maximise asteroid motion detectability.
+
+    Parameters
+    ----------
+    ra, dec : float
+        Field center in decimal degrees.
+    filter_band : str
+        PS1 filter (default 'r', most similar to IASC packages).
+    n_frames : int
+        Number of frames to return (IASC uses 4).
+    size_pixels : int
+        Cutout size in pixels.
+    config : dict, optional
+        Pipeline configuration dict.
+
+    Returns
+    -------
+    list[PS1WarpFrame]
+        Sequence of frames sorted by observation time, suitable for pipeline input.
+    """
+    warp_table = get_ps1_warp_list(ra, dec, filter_band, config)
+
+    if len(warp_table) == 0:
+        logger.warning("No PS1 warps available for RA=%.4f Dec=%.4f", ra, dec)
+        return []
+
+    # Select frames to maximise time span: take first, last, and middle ones
+    indices = _select_frame_indices(len(warp_table), n_frames)
+    selected = warp_table[indices]
+
+    logger.info("Downloading %d/%d PS1 warps (IASC-style sequence)", n_frames, len(warp_table))
+
+    frames: list[PS1WarpFrame] = []
+    for row in selected:
+        frame = download_ps1_cutout(ra, dec, row["filename"], size_pixels, config)
+        if frame is not None:
+            frames.append(frame)
+
+    # Sort by observation time
+    frames.sort(key=lambda f: f.obs_time.unix)
+
+    if len(frames) >= 2:
+        dt_min = (frames[-1].obs_time - frames[0].obs_time).to("min").value
+        logger.info(
+            "Sequence: %d frames, time span=%.1f min (target: ~30 min)",
+            len(frames), dt_min
+        )
+
+    return frames
+
+
+def _select_frame_indices(n_available: int, n_select: int) -> list[int]:
+    """Select n_select indices from n_available, spread to maximise time coverage."""
+    if n_available <= n_select:
+        return list(range(n_available))
+    # Always include first and last; fill middle evenly
+    if n_select == 1:
+        return [0]
+    step = (n_available - 1) / (n_select - 1)
+    return [round(i * step) for i in range(n_select)]

asteroidnet/data_access/skybot_client.py

@@ -0,0 +1,120 @@
+"""
+AsteroidNET SkyBoT Client (data_access.skybot_client)
+
+Queries the IMCCE SkyBoT service to find all known solar system objects
+in a given sky field at a given epoch. Used both for:
+  1. Removing known asteroids from candidate lists (catalog_matcher)
+  2. Labeling archival FITS frames for ML training (training.dataset_builder)
+
+SkyBoT covers ephemerides for all known SSOs from 1889 to 2060.
+
+Reference: https://ssp.imcce.fr/webservices/skybot/
+"""
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+import requests
+from astropy.coordinates import SkyCoord
+from astropy.table import Table
+from astropy.time import Time
+import astropy.units as u
+
+logger = logging.getLogger(__name__)
+
+_SKYBOT_URL = "https://ssp.imcce.fr/webservices/skybot/api/conesearch.php"
+
+
+def query_skybot(
+    center: SkyCoord,
+    radius: u.Quantity,
+    epoch: Time,
+    observer: str = "500",
+    config: Optional[dict] = None,
+) -> Table:
+    """
+    Cone-search the SkyBoT service for known solar system objects.
+
+    Parameters
+    ----------
+    center : SkyCoord
+        Field center.
+    radius : Quantity
+        Search radius (e.g. 30*u.arcmin).
+    epoch : Time
+        UTC observation epoch (use mid-exposure time).
+    observer : str
+        MPC observatory code. '500' = geocenter.
+        Use 'F51' for Pan-STARRS, '695' for Palomar.
+    config : dict, optional
+        Pipeline configuration dict.
+
+    Returns
+    -------
+    Table
+        Astropy Table with columns: Number, Name, RA(h), DE(deg),
+        Type, Mv, posunc, ErrRA, ErrDE, d, dRA, dDE, Rgeo, Rhel
+        Returns empty Table if no objects found or service unavailable.
+    """
+    url = (config or {}).get("data_access", {}).get("skybot", {}).get("url", _SKYBOT_URL)
+    observer = (config or {}).get("data_access", {}).get("skybot", {}).get(
+        "default_observer", observer)
+
+    # SkyBoT expects RA in degrees, Dec in degrees, epoch as JD (UTC)
+    ra_deg  = center.icrs.ra.deg
+    dec_deg = center.icrs.dec.deg
+    radius_deg = radius.to(u.deg).value
+    jd_utc = epoch.utc.jd
+
+    params = {
+        "EPOCH": f"{jd_utc:.6f}",
+        "-ra": f"{ra_deg:.6f}",
+        "-dec": f"{dec_deg:.6f}",
+        "-bd": f"{radius_deg:.4f}",
+        "-loc": observer,
+        "-mime": "votable",
+        "-filter": "0",          # 0 = all object types
+        "-refsys": "EQJ2000",
+    }
+
+    logger.info(
+        "SkyBoT query: RA=%.4f Dec=%.4f r=%.2f' epoch=%s obs=%s",
+        ra_deg, dec_deg, radius.to(u.arcmin).value, epoch.utc.isot, observer
+    )
+
+    try:
+        resp = requests.get(url, params=params, timeout=30)
+        resp.raise_for_status()
+
+        if "No solar system object" in resp.text or len(resp.text) < 100:
+            logger.debug("SkyBoT: no known SSOs in field")
+            return Table()
+
+        from astropy.io.votable import parse_single_table
+        import io
+        votable = parse_single_table(io.BytesIO(resp.content))
+        table = votable.to_table()
+        logger.info("SkyBoT: found %d known SSOs in field", len(table))
+        return table
+
+    except requests.exceptions.RequestException as exc:
+        logger.warning("SkyBoT query failed (network): %s — proceeding without known-object removal", exc)
+        return Table()
+    except Exception as exc:
+        logger.warning("SkyBoT parse error: %s — proceeding without known-object removal", exc)
+        return Table()
+
+
+def skybot_table_to_skycoord(table: Table) -> Optional[SkyCoord]:
+    """Convert SkyBoT result table to SkyCoord array for cross-matching."""
+    if len(table) == 0:
+        return None
+    ra_col  = "RA(h)" if "RA(h)" in table.colnames else "RA"
+    dec_col = "DE(deg)" if "DE(deg)" in table.colnames else "Dec"
+    return SkyCoord(
+        ra=table[ra_col],
+        dec=table[dec_col],
+        unit=(u.hourangle if "h" in ra_col else u.deg, u.deg),
+        frame="icrs",
+    )

asteroidnet/data_access/ztf_client.py

@@ -0,0 +1,212 @@
+"""
+AsteroidNET ZTF Data Access (data_access.ztf_client)
+
+Downloads multi-epoch science FITS images from the Zwicky Transient Facility
+via IRSA's Image Browser Engine (IBE).
+
+Key facts:
+  - ZTF pixel scale: 1.012 arcsec/pixel
+  - ZTF times: UTC (OBSJD, OBSMJD, DATE-OBS)
+  - 16 CCDs × 4 quadrants = 64 quadrant images per exposure
+  - Three filters: g (1), r (2), i (3)
+  - Science images: sciimg.fits; difference images: scimrefdiffimg.fits.fz
+  - Photometric ZP varies by quadrant — always use MAGZP header keyword
+
+Reference:
+  https://irsa.ipac.caltech.edu/docs/program_interface/ztf_api.html
+"""
+from __future__ import annotations
+
+import io, logging, time
+from dataclasses import dataclass
+from typing import Optional
+
+import numpy as np
+import requests
+from astropy.io import fits
+from astropy.time import Time
+
+from asteroidnet.utils.time_utils import parse_fits_time
+
+logger = logging.getLogger(__name__)
+
+_IBE_SEARCH = "https://irsa.ipac.caltech.edu/ibe/search/ztf/products/sci"
+_IBE_DATA   = "https://irsa.ipac.caltech.edu/ibe/data/ztf/products/sci"
+
+ZTF_PIXEL_SCALE_ARCSEC = 1.012
+ZTF_FILTERS = {"g": 1, "r": 2, "i": 3}
+
+
+@dataclass
+class ZTFFrame:
+    """A single ZTF science FITS frame ready for the pipeline."""
+    filename: str
+    data: np.ndarray     # float32, native byte-order
+    header: fits.Header
+    obs_time: Time       # UTC mid-exposure
+    exptime_s: float
+    filter_band: str
+    ra_center: float
+    dec_center: float
+    field_id: int
+    ccd_id: int
+    quad_id: int
+    mag_zp: float        # photometric zero point (varies per quadrant)
+    mag_zp_rms: float
+
+
+def search_ztf_images(
+    ra: float,
+    dec: float,
+    filter_band: str = "r",
+    size_deg: float = 0.08,
+    config: Optional[dict] = None,
+) -> list[dict]:
+    """
+    Search for ZTF science images covering a sky position.
+
+    Returns list of metadata dicts usable with download_ztf_cutout().
+    """
+    url = (config or {}).get("data_access", {}).get("ztf", {}).get(
+        "ibe_search_url", _IBE_SEARCH)
+    timeout = (config or {}).get("data_access", {}).get("ztf", {}).get("timeout_s", 90)
+
+    params = {
+        "POS": f"{ra},{dec}",
+        "SIZE": str(size_deg),
+        "ct": "csv",
+    }
+    if filter_band:
+        params["WHERE"] = f"filtercode='z{filter_band}'"
+
+    logger.info("Searching ZTF images: RA=%.4f Dec=%.4f filter=z%s", ra, dec, filter_band)
+    resp = requests.get(url, params=params, timeout=timeout)
+    resp.raise_for_status()
+
+    if not resp.text.strip() or resp.text.strip().startswith("Error"):
+        logger.warning("No ZTF images found at RA=%.4f Dec=%.4f", ra, dec)
+        return []
+
+    import csv
+    reader = csv.DictReader(io.StringIO(resp.text))
+    rows = list(reader)
+    logger.info("Found %d ZTF images", len(rows))
+    return rows
+
+
+def download_ztf_cutout(
+    ra: float,
+    dec: float,
+    metadata: dict,
+    size_arcsec: float = 300.0,
+    config: Optional[dict] = None,
+) -> Optional[ZTFFrame]:
+    """
+    Download a cutout from a ZTF science image using IBE.
+
+    Parameters
+    ----------
+    metadata : dict
+        Row from search_ztf_images() — contains filefracday, field, ccdid, qid, filtercode.
+    size_arcsec : float
+        Cutout size in arcseconds.
+    """
+    timeout = (config or {}).get("data_access", {}).get("ztf", {}).get("timeout_s", 90)
+    base = (config or {}).get("data_access", {}).get("ztf", {}).get("ibe_data_url", _IBE_DATA)
+
+    # Construct ZTF file path from metadata
+    try:
+        filefracday = metadata["filefracday"]
+        year = filefracday[:4]
+        monthday = filefracday[4:8]
+        fracday = filefracday[8:]
+        field = metadata.get("field", "").zfill(6)
+        filt = metadata.get("filtercode", "zr")
+        ccd = metadata.get("ccdid", "01").zfill(2)
+        q = metadata.get("qid", "1")
+        fname = f"ztf_{filefracday}_{field}_{filt}_c{ccd}_o_q{q}_sciimg.fits"
+        path = f"{year}/{monthday}/{fracday}/{fname}"
+    except KeyError as exc:
+        logger.error("ZTF metadata missing key: %s", exc)
+        return None
+
+    url = f"{base}/{path}"
+    params = {
+        "center": f"{ra},{dec}",
+        "size": f"{size_arcsec}arcsec",
+        "gzip": "false",
+    }
+
+    logger.info("Downloading ZTF cutout: %s", fname)
+    t0 = time.perf_counter()
+    try:
+        resp = requests.get(url, params=params, timeout=timeout, stream=True)
+        resp.raise_for_status()
+        fits_bytes = resp.content
+    except requests.exceptions.RequestException as exc:
+        logger.error("ZTF download failed for %s: %s", fname, exc)
+        return None
+
+    logger.debug("Downloaded %.1f KB in %.2fs", len(fits_bytes) / 1024, time.perf_counter() - t0)
+
+    try:
+        with fits.open(io.BytesIO(fits_bytes)) as hdul:
+            hdu = hdul[0]
+            header = hdu.header.copy()
+            raw_data = hdu.data
+
+        if raw_data is None:
+            return None
+
+        # Byte-swap to native endianness (FITS is big-endian; bottleneck needs native)
+        data = np.array(raw_data, dtype=np.float32)
+
+        obs_time = parse_fits_time(header, survey="ztf")
+        exptime_s = float(header.get("EXPTIME", 30.0))
+        filter_band = str(header.get("FILTERID", header.get("FILTER", "r")))
+        if filter_band.isdigit():
+            filter_band = {1: "g", 2: "r", 3: "i"}.get(int(filter_band), "r")
+
+        return ZTFFrame(
+            filename=fname,
+            data=data,
+            header=header,
+            obs_time=obs_time,
+            exptime_s=exptime_s,
+            filter_band=filter_band,
+            ra_center=ra,
+            dec_center=dec,
+            field_id=int(field),
+            ccd_id=int(ccd),
+            quad_id=int(q),
+            mag_zp=float(header.get("MAGZP", 26.0)),
+            mag_zp_rms=float(header.get("MAGZPRMS", 0.0)),
+        )
+
+    except Exception as exc:
+        logger.error("Failed to parse ZTF FITS %s: %s", fname, exc)
+        return None
+
+
+def get_ztf_sequence(
+    ra: float, dec: float,
+    filter_band: str = "r",
+    n_frames: int = 4,
+    size_arcsec: float = 300.0,
+    config: Optional[dict] = None,
+) -> list[ZTFFrame]:
+    """Download N ZTF frames covering a field, sorted by time."""
+    rows = search_ztf_images(ra, dec, filter_band, config=config)
+    if not rows:
+        return []
+
+    # Select spread across available epochs
+    from asteroidnet.data_access.ps1_client import _select_frame_indices
+    indices = _select_frame_indices(len(rows), n_frames)
+    frames = []
+    for i in indices:
+        f = download_ztf_cutout(ra, dec, rows[i], size_arcsec, config)
+        if f is not None:
+            frames.append(f)
+    frames.sort(key=lambda x: x.obs_time.unix)
+    return frames

asteroidnet/fits_ingestor/ingestor.py

@@ -0,0 +1,206 @@
+"""
+AsteroidNET FITS Ingestor (fits_ingestor.ingestor)
+
+Ingests FITS files from disk (IASC packages, PS1 downloads, ZTF downloads)
+and returns validated FITSFrame objects ready for the preprocessing stage.
+
+Critical real-data fixes applied here:
+  1. Byte-swap to native float32 — FITS big-endian silently breaks bottleneck
+  2. Survey auto-detection from headers
+  3. TAI/UTC time scale correct parsing (PS1 vs ZTF)
+  4. Mask application (DQ arrays, NaN pixels)
+  5. Header validation with graceful fallbacks for missing keywords
+"""
+from __future__ import annotations
+
+import logging
+import warnings
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+from astropy.io import fits
+from astropy.time import Time
+from astropy.wcs import WCS, FITSFixedWarning
+
+from asteroidnet.utils.time_utils import (
+    detect_survey_from_header,
+    fix_ps1_header,
+    parse_fits_time,
+)
+
+logger = logging.getLogger(__name__)
+
+MANDATORY_SOFT = ["DATE-OBS", "EXPTIME", "FILTER", "TELESCOP"]
+MANDATORY_HARD = []  # none — be maximally lenient with real data
+
+
+@dataclass
+class FITSFrame:
+    path: Path
+    data: np.ndarray          # float32, native byte-order, NaN where masked
+    header: fits.Header
+    wcs: WCS
+    obs_time: Time            # UTC mid-exposure
+    exptime_s: float
+    filter_band: str
+    telescope: str
+    instrument: str
+    survey: str               # 'ps1' | 'ztf' | 'generic'
+    pixel_scale_arcsec: float
+    warnings: list[str] = field(default_factory=list)
+
+
+def ingest_fits_file(
+    path: str | Path,
+    config: Optional[dict] = None,
+) -> FITSFrame:
+    """
+    Load and validate a FITS file, applying all survey-specific fixes.
+
+    Parameters
+    ----------
+    path : str or Path
+        Path to FITS file (may be .fits or .fits.fz for fpack-compressed).
+    config : dict, optional
+        Pipeline configuration.
+
+    Returns
+    -------
+    FITSFrame
+        Validated frame. Emits warnings for missing-but-recoverable keywords.
+
+    Raises
+    ------
+    ValueError
+        If the file cannot be read or has critically bad data.
+    """
+    path = Path(path)
+    if not path.exists():
+        raise FileNotFoundError(f"FITS file not found: {path}")
+
+    w_list: list[str] = []
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", FITSFixedWarning)
+        with fits.open(path, memmap=True) as hdul:
+            # Find the primary image extension
+            hdu = _find_image_hdu(hdul)
+            header = hdu.header.copy()
+            raw_data = hdu.data
+            mask_data = _find_mask_hdu(hdul)
+
+    if raw_data is None:
+        raise ValueError(f"No image data in {path}")
+
+    # ── FIX 1: byte-swap + float32 ──────────────────────────────────────────
+    data = np.array(raw_data, dtype=np.float32)
+
+    # ── FIX 2: apply mask (NaN masked pixels) ───────────────────────────────
+    if mask_data is not None:
+        data[mask_data.astype(bool)] = np.nan
+
+    # Replace Inf with NaN
+    data[~np.isfinite(data)] = np.nan
+
+    # ── FIX 3: survey detection and header corrections ───────────────────────
+    survey = detect_survey_from_header(header)
+    if survey == "ps1":
+        fix_ps1_header(header)
+
+    # ── FIX 4: WCS ───────────────────────────────────────────────────────────
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", FITSFixedWarning)
+        wcs = WCS(header, naxis=2)
+
+    # ── FIX 5: time parsing with correct scale ───────────────────────────────
+    try:
+        obs_time = parse_fits_time(header, survey=survey)
+    except (KeyError, ValueError) as exc:
+        w_list.append(f"Could not parse observation time: {exc}. Using J2000.0")
+        obs_time = Time("2000-01-01T00:00:00", scale="utc")
+
+    # ── Soft keyword validation ──────────────────────────────────────────────
+    for kw in MANDATORY_SOFT:
+        if kw not in header:
+            w_list.append(f"Missing optional keyword: {kw}")
+
+    exptime_s     = float(header.get("EXPTIME", 30.0))
+    filter_band   = str(header.get("FILTER", header.get("FILTERID", "?"))).strip()
+    telescope     = str(header.get("TELESCOP", "?")).strip()
+    instrument    = str(header.get("INSTRUME", "?")).strip()
+
+    # Estimate pixel scale from WCS if CDELT keywords missing
+    pixel_scale_arcsec = _estimate_pixel_scale(wcs, survey)
+
+    if w_list:
+        logger.warning("Ingested %s with %d warning(s): %s", path.name, len(w_list), "; ".join(w_list))
+    else:
+        logger.info(
+            "Ingested %s | %s | survey=%s | filter=%s | scale=%.3f\"/px",
+            path.name, obs_time.isot, survey, filter_band, pixel_scale_arcsec
+        )
+
+    return FITSFrame(
+        path=path,
+        data=data,
+        header=header,
+        wcs=wcs,
+        obs_time=obs_time,
+        exptime_s=exptime_s,
+        filter_band=filter_band,
+        telescope=telescope,
+        instrument=instrument,
+        survey=survey,
+        pixel_scale_arcsec=pixel_scale_arcsec,
+        warnings=w_list,
+    )
+
+
+def ingest_fits_sequence(
+    paths: list[str | Path],
+    config: Optional[dict] = None,
+) -> list[FITSFrame]:
+    """Ingest multiple FITS files, sort by observation time."""
+    frames = []
+    for p in paths:
+        try:
+            frames.append(ingest_fits_file(p, config))
+        except Exception as exc:
+            logger.error("Failed to ingest %s: %s", p, exc)
+    frames.sort(key=lambda f: f.obs_time.unix)
+    logger.info("Ingested %d/%d frames", len(frames), len(paths))
+    return frames
+
+
+def _find_image_hdu(hdul: fits.HDUList) -> fits.ImageHDU:
+    """Return the first HDU containing a 2D image array."""
+    for hdu in hdul:
+        if isinstance(hdu, (fits.PrimaryHDU, fits.ImageHDU, fits.CompImageHDU)):
+            if hdu.data is not None and hdu.data.ndim >= 2:
+                return hdu
+    return hdul[0]
+
+
+def _find_mask_hdu(hdul: fits.HDUList) -> Optional[np.ndarray]:
+    """Return mask array if a mask/DQ extension is present."""
+    for hdu in hdul:
+        extname = str(hdu.header.get("EXTNAME", "")).upper()
+        if extname in ("MASK", "DQ", "FLAGS", "BPM") and hdu.data is not None:
+            return hdu.data
+    return None
+
+
+def _estimate_pixel_scale(wcs: WCS, survey: str) -> float:
+    """Estimate pixel scale in arcsec/pixel from WCS or survey defaults."""
+    if survey == "ps1":
+        return 0.25
+    if survey == "ztf":
+        return 1.012
+    try:
+        from astropy.wcs.utils import proj_plane_pixel_scales
+        scales = proj_plane_pixel_scales(wcs)
+        return float(np.mean(scales) * 3600.0)
+    except Exception:
+        return 1.0

asteroidnet/image_preprocessor/preprocessor.py

@@ -0,0 +1,250 @@
+"""
+AsteroidNET Image Preprocessor (image_preprocessor.preprocessor)
+
+Two-pass background subtraction with source masking, cosmic-ray rejection,
+and multi-frame WCS alignment.
+
+Two-pass background is CRITICAL for real data:
+  - First pass gives rough background → build source mask
+  - Second pass with mask gives unbiased background (sources don't inflate estimate)
+  - This matters especially in crowded fields near galactic plane
+
+Byte-order note: data must already be float32 native (done in ingestor).
+"""
+from __future__ import annotations
+
+import logging
+import warnings
+from typing import Optional
+
+import numpy as np
+from astropy.io import fits
+from astropy.stats import SigmaClip
+from astropy.wcs import FITSFixedWarning
+
+from asteroidnet.fits_ingestor.ingestor import FITSFrame
+
+logger = logging.getLogger(__name__)
+
+
+def preprocess_frame(
+    frame: FITSFrame,
+    config: Optional[dict] = None,
+) -> tuple[np.ndarray, np.ndarray]:
+    """
+    Full preprocessing pipeline for a single frame.
+
+    Stages:
+      1. Cosmic-ray rejection (astroscrappy L.A.Cosmic)
+      2. Two-pass background subtraction with source masking
+      3. Returns (background-subtracted data, background RMS map)
+
+    Parameters
+    ----------
+    frame : FITSFrame
+        Ingested frame with float32 native data.
+    config : dict, optional
+        Pipeline configuration.
+
+    Returns
+    -------
+    data_sub : ndarray
+        Background-subtracted data (NaN where masked).
+    bkg_rms : ndarray
+        Per-pixel background RMS (for SNR threshold computation).
+    """
+    cfg = (config or {}).get("preprocessing", {})
+    data = frame.data.copy()
+
+    # ── Stage 1: Cosmic ray rejection ───────────────────────────────────────
+    data, cr_mask = _reject_cosmic_rays(data, cfg, frame.exptime_s)
+
+    # ── Stage 2: Two-pass background subtraction ─────────────────────────────
+    data_sub, bkg_rms = _subtract_background(data, cfg)
+
+    logger.debug(
+        "Preprocessed %s: CR_mask=%.3f%%, bkg_median=%.2f, bkg_rms_median=%.2f",
+        frame.path.name,
+        100 * cr_mask.sum() / cr_mask.size,
+        float(np.nanmedian(data_sub + bkg_rms)),  # approx background level
+        float(np.nanmedian(bkg_rms)),
+    )
+    return data_sub, bkg_rms
+
+
+def align_frames(
+    frames: list[FITSFrame],
+    data_list: list[np.ndarray],
+    config: Optional[dict] = None,
+) -> list[np.ndarray]:
+    """
+    Reproject all frames to the WCS of the first frame.
+
+    Uses reproject_adaptive with conserve_flux=True — the recommended
+    general-purpose algorithm that handles pixel scale differences.
+
+    Returns aligned data arrays (same WCS as frames[0]).
+    """
+    if len(frames) < 2:
+        return data_list
+
+    try:
+        from reproject import reproject_adaptive
+    except ImportError:
+        logger.warning("reproject not installed — skipping alignment")
+        return data_list
+
+    ref_header = frames[0].header
+    aligned = [data_list[0]]
+
+    for i, (frame, data) in enumerate(zip(frames[1:], data_list[1:]), 1):
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", FITSFixedWarning)
+            try:
+                reprojected, footprint = reproject_adaptive(
+                    (data, frame.header),
+                    ref_header,
+                    conserve_flux=True,
+                    kernel="gaussian",
+                )
+                # Mask pixels outside footprint
+                reprojected[footprint < 0.5] = np.nan
+                aligned.append(reprojected.astype(np.float32))
+                logger.debug("Aligned frame %d/%d to reference WCS", i, len(frames) - 1)
+            except Exception as exc:
+                logger.warning("Frame %d alignment failed: %s — using unaligned", i, exc)
+                aligned.append(data)
+
+    return aligned
+
+
+# ── Private helpers ──────────────────────────────────────────────────────────
+
+def _reject_cosmic_rays(
+    data: np.ndarray,
+    cfg: dict,
+    exptime_s: float,
+) -> tuple[np.ndarray, np.ndarray]:
+    """Apply L.A.Cosmic cosmic-ray rejection via astroscrappy."""
+    try:
+        import astroscrappy
+        sigclip = float(cfg.get("cosmic_ray_sigclip", 4.5))
+        objlim  = float(cfg.get("cosmic_ray_objlim", 5.0))
+
+        # Readnoise from config or typical survey default
+        readnoise = float(cfg.get("readnoise_e", 10.0))
+
+        # astroscrappy requires no NaN — replace with median
+        nan_mask  = ~np.isfinite(data)
+        fill_val  = float(np.nanmedian(data))
+        data_fill = np.where(nan_mask, fill_val, data).astype(np.float32)
+
+        cr_mask, cleaned = astroscrappy.detect_cosmics(
+            data_fill,
+            sigclip=sigclip,
+            sigfrac=0.3,
+            objlim=objlim,
+            readnoise=readnoise,
+            gain=1.0,
+            verbose=False,
+        )
+        # Restore original NaN positions
+        cleaned[nan_mask] = np.nan
+        cleaned = cleaned.astype(np.float32)
+        cr_mask |= nan_mask
+
+        n_cr = int(cr_mask.sum()) - int(nan_mask.sum())
+        if n_cr > 0:
+            logger.debug("Rejected %d cosmic rays", n_cr)
+        return cleaned, cr_mask
+
+    except ImportError:
+        logger.debug("astroscrappy not installed — skipping CR rejection")
+        nan_mask = ~np.isfinite(data)
+        return data, nan_mask
+
+
+def _subtract_background(
+    data: np.ndarray,
+    cfg: dict,
+) -> tuple[np.ndarray, np.ndarray]:
+    """
+    Two-pass sigma-clipped 2D background subtraction with source masking.
+
+    Pass 1: rough background → detect sources → build mask
+    Pass 2: re-estimate background with masked sources → final subtraction
+
+    The two-pass approach is critical for crowded fields: sources bias the
+    background estimate upward, causing under-subtraction and spurious detections.
+    """
+    try:
+        from photutils.background import Background2D, SExtractorBackground
+        from photutils.segmentation import detect_sources
+    except ImportError:
+        logger.warning("photutils not installed — using sigma-clipped median background")
+        from astropy.stats import sigma_clipped_stats
+        _, med, std = sigma_clipped_stats(data[np.isfinite(data)])
+        return (data - med).astype(np.float32), np.full_like(data, std)
+
+    box_size    = int(cfg.get("background_box_size", 64))
+    filter_size = int(cfg.get("background_filter_size", 3))
+    sigma       = float(cfg.get("sigma_clip_sigma", 3.0))
+    maxiters    = int(cfg.get("sigma_clip_maxiters", 10))
+    mask_snr    = float(cfg.get("source_mask_snr", 2.0))
+
+    sc = SigmaClip(sigma=sigma, maxiters=maxiters)
+    nan_mask = ~np.isfinite(data)
+
+    # Combine NaN mask with user mask
+    edge_mask = nan_mask.copy()
+
+    # ── Pass 1: rough background ──────────────────────────────────────────
+    try:
+        bkg1 = Background2D(
+            data,
+            box_size=box_size,
+            filter_size=filter_size,
+            sigma_clip=sc,
+            bkg_estimator=SExtractorBackground(),
+            mask=edge_mask,
+            fill_value=0.0,
+        )
+        rough_sub = data - bkg1.background
+
+        # Build source mask from first-pass subtraction
+        threshold1 = mask_snr * bkg1.background_rms
+        source_mask = np.zeros_like(data, dtype=bool)
+        try:
+            segm = detect_sources(rough_sub, threshold1, npixels=5)
+            if segm is not None:
+                source_mask = segm.data > 0
+        except Exception:
+            pass
+        combined_mask = edge_mask | source_mask
+
+        # ── Pass 2: refined background with source mask ──────────────────
+        bkg2 = Background2D(
+            data,
+            box_size=box_size,
+            filter_size=filter_size,
+            sigma_clip=sc,
+            bkg_estimator=SExtractorBackground(),
+            mask=combined_mask,
+            fill_value=0.0,
+        )
+        data_sub = (data - bkg2.background).astype(np.float32)
+        data_sub[nan_mask] = np.nan
+        bkg_rms = bkg2.background_rms.astype(np.float32)
+
+        logger.debug(
+            "Two-pass background: src_mask=%.2f%%, bkg_rms_median=%.3f",
+            100 * source_mask.sum() / source_mask.size,
+            float(np.nanmedian(bkg_rms)),
+        )
+        return data_sub, bkg_rms
+
+    except Exception as exc:
+        logger.warning("Background2D failed (%s) — falling back to constant", exc)
+        from astropy.stats import sigma_clipped_stats
+        _, med, std = sigma_clipped_stats(data[np.isfinite(data)])
+        return (data - med).astype(np.float32), np.full_like(data, std, dtype=np.float32)

asteroidnet/orbit_determination/gauss_method.py

@@ -0,0 +1,120 @@
+"""AsteroidNET Orbit Determination (orbit_determination.gauss_method)."""
+from __future__ import annotations
+import logging, math
+from dataclasses import dataclass
+from typing import Optional
+
+import numpy as np
+from astropy.time import Time
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class OrbitalElements:
+    a_au: float
+    e: float
+    i_deg: float
+    omega_deg: float
+    Omega_deg: float
+    M_deg: float
+    epoch: Time
+    sigma_a: float = 0.0
+    sigma_e: float = 0.0
+    sigma_i: float = 0.0
+    arc_hours: float = 0.0
+    method: str = "gauss"
+    object_class: str = "UNKNOWN"
+
+
+def determine_orbit(
+    tracklet,
+    observatory_location: Optional[str] = None,
+    config: Optional[dict] = None,
+) -> Optional[OrbitalElements]:
+    """Attempt orbit determination for a confirmed tracklet."""
+    dets = tracklet.detections
+    if len(dets) < 3:
+        return _analytical_estimate(tracklet)
+
+    try:
+        from astroquery.jplhorizons import Horizons
+        result = _gauss_sbpy(tracklet, observatory_location)
+        if result is not None:
+            return result
+    except ImportError:
+        pass
+
+    return _analytical_estimate(tracklet)
+
+
+def _analytical_estimate(tracklet) -> OrbitalElements:
+    """Fast analytical estimate from velocity for single-night arcs."""
+    vel = tracklet.velocity_arcsec_s
+    arc_h = tracklet.time_span_min / 60.0
+
+    # Very rough: v ~ sqrt(GM/r) / delta; assume delta ~ 1 AU
+    # For 1 AU target, typical vel ~1-3 arcsec/s → a ~ 1-4 AU
+    if vel > 0.01:
+        a_est = max(0.5, min(6.0, 2.5 / (vel**0.3)))
+    else:
+        a_est = 3.0
+
+    e_est = 0.1 + 0.1 * (vel > 2.0)
+    i_est = 5.0 + 10.0 * (vel > 1.0)
+
+    # Uncertainty scales inversely with arc length
+    scale = max(1.0, 2.0 / max(arc_h, 0.1))
+    obj_class = _classify(a_est, e_est)
+
+    return OrbitalElements(
+        a_au=round(a_est, 4),
+        e=round(e_est, 4),
+        i_deg=round(i_est, 2),
+        omega_deg=0.0,
+        Omega_deg=0.0,
+        M_deg=0.0,
+        epoch=Time.now(),
+        sigma_a=round(0.05 * scale, 4),
+        sigma_e=round(0.05 * scale, 4),
+        sigma_i=round(2.0 * scale, 2),
+        arc_hours=arc_h,
+        method="analytical_estimate",
+        object_class=obj_class,
+    )
+
+
+def _gauss_sbpy(tracklet, obs_location: Optional[str]) -> Optional[OrbitalElements]:
+    """Full Gauss method via sbpy (requires ≥3 astrometric positions)."""
+    try:
+        import sbpy.data as sbd
+        from sbpy.orbit import Orbit
+        dets = sorted(tracklet.detections, key=lambda d: d["time_unix"])
+        # Build observation table
+        times = [Time(d["time_unix"], format="unix") for d in dets[:3]]
+        ras   = [d["ra"]  for d in dets[:3]]
+        decs  = [d["dec"] for d in dets[:3]]
+        from astropy.coordinates import SkyCoord
+        import astropy.units as u
+        coords = SkyCoord(ra=ras, dec=decs, unit=u.deg)
+        # sbpy Gauss method — simplified call
+        # Full implementation requires topocentric correction and observer ephemeris
+        logger.debug("sbpy Gauss method not fully implemented — using analytical estimate")
+        return None
+    except Exception as exc:
+        logger.debug("sbpy Gauss failed: %s", exc)
+        return None
+
+
+def _classify(a: float, e: float) -> str:
+    if e >= 1.0:
+        return "HYPERBOLIC"
+    if a < 1.3:
+        return "NEO_APOLLO_ATEN_AMOR"
+    if a < 2.0:
+        return "MARS_CROSSER"
+    if a < 3.2:
+        return "MAIN_BELT"
+    if a < 5.2:
+        return "OUTER_BELT"
+    return "TROJAN_TNO"

asteroidnet/pipeline/runner.py

@@ -0,0 +1,132 @@
+"""AsteroidNET Pipeline Runner — orchestrates all 6 stages."""
+from __future__ import annotations
+import logging, time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+
+from asteroidnet.config.loader import load_config
+from asteroidnet.fits_ingestor.ingestor import ingest_fits_sequence, FITSFrame
+from asteroidnet.image_preprocessor.preprocessor import preprocess_frame, align_frames
+from asteroidnet.source_extractor.detector import extract_sources
+from asteroidnet.catalog_matcher.matcher import remove_known_sources
+from asteroidnet.tracklet_linker.linker import link_tracklets
+from asteroidnet.candidate_classifier.classifier import classify_tracklets
+from asteroidnet.orbit_determination.gauss_method import determine_orbit
+from asteroidnet.reporting.mpc_formatter import format_tracklet_records
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class PipelineResult:
+    run_id: str
+    n_frames: int
+    n_candidates: int
+    n_confirmed: int
+    classifications: list
+    mpc_records: list[str]
+    elapsed_s: float
+    config_hash: str
+    warnings: list[str] = field(default_factory=list)
+
+
+def run_pipeline(
+    fits_paths: list[str | Path],
+    config_path: Optional[str | Path] = None,
+    observatory_code: Optional[str] = None,
+) -> PipelineResult:
+    """
+    Run the complete AsteroidNET detection pipeline on a set of FITS files.
+
+    Parameters
+    ----------
+    fits_paths : list
+        Paths to FITS files (4 frames recommended, as in IASC packages).
+    config_path : str or Path, optional
+        Path to YAML config file (uses defaults if None).
+    observatory_code : str, optional
+        MPC 3-char observatory code for report generation.
+
+    Returns
+    -------
+    PipelineResult
+        Complete results including classifications and MPC records.
+    """
+    t_start = time.perf_counter()
+    cfg = load_config(config_path)
+    obs_code = observatory_code or cfg.get("reporting", {}).get("observatory_code", "???")
+
+    import hashlib, uuid
+    run_id = str(uuid.uuid4())[:8]
+    logger.info("=== AsteroidNET Pipeline Run %s ===", run_id)
+
+    # ── Stage 1: Ingest FITS ─────────────────────────────────────────────────
+    logger.info("Stage 1: Ingesting %d FITS frames", len(fits_paths))
+    frames: list[FITSFrame] = ingest_fits_sequence(fits_paths, cfg)
+    if len(frames) < 2:
+        return PipelineResult(run_id, len(frames), 0, 0, [], [],
+                              time.perf_counter() - t_start, cfg["_hash"],
+                              ["Need ≥2 frames"])
+
+    # ── Stage 2: Preprocess ─────────────────────────────────────────────────
+    logger.info("Stage 2: Preprocessing %d frames", len(frames))
+    data_subs = []
+    bkg_rmss  = []
+    for f in frames:
+        ds, br = preprocess_frame(f, cfg)
+        data_subs.append(ds)
+        bkg_rmss.append(br)
+
+    # Align frames to common WCS
+    data_aligned = align_frames(frames, data_subs, cfg)
+
+    # ── Stage 3: Source extraction ───────────────────────────────────────────
+    logger.info("Stage 3: Extracting sources from %d frames", len(frames))
+    catalogs = []
+    for frame, data, rms in zip(frames, data_aligned, bkg_rmss):
+        cat = extract_sources(data, rms, frame.wcs, cfg)
+        catalogs.append(cat)
+
+    # ── Stage 4: Catalog matching (remove stars + known SSOs) ────────────────
+    logger.info("Stage 4: Removing known sources from %d catalogs", len(catalogs))
+    filtered_catalogs = []
+    for frame, cat in zip(frames, catalogs):
+        filtered = remove_known_sources(cat, frame.obs_time, cfg)
+        filtered_catalogs.append(filtered)
+
+    # ── Stage 5: Tracklet linking ────────────────────────────────────────────
+    logger.info("Stage 5: Linking tracklets across %d frames", len(frames))
+    times = [f.obs_time for f in frames]
+    tracklets = link_tracklets(filtered_catalogs, times, cfg)
+
+    n_candidates = len(tracklets)
+
+    # ── Stage 6: Classification ──────────────────────────────────────────────
+    logger.info("Stage 6: Classifying %d tracklet candidates", n_candidates)
+    classifications = classify_tracklets(tracklets, data_aligned, cfg)
+    n_confirmed = len(classifications)
+
+    # ── Reporting ────────────────────────────────────────────────────────────
+    mpc_records = []
+    for cl in classifications:
+        records = format_tracklet_records(cl.tracklet, obs_code)
+        mpc_records.extend(records)
+
+    elapsed = time.perf_counter() - t_start
+    logger.info(
+        "=== Run %s complete: %d frames → %d candidates → %d confirmed "
+        "(%.2fs) ===",
+        run_id, len(frames), n_candidates, n_confirmed, elapsed
+    )
+
+    return PipelineResult(
+        run_id=run_id,
+        n_frames=len(frames),
+        n_candidates=n_candidates,
+        n_confirmed=n_confirmed,
+        classifications=classifications,
+        mpc_records=mpc_records,
+        elapsed_s=elapsed,
+        config_hash=cfg["_hash"],
+    )

asteroidnet/reporting/mpc_formatter.py

@@ -0,0 +1,102 @@
+"""AsteroidNET MPC Formatter — exact 80-column astrometric records."""
+from __future__ import annotations
+import logging
+from astropy.coordinates import SkyCoord
+from astropy.time import Time
+import astropy.units as u
+
+logger = logging.getLogger(__name__)
+
+
+def format_mpc_record(
+    designation: str,
+    ra_deg: float,
+    dec_deg: float,
+    obs_time: Time,
+    magnitude: float,
+    filter_band: str,
+    observatory_code: str,
+) -> str:
+    """
+    Generate an exact 80-column MPC optical astrometry record.
+
+    Reference: https://www.minorplanetcenter.net/iau/info/OpticalObs.html
+    """
+    observatory_code = observatory_code.strip()
+    if len(observatory_code) != 3:
+        raise ValueError(f"Observatory code must be exactly 3 chars, got: {repr(observatory_code)}")
+
+    coord = SkyCoord(ra=ra_deg * u.deg, dec=dec_deg * u.deg)
+
+    ra_h = int(coord.ra.hms.h)
+    ra_m = int(coord.ra.hms.m)
+    ra_s = coord.ra.hms.s
+    ra_str = f"{ra_h:02d} {ra_m:02d} {ra_s:05.2f}"
+
+    sign = "+" if coord.dec.deg >= 0 else "-"
+    d = abs(coord.dec.deg)
+    dd = int(d)
+    dm = int((d - dd) * 60)
+    ds = ((d - dd) * 60 - dm) * 60
+    dec_str = f"{sign}{dd:02d} {dm:02d} {ds:04.1f}"
+
+    # Date: YYYY MM DD.ddddd
+    t = obs_time.utc
+    frac_day = t.mjd - int(t.mjd)
+    year = int(t.strftime("%Y"))
+    month = int(t.strftime("%m"))
+    day = int(t.strftime("%d"))
+    date_str = f"{year:04d} {month:02d} {day + frac_day:08.5f}"
+
+    try:
+        mag_str = f"{float(magnitude):4.1f}"
+    except Exception:
+        mag_str = "    "
+
+    band = (filter_band.strip()[0] if filter_band.strip() else "R")
+
+    # Build 80-byte buffer
+    buf = bytearray(b" " * 80)
+
+    def place(s: str, start: int, width: int) -> None:
+        enc = s[:width].ljust(width).encode("ascii", errors="replace")
+        buf[start:start + width] = enc
+
+    place(str(designation)[:5].ljust(5), 0, 5)
+    place("C", 8, 1)                  # Note 2: CCD observation
+    place(date_str[:8], 9, 8)         # Date: YYYY MM
+    place(date_str[8:16], 17, 8)      # Date: DD.ddddd
+    place(ra_str, 26, 11)
+    place(dec_str, 37, 11)
+    place(mag_str, 56, 4)
+    place(band, 61, 1)
+    place(observatory_code.rjust(3), 77, 3)
+
+    line = buf.decode("ascii")
+    assert len(line) == 80, f"MPC record length {len(line)} ≠ 80"
+    return line
+
+
+def format_tracklet_records(
+    tracklet,
+    observatory_code: str,
+    designation: str = "     ",
+) -> list[str]:
+    """Generate one MPC record per detection in a tracklet."""
+    lines = []
+    for det in tracklet.detections:
+        t = Time(det["time_unix"], format="unix")
+        try:
+            line = format_mpc_record(
+                designation=designation,
+                ra_deg=det["ra"],
+                dec_deg=det["dec"],
+                obs_time=t,
+                magnitude=det.get("mag", 99.0),
+                filter_band=det.get("filter", "R"),
+                observatory_code=observatory_code,
+            )
+            lines.append(line)
+        except Exception as exc:
+            logger.warning("Could not format MPC record for detection: %s", exc)
+    return lines

asteroidnet/source_extractor/detector.py

@@ -0,0 +1,178 @@
+"""
+AsteroidNET Source Extractor (source_extractor.detector)
+
+Two-pass source detection: bright pass at 5σ to build PSF model,
+faint pass at 3σ for asteroid candidates.
+"""
+from __future__ import annotations
+import logging
+from typing import Optional
+
+import numpy as np
+from astropy.table import Table
+from astropy.wcs import WCS
+import astropy.units as u
+
+logger = logging.getLogger(__name__)
+
+_CATALOG_COLS = [
+    "x_pixel", "y_pixel", "ra_deg", "dec_deg",
+    "flux_adu", "flux_err", "mag", "snr",
+    "fwhm", "roundness", "sharpness",
+]
+
+
+def extract_sources(
+    data_sub: np.ndarray,
+    bkg_rms: np.ndarray,
+    wcs: WCS,
+    config: Optional[dict] = None,
+) -> Table:
+    """
+    Detect sources in a background-subtracted image and return a catalog.
+
+    Parameters
+    ----------
+    data_sub : ndarray
+        Background-subtracted image (float32, NaN where masked).
+    bkg_rms : ndarray
+        Per-pixel background RMS (for threshold computation).
+    wcs : WCS
+        Frame WCS for pixel→sky coordinate conversion.
+    config : dict, optional
+        Pipeline configuration.
+
+    Returns
+    -------
+    Table
+        Source catalog with standardized columns.
+    """
+    cfg  = (config or {}).get("detection", {})
+    thresh_hi = float(cfg.get("bright_threshold_sigma", 5.0))
+    thresh_lo = float(cfg.get("threshold_sigma", 3.0))
+    fwhm_lo, fwhm_hi = cfg.get("fwhm_range", [2.0, 8.0])
+
+    rms_median = float(np.nanmedian(bkg_rms))
+    if rms_median <= 0:
+        rms_median = float(np.nanstd(data_sub[np.isfinite(data_sub)])) or 1.0
+
+    try:
+        from photutils.detection import DAOStarFinder
+        from photutils.aperture import CircularAperture, aperture_photometry
+
+        # Estimate typical FWHM from bright sources
+        fwhm_est = _estimate_fwhm(data_sub, bkg_rms, thresh_hi, fwhm_lo, fwhm_hi)
+
+        # Low-threshold pass for asteroid candidates
+        finder = DAOStarFinder(
+            fwhm=fwhm_est,
+            threshold=thresh_lo * rms_median,
+            sharplo=0.2, sharphi=1.0,
+            roundlo=-1.0, roundhi=1.0,
+            exclude_border=True,
+        )
+        # Replace NaN with 0 for detection only
+        data_clean = np.nan_to_num(data_sub, nan=0.0)
+        sources = finder(data_clean)
+
+        if sources is None or len(sources) == 0:
+            logger.info("No sources detected (threshold=%.1fσ)", thresh_lo)
+            return _empty_catalog()
+
+        # Aperture photometry
+        positions = np.column_stack([sources["xcentroid"], sources["ycentroid"]])
+        ap = CircularAperture(positions, r=fwhm_est)
+        phot = aperture_photometry(data_sub, ap, error=bkg_rms)
+
+        flux = np.array(phot["aperture_sum"], dtype=float)
+        flux_err = np.array(phot["aperture_sum_err"], dtype=float) if "aperture_sum_err" in phot.colnames else np.full(len(flux), rms_median * np.sqrt(np.pi * fwhm_est**2))
+        flux = np.maximum(flux, 0.0)
+
+        snr = np.where(flux_err > 0, flux / flux_err, 0.0)
+
+        # Sky coordinates via WCS
+        sky = wcs.pixel_to_world(sources["xcentroid"], sources["ycentroid"])
+        ra  = np.atleast_1d(sky.icrs.ra.deg)
+        dec = np.atleast_1d(sky.icrs.dec.deg)
+
+        # Magnitude (relative, no ZP needed for detection)
+        with np.errstate(invalid="ignore", divide="ignore"):
+            mag = np.where(flux > 0, -2.5 * np.log10(flux), 99.0)
+
+        catalog = Table({
+            "x_pixel":   np.array(sources["xcentroid"]),
+            "y_pixel":   np.array(sources["ycentroid"]),
+            "ra_deg":    ra,
+            "dec_deg":   dec,
+            "flux_adu":  flux,
+            "flux_err":  flux_err,
+            "mag":       mag,
+            "snr":       snr,
+            "fwhm":      np.full(len(flux), fwhm_est),
+            "roundness": np.array(sources["roundness1"]),
+            "sharpness": np.array(sources["sharpness"]),
+        })
+
+        logger.info("Extracted %d sources (fwhm=%.2fpx, threshold=%.1fσ)",
+                    len(catalog), fwhm_est, thresh_lo)
+        return catalog
+
+    except ImportError:
+        logger.warning("photutils not available — falling back to sigma-clip peak finder")
+        return _fallback_extract(data_sub, bkg_rms, wcs, thresh_lo)
+
+
+def _estimate_fwhm(data, bkg_rms, thresh, lo, hi):
+    """Estimate FWHM from bright sources, clamped to [lo, hi]."""
+    try:
+        from photutils.detection import DAOStarFinder
+        rms = float(np.nanmedian(bkg_rms))
+        finder = DAOStarFinder(fwhm=3.5, threshold=thresh * rms,
+                               sharplo=0.3, sharphi=0.9,
+                               roundlo=-0.5, roundhi=0.5,
+                               exclude_border=True)
+        data_clean = np.nan_to_num(data, nan=0.0)
+        sources = finder(data_clean)
+        if sources and len(sources) > 5:
+            fwhm = float(np.median(sources["fwhm"]))
+            return float(np.clip(fwhm, lo, hi))
+    except Exception:
+        pass
+    return 3.5  # default
+
+
+def _empty_catalog() -> Table:
+    return Table({c: [] for c in _CATALOG_COLS})
+
+
+def _fallback_extract(data, bkg_rms, wcs, thresh):
+    """Simple connected-component fallback when photutils absent."""
+    try:
+        from scipy.ndimage import label, center_of_mass
+        rms = float(np.nanmedian(bkg_rms))
+        binary = np.nan_to_num(data) > thresh * rms
+        labeled, n = label(binary)
+        if n == 0:
+            return _empty_catalog()
+        indices = list(range(1, n + 1))
+        coms = center_of_mass(data, labeled, indices)
+        xs = np.array([c[1] for c in coms])
+        ys = np.array([c[0] for c in coms])
+        sky = wcs.pixel_to_world(xs, ys)
+        ra  = np.atleast_1d(sky.icrs.ra.deg)
+        dec = np.atleast_1d(sky.icrs.dec.deg)
+        flux = np.array([float(np.nansum(data[labeled == i])) for i in indices])
+        flux = np.maximum(flux, 0.0)
+        return Table({
+            "x_pixel": xs, "y_pixel": ys,
+            "ra_deg": ra, "dec_deg": dec,
+            "flux_adu": flux, "flux_err": np.full(n, rms),
+            "mag": np.where(flux > 0, -2.5 * np.log10(np.maximum(flux, 1e-10)), 99.0),
+            "snr": flux / rms,
+            "fwhm": np.full(n, 3.5),
+            "roundness": np.zeros(n),
+            "sharpness": np.zeros(n),
+        })
+    except Exception as exc:
+        logger.error("Fallback extraction failed: %s", exc)
+        return _empty_catalog()

asteroidnet/tracklet_linker/linker.py

@@ -0,0 +1,251 @@
+"""
+AsteroidNET Tracklet Linker (tracklet_linker.linker)
+
+Links detections across frames into tracklets using:
+  1. Hough-transform pair generation (velocity-direction clustering)
+  2. KD-tree spatial extension to ≥3 frames
+  3. Kinematic validation (velocity, residuals, time span)
+"""
+from __future__ import annotations
+import logging, math
+from dataclasses import dataclass, field
+from typing import Optional
+
+import numpy as np
+from astropy.table import Table
+from astropy.time import Time
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class Tracklet:
+    detections: list[dict]        # sorted by time
+    velocity_ra_arcsec_s: float
+    velocity_dec_arcsec_s: float
+    velocity_arcsec_s: float
+    position_angle_deg: float
+    rms_residual_arcsec: float
+    time_span_min: float
+    frame_ids: list[int]
+    score: float = 0.0
+
+
+def link_tracklets(
+    catalogs: list[Table],
+    times: list[Time],
+    config: Optional[dict] = None,
+) -> list[Tracklet]:
+    """
+    Link multi-frame detections into moving-object tracklets.
+
+    Parameters
+    ----------
+    catalogs : list[Table]
+        One source catalog per frame (from catalog_matcher output).
+    times : list[Time]
+        Observation times (UTC, mid-exposure) for each frame.
+
+    Returns
+    -------
+    list[Tracklet]
+        Validated tracklets sorted by score (descending).
+    """
+    if len(catalogs) < 2:
+        logger.warning("Need ≥2 frames for tracklet linking")
+        return []
+
+    cfg = (config or {}).get("tracking", {})
+    min_dets   = int(cfg.get("min_detections", 3))
+    vel_min    = float(cfg.get("velocity_range_arcsec_s", [0.01, 10.0])[0])
+    vel_max    = float(cfg.get("velocity_range_arcsec_s", [0.01, 10.0])[1])
+    pos_tol    = float(cfg.get("position_tolerance_arcsec", 3.0))
+    max_resid  = float(cfg.get("max_motion_residual_arcsec", 1.0))
+    min_tspan  = float(cfg.get("min_time_span_minutes", 30.0))
+
+    # Convert catalogs to detection lists with time info
+    frame_dets = []
+    for i, (cat, t) in enumerate(zip(catalogs, times)):
+        dets = []
+        for row in cat:
+            dets.append({
+                "frame_id": i,
+                "ra":  float(row["ra_deg"]),
+                "dec": float(row["dec_deg"]),
+                "x":   float(row["x_pixel"]),
+                "y":   float(row["y_pixel"]),
+                "snr": float(row["snr"]),
+                "mag": float(row["mag"]),
+                "time_unix": t.unix,
+            })
+        frame_dets.append(dets)
+
+    # Generate seed pairs between frame 0 and frame 1
+    seeds = _generate_pairs(frame_dets[0], frame_dets[1], vel_min, vel_max)
+    logger.debug("Generated %d seed pairs from frames 0→1", len(seeds))
+
+    # Extend each seed to remaining frames
+    tracklets: list[Tracklet] = []
+    for seed in seeds:
+        extended = _extend_tracklet(seed, frame_dets[2:], pos_tol)
+        if len(extended) < min_dets:
+            continue
+
+        t = _build_tracklet(extended)
+        if t is None:
+            continue
+
+        # Validate
+        if t.rms_residual_arcsec > max_resid:
+            continue
+        if t.time_span_min < min_tspan:
+            continue
+        if not (vel_min <= t.velocity_arcsec_s <= vel_max):
+            continue
+
+        tracklets.append(t)
+
+    # Deduplicate overlapping tracklets (keep highest-score)
+    tracklets = _deduplicate(tracklets)
+    tracklets.sort(key=lambda x: x.score, reverse=True)
+
+    logger.info("Linked %d valid tracklets from %d frames", len(tracklets), len(catalogs))
+    return tracklets
+
+
+# ── Private helpers ──────────────────────────────────────────────────────────
+
+def _generate_pairs(
+    dets0: list[dict],
+    dets1: list[dict],
+    vel_min: float,
+    vel_max: float,
+) -> list[list[dict]]:
+    """Generate all pairs between two frames consistent with velocity bounds."""
+    pairs = []
+    if not dets0 or not dets1:
+        return pairs
+
+    for d0 in dets0:
+        for d1 in dets1:
+            dt = abs(d1["time_unix"] - d0["time_unix"])
+            if dt < 1.0:
+                continue
+            dra  = (d1["ra"]  - d0["ra"]) * 3600.0 * math.cos(math.radians(d0["dec"]))
+            ddec = (d1["dec"] - d0["dec"]) * 3600.0
+            dist = math.sqrt(dra**2 + ddec**2)
+            vel  = dist / dt
+            if vel_min <= vel <= vel_max:
+                pairs.append([d0, d1])
+    return pairs
+
+
+def _extend_tracklet(
+    seed: list[dict],
+    remaining_frames: list[list[dict]],
+    pos_tol_arcsec: float,
+) -> list[dict]:
+    """Extend a seed pair to additional frames using linear prediction."""
+    if len(seed) < 2:
+        return seed
+
+    extended = list(seed)
+
+    for frame_dets in remaining_frames:
+        if not frame_dets:
+            continue
+        # Predict position via linear extrapolation
+        t0, t1 = extended[-2]["time_unix"], extended[-1]["time_unix"]
+        dt = t1 - t0
+        if abs(dt) < 1.0:
+            continue
+        vra  = (extended[-1]["ra"]  - extended[-2]["ra"])  / dt
+        vdec = (extended[-1]["dec"] - extended[-2]["dec"]) / dt
+
+        t_pred = frame_dets[0]["time_unix"]
+        ra_pred  = extended[-1]["ra"]  + vra  * (t_pred - t1)
+        dec_pred = extended[-1]["dec"] + vdec * (t_pred - t1)
+
+        # Find closest detection within tolerance
+        best_sep = pos_tol_arcsec / 3600.0
+        best_det = None
+        for d in frame_dets:
+            sep = math.sqrt(
+                ((d["ra"]  - ra_pred)  * math.cos(math.radians(dec_pred)))**2
+                + (d["dec"] - dec_pred)**2
+            )
+            if sep < best_sep:
+                best_sep = sep
+                best_det = d
+
+        if best_det is not None:
+            extended.append(best_det)
+
+    return extended
+
+
+def _build_tracklet(dets: list[dict]) -> Optional[Tracklet]:
+    """Fit a linear motion model and compute tracklet statistics."""
+    if len(dets) < 2:
+        return None
+
+    dets = sorted(dets, key=lambda d: d["time_unix"])
+    times = np.array([d["time_unix"] for d in dets])
+    ras   = np.array([d["ra"]  for d in dets])
+    decs  = np.array([d["dec"] for d in dets])
+    cos_dec = math.cos(math.radians(float(np.mean(decs))))
+
+    # Linear fit
+    dt = times - times[0]
+    try:
+        pra  = np.polyfit(dt, ras, 1)
+        pdec = np.polyfit(dt, decs, 1)
+    except Exception:
+        return None
+
+    ra_fit  = np.polyval(pra, dt)
+    dec_fit = np.polyval(pdec, dt)
+
+    res_arcsec = np.sqrt(
+        ((ras - ra_fit) * cos_dec)**2 + (decs - dec_fit)**2
+    ) * 3600.0
+    rms = float(np.sqrt(np.mean(res_arcsec**2)))
+
+    vel_ra_s  = float(pra[0]) * cos_dec * 3600.0   # arcsec/s
+    vel_dec_s = float(pdec[0]) * 3600.0             # arcsec/s
+    vel       = math.sqrt(vel_ra_s**2 + vel_dec_s**2)
+    pa        = math.degrees(math.atan2(vel_ra_s, vel_dec_s)) % 360.0
+    tspan_min = float(times[-1] - times[0]) / 60.0
+
+    # Score: higher SNR, longer arc, lower residuals
+    mean_snr = float(np.mean([d["snr"] for d in dets]))
+    score = mean_snr * len(dets) * tspan_min / max(rms, 0.1)
+
+    return Tracklet(
+        detections=dets,
+        velocity_ra_arcsec_s=vel_ra_s,
+        velocity_dec_arcsec_s=vel_dec_s,
+        velocity_arcsec_s=vel,
+        position_angle_deg=pa,
+        rms_residual_arcsec=rms,
+        time_span_min=tspan_min,
+        frame_ids=[d["frame_id"] for d in dets],
+        score=score,
+    )
+
+
+def _deduplicate(tracklets: list[Tracklet]) -> list[Tracklet]:
+    """Remove tracklets sharing >50% of detections with a higher-score one."""
+    kept: list[Tracklet] = []
+    sorted_t = sorted(tracklets, key=lambda t: t.score, reverse=True)
+    used_positions: set[tuple] = set()
+
+    for t in sorted_t:
+        positions = {(d["frame_id"], round(d["ra"], 5), round(d["dec"], 5))
+                     for d in t.detections}
+        overlap = len(positions & used_positions)
+        if overlap <= len(positions) * 0.5:
+            kept.append(t)
+            used_positions |= positions
+
+    return kept

asteroidnet/training/dataset_builder.py

@@ -0,0 +1,349 @@
+"""
+AsteroidNET Training Dataset Builder (training.dataset_builder)
+
+Mines Pan-STARRS and ZTF archives to create labeled asteroid detection
+training data using confirmed MPC objects as ground truth.
+
+Methodology (from research):
+  1. Select sky fields and time ranges
+  2. Query SkyBoT to find all known SSOs in each field at each epoch
+  3. Query JPL Horizons for high-precision positions of found SSOs
+  4. Download FITS frames covering those fields
+  5. Project ephemeris positions → pixel coordinates via WCS
+  6. Extract 63×63 px cutouts centered on confirmed positions (positives)
+  7. Extract random cutouts away from known objects (negatives)
+  8. Save as .npz training dataset
+
+This approach mirrors how ALeRCE built their stamp classifier training set
+from ZTF alerts cross-matched with the MPC.
+"""
+from __future__ import annotations
+
+import logging
+import hashlib
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+from astropy.coordinates import SkyCoord
+from astropy.time import Time
+import astropy.units as u
+
+logger = logging.getLogger(__name__)
+
+CUTOUT_SIZE = 63       # ALeRCE-compatible stamp size
+NEG_PER_POS = 3       # negative examples per positive
+
+
+@dataclass
+class TrainingExample:
+    cutout: np.ndarray          # shape (CUTOUT_SIZE, CUTOUT_SIZE)
+    label: int                  # 1 = asteroid, 0 = non-asteroid
+    snr: float
+    velocity_arcsec_s: float    # 0 for negatives
+    source_survey: str
+    field_ra: float
+    field_dec: float
+    epoch: str
+
+
+def build_training_dataset(
+    sky_fields: list[tuple[float, float]],
+    date_range: tuple[str, str],
+    output_path: str | Path,
+    surveys: list[str] = ("ps1", "ztf"),
+    neg_per_pos: int = NEG_PER_POS,
+    config: Optional[dict] = None,
+) -> Path:
+    """
+    Build a labeled training dataset from archival survey data.
+
+    Parameters
+    ----------
+    sky_fields : list of (ra, dec) tuples in degrees
+        Sky fields to mine. Random high-galactic-latitude positions work well.
+    date_range : (start_iso, end_iso)
+        Date range for archival frames (e.g. ('2023-01-01', '2024-06-01')).
+    output_path : Path
+        Where to save the .npz dataset.
+    surveys : list
+        Which surveys to use: 'ps1', 'ztf'.
+    neg_per_pos : int
+        Number of negative examples to generate per positive.
+    config : dict, optional
+        Pipeline configuration.
+
+    Returns
+    -------
+    Path
+        Path to the saved .npz dataset.
+    """
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    examples: list[TrainingExample] = []
+
+    for ra, dec in sky_fields:
+        for survey in surveys:
+            try:
+                field_examples = _mine_field(
+                    ra, dec, survey, date_range, neg_per_pos, config
+                )
+                examples.extend(field_examples)
+                logger.info(
+                    "Field (%.2f, %.2f) survey=%s: +%d examples (total %d)",
+                    ra, dec, survey, len(field_examples), len(examples)
+                )
+            except Exception as exc:
+                logger.warning("Failed to mine field (%.2f, %.2f) %s: %s", ra, dec, survey, exc)
+
+    if not examples:
+        logger.error("No training examples collected")
+        return output_path
+
+    # Save as compressed numpy archive
+    cutouts  = np.stack([e.cutout for e in examples])
+    labels   = np.array([e.label for e in examples], dtype=np.int8)
+    snrs     = np.array([e.snr   for e in examples], dtype=np.float32)
+    vels     = np.array([e.velocity_arcsec_s for e in examples], dtype=np.float32)
+    surveys_ = np.array([e.source_survey for e in examples])
+
+    np.savez_compressed(
+        output_path,
+        cutouts=cutouts,
+        labels=labels,
+        snr=snrs,
+        velocity=vels,
+        survey=surveys_,
+    )
+
+    n_pos = int(labels.sum())
+    n_neg = len(labels) - n_pos
+    logger.info(
+        "Dataset saved: %d examples (%d positive, %d negative) → %s",
+        len(examples), n_pos, n_neg, output_path
+    )
+    return output_path
+
+
+def _mine_field(
+    ra: float,
+    dec: float,
+    survey: str,
+    date_range: tuple[str, str],
+    neg_per_pos: int,
+    config: Optional[dict],
+) -> list[TrainingExample]:
+    """Mine a single sky field for labeled training examples."""
+    from asteroidnet.fits_ingestor.ingestor import ingest_fits_file
+    from asteroidnet.image_preprocessor.preprocessor import preprocess_frame, align_frames
+
+    # Download FITS frames
+    frames_raw = _download_frames(ra, dec, survey, date_range, config)
+    if not frames_raw:
+        return []
+
+    # Ingest + preprocess
+    fits_frames = []
+    data_subs = []
+    bkg_rmss = []
+    for frame_data, frame_header, frame_time, frame_survey in frames_raw:
+        # Build a minimal FITSFrame-like object
+        from asteroidnet.fits_ingestor.ingestor import FITSFrame
+        from astropy.wcs import WCS
+        import warnings
+        from astropy.wcs import FITSFixedWarning
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", FITSFixedWarning)
+            wcs = WCS(frame_header, naxis=2)
+        ff = FITSFrame(
+            path=Path(f"synthetic_{ra:.2f}_{dec:.2f}"),
+            data=frame_data,
+            header=frame_header,
+            wcs=wcs,
+            obs_time=frame_time,
+            exptime_s=float(frame_header.get("EXPTIME", 30.0)),
+            filter_band=str(frame_header.get("FILTER", "r")),
+            telescope=str(frame_header.get("TELESCOP", "?")),
+            instrument=str(frame_header.get("INSTRUME", "?")),
+            survey=frame_survey,
+            pixel_scale_arcsec=0.25 if frame_survey == "ps1" else 1.012,
+        )
+        from asteroidnet.image_preprocessor.preprocessor import preprocess_frame
+        data_sub, bkg_rms = preprocess_frame(ff, config)
+        fits_frames.append(ff)
+        data_subs.append(data_sub)
+        bkg_rmss.append(bkg_rms)
+
+    if not fits_frames:
+        return []
+
+    # Query SkyBoT for known SSOs at each epoch
+    examples = []
+    rng = np.random.default_rng(
+        seed=int(hashlib.md5(f"{ra}{dec}{survey}".encode()).hexdigest()[:8], 16)
+    )
+
+    for ff, data_sub, bkg_rms in zip(fits_frames, data_subs, bkg_rmss):
+        positives = _extract_positive_cutouts(ff, data_sub, bkg_rms, config)
+        negatives = _extract_negative_cutouts(
+            ff, data_sub, bkg_rms, positives, neg_per_pos, rng
+        )
+        examples.extend(positives)
+        examples.extend(negatives)
+
+    return examples
+
+
+def _download_frames(
+    ra: float,
+    dec: float,
+    survey: str,
+    date_range: tuple[str, str],
+    config: Optional[dict],
+) -> list:
+    """Download raw FITS data for a field. Returns list of (data, header, time, survey)."""
+    results = []
+    try:
+        if survey == "ps1":
+            from asteroidnet.data_access.ps1_client import get_ps1_warp_list, download_ps1_cutout
+            warps = get_ps1_warp_list(ra, dec, filter_band="r", config=config)
+            if len(warps) == 0:
+                return []
+            # Sample a few frames
+            for row in warps[:4]:
+                frame = download_ps1_cutout(ra, dec, row["filename"], size_pixels=300, config=config)
+                if frame:
+                    results.append((frame.data, frame.header, frame.obs_time, "ps1"))
+
+        elif survey == "ztf":
+            from asteroidnet.data_access.ztf_client import search_ztf_images, download_ztf_cutout
+            rows = search_ztf_images(ra, dec, filter_band="r", config=config)
+            for row in rows[:4]:
+                frame = download_ztf_cutout(ra, dec, row, size_arcsec=180, config=config)
+                if frame:
+                    results.append((frame.data, frame.header, frame.obs_time, "ztf"))
+
+    except Exception as exc:
+        logger.warning("Frame download failed for (%s %.2f %.2f): %s", survey, ra, dec, exc)
+
+    return results
+
+
+def _extract_positive_cutouts(
+    ff,
+    data_sub: np.ndarray,
+    bkg_rms: np.ndarray,
+    config: Optional[dict],
+) -> list[TrainingExample]:
+    """Find known SSOs in this frame and extract cutouts around them."""
+    examples = []
+    try:
+        from asteroidnet.data_access.skybot_client import query_skybot, skybot_table_to_skycoord
+        center = SkyCoord(ra=ff.ra_center if hasattr(ff, "ra_center") else 180.0,
+                          dec=ff.dec_center if hasattr(ff, "dec_center") else 0.0,
+                          unit=u.deg)
+        table = query_skybot(center, 15*u.arcmin, ff.obs_time,
+                             observer="F51" if ff.survey == "ps1" else "I41",
+                             config=config)
+        if len(table) == 0:
+            return []
+
+        sso_coords = skybot_table_to_skycoord(table)
+        if sso_coords is None:
+            return []
+
+        for coord in sso_coords:
+            try:
+                x, y = ff.wcs.world_to_pixel(coord)
+                x, y = int(round(float(x))), int(round(float(y)))
+                cutout = _extract_cutout(data_sub, x, y, CUTOUT_SIZE)
+                if cutout is None:
+                    continue
+
+                snr_val = float(np.nanmax(cutout) / (np.nanmedian(bkg_rms) + 1e-10))
+                examples.append(TrainingExample(
+                    cutout=cutout,
+                    label=1,
+                    snr=max(snr_val, 0.0),
+                    velocity_arcsec_s=0.5,  # approximate
+                    source_survey=ff.survey,
+                    field_ra=float(coord.ra.deg),
+                    field_dec=float(coord.dec.deg),
+                    epoch=ff.obs_time.isot,
+                ))
+            except Exception:
+                continue
+
+    except Exception as exc:
+        logger.debug("Positive cutout extraction failed: %s", exc)
+
+    return examples
+
+
+def _extract_negative_cutouts(
+    ff,
+    data_sub: np.ndarray,
+    bkg_rms: np.ndarray,
+    positives: list[TrainingExample],
+    n: int,
+    rng: np.random.Generator,
+) -> list[TrainingExample]:
+    """Extract random cutouts avoiding known object positions."""
+    h, w = data_sub.shape
+    half = CUTOUT_SIZE // 2
+    examples = []
+    attempts = 0
+
+    while len(examples) < n * len(positives) and attempts < 1000:
+        attempts += 1
+        x = int(rng.integers(half, w - half))
+        y = int(rng.integers(half, h - half))
+
+        cutout = _extract_cutout(data_sub, x, y, CUTOUT_SIZE)
+        if cutout is None:
+            continue
+        if np.isnan(cutout).mean() > 0.3:
+            continue
+
+        examples.append(TrainingExample(
+            cutout=cutout,
+            label=0,
+            snr=0.0,
+            velocity_arcsec_s=0.0,
+            source_survey=ff.survey,
+            field_ra=0.0,
+            field_dec=0.0,
+            epoch=ff.obs_time.isot,
+        ))
+
+    return examples
+
+
+def _extract_cutout(
+    data: np.ndarray,
+    x: int,
+    y: int,
+    size: int,
+) -> Optional[np.ndarray]:
+    """Extract a square cutout, normalizing to [0,1]. Returns None if out of bounds."""
+    half = size // 2
+    h, w = data.shape
+    if x - half < 0 or y - half < 0 or x + half >= w or y + half >= h:
+        return None
+    cutout = data[y - half:y + half + 1, x - half:x + half + 1].copy()
+    if cutout.shape != (size, size):
+        return None
+    # Normalize: subtract median, divide by MAD
+    finite = cutout[np.isfinite(cutout)]
+    if len(finite) == 0:
+        return None
+    med = np.median(finite)
+    mad = np.median(np.abs(finite - med))
+    if mad < 1e-10:
+        mad = 1.0
+    cutout = (cutout - med) / (3.0 * mad)
+    cutout = np.clip(cutout, -3.0, 3.0)
+    cutout = np.nan_to_num(cutout, nan=0.0).astype(np.float32)
+    return cutout

asteroidnet/utils/synthetic.py

@@ -0,0 +1,153 @@
+"""AsteroidNET Synthetic FITS generator for testing."""
+from __future__ import annotations
+import math, warnings
+from pathlib import Path
+from typing import Optional
+
+import numpy as np
+from astropy.io import fits
+from astropy.time import Time
+from astropy.wcs import WCS
+
+
+def make_synthetic_fits(
+    path: str | Path,
+    n_stars: int = 500,
+    n_asteroids: int = 3,
+    image_size: int = 512,
+    obs_time: Optional[Time] = None,
+    snr_asteroid: float = 8.0,
+    seed: int = 42,
+) -> Path:
+    """Create a synthetic FITS file with stars and planted asteroid-like sources."""
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    rng = np.random.default_rng(seed)
+
+    if obs_time is None:
+        obs_time = Time("2024-03-15T04:30:00", scale="utc")
+
+    # Background + noise
+    sky_bg  = 1000.0
+    readout = 10.0
+    data = rng.poisson(sky_bg, size=(image_size, image_size)).astype(np.float32)
+    data += rng.normal(0, readout, size=(image_size, image_size)).astype(np.float32)
+
+    # Plant stars
+    for _ in range(n_stars):
+        x, y = rng.uniform(10, image_size - 10, 2)
+        flux = 10 ** rng.uniform(2, 5)
+        _plant_psf(data, x, y, flux, fwhm=3.5)
+
+    # Plant asteroid-like sources
+    ast_positions = []
+    for i in range(n_asteroids):
+        x, y = rng.uniform(50, image_size - 50, 2)
+        noise_rms = math.sqrt(sky_bg + readout**2)
+        flux = snr_asteroid * noise_rms * math.pi * 3.5
+        _plant_psf(data, x, y, flux, fwhm=3.5)
+        ast_positions.append((x, y, flux))
+
+    # Build WCS (simple TAN at RA=180, Dec=0)
+    w = WCS(naxis=2)
+    w.wcs.crpix = [image_size / 2, image_size / 2]
+    w.wcs.cdelt = [-0.000069444, 0.000069444]  # 0.25 arcsec/px in degrees
+    w.wcs.crval = [180.0, 0.0]
+    w.wcs.ctype = ["RA---TAN", "DEC--TAN"]
+
+    hdr = w.to_header()
+    hdr["DATE-OBS"] = obs_time.utc.isot
+    hdr["MJD-OBS"]  = obs_time.utc.mjd
+    hdr["EXPTIME"]  = 30.0
+    hdr["FILTER"]   = "r"
+    hdr["TELESCOP"] = "Pan-STARRS"
+    hdr["INSTRUME"] = "GPC1"
+    hdr["TIMESYS"]  = "TAI"
+    hdr["RADESYS"]  = "FK5"
+    hdr["BKGND"]    = sky_bg
+    hdr["RDNOISE"]  = readout
+    hdr["NAST"]     = n_asteroids
+
+    # Store planted positions in header
+    for i, (x, y, f) in enumerate(ast_positions):
+        hdr[f"ASTX{i:02d}"] = round(x, 2)
+        hdr[f"ASTY{i:02d}"] = round(y, 2)
+        hdr[f"ASTF{i:02d}"] = round(f, 1)
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        hdul = fits.HDUList([fits.PrimaryHDU(data=data.astype(np.float32), header=hdr)])
+        hdul.writeto(path, overwrite=True, output_verify="silentfix")
+
+    return path
+
+
+def make_synthetic_sequence(
+    output_dir: str | Path,
+    n_frames: int = 4,
+    n_stars: int = 500,
+    n_asteroids: int = 3,
+    image_size: int = 512,
+    velocity_arcsec_s: float = 0.3,
+    cadence_min: float = 10.0,
+    seed: int = 42,
+) -> list[Path]:
+    """Generate a sequence of FITS frames with moving asteroid trails."""
+    output_dir = Path(output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    rng = np.random.default_rng(seed)
+
+    t0 = Time("2024-03-15T04:00:00", scale="utc")
+    paths = []
+
+    # Initial asteroid positions
+    ast0 = [(float(rng.uniform(80, image_size - 80)),
+             float(rng.uniform(80, image_size - 80)))
+            for _ in range(n_asteroids)]
+    # Random velocities in pixels/s for each asteroid
+    pixel_scale = 0.25  # arcsec/px
+    vel_pix_s   = velocity_arcsec_s / pixel_scale
+    ast_vel = [(float(rng.normal(vel_pix_s * 0.7, vel_pix_s * 0.3)),
+                float(rng.normal(vel_pix_s * 0.3, vel_pix_s * 0.1)))
+               for _ in range(n_asteroids)]
+
+    for i in range(n_frames):
+        # TimeDelta used properly below — this line replaced
+        from astropy.time import TimeDelta
+        t = t0 + TimeDelta(i * cadence_min * 60, format="sec")
+        dt = i * cadence_min * 60  # seconds from t0
+
+        path = output_dir / f"frame_{i:02d}.fits"
+        make_synthetic_fits(path, n_stars, 0, image_size, t, seed=seed + i)
+
+        # Overwrite with moving asteroids at their updated positions
+        with fits.open(path, mode="update") as hdul:
+            data = hdul[0].data
+            for j, ((x0, y0), (vx, vy)) in enumerate(zip(ast0, ast_vel)):
+                x = x0 + vx * dt
+                y = y0 + vy * dt
+                noise_rms = math.sqrt(1000.0 + 100.0)
+                flux = 8.0 * noise_rms * math.pi * 3.5
+                _plant_psf(data, x, y, flux, fwhm=3.5)
+                hdul[0].header[f"ASTX{j:02d}"] = round(x, 2)
+                hdul[0].header[f"ASTY{j:02d}"] = round(y, 2)
+            hdul.flush()
+
+        paths.append(path)
+
+    return paths
+
+
+def _plant_psf(data: np.ndarray, cx: float, cy: float, flux: float, fwhm: float = 3.5):
+    """Plant a 2D Gaussian PSF at (cx, cy) with given total flux."""
+    sigma = fwhm / 2.355
+    h, w = data.shape
+    r = int(4 * sigma) + 1
+
+    x0, y0 = int(round(cx)), int(round(cy))
+    for dy in range(-r, r + 1):
+        for dx in range(-r, r + 1):
+            xi, yi = x0 + dx, y0 + dy
+            if 0 <= xi < w and 0 <= yi < h:
+                g = math.exp(-0.5 * ((dx**2 + dy**2) / sigma**2))
+                data[yi, xi] += flux * g / (2 * math.pi * sigma**2)

asteroidnet/utils/time_utils.py

@@ -0,0 +1,171 @@
+"""
+AsteroidNET Time Utilities (utils.time_utils)
+
+Handles the TAI vs UTC mismatch between Pan-STARRS and ZTF — the most
+critical gotcha when processing real survey data.
+
+Key facts:
+  - Pan-STARRS MJD-OBS is in TAI (full skycell images MISSING 'TIMESYS' header)
+  - ZTF OBSJD / OBSMJD / DATE-OBS are in UTC
+  - TAI is ahead of UTC by 37 seconds (as of 2017, PS1 DR2 era)
+  - 37 seconds = ~0.5–2 arcseconds of asteroid apparent motion
+    (enough to put a predicted position outside the detection aperture)
+  - For asteroid ephemeris queries, JPL Horizons expects UTC
+  - Mid-exposure time = start_time + 0.5 * EXPTIME
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from astropy.time import Time
+
+import astropy.units as u
+
+logger = logging.getLogger(__name__)
+
+# TAI is ahead of UTC by this many seconds during the PS1 observation period.
+# Updated per IERS bulletins — 37 seconds since 2017-01-01.
+_PS1_TAI_UTC_OFFSET_S = 37.0
+
+# Survey-specific time scale lookup
+_SURVEY_TIMESCALE = {
+    "ps1":       "tai",   # Pan-STARRS — TAI (header often silent about this)
+    "ztf":       "utc",   # ZTF — UTC
+    "catalina":  "utc",
+    "atlas":     "utc",
+    "generic":   "utc",
+}
+
+
+def parse_fits_time(
+    header,
+    survey: str = "generic",
+    exptime_key: str = "EXPTIME",
+    obs_time_key: Optional[str] = None,
+    return_midexp: bool = True,
+) -> Time:
+    """
+    Parse observation time from a FITS header, correctly handling survey-specific
+    time scale conventions (most importantly Pan-STARRS TAI vs ZTF UTC).
+
+    Parameters
+    ----------
+    header : fits.Header or dict
+        FITS header containing time keywords.
+    survey : str
+        Survey identifier: 'ps1', 'ztf', 'catalina', 'atlas', 'generic'.
+        Controls which time scale is assumed.
+    exptime_key : str
+        Header keyword for exposure time in seconds.
+    obs_time_key : str, optional
+        Override the time keyword to read. If None, auto-detected from survey.
+    return_midexp : bool
+        If True, add half the exposure time to get mid-exposure time.
+
+    Returns
+    -------
+    Time
+        Observation time in UTC scale, at mid-exposure if return_midexp=True.
+    """
+    survey = survey.lower()
+    scale = _SURVEY_TIMESCALE.get(survey, "utc")
+
+    # Determine which header keyword holds the observation time
+    if obs_time_key is not None:
+        time_key = obs_time_key
+    elif survey == "ps1":
+        # Pan-STARRS: MJD-OBS is TAI (even if TIMESYS is absent/wrong)
+        time_key = "MJD-OBS"
+    elif survey == "ztf":
+        # ZTF: prefer OBSMJD (MJD), fall back to DATE-OBS (ISO)
+        time_key = "OBSMJD" if "OBSMJD" in header else "DATE-OBS"
+    else:
+        # Generic: try common keywords in order
+        for k in ("DATE-OBS", "MJD-OBS", "OBSMJD", "OBSJD"):
+            if k in header:
+                time_key = k
+                break
+        else:
+            raise KeyError(f"No recognized time keyword in header. Keys: {list(header.keys())[:20]}")
+
+    raw_time = header[time_key]
+
+    # Parse according to format
+    if isinstance(raw_time, (int, float)):
+        # Floating-point MJD or JD
+        fmt = "jd" if raw_time > 2400000 else "mjd"
+        t = Time(raw_time, format=fmt, scale=scale)
+    else:
+        # ISO string
+        t = Time(str(raw_time), format="isot", scale=scale)
+
+    # Convert to UTC (all downstream code expects UTC)
+    t_utc = t.utc
+
+    # Add half exposure time to get mid-exposure
+    if return_midexp and exptime_key in header:
+        exptime_s = float(header[exptime_key])
+        t_utc = t_utc + (exptime_s / 2.0) * u.s
+        logger.debug(
+            "Mid-exposure time (%s, scale=%s→utc): %s (exptime=%.1fs)",
+            survey, scale, t_utc.isot, exptime_s
+        )
+    else:
+        logger.debug("Start-of-exposure time (%s→utc): %s", survey, t_utc.isot)
+
+    return t_utc
+
+
+def detect_survey_from_header(header) -> str:
+    """
+    Attempt to identify the survey from FITS header keywords.
+
+    Returns one of: 'ps1', 'ztf', 'catalina', 'atlas', 'generic'
+    """
+    telescop = str(header.get("TELESCOP", "")).lower()
+    instrume = str(header.get("INSTRUME", "")).lower()
+    origin   = str(header.get("ORIGIN", "")).lower()
+    
+    if any(x in telescop for x in ("ps1", "panstarrs", "pan-starrs", "haleakala", "p60")):
+        return "ps1"
+    if any(x in telescop for x in ("p48", "palomar48", "ztf")):
+        return "ztf"
+    if "catalina" in telescop or "css" in instrume:
+        return "catalina"
+    if "atlas" in telescop:
+        return "atlas"
+    if "ps1" in str(header.get("FILENAME", "")).lower():
+        return "ps1"
+    if "ztf_" in str(header.get("FILENAME", "")).lower():
+        return "ztf"
+
+    return "generic"
+
+
+def fix_ps1_header(header) -> None:
+    """
+    In-place fix for Pan-STARRS full skycell FITS headers.
+
+    Known PS1 DR2 issues (from STScI documentation):
+      1. TIMESYS keyword absent �� should be 'TAI'
+      2. RADESYS keyword absent — should be 'FK5'
+      3. WCS uses obsolete PC001001 naming instead of PC1_1
+
+    The fitscut.cgi service corrects these, but full skycell downloads don't.
+    """
+    if "TIMESYS" not in header:
+        header["TIMESYS"] = ("TAI", "Time system [added by AsteroidNET]")
+        logger.debug("Added missing TIMESYS=TAI to PS1 header")
+
+    if "RADESYS" not in header:
+        header["RADESYS"] = ("FK5", "Celestial coordinate system [added by AsteroidNET]")
+        logger.debug("Added missing RADESYS=FK5 to PS1 header")
+
+    # Fix obsolete PC matrix naming (PC001001 → PC1_1 etc.)
+    for old, new in [("PC001001", "PC1_1"), ("PC001002", "PC1_2"),
+                     ("PC002001", "PC2_1"), ("PC002002", "PC2_2")]:
+        if old in header and new not in header:
+            header[new] = header[old]
+            logger.debug("Renamed WCS keyword %s → %s", old, new)