#8438

Committed 16 Jun 2025 06:48PM UTC coverage: 90.994% (+0.1%) from 90.89%

Build # #8438

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

IGVF-2795-ethnicities-african-enums (#1584)

Coverage Stats

8912 of 9794 relevant lines covered (90.99%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.34

/src/igvfd/audit/construct_library_set.py

from snovault.auditor import (
    AuditFailure,
    audit_checker
)
from .formatter import (
    audit_link,
    path_to_text,
    get_audit_message
)


def get_assay_terms(value, system):
    assay_terms = set()
    file_sets = value.get('file_sets', [])
    for file_set in file_sets:
        if file_set.startswith('/measurement-sets/'):
            file_set_object = system.get('request').embed(file_set, '@@object?skip_calculated=true')
            assay_terms.add(file_set_object['assay_term'])
    return list(assay_terms)


def audit_construct_library_set_associated_phenotypes(value, system):
    '''
    [
        {
            "audit_description": "Construct library sets with a selection criteria of phenotype-associated variants are expected to have associated phenotype(s).",
            "audit_category": "missing associated phenotypes",
            "audit_level": "NOT_COMPLIANT"
        }
    ]
    '''
    audit_message = get_audit_message(audit_construct_library_set_associated_phenotypes)
    detail = ''
    selection_criteria_list = value.get('selection_criteria', [])
    if 'phenotype-associated variants' in selection_criteria_list:
        assoc_pheno = value.get('associated_phenotypes', [])
        if assoc_pheno != []:
            return
        else:
            detail = (
                f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} '
                f'has phenotype-associated variants listed in its `selection_criteria`, '
                f'but no phenotype term specified in `associated_phenotypes`.'
            )
            yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))


def audit_construct_library_set_plasmid_map(value, system):
    '''
    [
        {
            "audit_description": "Construct library sets are expected to be associated with a plasmid map document.",
            "audit_category": "missing plasmid map",
            "audit_level": "NOT_COMPLIANT"
        }
    ]
    '''
    audit_message = get_audit_message(audit_construct_library_set_plasmid_map)
    map_counter = 0
    detail = (
        f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} '
        f'does not have a plasmid map attached in `documents`.'
    )
    documents = value.get('documents', [])
    if documents == []:
        yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))
    else:
        for document in documents:
            document_obj = system.get('request').embed(document, '@@object?skip_calculated=true')
            if document_obj['document_type'] == 'plasmid map':
                map_counter += 1
                break
            else:
                continue
        if map_counter == 0:
            yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))


def audit_construct_library_set_scope(value, system):
    '''
    [
        {
            "audit_description": "Construct library sets with a scope of tile or exon are expected to only link to one gene.",
            "audit_category": "inconsistent scope",
            "audit_level": "WARNING"
        }
    ]
    '''
    audit_message = get_audit_message(audit_construct_library_set_scope)
    detail = ''
    if value.get('scope') in ['exon', 'tile']:
        if len(value.get('small_scale_gene_list', [])) > 1:
            detail = (
                f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} '
                f'specifies it has a `scope` of {value["scope"]}, but multiple genes are listed in '
                f'`small_scale_gene_list`.'
            )
            yield AuditFailure(audit_message.get('audit_category', ''),
                               f'{detail} {audit_message.get("audit_description", "")})', level=audit_message.get('audit_level', ''))


