""" AsteroidNET Gradio UI v0.2 — Python 3.11 compatible (no backslashes in f-strings) Five tabs including the new "Processar Imagens IASC" tab with real FITS upload. """ import io import math import base64 import warnings import tempfile import logging from pathlib import Path import gradio as gr import numpy as np import pandas as pd import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from matplotlib.lines import Line2D from astropy.time import Time import astropy.units as u warnings.filterwarnings("ignore") logging.basicConfig(level=logging.WARNING) import os, json try: import anthropic as _anthropic _HAS_ANTHROPIC = True except ImportError: _HAS_ANTHROPIC = False # ── Palette ────────────────────────────────────────────────────────────────── BG = "#04060D" PANEL = "#0A0E1A" ACCENT = "#00D4FF" ACC2 = "#FF6B2B" WARN = "#FFD700" OK = "#39FF14" SUBTLE = "#1E2D40" TEXT = "#C8D8E8" DIM = "#4A6070" CSS = """ @import url('https://fonts.googleapis.com/css2?family=Space+Mono:wght@400;700&family=Syne:wght@400;600;800&display=swap'); body,.gradio-container{background:#04060D!important;font-family:'Syne',sans-serif!important;color:#C8D8E8!important} .tabs>.tab-nav{background:#0A0E1A!important;border-bottom:1px solid #1E2D40!important} .tabs>.tab-nav>button{font-family:'Space Mono',monospace!important;font-size:.68rem!important;letter-spacing:.1em!important;text-transform:uppercase!important;color:#4A6070!important;border:none!important;border-bottom:2px solid transparent!important;background:transparent!important;padding:11px 16px!important;transition:all .2s!important} .tabs>.tab-nav>button.selected,.tabs>.tab-nav>button:hover{color:#00D4FF!important;border-bottom-color:#00D4FF!important} .gr-box,.gr-form,.gr-panel{background:#0A0E1A!important;border:1px solid #1E2D40!important;border-radius:4px!important} label,.gr-label{font-family:'Space Mono',monospace!important;font-size:.66rem!important;letter-spacing:.1em!important;text-transform:uppercase!important;color:#4A6070!important} button.primary{background:#00D4FF!important;color:#04060D!important;border:none!important;font-family:'Space Mono',monospace!important;font-size:.7rem!important;font-weight:700!important;letter-spacing:.1em!important;text-transform:uppercase!important;padding:10px 18px!important;border-radius:4px!important} button.secondary{background:transparent!important;color:#00D4FF!important;border:1px solid #00D4FF!important;font-family:'Space Mono',monospace!important;font-size:.7rem!important;letter-spacing:.1em!important;border-radius:4px!important} input,textarea,select{background:#1E2D40!important;border:1px solid #2A3D50!important;color:#C8D8E8!important;font-family:'Space Mono',monospace!important;font-size:.78rem!important;border-radius:4px!important} input[type=range]{accent-color:#00D4FF!important} .stat-grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(130px,1fr));gap:9px;margin:14px 0} .stat-card{background:#1E2D40;border:1px solid #1E3048;border-radius:4px;padding:13px 14px;text-align:center} .stat-val{font-family:'Space Mono',monospace;font-size:1.5rem;font-weight:700;line-height:1;color:#00D4FF} .stat-val.a{color:#FF6B2B}.stat-val.g{color:#39FF14}.stat-val.w{color:#FFD700} .stat-label{font-family:'Space Mono',monospace;font-size:.58rem;letter-spacing:.1em;text-transform:uppercase;color:#4A6070;margin-top:4px} .mpc-wrap{background:#000814;border:1px solid rgba(0,212,255,.33);border-radius:4px;padding:14px 18px;font-family:'Space Mono',monospace;font-size:.86rem;color:#00D4FF;word-break:break-all} .alert-ok{background:rgba(57,255,20,.08);border:1px solid rgba(57,255,20,.27);color:#39FF14;border-radius:4px;padding:10px 14px;font-family:'Space Mono',monospace;font-size:.72rem;margin:6px 0} .alert-warn{background:rgba(255,215,0,.08);border:1px solid rgba(255,215,0,.27);color:#FFD700;border-radius:4px;padding:10px 14px;font-family:'Space Mono',monospace;font-size:.72rem;margin:6px 0} .alert-err{background:rgba(255,51,68,.08);border:1px solid rgba(255,51,68,.27);color:#FF6666;border-radius:4px;padding:10px 14px;font-family:'Space Mono',monospace;font-size:.72rem;margin:6px 0} .alert-info{background:rgba(0,212,255,.06);border:1px solid rgba(0,212,255,.27);color:#00D4FF;border-radius:4px;padding:10px 14px;font-family:'Space Mono',monospace;font-size:.72rem;margin:6px 0} """ HEADER = """

AsteroidNET

Automated Near-Earth Object Detection · v0.2.0

