fixes to shef exporter

.github/workflows/tests.yml

@@ -0,0 +1,40 @@
+name: Tests
+
+on:
+  pull_request:
+    branches:
+      - master
+  workflow_dispatch:
+
+jobs:
+  test:
+    name: Run Tests
+    runs-on: ubuntu-latest
+
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.9', '3.10', '3.11', '3.12']
+
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Set Up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install Poetry
+        uses: abatilo/actions-poetry@v4
+
+      - name: Cache Virtual Environment
+        uses: actions/cache@v4
+        with:
+          path: ./.venv
+          key: venv-${{ matrix.python-version }}-${{ hashFiles('poetry.lock') }}
+
+      - name: Install Dependencies
+        run: poetry install --all-extras
+
+      - name: Run Tests
+        run: poetry run pytest

pyproject.toml

@@ -17,12 +17,16 @@ authors = ["Hydrologic Engineering Center"]
 [tool.poetry.dependencies]
 python = "^3.9"
 click = "^8.1"
+cwms-python = { version = "^1.0.8", optional = true }
+hecdss = { version = "^0.1.29", optional = true }
 sphinx = { version = "^7.0", optional = true }
 sphinx_rtd_theme = { version = "^2.0", optional = true }
 sphinx-design = { version = "^0.5", optional = true }
 
 [tool.poetry.extras]
 docs = ["sphinx", "sphinx_rtd_theme", "sphinx-design"]
+cda = ["cwms-python"]
+dss = ["hecdss"]
 
 [tool.poetry.group.dev.dependencies]
 black = "^24.2.0"

shef/exporters/abstract_exporter.py

@@ -5,8 +5,6 @@
 from io import BufferedRandom, StringIO
 from typing import Optional, TextIO, Union
 
-import cwms  # type: ignore
-
 from shef.loaders import abstract_loader, shared
 
 

shef/exporters/cda_exporter.py

@@ -35,7 +35,6 @@ def __init__(self, cda_url: str, office: str):
         self._office = office
         self._cda_loader = loaders.cda_loader.CdaLoader(self.logger, sys.stdout)
         self._cda_loader.set_options(f"[{cda_url}][][{office}]")
-        self._cda_loader.make_export_transforms()
 
     def export(self, timeseries_or_group: str) -> None:
         """
@@ -45,35 +44,50 @@ def export(self, timeseries_or_group: str) -> None:
             timeseries_or_group (str): If a time series ID, export that time series; If a time series group ID, export each time series in that group
         """
         if len(timeseries_or_group.split(".")) == 6:
+            self._cda_loader.make_export_transforms()
             timeseries_ids = [timeseries_or_group]
-        elif timeseries_or_group in self._cda_loader._export_groups:
+        else:
+            self._cda_loader.make_export_transforms(group_id=timeseries_or_group)
+            if timeseries_or_group not in self._cda_loader._export_groups:
+                raise shared.LoaderException(
+                    f"Time series group [{timeseries_or_group}] not found under the SHEF Export category for office [{self._office}]"
+                )
             timeseries_ids = self._cda_loader._export_groups[timeseries_or_group][
                 "timeseries"
             ]
         total_value_count: int = 0
-        first = True
-        data = StringIO()
-        data.write("[")
+        ts_payloads: list[Any] = []
         for tsid in timeseries_ids:
-            unit = self._cda_loader._transforms[tsid].units
-            ts = cwms.get_timeseries(
-                ts_id=tsid,
-                office_id=self._cda_loader._office_id,
-                unit=unit,
-                begin=self._start_time,
-                end=self._end_time,
-            )
-            value_count = len(ts.json["values"])
-            if value_count > 0:
-                if not first:
-                    data.write(",")
-                data.write(json.dumps(ts.json))
-                total_value_count += value_count
-                first = False
-        data.write("]")
-        to_unload = data.getvalue()
-        data.close()
-        if (total_value_count) > 0:
+            try:
+                unit = self._cda_loader._transforms[tsid].units
+                ts = cwms.get_timeseries(
+                    ts_id=tsid,
+                    office_id=self._cda_loader._office_id,
+                    unit=unit,
+                    begin=self._start_time,
+                    end=self._end_time,
+                )
+            except Exception as e:
+                self.logger.warning(
+                    f"Skipping time series [{tsid}]: error fetching from CDA: {e}"
+                )
+                continue
+            ts_json = ts.json if ts is not None else None
+            if not isinstance(ts_json, dict):
+                self.logger.warning(
+                    f"Skipping time series [{tsid}]: CDA response is not a JSON object"
+                )
+                continue
+            values = ts_json.get("values") or []
+            if not values:
+                self.logger.info(
+                    f"Skipping time series [{tsid}]: no values in window {self._start_time} to {self._end_time}"
+                )
+                continue
+            ts_payloads.append(ts_json)
+            total_value_count += len(values)
+        to_unload = json.dumps(ts_payloads)
+        if total_value_count > 0:
             try:
                 old_output = self._cda_loader._output
                 self._cda_loader._output = self._output
