diff --git a/.gitignore b/.gitignore index f93f57e..b464af1 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,6 @@ hecdss.dll libhecdss.so .idea .ipynb_checkpoints +.venv/ +tests/csv_testing/ + diff --git a/src/hecdss/dss_csv.py b/src/hecdss/dss_csv.py new file mode 100644 index 0000000..c6d8e79 --- /dev/null +++ b/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 diff --git a/src/hecdss/dsspath.py b/src/hecdss/dsspath.py index f5ea6bf..f58d954 100644 --- a/src/hecdss/dsspath.py +++ b/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) \ No newline at end of file + 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] \ No newline at end of file diff --git a/src/hecdss/irregular_timeseries.py b/src/hecdss/irregular_timeseries.py index 8832b1c..328bc2c 100644 --- a/src/hecdss/irregular_timeseries.py +++ b/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): """ diff --git a/src/hecdss/regular_timeseries.py b/src/hecdss/regular_timeseries.py index bd61f79..0591612 100644 --- a/src/hecdss/regular_timeseries.py +++ b/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,17 +223,16 @@ 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]) @@ -226,7 +240,21 @@ def _generate_times(self): 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 \ No newline at end of file + return rts diff --git a/tests/test_csv.py b/tests/test_csv.py new file mode 100644 index 0000000..4f783a9 --- /dev/null +++ b/tests/test_csv.py @@ -0,0 +1,547 @@ +import unittest +from datetime import datetime +from unittest.mock import mock_open, patch + +from file_manager import FileManager + +from hecdss import HecDss +from hecdss.irregular_timeseries import IrregularTimeSeries +from hecdss.regular_timeseries import RegularTimeSeries + + +class TestCSV(unittest.TestCase): + + def setUp(self) -> None: + self.test_files = FileManager() + + def tearDown(self) -> None: + self.test_files.cleanup() + + def test_to_csv_writes_correct_structure(self): + # Create a dummy RegularTimeSeries instance + rts = RegularTimeSeries.create( + values=[10.5, 20.0], + times=[datetime(2021, 9, 1, 6, 0), datetime(2021, 9, 1, 12, 0)], + units="CFS", + data_type="INST-VAL", + path="/A/B/C/01Sep2021/6Hour/F/", + ) + + # Mock 'open' and capture written content + mock_file = mock_open() + with patch("builtins.open", mock_file): + rts.to_csv("fake_path.csv", with_metadata=True) + + # Assert that open was called with correct parameters + mock_file.assert_called_once_with( + "fake_path.csv", "w", newline="", encoding="utf-8" + ) + + # Extract all written data + handle = mock_file() + written_data = "".join(call.args[0] for call in handle.write.call_args_list) + + # Assertions on the CSV content structure + self.assertIn("A,,,A", written_data) + self.assertIn("B,,,B", written_data) + self.assertIn("C,,,C", written_data) + self.assertIn("Units,,,CFS", written_data) + self.assertIn("Type,Date/Time,INST-VAL", written_data) + self.assertIn("1,01Sep2021 0600,10.5", written_data) + self.assertIn("2,01Sep2021 1200,20.0", written_data) + + def test_to_csv_without_metadata(self): + """No metadata rows should be written; only data rows.""" + rts = RegularTimeSeries.create( + values=[1.0, 2.0], + times=[datetime(2021, 9, 1, 6, 0), datetime(2021, 9, 1, 12, 0)], + units="CFS", + data_type="INST-VAL", + path="/A/B/C//6Hour/F/", + ) + mock_file = mock_open() + with patch("builtins.open", mock_file): + rts.to_csv("fake.csv", with_metadata=False) + handle = mock_file() + written = "".join(call.args[0] for call in handle.write.call_args_list) + self.assertNotIn("Units", written) + self.assertNotIn("Type,Date/Time", written) + self.assertIn("1,01Sep2021 0600,1.0", written) + + def test_to_csv_empty_times(self): + rts = RegularTimeSeries.create( + values=[], + times=[], + units="CFS", + data_type="INST-VAL", + path="/A/B/C//6Hour/F/", + ) + + mock_file = mock_open() + with patch("builtins.open", mock_file): + rts.to_csv("fake_path.csv", with_metadata=True) + + handle = mock_file() + written_data = "".join(call.args[0] for call in handle.write.call_args_list) + + self.assertIn("Units,,,CFS", written_data) + self.assertIn("Type,Date/Time,INST-VAL", written_data) + self.assertNotIn("1,", written_data) # No data rows should be present + + def test_to_csv_second_precision(self): + rts = RegularTimeSeries.create( + values=[i for i in range(10)], + times=[datetime(2021, 9, 1, 6, 0, i) for i in range(10)], + units="CFS", + data_type="INST-VAL", + path="/A/B/C//1Second/F/", + ) + + mock_file = mock_open() + with patch("builtins.open", mock_file): + rts.to_csv("fake_path.csv", with_metadata=True) + + handle = mock_file() + written_data = "".join(call.args[0] for call in handle.write.call_args_list) + self.assertIn("Type,Date/Time,INST-VAL", written_data) + for i in range(10): + self.assertIn(f"{i + 1},01Sep2021 06000{i},", written_data) + + def test_to_csv_with_quality(self): + """When quality is present, header gets 'Quality' col and rows get flags.""" + rts = RegularTimeSeries.create( + values=[1.0, 2.0], + times=[datetime(2021, 9, 1, 6, 0), datetime(2021, 9, 1, 12, 0)], + quality=[0, 5], + units="CFS", + data_type="INST-VAL", + path="/A/B/C//6Hour/F/", + ) + mock_file = mock_open() + with patch("builtins.open", mock_file): + rts.to_csv("fake.csv", with_metadata=True) + handle = mock_file() + written = "".join(call.args[0] for call in handle.write.call_args_list) + self.assertIn("Type,Date/Time,INST-VAL,Quality", written) + self.assertIn("1,01Sep2021 0600,1.0,0", written) + self.assertIn("2,01Sep2021 1200,2.0,5", written) + + def read_rts_from_string(self, content): + """Helper to run read_csv against an in-memory CSV string.""" + m = mock_open(read_data=content) + with patch("builtins.open", m): + return RegularTimeSeries.read_csv("fake.csv") + + def test_read_csv_basic(self): + content = ( + "A,,,A\n" + "B,,,B\n" + "C,,,FLOW\n" + "E,,,6Hour\n" + "F,,,F\n" + "Units,,,CFS\n" + "Type,Date/Time,INST-VAL\n" + "1,01Sep2021 0600,10.5\n" + "2,01Sep2021 1200,20.0\n" + ) + rts = self.read_rts_from_string(content) + self.assertEqual(rts.units, "CFS") + self.assertEqual(rts.data_type, "INST-VAL") + self.assertEqual(rts.values.tolist(), [10.5, 20.0]) + self.assertEqual(rts.times[0], datetime(2021, 9, 1, 6, 0)) + + def test_read_csv_midnight_2400_rolls_to_next_day(self): + content = ( + "E,,,1Day\n" + "Type,Date/Time,INST-VAL\n" + "1,31Aug2021 2400,10.5\n" + "2,01Sep2021 2400,20.0\n" + ) + rts = self.read_rts_from_string(content) + self.assertEqual( + rts.times, [datetime(2021, 9, 1, 0, 0), datetime(2021, 9, 2, 0, 0)] + ) + self.assertEqual(rts.values.tolist(), [10.5, 20.0]) + + def test_read_csv_with_quality(self): + content = ( + "Type,Date/Time,INST-VAL,Quality\n" + "1,01Sep2021 0600,10.5,0\n" + "2,01Sep2021 1200,20.0,5\n" + ) + rts = self.read_rts_from_string(content) + self.assertEqual(rts.values.tolist(), [10.5, 20.0]) + self.assertEqual(rts.quality, [0, 5]) + + def test_read_csv_with_partial_quality(self): + content: tuple[str] = ( + "Type,Date/Time,INST-VAL,Quality\n" + "1,05Nov2004 0200,8,0\n" + "2,05Nov2004 0300,9\n" # missing quality! + "3,05Nov2004 0400,10,1\n" + ) + rts = self.read_rts_from_string(content) + self.assertEqual(rts.values.tolist()[0], 8) + self.assertEqual(rts.values.tolist()[2], 10) + self.assertEqual(rts.quality[0], 0) + self.assertEqual(rts.quality[2], 1) + + def test_read_csv_skips_malformed_rows(self): + content = ( + "Type,Date/Time,INST-VAL\n" + "1,01Sep2021 0600,10.5\n" + "2,not-a-date,20.0\n" + "3,01Sep2021 1200,not-a-number\n" + "4,01Sep2021 1800,30.0\n" + ) + rts = self.read_rts_from_string(content) + self.assertEqual(rts.values.tolist(), [10.5, 30.0]) + self.assertEqual( + rts.times, [datetime(2021, 9, 1, 6, 0), datetime(2021, 9, 1, 18, 0)] + ) + + def test_read_csv_seconds_precision_basic(self): + """Reading HHMMSS (seconds-precision) timestamps with no 2400 rollover involved. + Uses a 15-second gap (a standard DSS interval) so RegularTimeSeries can infer + the interval from the deltas without a metadata E row.""" + content = ( + "Type,Date/Time,INST-VAL\n" + "1,01Sep2021 060000,10.5\n" + "2,01Sep2021 060015,20.0\n" + ) + rts = self.read_rts_from_string(content) + self.assertEqual(rts.values.tolist(), [10.5, 20.0]) + self.assertEqual( + rts.times, + [datetime(2021, 9, 1, 6, 0, 0), datetime(2021, 9, 1, 6, 0, 15)], + ) + + def test_read_csv_skips_wrong_length_time(self): + """A time field that isn't 4 (minutes) or 6 (seconds) digits doesn't match + either DSS format, so the row should be skipped rather than raise.""" + content = ( + "Type,Date/Time,INST-VAL\n" + "1,01Sep2021 0600,10.5\n" + "2,01Sep2021 12345,20.0\n" # 5-digit clock -- not a valid DSS format + "3,01Sep2021 1800,30.0\n" + ) + rts = self.read_rts_from_string(content) + self.assertEqual(rts.values.tolist(), [10.5, 30.0]) + self.assertEqual( + rts.times, [datetime(2021, 9, 1, 6, 0), datetime(2021, 9, 1, 18, 0)] + ) + + def test_read_csv_single_digit_day_skipped(self): + """The format regex requires a zero-padded 2-digit day (matching this + library's own writer output), so a single-digit day doesn't match + and the row is skipped.""" + content = ( + "Type,Date/Time,INST-VAL\n" + "1,1Sep2021 0600,10.5\n" # single-digit day + "2,01Sep2021 1200,20.0\n" + ) + rts = self.read_rts_from_string(content) + self.assertEqual(rts.values.tolist(), [20.0]) + self.assertEqual(rts.times, [datetime(2021, 9, 1, 12, 0)]) + + def test_read_csv_month_case_mismatch_skipped(self): + """The format regex requires a title-case month abbreviation (matching + this library's own writer output); other casings don't match and are + skipped, even though datetime.strptime itself would accept them.""" + content = ( + "Type,Date/Time,INST-VAL\n" + "1,01SEP2021 0600,10.5\n" # all-caps month + "2,01sep2021 1200,20.0\n" # all-lowercase month + "3,01Sep2021 1800,30.0\n" + ) + rts = self.read_rts_from_string(content) + self.assertEqual(rts.values.tolist(), [30.0]) + self.assertEqual(rts.times, [datetime(2021, 9, 1, 18, 0)]) + + def test_read_csv_2400_with_nonzero_seconds_not_treated_as_rollover(self): + """24:00:15 is not a valid DSS midnight-rollover (only 24:00:00 is) and + should be rejected as malformed rather than silently rolled to the + next day with the seconds preserved. Covers the same edge case as + test_year_is_2400, but for the RegularTimeSeries read path.""" + content = ( + "Type,Date/Time,INST-VAL\n" + "1,15Sep2021 240015,10.5\n" + ) + rts = self.read_rts_from_string(content) + self.assertEqual(rts.values.tolist(), []) + self.assertEqual(rts.times, []) + + def test_read_csv_empty_file(self): + rts = self.read_rts_from_string("") + self.assertEqual(rts.values.tolist(), []) + self.assertEqual(rts.times, []) + + def test_read_csv_single_row_uses_path_interval(self): + content = "E,,,6Hour\n" "Type,Date/Time,INST-VAL\n" "1,01Sep2021 0600,10.5\n" + rts = self.read_rts_from_string(content) + self.assertEqual(rts.values.tolist(), [10.5]) + self.assertEqual(rts.times, [datetime(2021, 9, 1, 6, 0)]) + + def test_read_csv_irregular_interval_raises(self): + """RegularTimeSeries.read_csv expects a genuinely regular interval; + data implying an irregular gap (with no usable E row) is not valid + input for this class (that's what IrregularTimeSeries is for).""" + content = ( + "Type,Date/Time,INST-VAL\n" + "1,01Sep2021 0600,10.5\n" + "2,01Sep2021 0637,20.0\n" + ) + with self.assertRaises(ValueError): + self.read_rts_from_string(content) + + def test_round_trip_basic(self): + path = self.test_files.create_test_file(".csv") + rts = RegularTimeSeries.create( + values=[10.5, 20.0], + times=[datetime(2021, 9, 1, 6, 0), datetime(2021, 9, 1, 12, 0)], + units="CFS", + data_type="INST-VAL", + path="/A/B/C/01Sep2021/6Hour/F/", + ) + rts.to_csv(path, with_metadata=True) + result = RegularTimeSeries.read_csv(path) + + self.assertEqual(result.units, "CFS") + self.assertEqual(result.data_type, "INST-VAL") + self.assertEqual(result.interval, "6Hour") + self.assertEqual(result.id, "/A/B/C//6Hour/F/") + self.assertEqual(result.values.tolist(), [10.5, 20.0]) + self.assertEqual( + result.times, + [datetime(2021, 9, 1, 6, 0), datetime(2021, 9, 1, 12, 0)], + ) + + def test_round_trip_with_quality(self): + path = self.test_files.create_test_file(".csv") + rts = RegularTimeSeries.create( + values=[1.0, 2.0, 3.0], + times=[ + datetime(2021, 9, 1, 6, 0), + datetime(2021, 9, 1, 12, 0), + datetime(2021, 9, 1, 18, 0), + ], + quality=[0, 5, 10], + units="CFS", + data_type="INST-VAL", + path="/A/B/C//6Hour/F/", + ) + rts.to_csv(path, with_metadata=True) + result = RegularTimeSeries.read_csv(path) + + self.assertEqual(result.values.tolist(), [1.0, 2.0, 3.0]) + self.assertEqual(result.quality, [0, 5, 10]) + + def test_round_trip_without_metadata_infers_interval_from_times(self): + """With no metadata rows, units/data_type come back empty but values, + times, and the interval/id (inferred from the time deltas) are still + recovered correctly.""" + path = self.test_files.create_test_file(".csv") + rts = RegularTimeSeries.create( + values=[1.0, 2.0], + times=[datetime(2021, 9, 1, 6, 0), datetime(2021, 9, 1, 12, 0)], + units="CFS", + data_type="INST-VAL", + path="/A/B/C//6Hour/F/", + ) + rts.to_csv(path, with_metadata=False) + result = RegularTimeSeries.read_csv(path) + + self.assertEqual(result.units, "") + self.assertEqual(result.data_type, "") + self.assertEqual(result.interval, 21600) + self.assertEqual(result.id, "/////6Hour//") + self.assertEqual(result.values.tolist(), [1.0, 2.0]) + self.assertEqual( + result.times, + [datetime(2021, 9, 1, 6, 0), datetime(2021, 9, 1, 12, 0)], + ) + + # IRREGULAR TIME SERIES TESTS: + + def test_basic_to_csv_irregular(self): + """ + Basic structure test for irregular time series to_csv + """ + # Create a dummy IrregularTimeSeries instance + its = IrregularTimeSeries.create( + values=[10.5, 20.0, 42.0], + times=[ + datetime(2021, 9, 1, 0, 0), + datetime(2021, 9, 2, 0, 0), + datetime(2021, 9, 4, 0, 0), + ], # inconsistent time interval + units="CFS", + data_type="INST-VAL", + path="/A/B/C/01Sep2021/E/F/", + ) + + # Mock 'open' and capture written content + mock_file = mock_open() + with patch("builtins.open", mock_file): + its.to_csv("fake_path.csv", with_metadata=True) + + # Assert that open was called with correct parameters + mock_file.assert_called_once_with( + "fake_path.csv", "w", newline="", encoding="utf-8" + ) + + # Extract all written data + handle = mock_file() + written_data = "".join(call.args[0] for call in handle.write.call_args_list) + + # Assertions on the CSV content structure + self.assertIn("A,,,A", written_data) + self.assertIn("B,,,B", written_data) + self.assertIn("C,,,C", written_data) + self.assertIn("Units,,,CFS", written_data) + self.assertIn("Type,Date/Time,INST-VAL", written_data) + self.assertIn("1,01Sep2021 0000,10.5", written_data) + self.assertIn("2,02Sep2021 0000,20.0", written_data) + self.assertIn("3,04Sep2021 0000,42.0", written_data) + + def test_irregular_to_csv_without_metadata(self): + """No metadata rows should be written; only data rows.""" + its = IrregularTimeSeries.create( + values=[10.5, 20.0, 42.0], + times=[ + datetime(2021, 9, 1, 0, 0), + datetime(2021, 9, 2, 0, 0), + datetime(2021, 9, 4, 0, 0), + ], # inconsistent time interval + units="CFS", + data_type="INST-VAL", + path="/A/B/C/01Sep2021/E/F/", + ) + mock_file = mock_open() + with patch("builtins.open", mock_file): + its.to_csv("fake.csv", with_metadata=False) + handle = mock_file() + written = "".join(call.args[0] for call in handle.write.call_args_list) + self.assertNotIn("Units", written) + self.assertNotIn("Type,Date/Time", written) + self.assertIn("1,01Sep2021 0000,10.5", written) + self.assertIn("2,02Sep2021 0000,20.0", written) + self.assertIn("3,04Sep2021 0000,42.0", written) + + def test_irregular_to_csv_empty_times(self): + its = IrregularTimeSeries.create( + values=[], + times=[], + units="CFS", + data_type="INST-VAL", + path="/A/B/C//E/F/", + ) + + mock_file = mock_open() + with patch("builtins.open", mock_file): + its.to_csv("fake_path.csv", with_metadata=True) + + handle = mock_file() + written_data = "".join(call.args[0] for call in handle.write.call_args_list) + + self.assertIn("Units,,,CFS", written_data) + self.assertIn("Type,Date/Time,INST-VAL", written_data) + self.assertNotIn("1,", written_data) # No data rows should be present + + def test_irregular_to_csv_second_precision(self): + its = IrregularTimeSeries.create( + values=[i for i in range(3)], + times=[ + datetime(2021, 9, 1, 6, 0, 0), + datetime(2021, 9, 1, 6, 0, 1), + datetime(2021, 9, 1, 6, 0, 4), + ], + units="CFS", + data_type="INST-VAL", + path="/A/B/C//E/F/", + ) + + mock_file = mock_open() + with patch("builtins.open", mock_file): + its.to_csv("fake_path.csv", with_metadata=True) + + handle = mock_file() + written_data = "".join(call.args[0] for call in handle.write.call_args_list) + self.assertIn("Type,Date/Time,INST-VAL", written_data) + self.assertIn("1,01Sep2021 060000,0", written_data) + self.assertIn("2,01Sep2021 060001,1", written_data) + self.assertIn("3,01Sep2021 060004,2", written_data) + + def read_its_from_string(self, content): + """Helper to run read_csv against an in-memory CSV string.""" + m = mock_open(read_data=content) + with patch("builtins.open", m): + return IrregularTimeSeries.read_csv("fake.csv") + + def test_irregular_read_csv_basic(self): + content = ( + "A,,,A\n" + "B,,,B\n" + "C,,,FLOW\n" + "E,,,IR-Year\n" + "F,,,F\n" + "Units,,,CFS\n" + "Type,Date/Time,INST-VAL\n" + "1,01Sep2021 0000,10.5\n" + "2,02Sep2021 0000,20.0\n" + "3,04Sep2021 0000,20.0\n" + ) + its = self.read_its_from_string(content) + self.assertEqual(its.units, "CFS") + self.assertEqual(its.data_type, "INST-VAL") + self.assertEqual(its.values.tolist(), [10.5, 20.0, 20.0]) + self.assertEqual( + its.times, + [datetime(2021, 9, 1), datetime(2021, 9, 2), datetime(2021, 9, 4)], + ) + + # EDGE CASE TESTS: + + def test_roll_day_edge_case(self): + content = ( + "A,,,A\n" + "B,,,B\n" + "C,,,FLOW\n" + "E,,,IR-Day\n" + "F,,,F\n" + "Units,,,CFS\n" + "Type,Date/Time,INST-VAL\n" + "1,01Sep2021 002400,1\n" # 12:24 AM + "2,02Sep2021 024000,1\n" # 2:40 AM + "3,03Sep2021 240000,1\n" # 12:00 AM Next day + ) + its = self.read_its_from_string(content) + self.assertEqual(its.units, "CFS") + self.assertEqual(its.data_type, "INST-VAL") + self.assertEqual(its.values.tolist(), [1, 1, 1]) + self.assertEqual(its.times, [datetime(2021, 9, 1, 0, 24, 0), datetime( + 2021, 9, 2, 2, 40, 0), datetime(2021, 9, 4, 0, 0, 0)]) + + def test_year_is_2400(self): + content = ( + "A,,,A\n" + "B,,,B\n" + "C,,,FLOW\n" + "E,,,IR-Day\n" + "F,,,F\n" + "Units,,,CFS\n" + "Type,Date/Time,INST-VAL\n" + "1,01Sep2400 000000,1\n" + "2,01Sep2400 240000,1\n" + "3,01Sep2400 240015,1\n" # Should not be accepted + "4,01Sep2400 24000,1\n" # Should not be accepted + ) + its = self.read_its_from_string(content) + self.assertEqual(its.units, "CFS") + self.assertEqual(its.data_type, "INST-VAL") + self.assertEqual(its.values.tolist(), [1, 1]) + self.assertEqual(its.times, [datetime(2400, 9, 1, 0, 0, 0), datetime(2400, 9, 2, 0, 0, 0)]) + + +if __name__ == "__main__": + unittest.main()