def audit_integrated_content_files(value, system):
    '''
    [
        {
            "audit_description": "Guide libraries used in CRISPR assays are expected to link to an integrated content file of guide RNA sequences.",
            "audit_category": "missing guide RNA sequences",
            "audit_level": "NOT_COMPLIANT"
        },
        {
            "audit_description": "Reporter libraries used in MPRA assays are expected to link to an integrated content file of MPRA sequence designs.",
            "audit_category": "missing MPRA sequence designs",
            "audit_level": "NOT_COMPLIANT"
        }
    ]
    '''
    audit_message_guide = get_audit_message(audit_integrated_content_files, index=0)
    audit_message_reporter = get_audit_message(audit_integrated_content_files, index=1)
    assay_terms = get_assay_terms(value, system)
    CRISPR_assays = [
        '/assay-terms/OBI_0003659/',  # in vitro CRISPR screen assay
        '/assay-terms/OBI_0003660/',  # in vitro CRISPR screen using single-cell RNA-seq
        '/assay-terms/OBI_0003661/'  # in vitro CRISPR screen using flow cytometry
    ]
    MPRA_assays = [
        '/assay-terms/OBI_0002675/'  # massively parallel reporter assay
    ]
    library_expectation = {
        'guide library': ('guide RNA sequences', audit_message_guide, CRISPR_assays),
        'reporter library': ('MPRA sequence designs', audit_message_reporter, MPRA_assays),
    }
    integrated_content_files = value.get('integrated_content_files', '')
    library_type = value.get('file_set_type', '')
    if library_type in library_expectation and any(assay_term in library_expectation[library_type][2] for assay_term in assay_terms):
        file_expectation = library_expectation[library_type][0]
        audit_message = library_expectation[library_type][1]
        if integrated_content_files:
            files = [system.get('request').embed(file, '@@object?skip_calculated=true')
                     for file in integrated_content_files]
            if not ([file for file in files if file['content_type'] == file_expectation]):
                detail = (f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} has no '
                          f'linked files in `integrated_content_files` with `content_type` {file_expectation}.')
                yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))
        else:
            detail = (f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} has no '
                      f'`integrated_content_files`.')
            yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))


def audit_construct_library_set_orf_gene(value, system):
    '''
    [
        {
            "audit_description": "Genes listed in the library are expected to match the open read frame gene.",
            "audit_category": "inconsistent genes",
            "audit_level": "ERROR"
        }
    ]
    '''
    audit_message = get_audit_message(audit_construct_library_set_orf_gene)
    library_genes = set()
    orf_genes = set()
    if ('small_scale_gene_list' in value) and ('orf_list' in value):
        library_genes = set(value['small_scale_gene_list'])
        orf_ids = value.get('orf_list')
        for o in orf_ids:
            orf_object = system.get('request').embed(o + '@@object?skip_calculated=true')
            if 'genes' in orf_object:
                for d in orf_object['genes']:
                    orf_genes.add(d)

        if orf_genes != library_genes:
            detail = (
                f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} '
                f'has a `small_scale_gene_list` which does not match the genes of its associated `orf_list`.'
            )
            yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))


function_dispatcher_construct_library_set_object = {
    'audit_construct_library_set_associated_phenotypes': audit_construct_library_set_associated_phenotypes,
    'audit_construct_library_set_plasmid_map': audit_construct_library_set_plasmid_map,
    'audit_construct_library_set_scope': audit_construct_library_set_scope,
    'audit_integrated_content_files': audit_integrated_content_files,
    'audit_construct_library_set_orf_gene': audit_construct_library_set_orf_gene
}


@audit_checker('ConstructLibrarySet', frame='object')
def audit_construct_library_set_object_dispatcher(value, system):
    for function_name in function_dispatcher_construct_library_set_object.keys():
        for failure in function_dispatcher_construct_library_set_object[function_name](value, system):
            yield failure

