sane *_updated_timestamp field names

Author: melange396

integrations/acquisition/covidcast/test_csv_uploading.py

@@ -192,10 +192,10 @@ def apply_lag(expected_epidata):
     })
 
     # verify timestamps and default values are reasonable
-    self.cur.execute('select timestamp1, timestamp2, direction from covidcast')
-    for timestamp1, timestamp2, direction in self.cur:
-      self.assertGreater(timestamp1, 0)
-      self.assertEqual(timestamp2, 0)
+    self.cur.execute('select value_updated_timestamp, direction_updated_timestamp, direction from covidcast')
+    for value_updated_timestamp, direction_updated_timestamp, direction in self.cur:
+      self.assertGreater(value_updated_timestamp, 0)
+      self.assertEqual(direction_updated_timestamp, 0)
       self.assertIsNone(direction)
 
     # verify that the CSVs were archived

integrations/acquisition/covidcast/test_direction_updating.py

@@ -188,7 +188,7 @@ def test_uploading(self):
     })
 
     # verify secondary timestamps were updated
-    self.cur.execute('select timestamp2 from covidcast order by id asc')
+    self.cur.execute('select direction_updated_timestamp from covidcast order by id asc')
     timestamps = [t for (t,) in self.cur]
     for t in timestamps[:6]:
       # first 6 rows had `direction` updated

src/acquisition/covidcast/database.py

