Gtex Database

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""CLI wrapper for GTEx API V2.

Follows GTEx Portal Terms of Use by fetching sequentially and handling
pagination.
"""

# /// script
# requires-python = ">=3.10"
# dependencies = [
#   "scienceskillscommon",
# ]
# [tool.uv.sources]
# scienceskillscommon = { path = "../../scienceskillscommon" }
# ///

import argparse
import json
import sys
import urllib.parse

from science_skills.scienceskillscommon import http_client

BASE_URL = 'https://gtexportal.org/api/v2'
DATASET_ID = 'gtex_v10'
GENCODE_VERSION = 'v39'
CLIENT = http_client.HttpClient(BASE_URL, qps=1.0)

# Optional cache for tissues to avoid fetching repeatedly
TISSUE_CACHE = None


def _fetch(url, params=None):
  """Fetches URL, with optional query parameters, using HttpClient."""
  if params:
    # Filter out None values
    params = {k: v for k, v in params.items() if v is not None}
    query_string = urllib.parse.urlencode(params, doseq=True)
    full_url = f'{url}?{query_string}'
  else:
    full_url = url

  return CLIENT.fetch_json(full_url)


def _fetch_paginated(url, params=None):
  """Fetches all pages from a paginated endpoint."""
  if params is None:
    params = {}

  all_data = []
  page = 0
  while True:
    params['page'] = page
    response = _fetch(url, params)

    # Some endpoints don't use 'data' wrapping or 'paging_info'
    if 'paging_info' not in response:
      return response

    data = response.get('data', [])
    all_data.extend(data)

    paging = response.get('paging_info', {})
    total_pages = paging.get('numberOfPages', 0)

    if page >= total_pages - 1:
      break
    page += 1

  return all_data


def get_tissue_mapping():
  """Fetches the list of tissues and maps names to tissueSiteDetailId."""
  global TISSUE_CACHE
  if TISSUE_CACHE is not None:
    return TISSUE_CACHE

  url = f'{BASE_URL}/dataset/tissueSiteDetail'
  data = _fetch_paginated(url, {'datasetId': DATASET_ID})

  mapping = {}
  for t in data:
    id_ = t.get('tissueSiteDetailId')
    name = t.get('tissueSiteDetail')
    if id_ and name:
      mapping[id_.lower()] = id_
      mapping[name.lower()] = id_
      # Allow "Esophagus - Muscularis" instead of "Esophagus_Muscularis", etc.
      mapping[id_.replace('_', ' ').lower()] = id_
      mapping[name.replace('-', ' ').lower()] = id_

  TISSUE_CACHE = mapping
  return mapping


def resolve_tissue(tissue_str):
  mapping = get_tissue_mapping()
  cleaned = tissue_str.strip().lower()
  if cleaned in mapping:
    return mapping[cleaned]
  # Try more fuzzy matching if needed
  cleaned_no_hyphen = cleaned.replace('-', ' ')
  if cleaned_no_hyphen in mapping:
    return mapping[cleaned_no_hyphen]
  sys.stderr.write(f"Error: Unknown tissue '{tissue_str}'.\n")
  sys.exit(1)


def resolve_gencode_id(gene_symbol, output_file):
  """Maps a standard gene symbol to its Versioned GENCODE ID."""
  url = f'{BASE_URL}/reference/gene'
  data = _fetch_paginated(
      url, {'geneId': gene_symbol, 'gencodeVersion': GENCODE_VERSION}
  )
  if not data:
    sys.stderr.write(f"Error: Could not find GENCODE ID for '{gene_symbol}'.\n")
    sys.exit(1)

  # Return the first matching exact symbol if possible, else just the first one
  best_match = data[0]
  for d in data:
    if d.get('geneSymbol', '').lower() == gene_symbol.lower():
      best_match = d
      break

  result = {
      'gene_symbol': best_match.get('geneSymbol'),
      'gencode_id': best_match.get('gencodeId'),
      'chro