added a function to retrieve the environment type for a barcode

knimin/lib/data_access.py

@@ -2214,7 +2214,10 @@ def getAGBarcodeDetails(self, barcode):
         if not results:
             return {}
         else:
-            return dict(results)
+            res = dict(results)
+            res['sample_environments'] = \
+                ",".join(self.get_barcode_environment_types(barcode))
+            return res
 
     def get_barcode_info_by_kit_id(self, ag_kit_id):
         sql = """SELECT DISTINCT cast(ag_kit_barcode_id as varchar(100)) as
@@ -2437,6 +2440,63 @@ def _revert_ready(self, barcodes):
                  WHERE barcode IN %s"""
         self._con.execute(sql, [tuple(barcodes)])
 
+    def get_barcode_environment_types(self, barcode):
+        """ Returns the environments of the given barcode.
+
+        Parameters
+        ----------
+        barcode : str
+            A single AG barcode ID
+
+        Returns
+        -------
+        list of environments, i.e. survey types that have been associated
+        with this barcode.
+
+        Raises
+        ------
+        ValueError
+            given barcode not in DB."""
+
+        environments = []
+
+        # check if the barcode exists in the DB at all
+        sql = """SELECT barcode
+                 FROM ag.ag_kit_barcodes
+                 WHERE barcode = %s"""
+        res = self._con.execute_fetchone(sql, [barcode])
+        if res is None:
+            raise ValueError("Barcode '%s' not in DB." % barcode)
+
+        sql = """SELECT DISTINCT ags.survey_id
+                 FROM ag.ag_kit_barcodes
+                 JOIN ag.survey_answers USING (survey_id)
+                 JOIN ag.group_questions USING (survey_question_id)
+                 JOIN ag.surveys ags USING (survey_group)
+                 WHERE barcode = %s"""
+        res = self._con.execute_fetchall(sql, [barcode])
+        if len(res) > 0:
+            if res[0] == [1]:
+                environments.append('Human')
+            elif res[0] == [2]:
+                # find out what species it is
+                sql = """SELECT response
+                         FROM ag.ag_kit_barcodes
+                         JOIN ag.survey_answers USING (survey_id)
+                         WHERE barcode = %s AND survey_question_id = 128"""
+                res = self._con.execute_fetchone(sql, [barcode])
+                environments.append('Animal(%s)' % res[0])
+
+        sql = """SELECT environment_sampled
+                 FROM ag.ag_kit_barcodes
+                 WHERE barcode = %s"""
+        res = self._con.execute_fetchall(sql, [barcode])
+
+        if (len(res) > 0) and (res[0][0] is not None):
+            environments.append('Environment(%s)' % res[0][0])
+
+        return environments
+
     def ut_remove_external_survey(self, name, description, url):
         """ Remove an external survey from DB.
         For unit testing only!

knimin/lib/tests/test_data_access.py

@@ -329,6 +329,43 @@ def test_get_ag_barcode_details(self):
             self.assertEqual({k: obs[key][k] for k in exp[key]}, exp[key])
             self.assertIn(obs[key]['participant_name'], participant_names)
 
+    def test_get_barcode_environment_types(self):
+        # check that error is raised, if barcode is not in database
+        barcode = '9notInDB1'
+        with self.assertRaises(ValueError) as e:
+            db.get_barcode_environment_types(barcode)
+        self.assertEqual(e.exception.message,
+                         u"Barcode '%s' not in DB." % barcode)
+
+        # barcode that is not assigned to a survey, result should be the empty
+        # list
+        barcode = '000027834'
+        exp = []
+        obs = db.get_barcode_environment_types(barcode)
+        self.assertEqual(obs, exp)
+
+        # environmental barcode
+        barcode = '000015774'
+        exp = [u'Environment(Sole of shoe)']
+        obs = db.get_barcode_environment_types(barcode)
+        self.assertEqual(obs, exp)
+
+        # human barcode
+        barcode = '000004216'
+        exp = [u'Human']
+        obs = db.get_barcode_environment_types(barcode)
+        self.assertEqual(obs, exp)
+
+        # animal barcodes
+        barcode = '000002012'
+        exp = [u'Human']
+        obs = db.get_barcode_environment_types(barcode)
+        self.assertEqual(obs, exp)
+        barcode = '000013439'
+        exp = [u'Animal(Large Mammal)']
+        obs = db.get_barcode_environment_types(barcode)
+        self.assertEqual(obs, exp)
+
 
 if __name__ == "__main__":
     main()

knimin/lib/tests/test_geocoder.py

@@ -281,5 +281,6 @@ def test_geocode_bad_address(self):
    "status" : "INVALID_REQUEST"
 }'''
 
+
 if __name__ == '__main__':
     main()

knimin/templates/barcode_util.html

@@ -144,6 +144,7 @@ <h2>{{parent_project}} Details</h2>
              <li>
              Sample details: <br/>
              <table>
+             <tr><td>Sample Environment</td><td>{{proj_barcode_info['sample_environments']}}</td></tr>
              <tr><td>Sample Date</td><td>{{proj_barcode_info['sample_date']}}</td></tr>
              <tr><td>Sample Time</td><td>{{proj_barcode_info['sample_time']}}</td></tr>
              <tr><td>Sample Site</td><td>{{proj_barcode_info['site_sampled']}}</td></tr>

knimin/tests/test_ag_search.py

@@ -61,8 +61,8 @@ def test_post_barcode(self):
             for sample in db.get_barcode_info_by_kit_id(kit_id):
                 for field in sample:
                     if sample[field] is not None:
-                        if (field == 'ag_kit_id') or (
-                             field == 'ag_kit_barcode_id'):
+                        if (field == 'ag_kit_id') or
+                        (field == 'ag_kit_barcode_id'):
                             self.assertNotIn(sample[field], response.body)
                         else:
                             exp = xhtml_escape_recursive(sample[field])

knimin/tests/test_barcode_util.py

@@ -318,9 +318,8 @@ def test_get_ag_details_SJ(self):
         del ag_details['overloaded']
         del ag_details['other']
         del ag_details['deposited']
-        exp = {
+        exp = {'environment_sampled': '', 'withdrawn': '',
                # 'login_user': 'REMOVED',
-               'environment_sampled': '', 'withdrawn': '',
                'ag_kit_id': 'd8592c74-7ddb-2135-e040-8a80115d6401',
                'overloaded_checked': '',
                # 'participant_name': 'REMOVED-0',
@@ -335,15 +334,17 @@ def test_get_ag_details_SJ(self):
                'sample_time': time(6, 50),
                # 'notes': 'REMOVED',
                # 'email': 'REMOVED'
-              }
+               }
         # only look at those fields, that are not subject to scrubbing
         self.assertEqual({k: ag_details[k] for k in exp}, exp)
         exp_keys = ['login_user', 'environment_sampled', 'withdrawn',
                     'ag_kit_id', 'overloaded_checked', 'participant_name',
                     'ag_kit_barcode_id', 'sample_date', 'other_checked',
                     'status', 'refunded', 'other_text', 'barcode',
                     'moldy_checked', 'date_of_last_email', 'site_sampled',
-                    'email_type', 'name', 'sample_time', 'notes', 'email']
+                    'email_type', 'name', 'sample_time', 'notes', 'email',
+                    u'sample_environments']
+        self.maxDiff = None
         self.assertEqual(sorted(ag_details.keys()), sorted(exp_keys))
 
         # check that None values are set to ''