Signal Documentation Coverage Endpoint

integrations/acquisition/covidcast/test_coverage_crossref_update.py

@@ -0,0 +1,105 @@
+"""Integration tests for the covidcast `geo_coverage` endpoint."""
+
+# standard library
+import json
+import unittest
+
+# third party
+import mysql.connector
+import requests
+
+# first party
+from delphi_utils import Nans
+from delphi.epidata.client.delphi_epidata import Epidata
+import delphi.operations.secrets as secrets
+import delphi.epidata.acquisition.covidcast.database as live
+from delphi.epidata.maintenance.coverage_crossref_updater import main
+from delphi.epidata.acquisition.covidcast.test_utils import CovidcastBase
+
+# use the local instance of the Epidata API
+BASE_URL = 'http://delphi_web_epidata/epidata' # NOSONAR
+
+
+class CoverageCrossrefTests(CovidcastBase):
+  """Tests coverage crossref updater."""
+
+  def localSetUp(self):
+    """Perform per-test setup."""
+    self._db._cursor.execute('TRUNCATE TABLE `coverage_crossref`')
+
+  @staticmethod
+  def _make_request(params=None):
+    if params is None:
+        params = {'geo': 'state:*'}
+    response = requests.get(f"{Epidata.BASE_URL}/covidcast/geo_coverage", params=params, auth=Epidata.auth)
+    response.raise_for_status()
+    return response.json()
+
+  def test_caching(self):
+    """Populate, query, cache, query, and verify the cache."""
+
+    # insert dummy data
+    self._db._cursor.execute('''
+      INSERT INTO `signal_dim` (`signal_key_id`, `source`, `signal`)
+      VALUES
+        (42, 'src', 'sig');
+    ''')
+    self._db._cursor.execute('''
+      INSERT INTO `geo_dim` (`geo_key_id`, `geo_type`, `geo_value`)
+      VALUES
+        (96, 'state', 'pa'),
+        (97, 'state', 'wa');
+    ''')
+    self._db._cursor.execute(f'''
+      INSERT INTO
+        `epimetric_latest` (`epimetric_id`, `signal_key_id`, `geo_key_id`, `time_type`,
+	      `time_value`, `value_updated_timestamp`,
+        `value`, `stderr`, `sample_size`,
+        `issue`, `lag`, `missing_value`,
+        `missing_stderr`,`missing_sample_size`)
+      VALUES
+        (15, 42, 96, 'day', 20200422,
+          123, 1, 2, 3, 20200422, 0, {Nans.NOT_MISSING}, {Nans.NOT_MISSING}, {Nans.NOT_MISSING}),
+        (16, 42, 97, 'day', 20200422,
+          789, 1, 2, 3, 20200423, 1, {Nans.NOT_MISSING}, {Nans.NOT_MISSING}, {Nans.NOT_MISSING})
+    ''')
+    self._db.commit()
+
+    results = self._make_request()
+
+    # make sure the tables are empty
+    self.assertEqual(results, {
+      'result': -2,
+      'epidata': [],
+      'message': 'no results',
+    })
+
+    # update the coverage crossref table
+    main()
+
+    results = self._make_request()
+
+    # make sure the data was actually served
+    self.assertEqual(results, {
+      'result': 1,
+      'epidata': [{'signal': 'sig', 'source': 'src'}],
+      'message': 'success',
+    })
+
+    results = self._make_request(params = {'geo': 'hrr:*'})
+
+    # make sure the tables are empty
+    self.assertEqual(results, {
+      'result': -2,
+      'epidata': [],
+      'message': 'no results',
+    })
+
+    results = self._make_request(params = {'geo': 'state:pa'})
+
+    # make sure the data was actually served
+    self.assertEqual(results, {
+      'result': 1,
+      'epidata': [{'signal': 'sig', 'source': 'src'}],
+      'message': 'success',
+    })

src/acquisition/covidcast/database.py

@@ -561,3 +561,36 @@ def retrieve_covidcast_meta_cache(self):
     for entry in cache:
       cache_hash[(entry['data_source'], entry['signal'], entry['time_type'], entry['geo_type'])] = entry
     return cache_hash
+
+  def compute_coverage_crossref(self):
+    """Compute coverage_crossref table, for looking up available signals per geo or vice versa."""
+
+    logger = get_structured_logger("compute_coverage_crossref")
+
+    coverage_crossref_delete_sql = '''
+      DELETE FROM coverage_crossref;
+    '''
+
+    coverage_crossref_update_sql = '''
+      INSERT INTO coverage_crossref (signal_key_id, geo_key_id, min_time_value, max_time_value)
+      SELECT
+          signal_key_id,
+          geo_key_id,
+          MIN(time_value) AS min_time_value,
+          MAX(time_value) AS max_time_value
+      FROM covid.epimetric_latest
+      GROUP BY signal_key_id, geo_key_id;
+    '''
+
+    self._connection.start_transaction()
+
+    self._cursor.execute(coverage_crossref_delete_sql)
+    logger.info("coverage_crossref_delete", rows=self._cursor.rowcount)
+
+    self._cursor.execute(coverage_crossref_update_sql)
+    logger.info("coverage_crossref_update", rows=self._cursor.rowcount)
+
+    self.commit()
+    logger.info("coverage_crossref committed")
+
+    return self._cursor.rowcount

