| """
|
| GDC Data Portal Client
|
| Download and parse cancer genomics data from GDC
|
| """
|
|
|
| import os
|
| import json
|
| import requests
|
| from typing import Dict, List, Optional, Any
|
| from pathlib import Path
|
| import yaml
|
| from dataclasses import dataclass
|
| import logging
|
|
|
| logging.basicConfig(level=logging.INFO)
|
| logger = logging.getLogger(__name__)
|
|
|
|
|
| @dataclass
|
| class GDCFile:
|
| """Represents a file from GDC Portal"""
|
| file_id: str
|
| file_name: str
|
| file_size: int
|
| data_type: str
|
| data_format: str
|
| experimental_strategy: str
|
| case_id: str
|
| project_id: str
|
|
|
|
|
| class GDCClient:
|
| """Client for interacting with GDC Data Portal API"""
|
|
|
| def __init__(self, config_path: str = "config.yml"):
|
| with open(config_path, 'r') as f:
|
| self.config = yaml.safe_load(f)['gdc']
|
|
|
| self.api_url = self.config['api_url']
|
| self.download_dir = Path(self.config['download_dir'])
|
| self.download_dir.mkdir(parents=True, exist_ok=True)
|
|
|
| self.session = requests.Session()
|
| self.session.headers.update({
|
| 'Content-Type': 'application/json'
|
| })
|
|
|
| def search_files(
|
| self,
|
| filters: Optional[Dict] = None,
|
| size: int = 100,
|
| fields: Optional[List[str]] = None
|
| ) -> List[GDCFile]:
|
| """
|
| Search for files in GDC
|
|
|
| Args:
|
| filters: GDC filter query
|
| size: Number of results to return
|
| fields: Fields to include in response
|
| """
|
| endpoint = f"{self.api_url}/files"
|
|
|
| if fields is None:
|
| fields = [
|
| 'file_id', 'file_name', 'file_size', 'data_type',
|
| 'data_format', 'experimental_strategy', 'cases.case_id',
|
| 'cases.project.project_id'
|
| ]
|
|
|
| params = {
|
| 'size': size,
|
| 'fields': ','.join(fields)
|
| }
|
|
|
| if filters:
|
| params['filters'] = json.dumps(filters)
|
|
|
| try:
|
| response = self.session.get(endpoint, params=params)
|
| response.raise_for_status()
|
| data = response.json()
|
|
|
| files = []
|
| for hit in data.get('data', {}).get('hits', []):
|
| gdc_file = GDCFile(
|
| file_id=hit.get('file_id'),
|
| file_name=hit.get('file_name'),
|
| file_size=hit.get('file_size', 0),
|
| data_type=hit.get('data_type'),
|
| data_format=hit.get('data_format'),
|
| experimental_strategy=hit.get('experimental_strategy'),
|
| case_id=hit.get('cases', [{}])[0].get('case_id') if hit.get('cases') else None,
|
| project_id=hit.get('cases', [{}])[0].get('project', {}).get('project_id') if hit.get('cases') else None
|
| )
|
| files.append(gdc_file)
|
|
|
| logger.info(f"Found {len(files)} files")
|
| return files
|
|
|
| except Exception as e:
|
| logger.error(f"Error searching files: {e}")
|
| return []
|
|
|
| def download_file(
|
| self,
|
| file_id: str,
|
| output_dir: Optional[Path] = None
|
| ) -> Optional[Path]:
|
| """
|
| Download a file from GDC
|
|
|
| Args:
|
| file_id: GDC file UUID
|
| output_dir: Directory to save file (defaults to config download_dir)
|
|
|
| Returns:
|
| Path to downloaded file or None if failed
|
| """
|
| if output_dir is None:
|
| output_dir = self.download_dir
|
|
|
| output_dir = Path(output_dir)
|
| output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
| endpoint = f"{self.api_url}/data/{file_id}"
|
|
|
| try:
|
| logger.info(f"Downloading file {file_id}")
|
| response = self.session.get(endpoint, stream=True)
|
| response.raise_for_status()
|
|
|
|
|
| content_disposition = response.headers.get('content-disposition', '')
|
| if 'filename=' in content_disposition:
|
| filename = content_disposition.split('filename=')[1].strip('"')
|
| else:
|
| filename = file_id
|
|
|
| output_path = output_dir / filename
|
|
|
| with open(output_path, 'wb') as f:
|
| for chunk in response.iter_content(chunk_size=8192):
|
| f.write(chunk)
|
|
|
| logger.info(f"Downloaded to {output_path}")
|
| return output_path
|
|
|
| except Exception as e:
|
| logger.error(f"Error downloading file {file_id}: {e}")
|
| return None
|
|
|
| def get_project_files(
|
| self,
|
| project_id: str,
|
| data_type: Optional[str] = None,
|
| limit: int = 100
|
| ) -> List[GDCFile]:
|
| """
|
| Get files for a specific project
|
|
|
| Args:
|
| project_id: GDC project ID (e.g., TCGA-BRCA)
|
| data_type: Filter by data type
|
| limit: Maximum number of files
|
| """
|
| filters = {
|
| "op": "and",
|
| "content": [
|
| {
|
| "op": "in",
|
| "content": {
|
| "field": "cases.project.project_id",
|
| "value": [project_id]
|
| }
|
| }
|
| ]
|
| }
|
|
|
| if data_type:
|
| filters["content"].append({
|
| "op": "in",
|
| "content": {
|
| "field": "data_type",
|
| "value": [data_type]
|
| }
|
| })
|
|
|
| return self.search_files(filters=filters, size=limit)
|
|
|
| def get_mutation_data(self, project_id: str, limit: int = 100) -> List[GDCFile]:
|
| """Get mutation/variant calling files for a project"""
|
| return self.get_project_files(
|
| project_id=project_id,
|
| data_type="Simple Nucleotide Variation",
|
| limit=limit
|
| )
|
|
|
| def get_gene_expression_data(self, project_id: str, limit: int = 100) -> List[GDCFile]:
|
| """Get gene expression data for a project"""
|
| return self.get_project_files(
|
| project_id=project_id,
|
| data_type="Gene Expression Quantification",
|
| limit=limit
|
| )
|
|
|
| def search_cases(
|
| self,
|
| project_id: str,
|
| filters: Optional[Dict] = None,
|
| size: int = 100
|
| ) -> List[Dict]:
|
| """
|
| Search for cases (patients) in GDC
|
|
|
| Args:
|
| project_id: GDC project ID
|
| filters: Additional filter criteria
|
| size: Number of results
|
| """
|
| endpoint = f"{self.api_url}/cases"
|
|
|
| base_filters = {
|
| "op": "in",
|
| "content": {
|
| "field": "project.project_id",
|
| "value": [project_id]
|
| }
|
| }
|
|
|
| if filters:
|
| filter_query = {
|
| "op": "and",
|
| "content": [base_filters, filters]
|
| }
|
| else:
|
| filter_query = base_filters
|
|
|
| params = {
|
| 'size': size,
|
| 'filters': json.dumps(filter_query),
|
| 'fields': 'case_id,project.project_id,demographic,diagnoses'
|
| }
|
|
|
| try:
|
| response = self.session.get(endpoint, params=params)
|
| response.raise_for_status()
|
| data = response.json()
|
|
|
| cases = data.get('data', {}).get('hits', [])
|
| logger.info(f"Found {len(cases)} cases")
|
| return cases
|
|
|
| except Exception as e:
|
| logger.error(f"Error searching cases: {e}")
|
| return []
|
|
|
|
|
| class GDCDataParser:
|
| """Parse downloaded GDC data files"""
|
|
|
| @staticmethod
|
| def parse_maf(file_path: Path) -> List[Dict]:
|
| """
|
| Parse MAF (Mutation Annotation Format) file
|
|
|
| Returns list of mutation records
|
| """
|
| mutations = []
|
|
|
| try:
|
| with open(file_path, 'r') as f:
|
|
|
| for line in f:
|
| if not line.startswith('#'):
|
| header_line = line.strip()
|
| break
|
|
|
| headers = header_line.split('\t')
|
|
|
| for line in f:
|
| if line.startswith('#'):
|
| continue
|
|
|
| values = line.strip().split('\t')
|
| if len(values) == len(headers):
|
| mutation = dict(zip(headers, values))
|
| mutations.append(mutation)
|
|
|
| logger.info(f"Parsed {len(mutations)} mutations from {file_path}")
|
| return mutations
|
|
|
| except Exception as e:
|
| logger.error(f"Error parsing MAF file: {e}")
|
| return []
|
|
|
| @staticmethod
|
| def parse_vcf(file_path: Path) -> List[Dict]:
|
| """
|
| Parse VCF (Variant Call Format) file
|
|
|
| Returns list of variant records
|
| """
|
| variants = []
|
|
|
| try:
|
| with open(file_path, 'r') as f:
|
| for line in f:
|
| if line.startswith('##'):
|
| continue
|
| if line.startswith('#CHROM'):
|
| headers = line.strip().split('\t')
|
| continue
|
|
|
| values = line.strip().split('\t')
|
| variant = {
|
| 'chrom': values[0],
|
| 'pos': values[1],
|
| 'id': values[2],
|
| 'ref': values[3],
|
| 'alt': values[4],
|
| 'qual': values[5],
|
| 'filter': values[6],
|
| 'info': values[7]
|
| }
|
| variants.append(variant)
|
|
|
| logger.info(f"Parsed {len(variants)} variants from {file_path}")
|
| return variants
|
|
|
| except Exception as e:
|
| logger.error(f"Error parsing VCF file: {e}")
|
| return []
|
|
|
| @staticmethod
|
| def parse_clinical_data(data: Dict) -> Dict:
|
| """Parse clinical data from GDC case"""
|
| clinical = {
|
| 'case_id': data.get('case_id'),
|
| 'project_id': data.get('project', {}).get('project_id'),
|
| 'demographic': {},
|
| 'diagnoses': []
|
| }
|
|
|
|
|
| demo = data.get('demographic', {})
|
| clinical['demographic'] = {
|
| 'age_at_index': demo.get('age_at_index'),
|
| 'gender': demo.get('gender'),
|
| 'race': demo.get('race'),
|
| 'ethnicity': demo.get('ethnicity')
|
| }
|
|
|
|
|
| for diag in data.get('diagnoses', []):
|
| diagnosis = {
|
| 'diagnosis_id': diag.get('diagnosis_id'),
|
| 'primary_diagnosis': diag.get('primary_diagnosis'),
|
| 'tumor_stage': diag.get('tumor_stage'),
|
| 'age_at_diagnosis': diag.get('age_at_diagnosis'),
|
| 'vital_status': diag.get('vital_status')
|
| }
|
| clinical['diagnoses'].append(diagnosis)
|
|
|
| return clinical
|
|
|
