Merge remote-tracking branch 'origin/main' into hhs-covidcast

Author: krivard

docs/api/covidcast-signals/fb-survey.md

@@ -324,8 +324,10 @@ also available. These have names beginning `smoothed_w`, such as
 
 | Signal | Description | Survey Item |
 | --- | --- | --- |
+| `smoothed_covid_vaccinated_or_accept` | Estimated percentage of respondents who *either* have already received a COVID vaccine *or* would definitely or probably choose to get vaccinated, if a vaccine were offered to them today. <br/> **Earliest date available:** 2021-01-06 | V1 and V3 |
 | `smoothed_accept_covid_vaccine` | Estimated percentage of respondents who would definitely or probably choose to get vaccinated, if a COVID-19 vaccine were offered to them today. **Note:** Until January 6, 2021, all respondents answered this question; beginning on that date, only respondents who said they have not received a COVID vaccine are asked this question. <br/> **Earliest date available:** 2021-01-01 | V3 |
 | `smoothed_covid_vaccinated` | Estimated percentage of respondents who have already received a vaccine for COVID-19. **Note:** The Centers for Disease Control compiles data on vaccine administration across the United States. This signal may differ from CDC data because of survey biases and should not be treated as authoritative. However, the survey signal is not subject to the lags and reporting problems in official vaccination data. <br/> **Earliest date available:** 2021-01-06 | V1 |
+| `smoothed_worried_vaccine_side_effects` | Estimated percentage of respondents who are very or moderately concerned that they would "experience a side effect from a COVID-19 vaccination." (Asked of all respondents, including those who have already received one or more doses of a COVID-19 vaccine.) <br/> **Earliest date available:** 2021-01-12 | V9 |
 | `smoothed_vaccine_likely_friends` | Estimated percentage of respondents who would be more likely to get a COVID-19 vaccine if it were recommended to them by friends and family, among respondents who have not yet been vaccinated. <br/> **Earliest date available:** 2021-01-20 | V4 |
 | `smoothed_vaccine_likely_local_health` | Estimated percentage of respondents who would be more likely to get a COVID-19 vaccine if it were recommended to them by local health workers, among respondents who have not yet been vaccinated. <br/> **Earliest date available:** 2021-01-20 | V4 |
 | `smoothed_vaccine_likely_who` | Estimated percentage of respondents who would be more likely to get a COVID-19 vaccine if it were recommended to them by the World Health Organization, among respondents who have not yet been vaccinated. <br/> **Earliest date available:** 2021-01-20 | V4 |
@@ -335,9 +337,9 @@ also available. These have names beginning `smoothed_w`, such as
 These indicators are based on questions added in Wave 6 of the survey,
 introduced on December 19, 2020; however, Delphi only enabled item V1 beginning
 January 6, 2021. **Note:** As of January 2021, vaccination items on the survey
-are being revised in preparation for Wave 7. We may replace the signals above
-with new signals (with different names) if the underlying survey items change
-significantly.
+are being revised in preparation for later waves. We may replace the signals
+above with new signals (with different names) if the underlying survey items
+change significantly.
 
 Weighted versions of these signals, using the [survey weighting described
 below](#survey-weighting) to be more representative of state demographics, are

integrations/acquisition/covidcast_nowcast/test_csv_uploading.py

@@ -68,9 +68,9 @@ def test_uploading(self):
     # print full diff if something unexpected comes out
     self.maxDiff=None
 
-    receiving_dir = '/common/covidcast_nowcast/receiving/src/'
-    success_dir = '/common/covidcast_nowcast/archive/successful/src/'
-    failed_dir = '/common/covidcast_nowcast/archive/failed/src/'
+    receiving_dir = '/common/covidcast_nowcast/receiving/issue_20200421/src/'
+    success_dir = '/common/covidcast_nowcast/archive/successful/issue_20200421/src/'
+    failed_dir = '/common/covidcast_nowcast/archive/failed/issue_20200421/src/'
     os.makedirs(receiving_dir, exist_ok=True)
 
     # valid
@@ -125,7 +125,7 @@ def test_duplicate_row(self):
     # print full diff if something unexpected comes out
     self.maxDiff=None
 
-    receiving_dir = '/common/covidcast_nowcast/receiving/src/'
+    receiving_dir = '/common/covidcast_nowcast/receiving/issue_20200425/src/'
     os.makedirs(receiving_dir, exist_ok=True)
 
     with open(receiving_dir + '20200419_state_sig.csv', 'w') as f:
@@ -146,8 +146,8 @@ def test_duplicate_row(self):
         'time_value': 20200419,
         'geo_value': 'ca',
         'value': 2,
-        'issue': 20200421,
-        'lag': 2,
+        'issue': 20200425,
+        'lag': 6,
         'signal': 'sig',
       }],
       'message': 'success',

integrations/client/test_delphi_epidata.py

@@ -471,3 +471,48 @@ def test_covidcast_nowcast(self):
       'src', 'sig1', 'sensor', 'day', 'county', 22222222, '01001')
 
     self.assertEqual(response, {'result': -2, 'message': 'no results'})
