From 54ff59c186b2f5d454cdf1181761228f4e304f3f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Karl=20Sv=C3=A4rd?= Date: Thu, 29 Aug 2024 14:04:55 +0200 Subject: [PATCH] Format with black and isort --- .../files/sample_sheet/create_sample_sheet.py | 4 +- .../udf/calculate/adjust_missing_reads.py | 131 +++++++++++++----- cg_lims/EPPs/udf/calculate/base.py | 2 +- cg_lims/options.py | 29 +++- 4 files changed, 123 insertions(+), 43 deletions(-) diff --git a/cg_lims/EPPs/files/sample_sheet/create_sample_sheet.py b/cg_lims/EPPs/files/sample_sheet/create_sample_sheet.py index bb768c70..3ed0a022 100644 --- a/cg_lims/EPPs/files/sample_sheet/create_sample_sheet.py +++ b/cg_lims/EPPs/files/sample_sheet/create_sample_sheet.py @@ -152,7 +152,9 @@ def calculate_index_hamming_distance( return string_hamming_distance( index_1=index_1.sequence[-len(index_2.sequence) :], index_2=index_2.sequence ) - message: str = f"Non-supported index type identified for indexes {index_1.sequence} and {index_2.sequence}: '{index_1.type}'." + message: str = ( + f"Non-supported index type identified for indexes {index_1.sequence} and {index_2.sequence}: '{index_1.type}'." + ) LOG.error(message) raise InvalidValueError(message) diff --git a/cg_lims/EPPs/udf/calculate/adjust_missing_reads.py b/cg_lims/EPPs/udf/calculate/adjust_missing_reads.py index 9798e96f..5c33420c 100644 --- a/cg_lims/EPPs/udf/calculate/adjust_missing_reads.py +++ b/cg_lims/EPPs/udf/calculate/adjust_missing_reads.py @@ -2,38 +2,43 @@ import sys import click +from cg_lims import options from cg_lims.exceptions import LimsError, MissingCgFieldError, MissingUDFsError from cg_lims.get.artifacts import get_artifacts from cg_lims.get.samples import get_one_sample_from_artifact from cg_lims.status_db_api import StatusDBAPI from genologics.entities import Artifact from requests.exceptions import ConnectionError -from cg_lims import options LOG = logging.getLogger(__name__) + def calculate_adjusted_reads(artifact: Artifact, factor: str) -> float: """A function to calculate the adjusted reads to sequence for each artifact with the desired apptag""" reads = artifact.udf.get("Reads to sequence (M)") - return round(float(reads)*float(factor), 1) + return round(float(reads) * float(factor), 1) -def adjust_wgs_topups(artifact: Artifact, factor_wgs_lower: str, factor_wgs_higher: str, threshold_reads: str) -> None: - """A function that calculates adjusted reads to sequence for WGS topups, where the 'topup' factor is determined + +def adjust_wgs_topups( + artifact: Artifact, factor_wgs_lower: str, factor_wgs_higher: str, threshold_reads: str +) -> None: + """A function that calculates adjusted reads to sequence for WGS topups, where the 'topup' factor is determined by a threshold for the reads to sequence. This is specified in the cli""" - + valid_value = validate_udf_values(artifact=artifact) if valid_value: reads = float(artifact.udf.get("Reads to sequence (M)")) if reads < float(threshold_reads): - adjusted_reads = round(float(reads)*float(factor_wgs_lower), 1) + adjusted_reads = round(float(reads) * float(factor_wgs_lower), 1) else: - adjusted_reads = round(float(reads)*float(factor_wgs_higher), 1) + adjusted_reads = round(float(reads) * float(factor_wgs_higher), 1) artifact.udf["Reads to sequence (M)"] = str(adjusted_reads) artifact.put() + def reset_microbial_reads(artifact: Artifact, reset_microbial_reads: str) -> None: """A function that resets the reads to sequence for microbial samples, and the threshold_reads specifies what they are supposed to be reset to""" @@ -44,8 +49,9 @@ def reset_microbial_reads(artifact: Artifact, reset_microbial_reads: str) -> Non artifact.udf["Reads to sequence (M)"] = reset_microbial_reads artifact.put() + def is_topup(artifact: Artifact) -> bool: - """A function that determines whether an artifact has already been sequenced before or not, and therefore is a topup + """A function that determines whether an artifact has already been sequenced before or not, and therefore is a topup sample/artifact or not""" output = False @@ -53,6 +59,7 @@ def is_topup(artifact: Artifact) -> bool: output = True return output + def is_adjusted(artifact: Artifact) -> bool: """A function that checks if the process UDF Adjusted Reads to Sequence is set/true. This will be updated after the EPP to adjust the reads to sequence has run one time""" @@ -63,18 +70,23 @@ def is_adjusted(artifact: Artifact) -> bool: output = True return output + def validate_udf_values(artifact: Artifact) -> bool: """A function checking whether Reads to Sequence (M) has a negative/no value. Then the function returns the output as 'False' and logs all those sample IDs in the EPP log""" output = True - if not artifact.udf.get("Reads to sequence (M)") or float(artifact.udf.get("Reads to sequence (M)")) < 0: + if ( + not artifact.udf.get("Reads to sequence (M)") + or float(artifact.udf.get("Reads to sequence (M)")) < 0 + ): output = False LOG.info( f"Sample {artifact.samples[0].id} has no or a negative value for Reads to sequence (M). Skipping." ) return output + def adjust_reads(artifact: Artifact, apptags: tuple, factor: str) -> None: """Only artifacts that have passed the validation of acceptable Reads to Sequence (M) values will be adjusted""" @@ -85,24 +97,43 @@ def adjust_reads(artifact: Artifact, apptags: tuple, factor: str) -> None: artifact.udf["Reads to sequence (M)"] = str(adjusted_reads) artifact.put() + @click.command() @options.apptag_wgs(help="String of UDF Sequencing Analysis, also known as apptag, for WGS samples") -@options.apptag_wgs_tumor(help="String of UDF Sequencing Analysis, also known as apptag, for WGS tumor samples") +@options.apptag_wgs_tumor( + help="String of UDF Sequencing Analysis, also known as apptag, for WGS tumor samples" +) @options.apptag_tga(help="String of UDF Sequencing Analysis, also known as apptag, for TGA samples") -@options.apptag_micro(help="String of UDF Sequencing Analysis, also known as apptag, for micro samples") +@options.apptag_micro( + help="String of UDF Sequencing Analysis, also known as apptag, for micro samples" +) @options.apptag_rml(help="String of UDF Sequencing Analysis, also known as apptag, for RML samples") -@options.apptag_virus(help="String of UDF Sequencing Analysis, also known as apptag, for virus samples") +@options.apptag_virus( + help="String of UDF Sequencing Analysis, also known as apptag, for virus samples" +) @options.apptag_rna(help="String of UDF Sequencing Analysis, also known as apptag, for RNA samples") -@options.factor_wgs_tumor(help= "Factor to multiply Reads to sequence (M) with for WGS tumor samples") -@options.factor_tga(help= "Factor to multiply Reads to sequence (M) with for TGA samples") -@options.factor_micro(help= "Factor to multiply Reads to sequence (M) with for micro samples") -@options.factor_rml(help= "Factor to multiply Reads to sequence (M) with for RML samples") -@options.factor_rna(help= "Factor to multiply Reads to sequence (M) with for RNA samples") -@options.factor_rna_topups(help= "Factor to multiply Reads to sequence (M) with for RNA topup samples") -@options.factor_rml_topups(help= "Factor to multiply Reads to sequence (M) with for RML topup samples") -@options.factor_tga_topups(help= "Factor to multiply Reads to sequence (M) with for TGA topup samples") -@options.factor_wgs_lower(help= "Lower factor to multiply Reads to sequence (M) with for WGS samples") -@options.factor_wgs_higher(help= "Higher factor to multiply Reads to sequence (M) with for WGS samples") +@options.factor_wgs_tumor( + help="Factor to multiply Reads to sequence (M) with for WGS tumor samples" +) +@options.factor_tga(help="Factor to multiply Reads to sequence (M) with for TGA samples") +@options.factor_micro(help="Factor to multiply Reads to sequence (M) with for micro samples") +@options.factor_rml(help="Factor to multiply Reads to sequence (M) with for RML samples") +@options.factor_rna(help="Factor to multiply Reads to sequence (M) with for RNA samples") +@options.factor_rna_topups( + help="Factor to multiply Reads to sequence (M) with for RNA topup samples" +) +@options.factor_rml_topups( + help="Factor to multiply Reads to sequence (M) with for RML topup samples" +) +@options.factor_tga_topups( + help="Factor to multiply Reads to sequence (M) with for TGA topup samples" +) +@options.factor_wgs_lower( + help="Lower factor to multiply Reads to sequence (M) with for WGS samples" +) +@options.factor_wgs_higher( + help="Higher factor to multiply Reads to sequence (M) with for WGS samples" +) @options.threshold_reads(help="Threshold for Reads to sequence (M) during adjustment") @options.reset_micro_reads(help="A value to re-set Reads to sequence (M) for microbial samples") @options.reset_virus_reads(help="A value to re-set Reads to sequence (M) for virus samples") @@ -130,7 +161,7 @@ def adjust_missing_reads( reset_micro_reads: str, reset_virus_reads: str, ): - """Script to calculate the adjusted Reads to sequence (M) with a specific factor for specific apptags, + """Script to calculate the adjusted Reads to sequence (M) with a specific factor for specific apptags, specified in the command line""" process = ctx.obj["process"] @@ -141,41 +172,67 @@ def adjust_missing_reads( for artifact in artifacts: if not is_adjusted(artifact=artifact): for app in apptag_wgs: - if app in artifact.samples[0].udf.get("Sequencing Analysis") and is_topup(artifact=artifact): - adjust_wgs_topups(artifact=artifact, factor_wgs_lower=factor_wgs_lower, - factor_wgs_higher=factor_wgs_higher, threshold_reads=threshold_reads) + if app in artifact.samples[0].udf.get("Sequencing Analysis") and is_topup( + artifact=artifact + ): + adjust_wgs_topups( + artifact=artifact, + factor_wgs_lower=factor_wgs_lower, + factor_wgs_higher=factor_wgs_higher, + threshold_reads=threshold_reads, + ) for app in apptag_wgs_tumor: if app in artifact.samples[0].udf.get("Sequencing Analysis"): if is_topup(artifact=artifact): - adjust_wgs_topups(artifact=artifact, factor_wgs_lower=factor_wgs_lower, - factor_wgs_higher=factor_wgs_higher, threshold_reads=threshold_reads) - else: - adjust_reads(artifact=artifact, apptags=apptag_wgs_tumor, factor=factor_wgs_tumor) + adjust_wgs_topups( + artifact=artifact, + factor_wgs_lower=factor_wgs_lower, + factor_wgs_higher=factor_wgs_higher, + threshold_reads=threshold_reads, + ) + else: + adjust_reads( + artifact=artifact, apptags=apptag_wgs_tumor, factor=factor_wgs_tumor + ) for app in apptag_tga: if app in artifact.samples[0].udf.get("Sequencing Analysis"): if is_topup(artifact=artifact): - adjust_reads(artifact=artifact, apptags=apptag_tga, factor=factor_tga_topups) + adjust_reads( + artifact=artifact, apptags=apptag_tga, factor=factor_tga_topups + ) else: adjust_reads(artifact=artifact, apptags=apptag_tga, factor=factor_tga) for app in apptag_micro: if app in artifact.samples[0].udf.get("Sequencing Analysis"): if is_topup(artifact=artifact): - reset_microbial_reads(artifact=artifact, reset_microbial_reads=reset_micro_reads) + reset_microbial_reads( + artifact=artifact, reset_microbial_reads=reset_micro_reads + ) else: - adjust_reads(artifact=artifact, apptags=apptag_micro, factor=factor_micro) + adjust_reads( + artifact=artifact, apptags=apptag_micro, factor=factor_micro + ) for app in apptag_virus: - if app in artifact.samples[0].udf.get("Sequencing Analysis") and is_topup(artifact=artifact): - reset_microbial_reads(artifact=artifact, reset_microbial_reads=reset_virus_reads) + if app in artifact.samples[0].udf.get("Sequencing Analysis") and is_topup( + artifact=artifact + ): + reset_microbial_reads( + artifact=artifact, reset_microbial_reads=reset_virus_reads + ) for app in apptag_rml: if app in artifact.samples[0].udf.get("Sequencing Analysis"): if is_topup(artifact=artifact): - adjust_reads(artifact=artifact, apptags=apptag_rml, factor=factor_rml_topups) + adjust_reads( + artifact=artifact, apptags=apptag_rml, factor=factor_rml_topups + ) else: adjust_reads(artifact=artifact, apptags=apptag_rml, factor=factor_rml) for app in apptag_rna: if app in artifact.samples[0].udf.get("Sequencing Analysis"): if is_topup(artifact=artifact): - adjust_reads(artifact=artifact, apptags=apptag_rna, factor=factor_rna_topups) + adjust_reads( + artifact=artifact, apptags=apptag_rna, factor=factor_rna_topups + ) else: adjust_reads(artifact=artifact, apptags=apptag_rna, factor=factor_rna) if is_adjusted(artifact=artifact): diff --git a/cg_lims/EPPs/udf/calculate/base.py b/cg_lims/EPPs/udf/calculate/base.py index 2ec3708c..3a7fcf95 100644 --- a/cg_lims/EPPs/udf/calculate/base.py +++ b/cg_lims/EPPs/udf/calculate/base.py @@ -1,6 +1,7 @@ #!/usr/bin/env python import click +from cg_lims.EPPs.udf.calculate.adjust_missing_reads import adjust_missing_reads from cg_lims.EPPs.udf.calculate.aliquot_volume import aliquot_volume from cg_lims.EPPs.udf.calculate.calculate_amount_ng import calculate_amount_ng from cg_lims.EPPs.udf.calculate.calculate_amount_ng_fmol import calculate_amount_ng_fmol @@ -28,7 +29,6 @@ from cg_lims.EPPs.udf.calculate.sum_missing_reads_in_pool import missing_reads_in_pool from cg_lims.EPPs.udf.calculate.twist_aliquot_amount import twist_aliquot_amount from cg_lims.EPPs.udf.calculate.twist_get_volumes_from_buffer import get_volumes_from_buffer -from cg_lims.EPPs.udf.calculate.adjust_missing_reads import adjust_missing_reads # commands from cg_lims.EPPs.udf.calculate.twist_pool import twist_pool diff --git a/cg_lims/options.py b/cg_lims/options.py index 9c478a94..fb4369ca 100644 --- a/cg_lims/options.py +++ b/cg_lims/options.py @@ -435,6 +435,7 @@ def apptag( help=help, ) + def apptag_wgs( help: str = "String of UDF Sequencing Analysis, also known as apptag, for WGS samples", ) -> click.option: @@ -445,6 +446,7 @@ def apptag_wgs( help=help, ) + def apptag_wgs_tumor( help: str = "String of UDF Sequencing Analysis, also known as apptag, for WGS tumor samples", ) -> click.option: @@ -455,6 +457,7 @@ def apptag_wgs_tumor( help=help, ) + def apptag_tga( help: str = "String of UDF Sequencing Analysis, also known as apptag, for TGA samples", ) -> click.option: @@ -465,6 +468,7 @@ def apptag_tga( help=help, ) + def apptag_micro( help: str = "String of UDF Sequencing Analysis, also known as apptag, for micro samples", ) -> click.option: @@ -475,6 +479,7 @@ def apptag_micro( help=help, ) + def apptag_rml( help: str = "String of UDF Sequencing Analysis, also known as apptag, for RML samples", ) -> click.option: @@ -485,6 +490,7 @@ def apptag_rml( help=help, ) + def apptag_virus( help: str = "String of UDF Sequencing Analysis, also known as apptag, for virus samples", ) -> click.option: @@ -495,6 +501,7 @@ def apptag_virus( help=help, ) + def apptag_rna( help: str = "String of UDF Sequencing Analysis, also known as apptag, for RNA samples", ) -> click.option: @@ -505,6 +512,7 @@ def apptag_rna( help=help, ) + def factor( help: str = "Factor to multiply Reads to sequence (M) with", ) -> click.option: @@ -515,6 +523,7 @@ def factor( help=help, ) + def factor_wgs_tumor( help: str = "Factor to multiply Reads to sequence (M) with for WGS tumor samples", ) -> click.option: @@ -525,6 +534,7 @@ def factor_wgs_tumor( help=help, ) + def factor_tga( help: str = "Factor to multiply Reads to sequence (M) with for TGA samples", ) -> click.option: @@ -535,6 +545,7 @@ def factor_tga( help=help, ) + def factor_micro( help: str = "Factor to multiply Reads to sequence (M) with for micro samples", ) -> click.option: @@ -545,6 +556,7 @@ def factor_micro( help=help, ) + def factor_rml( help: str = "Factor to multiply Reads to sequence (M) with for RML samples", ) -> click.option: @@ -555,6 +567,7 @@ def factor_rml( help=help, ) + def factor_rna( help: str = "Factor to multiply Reads to sequence (M) with for RNA samples", ) -> click.option: @@ -565,8 +578,9 @@ def factor_rna( help=help, ) + def factor_rna_topups( - help: str = "Factor to multiply Reads to sequence (M) with for RNA topup samples", + help: str = "Factor to multiply Reads to sequence (M) with for RNA topup samples", ) -> click.option: return click.option( "--factor-rna-topups", @@ -575,8 +589,9 @@ def factor_rna_topups( help=help, ) + def factor_rml_topups( - help: str = "Factor to multiply Reads to sequence (M) with for RML topup samples", + help: str = "Factor to multiply Reads to sequence (M) with for RML topup samples", ) -> click.option: return click.option( "--factor-rml-topups", @@ -585,8 +600,9 @@ def factor_rml_topups( help=help, ) + def factor_tga_topups( - help: str = "Factor to multiply Reads to sequence (M) with for TGA topup samples", + help: str = "Factor to multiply Reads to sequence (M) with for TGA topup samples", ) -> click.option: return click.option( "--factor-tga-topups", @@ -595,6 +611,7 @@ def factor_tga_topups( help=help, ) + def factor_wgs_lower( help: str = "Lower factor to multiply Reads to sequence (M) with for WGS samples", ) -> click.option: @@ -605,6 +622,7 @@ def factor_wgs_lower( help=help, ) + def factor_wgs_higher( help: str = "Higher factor to multiply Reads to sequence (M) with for WGS samples", ) -> click.option: @@ -615,6 +633,7 @@ def factor_wgs_higher( help=help, ) + def threshold_reads( help: str = "Threshold for Reads to sequence (M) during adjustment", ) -> click.option: @@ -625,6 +644,7 @@ def threshold_reads( help=help, ) + def reset_micro_reads( help: str = "A value to re-set Reads to sequence (M) for microbial samples", ) -> click.option: @@ -635,6 +655,7 @@ def reset_micro_reads( help=help, ) + def reset_virus_reads( help: str = "A value to re-set Reads to sequence (M) for virus samples", ) -> click.option: @@ -643,4 +664,4 @@ def reset_virus_reads( required=True, multiple=False, help=help, - ) \ No newline at end of file + )