Add process_elev script and changed site_key to a hardwired name station_id.

This simplifies away site_id.

Author: Eli

dms_datastore/__main__.py

@@ -30,6 +30,7 @@
 from dms_datastore.rationalize_time_partitions import rationalize_time_partitions_cli
 from dms_datastore.spot_check import spot_check_cli
 from dms_datastore.populate_daily import populate_daily_cli
+from dms_datastore.processed.process_elev_data import process_elev_cli
 
 @click.group(help="DMS CLI tools for data processing and extraction.")
 @click.help_option("-h", "--help")  # Add the help option at the group level
@@ -69,6 +70,7 @@ def cli():
 cli.add_command(rationalize_time_partitions_cli, "rationalize_time_partitions")
 cli.add_command(populate_daily_cli, "populate_daily")
 cli.add_command(spot_check_cli, "spot_check")
+cli.add_command(process_elev_cli, "process_elev")
 
 
 if __name__ == "__main__":

dms_datastore/config_data/dstore_config.yaml

@@ -3,26 +3,25 @@ variable_mappings: variable_mappings.csv
 variable_definitions: variables.csv
 sublocations: station_subloc.csv
 ncro_inventory: ncro_inventory_full.csv
-
 daily_data_manifest: daily_data_manifest.csv
 daily_dropbox_spec: daily_dropbox.yaml
-
 file_deletion_list: non_15_min_files_checked.txt
 compare_excepts_formatted: compare_excepts_formatted.txt
 des_rationalize_time_spec: des_rationalize_cfg.yaml
-
 repo_base: "//cnrastore-bdo/Modeling_Data/repo"
-
-default_repo: screened
-
 spot_check_spec: spot_check_spec.yaml
 
-# registry files available to repos
+# Repository info
+default_repo: screened
 
 
+# aliases for registry files available to repos in their "registry"
+# 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.
 registries:
   continuous: station_dbase.csv
-  processed: processed_series.csv
+  processed_synthetic: processed_registry.csv
   structures: structures_registry.csv
   daily: station_dbase.csv
 
@@ -31,15 +30,10 @@ repos:
   formatted:
     root: "//cnrastore-bdo/Modeling_Data/repo/continuous/formatted"
 
-    # Registry providing supplementary metadata (e.g., agency, location)
-    registry: continuous
-
-    # Primary identifier for a site (station, structure, etc.). This will be a key for the registry 
-    # and is always part of filename templates, usually in the second slot
-    # Examples: structure_id or station_id
-    site_key: station_id
-
-    # Depending on settings for the repo, multiple sources (e.g., ncro, cdec) may supply the same series
+    # Registry providing supplementary metadata (e.g., agency, location) for stations
+    registry: continuous   # this is an alias set up above for station_dbase.csv. 
+                        
+    # Depending on settings for the repo, multiple sources (e.g., ncro, cdec) may supply the same time series
     # For instance Bacon Island conductance is observed by NCRO but distributed both by NCRO (QA/QC'd data on Water Data Library)
     # and on CDEC (realtime, not QA/QC). This field, in this case {source} appears in the filename template
     # (typically first slot) and drives globbing and merge order
@@ -88,7 +82,6 @@ repos:
   screened:
     root: "//cnrastore-bdo/Modeling_Data/repo/continuous/screened"
     registry: continuous
-    site_key: station_id
     # screened data is assumed to be grouped according to the original agency/program that manages the station.
     # so whereas in formatted you might have both "ncro" and "cdec" sources for a station, here they will be unified under ncro.    
     provider_key: agency
@@ -102,8 +95,8 @@ repos:
 
   processed:
     root: "//cnrastore-bdo/Modeling_Data/repo/continuous/processed"
-    registry: processed
-    site_key: station_id
+    registry: [continuous, processed_synthetic]
+
     provider_key: processor
     provider_resolution_mode: assume_unique     # This may have to be adjusted for cases like SF harmonic tides where there is a noaa and dwr version
     filename_templates:
@@ -113,23 +106,21 @@ repos:
       - "{processor}_{station_id@subloc}_{param}_{year}.csv"
       - "{processor}_{station_id@subloc}_{param}.csv"
     file_key: [processor, station_id, subloc, param, modifier, shard]
-    data_key: [station_id, subloc, param, modifier]
+    data_key: [processor, subloc, param, modifier]
 
   structures_formatted:
     root: "//cnrastore-bdo/Modeling_Data/repo/structures/formatted"
     registry: structures
-    site_key: structure_id
     provider_key: processor
     provider_resolution_mode: assume_unique
     filename_templates:
