WQP water level data | OSE Roswell data from ST2 | decrease WQP chunk size

CHANGELOG.md

@@ -4,7 +4,15 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## Unrelease
+## Unreleased: 0.8.0
+
+### Added
+- water level for WQP
+
+### Changed
+- NM OSE Roswell data is now pulled from ST2 and not CKAN
+
+### Fixed
 
 
 ## 0.7.0

README.md

@@ -39,7 +39,7 @@ Data comes from the following sources. We are continuously adding new sources as
 - [USGS (NWIS)](https://waterdata.usgs.gov/nwis)
   - Available data: `water levels`
 - [Water Quality Portal (WQP)](https://www.waterqualitydata.us/)
-  - Available data: `water quality`
+  - Available data: `water levels`, `water quality`
 
 ## Usage
 
@@ -66,7 +66,7 @@ where `{parameter}` is the name of the parameter whose data is to be retrieved,
 | **nmose-roswell**          | X           | -       | -           | -       | -         | -        | -        | -         | -       | -   | -         | -      | -      | -       | -   | -       |
 | **nwis**                   | X           | -       | -           | -       | -         | -        | -        | -         | -       | -   | -         | -      | -      | -       | -   | -       |
 | **pvacd**                  | X           | -       | -           | -       | -         | -        | -        | -         | -       | -   | -         | -      | -      | -       | -   | -       |
-| **wqp**                    | -           | X       | X           | X       | X         | X        | X        | X         | X       | X   | X         | X      | X      | X       | X   | X       |
+| **wqp**                    | X           | X       | X           | X       | X         | X        | X        | X         | X       | X   | X         | X      | X      | X       | X   | X       |
 
 ### Output
 The `--output` option is required and used to set the output type:

backend/config.py

@@ -55,9 +55,11 @@
     BernCoWaterLevelSource,
     CABQSiteSource,
     CABQWaterLevelSource,
+    NMOSERoswellSiteSource,
+    NMOSERoswellWaterLevelSource,
 )
 from .connectors.usgs.source import NWISSiteSource, NWISWaterLevelSource
-from .connectors.wqp.source import WQPSiteSource, WQPAnalyteSource
+from .connectors.wqp.source import WQPSiteSource, WQPAnalyteSource, WQPWaterLevelSource
 
 SOURCE_KEYS = (
     "bernco",
@@ -90,7 +92,7 @@ def get_source(source):
     elif source == "nmose_isc_seven_rivers":
         return ISCSevenRiversSiteSource()
     elif source == "nmose_roswell":
-        return OSERoswellSiteSource(HONDO_RESOURCE_ID)
+        return NMOSERoswellSiteSource()
     elif source == "nwis":
         return NWISSiteSource()
     elif source == "pvacd":
@@ -218,24 +220,7 @@ def water_level_sources(self):
             sources.append((NWISSiteSource(), NWISWaterLevelSource()))
 
         if self.use_source_nmose_roswell:
-            sources.append(
-                (
-                    OSERoswellSiteSource(HONDO_RESOURCE_ID),
-                    OSERoswellWaterLevelSource(HONDO_RESOURCE_ID),
-                )
-            )
-            sources.append(
-                (
-                    OSERoswellSiteSource(FORT_SUMNER_RESOURCE_ID),
-                    OSERoswellWaterLevelSource(FORT_SUMNER_RESOURCE_ID),
-                )
-            )
-            sources.append(
-                (
-                    OSERoswellSiteSource(ROSWELL_RESOURCE_ID),
-                    OSERoswellWaterLevelSource(ROSWELL_RESOURCE_ID),
-                )
-            )
+            sources.append((NMOSERoswellSiteSource(), NMOSERoswellWaterLevelSource()))
         if self.use_source_pvacd:
             sources.append((PVACDSiteSource(), PVACDWaterLevelSource()))
         if self.use_source_bernco:
@@ -244,6 +229,8 @@ def water_level_sources(self):
             sources.append((EBIDSiteSource(), EBIDWaterLevelSource()))
         if self.use_source_cabq:
             sources.append((CABQSiteSource(), CABQWaterLevelSource()))
+        if self.use_source_wqp:
+            sources.append((WQPSiteSource(), WQPWaterLevelSource()))
 
         for s, ss in sources:
             s.set_config(self)

backend/connectors/mappings.py

@@ -170,7 +170,7 @@
         "Total Sulfate",
     ],
     TDS: ["Total dissolved solids"],
-    URANIUM: ["Uranium", "Uranium-238"],
+    URANIUM: ["Uranium"],
     PH: ["pH"],
 }
 # BOR ===============================================================================

