diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..17b0c37 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,4 @@ +[tool.isort] +profile = "black" +multi_line_output = 3 + diff --git a/src/crate_builder_script/api_calls.py b/src/crate_builder_script/api_calls.py new file mode 100644 index 0000000..5201a02 --- /dev/null +++ b/src/crate_builder_script/api_calls.py @@ -0,0 +1,90 @@ +"""Handle requests to HuBMAP APIs.""" +import logging +import os +from functools import lru_cache +from pprint import pformat +from typing import Any + +import requests + +LOGGER = logging.getLogger(__name__) + +ENTITY_API = "https://entity.api.hubmapconsortium.org" +ASSETS_API = "https://assets.hubmapconsortium.org" +UUID_API = "https://uuid.api.hubmapconsortium.org" + +AUTH_TOK = os.environ["AUTH_TOK"] + +HUBMAP = "https://hubmapconsortium.org/" + +HUBMAP_ORG_ENTITY = HUBMAP # for lack of a better choice + +ENTITY_BASE = "https://portal.hubmapconsortium.org/browse/" + +HUBMAP_ORG_ENTITY = HUBMAP # for lack of a better choice + + +def entity_iri(entity: Any) -> str: + """Return a resolvable IRI for a HuBMAP entity (dataset / sample / donor). + + Raw datasets and multi-assay splits carry `doi_url`; processed datasets, samples + and donors do not, and fall back to the portal browse URL, which resolves by + HuBMAP ID for every entity type. + """ + return entity.get("doi_url") or ENTITY_BASE + (entity.get("hubmap_id") or "") + + +@lru_cache(maxsize=8) +def fetch_entity_info(target_id: str) -> dict[str, Any]: + """Fetch a dataset's entity information.""" + LOGGER.debug("FETCH %s", target_id) + try: + resp = requests.get( + ENTITY_API + f"/entities/{target_id}", + headers={"Authorization": f"Bearer {AUTH_TOK}"}, + ) + resp.raise_for_status() + except Exception as excp: + raise RuntimeError(f"Fetch of entity {target_id} failed: {excp}") from excp + ds_info = resp.json() + LOGGER.debug("TOP LEVEL for %s:\n%s", target_id, pformat(ds_info, depth=1)) + LOGGER.debug( + "INGEST METADATA:\n%s", pformat(ds_info.get("ingest_metadata", {}), depth=2) + ) + LOGGER.debug("METADATA:\n%s", pformat(ds_info.get("metadata", {}), depth=2)) + LOGGER.debug( + "DIRECT ANCESTORS: %s\n%s", + [elt["hubmap_id"] for elt in ds_info.get("direct_ancestors", [])], + pformat(ds_info.get("direct_ancestors"), depth=2), + ) + LOGGER.debug( + "DIRECT ANCESTOR:\n%s", pformat(ds_info.get("direct_ancestor"), depth=2) + ) + if "direct_ancestors" in ds_info: + first_ancestor = ds_info["direct_ancestors"][0] + elif "direct_ancestor" in ds_info: + first_ancestor = ds_info["direct_ancestor"] + else: + first_ancestor = {} + LOGGER.debug( + "ANCESTOR INGEST MD\n%s", pformat(first_ancestor.get("ingest_metadata", {})) + ) + LOGGER.debug("ANCESTOR MD\n%s", pformat(first_ancestor.get("metadata", {}))) + return ds_info + + +@lru_cache(maxsize=8) +def fetch_uuid_files_info(target_id: str) -> dict[str, Any]: + """Fetch files informaion from the UUID API.""" + resp = requests.get( + UUID_API + f"/{target_id}/files", + headers={"Authorization": f"Bearer {AUTH_TOK}"}, + ) + resp.raise_for_status() + # LOGGER.debug("UUID FILES first 10:\n%s", pformat(resp.json()[:10])) + return resp.json() + + +def asset_url(uuid: str, rel_path: str) -> str: + """Build the URL by which a file should be available from the ASSETS API.""" + return f"{ASSETS_API}/{uuid}/{rel_path}" diff --git a/src/crate_builder_script/build_crate_from_dataset.py b/src/crate_builder_script/build_crate_from_dataset.py index cfb6e28..05e42d5 100644 --- a/src/crate_builder_script/build_crate_from_dataset.py +++ b/src/crate_builder_script/build_crate_from_dataset.py @@ -1,14 +1,28 @@ +"""Build an RO-Crate and a Croissant for a dataset.""" import argparse +import logging import os from collections import defaultdict from datetime import datetime, timezone -import logging from pprint import pformat -from typing import Any, List - -import requests -from rocrate.model.contextentity import ContextEntity -from rocrate.model.person import Person +from tempfile import TemporaryDirectory +from typing import List + +from api_calls import ( + HUBMAP_ORG_ENTITY, + asset_url, + fetch_entity_info, + fetch_uuid_files_info, +) +from croissant_wrapper import CroissantWrapper +from extractors import WrappedEntity +from rocrate.model import ( + ComputerLanguage, + ContextEntity, + Dataset, + Person, + SoftwareApplication, +) from rocrate.rocrate import ROCrate logging.basicConfig( @@ -17,50 +31,49 @@ LOGGER = logging.getLogger(__name__) -ENTITY_API = "https://entity.api.hubmapconsortium.org" -ASSETS_API = "https://assets.hubmapconsortium.org" -UUID_API = "https://uuid.api.hubmapconsortium.org" - DEFAULT_OUTPUT_PATH = "/tmp/crate_test" +CROISSANT_FILENAME = "croissant.json" + +APACHE_ORG_ENTITY = "https://apache.org" # Externally defined identifiers NIH_URI = "https://ror.org/01cwqze88" ORCID_URI = "https://orcid.org" OBOLIB_URI = "http://purl.obolibrary.org/obo" -# TARGET_ID = "HBM567.VCBK.562" -# TARGET_ID = "HBM487.HJZB.546" - -AUTH_TOK = os.environ["AUTH_TOK"] - - -def fetch_entity_info(target_id: str) -> dict[str, Any]: - resp = requests.get(ENTITY_API + f"/entities/{target_id}") - resp.raise_for_status() - ds_info = resp.json() - LOGGER.debug("TOP LEVEL:\n%s", pformat(ds_info, depth=1)) - LOGGER.debug("INGEST METADATA:\n%s", - pformat(ds_info.get("ingest_metadata", {})) - ) - LOGGER.debug("DIRECT ANCESTORS:\n%s", pformat(ds_info["direct_ancestors"], depth=2)) - return ds_info - - -def fetch_uuid_files_info(target_id: str) -> dict[str, Any]: - resp = requests.get( - UUID_API + f"/{target_id}/files", - headers={"Authorization": f"Bearer {AUTH_TOK}"}, - ) - resp.raise_for_status() - LOGGER.debug("UUID FILES first 10:\n%s", pformat(resp.json()[:10])) - return resp.json() - - -def asset_url(uuid: str, rel_path: str) -> str: - return f"{ASSETS_API}/{uuid}/{rel_path}" +AIRFLOW_VERSION = "2.11.0" +CWL_VERSION = "v1.1" + +DEFAULT_INGEST_PIPELINE_URL = "https://github.com/hubmapconsortium/ingest-pipeline" + +############### +# Notes- +# - count_versions() is essentially untested, for lack of an example +# - croissant cite_as uses the DOI, and that only gets set for primary datasets. Do we +# want to reference the primary dataset's DOI in the derived dataset's cite_as? +# - Writing a croissant for a file in an unpublished dataset results in an error at +# validation time because the file block information from uuid-api has not yet been +# set so the sha256 code is not known. +# - We need the Airflow version to write a valid provenance block, but +# AFAIK it is not maintained in the entity information. Likewise, some +# datasets were produced with python versions before 3.11. Likewise, I've +# hard-coded the CWL version, but ours is actually modified- the CWL language +# definition entity should point at ours rather than at default CWL. +# - the workflow_instance lacks start and end dates +# - "homo sapiens" is hard-coded into the keywords. That will have to be +# fixed for SenNet. +# - unpublished examples: +# TARGET_ID = "HBM567.VCBK.562" +# TARGET_ID = "HBM487.HJZB.546" # primary dataset +# - published examples: +# TARGET_ID = "HBM866.VMBK.952" +# TARGET_ID = "HBM473.QLDT.264" # primary dataset +# TARGET_ID = "HBM748.CHWC.963" # derived from snare-seq +############### def build_funder_entity(crate: ROCrate) -> ContextEntity: + """Build an entity representing the NIH.""" funder_props = { "@id": NIH_URI, "@type": "Organization", @@ -71,6 +84,7 @@ def build_funder_entity(crate: ROCrate) -> ContextEntity: def build_license_entity(crate: ROCrate) -> ContextEntity: + """Build an entity representing the Creative Commons license.""" license_props = { "@type": "CreativeWork", "name": "Creative Commons Atribution 4.0 International", @@ -88,6 +102,7 @@ def build_license_entity(crate: ROCrate) -> ContextEntity: def build_pi_entity(crate: ROCrate) -> ContextEntity: + """Build an entity representing the Principal Investigator role.""" props = { "@id": "#role-principal-investigator", "@type": "Role", @@ -99,6 +114,7 @@ def build_pi_entity(crate: ROCrate) -> ContextEntity: def build_contact_entity(crate: ROCrate) -> ContextEntity: + """Build an entity representing the contact person for the dataset.""" props = { "@id": "#role-contact", "@type": "Role", @@ -115,6 +131,7 @@ def build_contact_entity(crate: ROCrate) -> ContextEntity: def build_ia_entity(crate: ROCrate) -> ContextEntity: + """Build an entity representing the operator for a dataset.""" props = { "@id": "#role-investigative-agent", "@type": "Role", @@ -132,14 +149,24 @@ def build_ia_entity(crate: ROCrate) -> ContextEntity: def build_contributors(crate: ROCrate, contributors: List[dict]) -> List[ContextEntity]: + """Construct a representation of contributors information.""" ent_l = [] role_d = {} role_list_d = defaultdict(list) + for contrib in contributors: + orcid = contrib.get("orcid", contrib.get("orcid_id")) + if not orcid: + LOGGER.error("Contributor with no orcid ID: %s", pformat(contrib)) + raise RuntimeError("Missing contributor orcid ID") + display_name = contrib.get("display_name", contrib.get("name")) + if not display_name: + LOGGER.error("Contributor with no display name: %s", pformat(contrib)) + raise RuntimeError("Missing contributor display name") props = { "@type": "Person", - "@id": f"{ORCID_URI}/{contrib['orcid']}", - "name": contrib["display_name"], + "@id": f"{ORCID_URI}/{orcid}", + "name": display_name, } person = Person(crate, identifier=props["@id"], properties=props) add_to_entities = True @@ -148,7 +175,7 @@ def build_contributors(crate: ROCrate, contributors: List[dict]) -> List[Context ("is_contact", build_contact_entity), ("is_operator", build_ia_entity), ]: - if contrib[match_key].lower() == "yes": + if contrib.get(match_key, "").lower() == "yes": entity = role_d.get(match_key) or builder(crate) role_d.setdefault(match_key, entity) role_list_d[match_key].append(crate.add(person)) @@ -161,11 +188,302 @@ def build_contributors(crate: ROCrate, contributors: List[dict]) -> List[Context return ent_l +def build_hubmap_org_entity(crate: ROCrate) -> ContextEntity: + """Build an entity representing the HuBMAP organization.""" + hubmap_org = crate.add( + ContextEntity( + crate, + HUBMAP_ORG_ENTITY, + properties={ + "@type": "Organization", + "name": "HuBMAP Consortium", + "url": HUBMAP_ORG_ENTITY, + }, + ) + ) + return hubmap_org + + +def build_primary_prov(ds_entity: WrappedEntity, crate: ROCrate) -> ContextEntity: + """Construct minimal dummy provenance.""" + hubmap_org = build_hubmap_org_entity(crate) + agent = crate.add( + SoftwareApplication( + crate, + "HuBMAP Process", + properties={"parentOrganization": {"@id": hubmap_org.id}}, + ) + ) + instrument_model = ds_entity["metadata"].get("acquisition_instrument_model", "") + instrument_entity = crate.add( + ContextEntity( + crate, + "#instrument", + properties={ + "@type": ["IndividualProduct", "ComputationalWorkflow"], + "name": instrument_model, + }, + ) + ) + props = { + "@id": "#some_workflow", + "@type": "CreateAction", + "name": "the-create-action", + "agent": {"@id": agent.id}, + "instrument": {"@id": instrument_entity.id}, + "object": [], + "result": [{"@id": "./"}], # the target dataset + "actionStatus": "CompletedActionStatus", + } + return ContextEntity(crate, identifier=props["@id"], properties=props) + + +def build_cwl_entity(crate: ROCrate) -> ContextEntity: + """Build an entity representing the CWL language.""" + props = { + "@id": "https://w3id.org/workflowhub/workflow-ro-crate#cwl", + "@type": "ComputerLanguage", + "name": "Common Workflow Language", + "alternateName": "CWL", + "identifier": {"@id": f"https://w3id.org/cwl/{CWL_VERSION}/"}, + "url": "https://www.commonwl.org/", + } + return crate.add(ContextEntity(crate, identifier=props["@id"], properties=props)) + + +def build_step_entity( + step: dict, + idx: int, + cwl_entity: ContextEntity, + python_entity: ContextEntity, + crate: ROCrate, +) -> ContextEntity: + """Build an entity representing one step of the provenance chain.""" + pos = idx + 1 + id_str = f"#step_{pos}" + hash = step["commit"] + if step["cwl"]: + return crate.add( + ContextEntity( + crate, + id_str, + properties={ + "position": pos, + "@type": "HowToStep", + "name": step["cwl"], + "description": step["name"], + "version": hash, + "url": step["repo"], + "codeRepository": step["repo"], + "programmingLanguage": cwl_entity.id, + }, + ) + ) + else: + return crate.add( + ContextEntity( + crate, + id_str, + properties={ + "position": pos, + "@type": "HowToStep", + "name": step["name"], + "description": step["name"], + "version": hash, + "url": step["repo"], + "codeRepository": step["repo"], + "programmingLanguage": python_entity.id, + }, + ) + ) + + +def build_derived_prov(ds_entity: WrappedEntity, crate: ROCrate) -> ContextEntity: + """Build a set of entities representing the provenance by which a derived dataset is produced from one or more primary datasets.""" + prov_chain = ds_entity.walk_ancestors(lambda d: d["entity_type"] == "Dataset") + assert len(prov_chain) == 1 + assert len(prov_chain[0]) == 3 + parent_chain = prov_chain[0][2] + parent_id_list = [] + for tuple in parent_chain: + parent_id, parent_info = tuple[0:2] + crate.add( + Dataset( + crate, + parent_info["doi_url"], + properties={ + "name": parent_id, + "description": parent_info.get("description", ""), + }, + ) + ) + parent_id_list.append(parent_info["doi_url"]) + hubmap_org = build_hubmap_org_entity(crate) + apache_org = crate.add( + ContextEntity( + crate, + APACHE_ORG_ENTITY, + properties={ + "@type": "Organization", + "name": "Apache Software Foundation", + "url": APACHE_ORG_ENTITY, + }, + ) + ) + airflow = crate.add( + SoftwareApplication( + crate, + "Apache Airflow", + properties={ + "version": AIRFLOW_VERSION, + "description": ( + "Apache Airflow - A platform to programmatically author," + " schedule, and monitor workflows" + ), + "publisher": {"@id": apache_org.id}, + }, + ) + ) + python_lang = crate.add( + ComputerLanguage( + crate, + identifier="https://python.org", + properties={ + "name": "Python", + "version": "3.11", + "url": "https://python.org", + }, + ) + ) + cwl_entity = build_cwl_entity(crate) + if all_steps := ds_entity.pipeline_steps(): + assert all_steps[0]["name"] == "ingest-pipeline" and not all_steps[0]["cwl"] + ingest_pipeline_info = all_steps[0] + hash = ingest_pipeline_info["commit"] + desc = ( + ds_entity["ingest_metadata"]["workflow_description"] + if "ingest_metadata" in ds_entity + and "workflow_description" in ds_entity["ingest_metadata"] + else "Processing steps implemented by an ingest-pipeline DAG" + ) + cwl_steps = all_steps[1:] + step_list = [ + build_step_entity(step, idx, cwl_entity, python_lang, crate) + for idx, step in enumerate(cwl_steps) + ] + workflow = crate.add( + ContextEntity( + crate, + "#workflow", + properties={ + "@type": ["ComputationalWorkflow", "SoftwareApplication"], + "name": "ingest-pipeline dag workflow", + "description": desc, + "steps": step_list, + "version": hash, + "url": ingest_pipeline_info["repo"], + "codeRepository": ingest_pipeline_info["repo"], + "downloadUrl": f"{ingest_pipeline_info['repo']}/archive/{hash}.zip", + "publisher": {"@id": hubmap_org.id}, + "programmingLanguage": {"@id": python_lang.id}, + "softwareRequirements": {"@id": airflow.id}, + }, + ) + ) + props = { + "@id": "#workflow_instance", + "@type": "CreateAction", + "name": "the-create-action", + "agent": {"@id": workflow.id}, + "instrument": {"@id": workflow.id}, + "object": [{"@id": this_id} for this_id in parent_id_list], + "result": [{"@id": "./"}], # the target dataset + "actionStatus": "CompletedActionStatus", + } + else: + workflow = crate.add( + ContextEntity( + crate, + "#workflow", + properties={ + "@type": ["ComputationalWorkflow", "SoftwareApplication"], + "name": "ingest-pipeline dag workflow", + "description": "Processing steps implemented by an ingest-pipeline DAG", + "url": DEFAULT_INGEST_PIPELINE_URL, + "codeRepository": DEFAULT_INGEST_PIPELINE_URL, + "publisher": {"@id": hubmap_org.id}, + "programmingLanguage": {"@id": python_lang.id}, + "softwareRequirements": {"@id": airflow.id}, + }, + ) + ) + props = { + "@id": "#workflow_instance", + "@type": "CreateAction", + "name": "the-create-action", + "agent": {"@id": workflow.id}, + "instrument": {"@id": workflow.id}, + "object": [{"@id": this_id} for this_id in parent_id_list], + "result": [{"@id": "./"}], # the target dataset + "actionStatus": "CompletedActionStatus", + } + return ContextEntity(crate, identifier=props["@id"], properties=props) + + +def build_profiles(crate: ROCrate) -> tuple: + """Build ContextElements for several profiles needed to describe a workflow.""" + base_crate_ctx_id = f"https://w3id.org/ro/crate/{crate.version}" + crate_profile = crate.add( + ContextEntity( + crate, + base_crate_ctx_id, + properties={ + "@type": ["CreativeWork", "Profile"], + "name": "RO-Crate Profile", + "version": crate.version, + }, + ) + ) + proc_profile = crate.add( + ContextEntity( + crate, + "https://w3id.org/ro/wfrun/process/0.5", + properties={ + "@type": ["CreativeWork", "Profile"], + "name": "Process Run Crate Profile", + "version": "0.5", + }, + ) + ) + # # These two profiles are needed for full software workflows + # + # wf_profile = crate.add(ContextEntity( + # crate, + # "https://w3id.org/ro/wfrun/workflow/0.5", + # properties={ + # "@type": ["CreativeWork", "Profile"], + # "name": "Workflow Run Crate Profile", + # "version": "0.5", + # # "isProfileOf": {"@id": proc_profile.id} + # } + # )) + # wfc_profile = crate.add(ContextEntity( + # crate, + # "https://w3id.org/workflowhub/workflow-ro-crate/1.0", + # properties={ + # "@type": ["CreativeWork", "Profile"], + # "name": "Workflow Run RO-Crate", + # "version": "1.0" + # } + # )) + # return (crate_profile, proc_profile, wf_profile, wfc_profile) + return (crate_profile, proc_profile) + + def main() -> None: + """Parse command line and carry out construction of the RO-Crate and Croissant files.""" parser = argparse.ArgumentParser() - parser.add_argument( - "target_id", help="Build an RO-Crate for this published dataset" - ) + parser.add_argument("target_id", help="Build an RO-Crate for this dataset") parser.add_argument( "--outdir", "-o", @@ -175,11 +493,30 @@ def main() -> None: ), default=DEFAULT_OUTPUT_PATH, ) + parser.add_argument( + "--debug", + "-d", + action="store_true", + help="Enable debug logging", + ) + parser.add_argument( + "--include_all_files", + action="store_true", + help="Include all files, even if some are not data product or qa resources", + ) args = parser.parse_args() target_id = args.target_id outdir = args.outdir + debug = args.debug + include_all_files = args.include_all_files - ds_info = fetch_entity_info(target_id) + if debug: + LOGGER.setLevel(logging.DEBUG) + logging.getLogger("requests").setLevel(logging.DEBUG) + logging.getLogger("urllib3").setLevel(logging.DEBUG) + logging.getLogger("api_calls").setLevel(logging.DEBUG) + logging.getLogger("croissant_wrapper").setLevel(logging.DEBUG) + ds_entity = WrappedEntity(fetch_entity_info(target_id)) uuid_files = fetch_uuid_files_info(target_id) blk_idx = {} @@ -189,45 +526,117 @@ def main() -> None: # The dataset DOIs point to the Portal, which is basically a landing page, which is # forbidden as the direct link for a dataset under FAIR. So we can't use the DOI # as the crate root dataset id. - crate = ROCrate() + # The version needs to be 1.1 to avoid compatibility problems with the workflow + # profile that seem to exist for crate profile 1.2. + crate = ROCrate(version="1.1") - if "doi_url" in ds_info: - doi_url = ds_info["doi_url"] - crate.root_dataset["identifier"] = doi_url - crate.root_dataset["sameAs"] = doi_url + for profile in build_profiles(crate): + crate.root_dataset.append_to("conformsTo", {"@id": profile.id}) + + crate.metadata.extra_contexts.append( + "https://w3id.org/ro/terms/workflow-run/context" + ) + + wrapped_croissant = CroissantWrapper(target_id, ds_entity["title"]) crate.root_dataset["name"] = target_id - crate.root_dataset["description"] = ds_info["title"] - crate.root_dataset["datePublished"] = str( - datetime.fromtimestamp(ds_info["published_timestamp"] // 1000).astimezone( - timezone.utc + crate.root_dataset["description"] = ds_entity["title"] + if "doi_url" in ds_entity: + doi_url = ds_entity["doi_url"] + crate.root_dataset["identifier"] = doi_url + crate.root_dataset["sameAs"] = doi_url + wrapped_croissant.cite_as_doi = doi_url + elif "doi_url" in ds_entity["direct_ancestors"][0]: + wrapped_croissant.cite_as_doi = ds_entity["direct_ancestors"][0]["doi_url"] + + if "published_timestamp" in ds_entity: + date_published = str( + datetime.fromtimestamp(ds_entity["published_timestamp"] // 1000).astimezone( + timezone.utc + ) ) - ) + crate.root_dataset["datePublished"] = date_published + wrapped_croissant.date_published = date_published + + license_entity = build_license_entity(crate) + crate.root_dataset["license"] = crate.add(license_entity) + wrapped_croissant.license = license_entity.properties()["url"] - crate.root_dataset["license"] = crate.add(build_license_entity(crate)) crate.root_dataset["funder"] = crate.add(build_funder_entity(crate)) - if contributors := ds_info.get("contributors"): + + ds_version = ds_entity.count_versions() + crate.root_dataset["version"] = ds_version + wrapped_croissant.version = ds_version + + if kwds := ds_entity.keywords(): + crate.root_dataset["keywords"] = kwds + wrapped_croissant.keywords = kwds + + if contributors := ds_entity.get("contributors"): crate.add(build_pi_entity(crate)) ent_l = build_contributors(crate, contributors) [crate.add(ent) for ent in ent_l] crate.root_dataset["contributor"] = ent_l - if "files" in ds_info: + if ds_entity.is_processed: + crate.add(build_derived_prov(ds_entity, crate)) + else: + crate.add(build_primary_prov(ds_entity, crate)) + + if "files" in ds_entity: # This is a derived dataset- include only data products and qa_qc files - for fl in ds_info["files"]: - if fl["is_data_product"] or fl["is_qa_qc"]: + for fl in ds_entity["files"]: + if ( + fl.get("is_data_product", False) + or fl.get("is_qa_qc", False) + or include_all_files + ): LOGGER.debug(f"Adding {fl['rel_path']}") crate.add_file( - asset_url(ds_info["uuid"], fl["rel_path"]), validate_url=True + asset_url(ds_entity["uuid"], fl["rel_path"]), validate_url=True + ) + wrapped_croissant.add_file( + ds_entity["uuid"], fl, blk_idx.get(fl["rel_path"]) ) else: - LOGGER.debug(f"{fl['rel_path']} is not a data product") + LOGGER.debug(f"{fl['rel_path']} is not a data product or qa_qc") else: for fl_blk in blk_idx.values(): crate.add_file( - asset_url(ds_info["uuid"], fl_blk["path"]), validate_url=True + asset_url(ds_entity["uuid"], fl_blk["path"]), validate_url=True ) - crate.write(outdir) + # We have no descriptive info for these files, so it's hard + # to see how we could add them to the Croissant object + + tmpdir = TemporaryDirectory() + + if not os.path.isdir(outdir): + os.makedirs(outdir, exist_ok=True) + wrapped_croissant.write(os.path.join(tmpdir.name, CROISSANT_FILENAME)) + crate.add( + ContextEntity( + crate, + "http://mlcommons.org/croissant/1.0", + properties={ + "@type": ["CreativeWork", "Profile"], + "name": "MLCommons Croissant Format Specification", + "version": "1.0", + "url": "https://docs.mlcommons.org/croissant/docs/crossant-spec-1.0.html", + }, + ) + ) + crate.add_file( + os.path.join(tmpdir.name, CROISSANT_FILENAME), + properties={ + "name": "Croissant Metadata Descriptor", + "description": "Machine learning data-loading configurations for this dataset.", + "encodingFormat": "application/ld+json", + "conformsTo": {"@id": "http://mlcommons.org/croissant/1.0"}, + }, + ) + + crate.write_zip(os.path.join(outdir, f"{target_id}_crate.zip")) + tmpdir.cleanup() if __name__ == "__main__": diff --git a/src/crate_builder_script/croissant_wrapper.py b/src/crate_builder_script/croissant_wrapper.py new file mode 100644 index 0000000..6479bd0 --- /dev/null +++ b/src/crate_builder_script/croissant_wrapper.py @@ -0,0 +1,369 @@ +"""Provides functions to build the Croissant file.""" +import json +import logging +from datetime import datetime, timezone +from pathlib import Path +from pprint import pformat + +import mlcroissant as mlc +from api_calls import asset_url, entity_iri, fetch_entity_info +from extractors import WrappedEntity + +LOGGER = logging.getLogger(__name__) + +EDAM_INFO = { + ("EDAM_1.24.format_3727", None): {"desc": "tiff", "mime": "image/tiff"}, + ("EDAM_1.24.format_3464", None): {"desc": "json", "mime": "application/json"}, + ("EDAM_1.24.format_3508", None): {"desc": "pdf", "mime": "application/pdf"}, + ("EDAM_1.24.format_3590", None): {"desc": "hdf5", "mime": "application/x-hdf5"}, + ("EDAM_1.24.format_3987", None): {"desc": "zip", "mime": "application/zip"}, + ("EDAM_1.24.format_3752", None): {"desc": "csv", "mime": "text/csv"}, + ("EDAM_1.24.format_3755", None): { + "desc": "tsv", + "mime": "text/tab-separated-values", + }, + ("EDAM_1.24.format_3790", None): { + "desc": "h5ad (anndata)", + "mime": "application/x-hdf5", + }, + ("EDAM_1.24.format_3915", None): {"desc": "zarr", "mime": "application/vnd.zarr"}, + ("EDAM_1.24.format_4006", None): { + "desc": "zarr (spatialdata)", + "mime": "application/vnd.zarr", + }, + ("EDAM_1.24.data_3671", None): {"desc": "plain text", "mime": "text/plain"}, + ("EDAM_1.24.format_3916", ".mtx"): { + "desc": "adjacency matrix", + "mime": "text/plain", + }, + (None, ".html"): {"desc": "unknown but html encoded", "mime": "text/html"}, + (None, ".tsv"): { + "desc": "unknown but tsv encoded", + "mime": "text/tab-separated-values", + }, + (None, ".csv"): {"desc": "unknown but csv encoded", "mime": "text/csv"}, + ("EDAM_1.24.format_2333", None): { + "desc": "binary format", + "mime": "application/octet-stream", + }, + ("EDAM_1.24.format_3003", None): { + "desc": "BED (Browser Extensible Data)", + "mime": "text/x-bed", + }, +} + + +def _ts_to_iso(ms): + return ( + datetime.fromtimestamp(ms / 1000, tz=timezone.utc).date().isoformat() + if ms + else None + ) + + +def _creators(entity: dict) -> list[dict]: + out = [] + for c in entity.get("contributors", []): + name = ( + c.get("name") + or f"{c.get('first_name', '')} {c.get('last_name', '')}".strip() + ) + if not name: + continue + person = {"name": name} + orcid = (c.get("orcid_id") or c.get("orcid") or "").strip() + if orcid: + person["url"] = ( + orcid if orcid.startswith("http") else f"https://orcid.org/{orcid}" + ) + out.append(person) + return out + + +def _build_citation(entity: WrappedEntity, doi) -> str: + authors = "; ".join(c["name"] for c in _creators(entity)) or "HuBMAP Consortium" + year = (_ts_to_iso(entity.get("published_timestamp")) or "")[:4] + title = entity.get("title") or entity.get("hubmap_id", "") + keystr = f"HUBMAP_{entity['hubmap_id']}" + short_doi = ( + doi.replace("doi.org/", "").replace("http://", "").replace("https://", "") + ) + return ( + "@data{" + f"{keystr}, " + f"author={{{authors}}}, " + f"title={{{title}}}, " + f"year={{{year}}}, " + f"version={{{entity.count_versions()}}}, " + f"doi={short_doi}, " + f"url={{{doi}}}, " + "publisher={HuBMAP}}" + ) + + +def _protocol_dois(md: dict) -> list[str]: + dois = [] + for k in ( + "preparation_protocol_doi", + "reagent_prep_protocols_io_doi", + "section_prep_protocols_io_doi", + ): + v = (md.get(k) or "").strip() + if not v: + continue + dois.append(f"https://dx.doi.org/{v}" if v.startswith("10.") else v) + return dois + + +def _acquisition_activity(entity: WrappedEntity, md: dict) -> dict: + def agent(name, role=None, org=False): + n = { + "@type": "prov:Organization" if org else "prov:Person", + "schema:name": name, + } + if role: + n["prov:role"] = role + return n + + assoc = [] + if md.get("pi"): + assoc.append(agent(md["pi"], role="principal investigator")) + if md.get("operator"): + assoc.append(agent(md["operator"], role="operator")) + if entity.get("group_name"): + assoc.append(agent(entity["group_name"], org=True)) + act = { + "@type": "prov:Activity", + "schema:name": f"{entity.get('dataset_type', 'assay')} acquisition", + "hubmap:instrument": " ".join( + filter( + None, + [ + md.get("acquisition_instrument_vendor"), + md.get("acquisition_instrument_model"), + ], + ) + ), + "hubmap:numberOfAntibodies": md.get("number_of_antibodies"), + "hubmap:numberOfImagingRounds": md.get("number_of_biomarker_imaging_rounds"), + "hubmap:numberOfChannels": md.get("number_of_channels"), + "prov:wasAssociatedWith": assoc, + } + if md.get("execution_datetime"): + act["prov:startedAtTime"] = md["execution_datetime"] + protocols = [ + { + "@type": ["prov:Entity", "schema:CreativeWork"], + "@id": d, + "prov:role": "protocol", + } + for d in _protocol_dois(md) + ] + if protocols: + act["prov:used"] = protocols + return {k: v for k, v in act.items() if v not in (None, "", [])} + + +def _specimen_chain(entity: WrappedEntity) -> dict: + def node(anc): + n = { + "@type": "prov:Entity", + "@id": entity_iri(anc), + "schema:name": anc.get("hubmap_id"), + "hubmap:entityType": anc.get("entity_type"), + "hubmap:sampleCategory": anc.get("sample_category"), + } + rui = anc.get("rui_location") + if rui: + r = json.loads(rui) if isinstance(rui, str) else rui + n["hubmap:ccfAnnotations"] = r.get("ccf_annotations") + n["hubmap:dimensions"] = { + "x": r.get("x_dimension"), + "y": r.get("y_dimension"), + "z": r.get("z_dimension"), + "unit": r.get("dimension_units"), + } + return {k: v for k, v in n.items() if v not in (None, "", [])} + + ancs = entity.list_ancestors(omit_test=lambda dct: dct["entity_type"] == "Dataset") + order = {"section": 0, "block": 1, "organ": 2} + ancs.sort(key=lambda a: order.get(a.get("sample_category"), 4)) + derived = None + for anc in reversed(ancs): + n = node(anc) + if derived: + n["prov:wasDerivedFrom"] = derived + derived = n + return derived + + +def _pipeline_activity(entity: WrappedEntity) -> dict: + agents = [] + for st in entity.pipeline_steps(): + a = { + "@type": ["prov:SoftwareAgent", "schema:SoftwareApplication"], + "schema:name": st["name"] + (f" [{st['cwl']}]" if st["cwl"] else ""), + "schema:codeRepository": st["repo"], + "hubmap:commit": st["commit"], + } + agents.append({k: v for k, v in a.items() if v}) + act = { + "@type": "prov:Activity", + "schema:name": "HuBMAP uniform processing pipeline", + } + if agents: + act["prov:wasAssociatedWith"] = agents + return act + + +def _build_raw_node(raw_entity: WrappedEntity, raw_md: dict | None) -> dict: + return { + "@type": "prov:Entity", + "@id": entity_iri(raw_entity), + "schema:name": raw_entity.get("hubmap_id"), + "hubmap:datasetType": raw_entity.get("dataset_type"), + "prov:wasGeneratedBy": _acquisition_activity(raw_entity, raw_md or {}), + "prov:wasDerivedFrom": _specimen_chain(raw_entity), + } + + +def build_embedded_provenance( + entity: WrappedEntity, md: dict | None, descendants: list | None = None +) -> dict: + """ + Build embedded provenance for the given entity. + + PROCESSED subject: wasGeneratedBy its own pipeline; wasDerivedFrom the raw parent + (which carries the acquisition activity + specimen chain). RAW subject: wasGeneratedBy + acquisition; wasDerivedFrom specimen chain; + a light forward pointer to processed versions. + """ + if entity.is_processed: + hubmap_id = entity["hubmap_id"] + ancestor_chain = entity.walk_ancestors(lambda d: d["entity_type"] == "Dataset") + assert len(ancestor_chain) == 1, "internal error walking ancestors" + check_id, ignored_entity, ancestors = ancestor_chain[0] + assert check_id == hubmap_id + raw_node_l = [] + for anc in ancestors: + anc_entity = WrappedEntity(anc[1]) # each ancestor is a 3-tuple + anc_md = anc_entity.get("metadata") + raw_node_l.append(_build_raw_node(anc_entity, anc_md)) + return { + "prov:wasGeneratedBy": _pipeline_activity(entity), + "prov:wasDerivedFrom": ( + raw_node_l[0] if len(raw_node_l) == 1 else raw_node_l + ), + } + + provo = {"prov:wasGeneratedBy": _acquisition_activity(entity, md)} + if chain := _specimen_chain(entity): + provo["prov:wasDerivedFrom"] = chain + if descendants: + provo["hubmap:hasProcessedDataset"] = [ + { + "@type": "prov:Entity", + "@id": entity_iri(d), + "schema:name": d.get("hubmap_id"), + "hubmap:datasetType": d.get("dataset_type"), + } + for d in descendants + ] + return provo + + +class CroissantWrapper: + """Wraps Croissant generation functionality.""" + + @classmethod + def test(cls, entity_dict: dict) -> None: + """Provide a test of utility methods.""" + LOGGER.info( + "Testing CroissantWrapper:\n%s", + pformat( + build_embedded_provenance( + entity_dict, + md=entity_dict.get("metadata"), + descendants=entity_dict.get("direct_descendants", []), + ) + ), + ) + + def __init__(self, name: str, description: str): + """Construct a CroissantWrapper instance.""" + self.name = name + self.description = description + self.file_objects = [] + self.record_sets = [] + self.cite_as_doi = None + self.date_published = None + self.license = None + self.version = None + self.keywords = None + + def add_file(self, ds_uuid: str, file_info: dict, file_blk: dict | None) -> None: + """Define and describe a single file in the dataset.""" + args = { + "id": file_info["rel_path"], + "name": file_info["rel_path"], + "description": file_info["description"], + "content_url": asset_url(ds_uuid, file_info["rel_path"]), + } + path_ext = Path(file_info["rel_path"]).suffix + if edam := file_info.get("edam_term"): + if (edam, None) in EDAM_INFO: + args["encoding_formats"] = [EDAM_INFO[(edam, None)]["mime"]] + elif (edam, path_ext) in EDAM_INFO: + args["encoding_formats"] = [EDAM_INFO[(edam, path_ext)]["mime"]] + elif (None, path_ext) in EDAM_INFO: + args["encoding_formats"] = [EDAM_INFO[(None, path_ext)]["mime"]] + else: + LOGGER.warning(f"Unknown EDAM format {edam} for {pformat(file_info)}") + args["encoding_formats"] = ["application/octet-stream"] + else: + args["encoding_formats"] = ["application/octet-stream"] + if file_blk: + args["sha256"] = file_blk["sha256_checksum"] + self.file_objects.append(mlc.FileObject(**args)) + + # def add_record_set(self, record_set: mlc.RecordSet): + # self.record_sets.append(record_set) + + def write(self, croissant_filename: str): + """Write the Croissant file.""" + entity = WrappedEntity(fetch_entity_info(self.name)) + args = { + "id": "croissant-spec", + "name": self.name, + "description": self.description, + "distribution": self.file_objects, + "record_sets": self.record_sets, + "ctx": mlc.Context(is_live_dataset=False), + } + if self.date_published: + args["date_published"] = self.date_published + if self.license: + args["license"] = self.license + if self.version: + args["version"] = self.version + if self.cite_as_doi: + args["cite_as"] = _build_citation(entity, self.cite_as_doi) + if self.keywords: + args["keywords"] = self.keywords + croissant_meta = mlc.Metadata(**args).to_json() + croissant_meta["@context"].update( + { + "hubmap": "https://hubmapconsortium.org/", + "mlc": "https://mlcommons.org/", + "prov": "http://www.w3.org/ns/prov#", + "schema": "http://schema.org/", + } + ) + croissant_meta.update( + build_embedded_provenance( + entity, + md=entity.get("metadata"), # md + descendants=entity.get("direct_descendants", []), + ) + ) + with open(croissant_filename, "w", encoding="utf-8") as f: + json.dump(croissant_meta, f, indent=2) diff --git a/src/crate_builder_script/extractors.py b/src/crate_builder_script/extractors.py new file mode 100644 index 0000000..567f8ac --- /dev/null +++ b/src/crate_builder_script/extractors.py @@ -0,0 +1,237 @@ +"""Define utilities to extract relevant subsets of entity data.""" + +import json +import logging +from collections.abc import Callable +from pprint import pformat +from typing import Any + +from api_calls import fetch_entity_info + +LOGGER = logging.getLogger(__name__) + +# HuBMAP two-letter organ codes -> (label, UBERON term) +ORGAN_MAP = { + "LY": ("Lymph Node", "UBERON:0000029"), + "SP": ("Spleen", "UBERON:0002106"), + "TH": ("Thymus", "UBERON:0002370"), + "BM": ("Bone Marrow", "UBERON:0002371"), + "LK": ("Kidney (left)", "UBERON:0004538"), + "RK": ("Kidney (right)", "UBERON:0004539"), + "HT": ("Heart", "UBERON:0000948"), + "LI": ("Large Intestine", "UBERON:0000059"), + "SI": ("Small Intestine", "UBERON:0002108"), + "LL": ("Lung (left)", "UBERON:0002168"), + "RL": ("Lung (right)", "UBERON:0002167"), + "LV": ("Liver", "UBERON:0002107"), + "PA": ("Pancreas", "UBERON:0001264"), + "BL": ("Bladder", "UBERON:0001255"), + "SK": ("Skin", "UBERON:0002097"), + "LF": ("Fallopian tube (left)", "UBERON:0001303"), + "RF": ("Fallopian tube (right)", "UBERON:0001302"), +} + + +def walk_ancestors( + entity: dict, continue_test: Callable[[dict], bool] = lambda ent: True +) -> list[tuple]: + """ + Extract the chain of ancestors for a dataset. + + Given an entity dictionary, return a list of tuples. Each tuple has + the form: + (hubmap_id entity_dict list-of-ancestors) + where list-of-ancestors is None or a list of tuples of the same form. + + continue_test takes an entity dict as a parameter and returns True if + the descent should continue to the children of that entity, False otherwise. + """ + rslt = [] + if not continue_test(entity): + return rslt + e_type = entity.get("entity_type") + e_id = entity.get("hubmap_id") + LOGGER.debug(f"walk_ancestors {e_id} {e_type}") + if e_type == "Dataset": + if not entity.get("direct_ancestors"): + entity = WrappedEntity(fetch_entity_info(e_id)) + assert entity["hubmap_id"] == e_id + assert entity["entity_type"] == e_type + assert entity[ + "direct_ancestors" + ], f"{e_id} is a dataset with no direct ancestors" + ancs = [ + walk_ancestors(anc, continue_test) + for anc in entity.get("direct_ancestors", []) + ] + all_tuples = [] + for sub_list in ancs: + assert isinstance(sub_list, list) + all_tuples.extend(sub_list) + rslt.append((e_id, entity, all_tuples)) + elif e_type in ("Sample", "Donor"): + e_cat = entity.get("sample_category", "UNKNOWN SAMPLE CATEGORY") + LOGGER.debug(f"walk_ancestors sample category is {e_cat}") + if "direct_ancestor" not in entity: + LOGGER.debug(f"walk_ancestors fetching dead-end sample {e_id}") + entity = fetch_entity_info(e_id) + LOGGER.debug("walk_ancestors fetch yielded:\n%s", pformat(entity, depth=2)) + LOGGER.debug("walk_ancestors end of walk jump result") + new_entity = entity.get("direct_ancestor", {}) + if e_type == "Donor": + rslt.append((e_id, entity, None)) + else: + rslt.append((e_id, entity, walk_ancestors(new_entity, continue_test))) + else: + LOGGER.warning(f"walk_ancestors UNKNOWN ETYPE {e_type} for {e_id}") + return rslt + + +def listify(ancestor_chain: list, omit_test: Callable[[dict], bool]) -> list: + """Transform the output of walk_ancestors() to a list.""" + assert len(ancestor_chain) == 1, "listify must start on a 1-tuple chain" + hubmap_id, entity_dict, ancestors = ancestor_chain[0] + rslt = [] + if not omit_test(entity_dict): + rslt.append(entity_dict) + if ancestors: + for anc in ancestors: + rslt.extend(listify([anc], omit_test)) + return rslt + + +def is_processed(entity: dict) -> bool: + """ + Distinguished processed datasets from raw (primary) datasets. + + Raw vs processed: `creation_action` is 'Create Dataset Activity' vs 'Central Process'. + """ + return "process" in entity.get("creation_action", "").lower() + + +def _own_dag(entity: dict) -> list: + return entity.get("ingest_metadata", {}).get("dag_provenance_list", []) + + +def pipeline_steps(entity: dict) -> list[dict]: + """Return a list of dicts describing the steps by which a derived dataset is created.""" + dag_list = _own_dag(entity) + steps, seen = [], set() + for s in dag_list or []: + repo = s.get("origin", "").strip().replace(".git", "") + name = repo.rsplit("/", 1)[-1] if repo else s.get("name", "") + commit = s.get("hash", "")[:7] + cwl = s.get("name", "") + key = (name, commit, cwl) + if not name or key in seen: + continue + seen.add(key) + steps.append({"name": name, "repo": repo, "commit": commit, "cwl": cwl}) + return steps + + +class WrappedEntity: + """Provide a convenient wrapper for entity information.""" + + def __init__(self, entity: dict): + """Construct a WrappedEntity.""" + self._entity = entity + + def get(self, key: Any, default=None) -> Any: + """Pass references to the get method to the internal dict.""" + return self._entity.get(key, default) + + def __getitem__(self, key: Any) -> Any: + """Implement square brackets for the wrapper class.""" + return self._entity[key] + + def __contains__(self, key: Any) -> bool: + """Implement 'in' for the wrapper class.""" + return key in self._entity + + def walk_ancestors( + self, continue_test: Callable[[dict], bool] = lambda ent: True + ) -> list[tuple]: + """ + Walk the ancestors of the entity. + + Return a list of tuples of the form: + (hubmap_id entity_dict list-of-ancestors) + where list-of-ancestors is None or a list of tuples of the same form. + + continue_test takes an entity dict as a parameter and returns True if + the descent should continue to the children of that entity, False otherwise. + """ + return walk_ancestors(self._entity, continue_test) + + def list_ancestors( + self, + continue_test: Callable[[dict], bool] = lambda ent: True, + omit_test: Callable[[dict], bool] = lambda end: False, + ) -> list: + """Return ancestor information in an expanded, non-recursive list.""" + return listify(self.walk_ancestors(continue_test), omit_test) + + @property + def is_processed(self): + """True if this a processed dataset, as opposed to a raw (primary) dataset.""" + return is_processed(self._entity) + + def pipeline_steps(self) -> list[dict]: + """Return the pipeline steps of the wrapped entity.""" + return pipeline_steps(self._entity) + + def count_versions(self) -> int: + """Return a version number for the wrapped entity.""" + if "previous_revision_uuid" in self: + prev_ent = WrappedEntity(fetch_entity_info(self["previous_revision_uuid"])) + return prev_ent.count_versions() + 1 + else: + return 1 + + def _organ(self, ancs: list[dict]) -> tuple: + for anc in ancs: + if anc.get("organ"): + return ORGAN_MAP.get(anc["organ"], (anc["organ"], None)) + return None, None + + @staticmethod + def _assay_info_kwds(ancs: list) -> list[str]: + rslt = [] + for anc in ancs: + md = anc.get("metadata", {}) + for v in [ + anc.get("dataset_type"), + md.get("assay_category"), + md.get("analyte_class"), + ]: + if v and v not in rslt: + rslt.append(v) + return rslt + + def keywords(self) -> list[str]: + """Return a list of keywords appropriate to the dataset.""" + kws: list[str] = [] + ancs = self.list_ancestors() + kws.extend(self._assay_info_kwds(ancs)) + organ_label, uberon = self._organ(ancs) + if organ_label: + kws.append(organ_label) + if uberon: + kws.append(uberon) + for anc in ancs: + rui = anc.get("rui_location") + if rui: + r = json.loads(rui) if isinstance(rui, str) else rui + for term in r.get("ccf_annotations", []) or []: + short = ( + "UBERON:" + term.rsplit("_", 1)[-1] + if "UBERON" in term + else term + ) + if short not in kws: + kws.append(short) + break + kws += ["Homo sapiens", "NCBITaxon:9606", "HuBMAP"] + LOGGER.debug(f"KEYWORDS for {self['hubmap_id']}:\n{pformat(kws)}") + return kws diff --git a/src/cwl_example/build_rocrate.py b/src/cwl_example/build_rocrate.py index c0b90d3..108ac1a 100755 --- a/src/cwl_example/build_rocrate.py +++ b/src/cwl_example/build_rocrate.py @@ -1,4 +1,5 @@ #!/bin/env python +"""Demonstrate the use of the --provenance option to cwltool.""" from pathlib import Path from subprocess import run @@ -8,6 +9,7 @@ def main(): + """Demonstrate the use of the --provenance option to cwltool.""" run("cwltool --provenance prov1 hello_world.cwl", shell=True, check=True) run( "cwltool --provenance prov2 hello_world.cwl --message 'hola'", diff --git a/src/requirements-dev.txt b/src/requirements-dev.txt index 24b8558..c1b70dc 100644 --- a/src/requirements-dev.txt +++ b/src/requirements-dev.txt @@ -1,4 +1,4 @@ -black==25.1.0 +black==23.11.0 flake8==7.1.0 isort==7.0.0 semantic-version==2.10 diff --git a/src/requirements.txt b/src/requirements.txt index d8d9c6d..f018eba 100644 --- a/src/requirements.txt +++ b/src/requirements.txt @@ -1,9 +1,9 @@ jsonschema==4.23.0 -pandas==2.2.2 +pandas>=2.2.2 PyYAML==6.0.2 requests==2.32.3 rocrate>=0.15.0 bagit -mlcroissant +mlcroissant[dev]>=1.1.0 git+https://github.com/hubmapconsortium/cwltool.git@docker-gpu#egg=cwltool setuptools<82