scout/server/blueprints/variants/views.py
def variants(institute_id, case_name):
    """Display a list of SNV variants."""
    page = int(request.form.get('page', 1))

    institute_obj, case_obj = institute_and_case(store, institute_id, case_name)
    variant_type = request.args.get('variant_type', 'clinical')

    # Update filter settings if Clinical Filter was requested

    default_panels = []
    for panel in case_obj['panels']:
        if panel.get('is_default'):
            default_panels.append(panel['panel_name'])

    request.form.get('gene_panels')
    if bool(request.form.get('clinical_filter')):
        clinical_filter = MultiDict({
            'variant_type': 'clinical',
            'region_annotations': ['exonic','splicing'],
            'functional_annotations': SEVERE_SO_TERMS,
            'clinsig': [4,5],
            'clinsig_confident_always_returned': True,
            'gnomad_frequency': str(institute_obj['frequency_cutoff']),
            'variant_type': 'clinical',
            'gene_panels': default_panels
             })

    if(request.method == "POST"):
        if bool(request.form.get('clinical_filter')):
            form = FiltersForm(clinical_filter)
            form.csrf_token = request.args.get('csrf_token')
        else:
            form = FiltersForm(request.form)
    else:
        form = FiltersForm(request.args)

    # populate available panel choices
    available_panels = case_obj.get('panels', []) + [
        {'panel_name': 'hpo', 'display_name': 'HPO'}]

    panel_choices = [(panel['panel_name'], panel['display_name'])
                     for panel in available_panels]

    form.gene_panels.choices = panel_choices

    # upload gene panel if symbol file exists
    if (request.files):
        file = request.files[form.symbol_file.name]

    if request.files and file and file.filename != '':
        log.debug("Upload file request files: {0}".format(request.files.to_dict()))
        try:
            stream = io.StringIO(file.stream.read().decode('utf-8'), newline=None)
        except UnicodeDecodeError as error:
            flash("Only text files are supported!", 'warning')
            return redirect(request.referrer)

        hgnc_symbols_set = set(form.hgnc_symbols.data)
        log.debug("Symbols prior to upload: {0}".format(hgnc_symbols_set))
        new_hgnc_symbols = controllers.upload_panel(store, institute_id, case_name, stream)
        hgnc_symbols_set.update(new_hgnc_symbols)
        form.hgnc_symbols.data = hgnc_symbols_set
        # reset gene panels
        form.gene_panels.data = ''

    # update status of case if vistited for the first time
    if case_obj['status'] == 'inactive' and not current_user.is_admin:
        flash('You just activated this case!', 'info')
        user_obj = store.user(current_user.email)
        case_link = url_for('cases.case', institute_id=institute_obj['_id'],
                            case_name=case_obj['display_name'])
        store.update_status(institute_obj, case_obj, user_obj, 'active', case_link)

    # check if supplied gene symbols exist
    hgnc_symbols = []
    non_clinical_symbols = []
    not_found_symbols = []
    not_found_ids = []
    if (form.hgnc_symbols.data) and len(form.hgnc_symbols.data) > 0:
        is_clinical = form.data.get('variant_type', 'clinical') == 'clinical'
        clinical_symbols = store.clinical_symbols(case_obj) if is_clinical else None
        for hgnc_symbol in form.hgnc_symbols.data:
            if hgnc_symbol.isdigit():
                hgnc_gene = store.hgnc_gene(int(hgnc_symbol))
                if hgnc_gene is None:
                    not_found_ids.append(hgnc_symbol)
                else:
                    hgnc_symbols.append(hgnc_gene['hgnc_symbol'])
            elif store.hgnc_genes(hgnc_symbol).count() == 0:
                  not_found_symbols.append(hgnc_symbol)
            elif is_clinical and (hgnc_symbol not in clinical_symbols):
                 non_clinical_symbols.append(hgnc_symbol)
            else:
                hgnc_symbols.append(hgnc_symbol)

    if (not_found_ids):
        flash("HGNC id not found: {}".format(", ".join(not_found_ids)), 'warning')
    if (not_found_symbols):
        flash("HGNC symbol not found: {}".format(", ".join(not_found_symbols)), 'warning')
    if (non_clinical_symbols):
        flash("Gene not included in clinical list: {}".format(", ".join(non_clinical_symbols)), 'warning')
    form.hgnc_symbols.data = hgnc_symbols

    # handle HPO gene list separately
    if form.data['gene_panels'] == ['hpo']:
        hpo_symbols = list(set(term_obj['hgnc_symbol'] for term_obj in
                               case_obj['dynamic_gene_list']))
        form.hgnc_symbols.data = hpo_symbols

    variants_query = store.variants(case_obj['_id'], query=form.data)
    data = {}

    if request.form.get('export'):
        document_header = controllers.variants_export_header(case_obj)
        export_lines = []
        if form.data['chrom'] == 'MT':
            # Return all MT variants
            export_lines = controllers.variant_export_lines(store, case_obj, variants_query)
        else:
            # Return max 500 variants
            export_lines = controllers.variant_export_lines(store, case_obj, variants_query.limit(500))

        def generate(header, lines):
            yield header + '\n'
            for line in lines:
                yield line + '\n'

        headers = Headers()
        headers.add('Content-Disposition','attachment', filename=str(case_obj['display_name'])+'-filtered_variants.csv')

        # return a csv with the exported variants
        return Response(generate(",".join(document_header), export_lines), mimetype='text/csv',
                        headers=headers)

    data = controllers.variants(store, institute_obj, case_obj, variants_query, page)

    return dict(institute=institute_obj, case=case_obj, form=form,
                    severe_so_terms=SEVERE_SO_TERMS, page=page, **data)
