<jupyter_start><jupyter_text>Orthogonal Matching Pursuit (OMP) 3rd Attempt - Parallelize the Signals  Imports<jupyter_code># Preparing the Spark Environment
from pyspark.sql import SparkSession
from pyspark import SparkContext
sc = SparkContext.getOrCreate()
spark = SparkSession(sc)

# Mllib items
from pyspark.mllib.linalg import Matrices
from pyspark.mllib.linalg.distributed import BlockMatrix
from pyspark.mllib.regression import *

# System, Numpy and WAV reading
import os
from scipy.io import wavfile
import numpy as np

# Plots
import matplotlib.pyplot as plt
import matplotlib as mpl
mpl.rcParams['savefig.dpi'] = 80
mpl.rcParams['figure.dpi'] = 80
mpl.rcParams['figure.figsize'] = (16,9)

# Normalization and Metrics
from sklearn.preprocessing import normalize
from sklearn.metrics import mean_squared_error<jupyter_output><empty_output><jupyter_text>Creating the Dictionary (matrix)<jupyter_code># Setting the directories and Listing files

violaoWaveDir = 'audio/violao/used/'
violaoWaveList = sorted(os.listdir(violaoWaveDir))

pianoWaveDir = 'audio/piano/used/'
pianoWaveList = sorted(os.listdir(pianoWaveDir))
# Importing WAV files

violaoDict = []
for audioFile in violaoWaveList:
    violaoDict.append(wavfile.read(violaoWaveDir + audioFile)[1])
    
pianoDict = []
for audioFile in pianoWaveList:
    pianoDict.append(wavfile.read(pianoWaveDir + audioFile)[1])
# Creating the basis for the Dictionary

Dict = []
Dict.extend(violaoDict)
Dict.extend(pianoDict)

Dict = np.asarray(Dict)
Dict.shape
# Function to remove the null signals

def clearSilence (signals, threshold=0.1):    
    s = []
    for i in range(signals.shape[0]):
        if any(elem > threshold for elem in signals[i]):
            s.append(signals[i])
    s = np.asarray(s)
    
    return s
# Creating the Dictionary (D) with "clean" signals only

D = np.transpose(clearSilence(Dict))<jupyter_output><empty_output><jupyter_text>Preparing the Test Data<jupyter_code># Setting the directories and listing files

testWaveDir = 'audio/TEST/'
testWaveList = sorted(os.listdir(testWaveDir))
# Importing WAV files

testData = []
for audioFile in testWaveList:
    testData.append(wavfile.read(testWaveDir + audioFile)[1])
# Normalizing all data

testData = normalize(testData, axis=1, norm='l2')
D = normalize(D, axis=0, norm='l2')<jupyter_output><empty_output><jupyter_text>The OMP function<jupyter_code># OMP function

def omp(D,s,k):
    """
    Orthogonal Matching Pursuit (OMP)
        
    Inputs
        D: dictionary (matrix)
        s: signal
        k: sparsity level
        
    Output
        x: coeff vector for sparse representation
    """
    
    [l, c] = D.shape
    
    x = np.zeros([c, 1]) # coefficient (output)
    r = s # residual of s
    omega = np.zeros([k, 1]) # selected support
    D_omega = np.zeros([l, 1]) # correspondng columns of D
    
    for cnt in range(k): # choose k atoms
        #print("Iteration: ", cnt)
        x_tmp = np.zeros([c, 1])
        inds = np.setdiff1d(np.arange(0,c-1), omega)
        
        for i in inds:
            t = np.transpose(D[:, i])
            n = np.linalg.norm(D[:, i])
            x_tmp[i] = np.matmul(t,r) / n
        
        ichosen = np.argmax(abs(x_tmp))
        omega[cnt] = ichosen
        D_omega = np.column_stack([D_omega, D[:, ichosen]])
        if (cnt == 0):
            D_omega = np.delete(D_omega, 0, 1)
        x_ls = np.linalg.lstsq(D_omega, s, rcond=None)[0]
        r = s - np.matmul(D_omega, x_ls)
    
    for a in range(k):
        x[int(omega[a])] = x_ls[a]

    return x<jupyter_output><empty_output><jupyter_text>Tests  Non-Parallel<jupyter_code>%%time
RES_SER = [0] * testData.shape[0]
for i in range(testData.shape[0]):
    RES_SER[i] = omp(D[:,:], testData[i], 20)<jupyter_output>CPU times: user 3h 26min 29s, sys: 3h 34min 22s, total: 7h 51s
Wall time: 2h 2min 10s<jupyter_text>Parallel<jupyter_code># Parallelize the data

rdd_data = sc.parallelize(list(enumerate(testData)), 32)
%%time
RES = (
    rdd_data
    .map(lambda x: omp(D[:,:], x[1], 20))
    .collect()
)<jupyter_output>CPU times: user 6.58 s, sys: 3.6 s, total: 10.2 s
Wall time: 48min 57s
{'fraction_non_alphanumeric': 0.12053140096618357, 'fraction_numerical': 0.01642512077294686, 'mean_word_length': 3.652808988764045, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 24, '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': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '568652', 'license': 'MIT', 'n_tokens_mistral': 1599, 'n_tokens_neox': 1519, 'n_words': 366, 'path': 'PROJECT/Attempt 3/par_OMP_Attempt_3.ipynb', 'repo_name': 'titocaco/BIGDATA2018', 'size': '4140'}