Merge pull request #26 from HumanCellAtlas/ds-get-sample-id-v5-598

Support getting sample id from v5 metadata

Author: David Shiga

adapter_pipelines/Optimus/adapter.wdl

@@ -11,43 +11,19 @@ task GetInputs {
 
   command <<<
     python <<CODE
-    from pipeline_tools import dcp_utils
     from pipeline_tools import input_utils
 
-    # Get bundle manifest
-    uuid = '${bundle_uuid}'
-    version = '${bundle_version}'
-    dss_url = '${dss_url}'
-    retry_seconds = ${retry_seconds}
-    timeout_seconds = ${timeout_seconds}
-    print('Getting bundle manifest for id {0}, version {1}'.format(uuid, version))
-    manifest = dcp_utils.get_manifest(uuid, version, dss_url, timeout_seconds, retry_seconds)
-    manifest_files = dcp_utils.get_manifest_file_dicts(manifest)
-
-    input_metadata_file_uuid = input_utils.get_input_metadata_file_uuid(manifest_files)
-    input_metadata_json = dcp_utils.get_file_by_uuid(input_metadata_file_uuid, dss_url)
-
-    # Parse inputs from metadata and write to fastq_inputs
-    print('Writing fastq inputs to fastq_inputs.tsv')
-    sample_id = input_utils.get_sample_id(input_metadata_json)
-    lanes = input_utils.get_optimus_lanes(input_metadata_json)
-    r1, r2, i1 = input_utils.get_optimus_inputs(lanes, manifest_files)
-    fastq_inputs = [list(i) for i in zip(r1, r2, i1)]
-    print(fastq_inputs)
-
-    with open('fastq_inputs.tsv', 'w') as f:
-        for line in fastq_inputs:
-            f.write('\t'.join(line) +'\n')
-
-    print('Writing sample ID to inputs.tsv')
-    sample_id = input_utils.get_sample_id(input_metadata_json)
-    with open('inputs.tsv', 'w') as f:
-        f.write('{0}'.format(sample_id))
-    print('Wrote input map')
+    input_utils.create_optimus_input_tsv(
+                    "${bundle_manifest}",
+                    "${bundle_version}",
+                    "${dss_url}",
+                    ${retry_seconds},
+                    ${timeout_seconds})
+
     CODE
   >>>
   runtime {
-    docker: "quay.io/humancellatlas/secondary-analysis-pipeline-tools:v0.1.11"
+    docker: "quay.io/humancellatlas/secondary-analysis-pipeline-tools:v0.12.0"
   }
   output {
     String sample_id = read_string("inputs.tsv")
@@ -88,7 +64,7 @@ task inputs_for_submit {
     >>>
 
     runtime {
-      docker: "quay.io/humancellatlas/secondary-analysis-pipeline-tools:v0.1.11"
+      docker: "quay.io/humancellatlas/secondary-analysis-pipeline-tools:v0.12.0"
     }
 
     output {
@@ -127,7 +103,7 @@ task outputs_for_submit {
     >>>
 
     runtime {
-      docker: "quay.io/humancellatlas/secondary-analysis-pipeline-tools:v0.1.11"
+      docker: "quay.io/humancellatlas/secondary-analysis-pipeline-tools:v0.12.0"
     }
 
     output {

adapter_pipelines/ss2_single_sample/adapter.wdl

@@ -11,36 +11,19 @@ task GetInputs {
 
   command <<<
     python <<CODE
-    from pipeline_tools import dcp_utils
     from pipeline_tools import input_utils
 
-    # Get bundle manifest
-    uuid = "${bundle_uuid}"
-    version = "${bundle_version}"
-    dss_url = "${dss_url}"
-    retry_seconds = ${retry_seconds}
-    timeout_seconds = ${timeout_seconds}
-    print("Getting bundle manifest for id {0}, version {1}".format(uuid, version))
-    manifest = dcp_utils.get_manifest(uuid, version, dss_url, timeout_seconds, retry_seconds)
-    manifest_files = dcp_utils.get_manifest_file_dicts(manifest)
+    input_utils.create_ss2_input_tsv(
+                    "${bundle_uuid}",
+                    "${bundle_version}",
+                    "${dss_url}",
+                    ${retry_seconds},
+                    ${timeout_seconds})
 
-    inputs_metadata_file_uuid = input_utils.get_input_metadata_file_uuid(manifest_files)
-    inputs_metadata_json = dcp_utils.get_file_by_uuid(inputs_metadata_file_uuid, dss_url)
-
-    sample_id = input_utils.get_sample_id(inputs_metadata_json)
-    fastq_1_name, fastq_2_name = input_utils.get_smart_seq_2_fastq_names(inputs_metadata_json)
-    fastq_1_url = dcp_utils.get_file_url(manifest_files, fastq_1_name)
-    fastq_2_url = dcp_utils.get_file_url(manifest_files, fastq_2_name)
-
-    print("Creating input map")
-    with open("inputs.tsv", "w") as f:
-        f.write("fastq_1\tfastq_2\tsample_id\n")
-        f.write("{0}\t{1}\t{2}\n".format(fastq_1_url, fastq_2_url, sample_id))
-    print("Wrote input map")
     CODE
   >>>
   runtime {
-    docker: "quay.io/humancellatlas/secondary-analysis-pipeline-tools:v0.1.11"
+    docker: "quay.io/humancellatlas/secondary-analysis-pipeline-tools:v0.12.0"
   }
   output {
     Object inputs = read_object("inputs.tsv")

pipeline_tools/input_utils.py

@@ -1,45 +1,82 @@
 from pipeline_tools import dcp_utils
 
 
-def get_sample_id(metadata, version='4'):
+def get_sample_id(metadata, version):
     """Return the sample id from the given metadata"""
-    if version == '4':
+    if version.startswith('4.'):
         return _get_sample_id_v4(metadata)
+    elif version.startswith('5.'):
+        return _get_sample_id_v5(metadata)
     else:
-        raise NotImplementedError('Only implemented for v4 metadata')
+        raise NotImplementedError('Only implemented for v4 and v5 metadata')
+
+
+def _get_sample_id_v5(links_json):
+    """Return sample id from links json"""
+    links = links_json['links']
+    def is_sample_link(link):
+        return link['source_type'] == 'biomaterial' and link['destination_type'] == 'sequencing_process'
+    sample_links = list(filter(lambda x: is_sample_link(x), links))
+    num_links = len(sample_links)
+    if num_links == 0:
+        raise ValueError('No sample link found')
+    elif num_links > 1:
+        raise ValueError('Expecting one sample link. {0} found'.format(num_links))
+    return sample_links[0]['source_id']
 
 
 def _get_sample_id_v4(assay_json):
     """Return sample id from assay json"""
     return assay_json["has_input"]
 
 
-def get_input_metadata_file_uuid(manifest_files, version='4'):
+def get_input_metadata_file_uuid(manifest_files, version):
     """Get the uuid of the file containing metadata about pipeline input files,
     e.g. assay.json in v4"""
-    if version == '5':
+    if version.startswith('5.'):
         return _get_input_metadata_file_uuid_v5(manifest_files)
-    elif version == '4':
+    elif version.startswith('4.'):
         return _get_input_metadata_file_uuid_v4(manifest_files)
     else:
         raise NotImplementedError('Only implemented for v4 and v5 metadata')
 
 
 def _get_input_metadata_file_uuid_v5(manifest_files):
     """Get the uuid of the files.json file"""
-    return dcp_utils.get_file_uuid(manifest_files, 'files.json')
+    return dcp_utils.get_file_uuid(manifest_files, 'file.json')
 
 
 def _get_input_metadata_file_uuid_v4(manifest_files):
     """Get the uuid of the assay.json file"""
     return dcp_utils.get_file_uuid(manifest_files, 'assay.json')
 
 
-def get_smart_seq_2_fastq_names(metadata, version='4'):
+def get_sample_id_file_uuid(manifest_files, version):
+    """Get the uuid of the file containing the sample id,
+    e.g. assay.json in v4"""
+    if version.startswith('5.'):
+        return _get_sample_id_file_uuid_v5(manifest_files)
+    elif version.startswith('4.'):
+        return _get_sample_id_file_uuid_v4(manifest_files)
+    else:
+        raise NotImplementedError('Only implemented for v4 and v5 metadata')
+
+
+def _get_sample_id_file_uuid_v5(manifest_files):
+    """Get the uuid of the links.json file"""
+    return dcp_utils.get_file_uuid(manifest_files, 'links.json')
+
+
+def _get_sample_id_file_uuid_v4(manifest_files):
+    """Get the uuid of the assay.json file"""
+    return dcp_utils.get_file_uuid(manifest_files, 'assay.json')
+
+
+def get_smart_seq_2_fastq_names(metadata, version):
     """Get the fastq file names from the given metadata"""
-    if version == '5':
+    if version.startswith('5.'):
         return _get_smart_seq_2_fastq_names_v5(metadata)
-    elif version == '4':
+    elif version.startswith('4.'):
         return _get_smart_seq_2_fastq_names_v4(metadata)
     else:
         raise NotImplementedError('Only implemented for v4 and v5 metadata')
@@ -62,9 +99,9 @@ def _get_smart_seq_2_fastq_names_v4(assay_json):
     return fastq_1_name, fastq_2_name
 
 
-def get_optimus_lanes(metadata_json, version='4'):
+def get_optimus_lanes(metadata_json, version):
     """Get the lane metadata"""
-    if version == '4':
+    if version.startswith('4.'):
         return _get_optimus_lanes_v4(metadata_json)
     else:
         raise NotImplementedError('Only implemented for v4 metadata')
@@ -85,3 +122,89 @@ def get_optimus_inputs(lanes, manifest_files):
     i1 = [manifest_files['name_to_meta'][lane['i1']]['url'] for lane in lanes]
 
     return r1, r2, i1
+
+
+def is_v5_or_higher(name_to_meta):
+    """Return true if schema of bundle metadata is at least 5.0.0"""
+    if 'assay.json' in name_to_meta:
+        return False
+    elif 'file.json' in name_to_meta:
+        return True
+    else:
+        raise ValueError('No assay.json and no process.json. Cannot determine metadata schema version.')
+
+
+def detect_schema_version(file_json):
+    """Return the bundle's metadata schema version"""
+    files = file_json['files']
+    if len(files) == 0:
+        raise ValueError('No files in bundle')
+    schema_url = files[0]['content']['describedBy']
+    version = schema_url.split('/')[-2]
+    return version
+
+
+def get_metadata_to_process(manifest_files, dss_url, is_v5_or_higher):
+    """Return the metadata json that we need to parse to set up pipeline inputs"""
+    if not is_v5_or_higher:
+        schema_version = '4.x'
+        sample_id_file_uuid = dcp_utils.get_file_uuid(manifest_files, 'sample.json')
+        inputs_metadata_file_uuid = dcp_utils.get_file_uuid(manifest_files, 'assay.json')
+        inputs_metadata_json = dcp_utils.get_file_by_uuid(inputs_metadata_file_uuid, dss_url)
+        sample_id_file_json = inputs_metadata_json
+    else:
+        sample_id_file_uuid = dcp_utils.get_file_uuid(manifest_files, 'links.json')
+        inputs_metadata_file_uuid = dcp_utils.get_file_uuid(manifest_files, 'file.json')
+        inputs_metadata_json = dcp_utils.get_file_by_uuid(inputs_metadata_file_uuid, dss_url)
+        schema_version = detect_schema_version(inputs_metadata_json)
+        sample_id_file_json = dcp_utils.get_file_by_uuid(sample_id_file_uuid, dss_url)
+    return inputs_metadata_json, sample_id_file_json, schema_version
+
+
+def create_ss2_input_tsv(uuid, version, dss_url, retry_seconds, timeout_seconds):
+    """Create tsv of Smart-seq2 inputs"""
+    # Get bundle manifest
+    print("Getting bundle manifest for id {0}, version {1}".format(uuid, version))
+    manifest = dcp_utils.get_manifest(uuid, version, dss_url, timeout_seconds, retry_seconds)
+    manifest_files = dcp_utils.get_manifest_file_dicts(manifest)
+
+    inputs_metadata_json, sample_id_file_json, schema_version = get_metadata_to_process(
+        manifest_files, dss_url, is_v5_or_higher(manifest_files['name_to_meta']))
+
+    sample_id = get_sample_id(sample_id_file_json, schema_version)
+    fastq_1_name, fastq_2_name = get_smart_seq_2_fastq_names(inputs_metadata_json, schema_version)
+    fastq_1_url = dcp_utils.get_file_url(manifest_files, fastq_1_name)
+    fastq_2_url = dcp_utils.get_file_url(manifest_files, fastq_2_name)
+
+    print("Creating input map")
+    with open("inputs.tsv", "w") as f:
+        f.write("fastq_1\tfastq_2\tsample_id\n")
+        f.write("{0}\t{1}\t{2}\n".format(fastq_1_url, fastq_2_url, sample_id))
+    print("Wrote input map")
+
+
+def create_optimus_input_tsv(uuid, version, dss_url, retry_seconds, timeout_seconds):
+    """Create tsv of Optimus inputs"""
+    # Get bundle manifest
+    print('Getting bundle manifest for id {0}, version {1}'.format(uuid, version))
+    manifest = dcp_utils.get_manifest(uuid, version, dss_url, timeout_seconds, retry_seconds)
+    manifest_files = dcp_utils.get_manifest_file_dicts(manifest)
+
+    inputs_metadata_json, sample_id_file_json, schema_version = get_metadata_to_process(
+        manifest_files, dss_url, is_v5_or_higher(manifest_files['name_to_meta']))
+
+    # Parse inputs from metadata and write to fastq_inputs
+    print('Writing fastq inputs to fastq_inputs.tsv')
+    sample_id = get_sample_id(sample_id_file_json, schema_version)
+    lanes = get_optimus_lanes(input_metadata_json, schema_version)
+    r1, r2, i1 = get_optimus_inputs(lanes, manifest_files)
+    fastq_inputs = [list(i) for i in zip(r1, r2, i1)]
+    print(fastq_inputs)
+
+    with open('fastq_inputs.tsv', 'w') as f:
+        for line in fastq_inputs:
+            f.write('\t'.join(line) +'\n')
+        print('Writing sample ID to inputs.tsv')
+        f.write('{0}'.format(sample_id))
+
+    print('Wrote input map')

pipeline_tools/tests/data/metadata/v5/ss2_links.json

@@ -0,0 +1,49 @@
+{
+    "links": [
+        {
+            "source_id": "7c72f20a-8199-497a-bba7-1316520ba847", 
+            "source_type": "biomaterial", 
+            "destination_id": "336c7184-70a7-41b3-ab31-98f78b0ab092", 
+            "destination_type": "dissociation_process"
+        }, 
+        {
+            "source_id": "336c7184-70a7-41b3-ab31-98f78b0ab092", 
+            "source_type": "dissociation_process", 
+            "destination_id": "94049366-4e4b-4076-8474-b2a00d7aa08d", 
+            "destination_type": "biomaterial"
+        }, 
+        {
+            "source_id": "8c01ac9e-751b-4426-9032-c4235503a164", 
+            "source_type": "process", 
+            "destination_id": "7c72f20a-8199-497a-bba7-1316520ba847", 
+            "destination_type": "biomaterial"
+        }, 
+        {
+            "source_id": "cd9d52fa-e89d-4c37-bb0e-997a2affe1dc", 
+            "source_type": "biomaterial", 
+            "destination_id": "8c01ac9e-751b-4426-9032-c4235503a164", 
+            "destination_type": "process"
+        }, 
+        {
+            "source_id": "test_sample_id", 
+            "source_type": "biomaterial", 
+            "destination_id": "adebcaaa-0cea-44af-8cce-fdc5dcadb286", 
+            "destination_type": "sequencing_process"
+        }, 
+        {
+            "source_id": "adebcaaa-0cea-44af-8cce-fdc5dcadb286", 
+            "source_type": "sequencing_process", 
+            "destination_id": "cb4f1900-cdcf-4e3d-834e-21c6b3e59237", 
+            "destination_type": "file"
+        }, 
+        {
+            "source_id": "adebcaaa-0cea-44af-8cce-fdc5dcadb286", 
+            "source_type": "sequencing_process", 
+            "destination_id": "ed7989fb-1ca0-48ba-9b81-f83c4af7b091", 
+            "destination_type": "file"
+        }
+    ], 
+    "describedBy": "https://schema.humancellatlas.org/bundle/1.0.0/links", 
+    "schema_version": "1.0.0", 
+    "schema_type": "link_bundle"
+}