Source code for tooluniverse.fourdn_tool
"""
4DN Data Portal API Tool
This tool provides access to the 4D Nucleome Data Portal, which hosts
350+ uniformly processed Hi-C contact files and chromatin conformation data.
"""
import requests
from typing import Dict, Any
from .base_tool import BaseTool
from .tool_registry import register_tool
FOURDN_BASE_URL = "https://data.4dnucleome.org"
def _format_request_error(exc: Exception) -> Dict[str, Any]:
"""Turn a request exception into an actionable error payload.
The 4DN portal has periodically served an expired TLS certificate
(a server-side operational lapse, not a client problem). Detect that
case and explain it instead of leaking a raw connection-pool traceback,
so a caller knows the failure is transient and infrastructure-side rather
than a malformed query. Certificate verification is never disabled.
"""
msg = str(exc)
if "CERTIFICATE_VERIFY_FAILED" in msg or "certificate has expired" in msg:
return {
"status": "error",
"data": {
"error": (
"The 4DN Data Portal (data.4dnucleome.org) is presenting an "
"invalid/expired TLS certificate, so the request could not be "
"completed securely. This is a transient server-side issue at "
"4DN, not a problem with the query -- retry later or check "
"https://data.4dnucleome.org status."
)
},
}
return {"status": "error", "data": {"error": msg}}
[docs]
@register_tool("FourDNTool")
class FourDNTool(BaseTool):
"""
4DN Data Portal API tool for accessing Hi-C and
chromatin conformation data.
"""
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the tool with given arguments."""
try:
# Fix-R39A-2: "operation" is optional in the schema (each config
# exposes exactly one valid value via enum+default), so a caller
# who omits it must still resolve to that tool instance's own
# operation -- the old bare "search" literal fallback was only
# coincidentally correct for FourDN_search_data and would
# silently mis-route the other 3 tools to a search instead of
# their own operation.
operation = arguments.get("operation") or self.get_schema_const_operation()
if operation == "search":
return self._search(arguments)
elif operation == "get_file_metadata":
return self._get_file_metadata(arguments)
elif operation == "get_experiment_metadata":
return self._get_experiment_metadata(arguments)
elif operation == "download_file_url":
return self._download_file_url(arguments)
else:
return {
"status": "error",
"data": {"error": f"Unknown operation: {operation}"},
}
except Exception as e:
return _format_request_error(e)
[docs]
def _search(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search 4DN data portal for files or experiments."""
try:
query = arguments.get("query", "*")
item_type = arguments.get("item_type", "File")
limit = int(arguments.get("limit", 25))
offset = int(arguments.get("offset", 0))
# Build search URL
url = f"{FOURDN_BASE_URL}/search/"
params = {
"type": item_type,
"q": query,
"limit": limit,
"from": offset,
"format": "json",
}
# Add filters
if "file_type" in arguments:
params["file_type"] = arguments["file_type"]
if "assay_title" in arguments:
params["assay_title"] = arguments["assay_title"]
if "biosource_name" in arguments:
params["biosource_name"] = arguments["biosource_name"]
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
api_data = response.json()
results = api_data.get("@graph", [])
# The portal uses Elasticsearch's 10,000-hit sentinel on later
# pages: for a large query the first page reports the exact total,
# while otherwise-identical requests with ``from > 0`` report
# exactly 10,000. A cheap first-page count probe (limit=0) gives
# the exact value without downloading duplicate records. If that
# auxiliary probe fails, preserve the requested page and mark the
# upstream count as inexact rather than presenting 10,000 as fact.
upstream_page_total = int(api_data.get("total", 0))
total = upstream_page_total
total_is_exact = True
total_warning = None
if offset > 0 and upstream_page_total == 10000:
count_params = dict(params)
count_params.update({"limit": 0, "from": 0})
try:
count_response = requests.get(url, params=count_params, timeout=30)
count_response.raise_for_status()
total = int(count_response.json().get("total", upstream_page_total))
except Exception:
total_is_exact = False
total_warning = (
"4DN returned its 10,000-hit sentinel for this later page, "
"and the exact-count probe failed; total is a lower bound."
)
returned = len(results)
candidate_next_offset = offset + returned
has_more = (
candidate_next_offset < total
if total_is_exact
else returned == limit and returned > 0
)
result = {
"status": "success",
"num_results": returned,
"returned": returned,
"offset": offset,
"total": total,
"total_is_exact": total_is_exact,
"upstream_page_total": upstream_page_total,
"has_more": has_more,
"next_offset": candidate_next_offset if has_more else None,
"results": results,
"search_url": response.url,
}
if total_warning:
result["warning"] = total_warning
return {"status": "success", "data": result}
except Exception as e:
return _format_request_error(e)
[docs]
def _get_file_metadata(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get metadata for a specific file."""
try:
file_accession = arguments.get("file_accession")
include_full_metadata = arguments.get("include_full_metadata", False)
if not file_accession:
return {
"status": "error",
"data": {"error": "file_accession is required"},
}
# Use frame=object for smaller response (84% reduction)
url = f"{FOURDN_BASE_URL}/{file_accession}/?format=json&frame=object"
response = requests.get(url, timeout=30)
response.raise_for_status()
api_data = response.json()
# Handle file_format - can be dict or string with frame=object
file_format = api_data.get("file_format")
if isinstance(file_format, dict):
file_format = file_format.get("display_title")
elif isinstance(file_format, str):
# Extract format from URL like /file-formats/bw/
file_format = (
file_format.split("/")[-2] if "/" in file_format else file_format
)
result = {
"status": "success",
"accession": file_accession,
"file_type": api_data.get("file_type"),
"file_format": file_format,
"file_size": api_data.get("file_size"),
"description": api_data.get("description"),
"data_status": api_data.get("status"),
"biosource": api_data.get("biosource"),
"experiment": api_data.get("experiment"),
"download_url": (f"{FOURDN_BASE_URL}/{file_accession}/@@download"),
}
# Only include full metadata if explicitly requested
# This reduces response size by ~97% in typical cases
if include_full_metadata:
result["metadata"] = api_data
return {"status": "success", "data": result}
except Exception as e:
return _format_request_error(e)
[docs]
def _get_experiment_metadata(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get metadata for a specific experiment."""
try:
experiment_accession = arguments.get("experiment_accession")
include_full_metadata = arguments.get("include_full_metadata", False)
if not experiment_accession:
return {
"status": "error",
"data": {"error": "experiment_accession is required"},
}
# Use frame=object for smaller response
url = f"{FOURDN_BASE_URL}/{experiment_accession}/?format=json&frame=object"
response = requests.get(url, timeout=30)
response.raise_for_status()
api_data = response.json()
# Extract biosource from biosample
# (biosource is nested inside biosample)
biosample = api_data.get("biosample", {})
# Handle biosample - can be dict or string with frame=object
if isinstance(biosample, str):
biosample = {"@id": biosample}
biosource_list = []
biosource_summary = ""
else:
biosource_list = biosample.get("biosource", [])
biosource_summary = biosample.get("biosource_summary", "")
# Combine files from different possible fields
files = api_data.get("files", [])
if not files:
files = api_data.get("processed_files", [])
# Handle experiment_type - can be dict or string
exp_type = api_data.get("experiment_type")
if isinstance(exp_type, dict):
exp_type_name = exp_type.get("display_title")
elif isinstance(exp_type, str):
# Extract from URL like /experiment-types/in-situ-hi-c/
exp_type_name = (
exp_type.split("/")[-2].replace("-", " ")
if "/" in exp_type
else exp_type
)
else:
exp_type_name = None
result = {
"status": "success",
"accession": experiment_accession,
"experiment_type": exp_type_name,
"biosample": biosample,
"biosource": biosource_list,
"biosource_summary": biosource_summary,
"description": api_data.get("description"),
"data_status": api_data.get("status"),
"files": files,
}
# Only include full metadata if explicitly requested
if include_full_metadata:
result["metadata"] = api_data
return {"status": "success", "data": result}
except Exception as e:
return _format_request_error(e)
[docs]
def _download_file_url(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get download URL for file (requires authentication)."""
try:
file_accession = arguments.get("file_accession")
if not file_accession:
return {
"status": "error",
"data": {"error": "file_accession is required"},
}
# Get DRS API information
drs_url = f"{FOURDN_BASE_URL}/ga4gh/drs/v1/objects/{file_accession}"
result = {
"status": "success",
"accession": file_accession,
"download_url": (f"{FOURDN_BASE_URL}/{file_accession}/@@download"),
"drs_url": drs_url,
"note": (
"File download requires authentication. "
"Use curl with access key: "
"curl -O -L --user <key>:<secret> <download-url>"
),
"instruction": ("Create access key at https://data.4dnucleome.org/me"),
}
return {"status": "success", "data": result}
except Exception as e:
return _format_request_error(e)