-      - "{processor}_{structure_id@subloc}_{param@modifier}.csv"
-      - "{processor}_{structure_id@subloc}_{param@modifier}_{year}.csv"
+      - "{processor}_{station_id@subloc}_{param@modifier}.csv"
+      - "{processor}_{station_id@subloc}_{param@modifier}_{year}.csv"
 
 
   daily_formatted:
     root: "//cnrastore-bdo/Modeling_Data/repo/daily/formatted"
     registry: continuous  # aliases the station data base that provides supplementary metadata like geolocation and agency
-    site_key: station_id  # registry and template use this term for the site's identifier (was "key")
     provider_key: source  # provider is the agency mainly involved in delivering the data. 
                           # Generic for "source" (datamart) "agency" (collecting agency) and
 #                         #  "processor" (group synthesizing, manipulating or filling data)

dms_datastore/dropbox_data.py

@@ -175,7 +175,7 @@ def _resolve_metadata_value(field_name, raw_value, inferred_meta, registry_row):
 
 def _registry_row_for_metadata(name, merged, repo_name):
     rcfg = repo_config(repo_name)
-    site_key = rcfg["site_key"]
+    site_key = "station_id"
     slookup = repo_registry(repo_name)
 
     if site_key in merged and merged[site_key] not in (None, "", "infer_from_filename"):
@@ -197,7 +197,7 @@ def _registry_row_for_metadata(name, merged, repo_name):
 
 def populate_meta(fpath, listing, repo_name, meta_out=None):
     rcfg = repo_config(repo_name)
-    site_key = rcfg["site_key"]
+    site_key = "station_id"
 
     meta = dict(listing.get("metadata", {}) or {})
     name = listing.get("name", "<unnamed>")
@@ -330,7 +330,7 @@ def _maybe_rename_value_column(ts, splice_args):
 
 def _check_metadata(meta, repo_name):
     rcfg = repo_config(repo_name)