def variants(institute_id, case_name):
    """Display a list of SNV variants."""
    page = int(request.form.get('page', 1))

    institute_obj, case_obj = institute_and_case(store, institute_id, case_name)
    variant_type = request.args.get('variant_type', 'clinical')

    # Update filter settings if Clinical Filter was requested

    default_panels = []
    for panel in case_obj['panels']:
        if panel.get('is_default'):
            default_panels.append(panel['panel_name'])

    request.form.get('gene_panels')
    if bool(request.form.get('clinical_filter')):
        clinical_filter = MultiDict({
            'variant_type': 'clinical',
            'region_annotations': ['exonic','splicing'],
            'functional_annotations': SEVERE_SO_TERMS,
            'clinsig': [4,5],
            'clinsig_confident_always_returned': True,
            'gnomad_frequency': str(institute_obj['frequency_cutoff']),
            'variant_type': 'clinical',
            'gene_panels': default_panels
             })

    if(request.method == "POST"):
        if bool(request.form.get('clinical_filter')):
            form = FiltersForm(clinical_filter)
            form.csrf_token = request.args.get('csrf_token')
        else:
            form = FiltersForm(request.form)
    else:
        form = FiltersForm(request.args)

    # populate available panel choices
    available_panels = case_obj.get('panels', []) + [
        {'panel_name': 'hpo', 'display_name': 'HPO'}]

    panel_choices = [(panel['panel_name'], panel['display_name'])
                     for panel in available_panels]

    form.gene_panels.choices = panel_choices

    # upload gene panel if symbol file exists
    if (request.files):
        file = request.files[form.symbol_file.name]

    if request.files and file and file.filename != '':
        log.debug("Upload file request files: {0}".format(request.files.to_dict()))
        try:
            stream = io.StringIO(file.stream.read().decode('utf-8'), newline=None)
        except UnicodeDecodeError as error:
            flash("Only text files are supported!", 'warning')
            return redirect(request.referrer)

        hgnc_symbols_set = set(form.hgnc_symbols.data)
        log.debug("Symbols prior to upload: {0}".format(hgnc_symbols_set))
        new_hgnc_symbols = controllers.upload_panel(store, institute_id, case_name, stream)
        hgnc_symbols_set.update(new_hgnc_symbols)
        form.hgnc_symbols.data = hgnc_symbols_set
        # reset gene panels
        form.gene_panels.data = ''

    # update status of case if vistited for the first time
    if case_obj['status'] == 'inactive' and not current_user.is_admin:
        flash('You just activated this case!', 'info')
        user_obj = store.user(current_user.email)
        case_link = url_for('cases.case', institute_id=institute_obj['_id'],
                            case_name=case_obj['display_name'])
        store.update_status(institute_obj, case_obj, user_obj, 'active', case_link)

    # check if supplied gene symbols exist
    hgnc_symbols = []
    non_clinical_symbols = []
    not_found_symbols = []
    not_found_ids = []
    if (form.hgnc_symbols.data) and len(form.hgnc_symbols.data) > 0:
        is_clinical = form.data.get('variant_type', 'clinical') == 'clinical'
        clinical_symbols = store.clinical_symbols(case_obj) if is_clinical else None
        for hgnc_symbol in form.hgnc_symbols.data:
            if hgnc_symbol.isdigit():
                hgnc_gene = store.hgnc_gene(int(hgnc_symbol))
                if hgnc_gene is None:
                    not_found_ids.append(hgnc_symbol)
                else:
                    hgnc_symbols.append(hgnc_gene['hgnc_symbol'])
            elif store.hgnc_genes(hgnc_symbol).count() == 0:
                  not_found_symbols.append(hgnc_symbol)
            elif is_clinical and (hgnc_symbol not in clinical_symbols):
                 non_clinical_symbols.append(hgnc_symbol)
            else:
                hgnc_symbols.append(hgnc_symbol)

    if (not_found_ids):
        flash("HGNC id not found: {}".format(", ".join(not_found_ids)), 'warning')
    if (not_found_symbols):
        flash("HGNC symbol not found: {}".format(", ".join(not_found_symbols)), 'warning')
    if (non_clinical_symbols):
        flash("Gene not included in clinical list: {}".format(", ".join(non_clinical_symbols)), 'warning')
    form.hgnc_symbols.data = hgnc_symbols

    # handle HPO gene list separately
    if form.data['gene_panels'] == ['hpo']:
        hpo_symbols = list(set(term_obj['hgnc_symbol'] for term_obj in
                               case_obj['dynamic_gene_list']))
        form.hgnc_symbols.data = hpo_symbols

    variants_query = store.variants(case_obj['_id'], query=form.data)
    data = {}

    if request.form.get('export'):
        document_header = controllers.variants_export_header(case_obj)
        export_lines = []
        if form.data['chrom'] == 'MT':
            # Return all MT variants
            export_lines = controllers.variant_export_lines(store, case_obj, variants_query)
        else:
            # Return max 500 variants
            export_lines = controllers.variant_export_lines(store, case_obj, variants_query.limit(500))

        def generate(header, lines):
            yield header + '\n'
            for line in lines:
                yield line + '\n'

        headers = Headers()
        headers.add('Content-Disposition','attachment', filename=str(case_obj['display_name'])+'-filtered_variants.csv')

        # return a csv with the exported variants
        return Response(generate(",".join(document_header), export_lines), mimetype='text/csv',
                        headers=headers)

    data = controllers.variants(store, institute_obj, case_obj, variants_query, page)

    return dict(institute=institute_obj, case=case_obj, form=form,
                    severe_so_terms=SEVERE_SO_TERMS, page=page, **data)