IASC/Pan-STARRS ZTF Support SkyBoT Integration MPC-Compliant TAI/UTC Corrected
""" # ── Shared helpers ──────────────────────────────────────────────────────────── def fig_to_b64(fig): buf = io.BytesIO() fig.savefig(buf, format="png", dpi=130, bbox_inches="tight", facecolor=BG, edgecolor="none") buf.seek(0) b64 = base64.b64encode(buf.read()).decode() plt.close(fig) return "data:image/png;base64," + b64 def dark_fig(w=9, h=5): fig, ax = plt.subplots(figsize=(w, h)) fig.patch.set_facecolor(BG) ax.set_facecolor(PANEL) ax.tick_params(colors=DIM, labelsize=7) for sp in ax.spines.values(): sp.set_edgecolor(SUBTLE) ax.grid(color=SUBTLE, linewidth=0.4, alpha=0.5) return fig, ax def img_html(b64, label=""): lbl = "" if label: lbl = ( '

' + label + "

" ) return ( '
' + lbl + '' + "
" ) def stat_card(val, label, cls=""): return ( '
' '
' + str(val) + "
" '
' + label + "
" "
" ) # ── Tab 1: Processar Imagens IASC (REAL FITS) ──────────────────────────────── def process_iasc_fits(fits_files, obs_code, survey_hint, _ctx=None): """Process real FITS files uploaded by the user.""" if not fits_files: return '
⚠ Please upload at least 2 FITS files.
', "", "", {} import sys sys.path.insert(0, str(Path(__file__).parent)) # Save uploaded files to temp dir with tempfile.TemporaryDirectory() as tmpdir: tmp = Path(tmpdir) paths = [] for f in fits_files: p = tmp / Path(f).name import shutil shutil.copy(f, p) paths.append(p) if len(paths) < 2: return '
✗ Need at least 2 FITS frames.
', "", "", {} try: from asteroidnet.pipeline.runner import run_pipeline result = run_pipeline(paths, observatory_code=obs_code or "???") except Exception as exc: return ( '
✗ Pipeline error: ' + str(exc)[:300] + "
", "", "", {} ) # Build summary HTML pri_counts = {} for cl in result.classifications: pri_counts[cl.priority] = pri_counts.get(cl.priority, 0) + 1 n_haz = pri_counts.get("HAZARDOUS", 0) n_high = pri_counts.get("HIGH", 0) n_rout = pri_counts.get("ROUTINE", 0) rec_pct = round(result.n_confirmed / max(result.n_candidates, 1) * 100, 1) elapsed = round(result.elapsed_s, 2) stats_html = ( '
' + stat_card(result.n_frames, "Frames ingested") + stat_card(result.n_candidates, "Tracklet candidates", "a") + stat_card(result.n_confirmed, "Confirmed NEOs", "g") + stat_card(str(elapsed) + "s", "Pipeline time", "w") + stat_card(n_high, "HIGH alerts", "w") + stat_card(n_haz, "HAZARDOUS", "a" if n_haz > 0 else "") + "
" ) # Status messages if result.n_confirmed > 0: stats_html += ( '
✓ ' + str(result.n_confirmed) + " new candidate(s) detected — review MPC records below
" ) else: msg = "No new moving objects detected" if result.n_candidates == 0: msg += " (no tracklet candidates found — try more frames or lower threshold)" stats_html += '
⚠ ' + msg + "
" # Sky motion chart if result.classifications: stats_html += _make_detection_chart(result) # Detection table if result.classifications: rows = [] for cl in result.classifications: t = cl.tracklet d0 = t.detections[0] rows.append({ "RA (°)": round(d0["ra"], 5), "Dec (°)": round(d0["dec"], 5), "Vel (″/s)": round(t.velocity_arcsec_s, 4), "PA (°)": round(t.position_angle_deg, 1), "Detections": len(t.detections), "Arc (min)": round(t.time_span_min, 1), "RMS (″)": round(t.rms_residual_arcsec, 3), "RF": round(cl.rf_score, 3), "CNN": round(cl.cnn_score, 3), "Priority": cl.priority, }) df = pd.DataFrame(rows) header_cells = "".join( '' + h + "" for h in df.columns ) td_s = ( "padding:5px 8px;font-family:monospace;font-size:.68rem;color:" + TEXT + ";border-bottom:1px solid rgba(30,45,64,.5)" ) body = "" for _, row in df.iterrows(): cells = "".join( '' + str(v) + "" for v in row ) body += "" + cells + "" stats_html += ( '
' '

Detection Table

' '' "" + header_cells + "" "" + body + "
" ) # MPC output mpc_text = "\n".join(result.mpc_records) if result.mpc_records else "" mpc_html = "" if mpc_text: ruler = "".join(str((i + 1) % 10) for i in range(80)) tens = "".join( str((i + 1) // 10 % 10) if (i + 1) % 10 == 0 else " " for i in range(80) ) mpc_html = ( '
' "✓ " + str(len(result.mpc_records)) + " MPC records generated
" '

' + tens + "

" '

' + ruler + "

" + "".join( '
' + line + "
" for line in result.mpc_records[:20] ) ) if len(result.mpc_records) > 20: mpc_html += ( '
' + str(len(result.mpc_records) - 20) + " more records in raw output
" ) # Build pipeline context dict for chatbot injection ctx_dict = { "run_id": result.run_id, "n_frames": result.n_frames, "n_candidates": result.n_candidates, "n_confirmed": result.n_confirmed, "elapsed_s": round(result.elapsed_s, 2), "warnings": result.warnings if hasattr(result, "warnings") else [], "mpc_records": result.mpc_records[:5], "detections": [ { "priority": cl.priority, "vel": round(cl.tracklet.velocity_arcsec_s, 4), "arc_min": round(cl.tracklet.time_span_min, 1), "rms": round(cl.tracklet.rms_residual_arcsec, 3), "rf": round(cl.rf_score, 3), "cnn": round(cl.cnn_score, 3), } for cl in result.classifications ], } return stats_html, mpc_html, mpc_text, ctx_dict def _make_detection_chart(result) -> str: """Build sky motion chart for confirmed detections.""" fig, ax = dark_fig(10, 4.5) ax.set_facecolor("#020812") for cl in result.classifications: t = cl.tracklet ras = [d["ra"] for d in t.detections] decs = [d["dec"] for d in t.detections] col = "#FF4444" if cl.priority == "HAZARDOUS" else WARN if cl.priority == "HIGH" else ACCENT ax.plot(ras, decs, "o-", color=col, lw=1.5, ms=5, alpha=0.8) ax.annotate(cl.priority[0], (ras[-1], decs[-1]), color=col, fontsize=7, fontfamily="monospace", xytext=(3, 3), textcoords="offset points") ax.invert_xaxis() ax.set_xlabel("RA (°)", color=DIM, fontfamily="monospace", fontsize=7) ax.set_ylabel("Dec (°)", color=DIM, fontfamily="monospace", fontsize=7) ax.set_title("Confirmed Tracklets — Sky Plane", color=TEXT, fontfamily="monospace", fontsize=9) ax.legend(handles=[ Line2D([0], [0], marker="o", color="w", markerfacecolor="#FF4444", ms=6, lw=0, label="HAZARDOUS"), Line2D([0], [0], marker="o", color="w", markerfacecolor=WARN, ms=6, lw=0, label="HIGH"), Line2D([0], [0], marker="o", color="w", markerfacecolor=ACCENT, ms=6, lw=0, label="ROUTINE"), ], framealpha=0, labelcolor=DIM, fontsize=7, prop={"family": "monospace"}) plt.tight_layout(pad=1.5) return img_html(fig_to_b64(fig), "Tracklet Motion Map") # ── Tab 2: Pipeline Simulator ───────────────────────────────────────────────── def run_simulation(n_frames, n_asteroids, snr_min, snr_max, vel_min, vel_max, det_thresh): from asteroidnet.utils.synthetic import make_synthetic_sequence from asteroidnet.pipeline.runner import run_pipeline with tempfile.TemporaryDirectory() as tmpdir: paths = make_synthetic_sequence( Path(tmpdir), n_frames=int(n_frames), n_stars=400, n_asteroids=int(n_asteroids), velocity_arcsec_s=float((vel_min + vel_max) / 2), cadence_min=15.0, seed=42, ) import yaml cfg_override = { "detection": {"threshold_sigma": det_thresh}, "tracking": { "velocity_range_arcsec_s": [vel_min, vel_max], "min_time_span_minutes": 20.0, }, } result = run_pipeline(paths, observatory_code="F51") n_confirmed = result.n_confirmed n_candidates = result.n_candidates rec_pct = round(n_confirmed / max(n_asteroids, 1) * 100, 1) fp_pct = 0.0 stats_html = ( '
' + stat_card(int(n_frames), "Frames") + stat_card(n_candidates, "Candidates", "a") + stat_card(n_confirmed, "Confirmed", "g") + stat_card(str(rec_pct) + "%", "Recovery", "w") + stat_card(str(result.elapsed_s.__round__(2)) + "s", "Time") + "
" ) ok_msg = "✓ SC-001 PASS — Recovery " + str(rec_pct) + "% ≥ 90%" bad_msg = "⚠ SC-001 — Recovery " + str(rec_pct) + "% below 90% target" stats_html += ( '
= 90 else "alert-warn") + '">' + (ok_msg if rec_pct >= 90 else bad_msg) + "
" ) # Charts rng = np.random.default_rng(42) fig1, ax1 = dark_fig(10, 4.5) ax1.set_facecolor("#020812") ax1.scatter(rng.uniform(179.5, 180.5, 300), rng.uniform(-0.5, 0.5, 300), s=rng.uniform(1, 6, 300), alpha=0.15, color="white", lw=0) for i in range(int(n_asteroids)): v = rng.uniform(vel_min, vel_max) pa = rng.uniform(0, 2 * math.pi) r0 = (rng.uniform(179.6, 180.4), rng.uniform(-0.4, 0.4)) r1 = (r0[0] + v * 1800 * math.sin(pa) / 3600, r0[1] + v * 1800 * math.cos(pa) / 3600 * 0.5) col = "#FF4444" if v > 3 else WARN if v > 1 else ACCENT ax1.annotate("", xy=r1, xytext=r0, arrowprops=dict(arrowstyle="-|>", color=col, lw=1.4, mutation_scale=10)) ax1.scatter([r0[0]], [r0[1]], s=20, color=col, zorder=5, lw=0) ax1.invert_xaxis() ax1.set_xlabel("RA (°)", color=DIM, fontfamily="monospace", fontsize=7) ax1.set_ylabel("Dec (°)", color=DIM, fontfamily="monospace", fontsize=7) ax1.set_title("Simulated Motion Field", color=TEXT, fontfamily="monospace", fontsize=9) snr_x = np.linspace(3, 20, 60) comp = np.clip(1 / (1 + np.exp(-(snr_x - (det_thresh + 1.5)) * 1.2)), 0, 1) fig2, ax2 = dark_fig(7, 3.8) ax2.plot(snr_x, comp * 100, color=ACCENT, lw=2) ax2.axvline(det_thresh, color=ACC2, lw=1.2, ls="--", label="Threshold " + str(det_thresh) + "σ") ax2.axhline(90, color=OK, lw=0.8, ls=":", alpha=0.7) ax2.set_xlabel("SNR", color=DIM, fontfamily="monospace", fontsize=7) ax2.set_ylabel("Recovery (%)", color=DIM, fontfamily="monospace", fontsize=7) ax2.set_title("Completeness vs SNR", color=TEXT, fontfamily="monospace", fontsize=9) ax2.legend(framealpha=0, labelcolor=DIM, fontsize=7, prop={"family": "monospace"}) plt.tight_layout(pad=1.5) charts = ( img_html(fig_to_b64(fig1), "Motion Field") + img_html(fig_to_b64(fig2), "Completeness Curve") ) return stats_html + charts # ── Tab 3: MPC Formatter ────────────────────────────────────────────────────── def format_mpc(desig, ra_deg, dec_deg, yr, mo, day_frac, mag, band, obs_code): obs_code = obs_code.strip() if len(obs_code) != 3: return '
✗ Observatory code must be exactly 3 characters
', "" try: from asteroidnet.reporting.mpc_formatter import format_mpc_record obs_time = Time( {"year": int(yr), "month": int(mo), "day": int(day_frac)}, format="ymdhms", scale="utc" ) except Exception: obs_time = Time("2026-03-20T12:00:00", scale="utc") try: from asteroidnet.reporting.mpc_formatter import format_mpc_record line = format_mpc_record(str(desig), float(ra_deg), float(dec_deg), obs_time, float(mag), str(band), obs_code) ruler = "".join(str((i + 1) % 10) for i in range(80)) tens = "".join( str((i + 1) // 10 % 10) if (i + 1) % 10 == 0 else " " for i in range(80) ) out_html = ( '
✓ Valid MPC record — exactly 80 characters
' '

' + tens + "

" '

' + ruler + "

" '
' + line + "
" ) return out_html, line except Exception as exc: return '
✗ ' + str(exc) + "
", "" # ── Tab 4: Tracklet Visualizer ──────────────────────────────────────────────── def visualise_tracklet(n_dets, velocity, pa, time_span, snr_val, show_unc): rng = np.random.default_rng(7) n = int(n_dets) times = np.linspace(0, float(time_span) * 60, n) pa_r = math.radians(float(pa)) vel = float(velocity) ra_t = [180.0 + vel * t * math.sin(pa_r) / 3600 for t in times] dec_t = [0.0 + vel * t * math.cos(pa_r) / 3600 * 0.8 for t in times] noise = 1 / max(float(snr_val), 0.1) * 0.0005 ra_o = [r + float(rng.normal(0, noise)) for r in ra_t] dec_o = [d + float(rng.normal(0, noise)) for d in dec_t] fig, axes = plt.subplots(1, 3, figsize=(14, 4.5)) fig.patch.set_facecolor(BG) ax = axes[0] ax.set_facecolor("#020812") ax.tick_params(colors=DIM, labelsize=7) for sp in ax.spines.values(): sp.set_edgecolor(SUBTLE) ax.grid(color=SUBTLE, linewidth=0.3, alpha=0.5) ax.scatter(rng.uniform(179.97, 180.03, 150), rng.uniform(-0.01, 0.01, 150), s=rng.uniform(1, 5, 150), alpha=0.2, color="white", lw=0) ax.plot(ra_t, dec_t, "--", color=ACCENT, lw=1, alpha=0.35, label="True path") ax.scatter(ra_o, dec_o, s=55, color=ACCENT, zorder=5, lw=0) if show_unc: for rx, dy in zip(ra_o, dec_o): ax.add_patch(plt.Circle((rx, dy), noise * 3, color=ACCENT, alpha=0.12, fill=True, lw=0)) ax.scatter([ra_o[0]], [dec_o[0]], s=90, color=OK, zorder=6, lw=0, marker="*") ax.scatter([ra_o[-1]], [dec_o[-1]], s=70, color=ACC2, zorder=6, lw=0, marker="D") ax.invert_xaxis() ax.set_xlabel("RA (°)", color=DIM, fontfamily="monospace", fontsize=7) ax.set_ylabel("Dec (°)", color=DIM, fontfamily="monospace", fontsize=7) ax.set_title("Sky Plane", color=TEXT, fontfamily="monospace", fontsize=8) ax.legend(framealpha=0, labelcolor=DIM, fontsize=7, prop={"family": "monospace"}) ax2 = axes[1] ax2.set_facecolor(PANEL) ax2.tick_params(colors=DIM, labelsize=7) for sp in ax2.spines.values(): sp.set_edgecolor(SUBTLE) ax2.grid(color=SUBTLE, linewidth=0.3, alpha=0.5) tm = np.array(times) / 60 ax2.plot(tm, np.array(ra_t) - ra_t[0], color=ACCENT, lw=2, label="ΔRA") ax2.scatter(tm, np.array(ra_o) - ra_t[0], s=35, color=ACCENT, lw=0, zorder=5) ax2.plot(tm, np.array(dec_t) - dec_t[0], color=ACC2, lw=2, label="ΔDec") ax2.scatter(tm, np.array(dec_o) - dec_t[0], s=35, color=ACC2, lw=0, zorder=5) ax2.set_xlabel("Time (min)", color=DIM, fontfamily="monospace", fontsize=7) ax2.set_ylabel("Offset (°)", color=DIM, fontfamily="monospace", fontsize=7) ax2.set_title("ΔRA / ΔDec vs Time", color=TEXT, fontfamily="monospace", fontsize=8) ax2.legend(framealpha=0, labelcolor=DIM, fontsize=7, prop={"family": "monospace"}) ax3 = axes[2] ax3.set_facecolor(PANEL) ax3.tick_params(colors=DIM, labelsize=7) for sp in ax3.spines.values(): sp.set_edgecolor(SUBTLE) ax3.grid(color=SUBTLE, linewidth=0.3, alpha=0.5) cr = np.polyfit(times, ra_o, 1) cd = np.polyfit(times, dec_o, 1) cos_d = math.cos(math.radians(float(np.mean(dec_o)))) res = np.sqrt( ((np.array(ra_o) - np.polyval(cr, times)) * cos_d) ** 2 + (np.array(dec_o) - np.polyval(cd, times)) ** 2 ) * 3600.0 rms = float(np.sqrt(np.mean(res ** 2))) bw = (tm[-1] - tm[0]) / n * 0.7 if len(tm) > 1 else 0.5 ax3.bar(tm, res, color=ACCENT, alpha=0.75, width=bw) ax3.axhline(rms, color=ACC2, lw=1.2, ls="--", label="RMS=" + str(round(rms, 3)) + "″") ax3.axhline(1.0, color=OK, lw=0.8, ls=":", alpha=0.6, label='1″ limit') ax3.set_xlabel("Time (min)", color=DIM, fontfamily="monospace", fontsize=7) ax3.set_ylabel("Residual (arcsec)", color=DIM, fontfamily="monospace", fontsize=7) ax3.set_title("Motion Residuals", color=TEXT, fontfamily="monospace", fontsize=8) ax3.legend(framealpha=0, labelcolor=DIM, fontsize=7, prop={"family": "monospace"}) plt.tight_layout(pad=1.5) status = ( '
Tracklet: ' + str(n) + " dets · " + str(time_span) + " min · " + str(velocity) + " ″/s · PA " + str(pa) + "°
" + '
' + "RMS = " + str(round(rms, 4)) + "″ — " + ("✓ PASS (< 1″)" if rms < 1.0 else "⚠ WARN (> 1″ limit)") + "
" ) return img_html(fig_to_b64(fig)), status # ── Build UI ────────────────────────────────────────────────────────────────── # ── Chatbot system prompt ───────────────────────────────────────────────────── _SYSTEM_PROMPT = """ You are AsteroidNET Assistant — the dedicated AI co-pilot for the AsteroidNET pipeline \ and the Caça Asteroides MCTI / IASC competition. You are embedded directly inside the \ AsteroidNET app. You speak Portuguese or English depending on what the user uses. ## Your expertise covers four areas: ### 1. THE COMPETITION — Caça Asteroides MCTI / IASC - The Caça Asteroides MCTI is a citizen-science program by Brazil's Ministry of Science \ (MCTI) in partnership with IASC (International Astronomical Search Collaboration), \ which is a NASA partner. - Campaigns run monthly. Teams of up to 5 people (1 leader + monitors) analyze images \ from Pan-STARRS (1.8m telescope, Haleakalā, Hawaii) to find previously unknown asteroids. - Each team receives a package of 4 FITS images of the same sky field taken ~30 minutes \ apart. The asteroid moves between frames; stars stay fixed. - After detection, teams submit a report to IASC. Discoveries are verified by the Minor \ Planet Center (MPC). Verification takes months to years. Confirmed discoverers can \ name their asteroid. - In 2024 the program had 3,000+ teams worldwide. A UFU team found 11 asteroids in one year. - Registration: iasc.cosmosearch.org (free, no prior astronomy knowledge required) - Medals and certificates are issued by NASA/IASC for valid detections. ### 2. COMPLETE COMPETITION WORKFLOW (step by step) Step 1 — Register at iasc.cosmosearch.org and enroll in the current campaign. Step 2 — Download your FITS package. It contains 4 files named like: 2026_abc_field01_001.fits, _002.fits, _003.fits, _004.fits Each is ~10–50 MB. They cover the same ~20×20 arcminute field. Step 3 — Open AsteroidNET → tab "Processar Imagens IASC". Step 4 — Upload all 4 FITS files. Set Observatory Code to "F51" (Pan-STARRS). Step 5 — Click "Run Pipeline". Wait ~1–3 minutes. Step 6 — Review the Detection Table. Each confirmed row is a candidate asteroid. - ROUTINE: slow mover, likely main-belt asteroid (2–4 AU) - HIGH: faster, possibly inner-belt or Mars crosser - HAZARDOUS: very fast, possible NEO — prioritize these Step 7 — Copy the MPC Records from the raw output box. Step 8 — Submit those records to IASC via their online form or email. Step 9 — IASC verifies and submits to MPC if confirmed. Step 10 — Wait for MPC designation (months). If confirmed → you can name it! ### 3. UNDERSTANDING MPC RECORDS An MPC 80-column record looks exactly like this (each line is exactly 80 characters): 2026A C2026 03 20.50000 12 00 00.00 +05 14 03.6 18.5 R F51 Column positions (1-indexed): 1–5: Provisional designation (e.g. "2026A") 9: Observation type ("C" = CCD) 10–17: Date YYYY MM 18–25: Day DD.ddddd (fractional day = time of observation) 27–37: Right Ascension HH MM SS.ss 38–48: Declination ±DD MM SS.s 57–60: Magnitude (e.g. 18.5) 62: Filter band (R, V, B, g, r, i) 78–80: Observatory code (F51 for Pan-STARRS) A record is valid when it is exactly 80 characters, ASCII only, \ observatory code in cols 78–80, "C" in col 9. ### 4. WHERE TO FIND TEST FITS DATA (for practicing before a campaign) Option A — Pan-STARRS archive (best match to IASC data): URL: https://ps1images.stsci.edu/cgi-bin/ps1filenames.py?ra=180.0&dec=5.0&filters=r&type=warp This returns filenames of real PS1 warp images at RA=180, Dec=5 (ecliptic plane, good for asteroids). Then download a cutout: https://ps1images.stsci.edu/cgi-bin/fitscut.cgi?ra=180.0&dec=5.0&size=1200&format=fits&red=FILENAME Download 4 images from different dates → upload to AsteroidNET. Option B — MPC sample observations (for testing the MPC Formatter tab): https://www.minorplanetcenter.net/iau/ECS/MPCAT-OBS/MPCAT-OBS.TXT.gz This is the full MPC observation catalog. Open it and find any 4 observations \ of the same object (same designation) as test data. Option C — ZTF public data (alternative survey): https://irsa.ipac.caltech.edu/ibe/search/ztf/products/sci?POS=180,5&SIZE=0&ct=csv Returns metadata for ZTF images at that position. Use the IBE API to download cutouts. Option D — IASC sample packages: IASC sometimes posts sample packages on their website for practice campaigns. Check: iasc.cosmosearch.org/Home/FAQ ### 5. HOW TO USE EACH APP TAB Tab "Processar Imagens IASC": The main competition tab. Upload your 4 FITS files \ here. Set obs code to F51. Results show detected tracklets, priority, velocity, \ arc length, RF/CNN scores. The MPC raw output is what you submit to IASC. Tab "Pipeline Simulator": Practice mode. Generates synthetic data with planted \ asteroids. Use this to understand what good detections look like before your \ first real campaign. Tab "MPC Formatter": Build a single MPC record manually. Useful for checking \ format compliance or correcting a record before submission. Tab "Tracklet Visualizer": Inspect the sky motion of a detected tracklet. \ Shows sky plane, ΔRA/ΔDec vs time, and residual bars. A good tracklet has \ linear motion and RMS < 1 arcsecond. Tab "Assistente IA": You are here. Ask me anything about the competition, \ pipeline outputs, or how to interpret results. ### 6. INTERPRETING PIPELINE RESULTS n_candidates = 0: No tracklets formed. Possible causes: - Frames not covering the same field (check RA/Dec in headers) - Too few frames (need ≥ 2, recommend 4) - All objects removed by SkyBoT (known asteroids) — this is correct behavior - Very sparse field with few sources n_confirmed = 0 but n_candidates > 0: Tracklets found but failed RF/CNN threshold. Possible causes: slow velocity below 0.01″/s cutoff, high residuals, \ not enough detections per tracklet. Good detection metrics: - Velocity: 0.1–3.0 ″/s (main belt: ~0.3–1.0, NEOs: 1–10) - Arc: ≥ 30 minutes (longer = better orbit determination) - RMS: < 1.0 arcsecond (linear motion quality) - Detections: ≥ 3 (minimum for reliable tracklet) - RF score: ≥ 0.7, CNN score: ≥ 0.9 ### 7. TRAINING THE CLASSIFIER The RF and CNN models ship untrained (heuristic fallback). To train them: Step 1 — Build a training dataset using the dataset_builder module: from asteroidnet.training.dataset_builder import build_training_dataset build_training_dataset( sky_fields=[(180.0, 5.0), (270.0, -15.0), (45.0, 10.0)], date_range=("2023-01-01", "2024-12-31"), output_path="training_data/dataset.npz", surveys=["ps1", "ztf"], ) This mines PS1/ZTF archives, uses SkyBoT to label known asteroids as positives, \ and random background cutouts as negatives. Takes 30–60 minutes. Step 2 — Train the Random Forest: import numpy as np from sklearn.ensemble import RandomForestClassifier import joblib data = np.load("training_data/dataset.npz") # RF uses kinematic features, not cutouts # Extract features from your tracklets and train rf = RandomForestClassifier(n_estimators=200, random_state=42) # rf.fit(X_train, y_train) joblib.dump(rf, "models/rf_classifier.pkl") Step 3 — Set model paths in config: ASTEROIDNET_CLASSIFIER__RF_MODEL_PATH=models/rf_classifier.pkl Until the model is trained, the pipeline uses smart heuristics (velocity range, \ SNR, residual quality) that already work well for competition use. ### 8. COMMON ERRORS AND FIXES "Need at least 2 FITS frames" → Upload more files. Need ≥ 2, recommend 4. "Pipeline error: No image data" → File may be corrupted or not a FITS image. \ Try opening with astropy: fits.open("file.fits")[0].data "No tracklet candidates" → Reduce detection threshold (default 3σ). Check that \ all 4 frames cover the same sky field. "Observatory code must be exactly 3 chars" → Use F51 (Pan-STARRS), 695 (Palomar), \ 500 (geocenter/generic), or your registered MPC code. Slow pipeline (>5 min) → Normal for first run (package imports). Subsequent runs \ are faster. The HF Space has limited CPU. ## Response style - Be concise and actionable during competition time pressure - When the user shares pipeline output, analyze it specifically - Use Portuguese when the user writes in Portuguese - For MPC record questions, always show the exact 80-column format - Never make up asteroid designations or MPC codes — always say "check at minorplanetcenter.net" - You have access to the most recent pipeline run context (provided below when available) """ def _build_context_block(ctx: dict) -> str: "Format the pipeline context for injection into the chat." if not ctx: return "" lines = ["\n--- LAST PIPELINE RUN ---"] lines.append("Frames ingested: " + str(ctx.get("n_frames", "?"))) lines.append("Tracklet candidates: " + str(ctx.get("n_candidates", "?"))) lines.append("Confirmed detections: " + str(ctx.get("n_confirmed", "?"))) lines.append("Elapsed: " + str(ctx.get("elapsed_s", "?")) + "s") lines.append("Run ID: " + str(ctx.get("run_id", "?"))) if ctx.get("warnings"): lines.append("Warnings: " + "; ".join(ctx["warnings"])) if ctx.get("detections"): lines.append("Detections:") for d in ctx["detections"][:10]: lines.append( " " + d.get("priority", "?") + " | vel=" + str(d.get("vel", "?")) + "″/s" + " | arc=" + str(d.get("arc_min", "?")) + "min" + " | RMS=" + str(d.get("rms", "?")) + "″" + " | RF=" + str(d.get("rf", "?")) + " | CNN=" + str(d.get("cnn", "?")) ) if ctx.get("mpc_records"): lines.append("MPC records generated: " + str(len(ctx["mpc_records"]))) lines.append("First record: " + ctx["mpc_records"][0]) lines.append("--- END PIPELINE RUN ---") return "\n".join(lines) def chat_with_claude(message: str, history: list, pipeline_ctx: dict) -> tuple: "Send a message to Claude with pipeline context injected." if not message.strip(): return history, "" api_key = os.environ.get("ANTHROPIC_API_KEY", "") if not api_key or not _HAS_ANTHROPIC: # Graceful fallback: show setup instructions reply = ( "⚠️ **API key not configured.**\n\n" "To enable the AI assistant:\n" "1. Go to your HuggingFace Space settings\n" "2. Click **Settings → Variables and secrets**\n" "3. Add a secret named `ANTHROPIC_API_KEY` with your Anthropic API key\n" "4. Restart the Space\n\n" "Get an API key at: [console.anthropic.com](https://console.anthropic.com)\n\n" "---\n" "**Running locally?** Set the environment variable:\n" "`export ANTHROPIC_API_KEY=sk-ant-...`\n" "then restart `python app.py`" ) history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": reply}) return history, "" # Build system prompt with optional pipeline context ctx_block = _build_context_block(pipeline_ctx) system = _SYSTEM_PROMPT if ctx_block: system = system + "\n\nCURRENT SESSION PIPELINE CONTEXT:" + ctx_block # Convert Gradio history to Anthropic messages format messages = [] for turn in history: if isinstance(turn, dict): messages.append({"role": turn["role"], "content": turn["content"]}) messages.append({"role": "user", "content": message}) try: client = _anthropic.Anthropic(api_key=api_key) response = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=1024, system=system, messages=messages, ) reply = response.content[0].text except Exception as exc: reply = "Error calling Claude API: " + str(exc)[:200] history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": reply}) return history, "" SUB_STYLE = ( 'style="font-family:monospace;font-size:.66rem;color:' + DIM + ';letter-spacing:.1em;text-transform:uppercase;padding:10px 0 2px"' ) with gr.Blocks(css=CSS, theme=gr.themes.Base()) as demo: gr.HTML(HEADER) # Shared pipeline context — updated on each real FITS run, injected into chatbot pipeline_ctx = gr.State({}) with gr.Tabs(): # ── Tab 1: REAL DATA ───────────────────────────────────────────────── with gr.Tab("⬡ Processar Imagens IASC"): gr.HTML("

