<jupyter_start><jupyter_text>Table of Contents<jupyter_code>import librosa
import os
import csv
from scipy import signal 
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# Helper functions to spectral filtering

def readfilter(path_filter):
  '''
    read file with spectral filter and output vector with filter
  '''
  line =[]
  with open(path_filter,'r') as f:
    reader = csv.reader(f,delimiter=',')
    for row in reader:
      line.append(row)
      if len(line[-1]) == 1:
        print(line[-1])
      else:
        filter_spec = np.array(np.float32(line[-1]) )
  return filter_spec



def load_binary_tagged_data(data_path,tag):
  '''
   Data
  '''  
  nomesref = [f for f in os.listdir(data_path)]
  data_fs=[]
  nome = []
  for f in nomesref:
    if f[0:2] == tag:
      data,fs = librosa.load(data_path+f,sr=None)
      data_fs.append(data)
      nome.append(f)
    
  return data_fs,nome

# FFT spectral filtering

def filtro_spect(data_path,tag, filtered_output_path = '/home/stattus4dpenalva/stattus4/filterspec_sv_data/',filter_norm = 'size', save_filter = True, train = True, spect_filter = '/home/stattus4dpenalva/stattus4/spectrum_filter_fft'):
  '''
     Filtro de ruido espectral do sensor de signal, na hipótese de todas medidas sem vazamente apresentarem mesmas
   componentes espectrais provenientes do sensor digital de signal. Amostras são consideradas de mesmo tamanho.
   
   filter_norm: 'size' for the size of the frequency domain, other case will default for the sum of the coeficients
  '''
  data_fs,nome = load_binary_tagged_data(data_path,tag)
  fft_normalized_signal = []
  fftnorms = []
  ldata = len(data_fs[0])
  if train:
    freq_hist = np.zeros(shape=ldata)
  else:
    freq_hist = readfilter(spect_filter)
    freq_hist = freq_hist[:ldata]
    
  signalfreq = []
  for i in range(len(data_fs)):
    signalfreq.append(np.angle( np.fft.fft(data_fs[i],n=ldata) ))
    svfreq = np.abs(np.fft.fft(data_fs[i],n=ldata))  
    fftnorms.append(np.sum(svfreq)) 
    svfreq /= fftnorms[-1]  
    fft_normalized_signal.append(svfreq)
    

  if train:
    for i in range(len(data_fs)):
      freq_hist = freq_hist + fft_normalized_signal[i]*fftnorms[i]

  if filter_norm == 'size':
    freq_hist_N = freq_hist/len(freq_hist)
    freq_hist = freq_hist_N
  else:
    freq_hist /= np.sum(freq_hist)    
    
  if save_filter:
    if filter_norm == 'size':      
      with open('/home/stattus4dpenalva/stattus4/spectrum_filter_fft','w') as f:
        writer = csv.writer(f,delimiter=',')
        writer.writerow(['filter_size_normed'])
        writer.writerow(freq_hist_N)  
    else:
      with open('/home/stattus4dpenalva/stattus4/spectrum_filter_fft','w') as f:
        writer = csv.writer(f,delimiter=',')
        writer.writerow(['filter_coeficient_normed'])
        writer.writerow(freq_hist)
      

  dist = []
  spec_filtered_list = []
  spec_filtered = np.zeros(shape=ldata)
  for i in range(len(data_fs)):
    d = fft_normalized_signal[i]-freq_hist
    d[d < 0.0] = 0.0  
    spec_filtered = d*fftnorms[i]*np.exp(signalfreq[i]*1j)      
  librosa.output.write_wav(filtered_output_path+'filtered_{}'.format(nome[i]),np.real(np.fft.ifft(spec_filtered)),fs)
   