1	from snovault.auditor import (	1✔
2	AuditFailure,
3	audit_checker
4	)
5	from .formatter import (	1✔
6	audit_link,
7	path_to_text,
8	get_audit_message
9	)
10
11
12	def get_assay_terms(value, system):	1✔
13	assay_terms = set()	1✔
14	file_sets = value.get('file_sets', [])	1✔
15	for file_set in file_sets:	1✔
16	if file_set.startswith('/measurement-sets/'):	1✔
17	file_set_object = system.get('request').embed(file_set, '@@object?skip_calculated=true')	1✔
18	assay_terms.add(file_set_object['assay_term'])	1✔
19	return list(assay_terms)	1✔
20
21
22	def audit_construct_library_set_associated_phenotypes(value, system):	1✔
23	'''
24	[
25	{
26	"audit_description": "Construct library sets with a selection criteria of phenotype-associated variants are expected to have associated phenotype(s).",
27	"audit_category": "missing associated phenotypes",
28	"audit_level": "NOT_COMPLIANT"
29	}
30	]
31	'''
32	audit_message = get_audit_message(audit_construct_library_set_associated_phenotypes)	1✔
33	detail = ''	1✔
34	selection_criteria_list = value.get('selection_criteria', [])	1✔
35	if 'phenotype-associated variants' in selection_criteria_list:	1✔
36	assoc_pheno = value.get('associated_phenotypes', [])	1✔
37	if assoc_pheno != []:	1✔
38	return	×
39	else:
40	detail = (	1✔
41	f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} '
42	f'has phenotype-associated variants listed in its `selection_criteria`, '
43	f'but no phenotype term specified in `associated_phenotypes`.'
44	)
45	yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))	1✔
46
47
48	def audit_construct_library_set_plasmid_map(value, system):	1✔
49	'''
50	[
51	{
52	"audit_description": "Construct library sets are expected to be associated with a plasmid map document.",
53	"audit_category": "missing plasmid map",
54	"audit_level": "NOT_COMPLIANT"
55	}
56	]
57	'''
58	audit_message = get_audit_message(audit_construct_library_set_plasmid_map)	1✔
59	map_counter = 0	1✔
60	detail = (	1✔
61	f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} '
62	f'does not have a plasmid map attached in `documents`.'
63	)
64	documents = value.get('documents', [])	1✔
65	if documents == []:	1✔
66	yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))	1✔
67	else:
68	for document in documents:	1✔
69	document_obj = system.get('request').embed(document, '@@object?skip_calculated=true')	1✔
70	if document_obj['document_type'] == 'plasmid map':	1✔
71	map_counter += 1	1✔
72	break	1✔
73	else:
74	continue	×
75	if map_counter == 0:	1✔
76	yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))	×
77
78
79	def audit_construct_library_set_scope(value, system):	1✔
80	'''
81	[
82	{
83	"audit_description": "Construct library sets with a scope of tile or exon are expected to only link to one gene.",
84	"audit_category": "inconsistent scope",
85	"audit_level": "WARNING"
86	}
87	]
88	'''
89	audit_message = get_audit_message(audit_construct_library_set_scope)	1✔
90	detail = ''	1✔
91	if value.get('scope') in ['exon', 'tile']:	1✔
92	if len(value.get('small_scale_gene_list', [])) > 1:	1✔
93	detail = (	1✔
94	f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} '
95	f'specifies it has a `scope` of {value["scope"]}, but multiple genes are listed in '
96	f'`small_scale_gene_list`.'
97	)
98	yield AuditFailure(audit_message.get('audit_category', ''),	1✔
99	f'{detail} {audit_message.get("audit_description", "")})', level=audit_message.get('audit_level', ''))
100
101
102	def audit_integrated_content_files(value, system):	1✔
103	'''
104	[
105	{
106	"audit_description": "Guide libraries used in CRISPR assays are expected to link to an integrated content file of guide RNA sequences.",
107	"audit_category": "missing guide RNA sequences",
108	"audit_level": "NOT_COMPLIANT"
109	},
110	{
111	"audit_description": "Reporter libraries used in MPRA assays are expected to link to an integrated content file of MPRA sequence designs.",
112	"audit_category": "missing MPRA sequence designs",
113	"audit_level": "NOT_COMPLIANT"
114	}
115	]
116	'''
117	audit_message_guide = get_audit_message(audit_integrated_content_files, index=0)	1✔
118	audit_message_reporter = get_audit_message(audit_integrated_content_files, index=1)	1✔
119	assay_terms = get_assay_terms(value, system)	1✔
120	CRISPR_assays = [	1✔
121	'/assay-terms/OBI_0003659/', # in vitro CRISPR screen assay
122	'/assay-terms/OBI_0003660/', # in vitro CRISPR screen using single-cell RNA-seq
123	'/assay-terms/OBI_0003661/' # in vitro CRISPR screen using flow cytometry
124	]
125	MPRA_assays = [	1✔
126	'/assay-terms/OBI_0002675/' # massively parallel reporter assay
127	]
128	library_expectation = {	1✔
129	'guide library': ('guide RNA sequences', audit_message_guide, CRISPR_assays),
130	'reporter library': ('MPRA sequence designs', audit_message_reporter, MPRA_assays),
131	}
132	integrated_content_files = value.get('integrated_content_files', '')	1✔
133	library_type = value.get('file_set_type', '')	1✔
134	if library_type in library_expectation and any(assay_term in library_expectation[library_type][2] for assay_term in assay_terms):	1✔
135	file_expectation = library_expectation[library_type][0]	1✔
136	audit_message = library_expectation[library_type][1]	1✔
137	if integrated_content_files:	1✔
138	files = [system.get('request').embed(file, '@@object?skip_calculated=true')	1✔
139	for file in integrated_content_files]
140	if not ([file for file in files if file['content_type'] == file_expectation]):	1✔
141	detail = (f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} has no '	1✔
142	f'linked files in `integrated_content_files` with `content_type` {file_expectation}.')
143	yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))	1✔
144	else:
145	detail = (f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} has no '	1✔
146	f'`integrated_content_files`.')
147	yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))	1✔
148
149
150	def audit_construct_library_set_orf_gene(value, system):	1✔
151	'''
152	[
153	{
154	"audit_description": "Genes listed in the library are expected to match the open read frame gene.",
155	"audit_category": "inconsistent genes",
156	"audit_level": "ERROR"
157	}
158	]
159	'''
160	audit_message = get_audit_message(audit_construct_library_set_orf_gene)	1✔
161	library_genes = set()	1✔
162	orf_genes = set()	1✔
163	if ('small_scale_gene_list' in value) and ('orf_list' in value):	1✔
164	library_genes = set(value['small_scale_gene_list'])	1✔
165	orf_ids = value.get('orf_list')	1✔
166	for o in orf_ids:	1✔
167	orf_object = system.get('request').embed(o + '@@object?skip_calculated=true')	1✔
168	if 'genes' in orf_object:	1✔
169	for d in orf_object['genes']:	1✔
170	orf_genes.add(d)	1✔
171
172	if orf_genes != library_genes:	1✔
173	detail = (	1✔
174	f'Construct library set {audit_link(path_to_text(value["@id"]), value["@id"])} '
175	f'has a `small_scale_gene_list` which does not match the genes of its associated `orf_list`.'
176	)
177	yield AuditFailure(audit_message.get('audit_category', ''), f'{detail} {audit_message.get("audit_description", "")}', level=audit_message.get('audit_level', ''))	1✔
178
179
180	function_dispatcher_construct_library_set_object = {	1✔
181	'audit_construct_library_set_associated_phenotypes': audit_construct_library_set_associated_phenotypes,
182	'audit_construct_library_set_plasmid_map': audit_construct_library_set_plasmid_map,
183	'audit_construct_library_set_scope': audit_construct_library_set_scope,
184	'audit_integrated_content_files': audit_integrated_content_files,
185	'audit_construct_library_set_orf_gene': audit_construct_library_set_orf_gene
186	}
187
188
189	@audit_checker('ConstructLibrarySet', frame='object')	1✔
190	def audit_construct_library_set_object_dispatcher(value, system):	1✔
191	for function_name in function_dispatcher_construct_library_set_object.keys():	1✔
192	for failure in function_dispatcher_construct_library_set_object[function_name](value, system):	1✔
193	yield failure	1✔

IGVF-DACC / igvfd / #8438

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous