Videos can be made of sparse filters evolving over time. Below is a code snippet implementing the K-SVD algorithm. The purpose of the snippet is to visualize the state of sparse basis functions at they are iteratively refined.
1import matplotlib.pyplot as plt
2import numpy as np
3import scipy
4import sklearn.linear_model
5from matplotlib import gridspec
6from sklearn.feature_extraction import image
7
8import skvideo.datasets
9
10try:
11 xrange
12except NameError:
13 xrange = range
14
15np.random.seed(0)
16
17# use greedy K-SVD algorithm with OMP
18def code_step(X, D):
19 model = sklearn.linear_model.OrthogonalMatchingPursuit(
20 n_nonzero_coefs=5, fit_intercept=False, normalize=False
21 )
22 #C = sklearn.
23 model.fit(D.T, X.T)
24 return model.coef_
25
26def dict_step(X, C, D):
27 unused_indices = []
28 for k in xrange(D.shape[0]):
29 usedidx = np.abs(C[:, k])>0
30
31 if np.sum(usedidx) <= 1:
32 print("Skipping filter #%d" % (k,))
33 unused_indices.append(k)
34 continue
35
36 selectNotK = np.arange(D.shape[0]) != k
37 used_coef = C[usedidx, :][:, selectNotK]
38
39 E_kR = X[usedidx, :].T - np.dot(used_coef, D[selectNotK, :]).T
40
41 U, S, V = scipy.sparse.linalg.svds(E_kR, k=1)
42
43 # choose sign based on largest dot product
44 choicepos = np.dot(D[k,:], U[:, 0])
45 choiceneg = np.dot(D[k,:], -U[:, 0])
46
47 if choicepos > choiceneg:
48 D[k, :] = U[:, 0]
49 C[usedidx, k] = S[0] * V[0, :]
50 else:
51 D[k, :] = -U[:, 0]
52 C[usedidx, k] = -S[0] * V[0, :]
53
54
55 # re-randomize filters that were not used
56 for i in unused_indices:
57 D[i, :] = np.random.normal(size=D.shape[1])
58 D[i, :] /= np.sqrt(np.dot(D[i,:], D[i,:]))
59
60 return D
61
62def plot_weights(basis):
63 n_filters, n_channels, height, width = basis.shape
64 ncols = 10
65 nrows = 10
66 fig = plt.figure()
67 gs = gridspec.GridSpec(nrows, ncols)
68 rown = 0
69 coln = 0
70 for filter in xrange(n_filters):
71 ax = fig.add_subplot(gs[rown, coln])
72 mi = np.min(basis[filter, 0, :, :])
73 ma = np.max(basis[filter, 0, :, :])
74 ma = np.max((np.abs(mi), np.abs(ma)))
75 mi = -ma
76 ax.imshow(basis[filter, 0, :, :], vmin=mi, vmax=ma, cmap='Greys_r', interpolation='none')
77 ax.xaxis.set_major_locator(plt.NullLocator())
78 ax.yaxis.set_major_locator(plt.NullLocator())
79 coln += 1
80 if coln >= ncols:
81 coln = 0
82 rown += 1
83 gs.tight_layout(fig, pad=0, h_pad=0, w_pad=0)
84 fig.canvas.draw()
85 buf, sz = fig.canvas.print_to_buffer()
86 data = np.fromstring(buf, dtype=np.uint8).reshape(sz[1], sz[0], -1)[:, :, :3]
87 plt.close()
88 return data
89
90# a 5 fps video encoded using x264
91writer = skvideo.io.FFmpegWriter("sparsity.mp4",
92 inputdict={
93 "-r": "10"
94 },
95 outputdict={
96 '-vcodec': 'libx264', '-b': '30000000'
97})
98
99# open the first frame of bigbuckbunny
100filename = skvideo.datasets.bigbuckbunny()
101vidframe = skvideo.io.vread(filename, outputdict={"-pix_fmt": "gray"})[0, :, :, 0]
102
103# initialize D
104D = np.random.normal(size=(100, 7*7))
105for i in range(D.shape[0]):
106 D[i, :] /= np.sqrt(np.dot(D[i,:], D[i,:]))
107
108
109X = image.extract_patches_2d(vidframe, (7, 7))
110
111X = X.reshape(X.shape[0], -1).astype(float)
112
113# sumsample about 10000 patches
114X = X[np.random.permutation(X.shape[0])[:10000]]
115
116for i in range(200):
117 print("Iteration %d / %d" % (i, 200))
118 C = code_step(X, D)
119 D = dict_step(X, C, D)
120 frame = plot_weights(D.reshape(100, 1, 7, 7))
121 writer.writeFrame(frame)
122writer.close()
The video output for 200 iterations of the K-SVD algorithm:
If you want to create a corrupted version of a video, you can use the FFmpegReader/FFmpegWriter in combination. Just make sure that you pass the video metadata along, or you may get incorrect output video (such as incorrect framerate). Provided below is an example corrupting one frame from the source video with white noise:
1import numpy as np
2
3import skvideo.datasets
4
5filename = skvideo.datasets.bigbuckbunny()
6
7vid_in = skvideo.io.FFmpegReader(filename)
8data = skvideo.io.ffprobe(filename)['video']
9rate = data['@r_frame_rate']
10T = int(data['@nb_frames'])
11
12vid_out = skvideo.io.FFmpegWriter("corrupted_video.mp4", inputdict={
13 '-r': rate,
14 },
15 outputdict={
16 '-vcodec': 'libx264',
17 '-pix_fmt': 'yuv420p',
18 '-r': rate,
19})
20for idx, frame in enumerate(vid_in.nextFrame()):
21 print("Writing frame %d/%d" % (idx, T))
22 if (idx >= (T/2)) & (idx <= (T/2 + 10)):
23 frame = np.random.normal(128, 128, size=frame.shape).astype(np.uint8)
24 vid_out.writeFrame(frame)
25vid_out.close()
Video output of the corrupted BigBuckBunny sequence: