Skip to content

Commit 687a3e9

Browse files
wei-qluwei-qlu
committed
tdms detect_config tests
1 parent f684288 commit 687a3e9

1 file changed

Lines changed: 376 additions & 0 deletions

File tree

‎python/lib/sift_client/_tests/_internal/test_tdms.py‎

Lines changed: 376 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,376 @@
1+
"""Tests for detect_config (TDMS)."""
2+
3+
import numpy as np
4+
import pytest
5+
from nptdms import ChannelObject, GroupObject, RootObject, TdmsWriter
6+
7+
from sift_client._internal.util.tdms import detect_config
8+
from sift_client.sift_types.channel import ChannelDataType
9+
from sift_client.sift_types.data_import import TdmsComplexComponent, TdmsFallbackMethod
10+
11+
12+
@pytest.fixture
13+
def create_tdms_file(tmp_path):
14+
"""Return a helper that writes a TDMS file and returns its path."""
15+
file_path = tmp_path / "test.tdms"
16+
17+
def _create(root_props=None, groups=None):
18+
"""Write a TDMS file.
19+
20+
Args:
21+
root_props: dict of root-level file properties.
22+
groups: list of (group_name, channels) tuples where channels is a list of
23+
ChannelObject instances.
24+
"""
25+
root = RootObject(properties=root_props or {})
26+
with TdmsWriter(file_path) as writer:
27+
for group_name, channels in groups or []:
28+
group = GroupObject(group_name)
29+
writer.write_segment([root, group, *channels])
30+
return file_path
31+
32+
return _create
33+
34+
35+
class TestDetectConfig:
36+
def test_waveform_channels(self, create_tdms_file):
37+
"""Channels with wf_start_offset and wf_increment are detected as waveform channels."""
38+
path = create_tdms_file(
39+
groups=[
40+
(
41+
"sensors",
42+
[
43+
ChannelObject(
44+
"sensors",
45+
"voltage",
46+
np.array([1.0, 2.0, 3.0], dtype="float64"),
47+
properties={
48+
"wf_start_offset": 0.0,
49+
"wf_increment": 0.001,
50+
"wf_start_time": np.datetime64("2024-01-01T00:00:00"),
51+
},
52+
),
53+
],
54+
)
55+
]
56+
)
57+
58+
config = detect_config(path)
59+
60+
assert len(config.data) == 1
61+
assert config.data[0].name == "sensors.voltage"
62+
assert config.data[0].data_type == ChannelDataType.DOUBLE
63+
assert config.data[0].time_channel_name is None
64+
assert config.data[0].group_name == "sensors"
65+
assert config.data[0].channel_name == "voltage"
66+
67+
def test_time_channel_detection(self, create_tdms_file):
68+
"""A channel with TimeStamp type is used as the time source and excluded from data."""
69+
path = create_tdms_file(
70+
groups=[
71+
(
72+
"group1",
73+
[
74+
ChannelObject(
75+
"group1",
76+
"timestamp",
77+
np.array(
78+
["2024-01-01", "2024-01-02"],
79+
dtype="datetime64[ns]",
80+
),
81+
),
82+
ChannelObject(
83+
"group1",
84+
"temperature",
85+
np.array([20.5, 21.0], dtype="float32"),
86+
),
87+
],
88+
)
89+
]
90+
)
91+
92+
config = detect_config(path)
93+
94+
channel_names = [d.name for d in config.data]
95+
assert "group1.timestamp" not in channel_names
96+
assert "group1.temperature" in channel_names
97+
assert config.data[0].time_channel_name == "timestamp"
98+
assert config.data[0].data_type == ChannelDataType.FLOAT
99+
100+
def test_common_time_name_detection(self, create_tdms_file):
101+
"""Channels named 'time', 'Time', etc. are detected as time channels."""
102+
path = create_tdms_file(
103+
groups=[
104+
(
105+
"data",
106+
[
107+
ChannelObject(
108+
"data",
109+
"time",
110+
np.array([0.0, 0.1, 0.2], dtype="float64"),
111+
),
112+
ChannelObject(
113+
"data",
114+
"pressure",
115+
np.array([101.3, 101.4, 101.5], dtype="float64"),
116+
),
117+
],
118+
)
119+
]
120+
)
121+
122+
config = detect_config(path)
123+
124+
channel_names = [d.name for d in config.data]
125+
assert "data.time" not in channel_names
126+
assert "data.pressure" in channel_names
127+
assert config.data[0].time_channel_name == "time"
128+
129+
def test_complex_channels_split(self, create_tdms_file):
130+
"""Complex-valued channels are split into .real and .imag entries."""
131+
path = create_tdms_file(
132+
groups=[
133+
(
134+
"rf",
135+
[
136+
ChannelObject(
137+
"rf",
138+
"signal",
139+
np.array([1 + 2j, 3 + 4j], dtype="complex128"),
140+
properties={
141+
"wf_start_offset": 0.0,
142+
"wf_increment": 0.001,
143+
"wf_start_time": np.datetime64("2024-01-01T00:00:00"),
144+
},
145+
),
146+
],
147+
)
148+
]
149+
)
150+
151+
config = detect_config(path)
152+
153+
assert len(config.data) == 2
154+
names = [d.name for d in config.data]
155+
assert "rf.signal.real" in names
156+
assert "rf.signal.imag" in names
157+
158+
real_col = next(d for d in config.data if d.name == "rf.signal.real")
159+
imag_col = next(d for d in config.data if d.name == "rf.signal.imag")
160+
assert real_col.complex_component == TdmsComplexComponent.REAL
161+
assert imag_col.complex_component == TdmsComplexComponent.IMAGINARY
162+
assert real_col.data_type == ChannelDataType.DOUBLE
163+
assert imag_col.data_type == ChannelDataType.DOUBLE
164+
165+
def test_unit_and_description_detection(self, create_tdms_file):
166+
"""Units and descriptions are read from TDMS channel properties."""
167+
path = create_tdms_file(
168+
groups=[
169+
(
170+
"sensors",
171+
[
172+
ChannelObject(
173+
"sensors",
174+
"voltage",
175+
np.array([1.0, 2.0], dtype="float64"),
176+
properties={
177+
"wf_start_offset": 0.0,
178+
"wf_increment": 0.001,
179+
"wf_start_time": np.datetime64("2024-01-01T00:00:00"),
180+
"unit_string": "V",
181+
"description": "Supply voltage",
182+
},
183+
),
184+
],
185+
)
186+
]
187+
)
188+
189+
config = detect_config(path)
190+
191+
assert config.data[0].units == "V"
192+
assert config.data[0].description == "Supply voltage"
193+
194+
def test_fallback_fail_on_error(self, create_tdms_file):
195+
"""Channels without timing info raise ValueError when fallback is FAIL_ON_ERROR."""
196+
path = create_tdms_file(
197+
groups=[
198+
(
199+
"data",
200+
[
201+
ChannelObject(
202+
"data",
203+
"orphan",
204+
np.array([1.0, 2.0], dtype="float64"),
205+
),
206+
],
207+
)
208+
]
209+
)
210+
211+
with pytest.raises(ValueError, match="No timing information"):
212+
detect_config(path, fallback_method=TdmsFallbackMethod.FAIL_ON_ERROR)
213+
214+
def test_fallback_ignore_error(self, create_tdms_file):
215+
"""Channels without timing info are silently skipped when fallback is IGNORE_ERROR."""
216+
path = create_tdms_file(
217+
groups=[
218+
(
219+
"data",
220+
[
221+
ChannelObject(
222+
"data",
223+
"orphan",
224+
np.array([1.0, 2.0], dtype="float64"),
225+
),
226+
],
227+
)
228+
]
229+
)
230+
231+
config = detect_config(path, fallback_method=TdmsFallbackMethod.IGNORE_ERROR)
232+
233+
assert len(config.data) == 0
234+
assert config.fallback_method == TdmsFallbackMethod.IGNORE_ERROR
235+
236+
def test_multiple_groups(self, create_tdms_file):
237+
"""Channels from multiple groups are all detected with correct group_name."""
238+
path = create_tdms_file(
239+
groups=[
240+
(
241+
"group_a",
242+
[
243+
ChannelObject(
244+
"group_a",
245+
"ch1",
246+
np.array([1.0, 2.0], dtype="float64"),
247+
properties={
248+
"wf_start_offset": 0.0,
249+
"wf_increment": 0.001,
250+
"wf_start_time": np.datetime64("2024-01-01T00:00:00"),
251+
},
252+
),
253+
],
254+
),
255+
(
256+
"group_b",
257+
[
258+
ChannelObject(
259+
"group_b",
260+
"ch2",
261+
np.array([3, 4], dtype="int32"),
262+
properties={
263+
"wf_start_offset": 0.0,
264+
"wf_increment": 0.001,
265+
"wf_start_time": np.datetime64("2024-01-01T00:00:00"),
266+
},
267+
),
268+
],
269+
),
270+
]
271+
)
272+
273+
config = detect_config(path)
274+
275+
assert len(config.data) == 2
276+
assert config.data[0].group_name == "group_a"
277+
assert config.data[0].name == "group_a.ch1"
278+
assert config.data[0].data_type == ChannelDataType.DOUBLE
279+
assert config.data[1].group_name == "group_b"
280+
assert config.data[1].name == "group_b.ch2"
281+
assert config.data[1].data_type == ChannelDataType.INT_32
282+
283+
def test_enum_channel_detection(self, create_tdms_file):
284+
"""Channels with enum_config property are detected as ENUM type with enum_types populated."""
285+
import json
286+
287+
enum_config = json.dumps({"0": "Off", "1": "On", "2": "Error"})
288+
path = create_tdms_file(
289+
groups=[
290+
(
291+
"status",
292+
[
293+
ChannelObject(
294+
"status",
295+
"state",
296+
np.array([0, 1, 2], dtype="uint32"),
297+
properties={
298+
"wf_start_offset": 0.0,
299+
"wf_increment": 1.0,
300+
"wf_start_time": np.datetime64("2024-01-01T00:00:00"),
301+
"enum_config": enum_config,
302+
},
303+
),
304+
],
305+
)
306+
]
307+
)
308+
309+
config = detect_config(path)
310+
311+
assert len(config.data) == 1
312+
assert config.data[0].data_type == ChannelDataType.ENUM
313+
assert config.data[0].enum_types == {"Off": 0, "On": 1, "Error": 2}
314+
315+
def test_asset_name_passthrough(self, create_tdms_file):
316+
"""The asset_name parameter is set on the returned config."""
317+
path = create_tdms_file(
318+
groups=[
319+
(
320+
"g",
321+
[
322+
ChannelObject(
323+
"g",
324+
"ch",
325+
np.array([1.0], dtype="float64"),
326+
properties={
327+
"wf_start_offset": 0.0,
328+
"wf_increment": 0.001,
329+
"wf_start_time": np.datetime64("2024-01-01T00:00:00"),
330+
},
331+
),
332+
],
333+
)
334+
]
335+
)
336+
337+
config = detect_config(path, asset_name="my-asset")
338+
339+
assert config.asset_name == "my-asset"
340+
341+
def test_xchannel_property(self, create_tdms_file):
342+
"""Group-level 'xchannel' property overrides time channel detection."""
343+
path = create_tdms_file(
344+
groups=[
345+
(
346+
"data",
347+
[
348+
ChannelObject(
349+
"data",
350+
"custom_time",
351+
np.array([0.0, 1.0, 2.0], dtype="float64"),
352+
),
353+
ChannelObject(
354+
"data",
355+
"value",
356+
np.array([10.0, 20.0, 30.0], dtype="float64"),
357+
),
358+
],
359+
)
360+
]
361+
)
362+
363+
# nptdms TdmsWriter doesn't support group-level properties directly in segments,
364+
# so we write the file and then patch the group property by re-reading/writing.
365+
# Instead, test via the find_time_channel helper.
366+
from nptdms import TdmsFile
367+
368+
from sift_client._internal.util.tdms import find_time_channel
369+
370+
with TdmsFile.open(path) as tdms_file:
371+
group = tdms_file["data"]
372+
# Simulate xchannel property
373+
group.properties["xchannel"] = "custom_time"
374+
result = find_time_channel(group)
375+
376+
assert result == "custom_time"

0 commit comments

Comments
 (0)