aligned preprocessing to match current & rollback write for consistent result

Author: aysim319

doctor_visits/delphi_doctor_visits/process_data.py

@@ -81,32 +81,54 @@ def format_df(df: pd.DataFrame, geo_id: str, se: bool, logger):
     return df
 
 
-def write_to_csv(output_df: pd.DataFrame, prefix: str, geo_id: str, weekday: bool, se: bool, logger, output_path="."):
+def write_to_csv(
+    output_df: pd.DataFrame, prefix: str, geo_level: str, weekday: bool, se: bool, logger, output_path="."
+):
     """
     Write sensor values to csv.
 
     Args:
       output_dict: dictionary containing sensor rates, se, unique dates, and unique geo_id
-      geo_id: geographic resolution, one of ["county", "state", "msa", "hrr", "nation", "hhs"]
+      geo_level: geographic resolution, one of ["county", "state", "msa", "hrr", "nation", "hhs"]
       se: boolean to write out standard errors, if true, use an obfuscated name
       out_name: name of the output file
       output_path: outfile path to write the csv (default is current directory)
     """
-    out_name = format_outname(prefix, se, weekday)
-    filtered_df = format_df(output_df, geo_id, se, logger)
+    # out_name = format_outname(prefix, se, weekday)
 
+    # write out results
+    out_name = "smoothed_adj_cli" if weekday else "smoothed_cli"
     if se:
-        logger.info(f"========= WARNING: WRITING SEs TO {out_name} =========")
+        assert prefix is not None, "template has no obfuscated prefix"
+        out_name = prefix + "_" + out_name
 
-    dates = set(list(output_df["date"]))
-    grouped = filtered_df.groupby("date")
-    for d in dates:
-        filename = "%s/%s_%s_%s.csv" % (output_path, (d + Config.DAY_SHIFT).strftime("%Y%m%d"), geo_id, out_name)
-        single_date_df = grouped.get_group(d)
-        single_date_df = single_date_df.drop(columns=["date"])
-        single_date_df.to_csv(filename, index=False, na_rep="NA")
+    if se:
+        logger.info(f"========= WARNING: WRITING SEs TO {out_name} =========")
 
-        logger.debug(f"wrote {len(single_date_df)} rows for {geo_id}")
+    out_n = 0
+    for d in set(output_df["date"]):
+        filename = "%s/%s_%s_%s.csv" % (output_path, (d + Config.DAY_SHIFT).strftime("%Y%m%d"), geo_level, out_name)
+        single_date_df = output_df[output_df["date"] == d]
+        with open(filename, "w") as outfile:
+            outfile.write("geo_id,val,se,direction,sample_size\n")
+
+            for line in single_date_df.itertuples():
+                geo_id = line.geo_id
+                sensor = 100 * line.val  # report percentages
+                se_val = 100 * line.se
+                assert not np.isnan(sensor), "sensor value is nan, check pipeline"
+                assert sensor < 90, f"strangely high percentage {geo_level, sensor}"
+                if not np.isnan(se_val):
+                    assert se_val < 5, f"standard error suspiciously high! investigate {geo_level}"
+
+                if se:
+                    assert sensor > 0 and se_val > 0, "p=0, std_err=0 invalid"
+                    outfile.write("%s,%f,%s,%s,%s\n" % (geo_id, sensor, se_val, "NA", "NA"))
+                else:
+                    # for privacy reasons we will not report the standard error
+                    outfile.write("%s,%f,%s,%s,%s\n" % (geo_id, sensor, "NA", "NA", "NA"))
+                out_n += 1
+    logger.debug(f"wrote {out_n} rows for {geo_level}")
 
 
 def csv_to_df(filepath: str, startdate: datetime, enddate: datetime, dropdate: datetime, logger) -> pd.DataFrame:
@@ -131,29 +153,29 @@ def csv_to_df(filepath: str, startdate: datetime, enddate: datetime, dropdate: d
         dtype=Config.DTYPES,
         blocksize=None,
     )
-
-    ddata = ddata.dropna()
     # rename inconsistent column names to match config column names
     ddata = ddata.rename(columns=Config.DEVIANT_COLS_MAP)
-
     ddata = ddata[Config.FILT_COLS]
-    ddata[Config.DATE_COL] = dd.to_datetime(ddata[Config.DATE_COL])
-
-    # restrict to training start and end date
-    startdate = startdate - Config.DAY_SHIFT
 
-    assert startdate > Config.FIRST_DATA_DATE, "Start date <= first day of data"
-    assert startdate < enddate, "Start date >= end date"
-    assert enddate <= dropdate, "End date > drop date"
+    ddata = ddata.dropna()
 
-    date_filter = (ddata[Config.DATE_COL] >= Config.FIRST_DATA_DATE) & (ddata[Config.DATE_COL] < dropdate)
+    ddata[Config.DATE_COL] = dd.to_datetime(ddata[Config.DATE_COL])
 
-    df = ddata[date_filter].compute()
+    df = ddata.compute()
 
     # aggregate age groups (so data is unique by service date and FIPS)
     df = df.groupby([Config.DATE_COL, Config.GEO_COL]).sum(numeric_only=True).reset_index()
     assert np.sum(df.duplicated()) == 0, "Duplicates after age group aggregation"
     assert (df[Config.COUNT_COLS] >= 0).all().all(), "Counts must be nonnegative"
 
+    # restrict to training start and end date
+    startdate = startdate - Config.DAY_SHIFT
+
+    assert startdate > Config.FIRST_DATA_DATE, "Start date <= first day of data"
+    assert startdate < enddate, "Start date >= end date"
+    assert enddate <= dropdate, "End date > drop date"
+
+    date_filter = (df[Config.DATE_COL] >= Config.FIRST_DATA_DATE) & (df[Config.DATE_COL] < dropdate)
+    df = df[date_filter]
     logger.info(f"Done processing {filepath}")
     return df