add text type

Author: ktarbet

src/hecdss/download_hecdss.py

@@ -45,7 +45,7 @@ def download_and_unzip(url, zip_file, destination_dir):
         logger.error(error_msg)
 
 base_url = "https://www.hec.usace.army.mil/nexus/repository/maven-public/mil/army/usace/hec/hecdss/"
-version = "7-IU-16"
+version = "7-IW-3"
 
 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

@@ -8,6 +8,7 @@
 import hecdss.record_type
 from hecdss.array_container import ArrayContainer
 from hecdss.location_info import LocationInfo
+from hecdss.text import Text
 from hecdss.paired_data import PairedData
 from hecdss.native import _Native
 from hecdss.dateconverter import DateConverter
@@ -179,8 +180,32 @@ def get(self, pathname: str, startdatetime=None, enddatetime=None, trim=False):
             return self._get_array(pathname)
         elif type == RecordType.LocationInfo:
             return self._get_location_info(pathname)
+        elif type == RecordType.Text:
+            return self._get_text(pathname)
         return None
 
+    def _get_text(self, pathname: str):
+        textLength = 1024
+
+
+        BUFFER_TOO_SMALL = -1
+        textArray = []
+        status = self._native.hec_dss_textRetrieve(pathname, textArray, textLength)
+        while status == BUFFER_TOO_SMALL:
+            textLength *= 2
+            status = self._native.hec_dss_textRetrieve(pathname, textArray, textLength)
+            if textLength > 2*1048576:  # 2 MB
+                logger.error("Text record too large to read from '%s'", pathname)
+                return None
+
+        if status != 0:
+            logger.error("Error reading text from '%s'", pathname)
+            return None
+        text = Text()
+        text.id = pathname
+        text.text = textArray[0]
+        return text
+
     def _get_array(self, pathname: str):
         intValuesCount = [0]
         floatValuesCount = [0]
@@ -682,6 +707,10 @@ def put(self, container) -> int:
         elif type(container) is LocationInfo:
             status = self._native.hec_dss_locationStore(container,1)
             self._catalog = None
+        elif type(container) is Text:
+            text = container
+            status = self._native.hec_dss_textStore(text.id, text.text, len(text.text))
+            self._catalog = None
         else:
             raise NotImplementedError(f"unsupported record_type: {type(container)}. Expected types are: {RecordType.SUPPORTED_RECORD_TYPES.value}")
 
@@ -812,6 +841,8 @@ def set_debug_level(self, level) -> int:
             int: _description_
         """
         return self._native.hec_dss_set_debug_level(level)
+    
+    def 
 
 
 if __name__ == "__main__":

src/hecdss/native.py

@@ -1223,3 +1223,51 @@ def hec_dss_delete(self, pathname: str) -> int:
                 logger.error("Function call failed with result: %s", result)
 
             return result
+
+    def hec_dss_textStore(self, pathname, text, length=None):
+        """
+        Store text data in a DSS file.
+        Args:
+            pathname (str): The DSS pathname where the text will be stored.
+            text (str): The text data to store.
+            length (int, optional): The length of the text. If None, it will be set to the length of the text.
+        """
+        f = self.dll.hec_dss_textStore
+        f.argtypes = [
+            c_void_p,  # dss
+            c_char_p,  # pathname
+            c_char_p,  # text
+            c_int      # length
+        ]
+        f.restype = c_int
+
+        result = f(self.handle, pathname.encode("utf-8"),
+                    text.encode("utf-8"), 
+                    length if length is not None else len(text))
+    
+        return result
+    
+    def hec_dss_textRetrieve(self, pathname, buffer :List[str], buff_size: int) -> int:
+        """
+        Store text data in a DSS file.
+        Args:
+            pathname (str): The DSS pathname where the text will be stored.
+            text (str): The text data to store.
+            length (int, optional): The length of the text. If None, it will be set to the length of the text.
+        """
+        f = self.dll.hec_dss_textRetrieve
+        f.argtypes = [
+            c_void_p,  # dss
+            c_char_p,  # pathname
+            c_char_p,  # buffer
+            c_int      # buff_size
+        ]
+        f.restype = c_int
+
+        c_buffer = create_string_buffer(buff_size)
+        result = f(self.handle, pathname.encode("utf-8"),
+                    c_buffer, 
+                    buff_size)
+    
+        buffer[0] = c_buffer.value.decode("utf-8")
+        return result

src/hecdss/record_type.py

@@ -23,6 +23,7 @@ class RecordType(Enum):
         "PairedData",
         "GriddedData",
         "ArrayContainer",
+        "Text"
     ]
 
     @staticmethod

src/hecdss/text.py

@@ -0,0 +1,8 @@
+
+class Text:
+    def __init__(self):
+        """
+        Initialize a Text object with default values.
+        """
+        self.id = None
+        self.text = ""