|
| 1 | +# MIT License |
| 2 | +# Copyright (c) 2024, Technical University of Denmark (DTU) |
| 3 | +# |
| 4 | +# Permission is hereby granted, free of charge, to any person obtaining a copy |
| 5 | +# of this software and associated documentation files (the "Software"), to deal |
| 6 | +# in the Software without restriction, including without limitation the rights |
| 7 | +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell |
| 8 | +# copies of the Software, and to permit persons to whom the Software is |
| 9 | +# furnished to do so, subject to the following conditions: |
| 10 | +# |
| 11 | +# The above copyright notice and this permission notice shall be included in all |
| 12 | +# copies or substantial portions of the Software. |
| 13 | + |
| 14 | +"""Easy to use Benchling functions to fetch sequences and objects.""" |
| 15 | + |
| 16 | +import datetime |
| 17 | +import os |
| 18 | + |
| 19 | +import Bio |
| 20 | +import pandas as pd |
| 21 | +import pydna |
| 22 | +from benchlingapi import Session |
| 23 | +from dotenv import find_dotenv, load_dotenv |
| 24 | + |
| 25 | +from teemi.utils import ( |
| 26 | + nest_dict, |
| 27 | + rename_dict_keys, |
| 28 | + split_based_on_keys, |
| 29 | + start_end_to_location, |
| 30 | +) |
| 31 | + |
| 32 | + |
| 33 | +load_dotenv(find_dotenv()) |
| 34 | + |
| 35 | +api_key = os.environ.get("API_KEY") |
| 36 | +home_url = os.environ.get("HOME_url") |
| 37 | +session = Session(api_key=api_key, home=home_url) |
| 38 | + |
| 39 | + |
| 40 | +def sequence_to_benchling(folder_name, oligo_name, oligo_bases, schema): |
| 41 | + """Upload sequences to Benchling.""" |
| 42 | + folder = session.Folder.find_by_name(folder_name) |
| 43 | + dna = session.DNASequence( |
| 44 | + name=oligo_name, bases=oligo_bases, folder_id=folder.id, is_circular=False |
| 45 | + ) |
| 46 | + dna.save() |
| 47 | + |
| 48 | + schemas = ( |
| 49 | + "Primer", |
| 50 | + "DNA Fragment", |
| 51 | + "Plasmid", |
| 52 | + "Gene", |
| 53 | + "gRNA", |
| 54 | + "Marker", |
| 55 | + "Promoter", |
| 56 | + "Terminator", |
| 57 | + "Tag", |
| 58 | + "Origin of Replication", |
| 59 | + ) |
| 60 | + if schema in schemas: |
| 61 | + dna.set_schema(schema) |
| 62 | + |
| 63 | + dna.register() |
| 64 | + |
| 65 | + |
| 66 | +def from_benchling(bname: str, schema: str = ""): |
| 67 | + """Extract information of an object on Benchling.""" |
| 68 | + bench_dict = session.DNASequence.find_by_name(bname).dump() |
| 69 | + |
| 70 | + trans_dict = { |
| 71 | + "bases": "seq", |
| 72 | + "id": "id", |
| 73 | + "annotations": "features", |
| 74 | + "name": "name", |
| 75 | + "fields": "annotations", |
| 76 | + } |
| 77 | + trans_bench_dict = rename_dict_keys(bench_dict, trans_dict) |
| 78 | + |
| 79 | + translated_bench_dict_sel, translated_bench_dict_other = split_based_on_keys( |
| 80 | + trans_bench_dict, trans_dict.values() |
| 81 | + ) |
| 82 | + |
| 83 | + translated_bench_dict_sel["annotations"].update( |
| 84 | + translated_bench_dict_other["customFields"] |
| 85 | + ) |
| 86 | + translated_bench_dict_sel["annotations"].update( |
| 87 | + {"topology": translated_bench_dict_other["isCircular"]} |
| 88 | + ) |
| 89 | + |
| 90 | + date = datetime.datetime.today().strftime("%d-%b-%Y").upper() |
| 91 | + comment = translated_bench_dict_sel["annotations"].pop("comment", None) |
| 92 | + translated_bench_dict_sel["annotations"].update( |
| 93 | + { |
| 94 | + "data_file_division": "PLN", |
| 95 | + "date": date, |
| 96 | + "molecule_type": "DNA", |
| 97 | + "location": "unknown", |
| 98 | + "commentary": comment, |
| 99 | + } |
| 100 | + ) |
| 101 | + |
| 102 | + translated_bench_dict_sel["seq"] = Bio.Seq.Seq(translated_bench_dict_sel["seq"]) |
| 103 | + seq_length = len(translated_bench_dict_sel["seq"]) |
| 104 | + |
| 105 | + translated_bench_dict_sel["features"] = [ |
| 106 | + start_end_to_location(feature_dict, seq_length) |
| 107 | + for feature_dict in translated_bench_dict_sel["features"] |
| 108 | + ] |
| 109 | + for feature in translated_bench_dict_sel["features"]: |
| 110 | + feature.update({"label": feature.get("name", "")}) |
| 111 | + translated_bench_dict_sel["features"] = [ |
| 112 | + nest_dict( |
| 113 | + feature_dict, |
| 114 | + first_order_keys=["location", "type", "strand"], |
| 115 | + key_for_nested_dict="qualifiers", |
| 116 | + ) |
| 117 | + for feature_dict in translated_bench_dict_sel["features"] |
| 118 | + ] |
| 119 | + translated_bench_dict_sel["features"] = [ |
| 120 | + Bio.SeqFeature.SeqFeature(**feature_dict) |
| 121 | + for feature_dict in translated_bench_dict_sel["features"] |
| 122 | + ] |
| 123 | + |
| 124 | + seqRecord = Bio.SeqRecord.SeqRecord(**translated_bench_dict_sel) |
| 125 | + seqRecord.name = bname |
| 126 | + seqRecord = update_loc_vol_conc(seqRecord) |
| 127 | + |
| 128 | + if schema == "Primer": |
| 129 | + seqRecord = pydna.primer.Primer(seqRecord) |
| 130 | + |
| 131 | + return seqRecord |
| 132 | + |
| 133 | + |
| 134 | +def update_loc_vol_conc(seqRecord, DBpath: str = ""): |
| 135 | + """Update location, volume, and concentration information from a CSV export.""" |
| 136 | + DB = pd.read_csv(DBpath) |
| 137 | + |
| 138 | + seqRecord.annotations["batches"] = [] |
| 139 | + for _, row in DB.iterrows(): |
| 140 | + if row["batchEntId"] == seqRecord.id: |
| 141 | + location = row["parentBoxPlateName"] + "_" + row["parentBoxPlatePos"] |
| 142 | + seqRecord.annotations["batches"].append( |
| 143 | + { |
| 144 | + "box": row["parentBoxPlateName"], |
| 145 | + "position": row["parentBoxPlatePos"], |
| 146 | + "volume": int(row["volume"]), |
| 147 | + "concentration": int(row["Concentration (ng/ul)"]), |
| 148 | + "location": location, |
| 149 | + } |
| 150 | + ) |
| 151 | + return seqRecord |
0 commit comments