diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 515400d..33e58af 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -8,10 +8,6 @@ repos: - id: check-toml - id: end-of-file-fixer - id: requirements-txt-fixer - - repo: https://github.com/asottile/reorder_python_imports - rev: v3.12.0 - hooks: - - id: reorder-python-imports - repo: https://github.com/PyCQA/doc8 rev: v1.1.1 hooks: diff --git a/src/pyrvt/peak_calculators.py b/src/pyrvt/peak_calculators.py index 0b51c84..695310e 100644 --- a/src/pyrvt/peak_calculators.py +++ b/src/pyrvt/peak_calculators.py @@ -8,8 +8,10 @@ import ctypes import itertools import pathlib -from abc import ABC, abstractmethod -from typing import List, Tuple +from abc import ABC +from abc import abstractmethod +from typing import List +from typing import Tuple import numba import numpy as np @@ -140,13 +142,10 @@ def calc_moments( moment : list Computed spectral moments. """ - squared_fa = np.square(fourier_amps) # Use trapzoidal integration to compute the requested moments. - moments = [ - 2.0 * trapz(freqs, np.power(2 * np.pi * freqs, o) * squared_fa) for o in orders - ] + moments = [2.0 * trapz(freqs, np.power(2 * np.pi * freqs, o) * squared_fa) for o in orders] return moments @@ -164,12 +163,13 @@ class SquaredSpectrum: Amplitude of the Fourier amplitude spectrum. """ - def __init__(self, freqs, fourier_amps): + def __init__(self, freqs: npt.ArrayLike, fourier_amps: npt.ArrayLike): + """Initialize SquaredSpectrum.""" self._freqs = freqs self._squared_fa = np.square(fourier_amps) self._moments = {} - def moment(self, num): + def moment(self, num: int) -> float: """Compute the spectral moments. The spectral moment is computed using the squared Fourier amplitude @@ -191,7 +191,14 @@ def moment(self, num): return moment - def moments(self, *nums): + def moments(self, *nums) -> List[float]: + """Return the computed moments. + + Returns + ------- + moments : list[float] + Computed spectral moments. + """ return [self.moment(n) for n in nums] @@ -379,9 +386,7 @@ def _calc_peak_factor(self, duration: float, **kwargs) -> float: bandwidth = np.sqrt(1 - (m1 * m1) / (m0 * m2)) bandwidth_eff = bandwidth**1.2 - num_zero_crossings = self.limited_num_zero_crossings( - duration * np.sqrt(m2 / m0) / np.pi - ) + num_zero_crossings = self.limited_num_zero_crossings(duration * np.sqrt(m2 / m0) / np.pi) # The expected peak factor is computed as the integral of the # complementary CDF (1 - CDF(x)). peak_factor = quad( @@ -399,9 +404,7 @@ def _calc_peak_factor(self, duration: float, **kwargs) -> float: return peak_factor @classmethod - def nonstationarity_factor( - cls, osc_damping: float, osc_freq: float, duration: float - ) -> float: + def nonstationarity_factor(cls, osc_damping: float, osc_freq: float, duration: float) -> float: """Compute nonstationarity factor to modify duration. Parameters @@ -457,9 +460,7 @@ def _calc_peak_factor(self, duration: float, **kwargs) -> float: m0, m2 = self._spectrum.moments(0, 2) # Compute the number of zero crossings - num_zero_crossings = self.limited_num_zero_crossings( - duration * np.sqrt(m2 / m0) / np.pi - ) + num_zero_crossings = self.limited_num_zero_crossings(duration * np.sqrt(m2 / m0) / np.pi) peak_factor = self.asymtotic_approx(num_zero_crossings) @@ -593,9 +594,7 @@ def _calc_peak_factor(self, duration: float, **kwargs) -> float: osc_freq = kwargs.get("osc_freq", None) osc_damping = kwargs.get("osc_damping", None) if osc_freq and osc_damping: - peak_factor *= Vanmarcke1975.nonstationarity_factor( - osc_damping, osc_freq, duration - ) + peak_factor *= Vanmarcke1975.nonstationarity_factor(osc_damping, osc_freq, duration) return peak_factor @@ -645,9 +644,7 @@ def _calc_peak_factor(self, duration: float, **kwargs) -> float: # Compute the peak factor by the indefinite integral. peak_factor = ( np.sqrt(2.0) - * quad( - _calc_cartwright_pf.ctypes, 0, np.inf, args=(num_extrema, bandwidth) - )[0] + * quad(_calc_cartwright_pf.ctypes, 0, np.inf, args=(num_extrema, bandwidth))[0] ) return peak_factor @@ -702,9 +699,7 @@ def _calc_duration_rms(self, duration, **kwargs): coef = 1.0 / 3.0 # This equation was rewritten in Boore and Thompson (2012). foo = 1.0 / (osc_freq * duration) - dur_ratio = 1 + 1.0 / (2 * np.pi * osc_damping) * ( - foo / (1 + coef * foo**power) - ) + dur_ratio = 1 + 1.0 / (2 * np.pi * osc_damping) * (foo / (1 + coef * foo**power)) duration *= dur_ratio return duration @@ -760,9 +755,7 @@ def _calc_duration_rms(self, duration, **kwargs): # Same model as used in Boore and Joyner (1984). This equation was # rewritten in Boore and Thompson (2012). foo = 1.0 / (osc_freq * duration) - dur_ratio = 1 + 1.0 / (2 * np.pi * osc_damping) * ( - foo / (1 + coef * foo**power) - ) + dur_ratio = 1 + 1.0 / (2 * np.pi * osc_damping) * (foo / (1 + coef * foo**power)) duration *= dur_ratio return duration @@ -787,10 +780,7 @@ def _make_bt_interpolator(region, ref): Interpolator for the data. """ - - fpath = pathlib.Path(__file__).parent.joinpath( - "data", f"{region}_{ref}_trms4osc.pars.gz" - ) + fpath = pathlib.Path(__file__).parent.joinpath("data", f"{region}_{ref}_trms4osc.pars.gz") d = np.rec.fromrecords( np.loadtxt(str(fpath), skiprows=4, usecols=range(9)), names="mag,dist,c1,c2,c3,c4,c5,c6,c7", @@ -1005,9 +995,7 @@ def _calc_duration_rms(self, duration, **kwargs): m = self.COEFS.d * af_ratio + self.COEFS.e * af_ratio**2 incr_max = c * np.exp(-duration / m) - incr = incr_max * np.exp( - -((np.log(osc_freq / modes_f)) ** 2) / (2 * self.COEFS.sd**2) - ) + incr = incr_max * np.exp(-((np.log(osc_freq / modes_f)) ** 2) / (2 * self.COEFS.sd**2)) duration_rms += incr.sum() return duration_rms diff --git a/src/pyrvt/runner.py b/src/pyrvt/runner.py index 77d9462..a8ec418 100644 --- a/src/pyrvt/runner.py +++ b/src/pyrvt/runner.py @@ -7,14 +7,13 @@ from pyrvt import __version__ from pyrvt.motions import DEFAULT_CALC -from pyrvt.tools import operation_fa2psa, operation_psa2fa +from pyrvt.tools import operation_fa2psa +from pyrvt.tools import operation_psa2fa parser = argparse.ArgumentParser( prog="pyrvt", description="Compute response or Fourier amplitude spectra using RVT." ) -parser.add_argument( - "--version", action="version", version="%(prog)s version " + str(__version__) -) +parser.add_argument("--version", action="version", version="%(prog)s version " + str(__version__)) parser.add_argument( "operation", help="""Operation to be performed: [psa2fa] converts from pseudo-spectral @@ -72,13 +71,9 @@ def main(): args = parser.parse_args() if args.operation == "psa2fa": - operation_psa2fa( - args.input, args.output, args.damping, args.method, args.fixed_spacing - ) + operation_psa2fa(args.input, args.output, args.damping, args.method, args.fixed_spacing) elif args.operation == "fa2psa": - operation_fa2psa( - args.input, args.output, args.damping, args.method, args.fixed_spacing - ) + operation_fa2psa(args.input, args.output, args.damping, args.method, args.fixed_spacing) else: raise NotImplementedError diff --git a/src/pyrvt/tools.py b/src/pyrvt/tools.py index d80c8e5..3ee164c 100644 --- a/src/pyrvt/tools.py +++ b/src/pyrvt/tools.py @@ -5,13 +5,14 @@ import functools import multiprocessing from pathlib import Path +from typing import List import numpy as np import numpy.typing as npt import pyexcel - from pyrvt import motions -from pyrvt.peak_calculators import get_peak_calculator, get_region +from pyrvt.peak_calculators import get_peak_calculator +from pyrvt.peak_calculators import get_region PARAMETER_NAMES = [ ("magnitude", "Magnitude"), @@ -23,13 +24,12 @@ ] -def get_fpaths(s): - return Path(".").glob(s) if "*" in s else [Path(s)] +def get_fpaths(pat: str) -> List[Path]: + """Iterate over file paths.""" + return Path(".").glob(pat) if "*" in pat else [Path(pat)] -def read_events( - fpath: str | Path, response_type: str -) -> tuple[str, np.ndarray, list[dict]]: +def read_events(fpath: str | Path, response_type: str) -> tuple[str, np.ndarray, list[dict]]: """Read data from the file an Excel work book. Parameters @@ -206,9 +206,7 @@ def calc_compatible_spectra( """ target_freqs = 1.0 / periods with multiprocessing.Pool() as pool: - results = pool.map( - functools.partial(_calc_fa, target_freqs, damping, method), events - ) + results = pool.map(functools.partial(_calc_fa, target_freqs, damping, method), events) # Copy values back into the dictionary modified = [] @@ -234,7 +232,7 @@ def operation_psa2fa( fixed_spacing: bool = True, verbose: bool = True, ): - """Compute the accel. response spectrum from a Fourier amplitude spectrum. + """Compute the acceleration response spectrum from a Fourier amplitude spectrum. Parameters ---------- @@ -256,7 +254,6 @@ def operation_psa2fa( Print status of calculation. """ - dst = Path(dst) dst.mkdir(parents=True, exist_ok=True) @@ -275,9 +272,7 @@ def operation_psa2fa( periods = _periods # Compute the FA from the PSA - freqs, events = calc_compatible_spectra( - method, periods, events, damping=damping - ) + freqs, events = calc_compatible_spectra(method, periods, events, damping=damping) basename = fpath.stem.rsplit("_", 1)[0]