#!/usr/bin/env python3 """Pool a set of oligos by cartridge and chip, then queue them in a sensible order. python3 kb_queue_assay.py --host kilobaser.lab.example.org --user apibot \\ --insecure --dry-run kb_lamp_primers.fasta Give it a FASTA file. Records default to a standard unmodified oligo; a record that needs something else says so in its header: >LAMP-042_RPP30-P cartridge=3 chip=3-BHQ1 AGCCTGACTTGCAAGGTCATGCTT Why pool at all, when the machine inserts the cartridge and chip changes for you? Because those changes are not free: * a chip is single-use, so every oligo costs one regardless; * swapping a cartridge **discards the one that was installed**, along with whatever is left in it; * swapping *away from* a fluorophore cartridge additionally forces a manual cleaning run, which somebody has to stand at the machine and perform. Submitting one oligo at a time in the order a user happened to click them can turn a two-step run into a nine-step one, over and over. Grouping by cartridge and chip, and leaving the fluorophore cartridges until last, makes each change happen once. `--dry-run` prints the plan and the cartridge shopping list without touching the queue. Run it first. """ import argparse import getpass import re import sys from collections import Counter, OrderedDict from kb_client import Kilobaser, KilobaserError HEADER_OPTION = re.compile(r"\b(cartridge|chip)=(\S+)") def parse_fasta(text, default_cartridge, default_chip): """Yield (title, sequence, cartridge, chip) per record. The device's own FASTA import takes the whole header line as the title; here we strip the `cartridge=`/`chip=` tokens back out so they do not end up in it. """ oligos, title, options, chunks = [], None, {}, [] def flush(): if title is not None: oligos.append(( title, "".join(chunks).upper(), options.get("cartridge", default_cartridge), options.get("chip", default_chip), )) for line in text.splitlines(): line = line.strip() if not line: continue if line[0] in ">;": flush() header = line[1:].strip() options = dict(HEADER_OPTION.findall(header)) title = HEADER_OPTION.sub("", header).strip() chunks = [] elif title is not None: chunks.append(re.sub(r"[^A-Za-z]", "", line)) flush() return oligos def pool(oligos, ccsettings): """Group oligos by (cartridge, chip), ordered to make each change happen once. Groups sharing a cartridge are kept adjacent so the cartridge is installed once and its chips swapped within it. Fluorophore cartridges go last, because leaving one is what triggers the manual cleaning -- ending on it means never paying that. """ cartridges = ccsettings["cartridges"] def is_fluorophore(cartridge_type): # Fluorophore cartridges are the ones that offer a labelled chip: their # compatible chips include something other than the plain/placeholder set. chips = set(ccsettings["operations"].get(cartridge_type, {})) return bool(chips - {"1", "2", "2-nocap", "1000"}) groups = OrderedDict() for oligo in oligos: groups.setdefault((oligo[2], oligo[3]), []).append(oligo) def sort_key(pair): cartridge, chip = pair return (is_fluorophore(cartridge), cartridges.get(cartridge, {}).get("cartridgePriority", 0), cartridge, chip) return OrderedDict((key, groups[key]) for key in sorted(groups, key=sort_key)) def check(pair, oligos, ccsettings): """Return a list of problems with this pool, in plain language. Validating against `operations` locally is worth the few lines: the device rejects an impossible cartridge/chip pair with "provided processType is invalid", which sends you looking at the wrong field entirely. """ cartridge, chip = pair problems = [] operations = ccsettings["operations"] if cartridge not in operations: return [f"unknown cartridge type {cartridge!r}"] if chip not in operations[cartridge]: usable = ", ".join(sorted(operations[cartridge])) return [f"chip {chip!r} cannot run on cartridge {cartridge!r}; that " f"cartridge accepts: {usable}"] spec = ccsettings["cartridges"][cartridge] total = sum(len(sequence) for _, sequence, _, _ in oligos) per_base = Counter() for _, sequence, _, _ in oligos: per_base.update(sequence) capacity = spec.get("totalBasesCount") or 0 if capacity and total > capacity: problems.append( f"{total} bases needs {-(-total // capacity)} cartridges " f"({spec['name']} holds {capacity})") for base, count in sorted(per_base.items()): limit = (spec.get("baseCount") or {}).get(base) if limit and count > limit: problems.append( f"{count}x {base} exceeds the {limit} available per cartridge -- a " "sequence set skewed towards one base runs out before the total does") return problems def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("fasta", help="FASTA file of oligos to synthesise") parser.add_argument("--host", required=True) parser.add_argument("--user", required=True) parser.add_argument("--password") parser.add_argument("--insecure", action="store_true") parser.add_argument("--cartridge", default="2", help="default cartridge type") parser.add_argument("--chip", default="2", help="default chip kind") parser.add_argument("--dry-run", action="store_true", help="print the plan without queueing anything") args = parser.parse_args() with open(args.fasta) as handle: oligos = parse_fasta(handle.read(), args.cartridge, args.chip) if not oligos: sys.exit(f"no records found in {args.fasta}") password = args.password or getpass.getpass(f"password for {args.user}: ") kb = Kilobaser(args.host, args.user, password, verify=not args.insecure) kb.login() ccsettings = kb.cartridge_chip_settings() pools = pool(oligos, ccsettings) failed = False for pair, members in pools.items(): cartridge, chip = pair names = ccsettings["cartridges"].get(cartridge, {}).get("name", cartridge) chip_name = ccsettings["chips"].get(chip, {}).get("name", chip) total = sum(len(sequence) for _, sequence, _, _ in members) print(f"\n{names} + {chip_name} chip ({cartridge} / {chip})") plural = "" if len(members) == 1 else "s" print(f" {len(members)} oligo{plural}, {total} bases") for problem in check(pair, members, ccsettings): print(f" ! {problem}") failed = True for title, sequence, _, _ in members: print(f" {title:<28} {len(sequence):>3} nt") if args.dry_run: print("\ndry run -- nothing was queued") return if failed: sys.exit("\nrefusing to queue: fix the problems above first") print() created = [] for (cartridge, chip), members in pools.items(): fasta = "".join(f">{title}\n{sequence}\n" for title, sequence, _, _ in members) try: entries = kb.import_fasta(fasta, cartridge_type=cartridge, chip_kind=chip) except KilobaserError as exc: sys.exit(f"queueing {cartridge}/{chip} failed: {exc}") created.extend(entries) for entry in entries: print(f"queued {entry['title']:<28} {entry['id']}") # Read the queue back one final time. The machine replans after every change and # drops anything it can no longer schedule, so this is what actually got queued # -- not what we sent. final = {entry["id"] for entry in kb.queue()} missing = [entry["title"] for entry in created if entry["id"] not in final] if missing: print(f"\n! the machine dropped: {', '.join(missing)}") print(f"\n{len(final)} entries in the queue") if __name__ == "__main__": main()