Merge pull request #167 from melange396/meta_cache_breakdown

Meta cache breakdown

Author: krivard

integrations/acquisition/covidcast/test_covidcast_meta_caching.py

@@ -66,14 +66,14 @@ def test_caching(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'state', 20200422, 'pa',
-          123, 1, 2, 3, 456, 1, 20200422, 0, 1),
+          123, 1, 2, 3, 456, 1, 20200422, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200422, 'wa',
-          789, 1, 2, 3, 456, 1, 20200423, 1, 1)
+          789, 1, 2, 3, 456, 1, 20200423, 1, 1, False)
     ''')
     self.cur.execute('''
       insert into covidcast values
         (100, 'src', 'wip_sig', 'day', 'state', 20200422, 'pa',
-          456, 4, 5, 6, 789, -1, 20200422, 0, 1)
+          456, 4, 5, 6, 789, -1, 20200422, 0, 1, True)
     ''')
 
     self.cnx.commit()

integrations/acquisition/covidcast/test_direction_updating.py

@@ -63,29 +63,29 @@ def test_uploading(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig1', 'day', 'state', 20201028, '1111',
-          123, 2, 0, 0, 0, -1, 20201028, 0, 1),
+          123, 2, 0, 0, 0, -1, 20201028, 0, 1, False),
         (0, 'src', 'sig1', 'day', 'state', 20201029, '1111',
-          123, 6, 0, 0, 0, 0, 20201029, 0, 1),
+          123, 6, 0, 0, 0, 0, 20201029, 0, 1, False),
         (0, 'src', 'sig1', 'day', 'state', 20201030, '1111',
-          123, 5, 0, 0, 0, 1, 20201030, 0, 1),  
+          123, 5, 0, 0, 0, 1, 20201030, 0, 1, False),  
         (0, 'src', 'sig', 'day', 'state', 20200228, 'ca',
-          123, 2, 0, 0, 0, NULL, 20200228, 0, 1),
+          123, 2, 0, 0, 0, NULL, 20200228, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200229, 'ca',
-          123, 6, 0, 0, 0, NULL, 20200229, 0, 1),
+          123, 6, 0, 0, 0, NULL, 20200229, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200301, 'ca',
-          123, 5, 0, 0, 0, NULL, 20200301, 0, 1),
+          123, 5, 0, 0, 0, NULL, 20200301, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200511, 'fl',
-          123, 1, 0, 0, 0, NULL, 20200511, 0, 1),
+          123, 1, 0, 0, 0, NULL, 20200511, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200512, 'fl',
-          123, 2, 0, 0, 0, NULL, 20200512, 0, 1),
+          123, 2, 0, 0, 0, NULL, 20200512, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200517, 'fl',
-          123, 2, 0, 0, 0, NULL, 20200517, 0, 1),
+          123, 2, 0, 0, 0, NULL, 20200517, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200615, 'tx',
-          123, 9, 0, 0, 456, NULL, 20200615, 0, 1),
+          123, 9, 0, 0, 456, NULL, 20200615, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200616, 'tx',
-          123, 5, 0, 0, 456, NULL, 20200616, 0, 1),
+          123, 5, 0, 0, 456, NULL, 20200616, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200617, 'tx',
-          123, 1, 0, 0, 456, 1, 20200617, 0, 1)
+          123, 1, 0, 0, 456, 1, 20200617, 0, 1, False)
     ''')
     self.cnx.commit()
 

integrations/acquisition/covidcast/test_fill_is_latest_issue.py

