diff --git a/packages/opencloning/src/opencloning/batch_cloning/index.html b/packages/opencloning/src/opencloning/batch_cloning/index.html index 1a5d1384..3ae2b589 100644 --- a/packages/opencloning/src/opencloning/batch_cloning/index.html +++ b/packages/opencloning/src/opencloning/batch_cloning/index.html @@ -41,8 +41,8 @@

Batch Cloning Methods

- S. pombe Gene Cloning -

Batch cloning of S. pombe genes using homologous recombination.

+ Yeast PCR-based Cloning primer design +

Design PCR-based cloning strategies for yeast in batch (Bähler et al. 1998).

CRISPR Guide RNA Cloning diff --git a/packages/opencloning/src/opencloning/batch_cloning/pombe/__init__.py b/packages/opencloning/src/opencloning/batch_cloning/pombe/__init__.py index f1779344..7f7d4006 100644 --- a/packages/opencloning/src/opencloning/batch_cloning/pombe/__init__.py +++ b/packages/opencloning/src/opencloning/batch_cloning/pombe/__init__.py @@ -1,79 +1,163 @@ from fastapi import Form, File, UploadFile, HTTPException from typing import Annotated, Literal -from tempfile import TemporaryDirectory import os -from fastapi.responses import FileResponse -from .pombe_get_primers import main as pombe_primers +import tempfile +from fastapi.responses import FileResponse, HTMLResponse from .pombe_clone import main as pombe_clone from .pombe_summary import main as pombe_summary from .pombe_gather import main as pombe_gather +from .pombe_primer_table import build_primer_summary_df, primer_summary_to_html import shutil -import traceback from ...get_router import get_router from fastapi import Request +from opencloning.dna_functions import get_sequence_from_euroscarf_url, request_from_addgene, request_from_snapgene +from pydna.primer import Primer +from pydna.opencloning_models import UploadedFileSource +from pydna.parsers import parse as pydna_parse +from pydna.parsers import parse_snapgene router = get_router() -@router.get('/batch_cloning/pombe') +@router.get('/batch_cloning/yeast_primer_design') async def get_batch_cloning_page(request: Request): return FileResponse(os.path.join(os.path.dirname(__file__), 'index.html')) -@router.post('/batch_cloning/pombe') +DEFAULT_PLASMID_OPTIONS = { + 'gene_deletion': { + 'kanmx6': ('addgene', '39296', None), + 'natmx6': ('snapgene', 'yeast_plasmids', 'pFA6a-natMX6'), + 'hphmx6': ('snapgene', 'yeast_plasmids', 'pFA6a-hphMX6'), + }, + 'gene_cterm_tagging': { + 'kanmx6': ('addgene', '87023', None), + 'natmx6': ('addgene', '52693', None), + 'hphmx6': ('addgene', '105156', None), + }, + 'promoter_not_tag': { + 'kanmx6': ('addgene', '39280', None), # pFA6a-kanMX6-P3nmt1 + 'natmx6': ('euroscarf', 'P30425', None), # pFA6a-natMX6-P3nmt1 + 'hphmx6': ('addgene', '105162', None), # pFA6a-hphMX6-3nmt1 + }, +} + + +def raise_plasmid_import_error(exception: Exception, mode) -> None: + raise HTTPException(status_code=503, detail=f'Failed to import plasmid from {mode}: {exception}') from exception + + +@router.post('/batch_cloning/yeast_primer_design') async def post_batch_cloning( + cloning_type: Annotated[ + Literal['gene_deletion', 'gene_cterm_tagging', 'promoter_not_tag', 'promoter_with_tag'], Form(...) + ], + desired_output: Annotated[Literal['simulate_cloning', 'primers_only'], Form(...)], + assembly_accession: str = Form(..., pattern=r'^GC[AF]_[0-9.]+$', min_length=1), gene_list: str = Form(...), + integration_binding_forward: str = Form(..., pattern=r'^[ACGTacgt]+$', min_length=1), + integration_binding_reverse: str = Form(..., pattern=r'^[ACGTacgt]+$', min_length=1), plasmid_file: UploadFile | None = File(None), addgene_id: str | None = Form(None), - plasmid_option: Annotated[Literal['addgene', 'file'], Form(...)] = None, - checking_primer_forward: str = Form(..., pattern=r'^[ACGTacgt]+$', min_length=1), - checking_primer_reverse: str = Form(..., pattern=r'^[ACGTacgt]+$', min_length=1), + plasmid_option: Annotated[Literal['addgene', 'file', 'default'], Form(...)] = None, + checking_primer_forward: str = Form('', pattern=r'^[ACGTacgt]*$'), + checking_primer_reverse: str = Form('', pattern=r'^[ACGTacgt]*$'), + resistance_marker: Annotated[Literal['kanmx6', 'natmx6', 'hphmx6', 'other'], Form(...)] = None, ): - - plasmid = plasmid_file if plasmid_option == 'file' else addgene_id - if plasmid is None: - raise HTTPException(status_code=400, detail='No plasmid provided') - genes = [gene.strip() for gene in gene_list.split() if gene.strip()] if not genes: raise HTTPException(status_code=400, detail='No valid genes provided') - with TemporaryDirectory() as temp_dir: - if plasmid_option == 'file': - # Write the plasmid to the temp dir - with open(os.path.join(temp_dir, plasmid_file.filename), 'wb') as f: - shutil.copyfileobj(plasmid_file.file, f) - - # Write the checking primers - with open(os.path.join(temp_dir, 'checking_primers.fa'), 'w') as f: - f.write(f'>common_insert_fwd\n{checking_primer_forward}\n>common_insert_rvs\n{checking_primer_reverse}') - - for gene in genes: - try: - await pombe_primers(gene, temp_dir) - except Exception: - raise HTTPException(status_code=404, detail=f'Primers for {gene} not found') - try: - if plasmid_option == 'file': - with open(os.path.join(temp_dir, plasmid_file.filename), 'rb') as f: - await pombe_clone( - gene, 'GCF_000002945.2', temp_dir, UploadFile(file=f, filename=plasmid_file.filename) - ) - else: - await pombe_clone(gene, 'GCF_000002945.2', temp_dir, addgene_id) - except Exception: - # Show the stack trace in console - print(f'Error occurred while cloning {gene}:') - traceback.print_exc() - raise HTTPException(status_code=400, detail=f'Clone for {gene} failed') + if desired_output == 'primers_only': + gene_primers = [] + try: + for gene in genes: + primers = await pombe_clone( + gene, + assembly_accession, + integration_binding_forward, + integration_binding_reverse, + cloning_type, + primers_only=True, + ) + gene_primers.append((gene, cloning_type, primers)) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + primer_df = build_primer_summary_df(gene_primers) + return HTMLResponse(content=primer_summary_to_html(primer_df)) + + common_primer_forward = ( + Primer(checking_primer_forward, name='common_insert_fwd') if checking_primer_forward else None + ) + common_primer_reverse = ( + Primer(checking_primer_reverse, name='common_insert_rvs') if checking_primer_reverse else None + ) + + if plasmid_option == 'default': + try: + mode, first, second = DEFAULT_PLASMID_OPTIONS[cloning_type][resistance_marker] + if mode == 'addgene': + plasmid = await request_from_addgene(first) + elif mode == 'euroscarf': + plasmid = await get_sequence_from_euroscarf_url(first) + else: + plasmid = await request_from_snapgene(first, second) + except KeyError: + raise HTTPException( + status_code=400, detail=f'Resistance marker {resistance_marker} is not supported for default plasmid' + ) + except Exception as e: + raise_plasmid_import_error(e, mode) + + elif plasmid_option == 'file': + try: + assert plasmid_file is not None + assert plasmid_file.filename is not None + file_content = await plasmid_file.read() + if plasmid_file.filename.endswith('.dna'): + plasmid = parse_snapgene(file_content)[0] + else: + plasmid = pydna_parse(file_content.decode('utf-8'))[0] + plasmid.source = UploadedFileSource( + file_name=plasmid_file.filename, + sequence_file_format=plasmid.annotations['pydna_parse_sequence_file_format'], + index_in_file=0, + ) + except Exception as e: + raise HTTPException(status_code=400, detail=f'Plasmid loading failed: {e}') + + if plasmid_option == 'addgene': + assert addgene_id is not None + try: + plasmid = await request_from_addgene(addgene_id) + except Exception as e: + raise_plasmid_import_error(e, 'addgene') + + with tempfile.TemporaryDirectory() as temp_dir: + try: + for gene in genes: + await pombe_clone( + gene, + assembly_accession, + integration_binding_forward, + integration_binding_reverse, + cloning_type, + output_dir=temp_dir, + plasmid=plasmid, + common_primer_forward=common_primer_forward, + common_primer_reverse=common_primer_reverse, + ) + except ValueError as e: + raise HTTPException(status_code=400, detail=f'Cloning failed: {e}') + try: pombe_summary(temp_dir) pombe_gather(temp_dir) except Exception as e: raise HTTPException(status_code=400, detail=f'Summary failed: {e}') - # zip the temp dir and return it zip_filename = f'{temp_dir}_archive' shutil.make_archive(zip_filename, 'zip', temp_dir) zip_file = f'{zip_filename}.zip' diff --git a/packages/opencloning/src/opencloning/batch_cloning/pombe/index.html b/packages/opencloning/src/opencloning/batch_cloning/pombe/index.html index a3373e9d..a8a93f5a 100644 --- a/packages/opencloning/src/opencloning/batch_cloning/pombe/index.html +++ b/packages/opencloning/src/opencloning/batch_cloning/pombe/index.html @@ -8,25 +8,120 @@ -

Batch Cloning Gene List

-

-

Design gene-deletion clones for Schizosaccharomyces pombe in batch.

-

-

How to use

- - -

What it returns

-

After a delay (can be long), it returns a ZIP archive with:

- - - -

Best way to get started

- -

- Just try with the below example, it will return a zip file and you can drag-and-drop the json file - into OpenCloning to visualize the - strategy. - Check also the summary.xlsx file to see the primers. -

- -
-
-

- S. pombe systematic IDs only (e.g. SPAPB1A10.09, SPBC15D4.01c). One gene ID per line; empty lines are - ignored. +

+

Yeast cloning batch primer design

+

+ Design PCR-based cloning strategies for yeast in batch.

-
-

-
-

- The plasmid supplies the marker cassette (e.g. KanMX6, NatMX6) to be PCR-amplified and integrated into each - gene locus via homologous recombination. +

+ Powered by OpenCloning. + To learn more about all OpenCloning features, see docs.opencloning.org. + If you have feedback, reach out to + manuel.lera-ramirez@ucl.ac.uk.

-
- - - -
- -
- - -

- - - -
-
-

- Numeric Addgene ID of the plasmid (e.g. 19343 for pFA6a-KanMX6). Find IDs at - addgene.org. -

-

-
- -
-
-

- Primers used for PCR verification of marker cassette integration. Choose NatMX6 or KanMX6 for - standard cassettes, or "Other" to enter custom primers (e.g. for HygMX6 or another marker). -

- - -
- - -
- - -
-
-
-
-

- Aligns to the top strand of the modified allele and pairs with a right-side primer. DNA bases only - (A, C, G, T). -

-

- -
-

- Aligns to the bottom strand of the modified allele and pairs with a left-side primer. DNA bases only - (A, C, G, T). -

-

-
- -

- Submit to run the pipeline. You will receive a ZIP file containing per-gene folders (primers, cloning - strategy, GenBank sequences) and summary tables (TSV, Excel, YAML). +

How to use

+

You can check out this video for a + quick overview.

+
    +
  • Select the genome assembly
    • +
  • Select the cloning type
    • +
  • Enter yeast systematic gene IDs (e.g. SPBC15D4.01c).
    • +
  • Select a marker plasmid.
    • +
  • Specify verification primers.
    • +
  • The pipeline designs primers for integration and checking.
    • +
  • You can decide if you just want the primers or if you want to simulate the cloning.
    • +
+ +

What it returns

+ +

Primers only

+

Shows you a table with the primers.

+

Simulate cloning

+

After a delay (can be long), it returns a ZIP archive with:

+ +
    +
  • Per-gene folders that contain: +
      +
    • Genbank files with the sequences involved
      • +
    • A json file containing the entire cloning strategy. You can drag-and-drop + it into the OpenCloning + website to visualize the strategy. + It will look similar to this. +
      • + +
    +
    • + +
  • Summary tables (TSV, Excel, YAML)
    • +
+ +

Best way to get started

+ +

+ Just try with the below example, it will return a zip file and you can drag-and-drop the json file + SPAPB1A10.09/cloning_strategy.json + into OpenCloning to visualize the + strategy. + Check also the file summary.xlsx file to see the primers, and the multiple genbank files with + the sequences involved, you can open those in SnapGene, Benchling, etc.

-
- - - - + } + + // --- init --- + function init() { + cacheDOM(); + syncIntegrationBinding(); + syncForm(); + DOM.form.addEventListener("change", onFormChange); + DOM.restoreIntegrationBinding.addEventListener("click", restoreIntegrationBinding); + DOM.form.addEventListener("submit", onFormSubmit); + restoreCloningSimulationDefaults(); + } + + document.addEventListener("DOMContentLoaded", init); + + +
+ +
+

Primer summary

+
+
diff --git a/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_clone.py b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_clone.py index cdf60835..9f1c5199 100644 --- a/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_clone.py +++ b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_clone.py @@ -1,109 +1,224 @@ +import json import os -from pydna.assembly2 import homologous_recombination_integration, pcr_assembly -from opencloning.dna_functions import request_from_addgene +from dataclasses import dataclass + +from Bio.SeqFeature import SeqFeature +from opencloning.dna_utils import trim_location from opencloning.ncbi_requests import get_annotations_from_query, get_genome_region_from_annotation -import asyncio -from Bio import SeqIO -from pydna.primer import Primer +from opencloning.primer_design import primer_in_region +from pydna.assembly2 import homologous_recombination_integration, pcr_assembly +from pydna.dseqrecord import Dseqrecord from pydna.opencloning_models import CloningStrategy -from fastapi.datastructures import UploadFile -from pydna.parsers import parse as pydna_parse -from opencloning.dna_functions import custom_file_parser -import io +from pydna.primer import Primer +from pydna.utils import location_boundaries + +from .pombe_naming import CloningType, allele_name, integration_primer_name + + +@dataclass +class GeneLocus: + gene: str + locus: Dseqrecord + feature: SeqFeature + cloning_type: CloningType + + +def get_homology_arms(locus: Dseqrecord, feature: SeqFeature, cloning_type: str) -> tuple[str, str]: + if cloning_type == 'gene_deletion': + start, end = (int(i) for i in location_boundaries(feature.location)) + left_homology_arm = str(locus.seq[start - 80 : start]) + right_homology_arm = str(locus.seq[end : end + 80].reverse_complement()) + elif cloning_type == 'gene_cterm_tagging': + feature.location = trim_location(feature.location, 3, from_end=True) + start, end = (int(i) for i in location_boundaries(feature.location)) + left_homology_arm = str(locus.seq[end - 80 : end]) + right_homology_arm = str(locus.seq[end : end + 80].reverse_complement()) + elif cloning_type in ('promoter_not_tag', 'promoter_with_tag'): + start, end = (int(i) for i in location_boundaries(feature.location)) + left_homology_arm = str(locus.seq[start - 80 : start]) + right_homology_arm = str(locus.seq[start : start + 80].reverse_complement()) + else: + raise ValueError(f'Unsupported cloning type: {cloning_type}') + + return left_homology_arm.lower(), right_homology_arm.lower() + + +def get_checking_primers( + locus: Dseqrecord, feature: SeqFeature, gene: str, cloning_type: str, region_length: int = 200 +) -> tuple[Primer, Primer]: + start, end = (int(i) for i in location_boundaries(feature.location)) + + def get_upstream_primer(): + padding = 200 + primer = primer_in_region(locus[(start - padding - region_length) : (start - padding)], forward=True) + primer.name = f'{gene}_check_upstream_fwd' + return primer + + def get_downstream_primer(): + padding = 200 + primer = primer_in_region(locus[(end + padding) : (end + padding + region_length)], forward=False) + primer.name = f'{gene}_check_downstream_rvs' + return primer + + def get_inside_forward_primer(): + padding = 200 + primer = primer_in_region(locus[(end - padding - region_length) : (end - padding)], forward=True) + primer.name = f'{gene}_check_inside_fwd' + return primer + + def get_inside_reverse_primer(): + padding = 200 + primer = primer_in_region(locus[(start + padding) : (start + padding + region_length)], forward=False) + primer.name = f'{gene}_check_inside_rvs' + return primer + + if cloning_type == 'gene_deletion': + fwd = get_upstream_primer() + rvs = get_downstream_primer() + elif cloning_type == 'gene_cterm_tagging': + fwd = get_inside_forward_primer() + rvs = get_downstream_primer() + elif cloning_type in ('promoter_not_tag', 'promoter_with_tag'): + fwd = get_upstream_primer() + rvs = get_inside_reverse_primer() + else: + raise ValueError(f'Unsupported cloning type: {cloning_type}') + return fwd, rvs -async def main( - gene: str, - assembly_accession: str, - output_dir: str, - plasmid_input: UploadFile | str = '19343', - padding: int = 1000, -): - print(f"\033[92mCloning {gene}\033[0m") - # Parse primers ================================================================================= - primers = [Primer(p) for p in SeqIO.parse(os.path.join(output_dir, gene, 'primers.fa'), 'fasta')] - common_primers = [Primer(p) for p in SeqIO.parse(os.path.join(output_dir, 'checking_primers.fa'), 'fasta')] - - # Get plasmid sequence ================================================================================= - if isinstance(plasmid_input, UploadFile): - file_content = await plasmid_input.read() - if plasmid_input.filename.endswith('.dna'): - file_streamer = io.BytesIO(file_content) - plasmid = custom_file_parser(file_streamer, 'snapgene')[0] - else: - plasmid = pydna_parse(file_content)[0] - - else: - plasmid = await request_from_addgene(plasmid_input) - # Get genome region ===================================================================== +async def resolve_gene_locus(gene: str, assembly_accession: str, cloning_type: CloningType) -> GeneLocus: annotations = await get_annotations_from_query(gene, assembly_accession) if len(annotations) == 0: raise ValueError(f'No annotations found for {gene}') annotations = [a for a in annotations if gene.upper() in a['locus_tag'].upper()] - if len(annotations) != 1: + if len(annotations) == 0: raise ValueError(f'No right annotation found for {gene}') + if len(annotations) > 1: + raise ValueError(f'Multiple annotations found for {gene}') + + annotation = annotations[0] + if annotation['gene_type'] != 'protein-coding': + raise ValueError(f'{gene} is not a protein-coding gene') + if 'gene_id' not in annotation or not annotation['gene_id']: + raise ValueError(f'{gene} has no gene_id') + + locus = await get_genome_region_from_annotation(annotation, 1000, 1000) + feature = next( + f + for f in locus.features + if (f.type == 'CDS') and (f"GeneID:{annotation['gene_id']}" in f.qualifiers['db_xref']) + ) + return GeneLocus(gene=gene, locus=locus, feature=feature, cloning_type=cloning_type) - locus = await get_genome_region_from_annotation(annotations[0], 1000, 1000) - - # PCR ================================================================================================ - pcr_products = pcr_assembly(plasmid, primers[0], primers[1], limit=14, mismatches=0) - pcr_products[0].name = 'amplified_marker' - alleles = homologous_recombination_integration(locus, [pcr_products[0]], 40) - pcr_check1 = pcr_assembly(alleles[0], primers[2], common_primers[1], limit=14, mismatches=0)[0] - pcr_check1.name = 'check_pcr_left' - pcr_check2 = pcr_assembly(alleles[0], primers[3], common_primers[0], limit=14, mismatches=0)[0] - pcr_check2.name = 'check_pcr_right' - alleles[0].name = 'deletion_allele' - - cs = CloningStrategy.from_dseqrecords([pcr_check1, pcr_check2]) - - if not os.path.exists(os.path.join(output_dir, gene)): - os.makedirs(os.path.join(output_dir, gene)) - - with open(os.path.join(output_dir, gene, 'cloning_strategy.json'), 'w') as f: - f.write(cs.model_dump_json(indent=2)) - - -if __name__ == '__main__': - import argparse - parser = argparse.ArgumentParser(description='List of genes to delete from S. pombe') - parser.add_argument( - '--genes', type=str, required=True, help='Path to a file containing a list of genes, one per line' +def design_gene_primers( + locus_ctx: GeneLocus, + integration_binding_forward: str, + integration_binding_reverse: str, +) -> tuple[Primer, Primer, Primer, Primer]: + left_homology_arm, right_homology_arm = get_homology_arms( + locus_ctx.locus, locus_ctx.feature, locus_ctx.cloning_type ) - args = parser.parse_args() - parser.add_argument( - '--assembly_accession', - type=str, - default='GCF_000002945.2', - help='Assembly accession for S. pombe genome (default: GCF_000002945.2)', + left_primer = Primer( + left_homology_arm + integration_binding_forward.upper(), + name=integration_primer_name(locus_ctx.gene, 'fwd', locus_ctx.cloning_type), ) - - parser.add_argument( - '--output_dir', - type=str, - default='batch_cloning_output', - help='Directory to save the output files (default: batch_cloning_output)', + right_primer = Primer( + right_homology_arm + integration_binding_reverse.upper(), + name=integration_primer_name(locus_ctx.gene, 'rvs', locus_ctx.cloning_type), ) - - parser.add_argument( - '--plasmid', - type=str, - default='19343', - help='Addgene ID for the plasmid (default: 19343)', + left_check_primer, right_check_primer = get_checking_primers( + locus_ctx.locus, locus_ctx.feature, locus_ctx.gene, locus_ctx.cloning_type ) + return left_primer, right_primer, left_check_primer, right_check_primer - args = parser.parse_args() - assembly_accession = args.assembly_accession - with open(args.genes, 'r') as f: - genes = [line.strip() for line in f if line.strip()] +def simulate_and_write( + locus_ctx: GeneLocus, + primers: tuple[Primer, Primer, Primer, Primer], + plasmid: Dseqrecord, + common_primer_forward: Primer | None, + common_primer_reverse: Primer | None, + output_dir: str, +) -> None: + left_primer, right_primer, left_check_primer, right_check_primer = primers + + pcr_products = pcr_assembly(plasmid, left_primer, right_primer, limit=14, mismatches=0) + if len(pcr_products) == 0: + raise ValueError('No PCR products when amplifying from the plasmid') + if len(pcr_products) > 1: + raise ValueError('Multiple PCR products when amplifying from the plasmid') + pcr_products[0].name = 'amplified_marker' + alleles = homologous_recombination_integration(locus_ctx.locus, [pcr_products[0]], 80) + if len(alleles) == 0: + raise ValueError(f'No insertions possible for {locus_ctx.gene}') + if len(alleles) > 1: + raise ValueError(f'Multiple insertions possible for {locus_ctx.gene}') + modified_allele_name = allele_name(locus_ctx.gene, locus_ctx.cloning_type) + alleles[0].name = modified_allele_name + terminals = [alleles[0]] + if common_primer_reverse is not None: + pcr_check_left_products = pcr_assembly( + alleles[0], left_check_primer, common_primer_reverse, limit=14, mismatches=0 + ) + if len(pcr_check_left_products) == 0: + raise ValueError(f'No PCR products with the left check primer for {locus_ctx.gene}') + pcr_check_left = pcr_check_left_products[0] + pcr_check_left.name = 'check_pcr_left' + terminals.append(pcr_check_left) + + if common_primer_forward is not None: + pcr_check_right_products = pcr_assembly( + alleles[0], common_primer_forward, right_check_primer, limit=14, mismatches=0 + ) + if len(pcr_check_right_products) == 0: + raise ValueError(f'No PCR products with the right check primer for {locus_ctx.gene}') + pcr_check_right = pcr_check_right_products[0] + pcr_check_right.name = 'check_pcr_right' + terminals.append(pcr_check_right) + + cs = CloningStrategy.from_dseqrecords(terminals) + if not os.path.exists(os.path.join(output_dir, locus_ctx.gene)): + os.makedirs(os.path.join(output_dir, locus_ctx.gene)) + + gene_dir = os.path.join(output_dir, locus_ctx.gene) + with open(os.path.join(gene_dir, 'cloning_strategy.json'), 'w') as f: + f.write(cs.model_dump_json(indent=2)) + with open(os.path.join(gene_dir, 'metadata.json'), 'w') as f: + json.dump( + { + 'cloning_type': locus_ctx.cloning_type, + 'allele_name': modified_allele_name, + 'check_pcr_left': common_primer_reverse is not None, + 'check_pcr_right': common_primer_forward is not None, + }, + f, + ) - if not os.path.exists(args.output_dir): - os.makedirs(args.output_dir) - for gene in genes: - asyncio.run(main(gene, assembly_accession, args.output_dir, args.plasmid)) +async def main( + gene: str, + assembly_accession: str, + integration_binding_forward: str, + integration_binding_reverse: str, + cloning_type: CloningType, + *, + output_dir: str | None = None, + plasmid: Dseqrecord | None = None, + common_primer_forward: Primer | None = None, + common_primer_reverse: Primer | None = None, + primers_only: bool = False, +) -> tuple[Primer, Primer, Primer, Primer] | None: + print(f"\033[92mCloning {gene}\033[0m") + locus_ctx = await resolve_gene_locus(gene, assembly_accession, cloning_type) + primers = design_gene_primers(locus_ctx, integration_binding_forward, integration_binding_reverse) + if primers_only: + return primers + + assert output_dir is not None + assert plasmid is not None + simulate_and_write(locus_ctx, primers, plasmid, common_primer_forward, common_primer_reverse, output_dir) + return None diff --git a/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_gather.py b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_gather.py index 16ebf3d8..bc002e04 100644 --- a/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_gather.py +++ b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_gather.py @@ -32,18 +32,20 @@ def main(input_dir: str): # Primer summary table: primer_summary = list() - primer_names = ['primer_fwd_check', 'primer_fwd', 'primer_rvs', 'primer_rvs_check'] + primer_names = ['primer_fwd', 'primer_rvs', 'primer_fwd_check', 'primer_rvs_check'] for _, row in df.iterrows(): - gene = row['gene'] for primer_name in primer_names: + if primer_name not in row or row[primer_name] is None: + continue primer_row = OrderedDict() - primer_row['name'] = primer_name.replace('primer_', gene + '_') + primer_row['name'] = row[primer_name + '_name'] primer_row['sequence'] = row[primer_name] primer_row['bound'] = row[primer_name + '_bound'] primer_row['tm'] = row[primer_name + '_tm'] primer_summary.append(primer_row) primer_df = pd.DataFrame(primer_summary) + primer_df['tm'] = primer_df['tm'].round(1) df.to_csv(os.path.join(input_dir, 'summary.tsv'), index=False, sep='\t') primer_df.to_csv(os.path.join(input_dir, 'primer_summary.tsv'), index=False, sep='\t') diff --git a/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_get_primers.py b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_get_primers.py deleted file mode 100644 index 51b2a7d8..00000000 --- a/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_get_primers.py +++ /dev/null @@ -1,114 +0,0 @@ -from bs4 import BeautifulSoup -import asyncio -import re -import os -import argparse -from Bio import SeqIO -from Bio.Seq import Seq -from Bio.SeqRecord import SeqRecord -from ...http_client import get_http_client, Response - -default_settings_primer_design = { - 'length': 80, - 'increment': 40, - 'add_seq': 400, - 'opt_len': 22, - 'min_len': 20, - 'max_len': 28, - 'opt_tm': 60.0, - 'min_tm': 57.0, - 'max_tm': 63.0, - 'min_gc': 30, - 'max_gc': 60, - '.submit': 'Submit', -} - - -async def async_post(url, headers, data, params=None) -> Response: - async with get_http_client() as client: - return await client.post(url, headers=headers, data=data, params=params, timeout=20.0) - - -async def get_primers(gene): - print(f"\033[92mGetting primers for {gene}\033[0m") - # A first request to access the primers - data = { - 'gene': gene, - 'length': 80, - 'plasmid': 'pFA6a', - 'increment': 40, - '.submit': 'Submit', - '.cgifields': 'plasmid', - } - - url = 'http://bahlerweb.cs.ucl.ac.uk/cgi-bin/PPPP/pppp_deletion.pl' - headers = None - resp = await async_post(url, headers, data) - # Parse with BeautifulSoup - soup = BeautifulSoup(resp.text, 'html.parser') - # select forward and reverse primers by default - forward_primer = soup.find('input', {'name': 'for_sel', 'checked': True}) - reverse_primer = soup.find('input', {'name': 'rev_sel', 'checked': True}) - forward_primer_seq = re.sub(r'[^a-zA-Z]', '', forward_primer['value']) - reverse_primer_seq = re.sub(r'[^a-zA-Z]', '', reverse_primer['value']) - # Make a second request to get the checking primers - data = { - 'gene': gene, - 'for_sel': forward_primer['value'], - 'rev_sel': reverse_primer['value'], - } - data.update(default_settings_primer_design) - - url = 'http://bahlerweb.cs.ucl.ac.uk/cgi-bin/PPPP/pppp_checking.pl' - resp = await async_post(url, headers, data) - # Parse with BeautifulSoup - soup = BeautifulSoup(resp.text, 'html.parser') - # Find a pre tag with the text "Left Primer:" - left_check_primer = soup.find('pre', string=re.compile(r'Left Primer:\s+Sequence:')).get_text().strip() - right_check_primer = soup.find('pre', string=re.compile(r'Right Primer:\s+Sequence:')).get_text().strip() - - pattern = r'Sequence:\s+(\S+)' - left_check_primer_seq = re.search(pattern, left_check_primer).group(1) - right_check_primer_seq = re.search(pattern, right_check_primer).group(1) - return forward_primer_seq, reverse_primer_seq, left_check_primer_seq, right_check_primer_seq - - -async def main(gene: str, output_dir: str): - gene_dir = os.path.join(output_dir, gene) - os.makedirs(gene_dir, exist_ok=True) - - forward_primer, reverse_primer, forward_check, reverse_check = await get_primers(gene) - - primers = [ - SeqRecord(Seq(forward_primer), id=f"{gene}_fwd", description=''), - SeqRecord(Seq(reverse_primer), id=f"{gene}_rvs", description=''), - SeqRecord(Seq(forward_check), id=f"{gene}_fwd_check", description=''), - SeqRecord(Seq(reverse_check), id=f"{gene}_rvs_check", description=''), - ] - - with open(os.path.join(gene_dir, 'primers.fa'), 'w') as f: - SeqIO.write(primers, f, 'fasta') - - -if __name__ == '__main__': - parser = argparse.ArgumentParser(description='Get primers for S. pombe genes') - parser.add_argument( - '--genes', type=str, required=True, help='Path to a file containing a list of genes, one per line' - ) - parser.add_argument( - '--output_dir', - type=str, - default='batch_cloning_output', - help='Directory to save the output files (default: batch_cloning_output)', - ) - - args = parser.parse_args() - - with open(args.genes, 'r') as f: - genes = [line.strip() for line in f if line.strip()] - - if not os.path.exists(args.output_dir): - os.makedirs(args.output_dir) - - for gene in genes: - asyncio.run(main(gene, args.output_dir)) diff --git a/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_naming.py b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_naming.py new file mode 100644 index 00000000..82ebde5b --- /dev/null +++ b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_naming.py @@ -0,0 +1,29 @@ +from typing import Literal + +CloningType = Literal['gene_deletion', 'gene_cterm_tagging', 'promoter_not_tag', 'promoter_with_tag'] + +INTEGRATION_PRIMER_LABELS = { + 'gene_deletion': 'deletion', + 'gene_cterm_tagging': 'tag', + 'promoter_not_tag': 'promoter', + 'promoter_with_tag': 'promoter_tag', +} + +ALLELE_NAMES = { + 'gene_deletion': lambda gene: f'{gene}Δ', + 'gene_cterm_tagging': lambda gene: f'{gene}-tagged', + 'promoter_not_tag': lambda gene: f'{gene}-promoter', + 'promoter_with_tag': lambda gene: f'{gene}-promoter-tagged', +} + + +def integration_primer_label(cloning_type: CloningType) -> str: + return INTEGRATION_PRIMER_LABELS[cloning_type] + + +def allele_name(gene: str, cloning_type: CloningType) -> str: + return ALLELE_NAMES[cloning_type](gene) + + +def integration_primer_name(gene: str, direction: Literal['fwd', 'rvs'], cloning_type: CloningType) -> str: + return f'{gene}_{integration_primer_label(cloning_type)}_{direction}' diff --git a/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_primer_table.py b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_primer_table.py new file mode 100644 index 00000000..bae1ceb2 --- /dev/null +++ b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_primer_table.py @@ -0,0 +1,54 @@ +from collections import OrderedDict + +import pandas as pd +from pydna import tm +from pydna.primer import Primer + +from .pombe_naming import CloningType + +PRIMER_KEYS = ['primer_fwd', 'primer_rvs', 'primer_fwd_check', 'primer_rvs_check'] + +PRIMERS_BY_KEY = { + 'primer_fwd': 0, + 'primer_rvs': 1, + 'primer_fwd_check': 2, + 'primer_rvs_check': 3, +} + + +def _primer_tm_region(primer: Primer) -> str: + seq = str(primer.seq) + uppercase = ''.join(c for c in seq if c.isupper()) + return uppercase if uppercase else seq.upper() + + +def build_primer_summary_df( + gene_results: list[tuple[str, CloningType, tuple[Primer, Primer, Primer, Primer]]], +) -> pd.DataFrame: + primer_summary = [] + for _, _, primers in gene_results: + for primer_key in PRIMER_KEYS: + primer = primers[PRIMERS_BY_KEY[primer_key]] + tm_region = _primer_tm_region(primer) + primer_summary.append( + OrderedDict( + { + 'name': primer.name, + 'sequence': str(primer.seq), + 'tm': tm.tm_default(tm_region), + } + ) + ) + + primer_df = pd.DataFrame(primer_summary) + primer_df['tm'] = primer_df['tm'].round(1) + return primer_df + + +def primer_summary_to_html(primer_df: pd.DataFrame) -> str: + return primer_df.to_html( + index=False, + border=0, + classes='primer-summary-table', + float_format='%.1f', + ) diff --git a/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_summary.py b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_summary.py index 8e5c9d60..b5a4b25f 100644 --- a/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_summary.py +++ b/packages/opencloning/src/opencloning/batch_cloning/pombe/pombe_summary.py @@ -1,10 +1,6 @@ from pydna.utils import location_boundaries from ...pydantic_models import BaseCloningStrategy -from opencloning_linkml.datamodel import ( - Primer as PrimerModel, - PCRSource, - HomologousRecombinationSource, -) +from opencloning_linkml.datamodel import Primer as PrimerModel, PCRSource from pydna.parsers import parse as pydna_parse import os import json @@ -13,87 +9,115 @@ import argparse from Bio.SeqFeature import Location -chromosomes = { - 'NC_003424.3': 'I', - 'NC_003423.3': 'II', - 'NC_003421.2': 'III', - 'NC_088682.1': 'MT', +SEQUENCE_LENGTH_KEYS = { + 'amplified_marker': 'amplified_marker_length', + 'check_pcr_left': 'check_pcr_left_length', + 'check_pcr_right': 'check_pcr_right_length', } -def find_primer_aligned_sequence(pcr_sources: list[PCRSource], primer: PrimerModel) -> str: - for source in pcr_sources: - if source.input[0].sequence == primer.id: - loc = Location.fromstring(source.input[0].right_location) +def _sequence_names_by_id(strategy: BaseCloningStrategy) -> dict[int, str]: + names = {source.id: source.output_name for source in strategy.sources if source.output_name is not None} + for sequence in strategy.sequences: + if sequence.id not in names: + names[sequence.id] = pydna_parse(sequence.file_content)[0].name + return names + + +def _primer_binding(source: PCRSource, primer: PrimerModel) -> str: + for fragment in source.input: + if fragment.sequence != primer.id: + continue + if fragment.right_location is not None: + loc = Location.fromstring(fragment.right_location) return loc.extract(primer.sequence) - if source.input[-1].sequence == primer.id: - loc = Location.fromstring(source.input[-1].left_location) - return loc.extract(reverse_complement(primer.sequence)) - raise ValueError(f"Primer {primer.id} not found in any PCR source") + loc = Location.fromstring(fragment.left_location) + return loc.extract(reverse_complement(primer.sequence)) + raise ValueError(f'Primer {primer.id} not found in PCR source {source.id}') + + +def _primer_summary_entry(key: str, primer: PrimerModel, source: PCRSource) -> dict: + bound = _primer_binding(source, primer) + if 'rvs' in key: + bound = reverse_complement(bound) + return { + key: primer.sequence, + f'{key}_name': primer.name, + f'{key}_bound': bound, + f'{key}_tm': tm.tm_default(bound), + } + + +def _add_pcr_primers( + primer_dict: dict, + source: PCRSource, + output_name: str, + primer_by_id: dict[int, PrimerModel], +) -> None: + primer_ids = [inp.sequence for inp in source.input if inp.sequence in primer_by_id] + if output_name == 'amplified_marker': + if len(primer_ids) < 2: + return + primer_dict.update(_primer_summary_entry('primer_fwd', primer_by_id[primer_ids[0]], source)) + primer_dict.update(_primer_summary_entry('primer_rvs', primer_by_id[primer_ids[1]], source)) + elif output_name == 'check_pcr_left' and primer_ids: + primer_dict.update(_primer_summary_entry('primer_fwd_check', primer_by_id[primer_ids[0]], source)) + elif output_name == 'check_pcr_right' and primer_ids: + primer_dict.update(_primer_summary_entry('primer_rvs_check', primer_by_id[primer_ids[-1]], source)) + + +def extract_primers_from_strategy(strategy: BaseCloningStrategy) -> dict: + primer_by_id = {primer.id: primer for primer in strategy.primers} + seq_id_to_name = _sequence_names_by_id(strategy) + primer_dict: dict = {} + + for source in strategy.sources: + if source.type != 'PCRSource': + continue + output_name = seq_id_to_name.get(source.id) + if output_name is None: + continue + _add_pcr_primers(primer_dict, source, output_name, primer_by_id) + + return primer_dict def process_folder(working_dir: str): with open(os.path.join(working_dir, 'cloning_strategy.json'), 'r') as f: strategy = BaseCloningStrategy.model_validate(json.load(f)) - pcr_sources = [s for s in strategy.sources if s.type == 'PCRSource'] - # We do this to have action to .end and .start - pcr_sources = [PCRSource.model_validate(s.model_dump()) for s in pcr_sources] locus_source = next(s for s in strategy.sources if s.type == 'GenomeCoordinatesSource') locus_location = Location.fromstring(locus_source.coordinates) hrec_source = next(s for s in strategy.sources if s.type == 'HomologousRecombinationSource') - # We do this to have action to .end and .start - hrec_source: HomologousRecombinationSource = HomologousRecombinationSource.model_validate(hrec_source.model_dump()) - chromosome = chromosomes[locus_source.repository_id] - insertion_start = ( - locus_location.start + location_boundaries(Location.fromstring(hrec_source.input[0].right_location))[1] - ) - insertion_end = ( - locus_location.start + location_boundaries(Location.fromstring(hrec_source.input[-1].left_location))[0] - ) - - # Write out the sequences in genbank format and extract some relevant info - sequences = [pydna_parse(sequence.file_content)[0] for sequence in strategy.sequences] - for seq in sequences: - with open(os.path.join(working_dir, f"{seq.name}.gb"), 'w') as f: - f.write(seq.format('genbank')) - - if seq.name == 'amplified_marker': - amplified_marker_length = len(seq.seq) - - if seq.name == 'check_pcr_left': - check_pcr_left_length = len(seq.seq) - - if seq.name == 'check_pcr_right': - check_pcr_right_length = len(seq.seq) - - primer_dict = dict() - - primer_names = ['primer_fwd_check', 'primer_fwd', 'primer_rvs', None, 'primer_rvs_check', None] - for i, primer in enumerate(strategy.primers): - if primer_names[i] is None: - continue - name = primer_names[i] - primer_dict[name] = primer.sequence - # Find what the alignment bit of the primer is - aligned_seq = find_primer_aligned_sequence(pcr_sources, primer) - if 'rvs' in name: - aligned_seq = reverse_complement(aligned_seq) - primer_dict[name + '_bound'] = aligned_seq - primer_dict[name + '_tm'] = tm.tm_default(aligned_seq) + metadata_path = os.path.join(working_dir, 'metadata.json') + metadata = {} + if os.path.exists(metadata_path): + with open(metadata_path, 'r') as f: + metadata = json.load(f) summary = { 'gene': os.path.basename(working_dir), - 'chromosome': chromosome, - 'insertion_start': insertion_start, - 'insertion_end': insertion_end, - 'amplified_marker_length': amplified_marker_length, - 'check_pcr_left_length': check_pcr_left_length, - 'check_pcr_right_length': check_pcr_right_length, + 'cloning_type': metadata.get('cloning_type', 'gene_deletion'), + 'allele_name': metadata.get('allele_name'), + 'chromosome': locus_source.repository_id, + 'insertion_start': ( + locus_location.start + location_boundaries(Location.fromstring(hrec_source.input[0].right_location))[1] + ), + 'insertion_end': ( + locus_location.start + location_boundaries(Location.fromstring(hrec_source.input[-1].left_location))[0] + ), } - summary.update(primer_dict) + for sequence in strategy.sequences: + seq = pydna_parse(sequence.file_content)[0] + with open(os.path.join(working_dir, f'{seq.name}.gb'), 'w') as f: + f.write(seq.format('genbank')) + length_key = SEQUENCE_LENGTH_KEYS.get(seq.name) + if length_key is not None: + summary[length_key] = len(seq.seq) + + summary.update(extract_primers_from_strategy(strategy)) with open(os.path.join(working_dir, 'summary.json'), 'w') as f: json.dump(summary, f, indent=4) diff --git a/packages/opencloning/src/opencloning/dna_utils.py b/packages/opencloning/src/opencloning/dna_utils.py index 0a526cd0..4c6a9100 100644 --- a/packages/opencloning/src/opencloning/dna_utils.py +++ b/packages/opencloning/src/opencloning/dna_utils.py @@ -14,8 +14,8 @@ from pairwise_alignments_to_msa.alignment import aligned_tuples_to_MSA from copy import deepcopy import numpy as np -from Bio.SeqFeature import CompoundLocation from pydna.utils import location_boundaries, shift_location +from Bio.SeqFeature import SimpleLocation, CompoundLocation aligner = PairwiseAligner(scoring='blastn') @@ -196,3 +196,46 @@ def compound_location_to_capitalized_str(sequence: Dseqrecord, location: Compoun if location.strand == -1: result = reverse_complement(result) return result + + +def trim_location( + location: SimpleLocation | CompoundLocation, amount: int, from_end: bool = True +) -> SimpleLocation | CompoundLocation: + + if not isinstance(location, (SimpleLocation, CompoundLocation)): + raise ValueError('Location must be a SimpleLocation or CompoundLocation') + if amount > len(location): + raise ValueError('Amount is greater than the length of the location') + + if from_end: + iter = (part for part in location.parts[::-1]) + else: + iter = (part for part in location.parts) + left_to_cut = amount + new_parts = [] + for part in iter: + if left_to_cut == 0: + new_parts.append(part) + continue + if len(part) > left_to_cut: + new_parts.append(SimpleLocation(part.start, part.end - left_to_cut)) + left_to_cut = 0 + elif len(part) == left_to_cut: + left_to_cut = 0 + else: + left_to_cut -= len(part) + if from_end: + new_parts = new_parts[::-1] + + if len(new_parts) == 1: + return SimpleLocation(new_parts[0].start, new_parts[0].end) + + return CompoundLocation(new_parts) + + +# For an eventual test +# for loc_str in ['10..20', 'join(10..20, 30..40)', 'complement(join(10..20, 30..40))']: +# loc = Location.fromstring(loc_str) +# print(loc) +# print(trim_location(loc, 5)) +# print() diff --git a/packages/opencloning/src/opencloning/primer_design.py b/packages/opencloning/src/opencloning/primer_design.py index fe4feb99..bfddddfe 100644 --- a/packages/opencloning/src/opencloning/primer_design.py +++ b/packages/opencloning/src/opencloning/primer_design.py @@ -8,7 +8,7 @@ from Bio.Restriction.Restriction import RestrictionType from Bio.Data.IUPACData import ambiguous_dna_values as _ambiguous_dna_values from typing import Callable -from .primer3_functions import primer3_calc_tm, PrimerDesignSettings +from .primer3_functions import primer3_calc_tm, PrimerDesignSettings, primer3_design_primers ambiguous_dna_values = _ambiguous_dna_values.copy() # Remove acgt @@ -246,38 +246,74 @@ def simple_pair_primers( return (Primer(fwd_primer_seq, name=fwd_primer_name), Primer(rvs_primer_seq, name=rvs_primer_name)) -# def gateway_attB_primers( -# template: Dseqrecord, -# minimal_hybridization_length: int, -# target_tm: float, -# sites: tuple[str, str], -# spacers: tuple[str, str], -# filler_bases: str = 'GGGG', -# ) -> tuple[PrimerModel, PrimerModel]: -# if spacers is None: -# spacers = ['', ''] - -# if len(spacers) != 2: -# raise ValueError("The 'spacers' list must contain exactly two elements.") - -# if sites[0] not in primer_design_attB or sites[1] not in primer_design_attB: -# raise ValueError('Invalid attB site.') - -# amplicon = primer_design(template, limit=minimal_hybridization_length, target_tm=target_tm) -# fwd_primer, rvs_primer = amplicon.primers() - -# if fwd_primer is None or rvs_primer is None: -# raise ValueError('Primers could not be designed, try changing settings.') +def primer_to_amplify_fragment_of_given_size_knowing_other_primer( + template: Dseqrecord, + known_primer: Primer, + known_is_forward: bool, + fragment_size_range: list[int, int], +) -> Primer: + """ + Design primers to amplify a DNA fragment of a given size, knowing one primer. + """ + settings = PrimerDesignSettings() -# template_name = template.name if template.name != 'name' else f'seq_{template.id}' + if known_is_forward: + seq_args = { + 'SEQUENCE_PRIMER': str(known_primer.seq), + } + else: + seq_args = { + 'SEQUENCE_PRIMER_REVCOMP': str(known_primer.seq), + } + + result = primer3_design_primers( + str(template.seq).upper(), + seq_args=seq_args, + global_args={ + 'PRIMER_PRODUCT_SIZE_RANGE': [fragment_size_range], # ~500 bp band + 'PRIMER_OPT_SIZE': 20, + 'PRIMER_MIN_SIZE': 18, + 'PRIMER_MAX_SIZE': 30, + 'PRIMER_PAIR_MAX_DIFF_TM': 3, + 'PRIMER_PICK_ANYWAY': 1, + 'GC_CONTENT_MIN': 10, + 'GC_CONTENT_MAX': 90, + **settings.to_primer3_args(), + }, + ) + try: + if known_is_forward: + return Primer(result['PRIMER_RIGHT'][0]['SEQUENCE'], name=f'{template.name}_fwd') + else: + return Primer(result['PRIMER_LEFT'][0]['SEQUENCE'], name=f'{template.name}_rvs') + except IndexError: + print(result) + raise ValueError(f'Primers for checking could not be designed: {result}') -# left_site = primer_design_attB[sites[0]] -# right_site = primer_design_attB[sites[1]] -# fwd_primer_seq = filler_bases + left_site + spacers[0] + fwd_primer.seq -# rvs_primer_seq = filler_bases + right_site + reverse_complement(spacers[1]) + rvs_primer.seq +def primer_in_region(sequence: Dseqrecord, forward: bool) -> Primer: + """ + Design a forward or reverse primer constrained to lie within a subsequence of a template. + """ -# return ( -# PrimerModel(id=0, name=f'{template_name}_{sites[0]}_fwd', sequence=str(fwd_primer_seq)), -# PrimerModel(id=0, name=f'{template_name}_{sites[1]}_rvs', sequence=str(rvs_primer_seq)), -# ) + settings = PrimerDesignSettings() + + result = primer3_design_primers( + str(sequence.seq).upper(), + seq_args={}, + global_args={ + 'PRIMER_PICK_LEFT_PRIMER': int(forward), + 'PRIMER_PICK_RIGHT_PRIMER': int(not forward), + 'PRIMER_OPT_TM': 55, + 'PRIMER_MIN_TM': 52, + 'PRIMER_MAX_TM': 60, + **settings.to_primer3_args(), + }, + ) + try: + if forward: + return Primer(result['PRIMER_LEFT'][0]['SEQUENCE'], name=f'{sequence.name}_fwd') + else: + return Primer(result['PRIMER_RIGHT'][0]['SEQUENCE'], name=f'{sequence.name}_rvs') + except (IndexError, KeyError): + raise ValueError(f'Primer could not be designed in region: {result}') diff --git a/pombe_all.sh b/pombe_all.sh deleted file mode 100644 index e396ba45..00000000 --- a/pombe_all.sh +++ /dev/null @@ -1,9 +0,0 @@ -set -e -# Clear the output directory except for the gene_list.txt -find batch_cloning_output -type f ! -name 'gene_list.txt' -delete -find batch_cloning_output -type d -empty -delete - -python -m opencloning.batch_cloning.pombe.pombe_get_primers --genes batch_cloning_output/gene_list.txt -python -m opencloning.batch_cloning.pombe.pombe_clone --genes batch_cloning_output/gene_list.txt -python -m opencloning.batch_cloning.pombe.pombe_summary -python -m opencloning.batch_cloning.pombe.pombe_gather diff --git a/uv.lock b/uv.lock index 5d9f6fbc..bbba585d 100644 --- a/uv.lock +++ b/uv.lock @@ -1185,7 +1185,7 @@ wheels = [ [[package]] name = "opencloning" -version = "1.9.3" +version = "1.9.4" source = { editable = "packages/opencloning" } dependencies = [ { name = "beautifulsoup4" }, @@ -1236,7 +1236,7 @@ requires-dist = [ [[package]] name = "opencloning-cli" -version = "1.9.3" +version = "1.9.4" source = { editable = "packages/opencloning-cli" } dependencies = [ { name = "opencloning-db" }, @@ -1251,7 +1251,7 @@ requires-dist = [ [[package]] name = "opencloning-db" -version = "1.9.3" +version = "1.9.4" source = { editable = "packages/opencloning-db" } dependencies = [ { name = "alembic" },