Scientific Marine Ecology

2. WoRMS 分類学検索

WORMS_BASE = "https://www.marinespecies.org/rest"


def worms_taxon_search(name, fuzzy=True, marine_only=True):
    """
    WoRMS — 海洋生物分類学的検索。

    Parameters:
        name: str — 種名/属名
        fuzzy: bool — ファジー検索
        marine_only: bool — 海洋種のみ
    """
    url = f"{WORMS_BASE}/AphiaRecordsByName/{name}"
    params = {
        "like": str(fuzzy).lower(),
        "marine_only": str(marine_only).lower(),
    }
    resp = requests.get(url, params=params, timeout=30)
    resp.raise_for_status()
    data = resp.json()

    results = []
    for taxon in data if isinstance(data, list) else [data]:
        results.append({
            "aphia_id": taxon.get("AphiaID", ""),
            "scientific_name": taxon.get("scientificname", ""),
            "authority": taxon.get("authority", ""),
            "status": taxon.get("status", ""),
            "rank": taxon.get("rank", ""),
            "valid_name": taxon.get("valid_name", ""),
            "kingdom": taxon.get("kingdom", ""),
            "phylum": taxon.get("phylum", ""),
            "class": taxon.get("class", ""),
            "order": taxon.get("order", ""),
            "family": taxon.get("family", ""),
            "genus": taxon.get("genus", ""),
            "is_marine": taxon.get("isMarine", 0),
            "is_brackish": taxon.get("isBrackish", 0),
            "is_freshwater": taxon.get("isFreshwater", 0),
        })

    df = pd.DataFrame(results)
    print(f"WoRMS: '{name}' → {len(df)} taxa")
    return df


def worms_classification(aphia_id):
    """
    WoRMS — 完全分類階層取得。

    Parameters:
        aphia_id: int — AphiaID
    """
    url = f"{WORMS_BASE}/AphiaClassificationByAphiaID/{aphia_id}"
    resp = requests.get(url, timeout=30)
    resp.raise_for_status()
    data = resp.json()

    hierarchy = []
    node = data
    while node:
        hierarchy.append({
            "aphia_id": node.get("AphiaID", ""),
            "name": node.get("scientificname", ""),
            "rank": node.get("rank", ""),
        })
        node = node.get("child")

    df = pd.DataFrame(hierarchy)
    print(f"WoRMS classification: {len(df)} levels")
    return df

3. GBIF 生物多様性レコード

GBIF_BASE = "https://api.gbif.org/v1"


def gbif_species_search(name, limit=20):
    """
    GBIF — 種名検索・分類マッチング。

    Parameters:
        name: str — 種名
        limit: int — 結果上限
    """
    url = f"{GBIF_BASE}/species/search"
    params = {"q": name, "limit": limit}
    resp = requests.get(url, params=params, timeout=30)
    resp.raise_for_status()
    data = resp.json()

    results = []
    for sp in data.get("results", []):
        results.append({
            "taxon_key": sp.get("key", ""),
            "scientific_name": sp.get("scientificName", ""),
            "canonical_name": sp.get("canonicalName", ""),
            "status": sp.get("taxonomicStatus", ""),
            "rank": sp.get("rank", ""),
            "kingdom": sp.get("kingdom", ""),
            "phylum": sp.get("phylum", ""),
            "class": sp.get("class", ""),
            "order": sp.get("order", ""),
            "family": sp.get("family", ""),
            "num_occurrences": sp.get("numOccurrences", 0),
        })

    df = pd.DataFrame(results)
    print(f"GBIF species: '{name}' → {len(df)} taxa")
    return df


def gbif_occurrence_search(taxon_key=None, country=None,
                            year_range=None, limit=300):
    """
    GBIF — 出現記録検索。

    Parameters:
        taxon_key: int — GBIF taxon key
        country: str — ISO 国コード (例: "JP")
        year_range: tuple — (start, end)
        limit: int — 最大件数
    """
    url = f"{GBIF_BASE}/occurrence/search"
    params = {"limit": min(limit, 300)}

    if taxon_key:
        params["taxonKey"] = taxon_key
    if country:
        params["country"] = country
    if year_range:
        params["year"] = f"{year_range[0]},{year_range[1]}"

    resp = requests.get(url, params=params, timeout=60)
    resp.raise_for_status()
    data = resp.json()

    records = []
    for rec in data.get("results", []):
        records.append({
            "gbif_id": rec.get("gbifID", ""),
            "scientific_name": rec.get("scientificName", ""),
            "latitude": rec.get("decimalLatitude", None),
            "longitude": rec.get("decimalLongitude", None),
            "country": rec.get("country", ""),
            "year": rec.get("year", ""),
            "basis_of_record": rec.get("basisOfRecord", ""),
            "institution": rec.get("institutionCode", ""),
        })

    df = pd.DataFrame(records)
    print(f"GBIF occurrences: {len(df)} records (total: {data.get('count', 0)})")
    return df

