Coverage for C: \ Users \ peaco \ OneDrive \ Documents \ GitHub \ mt_metadata \ mt_metadata \ timeseries \ filters \ time_delay_filter.py: 92%

1# ===================================================== 

2# Imports 

3# ===================================================== 

4from typing import Annotated 

5 

6import numpy as np 

7from loguru import logger 

8from pydantic import Field, field_validator, PrivateAttr, ValidationInfo 

9 

10from mt_metadata.base.helpers import requires 

11from mt_metadata.timeseries.filters import FilterBase, get_base_obspy_mapping 

12 

13 

14try: 

15 from obspy.core import inventory 

16except ImportError: 

17 inventory = None 

18 

19 

20# ===================================================== 

21class TimeDelayFilter(FilterBase): 

22 _filter_type: str = PrivateAttr("time delay") 

23 type: Annotated[ 

24 str, 

25 Field( 

26 default="time delay", 

27 description="Type of filter. Must be 'fir'", 

28 alias=None, 

29 json_schema_extra={ 

30 "units": None, 

31 "required": True, 

32 "examples": "time delay", 

33 }, 

34 ), 

35 ] 

36 delay: Annotated[ 

37 float, 

38 Field( 

39 default=0.0, 

40 description="The delay interval of the filter. This should be a single number.", 

41 alias=None, 

42 json_schema_extra={ 

43 "units": "second", 

44 "required": True, 

45 "examples": "-0.201", 

46 }, 

47 ), 

48 ] 

49 

50 @field_validator("type", mode="before") 

51 @classmethod 

52 def validate_type(cls, value, info: ValidationInfo) -> str: 

53 """ 

54 Validate that the type of filter is set to "time delay" 

55 """ 

56 if value not in ["time delay", "time_delay"]: 

57 logger.warning( 

58 f"Filter type is set to {value}, but should be 'time delay' for TimeDelayFilter." 

59 ) 

60 return "time delay" 

61 

62 def make_obspy_mapping(self): 

63 mapping = get_base_obspy_mapping() 

64 mapping["decimation_delay"] = "delay" 

65 return mapping 

66 

67 @requires(obspy=inventory) 

68 def to_obspy(self, stage_number=1, sample_rate=1, normalization_frequency=0): 

69 """ 

70 Convert to an obspy stage 

71 

72 :param stage_number: sequential stage number, defaults to 1 

73 :type stage_number: integer, optional 

74 :param normalization_frequency: Normalization frequency, defaults to 1 

75 :type normalization_frequency: float, optional 

76 :param sample_rate: sample rate, defaults to 1 

77 :type sample_rate: float, optional 

78 :return: Obspy stage filter 

79 :rtype: :class:`obspy.core.inventory.CoefficientsTypeResponseStage` 

80 

81 """ 

82 

83 stage = inventory.CoefficientsTypeResponseStage( 

84 stage_number, 

85 self.gain, 

86 normalization_frequency, 

87 self.units_in_object.symbol, 

88 self.units_out_object.symbol, 

89 "DIGITAL", 

90 name=self.name, 

91 decimation_input_sample_rate=sample_rate, 

92 decimation_factor=1, 

93 decimation_offset=0, 

94 decimation_delay=self.delay, 

95 decimation_correction=0, 

96 numerator=[1], 

97 denominator=[], 

98 description=self.get_filter_description(), 

99 input_units_description=self.units_in_object.name, 

100 output_units_description=self.units_out_object.name, 

101 ) 

102 

103 return stage 

104 

105 def complex_response(self, frequencies, **kwargs): 

106 """ 

107 Computes complex response for given frequency range 

108 :param frequencies: array of frequencies to estimate the response 

109 :type frequencies: np.ndarray 

110 

111 :return: complex response 

112 :rtype: np.ndarray 

113 

114 """ 

115 logger.debug( 

116 "USING FREQUENCY DOMAIN VERSION OF TIME DELAY FILTER NOT RECOMMENDED FOR MT PROCESSING" 

117 ) 

118 

119 if isinstance(frequencies, (float, int)): 

120 frequencies = np.array([frequencies]) 

121 w = 2 * np.pi * frequencies 

122 exponent = -1.0j * w * self.delay 

123 spectral_shift_multiplier = np.exp(exponent) 

124 return spectral_shift_multiplier