Upload real FITS files from an IASC campaign package (4 frames recommended)

") with gr.Row(): with gr.Column(scale=1, min_width=280): iasc_files = gr.File( label="FITS Files (upload 4 frames)", file_count="multiple", file_types=[".fits", ".fit", ".fts", ".fits.gz"], ) iasc_obs = gr.Textbox(label="Observatory Code (3 chars)", value="F51", max_lines=1) iasc_survey = gr.Dropdown( label="Survey hint", choices=["auto", "ps1", "ztf", "generic"], value="auto", ) iasc_btn = gr.Button("▶ Run Pipeline on FITS", variant="primary") gr.HTML( '
' "Tip: IASC packages contain 4 FITS frames of the same field " "~30 min apart. Download them from iasc.cosmosearch.org after " "registering for a campaign.
" ) with gr.Column(scale=3): iasc_stats = gr.HTML() iasc_mpc_html = gr.HTML() iasc_mpc_raw = gr.Textbox( label="Raw MPC Records (copy for submission)", interactive=False, lines=6, ) iasc_btn.click( fn=process_iasc_fits, inputs=[iasc_files, iasc_obs, iasc_survey], outputs=[iasc_stats, iasc_mpc_html, iasc_mpc_raw, pipeline_ctx], api_name=False, ) # ── Tab 2: Simulator ───────────────────────────────────────────────── with gr.Tab("⬡ Pipeline Simulator"): gr.HTML("

