Distinguish better between agency_id and lookup id

Distinguish between source site id and agency_id for cdec

Eliminate hard-wired conception of series_id based on stations database

Author: Eli

dms_datastore/config_data/dstore_config.yaml

@@ -79,7 +79,7 @@ repos:
 
     # Data identity: uniquely identifies a logical time series independent of provider
     # Multiple files (different sources) may map to the same data_key
-    data_key: [station_id, subloc, param, shard]
+    data_key: [station_id, subloc, param]
 
 
   screened:
@@ -109,7 +109,7 @@ 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, shard]
+    data_key: [station_id, subloc, param, modifier]
 
   structures:
     root: "//cnrastore-bdo/Modeling_Data/repo/structures"

dms_datastore/download_cdec.py

@@ -12,6 +12,7 @@
 import zipfile
 import os
 import string
+from pathlib import Path
 import datetime as dt
 import time
 import numpy as np
@@ -39,7 +40,7 @@ def download_station_data(
 ):
     station = row.station_id
     try:
-        cdec_id = row.cdec_id.lower()
+        cdec_id = row.src_site_id.upper()
     except Exception:
         cdec_id = station
 
@@ -266,7 +267,8 @@ def download_cdec(
     df = attach_agency_id(
         df,
         repo_name="formatted",
-        agency_id_col="cdec_id",
+        agency_id_col="agency_id",
+        src_site_id_col="cdec_id",
         on_missing="drop",
     )
     vlookup = dstore_config.config_file("variable_mappings")
@@ -329,6 +331,9 @@ def download_cdec(
     help="Frequency code(s): E (event), H (hourly), D (daily), M (monthly). "
      "Default is E,H. Multiple values allowed as comma-separated list (e.g., E,H,D)."
 )
+@click.option("--logdir", type=click.Path(path_type=Path), default=None)
+@click.option("--debug", is_flag=True)
+@click.option("--quiet", is_flag=True)
 @click.argument("stationfile", nargs=-1)
 def download_cdec_cli(
     dest_dir,
@@ -341,9 +346,25 @@ def download_cdec_cli(
     stationfile,
     overwrite,
     freq,
+    logdir,
+    debug,
+    quiet,
 ):
     """CLI for downloading CDEC water data."""
-
+    level, console = resolve_loglevel(
+        debug=debug,
+        quiet=quiet,
+    )
+    logger.debug(f"Logging level set to {logging.getLevelName(level)} and console={console}")
+
+    configure_logging(
+          package_name="dms_datastore",
+          level=level,
+          console=not quiet,
+          logdir=logdir,
+          logfile_prefix="download_ncro"
+    )
+    logger.debug("Starting NCRO download")
     download_cdec(
         dest_dir,
         id_col,

dms_datastore/dstore_config.py

@@ -426,6 +426,7 @@ def repo_registry(repo=None, repo_cfg=None):
 
     db = db.set_index(site_key, drop=False)
     db.index = db.index.astype(str)
+    db.index.name = "site_id"
     return db
 
 
@@ -590,6 +591,17 @@ def registry_df(registry_name):
             db["agency_id"].astype(str).str.replace("'", "", regex=True).str.strip()
         )
 
+    if "agency_id" in db.columns:
+        db["agency_id"] = (
+            db["agency_id"].astype(str).str.replace("'", "", regex=True).str.strip()
+        )
+
+        missing_agency = db["agency_id"].isna() | (db["agency_id"] == "") | (db["agency_id"].str.lower() == "nan")
+        if missing_agency.any():
+            bad_rows = db.loc[missing_agency]
+            raise ValueError(
+                f"Registry {registry_name} has missing agency_id values in {len(bad_rows)} row(s)"
+            )
     _registry_cache[registry_name] = db
     return db
 

dms_datastore/inventory.py

@@ -58,36 +58,35 @@ def _inventory_files(root):
 def _parse_inventory_meta(allfiles, repo_cfg=None):
     return [interpret_fname(fname, repo_cfg=repo_cfg) for fname in allfiles]
 
+
 def series_id_from_meta(meta, repo_cfg=None, remove_provider=False):
     """
     Construct a stable logical series identifier from parsed metadata.
 
     Parameters
     ----------
-    meta : dict
-        Parsed filename metadata
+    meta : dict-like
+        Parsed filename metadata or row-like object.
     repo_cfg : dict or None
-        Repo configuration (defines provider_key, site_key)
+        Repo configuration. When provided, identity is driven by repo_cfg["data_key"].
+        When omitted, a legacy station-style fallback is used.
     remove_provider : bool
-        If True, omit provider from identity
+        If True, omit provider from identity.
 
     Returns
     -------
     str
         series_id
     """
-
-    # --- resolve keys ---
     if repo_cfg is None:
-        provider_key = "agency"       # legacy fallback
-        site_key = "station_id"
+        provider_key = "agency"
+        data_key = ["station_id", "subloc", "param"]
     else:
         provider_key = repo_cfg["provider_key"]
-        site_key = repo_cfg["site_key"]
+        data_key = repo_cfg["data_key"]  # let missing data_key fail naturally for now
 
     parts = []
 
-    # --- provider ---
     if not remove_provider:
         provider_val = meta.get(provider_key)
         if provider_val is None:
@@ -96,29 +95,19 @@ def series_id_from_meta(meta, repo_cfg=None, remove_provider=False):
             )
         parts.append(str(provider_val))
 
-    # --- site ---
-    site_val = meta.get(site_key)
-    if site_val is None:
-        raise ValueError(
-            f"Missing site field {site_key!r} in metadata: {meta}"
-        )
-    parts.append(str(site_val))
-
-    # --- subloc ---
-    subloc = meta.get("subloc")
-    if subloc not in (None, "default"):
-        parts.append(str(subloc))
-
-    # --- param ---
-    param = meta.get("param")
-    if param is None:
-        raise ValueError(f"Missing param in metadata: {meta}")
-    parts.append(str(param))
-
-    # --- modifier ---
-    modifier = meta.get("modifier")
-    if modifier is not None:
-        parts.append(str(modifier))
+    for key in data_key:
+        val = meta.get(key)
+
+        # suppress conventional empty sublocation/modifier noise
+        if key in ("subloc", "modifier") and val in (None, "", "default", "none"):
+            continue
+
+        if val is None:
+            raise ValueError(
+                f"Missing data_key field {key!r} in metadata: {meta}"
+            )
+
+        parts.append(str(val))
 
     return "|".join(parts)
 

dms_datastore/populate_repo.py

@@ -289,47 +289,70 @@ def revise_filename_syear_eyear(pat, force=True, outfile="rename.txt"):
             logger.info(b)
     logger.info(f"Renaming complete for pattern: {pat}")
 
-
 def populate_repo(
     agency, param, dest, start, end, overwrite=False, ignore_existing=None
 ):
     """Populate repository for the given agency/source and parameter."""
     maximize_subloc = False
 
-    slookup = dstore_config.config_file("station_dbase")
     if "ncro" in agency:
         vlookup = mapping_df
         agency = "ncro"
     else:
         vlookup = dstore_config.config_file("variable_mappings")
 
-    subloclookup = dstore_config.config_file("sublocations")
-    df = pd.read_csv(slookup, sep=",", comment="#", header=0, dtype={"agency_id": str})
+    # Use repo-aware registry access instead of reading station_dbase CSV directly.
+    slookup = dstore_config.repo_registry("formatted").copy()
+
     filter_agency = "dwr_ncro" if agency == "ncro" else agency
-    df = df.loc[df.agency.str.lower() == filter_agency, :]
-    df["agency_id"] = df["agency_id"].str.replace("'", "", regex=True)
+    slookup = slookup.loc[slookup.agency.str.lower() == filter_agency, :]
+    name_lookup = (
+        slookup.loc[:, ["station_id", "name"]]
+        .drop_duplicates(subset=["station_id"])
+    )
 
-    dfsub = read_station_subloc(subloclookup)
-    df = merge_station_subloc(df, dfsub, default_z=-0.5)
 
-    df = df.reset_index()
+    dfsub = read_station_subloc(dstore_config.config_file("sublocations"))
+    slookup = merge_station_subloc(slookup, dfsub, default_z=-0.5)
 
     if ignore_existing is not None:
-        df = df[~df["station_id"].isin(ignore_existing)]
+        slookup = slookup[~slookup["station_id"].isin(ignore_existing)]
 
     dest_dir = dest
     source = "cdec" if agency in ["dwr", "usbr"] else agency
-    agency_id_col = "cdec_id" if source == "cdec" else "agency_id"
+    agency_id_col = "agency_id"
+    src_site_id_col = "cdec_id" if source == "cdec" else None
 
-    df = df[["station_id", "subloc"]]
+    # Preserve only the station display name, and only outside the standard
+    # request-building pipeline.
+    slookup = slookup.reset_index()
+    df_req = slookup.loc[:, ["station_id", "subloc"]]
 
     stationlist = normalize_station_request(
-        stationframe=df,
+        stationframe=df_req,
         param=param,
         default_subloc="default",
     )
-    stationlist = attach_agency_id(stationlist, repo_name="formatted", agency_id_col=agency_id_col)
+    stationlist = attach_agency_id(
+        stationlist,
+        repo_name="formatted",
+        agency_id_col=agency_id_col,
+        src_site_id_col=src_site_id_col,
+        on_missing="drop" if src_site_id_col is not None else "raise",
+    )
     stationlist = attach_src_var_id(stationlist, vlookup, source=source)
+
+    stationlist = stationlist.merge(
+        name_lookup,
+        on="station_id",
+        how="left",
+        validate="many_to_one",
+    )
+
+    if stationlist["name"].isna().any():
+        missing = stationlist.loc[stationlist["name"].isna(), "station_id"].tolist()
+        raise ValueError(f"Missing station name for station_id(s): {missing}")
+
     if maximize_subloc:
         stationlist["subloc"] = "default"
         if param not in ["flow", "elev"]:
@@ -386,7 +409,8 @@ def supplement_ncro_with_cdec(df, dest, start, overwrite=False, ignore_existing=
         df = df[~df["station_id"].isin(ignore_existing)]
 
     source = "cdec"
-    agency_id_col = "cdec_id"
+    agency_id_col = "agency_id"
+    src_site_id_col = "cdec_id"
 
     stationlist = normalize_station_request(
         stationframe=df,
@@ -397,6 +421,7 @@ def supplement_ncro_with_cdec(df, dest, start, overwrite=False, ignore_existing=
         stationlist,
         repo_name="formatted",
         agency_id_col=agency_id_col,
+        src_site_id_col=src_site_id_col,
         on_missing="drop",
     )