@@ -53,19 +53,19 @@ def test_fill_is_latest_issue(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'state', 20200228, 'ca',
-          123, 2, 5, 5, 5, NULL, 20200228, 0, 1),
+          123, 2, 5, 5, 5, NULL, 20200228, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200228, 'ca',
-          123, 2, 0, 0, 0, NULL, 20200229, 1, 1),
+          123, 2, 0, 0, 0, NULL, 20200229, 1, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200229, 'ca',
-          123, 6, 0, 0, 0, NULL, 20200301, 1, 1),
+          123, 6, 0, 0, 0, NULL, 20200301, 1, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200229, 'ca',
-          123, 6, 9, 9, 9, NULL, 20200229, 0, 1),
+          123, 6, 9, 9, 9, NULL, 20200229, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200301, 'ca',
-          123, 5, 0, 0, 0, NULL, 20200303, 2, 1),
+          123, 5, 0, 0, 0, NULL, 20200303, 2, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200301, 'ca',
-          123, 5, 5, 5, 5, NULL, 20200302, 1, 1),
+          123, 5, 5, 5, 5, NULL, 20200302, 1, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200301, 'ca',
-          123, 5, 9, 8, 7, NULL, 20200301, 0, 1)
+          123, 5, 9, 8, 7, NULL, 20200301, 0, 1, False)
     ''')
     self.cnx.commit()
 
@@ -76,19 +76,19 @@ def test_fill_is_latest_issue(self):
     result = list(self.cur)
     expected = [
         (1, 'src', 'sig', 'day', 'state', 20200228, 'ca',
-          123, 2, 5, 5, 5, None, 20200228, 0, bytearray(b'0')),
+          123, 2, 5, 5, 5, None, 20200228, 0, bytearray(b'0'), bytearray(b'0')),
         (2, 'src', 'sig', 'day', 'state', 20200228, 'ca',
-          123, 2, 0, 0, 0, None, 20200229, 1, bytearray(b'1')),
+          123, 2, 0, 0, 0, None, 20200229, 1, bytearray(b'1'), bytearray(b'0')),
         (3, 'src', 'sig', 'day', 'state', 20200229, 'ca',
-          123, 6, 0, 0, 0, None, 20200301, 1, bytearray(b'1')),
+          123, 6, 0, 0, 0, None, 20200301, 1, bytearray(b'1'), bytearray(b'0')),
         (4, 'src', 'sig', 'day', 'state', 20200229, 'ca',
-          123, 6, 9, 9, 9, None, 20200229, 0, bytearray(b'0')),
+          123, 6, 9, 9, 9, None, 20200229, 0, bytearray(b'0'), bytearray(b'0')),
         (5, 'src', 'sig', 'day', 'state', 20200301, 'ca',
-          123, 5, 0, 0, 0, None, 20200303, 2, bytearray(b'1')),
+          123, 5, 0, 0, 0, None, 20200303, 2, bytearray(b'1'), bytearray(b'0')),
         (6, 'src', 'sig', 'day', 'state', 20200301, 'ca',
-          123, 5, 5, 5, 5, None, 20200302, 1, bytearray(b'0')),
+          123, 5, 5, 5, 5, None, 20200302, 1, bytearray(b'0'), bytearray(b'0')),
         (7, 'src', 'sig', 'day', 'state', 20200301, 'ca',
-          123, 5, 9, 8, 7, None, 20200301, 0, bytearray(b'0'))
+          123, 5, 9, 8, 7, None, 20200301, 0, bytearray(b'0'), bytearray(b'0'))
     ]
 
     self.assertEqual(result, expected)

integrations/client/test_delphi_epidata.py

@@ -50,11 +50,11 @@ def test_covidcast(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0, 0),
+          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0, 0, False),
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          456, 5.5, 1.2, 10.5, 789, 0, 20200415, 1, 0),
+          456, 5.5, 1.2, 10.5, 789, 0, 20200415, 1, 0, False),
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          345, 6.5, 2.2, 11.5, 678, 0, 20200416, 2, 1)
+          345, 6.5, 2.2, 11.5, 678, 0, 20200416, 2, 1, False)
     ''')
     self.cnx.commit()
 
@@ -157,11 +157,11 @@ def test_covidcast_meta(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0, 0),
+          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0, 0, False),
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          345, 6.0, 2.2, 11.5, 678, 0, 20200416, 2, 1),
+          345, 6.0, 2.2, 11.5, 678, 0, 20200416, 2, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200415, '01234',
-          345, 7.0, 2.0, 12.5, 678, 0, 20200416, 1, 1)
+          345, 7.0, 2.0, 12.5, 678, 0, 20200416, 1, 1, False)
     ''')
     self.cnx.commit()
 

integrations/server/test_covidcast.py

@@ -45,7 +45,7 @@ def test_round_trip(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0, 1)
+          123, 1.5, 2.5, 3.5, 456, 4, 20200414, 0, 1, False)
     ''')
     self.cnx.commit()
 
