import sys
import os
from numpy import *
from scipy.linalg import *
from scipy.io import * 
from pyspark import SparkContext
import logging

if len(sys.argv) < 6:
  print >> sys.stderr, \
  "(sta) usage: delay <master> <inputFile_A> <inputFile_y> <outputFile> <mxLag>"
  exit(-1)

def parseVector(line):
	vec = [float(x) for x in line.split(' ')]
	ts = array(vec[3:])
	#ts = (ts - mean(ts))/std(ts)
	med = median(ts)
	ts = (ts - med) / (med)
	ts = (ts - mean(ts)) / std(ts)
	return ts

def getSta(x,y,lags):
	w = zeros((len(lags),1))
	for i in arange(len(lags)):
		w[i] = mean(x * roll(y,int(lags[i])))
	co = norm(w)
	wscale = w - min(w)
	wscale = wscale / max(wscale)
	ph = dot(lags,wscale/sum(wscale))
	return [ph,co]

# parse inputs
sc = SparkContext(sys.argv[1], "delay")
inputFile_A = str(sys.argv[2])
inputFile_y = str(sys.argv[3])
mxLag = int(sys.argv[5])
outputFile = str(sys.argv[4]) + "-delay-mxLag-" + str(mxLag)
lags = arange(2*mxLag) - floor(2*mxLag/2)
if not os.path.exists(outputFile):
    os.makedirs(outputFile)
logging.basicConfig(filename=outputFile+'/'+'stdout.log',level=logging.INFO,format='%(asctime)s %(message)s',datefmt='%m/%d/%Y %I:%M:%S %p')

# parse data
logging.info("(delay) loading data")
lines_A = sc.textFile(inputFile_A)
lines_y = sc.textFile(inputFile_y)
y = array([float(x) for x in lines_y.collect()[0].split(' ')])
y = (y - mean(y))/std(y)
A = lines_A.map(parseVector)

# compute sta
logging.info('(delay) computing delay')
sta = A.map(lambda x : getSta(x,y,lags))
logging.info('(delay) saving results...')
savemat(outputFile+"/"+"ph.mat",mdict={'ph':sta.map(lambda x : x[0]).collect()},oned_as='column',do_compression='true')
savemat(outputFile+"/"+"co.mat",mdict={'co':sta.map(lambda x : x[1]).collect()},oned_as='column',do_compression='true')

@@ -223,17 +223,17 @@
 putFile_
-A
+X
 %3E %3Cinput
@@ -483,243 +483,8 @@
 ts%0A%0A
-def getSta(x,y,lags):%0A%09w = zeros((len(lags),1))%0A%09for i in arange(len(lags)):%0A%09%09w%5Bi%5D = mean(x * roll(y,int(lags%5Bi%5D)))%0A%09co = norm(w)%0A%09wscale = w - min(w)%0A%09wscale = wscale / max(wscale)%0A%09ph = dot(lags,wscale/sum(wscale))%0A%09return %5Bph,co%5D%0A%0A
 # pa
@@ -544,17 +544,17 @@
 putFile_
-A
+X
  = str(s
@@ -703,16 +703,18 @@
 (2*mxLag
++1
 ) - floo
@@ -987,17 +987,17 @@
 )%0Alines_
-A
+X
  = sc.te
@@ -1169,16 +1169,210 @@
 ector)%0A%0A
+# make predictor%0AY = zeros((len(lags),len(y)))%0Afor i in arange(len(lags)):%0A%09%09lag = int(lags%5Bi%5D)%0A%09%09tmp = roll(y,lag)%0A%09%09if lag %3C 0:%0A%09%09%09tmp%5Blag:%5D = 0%0A%09%09if lag %3E 0:%0A%09%09%09tmp%5B:lag%5D = 0%0A%09%09Y%5Bi,:%5D = tmp%0A%0A
 # comput
@@ -1444,25 +1444,63 @@
 x : 
-getSta(x,y,lags))
+dot(Y,x))%0Anorm = sta.map(lambda x : norm(x)).collect()%0A
 %0Alog
@@ -1562,18 +1562,20 @@
 le+%22/%22+%22
-ph
+norm
 .mat%22,md
@@ -1584,46 +1584,18 @@
 t=%7B'
-ph':sta.map(lambda x : x%5B0%5D).collect()
+norm':norm
 %7D,on
@@ -1636,86 +1636,142 @@
 e')%0A
-savemat(outputFile+%22/%22+%22co.mat%22,mdict=%7B'co':sta.map(lambda x : x%5B1%5D).collect()
+for i in arange(len(lags)):%0A%09filt = sta.map(lambda x : x%5Bi%5D).collect()%0A%09savemat(outputFile+%22/%22+%22filt%22+str(i)+%22.mat%22,mdict=%7B'filt':filt
 %7D,on

Build predictor as a matrix