Merge pull request #25 from DataIntegrationGroup/dev/jab

Version 0.4.0

Author: jirhiker

UNIT_CONVERSIONS.md

@@ -0,0 +1,58 @@
+# Successful Unit Conversions
+The conversion factor is applied to the reported value from the source to obtain a value with standardized units.
+
+| SOURCE PARAMETER NAME | SOURCE UNITS | DIE PARAMETER NAME | DIE PARAMETER UNITS | CONVERSION FACTOR | NOTES |
+| :-------------------- | :----------- | :----------------- | :------------------ | :---------------- | :---- |
+| -                     | m            | -                  | ft                  | 3.28084           | Applies to all records where source and die parameter names are equivalent |
+| -                     | ft           | -                  | m                   | 0.3048            | Applies to all records where source and die parameter names are equivalent |
+| -                     | ppm          | -                  | mg/L                | 1                 | Applies to all records where source and die parameter names are equivalent |
+| -                     | ug/L         | -                  | mg/L                | 0.001             | Applies to all records where source and die parameter names are equivalent |
+| -                     | ton/ac-ft    | -                  | mg/L                | 735.47            | Applies to all records where source and die parameter names are equivalent |
+| bicarbonate           | mg/L as CaCO3 | bicarbonate       | mg/L                | 1.22              | equivalent mass HCO3 - = 61 |
+| calcium               | mg/L as CaCO3 | calcium           | mg/L                | 0.4               | equivalent mass Ca 2+ = 20 |
+| carbonate             | mg/L as CaCO3 | carbonate         | mg/L                | 0.6               | equivalent mass CO3 2- = 30 |
+| nitrate as n          | mg/L as N    | nitrate            | mg/L                | 4.4268            | -     |
+| nitrate               | mg/L as N    | nitrate            | mg/L                | 4.4268            | -     |
+| nitrate               | ug/L as N    | nitrate            | mg/L                | 0.0044268         | -     |
+| nitrate               | mg/L as NO3  | nitrate            | mg/L                | 1                 | -     |
+| sulfate as SO4        | mg/L         | sulfate            | mg/L                | 1                 | -     |
+| sulfur sulfate        | mg/L         | sulfate            | mg/L                | 1                 | -     |
+| uranium               | pCi/L        | uranium            | mg/L                | 0.00149           | [conversion factor source](https://www.epa.gov/sites/default/files/2015-09/documents/qa_rad_webcast.pdf) |
+
+## Converting from mg/L as CaCO3 to mg/L
+
+```
+mg/L as CaCO3 = mg/L * (equivalent mass CaCO3/equivalent mass analyte)
+
+mg/L = mg/L as CaCO3 * (equivalent mass analyte/equivalent mass CaCO3)
+```
+
+where **equivalent mass analyte = atomic mass analyte/valency analyte**
+
+so **equivalent mass CaCO3 = (equivalent mass Ca 2+) + (equivalent mass CO3 2-) = 40/2 + (12+3*16)/2 = 50**
+
+
+## Converting from mg/L as N to mg/L
+
+```
+mg/L * (molecular weight N/molecular weight NO3) = mg/L as N 
+mg/L = mg/L as N * (molecuar weight NO3/molecular weight N)
+```
+
+where **molecular weight N = 14.007 g/mol** and **molecular weight NO3 = (14.007+15.999*3) = 62.004 g/mol**
+
+## Converting from pCi/L to mg/L
+
+```
+ug/L = pCi/L * 1.49
+```
+
+# Skipped Unit Conversions
+
+Records with the following source parameter names, source units, die parameter names, and die parameter units are not included in the output.
+
+| SOURCE PARAMETER NAME | SOURCE UNITS | NOTES |
+| :-------------------- | :----------- | :---- |
+| -                     | tons/day     | This is not a unit of concentration |
+| -                     | mg/kg        | This is not a unit of aqueous concentration |
+| sodium                | %            | This is "Sodium fraction of cations, water, percent in equivalents of major cations" - see USGS pcode 00932 |

backend/config.py

@@ -421,6 +421,11 @@ def _make_output_path(self):
         if not os.path.exists(self.output_path):
             os.mkdir(self.output_path)
 
+    def _update_output_units(self):
+        parameter = self.parameter.lower()
+        if parameter == "ph":
+            self.analyte_output_units = ""
+
     @property
     def start_dt(self):
         return self._extract_date(self.start_date)

backend/connectors/README.md

@@ -33,7 +33,7 @@ The following methods need to be defined for Faux. See `BaseAnalyteSource` for d
 
 - `get_records`
 - `_extract_site_records`
-- `_extract_parameter_units`
+- `_extract_source_parameter_units`
 - `_extract_most_recent`
 - `_extract_parameter_result`
 - `_extract_parameter_record`
@@ -48,7 +48,7 @@ The following methods need to be defined for Faux. See `BaseWaterLevelSource` fo
 
 - `get_records`
 - `_extract_site_records`
-- `_extract_parameter_units`
+- `_extract_source_parameter_units`
 - `_extract_most_recent`
 - `_extract_parameter_result`
 - `_extract_parameter_record`

backend/connectors/bor/source.py

@@ -21,14 +21,11 @@
 from backend.connectors.bor.transformer import BORSiteTransformer, BORAnalyteTransformer
 from backend.connectors.mappings import BOR_ANALYTE_MAPPING
 from backend.constants import (
-    TDS,
-    URANIUM,
-    ARSENIC,
-    SULFATE,
-    FLUORIDE,
-    CHLORIDE,
+    PARAMETER_NAME,
     PARAMETER_VALUE,
     PARAMETER_UNITS,
+    SOURCE_PARAMETER_NAME,
+    SOURCE_PARAMETER_UNITS,
     DT_MEASURED,
 )
 
@@ -68,32 +65,42 @@ def parse_dt(dt):
 class BORAnalyteSource(BaseAnalyteSource):
     transformer_klass = BORAnalyteTransformer
     _catalog_item_idx = None
+    _source_parameter_name = None
 
     def __repr__(self):
         return "BORAnalyteSource"
 
     def _extract_parameter_record(self, record):
+        record[PARAMETER_NAME] = self.config.parameter
         record[PARAMETER_VALUE] = record["attributes"]["result"]
-        record[PARAMETER_UNITS] = record["attributes"]["resultAttributes"]["units"]
+        record[PARAMETER_UNITS] = self.config.analyte_output_units
         record[DT_MEASURED] = parse_dt(record["attributes"]["dateTime"])
+        record[SOURCE_PARAMETER_NAME] = self._source_parameter_name
+        record[SOURCE_PARAMETER_UNITS] = record["attributes"]["resultAttributes"][
+            "units"
+        ]
         return record
 
-    def _extract_parameter_results(self, rs):
+    def _extract_source_parameter_results(self, rs):
         return [ri["attributes"]["result"] for ri in rs]
 
-    def _extract_parameter_units(self, records):
+    def _extract_source_parameter_units(self, records):
         return [ri["attributes"]["resultAttributes"]["units"] for ri in records]
 
     def _extract_parameter_dates(self, records):
         return [parse_dt(ri["attributes"]["dateTime"]) for ri in records]
 
+    def _extract_source_parameter_names(self, records):
+        return [self._source_parameter_name for ri in records]
+
     def _extract_most_recent(self, rs):
 
         record = get_most_recent(rs, "attributes.dateTime")
         return {
             "value": record["attributes"]["result"],
             "datetime": parse_dt(record["attributes"]["dateTime"]),
-            "units": record["attributes"]["resultAttributes"]["units"],
+            "source_parameter_units": record["attributes"]["resultAttributes"]["units"],
+            "source_parameter_name": self._source_parameter_name,
         }
 
     def _extract_site_records(self, records, site_record):
@@ -126,6 +133,11 @@ def get_records(self, site_record):
                 if not self._catalog_item_idx:
                     self._catalog_item_idx = i
 
+                if self._source_parameter_name is None:
+                    self._source_parameter_name = data["attributes"][
+                        "parameterSourceCode"
+                    ]
+
                 return self._execute_json_request(
                     "https://data.usbr.gov/rise/api/result",
                     params={"itemId": data["attributes"]["_id"]},

backend/connectors/ckan/source.py

@@ -36,9 +36,11 @@
     DTW,
     DTW_UNITS,
     DT_MEASURED,
-    PARAMETER,
+    PARAMETER_NAME,
     PARAMETER_UNITS,
     PARAMETER_VALUE,
+    SOURCE_PARAMETER_NAME,
+    SOURCE_PARAMETER_UNITS,
 )
 from backend.source import (
     BaseSource,
@@ -133,21 +135,31 @@ def _parse_response(self, site_record, resp):
         records = resp.json()["result"]["records"]
         return [record for record in records if record["Site_ID"] == site_record.id]
 
-    def _extract_parameter_results(self, records):
+    def _extract_source_parameter_results(self, records):
         return [float(r["DTWGS"]) for r in records]
 
     def _extract_most_recent(self, records):
         record = get_most_recent(records, tag="Date")
-        return {"value": record["DTWGS"], "datetime": record["Date"], "units": FEET}
+        return {
+            "value": record["DTWGS"],
+            "datetime": record["Date"],
+            "source_parameter_units": FEET,
+            "source_parameter_name": "DTWGS",
+        }
 
     def _extract_parameter_dates(self, records: list) -> list:
         return [r["Date"] for r in records]
 
+    def _extract_source_parameter_names(self, records):
+        return ["" for r in records]
+
     def _extract_parameter_record(self, record):
-        record[PARAMETER] = DTW
+        record[PARAMETER_NAME] = DTW
         record[PARAMETER_VALUE] = float(record["DTWGS"])
-        record[PARAMETER_UNITS] = FEET
+        record[PARAMETER_UNITS] = self.config.waterlevel_output_units
         record[DT_MEASURED] = record["Date"]
+        record[SOURCE_PARAMETER_NAME] = "DTWGS"
+        record[SOURCE_PARAMETER_UNITS] = FEET
         return record
 
     def _clean_records(self, records: list) -> list:

backend/connectors/isc_seven_rivers/source.py

@@ -20,18 +20,14 @@
 from backend.connectors import ISC_SEVEN_RIVERS_BOUNDING_POLYGON
 from backend.connectors.mappings import ISC_SEVEN_RIVERS_ANALYTE_MAPPING
 from backend.constants import (
-    TDS,
     FEET,
-    URANIUM,
-    SULFATE,
-    FLUORIDE,
-    CHLORIDE,
     DT_MEASURED,
-    DTW_UNITS,
     DTW,
-    PARAMETER,
+    PARAMETER_NAME,
     PARAMETER_VALUE,
     PARAMETER_UNITS,
+    SOURCE_PARAMETER_NAME,
+    SOURCE_PARAMETER_UNITS,
 )
 from backend.connectors.isc_seven_rivers.transformer import (
     ISCSevenRiversSiteTransformer,
@@ -93,11 +89,12 @@ def get_records(self):
 class ISCSevenRiversAnalyteSource(BaseAnalyteSource):
     transformer_klass = ISCSevenRiversAnalyteTransformer
     _analyte_ids = None
+    _source_parameter_name = None
 
     def __repr__(self):
         return "ISCSevenRiversAnalyteSource"
 
-    def _get_analyte_id(self, analyte):
+    def _get_analyte_id_and_name(self, analyte):
         """ """
         if self._analyte_ids is None:
 
@@ -107,13 +104,20 @@ def _get_analyte_id(self, analyte):
 
         analyte = get_analyte_search_param(analyte, ISC_SEVEN_RIVERS_ANALYTE_MAPPING)
         if analyte:
-            return self._analyte_ids.get(analyte)
+            id_and_name = {
+                "id": self._analyte_ids.get(analyte),
+                "name": analyte,
+            }
+            return id_and_name
 
     def _extract_parameter_record(self, record):
-        record[PARAMETER] = self.config.parameter
+        record[PARAMETER_NAME] = self.config.parameter
         record[PARAMETER_VALUE] = record["result"]
-        record[PARAMETER_UNITS] = record["units"]
+        record[PARAMETER_UNITS] = self.config.analyte_output_units
         record[DT_MEASURED] = get_datetime(record)
+        record[SOURCE_PARAMETER_NAME] = self._source_parameter_name
+        record[SOURCE_PARAMETER_UNITS] = record["units"]
+
         return record
 
     def _extract_most_recent(self, records):
@@ -122,25 +126,30 @@ def _extract_most_recent(self, records):
         return {
             "value": record["result"],
             "datetime": get_datetime(record),
-            "units": record["units"],
+            "source_parameter_units": record["units"],
+            "source_parameter_name": self._source_parameter_name,
         }
 
     def _clean_records(self, records):
         return [r for r in records if r["result"] is not None]
 
-    def _extract_parameter_results(self, records):
+    def _extract_source_parameter_results(self, records):
         return [r["result"] for r in records]
 
-    def _extract_parameter_units(self, records):
+    def _extract_source_parameter_units(self, records):
         return [r["units"] for r in records]
 
     def _extract_parameter_dates(self, records: list) -> list:
         return [get_datetime(r) for r in records]
 
+    def _extract_source_parameter_names(self, records: list) -> list:
+        return [self._source_parameter_name for r in records]
+
     def get_records(self, site_record):
         config = self.config
-        analyte_id = self._get_analyte_id(config.parameter)
-        if analyte_id:
+        analyte_id_and_name = self._get_analyte_id_and_name(config.parameter)
+        if analyte_id_and_name:
+            analyte_id = analyte_id_and_name["id"]
             params = {
                 "monitoringPointId": site_record.id,
                 "analyteId": analyte_id,
@@ -149,13 +158,18 @@ def get_records(self, site_record):
             }
             params.update(get_date_range(config))
 
+            if self._source_parameter_name is None:
+                self._source_parameter_name = analyte_id_and_name["name"]
+
             return self._execute_json_request(
                 _make_url("getReadings.ashx"), params=params
             )
 
 
 class ISCSevenRiversWaterLevelSource(BaseWaterLevelSource):
     transformer_klass = ISCSevenRiversWaterLevelTransformer
+    _source_parameter_name = "depthToWaterFeet"
+    _source_parameter_units = FEET
 
     def get_records(self, site_record):
         params = {
@@ -174,24 +188,37 @@ def _clean_records(self, records):
         return [r for r in records if r["depthToWaterFeet"] is not None]
 
     def _extract_parameter_record(self, record):
-        record[PARAMETER] = DTW
+        record[PARAMETER_NAME] = DTW
         record[PARAMETER_VALUE] = record["depthToWaterFeet"]
-        record[PARAMETER_UNITS] = FEET
+        record[PARAMETER_UNITS] = self.config.waterlevel_output_units
         record[DT_MEASURED] = get_datetime(record)
+        record[SOURCE_PARAMETER_NAME] = self._source_parameter_name
+        record[SOURCE_PARAMETER_UNITS] = self._source_parameter_units
         return record
 
-    def _extract_parameter_results(self, records):
+    def _extract_source_parameter_results(self, records):
         return [
             r["depthToWaterFeet"] for r in records if not r["invalid"] and not r["dry"]
         ]
 
     def _extract_parameter_dates(self, records: list) -> list:
         return [get_datetime(r) for r in records]
 
+    def _extract_source_parameter_names(self, records):
+        return [self._source_parameter_name for r in records]
+
+    def _extract_source_parameter_units(self, records):
+        return [self._source_parameter_units for r in records]
+
     def _extract_most_recent(self, records):
         record = get_most_recent(records, "dateTime")
         t = get_datetime(record)
-        return {"value": record["depthToWaterFeet"], "datetime": t, "units": FEET}
+        return {
+            "value": record["depthToWaterFeet"],
+            "datetime": t,
+            "source_parameter_units": self._source_parameter_units,
+            "source_parameter_name": DTW,
+        }
 
 
 # ============= EOF =============================================