Gautam's edits #1

Summary: In english.py, I added missing English translations of corresponding protocols that were in protocol.py. Additionally, I added a line to pytest.ini to allow for doctring tests. Finally, I am also submitting my testing file, AP2En_test.py.

Test Plan: I ran doctests through the PyTest testing format to check the correctness of the parser's standard output (see AP2En_test.py).

Reviewers: peter, yangchoo

Differential Revision: https://work.r23s.net/D5009

pytest.ini

@@ -3,3 +3,4 @@ minversion = 2.3.3
 python_files = "test/*_test.py" "test/test_*.py"
 python_classes = Test
 norecursedirs = ".tox" "build" "docs"
+addopts = --doctest-glob='*_test.py'

test/AP2En_test.py

@@ -0,0 +1,184 @@
+import unittest
+import doctest
+import json
+from autoprotocol.container import Container, WellGroup
+from autoprotocol.instruction import Thermocycle, Incubate, Spin
+from autoprotocol.pipette_tools import *  # flake8: noqa
+from autoprotocol.protocol import Protocol, Ref
+from autoprotocol.unit import Unit
+from transcriptic import english
+from transcriptic import cli
+
+
+class AP2EnTestCase(unittest.TestCase):
+
+    maxDiff = None
+
+    def test_measure_suite(self):
+        """ Doctest:
+>>> test_measure_suite()
+1. Measure concentration of 2.0 microliters DNA source aliquots
+2. Measure mass of test_plate3
+3. Mesaure volume of 12 wells from test_plate
+4. Mesaure volume of 8 wells from test_plate2
+"""
+        p = Protocol()
+        test_plate = p.ref("test_plate", None, "96-flat", storage="cold_4")
+        test_plate2 = p.ref("test_plate2", id=None,
+                            cont_type="96-flat", storage=None, discard=True)
+        for well in test_plate2.all_wells():
+            well.set_volume("150:microliter")
+
+        p.measure_concentration(wells=test_plate2.wells_from(
+            0, 96), dataref="mc_test", measurement="DNA", volume=Unit(2, "microliter"))
+        p.measure_mass(test_plate2, "test_ref")
+        p.measure_volume(test_plate.wells_from(0, 12), "test_ref")
+        p.measure_volume(test_plate2.wells_from(
+            1, 8), "test_ref")  # is this correct??
+
+        pjsonString = json.dumps(p.as_dict(), indent=2)
+        pjson = json.loads(pjsonString)
+        english.AutoprotocolParser(pjson)
+
+    def test_mag_suite(self):
+        """ Doctest:
+>>> test_mag_suite()
+1. Magnetically mix pcr_0 beads for 30.0 seconds at an aplitude of 0
+2. Magnetically collect pcr_1 beads for 5 cycles with a pause duration of 30.0 seconds
+3. Magnetically incubate pcr_0 for 30.0 minutes with a tip position of 1.5
+4. Magnetically dry pcr_0 for 30.0 minutes
+5. Distribute from test/1 into wells test/7, test/8, test/9
+6. Distribute from test/2 into wells test/10
+7. Distribute from test/0 into wells test/1
+8. Magnetically release pcr_0 beads for 30.0 seconds at an aplitude of 0
+"""
+        p = Protocol()
+        pcrs = [p.ref("pcr_%s" % i, None, "96-pcr", storage="cold_20")
+                for i in range(7)]
+        pcr = pcrs[0]
+
+        p.mag_mix("96-pcr", pcr, "30:second", "60:hertz",
+                  center=float(100) / 100, amplitude=0)
+        p.mag_collect("96-pcr", pcr, 5, "30:second",
+                      bottom_position=float(5) / 100)
+        p.mag_incubate("96-pcr", pcr, "30:minute", temperature="30:celsius")
+        p.mag_dry("96-pcr", pcr, "30:minute",
+                  new_tip=False, new_instruction=False)
+
+        c = p.ref("test", None, "96-flat", discard=True)
+        srcs = c.wells_from(1, 2).set_volume("100:microliter")
+        dests = c.wells_from(7, 4)
+        p.distribute(srcs, dests, "30:microliter", allow_carryover=True)
+        # test distribute from Well to Well
+        p.distribute(c.well("A1").set_volume(
+            "20:microliter"), c.well("A2"), "5:microliter")
+        p.mag_release("96-pcr", pcr, "30:second", "1:hertz",
+                      center=float(5) / 100, amplitude=0)
+
+        pjsonString = json.dumps(p.as_dict(), indent=2)
+        pjson = json.loads(pjsonString)
+        a = english.AutoprotocolParser(pjson)
+        print(a)
+
+    def test_purify(self):
+        """ Doctest:
+>>> test_purify()
+1. Perform gel purification on the 0.8% agarose gel with band range(s) 0-10
+2. Perform gel purification on the 0.8% agarose gel with band range(s) 0-10
+3. Perform gel purification on the 0.8% agarose gel with band range(s) 0-10
+"""
+        p = Protocol()
+        sample_wells = p.ref("sample_wells", None, "96-pcr",
+                             discard=True).wells_from(0, 20)
+        extract_wells = [p.ref("extract_%s" % i, None, "micro-1.5",
+                               storage="cold_4").well(0) for i in sample_wells]
+        extract = [
+            {
+                "source": sample_wells[i],
+                "band_list": [{
+                    "band_size_range": {"min_bp": 0, "max_bp": 10},
+                    "elution_volume": Unit("5:microliter"),
+                    "elution_buffer": "water",
+                    "destination": d
+                }],
+                "lane": None,
+                "gel": None
+            } for i, d in enumerate(extract_wells)
+        ]
+
+        p.gel_purify(extract, "10:microliter",
+                     "size_select(8,0.8%)", "ladder1", "gel_purify_test")
+
+        pjsonString = json.dumps(p.as_dict(), indent=2)
+        pjson = json.loads(pjsonString)
+        a = english.AutoprotocolParser(pjson)
+        print(a)
+
+    def test_dispense_suite(self):
+        """ Doctest:
+>>> test_dispense_suite()
+1. Dispense 100 microliters of water to the full plate of sample_plate5
+2. Dispense corresponding amounts of water to 12 column(s) of sample_plate5
+"""
+        p = Protocol()
+        sample_plate5 = p.ref("sample_plate5", None,
+                              "96-flat", storage="warm_37")
+
+        p.dispense_full_plate(sample_plate5, "water", "100:microliter")
+        p.dispense(sample_plate5,
+                   "water",
+                   [{"column": 0, "volume": "10:microliter"},
+                    {"column": 1, "volume": "20:microliter"},
+                    {"column": 2, "volume": "30:microliter"},
+                    {"column": 3, "volume": "40:microliter"},
+                    {"column": 4, "volume": "50:microliter"},
+                    {"column": 5, "volume": "60:microliter"},
+                    {"column": 6, "volume": "70:microliter"},
+                    {"column": 7, "volume": "80:microliter"},
+                    {"column": 8, "volume": "90:microliter"},
+                    {"column": 9, "volume": "100:microliter"},
+                    {"column": 10, "volume": "110:microliter"},
+                    {"column": 11, "volume": "120:microliter"}
+                    ])
+        pjsonString = json.dumps(p.as_dict(), indent=2)
+        pjson = json.loads(pjsonString)
+        a = english.AutoprotocolParser(pjson)
+        print(a)
+
+    def test_illumina(self):
+        """ Doctest:
+>>> test_illumina()
+1. Illumina sequence wells test_plate6/0, test_plate6/1 with library size 34
+"""
+        p = Protocol()
+        sample_wells6 = p.ref(
+            "test_plate6", None, "96-pcr", discard=True).wells_from(0, 8)
+        p.illuminaseq("PE", [{"object": sample_wells6[0], "library_concentration": 1.0},
+                             {"object": sample_wells6[1], "library_concentration": 2}],
+                      "nextseq", "mid", 'none', 34, "dataref")
+        pjsonString = json.dumps(p.as_dict(), indent=2)
+        pjson = json.loads(pjsonString)
+        english.AutoprotocolParser(pjson)
+
+    def test_flow(self):
+        """ Doctest:
+>>> test_flow()
+1. Perform flow cytometry on well0 with the respective FSC and SSC channel parameters
+"""
+        p = Protocol()
+        dataref = "flow_test"  # dummy variable
+        FSC = {"voltage_range": {"low": "230:volt", "high": "280:volt"},
+               "area": True, "height": True, "weight": False}
+        SSC = {"voltage_range": {"low": "230:volt", "high": "280:volt"},
+               "area": True, "height": True, "weight": False}
+        neg_controls = {"well": "well0", "volume": "100:microliter",
+                        "captured_events": 5, "channel": "channel0"}
+        samples = [
+            {"well": "well0", "volume": "100:microliter", "captured_events": 9}]
+
+        p.flow_analyze(dataref, FSC, SSC, neg_controls,
+                       samples, colors=None, pos_controls=None)
+
+        pjsonString = json.dumps(p.as_dict(), indent=2)
+        pjson = json.loads(pjsonString)
+        english.AutoprotocolParser(pjson)

