-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add support for 25B and 1.5B flow cells (#453)(minor)
### Added - New constants for 25B and 1.5B flow cells - New EPP for checking that given process UDFs have been set, cg_lims/EPPs/udf/check/check_process_udfs.py - New constants file for the `set` EPPs, cg_lims/EPPs/udf/set/constants.py - New EPP for automatically filling in the default sequencing settings for a given flow cell type, cg_lims/EPPs/udf/set/set_sequencing_settings.py ### Changed - Reworked the way constants were used in cg_lims/EPPs/udf/calculate/novaseq_x_denaturation.py - updated cg_lims/EPPs/udf/calculate/novaseq_x_volumes.py to use the new constants
- Loading branch information
1 parent
1e838a6
commit 5222040
Showing
8 changed files
with
247 additions
and
22 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| import logging | ||
| import sys | ||
| from typing import List | ||
| import click | ||
| from genologics.entities import Process | ||
| from cg_lims import options | ||
| from cg_lims.exceptions import LimsError, MissingUDFsError | ||
|
|
||
| LOG = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| def check_udfs(process: Process, process_udfs: List[str]) -> None: | ||
| """Check that process UDFs are set.""" | ||
|
|
||
| warning = [] | ||
| for udf in process_udfs: | ||
| if process.udf.get(udf) is None: | ||
| warning.append(f"UDF: '{udf}' is missing for the step.") | ||
| if warning: | ||
| LOG.warning(" ".join(warning)) | ||
| raise MissingUDFsError(message=" ".join(warning)) | ||
| LOG.info("Process UDFs were all set.") | ||
|
|
||
|
|
||
| @click.command() | ||
| @options.process_udfs() | ||
| @click.pass_context | ||
| def check_process_udfs( | ||
| ctx: click.Context, | ||
| process_udfs: List[str], | ||
| ): | ||
| """Script to check that process UDFs are set.""" | ||
|
|
||
| LOG.info(f"Running {ctx.command_path} with params: {ctx.params}") | ||
| process: Process = ctx.obj["process"] | ||
| try: | ||
| check_udfs(process=process, process_udfs=process_udfs) | ||
| click.echo("Process UDFs were checked.") | ||
| except LimsError as e: | ||
| sys.exit(e.message) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| from cg_lims.enums import IntEnum | ||
|
|
||
|
|
||
| class DefaultReadLength(IntEnum): | ||
| """Default read length most usually used by each flow cell type.""" | ||
|
|
||
| FLOW_CELL_10B: int = 151 | ||
| FLOW_CELL_25B: int = 151 | ||
| FLOW_CELL_15B: int = 51 | ||
|
|
||
|
|
||
| class DefaultIndexLength(IntEnum): | ||
| """Default index length most usually used by each flow cell type.""" | ||
|
|
||
| FLOW_CELL_10B: int = 10 | ||
| FLOW_CELL_25B: int = 10 | ||
| FLOW_CELL_15B: int = 8 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| import logging | ||
| import sys | ||
| import click | ||
|
|
||
| from typing import List, Optional | ||
| from genologics.lims import Artifact, Process | ||
| from cg_lims.exceptions import LimsError, MissingUDFsError | ||
| from cg_lims.get.artifacts import get_artifacts | ||
|
|
||
| from cg_lims.EPPs.udf.calculate.constants import FlowCellTypes | ||
| from cg_lims.EPPs.udf.set.constants import DefaultReadLength, DefaultIndexLength | ||
|
|
||
| LOG = logging.getLogger(__name__) | ||
|
|
||
| DEFAULT_INDEX_LENGTHS = { | ||
| FlowCellTypes.FLOW_CELL_10B: DefaultIndexLength.FLOW_CELL_10B, | ||
| FlowCellTypes.FLOW_CELL_15B: DefaultIndexLength.FLOW_CELL_15B, | ||
| FlowCellTypes.FLOW_CELL_25B: DefaultIndexLength.FLOW_CELL_25B, | ||
| } | ||
|
|
||
| DEFAULT_READ_LENGTHS = { | ||
| FlowCellTypes.FLOW_CELL_10B: DefaultReadLength.FLOW_CELL_10B, | ||
| FlowCellTypes.FLOW_CELL_15B: DefaultReadLength.FLOW_CELL_15B, | ||
| FlowCellTypes.FLOW_CELL_25B: DefaultReadLength.FLOW_CELL_25B, | ||
| } | ||
|
|
||
|
|
||
| def get_library_tube_strip(process: Process) -> str: | ||
| """Return the Library Tube Strip ID from a process.""" | ||
| library_tube_strip: str = process.udf.get("Library Tube Strip ID") | ||
| if not library_tube_strip: | ||
| LOG.error(f"Process {process.id} is missing UDF 'Library Tube Strip ID'") | ||
| raise MissingUDFsError(f"UDF 'Library Tube Strip ID' missing from previous step.") | ||
| return library_tube_strip | ||
|
|
||
|
|
||
| def get_flow_cell_type(process: Process) -> str: | ||
| """Return the Flow Cell Type from a process.""" | ||
| flow_cell_type: str = process.udf.get("Flow Cell Type") | ||
| if not flow_cell_type: | ||
| LOG.error(f"Process {process.id} is missing UDF 'Flow Cell Type'") | ||
| raise MissingUDFsError(f"UDF 'Flow Cell Type' missing from previous step.") | ||
| return flow_cell_type | ||
|
|
||
|
|
||
| def get_flow_cell_name(process: Process) -> str: | ||
| """Return the flow cell name of the step.""" | ||
| containers = process.output_containers() | ||
| return containers[0].name | ||
|
|
||
|
|
||
| def set_process_udfs(process: Process, parent_process: Process) -> None: | ||
| """Set Prepare for Sequencing (NovaSeq X) process UDFs.""" | ||
| library_tube_strip: str = get_library_tube_strip(process=parent_process) | ||
| flow_cell_type: str = get_flow_cell_type(process=parent_process) | ||
| flow_cell_name: str = get_flow_cell_name(process=process) | ||
| read_length: int = DEFAULT_READ_LENGTHS[flow_cell_type].value | ||
| index_length: int = DEFAULT_INDEX_LENGTHS[flow_cell_type].value | ||
| process.udf["Library Tube Strip ID"] = library_tube_strip | ||
| process.udf["Run Mode"] = flow_cell_type | ||
| process.udf["Read 1 Cycles"] = read_length | ||
| process.udf["Read 2 Cycles"] = read_length | ||
| process.udf["Index Read 1"] = index_length | ||
| process.udf["Index Read 2"] = index_length | ||
| process.udf["BaseSpace Run Name"] = flow_cell_name | ||
| process.put() | ||
|
|
||
|
|
||
| def get_parent_process(process: Process) -> Optional[Process]: | ||
| """Get the parent process of another process, assuming all input artifacts come from the same step.""" | ||
| input_artifacts: List[Artifact] = get_artifacts(process=process, input=True) | ||
| if not input_artifacts: | ||
| LOG.info(f"No input artifacts found for process {process.id}.") | ||
| return None | ||
| return input_artifacts[0].parent_process | ||
|
|
||
|
|
||
| @click.command() | ||
| @click.pass_context | ||
| def set_sequencing_settings(ctx): | ||
| """Sets the settings required for sequencing of NovaSeq X flow cells.""" | ||
|
|
||
| LOG.info(f"Running {ctx.command_path} with params: {ctx.params}") | ||
|
|
||
| process: Process = ctx.obj["process"] | ||
|
|
||
| try: | ||
| parent_process: Process = get_parent_process(process=process) | ||
| set_process_udfs(process=process, parent_process=parent_process) | ||
| message: str = "Sequencing settings have been successfully set." | ||
| LOG.info(message) | ||
| click.echo(message) | ||
| except LimsError as e: | ||
| LOG.error(e.message) | ||
| sys.exit(e.message) |