@@ -42,7 +42,7 @@ def __init__(self, source, signal, time_type, geo_type, time_value, geo_value, v
     self.value = value              # ...
     self.stderr = stderr            # ...
     self.sample_size = sample_size  # from CSV row
-    self.timestamp2 = 0
+    self.direction_updated_timestamp = 0
     self.direction = None
     self.issue = issue
     self.lag = lag
@@ -101,21 +101,21 @@ def insert_or_update_batch(self, cc_rows, batch_size=0, commit_partial=False):
     sql = '''
       INSERT INTO `covidcast`
         (`id`, `source`, `signal`, `time_type`, `geo_type`, `time_value`, `geo_value`,
-        `timestamp1`, `value`, `stderr`, `sample_size`,
-        `timestamp2`, `direction`,
+        `value_updated_timestamp`, `value`, `stderr`, `sample_size`,
+        `direction_updated_timestamp`, `direction`,
         `issue`, `lag`)
       VALUES
         (0, %s, %s, %s, %s, %s, %s,
         UNIX_TIMESTAMP(NOW()), %s, %s, %s,
         0, NULL,
         %s, %s)
       ON DUPLICATE KEY UPDATE
-        `timestamp1` = VALUES(`timestamp1`),
+        `value_updated_timestamp` = VALUES(`value_updated_timestamp`),
         `value` = VALUES(`value`),
         `stderr` = VALUES(`stderr`),
         `sample_size` = VALUES(`sample_size`)
     '''
-    # TODO: ^ do we want to reset `timestamp2` and `direction` in the duplicate key case?
+    # TODO: ^ do we want to reset `direction_updated_timestamp` and `direction` in the duplicate key case?
 
     # TODO: consider handling cc_rows as a generator instead of a list
     num_rows = len(cc_rows)
@@ -175,7 +175,7 @@ def insert_or_update(
       INSERT INTO `covidcast` VALUES
         (0, %s, %s, %s, %s, %s, %s, UNIX_TIMESTAMP(NOW()), %s, %s, %s, 0, NULL, %s, %s)
       ON DUPLICATE KEY UPDATE
-        `timestamp1` = VALUES(`timestamp1`),
+        `value_updated_timestamp` = VALUES(`value_updated_timestamp`),
         `value` = VALUES(`value`),
         `stderr` = VALUES(`stderr`),
         `sample_size` = VALUES(`sample_size`)
@@ -246,7 +246,7 @@ def update_direction(
       UPDATE
         `covidcast`
       SET
-        `timestamp2` = UNIX_TIMESTAMP(NOW()),
+        `direction_updated_timestamp` = UNIX_TIMESTAMP(NOW()),
         `direction` = %s
       WHERE
         `source` = %s AND
@@ -286,9 +286,9 @@ def get_all_record_values_of_timeseries_with_potentially_stale_direction(self, t
       `geo_type` varchar(12),
       `geo_value` varchar(12),
       `time_value` int(11),
-      `timestamp1` int(11),
+      `value_updated_timestamp` int(11),
       `value` double,
-      `timestamp2` int(11),
+      `direction_updated_timestamp` int(11),
       `direction` int(11),
       PRIMARY KEY(`id`)
     ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
@@ -305,9 +305,9 @@ def get_all_record_values_of_timeseries_with_potentially_stale_direction(self, t
       `geo_type`,
       `geo_value`,
       `time_value`,
-      `timestamp1`,
+      `value_updated_timestamp`,
       `value`,
-      `timestamp2`,
+      `direction_updated_timestamp`,
       `direction`
     FROM
     (
@@ -358,7 +358,7 @@ def get_all_record_values_of_timeseries_with_potentially_stale_direction(self, t
         `geo_type`,
         `geo_value`
       HAVING
-        MAX(`timestamp1`) > MIN(`timestamp2`)
+        MAX(`value_updated_timestamp`) > MIN(`direction_updated_timestamp`)
     '''
 
     # A query that selects rows of the time-series selected by stale_ts_key_sql query.
@@ -372,9 +372,9 @@ def get_all_record_values_of_timeseries_with_potentially_stale_direction(self, t
       `geo_type`,
       `geo_value`,
       `time_value`,
-      `timestamp1`,
+      `value_updated_timestamp`,
       `value`,
-      `timestamp2`,
+      `direction_updated_timestamp`,
       `direction`
     FROM ({stale_ts_key_sql}) AS t2
     LEFT JOIN `latest_issues` AS t3
@@ -424,8 +424,8 @@ def drop_temporary_table(self, tmp_table_name):
     sql = f'DROP TEMPORARY TABLE `{tmp_table_name}`;'
     self._cursor.execute(sql)
 
-  def update_timestamp2_from_temporary_table(self, tmp_table_name):
-    """Updates the `timestamp2` column of `covidcast` table for all the rows with id value in `tmp_table_name`.
+  def update_direction_updated_timestamp_from_temporary_table(self, tmp_table_name):
+    """Updates the `direction_updated_timestamp` column of `covidcast` table for all the rows with id value in `tmp_table_name`.
 
     `tmp_table_name`: name of the temporary table.
     """
@@ -437,7 +437,7 @@ def update_timestamp2_from_temporary_table(self, tmp_table_name):
       ON
         `covidcast`.id=t.id
       SET
-        `covidcast`.timestamp2=UNIX_TIMESTAMP(NOW())
+        `covidcast`.direction_updated_timestamp=UNIX_TIMESTAMP(NOW())
       '''
     self._cursor.execute(sql)
 
@@ -457,8 +457,8 @@ def get_keys_with_potentially_stale_direction(self):
         `signal`,
         `geo_type`,
         `geo_value`,
-        MAX(`timestamp1`) AS `max_timestamp1`,
-        MIN(`timestamp2`) AS `min_timestamp2`,
+        MAX(`value_updated_timestamp`) AS `max_value_updated_timestamp`,
+        MIN(`direction_updated_timestamp`) AS `min_direction_updated_timestamp`,
         MIN(`time_value`) AS `min_day`,
         MAX(`time_value`) AS `max_day`,
         COUNT(1) AS `series_length`
@@ -473,7 +473,7 @@ def get_keys_with_potentially_stale_direction(self):
         `geo_type`,
         `geo_value`
       HAVING
-        MAX(`timestamp1`) > MIN(`timestamp2`)
+        MAX(`value_updated_timestamp`) > MIN(`direction_updated_timestamp`)
     '''
 
     self._cursor.execute(sql)
@@ -488,8 +488,8 @@ def get_daily_timeseries_for_direction_update(
         DATEDIFF(`time_value`, %s) AS `offset`,
         `time_value` AS `day`,
         `value`,
-        `timestamp1`,
-        `timestamp2`
+        `value_updated_timestamp`,
+        `direction_updated_timestamp`
       FROM
         `covidcast`
       WHERE
@@ -507,9 +507,9 @@ def get_daily_timeseries_for_direction_update(
     self._cursor.execute(sql, args)
     return list(self._cursor)
 
-  def update_timeseries_timestamp2(
+  def update_timeseries_direction_updated_timestamp(
       self, source, signal, time_type, geo_type, geo_value):
-    """Update the `timestamp2` column for an entire time-series.
+    """Update the `direction_updated_timestamp` column for an entire time-series.
 
     For daily time-series, this implies that all `direction` values in the
     specified time-series are confirmed fresh as of the current time. Even if
@@ -523,7 +523,7 @@ def update_timeseries_timestamp2(
       UPDATE
         `covidcast`
       SET
-        `timestamp2` = UNIX_TIMESTAMP(NOW())
+        `direction_updated_timestamp` = UNIX_TIMESTAMP(NOW())
       WHERE
         `source` = %s AND
         `signal` = %s AND
@@ -552,7 +552,7 @@ def get_covidcast_meta(self):
         MAX(`value`) AS `max_value`,
         ROUND(AVG(`value`),7) AS `mean_value`,
         ROUND(STD(`value`),7) AS `stdev_value`,
-        MAX(`timestamp1`) AS `last_update`,
+        MAX(`value_updated_timestamp`) AS `last_update`,
         MAX(`issue`) as `max_issue`,
         MIN(`lag`) as `min_lag`,
         MAX(`lag`) as `max_lag`

src/acquisition/covidcast/direction.py

@@ -64,18 +64,18 @@ def scan_timeseries(
       offsets,
       days,
       values,
-      timestamp1s,
-      timestamp2s,
+      value_updated_timestamps,
+      direction_updated_timestamps,
       get_direction_impl):
     """Scan an entire time-series and return fresh direction updates.
 
     `offsets`: time offset, relative to any arbitrary fixed point in time, for
       each day in `days`
     `days`: day (YYYYMMDD) corresponding to each row in the other arrays
     `values`: value of the signal on each day
-    `timestamp1s`: primary timestamp for each row (i.e. when `value` was
+    `value_updated_timestamps`: primary timestamp for each row (i.e. when `value` was
       updated)
-    `timestamp2s`: secondary timestamp for each row (i.e. when `direction` was
+    `direction_updated_timestamps`: secondary timestamp for each row (i.e. when `direction` was
       last deemed to be fresh, relative to associated `value`s)
     `get_direction_impl`: a function which takes two arrays (time and value)
       and returns a classification of the direction (i.e. as -1, 0, +1)
@@ -100,8 +100,8 @@ def scan_timeseries(
         start += 1
 
       # check whether this row needs an update
-      direction_time = timestamp2s[end]
-      value_time = np.max(timestamp1s[start:end + 1])
+      direction_time = direction_updated_timestamps[end]
+      value_time = np.max(value_updated_timestamps[start:end + 1])
       if direction_time > value_time:
         # this row is fresh
         continue

src/acquisition/covidcast/direction_updater.py

@@ -79,8 +79,8 @@ def update_loop(database, direction_impl=Direction):
       signal,
       geo_type,
       geo_value,
-      max_timestamp1,
-      min_timestamp2,
+      max_value_updated_timestamp,
+      min_direction_updated_timestamp,
       min_day,
       max_day,
       series_length,
@@ -108,7 +108,7 @@ def update_loop(database, direction_impl=Direction):
         min_day,
         max_day,
         series_length,
-        max_timestamp1 - min_timestamp2,
+        max_value_updated_timestamp - min_direction_updated_timestamp,
       )
       print(msg % args)
 
@@ -118,12 +118,12 @@ def update_loop(database, direction_impl=Direction):
 
     # transpose result set and cast data types
     data = np.array(timeseries_rows)
-    offsets, days, values, timestamp1s, timestamp2s = data.T
+    offsets, days, values, value_updated_timestamps, direction_updated_timestamps = data.T
     offsets = offsets.astype(np.int64)
     days = days.astype(np.int64)
     values = values.astype(np.float64)
-    timestamp1s = timestamp1s.astype(np.int64)
-    timestamp2s = timestamp2s.astype(np.int64)
+    value_updated_timestamps = value_updated_timestamps.astype(np.int64)
+    direction_updated_timestamps = direction_updated_timestamps.astype(np.int64)
 
     # create a direction classifier for this signal
     data_stdev = data_stdevs[source][signal][geo_type]
@@ -135,7 +135,7 @@ def get_direction_impl(x, y):
 
     # recompute any stale directions
     days, directions = direction_impl.scan_timeseries(
-      offsets, days, values, timestamp1s, timestamp2s, get_direction_impl)
+      offsets, days, values, value_updated_timestamps, direction_updated_timestamps, get_direction_impl)
 
     if be_verbose:
       print(' computed %d direction updates' % len(directions))
@@ -148,7 +148,7 @@ def get_direction_impl(x, y):
         source, signal, 'day', geo_type, day, geo_value, direction)
 
     # mark the entire time-series as fresh with respect to direction
-    database.update_timeseries_timestamp2(
+    database.update_timeseries_direction_updated_timestamp(
       source, signal, 'day', geo_type, geo_value)
 
 
@@ -162,7 +162,7 @@ def optimized_update_loop(database, direction_impl=Direction):
 
   # A pandas DataFrame that will hold all rows from potentially stale time-series
   df_all = pd.DataFrame(columns=['id', 'source', 'signal', 'time_type', 'geo_type', 'geo_value', 'time_value',
-                                 'timestamp1', 'value', 'timestamp2', 'direction'],
+                                 'value_updated_timestamp', 'value', 'direction_updated_timestamp', 'direction'],
                         data=database.get_all_record_values_of_timeseries_with_potentially_stale_direction(
                           tmp_table_name))
   df_all.drop(columns=['time_type'], inplace=True)
@@ -220,15 +220,15 @@ def optimized_update_loop(database, direction_impl=Direction):
           ts_rows.time_value.min(),
           ts_rows.time_value.max(),
           len(ts_rows),
-          ts_rows.timestamp1.max() - ts_rows.timestamp2.min()
+          ts_rows.value_updated_timestamp.max() - ts_rows.direction_updated_timestamp.min()
       )
       print(msg % args)
 
     offsets = ts_rows.offsets.values.astype(np.int64)
     days = ts_rows.time_value.values.astype(np.int64)
     values = ts_rows.value.values.astype(np.float64)
-    timestamp1s = ts_rows.timestamp1.values.astype(np.int64)
-    timestamp2s = ts_rows.timestamp2.values.astype(np.int64)
+    value_updated_timestamps = ts_rows.value_updated_timestamp.values.astype(np.int64)
+    direction_updated_timestamps = ts_rows.direction_updated_timestamp.values.astype(np.int64)
 
     # create a direction classifier for this signal
     data_stdev = data_stdevs[source][signal][geo_type]
@@ -240,7 +240,7 @@ def get_direction_impl(x, y):
 
     # recompute any stale directions
     days, directions = direction_impl.scan_timeseries(
-      offsets, days, values, timestamp1s, timestamp2s, get_direction_impl)
+      offsets, days, values, value_updated_timestamps, direction_updated_timestamps, get_direction_impl)
 
     if be_verbose:
       print(' computed %d direction updates' % len(directions))
@@ -264,8 +264,8 @@ def get_direction_impl(x, y):
   for v, id_list in changed_rows.items():
     database.batched_update_direction(v, id_list)
 
-  # Updating timestamp2
-  database.update_timestamp2_from_temporary_table(tmp_table_name)
+  # Updating direction_updated_timestamp
+  database.update_direction_updated_timestamp_from_temporary_table(tmp_table_name)
   # Dropping temporary table
   database.drop_temporary_table(tmp_table_name)