transcriptic/english.py

@@ -28,7 +28,8 @@ def parse(self, obj):
             except AttributeError:
                 parsed_output.append("[Unknown instruction]")
         for i, p in enumerate(parsed_output):
-            print("%d. %s" % (i+1, p))
+            print("%d. %s" % (i + 1, p))
+    # add a graph maker and maybe take out the integer count for each step...?
 
     def absorbance(self, opts):
         return ("Measure absorbance at %s for %s of plate %s" %
@@ -44,7 +45,6 @@ def acoustic_transfer(self, opts):
         return transfers
 
     def autopick(self, opts):
-
         picks = []
         for i, g in enumerate(opts['groups']):
             picks.extend(["Pick %s colonies from %s %s: %s to %s, %s" %
@@ -54,18 +54,31 @@ def autopick(self, opts):
                            self.well_list(g['to']),
                            ("data saved at '%s'" % opts["dataref"]
                             if i is 0 else "analyzed with previous"))])
-
         return picks
 
     @staticmethod
     def cover(opts):
         return "Cover %s with a %s lid" % (opts['object'], opts['lid'])
 
-    @staticmethod
-    def dispense(opts):
-        return "Dispense %s to %d column(s) of %s" % (opts['reagent'],
-                                                      len(opts['columns']),
-                                                      opts['object'])
+    # @staticmethod
+    # def dispense(opts):
+    #     return "Dispense %s to %d column(s) of %s" % (opts['reagent'],
+    #                                                   len(opts['columns']),
+    #                                                   opts['object'])
+
+    # ----- Gautam's edits 6/7 -----
+    def dispense(self, opts):
+        unique_vol = []
+        for col in opts['columns']:
+            vol = self.unit(col["volume"])
+            if vol not in unique_vol:
+                unique_vol.append(vol)
+
+        if len(opts['columns']) == 12 and len(unique_vol) == 1:
+            return "Dispense %s of %s to the full plate of %s" % (unique_vol[0], opts['reagent'], opts['object'])
+        else:
+            return "Dispense corresponding amounts of %s to %d column(s) of %s" % (opts['reagent'], len(opts['columns']), opts['object'])
+    # ------------------------------
 
     def flash_freeze(self, opts):
         return ("Flash freeze %s for %s" %
@@ -84,6 +97,25 @@ def gel_separate(self, opts):
                 "a %s agarose gel for %s" % (opts['matrix'].split(',')[1][:-1],
                                              self.unit(opts['duration'])))
 
