Feature/ega2 catalog

examples/ega/example_config.toml

@@ -0,0 +1,18 @@
+[dataset]
+identifier = "https://www.example.com/img-123"
+title = "Example Imaging Dataset Title"
+description = "This is imaging data description"
+theme = ["http://publications.europa.eu/resource/authority/data-theme/HEAL"]
+keyword = ["list", "of", "key", "words"]
+access_rights = "http://publications.europa.eu/resource/authority/access-right/PUBLIC"
+applicable_legislation = ["http://publications.europa.eu/resource/authority/access-right/NON_PUBLIC"]
+
+[dataset.publisher]
+name = ["Example publisher list"]
+identifier = ["http://example.com"]
+mbox = "mailto:publisher@example.com"
+homepage = "http://www.example.com"
+
+[dataset.contact_point]
+formatted_name = "Example Data Management office"
+email = "mailto:datamanager@example.com"

pyproject.toml

@@ -34,6 +34,7 @@ dependencies = [
   "sparqlwrapper~=2.0",
   "fairclient >= 1.0.0",
   "pandas >= 2.0.0",
+  "requests >= 2.34.2",
 ]
 
 [project.entry-points]

src/img2catalog/cli_app.py

@@ -14,6 +14,8 @@
 from img2catalog.configmanager import load_img2catalog_configuration
 
 from img2catalog.const import (
+    EGA_API_URL_ENV,
+    EGA_DEFAULT_API_URL,
     FDP_PASS_ENV,
     FDP_SERVER_ENV,
     FDP_USER_ENV,
@@ -28,6 +30,8 @@
 from img2catalog.mappings.xnat import map_xnat_to_healthriv2
 from img2catalog.inputs.csv_reader import read_csv
 from img2catalog.mappings.xds import map_xds_to_healthri_dcat_dataset
+from img2catalog.inputs.ega import fetch_ega_datasets
+from img2catalog.mappings.ega import map_ega_to_healthri_dcat_dataset
 from img2catalog.outputs.fdp import FDPOutput
 from img2catalog.outputs.rdf import RDFOutput
 
@@ -350,5 +354,57 @@
 input_xds.add_command(mapping_xds)
 mapping_xds.add_command(output_fdp)
 
+
+@click.group(name="ega")
+@click.option(
+    "-a",
+    "--dataset-id",
+    "dataset_ids",
+    type=str,
+    multiple=True,
+    required=True,
+    help="EGA dataset dataset to import (e.g. EGAD00001005083). Can be repeated to import multiple datasets.",
+)
+@click.option(
+    "--api-url",
+    envvar=EGA_API_URL_ENV,
+    type=str,
+    default=EGA_DEFAULT_API_URL,
+    help=f"Base URL of the EGA metadata API. Defaults to {EGA_DEFAULT_API_URL}.",
+)
+@click.pass_context
+def input_ega(ctx: click.Context, dataset_ids: tuple, api_url: str):
+    """Extract dataset metadata from the EGA (European Genome-phenome Archive) metadata API."""
+    ega_datasets = fetch_ega_datasets(list(dataset_ids), api_url)
+    ctx.obj['unmapped_objects'] = {
+        'dataset': ega_datasets
+    }
+
+cli_click.add_command(input_ega)
+
+
+@click.group("map-ega-hriv2")
+@click.pass_context
+def mapping_ega_healthriv2(ctx: click.Context):
+    """Map metadata from EGA to the Health-RI model."""
+    config = ctx.obj["config"]
+    unmapped_objects = ctx.obj['unmapped_objects']
+
+    datasets = []
+    for ega_dataset in unmapped_objects['dataset']:
+        dataset = map_ega_to_healthri_dcat_dataset(ega_dataset, config)
+        datasets.append({
+            'uri': URIRef(f"http://img2catalog.internal/dataset/{ega_dataset['accession_id']}"),
+            'model_object': dataset
+        })
+
+    ctx.obj['mapped_objects'] = {
+        'dataset': datasets
+    }
+
+input_ega.add_command(mapping_ega_healthriv2)
+mapping_ega_healthriv2.add_command(output_rdf)
+mapping_ega_healthriv2.add_command(output_fdp)
+
 if __name__ == "__main__":
     cli_click()

src/img2catalog/const.py

@@ -15,6 +15,9 @@
 
 SPARQL_ENV = "IMG2CATALOG_SPARQL_ENDPOINT"
 
+EGA_API_URL_ENV = "IMG2CATALOG_EGA_API_URL"
+EGA_DEFAULT_API_URL = "https://metadata.ega-archive.org"
+
 # Default setting
 REMOVE_OPTIN_KEYWORD = True
 INCLUDE_PRIVATE = False

src/img2catalog/inputs/ega.py

@@ -0,0 +1,23 @@
+import logging
+from typing import Dict, List, Optional
+
+import requests
+
+logger = logging.getLogger(__name__)
+
+def fetch_ega_dataset(dataset_id: str, api_url: str) -> Dict:
+    response = requests.get(f"{api_url}/datasets/{dataset_id}", timeout=30)
+    response.raise_for_status()
+
+    return response.json()
+
+
+def fetch_ega_datasets(dataset_ids: List[str], api_url: str) -> List[Dict]:
+    datasets = []
+    for dataset_id in dataset_ids:
+        try:
+            datasets.append(fetch_ega_dataset(dataset_id, api_url))
+        except requests.RequestException as e:
+            logger.warning("Error fetching EGA dataset %s: %s", dataset_id, e)
+
+    return datasets

src/img2catalog/mappings/ega.py

@@ -0,0 +1,88 @@
+import logging
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from pydantic import AnyHttpUrl
+from rdflib import URIRef
+from sempyro import LiteralField
+from sempyro.dcat import AccessRights
+from sempyro.hri_dcat import DatasetStatus, DatasetTheme, HRIAgent, HRIDataset, HRIVCard
+
+logger = logging.getLogger(__name__)
+
+# Source used:
+# https://healthri.sharepoint.com/:x:/r/sites/hri-team022/_layouts/15/Doc.aspx?sourcedoc=%7BE3EC5B3F-6BB2-404B-9DA9-489A90BAC077%7D&file=EGA%20Health-RI%20Core%20mapping.xlsx&action=default&mobileredirect=true
+
+def get_identifier(ega_dataset: Dict) -> str:
+    """Build the identifiers.org URI for an EGA dataset's accession_id."""
+    return f"http://identifiers.org/ega.dataset:{ega_dataset['accession_id']}"
+
+def get_title(ega_dataset: Dict) -> str:
+    return ega_dataset["title"]
+
+def get_description(ega_dataset: Dict) -> str:
+    return ega_dataset["description"]
+
+def get_number_of_records(ega_dataset: Dict) -> Optional[int]:
+    return ega_dataset.get("num_samples")
+
+def get_release_date(ega_dataset: Dict) -> Optional[datetime]:
+    released_date = ega_dataset.get("released_date")
+    if released_date is None:
+        return None
+
+    try:
+        return datetime.fromisoformat(released_date)
+    except ValueError:
+        logger.error("Could not parse EGA release date %r", released_date)
+        return released_date
+
+def get_keyword(ega_dataset: Dict) -> List[LiteralField]:
+    """Map EGA's free-text `technologies` field to DCAT-AP keywords."""
+    return [LiteralField(value=technology) for technology in ega_dataset.get("technologies", [])]
+
+def map_ega_to_healthri_dcat_dataset(ega_dataset: Dict, config: Dict) -> HRIDataset:
+    dataset_config = config["dataset"]
+    publisher_config = dataset_config["publisher"]
+    contact_point_config = dataset_config["contact_point"]
+
+    dataset_themes = [DatasetTheme(URIRef(theme)) for theme in dataset_config["theme"]]
+
+    dataset_keywords = get_keyword(ega_dataset)
+    dataset_keywords.extend(LiteralField(value=keyword) for keyword in dataset_config.get("keyword", []))
+
+    dataset_applicable_legislation = [AnyHttpUrl(url) for url in dataset_config["applicable_legislation"]]
+
+    publisher_identifiers = [LiteralField(value=identifier) for identifier in publisher_config["identifier"]]
+
+    publisher = HRIAgent(
+        name=[LiteralField(value=name) for name in publisher_config["name"]],
+        identifier=publisher_identifiers,
+        mbox=publisher_config["mbox"],
+        homepage=publisher_config["homepage"],
+    )
+
+    contact_point = HRIVCard(
+        hasEmail=contact_point_config["email"],
+        formatted_name=contact_point_config["formatted_name"],
+    )
+
+    dataset = HRIDataset(
+        # Directly mapped from EGA
+        identifier=LiteralField(value=get_identifier(ega_dataset)),
+        title=[LiteralField(value=get_title(ega_dataset))],
+        description=[LiteralField(value=get_description(ega_dataset))],
+        release_date=get_release_date(ega_dataset),
+        number_of_records=get_number_of_records(ega_dataset),
+        keyword=dataset_keywords,
+        # Not present in EGA metadata, supplied from local node configuration (see
+        # docs/ega_mapping.md for the fields that are not (yet) mapped from EGA)
+        publisher=publisher,
+        contact_point=contact_point,
+        creator=[publisher],
+        theme=dataset_themes,
+        applicable_legislation=dataset_applicable_legislation,
+        access_rights=AccessRights(URIRef(dataset_config["access_rights"])),
+    )
+
+    return dataset

tests/conftest.py

@@ -14,10 +14,13 @@
 
 
 TEST_CONFIG = pathlib.Path(__file__).parent / "img2catalog" / "examples" / "xnat" / "example-config.toml"
-pytest_plugins = "tests.img2catalog.xnatpy_fixtures"
+pytest_plugins = [
+    "tests.img2catalog.xnatpy_fixtures",
+    "tests.img2catalog.ega_fixtures",
+]
 
 
 @pytest.fixture()
 def config():
     """Loads the default configuration TOML"""
     config_path = TEST_CONFIG
@@ -28,19 +31,19 @@
     return config
 
 
 @pytest.fixture()
 def mock_catalog():
     catalog = DCATCatalog(title=["Example XNAT catalog"], description=["This is an example XNAT catalog description"])
     return catalog
 
 
 @pytest.fixture()
 def mock_dataset():
     dataset = DCATDataset(title=["test project"], description=["test description"])
     return dataset
 
 
 @pytest.fixture()
 def empty_graph():
     graph = Graph()
     graph.bind("dcat", DCAT)
@@ -49,7 +52,7 @@
     return graph
 
 
 @pytest.fixture()
 def second_empty_graph():
     graph = Graph()
     graph.bind("dcat", DCAT)
@@ -58,7 +61,7 @@
     return graph
 
 
 @pytest.fixture()
 def toml_patch_target():
     # Python 3.11 and up has tomllib built-in, for 3.10 and lower we use tomli which provides
     # the same functonality. We check if it's Python 3.10 or lower to patch the correct target.

tests/img2catalog/ega_fixtures.py

@@ -0,0 +1,31 @@
+import pytest
+
+
+@pytest.fixture
+def default_ega_dataset():
+    """A dataset description as returned by the EGA metadata API, taken from the ticket."""
+    return {
+        "accession_id": "EGAD00001005083",
+        "title": "300-Obese cohort gut microbiome data",
+        "description": (
+            "300-Obese cohort, Nijmegen, the Netherlands. Dataset contains gut microbiome data "
+            "generated by metagenomic sequencing."
+        ),
+        "dataset_types": ["Whole genome sequencing"],
+        "technologies": ["Illumina HiSeq 2000"],
+        "num_samples": 297,
+        "access_type": "controlled",
+        "is_in_beacon": False,
+        "is_released": True,
+        "released_date": "2001-01-01T00:00:00+01:00",
+        "is_deprecated": False,
+        "policy_accession_id": "EGAP00001001117",
+    }
+
+
+@pytest.fixture
+def missing_ega_dataset():
+    """An EGA dataset description missing mandatory fields (title, description)."""
+    return {
+        "accession_id": "EGAD00001005083",
+    }

tests/img2catalog/inputs/test_ega.py

@@ -0,0 +1,51 @@
+import pytest
+import requests
+
+from img2catalog.inputs.ega import fetch_ega_dataset, fetch_ega_datasets
+
+API_URL = "https://metadata.ega-archive.org"
+
+
+def test_fetch_ega_dataset_returns_json(requests_mock, default_ega_dataset):
+    # Arrange
+    requests_mock.get(f"{API_URL}/datasets/EGAD00001005083", json=default_ega_dataset)
+
+    # Act
+    result = fetch_ega_dataset("EGAD00001005083", API_URL)
+
+    # Assert
+    assert result == default_ega_dataset
+
+
+def test_fetch_ega_dataset_raises_on_404(requests_mock):
+    # Arrange
+    requests_mock.get(f"{API_URL}/datasets/EGAD00000000000", status_code=404)
+
+    # Act & Assert
+    with pytest.raises(requests.HTTPError):
+        fetch_ega_dataset("EGAD00000000000", API_URL)
+
+
+def test_fetch_ega_datasets_returns_all_on_success(requests_mock, default_ega_dataset):
+    # Arrange
+    other_dataset = {**default_ega_dataset, "dataset_id": "EGAD00001005084"}
+    requests_mock.get(f"{API_URL}/datasets/EGAD00001005083", json=default_ega_dataset)
+    requests_mock.get(f"{API_URL}/datasets/EGAD00001005084", json=other_dataset)
+
+    # Act
+    result = fetch_ega_datasets(["EGAD00001005083", "EGAD00001005084"], API_URL)
+
+    # Assert
+    assert result == [default_ega_dataset, other_dataset]
+
+
+def test_fetch_ega_datasets_skips_failed_dataset(requests_mock, default_ega_dataset):
+    # Arrange
+    requests_mock.get(f"{API_URL}/datasets/EGAD00001005083", json=default_ega_dataset)
+    requests_mock.get(f"{API_URL}/datasets/EGAD00000000000", status_code=404)
+
+    # Act
+    result = fetch_ega_datasets(["EGAD00001005083", "EGAD00000000000"], API_URL)
+
+    # Assert
+    assert result == [default_ega_dataset]