diff --git a/pyproject.toml b/pyproject.toml index f3057d4..488fd9e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "hec-dss-python" -version = "0.1.29" +version = "0.1.30" description = "Python wrapper for the HEC-DSS file database C library." authors = ["Hydrologic Engineering Center"] license = "MIT" diff --git a/setup.cfg b/setup.cfg index 0e81e2c..e99061b 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,6 +1,6 @@ [metadata] name = hecdss -version = 0.1.29 +version = 0.1.30 author = Hydrologic Engineering Center author_email =hec.dss@usace.army.mil description = Python wrapper for the HEC-DSS file database C library. diff --git a/src/hecdss/dss_csv.py b/src/hecdss/dss_csv.py index 6386268..3a18e70 100644 --- a/src/hecdss/dss_csv.py +++ b/src/hecdss/dss_csv.py @@ -14,33 +14,21 @@ DEFAULT_MISSING_VALUE = None -def timeseries_to_csv( - series: RegularTimeSeries | IrregularTimeSeries, path: str, with_metadata: bool -) -> None: +def timeseries_to_csv(series: RegularTimeSeries | IrregularTimeSeries, path: str) -> 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") 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(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([letter, "", "", value]) # Writing metadata rows - writer.writerow(["Units", "", "", series.units]) - - header: list[str] = ["Type", "Date/Time", series.data_type] + + header: list[str] = ["Date/Time", "Value"] has_quality: bool = len(series.quality) > 0 has_notes: bool = len(series.notes) > 0 @@ -50,14 +38,17 @@ def timeseries_to_csv( if has_notes: header.append("Notes") + header.append("Units") + header.append("Type") + header.append("Path") + writer.writerow(header) time_format: str = ("%d%b%Y %H%M%S" if _needs_second_precision(series) else "%d%b%Y %H%M") for i, (time, value) in enumerate(zip(series.times, series.values)): - ordinate: str = str(i + 1) formatted_time: str = time.strftime(time_format) - row: list[str] = [ordinate, formatted_time, value] + row: list[str] = [formatted_time, value] if has_quality: try: @@ -69,6 +60,12 @@ def timeseries_to_csv( row.append(series.notes[i]) except IndexError: row.append(DEFAULT_MISSING_VALUE) + + if i == 0: # Add metadata to only first row + row.append(series.units) + row.append(series.data_type) + row.append(series.id) + writer.writerow(row) @@ -86,72 +83,87 @@ def timeseries_read_csv(cls: type[RegularTimeSeries] | type[IrregularTimeSeries] if cls not in (RegularTimeSeries, IrregularTimeSeries): raise TypeError("cls must be RegularTimeSeries or IrregularTimeSeries") - times, values, quality, notes, units, data_type = [], [], [], [], "", "" - path_parts: dict[str, str] = _empty_path_parts() + times, values, quality, notes, units, data_type, id = [], [], [], [], "", "", "" column_index: dict[str, int] = {} with open(path, "r", newline="", encoding="utf-8") as f: + is_first_data_row: bool = True + header_detected: bool = False + 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 == "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', 'Notes'] - column_index = { - name.strip(): i for i, name in enumerate(row) if i >= 3} - else: # Data row - if len(row) < 3: - continue # csv is malformed, something is missing - - raw_time: str = row[1].strip() - time_format: str = _get_time_format(raw_time) - if time_format is None: - # Time format is unrecognized - continue - - 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 first_column_item == "Date/Time": + header_detected = True + column_index = {name.strip(): i for i, name in enumerate(row)} + continue + + if len(row) < 2: + continue # need at least Date/Time and Value + + if not header_detected: + raise ValueError("No header found!") + + if is_first_data_row: + is_first_data_row = False + + units_idx: int = column_index.get("Units") + if units_idx is not None and len(row) > units_idx: + units = row[units_idx].strip() + type_idx: int = column_index.get("Type") + if type_idx is not None and len(row) > type_idx: + data_type = row[type_idx].strip() + path_idx: int = column_index.get("Path") + if path_idx is not None and len(row) > path_idx: + id = row[path_idx].strip() + + time_idx: int = column_index.get("Date/Time") + raw_time: str = row[time_idx].strip() + time_format: str = _get_time_format(raw_time) + if time_format is None: + # Time format is unrecognized + continue - if roll_day: # Convert a 24:00 time to 00:00 of the next day - time += timedelta(days=1) + 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") - value_str: str = row[2].strip() - try: - value: float = float(value_str) if value_str else DEFAULT_MISSING_VALUE - except ValueError: - value: float = DEFAULT_MISSING_VALUE # If datetime is valid but not value, use DEFAULT_MISSING_VALUE + try: + time: datetime = datetime.strptime(raw_time, time_format) + except ValueError: + continue # Skip a malformed date - times.append(time) - values.append(value) + if roll_day: # Convert a 24:00 time to 00:00 of the next day + time += timedelta(days=1) - quality_idx = column_index.get("Quality") - if quality_idx is not None: - quality_str = row[quality_idx].strip() if len(row) > quality_idx else "" - try: - quality.append(int(quality_str) if quality_str else 0) - except ValueError: - quality.append(0) + value_idx: int = column_index.get("Value") + value_str: str = row[value_idx].strip() + try: + value: float = float(value_str) if value_str else DEFAULT_MISSING_VALUE + except ValueError: + value: float = DEFAULT_MISSING_VALUE # If datetime is valid but not value, use DEFAULT_MISSING_VALUE - notes_idx = column_index.get("Notes") - if notes_idx is not None: - notes.append(row[notes_idx].strip() if len(row) > notes_idx else "") + times.append(time) + values.append(value) - id_path: str = _path_parts_to_id(path_parts) + quality_idx: int = column_index.get("Quality") + if quality_idx is not None: + quality_str = row[quality_idx].strip() if len(row) > quality_idx else "" + try: + quality.append(int(quality_str) if quality_str else 0) + except ValueError: + quality.append(0) + + notes_idx: int = column_index.get("Notes") + if notes_idx is not None: + notes.append(row[notes_idx].strip() if len(row) > notes_idx else "") + + path_parts: dict = _id_to_path_parts(id) interval: str | int = path_parts["E"] return cls.create( @@ -162,18 +174,17 @@ def timeseries_read_csv(cls: type[RegularTimeSeries] | type[IrregularTimeSeries] units=units, data_type=data_type, interval=interval, - path=id_path, + path=id or None, ) -def paired_data_to_csv(paired_data: PairedData, path: str, with_metadata: bool) -> None: +def paired_data_to_csv(paired_data: PairedData, path: str) -> 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 @@ -183,27 +194,29 @@ def paired_data_to_csv(paired_data: PairedData, path: str, with_metadata: bool) 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 + + path_parts: dict = _id_to_path_parts(paired_data.id) + c_part: str = path_parts["C"] + + curve_count: int = paired_data.curve_count() + header_labels: list[str] = list(paired_data.labels) + [""] * \ + (curve_count - len(paired_data.labels)) + + writer.writerow([_prefix_before_dash(c_part)] + header_labels + ["X Units", + "Y Units", "X Type", "Y Type", "Path"]) + 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 + is_first_data_row: bool = True 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 + if is_first_data_row: + is_first_data_row = False + full_row.extend([paired_data.units_independent, paired_data.units_dependent, + paired_data.type_independent, paired_data.type_dependent, paired_data.id]) + writer.writerow(full_row) + return @@ -228,58 +241,81 @@ def paired_data_read_csv(cls: type[PairedData], path: str) -> PairedData: x_type: str = "" y_units: str = "" y_type: str = "" - path_parts: dict[str, str] = _empty_path_parts() + id: str = "" + column_index: dict[str, int] = {} with open(path, "r", newline="", encoding="utf-8") as f: + is_first_data_row: bool = True + header_detected: bool = False + 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 + + if row[-1] == "Path": + header_detected = True + column_index = {name.strip(): i for i, name in enumerate(row)} + + # The index of the x units column is the number of curves + 1 (1 is the x column) + x_units_col: int = column_index.get("X Units", len(row) - 5) + curve_count: int = x_units_col - 1 + + for label in row[1:x_units_col]: 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() + continue + + if len(row) < 2: + continue + + if not header_detected: + raise ValueError("No header found!") + + if not is_first_data_row and curve_count != len(row) - 1: + raise ValueError("Curve count mismatch!") + + if is_first_data_row: + is_first_data_row = False + + if len(row) - 1 < curve_count: + raise ValueError("Curve count mismatch!") + + x_units_idx: int = column_index.get("X Units") + if x_units_idx is not None and len(row) > x_units_idx: + x_units = row[x_units_idx].strip() + y_units_idx: int = column_index.get("Y Units") + if y_units_idx is not None and len(row) > y_units_idx: + y_units = row[y_units_idx].strip() + x_type_idx: int = column_index.get("X Type") + if x_type_idx is not None and len(row) > x_type_idx: + x_type = row[x_type_idx].strip() + y_type_idx: int = column_index.get("Y Type") + if y_type_idx is not None and len(row) > y_type_idx: + y_type = row[y_type_idx].strip() + path_idx: int = column_index.get("Path") + if path_idx is not None and len(row) > path_idx: + id = row[path_idx].strip() + + # If we make it here, we are a data row + x_column_idx: int = 0 + x_str: str = row[x_column_idx].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[1:curve_count + 1] + y_str: str + for y_str in y_row_str: try: - x: float = float(x_str) if x_str else DEFAULT_MISSING_VALUE + y: float = float(y_str) if y_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) + y: float = DEFAULT_MISSING_VALUE + y_row.append(y) + y_values.append(y_row) return cls.create( x_values=x_values, @@ -289,7 +325,7 @@ def paired_data_read_csv(cls: type[PairedData], path: str) -> PairedData: x_type=x_type, y_units=y_units, y_type=y_type, - path=id_path, + path=id, ) @@ -331,19 +367,6 @@ def _id_to_path_parts(dss_id: str | None) -> dict[str, str]: 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 @@ -399,3 +422,16 @@ def _need_roll_day(time_format: str, raw_time: str) -> bool: return True return False + + +def _prefix_before_dash(s: str) -> str: + """ + Helper to determine column name for independent column of PairedData. + + Parameters: + s (str): Input string (c-part of path) + + Returns: + str: Substring of c-path part that contains independent column name + """ + return s.split("-", 1)[0] if "-" in s else 'X' diff --git a/src/hecdss/dsspath.py b/src/hecdss/dsspath.py index b086f4b..06607e5 100644 --- a/src/hecdss/dsspath.py +++ b/src/hecdss/dsspath.py @@ -89,15 +89,6 @@ def print(self): print("e:" + self.path.E) 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] - def path_to_dict(self) -> dict[str, str]: """ Convert the DSS path to a dictionary of its components. diff --git a/src/hecdss/irregular_timeseries.py b/src/hecdss/irregular_timeseries.py index d660c9d..5bab014 100644 --- a/src/hecdss/irregular_timeseries.py +++ b/src/hecdss/irregular_timeseries.py @@ -99,16 +99,15 @@ def print_to_console(self): line += f", Note: {self.notes[i]}" print(line) - def to_csv(self, file_path: str, with_metadata: bool = True) -> None: + def to_csv(self, file_path: str) -> 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) + timeseries_to_csv(self, file_path) print(f"Wrote IrregularTimeSeries to .csv file at {file_path}.") @staticmethod diff --git a/src/hecdss/paired_data.py b/src/hecdss/paired_data.py index 6f85ac1..ace55ef 100644 --- a/src/hecdss/paired_data.py +++ b/src/hecdss/paired_data.py @@ -31,16 +31,15 @@ def curve_count(self): # 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: + def to_csv(self, file_path: str) -> 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) + paired_data_to_csv(self, file_path) print(f"Wrote PairedData to .csv file at {file_path}.") # def to_data_frame(self, include_index=False): diff --git a/src/hecdss/regular_timeseries.py b/src/hecdss/regular_timeseries.py index 38bbae5..f97a3dd 100644 --- a/src/hecdss/regular_timeseries.py +++ b/src/hecdss/regular_timeseries.py @@ -109,16 +109,15 @@ def print_to_console(self): row += f", {self.notes[i]}" print(row) - def to_csv(self, file_path: str, with_metadata: bool = True) -> None: + def to_csv(self, file_path: str) -> 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) + timeseries_to_csv(self, file_path) print(f"Wrote RegularTimeSeries to .csv file at {file_path}.") def _get_interval_interval(self): diff --git a/tests/test_pd_csv.py b/tests/test_pd_csv.py index d0654fa..b7e6604 100644 --- a/tests/test_pd_csv.py +++ b/tests/test_pd_csv.py @@ -1,6 +1,6 @@ +# import pandas as pd import math import unittest -from datetime import datetime from unittest.mock import mock_open, patch from file_manager import FileManager @@ -8,6 +8,7 @@ from hecdss import HecDss from hecdss.dss_csv import ( DEFAULT_MISSING_VALUE, + _prefix_before_dash, paired_data_read_csv, paired_data_to_csv, ) @@ -41,20 +42,19 @@ def read_pd_from_string(self, content: str) -> PairedData: with patch("builtins.open", m): return PairedData.read_csv("fake.csv") - def write_pd_to_string(self, pd: PairedData, with_metadata: bool = True) -> str: + def write_pd_to_string(self, pd: PairedData) -> str: """ Helper to run to_csv against a mocked file and return everything written. Parameters: pd (PairedData): the object to export - with_metadata (bool): whether metadata rows are written Returns: str: the full CSV text that to_csv wrote """ mock_file = mock_open() with patch("builtins.open", mock_file): - pd.to_csv("fake_path.csv", with_metadata=with_metadata) + pd.to_csv("fake_path.csv") handle = mock_file() return "".join(call.args[0] for call in handle.write.call_args_list) @@ -97,7 +97,7 @@ def test_multiple_curves_to_csv(self): mock_file = mock_open() with patch("builtins.open", mock_file): - pd.to_csv("fake_path.csv", with_metadata=True) + pd.to_csv("fake_path.csv") mock_file.assert_called_once_with( "fake_path.csv", "w", newline="", encoding="utf-8" @@ -106,132 +106,112 @@ def test_multiple_curves_to_csv(self): handle = mock_file() written_data = "".join(call.args[0] for call in handle.write.call_args_list) - self.assertIn("A,,A", written_data) - self.assertIn("B,,B", written_data) - self.assertIn("C,,C", written_data) - self.assertIn("1,1.0,0.0,1.0,2.0", written_data) - self.assertIn("2,1.5,0.0,1.5,3.0", written_data) - self.assertIn("3,4.0,0.0,4.0,8.0", written_data) - - def test_to_csv_without_metadata(self): - num_curves: int = 3 - x_values: list[float] = [1.0, 1.5, 4.0] - y_values: list[list[float]] = [[x * i for i in range(num_curves)] for x in x_values] - labels: list[str] = [f"x times {i}" for i in range(num_curves)] - path: str = "/A/B/C///Source: I made it up/" - pd: PairedData = PairedData.create( - x_values=x_values, - y_values=y_values, - labels=labels, - path=path - ) - - mock_file = mock_open() - with patch("builtins.open", mock_file): - pd.to_csv("fake_path.csv", with_metadata=False) - - mock_file.assert_called_once_with( - "fake_path.csv", "w", newline="", encoding="utf-8" - ) - - handle = mock_file() - written_data = "".join(call.args[0] for call in handle.write.call_args_list) - - self.assertNotIn("A,,A", written_data) - self.assertNotIn("B,,B", written_data) - self.assertNotIn("C,,C", written_data) - self.assertNotIn("F,,Source: I made it up", written_data) - self.assertNotIn("Labels", written_data) - self.assertIn("Type", written_data) - self.assertIn("1,1.0,0.0,1.0,2.0", written_data) - self.assertIn("2,1.5,0.0,1.5,3.0", written_data) - self.assertIn("3,4.0,0.0,4.0,8.0", written_data) - - # def test_temp(self): - # file_path: str = "./tests/data/examples-all-data-types.dss" - # with HecDss(file_path) as dss: - # data_path: str = "/paired-data-multi-column/RIVERDALE/FREQ-FLOW/MAX ANALYTICAL//1969-01 H33(MAX)/" - # data: PairedData = dss.get(data_path) - # export_path: str = "./tests/csv_testing/paired_data_test.csv" - # data.to_csv(export_path) - - def test_to_csv_metadata_rows_present(self): - """Every metadata row (path parts, labels, units, type) is written.""" - written = self.write_pd_to_string(self.make_pd(), with_metadata=True) - self.assertIn("A,,A", written) - self.assertIn("B,,B", written) - self.assertIn("C,,C", written) - self.assertIn("E,,E", written) - self.assertIn("F,,F", written) - self.assertIn("Labels,,L1,L2", written) - self.assertIn("Units,FEET,CFS", written) - self.assertIn("Type,UNT,LINEAR", written) - - def test_to_csv_skips_d_part(self): - """The D (date) path part is dropped by convention; E is still written.""" + # C part "C" has no dash, so the x column name falls back to "X". + self.assertIn( + "X,x times 0,x times 1,x times 2,X Units,Y Units,X Type,Y Type,Path", written_data) + # The full path is preserved verbatim in the first data row. + self.assertIn("/A/B/C///Source: I made it up/", written_data) + # Data rows: x then the row-major y values (no leading counter). + self.assertIn("1.0,0.0,1.0,2.0", written_data) + self.assertIn("1.5,0.0,1.5,3.0", written_data) + self.assertIn("4.0,0.0,4.0,8.0", written_data) + + def test_to_csv_metadata_present(self): + """The header carries labels + column names; the first data row carries the + units, types, and full path.""" + written = self.write_pd_to_string(self.make_pd()) + self.assertIn("X,L1,L2,X Units,Y Units,X Type,Y Type,Path", written) + self.assertIn("FEET,CFS,UNT,LINEAR,/A/B/C/DATE/E/F/", written) + + def test_to_csv_preserves_full_path(self): + """The whole DSS path (including the D part) is written to the Path cell.""" pd = self.make_pd(path="/A/B/C/DPART/EPART/FPART/") - written = self.write_pd_to_string(pd, with_metadata=True) - self.assertNotIn("DPART", written) - self.assertIn("E,,EPART", written) - self.assertIn("F,,FPART", written) + written = self.write_pd_to_string(pd) + self.assertIn("/A/B/C/DPART/EPART/FPART/", written) def test_to_csv_data_rows(self): - """Data rows are index, x, then the row-major y values.""" - written = self.write_pd_to_string(self.make_pd(), with_metadata=True) - self.assertIn("1,1.0,10.0,100.0", written) - self.assertIn("2,2.0,20.0,200.0", written) + """Data rows are x then the row-major y values.""" + written = self.write_pd_to_string(self.make_pd()) + self.assertIn("1.0,10.0,100.0", written) + self.assertIn("2.0,20.0,200.0", written) - def test_to_csv_ordinate_counter_increments(self): - """The leading index column counts up from 1, one per ordinate.""" + def test_to_csv_has_no_counter_column(self): + """There is no leading index column; a data row starts with the x value.""" pd = self.make_pd( x_values=[5.0, 6.0, 7.0], y_values=[[1.0], [2.0], [3.0]], ) - written = self.write_pd_to_string(pd, with_metadata=False) - self.assertIn("1,5.0,1.0", written) - self.assertIn("2,6.0,2.0", written) - self.assertIn("3,7.0,3.0", written) + written = self.write_pd_to_string(pd) + self.assertIn("5.0,1.0", written) + self.assertIn("6.0,2.0", written) + self.assertIn("7.0,3.0", written) + # A leading "1,5.0" counter would look like this - make sure it is absent. + self.assertNotIn("1,5.0", written) def test_to_csv_single_curve(self): """A single dependent column round-trips through the writer fine.""" pd = self.make_pd(y_values=[[10.0], [20.0]], labels=["only"]) - written = self.write_pd_to_string(pd, with_metadata=True) - self.assertIn("Labels,,only", written) - self.assertIn("1,1.0,10.0", written) - self.assertIn("2,2.0,20.0", written) + written = self.write_pd_to_string(pd) + self.assertIn("X,only,X Units,Y Units,X Type,Y Type,Path", written) + self.assertIn("1.0,10.0", written) + self.assertIn("2.0,20.0", written) def test_to_csv_rounds_to_four_decimals(self): """x and y are rounded to ROUND_PRECISION (4) decimal places.""" pd = self.make_pd(x_values=[1.23456], y_values=[[9.87654]], labels=["a"]) - written = self.write_pd_to_string(pd, with_metadata=False) - self.assertIn("1,1.2346,9.8765", written) + written = self.write_pd_to_string(pd) + self.assertIn("1.2346,9.8765", written) def test_to_csv_negative_values(self): """Negative x/y values are written verbatim.""" pd = self.make_pd(x_values=[-1.0], y_values=[[-5.5, -0.25]], labels=["a", "b"]) - written = self.write_pd_to_string(pd, with_metadata=False) - self.assertIn("1,-1.0,-5.5,-0.25", written) + written = self.write_pd_to_string(pd) + self.assertIn("-1.0,-5.5,-0.25", written) def test_to_csv_empty_ordinates_writes_no_data_rows(self): - """No ordinates -> metadata is present but there are no numbered data rows.""" + """No ordinates -> only the header is written. Because units/types/path ride + on the first data row, they are absent when there are no data rows.""" # Labels avoid digits so the "no data row" check can't collide with a label. pd = self.make_pd(x_values=[], y_values=[], labels=["aa", "bb"]) - written = self.write_pd_to_string(pd, with_metadata=True) - self.assertIn("Units,FEET,CFS", written) - self.assertIn("Type,UNT,LINEAR", written) - self.assertNotIn("1,", written) # no data rows + written = self.write_pd_to_string(pd) + self.assertIn("X,aa,bb,X Units,Y Units,X Type,Y Type,Path", written) + self.assertNotIn("FEET", written) # no data row -> no metadata written + self.assertNotIn("1.0", written) # no data rows - def test_to_csv_none_id_writes_empty_path_parts(self): - """A None id still emits empty A-F rows rather than raising.""" + def test_to_csv_none_id_writes_empty_path(self): + """A None id writes an empty Path cell rather than raising or writing 'None'.""" pd = self.make_pd(path=None) - written = self.write_pd_to_string(pd, with_metadata=True) - for letter in ("A", "B", "C", "E", "F"): - self.assertIn(f"{letter},,", written) + written = self.write_pd_to_string(pd) + # metadata tail is present, with an empty trailing Path cell + self.assertIn("FEET,CFS,UNT,LINEAR,", written) + self.assertNotIn("None", written) + + def test_to_csv_x_column_name_from_c_part(self): + """The x column name in the header is the C path part up to its first dash.""" + pd = self.make_pd(path="/A/B/STAGE-FLOW/D/E/F/") + written = self.write_pd_to_string(pd) + self.assertIn("STAGE,L1,L2,X Units", written) + + def test_to_csv_x_column_name_defaults_to_X(self): + """A C part with no dash (or an empty C part) falls back to 'X'.""" + no_dash = self.write_pd_to_string(self.make_pd(path="/A/B/NODASH/D/E/F/")) + self.assertIn("X,L1,L2,X Units", no_dash) + + empty_c = self.write_pd_to_string(self.make_pd(path="/A/B//D/E/F/")) + self.assertIn("X,L1,L2,X Units", empty_c) + + def test_prefix_before_dash_helper(self): + """The x-column-name helper takes the substring before the first dash.""" + self.assertEqual(_prefix_before_dash("STAGE-FLOW"), "STAGE") + self.assertEqual(_prefix_before_dash("FREQ-FLOW"), "FREQ") + self.assertEqual(_prefix_before_dash("A-B-C"), "A") + self.assertEqual(_prefix_before_dash("NODASH"), "X") + self.assertEqual(_prefix_before_dash(""), "X") def test_to_csv_raises_on_non_paireddata(self): """paired_data_to_csv rejects anything that isn't a PairedData.""" with self.assertRaises(TypeError): - paired_data_to_csv("not a paired data", "fake.csv", True) + paired_data_to_csv("not a paired data", "fake.csv") # ================================================================== # # read_csv (read) @@ -239,17 +219,10 @@ def test_to_csv_raises_on_non_paireddata(self): def test_basic_read_csv(self): content: str = ( - "A,,paired-data-test\n" - "B,,berkeley\n" - "C,,FREQ-FLOW\n" - "E,,\n" - "F,,Source: I made it up\n" - "Labels,,IMAGINARY,X,Y,Z\n" - "Units,DAYS,PPG\n" - "Type,TIME,POINTS\n" - "1,0,1,5,6,8\n" - "2,1,3,5,6,9\n" - "3,5,1,2,3,4\n" + "FREQ,IMAGINARY,X,Y,Z,X Units,Y Units,X Type,Y Type,Path\n" + "0,1,5,6,8,DAYS,PPG,TIME,POINTS,/paired-data-test/berkeley/FREQ-FLOW///Source: I made it up/\n" + "1,3,5,6,9\n" + "5,1,2,3,4\n" ) pd: PairedData = self.read_pd_from_string(content) self.assertEqual(pd.labels, ['IMAGINARY', 'X', 'Y', 'Z']) @@ -266,56 +239,72 @@ def test_basic_read_csv(self): def test_read_csv_single_curve(self): content = ( - "Type,TIME,POINTS\n" - "1,0,10\n" - "2,1,20\n" + "FLOW,only,X Units,Y Units,X Type,Y Type,Path\n" + "0,10\n" + "1,20\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.ordinates.tolist(), [0.0, 1.0]) self.assertEqual(pd.values.tolist(), [[10.0], [20.0]]) self.assertEqual(pd.curve_count(), 1) - def test_read_csv_data_only_no_metadata(self): - """Data rows with no metadata rows still populate ordinates/values.""" + def test_read_csv_no_header_raises(self): + """Data rows with no header row (nothing ending in 'Path') are rejected.""" content = ( "1,0,10,100\n" "2,1,20,200\n" ) - pd = self.read_pd_from_string(content) - self.assertEqual(pd.ordinates.tolist(), [0.0, 1.0]) - self.assertEqual(pd.values.tolist(), [[10.0, 100.0], [20.0, 200.0]]) - self.assertEqual(pd.labels, []) - self.assertEqual(pd.units_independent, "") + with self.assertRaises(ValueError): + self.read_pd_from_string(content) - def test_read_csv_builds_id_from_path_parts(self): + def test_read_csv_reads_full_path(self): + """The id is read verbatim from the Path cell of the first data row, + including its D and E parts.""" content = ( - "A,,SITE\n" - "B,,LOC\n" - "C,,STAGE-FLOW\n" - "E,,\n" - "F,,MADEUP\n" - "Type,STAGE,FLOW\n" - "1,0,10\n" + "STAGE,only,X Units,Y Units,X Type,Y Type,Path\n" + "0,10,STAGE,FLOW,,,/SITE/LOC/STAGE-FLOW/DPART/EPART/MADEUP/\n" ) pd = self.read_pd_from_string(content) - # D is never present in a paired-data CSV, so both D and E are empty here. - self.assertEqual(pd.id, "/SITE/LOC/STAGE-FLOW///MADEUP/") + self.assertEqual(pd.id, "/SITE/LOC/STAGE-FLOW/DPART/EPART/MADEUP/") def test_read_csv_curve_count(self): content = ( - "Type,TIME,POINTS\n" - "1,0,1,2,3\n" - "2,1,4,5,6\n" + "TIME,a,b,c,X Units,Y Units,X Type,Y Type,Path\n" + "0,1,2,3\n" + "1,4,5,6\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.curve_count(), 3) + def test_read_csv_gets_labels_from_header(self): + """Labels come from the header columns between the x column and 'X Units'.""" + content = ( + "STAGE,alpha,beta,X Units,Y Units,X Type,Y Type,Path\n" + "0,1,2\n" + ) + pd = self.read_pd_from_string(content) + self.assertEqual(pd.labels, ["alpha", "beta"]) + self.assertEqual(pd.curve_count(), 2) + + def test_read_csv_curve_count_independent_of_labels(self): + """Curve count is derived from the 'X Units' boundary, not the label count, + so a header with blank label cells still reads every curve.""" + content = ( + "X,,,X Units,Y Units,X Type,Y Type,Path\n" + "0,10,100\n" + "1,20,200\n" + ) + pd = self.read_pd_from_string(content) + self.assertEqual(pd.curve_count(), 2) + self.assertEqual(pd.values.tolist(), [[10.0, 100.0], [20.0, 200.0]]) + self.assertEqual(pd.labels, ["", ""]) + def test_read_csv_malformed_x_defaults_missing(self): - """A non-numeric x cell falls back to the missing-value default (0.0).""" + """A non-numeric x cell falls back to the missing-value default.""" content = ( - "Type,TIME,POINTS\n" - "1,not-a-number,10,100\n" - "2,2,20,200\n" + "X,a,b,X Units,Y Units,X Type,Y Type,Path\n" + "not-a-number,10,100\n" + "2,20,200\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.ordinates.tolist(), [DEFAULT_MISSING_VALUE, 2.0]) @@ -325,30 +314,30 @@ def test_read_csv_malformed_y_defaults_missing(self): """A non-numeric y cell falls back to the default WITHOUT shortening the row, so the value matrix stays rectangular.""" content = ( - "Type,TIME,POINTS\n" - "1,1,bad,3\n" - "2,2,5,6\n" + "X,a,b,X Units,Y Units,X Type,Y Type,Path\n" + "1,bad,3\n" + "2,5,6\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.values.tolist(), [[DEFAULT_MISSING_VALUE, 3.0], [5.0, 6.0]]) def test_read_csv_empty_cells_default_missing(self): - """Blank x and y cells become the missing-value default (0.0).""" + """Blank x and y cells become the missing-value default.""" content = ( - "Type,TIME,POINTS\n" - "1,,\n" - "2,2,\n" + "X,a,X Units,Y Units,X Type,Y Type,Path\n" + ",\n" + "2,\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.ordinates.tolist(), [DEFAULT_MISSING_VALUE, 2.0]) self.assertEqual(pd.values.tolist(), [[DEFAULT_MISSING_VALUE], [DEFAULT_MISSING_VALUE]]) def test_read_csv_skips_short_rows(self): - """A data row with fewer than 3 columns is malformed and skipped.""" + """A data row with fewer than 2 columns is malformed and skipped.""" content = ( - "Type,TIME,POINTS\n" - "1,5\n" # only 2 columns -> skipped - "2,2,9\n" + "X,a,X Units,Y Units,X Type,Y Type,Path\n" + "5\n" # only 1 column -> skipped + "2,9\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.ordinates.tolist(), [2.0]) @@ -357,50 +346,47 @@ def test_read_csv_skips_short_rows(self): def test_read_csv_strips_numeric_whitespace(self): """Surrounding whitespace around numeric cells is stripped before parsing.""" content = ( - "Type,TIME,POINTS\n" - "1, 1 , 2 , 3 \n" + "X,a,b,X Units,Y Units,X Type,Y Type,Path\n" + " 1 , 2 , 3 \n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.ordinates.tolist(), [1.0]) self.assertEqual(pd.values.tolist(), [[2.0, 3.0]]) def test_read_csv_labels_preserve_whitespace(self): - """Labels are captured verbatim from row[2:] (not stripped).""" + """Labels are captured verbatim from the header (not stripped).""" content = ( - "Labels,, spacey ,X\n" - "Type,TIME,POINTS\n" - "1,0,1,2\n" + "FREQ, spacey ,X,X Units,Y Units,X Type,Y Type,Path\n" + "0,1,2\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.labels, [" spacey ", "X"]) - def test_read_csv_partial_units_row(self): - """A Units row missing the dependent column leaves y_units empty, no crash.""" + def test_read_csv_partial_units(self): + """A first data row missing the dependent-units cell leaves y_units empty.""" content = ( - "Units,DAYS\n" - "Type,TIME,POINTS\n" - "1,0,10\n" + "DAY,a,X Units,Y Units,X Type,Y Type,Path\n" + "0,10,DAYS\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.units_independent, "DAYS") self.assertEqual(pd.units_dependent, "") - def test_read_csv_partial_type_row(self): - """A Type row missing the dependent column leaves y_type empty, no crash.""" + def test_read_csv_partial_type(self): + """A first data row missing the dependent-type cell leaves y_type empty.""" content = ( - "Type,TIME\n" - "1,0,10\n" + "TIME,a,X Units,Y Units,X Type,Y Type,Path\n" + "0,10,,,TIME\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.type_independent, "TIME") self.assertEqual(pd.type_dependent, "") - def test_read_csv_units_first_column_only(self): - """A bare 'Units' row (no values at all) is ignored gracefully.""" + def test_read_csv_no_metadata_cells_leaves_fields_empty(self): + """A first data row carrying only x and y leaves the metadata fields empty.""" content = ( - "Units\n" - "Type,TIME,POINTS\n" - "1,0,10\n" + "X,a,X Units,Y Units,X Type,Y Type,Path\n" + "0,10\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.units_independent, "") @@ -415,27 +401,43 @@ def test_read_csv_empty_file(self): def test_read_csv_blank_lines_ignored(self): content = ( - "Type,TIME,POINTS\n" + "X,a,X Units,Y Units,X Type,Y Type,Path\n" "\n" - "1,0,10\n" + "0,10\n" "\n" - "2,1,20\n" + "1,20\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.ordinates.tolist(), [0.0, 1.0]) self.assertEqual(pd.values.tolist(), [[10.0], [20.0]]) def test_read_csv_ragged_rows_raise(self): - """Genuinely ragged data (rows with different column counts) can't form a - rectangular value matrix, so PairedData.create raises when numpy builds it.""" + """A later data row whose width disagrees with the curve count is rejected.""" content = ( - "Type,TIME,POINTS\n" - "1,0,1,2\n" # 2 y-values - "2,1,3\n" # 1 y-value + "X,a,b,X Units,Y Units,X Type,Y Type,Path\n" + "0,1,2\n" # 2 y-values + "1,3\n" # 1 y-value ) with self.assertRaises(ValueError): self.read_pd_from_string(content) + def test_read_csv_short_first_row_is_rejected(self): + content = ( + "X,a,b,X Units,Y Units,X Type,Y Type,Path\n" + "1,7\n" # header implies 2 curves, only 1 y-value present + ) + with self.assertRaises(ValueError): + self.read_pd_from_string(content) + + def test_read_csv_x_column_name_colliding_with_label(self): + content = ( + "STAGE,STAGE,X Units,Y Units,X Type,Y Type,Path\n" + "5,10\n" + ) + pd = self.read_pd_from_string(content) + self.assertEqual(pd.ordinates.tolist(), [5.0]) + self.assertEqual(pd.values.tolist(), [[10.0]]) + def test_read_csv_wrong_cls_raises(self): """paired_data_read_csv only builds PairedData; other classes are rejected.""" with self.assertRaises(TypeError): @@ -448,7 +450,7 @@ def test_read_csv_wrong_cls_raises(self): def test_round_trip_basic(self): path = self.test_files.create_test_file(".csv") pd = self.make_pd() - pd.to_csv(path, with_metadata=True) + pd.to_csv(path) result = PairedData.read_csv(path) self.assertEqual(result.ordinates.tolist(), [1.0, 2.0]) @@ -458,8 +460,8 @@ def test_round_trip_basic(self): self.assertEqual(result.units_dependent, "CFS") self.assertEqual(result.type_independent, "UNT") self.assertEqual(result.type_dependent, "LINEAR") - # The D (date) part is dropped by the writer, so it comes back empty. - self.assertEqual(result.id, "/A/B/C//E/F/") + # The full path (including the D part) survives the round trip. + self.assertEqual(result.id, "/A/B/C/DATE/E/F/") def test_round_trip_multiple_curves(self): path = self.test_files.create_test_file(".csv") @@ -468,7 +470,7 @@ def test_round_trip_multiple_curves(self): y_values=[[1.0, 2.0, 3.0], [4.0, 5.0, 6.0], [7.0, 8.0, 9.0]], labels=["a", "b", "c"], ) - pd.to_csv(path, with_metadata=True) + pd.to_csv(path) result = PairedData.read_csv(path) self.assertEqual(result.curve_count(), 3) @@ -478,30 +480,26 @@ def test_round_trip_multiple_curves(self): ) self.assertEqual(result.labels, ["a", "b", "c"]) - def test_round_trip_without_metadata(self): - """Without metadata, units/labels/id are lost, but data and (because the - Type row is written unconditionally) the types survive.""" + def test_round_trip_no_labels(self): + """A PairedData written with no labels still round-trips its curves; the + header pads a blank label column per curve so the count survives.""" path = self.test_files.create_test_file(".csv") - pd = self.make_pd() - pd.to_csv(path, with_metadata=False) + pd = self.make_pd(labels=[]) + pd.to_csv(path) result = PairedData.read_csv(path) - self.assertEqual(result.ordinates.tolist(), [1.0, 2.0]) + self.assertEqual(result.curve_count(), 2) self.assertEqual(result.values.tolist(), [[10.0, 100.0], [20.0, 200.0]]) - self.assertEqual(result.labels, []) - self.assertEqual(result.units_independent, "") - self.assertEqual(result.units_dependent, "") - self.assertEqual(result.id, "///////") - # Type row is always written, so type survives the no-metadata round trip. - self.assertEqual(result.type_independent, "UNT") - self.assertEqual(result.type_dependent, "LINEAR") + self.assertEqual(result.ordinates.tolist(), [1.0, 2.0]) + # Padded blank labels come back, one per curve. + self.assertEqual(result.labels, ["", ""]) def test_round_trip_rounding(self): """Values are rounded to 4 decimals on write, and that rounded value is exactly what is read back.""" path = self.test_files.create_test_file(".csv") pd = self.make_pd(x_values=[1.23456], y_values=[[2.34567]], labels=["a"]) - pd.to_csv(path, with_metadata=True) + pd.to_csv(path) result = PairedData.read_csv(path) self.assertEqual(result.ordinates.tolist(), [1.2346]) @@ -514,18 +512,24 @@ def test_round_trip_negative_values(self): y_values=[[-5.5], [0.0], [5.5]], labels=["a"], ) - pd.to_csv(path, with_metadata=True) + pd.to_csv(path) result = PairedData.read_csv(path) self.assertEqual(result.ordinates.tolist(), [-1.0, 0.0, 1.0]) self.assertEqual(result.values.tolist(), [[-5.5], [0.0], [5.5]]) + def test_round_trip_x_column_name(self): + """The x column name written from the C part is exactly what appears as the + first header cell (it is not stored on PairedData, only in the CSV header).""" + path = self.test_files.create_test_file(".csv") + pd = self.make_pd(path="/A/B/STAGE-FLOW/D/E/F/") + pd.to_csv(path) + with open(path, "r", encoding="utf-8") as f: + first_line = f.readline() + self.assertTrue(first_line.startswith("STAGE,")) + # ================================================================== # - # weird / adversarial edge cases - # - # These probe surprising inputs. Tests that DOCUMENT current behavior - # (even when that behavior is a footgun) pass; two known bugs are pinned - # with @unittest.expectedFailure so they stay visible until fixed. + # edge cases # ================================================================== # def test_read_csv_quoted_label_with_comma_round_trips(self): @@ -534,51 +538,31 @@ def test_read_csv_quoted_label_with_comma_round_trips(self): pd = self.make_pd( x_values=[1.0], y_values=[[2.0]], labels=["flow, cfs"], ) - pd.to_csv(path, with_metadata=True) + pd.to_csv(path) result = PairedData.read_csv(path) self.assertEqual(result.labels, ["flow, cfs"]) - def test_read_csv_metadata_after_data_still_applies(self): - """The reader is order-independent: a Units row placed AFTER the data rows - still populates the units (nothing forces metadata to come first).""" + def test_read_csv_metadata_only_from_first_data_row(self): + """Metadata is positional: it is read from the first data row (the one right + after the header), and later rows need not repeat it.""" content = ( - "1,0,10\n" - "Units,DAYS,PPG\n" + "X,a,X Units,Y Units,X Type,Y Type,Path\n" + "0,10,DAYS,PPG,TIME,POINTS,/the/id/\n" + "1,20\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.units_independent, "DAYS") self.assertEqual(pd.units_dependent, "PPG") - self.assertEqual(pd.values.tolist(), [[10.0]]) - - def test_read_csv_ignores_counter_column(self): - """The leading index column is never validated on read - non-sequential or - even non-numeric counters are ignored entirely (x comes from column 2).""" - content = ( - "Type,,\n" - "999,0,10\n" - "hello,1,20\n" - ) - pd = self.read_pd_from_string(content) - self.assertEqual(pd.ordinates.tolist(), [0.0, 1.0]) + self.assertEqual(pd.type_independent, "TIME") + self.assertEqual(pd.type_dependent, "POINTS") + self.assertEqual(pd.id, "/the/id/") self.assertEqual(pd.values.tolist(), [[10.0], [20.0]]) - def test_read_csv_duplicate_units_last_wins(self): - """When a keyword row appears twice, the last occurrence overwrites.""" - content = ( - "Units,AA,BB\n" - "Units,CC,DD\n" - "Type,,\n" - "1,0,10\n" - ) - pd = self.read_pd_from_string(content) - self.assertEqual(pd.units_independent, "CC") - self.assertEqual(pd.units_dependent, "DD") - def test_read_csv_whitespace_only_cells_default_missing(self): """Cells that are only whitespace strip to empty and become the default.""" content = ( - "Type,,\n" - "1, , \n" + "X,a,X Units,Y Units,X Type,Y Type,Path\n" + " , \n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.ordinates.tolist(), [DEFAULT_MISSING_VALUE]) @@ -587,78 +571,67 @@ def test_read_csv_whitespace_only_cells_default_missing(self): def test_read_csv_inf_and_nan_parse_through(self): """float() accepts 'inf'/'nan', so these land in the array as real inf/NaN rather than being treated as missing. Surprising, given the missing-value - default is 0.0 - documented here so a future change is caught.""" + default - documented here so a future change is caught.""" content = ( - "Type,,\n" - "1,inf,nan\n" - "2,-inf,5\n" + "X,a,X Units,Y Units,X Type,Y Type,Path\n" + "inf,nan\n" + "-inf,5\n" ) pd = self.read_pd_from_string(content) self.assertTrue(math.isinf(pd.ordinates[0])) self.assertTrue(math.isnan(pd.values[0][0])) self.assertTrue(math.isinf(pd.ordinates[1]) and pd.ordinates[1] < 0) - def test_read_csv_miscased_keyword_becomes_phantom_data_row(self): - """FOOTGUN: keyword matching is case-sensitive, so a mis-cased 'units' row - is not recognized as metadata. It falls through to the data branch, its - text cells fail to parse, and it is silently injected as a (0.0, [0.0]) - data point instead of raising or being ignored.""" + def test_read_csv_miscased_path_header_not_detected(self): + """FOOTGUN: header detection is case-sensitive on 'Path'. A mis-cased 'path' + is not recognized as the header, so the reader sees a data row before any + header and raises.""" content = ( - "units,DAYS,PPG\n" # lowercase -> NOT recognized as a Units row - "Type,TIME,POINTS\n" - "1,5,10\n" + "X,a,X Units,Y Units,X Type,Y Type,path\n" # lowercase -> not a header + "0,10\n" ) - pd = self.read_pd_from_string(content) - # The real units were never captured... - self.assertEqual(pd.units_independent, "") - # ...and a bogus leading data point was manufactured from the keyword row. - self.assertEqual(pd.ordinates.tolist(), [DEFAULT_MISSING_VALUE, 5.0]) - self.assertEqual(pd.values.tolist(), [[DEFAULT_MISSING_VALUE], [10.0]]) - - def test_read_csv_extra_metadata_column_grabs_last_cell(self): - """FOOTGUN: path parts are read via the 'last cell' convention (row[-1]), - so a metadata row with an unexpected trailing column captures the trailing - value instead of the intended one.""" + with self.assertRaises(ValueError): + self.read_pd_from_string(content) + + def test_read_csv_id_from_path_column_not_last_cell(self): + """The id is read from the 'Path' column position, not the last cell, so a + stray trailing column on the first data row does not hijack the id.""" content = ( - "A,,SITE,EXTRA\n" # intended A=SITE, but row[-1] is 'EXTRA' - "Type,,\n" - "1,0,10\n" + "X,a,X Units,Y Units,X Type,Y Type,Path\n" + "0,10,U,V,XT,YT,/the/id/,EXTRA\n" ) pd = self.read_pd_from_string(content) - self.assertIn("EXTRA", pd.id) - self.assertNotIn("SITE", pd.id) - - def test_curve_count_on_dataless_read_should_be_zero(self): - """BUG: reading a CSV with metadata but no data rows yields an empty - PairedData, and curve_count() then does len(self.values[0]) -> IndexError. - It should return 0. Pinned as expectedFailure until curve_count guards - against an empty value array.""" + self.assertEqual(pd.id, "/the/id/") + self.assertNotIn("EXTRA", pd.id or "") + + def test_curve_count_on_dataless_read_is_zero(self): + """A header with no data rows yields an empty PairedData, and curve_count() + returns 0 (it guards against an empty value array).""" content = ( - "Units,DAYS,PPG\n" - "Type,TIME,POINTS\n" + "X,a,b,X Units,Y Units,X Type,Y Type,Path\n" ) pd = self.read_pd_from_string(content) self.assertEqual(pd.curve_count(), 0) def test_to_csv_missing_value_does_not_crash(self): - """A blank/malformed numeric cell is read back as DEFAULT_MISSING_VALUE - (None), producing an object-dtype array. The writer must tolerate a None - ordinate/value (write an empty cell) rather than crash trying to round it.""" + """A blank ordinate is read back as DEFAULT_MISSING_VALUE (None), producing + an object-dtype array. The writer must tolerate a None ordinate (write an + empty cell) rather than crash trying to round it.""" pd = self.read_pd_from_string( - "Type,TIME,POINTS\n" - "1,,10\n" # blank ordinate -> None - "2,2,20\n" + "X,a,X Units,Y Units,X Type,Y Type,Path\n" + ",10\n" # blank ordinate -> None + "2,20\n" ) - written = self.write_pd_to_string(pd, with_metadata=True) - self.assertIn("1,,10.0", written) - self.assertIn("2,2.0,20.0", written) + written = self.write_pd_to_string(pd) + self.assertIn(",10.0", written) # first row: empty x, then y + self.assertIn("2.0,20.0", written) def test_round_trip_nan_value_survives(self): """A NaN dependent value is written as 'nan' and read back as a real NaN (it is NOT collapsed into the missing-value default).""" path = self.test_files.create_test_file(".csv") pd = self.make_pd(x_values=[1.0], y_values=[[float("nan")]], labels=["a"]) - pd.to_csv(path, with_metadata=True) + pd.to_csv(path) result = PairedData.read_csv(path) self.assertTrue(math.isnan(result.values.tolist()[0][0])) diff --git a/tests/test_ts_csv.py b/tests/test_ts_csv.py index f0d9035..e8f0163 100644 --- a/tests/test_ts_csv.py +++ b/tests/test_ts_csv.py @@ -12,6 +12,17 @@ class TestCSV(unittest.TestCase): + """ + Tests for the CSV convention: + + Date/Time,Value[,Quality][,Notes],Units,Type,Path +