Simulate full pipeline on synthetic FITS data

") with gr.Row(): with gr.Column(scale=1, min_width=240): sim_nfr = gr.Slider(4, 20, value=4, step=1, label="FITS Frames") sim_nas = gr.Slider(5, 50, value=10, step=1, label="Injected Asteroids") sim_sm = gr.Slider(3, 20, value=5, step=0.5, label="SNR Min") sim_sx = gr.Slider(5, 50, value=25, step=1, label="SNR Max") sim_vm = gr.Slider(0.01, 2, value=0.1, step=0.01, label="Vel min (″/s)") sim_vx = gr.Slider(0.5, 10, value=5.0, step=0.1, label="Vel max (″/s)") sim_dt = gr.Slider(2.5, 8, value=3.0, step=0.5, label="Detection threshold (σ)") sim_btn = gr.Button("▶ Run Simulation", variant="primary") with gr.Column(scale=3): sim_out = gr.HTML() sim_btn.click( fn=run_simulation, inputs=[sim_nfr, sim_nas, sim_sm, sim_sx, sim_vm, sim_vx, sim_dt], outputs=[sim_out], api_name=False, ) # ── Tab 3: MPC Formatter ───────────────────────────────────────────── with gr.Tab("⬡ MPC Formatter"): gr.HTML("

