import math from jddb.file_repo import FileRepo from jddb.processor import Signal, Shot, ShotSet, BaseProcessor from typing import Optional, List from scipy.fftpack import fft from copy import deepcopy from scipy.interpolate import interp1d import pandas as pd import numpy as np from scipy import signal as sig from copy import deepcopy class SliceProcessor(BaseProcessor): """ input the point number of the window and overlap rate of the given window , then the sample rate is recalculated, return a signal of time window sequence """ def __init__(self, window_length: int, overlap: float): super().__init__() assert (0 <= overlap <= 1), "Overlap is not between 0 and 1." self.params.update({"WindowLength": window_length, "Overlap": overlap}) def transform(self, signal: Signal) -> Signal: window_length = self.params["WindowLength"] overlap = self.params["Overlap"] new_signal = deepcopy(signal) raw_sample_rate = new_signal.attributes["SampleRate"] step = round(window_length * (1 - overlap)) down_time = new_signal.time[-1] down_time = round(down_time, 3) idx = len(signal.data) window = list() while (idx - window_length) >= 0: window.append(new_signal.data[idx - window_length:idx]) idx -= step window.reverse() new_signal.attributes['SampleRate'] = raw_sample_rate * len(window) / (len(new_signal.data) - window_length + 1) new_signal.data = np.array(window) new_start_time = down_time - len(window) / new_signal.attributes['SampleRate'] new_signal.attributes['StartTime'] = round(new_start_time, 3) new_signal.attributes['OriginalSampleRate'] = raw_sample_rate return new_signal class FFTProcessor(BaseProcessor): """ processing signal by Fast Fourier Transform , return the maximum amplitude and the corresponding frequency """ def __init__(self): super().__init__() self.amp_signal = None self.signal_rate = None self.fre_signal = None def transform(self, signal: Signal): self.amp_signal = deepcopy(signal) self.signal_rate = signal.attributes['OriginalSampleRate'] self.fre_signal = deepcopy(signal) self.fft() self.amp_max() return self.amp_signal, self.fre_signal def fft(self): if self.amp_signal.data.ndim == 1: N = len(self.amp_signal.data) fft_y = fft(self.amp_signal.data) abs_y = np.abs(fft_y) normed_abs_y = abs_y / (N / 2) self.amp_signal.data = normed_abs_y[:int(N / 2)] elif self.amp_signal.data.ndim == 2: N = self.amp_signal.data.shape[1] R = self.amp_signal.data.shape[0] raw_cover = np.empty(shape=[0, int(N / 2)], dtype=float) for i in range(R): fft_y = fft(self.amp_signal.data[i]) abs_y = np.abs(fft_y) normed_abs_y = abs_y / (N / 2) raw_cover = np.append(raw_cover, [normed_abs_y[:int(N / 2)]], axis=0) self.amp_signal.data = raw_cover def amp_max(self): fs = self.signal_rate raw = self.amp_signal.data amp_cover = np.empty(shape=0, dtype=float) fre_cover = np.empty(shape=0, dtype=float) N = (raw.shape[1]) * 2 f = (np.linspace(start=0, stop=N - 1, num=N) / N) * fs f = f[:int(N / 2)] for j in range(raw.shape[0]): list_max = (raw[j, :]).tolist() raw_max = max(list_max) max_index = list_max.index(max(list_max)) f_rawmax = f[max_index] amp_cover = np.append(amp_cover, raw_max) fre_cover = np.append(fre_cover, f_rawmax) self.amp_signal.data = amp_cover self.fre_signal.data = fre_cover class Mean(BaseProcessor): """ Given a set of input signals, average each instant """ def __init__(self): super().__init__() def transform(self, *signal: Signal) -> Signal: new_signal = Signal(np.row_stack([sign.data for sign in signal.__iter__()]).T, signal.__getitem__(0).attributes) new_signal.data = np.mean(np.array(new_signal.data, dtype=np.float32), axis=1) return new_signal class Concatenate(BaseProcessor): """ calculate the mean and standard deviation of the given signal """ def __init__(self): super().__init__() def transform(self, *signal: Signal) -> Signal: new_signal = Signal(np.concatenate([sign.data for sign in signal.__iter__()], axis=0), signal.__getitem__(0).attributes) return new_signal class AlarmTag(BaseProcessor): """ Give arbitrary signals, extract downtime, timeline, and generate actual warning time labels """ def __init__(self, lead_time, disruption_label: str, downtime_label: str): super().__init__() self.lead_time = lead_time self._disruption_label = disruption_label self._downtime_label = downtime_label def transform(self, signal: Signal): copy_signal = deepcopy(signal) fs = copy_signal.attributes['SampleRate'] start_time = copy_signal.attributes['StartTime'] if self.params[self._disruption_label] == 1: undisrupt_number = int(fs * (self.params[self._downtime_label] - self.lead_time - start_time)) else: undisrupt_number = len(copy_signal.data) if undisrupt_number < len(copy_signal.data): # new_data = np.zeros(shape=undisrupt_number, dtype=int) new_data = np.zeros(shape=1, dtype=int) for i in range(len(copy_signal.data) - 1): if i <= undisrupt_number - 1: new_data = np.append(new_data, np.array(0)) else: new_data = np.append(new_data, np.array(1)) else: new_data = np.zeros(shape=len(copy_signal.data), dtype=int) new_signal = Signal(data=new_data, attributes=dict()) new_signal.attributes['SampleRate'] = fs new_signal.attributes['StartTime'] = start_time return new_signal class RadiatedFraction(BaseProcessor): """ Given the radiated power signal and input power signal to calculate the radiated fraction. """ def __init__(self, ): super().__init__() def transform(self, radiated_power_signal: Signal, input_power_signal: Signal) -> Signal: """ :param radiated_power_signal: :param input_power_signal: :return: """ resampled_attributes = deepcopy(input_power_signal.attributes) new_data = radiated_power_signal.data / input_power_signal.data return Signal(data=new_data, attributes=resampled_attributes) def find_tags(prefix, all_tags): """ find tags that start with the prefix param: prefix: The first few strings of the tags users need to look for all_tags: a list of all the tags that needed to be filtered :return: matching tags as a list[sting] """ return list(filter(lambda tag: tag.encode("utf-8").decode("utf-8", "ignore")[0:len(prefix)] == prefix, all_tags))