added a function to retrieve the environment type for a barcode

knimin/lib/configuration.py

@@ -89,4 +89,5 @@ def _get_email(self, config):
         self.smtp_user = config.get('email', 'USERNAME')
         self.smtp_password = config.get('email', 'PASSWORD')
 
+
 config = KniminConfig()

knimin/lib/constants.py

@@ -488,6 +488,7 @@
 def default_blank():
     return 'Not applicable'
 
+
 blanks_values = defaultdict(default_blank,
                             ASSIGNED_FROM_GEO="No",
                             COMMON_NAME="unclassified metagenome",

knimin/lib/data_access.py

@@ -2202,7 +2202,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
@@ -2422,3 +2425,61 @@ def _revert_ready(self, barcodes):
                  SET results_ready = NULL
                  WHERE barcode IN %s"""
         self._con.execute(sql, [tuple(barcodes)])
+
+    def get_barcode_environment_types(self, barcode):
+        # TODO: write test code once test data have been dumped into DB!
+        """ 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

knimin/lib/geocoder.py

@@ -15,6 +15,7 @@ class GoogleAPIRequestDenied(Exception):
 class GoogleAPIInvalidRequest(Exception):
     pass
 
+
 Location = namedtuple('Location', ['input', 'lat', 'long', 'elev', 'city',
                       'state', 'postcode', 'country'])
 

knimin/lib/mem_zip.py

@@ -60,6 +60,7 @@ def write_to_buffer(self):
         self.in_memory_zip.close()
         return self.in_memory_data.getvalue()
 
+
 if __name__ == "__main__":
     # Run a test
     imz = InMemoryZip()

knimin/lib/string_converter.py

@@ -59,4 +59,5 @@ def _convertToCamel(self, snake_cased_str, separator):
             camelCasedStr = components[0]
         return preffix + camelCasedStr + suffix
 
+
 converter = Converter()

knimin/lib/tests/test_configuration.py

@@ -47,6 +47,7 @@ def test_get_tornado(self):
         config = KniminConfig(self.config_fp)
         self.assertEqual(config.http_port, 8888)
 
+
 test_config = """[main]
 debug = True
 help_email = help@email.com

knimin/lib/tests/test_data_access.py

@@ -198,7 +198,8 @@ def test_getAGBarcodeDetails(self):
                'sample_date': datetime.date(2014, 8, 13),
                'date_of_last_email': datetime.date(2014, 8, 15),
                'other_text': 'REMOVED',
-               'site_sampled': 'Stool'}
+               'site_sampled': 'Stool',
+               'sample_environments': 'Human'}
         self.assertEqual(obs, exp)
 
     def test_get_barcode_info_by_kit_id(self):

knimin/lib/tests/test_geocoder.py

@@ -98,6 +98,7 @@ def test_geocode_bad_address(self):
                        None)
         self.assertEqual(obs, exp)
 
+
 # Results copied from Google API responses on 2015-10-25
 ok = '''{
    "results" : [
@@ -278,5 +279,6 @@ def test_geocode_bad_address(self):
    "status" : "INVALID_REQUEST"
 }'''
 
+
 if __name__ == '__main__':
     main()

knimin/lib/tests/test_mem_zip.py

@@ -50,5 +50,6 @@ def test_write_to_buffer(self):
         res_contents = zhandle.read(self.test_fname)
         self.assertEqual(res_contents, exp_contents)
 
+
 if __name__ == '__main__':
     unittest.main()

knimin/lib/tests/test_util.py

@@ -196,5 +196,6 @@ def test_fetch_url(self):
         self.assertTrue(isinstance(fetch_url('http://www.google.com'),
                                    StringIO))
 
+
 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_new_barcode.py

@@ -149,5 +149,6 @@ def test_post(self):
                       (num_barcodes, ", ".join(projects + [newProject])),
                       response.body)
 
+
 if __name__ == "__main__":
     main()

knimin/tests/test_ag_pulldown.py

@@ -108,5 +108,6 @@ def test_post(self):
         self.assertIn('failures.txt', response.body)
         self.assertIn('survey_1_md.txt', response.body)
 
+
 if __name__ == "__main__":
     main()

knimin/tests/test_ag_results_ready.py

@@ -52,5 +52,6 @@ def test_post(self):
         self.assertEqual(response.code, 200)
         self.assertIn('ERROR: No barcode results available', response.body)
 
+
 if __name__ == '__main__':
     main()

knimin/tests/test_ag_search.py

@@ -66,5 +66,6 @@ def test_post_barcode(self):
                         else:
                             self.assertIn(sample[field], response.body)
 
+
 if __name__ == '__main__':
     main()

knimin/tests/test_ag_stats.py

@@ -21,5 +21,6 @@ def test_stats_page(self):
             stat = '' if stat is None else stat
             self.assertIn('%s</td><td>%s' % (item, stat), response.body)
 
+
 if __name__ == '__main__':
     main()

knimin/tests/test_auth_handlers.py

@@ -72,5 +72,6 @@ def test_get(self):
         self.assertTrue(response.effective_url.endswith(
             '/login/?next=%2Fprojects%2Fsummary%2F'))
 
+
 if __name__ == "__main__":
     main()

knimin/tests/test_barcode_util.py

@@ -323,7 +323,8 @@ def test_get_ag_details_SJ(self):
                           'email_type': '1', 'name': 'REMOVED',
                           'sample_time': time(6, 50),
                           'notes': 'REMOVED',
-                          'email': 'REMOVED'})
+                          'email': 'REMOVED',
+                          'sample_environments': 'Human'})
 
         # check that None values are set to ''
         barcode = '000016744'

knimin/tests/test_consent_check.py

@@ -21,5 +21,6 @@ def test_post_no_consent(self):
         self.assertEqual(response.code, 200)
         self.assertIn('000004126</td><td style="color:red">', response.body)
 
+
 if __name__ == '__main__':
     main()