Add to_csv/read_csv support for RegularTimeSeries

.gitignore

@@ -14,3 +14,6 @@ hecdss.dll
 libhecdss.so
 .idea
 .ipynb_checkpoints
+.venv/
+tests/csv_testing/
+

src/hecdss/dss_csv.py

@@ -0,0 +1,203 @@
+import csv
+import re
+from datetime import datetime, timedelta
+
+from .dsspath import DssPath
+from .irregular_timeseries import IrregularTimeSeries
+from .regular_timeseries import RegularTimeSeries
+
+
+def timeseries_to_csv(
+    series: RegularTimeSeries | IrregularTimeSeries, path: str, with_metadata: bool
+) -> None:
+    """
+    Exports a timeseries (either regular or irregular) to a .csv file.
+
+    Parameters:
+        series: The timeseries object to export. Must be either a RegularTimeSeries or IrregularTimeSeries
+        path (str): The file path where the .csv file will be exported.
+        with_metadata (bool): Whether to include metadata in the .csv file.
+    """
+    if not isinstance(series, (RegularTimeSeries, IrregularTimeSeries)):
+        raise TypeError("series must be a RegularTimeSeries or IrregularTimeSeries")
+
+    metadata_rows: list[str] = ["A", "B", "C", "D", "E", "F"]
+    with open(path, "w", newline="", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        if with_metadata:
+            if series.id:
+                id_components: list[str] = DssPath(series.id).path_to_list()
+            else:
+                id_components: list[str] = [""] * 6
+            for i in range(len(metadata_rows)):
+                row: str = metadata_rows[i]
+                metadata_value: str = id_components[i]
+                if row == "D":  # Skip D by convention # ts pattern
+                    if metadata_value == "ts-pattern":
+                        print("Warning: ts-pattern skipped")
+                    continue
+                writer.writerow([row, "", "", metadata_value])  # Writing metadata rows
+            writer.writerow(["Units", "", "", series.units])
+            if len(series.quality) > 0:
+                # Write column names with quality
+                writer.writerow(["Type", "Date/Time", series.data_type, "Quality"])
+            else:
+                # Write column names without quality
+                writer.writerow(["Type", "Date/Time", series.data_type])
+
+        time_format: str = ("%d%b%Y %H%M%S" if _needs_second_precision(series) else "%d%b%Y %H%M")
+
+        ordinate: int = 1
+        if len(series.quality) > 0:
+            for time, value, quality in zip(series.times, series.values, series.quality):
+                formatted_time: str = time.strftime(time_format)
+                writer.writerow([ordinate, formatted_time, value, quality])
+                ordinate += 1
+        else:
+            for time, value in zip(series.times, series.values):
+                formatted_time: str = time.strftime(time_format)
+                writer.writerow([ordinate, formatted_time, value])
+                ordinate += 1
+
+
+def timeseries_read_csv(cls: type[RegularTimeSeries] | type[IrregularTimeSeries], path: str) -> RegularTimeSeries | IrregularTimeSeries:
+    """
+    Reads a .csv file and builds a new timeseries of the given type.
+
+    Parameters:
+        cls: the class to build - either RegularTimeSeries or IrregularTimeSeries
+        path (str): File path to .csv that we are reading from
+
+    Returns:
+        An instance of cls populated from the .csv file.
+    """
+    if cls not in (RegularTimeSeries, IrregularTimeSeries):
+        raise TypeError("cls must be RegularTimeSeries or IrregularTimeSeries")
+
+    times, values, quality, units, data_type = [], [], [], "", ""
+    path_parts: dict[str, str] = {"A": "", "B": "", "C": "", "D": "", "E": "", "F": ""}
+    has_quality: bool = False  # flags
+
+    with open(path, "r", newline="", encoding="utf-8") as f:
+        reader = csv.reader(f)
+        for row in reader:
+            if not row:
+                continue
+            # first item in the row we grabbed, the first column's item
+            first_column_item: str = row[0].strip()
+            # If the first column item is a path component (['A', 'B', 'C', 'D', 'E', 'F'])
+            if first_column_item in path_parts:
+                path_parts[first_column_item] = row[-1].strip()  # last cell in csv row (convention)
+            elif first_column_item == "Units":
+                units = row[-1].strip()
+            elif first_column_item == "Type":  # reached the header
+                if len(row) >= 3:  # ['Type', 'Date/Time', data_type, ...potentially more]
+                    data_type = row[2].strip()
+                # ['Type', 'Date/Time', data_type, 'Quality', ...potentially more]
+                if len(row) >= 4 and row[3].strip() == "Quality":
+                    has_quality = True
+            else:  # Data row
+                if len(row) < 3:
+                    continue  # csv is malformed, something is missing
+
+                raw_time: str = row[1].strip()
+                # Time format is determined by length of raw_time
+                time_format: str = _get_time_format(raw_time)
+                if time_format is None:
+                    # Time format is unrecognized
+                    continue
+
+                # Do we need to roll over the day date? Yes if time is 2400
+                roll_day: bool = _need_roll_day(time_format, raw_time)
+                if roll_day:  # 2400 isn't a valid hour, so roll it to 0000 before parsing and add a day after
+                    raw_time = raw_time.replace(" 2400", " 0000")
+
+                try:
+                    time: datetime = datetime.strptime(raw_time, time_format)
+                except ValueError:
+                    continue  # Skip a malformed date
+
+                if roll_day:  # Convert a 24:00 time to 00:00 of the next day
+                    time += timedelta(days=1)
+
+                value_str: str = row[2].strip()
+                try:
+                    value: float = float(value_str) if value_str else 0.0
+                except ValueError:
+                    continue  # Skip a malformed value
+
+                times.append(time)
+                values.append(value)
+                if has_quality:  # Always keep quality index-aligned with values, defaulting a missing cell to 0
+                    quality_str: str = row[3].strip() if len(row) >= 4 else ""
+                    quality.append(int(quality_str) if quality_str else 0)
+
+    id_path: str = f"/{path_parts['A']}/{path_parts['B']}/{path_parts['C']}/{path_parts['D']}/{path_parts['E']}/{path_parts['F']}/"
+    interval: str | int = path_parts["E"]
+
+    return cls.create(
+        values=values,
+        times=times,
+        quality=quality,
+        units=units,
+        data_type=data_type,
+        interval=interval,
+        path=id_path,
+    )
+
+
+def _needs_second_precision(series: RegularTimeSeries | IrregularTimeSeries) -> bool:
+    """
+    Returns True if any datetime in the series has a non-zero seconds component
+
+    Parameters:
+        series: must be either a RegularTimeSeries or IrregularTimeSeries
+
+    Returns:
+        bool: whether we need seconds precision or not
+    """
+    return any(getattr(t, "second", 0) != 0 for t in series.times)
+
+
+def _get_time_format(raw_time: str) -> str | None:
+    """
+    Given a raw DSS time string, detect and return the correct time format, whether it be minutes or seconds precision.
+
+    Parameters:
+        raw_time (str): time in DSS string format (TODO: ISO)
+
+    Returns:
+        str | None: time format to use to convert to datetime 
+    """
+    dss_minutes_pattern: str = r"^\d{2}[A-Z][a-z]{2}\d{4} \d{4}$"
+    dss_seconds_pattern: str = r"^\d{2}[A-Z][a-z]{2}\d{4} \d{6}$"
+
+    if re.fullmatch(dss_minutes_pattern, raw_time):
+        return "%d%b%Y %H%M"
+    elif re.fullmatch(dss_seconds_pattern, raw_time):
+        return "%d%b%Y %H%M%S"
+    else:
+        return None
+
+
+def _need_roll_day(time_format: str, raw_time: str) -> bool:
+    """
+    Primarily to deal with the DSS time sometimes being "2400" and needing to roll over.
+
+    Parameters:
+        time_format (str): time format being used
+        raw_time (str): raw time string extracted from csv
+
+    Returns: 
+        bool: whether or not we need to roll over to the next day
+    """
+    if time_format != r"%d%b%Y %H%M" and time_format != r"%d%b%Y %H%M%S":
+        return False  # Only DSS formats can return True
+
+    # This should correctly catch if we need to roll day
+    roll_day_pattern: str = r"^\d{2}[A-Z][a-z]{2}\d{4} 2400(00)?$"
+
+    if re.fullmatch(roll_day_pattern, raw_time):
+        return True
+
+    return False

src/hecdss/dsspath.py

@@ -86,4 +86,13 @@ def print(self):
         print("c:" + self.path.C)
         print("d:" + self.path.D)
         print("e:" + self.path.E)
-        print("f:" + self.path.F)
+        print("f:" + self.path.F)
+
+    def path_to_list(self) -> list[str]:
+        """
+        Convert the DSS path to a list of its components.
+
+        Returns:
+            list: A list containing the components of the DSS path.
+        """
+        return [self.A, self.B, self.C, self.D, self.E, self.F]

src/hecdss/irregular_timeseries.py

@@ -85,6 +85,33 @@ def print_to_console(self):
         print("dataType='" + self.data_type + "'")
         for time, value in zip(self.times, self.values):
             print(f"Time: {time}, Value: {value}")
+
+    def to_csv(self, file_path: str, with_metadata: bool = True) -> None:
+        """
+        Exports the IrregularTimeSeries object to a .csv file.
+
+        Parameters:
+            file_path (str): The path to the .csv file where the data will be exported.
+            with_metadata (bool): Whether to include metadata in the exported file.
+        """
+        from .dss_csv import timeseries_to_csv
+        timeseries_to_csv(self, file_path, with_metadata)
+        print(f"Wrote IrregularTimeSeries to .csv file at {file_path}.")
+
+    @staticmethod
+    def read_csv(file_path: str) -> "IrregularTimeSeries":
+        """
+        Reads a .csv file and creates an IrregularTimeSeries instance from the data.
+
+        Parameters:
+            file_path (str): The path to the .csv file to read
+
+        Returns:
+            IrregularTimeSeries: A new instance of IrregularTimeSeries populated with the data from the .csv file.
+        """
+        from .dss_csv import timeseries_read_csv
+        return timeseries_read_csv(IrregularTimeSeries, file_path)
+
     @staticmethod
     def create(values, times, quality=[], units="", data_type="", interval=0, start_date="", time_granularity_seconds=1, julian_base_date=None, time_zone_name="", path=None, location_info=None):
         """

src/hecdss/regular_timeseries.py

@@ -1,9 +1,11 @@
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
 import numpy as np
 
 from .dateconverter import DateConverter
 from .dsspath import DssPath
-from datetime import datetime, timedelta
-from zoneinfo import ZoneInfo
+
 
 class RegularTimeSeries:
     def __init__(self):
@@ -85,7 +87,7 @@ def print_to_console(self):
         Prints the time series data to the console.
         """
         print("dsspath='" + self.id + "'")
-        print("units='"+self.units+"'")
+        print("units='" + self.units + "'")
         print("dataType='" + self.data_type + "'")
         print("Time,Value,Flag")
         if not len(self.quality) > 0:
@@ -95,6 +97,18 @@ def print_to_console(self):
             for time, value, flag in zip(self.times, self.values, self.quality):
                 print(f"{time}, {value}, {flag}")
 
+    def to_csv(self, file_path: str, with_metadata: bool = True) -> None:
+        """
+        Exports the RegularTimeSeries data to a .csv file.
+
+        Args:
+            file_path (str): The path to the .csv file where the data will be exported.
+            with_metadata (bool): Whether to include metadata in the exported file.
+        """
+        from .dss_csv import timeseries_to_csv
+        timeseries_to_csv(self, file_path, with_metadata)
+        print(f"Wrote RegularTimeSeries to .csv file at {file_path}.")
+
     def _get_interval_interval(self):
         """
         Converts the interval string to seconds.
@@ -124,7 +138,7 @@ def _get_interval_times(self):
             int: The interval in seconds, or "empty" if there are fewer than two dates.
         """
         if len(self.times) > 1 and type(self.times[0]) == datetime:
-            interval = self.times[1]-self.times[0]
+            interval = self.times[1] - self.times[0]
             total_seconds = interval.total_seconds()
             if total_seconds > 86400:
                 return "empty"
@@ -159,8 +173,9 @@ def _interval_to_times(self, new_interval):
         Args:
             new_interval (int): The new interval in seconds.
         """
-        is_leap = lambda y: y % 4 == 0 and y % 100 != 0 or y % 400 == 0
-        last_day = lambda y, m: 31 if m in (1,3,5,7,8,10,12) else 30 if m in (4,6,9,11) else 29 if is_leap(y) else 28
+        def is_leap(y): return y % 4 == 0 and y % 100 != 0 or y % 400 == 0
+        def last_day(y, m): return 31 if m in (1, 3, 5, 7, 8, 10, 12) else 30 if m in (
+            4, 6, 9, 11) else 29 if is_leap(y) else 28
         if type(self.start_date) == datetime:
             tz = ZoneInfo(self.time_zone_name) if self.time_zone_name else None
             first_time = self.start_date.replace(microsecond=0, tzinfo=tz)
@@ -208,25 +223,38 @@ def _interval_to_times(self, new_interval):
             else:
                 raise ValueError(f"Invalid interval seconds: {new_interval}")
 
-
     def _generate_times(self):
         """
         Generates times for the time series based on the interval and start date.
         """
-        if(len(self.times) > 0 and self.start_date == ""):
+        if (len(self.times) > 0 and self.start_date == ""):
             self.start_date = self.times[0]
 
         x = [self._get_interval_times(), self._get_interval_path(), self._get_interval_interval()]
         x = [i for i in x if i != "empty"]
-        if(not all(i == x[0] for i in x)):
+        if (not all(i == x[0] for i in x)):
             raise ValueError("inconsistent interval within arguments")
         elif len(x) != 3 and len(x) != 0:
             self._interval_to_interval(x[0])
             self._interval_to_path(x[0])
             self._interval_to_times(x[0])
 
     @staticmethod
-    def create(values, times=[], quality=[], units="", data_type="", interval="", start_date="", time_granularity_seconds=1, julian_base_date=0, time_zone_name="", path=None, location_info = None):
+    def read_csv(file_path: str) -> "RegularTimeSeries":
+        """
+        Reads a .csv file and creates a RegularTimeSeries instance from the data.
+
+        Parameters:
+            file_path (str): The path to the .csv file to read.
+
+        Returns:
+            RegularTimeSeries: A new instance of RegularTimeSeries populated with the data from the .csv file.
+        """
+        from .dss_csv import timeseries_read_csv
+        return timeseries_read_csv(RegularTimeSeries, file_path)
+
+    @staticmethod
+    def create(values, times=[], quality=[], units="", data_type="", interval="", start_date="", time_granularity_seconds=1, julian_base_date=0, time_zone_name="", path=None, location_info=None):
         """
         Creates a new instance of the RegularTimeSeries class with the specified parameters.
 
@@ -261,4 +289,4 @@ def create(values, times=[], quality=[], units="", data_type="", interval="", st
         rts.location_info = location_info
         rts._generate_times()
 
-        return rts
+        return rts