+
+  def test_async_epidata(self):
+    # insert dummy data
+    self.cur.execute('''
+      insert into covidcast values
+        (0, 'src', 'sig', 'day', 'county', 20200414, '11111',
+          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, False),
+        (0, 'src', 'sig', 'day', 'county', 20200414, '33333',
+          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, False),
+        (0, 'src', 'sig', 'day', 'msa', 20200414, '22222',
+          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, False)
+    ''')
+    self.cnx.commit()
+    test_output = Epidata.async_epidata([
+      {
+        'source': 'covidcast',
+        'data_source': 'src',
+        'signals': 'sig',
+        'time_type': 'day',
+        'geo_type': 'county',
+        'geo_value': '11111',
+        'time_values': '20200414'
+      },
+      {
+        'source': 'covidcast',
+        'data_source': 'src',
+        'signals': 'sig',
+        'time_type': 'day',
+        'geo_type': 'county',
+        'geo_value': '00000',
+        'time_values': '20200414'
+      }
+    ], batch_size=10)
+    responses = [i[0] for i in test_output]*12
+    # check response is same as standard covidcast call, using 24 calls to test batch sizing
+    self.assertEqual(responses,
+                     [Epidata.covidcast('src', 'sig', 'day', 'county', 20200414, '11111'),
+                      Epidata.covidcast('src', 'sig', 'day', 'county', 20200414, '00000')]*12
+                     )

src/acquisition/covidcast_nowcast/load_sensors.py

@@ -1,4 +1,5 @@
 from shutil import move
+from datetime import datetime
 import os
 import time
 
@@ -36,7 +37,7 @@ def main(csv_path: str = SENSOR_CSV_PATH) -> None:
     """
     user, pw = secrets.db.epi
     engine = sqlalchemy.create_engine(f"mysql+pymysql://{user}:{pw}@{secrets.db.host}/{DB_NAME}")
-    for filepath, attribute in CsvImporter.find_csv_files(csv_path):
+    for filepath, attribute in CsvImporter.find_issue_specific_csv_files(csv_path):
         if attribute is None:
             _move_after_processing(filepath, success=False)
             continue

src/client/delphi_epidata.py

@@ -10,7 +10,10 @@
 
 # External modules
 import requests
+import asyncio
+import warnings
 
+from aiohttp import ClientSession
 from pkg_resources import get_distribution, DistributionNotFound
 
 # Obtain package version for the user-agent. Uses the installed version by
@@ -704,3 +707,29 @@ def covidcast_nowcast(
 
     # Make the API call
     return Epidata._request(params)
+
+  @staticmethod
+  def async_epidata(param_list, batch_size=100):
+    """Make asynchronous Epidata calls for a list of parameters."""
+    async def async_get(params, session):
+      """Helper function to make Epidata GET requests."""
+      async with session.get(Epidata.BASE_URL, params=params) as response:
+        return await response.json(), params
+
+    async def async_make_calls(param_combos):
+      """Helper function to asynchronously make and aggregate Epidata GET requests."""
+      tasks = []
+      async with ClientSession() as session:
+        for param in param_combos:
+          task = asyncio.ensure_future(async_get(param, session))
+          tasks.append(task)
+        responses = await asyncio.gather(*tasks)
+        return responses
+
+    batches = [param_list[i:i+batch_size] for i in range(0, len(param_list), batch_size)]
+    responses = []
+    for batch in batches:
+      loop = asyncio.get_event_loop()
+      future = asyncio.ensure_future(async_make_calls(batch))
+      responses += loop.run_until_complete(future)
+    return responses

src/client/packaging/pypi/setup.py

@@ -14,6 +14,7 @@
   url='https://github.com/cmu-delphi/delphi-epidata',
   packages=setuptools.find_packages(),
   install_requires=[
+    'aiohttp'
     'requests>=2.7.0',
   ],
   classifiers=[

tests/acquisition/covidcast_nowcast/test_load_sensors.py

@@ -24,9 +24,9 @@ def test_load_and_prepare_file(self):
                        "test_signal",
                        "test_time_type",
                        "test_geo_type",
-                       "test_time_value",
-                       "test_issue_value",
-                       "test_lag_value")
+                       20201231,
+                       20210102,
+                       3)
 
     test_df = load_and_prepare_file(StringIO("sensor_name,geo_value,value\ntestname,01001,1.5"), test_attributes)
     pd.testing.assert_frame_equal(test_df,
@@ -37,8 +37,8 @@ def test_load_and_prepare_file(self):
                                                 "signal": ["test_signal"],
                                                 "time_type": ["test_time_type"],
                                                 "geo_type": ["test_geo_type"],
-                                                "time_value": ["test_time_value"],
-                                                "issue": ["test_issue_value"],
-                                                "lag": ["test_lag_value"],
+                                                "time_value": [20201231],
+                                                "issue": [20210102],
+                                                "lag": [3],
                                                 "value_updated_timestamp": [12345]})
                                   )