+    # ----- Gautam's edits 6/7 -----
+    @staticmethod
+    def gel_purify(opts):
+        unique_bl = []
+        for ext in opts['extract']:
+            bl = ext["band_size_range"]
+            if bl not in unique_bl:
+                unique_bl.append(bl)
+        for i in range(len(unique_bl)):
+            unique_bl[i] = str(unique_bl[i]['min_bp']) + \
+                "-" + str(unique_bl[i]['max_bp'])
+
+        if len(unique_bl) <= 3:
+            return "Perform gel purification on the %s agarose gel with band range(s) %s" % (opts['matrix'].split(',')[1][:-1], ', '.join(unique_bl))
+        else:
+            return "Perform gel purification on the %s agarose gel with %s band ranges" % (opts['matrix'].split(',')[1][:-1], len(unique_bl))
+
+    # ------------------------------
+
     def incubate(self, opts):
         shaking = " (shaking)" if opts['shaking'] else ""
         return "Incubate %s at %s for %s%s" % (opts['object'],
@@ -96,19 +128,16 @@ def image_plate(opts):
         return "Take an image of %s" % opts['object']
 
     def luminescence(self, opts):
-        return ("Read luminescence of %s of plate %s" %
-                (self.well_list(opts['wells']), opts['object']))
+        return ("Read luminescence of %s of plate %s" % (self.well_list(opts['wells']), opts['object']))
 
     @staticmethod
     def oligosynthesize(opts):
-        return (["Oligosynthesize sequence '%s' into '%s'" %
-                 (o['sequence'], o['destination']) for o in opts['oligos']])
+        return (["Oligosynthesize sequence '%s' into '%s'" % (o['sequence'], o['destination']) for o in opts['oligos']])
 
     def provision(self, opts):
         provisions = []
         for t in opts['to']:
-            provisions.append("Provision %s of resource with ID %s to well %s \
-                              of container %s" %
+            provisions.append("Provision %s of resource with ID %s to well %s of container %s" %
                               (self.unit(t['volume']), opts['resource_id'],
                                self.well(t['well']), self.platename(t['well'])
                                ))
@@ -122,6 +151,40 @@ def sanger_sequence(self, opts):
         elif opts['type'] == "rca":
             return seq + " with %s" % self.platename(opts['primer'])
 
+    # ----- Gautam's edits 6/7 -----
+    def get_unique_wells(self, list_of_wells):
+        unique_wells = []
+        for well in list_of_wells:
+            w = well['object']
+            if w not in unique_wells:
+                unique_wells.append(w)
+        return unique_wells
+
+    def illumina_sequence(self, opts):
+        unique_wells = self.get_unique_wells(opts['lanes'])
+        unique_plates = self.get_unique_plates(unique_wells)
+
+        if len(unique_plates) == 1 and len(unique_wells) <= 3:
+            seq = "Illumina sequence wells %s" % (", ".join(unique_wells))
+        elif len(unique_plates) > 1 and len(unique_plates) <= 3:
+            seq = "Illumina sequence the corresponding wells of plates %s" % ", ".join(
+                unique_plates[0])
+        else:
+            seq = "Illumina sequence the corresponding wells of %s plates" % len(
+                unique_wells)
+
+        return seq + " with library size %s" % opts['library_size']
+
+    @staticmethod
+    def flow_analyze(opts):
+        wells = []
+        for sample in opts['samples']:
+            if sample['well'] not in wells:
+                wells.append(sample['well'])
+
+        return "Perform flow cytometry on %s with the respective FSC and SSC channel parameters" % ", ".join(wells)
+    # ------------------------------
+
     @staticmethod
     def seal(opts):
         return "Seal %s (%s)" % (opts['object'], opts['type'])
@@ -197,6 +260,49 @@ def pipette(self, opts):
                                      g[pip]['to']))
         return pipettes
 
