Tests for auxiliary-vector mask support on ADC channels.
Aux mask vectors are a site convention (the CDS dataValid aux FrVect):
a 16-bit integer vector on FrAdcData whose length evenly divides the data,
each value covering a block of consecutive samples (0 = good, nonzero =
invalid). Reading is opt-in via aux_mask=True; writing is opt-in per
channel with aux_mask=True, and the layout (1/16-second blocks for
integer rates that are a multiple of 16 Hz, otherwise one value per sample)
is derived from the sample rate. Non-conforming aux vectors are assumed to
not be masks and are ignored.
TestAuxMaskOrthogonality
allow_invalid (channel-level flag) and aux_mask are independent.
TestAuxMaskReadLeniency
Non-conforming aux vectors are assumed to not be masks and ignored.
test_non_conforming_channel_survives_multi_channel_read()
A non-conforming aux must not drop the channel from bulk reads.
Source code in gwframe/tests/test_aux_mask.py
| def test_non_conforming_channel_survives_multi_channel_read(self):
"""A non-conforming aux must not drop the channel from bulk reads."""
gwf_bytes = _write_raw_adc_with_aux(
np.ones(15, dtype=np.int16), _core.FrVect.FR_VECT_2S
)
channels = gwframe.read_bytes(gwf_bytes, channels=None, aux_mask=True)
assert "H1:RAW" in channels
assert channels["H1:RAW"].mask is None
|
TestAuxMaskRoundTrip
Masks written as aux vectors survive a write/read round-trip.
test_block_16hz_partial_block_is_conservative
test_block_16hz_partial_block_is_conservative()
Any masked sample marks its whole 1/16 s block invalid.
Source code in gwframe/tests/test_aux_mask.py
| def test_block_16hz_partial_block_is_conservative(self):
"""Any masked sample marks its whole 1/16 s block invalid."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.zeros(N_SAMPLES, dtype=bool)
mask[CDS_BLOCK + 5] = True # a single sample in block 1
gwf_bytes = _write_adc_aux(data, mask)
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", aux_mask=True)
assert ts.mask is not None
expected = np.zeros(N_SAMPLES, dtype=bool)
expected[CDS_BLOCK : 2 * CDS_BLOCK] = True
np.testing.assert_array_equal(ts.mask, expected)
|
test_block_16hz_roundtrip
test_block_16hz_roundtrip(sample_rate)
CDS layout: 16 aux values per second regardless of sample rate.
Source code in gwframe/tests/test_aux_mask.py
| @pytest.mark.parametrize("sample_rate", [4096, 16384])
def test_block_16hz_roundtrip(self, sample_rate):
"""CDS layout: 16 aux values per second regardless of sample rate."""
n_samples = int(DURATION * sample_rate)
block = sample_rate // 16
data = np.random.randn(n_samples).astype(np.float64)
mask = np.zeros(n_samples, dtype=bool)
mask[:block] = True # exactly the first block
masked = np.ma.MaskedArray(data, mask=mask)
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
frame.add_channel(
"H1:TEST",
masked,
sample_rate=sample_rate,
channel_type="adc",
aux_mask=True,
)
ts = gwframe.read_bytes(frame.write_bytes(), "H1:TEST", aux_mask=True)
assert ts.mask is not None
np.testing.assert_array_equal(ts.mask, mask)
np.testing.assert_array_equal(ts.array, data)
|
test_compression_roundtrip
test_compression_roundtrip(tmp_path)
Aux vectors survive frame compression.
Source code in gwframe/tests/test_aux_mask.py
| def test_compression_roundtrip(self, tmp_path):
"""Aux vectors survive frame compression."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.zeros(N_SAMPLES, dtype=bool)
mask[:CDS_BLOCK] = True
masked = np.ma.MaskedArray(data, mask=mask)
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
frame.add_channel(
"H1:TEST",
masked,
sample_rate=SAMPLE_RATE,
channel_type="adc",
aux_mask=True,
)
path = tmp_path / "aux_gzip.gwf"
frame.write(str(path), compression=Compression.GZIP)
ts = gwframe.read(str(path), "H1:TEST", aux_mask=True)
assert ts.mask is not None
np.testing.assert_array_equal(ts.mask, mask)
|
test_file_roundtrip_stream_path
test_file_roundtrip_stream_path(tmp_path)
The stream (file) reader decodes aux masks like the bytes reader.
Source code in gwframe/tests/test_aux_mask.py
| def test_file_roundtrip_stream_path(self, tmp_path):
"""The stream (file) reader decodes aux masks like the bytes reader."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.zeros(N_SAMPLES, dtype=bool)
mask[:CDS_BLOCK] = True
masked = np.ma.MaskedArray(data, mask=mask)
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
frame.add_channel(
"H1:TEST",
masked,
sample_rate=SAMPLE_RATE,
channel_type="adc",
aux_mask=True,
)
path = tmp_path / "aux.gwf"
frame.write(str(path))
ts = gwframe.read(str(path), "H1:TEST", aux_mask=True)
assert ts.mask is not None
np.testing.assert_array_equal(ts.mask, mask)
|
test_multi_frame_time_sliced_stitching
test_multi_frame_time_sliced_stitching(tmp_path)
Masks stitch correctly across frames in a time-sliced read.
Source code in gwframe/tests/test_aux_mask.py
| def test_multi_frame_time_sliced_stitching(self, tmp_path):
"""Masks stitch correctly across frames in a time-sliced read."""
path = tmp_path / "multi.gwf"
masks = []
with gwframe.FrameWriter(str(path)) as writer:
for i in range(3):
data = np.full(N_SAMPLES, float(i))
mask = np.zeros(N_SAMPLES, dtype=bool)
if i == 1:
mask[:CDS_BLOCK] = True # first 1/16 s of second frame
masks.append(mask)
writer.write(
{"H1:TEST": np.ma.MaskedArray(data, mask=mask)},
start=T0 + i * DURATION,
sample_rate=SAMPLE_RATE,
channel_type="adc",
aux_mask=True,
)
# Slice from mid-frame 0 to mid-frame 2
ts = gwframe.read(
str(path),
"H1:TEST",
start=T0 + 0.5,
end=T0 + 2.5,
aux_mask=True,
)
assert ts.mask is not None
expected = np.concatenate(masks)[
int(0.5 * SAMPLE_RATE) : int(2.5 * SAMPLE_RATE)
]
np.testing.assert_array_equal(ts.mask, expected)
|
test_per_sample_roundtrip
test_per_sample_roundtrip()
Rates that are not a multiple of 16 Hz get an exact 1-1 mask.
Source code in gwframe/tests/test_aux_mask.py
| def test_per_sample_roundtrip(self):
"""Rates that are not a multiple of 16 Hz get an exact 1-1 mask."""
rate = 100
data = np.random.randn(rate).astype(np.float64)
mask = np.zeros(rate, dtype=bool)
mask[10:20] = True
mask[-1] = True
gwf_bytes = _write_adc_aux(data, mask, sample_rate=rate)
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", aux_mask=True)
assert ts.mask is not None
np.testing.assert_array_equal(ts.mask, mask)
np.testing.assert_array_equal(ts.array, data)
|
test_read_frames_decodes_aux
test_read_frames_decodes_aux(tmp_path)
read_frames(aux_mask=True) yields frames with masked channels.
Source code in gwframe/tests/test_aux_mask.py
| def test_read_frames_decodes_aux(self, tmp_path):
"""read_frames(aux_mask=True) yields frames with masked channels."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.zeros(N_SAMPLES, dtype=bool)
mask[:CDS_BLOCK] = True
path = tmp_path / "frames.gwf"
masked = np.ma.MaskedArray(data, mask=mask)
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
frame.add_channel(
"H1:TEST",
masked,
sample_rate=SAMPLE_RATE,
channel_type="adc",
aux_mask=True,
)
frame.write(str(path))
with warnings.catch_warnings():
warnings.simplefilter("error") # no OnMaskLoss noise
frames = list(gwframe.read_frames(str(path), aux_mask=True))
assert len(frames) == 1
ts = frames[0]["H1:TEST"]
assert ts.mask is not None
np.testing.assert_array_equal(ts.mask, mask)
# Writing the reconstructed frame back preserves the mask in an aux
# vector (CDS layout, derived from the sample rate)
path2 = tmp_path / "frames_rewritten.gwf"
frames[0].write(str(path2))
ts2 = gwframe.read(str(path2), "H1:TEST", aux_mask=True)
assert ts2.mask is not None
np.testing.assert_array_equal(ts2.mask, mask)
stream = _core.IFrameFStream(str(path2))
adc = stream.read_fr_adc_data(0, "H1:TEST")
assert adc.get_aux_size() == 1
assert adc.get_aux_vector(0).get_n_data() == int(16 * DURATION)
|
TestAuxMaskWriteValidation
Invalid write requests fail loudly.
TestAuxMaskWrittenLayout
The written aux vector matches the CDS convention, with the layout
derived from the sample rate.
test_default_keeps_legacy_behavior
test_default_keeps_legacy_behavior()
aux_mask=False falls back to channel-level dataValid + warning.
Source code in gwframe/tests/test_aux_mask.py
| def test_default_keeps_legacy_behavior(self):
"""aux_mask=False falls back to channel-level dataValid + warning."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.zeros(N_SAMPLES, dtype=bool)
mask[10:20] = True
with pytest.warns(UserWarning, match="per-channel masking"):
gwf_bytes = _write_adc_aux(data, mask, aux_mask=False)
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", allow_invalid=True)
assert ts.mask is not None
assert ts.mask.all() # degraded to whole-channel mask
|
test_fully_masked_channel_uses_channel_flag
test_fully_masked_channel_uses_channel_flag(tmp_path)
All samples invalid: set the whole-channel flag, write no aux.
Source code in gwframe/tests/test_aux_mask.py
| def test_fully_masked_channel_uses_channel_flag(self, tmp_path):
"""All samples invalid: set the whole-channel flag, write no aux."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.ones(N_SAMPLES, dtype=bool)
with warnings.catch_warnings():
warnings.simplefilter("error") # nothing is lost, so no warning
gwf_bytes = _write_adc_aux(data, mask)
_stream, adc = _adc_from_bytes(gwf_bytes, tmp_path)
assert adc.get_aux_size() == 0
assert adc.get_data_valid() != 0
with pytest.raises(InvalidDataError):
gwframe.read_bytes(gwf_bytes, "H1:TEST", aux_mask=True)
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", aux_mask=True, allow_invalid=True)
assert ts.mask is not None
assert ts.mask.all()
|
test_no_on_mask_loss_warning_when_aux_encoded
test_no_on_mask_loss_warning_when_aux_encoded()
Aux encoding preserves the mask, so OnMaskLoss must not fire.
Source code in gwframe/tests/test_aux_mask.py
| def test_no_on_mask_loss_warning_when_aux_encoded(self):
"""Aux encoding preserves the mask, so OnMaskLoss must not fire."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.zeros(N_SAMPLES, dtype=bool)
mask[10:20] = True # partial mask: would warn without aux encoding
with warnings.catch_warnings():
warnings.simplefilter("error")
_write_adc_aux(data, mask)
|
test_non_dividing_length_falls_back_to_per_sample
test_non_dividing_length_falls_back_to_per_sample(tmp_path)
A sample count that is not a multiple of the 1/16 s block size
(frame duration not a multiple of 1/16 s) is written 1-1.
Source code in gwframe/tests/test_aux_mask.py
| def test_non_dividing_length_falls_back_to_per_sample(self, tmp_path):
"""A sample count that is not a multiple of the 1/16 s block size
(frame duration not a multiple of 1/16 s) is written 1-1."""
n = N_SAMPLES + 1
data = np.random.randn(n).astype(np.float64)
mask = np.zeros(n, dtype=bool)
mask[-1] = True
gwf_bytes = _write_adc_aux(data, mask)
_stream, adc = _adc_from_bytes(gwf_bytes, tmp_path)
assert adc.get_aux_size() == 1
assert adc.get_aux_vector(0).get_n_data() == n
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", aux_mask=True)
assert ts.mask is not None
np.testing.assert_array_equal(ts.mask, mask)
|
test_per_channel_dict_in_write
test_per_channel_dict_in_write(tmp_path)
write() accepts a per-channel dict of bools.
Source code in gwframe/tests/test_aux_mask.py
| def test_per_channel_dict_in_write(self, tmp_path):
"""write() accepts a per-channel dict of bools."""
mask = np.zeros(N_SAMPLES, dtype=bool)
mask[:CDS_BLOCK] = True
channels = {
"H1:AUX": np.ma.MaskedArray(np.ones(N_SAMPLES), mask=mask),
"H1:FLAG": np.ma.MaskedArray(np.ones(N_SAMPLES), mask=mask),
}
path = tmp_path / "dict.gwf"
with pytest.warns(UserWarning, match="per-channel masking"):
gwframe.write(
str(path),
channels,
start=T0,
sample_rate=SAMPLE_RATE,
channel_type="adc",
aux_mask={"H1:AUX": True},
)
stream = _core.IFrameFStream(str(path))
assert stream.read_fr_adc_data(0, "H1:AUX").get_aux_size() == 1
flag = stream.read_fr_adc_data(0, "H1:FLAG")
assert flag.get_aux_size() == 0
assert flag.get_data_valid() != 0
|
TestBytesReaderAuxOptIn
The bytes fast path only touches the aux vector when asked to.
TestReadFramesRewrite
read_frames(aux_mask=True) re-encodes ADC masks for lossless rewrites.
test_flag_only_channel_stays_flag
test_flag_only_channel_stays_flag(tmp_path)
A channel masked via the whole-channel flag (allow_invalid=True)
is rewritten with the flag, not an all-ones aux vector.
Source code in gwframe/tests/test_aux_mask.py
| def test_flag_only_channel_stays_flag(self, tmp_path):
"""A channel masked via the whole-channel flag (allow_invalid=True)
is rewritten with the flag, not an all-ones aux vector."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
gwf_bytes = _write_adc_aux(data, np.ones(N_SAMPLES, dtype=bool), aux_mask=False)
path2 = self._rewrite(gwf_bytes, tmp_path, allow_invalid=True)
stream = _core.IFrameFStream(str(path2))
adc = stream.read_fr_adc_data(0, "H1:TEST")
assert adc.get_aux_size() == 0
assert adc.get_data_valid() != 0
|
test_per_sample_file_rewritten_in_cds_layout
test_per_sample_file_rewritten_in_cds_layout(tmp_path)
A per-sample aux file at a 16-multiple rate is rewritten with 1/16 s
blocks: the mask is preserved conservatively (block-expanded).
Source code in gwframe/tests/test_aux_mask.py
| def test_per_sample_file_rewritten_in_cds_layout(self, tmp_path):
"""A per-sample aux file at a 16-multiple rate is rewritten with 1/16 s
blocks: the mask is preserved conservatively (block-expanded)."""
aux = np.zeros(N_SAMPLES, dtype=np.int16)
aux[5] = 1 # one bad sample in block 0
gwf_bytes = _write_raw_adc_with_aux(aux, _core.FrVect.FR_VECT_2S)
path2 = self._rewrite(gwf_bytes, tmp_path)
stream = _core.IFrameFStream(str(path2))
adc = stream.read_fr_adc_data(0, "H1:RAW")
assert adc.get_aux_size() == 1
assert adc.get_aux_vector(0).get_n_data() == int(16 * DURATION)
assert adc.get_data_valid() == 0
ts = gwframe.read(str(path2), "H1:RAW", aux_mask=True)
assert ts.mask is not None
assert ts.mask[:CDS_BLOCK].all() and not ts.mask[CDS_BLOCK:].any()
|
test_unknown_block_size_decodes
test_unknown_block_size_decodes()
A conforming aux with any block size decodes (reader is lenient).
Source code in gwframe/tests/test_aux_mask.py
| def test_unknown_block_size_decodes(self):
"""A conforming aux with any block size decodes (reader is lenient)."""
# 128 aux values over 4096 samples: 32-sample blocks (not 1, not 256)
aux = np.zeros(128, dtype=np.int16)
aux[0] = 1
gwf_bytes = _write_raw_adc_with_aux(aux, _core.FrVect.FR_VECT_2S)
ts = gwframe.read_bytes(gwf_bytes, "H1:RAW", aux_mask=True)
assert ts.mask is not None
assert ts.mask[:32].all() and not ts.mask[32:].any()
|
test_without_aux_mask_partial_mask_degrades
test_without_aux_mask_partial_mask_degrades(tmp_path)
read_frames(aux_mask=False) never decodes aux vectors, so the
rewritten channel is unmasked (legacy behavior).
Source code in gwframe/tests/test_aux_mask.py
| def test_without_aux_mask_partial_mask_degrades(self, tmp_path):
"""read_frames(aux_mask=False) never decodes aux vectors, so the
rewritten channel is unmasked (legacy behavior)."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
gwf_bytes = _write_adc_aux(data, self._mask())
path2 = self._rewrite(gwf_bytes, tmp_path, aux_mask=False)
stream = _core.IFrameFStream(str(path2))
adc = stream.read_fr_adc_data(0, "H1:TEST")
assert adc.get_aux_size() == 0
assert adc.get_data_valid() == 0
|