# '/home/stattus4dpenalva/stattus4/spectrum_filter_fft'
filtro_spect('/home/stattus4dpenalva/stattus4/wav_data_2018/','CV', filtered_output_path = '/home/stattus4dpenalva/stattus4/filterspec_cv_data/',filter_norm = 'size', save_filter = False, train = False, spect_filter = '/home/stattus4dpenalva/stattus4/spectrum_filter_fft')
class ProjectFT:  
    
  def __init__(self, dimensions_bounds, dimensions_resolution, number_of_bins = 10, mode = 'map size', **kwargs):
    '''
    Input:

     1- dimensions_bound is a two dimensional object with shape (2,2) for the map bounds

     2- dimensions_resolution is a object of shape (2,) with the minimal steps of frequency dimension and time dimension

      number_of_bins: an integer with the number of bins of the map.

      mode: how the bin projection are calculated, based on constrains implemented in __implement_constrain.

       - 'map size': constrain are calculated to match number_of_bins * bin_size == size freq x size time

       - 'kernel radius': calculate bins according to a kernel of given integer radius, the real radius are calculated by the following:
      r^2 = (delta_f_from_matrix + delta_f_from_nearest_neighbor_point)^2 + (delta_t_from_matrix + delta_t_from_nearest_neighbor_point)^2
      The bins are calculated from points in the kernel_centers list that must be of same size has number_of_bins.

     Optional key-word args:


     Freq time domain, must be a tuple of 2 dimensions, each with a list with 2 number, one specifing lower bounds of the domain other the upper bound
     Bin dict, the key must be a tuple of 2 dimensions.
    '''
    self.number_of_bins = number_of_bins
    self.mode = mode
    self.resolution = (dimensions_resolution[0], dimensions_resolution[1])
    self.freqtime_domain = ( (dimensions_bounds[0][0], dimensions_bounds[0][1]), (dimensions_bounds[1][0], dimensions_bounds[1][1]) ) 
    self.setup(dimensions_bounds, dimensions_resolution, mode, number_of_bins, **kwargs)
    
    
  def __call__(self, freq, time, **kwargs):    
    kwargs['mode'] = self.mode
    return _project(self, freq, time, **kwargs)

  def setup(self, dimensions_bounds, dimensions_resolution, mode, number_of_bins, **kwargs):
    '''
      Setup constrains through implementations.
      
     Constrainsts to the FT to bin projection mapping function.
   
     mode: how the bin projection are calculated, based on constrains implemented in __implement_constrain.

      -'map size': constrain are calculated to match number_of_bins * bin_size == size freq x size time

      output:
      
        (bin_space_shape, binsize, overlay)
      
        overlay = (overlay_x,overlay_y)
        bin_space_shape = (num_x_bins,num_y_bins)
        binsize = (binxsize,binysize)
        
    Projection with rectangular map size.
    
      output:
      
        (bin_points, bin_grid_location, original_map_index)      
      
        bin_points = (binfpos,bintpos)
        bin_grid_location = (binf,bint)              
        original_map_index = (f0 + (binf-1)*binxsize , t0 + (bint-1)*binysize)

      -'kernel radius': calculate bins according to a kernel of given integer radius, the real radius are calculated by the following:
      r^2 = (delta_f_from_matrix + delta_f_from_nearest_neighbor_point)^2 + (delta_t_from_matrix + delta_t_from_nearest_neighbor_point)^2
      The bins are calculated from points in the kernel_centers list that must be of same size has number_of_bins.
     
      output:
       
        #TODO: kernel mode setup, kernel_radius = 1, kernel_centers = [(0.1,0.3)], kernel_threshold = 0.1 
        
    Keyword argments:
        
      num x bins: number of divisions in x dimensions
    '''        
    if number_of_bins < 3:
      number_of_bins = 4
    if 'number_of_bins' not in kwargs:
      kwargs['num x bins'] = number_of_bins - 2
    
    self.constrains = _implement_constrain(dimensions_bounds, dimensions_resolution, mode, number_of_bins, **kwargs)        
#global module, implement constraining, statistical selection and projection. This must be imported by the Projection module.

def _project(self, freq, time, **kwargs):
  '''
   adsadsad  
  '''    
  mode = self.mode
                
  if 'return_original_index' not in kwargs: # assert x:
    return_original_index = False
  else:
    return_original_index = kwargs['return_original_index']        
    
  if 'mode' not in kwargs: # assert x:
    raise AssertionError('mode not specified.')
  else:
    mode = kwargs['mode']        
    
  if mode == 'map size':
    overlay = self.constrains[2]
    bin_space_shape = self.constrains[0]
    binxsize = self.constrains[1][0]
    binysize = self.constrains[1][1]
        
    f0,ff,t0,tf = (self.freqtime_domain[0][0],self.freqtime_domain[0][1],self.freqtime_domain[1][0],self.freqtime_domain[1][1])
    df,dt = (self.resolution[0],self.resolution[1])
      
       
    Nf=(freq-f0) // df
    rf=(freq-f0) % df
    Nt=(time-t0) // dt
    rt=(time-t0) % dt
        
    # Calculate bin number and bin location of the given frequency time point  
    binf=(freq-f0) // (df*binxsize)
    binf = binf + 1
    rbinf=(freq-f0) % (df*binxsize)
          
    bint=(time-t0) // (dt*binysize)
    bint = bint + 1  
    rbint=(time-t0) % (dt*binysize)
              
    if binf*binxsize > ff:
      binfpos = rbinf//df + overlay[0]
    else:
      binfpos = rbinf//df
        
    if bint*binysize > tf:
      bintpos = rbint//dt + overlay[1]
    else:
      bintpos = rbint//dt  
        
    bin_points = (binfpos,bintpos)
    bin_grid_location = (binf,bint)      
    original_map_index = (f0 + (binf-1)*binxsize , t0 + (bint-1)*binysize)
        
  return bin_points , bin_grid_location, original_map_index

