feat: added ability to transfer well screens and gwl observations

Author: jirhiker

api/observation.py

@@ -110,6 +110,7 @@ def get_groundwater_level_observations(
         sql = sql.where(Observation.observation_datetime <= end_time)
 
     sql = order_sort_filter(sql, Observation, sort, order, filter_)
+    sql = sql.order_by(Observation.observation_datetime.desc())
     return paginate(query=sql, conn=session)
 
 

api/search.py

@@ -64,23 +64,46 @@ def _get_contact_results(session: Session, q: str) -> list[dict]:
 
 def _get_thing_results(session: Session, q: str) -> list[dict]:
     vector = Thing.search_vector
-    query = search(select(Thing), q, vector=vector)
+    water_well_query = search(select(Thing).where(Thing.thing_type=='water well'), q, vector=vector)
+    spring_well_query = search(select(Thing).where(Thing.thing_type=='spring'), q, vector=vector)
 
-    things = session.scalars(query).all()
-    results = [
-        {
-            "label": t.name,
-            "group": "Things",
-            "properties": {
-                "well_type": t.well_type,
-                "thing_type": t.thing_type,
-                "id": t.id,
-            },
-        }
-        for t in things
-    ]
+    wells = session.scalars(water_well_query).all()
+    springs = session.scalars(spring_well_query).all()
 
-    return results
+    def _make_response(group: str, thing: Thing, properties: dict) -> dict:
+
+        if properties is None:
+            properties = {}
+
+        properties['thing_type'] = thing.thing_type
+        properties['id'] = thing.id
+        return {
+            "label": thing.name,
+            "group": group,
+            "properties": properties,
+        }
+    def make_well_response(thing: Thing) -> dict:
+        return _make_response(
+            'Wells',
+            thing,
+            {
+                'well_type': thing.well_type,
+                'well_depth': thing.well_depth,
+                'hole_depth': thing.hole_depth,
+            }
+        )
+    def make_spring_response(thing: Thing) -> dict:
+        return _make_response(
+            'Springs',
+            thing,
+            {
+                'spring_type': thing.spring_type,
+            }
+        )
+
+    return [func(item) for items, func in ((wells, make_well_response),
+                                (springs, make_spring_response))
+            for item in items]
 
 
 def _get_asset_results(session: Session, q: str) -> list[dict]:
@@ -118,7 +141,7 @@ def search_api(q: str, session: Session = Depends(get_db_session)):
     results.extend(_get_thing_results(session, q))
     results.extend(_get_asset_results(session, q))
 
-    return results
+    return {'items': results, 'total': len(results)}
 
 
 # ============= EOF =============================================

api/thing.py

@@ -136,10 +136,16 @@ async def get_springs(
 )
 async def get_well_screens(
     session: session_dependency,
+        thing_id: int = None,
 ) -> CustomPage[WellScreenResponse]:
     """
     Retrieve all well screens from the database.
     """
+    if thing_id:
+        sql = select(WellScreen).where(WellScreen.thing_id == thing_id)
+        return paginate(query=sql, conn=session)
+
+
     return paginated_all_getter(session, WellScreen)
 
 

db/thing.py

@@ -116,6 +116,9 @@ class WellScreen(Base, AutoBaseMixin):
     )
     screen_type = lexicon_term()  # e.g., "PVC", "Steel", etc.
 
+    screen_description = Column(
+        String(1000), nullable=True, info={"unit": "description of the screen"}
+    )
     # Define a relationship to well if needed
     # well = relationship("Well")
 

schemas/observation.py

@@ -96,7 +96,7 @@ class BaseObservationResponse(BaseModel):
 
 class GroundwaterLevelObservationResponse(BaseObservationResponse):
     depth_to_water: float
-    level_status: str
+    level_status: str | None
 
 
 class GeothermalObservationResponse(BaseObservationResponse):

schemas/thing.py

@@ -78,6 +78,7 @@ class CreateWellScreen(BaseModel):
     screen_depth_bottom: float
     screen_depth_top: float
     screen_type: str | None = None
+    screen_description: str | None = None
 
     @model_validator(mode="after")
     def validate_screen_type(self):

transfers/data/wellscreens.csv

@@ -15,15 +15,20 @@
 # ===============================================================================
 
 import time
+import uuid
+from datetime import datetime
 
 import numpy as np
 import pandas as pd
 import pyproj
+from pydantic import ValidationError
 from shapely import Point
 from shapely.ops import transform
+from sqlalchemy import select
 
-from db import Location, LocationThingAssociation
+from db import Location, LocationThingAssociation, adder, WellScreen, Thing, Observation, Sample
 from db.engine import session_ctx
+from schemas.thing import CreateWellScreen
 
 # from db.observation.groundwaterlevel import GroundwaterLevelObservation
 
@@ -67,7 +72,42 @@ def make_location(row):
     )
 
 