@@ -85,17 +85,17 @@ def test_location_wildcard(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '11111',
-          123, 10, 11, 12, 456, 13, 20200414, 0, 1),
+          123, 10, 11, 12, 456, 13, 20200414, 0, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200414, '22222',
-          123, 20, 21, 22, 456, 23, 20200414, 0, 1),
+          123, 20, 21, 22, 456, 23, 20200414, 0, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200414, '33333',
-          123, 30, 31, 32, 456, 33, 20200414, 0, 1),
+          123, 30, 31, 32, 456, 33, 20200414, 0, 1, False),
         (0, 'src', 'sig', 'day', 'msa', 20200414, '11111',
-          123, 40, 41, 42, 456, 43, 20200414, 0, 1),
+          123, 40, 41, 42, 456, 43, 20200414, 0, 1, False),
         (0, 'src', 'sig', 'day', 'msa', 20200414, '22222',
-          123, 50, 51, 52, 456, 53, 20200414, 0, 1),
+          123, 50, 51, 52, 456, 53, 20200414, 0, 1, False),
         (0, 'src', 'sig', 'day', 'msa', 20200414, '33333',
-          123, 60, 61, 62, 456, 634, 20200414, 0, 1)
+          123, 60, 61, 62, 456, 634, 20200414, 0, 1, False)
     ''')
     self.cnx.commit()
 
@@ -155,17 +155,17 @@ def test_location_timeline(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200411, '01234',
-          123, 10, 11, 12, 456, 13, 20200413, 2, 1),
+          123, 10, 11, 12, 456, 13, 20200413, 2, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200412, '01234',
-          123, 20, 21, 22, 456, 23, 20200413, 1, 1),
+          123, 20, 21, 22, 456, 23, 20200413, 1, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200413, '01234',
-          123, 30, 31, 32, 456, 33, 20200413, 0, 1),
+          123, 30, 31, 32, 456, 33, 20200413, 0, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200411, '11111',
-          123, 40, 41, 42, 456, 43, 20200413, 2, 1),
+          123, 40, 41, 42, 456, 43, 20200413, 2, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200412, '22222',
-          123, 50, 51, 52, 456, 53, 20200413, 1, 1),
+          123, 50, 51, 52, 456, 53, 20200413, 1, 1, False),
         (0, 'src', 'sig', 'day', 'county', 20200413, '33333',
-          123, 60, 61, 62, 456, 63, 20200413, 0, 1)
+          123, 60, 61, 62, 456, 63, 20200413, 0, 1, False)
     ''')
     self.cnx.commit()
 
@@ -225,7 +225,7 @@ def test_unique_key_constraint(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          0, 0, 0, 0, 0, 0, 20200414, 0, 1)
+          0, 0, 0, 0, 0, 0, 20200414, 0, 1, False)
     ''')
     self.cnx.commit()
 
@@ -234,14 +234,14 @@ def test_unique_key_constraint(self):
       self.cur.execute('''
         insert into covidcast values
           (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-            1, 1, 1, 1, 1, 1, 20200414, 0, 1)
+            1, 1, 1, 1, 1, 1, 20200414, 0, 1, False)
       ''')
 
     # succeed to insert different dummy data under a different issue
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          1, 1, 1, 1, 1, 1, 20200415, 1, 1)
+          1, 1, 1, 1, 1, 1, 20200415, 1, 1, False)
     ''')
 
   def test_nullable_columns(self):
@@ -251,7 +251,7 @@ def test_nullable_columns(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'day', 'county', 20200414, '01234',
-          123, 0.123, NULL, NULL, 456, NULL, 20200414, 0, 1)
+          123, 0.123, NULL, NULL, 456, NULL, 20200414, 0, 1, False)
     ''')
     self.cnx.commit()
 
@@ -291,15 +291,15 @@ def test_temporal_partitioning(self):
     self.cur.execute('''
       insert into covidcast values
         (0, 'src', 'sig', 'hour', 'state', 2020041714, 'vi',
-          123, 10, 11, 12, 456, 13, 2020041714, 0, 1),
+          123, 10, 11, 12, 456, 13, 2020041714, 0, 1, False),
         (0, 'src', 'sig', 'day', 'state', 20200417, 'vi',
-          123, 20, 21, 22, 456, 23, 20200417, 00, 1),
+          123, 20, 21, 22, 456, 23, 20200417, 00, 1, False),
         (0, 'src', 'sig', 'week', 'state', 202016, 'vi',
-          123, 30, 31, 32, 456, 33, 202016, 0, 1),
+          123, 30, 31, 32, 456, 33, 202016, 0, 1, False),
         (0, 'src', 'sig', 'month', 'state', 202004, 'vi',
-          123, 40, 41, 42, 456, 43, 202004, 0, 1),
+          123, 40, 41, 42, 456, 43, 202004, 0, 1, False),
         (0, 'src', 'sig', 'year', 'state', 2020, 'vi',
-          123, 50, 51, 52, 456, 53, 2020, 0, 1)
+          123, 50, 51, 52, 456, 53, 2020, 0, 1, False)
     ''')
     self.cnx.commit()
 

integrations/server/test_covidcast_meta.py

@@ -28,6 +28,7 @@ def setUp(self):
         database='epidata')
     cur = cnx.cursor()
     cur.execute('truncate table covidcast')
+    cur.execute('update covidcast_meta_cache set timestamp = 0, epidata = ""')
     cnx.commit()
     cur.close()
 
@@ -46,7 +47,7 @@ def test_round_trip(self):
     # insert dummy data and accumulate expected results (in sort order)
     template = '''
       insert into covidcast values