def _implement_constrain(dimensions_bounds, dimensions_resolution, mode, number_of_bins, **kwargs):
  '''

  '''
    
    ## mode assertion
  assert type(mode) == type('')
      
  if mode != 'map size' and mode != 'kernel radius':
    raise AssertionError('Mode '+mode+' not found.\n Use one of the following:\n - map size, kernel radius.')
  ## 
  
  if mode == 'map size':
      
    # calculate the size of the mapping
    nx = np.abs(dimensions_bounds[0][1]-dimensions_bounds[0][0])//dimensions_resolution[0]
    ny = np.abs(dimensions_bounds[1][1]-dimensions_bounds[1][0])//dimensions_resolution[1]
        
    binsize = np.floor(nx * ny/number_of_bins)
      
    num_x_bins = kwargs['num x bins']
        
    # calculate the number of pixel superpositions that is necessary to fit num_x_bins in the x dim
    overlay_x = nx % num_x_bins
    binxsize = nx // num_x_bins
    
    num_y_bins = nx * ny //(binsize * num_x_bins)
      
    if num_y_bins == 0:
      num_y_bins = 1
        
    # calculate the number of pixel superpositions that is necessary to fit num_y_bins in the y dim
    overlay_y = ny % num_y_bins
    binysize = ny // num_y_bins
      
    # sanity check for dimensionality    
    if num_y_bins > ny or num_x_bins > nx:
      raise AssertionError('Number of resulting bins is higher than the number of pixels./n Try another num_x_bins  value')
        
    overlay = (overlay_x,overlay_y)
    bin_space_shape = (num_x_bins,num_y_bins)
    binsize = (binxsize,binysize)    
    
    return bin_space_shape,binsize,overlay
        
  if mode == 'kernel radius':
      #TODO
    return None

def _select_statistics(data, **kwargs):    
  pass

# Bugs:
# constrains: bin space shape is smaller than the number of bins ** lower priority
# projection: bin grid location limits to bin space shape ** higher priority
# Verify mapping through plotting, mainly: bin points binsize and overlay and original_map_index ** higher priority
projector = ProjectFT( ((10.0,8000.0),(0.0,10.0)) , (20.0,1.5), number_of_bins = 20.0, mode = 'map size')
from abc import ABCMeta

class StatisticalEstimator(ABCMeta):
  __metaclass__ = ABCMeta

  def __init__(self, dataset, domain, est_type, **kwargs):
    ''' 
     Abstract method to generating estimatives of spectral statistics
    '''
    self.setup(dataset, domain, est_type, **kwargs)
    
  def setup(self, dataset, domain, est_type, **kwargs):
    pass  

  def apply(self):
    pass       
class SpectralStatistics(StatisticalEstimator):

  def __init__(self, dataset, domain, est_type, **kwargs):
    ''' 
     Abstract method to generating estimatives of spectral statistics
    '''
    super(SpectralStatistics).__init__(self, dataset, domain, est_type, **kwargs)
    
    
  def setup(self, dataset, domain, est_type, **kwargs):
    
    self.selector = _select_statistics(dataset, est_type, **kwargs)
    
  def apply(self,**kwargs):

    self.selector(**kwargs)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10306079664570231, 'fraction_numerical': 0.011907756813417191, 'mean_word_length': 3.2022551092318534, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '112626', 'license': 'MIT', 'n_tokens_mistral': 4149, 'n_tokens_neox': 3957, 'n_words': 1006, 'path': 'notebooks/.ipynb_checkpoints/Spectral_Filtering-checkpoint.ipynb', 'repo_name': 'Uiuran/stattus4-audio-models', 'size': '11925'}