Generate a Minor Planet Center 80-column astrometric record

") with gr.Row(): with gr.Column(scale=1): mpc_d = gr.Textbox(label="Provisional Designation", value="2026 AA1") mpc_ra = gr.Number(label="RA (decimal °)", value=180.0) mpc_dc = gr.Number(label="Dec (decimal °)", value=5.234) with gr.Row(): mpc_yr = gr.Number(label="Year", value=2026, precision=0) mpc_mo = gr.Number(label="Month", value=3, precision=0) mpc_dy = gr.Number(label="Day (DD.ddddd)", value=20.50000) mpc_mg = gr.Number(label="Magnitude", value=18.5) mpc_bd = gr.Textbox(label="Filter Band", value="R", max_lines=1) mpc_oc = gr.Textbox(label="Observatory Code", value="F51", max_lines=1) mpc_bt = gr.Button("Generate MPC Record", variant="primary") with gr.Column(scale=2): mpc_out = gr.HTML() mpc_raw = gr.Textbox(label="Raw 80-column line", interactive=False, lines=2) mpc_bt.click( fn=format_mpc, inputs=[mpc_d, mpc_ra, mpc_dc, mpc_yr, mpc_mo, mpc_dy, mpc_mg, mpc_bd, mpc_oc], outputs=[mpc_out, mpc_raw], api_name=False, ) # ── Tab 4: Tracklet Visualizer ─────────────────────────────────────── with gr.Tab("⬡ Tracklet Visualizer"): gr.HTML("

