""" Populates the species_detail table from static trait CSV downloads. Required files in wildlife/trait_data/: - PanTHERIA_1-0_WR05_Aug2008.txt (mammals) https://ecologicaldata.org/wiki/pantheria - EltonTraits_BirdFuncDat.txt (birds) https://figshare.com/articles/dataset/EltonTraits_1_0/949799 - EltonTraits_MamFuncDat.txt (mammals) same Figshare link above - AmphiBIO_v1.csv (amphibians) https://figshare.com/articles/dataset/AmphiBIO_v1/4644424 Place all files in: wildlife/trait_data/ """ import csv import sqlite3 from pathlib import Path DB_PATH = Path(__file__).parent.parent / "wilddex.db" TRAIT_DATA_DIR = Path(__file__).parent.parent / "wildlife" / "trait_data" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def get_species_map(conn: sqlite3.Connection) -> dict[str, int]: """Return scientific_name -> species_id for all rows in the species table.""" rows = conn.execute("SELECT species_id, scientific_name FROM species").fetchall() return {sci: sid for sid, sci in rows} def upsert_detail(conn: sqlite3.Connection, species_id: int, traits: dict) -> None: """Insert or update a species_detail row, skipping None values.""" traits = {k: v for k, v in traits.items() if v is not None} if not traits: return existing = conn.execute( "SELECT detail_id FROM species_detail WHERE species_id = ?", (species_id,) ).fetchone() if existing: sets = ", ".join(f"{col} = ?" for col in traits) conn.execute( f"UPDATE species_detail SET {sets} WHERE species_id = ?", (*traits.values(), species_id), ) else: cols = "species_id, " + ", ".join(traits.keys()) placeholders = ", ".join("?" * (len(traits) + 1)) conn.execute( f"INSERT INTO species_detail ({cols}) VALUES ({placeholders})", (species_id, *traits.values()), ) # --------------------------------------------------------------------------- # PanTHERIA — mammals # --------------------------------------------------------------------------- def import_pantheria(conn: sqlite3.Connection, species_map: dict[str, int]) -> None: path = TRAIT_DATA_DIR / "PanTHERIA_1-0_WR05_Aug2008.txt" if not path.exists(): print(f" Skipping PanTHERIA (file not found: {path})") return matched = 0 with open(path, newline="", encoding="utf-8") as f: reader = csv.DictReader(f, delimiter="\t") for row in reader: sci_name = row.get("MSW05_Binomial", "").strip() species_id = species_map.get(sci_name) if not species_id: continue mass_g = _float(row.get("5-1_AdultBodyMass_g")) longevity_m = _float(row.get("17-1_MaxLongevity_m")) traits = { "weight_range": f"{mass_g:g} g" if mass_g else None, "lifespan": f"{longevity_m / 12:.1f} years" if longevity_m else None, } upsert_detail(conn, species_id, traits) matched += 1 conn.commit() print(f" PanTHERIA: {matched} species matched") # --------------------------------------------------------------------------- # EltonTraits — birds and mammals # --------------------------------------------------------------------------- ELTON_DIET_COLS = [ ("Diet.Inv", "invertebrates"), ("Diet.Vend", "vertebrates (endotherm)"), ("Diet.Vect", "vertebrates (ectotherm)"), ("Diet.Vfish", "fish"), ("Diet.Vunk", "vertebrates (unknown)"), ("Diet.Scav", "scavenging"), ("Diet.Fruit", "fruit"), ("Diet.Nect", "nectar"), ("Diet.Seed", "seeds"), ("Diet.PlantO","plant matter"), ] def _elton_diet(row: dict) -> str | None: parts = [] for col, label in ELTON_DIET_COLS: val = _float(row.get(col)) if val and val > 10: # only include if >10% of diet parts.append(f"{label} ({val:.0f}%)") return ", ".join(parts) if parts else None def _elton_activity(row: dict) -> str | None: val = _float(row.get("Nocturnal")) if val is None: return None return "nocturnal" if val == 1 else "diurnal" def _import_elton_file(conn: sqlite3.Connection, species_map: dict[str, int], path: Path, sci_col: str) -> int: matched = 0 with open(path, newline="", encoding="latin-1") as f: reader = csv.DictReader(f, delimiter="\t") for row in reader: sci_name = row.get(sci_col, "").strip().replace("_", " ") species_id = species_map.get(sci_name) if not species_id: continue mass_g = _float(row.get("BodyMass.Value")) traits = { "weight_range": f"{mass_g:g} g" if mass_g else None, "diet": _elton_diet(row), "activity_period": _elton_activity(row), } upsert_detail(conn, species_id, traits) matched += 1 return matched def import_elton(conn: sqlite3.Connection, species_map: dict[str, int]) -> None: bird_path = TRAIT_DATA_DIR / "EltonTraits_BirdFuncDat.txt" mam_path = TRAIT_DATA_DIR / "EltonTraits_MamFuncDat.txt" total = 0 if bird_path.exists(): n = _import_elton_file(conn, species_map, bird_path, sci_col="Scientific") total += n print(f" EltonTraits birds: {n} species matched") else: print(f" Skipping EltonTraits birds (file not found)") if mam_path.exists(): n = _import_elton_file(conn, species_map, mam_path, sci_col="Scientific") total += n print(f" EltonTraits mammals: {n} species matched") else: print(f" Skipping EltonTraits mammals (file not found)") conn.commit() # --------------------------------------------------------------------------- # AmphiBIO — amphibians # --------------------------------------------------------------------------- def import_amphibio(conn: sqlite3.Connection, species_map: dict[str, int]) -> None: path = TRAIT_DATA_DIR / "AmphiBIO_v1.csv" if not path.exists(): print(f" Skipping AmphiBIO (file not found: {path})") return matched = 0 with open(path, newline="", encoding="latin-1") as f: reader = csv.DictReader(f) for row in reader: sci_name = row.get("Species", "").strip() species_id = species_map.get(sci_name) if not species_id: continue size_min = _float(row.get("Body_size_mm")) longevity = _float(row.get("Longevity_max_y")) diet = row.get("Prey") or None habitat = row.get("Habitat") or None activity = row.get("Diel_activity") or None traits = { "size_range": f"{size_min:g} mm" if size_min else None, "lifespan": f"{longevity:g} years" if longevity else None, "diet": diet, "habitat_type": habitat, "activity_period": activity, } upsert_detail(conn, species_id, traits) matched += 1 conn.commit() print(f" AmphiBIO: {matched} species matched") # --------------------------------------------------------------------------- # Utilities # --------------------------------------------------------------------------- def _float(val) -> float | None: try: f = float(val) return None if f < 0 else f # datasets use -999 for missing except (TypeError, ValueError): return None # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- if __name__ == "__main__": if not TRAIT_DATA_DIR.exists(): TRAIT_DATA_DIR.mkdir(parents=True) print(f"Created {TRAIT_DATA_DIR} — place trait CSVs here and re-run.") else: with sqlite3.connect(DB_PATH) as conn: species_map = get_species_map(conn) print(f"Loaded {len(species_map)} species from DB\n") import_pantheria(conn, species_map) import_elton(conn, species_map) import_amphibio(conn, species_map) print("\nDone.")