@@ -90,6 +104,7 @@ def get_groups(self) -> dict[str, str]:
         Returns:
             dict[str, str]: A dictionary of time series group descriptions keyed by time series group IDs
         """
+        self._cda_loader.make_export_transforms()
         return {
             group: self._cda_loader._export_groups[group]["description"]
             for group in self._cda_loader._export_groups
@@ -105,6 +120,7 @@ def get_time_series(self, group: str) -> list[str]:
         Returns:
             list[str]: The assigned time series IDs
         """
+        self._cda_loader.make_export_transforms(group_id=group)
         return [ts for ts in self._cda_loader._export_groups[group]["timeseries"]]
 
     def get_unit(self, tsid: str) -> Optional[str]:
@@ -117,6 +133,8 @@ def get_unit(self, tsid: str) -> Optional[str]:
         Returns:
             Optional[str]: The unit as specified in the time series alias
         """
+        if tsid not in self._cda_loader._transforms:
+            self._cda_loader.make_export_transforms()
         return self._cda_loader._transforms[tsid].units
 
 

shef/loaders/cda_loader.py

@@ -106,6 +106,8 @@ def __init__(
         self._value_error_count: int = 0
         self._write_tasks: list[Coroutine[Any, Any, Any]] = []
         self._configured_pe_codes = set()
+        self._loaded_export_group_ids: set[str] = set()
+        self._loaded_all_export_groups: bool = False
 
     def make_shef_transform(self, crit: dict[str, Any]) -> ShefTransform:
         """
@@ -462,55 +464,72 @@ def set_input(self, input_object: Union[StringIO, TextIO, str]) -> None:
                 f"Expected TextIOWrapper or str object, got [{input_object.__class__.__name__}]"
             )
 
-    def make_export_transforms(self) -> None:
+    def make_export_transforms(self, group_id: Optional[str] = None) -> None:
         if not self._office_id:
             raise shared.LoaderException(
                 f"Cannot unload without office specified, use options [api_root][api_key][office]"
             )
