From e4ba507b8d350fdc8fd613660271b650eca99506 Mon Sep 17 00:00:00 2001 From: Eric Novotny Date: Fri, 4 Sep 2026 09:13:55 -0700 Subject: [PATCH 1/2] add multiple offices --- README.md | 3 + rtd_docs/using-cda-loader.rst | 7 +- shef/loaders/cda_loader.py | 207 ++++++++++-- .../test_cda_loader_make_export_transforms.py | 301 ++++++++++++++++++ 4 files changed, 490 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index 61f4493..8253fe1 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,9 @@ pip install shef-parser ```sh #CWMS CDA loader shefParser -i input_filename --loader cda[$API_ROOT][$API_KEY] +# optional office scoping for transform lookup +shefParser -i input_filename --loader 'cda[$API_ROOT][$API_KEY][MVP]' +shefParser -i input_filename --loader 'cda[$API_ROOT][$API_KEY]["MVP","LRL","SWG"]' ``` ```sh diff --git a/rtd_docs/using-cda-loader.rst b/rtd_docs/using-cda-loader.rst index fd64f46..ba05701 100644 --- a/rtd_docs/using-cda-loader.rst +++ b/rtd_docs/using-cda-loader.rst @@ -13,14 +13,15 @@ This page does not cover unloading with the ``--unload`` command line option, wh Command Line ------------ -To use the :py:`CdaLoader` class, specify ``--loader cda[][]`` on the command line, where +To use the :py:`CdaLoader` class, specify ``--loader cda[][][]`` on the command line, where * ```` is the URL to the Cwms Data API (e.g., ``https://cwms-data-test.cwbi.us/cwms-data/`` for the CWBI test database) * ```` is your personal authentication key for the database referenced by the URL +* ```` is optional and can be a single office code (for example, ``MVP``) or a list of offices (for example, ``['MVP','LRL','SWG']`` or ``["MVP","LRL","SWG"]``) It is recommended to use environment variables to hold the URL root and API key so that your command line would look -something like ``run_shef_parser --loader cda[%CDA_URL_ROOT%][%CDA_API_KEY%]`` on Windows or ``run_shef_parser --loader cda[$CDA_URL_ROOT][$CDA_API_KEY]`` -on Linux +something like ``run_shef_parser --loader cda[%CDA_URL_ROOT%][%CDA_API_KEY%][MVP]`` on Windows or ``run_shef_parser --loader cda[$CDA_URL_ROOT][$CDA_API_KEY][MVP]`` +on Linux. For multiple offices, use ``run_shef_parser --loader 'cda[$CDA_URL_ROOT][$CDA_API_KEY]["MVP","LRL","SWG"]'``. Loading Configuration ---------------------- diff --git a/shef/loaders/cda_loader.py b/shef/loaders/cda_loader.py index 6e1df67..be1ee92 100644 --- a/shef/loaders/cda_loader.py +++ b/shef/loaders/cda_loader.py @@ -1,5 +1,6 @@ import asyncio import json +import ast import math import re import time @@ -99,6 +100,8 @@ def __init__( self._input: Optional[Union[BufferedRandom, TextIOWrapper]] = None self._message_count: int = 0 self._office_id: str = "" + # may be single office or list of offices for transform discovery + self._office_ids: list[str] = [] self._parsed_payloads: list[TimeseriesPayload] = [] self._payloads: list[TimeseriesPayload] = [] self._time_series_error_count: int = 0 @@ -108,6 +111,7 @@ def __init__( self._configured_pe_codes = set() self._loaded_export_group_ids: set[str] = set() self._loaded_all_export_groups: bool = False + self._office_load_stats: dict[str, dict[str, int]] = {} def make_shef_transform(self, crit: dict[str, Any]) -> ShefTransform: """ @@ -164,7 +168,36 @@ def set_options(self, options_str: Union[str, None]) -> None: options = self.get_options(options_str) if len(options) > 2: - self._office_id = options[2] + office_opt = options[2] + # accept JSON array, Python-style list, bracketed comma lists, or a single office. + # Common CLI forms include: [LRN,LRL], [[LRN,LRL]], ["MVP","LRL"], and plain LRL. + parsed_offices: list[str] = [] + text = (office_opt or "").strip() + while text.startswith("[") and text.endswith("]"): + text = text[1:-1].strip() + if text: + if text.startswith(("'", '"')) and text.endswith(("'", '"')): + text = text[1:-1].strip() + try: + parsed = json.loads(text) + except Exception: + try: + parsed = ast.literal_eval(text) + except Exception: + parsed = None + if isinstance(parsed, (list, tuple)): + parsed_offices = [str(x).strip().strip("'\"") for x in parsed if str(x).strip()] + elif "," in text: + parsed_offices = [ + x.strip().strip("'\"[]") + for x in text.split(",") + if x.strip() + ] + else: + parsed_offices = [text.strip().strip("'\"[]")] + + self._office_ids = [office for office in parsed_offices if office] + self._office_id = self._office_ids[0] if self._office_ids else "" if len(options) > 1: self._cda_url = options[0] cda_api_key = options[1] @@ -180,14 +213,43 @@ def set_options(self, options_str: Union[str, None]) -> None: api_root=self._cda_url, api_key=None ) # don't need api key if unloading + @staticmethod + def make_transform_key( + office: Optional[str], location: str, parameter_code: str + ) -> str: + """Create the dictionary key used to index SHEF transforms. + + The office is part of the key when present, so the same location/parameter + in different offices does not overwrite one another. The transform value + itself still holds the office for downstream logic. + """ + if office: + return f"{office}.{location}.{parameter_code}" + return f"{location}.{parameter_code}" + @property def transform_key(self) -> str: """ - The transform key for the current SHEF value + The transform key for the current SHEF value. """ self.assert_value_is_set() sv = cast(shared.ShefValue, self._shef_value) - return f"{sv.location}.{sv.parameter_code[:-1]}" + parameter_code = sv.parameter_code[:-1] + candidate_keys = [] + office_candidates: list[str] = [] + if self._office_ids: + office_candidates.extend(self._office_ids) + if self._office_id and self._office_id not in office_candidates: + office_candidates.append(self._office_id) + for office in office_candidates: + candidate_keys.append( + self.make_transform_key(office, sv.location, parameter_code) + ) + candidate_keys.append(self.make_transform_key(None, sv.location, parameter_code)) + for key in candidate_keys: + if key in self._transforms: + return key + raise KeyError(f"No transform found for {sv.location}.{parameter_code}") @property def transform(self) -> ShefTransform: @@ -196,27 +258,69 @@ def transform(self) -> ShefTransform: """ return self._transforms[self.transform_key] - def make_transforms(self) -> None: + def make_transforms( + self, office: Optional[Union[str, list[str]]] = None + ) -> None: """ - Makes the loading transforms + Makes the loading transforms. + + When a single office is requested, query only that office. When multiple or + no offices are requested, fetch the full assignment set in one call and then + filter down to the requested office IDs locally. """ - shef_group = cwms.get_timeseries_group( - group_office_id="CWMS", - category_office_id="CWMS", - group_id="SHEF Data Acquisition", - category_id="Data Acquisition", - ).json - for assigned_ts in shef_group["assigned-time-series"]: - if "timeseries-id" in assigned_ts and "alias-id" in assigned_ts: + requested_offices: list[str] = [] + if office is None: + requested_offices = self._office_ids if self._office_ids else [""] + elif isinstance(office, str): + requested_offices = [office] + else: + requested_offices = [str(o) for o in office] + + group_kwargs: dict[str, Any] = { + "group_office_id": "CWMS", + "category_office_id": "CWMS", + "group_id": "SHEF Data Acquisition", + "category_id": "Data Acquisition", + } + if len(requested_offices) == 1: + group_kwargs["office_id"] = requested_offices[0] + office_filter = set(requested_offices) if len(requested_offices) > 1 else None + + shef_group = cwms.get_timeseries_group(**group_kwargs).json + assigned_ts = shef_group.get("assigned-time-series", []) + if office_filter is not None: + assigned_ts = [ + item for item in assigned_ts if item.get("office-id") in office_filter + ] + if self._logger: + self._logger.debug(f"assigned_ts: {assigned_ts}") + self._logger.debug(f"office_filter: {office_filter}") + self._logger.debug(f"requested_offices: {requested_offices}") + for assigned_item in assigned_ts: + if "timeseries-id" in assigned_item and "alias-id" in assigned_item: try: - transform = self.make_shef_transform(assigned_ts) - transform_key = f"{transform.location}.{transform.parameter_code}" + transform = self.make_shef_transform(assigned_item) + transform_key = self.make_transform_key( + transform.office, + transform.location, + transform.parameter_code, + ) + if transform_key in self._transforms: + if self._logger: + self._logger.warning( + "Duplicate transform for office [%s], location [%s], parameter [%s]; overwriting prior mapping", + transform.office, + transform.location, + transform.parameter_code, + ) self._transforms[transform_key] = transform except Exception as e: if self._logger: self._logger.warning( - f"{str(e)} occurred while processing SHEF criteria for {assigned_ts['timeseries-id']}" + f"{str(e)} occurred while processing SHEF criteria for {assigned_item['timeseries-id']}" ) + if self._logger: + self._logger.debug(f"transforms: {list(self._transforms.keys())}") def get_additional_pe_codes(self, parser_recognized_pe_codes: set[str]) -> set[str]: """ @@ -234,8 +338,22 @@ def get_time_series_name(self, shef_value: Optional[shared.ShefValue]) -> str: raise shared.LoaderException("Empty SHEF value in get_time_series_name()") if not self._transforms: self.make_transforms() - transform_key = f"{shef_value.location}.{shef_value.parameter_code[:-1]}" - return self._transforms[transform_key].timeseries_id + parameter_code = shef_value.parameter_code[:-1] + candidate_keys = [] + office_candidates: list[str] = [] + if self._office_ids: + office_candidates.extend(self._office_ids) + if self._office_id and self._office_id not in office_candidates: + office_candidates.append(self._office_id) + for office in office_candidates: + candidate_keys.append( + self.make_transform_key(office, shef_value.location, parameter_code) + ) + candidate_keys.append(self.make_transform_key(None, shef_value.location, parameter_code)) + for key in candidate_keys: + if key in self._transforms: + return self._transforms[key].timeseries_id + raise KeyError(f"No transform found for {shef_value.location}.{parameter_code}") @staticmethod def get_unix_timestamp(timestamp: str) -> int: @@ -258,6 +376,7 @@ def load_time_series(self) -> None: if self._shef_value and self._time_series: sv = self._shef_value + transform = self.transform if self._logger: self._logger.debug(f"ts_name: {self.get_time_series_name(sv)}") self._logger.debug(f"shef_value: {sv}") @@ -269,16 +388,27 @@ def load_time_series(self) -> None: time_series.append(CdaValue(time, float(ts[1]), 0)) post_data: TimeseriesPayload = { "name": self.get_time_series_name(sv), - "office-id": self.transform.office, - "units": self.transform.units, + "office-id": transform.office, + "units": transform.units, "values": time_series, } match_index = self.find_matching_payload_index(post_data) - if not match_index: + if match_index is None: self._payloads.append(post_data) + office_stats = self._office_load_stats.setdefault( + self.transform.office, + {"time_series": 0, "value_count": 0}, + ) + office_stats["time_series"] += 1 + office_stats["value_count"] += len(time_series) else: match_payload = self._payloads[match_index] match_payload["values"].extend(time_series) + office_stats = self._office_load_stats.setdefault( + self.transform.office, + {"time_series": 0, "value_count": 0}, + ) + office_stats["value_count"] += len(time_series) self._time_series = [] def create_write_task( @@ -329,6 +459,10 @@ async def process_write_tasks(self) -> None: self._logger.info( f"CWMS-Data-API POST tasks complete ({process_time:.2f} seconds)" ) + self._logger.info( + "Loaded by office: %s", + self.get_office_summary(), + ) def find_matching_payload_index( self, payload: TimeseriesPayload @@ -442,7 +576,8 @@ def done(self) -> None: "--[Summary]-----------------------------------------------------------" ) self._logger.info( - f"{self._value_count} values posted in {self._time_series_count} time series" + "Loaded by office: %s", + self.get_office_summary(), ) if self._value_error_count > 0: self._logger.info( @@ -464,6 +599,25 @@ def set_input(self, input_object: Union[StringIO, TextIO, str]) -> None: f"Expected TextIOWrapper or str object, got [{input_object.__class__.__name__}]" ) + def get_office_summary(self) -> str: + """Return a human-readable breakdown of loaded series and values by office.""" + if not self._office_load_stats: + return "No office data loaded" + + summary_parts = [] + for office, stats in sorted(self._office_load_stats.items()): + time_series = stats.get("time_series", 0) + value_count = stats.get("value_count", 0) + summary_parts.append(f"{office}: {time_series} time series, {value_count} values") + return "; ".join(summary_parts) + + def _track_office_load(self, office_id: str, value_count: int = 0) -> None: + """Track the number of time series and values loaded for a given office.""" + if not office_id: + office_id = "DEFAULT" + office_stats = self._office_load_stats.setdefault(office_id, {"time_series": 0, "value_count": 0}) + office_stats["value_count"] += value_count + def make_export_transforms(self, group_id: Optional[str] = None) -> None: if not self._office_id: raise shared.LoaderException( @@ -503,8 +657,10 @@ def make_export_transforms(self, group_id: Optional[str] = None) -> None: continue try: transform = self.make_shef_transform(time_series) - transform_key = ( - f"{transform.location}.{transform.parameter_code}" + transform_key = self.make_transform_key( + self._office_id or transform.office, + transform.location, + transform.parameter_code, ) self._transforms[transform_key] = transform if transform.timeseries_id in tsids_used: @@ -709,9 +865,10 @@ def use_value(self) -> bool: loader_options = ( - "--loader cda[cda_url][cda_api_key]\n" + "--loader cda[cda_url][cda_api_key][office]\n" "* cda_url = the url of the CDA instance to be used, e.g. https://cwms-data.usace.army.mil/cwms-data/\n" "* cda_api_key = the api_key to use for CDA POST requests\n" + "* office = optional office code or list of office codes used to scope SHEF Data Acquisition transforms. Examples: MVP, ['MVP','LRL','SWG']\n" ) loader_description = ( "Used to import and export SHEF data through cwms-data-api.\n" diff --git a/tests/test_cda_loader_make_export_transforms.py b/tests/test_cda_loader_make_export_transforms.py index 16b9965..35050a3 100644 --- a/tests/test_cda_loader_make_export_transforms.py +++ b/tests/test_cda_loader_make_export_transforms.py @@ -18,6 +18,7 @@ def _fake_groups_response(assigned): def _make_loader(): loader = cda_loader.CdaLoader(logger=None) loader._office_id = "OFF" + loader._office_ids = ["OFF"] return loader @@ -74,6 +75,293 @@ def test_make_export_transforms_skips_empty_alias_and_keeps_others(monkeypatch): ] +def test_make_transforms_passes_office_filter_to_cwms(monkeypatch): + """make_transforms() should pass the office filter through to cwms.get_timeseries_group.""" + calls = [] + + def fake_get_group(**kwargs): + calls.append(kwargs) + return types.SimpleNamespace( + json={ + "assigned-time-series": [ + { + "timeseries-id": "MVP.Test.Flow.Inst.1Hour.0.Raw", + "office-id": "MVP", + "alias-id": "TEST.HG.RZ.1", + } + ] + } + ) + + monkeypatch.setattr(cda_loader.cwms, "get_timeseries_group", fake_get_group) + + loader = cda_loader.CdaLoader(logger=None) + loader.make_transforms(office="MVP") + + assert len(calls) == 1 + assert calls[0]["office_id"] == "MVP" + assert any( + t.timeseries_id == "MVP.Test.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + + +def test_transform_key_checks_all_requested_offices(): + """Multi-office lookups should not be biased to the first office in the list.""" + loader = cda_loader.CdaLoader(logger=None) + loader._office_ids = ["LRL", "LRN"] + loader._office_id = "LRL" + loader._transforms = { + "LRN.ALCT1.HGIRZZ": cda_loader.ShefTransform( + office="LRN", + location="ALCT1", + parameter_code="HGIRZZ", + timeseries_id="LRN.ALCT1.Flow.Inst.1Hour.0.Raw", + units="ft", + timezone=None, + dl_time=None, + ) + } + + loader._shef_value = types.SimpleNamespace( + location="ALCT1", + parameter_code="HGIRZZQ", + ) + value = loader._shef_value + assert loader.transform_key == "LRN.ALCT1.HGIRZZ" + assert loader.get_time_series_name(value) == "LRN.ALCT1.Flow.Inst.1Hour.0.Raw" + + +def test_make_transforms_accepts_multiple_offices(monkeypatch): + """make_transforms() should use a single unscoped API call and filter the response to the requested offices.""" + calls = [] + + def fake_get_group(**kwargs): + calls.append(kwargs) + return types.SimpleNamespace( + json={ + "assigned-time-series": [ + { + "timeseries-id": "MVP.Multi.Flow.Inst.1Hour.0.Raw", + "office-id": "MVP", + "alias-id": "MULTIMVP.HG.RZ.1", + }, + { + "timeseries-id": "LRL.Multi.Flow.Inst.1Hour.0.Raw", + "office-id": "LRL", + "alias-id": "MULTILRL.HG.RZ.1", + }, + { + "timeseries-id": "SWG.Multi.Flow.Inst.1Hour.0.Raw", + "office-id": "SWG", + "alias-id": "MULTISWG.HG.RZ.1", + }, + ] + } + ) + + monkeypatch.setattr(cda_loader.cwms, "get_timeseries_group", fake_get_group) + + loader = cda_loader.CdaLoader(logger=None) + loader.make_transforms(office=["MVP", "LRL"]) + + assert len(calls) == 1 + assert "office_id" not in calls[0] + assert any( + t.timeseries_id == "MVP.Multi.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + assert any( + t.timeseries_id == "LRL.Multi.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + assert all( + t.timeseries_id != "SWG.Multi.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + + +def test_set_options_parses_multi_office_cli_string(): + """The command-line office option should parse into a list of office IDs.""" + loader = cda_loader.CdaLoader(logger=None) + loader.set_options("[https://example.test/cwms-data/][abc123][\"MVP\",\"LRL\",\"SWG\"]") + + assert loader._cda_url == "https://example.test/cwms-data/" + assert loader._office_ids == ["MVP", "LRL", "SWG"] + assert loader._office_id == "MVP" + + +def test_set_options_parses_bracketed_multi_office_cli_string(): + """Bracketed comma-delimited office options should flatten into a clean office list.""" + loader = cda_loader.CdaLoader(logger=None) + loader.set_options("[https://example.test/cwms-data/][abc123][[LRN,LRL]]") + + assert loader._cda_url == "https://example.test/cwms-data/" + assert loader._office_ids == ["LRN", "LRL"] + assert loader._office_id == "LRN" + + loader = cda_loader.CdaLoader(logger=None) + loader.set_options("[https://example.test/cwms-data/][abc123][LRL,LRN]") + assert loader._office_ids == ["LRL", "LRN"] + assert loader._office_id == "LRL" + + +def test_make_transforms_filters_duplicate_alias_by_office(monkeypatch): + """A duplicate alias in another office should not be processed when office scoping is active.""" + calls = [] + + def fake_get_group(**kwargs): + calls.append(kwargs) + office = kwargs.get("office_id") + assigned = [ + { + "timeseries-id": f"{office}.Test.Flow.Inst.1Hour.0.Raw", + "office-id": office, + "alias-id": "TEST.HG.RZ.1", + } + ] + if office == "LRL": + assigned.append( + { + "timeseries-id": "LRL.Other.Flow.Inst.1Hour.0.Raw", + "office-id": "LRL", + "alias-id": "TEST.HG.RZ.1", + } + ) + return types.SimpleNamespace(json={"assigned-time-series": assigned}) + + monkeypatch.setattr(cda_loader.cwms, "get_timeseries_group", fake_get_group) + + loader = cda_loader.CdaLoader(logger=None) + loader.make_transforms(office="MVP") + + assert len(calls) == 1 + assert calls[0]["office_id"] == "MVP" + assert any( + t.timeseries_id == "MVP.Test.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + assert all( + t.timeseries_id != "LRL.Other.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + + +def test_make_transforms_processes_each_office_in_list(monkeypatch): + """When multiple offices are supplied, they should be filtered from one unscoped response.""" + calls = [] + + def fake_get_group(**kwargs): + calls.append(kwargs) + return types.SimpleNamespace( + json={ + "assigned-time-series": [ + { + "timeseries-id": "MVP.List.Flow.Inst.1Hour.0.Raw", + "office-id": "MVP", + "alias-id": "LISTMVP.HG.RZ.1", + }, + { + "timeseries-id": "LRL.List.Flow.Inst.1Hour.0.Raw", + "office-id": "LRL", + "alias-id": "LISTLRL.HG.RZ.1", + }, + ] + } + ) + + monkeypatch.setattr(cda_loader.cwms, "get_timeseries_group", fake_get_group) + + loader = cda_loader.CdaLoader(logger=None) + loader.make_transforms(office=["MVP", "LRL"]) + + assert len(calls) == 1 + assert "office_id" not in calls[0] + assert any( + t.timeseries_id == "MVP.List.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + assert any( + t.timeseries_id == "LRL.List.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + + +def test_make_transforms_without_office_uses_default_unscoped_lookup(monkeypatch): + """When no office is supplied, make_transforms should fall back to the default unscoped query.""" + calls = [] + + def fake_get_group(**kwargs): + calls.append(kwargs) + return types.SimpleNamespace( + json={ + "assigned-time-series": [ + { + "timeseries-id": "DEFAULT.Test.Flow.Inst.1Hour.0.Raw", + "office-id": "MVP", + "alias-id": "TEST.HG.RZ.1", + } + ] + } + ) + + monkeypatch.setattr(cda_loader.cwms, "get_timeseries_group", fake_get_group) + + loader = cda_loader.CdaLoader(logger=None) + loader._office_ids = [] + loader.make_transforms() + + assert len(calls) == 1 + assert calls[0]["office_id"] == "" + assert any( + t.timeseries_id == "DEFAULT.Test.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + + +def test_make_transforms_keeps_unique_entries_when_same_alias_appears_in_multiple_offices( + monkeypatch, +): + """The transform map should keep separate office-specific entries from a single unscoped response.""" + calls = [] + + def fake_get_group(**kwargs): + calls.append(kwargs) + return types.SimpleNamespace( + json={ + "assigned-time-series": [ + { + "timeseries-id": "MVP.SameAlias.Flow.Inst.1Hour.0.Raw", + "office-id": "MVP", + "alias-id": "SAME.HG.RZ.1", + }, + { + "timeseries-id": "LRL.SameAlias.Flow.Inst.1Hour.0.Raw", + "office-id": "LRL", + "alias-id": "SAME.HG.RZ.1", + }, + ] + } + ) + + monkeypatch.setattr(cda_loader.cwms, "get_timeseries_group", fake_get_group) + + loader = cda_loader.CdaLoader(logger=None) + loader.make_transforms(office=["MVP", "LRL"]) + + assert len(calls) == 1 + assert "office_id" not in calls[0] + assert set(loader._transforms) == {"MVP.SAME.HGURZ", "LRL.SAME.HGURZ"} + assert any( + t.timeseries_id == "MVP.SameAlias.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + assert any( + t.timeseries_id == "LRL.SameAlias.Flow.Inst.1Hour.0.Raw" + for t in loader._transforms.values() + ) + + def test_make_export_transforms_scopes_fetch_to_requested_group(monkeypatch): """When a group_id is given, only that group should be fetched from cwms (no warnings about other groups).""" calls = [] @@ -107,6 +395,19 @@ def fake_get_groups(**kwargs): assert len(calls) == 1 +def test_get_office_summary_reports_time_series_and_values_by_office(): + """The loader summary should break totals down by office for the loaded payloads.""" + loader = cda_loader.CdaLoader(logger=None) + loader._office_load_stats = { + "MVP": {"time_series": 2, "value_count": 5}, + "LRL": {"time_series": 1, "value_count": 3}, + } + + assert loader.get_office_summary() == ( + "LRL: 1 time series, 3 values; MVP: 2 time series, 5 values" + ) + + def test_make_export_transforms_skips_malformed_alias_and_keeps_others(monkeypatch): """A malformed alias-id (raises inside make_shef_transform) must not abort the rest.""" assigned = [ From 24d6b7e7f9461e08c196458da4d2a52903a524a0 Mon Sep 17 00:00:00 2001 From: Eric Novotny Date: Fri, 4 Sep 2026 15:11:25 -0700 Subject: [PATCH 2/2] update version. fix issues --- README.md | 6 +- pyproject.toml | 2 +- shef/loaders/cda_loader.py | 187 ++++++++++-------- .../test_cda_loader_make_export_transforms.py | 103 +++++++++- 4 files changed, 215 insertions(+), 83 deletions(-) diff --git a/README.md b/README.md index 8253fe1..a7831fe 100644 --- a/README.md +++ b/README.md @@ -18,11 +18,11 @@ pip install shef-parser ## Loading via command line ```sh -#CWMS CDA loader +#CWMS CDA loader will load for all offices shefParser -i input_filename --loader cda[$API_ROOT][$API_KEY] # optional office scoping for transform lookup -shefParser -i input_filename --loader 'cda[$API_ROOT][$API_KEY][MVP]' -shefParser -i input_filename --loader 'cda[$API_ROOT][$API_KEY]["MVP","LRL","SWG"]' +shefParser -i input_filename --loader cda[$API_ROOT][$API_KEY][MVP] +shefParser -i input_filename --loader cda[$API_ROOT][$API_KEY][MVP,LRL,SWG] ``` ```sh diff --git a/pyproject.toml b/pyproject.toml index f1ae4ff..43219b4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,7 +2,7 @@ name = "shef-parser" repository = "https://github.com/HydrologicEngineeringCenter/SHEF_processing" -version = "1.10.0" +version = "1.11.0" packages = [ diff --git a/shef/loaders/cda_loader.py b/shef/loaders/cda_loader.py index be1ee92..6f53b65 100644 --- a/shef/loaders/cda_loader.py +++ b/shef/loaders/cda_loader.py @@ -227,14 +227,22 @@ def make_transform_key( return f"{office}.{location}.{parameter_code}" return f"{location}.{parameter_code}" - @property - def transform_key(self) -> str: - """ - The transform key for the current SHEF value. - """ - self.assert_value_is_set() - sv = cast(shared.ShefValue, self._shef_value) - parameter_code = sv.parameter_code[:-1] + def get_matching_transforms( + self, shef_value: Optional[shared.ShefValue] + ) -> list[ShefTransform]: + """Return all matching office-scoped transforms for a SHEF value.""" + if shef_value is None: + return [] + + parameter_code = shef_value.parameter_code[:-1] + if not self._office_ids and not self._office_id: + return [ + transform + for transform in self._transforms.values() + if transform.location == shef_value.location + and transform.parameter_code == parameter_code + ] + candidate_keys = [] office_candidates: list[str] = [] if self._office_ids: @@ -243,20 +251,45 @@ def transform_key(self) -> str: office_candidates.append(self._office_id) for office in office_candidates: candidate_keys.append( - self.make_transform_key(office, sv.location, parameter_code) + self.make_transform_key(office, shef_value.location, parameter_code) ) - candidate_keys.append(self.make_transform_key(None, sv.location, parameter_code)) + candidate_keys.append( + self.make_transform_key(None, shef_value.location, parameter_code) + ) + + matches: list[ShefTransform] = [] + seen: set[str] = set() for key in candidate_keys: - if key in self._transforms: - return key - raise KeyError(f"No transform found for {sv.location}.{parameter_code}") + if key in self._transforms and key not in seen: + matches.append(self._transforms[key]) + seen.add(key) + return matches + + @property + def transform_key(self) -> str: + """ + The transform key for the current SHEF value. + """ + self.assert_value_is_set() + sv = cast(shared.ShefValue, self._shef_value) + matches = self.get_matching_transforms(sv) + if matches: + return self.make_transform_key( + matches[0].office, + sv.location, + sv.parameter_code[:-1], + ) + raise KeyError(f"No transform found for {sv.location}.{sv.parameter_code[:-1]}") @property def transform(self) -> ShefTransform: """ The ShefTransform object for the current SHEF value """ - return self._transforms[self.transform_key] + matches = self.get_matching_transforms(self._shef_value) + if not matches: + raise KeyError(f"No transform found for {self._shef_value.location}.{self._shef_value.parameter_code[:-1]}") + return matches[0] def make_transforms( self, office: Optional[Union[str, list[str]]] = None @@ -270,11 +303,11 @@ def make_transforms( """ requested_offices: list[str] = [] if office is None: - requested_offices = self._office_ids if self._office_ids else [""] + requested_offices = self._office_ids if self._office_ids else [] elif isinstance(office, str): - requested_offices = [office] + requested_offices = [office] if office else [] else: - requested_offices = [str(o) for o in office] + requested_offices = [str(o) for o in office if str(o).strip()] group_kwargs: dict[str, Any] = { "group_office_id": "CWMS", @@ -292,10 +325,6 @@ def make_transforms( assigned_ts = [ item for item in assigned_ts if item.get("office-id") in office_filter ] - if self._logger: - self._logger.debug(f"assigned_ts: {assigned_ts}") - self._logger.debug(f"office_filter: {office_filter}") - self._logger.debug(f"requested_offices: {requested_offices}") for assigned_item in assigned_ts: if "timeseries-id" in assigned_item and "alias-id" in assigned_item: try: @@ -321,7 +350,10 @@ def make_transforms( ) if self._logger: self._logger.debug(f"transforms: {list(self._transforms.keys())}") - + self._logger.debug(f"assigned_ts: {assigned_ts}") + self._logger.debug(f"office_filter: {office_filter}") + self._logger.debug(f"requested_offices: {requested_offices}") + self._logger.debug(f"kw: {group_kwargs}") def get_additional_pe_codes(self, parser_recognized_pe_codes: set[str]) -> set[str]: """ Return any PE codes recognized by this loader that aren't otherwised recognized by the parser @@ -332,28 +364,19 @@ def get_additional_pe_codes(self, parser_recognized_pe_codes: set[str]) -> set[s def get_time_series_name(self, shef_value: Optional[shared.ShefValue]) -> str: """ - Get the time series ID for the current SHEF value + Get the time series ID for the current SHEF value. + + When multiple office-specific transforms match the same SHEF value, + the first office in the requested list wins for this single-value API. """ if shef_value is None: raise shared.LoaderException("Empty SHEF value in get_time_series_name()") if not self._transforms: self.make_transforms() - parameter_code = shef_value.parameter_code[:-1] - candidate_keys = [] - office_candidates: list[str] = [] - if self._office_ids: - office_candidates.extend(self._office_ids) - if self._office_id and self._office_id not in office_candidates: - office_candidates.append(self._office_id) - for office in office_candidates: - candidate_keys.append( - self.make_transform_key(office, shef_value.location, parameter_code) - ) - candidate_keys.append(self.make_transform_key(None, shef_value.location, parameter_code)) - for key in candidate_keys: - if key in self._transforms: - return self._transforms[key].timeseries_id - raise KeyError(f"No transform found for {shef_value.location}.{parameter_code}") + matches = self.get_matching_transforms(shef_value) + if matches: + return matches[0].timeseries_id + raise KeyError(f"No transform found for {shef_value.location}.{shef_value.parameter_code[:-1]}") @staticmethod def get_unix_timestamp(timestamp: str) -> int: @@ -369,6 +392,12 @@ def get_unix_timestamp(timestamp: str) -> int: def get_python_datetime(unix_time: int) -> datetime: return datetime.fromtimestamp(unix_time / 1000, tz=timezone.utc) + @staticmethod + def normalize_office_id(office_id: Optional[str]) -> str: + """Normalize blank office IDs to a display label for logs and summaries.""" + office = (office_id or "").strip() + return office if office else "DEFAULT" + def load_time_series(self) -> None: """ Store SHEF values as CDA POST payloads grouped by time series ID @@ -376,9 +405,9 @@ def load_time_series(self) -> None: if self._shef_value and self._time_series: sv = self._shef_value - transform = self.transform + transforms = self.get_matching_transforms(sv) if self._logger: - self._logger.debug(f"ts_name: {self.get_time_series_name(sv)}") + self._logger.debug(f"ts_names: {[t.timeseries_id for t in transforms]}") self._logger.debug(f"shef_value: {sv}") self._logger.debug(f"time_series: {self._time_series}") if self._time_series: @@ -386,29 +415,31 @@ def load_time_series(self) -> None: for ts in self._time_series: time = self.get_unix_timestamp(ts[0]) time_series.append(CdaValue(time, float(ts[1]), 0)) - post_data: TimeseriesPayload = { - "name": self.get_time_series_name(sv), - "office-id": transform.office, - "units": transform.units, - "values": time_series, - } - match_index = self.find_matching_payload_index(post_data) - if match_index is None: - self._payloads.append(post_data) - office_stats = self._office_load_stats.setdefault( - self.transform.office, - {"time_series": 0, "value_count": 0}, - ) - office_stats["time_series"] += 1 - office_stats["value_count"] += len(time_series) - else: - match_payload = self._payloads[match_index] - match_payload["values"].extend(time_series) - office_stats = self._office_load_stats.setdefault( - self.transform.office, - {"time_series": 0, "value_count": 0}, - ) - office_stats["value_count"] += len(time_series) + for transform in transforms: + office_id = self.normalize_office_id(transform.office or self._office_id) + post_data: TimeseriesPayload = { + "name": transform.timeseries_id, + "office-id": office_id, + "units": transform.units, + "values": time_series, + } + match_index = self.find_matching_payload_index(post_data) + if match_index is None: + self._payloads.append(post_data) + office_stats = self._office_load_stats.setdefault( + office_id, + {"time_series": 0, "value_count": 0}, + ) + office_stats["time_series"] += 1 + office_stats["value_count"] += len(time_series) + else: + match_payload = self._payloads[match_index] + match_payload["values"].extend(time_series) + office_stats = self._office_load_stats.setdefault( + office_id, + {"time_series": 0, "value_count": 0}, + ) + office_stats["value_count"] += len(time_series) self._time_series = [] def create_write_task( @@ -453,16 +484,16 @@ async def process_write_tasks(self) -> None: self._value_count += value_count self._time_series_count += 1 if self._logger: - self._logger.info(f"Stored {value_count} values in {tsid}") + office_id = payload.get("office-id", "DEFAULT") + self._logger.info( + f"Stored {value_count} values in {tsid} [{office_id}]" + ) process_time = time.time() - start_time if self._logger: self._logger.info( f"CWMS-Data-API POST tasks complete ({process_time:.2f} seconds)" ) - self._logger.info( - "Loaded by office: %s", - self.get_office_summary(), - ) + self._logger.info(f"Loaded by office: {self.get_office_summary()}") def find_matching_payload_index( self, payload: TimeseriesPayload @@ -575,10 +606,7 @@ def done(self) -> None: self._logger.info( "--[Summary]-----------------------------------------------------------" ) - self._logger.info( - "Loaded by office: %s", - self.get_office_summary(), - ) + self._logger.info(f"Loaded by office: {self.get_office_summary()}") if self._value_error_count > 0: self._logger.info( f"Errors occurred for {self._value_error_count} values in {self._time_series_error_count} time series" @@ -606,16 +634,21 @@ def get_office_summary(self) -> str: summary_parts = [] for office, stats in sorted(self._office_load_stats.items()): + office_label = self.normalize_office_id(office) time_series = stats.get("time_series", 0) value_count = stats.get("value_count", 0) - summary_parts.append(f"{office}: {time_series} time series, {value_count} values") + summary_parts.append( + f"{office_label}: {time_series} time series, {value_count} values" + ) return "; ".join(summary_parts) def _track_office_load(self, office_id: str, value_count: int = 0) -> None: """Track the number of time series and values loaded for a given office.""" - if not office_id: - office_id = "DEFAULT" - office_stats = self._office_load_stats.setdefault(office_id, {"time_series": 0, "value_count": 0}) + office_key = self.normalize_office_id(office_id) + office_stats = self._office_load_stats.setdefault( + office_key, + {"time_series": 0, "value_count": 0}, + ) office_stats["value_count"] += value_count def make_export_transforms(self, group_id: Optional[str] = None) -> None: diff --git a/tests/test_cda_loader_make_export_transforms.py b/tests/test_cda_loader_make_export_transforms.py index 35050a3..1254354 100644 --- a/tests/test_cda_loader_make_export_transforms.py +++ b/tests/test_cda_loader_make_export_transforms.py @@ -288,7 +288,7 @@ def fake_get_group(**kwargs): def test_make_transforms_without_office_uses_default_unscoped_lookup(monkeypatch): - """When no office is supplied, make_transforms should fall back to the default unscoped query.""" + """When no office is supplied, make_transforms should omit the office filter and use the default unscoped query.""" calls = [] def fake_get_group(**kwargs): @@ -312,13 +312,49 @@ def fake_get_group(**kwargs): loader.make_transforms() assert len(calls) == 1 - assert calls[0]["office_id"] == "" + assert "office_id" not in calls[0] assert any( t.timeseries_id == "DEFAULT.Test.Flow.Inst.1Hour.0.Raw" for t in loader._transforms.values() ) +def test_get_matching_transforms_without_office_returns_all_office_matches(): + """Unscoped lookups should include all office-specific transforms for the same location and parameter.""" + loader = cda_loader.CdaLoader(logger=None) + loader._office_ids = [] + loader._office_id = "" + loader._transforms = { + "LRL.ALCT1.HGIRZZ": cda_loader.ShefTransform( + office="LRL", + location="ALCT1", + parameter_code="HGIRZZ", + timeseries_id="LRL.ALCT1.Flow.Inst.1Hour.0.Raw", + units="ft", + timezone=None, + dl_time=None, + ), + "LRN.ALCT1.HGIRZZ": cda_loader.ShefTransform( + office="LRN", + location="ALCT1", + parameter_code="HGIRZZ", + timeseries_id="LRN.ALCT1.Flow.Inst.1Hour.0.Raw", + units="ft", + timezone=None, + dl_time=None, + ), + } + + loader._shef_value = types.SimpleNamespace(location="ALCT1", parameter_code="HGIRZZQ") + matches = loader.get_matching_transforms(loader._shef_value) + + assert {m.office for m in matches} == {"LRL", "LRN"} + assert {m.timeseries_id for m in matches} == { + "LRL.ALCT1.Flow.Inst.1Hour.0.Raw", + "LRN.ALCT1.Flow.Inst.1Hour.0.Raw", + } + + def test_make_transforms_keeps_unique_entries_when_same_alias_appears_in_multiple_offices( monkeypatch, ): @@ -362,6 +398,42 @@ def fake_get_group(**kwargs): ) +def test_load_time_series_processes_all_matching_office_transforms(monkeypatch): + """A single location+parameter should process every office-specific transform that matches.""" + loader = cda_loader.CdaLoader(logger=None) + loader._office_ids = ["LRL", "LRN"] + loader._office_id = "LRL" + loader._transforms = { + "LRL.ALCT1.HGIRZZ": cda_loader.ShefTransform( + office="LRL", + location="ALCT1", + parameter_code="HGIRZZ", + timeseries_id="LRL.ALCT1.Flow.Inst.1Hour.0.Raw", + units="ft", + timezone=None, + dl_time=None, + ), + "LRN.ALCT1.HGIRZZ": cda_loader.ShefTransform( + office="LRN", + location="ALCT1", + parameter_code="HGIRZZ", + timeseries_id="LRN.ALCT1.Flow.Inst.1Hour.0.Raw", + units="ft", + timezone=None, + dl_time=None, + ), + } + loader._shef_value = types.SimpleNamespace(location="ALCT1", parameter_code="HGIRZZQ") + loader._time_series = [["2024-01-01 00:00:00", "15.0"]] + + loader.load_time_series() + + assert {payload["name"] for payload in loader._payloads} == { + "LRL.ALCT1.Flow.Inst.1Hour.0.Raw", + "LRN.ALCT1.Flow.Inst.1Hour.0.Raw", + } + + def test_make_export_transforms_scopes_fetch_to_requested_group(monkeypatch): """When a group_id is given, only that group should be fetched from cwms (no warnings about other groups).""" calls = [] @@ -408,6 +480,33 @@ def test_get_office_summary_reports_time_series_and_values_by_office(): ) +def test_load_time_series_tracks_blank_office_as_default(): + """A missing office should still report a valid office bucket instead of an empty label.""" + loader = cda_loader.CdaLoader(logger=None) + loader._shef_value = types.SimpleNamespace( + location="ALCT1", + parameter_code="HGIRZZQ", + ) + loader._time_series = [["2024-01-01 00:00:00", "15.0"]] + loader._transforms = { + "ALCT1.HGIRZZ": cda_loader.ShefTransform( + office="", + location="ALCT1", + parameter_code="HGIRZZ", + timeseries_id="ALCT1.Flow.Inst.1Hour.0.Raw", + units="ft", + timezone=None, + dl_time=None, + ) + } + loader._office_id = "" + + loader.load_time_series() + + assert loader.get_office_summary() == "DEFAULT: 1 time series, 1 values" + assert loader._payloads[0]["office-id"] == "DEFAULT" + + def test_make_export_transforms_skips_malformed_alias_and_keeps_others(monkeypatch): """A malformed alias-id (raises inside make_shef_transform) must not abort the rest.""" assigned = [