-#
+def transfer_water_levels(session):
+    wd = pd.read_csv("./data/water_levels.csv")
+    gwd = wd.groupby(["PointID"])
+
+    for index, group in gwd:
+        for row in group.itertuples():
+            if pd.isna(row.DepthToWater) or pd.isna(row.DateMeasured):
+                print(f"Skipping row {row.Index} due to missing data.")
+                continue
+
+            dt = datetime.fromisoformat(row.DateMeasured)
+            thing = session.query(Thing).where(Thing.name==row.PointID).first()
+            if thing is None:
+                print(f"Thing with PointID {row.PointID} not found. Skipping water level.")
+                continue
+
+            sample = Sample()
+            sample.sampler_name = 'unknown'
+            sample.sample_type = 'groundwater level'
+
+            sample.field_sample_id = str(uuid.uuid4())
+            sample.sample_date = dt
+            sample.thing = thing
+            session.add(sample)
+
+            obs = Observation()
+            obs.sensor_id = 1
+            obs.sample = sample
+            obs.observation_datetime = dt
+            obs.depth_to_water = row.DepthToWater
+            obs.observed_property = "groundwater level"
+            obs.unit = "ft"
+
+            session.add(obs)
+            session.commit()
+
 # def migrate_water_levels(session, limit=800):
 #     wd = pd.read_csv("./migration/data/water_levels.csv")
 #     p = pd.read_csv("./migration/data/welldata.csv")
@@ -150,14 +190,14 @@ def make_location(row):
 ADDED = []
 
 
-def transfer_springs(session, limit=10000):
+def transfer_springs(session, limit=None):
     ldf = pd.read_csv("./data/location.csv")
     ldf = ldf[ldf["SiteType"] == "SP"]
     ldf = ldf[ldf["Easting"].notna() & ldf["Northing"].notna()]
     n = len(ldf)
     start_time = time.time()
     for i, row in enumerate(ldf.itertuples()):
-        if i >= limit:
+        if limit and i >= limit:
             print(f"Reached limit of {limit} rows. Stopping migration.")
             break
 
@@ -185,7 +225,7 @@ def transfer_springs(session, limit=10000):
         session.add(assoc)
 
 
-def transfer_wells(session, limit=1000):
+def transfer_wells(session, limit=None):
     wdf = pd.read_csv("./data/welldata.csv")
     ldf = pd.read_csv("./data/location.csv")
 
@@ -200,7 +240,7 @@ def transfer_wells(session, limit=1000):
     start_time = time.time()
 
     for i, row in enumerate(wdf.itertuples()):
-        if i >= limit:
+        if limit and i >= limit:
             print("Reached limit of", limit, "rows. Stopping migration.")
             break
 
@@ -242,7 +282,51 @@ def transfer_wells(session, limit=1000):
         session.add(assoc)
         # break
 
+def transfer_wellscreens(session, limit=None):
+    wdf = pd.read_csv("./data/wellscreens.csv")
+    wdf = wdf.replace(pd.NA, None)
+    wdf = wdf.replace({np.nan: None})
 
+    n = len(wdf)
+    start_time = time.time()
+
+    for i, row in enumerate(wdf.itertuples()):
+        if limit and i >= limit:
+            print("Reached limit of", limit, "rows. Stopping migration.")
+            break
+
+        if i and not i % 100:
+            print(
+                f"Processing row {i} of {n}. {row.PointID},  avg rows per second: {i / (time.time() - start_time):.2f}"
+            )
+            session.commit()
+# thing_id: int
+# screen_depth_bottom: float
+# screen_depth_top: float
+# screen_type: str | None = None
+        # print(row)
+
+        sql = select(Thing).where(Thing.name == row.PointID)
+        thing = session.execute(sql).scalar_one_or_none()
+        if not thing:
+            print(f"Thing with PointID {row.PointID} not found. Skipping well screen.")
+            continue
+
+        well_screen_data = {
+            "thing_id": thing.id,
+            "screen_depth_top": row.ScreenTop,
+            "screen_depth_bottom": row.ScreenBottom,
+            # "screen_type": row.ScreenType,
+            "screen_description": row.ScreenDescription,
+            "release_status": 'draft'
+        }
+        try:
+            model = CreateWellScreen.model_validate(well_screen_data)
+            adder(session, WellScreen, model)
+        except ValidationError as e:
+            print(f"Validation error for row {i} with PointID {row.PointID}: {e}")
+            continue
+        # session.add(screen)
 # def reset_db():
 #     configure_mappers()
 #
@@ -256,8 +340,9 @@ def transfer_wells(session, limit=1000):
 if __name__ == "__main__":
     # reset_db()
     with session_ctx() as sess:
-        transfer_wells(sess, limit=10000)
-        transfer_springs(sess, limit=10000)
-        # migrate_water_levels(sess)
+        # transfer_wells(sess, 1000)
+        # transfer_springs(sess, limit=10000)
+        # transfer_wellscreens(sess)
+        transfer_water_levels(sess)
 
 # ============= EOF =============================================