-        if not self._transforms:
-            tsids_used: dict[str, list[str]] = {}
-            group_list = cwms.get_timeseries_groups(
-                office_id=self._office_id,
-                include_assigned=True,
-                timeseries_category_like="SHEF Export",
-                timeseries_group_like="^.+$",
-                category_office_id="CWMS",
-            ).json
-            for shef_group in group_list:
-                group_id = shef_group["id"]
-                if "description" not in shef_group:
-                    shef_group["description"] = ""
-                self._export_groups[group_id] = {
-                    "description": shef_group["description"],
-                    "timeseries": [],
-                }
-                try:
-                    for time_series in shef_group["assigned-time-series"]:
-                        transform = self.make_shef_transform(time_series)
-                        transform_key = (
-                            f"{transform.location}.{transform.parameter_code}"
-                        )
-                        self._transforms[transform_key] = transform
-                        if transform.timeseries_id in tsids_used:
-                            if self._logger:
-                                self._logger.warning(
-                                    f"Tranform for time seires {transform.timeseries_id} specified in group(s) "
-                                    f"{','.join(tsids_used[transform.timeseries_id])} is/are overriden by transform specified in group {group_id}"
-                                )
-                        self._export_groups[group_id]["timeseries"].append(
-                            transform.timeseries_id
-                        )
-                        tsids_used.setdefault(transform.timeseries_id, []).append(
-                            group_id
-                        )
-                        self._transforms[transform.timeseries_id] = (
-                            transform  # to be able to retrieve by time series ID
+        if self._loaded_all_export_groups:
+            return
+        if group_id is None:
+            group_filter = "^.+$"
+        elif group_id in self._loaded_export_group_ids:
+            return
+        else:
+            group_filter = f"^{re.escape(group_id)}$"
+        tsids_used: dict[str, list[str]] = {}
+        group_list = cwms.get_timeseries_groups(
+            office_id=self._office_id,
+            include_assigned=True,
+            timeseries_category_like="SHEF Export",
+            timeseries_group_like=group_filter,
+            category_office_id="CWMS",
+            group_office_id=self._office_id,
+        ).json
+        for shef_group in group_list:
+            shef_group_id = shef_group["id"]
+            if "description" not in shef_group:
+                shef_group["description"] = ""
+            self._export_groups[shef_group_id] = {
+                "description": shef_group["description"],
+                "timeseries": [],
+            }
+            for time_series in shef_group["assigned-time-series"]:
+                if not time_series.get("alias-id"):
+                    if self._logger:
+                        self._logger.warning(
+                            f"Skipping time series {time_series.get('timeseries-id')} in group {shef_group_id}: missing or empty alias-id"
                         )
+                    continue
+                try:
+                    transform = self.make_shef_transform(time_series)
+                    transform_key = (
+                        f"{transform.location}.{transform.parameter_code}"
+                    )
+                    self._transforms[transform_key] = transform
+                    if transform.timeseries_id in tsids_used:
+                        if self._logger:
+                            self._logger.warning(
+                                f"Tranform for time seires {transform.timeseries_id} specified in group(s) "
+                                f"{','.join(tsids_used[transform.timeseries_id])} is/are overriden by transform specified in group {shef_group_id}"
+                            )
+                    self._export_groups[shef_group_id]["timeseries"].append(
+                        transform.timeseries_id
+                    )
+                    tsids_used.setdefault(transform.timeseries_id, []).append(
+                        shef_group_id
+                    )
+                    self._transforms[transform.timeseries_id] = (
+                        transform  # to be able to retrieve by time series ID
+                    )
                 except Exception as e:
                     if self._logger:
                         self._logger.warning(
-                            f"{str(e)} occurred while processing SHEF criteria for {time_series['timeseries-id']}"
+                            f"{str(e)} occurred while processing SHEF criteria for {time_series.get('timeseries-id')}"
                         )
+            self._loaded_export_group_ids.add(shef_group_id)
+        if group_id is None:
+            self._loaded_all_export_groups = True
 
     def unload(self) -> None:
         """

shef/loaders/dss_loader.py

@@ -6,7 +6,14 @@
 from logging import Logger
 from typing import Any, Optional, TextIO, Union, cast
 
-from hecdss import HecDss, IrregularTimeSeries, RegularTimeSeries  # type: ignore
+try:
+    from hecdss import HecDss, IrregularTimeSeries, RegularTimeSeries  # type: ignore
+    HECDSS_AVAILABLE = True
+except ImportError:
+    HECDSS_AVAILABLE = False
+    HecDss = None
+    IrregularTimeSeries = None
+    RegularTimeSeries = None
 
 from shef.constants import PE_CONVERSIONS
 from shef.loaders import abstract_loader, shared
@@ -68,6 +75,8 @@ def set_options(self, options_str: Optional[str]) -> None:
         """
         Set the sensor and parameter file names
         """
+        if not HECDSS_AVAILABLE:
+            raise ImportError("The 'hecdss' library is required but not installed. Please install it to use DSS functionality.")
         if not options_str:
             raise shared.LoaderException(
                 f"Empty options on {self.loader_name}.set_options()"
@@ -617,6 +626,8 @@ def load_time_series(self) -> None:
         """
         Store the time series to HEC-DSS file
         """
+        if not HECDSS_AVAILABLE:
+            raise ImportError("The 'hecdss' library is required but not installed. Please install it to use DSS functionality.")
         if self._shef_value and self._time_series:
             if self._dss_file is None:
                 self._dss_file = HecDss(self._dss_file_name)