def variants ( institute_id , case_name ) : page = int ( request . form . get ( 'page' , 1 ) ) institute_obj , case_obj = institute_and_case ( store , institute_id , case_name ) variant_type = request . args . get ( 'variant_type' , 'clinical' ) # Update filter settings if Clinical Filter was requested default_panels = [ ] for panel in case_obj [ 'panels' ] : if panel . get ( 'is_default' ) : default_panels . append ( panel [ 'panel_name' ] ) request . form . get ( 'gene_panels' ) if bool ( request . form . get ( 'clinical_filter' ) ) : clinical_filter = MultiDict ( { 'variant_type' : 'clinical' , 'region_annotations' : [ 'exonic' , 'splicing' ] , 'functional_annotations' : SEVERE_SO_TERMS , 'clinsig' : [ 4 , 5 ] , 'clinsig_confident_always_returned' : True , 'gnomad_frequency' : str ( institute_obj [ 'frequency_cutoff' ] ) , 'variant_type' : 'clinical' , 'gene_panels' : default_panels } ) if ( request . method == "POST" ) : if bool ( request . form . get ( 'clinical_filter' ) ) : form = FiltersForm ( clinical_filter ) form . csrf_token = request . args . get ( 'csrf_token' ) else : form = FiltersForm ( request . form ) else : form = FiltersForm ( request . args ) # populate available panel choices available_panels = case_obj . get ( 'panels' , [ ] ) + [ { 'panel_name' : 'hpo' , 'display_name' : 'HPO' } ] panel_choices = [ ( panel [ 'panel_name' ] , panel [ 'display_name' ] ) for panel in available_panels ] form . gene_panels . choices = panel_choices # upload gene panel if symbol file exists if ( request . files ) : file = request . files [ form . symbol_file . name ] if request . files and file and file . filename != '' : log . debug ( "Upload file request files: {0}" . format ( request . files . to_dict ( ) ) ) try : stream = io . StringIO ( file . stream . read ( ) . decode ( 'utf-8' ) , newline = None ) except UnicodeDecodeError as error : flash ( "Only text files are supported!" , 'warning' ) return redirect ( request . referrer ) hgnc_symbols_set = set ( form . hgnc_symbols . data ) log . debug ( "Symbols prior to upload: {0}" . format ( hgnc_symbols_set ) ) new_hgnc_symbols = controllers . upload_panel ( store , institute_id , case_name , stream ) hgnc_symbols_set . update ( new_hgnc_symbols ) form . hgnc_symbols . data = hgnc_symbols_set # reset gene panels form . gene_panels . data = '' # update status of case if vistited for the first time if case_obj [ 'status' ] == 'inactive' and not current_user . is_admin : flash ( 'You just activated this case!' , 'info' ) user_obj = store . user ( current_user . email ) case_link = url_for ( 'cases.case' , institute_id = institute_obj [ '_id' ] , case_name = case_obj [ 'display_name' ] ) store . update_status ( institute_obj , case_obj , user_obj , 'active' , case_link ) # check if supplied gene symbols exist hgnc_symbols = [ ] non_clinical_symbols = [ ] not_found_symbols = [ ] not_found_ids = [ ] if ( form . hgnc_symbols . data ) and len ( form . hgnc_symbols . data ) > 0 : is_clinical = form . data . get ( 'variant_type' , 'clinical' ) == 'clinical' clinical_symbols = store . clinical_symbols ( case_obj ) if is_clinical else None for hgnc_symbol in form . hgnc_symbols . data : if hgnc_symbol . isdigit ( ) : hgnc_gene = store . hgnc_gene ( int ( hgnc_symbol ) ) if hgnc_gene is None : not_found_ids . append ( hgnc_symbol ) else : hgnc_symbols . append ( hgnc_gene [ 'hgnc_symbol' ] ) elif store . hgnc_genes ( hgnc_symbol ) . count ( ) == 0 : not_found_symbols . append ( hgnc_symbol ) elif is_clinical and ( hgnc_symbol not in clinical_symbols ) : non_clinical_symbols . append ( hgnc_symbol ) else : hgnc_symbols . append ( hgnc_symbol ) if ( not_found_ids ) : flash ( "HGNC id not found: {}" . format ( ", " . join ( not_found_ids ) ) , 'warning' ) if ( not_found_symbols ) : flash ( "HGNC symbol not found: {}" . format ( ", " . join ( not_found_symbols ) ) , 'warning' ) if ( non_clinical_symbols ) : flash ( "Gene not included in clinical list: {}" . format ( ", " . join ( non_clinical_symbols ) ) , 'warning' ) form . hgnc_symbols . data = hgnc_symbols # handle HPO gene list separately if form . data [ 'gene_panels' ] == [ 'hpo' ] : hpo_symbols = list ( set ( term_obj [ 'hgnc_symbol' ] for term_obj in case_obj [ 'dynamic_gene_list' ] ) ) form . hgnc_symbols . data = hpo_symbols variants_query = store . variants ( case_obj [ '_id' ] , query = form . data ) data = { } if request . form . get ( 'export' ) : document_header = controllers . variants_export_header ( case_obj ) export_lines = [ ] if form . data [ 'chrom' ] == 'MT' : # Return all MT variants export_lines = controllers . variant_export_lines ( store , case_obj , variants_query ) else : # Return max 500 variants export_lines = controllers . variant_export_lines ( store , case_obj , variants_query . limit ( 500 ) ) def generate ( header , lines ) : yield header + '\n' for line in lines : yield line + '\n' headers = Headers ( ) headers . add ( 'Content-Disposition' , 'attachment' , filename = str ( case_obj [ 'display_name' ] ) + '-filtered_variants.csv' ) # return a csv with the exported variants return Response ( generate ( "," . join ( document_header ) , export_lines ) , mimetype = 'text/csv' , headers = headers ) data = controllers . variants ( store , institute_obj , case_obj , variants_query , page ) return dict ( institute = institute_obj , case = case_obj , form = form , severe_so_terms = SEVERE_SO_TERMS , page = page , * * data )
Display a list of SNV variants.
Display a list of SNV variants .