Ts pattern

.github/workflows/publish-to-test-pypi.yml

@@ -98,6 +98,9 @@ jobs:
       - name: Install numpy
         run: pip install numpy
 
+      - name: Install tzdata (Python)
+        run: pip install tzdata
+
       - name: Run tests
         run: pytest tests/
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "hec-dss-python"
-version = "0.1.21"
+version = "0.1.22"
 description = "Python wrapper for the HEC-DSS file database C library."
 authors = ["Hydrologic Engineering Center"]
 license = "MIT"

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = hecdss
-version = 0.1.21
+version = 0.1.22
 author = Hydrologic Engineering Center
 author_email =hec.dss@usace.army.mil
 description = Python wrapper for the HEC-DSS file database C library.
@@ -20,6 +20,7 @@ python_requires = >=3.8
 include_package_data = True
 install_requires =
     numpy
+    tzdata
 
 [options.packages.find]
 where=src

src/hecdss/catalog.py

@@ -1,6 +1,7 @@
 from .record_type import RecordType
 from .dsspath import DssPath
 from datetime import datetime
+import re
 
 class Catalog:
     """manage list of objects inside a DSS database"""
@@ -47,7 +48,7 @@ def __create_condensed_catalog(self):
                 cleanPath = str(path.path_without_date())
                 raw_paths[cleanPath.lower()] = rawPath
                 self.recordTypeDict[cleanPath.lower()] = recordType
-                if(path.D != "TS-Pattern"):
+                if re.match(r"^\d{2}[A-Za-z]{3}\d{4}$", path.D):  # Check if path.D matches the format 'DDMMMYYYY'
                     tsRecords = self.timeSeriesDictNoDates.setdefault(cleanPath.lower(), [])
                     t = datetime.strptime(path.D,"%d%b%Y")
                     tsRecords.append(t)

src/hecdss/download_hecdss.py

@@ -36,7 +36,7 @@ def download_and_unzip(url, zip_file, destination_dir):
         print(f"Failed to download zip file. Status code: {response.status_code}")
 
 base_url = "https://www.hec.usace.army.mil/nexus/repository/maven-public/mil/army/usace/hec/hecdss/"
-version = "7-IU-10"
+version = "7-IU-16"
 
 destination_dir = Path(__file__).parent.joinpath("lib")
 zip_url = f"{base_url}{version}-win-x86_64/hecdss-{version}-win-x86_64.zip"

src/hecdss/hecdss.py

@@ -105,7 +105,11 @@ def get(self, pathname: str, startdatetime=None, enddatetime=None, trim=False):
             varies: RegularTimeSeries, PairedData, Grid, or Array.
         """
         if type == RecordType.RegularTimeSeries or type == RecordType.IrregularTimeSeries:
-            new_pathname = DssPath(pathname).path_without_date().__str__()
+            new_pathname = pathname
+            if(DssPath(pathname).D.lower() != "ts-pattern"):
+                new_pathname = DssPath(pathname).path_without_date().__str__()
+            elif type == RecordType.IrregularTimeSeries:
+                raise ValueError("ts-pattern is not fully supported for irregular time series")
             ts = self._get_timeseries(new_pathname, startdatetime, enddatetime, trim)
             return ts
         elif type == RecordType.PairedData:
@@ -315,7 +319,7 @@ def _get_julian_time_range(self, pathname, boolFullSet):
         firstSeconds = [0]
         lastValidJulian = [0]
         lastSeconds = [0]
-        self._native.hec_dss_tsGetDateTimeRange(
+        status = self._native.hec_dss_tsGetDateTimeRange(
             pathname,
             boolFullSet,
             firstValidJulian,
@@ -464,8 +468,17 @@ def _get_timeseries(self, pathname, startDateTime, endDateTime, trim):
         timeZoneName = timeZoneName[0]
         if(timeZoneName):
             new_times = [i.replace(tzinfo=ZoneInfo(timeZoneName)) for i in new_times]
+        elif (DssPath(pathname).D.lower() == "ts-pattern"):
+            new_times = []
+            start_date = _startDateTime - timedelta(seconds=interval_seconds)
+
         location_info = self._get_location_info(pathname)
         ts = ts.create(values=values, times=new_times, quality=quality, units=units, data_type=data_type, start_date=start_date, time_granularity_seconds=time_granularity_seconds, julian_base_date=julian_base_date, time_zone_name=timeZoneName, path=pathname, location_info=location_info)
+
+        if (DssPath(pathname).D.lower() == "ts-pattern"):
+            new_interval = ts._get_interval_path()
+            ts._interval_to_times(new_interval)
+
         return ts
 
     def _get_location_info(self, pathname: str):
@@ -561,6 +574,8 @@ def put(self, container) -> int:
             self._catalog = None
         elif type(container) is IrregularTimeSeries:
             its = container
+            if (DssPath(its.id).D.lower() == "ts-pattern"):
+                raise ValueError("ts-pattern is not fully supported for irregular time series")
             # def hec_dss_tsStoreRegular(dss, pathname, startDate, startTime, valueArray, qualityArray,
             #                           saveAsFloat, units, type):
             start_date_base = (datetime(1900, 1, 1)+timedelta(days=its.julian_base_date))

tests/test_basics.py

@@ -209,6 +209,29 @@ def test_delete_range(self):
             dss.delete(path, False, t1, t2)
             assert (not("//SACRAMENTO/PRECIP-INC/01Jan1882/1Day/OBS/" in dss.get_catalog().uncondensed_paths))
 
+    def test_Read_TS_Pattern_Regular(self):
+        with HecDss(self.test_files.get_copy("Depth_Area_01.dss")) as dss:
+            path = "//010010-B/FLOW-UNIT GRAPH/TS-Pattern/5Minute/DAA:Depth-Area 01>010005-C/"
+            tsc = dss.get(path)
+            assert(len(tsc.values) > 0)
+
+    def test_Write_TS_Pattern_Regular(self):
+        with HecDss(self.test_files.get_copy("Depth_Area_01.dss")) as dss:
+            path = "//010010-B/FLOW-UNIT GRAPH/TS-Pattern/5Minute/DAA:Depth-Area 01>010005-C/"
+            tsc = dss.get(path)
+            tsc.id = "//010010-B/FLOW-UNIT GRAPH/TS-Pattern/5Minute/DAA:Depth-Area 01>010005-C-modified2/"
+            tsc.times = [i.replace(year=2023) for i in tsc.times]
+            tsc.start_date = tsc.start_date.replace(year=2023)
+            dss.put(tsc)
+            tsc2 = dss.get(tsc.id)
+            assert(len(tsc2.values) > 0)
+
+    def test_path_empty_parts(self):
+        with HecDss(self.test_files.get_copy("Depth_Area_01.dss")) as dss:
+            path = "//010020-R/STORAGE-FLOW///DAA:Depth-Area 01>010025-R/"
+            tsc = dss.get(path)
+            assert (len(tsc.values) > 0)
+
 if __name__ == "__main__":
     unittest.main()
     # test_catalog()