Inspect multi-frame tracklet motion and linear residuals

") with gr.Row(): with gr.Column(scale=1): t_n = gr.Slider(3, 10, value=5, step=1, label="Detections") t_v = gr.Slider(0.01, 8, value=0.5, step=0.01, label="Velocity (″/s)") t_pa = gr.Slider(0, 360, value=135, step=1, label="Position Angle (°)") t_sp = gr.Slider(30, 240, value=90, step=5, label="Time Span (min)") t_sn = gr.Slider(3, 30, value=10, step=0.5, label="SNR") t_uc = gr.Checkbox(label="Show position uncertainties", value=True) t_bt = gr.Button("Plot Tracklet", variant="primary") with gr.Column(scale=3): t_img = gr.HTML() t_st = gr.HTML() t_bt.click( fn=visualise_tracklet, inputs=[t_n, t_v, t_pa, t_sp, t_sn, t_uc], outputs=[t_img, t_st], api_name=False, ) # ── Tab 5: Assistente IA ───────────────────────────────────────────── with gr.Tab("⬡ Assistente IA"): gr.HTML( "

Co-piloto de IA para a competição Caça Asteroides · " "conhece o pipeline, o fluxo IASC e os dados MPC

" ) gr.HTML( '
' "Dica: após rodar o pipeline na aba Processar Imagens IASC, " "volte aqui e pergunte sobre os resultados — o assistente já tem contexto " "da sua última execução.
" ) with gr.Row(): with gr.Column(scale=3): chatbot_ui = gr.Chatbot( label="", height=520, type="messages", placeholder=( "Olá! Sou o assistente AsteroidNET.\n\n" "Posso te ajudar com:\n" "• Fluxo completo da competição Caça Asteroides\n" "• Interpretar os resultados do pipeline\n" "• Entender os registros MPC gerados\n" "• Onde encontrar imagens FITS de teste\n" "• Como treinar os classificadores RF/CNN\n\n" "Pergunte em português ou inglês!" ), avatar_images=(None, "https://huggingface.co/front/assets/huggingface_logo-noborder.svg"), bubble_full_width=False, ) with gr.Row(): chat_input = gr.Textbox( placeholder="Escreva sua pergunta aqui...", show_label=False, lines=2, scale=5, container=False, ) chat_send = gr.Button("Enviar ➤", variant="primary", scale=1, min_width=100) chat_clear = gr.Button("🗑 Limpar conversa", variant="secondary", size="sm") with gr.Column(scale=1, min_width=220): gr.HTML( '
' + '

