code: def test_annotation_filtering(first_samp):
      """Test that annotations work properly with filtering."""
      # Create data with just a DC component
      data = np.ones((1, 1000))
      info = create_info(1, 1000.0, "eeg")
      raws = [
          RawArray(data * (ii + 1), info, first_samp=first_samp)
          for ii in range(4)
      ]
      kwargs_pass = dict(l_freq=None, h_freq=50.0, fir_design="firwin")
      kwargs_stop = dict(l_freq=50.0, h_freq=None, fir_design="firwin")
      # lowpass filter, which should not modify the data
      raws_pass = [raw.copy().filter(**kwargs_pass) for raw in raws]
      # highpass filter, which should zero it out
      raws_stop = [raw.copy().filter(**kwargs_stop) for raw in raws]
      # concat the original and the filtered segments
      raws_concat = concatenate_raws([raw.copy() for raw in raws])
      raws_zero = raws_concat.copy().apply_function(lambda x: x * 0)
      raws_pass_concat = concatenate_raws(raws_pass)
      raws_stop_concat = concatenate_raws(raws_stop)
      # make sure we did something reasonable with our individual-file filtering
      assert_allclose(
          raws_concat[0][0], raws_pass_concat[0][0], atol=1e-14
      )
      assert_allclose(
          raws_zero[0][0], raws_stop_concat[0][0], atol=1e-14
      )
      # ensure that our Annotations cut up the filtering properly
      raws_concat_pass = raws_concat.copy().filter(
          skip_by_annotation="edge", **kwargs_pass
      )
      assert_allclose(
          raws_concat[0][0], raws_concat_pass[0][0], atol=1e-14
      )
      raws_concat_stop = raws_concat.copy().filter(
          skip_by_annotation="edge", **kwargs_stop
      )
      assert_allclose(
          raws_zero[0][0], raws_concat_stop[0][0], atol=1e-14
      )
  
      # test notch_filtering
      raw_notch = concatenate_raws(
          [raws_concat.copy(), raws_concat.copy()]
      )
      raw_notch.annotations.append(
          3.0 + raw_notch._first_time, 0.2, "foo_notch"
      )
      raw_notch.annotations.append(
          5.0 + raw_notch._first_time, 0.2, "foo_notch"
      )
  
      n_times = raw_notch._data.shape[1]
      with catch_logging() as log:
          raw_notch.notch_filter(
              60.0,
              fir_design="firwin",
              trans_bandwidth=5.0,
              skip_by_annotation="foo_notch",
              verbose="info",
          )
      log = log.getvalue()
      assert "3 contiguous segment" in log
      # check that data has same shape before/after filtering
      assert n_times == raw_notch._data.shape[1]
  
      # one last test: let's cut out a section entirely:
      # here the 1-3 second window should be skipped
      raw = raws_concat.copy()
      raw.annotations.append(1.0 + raw._first_time, 2.0, "foo")
      with catch_logging() as log:
          raw.filter(
              l_freq=50.0,
              h_freq=None,
              fir_design="firwin",
              skip_by_annotation="foo",
              verbose="info",
          )
      log = log.getvalue()
      assert "2 contiguous segments" in log
      raw.annotations.append(
          2.0 + raw._first_time, 1.0, "foo"
      )  # shouldn't change
      with catch_logging() as log:
          raw.filter(
              l_freq=50.0,
              h_freq=None,
              fir_design="firwin",
              skip_by_annotation="foo",
              verbose="info",
          )
      log = log.getvalue()
      assert "2 contiguous segments" in log
      # our filter will zero out anything not skipped:
      mask = np.concatenate(
          (np.zeros(1000), np.ones(2000), np.zeros(1000))
      )
      expected_data = raws_concat[0][0][0] * mask
      assert_allclose(raw[0][0][0], expected_data, atol=1e-14)
  
      # Let's try another one
      raw = raws[0].copy()
      raw.set_annotations(Annotations([0.0], [0.5], ["BAD_ACQ_SKIP"]))
      my_data, times = raw.get_data(
          reject_by_annotation="omit", return_times=True
      )
      assert_allclose(times, raw.times[500:])
      assert my_data.shape == (1, 500)
      raw_filt = raw.copy().filter(
          skip_by_annotation="bad_acq_skip", **kwargs_stop
      )
      expected = data.copy()
      expected[:, 500:] = 0
      assert_allclose(raw_filt[:][0], expected, atol=1e-14)
  
      raw = raws[0].copy()
      raw.set_annotations(Annotations([0.5], [0.5], ["BAD_ACQ_SKIP"]))
      my_data, times = raw.get_data(
          reject_by_annotation="omit", return_times=True
      )
      assert_allclose(times, raw.times[:500])
      assert my_data.shape == (1, 500)
      raw_filt = raw.copy().filter(
          skip_by_annotation="bad_acq_skip", **kwargs_stop
      )
      expected = data.copy()
      expected[:, :500] = 0
      assert_allclose(raw_filt[:][0], expected, atol=1e-14)
  
imports: import sys
  from collections import OrderedDict
  from datetime import datetime, timedelta, timezone
  from itertools import repeat
  from pathlib import Path
  import numpy
  import pytest
  from numpy.testing import assert_allclose, assert_array_almost_equal, assert_array_equal, assert_equal
  from pytest import approx
  import mne
  from mne import Annotations, Epochs, annotations_from_events, count_annotations, create_info, events_from_annotations, read_annotations
  from mne.annotations import _handle_meas_date, _read_annotations_txt_parse_header, _sync_onset
  from mne.datasets import testing
  from mne.io import RawArray, concatenate_raws, read_raw_fif
  from mne.utils import _dt_to_stamp, _raw_annot, _record_warnings, _stamp_to_dt, assert_and_remove_boundary_annot, catch_logging, check_version
label_desc: Use image node_id 12 for calling a global function with example usage: test_annotation_filtering(first_samp) without return types