-        (0, "%s", "%s", "%s", "%s", %d, "%s", 123, %d, 0, 0, 456, 0, %d, 0, 1)
+        (0, "%s", "%s", "%s", "%s", %d, "%s", 123, %d, 0, 0, 456, 0, %d, 0, 1, %d)
     '''
     expected = []
     for src in ('src1', 'src2'):
@@ -72,7 +73,7 @@ def test_round_trip(self):
             })
             for tv in (1, 2):
               for gv, v in zip(('geo1', 'geo2'), (10, 20)):
-                self.cur.execute(template % (src, sig, tt, gt, tv, gv, v, tv))
+                self.cur.execute(template % (src, sig, tt, gt, tv, gv, v, tv, False))
     self.cnx.commit()
     update_cache(args=None)
 
@@ -94,14 +95,18 @@ def test_suppress_work_in_progress(self):
     # insert dummy data and accumulate expected results (in sort order)
     template = '''
       insert into covidcast values
-        (0, "%s", "%s", "%s", "%s", %d, "%s", 123, %d, 0, 0, 456, 0, %d, 0, 1)
+        (0, "%s", "%s", "%s", "%s", %d, "%s", 123, %d, 0, 0, 456, 0, %d, 0, 1, %d)
     '''
     expected = []
     for src in ('src1', 'src2'):
       for sig in ('sig1', 'sig2', 'wip_sig3'):
         for tt in ('day', 'week'):
           for gt in ('hrr', 'msa'):
-            if sig != 'wip_sig3':
+
+            if sig == 'wip_sig3':
+              is_wip = True
+            else:
+              is_wip = False
               expected.append({
                 'data_source': src,
                 'signal': sig,
@@ -121,7 +126,7 @@ def test_suppress_work_in_progress(self):
               })
             for tv in (1, 2):
               for gv, v in zip(('geo1', 'geo2'), (10, 20)):
-                self.cur.execute(template % (src, sig, tt, gt, tv, gv, v, tv))
+                self.cur.execute(template % (src, sig, tt, gt, tv, gv, v, tv, is_wip))
     self.cnx.commit()
     update_cache(args=None)
 

src/acquisition/covidcast/covidcast_meta_cache_updater.py

@@ -2,7 +2,6 @@
 
 # standard library
 import argparse
-import json
 import sys
 
 # first party
@@ -43,12 +42,9 @@ def main(args, epidata_impl=Epidata, database_impl=Database):
     print('unable to cache epidata')
     return False
 
-  # serialize the data
-  epidata_json = json.dumps(metadata)
-
   # update the cache
   try:
-    database.update_covidcast_meta_cache(epidata_json)
+    database.update_covidcast_meta_cache(metadata)
     print('successfully cached epidata')
   finally:
     # no catch block so that an exception above will cause the program to

src/acquisition/covidcast/csv_to_database.py

@@ -74,11 +74,16 @@ def archive_as_successful(path_src, filename, source):
 
     (source, signal, time_type, geo_type, time_value, issue, lag) = details
 
+    is_wip = False
+    if signal[:4].lower() == "wip_":
+      is_wip = True
+    print(signal, is_wip)
+
     csv_rows = csv_importer_impl.load_csv(path, geo_type)
 
     all_rows_valid = False
     try:
-      cc_rows = CovidcastRow.fromCsvRows(csv_rows, source, signal, time_type, geo_type, time_value, issue, lag)
+      cc_rows = CovidcastRow.fromCsvRows(csv_rows, source, signal, time_type, geo_type, time_value, issue, lag, is_wip)
       rows_list = list(cc_rows)
       if not rows_list:
         raise ValueError("No data")