Perguntas rápidas

' + '

Clique para perguntar:

' ) quick_q1 = gr.Button("Como funciona a competição?", size="sm", variant="secondary") quick_q2 = gr.Button("O que é um registro MPC?", size="sm", variant="secondary") quick_q3 = gr.Button("Onde baixar imagens FITS?", size="sm", variant="secondary") quick_q4 = gr.Button("Como interpretar os resultados?",size="sm", variant="secondary") quick_q5 = gr.Button("Como treinar o classificador?", size="sm", variant="secondary") quick_q6 = gr.Button("Zero detecções — o que fazer?", size="sm", variant="secondary") quick_q7 = gr.Button("What is HAZARDOUS priority?", size="sm", variant="secondary") quick_q8 = gr.Button("Como submeter ao IASC?", size="sm", variant="secondary") def send_message(msg, hist, ctx): return chat_with_claude(msg, hist or [], ctx) def quick_ask(question, hist, ctx): return chat_with_claude(question, hist or [], ctx) def clear_chat(): return [] chat_send.click( fn=send_message, inputs=[chat_input, chatbot_ui, pipeline_ctx], outputs=[chatbot_ui, chat_input], api_name=False, ) chat_input.submit( fn=send_message, inputs=[chat_input, chatbot_ui, pipeline_ctx], outputs=[chatbot_ui, chat_input], api_name=False, ) chat_clear.click(fn=clear_chat, outputs=[chatbot_ui], api_name=False) for qbtn, qtxt in [ (quick_q1, "Como funciona a competição Caça Asteroides?"), (quick_q2, "O que é um registro MPC e como lê-lo?"), (quick_q3, "Onde posso baixar imagens FITS para testar o pipeline?"), (quick_q4, "Como interpreto os resultados do pipeline? O que significam RF, CNN, velocity e RMS?"), (quick_q5, "Como treino o classificador RF e CNN com dados reais?"), (quick_q6, "O pipeline retornou zero detecções. O que pode estar errado?"), (quick_q7, "What does HAZARDOUS priority mean and what should I do with it?"), (quick_q8, "Como submeto os registros MPC ao IASC para obter medalhas?"), ]: qbtn.click( fn=lambda h, c, q=qtxt: quick_ask(q, h, c), inputs=[chatbot_ui, pipeline_ctx], outputs=[chatbot_ui, chat_input], api_name=False, ) # ── Tab 6: About ───────────────────────────────────────────────────── with gr.Tab("⬡ About"): gr.HTML("""

AsteroidNET v0.2

Automated NEO Detection • Dr. Matheus Machado Rech

What is new in v0.2

How to use with IASC

  1. Register at iasc.cosmosearch.org
  2. Download a campaign FITS package (4 frames, ~30 min cadence)
  3. Upload all 4 .fits files in the Processar Imagens IASC tab
  4. Enter your observatory code (F51 for Pan-STARRS; 500 for generic)
  5. Click Run Pipeline — MPC records generated automatically
  6. Submit records to IASC for verification and MPC submission
""") gr.HTML( '
' "AsteroidNET · Dr. Matheus Machado Rech · github.com/mmrech/asteroidnet
" ) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True)