Skip to content

test_roundtrip

Roundtrip tests for writing and reading various data types.

test_complex128_roundtrip 

test_complex128_roundtrip(tmp_path)

Test complex128 (FR_VECT_16C) write and read.

Source code in gwframe/tests/test_roundtrip.py 
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
def test_complex128_roundtrip(tmp_path):
    """Test complex128 (FR_VECT_16C) write and read."""
    t0 = 1234567890.0
    duration = 1.0
    sample_rate = 16384.0
    n_samples = int(duration * sample_rate)

    tmp_file = tmp_path / "test_c128.gwf"

    # Create complex128 data
    data_c128 = np.exp(1j * 2 * np.pi * np.linspace(0, 1, n_samples)).astype(
        np.complex128
    )

    # Write frame
    frame = gwframe.Frame(t0=t0, duration=duration, name="L1", run=1)
    frame.add_channel(
        "L1:TEST_C128",
        data_c128,
        sample_rate=sample_rate,
        unit="",
        channel_type="proc",
    )
    frame.write(str(tmp_file))

    # Read back
    ts = gwframe.read(str(tmp_file), "L1:TEST_C128")

    assert ts.array.dtype == np.complex128
    assert np.allclose(ts.array, data_c128)

test_complex64_roundtrip 

test_complex64_roundtrip(tmp_path)

Test complex64 (FR_VECT_8C) write and read.

Source code in gwframe/tests/test_roundtrip.py 
 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
def test_complex64_roundtrip(tmp_path):
    """Test complex64 (FR_VECT_8C) write and read."""
    t0 = 1234567890.0
    duration = 1.0
    sample_rate = 16384.0
    n_samples = int(duration * sample_rate)

    tmp_file = tmp_path / "test_c64.gwf"

    # Create complex64 data
    data_c64 = np.exp(1j * 2 * np.pi * np.linspace(0, 1, n_samples)).astype(
        np.complex64
    )

    # Write frame
    frame = gwframe.Frame(t0=t0, duration=duration, name="L1", run=1)
    frame.add_channel(
        "L1:TEST_C64", data_c64, sample_rate=sample_rate, unit="", channel_type="proc"
    )
    frame.write(str(tmp_file))

    # Read back
    ts = gwframe.read(str(tmp_file), "L1:TEST_C64")

    assert ts.array.dtype == np.complex64
    assert np.allclose(ts.array, data_c64)

test_float32_roundtrip 

test_float32_roundtrip(tmp_path)

Test writing and reading float32 data.

Source code in gwframe/tests/test_roundtrip.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def test_float32_roundtrip(tmp_path):
    """Test writing and reading float32 data."""
    t0 = 1234567890.0
    sample_rate = 1000.0
    n_samples = 1000

    tmp_file = tmp_path / "test_float32.gwf"

    data_f32 = np.random.randn(n_samples).astype(np.float32)

    gwframe.write(
        str(tmp_file),
        data_f32,
        t0=t0,
        sample_rate=sample_rate,
        name="L1:FLOAT32",
    )

    # Read back
    read_data = gwframe.read(str(tmp_file), "L1:FLOAT32")
    assert read_data.array.dtype == np.float32
    np.testing.assert_array_almost_equal(read_data.array, data_f32)

test_integer_dtypes_roundtrip 

test_integer_dtypes_roundtrip(tmp_path)

Test writing and reading various integer data types.

Source code in gwframe/tests/test_roundtrip.py 
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
def test_integer_dtypes_roundtrip(tmp_path):
    """Test writing and reading various integer data types."""
    t0 = 1234567890.0
    sample_rate = 1000.0
    n_samples = 1000

    tmp_file = tmp_path / "test_int_dtypes.gwf"

    # Create data with different integer dtypes
    channels = {
        "L1:INT16": np.random.randint(-100, 100, n_samples, dtype=np.int16),
        "L1:INT32": np.random.randint(-1000, 1000, n_samples, dtype=np.int32),
        "L1:INT64": np.random.randint(-10000, 10000, n_samples, dtype=np.int64),
        "L1:UINT16": np.random.randint(0, 200, n_samples, dtype=np.uint16),
        "L1:UINT32": np.random.randint(0, 2000, n_samples, dtype=np.uint32),
    }

    # Write all channels
    frame = gwframe.Frame(t0=t0, duration=n_samples / sample_rate, name="L1")
    for ch_name, data in channels.items():
        frame.add_channel(ch_name, data, sample_rate=sample_rate)
    frame.write(str(tmp_file))

    # Read back and verify dtypes are preserved
    for ch_name, original_data in channels.items():
        read_data = gwframe.read(str(tmp_file), ch_name)
        assert read_data.array.dtype == original_data.dtype
        assert len(read_data.array) == len(original_data)
        # For integer types, values should match exactly
        np.testing.assert_array_equal(read_data.array, original_data)

