Update from wis2box code

.gitignore

@@ -167,3 +167,15 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+__pycache__/
+.pytest_cache/
+.venv/
+tmp*/
+oregondb*
+oregon_load_metadata.json
+# dagster local sqlite dbs
+history/
+schedules/
+.env
+.hypothesis/

docker-compose.yml

@@ -1,14 +1,21 @@
 services:
   wqie-frost:
     container_name: wqie-frost
-    build:
-      context: ./docker/frost
+    image: hylkevds/frost-http-waterquality:latest
+    platform: linux/amd64
     environment:
       - serviceRootUrl=${WQIE_URL}/FROST-Server
       - persistence_db_url=jdbc:postgresql://wqie-database:5432/sensorthings
       - persistence_db_username=sensorthings
       - persistence_db_password=ChangeMe
+      - plugins_multiDatastream_enable=false
+      - http_cors_enable=true
+      - http_cors_allowed_origins=*
+      - persistence_db_driver=org.postgresql.Driver
       - persistence_autoUpdateDatabase=true
+      - persistence_idGenerationMode=ServerAndClientGenerated
+      - plugins_coreModel_idType=STRING
+      - mqtt_Enabled=false
     depends_on:
       - wqie-database
 

docker/dagster/Dockerfile

@@ -11,7 +11,7 @@ ENV DAGSTER_HOME=/opt/dagster/dagster_home/
 
 COPY ../dagster.yaml /opt/dagster/dagster_home/
 COPY ../pyproject.toml /opt/dagster/app/pyproject.toml
-COPY userCode /opt/dagster/app/userCode
+COPY src /opt/dagster/app/src
 
 WORKDIR /opt/dagster/app
 

docker/frost/Dockerfile

@@ -1,12 +1,10 @@
-FROM fraunhoferiosb/frost-server
+FROM hylkevds/frost-http-waterquality
 
 ENV plugins_multiDatastream_enable=false
 ENV http_cors_enable=true
 ENV http_cors_allowed_origins=*
 ENV persistence_db_driver=org.postgresql.Driver
 ENV persistence_autoUpdateDatabase=true
 ENV persistence_idGenerationMode=ServerAndClientGenerated
-ENV plugins_coreModel_idType_thing=STRING
-ENV plugins_coreModel_idType_location=STRING
-ENV plugins_coreModel_idType_datastream=STRING
+ENV plugins_coreModel_idType=STRING
 ENV mqtt_Enabled=false

pyproject.toml

@@ -18,15 +18,14 @@ dependencies = [
     "openpyxl>=3.1.5",
     "dagster-slack>=0.25.4",
     "pytest-dotenv>=0.5.2",
-    "hypothesis>=6.122.5",
+    "hypothesis>=6.122.5"
 ]
 
-[tool.setuptools]
-packages = ["userCode"]
+[tool.hatch.build.targets.wheel]
+packages = ["src.wqie"]
 
 [tool.dagster]
-module_name = "userCode"
-code_location_name = "userCode"
+module_name = "wqie"
 
 [tool.pytest.ini_options]
 env_files = [

requirements.txt

@@ -67,6 +67,7 @@ requests==2.32.3
 requests-toolbelt==1.0.0
 rich==13.9.4
 setuptools==75.8.0
+shapely==2.0.6
 six==1.17.0
 slack-sdk==3.34.0
 sniffio==1.3.1
@@ -87,4 +88,5 @@ uvloop==0.21.0
 watchdog==5.0.3
 watchfiles==1.0.4
 websockets==14.1
+-e file:///Users/benwebb/Library/CloudStorage/OneDrive-LincolnInstituteofLandPolicy/Documents/GitHub/cgs-earth/wqie
 yarl==1.18.3

src/wqie/__init__.py

@@ -0,0 +1,17 @@
+# =================================================================
+#
+# Copyright (c) 2025 Lincoln Institute of Land Policy
+#
+# Licensed under the MIT License.
+#
+# =================================================================
+
+from dagster import Definitions
+from wqie.assets import county_stations
+from wqie.jobs import process_county_stations
+
+
+defs = Definitions(
+    assets=[county_stations],
+    jobs=[process_county_stations],
+)

src/wqie/assets.py

@@ -0,0 +1,29 @@
+# =================================================================
+#
+# Authors: Ben Webb <[email protected]>
+#
+# Copyright (c) 2025 Lincoln Institute of Land Policy
+#
+# Licensed under the MIT License.
+#
+# =================================================================
+
+from dagster import asset
+from wqie.ops.fetch import fetch_station_metadata, fetch_site_metadata
+from wqie.ops.transform import transform_stations
+from wqie.partitions import county_partitions
+
+
+@asset(partitions_def=county_partitions)
+def county_stations(context):
+    """Fetch and process stations for a single county partition."""
+
+    # Extract the partition key (county)
+    county = context.asset_partition_key_for_output()
+
+    # Fetch and process data
+    stations = fetch_station_metadata(county)
+    site_details = fetch_site_metadata(
+        [s["MonitoringLocationIdentifier"] for s in stations]
+    )
+    return transform_stations(site_details)

src/wqie/env.py

@@ -0,0 +1,20 @@
+# =================================================================
+#
+# Authors: Colton Loftus <[email protected]>
+# Authors: Ben Webb <[email protected]>
+#
+# Copyright (c) 2025 Lincoln Institute of Land Policy
+#
+# Licensed under the MIT License.
+#
+# =================================================================
+
+from wqie.util import get_env
+
+API_BACKEND_URL = get_env(
+    "API_BACKEND_URL", fallback="http://localhost:8888/FROST-Server/v1.1"
+)
+GEOCONNEX_URL = "https://geoconnex.us"
+NLDI_URL = "https://labs.waterdata.usgs.gov/api/nldi"
+STATION_URL = 'https://www.waterqualitydata.us/data/Station/search'
+RESULTS_URL = 'https://www.waterqualitydata.us/data/Result/search'

src/wqie/jobs.py

@@ -0,0 +1,36 @@
+# =================================================================
+#
+# Authors: Ben Webb <[email protected]>
+#
+# Copyright (c) 2025 Lincoln Institute of Land Policy
+#
+# Licensed under the MIT License.
+#
+# =================================================================
+
+from dagster import job, op
+from wqie.ops.fetch import fetch_station_metadata, fetch_site_metadata
+from wqie.ops.transform import transform_stations, publish_station_collection
+from wqie.partitions import county_partitions
+
+
+@op
+def fetch_and_process_stations(context):
+    """Fetch and process stations for a single county partition."""
+
+    # Extract partition key (county)
+    county = context.partition_key
+
+    # Fetch and process data
+    stations = fetch_station_metadata(county)
+    site_details = fetch_site_metadata(
+        [s["MonitoringLocationIdentifier"] for s in stations]
+    )
+    sites = transform_stations(site_details)
+    return publish_station_collection(sites)
+
+
+@job(partitions_def=county_partitions)
+def process_county_stations():
+    """Job to process stations for a single county partition."""
+    fetch_and_process_stations()