code: #!/usr/bin/env python
  # -*- coding: utf-8 -*-
  
  import numpy as np
  
  from obspy.geodetics.base import gps2dist_azimuth
  
  from gmprocess.waveform_processing.clipping.clipping_ann import clipNet
  from gmprocess.waveform_processing.clipping.max_amp import Max_Amp
  from gmprocess.waveform_processing.clipping.histogram import Histogram
  from gmprocess.waveform_processing.clipping.ping import Ping
  
  M_TO_KM = 1.0 / 1000
  
  
  def check_clipping(st, origin, threshold=0.2):
      """Apply clicking check.
  
      Lower thresholds will pass fewer streams but will give less false negatives
      (i.e., streams in which clipping actually occurred but were missed).
  
      Args:
          st (StationStream):
             Trace of data.
          origin (ScalarEvent):
              ScalarEvent object.
          threshold (float):
              Threshold probability.
  
      Returns:
          StationStream checked for clipping.
  
      """
      # Don't bother with test for strong motion instruments
      chan_code = st.get_id().split(".")[2]
      if chan_code[1] == "N":
          return st
  
      # Don't bother with test if it has already failed
      if not st.passed:
          return st
  
      event_mag = origin.magnitude
      event_lon = origin.longitude
      event_lat = origin.latitude
      dist = (
          gps2dist_azimuth(
              lat1=event_lat,
              lon1=event_lon,
              lat2=st[0].stats["coordinates"]["latitude"],
              lon2=st[0].stats["coordinates"]["longitude"],
          )[0]
          * M_TO_KM
      )
  
      # Clip mag/dist to range of training dataset
      event_mag = np.clip(event_mag, 4.0, 8.8)
      dist = np.clip(dist, 0.0, 445.0)
  
      clip_nnet = clipNet()
  
      max_amp_method = Max_Amp(st, max_amp_thresh=6e6)
      hist_method = Histogram(st)
      ping_method = Ping(st)
      inputs = [
          event_mag,
          dist,
          max_amp_method.is_clipped,
          hist_method.is_clipped,
          ping_method.is_clipped,
      ]
      prob_clip = clip_nnet.evaluate(inputs)[0][0]
  
      if prob_clip >= threshold:
          for tr in st:
              tr.fail(f"Failed clipping check: prob_clip = {prob_clip:.2f}.")
  
      return st
   #!/usr/bin/env python
  # -*- coding: utf-8 -*-
  
  import os
  
  import numpy as np
  
  from gmprocess.core.streamcollection import StreamCollection
  from gmprocess.io.read import read_data
  from gmprocess.utils.test_utils import read_data_dir
  from gmprocess.utils.config import get_config
  
  from gmprocess.waveform_processing.windows import signal_split
  from gmprocess.waveform_processing.windows import signal_end
  from gmprocess.waveform_processing.windows import window_checks
  
  from gmprocess.waveform_processing.processing import get_corner_frequencies
  from gmprocess.waveform_processing.snr import compute_snr
  
  
  def test_corner_frequencies():
      # Default config has 'constant' corner frequency method, so the need
      # here is to force the 'snr' method.
      data_files, origin = read_data_dir("geonet", "us1000778i", "*.V1A")
      streams = []
      for f in data_files:
          streams += read_data(f)
  
      sc = StreamCollection(streams)
  
      config = get_config()
  
      window_conf = config["windows"]
  
      processed_streams = sc.copy()
      for st in processed_streams:
          if st.passed:
              # Estimate noise/signal split time
              event_time = origin.time
              event_lon = origin.longitude
              event_lat = origin.latitude
              st = signal_split(st, origin)
  
              # Estimate end of signal
              end_conf = window_conf["signal_end"]
              event_mag = origin.magnitude
              print(st)
              st = signal_end(
                  st,
                  event_time=event_time,
                  event_lon=event_lon,
                  event_lat=event_lat,
                  event_mag=event_mag,
                  **end_conf
              )
              wcheck_conf = window_conf["window_checks"]
              st = window_checks(
                  st,
                  min_noise_duration=wcheck_conf["min_noise_duration"],
                  min_signal_duration=wcheck_conf["min_signal_duration"],
              )
  
      pconfig = config["processing"]
  
      # Run SNR check
      # I think we don't do this anymore.
      test = [d for d in pconfig if list(d.keys())[0] == "compute_snr"]
      snr_config = test[0]["compute_snr"]
      snr_config["check"]["min_freq"] = 0.2
      for stream in processed_streams:
          stream = compute_snr(stream, mag=origin.magnitude, **snr_config)
  
      # Run get_corner_frequencies
      test = [d for d in pconfig if list(d.keys())[0] == "get_corner_frequencies"]
      cf_config = test[0]["get_corner_frequencies"]
      snr_config = cf_config["snr"]
  
      # With same_horiz False
      snr_config["same_horiz"] = False
  
      lp = []
      hp = []
      for stream in processed_streams:
          if not stream.passed:
              continue
          stream = get_corner_frequencies(stream, method="snr", snr=snr_config)
          if stream[0].hasParameter("corner_frequencies"):
              cfdict = stream[0].getParameter("corner_frequencies")
              lp.append(cfdict["lowpass"])
              hp.append(cfdict["highpass"])
      np.testing.assert_allclose(
          np.sort(hp), [0.00467919, 0.00584742, 0.01026485], atol=1e-6
      )
  
      st = processed_streams.select(station="HSES")[0]
      lps = [tr.getParameter("corner_frequencies")["lowpass"] for tr in st]
      hps = [tr.getParameter("corner_frequencies")["highpass"] for tr in st]
      np.testing.assert_allclose(np.sort(lps), [100.0, 100.0, 100.0], atol=1e-6)
      np.testing.assert_allclose(
          np.sort(hps), [0.00542478, 0.01026485, 0.02527502], atol=1e-6
      )
  
      # With same_horiz True
      snr_config["same_horiz"] = True
  
      lp = []
      hp = []
      for stream in processed_streams:
          if not stream.passed:
              continue
          stream = get_corner_frequencies(stream, method="snr", snr=snr_config)
          if stream[0].hasParameter("corner_frequencies"):
              cfdict = stream[0].getParameter("corner_frequencies")
              lp.append(cfdict["lowpass"])
              hp.append(cfdict["highpass"])
  
      np.testing.assert_allclose(
          np.sort(hp), [0.00467919, 0.01026485, 0.01787214], atol=1e-6
      )
  
      st = processed_streams.select(station="HSES")[0]
      lps = [tr.getParameter("corner_frequencies")["lowpass"] for tr in st]
      hps = [tr.getParameter("corner_frequencies")["highpass"] for tr in st]
      np.testing.assert_allclose(np.sort(lps), [100.0, 100.0, 100.0], atol=1e-6)
      np.testing.assert_allclose(np.sort(hps), [0.010265, 0.010265, 0.025275], atol=1e-6)
  
  
  if __name__ == "__main__":
      os.environ["CALLED_FROM_PYTEST"] = "True"
      test_corner_frequencies()
  
apis: numpy.clip
apis: numpy.sort