-    site_key = rcfg["site_key"]
+    site_key = "station_id"
 
     required = [
         site_key,

dms_datastore/dstore_config.py

@@ -402,31 +402,65 @@ def repo_registry(repo=None, repo_cfg=None):
     In contrast, :func:`registry_df` returns a raw table without
     any guarantees about identity or uniqueness.
     """
+    site_key = "station_id"
     cfg = coerce_repo_config(repo=repo, repo_cfg=repo_cfg)
-    registry_name = cfg.get("registry")
-    site_key = cfg["site_key"]
+    registry_names = cfg.get("registry")
 
-    if registry_name is None:
+    if registry_names is None:
         raise ValueError(f"Repo {cfg['name']!r} has no named registry")
 
-    db = registry_df(registry_name).copy()
+    if isinstance(registry_names, str):
+        registry_names = [registry_names]
 
-    if site_key not in db.columns:
-        raise ValueError(
-            f"Registry {registry_name} is missing key column {site_key}"
+    dfs = []
+    for rname in registry_names:
+        sub = registry_df(rname).copy()
+        if site_key not in sub.columns:
+            raise ValueError(
+                f"Registry {rname} is missing key column {site_key}"
+            )
+        sub[site_key] = (
+            sub[site_key].astype(str).str.replace("'", "", regex=True).str.strip()
         )
+        dfs.append(sub)
+
+    # Merge on common columns
+    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"
+            )
 
-    db[site_key] = (
-        db[site_key].astype(str).str.replace("'", "", regex=True).str.strip()
-    )
+        # Check for key overlap across sub-registries
+        all_keys = []
+        for i, sub in enumerate(dfs):
+            keys = set(sub[site_key].values)
+            overlap = keys & set(all_keys)
+            if overlap:
+                raise ValueError(
+                    f"Key overlap across registries: {sorted(overlap)}"
+                )
+            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()
     if dup.any():
-        raise ValueError(f"Registry {registry_name} has duplicate {site_key} keys")
+        duped = db.loc[dup, site_key].tolist()
+        raise ValueError(
+            f"Registry has duplicate {site_key} keys: {duped}"
+        )
 
     db = db.set_index(site_key, drop=False)
     db.index = db.index.astype(str)
-    db.index.name = "site_id"
+    db.index.name = None
     return db
 
 
@@ -487,7 +521,7 @@ def repo_config(repo_name):
 
     spec = dict(repos[repo_name])
 
-    required = ["site_key", "provider_key", "provider_resolution_mode", "filename_templates"]
+    required = ["provider_key", "provider_resolution_mode", "filename_templates"]
     missing = [k for k in required if k not in spec]
     if missing:
         raise ValueError(f"Repo {repo_name!r} missing required keys: {missing}")

dms_datastore/filename.py

@@ -12,7 +12,7 @@ def naming_spec(repo=None, repo_cfg=None, templates=None):
     if repo_cfg is not None:
         return {
             "filename_templates": list(repo_cfg.get("filename_templates", [])),
-            "site_key": repo_cfg["site_key"],
+            "site_key": "station_id",
             "provider_key": repo_cfg.get("provider_key"),
             "name": repo_cfg.get("name"),
         }
@@ -22,7 +22,7 @@ def naming_spec(repo=None, repo_cfg=None, templates=None):
         rcfg = dstore_config.repo_config(repo)
         return {
             "filename_templates": list(rcfg.get("filename_templates", [])),
-            "site_key": rcfg["site_key"],
+            "site_key": "station_id",
             "provider_key": rcfg.get("provider_key"),
             "name": rcfg.get("name"),
         }
@@ -128,7 +128,7 @@ def build_repo_globs(
     if not templates:
         raise ValueError(f"Repo {repo_cfg.get('name')!r} has no filename_templates")
 
-    site_key = repo_cfg["site_key"]
+    site_key = "station_id"
     provider_key = repo_cfg.get("provider_key")
 
     if providers is None:
@@ -264,7 +264,7 @@ def _template_regex_from_template(template, *, key_column="station_id"):
 def _interpret_fname_template(fname, repo_cfg):
     fname = os.path.split(fname)[1]
     templates = repo_cfg.get("filename_templates", [])
-    site_key = repo_cfg.get("site_key", "station_id")
+    site_key = "station_id"
     if not templates:
         raise ValueError(
             f"Repo {repo_cfg.get('name')!r} has no filename_templates for template parse"

dms_datastore/inventory.py

@@ -123,7 +123,7 @@ def repo_data_inventory(repo=None, *, repo_cfg=None, in_path=None, registry=None
     root = in_path if in_path is not None else repo_cfg["root"]
     registry = repo_registry(repo_cfg=repo_cfg)
 
-    site_key = repo_cfg["site_key"]
+    site_key = "station_id"
     provider_key = repo_cfg["provider_key"]
 
     allfiles = _inventory_files(root)
@@ -190,23 +190,12 @@ def repo_data_inventory(repo=None, *, repo_cfg=None, in_path=None, registry=None
 
     grouped[site_key] = grouped[site_key].astype(str).str.strip()
 
-    if registry.index.name != site_key:
-        if site_key not in registry.columns:
-            raise ValueError(
-                f"Registry missing join key column {site_key!r}; "
-                f"columns are {registry.columns.tolist()}"
-            )
-        registry = registry.copy()
-        registry[site_key] = registry[site_key].astype(str).str.strip()
-        registry = registry.set_index(site_key, drop=False)
-    else:
-        registry = registry.copy()
-        registry.index = registry.index.astype(str)
-        if site_key in registry.columns:
-            registry[site_key] = registry[site_key].astype(str).str.strip()
+    # Registry is already indexed by station_id from repo_registry()
+    # Drop the station_id column before join to avoid pandas ambiguity
+    reg_for_join = registry.drop(columns=[site_key], errors="ignore")
 
     metastat = grouped.join(
-        registry,
+        reg_for_join,
         on=site_key,
         rsuffix="_registry",
         how="left",
@@ -251,7 +240,7 @@ def repo_file_inventory(repo=None, *, repo_cfg=None, in_path=None):
     root = in_path if in_path is not None else repo_cfg["root"]
     registry = repo_registry(repo_cfg=repo_cfg)
 
-    site_key = repo_cfg["site_key"]
+    site_key = "station_id"
 
     allfiles = _inventory_files(root)
     allmeta = _parse_inventory_meta(allfiles, repo_cfg=repo_cfg)
@@ -309,8 +298,11 @@ def repo_file_inventory(repo=None, *, repo_cfg=None, in_path=None):
             f"Cannot join registry: grouped inventory missing site key {site_key!r}"
         )
 
+    # Drop the station_id column before join to avoid pandas ambiguity
+    reg_for_join = registry.drop(columns=[site_key], errors="ignore")
+
     metastat = grouped.join(
-        registry,
+        reg_for_join,
         on=site_key,
         rsuffix="_registry",
         how="left",

dms_datastore/process_station_variable.py

@@ -282,9 +282,8 @@ def attach_agency_id(
         )
 
     repo_cfg = dstore_config.repo_config(repo_name)
-    registry_name = repo_cfg["registry"]
-    site_key = repo_cfg["site_key"]
-    registry = dstore_config.registry_df(registry_name).copy()
+    site_key = "station_id"
+    registry = dstore_config.repo_registry(repo_cfg=repo_cfg)
 
     if site_key not in registry.columns:
         raise ValueError(f"Registry site key column {site_key!r} not found")
@@ -298,7 +297,7 @@ def attach_agency_id(
         raise ValueError(f"Requested registry column(s) not found: {missing_cols}")
 
     lookup = registry[[site_key] + required_cols].copy()
-    lookup = lookup.rename(columns={site_key: "station_id"})
+    lookup = lookup.reset_index(drop=True)
 
     merged = df.merge(lookup, on="station_id", how="left")