Commit 2d58eedd authored by Imanol Perez's avatar Imanol Perez
Browse files

Add tools to work with paths

parent 35f25ee5
import numpy as np
def lead_lag(mylist):
'''
Transforms a list by adding a delayed copy of the sameself.
Arguments:
mylist (list): List of dimensions len_stream x dim_stream.
Returns:
np.array: Lead-lag transform of mylist.
'''
leadlist = np.concatenate([[mylist[0]], mylist])
laglist = np.concatenate([mylist, [mylist[-1]]])
return np.concatenate([leadlist, laglist], axis=1)
def add_time(mylist, init_time = 0., total_time = 1.):
'''
adds a normalised time variable
should only use a half interval at begining and end
Arguments:
mylist (list): List of dimensions len_stream x dim_stream.
init_time (float): Initial time.
total_time (float): Total time of the stream.
Returns:
np.array: mylist with normalized time as the first dimension.
'''
ans = [[init_time + xn * total_time/(len(mylist)-1)] + list(x) for (xn,x) in enumerate(mylist)]
return np.array(ans) ## note that the elements of list have to be lists or tuples - a path in other words
def home_and_pen_off(mylist):
'''
adds a pen off co-ordinate at end and then returns the main signal to zero
Arguments:
mylist (list): List of dimensions len_stream x dim_stream.
Returns:
np.array: mylist with an extra dimension representing pen-on and pen-off,
as well as an extra final point.
'''
## pen on
ans = [list(x) + [1.] for x in mylist]
last = list(ans[-1])
## pen off
last[-1] = 0.
ans.append(last)
## home
ans.append([0 for item in last])
return np.array(ans)
def refocus(path, centre):
'''
premultiplies the stream path by the stream centre run backwards
Arguments:
path (list): First path.
centre (list): Second path.
Returns:
np.array: centre multiplied by path, run backwards.
'''
return np.concatenate((centre[::-1], path), axis=0)
if __name__ == "__main__":
path = np.array([[1., 2.], [4., 3.], [1., 1.]])
import numpy as np # Essential scientific library.
from esig import tosig # Computes signatures of streams of data.
from tqdm import tqdm # To show load bars
from multiprocessing import Pool # To allow multicore computation
from functools import partial # To avoid lambda and allow pickleable functions
import pickle # Serialises Python objects
import os # OS tools
def st2si(order, stream):
'''
Computes the signature of a stream.
Arguments:
X (iterable): Iterable of numpy arrays.
order (int): Order of the signature.
'''
if order > 1:
return(tosig.stream2sig(stream , order))
else:
if order == 1:
return np.concatenate((np.array([1]), stream[-1] - stream[0]), axis = 0)
else:
return np.array([1]);
def compute(X, order=2):
'''
Computes the signatures of a given dataset of streams of data.
Arguments:
X (iterable): Iterable of numpy arrays.
order (int): Order of the signature.
'''
func = partial(st2si,order)
pool = Pool()
n_samples = len(X)
signatures = []
try:
signatures = np.array(list(tqdm(pool.imap(func, X), total=n_samples)))
except Exception as e:
print('Failed to compute signatures: ' + repr(e))
signatures = []
return signatures
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment