Upload app.py with huggingface_hub

app.py

@@ -0,0 +1,267 @@
+
+import gradio as gr
+import cv2, math, json, requests, numpy as np
+from PIL import Image
+from ultralytics import YOLO
+
+MODEL = YOLO("model/best.pt")
+
+# ── API keys — set in HuggingFace Space Secrets ──
+K2_KEY  = "REPLACE_K2_KEY"
+KT_KEY  = "REPLACE_KEEPTRACK_KEY"
+K2_URL  = "https://api.k2think.ai/v1/chat/completions"
+K2_MDL  = "MBZUAI-IFM/K2-Think-v2"
+KT_BASE = "https://api.keeptrack.space/v4"
+
+# ── KeepTrack helpers (verified endpoints only) ───────────────────────
+def kt_get(path, timeout=15):
+    try:
+        r = requests.get(f"{KT_BASE}{path}",
+                         headers={"X-API-Key": KT_KEY}, timeout=timeout)
+        r.raise_for_status()
+        return r.json()
+    except:
+        return None
+
+def get_sat_lla(norad_id=25544):
+    r = kt_get(f"/sat/{norad_id}/lla")
+    return r if isinstance(r, dict) else {}
+
+def get_sat_eci(norad_id=25544):
+    r = kt_get(f"/sat/{norad_id}/eci")
+    return r if isinstance(r, dict) else {}
+
+def get_sat_rae(norad_id=25544, lat=28.61, lon=77.20, alt=0.2):
+    r = kt_get(f"/sat/{norad_id}/rae/{lat}/{lon}/{alt}")
+    return r if isinstance(r, dict) else {}
+
+def get_sat_metadata(norad_id=25544):
+    r = kt_get(f"/sat/{norad_id}")
+    return r if isinstance(r, dict) else {}
+
+def get_debris_count():
+    r = kt_get("/metrics/debris/count")
+    if isinstance(r, dict): return str(r.get("count", "N/A"))
+    return str(r) if r else "N/A"
+
+def get_active_count():
+    r = kt_get("/metrics/active/count")
+    if isinstance(r, dict): return str(r.get("count", "N/A"))
+    return str(r) if r else "N/A"
+
+def get_socrates():
+    r = kt_get("/socrates/latest", timeout=25)
+    if isinstance(r, list):   return r[:5]
+    if isinstance(r, dict):
+        return (r.get("conjunctions") or r.get("data") or [])[:5]
+    return []
+
+def parse_socrates_entry(c):
+    if not isinstance(c, dict): return {}
+    return {
+        "sat1Name": c.get("sat1Name") or c.get("OBJECT1_NAME") or c.get("sat1") or "?",
+        "sat2Name": c.get("sat2Name") or c.get("OBJECT2_NAME") or c.get("sat2") or "?",
+        "minRng":   c.get("minRng")   or c.get("MISS_DISTANCE") or c.get("missDistance") or "?",
+        "tcaTime":  c.get("tcaTime")  or c.get("TCA") or c.get("tca") or "?",
+    }
+
+def extract_meta(raw):
+    if not raw: return {}
+    return {
+        "name":    raw.get("SATNAME") or raw.get("name") or raw.get("satname") or "N/A",
+        "country": raw.get("COUNTRY") or raw.get("country") or "N/A",
+        "type":    raw.get("OBJECT_TYPE") or raw.get("objectType") or "N/A",
+    }
+
+def fmt(v, dp=2):
+    try:    return f"{float(v):.{dp}f}"
+    except: return str(v)
+
+# ── Zones ─────────────────────────────────────────────────────────────
+ZONE_MAP = {
+    (0,0):"TOP-LEFT",(1,0):"TOP-CENTER",(2,0):"TOP-RIGHT",
+    (0,1):"MID-LEFT",(1,1):"CENTER",(2,1):"MID-RIGHT",
+    (0,2):"BOT-LEFT",(1,2):"BOT-CENTER",(2,2):"BOT-RIGHT",
+}
+ZONE_CMDS = {
+    "TOP-LEFT":  ["MOVE RIGHT","MOVE DOWN"],
+    "TOP-CENTER":["MOVE DOWN","ADJUST PITCH DOWN"],
+    "TOP-RIGHT": ["MOVE LEFT","MOVE DOWN"],
+    "MID-LEFT":  ["MOVE RIGHT","ADJUST THRUST VECTOR RIGHT"],
+    "CENTER":    ["EMERGENCY REVERSE THRUST","RAPID EVASIVE MANEUVER"],
+    "MID-RIGHT": ["MOVE LEFT","ADJUST THRUST VECTOR LEFT"],
+    "BOT-LEFT":  ["MOVE RIGHT","MOVE UP"],
+    "BOT-CENTER":["MOVE UP","ADJUST PITCH UP"],
+    "BOT-RIGHT": ["MOVE LEFT","MOVE UP"],
+}
+SPEED = {"CRITICAL":"REDUCE SPEED IMMEDIATELY","HIGH":"REDUCE SPEED",
+         "MEDIUM":"MODERATE SPEED","LOW":"MAINTAIN SPEED"}
+COLORS_BGR = {"CRITICAL":(0,0,255),"HIGH":(0,100,255),"MEDIUM":(0,200,255),"LOW":(0,255,120)}
+
+def zone_fn(xc,yc,fw,fh):
+    c=0 if xc<fw/3 else(1 if xc<2*fw/3 else 2)
+    r=0 if yc<fh/3 else(1 if yc<2*fh/3 else 2)
+    return ZONE_MAP[(c,r)]
+
+def analyse(boxes,fw,fh):
+    cx,cy=fw/2,fh/2; objs=[]
+    for box in boxes:
+        x1,y1,x2,y2=box[:4]; conf=float(box[4]) if len(box)>4 else 1.0
+        xc,yc2=(x1+x2)/2,(y1+y2)/2
+        dist=math.hypot(xc-cx,yc2-cy); size=(x2-x1)*(y2-y1)
+        risk=(size/(dist+1))*conf
+        lvl="CRITICAL" if risk>800 else "HIGH" if risk>300 else "MEDIUM" if risk>80 else "LOW"
+        objs.append({"box":(int(x1),int(y1),int(x2),int(y2)),
+                     "zone":zone_fn(xc,yc2,fw,fh),"risk":round(risk,3),
+                     "level":lvl,"conf":round(conf,3),"dist":round(dist,1)})
+    return sorted(objs,key=lambda x:x["risk"],reverse=True)
+
+def decide(objs):
+    if not objs: return {"status":"STABLE","cmds":["MAINTAIN CURRENT ORBIT"],"level":"LOW"}
+    t=objs[0]; lvl=t["level"]
+    cmds=list(ZONE_CMDS.get(t["zone"],["HOLD POSITION"]))+[SPEED[lvl]]
+    if lvl=="CRITICAL": cmds.insert(0,"COLLISION ALERT")
+    return {"status":{"CRITICAL":"EMERGENCY","HIGH":"DANGER","MEDIUM":"CAUTION","LOW":"MONITOR"}[lvl],
+            "cmds":cmds,"level":lvl,"target":t,"all":objs}
+
+def k2_brief(dec, lla, eci, rae, socrates, debris_count, active_count, meta, k2_key):
+    if k2_key in ("REPLACE_K2_KEY",""):
+        return ""
+    soc_str = "; ".join(
+        f"{c.get('sat1Name','?')} -- {c.get('sat2Name','?')} miss:{c.get('minRng','?')}"
+        for c in socrates[:3]) or "none"
+    prompt=(
+        f"Satellite: {meta.get('name','Unknown')} ({meta.get('type','?')})\n"
+        f"LLA: lat={fmt(lla.get('lat','?'))}deg lon={fmt(lla.get('lon','?'))}deg "
+        f"alt={fmt(lla.get('alt','?'))}km\n"
+        f"ECI: x={eci.get('x','?')} y={eci.get('y','?')} z={eci.get('z','?')} km\n"
+        f"Vision status:{dec['status']} | Debris detected:{len(dec.get('all',[]))} | "
+        f"Zone:{dec.get('target',{}).get('zone','N/A')}\n"
+        f"Active sats:{active_count} | Total orbital debris:{debris_count}\n"
+        f"SOCRATES conjunctions: {soc_str}\n"
+        f"Commands:{dec['cmds']}\n"
+        "Write MISSION BRIEF (3 sentences) + RISK ASSESSMENT + FINAL INSTRUCTIONS (-> format). "
+        "Navigation only, no robotic arms."
+    )
+    try:
+        r=requests.post(K2_URL,
+            headers={"Authorization":f"Bearer {k2_key}","Content-Type":"application/json"},
+            json={"model":K2_MDL,"messages":[{"role":"user","content":prompt}],"stream":False},
+            timeout=30)
+        r.raise_for_status()
+        return r.json()["choices"][0]["message"]["content"]
+    except Exception as e:
+        return f"[K2 unavailable: {e}]"
+
+def render(frame,objs,dec,lla,socrates):
+    img=frame.copy(); h,w=img.shape[:2]
+    for o in objs:
+        x1,y1,x2,y2=o["box"]; col=COLORS_BGR[o["level"]]
+        cv2.rectangle(img,(x1,y1),(x2,y2),col,2)
+        cv2.putText(img,f"{o['level']}|{o['zone']}|{o['conf']:.2f}",
+                    (x1,max(y1-5,10)),cv2.FONT_HERSHEY_SIMPLEX,0.35,col,1)
+    ov=img.copy(); cv2.rectangle(ov,(0,0),(w,130),(8,8,25),-1)
+    img=cv2.addWeighted(ov,0.75,img,0.25,0)
+    cv2.putText(img,f"STATUS: {dec['status']}",(10,22),cv2.FONT_HERSHEY_SIMPLEX,0.62,(255,255,100),2)
+    cv2.putText(img,f"DEBRIS: {len(objs)}",(10,42),cv2.FONT_HERSHEY_SIMPLEX,0.5,(180,220,255),1)
+    if socrates:
+        cv2.putText(img,f"SOCRATES: {len(socrates)} conjunction(s)",(10,60),
+                    cv2.FONT_HERSHEY_SIMPLEX,0.42,(0,80,255),1)
+    if lla:
+        cv2.putText(img,
+            f"SAT: lat={fmt(lla.get('lat','?'))}deg lon={fmt(lla.get('lon','?'))}deg alt={fmt(lla.get('alt','?'))}km",
+            (10,78),cv2.FONT_HERSHEY_SIMPLEX,0.38,(150,200,255),1)
+    y=98
+    for c in dec["cmds"][:3]:
+        cv2.putText(img,f"  -> {c}",(10,y),cv2.FONT_HERSHEY_SIMPLEX,0.4,(100,255,180),1); y+=16
+    return img
+
+def process(input_img, norad_id, use_k2, use_kt):
+    norad_id = int(str(norad_id)) if str(norad_id).isdigit() else 25544
+    frame = cv2.cvtColor(np.array(input_img), cv2.COLOR_RGB2BGR)
+    res   = MODEL(frame, verbose=False)
+    boxes = res[0].boxes
+    raw   = []
+    if boxes and len(boxes):
+        xy=boxes.xyxy.cpu().numpy(); cf=boxes.conf.cpu().numpy()
+        raw=[(*xy[i],cf[i]) for i in range(len(xy))]
+    objs = analyse(raw, frame.shape[1], frame.shape[0])
+    dec  = decide(objs)
+
+    lla={}; eci={}; rae={}; meta={}
+    socrates=[]; debris_count="N/A"; active_count="N/A"
+
+    if use_kt and KT_KEY not in ("REPLACE_KEEPTRACK_KEY",""):
+        lla          = get_sat_lla(norad_id)
+        eci          = get_sat_eci(norad_id)
+        rae          = get_sat_rae(norad_id)
+        meta         = extract_meta(get_sat_metadata(norad_id))
+        debris_count = get_debris_count()
+        active_count = get_active_count()
+        socrates     = [parse_socrates_entry(c) for c in get_socrates()]
+        # Escalate if SOCRATES has close approaches
+        if socrates and dec["level"] != "CRITICAL":
+            dec["cmds"].insert(0,"SOCRATES CONJUNCTION WARNING")
+
+    brief = k2_brief(dec,lla,eci,rae,socrates,debris_count,active_count,meta,K2_KEY) if use_k2 else ""
+    ann   = render(frame, objs, dec, lla, socrates)
+    pil   = Image.fromarray(cv2.cvtColor(ann, cv2.COLOR_BGR2RGB))
+
+    lines = [
+        f"Debris Detected:   {len(objs)}",
+        f"Status:            {dec['status']}",
+        f"Risk Level:        {dec['level']}",
+    ]
+    if dec.get("target"):
+        t=dec["target"]
+        lines.append(f"Highest Risk:      {t['zone']}  [{t['level']}]  conf:{t['conf']}")
+    if meta.get("name") and meta["name"] != "N/A":
+        lines.append(f"Satellite:         {meta['name']} | {meta.get('country','?')} | {meta.get('type','?')}")
+    if lla:
+        lines.append(f"LLA Position:      lat:{fmt(lla.get('lat','?'))}deg "
+                     f"lon:{fmt(lla.get('lon','?'))}deg alt:{fmt(lla.get('alt','?'))}km")
+    if eci:
+        lines.append(f"ECI Position:      x:{eci.get('x','?')} y:{eci.get('y','?')} z:{eci.get('z','?')} km")
+    if rae:
+        lines.append(f"Ground RAE:        az:{fmt(rae.get('az','?'))}deg el:{fmt(rae.get('el','?'))}deg")
+    if debris_count != "N/A":
+        lines.append(f"Total Orbital Debris: {debris_count}")
+    if active_count != "N/A":
+        lines.append(f"Active Satellites:    {active_count}")
+    if socrates:
+        lines.append(f"SOCRATES Alerts:   {len(socrates)} conjunction(s)")
+        for s in socrates[:2]:
+            lines.append(f"  -> {s.get('sat1Name','?')} <-> {s.get('sat2Name','?')}  miss:{s.get('minRng','?')}km")
+    lines += ["", "Navigation Instructions:"]
+    for c in dec["cmds"]:
+        lines.append(f"  -> {c}")
+    if brief:
+        lines += ["", "=== K2 Think V2 Mission Brief ===", brief]
+    return pil, "\n".join(lines)
+
+with gr.Blocks(title="Satellite Debris Avoidance", theme=gr.themes.Monochrome()) as demo:
+    gr.Markdown("# Satellite Autonomous Debris Avoidance System v2.1")
+    gr.Markdown(
+        "YOLOv8m vision + KeepTrack API (LLA, ECI, RAE, SOCRATES, metrics) + K2 Think V2 AI\n\n"
+        "**Output:** navigation commands only — MOVE LEFT/RIGHT/UP/DOWN, ADJUST THRUST, REDUCE SPEED"
+    )
+    with gr.Row():
+        with gr.Column(scale=1):
+            inp    = gr.Image(type="pil", label="Input Frame")
+            norad  = gr.Textbox(value="25544", label="NORAD ID")
+            use_k2 = gr.Checkbox(label="Use K2 Think V2 AI Brief", value=False)
+            use_kt = gr.Checkbox(label="Use KeepTrack API (live orbital data)", value=True)
+            btn    = gr.Button("Analyse Frame", variant="primary")
+        with gr.Column(scale=1):
+            out_img = gr.Image(label="Annotated Output")
+            out_txt = gr.Textbox(label="Navigation Decision + Orbital Data", lines=22)
+    btn.click(process, inputs=[inp,norad,use_k2,use_kt], outputs=[out_img,out_txt])
+    gr.Markdown("""
+### KeepTrack Endpoints Used (verified working)
+`/sat/{id}/lla` · `/sat/{id}/eci` · `/sat/{id}/rae/{lat}/{lon}/{alt}` · `/sat/{id}` ·
+`/socrates/latest` · `/metrics/debris/count` · `/metrics/active/count`
+### CelesTrak Debris Groups
+Cosmos-1408 · Fengyun-1C · Iridium-33 · Cosmos-2251 · Active satellites
+""")
+
+demo.launch()