diff --git a/hail_search/queries/base.py b/hail_search/queries/base.py
index 0d284c6a59..84f2cab757 100644
--- a/hail_search/queries/base.py
+++ b/hail_search/queries/base.py
@@ -738,7 +738,12 @@ def _parse_intervals(self, intervals, gene_ids=None, variant_keys=None, variant_
return intervals
if variant_ids:
- intervals = [(chrom, pos, pos+1) for chrom, pos, _, _ in variant_ids]
+ first_chrom = variant_ids[0][0]
+ if all(first_chrom == v[0] for v in variant_ids):
+ positions = [pos for _, pos, _, _ in variant_ids]
+ intervals = [(first_chrom, min(positions), max(positions) + 1)]
+ else:
+ intervals = [(chrom, pos, pos+1) for chrom, pos, _, _ in variant_ids]
is_x_linked = self._inheritance_mode == X_LINKED_RECESSIVE
if not (intervals or is_x_linked):
diff --git a/seqr/management/commands/check_for_new_samples_from_pipeline.py b/seqr/management/commands/check_for_new_samples_from_pipeline.py
index fe68b47c25..5a7551a4a2 100644
--- a/seqr/management/commands/check_for_new_samples_from_pipeline.py
+++ b/seqr/management/commands/check_for_new_samples_from_pipeline.py
@@ -3,8 +3,6 @@
from django.contrib.postgres.aggregates import ArrayAgg
from django.core.management.base import BaseCommand, CommandError
-from django.db.models import Q
-from django.db.models.functions import JSONObject
import json
import logging
import re
@@ -186,7 +184,7 @@ def _load_new_samples(cls, metadata_path, genome_version, dataset_type, run_vers
sample_type = metadata['sample_type']
logger.info(f'Loading {len(sample_project_tuples)} {sample_type} {dataset_type} samples in {len(samples_by_project)} projects')
- updated_samples, inactivated_sample_guids, *args = match_and_update_search_samples(
+ updated_samples, new_samples, *args = match_and_update_search_samples(
projects=samples_by_project.keys(),
sample_project_tuples=sample_project_tuples,
sample_data={'data_source': run_version, 'elasticsearch_index': ';'.join(metadata['callsets'])},
@@ -196,9 +194,9 @@ def _load_new_samples(cls, metadata_path, genome_version, dataset_type, run_vers
)
# Send loading notifications and update Airtable PDOs
- update_sample_data_by_project = {
- s['individual__family__project']: s for s in updated_samples.values('individual__family__project').annotate(
- samples=ArrayAgg(JSONObject(sample_id='sample_id', individual_id='individual_id')),
+ new_sample_data_by_project = {
+ s['individual__family__project']: s for s in updated_samples.filter(id__in=new_samples).values('individual__family__project').annotate(
+ samples=ArrayAgg('sample_id', distinct=True),
family_guids=ArrayAgg('individual__family__guid', distinct=True),
)
}
@@ -207,15 +205,16 @@ def _load_new_samples(cls, metadata_path, genome_version, dataset_type, run_vers
split_project_pdos = {}
session = AirtableSession(user=None, no_auth=True)
for project, sample_ids in samples_by_project.items():
- project_sample_data = update_sample_data_by_project[project.id]
+ project_sample_data = new_sample_data_by_project[project.id]
is_internal = not project_has_anvil(project) or is_internal_anvil_project(project)
notify_search_data_loaded(
- project, is_internal, dataset_type, sample_type, inactivated_sample_guids,
- updated_samples=project_sample_data['samples'], num_samples=len(sample_ids),
+ project, is_internal, dataset_type, sample_type, project_sample_data['samples'],
+ num_samples=len(sample_ids),
)
project_families = project_sample_data['family_guids']
- updated_families.update(project_families)
- updated_project_families.append((project.id, project.name, project.genome_version, project_families))
+ if project_families:
+ updated_families.update(project_families)
+ updated_project_families.append((project.id, project.name, project.genome_version, project_families))
if is_internal and dataset_type == Sample.DATASET_TYPE_VARIANT_CALLS:
split_project_pdos[project.name] = cls._update_pdos(session, project.guid, sample_ids)
@@ -337,14 +336,21 @@ def _reload_shared_variant_annotations(data_type, genome_version, updated_varian
variant_id: {k: v for k, v in variant.items() if k not in {'familyGuids', 'genotypes'}}
for variant_id, variant in (updated_variants_by_id or {}).items()
}
- fetch_variant_ids = sorted(set(variants_by_id.keys()) - set(updated_variants_by_id.keys()))
+ fetch_variant_ids = set(variants_by_id.keys()) - set(updated_variants_by_id.keys())
if fetch_variant_ids:
- if not is_sv:
- fetch_variant_ids = [parse_valid_variant_id(variant_id) for variant_id in fetch_variant_ids]
- for i in range(0, len(fetch_variant_ids), MAX_LOOKUP_VARIANTS):
- updated_variants = hail_variant_multi_lookup(USER_EMAIL, fetch_variant_ids[i:i+MAX_LOOKUP_VARIANTS], data_type, genome_version)
- logger.info(f'Fetched {len(updated_variants)} additional variants')
- updated_variants_by_id.update({variant['variantId']: variant for variant in updated_variants})
+ if is_sv:
+ variant_ids_by_chrom = {'all': fetch_variant_ids}
+ else:
+ variant_ids_by_chrom = defaultdict(list)
+ for variant_id in fetch_variant_ids:
+ parsed_id = parse_valid_variant_id(variant_id)
+ variant_ids_by_chrom[parsed_id[0]].append(parsed_id)
+ for chrom, variant_ids in sorted(variant_ids_by_chrom.items()):
+ variant_ids = sorted(variant_ids)
+ for i in range(0, len(variant_ids), MAX_LOOKUP_VARIANTS):
+ updated_variants = hail_variant_multi_lookup(USER_EMAIL, variant_ids[i:i+MAX_LOOKUP_VARIANTS], data_type, genome_version)
+ logger.info(f'Fetched {len(updated_variants)} additional variants in chromosome {chrom}')
+ updated_variants_by_id.update({variant['variantId']: variant for variant in updated_variants})
updated_variant_models = []
for variant_id, variant in updated_variants_by_id.items():
diff --git a/seqr/management/tests/check_for_new_samples_from_pipeline_tests.py b/seqr/management/tests/check_for_new_samples_from_pipeline_tests.py
index 4b162f9d27..60a336f82b 100644
--- a/seqr/management/tests/check_for_new_samples_from_pipeline_tests.py
+++ b/seqr/management/tests/check_for_new_samples_from_pipeline_tests.py
@@ -423,7 +423,7 @@ def test_command(self, mock_email, mock_airtable_utils, mock_open_write_file, mo
{'individual_guid': 'I000017_na20889', 'family_guid': 'F000012_12', 'project_guid': 'R0003_test', 'affected': 'A', 'sample_id': 'NA20889', 'sample_type': 'WES'},
]}},
], reload_annotations_logs=[
- 'Reloading shared annotations for 3 SNV_INDEL GRCh38 saved variants (3 unique)', 'Fetched 1 additional variants', 'Fetched 1 additional variants', 'Updated 2 SNV_INDEL GRCh38 saved variants',
+ 'Reloading shared annotations for 3 SNV_INDEL GRCh38 saved variants (3 unique)', 'Fetched 1 additional variants in chromosome 1', 'Fetched 1 additional variants in chromosome 1', 'Updated 2 SNV_INDEL GRCh38 saved variants',
'No additional SV_WES GRCh38 saved variants to update',
], run_loading_logs={
'GRCh38/SNV_INDEL': 'Loading 4 WES SNV_INDEL samples in 2 projects',
@@ -559,10 +559,10 @@ def test_command(self, mock_email, mock_airtable_utils, mock_open_write_file, mo
self.assertEqual(annotation_updated_json['mainTranscriptId'], 'ENST00000505820')
self.assertEqual(len(annotation_updated_json['genotypes']), 3)
- self.mock_utils_logger.error.assert_called_with('Error in project Test Reprocessed Project: Bad Request')
+ self.mock_utils_logger.error.assert_called_with('Error reloading variants in Test Reprocessed Project: Bad Request')
self.mock_utils_logger.info.assert_has_calls([
- mock.call('Updated 0 variants for project Test Reprocessed Project'),
- mock.call('Updated 1 variants for project Non-Analyst Project'),
+ mock.call('Updated 0 variants in 1 families for project Test Reprocessed Project'),
+ mock.call('Updated 1 variants in 1 families for project Non-Analyst Project'),
mock.call('Reload Summary: '),
mock.call(' Non-Analyst Project: Updated 1 variants'),
mock.call('Reloading saved variants in 2 projects'),
diff --git a/seqr/management/tests/reload_saved_variant_annotations_tests.py b/seqr/management/tests/reload_saved_variant_annotations_tests.py
index 2301759b16..95227e3761 100644
--- a/seqr/management/tests/reload_saved_variant_annotations_tests.py
+++ b/seqr/management/tests/reload_saved_variant_annotations_tests.py
@@ -39,18 +39,24 @@ def test_command(self, mock_logger):
mock_logger.info.assert_has_calls([mock.call(log) for log in [
'Reloading shared annotations for 3 SNV_INDEL GRCh37 saved variants (3 unique)',
- 'Fetched 2 additional variants',
+ 'Fetched 2 additional variants in chromosome 1',
+ 'Fetched 2 additional variants in chromosome 21',
'Updated 2 SNV_INDEL GRCh37 saved variants',
]])
- self.assertEqual(len(responses.calls), 1)
- multi_lookup_request = responses.calls[0].request
- self.assertEqual(multi_lookup_request.url, f'{MOCK_HAIL_ORIGIN}:5000/multi_lookup')
- self.assertEqual(multi_lookup_request.headers.get('From'), 'manage_command')
- self.assertDictEqual(json.loads(multi_lookup_request.body), {
+ self.assertEqual(len(responses.calls), 2)
+ for call in responses.calls:
+ self.assertEqual(call.request.url, f'{MOCK_HAIL_ORIGIN}:5000/multi_lookup')
+ self.assertEqual(call.request.headers.get('From'), 'manage_command')
+ self.assertDictEqual(json.loads(responses.calls[0].request.body), {
'genome_version': 'GRCh37',
'data_type': 'SNV_INDEL',
- 'variant_ids': [['1', 248367227, 'TC', 'T'], ['1', 46859832, 'G', 'A'], ['21', 3343353, 'GAGA', 'G']],
+ 'variant_ids': [['1', 46859832, 'G', 'A'], ['1', 248367227, 'TC', 'T']],
+ })
+ self.assertDictEqual(json.loads(responses.calls[1].request.body), {
+ 'genome_version': 'GRCh37',
+ 'data_type': 'SNV_INDEL',
+ 'variant_ids': [['21', 3343353, 'GAGA', 'G']],
})
annotation_updated_json_1 = SavedVariant.objects.get(guid='SV0000002_1248367227_r0390_100').saved_variant_json
@@ -66,11 +72,12 @@ def test_command(self, mock_logger):
self.assertEqual(len(annotation_updated_json_2['genotypes']), 3)
# Test SVs
+ responses.calls.reset()
Sample.objects.filter(guid='S000147_na21234').update(individual_id=20)
call_command('reload_saved_variant_annotations', 'SV_WGS', 'GRCh37')
- self.assertEqual(len(responses.calls), 2)
- self.assertDictEqual(json.loads(responses.calls[1].request.body), {
+ self.assertEqual(len(responses.calls), 1)
+ self.assertDictEqual(json.loads(responses.calls[0].request.body), {
'genome_version': 'GRCh37',
'data_type': 'SV_WGS',
'variant_ids': ['prefix_19107_DEL'],
diff --git a/seqr/management/tests/reload_saved_variant_json_tests.py b/seqr/management/tests/reload_saved_variant_json_tests.py
index 4ceb4314b6..abe15b58d0 100644
--- a/seqr/management/tests/reload_saved_variant_json_tests.py
+++ b/seqr/management/tests/reload_saved_variant_json_tests.py
@@ -30,7 +30,7 @@ def test_with_param_command(self, mock_get_variants, mock_logger):
[family_1], ['1-46859832-G-A','21-3343353-GAGA-G'], user=None, user_email='manage_command')
logger_info_calls = [
- mock.call('Updated 2 variants for project 1kg project n\xe5me with uni\xe7\xf8de'),
+ mock.call('Updated 2 variants in 1 families for project 1kg project n\xe5me with uni\xe7\xf8de'),
mock.call('Reload Summary: '),
mock.call(' 1kg project n\xe5me with uni\xe7\xf8de: Updated 2 variants')
]
@@ -81,4 +81,4 @@ def test_with_param_command(self, mock_get_variants, mock_logger):
]
mock_logger.info.assert_has_calls(logger_info_calls)
- mock_logger.error.assert_called_with('Error in project 1kg project n\xe5me with uni\xe7\xf8de: Database error.')
+ mock_logger.error.assert_called_with('Error reloading variants in 1kg project n\xe5me with uni\xe7\xf8de: Database error.')
diff --git a/seqr/utils/search/add_data_utils.py b/seqr/utils/search/add_data_utils.py
index 805f176239..d910cb23bf 100644
--- a/seqr/utils/search/add_data_utils.py
+++ b/seqr/utils/search/add_data_utils.py
@@ -39,7 +39,7 @@ def add_new_es_search_samples(request_json, project, user, notify=False, expecte
request_json['mappingFilePath'], user) if request_json.get('mappingFilePath') else {}
ignore_extra_samples = request_json.get('ignoreExtraSamplesInCallset')
sample_project_tuples = [(sample_id, project.name) for sample_id in sample_ids]
- updated_samples, inactivated_sample_guids, num_skipped, updated_family_guids = match_and_update_search_samples(
+ updated_samples, new_samples, inactivated_sample_guids, num_skipped, updated_family_guids = match_and_update_search_samples(
projects=[project],
user=user,
sample_project_tuples=sample_project_tuples,
@@ -52,45 +52,41 @@ def add_new_es_search_samples(request_json, project, user, notify=False, expecte
)
if notify:
- updated_sample_data = updated_samples.values('sample_id', 'individual_id')
- _basic_notify_search_data_loaded(project, dataset_type, sample_type, inactivated_sample_guids, updated_sample_data)
+ _basic_notify_search_data_loaded(project, dataset_type, sample_type, new_samples.values())
return inactivated_sample_guids, updated_family_guids, updated_samples
-def _basic_notify_search_data_loaded(project, dataset_type, sample_type, inactivated_sample_guids, updated_samples, format_email=None, slack_channel=None, include_slack_detail=False):
- previous_loaded_individuals = set(Sample.objects.filter(guid__in=inactivated_sample_guids).values_list('individual_id', flat=True))
- new_sample_ids = [sample['sample_id'] for sample in updated_samples if sample['individual_id'] not in previous_loaded_individuals]
-
+def _basic_notify_search_data_loaded(project, dataset_type, sample_type, new_samples, email_template=None, slack_channel=None, include_slack_detail=False):
msg_dataset_type = '' if dataset_type == Sample.DATASET_TYPE_VARIANT_CALLS else f' {dataset_type}'
- num_new_samples = len(new_sample_ids)
+ num_new_samples = len(new_samples)
sample_summary = f'{num_new_samples} new {sample_type}{msg_dataset_type} samples'
return send_project_notification(
project,
notification=sample_summary,
- email_template=format_email(num_new_samples) if format_email else None,
+ email_template=email_template,
subject='New data available in seqr',
slack_channel=slack_channel,
- slack_detail=', '.join(sorted(new_sample_ids)) if include_slack_detail else None,
+ slack_detail=', '.join(sorted(new_samples)) if include_slack_detail else None,
)
-def notify_search_data_loaded(project, is_internal, dataset_type, sample_type, inactivated_sample_guids, updated_samples, num_samples):
+def notify_search_data_loaded(project, is_internal, dataset_type, sample_type, new_samples, num_samples):
if is_internal:
- format_email = None
+ email_template = None
else:
workspace_name = f'{project.workspace_namespace}/{project.workspace_name}'
- def format_email(num_new_samples):
- reload_summary = f' and {num_samples - num_new_samples} re-loaded samples' if num_samples > num_new_samples else ''
- return '\n'.join([
- f'We are following up on the request to load data from AnVIL on {project.created_date.date().strftime("%B %d, %Y")}.',
- f'We have loaded {{notification}}{reload_summary} from the AnVIL workspace {workspace_name} to the corresponding seqr project {{project_link}}.',
- 'Let us know if you have any questions.',
- ])
+ num_new_samples = len(new_samples)
+ reload_summary = f' and {num_samples - num_new_samples} re-loaded samples' if num_samples > num_new_samples else ''
+ email_template = '\n'.join([
+ f'We are following up on the request to load data from AnVIL on {project.created_date.date().strftime("%B %d, %Y")}.',
+ f'We have loaded {{notification}}{reload_summary} from the AnVIL workspace {workspace_name} to the corresponding seqr project {{project_link}}.',
+ 'Let us know if you have any questions.',
+ ])
url = _basic_notify_search_data_loaded(
- project, dataset_type, sample_type, inactivated_sample_guids, updated_samples, format_email=format_email,
+ project, dataset_type, sample_type, new_samples, email_template=email_template,
slack_channel=SEQR_SLACK_DATA_ALERTS_NOTIFICATION_CHANNEL if is_internal else SEQR_SLACK_ANVIL_DATA_LOADING_CHANNEL,
include_slack_detail=is_internal,
)
diff --git a/seqr/views/utils/dataset_utils.py b/seqr/views/utils/dataset_utils.py
index f75e93eb00..42882f5dd4 100644
--- a/seqr/views/utils/dataset_utils.py
+++ b/seqr/views/utils/dataset_utils.py
@@ -222,7 +222,10 @@ def match_and_update_search_samples(
Family.bulk_update(
user, {'analysis_status': Family.ANALYSIS_STATUS_ANALYSIS_IN_PROGRESS}, guid__in=family_guids_to_update)
- return updated_samples, inactivated_sample_guids, len(remaining_sample_keys), family_guids_to_update
+ previous_loaded_individuals = set(Sample.objects.filter(guid__in=inactivated_sample_guids).values_list('individual_id', flat=True))
+ new_samples = dict(updated_samples.exclude(individual_id__in=previous_loaded_individuals).values_list('id', 'sample_id'))
+
+ return updated_samples, new_samples, inactivated_sample_guids, len(remaining_sample_keys), family_guids_to_update
def _parse_tsv_row(row):
diff --git a/seqr/views/utils/variant_utils.py b/seqr/views/utils/variant_utils.py
index aaf65a129b..52b2fe500a 100644
--- a/seqr/views/utils/variant_utils.py
+++ b/seqr/views/utils/variant_utils.py
@@ -45,12 +45,13 @@ def update_projects_saved_variant_json(projects, user_email, **kwargs):
skipped[project_name] = True
else:
success[project_name] = len(updated_saved_variants)
- logger.info(f'Updated {len(updated_saved_variants)} variants for project {project_name}')
+ family_summary = f' in {len(family_guids)} families' if family_guids else ''
+ logger.info(f'Updated {len(updated_saved_variants)} variants{family_summary} for project {project_name}')
updated_variants_by_id.update({v.variant_id: v.saved_variant_json for v in updated_saved_variants.values()})
except Exception as e:
traceback_message = traceback.format_exc()
logger.error(traceback_message)
- logger.error(f'Error in project {project_name}: {e}')
+ logger.error(f'Error reloading variants in {project_name}: {e}')
error[project_name] = e
logger.info('Reload Summary: ')