src/ddl/v4_schema.sql

@@ -164,3 +164,24 @@ CREATE TABLE `covidcast_meta_cache` (
     PRIMARY KEY (`timestamp`)
 ) ENGINE=InnoDB;
 INSERT INTO covidcast_meta_cache VALUES (0, '[]');
+
+CREATE TABLE `coverage_crossref` (
+    `signal_key_id` bigint NOT NULL,
+    `geo_key_id` bigint NOT NULL,
+    `min_time_value` int NOT NULL,
+    `max_time_value` int NOT NULL,
+    UNIQUE INDEX coverage_crossref_geo_sig (`geo_key_id`, `signal_key_id`),
+    INDEX coverage_crossref_sig_geo (`signal_key_id`, `geo_key_id`)
+) ENGINE=InnoDB;
+
+CREATE OR REPLACE VIEW `coverage_crossref_v` AS
+SELECT
+    `sd`.`source`,
+    `sd`.`signal`,
+    `gd`.`geo_type`,
+    `gd`.`geo_value`,
+    `cc`.`min_time_value`,
+    `cc`.`max_time_value`
+FROM `coverage_crossref` `cc`
+JOIN `signal_dim` `sd` USING (`signal_key_id`)
+JOIN `geo_dim` `gd` USING (`geo_key_id`);

src/ddl/v4_schema_aliases.sql

@@ -8,3 +8,4 @@
 CREATE VIEW `epidata`.`epimetric_full_v`     AS SELECT * FROM `covid`.`epimetric_full_v`;
 CREATE VIEW `epidata`.`epimetric_latest_v`   AS SELECT * FROM `covid`.`epimetric_latest_v`;
 CREATE VIEW `epidata`.`covidcast_meta_cache` AS SELECT * FROM `covid`.`covidcast_meta_cache`;
+CREATE VIEW `epidata`.`coverage_crossref_v`  AS SELECT * FROM `covid`.`coverage_crossref_v`;

src/maintenance/coverage_crossref_updater.py

@@ -0,0 +1,44 @@
+"""Updates the table for the `coverage_crossref` endpoint."""
+
+# standard library
+import argparse
+import sys
+import time
+
+# first party
+from delphi.epidata.acquisition.covidcast.database import Database
+from delphi_utils import get_structured_logger
+from delphi.epidata.client.delphi_epidata import Epidata
+
+
+def main():
+  """Updates the table for the `coverage_crossref`."""
+
+  logger = get_structured_logger("coverage_crossref_updater")
+  start_time = time.time()
+  database = Database()
+  database.connect()
+
+  # compute and update coverage_crossref
+  try:
+    coverage = database.compute_coverage_crossref()
+  except:
+    # clean up before failing
+    database.disconnect(True)
+    raise
+
+  result = ("success",1)
+  if coverage==0:
+    result = ("no results",-2)
+
+  logger.info('coverage_crossref result: %s (code %d)' % result)
+
+
+  logger.info(
+      "Generated and updated covidcast geo/signal coverage",
+      total_runtime_in_seconds=round(time.time() - start_time, 2))
+  return True
+
+
+if __name__ == '__main__':
+  main()

src/server/endpoints/covidcast.py

@@ -542,6 +542,24 @@ def transform_row(row, proxy):
 
     return execute_query(q.query, q.params, fields_string, fields_int, [], transform=transform_row)
 
+@bp.route("/geo_coverage", methods=("GET", "POST"))
+def handle_geo_coverage():
+    """
+    For a specific geo returns the signal coverage (number of signals for a given geo_type)
+    """
+
+    geo_sets = parse_geo_sets()
+
+    q = QueryBuilder("coverage_crossref_v", "c")
+    fields_string = ["source", "signal"]
+
+    q.set_fields(fields_string)
+
+    q.apply_geo_filters("geo_type", "geo_value", geo_sets)
+    q.set_sort_order("source", "signal")
+    q.group_by = ["c." + field for field in fields_string] # this condenses duplicate results, similar to `SELECT DISTINCT`
+
+    return execute_query(q.query, q.params, fields_string, [], [])
 
 @bp.route("/anomalies", methods=("GET", "POST"))
 def handle_anomalies():

tests/acquisition/covidcast/test_database.py

@@ -78,6 +78,22 @@ def test_update_covidcast_meta_cache_query(self):
     self.assertIn('timestamp', sql)
     self.assertIn('epidata', sql)
 
+  def test_compute_coverage_crossref_query(self):
+    """Query to update the compute crossref looks sensible.
+
+    NOTE: Actual behavior is tested by integration test.
+    """
+
+    mock_connector = MagicMock()
+    database = Database()
+    database.connect(connector_impl=mock_connector)
+
+    database.compute_coverage_crossref()
+
+    connection = mock_connector.connect()
+    cursor = connection.cursor()
+    self.assertTrue(cursor.execute.called)
+
   def test_insert_or_update_batch_exception_reraised(self):
     """Test that an exception is reraised"""
     mock_connector = MagicMock()