4. FishBase 魚類データ

FISHBASE_BASE = "https://fishbase.ropensci.org"


def fishbase_species(genus=None, species=None, family=None):
    """
    FishBase — 魚類種データ取得。

    Parameters:
        genus: str — 属名
        species: str — 種小名
        family: str — 科名
    """
    url = f"{FISHBASE_BASE}/species"
    params = {"limit": 100}
    if genus:
        params["Genus"] = genus
    if species:
        params["Species"] = species
    if family:
        params["Family"] = family

    resp = requests.get(url, params=params, timeout=30)
    resp.raise_for_status()
    data = resp.json()

    records = []
    for fish in data.get("data", []):
        records.append({
            "spec_code": fish.get("SpecCode", ""),
            "genus": fish.get("Genus", ""),
            "species": fish.get("Species", ""),
            "family": fish.get("Family", ""),
            "body_shape": fish.get("BodyShapeI", ""),
            "max_length": fish.get("Length", None),
            "vulnerability": fish.get("Vulnerability", None),
            "importance": fish.get("Importance", ""),
            "habitat": fish.get("DemersPelag", ""),
            "depth_range": f"{fish.get('DepthRangeShallow', '')}-{fish.get('DepthRangeDeep', '')}",
        })

    df = pd.DataFrame(records)
    print(f"FishBase: {len(df)} species")
    return df

5. 海洋生態学統合パイプライン

def marine_ecology_pipeline(taxon_name, region_wkt=None,
                             output_dir="results"):
    """
    OBIS + WoRMS + GBIF + FishBase 統合パイプライン。

    Parameters:
        taxon_name: str — 分類群名
        region_wkt: str — 調査海域 WKT
        output_dir: str — 出力ディレクトリ
    """
    from pathlib import Path
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    # 1) WoRMS 分類検証
    taxonomy = worms_taxon_search(taxon_name)
    taxonomy.to_csv(output_dir / "worms_taxonomy.csv", index=False)

    aphia_id = None
    if len(taxonomy) > 0:
        aphia_id = taxonomy.iloc[0]["aphia_id"]
        classification = worms_classification(aphia_id)
        classification.to_csv(output_dir / "classification.csv", index=False)

    # 2) OBIS 海洋分布
    obis_data = obis_occurrence_search(
        taxon_name=taxon_name,
        taxon_id=aphia_id,
        geometry=region_wkt,
    )
    obis_data.to_csv(output_dir / "obis_occurrences.csv", index=False)

    # 3) GBIF 補完
    gbif_sp = gbif_species_search(taxon_name)
    if len(gbif_sp) > 0:
        taxon_key = gbif_sp.iloc[0]["taxon_key"]
        gbif_occ = gbif_occurrence_search(taxon_key=taxon_key)
        gbif_occ.to_csv(output_dir / "gbif_occurrences.csv", index=False)
    else:
        gbif_occ = pd.DataFrame()

    # 4) 多様性指標
    if len(obis_data) > 0:
        n_species = obis_data["scientific_name"].nunique()
        lat_range = (obis_data["latitude"].min(), obis_data["latitude"].max())
        depth_range = (obis_data["depth"].min(), obis_data["depth"].max())
        print(f"Diversity: {n_species} species, "
              f"lat {lat_range}, depth {depth_range}")

    print(f"Marine ecology pipeline: {output_dir}")
    return {
        "taxonomy": taxonomy,
        "obis_occurrences": obis_data,
        "gbif_occurrences": gbif_occ,
    }

ToolUniverse 連携

パイプライン統合

environmental-ecology → marine-ecology → phylogenetics
  (陸域生態学)          (OBIS/WoRMS/GBIF)  (系統解析)
       │                      │               ↓
  biodiversity-db ───────────┘         species-distribution
  (ENA/BOLD)           │               (空間分布モデル)
                       ↓
                  fisheries-management
                  (水産資源管理)

パイプライン出力