1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
|
# Copyright (c) DataLab Platform Developers, BSD 3-Clause license, see LICENSE file.
"""
DateTime CSV I/O Unit Test
==========================
Unit tests for reading CSV files with datetime columns.
"""
# pylint: disable=invalid-name # Allows short reference names like x, y, ...
from __future__ import annotations
import os
import os.path as osp
import tempfile
from datetime import datetime, timedelta
from pathlib import Path
import numpy as np
import pandas as pd
import pytest
from sigima.io import read_signal, read_signals, write_signal
from sigima.io.signal.formats import CSVSignalFormat
from sigima.objects import SignalObj, create_signal
from sigima.tests.env import execenv
from sigima.tests.helpers import get_test_fnames
def test_datetime_csv_io() -> None:
"""Test reading CSV file with datetime X column."""
execenv.print("Testing datetime CSV I/O...")
# Get path to the datetime test file
filenames = get_test_fnames("datetime.txt", in_folder="curve_formats")
assert len(filenames) > 0, "datetime.txt test file not found"
filename = filenames[0]
assert osp.exists(filename), f"Test file not found: {filename}"
# Read signals from file (read_signals returns a list, read_signal returns first)
signals = read_signals(filename)
# Should create multiple signals (one per Y column)
assert len(signals) > 0, "No signals were read from datetime.txt"
execenv.print(f" Read {len(signals)} signals from file")
# Test first signal (Temperature)
signal = signals[0]
execenv.print(f" First signal: {signal.title}")
# Check that datetime metadata was detected
assert signal.metadata.get("x_datetime", False), (
"DateTime metadata not detected in X column"
)
assert signal.xunit == "s", "DateTime unit should be 's' (seconds)"
# Check X data is float (timestamps)
assert isinstance(signal.x, np.ndarray), "X data should be numpy array"
assert signal.x.dtype in (np.float32, np.float64), "X data should be float"
execenv.print(f" X data type: {signal.x.dtype}")
execenv.print(f" X data shape: {signal.x.shape}")
execenv.print(f" First 5 X values: {signal.x[:5]}")
# Remove NaN values before checking monotonicity
x_clean = signal.x[~np.isnan(signal.x)]
execenv.print(f" Clean X shape (no NaNs): {x_clean.shape}")
execenv.print(f" First clean X value (timestamp): {x_clean[0]:.2f}")
# Check X values are monotonically increasing
assert np.all(np.diff(x_clean) >= 0), "X values should be monotonic"
# Check Y data exists and has correct length
assert isinstance(signal.y, np.ndarray), "Y data should be numpy array"
assert len(signal.x) == len(signal.y), "X and Y should have same length"
execenv.print(f" Y data shape: {signal.y.shape}")
execenv.print(f" First Y value: {signal.y[0]}")
# Test datetime conversion back
dt_values = signal.get_x_as_datetime()
assert dt_values is not None, "Should be able to get datetime values"
assert len(dt_values) == len(signal.x), (
"Datetime array should have same length as X"
)
execenv.print(f" First datetime value: {dt_values[0]}")
# Check that the datetime is reasonable (should be 2025-06-19 10:00:00)
# Convert to string to check
dt_str = pd.to_datetime(dt_values[0]).strftime("%Y-%m-%d %H:%M:%S")
expected_start = "2025-06-19 10:00:00"
assert dt_str == expected_start, (
f"Expected first datetime to be {expected_start}, got {dt_str}"
)
# Check labels were extracted correctly
assert signal.xlabel, "X label should be set"
assert signal.ylabel, "Y label should be set"
execenv.print(f" X label: {signal.xlabel}")
execenv.print(f" Y label: {signal.ylabel}")
# Check we have multiple signals (Temperature, Humidity, Dew Point)
assert len(signals) >= 3, "Should have at least 3 signals"
signal_titles = [s.ylabel for s in signals]
execenv.print(f" Signal Y labels: {signal_titles}")
# All signals should have the same datetime metadata
for sig in signals:
assert sig.is_x_datetime(), "All signals should have datetime X"
assert sig.xunit == "s"
# Check that all signals have the same X data (timestamps)
for sig in signals[1:]:
assert np.array_equal(sig.x, signals[0].x), (
"All signals should share the same X data"
)
execenv.print(" ✓ DateTime CSV I/O test passed")
def test_datetime_csv_write_with_datetime(tmp_path: Path):
"""Test writing CSV file with datetime X values"""
# Create signal with datetime X
timestamps = [
datetime(2025, 1, 1, 10, 0, 0),
datetime(2025, 1, 1, 10, 0, 1),
datetime(2025, 1, 1, 10, 0, 2),
]
signal = SignalObj()
signal.set_x_from_datetime(timestamps, unit="s")
signal.y = np.array([1.0, 2.0, 3.0])
signal.ylabel = "Temperature"
signal.xlabel = "Time"
# Write to CSV
csv_file = tmp_path / "datetime_signal.csv"
fmt = CSVSignalFormat()
fmt.write(str(csv_file), signal)
# Read file back and check contents
with open(csv_file, "r", encoding="utf-8") as f:
lines = f.readlines()
# Should have header + 3 data lines
assert len(lines) == 4
# Header should be "Time,Temperature"
assert "Time" in lines[0]
assert "Temperature" in lines[0]
# First data line should contain datetime string
assert "2025-01-01" in lines[1]
assert "10:00:00" in lines[1]
def test_datetime_csv_roundtrip() -> None:
"""Test that datetime signals can be written and read back."""
execenv.print("Testing datetime CSV roundtrip...")
# Create a signal with datetime X
base_time = datetime(2025, 10, 6, 10, 0, 0)
timestamps = [base_time + timedelta(minutes=i * 5) for i in range(20)]
values = 20 + np.random.randn(20) * 2
signal = create_signal("Test Temperature")
signal.set_x_from_datetime(timestamps, unit="s")
signal.y = values
signal.ylabel = "Temperature"
signal.yunit = "°C"
# Write to temporary file
with tempfile.NamedTemporaryFile(
mode="w", suffix=".csv", delete=False, newline=""
) as tmp:
tmp_path = tmp.name
try:
write_signal(tmp_path, signal)
execenv.print(f" Wrote signal to: {tmp_path}")
# Read it back
signal_read = read_signal(tmp_path)
execenv.print(" Signal read back successfully")
# Check datetime metadata is preserved
assert signal_read.is_x_datetime(), "DateTime metadata should be preserved"
# Check Y values match (X will be timestamps now, not exact datetime strings)
assert len(signal_read.y) == len(values), "Y length should match"
assert np.allclose(signal_read.y, values, atol=0.01), "Y values should match"
execenv.print(" ✓ DateTime CSV roundtrip test passed")
finally:
# Clean up temporary file
if osp.exists(tmp_path):
os.unlink(tmp_path)
if __name__ == "__main__":
pytest.main([__file__, "-v"])
|