backend/connectors/st2/source.py

@@ -24,6 +24,8 @@
     CABQ_BOUNDING_POLYGON,
 )
 from backend.connectors.st2.transformer import (
+    NMOSERoswellSiteTransformer,
+    NMOSERoswellWaterLevelTransformer,
     PVACDSiteTransformer,
     PVACDWaterLevelTransformer,
     EBIDSiteTransformer,
@@ -64,6 +66,14 @@ def _get_filters(self):
         return [f"properties/agency eq '{self.agency}'"]
 
 
+class NMOSERoswellSiteSource(ST2SiteSource):
+    transformer_klass = NMOSERoswellSiteTransformer
+    agency = "OSE-Roswell"
+
+    def __repr__(self):
+        return "NMOSERoswellSiteSource"
+
+
 class PVACDSiteSource(ST2SiteSource):
     transformer_klass = PVACDSiteTransformer
     agency = "PVACD"
@@ -172,6 +182,14 @@ def get_records(self, site_record, *args, **kw):
         return records
 
 
+class NMOSERoswellWaterLevelSource(ST2WaterLevelSource):
+    transformer_klass = NMOSERoswellWaterLevelTransformer
+    agency = "OSE-Roswell"
+
+    def __repr__(self):
+        return "NMOSERoswellWaterLevelSource"
+
+
 class PVACDWaterLevelSource(ST2WaterLevelSource):
     transformer_klass = PVACDWaterLevelTransformer
     agency = "PVACD"

backend/connectors/st2/transformer.py

@@ -26,6 +26,10 @@
 )
 
 
+class NMOSERoswellSiteTransformer(STSiteTransformer):
+    source_id = "ST2/NMOSE-Roswell"
+
+
 class PVACDSiteTransformer(STSiteTransformer):
     source_id = "ST2/PVACD"
 
@@ -103,6 +107,10 @@ def _transform_elevation(self, elevation, record):
 #     return rec
 
 
+class NMOSERoswellWaterLevelTransformer(WaterLevelTransformer):
+    source_tag = "ST2/NMOSE-Roswell"
+
+
 class PVACDWaterLevelTransformer(WaterLevelTransformer):
     source_tag = "ST2/PVACD"
 

backend/connectors/wqp/source.py

@@ -20,24 +20,23 @@
 from backend.connectors import NM_STATE_BOUNDING_POLYGON
 from backend.connectors.mappings import WQP_ANALYTE_MAPPING
 from backend.constants import (
-    TDS,
-    URANIUM,
-    NITRATE,
-    SULFATE,
-    ARSENIC,
-    CHLORIDE,
     PARAMETER_NAME,
     PARAMETER_VALUE,
     PARAMETER_UNITS,
     SOURCE_PARAMETER_NAME,
     SOURCE_PARAMETER_UNITS,
     DT_MEASURED,
 )
-from backend.connectors.wqp.transformer import WQPSiteTransformer, WQPAnalyteTransformer
+from backend.connectors.wqp.transformer import (
+    WQPSiteTransformer,
+    WQPAnalyteTransformer,
+    WQPWaterLevelTransformer,
+)
 from backend.source import (
-    BaseSource,
     BaseSiteSource,
     BaseAnalyteSource,
+    BaseWaterLevelSource,
+    BaseParameterSource,
     make_site_list,
     get_most_recent,
     get_analyte_search_param,
@@ -61,7 +60,7 @@ def get_date_range(config):
 
 class WQPSiteSource(BaseSiteSource):
     transformer_klass = WQPSiteTransformer
-    chunk_size = 100
+    chunk_size = 50
 
     bounding_polygon = NM_STATE_BOUNDING_POLYGON
 
@@ -93,6 +92,10 @@ def get_records(self):
             params["characteristicName"] = get_analyte_search_param(
                 config.parameter, WQP_ANALYTE_MAPPING
             )
+        else:
+            # every record with pCode 30210 (depth in m) has a corresponding
+            # record with pCode 72019 (depth in ft) but not vice versa
+            params["pCode"] = "30210"
 
         params.update(get_date_range(config))
 
@@ -103,17 +106,15 @@ def get_records(self):
             return parse_tsv(text)
 
 
-class WQPAnalyteSource(BaseAnalyteSource):
-    transformer_klass = WQPAnalyteTransformer
-
-    def __repr__(self):
-        return "WQPAnalyteSource"
+class WQPParameterSource(BaseParameterSource):
 
     def _extract_parameter_record(self, record):
         record[PARAMETER_NAME] = self.config.parameter
         record[PARAMETER_VALUE] = record["ResultMeasureValue"]
-        record[PARAMETER_UNITS] = self.config.analyte_output_units
-        record[DT_MEASURED] = record["ActivityStartDate"]
+        record[PARAMETER_UNITS] = self._parameter_units_hook()
+        record[DT_MEASURED] = (
+            f"{record['ActivityStartDate']} {record['ActivityStartTime/Time']}"
+        )
         record[SOURCE_PARAMETER_NAME] = record["CharacteristicName"]
         record[SOURCE_PARAMETER_UNITS] = record["ResultMeasure/MeasureUnitCode"]
         return record
@@ -148,22 +149,57 @@ def _extract_most_recent(self, records):
         }
 
     def get_records(self, site_record):
