Merge branch 'source_information_and_conversion_factor' into dev/jab

Author: jacob-a-brown

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,40 @@ 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 +131,9 @@ 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,30 @@ 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,36 @@ 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 =============================================