Simplification of config and unification of nomenclature in registries. Update dependencies.

Author: Eli

dms_datastore/config_data/dstore_config.yaml

@@ -15,10 +15,15 @@ spot_check_spec: spot_check_spec.yaml
 default_repo: screened
 
 
-# aliases for registry files available to repos in their "registry"
+# registry files available to repos and identified in their `registry` field.
 # these will be looked for in the dbase_config directory and are expected to be csv files. 
 # The registry files provide supplementary metadata that can be used for provider resolution, 
-# geographical and other purposes.
+# geographical and other purposes. The `column_map` field allows you to map from the column names
+# in the registry file to canonical metadata keys that are used in the data file headers and 
+# elsewhere in the config. The `crs` field specifies coordinate reference systems for geospatial 
+# data in the registry. If registries are joined as they are in the processed directory,
+# the column_map and crs fields for each registry will be applied to the relevant columns before merging.
+# and the crs must be compatible across registries.
 registries:
   continuous:
     file: station_dbase.csv
@@ -30,6 +35,9 @@ registries:
       lon: longitude
       x: projection_x_coordinate
       y: projection_y_coordinate
+    crs:
+      geographic: "epsg:4269"        # applies to latitude/longitude
+      projected: "epsg:26910"        # applies to projection_x_coordinate/projection_y_coordinate
   processed_synthetic: processed_registry.csv
   structures: structures_registry.csv
   daily: station_dbase.csv

dms_datastore/dropbox_data.py

@@ -180,8 +180,8 @@ def _get_float(col):
             )
         return float(val)
 
-    out["latitude"] = _get_float("agency_lat")
-    out["longitude"] = _get_float("agency_lon")
+    out["latitude"] = _get_float("lat")
+    out["longitude"] = _get_float("lon")
     out["projection_x_coordinate"] = _get_float("x")
     out["projection_y_coordinate"] = _get_float("y")
     out["projection_authority_id"] = "epsg:26910"

dms_datastore/dstore_config.py

@@ -54,6 +54,9 @@
     "repo_root",
     "resolve_repo_data_dir",
     "repo_names",
+    "registry_spec",
+    "registry_column_map",
+    "registry_crs",
     "registry_df",
     "repo_registry",
     "source_priority_group",
@@ -412,6 +415,23 @@ def repo_registry(repo=None, repo_cfg=None):
     if isinstance(registry_names, str):
         registry_names = [registry_names]
 
+    # Validate CRS consistency across sub-registries
+    if len(registry_names) > 1:
+        crs_values = []
+        for rname in registry_names:
+            crs = registry_crs(rname)
+            if crs is not None:
+                crs_values.append((rname, crs))
+        if len(crs_values) > 1:
+            first_name, first_crs = crs_values[0]
+            for other_name, other_crs in crs_values[1:]:
+                if other_crs != first_crs:
+                    raise ValueError(
+                        f"CRS mismatch across registries: "
+                        f"{first_name} declares {first_crs} but "
+                        f"{other_name} declares {other_crs}"
+                    )
+
     dfs = []
     for rname in registry_names:
         sub = registry_df(rname).copy()
@@ -424,19 +444,10 @@ def repo_registry(repo=None, repo_cfg=None):
         )
         dfs.append(sub)
 
-    # Merge on common columns
+    # Merge registries (union of columns, no key overlap allowed)
     if len(dfs) == 1:
         db = dfs[0]
     else:
-        common_cols = set(dfs[0].columns)
-        for sub in dfs[1:]:
-            common_cols &= set(sub.columns)
-        common_cols = sorted(common_cols)
-        if site_key not in common_cols:
-            raise ValueError(
-                f"Column {site_key!r} not in common columns across registries"
-            )
-
         # Check for key overlap across sub-registries
         all_keys = []
         for i, sub in enumerate(dfs):
@@ -448,7 +459,6 @@ def repo_registry(repo=None, repo_cfg=None):
                 )
             all_keys.extend(sub[site_key].values)
 
-        dfs = [sub[common_cols] for sub in dfs]
         db = pd.concat(dfs, ignore_index=True)
 
     dup = db[site_key].duplicated()
@@ -547,6 +557,39 @@ def repo_config(repo_name):
     _repo_cache[repo_name] = spec
     return spec
 
+def registry_spec(registry_name):
+    """
+    Return the raw registry specification dict for a named registry.
+
+    If the registry is declared as a bare string (legacy), returns
+    ``{"file": <string>}``.
+    """
+    registries = config.get("registries", {})
+    if registry_name not in registries:
+        raise ValueError(f"Registry not found: {registry_name}")
+    spec = registries[registry_name]
+    if isinstance(spec, str):
+        return {"file": spec}
+    return dict(spec)
+
+
+def registry_column_map(registry_name):
+    """
+    Return the column_map for a named registry, or empty dict if unset.
+    """
+    spec = registry_spec(registry_name)
+    return dict(spec.get("column_map", {}) or {})
+
+
+def registry_crs(registry_name):
+    """
+    Return the crs dict for a named registry, or None if unset.
+    """
+    spec = registry_spec(registry_name)
+    crs = spec.get("crs", None)
+    return dict(crs) if crs else None
+
+
 def registry_df(registry_name):
     """
     Load a raw registry table by name.
@@ -604,7 +647,9 @@ def registry_df(registry_name):
     if registry_name not in registries:
         raise ValueError(f"Registry not found: {registry_name}")
 
-    reg_path = _resolve_config_path(registries[registry_name])
+    spec = registries[registry_name]
+    fname = spec["file"] if isinstance(spec, dict) else spec
+    reg_path = _resolve_config_path(fname)
     if not os.path.exists(reg_path):
         raise ValueError(f"Registry file not found: {reg_path}")
 

environment.yml

@@ -1,14 +1,17 @@
 name: dms_datastore
 channels:
-  - cadwr-dms
   - conda-forge
+  - cadwr-dms
   - nodefaults
 dependencies:
   - python=3.12
   - pip
   - beautifulsoup4
-  - dask
-  - numba
+  - dask>=0.6.24
+  - numpy>=2.0
+  - pandas>=2  
+  - numba>=0.61
+  - vtools3>=3.9.21  
   - scipy
   - tqdm
   - openpyxl
@@ -17,11 +20,8 @@ dependencies:
   - paramiko # for sftp cimis requests
   - tabula-py>=2.9.0 # for pdf parsing
   - jpype1
-  - numpy
-  - pandas>=2
   - cfgrib
   - xarray
-  - vtools3
   - pyyaml
   - gdal
   - pytz # a new hrrr dependency ?

jenkins.environment.yml

@@ -1,14 +1,14 @@
 name: dms_datastore
 channels:
-  - cadwr-dms
   - conda-forge
+  - cadwr-dms
   - nodefaults
 dependencies:
   - python=3.12
   - pip
   - beautifulsoup4
-  - dask
-  - numba
+  - dask>=2024
+  - numba>=2.0
   - scipy
   - tqdm
   - openpyxl

pyproject.toml

@@ -40,9 +40,9 @@ dependencies = [
     "omegaconf>=2.3",
     "paramiko",
     "pandas>=2",
-    "numpy",
+    "numpy>2",
     "xarray",
-    "dask",
+    "dask>2024",
     "scikit-learn",
     "matplotlib",
     "cfgrib",