+        config = self.config
         sites = make_site_list(site_record)
 
         params = {
             "siteid": sites,
             "mimeType": "tsv",
-            "characteristicName": get_analyte_search_param(
-                self.config.parameter, WQP_ANALYTE_MAPPING
-            ),
         }
         params.update(get_date_range(self.config))
 
+        if config.parameter.lower() != "waterlevels":
+            params["characteristicName"] = get_analyte_search_param(
+                config.parameter, WQP_ANALYTE_MAPPING
+            )
+        else:
+            # every record with pCode 30210 (depth in m) has a corresponding
+            # record with pCode 72019 (depth in ft) but not vice versa
+            params["pCode"] = "30210"
+
+        params.update(get_date_range(config))
+
         text = self._execute_text_request(
-            "https://www.waterqualitydata.us/data/Result/search?", params
+            "https://www.waterqualitydata.us/data/Result/search?", params, timeout=30
         )
         if text:
             return parse_tsv(text)
 
+    def _parameter_units_hook(self):
+        raise NotImplementedError(
+            f"{self.__class__.__name__} must implement _parameter_units_hook"
+        )
+
+
+class WQPAnalyteSource(WQPParameterSource, BaseAnalyteSource):
+    transformer_klass = WQPAnalyteTransformer
+
+    def __repr__(self):
+        return "WQPAnalyteSource"
+
+    def _parameter_units_hook(self):
+        return self.config.analyte_output_units
+
+
+# inherit from WQPParameterSource first so that its _extract_souce_parameter_units method is used instead of BaseWaterLevelSource's method
+class WQPWaterLevelSource(WQPParameterSource, BaseWaterLevelSource):
+    transformer_klass = WQPWaterLevelTransformer
+
+    def __repr__(self):
+        return "WQPWaterLevelSource"
+
+    def _parameter_units_hook(self):
+        return self.config.waterlevel_output_units
+
 
 # ============= EOF =============================================

backend/connectors/wqp/transformer.py

@@ -16,7 +16,12 @@
 import pprint
 
 from backend.record import SiteRecord, AnalyteSummaryRecord
-from backend.transformer import BaseTransformer, SiteTransformer, AnalyteTransformer
+from backend.transformer import (
+    BaseTransformer,
+    SiteTransformer,
+    AnalyteTransformer,
+    WaterLevelTransformer,
+)
 
 
 class WQPSiteTransformer(SiteTransformer):
@@ -44,4 +49,8 @@ class WQPAnalyteTransformer(AnalyteTransformer):
     source_tag = "WQP"
 
 
+class WQPWaterLevelTransformer(WaterLevelTransformer):
+    source_tag = "WQP"
+
+
 # ============= EOF =============================================

frontend/cli.py

@@ -264,10 +264,10 @@ def weave(
         config.use_source_nmose_roswell = no_nmose_roswell
         config.use_source_nwis = no_nwis
         config.use_source_pvacd = no_pvacd
+        config.use_source_wqp = no_wqp
 
         config.use_source_bor = False
         config.use_source_nmed_dwb = False
-        config.use_source_wqp = False
 
     elif parameter == "carbonate":
         config.use_source_nmbgmr_amp = no_nmbgmr_amp

setup.py

@@ -21,7 +21,7 @@
 
 setup(
     name="nmuwd",
-    version="0.7.0",
+    version="0.7.1",
     author="Jake Ross",
     description="New Mexico Water Data Integration Engine",
     long_description=long_description,