PairedData bug fix and CSV functionality

src/hecdss/dss_csv.py

@@ -1,12 +1,19 @@
+# THESE ARE ALL HELPER FUNCTIONS, MEANT TO BE TREATED AS PRIVATE. These functions are internally called in their respective classes.
+
 import csv
 import re
 from datetime import datetime, timedelta
 
 from .dsspath import DssPath
 from .irregular_timeseries import IrregularTimeSeries
+from .paired_data import PairedData
 from .regular_timeseries import RegularTimeSeries
 
 
+ROUND_PRECISION: int = 4
+DEFAULT_MISSING_VALUE = None
+
+
 def timeseries_to_csv(
     series: RegularTimeSeries | IrregularTimeSeries, path: str, with_metadata: bool
 ) -> None:
@@ -21,29 +28,23 @@ def timeseries_to_csv(
     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":
+            id_components: dict[str, str] = _id_to_path_parts(series.id)
+            for letter, value in id_components.items():
+                if letter == 'D':  # Skip D by convention # ts pattern
+                    if value == "ts-pattern":
                         print("Warning: ts-pattern skipped")
                     continue
-                writer.writerow([row, "", "", metadata_value])  # Writing metadata rows
+                writer.writerow([letter, "", "", 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])
+        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")
 
@@ -75,7 +76,7 @@ def timeseries_read_csv(cls: type[RegularTimeSeries] | type[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": ""}
+    path_parts: dict[str, str] = _empty_path_parts()
     has_quality: bool = False  # flags
 
     with open(path, "r", newline="", encoding="utf-8") as f:
@@ -122,17 +123,20 @@ def timeseries_read_csv(cls: type[RegularTimeSeries] | type[IrregularTimeSeries]
 
                 value_str: str = row[2].strip()
                 try:
-                    value: float = float(value_str) if value_str else 0.0
+                    value: float = float(value_str) if value_str else DEFAULT_MISSING_VALUE
                 except ValueError:
-                    continue  # Skip a malformed value
+                    value: float = DEFAULT_MISSING_VALUE  # If datetime is valid but not value, use DEFAULT_MISSING_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)
+                    try:
+                        quality.append(int(quality_str) if quality_str else 0)
+                    except ValueError:
+                        quality.append(0)
 
-    id_path: str = f"/{path_parts['A']}/{path_parts['B']}/{path_parts['C']}/{path_parts['D']}/{path_parts['E']}/{path_parts['F']}/"
+    id_path: str = _path_parts_to_id(path_parts)
     interval: str | int = path_parts["E"]
 
     return cls.create(
@@ -146,6 +150,184 @@ def timeseries_read_csv(cls: type[RegularTimeSeries] | type[IrregularTimeSeries]
     )
 
 
+def paired_data_to_csv(paired_data: PairedData, path: str, with_metadata: bool) -> None:
+    """
+    Exports a PairedData object to a .csv file.
+
+    Parameters:
+        paired_data (PairedData): Paired Data object to convert to csv.
+        path: (str): file path to export csv to.
+        with_metadata (bool): whether or not to include metadata in the csv file.
+
+    Returns:
+        None
+    """
+    if not isinstance(paired_data, PairedData):
+        raise TypeError("paired_data must be a PairedData!")
+
+    with open(path, "w", newline="", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        if with_metadata:
+            id_components: dict[str, str] = _id_to_path_parts(paired_data.id)
+            for letter, value in id_components.items():
+                if letter == 'D':  # Skipping D by convention
+                    continue
+                writer.writerow([letter, "", value])  # Writing metadata rows
+            labels: list[str] = ["Labels", ""]
+            for label in paired_data.labels:
+                labels.append(label)
+            writer.writerow(labels)
+            writer.writerow(["Units", paired_data.units_independent, paired_data.units_dependent])
+        writer.writerow(["Type", paired_data.type_independent, paired_data.type_dependent])
+
+        counter: int = 1  # 1st column index
+        x: float  # x is the same as "ordinate"
+        y_row: list[float]  # Each y_row is one row of y_values, as paired_data.values is Row-Major
+        for x, y_row in zip(paired_data.ordinates, paired_data.values):
+            full_row: list[float] = [_round_or_none(x)] + \
+                [_round_or_none(y) for y in y_row]
+            writer.writerow([counter] + full_row)
+            counter += 1
+    return
+
+
+def paired_data_read_csv(cls: type[PairedData], path: str) -> PairedData:
+    """
+    Reads a .csv file and builds a new paired data of the given type.
+
+    Parameters:
+        cls: the class to build - PairedData
+        path (str): File path to .csv that we are reading from
+
+    Returns:
+        An instance of cls populated from the .csv file.
+    """
+    if cls is not PairedData:
+        raise TypeError("cls must be PairedData")
+
+    x_values: list[float] = []
+    y_values: list[list[float]] = []
+    labels: list[str] = []
+    x_units: str = ""
+    x_type: str = ""
+    y_units: str = ""
+    y_type: str = ""
+    path_parts: dict[str, str] = _empty_path_parts()
+
+    with open(path, "r", newline="", encoding="utf-8") as f:
+        reader = csv.reader(f)
+        row: list[str]
+        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 == "Labels":
+                for label in row[2:]:  # First two elements of row are not labels
+                    labels.append(label)
+            elif first_column_item == "Units":
+                if len(row) < 2:
+                    continue
+                x_units = row[1].strip()
+                if len(row) < 3:
+                    continue
+                y_units = row[2].strip()  # If units change to a list, will have to update this
+            elif first_column_item == "Type":
+                if len(row) < 2:
+                    continue
+                x_type = row[1].strip()
+                if len(row) < 3:
+                    continue
+                y_type = row[2].strip()  # If type changes to a list, will also have to update this
+            else:  # Data row
+                if len(row) < 3:
+                    continue  # csv is malformed, something is missing
+                x_str: str = row[1].strip()
+                try:
+                    x: float = float(x_str) if x_str else DEFAULT_MISSING_VALUE
+                except ValueError:
+                    x: float = DEFAULT_MISSING_VALUE
+                x_values.append(x)
+
+                y_row: list[float] = []
+                y_row_str: list[str] = row[2:]
+                y_str: str
+                for y_str in y_row_str:
+                    try:
+                        y: float = float(y_str) if y_str else DEFAULT_MISSING_VALUE
+                    except ValueError:
+                        y: float = DEFAULT_MISSING_VALUE
+                    y_row.append(y)
+                y_values.append(y_row)
+
+    id_path: str = _path_parts_to_id(path_parts)
+
+    return cls.create(
+        x_values=x_values,
+        y_values=y_values,
+        labels=labels,
+        x_units=x_units,
+        x_type=x_type,
+        y_units=y_units,
+        y_type=y_type,
+        path=id_path,
+    )
+
+
+def _round_or_none(value: float | None) -> float | None:
+    """
+    Rounds a numeric value to ROUND_PRECISION, passing None through unchanged.
+
+    Parameters:
+        value (float | None): the value to round, or None if missing
+
+    Returns:
+        float | None: the rounded value, or None if value was None
+    """
+    return round(value, ROUND_PRECISION) if value is not None else None
+
+
+def _empty_path_parts() -> dict[str, str]:
+    """
+    Returns a fresh dict of empty A-F DSS path components.
+
+    Returns:
+        dict[str, str]: mapping of each path part letter to an empty string
+    """
+    return {"A": "", "B": "", "C": "", "D": "", "E": "", "F": ""}
+
+
+def _id_to_path_parts(dss_id: str | None) -> dict[str, str]:
+    """
+    Splits a DSS id string into its A-F components, or returns empties if id is missing.
+
+    Parameters:
+        dss_id (str | None): the DSS pathname to split, or None
+
+    Returns:
+        dict[str, str]: mapping of each path part letter to its value
+    """
+    if dss_id:
+        return DssPath(dss_id).path_to_dict()
+    return _empty_path_parts()
+
+
+def _path_parts_to_id(path_parts: dict[str, str]) -> str:
+    """
+    Rebuilds a '/A/B/C/D/E/F/' DSS id string from its component dict.
+
+    Parameters:
+        path_parts (dict[str, str]): mapping of each path part letter to its value
+
+    Returns:
+        str: the reconstructed DSS pathname
+    """
+    return f"/{path_parts['A']}/{path_parts['B']}/{path_parts['C']}/{path_parts['D']}/{path_parts['E']}/{path_parts['F']}/"
+
+
 def _needs_second_precision(series: RegularTimeSeries | IrregularTimeSeries) -> bool:
     """
     Returns True if any datetime in the series has a non-zero seconds component
@@ -164,7 +346,7 @@ 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)
+        raw_time (str): time in DSS string format
 
     Returns:
         str | None: time format to use to convert to datetime 

src/hecdss/dsspath.py

@@ -2,6 +2,7 @@
 from .dss_type import DssType
 from .record_type import RecordType
 
+
 class DssPath:
     """
     Manage parts of DSS path /A/B/C/D/E/F/
@@ -11,7 +12,7 @@ class DssPath:
     _timeSeriesFamily = [RecordType.IrregularTimeSeries, RecordType.RegularTimeSeries,
                          RecordType.RegularTimeSeriesProfile]
 
-    def __init__(self, path: str, recType = 0):
+    def __init__(self, path: str, recType=0):
         """
         Initialize a DssPath object.
 
@@ -95,4 +96,13 @@ def path_to_list(self) -> list[str]:
         Returns:
             list: A list containing the components of the DSS path.
         """
-        return [self.A, self.B, self.C, self.D, self.E, self.F]
+        return [self.A, self.B, self.C, self.D, self.E, self.F]
+
+    def path_to_dict(self) -> dict[str, str]:
+        """
+        Convert the DSS path to a dictionary of its components.
+
+        Returns:
+            dict[str, str]: A dictionary containing the components of the DSS path.
+        """
+        return {'A': self.A, 'B': self.B, 'C': self.C, 'D': self.D, 'E': self.E, 'F': self.F}

src/hecdss/paired_data.py

@@ -7,15 +7,16 @@ def __init__(self):
         """
         Initialize a PairedData object with default values.
         """
-        self.id = None
-        self.ordinates = np.empty(0)
-        self.values = np.empty(0)
-        self.labels = []
-        self.type_independent = ""
-        self.type_dependent = ""
-        self.units_independent = ""
-        self.units_dependent = ""
-        self.time_zone_name = ""
+        self.id: str = None
+        self.ordinates: np.ndarray = np.empty(0)
+        # ROW-MAJOR, each sublist represents the values for each ordinate
+        self.values: np.ndarray = np.empty(0)
+        self.labels: list[str] = []
+        self.type_independent: str = ""
+        self.type_dependent: str = ""
+        self.units_independent: str = ""
+        self.units_dependent: str = ""
+        self.time_zone_name: str = ""
         self.location_info = None
 
     def curve_count(self):
@@ -25,7 +26,22 @@ def curve_count(self):
         Returns:
         int: The number of curves.
         """
-        return len(self.values)
+        if len(self.values) == 0:
+            return 0
+        # values is ROW-MAJOR, each sublist represents a row of curve values for each ordinate
+        return len(self.values[0])
+
+    def to_csv(self, file_path: str, with_metadata: bool = True) -> None:
+        """
+        Exports the PairedData instance 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 paired_data_to_csv
+        paired_data_to_csv(self, file_path, with_metadata)
+        print(f"Wrote PairedData to .csv file at {file_path}.")
 
     # def to_data_frame(self, include_index=False):
     #     """
@@ -47,6 +63,20 @@ def curve_count(self):
     #
     #     return pd.DataFrame(data)
 
+    @staticmethod
+    def read_csv(file_path: str) -> "PairedData":
+        """
+        Reads a .csv file and creates an PairedData instance from the data.
+
+        Parameters:
+            file_path (str): The path to the .csv file to read
+
+        Returns:
+            PairedData: A new instance of PairedData populated with the data from the .csv file.
+        """
+        from .dss_csv import paired_data_read_csv
+        return paired_data_read_csv(PairedData, file_path)
+
     @staticmethod
     def create(x_values, y_values, labels=[], x_units="", x_type="", y_units="", y_type="", time_zone_name="", path=None):
         """
@@ -75,4 +105,4 @@ def create(x_values, y_values, labels=[], x_units="", x_type="", y_units="", y_t
         pd.type_independent = x_type
         pd.type_dependent = y_type
         pd.time_zone_name = time_zone_name
-        return pd
+        return pd