Tests for masked array support (channel-level dataValid on ADC channels).
Tests cover channel-level FrAdcData.dataValid (frame spec v3-v8, ADC only)
and on_mask_loss behavior for v8 frames.
TestAdcChannelLevelMasking
ADC channels support channel-level dataValid on frame spec v3-v8.
test_all_valid_mask_no_flag
test_all_valid_mask_no_flag()
All-False mask should not set dataValid (no mask loss).
Source code in gwframe/tests/test_masked.py
| def test_all_valid_mask_no_flag(self):
"""All-False mask should not set dataValid (no mask loss)."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.zeros(N_SAMPLES, dtype=bool)
gwf_bytes = _write_adc_masked(data, mask)
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", allow_invalid=True)
assert ts.mask is None
|
test_error_message_content
test_error_message_content()
InvalidDataError message includes channel name and suggestion.
Source code in gwframe/tests/test_masked.py
| def test_error_message_content(self):
"""InvalidDataError message includes channel name and suggestion."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.ones(N_SAMPLES, dtype=bool)
gwf_bytes = _write_adc_masked(data, mask)
with pytest.raises(InvalidDataError) as exc_info:
gwframe.read_bytes(gwf_bytes, "H1:TEST")
msg = str(exc_info.value)
assert "H1:TEST" in msg
assert "allow_invalid=True" in msg
assert "100.0%" in msg
|
test_float32_roundtrip
Verify float32 dtype works with masked ADC.
Source code in gwframe/tests/test_masked.py
| def test_float32_roundtrip(self):
"""Verify float32 dtype works with masked ADC."""
data = np.random.randn(N_SAMPLES).astype(np.float32)
mask = np.ones(N_SAMPLES, dtype=bool)
gwf_bytes = _write_adc_masked(data, mask)
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", allow_invalid=True)
assert ts.mask is not None
np.testing.assert_array_equal(ts.array, data)
|
test_masked_roundtrip
Write masked ADC, read back with allow_invalid=True.
Source code in gwframe/tests/test_masked.py
| def test_masked_roundtrip(self):
"""Write masked ADC, read back with allow_invalid=True."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.zeros(N_SAMPLES, dtype=bool)
mask[100:200] = True
gwf_bytes = _write_adc_masked(data, mask, on_mask_loss="ignore")
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", allow_invalid=True)
assert ts.mask is not None
# Channel-level masking: entire channel is flagged invalid
assert ts.mask.all()
# Underlying data values are preserved
np.testing.assert_array_equal(ts.array, data)
|
test_multi_channel_mixed
test_multi_channel_mixed()
Frame with masked ADC + plain proc reads correctly.
Source code in gwframe/tests/test_masked.py
| def test_multi_channel_mixed(self):
"""Frame with masked ADC + plain proc reads correctly."""
adc_data = np.random.randn(N_SAMPLES).astype(np.float64)
proc_data = np.random.randn(N_SAMPLES).astype(np.float64)
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
masked_adc = np.ma.MaskedArray(adc_data, mask=np.ones(N_SAMPLES, dtype=bool))
frame.add_channel(
"H1:ADC", masked_adc, sample_rate=SAMPLE_RATE, channel_type="adc"
)
frame.add_channel(
"H1:PROC", proc_data, sample_rate=SAMPLE_RATE, channel_type="proc"
)
gwf_bytes = frame.write_bytes()
result = gwframe.read_bytes(gwf_bytes, channels=None, allow_invalid=True)
assert result["H1:ADC"].mask is not None
assert result["H1:ADC"].mask.all()
assert result["H1:PROC"].mask is None
np.testing.assert_array_equal(result["H1:PROC"].array, proc_data)
|
test_plain_array_unaffected
test_plain_array_unaffected()
Plain arrays return plain arrays even with allow_invalid=True.
Source code in gwframe/tests/test_masked.py
| def test_plain_array_unaffected(self):
"""Plain arrays return plain arrays even with allow_invalid=True."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
frame.add_channel("H1:TEST", data, sample_rate=SAMPLE_RATE, channel_type="adc")
gwf_bytes = frame.write_bytes()
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", allow_invalid=True)
assert ts.mask is None
np.testing.assert_array_equal(ts.array, data)
|
test_raises_by_default
InvalidDataError raised when dataValid != 0.
Source code in gwframe/tests/test_masked.py
| def test_raises_by_default(self):
"""InvalidDataError raised when dataValid != 0."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.ones(N_SAMPLES, dtype=bool)
gwf_bytes = _write_adc_masked(data, mask)
with pytest.raises(InvalidDataError) as exc_info:
gwframe.read_bytes(gwf_bytes, "H1:TEST")
assert exc_info.value.channel == "H1:TEST"
assert exc_info.value.n_invalid == N_SAMPLES
assert exc_info.value.n_total == N_SAMPLES
|
TestDefaultFrameSpec
Verify default frame spec behavior.
test_default_frame_spec_is_v8
test_default_frame_spec_is_v8()
Frame with no explicit frame_spec defaults to v8 behavior.
Source code in gwframe/tests/test_masked.py
| def test_default_frame_spec_is_v8(self):
"""Frame with no explicit frame_spec defaults to v8 behavior."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
masked = np.ma.MaskedArray(data, mask=np.ones(N_SAMPLES, dtype=bool))
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
# Default v8: proc should warn about mask loss
with pytest.warns(UserWarning, match="masking is not supported for proc"):
frame.add_channel(
"H1:TEST", masked, sample_rate=SAMPLE_RATE, channel_type="proc"
)
|
TestOnMaskLoss
Test on_mask_loss behavior for proc/sim/adc channels.
test_adc_all_masked_no_warn
test_adc_all_masked_no_warn()
ADC with all-True mask doesn't warn (no precision loss).
Source code in gwframe/tests/test_masked.py
| def test_adc_all_masked_no_warn(self):
"""ADC with all-True mask doesn't warn (no precision loss)."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.ones(N_SAMPLES, dtype=bool)
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
import warnings
with warnings.catch_warnings():
warnings.simplefilter("error")
frame.add_channel(
"H1:TEST",
data=np.ma.MaskedArray(data, mask=mask),
sample_rate=SAMPLE_RATE,
channel_type="adc",
)
|
test_adc_mixed_mask_warns
test_adc_mixed_mask_warns()
ADC with mixed mask warns about per-channel coarsening.
Source code in gwframe/tests/test_masked.py
| def test_adc_mixed_mask_warns(self):
"""ADC with mixed mask warns about per-channel coarsening."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.zeros(N_SAMPLES, dtype=bool)
mask[100:200] = True # partial mask
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
with pytest.warns(UserWarning, match="per-sample mask.*per-channel masking"):
frame.add_channel(
"H1:TEST",
data=np.ma.MaskedArray(data, mask=mask),
sample_rate=SAMPLE_RATE,
channel_type="adc",
)
|
test_proc_masked_ignore
test_proc_masked_ignore()
Proc channel with masked array is silent with on_mask_loss='ignore'.
Source code in gwframe/tests/test_masked.py
| def test_proc_masked_ignore(self):
"""Proc channel with masked array is silent with on_mask_loss='ignore'."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
masked = np.ma.MaskedArray(data, mask=np.ones(N_SAMPLES, dtype=bool))
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
# Should not warn
import warnings
with warnings.catch_warnings():
warnings.simplefilter("error")
frame.add_channel(
"H1:TEST",
masked,
sample_rate=SAMPLE_RATE,
channel_type="proc",
on_mask_loss=OnMaskLoss.IGNORE,
)
|
test_proc_masked_raises
test_proc_masked_raises()
Proc channel with masked array raises with on_mask_loss='raise'.
Source code in gwframe/tests/test_masked.py
| def test_proc_masked_raises(self):
"""Proc channel with masked array raises with on_mask_loss='raise'."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
masked = np.ma.MaskedArray(data, mask=np.ones(N_SAMPLES, dtype=bool))
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
with pytest.raises(ValueError, match="masking is not supported for proc"):
frame.add_channel(
"H1:TEST",
masked,
sample_rate=SAMPLE_RATE,
channel_type="proc",
on_mask_loss="raise",
)
|
test_proc_masked_warns
Proc channel with masked array emits warning by default.
Source code in gwframe/tests/test_masked.py
| def test_proc_masked_warns(self):
"""Proc channel with masked array emits warning by default."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
masked = np.ma.MaskedArray(data, mask=np.ones(N_SAMPLES, dtype=bool))
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
with pytest.warns(UserWarning, match="masking is not supported for proc"):
frame.add_channel(
"H1:TEST", masked, sample_rate=SAMPLE_RATE, channel_type="proc"
)
gwf_bytes = frame.write_bytes()
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST")
assert ts.mask is None
|
test_sim_masked_warns
Sim channel with masked array emits warning.
Source code in gwframe/tests/test_masked.py
| def test_sim_masked_warns(self):
"""Sim channel with masked array emits warning."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
masked = np.ma.MaskedArray(data, mask=np.ones(N_SAMPLES, dtype=bool))
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
with pytest.warns(UserWarning, match="masking is not supported for sim"):
frame.add_channel(
"H1:TEST", masked, sample_rate=SAMPLE_RATE, channel_type="sim"
)
|
TestToMasked
Tests for TimeSeries.to_masked() conversion method.
test_returns_masked_array_when_mask_present
test_returns_masked_array_when_mask_present()
to_masked() returns MaskedArray when mask is set.
Source code in gwframe/tests/test_masked.py
| def test_returns_masked_array_when_mask_present(self):
"""to_masked() returns MaskedArray when mask is set."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.ones(N_SAMPLES, dtype=bool)
gwf_bytes = _write_adc_masked(data, mask)
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", allow_invalid=True)
result = ts.to_masked()
assert isinstance(result, np.ma.MaskedArray)
assert result.mask.all()
np.testing.assert_array_equal(result.data, data)
|
test_returns_plain_array_when_no_mask
test_returns_plain_array_when_no_mask()
to_masked() returns plain ndarray when mask is None.
Source code in gwframe/tests/test_masked.py
| def test_returns_plain_array_when_no_mask(self):
"""to_masked() returns plain ndarray when mask is None."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
frame.add_channel("H1:TEST", data, sample_rate=SAMPLE_RATE, channel_type="adc")
gwf_bytes = frame.write_bytes()
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", allow_invalid=True)
result = ts.to_masked()
assert not isinstance(result, np.ma.MaskedArray)
assert isinstance(result, np.ndarray)
np.testing.assert_array_equal(result, data)
|
test_to_masked_data_matches_array
test_to_masked_data_matches_array()
to_masked().data matches ts.array.
Source code in gwframe/tests/test_masked.py
| def test_to_masked_data_matches_array(self):
"""to_masked().data matches ts.array."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.ones(N_SAMPLES, dtype=bool)
gwf_bytes = _write_adc_masked(data, mask)
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", allow_invalid=True)
result = ts.to_masked()
np.testing.assert_array_equal(result.data, ts.array)
|