test_mixed_proc_and_sim_roundtrip 

test_mixed_proc_and_sim_roundtrip(tmp_path)

Test file with both processed and simulated channels.

Source code in gwframe/tests/test_roundtrip.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def test_mixed_proc_and_sim_roundtrip(tmp_path):
    """Test file with both processed and simulated channels."""
    tmp_file = tmp_path / "mixed.gwf"
    frame = gwframe.Frame(t0=1234567890.0, duration=1.0, name="L1")

    # Add proc channel
    proc_data = np.random.randn(1000)
    frame.add_channel("L1:PROC_CHAN", proc_data, sample_rate=1000, channel_type="proc")

    # Add sim channel
    sim_data = np.random.randn(1000)
    frame.add_channel("L1:SIM_CHAN", sim_data, sample_rate=1000, channel_type="sim")

    frame.write(str(tmp_file))

    # Read back both channels by name
    proc_result = gwframe.read(str(tmp_file), "L1:PROC_CHAN")
    sim_result = gwframe.read(str(tmp_file), "L1:SIM_CHAN")

    assert proc_result.type == "proc"
    assert sim_result.type == "sim"
    assert len(proc_result.array) == len(proc_data)
    assert len(sim_result.array) == len(sim_data)

test_mixed_real_complex_roundtrip 

test_mixed_real_complex_roundtrip(tmp_path)

Test mixed real and complex data in same frame.

Source code in gwframe/tests/test_roundtrip.py 
 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
def test_mixed_real_complex_roundtrip(tmp_path):
    """Test mixed real and complex data in same frame."""
    t0 = 1234567890.0
    duration = 1.0
    sample_rate = 16384.0
    n_samples = int(duration * sample_rate)

    tmp_file = tmp_path / "test_mixed.gwf"

    # Create mixed data
    data_real = np.random.randn(n_samples)
    data_complex = (
        np.random.randn(n_samples) + 1j * np.random.randn(n_samples)
    ).astype(np.complex128)

    # Write frame with both types
    frame = gwframe.Frame(t0=t0, duration=duration, name="L1", run=1)
    frame.add_channel(
        "L1:REAL",
        data_real,
        sample_rate=sample_rate,
        unit="strain",
        channel_type="proc",
    )
    frame.add_channel(
        "L1:COMPLEX",
        data_complex,
        sample_rate=sample_rate,
        unit="",
        channel_type="proc",
    )
    frame.write(str(tmp_file))

    # Read back both channels
    ts_real = gwframe.read(str(tmp_file), "L1:REAL")
    ts_complex = gwframe.read(str(tmp_file), "L1:COMPLEX")

    assert ts_real.array.dtype == np.float64
    assert ts_complex.array.dtype == np.complex128
    assert np.allclose(ts_real.array, data_real)
    assert np.allclose(ts_complex.array, data_complex)

test_simulated_channel_roundtrip 

test_simulated_channel_roundtrip(tmp_path)

Test writing and reading simulated channels.

Source code in gwframe/tests/test_roundtrip.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def test_simulated_channel_roundtrip(tmp_path):
    """Test writing and reading simulated channels."""
    tmp_file = tmp_path / "sim_data.gwf"
    data = np.random.randn(1000)

    gwframe.write(
        str(tmp_file),
        data,
        t0=1234567890.0,
        sample_rate=1000,
        name="L1:SIM_STRAIN",
        channel_type="sim",
    )

    # Read back and verify
    read_data = gwframe.read(str(tmp_file), "L1:SIM_STRAIN")
    assert read_data.type == "sim"
    assert len(read_data.array) == len(data)
    np.testing.assert_array_almost_equal(read_data.array, data)