+    # ----- Gautam's edits 6/7 -----
+    def magnetic_transfer(self, opts):
+        # dry, incubate, collect, release, mix
+        specific_op = list(opts['groups'][0][0].keys())[0]
+        specs_dict = opts['groups'][0][0][specific_op]
+        seq = "Magnetically %s %s" % (specific_op, specs_dict["object"])
+
+        if specific_op == "dry":
+            return seq + " for %s" % self.unit(specs_dict["duration"])
+        elif specific_op == "incubate":
+            return seq + " for %s with a tip position of %s" % (self.unit(specs_dict["duration"]), specs_dict["tip_position"])
+        elif specific_op == "collect":
+            return seq + " beads for %s cycles with a pause duration of %s" % (specs_dict["cycles"], self.unit(specs_dict["pause_duration"]))
+        elif specific_op == "release" or "mix":
+            return seq + " beads for %s at an aplitude of %s" % (self.unit(specs_dict["duration"]), specs_dict["amplitude"])
+
+    def get_unique_plates(self, list_of_wells):
+        unique_plates = []
+        for well in list_of_wells:
+            loc = well.find('/')
+            if loc == -1:
+                plate = well
+            else:
+                plate = well[:loc]
+
+            if plate not in unique_plates:
+                unique_plates.append(plate)
+        return unique_plates
+
+    def measure_volume(self, opts):
+        unique_plates = self.get_unique_plates(opts['object'])
+        if len(unique_plates) <= 3:
+            return "Mesaure volume of %s wells from %s" % (len(opts['object']), ", ".join(unique_plates))
+        else:
+            return "Mesaure volume of %s wells from the %s plates" % (len(opts['object']), len(unique_plates))
+
+    def measure_mass(self, opts):
+        return "Measure mass of %s" % ", ".join(opts['object'])
+
+    def measure_concentration(self, opts):
+        return "Measure concentration of %s %s source aliquots" % (self.unit(opts['volume']), opts['measurement'])
+    # ------------------------------
+
     @staticmethod
     def uncover(opts):
         return "Uncover %s" % opts['object']