Spaces:
Sleeping
Sleeping
| # ───────────────────────────────────────────────────────────────────────────── | |
| # TRINETRA - Database Layer v2 | |
| # Queries planet_candidates_v2 — two-layer architecture enforced | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| import os | |
| from supabase import create_client, Client | |
| from typing import Optional, List, Dict, Any | |
| class TrinetraDB: | |
| def __init__(self): | |
| self._client: Optional[Client] = None | |
| def _get_client(self) -> Client: | |
| if self._client is None: | |
| url = os.environ.get("SUPABASE_URL", "").strip() | |
| key = os.environ.get("SUPABASE_KEY", "").strip() | |
| if not url or not key: | |
| raise ValueError("SUPABASE_URL and SUPABASE_KEY not set.") | |
| self._client = create_client(url, key) | |
| return self._client | |
| async def ping(self) -> bool: | |
| try: | |
| self._get_client().table("planet_candidates_v2").select("id").limit(1).execute() | |
| return True | |
| except Exception: | |
| return False | |
| # ── planet_candidates_v2 ────────────────────────────────────────────────── | |
| async def get_all_planets(self) -> List[Dict[str, Any]]: | |
| try: | |
| result = self._get_client().table("planet_candidates_v2").select("*").execute() | |
| return result.data or [] | |
| except Exception: | |
| return [] | |
| async def get_planet_by_id(self, star_id: str) -> Optional[Dict[str, Any]]: | |
| try: | |
| # Try planet_name first (Kepler-22b), then star_id (Kepler-22) | |
| result = ( | |
| self._get_client().table("planet_candidates_v2") | |
| .select("*") | |
| .eq("planet_name", star_id) | |
| .execute() | |
| ) | |
| if result.data: | |
| return result.data[0] | |
| # Fallback to star_id | |
| result = ( | |
| self._get_client().table("planet_candidates_v2") | |
| .select("*") | |
| .eq("star_id", star_id) | |
| .execute() | |
| ) | |
| return result.data[0] if result.data else None | |
| except Exception: | |
| return None | |
| async def get_planets_by_tier(self, tier: str) -> List[Dict[str, Any]]: | |
| try: | |
| result = ( | |
| self._get_client().table("planet_candidates_v2") | |
| .select("*") | |
| .eq("habitability_tier", tier) | |
| .execute() | |
| ) | |
| return result.data or [] | |
| except Exception: | |
| return [] | |
| async def get_planets_by_detection_status(self, status: str) -> List[Dict[str, Any]]: | |
| try: | |
| result = ( | |
| self._get_client().table("planet_candidates_v2") | |
| .select("*") | |
| .eq("detection_status", status) | |
| .execute() | |
| ) | |
| return result.data or [] | |
| except Exception: | |
| return [] | |
| async def get_validated_planets(self) -> List[Dict[str, Any]]: | |
| """Returns only planets that passed the detection gate.""" | |
| try: | |
| result = ( | |
| self._get_client().table("planet_candidates_v2") | |
| .select("*") | |
| .neq("detection_status", "FALSE_POSITIVE") | |
| .neq("habitability_tier", "NOT_EVALUATED") | |
| .execute() | |
| ) | |
| return result.data or [] | |
| except Exception: | |
| return [] | |
| # ── processed_stars ────────────────────────────────────────────────────── | |
| async def get_all_stars(self) -> List[Dict[str, Any]]: | |
| try: | |
| result = self._get_client().table("processed_stars").select("*").execute() | |
| return result.data or [] | |
| except Exception: | |
| return [] | |
| async def get_star_by_id(self, star_id: str) -> Optional[Dict[str, Any]]: | |
| try: | |
| result = ( | |
| self._get_client().table("processed_stars") | |
| .select("*") | |
| .eq("star_id", star_id) | |
| .execute() | |
| ) | |
| return result.data[0] if result.data else None | |
| except Exception: | |
| return None | |
| # ── tce_candidates ─────────────────────────────────────────────────────── | |
| async def get_all_tce(self) -> List[Dict[str, Any]]: | |
| try: | |
| result = self._get_client().table("tce_candidates").select("*").execute() | |
| return result.data or [] | |
| except Exception: | |
| return [] | |
| async def get_tce_by_id(self, star_id: str) -> Optional[Dict[str, Any]]: | |
| try: | |
| result = ( | |
| self._get_client().table("tce_candidates") | |
| .select("*") | |
| .eq("star_id", star_id) | |
| .execute() | |
| ) | |
| return result.data[0] if result.data else None | |
| except Exception: | |
| return None | |
| db = TrinetraDB() | |