Simplified config, better implementation and testing of filename templates, inventory and read_ts_repo. Format uses config based template.

Author: Eli

dms_datastore/auto_screen.py

@@ -18,6 +18,7 @@
 from dms_datastore.dstore_config import *
 from dms_datastore.inventory import *
 from dms_datastore.write_ts import *
+from dms_datastore.filename import meta_to_filename
 from schimpy.station import *
 import geopandas as gpd
 import numpy as np
@@ -148,22 +149,20 @@ def plot_anomalies(
     plt.close(fig)
 
 
+
+
 def filter_inventory_(inventory, stations, params):
     if stations is not None:
         if isinstance(stations, str):
             stations = [stations]
-        inventory = inventory.loc[
-            inventory.index.get_level_values("station_id").isin(stations), :
-        ]
+        inventory = inventory.loc[inventory["station_id"].isin(stations), :]
+
     if params is not None:
         if isinstance(params, str):
             params = [params]
-            logger.debug(f"params: {params}")
-        inventory = inventory.loc[
-            inventory.index.get_level_values("param").isin(params), :
-        ]
-    return inventory
+        inventory = inventory.loc[inventory["param"].isin(params), :]
 
+    return inventory
 
 def auto_screen(
     fpath="formatted",
@@ -208,29 +207,29 @@ def auto_screen(
 
     active = start_station is None
     station_db = station_dbase()
-    inventory = repo_data_inventory(fpath)
+
+    source_repo = "formatted"
+    actual_fpath = fpath if fpath is not None else repo_root(source_repo)
+    inventory = repo_data_inventory(repo="formatted",in_path=actual_fpath) # repo is the config repo, in_path is the data storage location
     inventory = filter_inventory_(inventory, stations, params)
     failed_read = []
 
     for index, row in inventory.iterrows():
-        station_id = index[0]
-        if not active:
-            if station_id == start_station:
-                active = True
-            else:
-                continue
-        subloc = index[1]
-        if type(subloc) == float:
+        station_id = row["station_id"]
+        subloc = row["subloc"]
+        param = row["param"]
+
+        if pd.isna(subloc) or subloc is None:
             subloc = "default"
-        param = index[2]
+
         if subloc is None:
             subloc = "default"
         if np.random.uniform() < 0.0:  # 0.95:
             logger.debug(f"Randomly rejecting: {station_id} {subloc} {param}")
             continue
-        filename = str(row.filename)
+
         station_info = station_db.loc[station_id, :]
-        agency = row.agency_dbase
+        agency = row["agency_registry"] if "agency_registry" in row.index else row["agency"]
         if agency.startswith("dwr_"):
             agency = agency[4:]  # todo: need to take care of des_ vs dwr_des etc
 
@@ -240,9 +239,11 @@ def auto_screen(
         # these may be lists
         try:
             # logger.debug(f"fetching {fpath},{station_id},{param}")
-            meta_ts = fetcher(fpath, station_id, param, subloc=subloc)
-        except:
-            logger.warning(f"Read failed for {fpath}, {station_id}, {param}, {subloc}")
+            meta_ts = fetcher(source_repo, station_id, param, subloc=subloc, data_path=actual_fpath)
+        except Exception as e:
+            logger.warning(f"Read failed for {actual_fpath}, {station_id}, {param}, {subloc}, storage loc = {actual_fpath}")
+            logger.exception(e)
+            print(e)
             meta_ts = None
 
         if meta_ts is None:
@@ -276,12 +277,23 @@ def auto_screen(
         if "value" in screened.columns:
             screened = screened[["value", "user_flag"]]
         meta["screen"] = proto
-        if subloc_actual and subloc_actual != "default":
-            output_fname = (
-                f"{agency}_{station_id}@{subloc_actual}_{row.agency_id}_{param}.csv"
-            )
-        else:
-            output_fname = f"{agency}_{station_id}_{row.agency_id}_{param}.csv"
+        
+        # Build output filename using configured naming spec for screened repo
+        output_meta = {
+            "agency": agency,
+            "station_id": station_id,
+            "subloc": subloc_actual if subloc_actual != "default" else None,
+            "param": param,
+            "agency_id": row.agency_id,
+        }
+        # Add year info if available from metadata
+        if "year" in meta:
+            output_meta["year"] = meta["year"]
+        elif "syear" in meta and "eyear" in meta:
+            output_meta["syear"] = meta["syear"]
+            output_meta["eyear"] = meta["eyear"]
+        
+        output_fname = meta_to_filename(output_meta, repo="screened",include_shard=False)
         output_fpath = os.path.join(dest, output_fname)
         logger.debug("start write")
         write_ts_csv(screened, output_fpath, meta, chunk_years=True)
@@ -479,24 +491,26 @@ def spatial_config(configfile, x, y):
     return checker.region_info(x, y)
 
 
-def ncro_fetcher(repo_path, station_id, param, subloc):
-    """Reads NCRO data, correctly folding together NCRO and CDEC by priority.
-    Celsius is converted to Farenheit
-    """
+def ncro_fetcher(repo, station_id, param, subloc, data_path=None):
     return read_ts_repo(
         station_id,
         param,
         subloc=subloc,
         src_priority=["ncro", "cdec"],
-        repo=repo_path,
+        repo=repo,
+        data_path=data_path,
         meta=True,
     )
 
-
-def general_fetcher(repo_path, station_id, param, subloc):
-    """Fetches from a well behaved and standard repo"""
-    return read_ts_repo(station_id, param, subloc=subloc, repo=repo_path, meta=True)
-
+def general_fetcher(repo, station_id, param, subloc, data_path=None):
+    return read_ts_repo(
+        station_id,
+        param,
+        subloc=subloc,
+        repo=repo,
+        data_path=data_path,
+        meta=True,
+    )
 
 def custom_fetcher(agency):
     if agency in ["ncro", "dwr_ncro"]:

dms_datastore/config_data/des_rationalize_cfg.yaml

@@ -68,4 +68,5 @@ rationalize:
       #   B supersedes A if B.syear <= A.syear and B.eyear >= A.eyear
       #
       # IMPORTANT: If any file would end up in both include and omit after expansion,
-      # that is an error. Be explicit instead.
+      # that is an error. Be explicit instead.
+      

dms_datastore/config_data/dstore_config.yaml

@@ -47,28 +47,25 @@ repos:
     search:
       use_source_slot: true
       shard_style: auto
-    parse:
-      style: legacy
+
 
   screened:
     root: "//cnrastore-bdo/Modeling_Data/repo/continuous/screened"
     registry: continuous
-    key_column: id
+    key_column: station_id
     source_priority_mode: by_registry_column
     source_priority_column: agency
     filename_templates:
-      - "{source}_{key@subloc}_{agency_id}_{param}_{syear}_{eyear}.csv"
-      - "{source}_{key@subloc}_{agency_id}_{param}_{year}.csv"
+      - "{agency}_{key@subloc}_{agency_id}_{param}_{year}.csv"
     search:
       use_source_slot: true
       shard_style: auto
-    parse:
-      style: legacy
+
 
   processed:
     root: "//cnrastore-bdo/Modeling_Data/repo/continuous/processed"
     registry: processed
-    key_column: id
+    key_column: station_id
     source_priority_mode: repo_default
     source_priority_group: processed_default
     filename_templates:
@@ -79,13 +76,11 @@ repos:
     search:
       use_source_slot: true
       shard_style: auto
-    parse:
-      style: template
 
   structures:
     root: "//cnrastore-bdo/Modeling_Data/repo/structures"
     registry: structures
-    key_column: id
+    key_column: structure_id
     source_priority_mode: repo_default
     source_priority_group: structures_default
     filename_templates:

dms_datastore/download_ncro.py

@@ -690,16 +690,9 @@ def test():
             param_lookup=vlookup,
             source="ncro",
         )
-        print(df)
         ncro_download(df, destdir, stime, etime, overwrite=overwrite)
 
 
-def test_read():
-    fname = "ncro_old_b95380_temp_2015_2024.csv"
-    fname = "ncro_orm_b95370_cla_*.csv"
-    ts = read_ts.read_ts(fname)
-    print(ts)
-
 
 @click.command(
     help=(

dms_datastore/dstore_config.py

@@ -73,7 +73,6 @@ def repo_root(repo=None, repo_cfg=None):
 
 def repo_registry(repo=None, repo_cfg=None):
     cfg = coerce_repo_config(repo=repo, repo_cfg=repo_cfg)
-    print(cfg)
     registry_name = cfg.get("registry")
     key_column = cfg.get("key_column", "id")
     if registry_name is None:
@@ -179,8 +178,6 @@ def repo_config(repo_name):
     )
     spec.setdefault("key_column", "id")
     spec.setdefault("source_priority_mode", "none")
-    spec.setdefault("parse", {"style": "legacy"})
-    spec.setdefault("search", {"use_source_slot": True, "shard_style": "auto"})
 
     templates = spec.get("filename_templates", [])
     if not templates:
@@ -232,7 +229,7 @@ def registry_df(registry_name, key_column="id"):
         )
 
     db[key_column] = db[key_column].astype(str).str.replace("'", "", regex=True).str.strip()
-    print(db)
+   
 
     if "agency_id" in db.columns:
         db["agency_id"] = db["agency_id"].astype(str).str.replace("'", "", regex=True).str.strip()
@@ -244,6 +241,8 @@ def registry_df(registry_name, key_column="id"):
         raise ValueError(f"Registry {registry_name} has duplicate {key_column} keys")
 
     db = db.set_index(key_column, drop=False)
+    # Ensure index is string type for consistent merging
+    db.index = db.index.astype(str)
     return db