Last active
March 19, 2019 19:37
-
-
Save dkohlsdorf/b200e33ac0be6ea612eeb280f3f325b4 to your computer and use it in GitHub Desktop.
Align A Bunch Of Videos. Second file using merge sort for joining the timestamps.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
''' | |
Align videos using Dynamic Time Warping | |
Given a set of video files with associated time stamp files, | |
the program creates a new set of aligned videos. | |
Copyright (c) <year> <copyright holders> | |
Permission is hereby granted, free of charge, to any person obtaining a copy | |
of this software and associated documentation files (the "Software"), to deal | |
in the Software without restriction, including without limitation the rights | |
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | |
copies of the Software, and to permit persons to whom the Software is | |
furnished to do so, subject to the following conditions: | |
The above copyright notice and this permission notice shall be included in all | |
copies or substantial portions of the Software. | |
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | |
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | |
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | |
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | |
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | |
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE | |
SOFTWARE. | |
by Daniel Kohlsdorf | |
''' | |
import numpy as np | |
import matplotlib.pyplot as plt | |
import cv2 | |
import os | |
import math | |
import time | |
import sklearn | |
class AlignedVideo: | |
''' | |
Manages aligned read / write of videos | |
''' | |
def __init__(self): | |
self.video = None | |
self.writer = None | |
self.ts = [] | |
self.pos = 0 | |
self.last = None | |
def read_video(self, file, output): | |
''' | |
Open video input and output | |
file: the file to read | |
output: the output video | |
''' | |
self.video = cv2.VideoCapture(file) | |
codec = 0x00000021 # TODO Maybe change | |
width = int(self.video.get(3)) | |
height = int(self.video.get(4)) | |
fps = int(self.video.get(5)) | |
print(file, output, width, height, fps) | |
self.writer = cv2.VideoWriter(output, codec, fps, (width, height)) | |
self.last = np.zeros((height, width, 3), dtype=np.uint8) | |
def read_timestamps(self, file): | |
''' | |
Read all time stamps from CSV | |
file: path to filename with one timestamp per row | |
''' | |
for line in open(file): | |
self.ts.append(float(line.strip())) | |
def current_timestamp(self): | |
''' | |
Check the current timestamp | |
''' | |
return self.ts[self.pos] | |
def nop(self): | |
''' | |
Write last image | |
''' | |
self.writer.write(self.last) | |
def next_frame(self): | |
''' | |
Read next frame from video and write it to the output. | |
Also increases pointer to timestamps | |
''' | |
_, frame = self.video.read() | |
self.writer.write(frame) | |
self.pos += 1 | |
self.last = frame | |
return frame | |
def euclidean(x, y): | |
''' | |
Squared euclidean distance between two values x and y. | |
If the first value is a sequence, we take the minimum distance of y to all elements in x | |
x: float of array-like | |
y: float | |
''' | |
if type(x) == list: | |
min_dist = float('inf') | |
for v in x: | |
d = (v - y) ** 2 | |
if d < min_dist: | |
min_dist = d | |
return d | |
return (x - y) ** 2 | |
def min3(match, insertion, deletion): | |
''' | |
Calculate the minimum walking direction | |
''' | |
if match <= insertion and match <= deletion: | |
return (match, -1, -1) | |
if insertion < match and insertion < deletion: | |
return (insertion, -1, 0) | |
if deletion < insertion and deletion < match: | |
return (deletion, 0, -1) | |
def align(x, y, w): | |
''' | |
Align two sequences using dynamic time warping. | |
https://en.wikipedia.org/wiki/Dynamic_time_warping | |
x: the first sequence, which can also be a collection of sequences. | |
y: a 1-dimensional sequence | |
w: the warping band see | |
''' | |
n = len(x) | |
m = len(y) | |
w = max(w, abs(n-m + 2)) | |
# Build dynamic programming matrix and save path for back tracking ... | |
dp = np.ones((n + 1, m + 1)) * float('inf') | |
dp[0][0] = 0.0 | |
bp = {} | |
for i in range(1, n + 1): | |
for j in range(1, m + 1): | |
last, di, dj = min3( | |
dp[i - 1][j - 1], | |
dp[i - 1][j], | |
dp[i][j - 1] | |
) | |
dp[i][j] = euclidean(x[i - 1], y[j - 1]) + last | |
bp[(i, j)] = (i + di, j + dj) | |
# Back track alignment path and fill gaps with previous sample ... | |
aligned_x = [] | |
aligned_y = [] | |
i, j = bp[(n, m)] | |
aligned_x.append(x[n - 1]) | |
aligned_y.append(y[m - 1]) | |
while i > 0 and j > 0: | |
aligned_x.append(x[i - 1]) | |
aligned_y.append(y[j - 1]) | |
i, j = bp[(i, j)] | |
aligned_x.reverse() | |
aligned_y.reverse() | |
return (aligned_x, aligned_y, dp[n][m]) | |
def merge(sequence, so_far, band): | |
''' | |
Merge two sequences by appending the aligned samples | |
''' | |
(a, b, _) = align(so_far, sequence, band) | |
result = [] | |
for k in range(0, len(a)): | |
result.append(a[k] + [b[k]]) | |
return result | |
def merge_all(sequences): | |
''' | |
Greedily merge all sequences into one | |
''' | |
so_far = [[x] for x in sequences[0]] | |
for i in range(1, len(sequences)): | |
print("MERGE: {} {}".format(i + 1, len(sequences))) | |
band = int(max(len(so_far), len(sequences[i])) / 10) | |
so_far = merge(sequences[i], so_far, band) | |
return so_far | |
# Read all video files with their timestamps into annotated video classes | |
files = {} | |
path = 'recordings/private/session.017/video/trial1/' # Should be next to the recordings folder ... otherwise adjust path | |
for file_name in os.listdir(path): | |
file = file_name.split('.')[0].replace('_ts', "") | |
if not file in files: | |
files[file] = AlignedVideo() | |
if 'mkv' in file_name: | |
files[file].read_video(path + file_name, "{}_annotation.mp4".format(file)) | |
else: | |
files[file].read_timestamps(path + file_name) | |
# Merge all time stamps into a single aligned time stamp sequence | |
l = len(files) | |
pos = {k: i for i, (k, _) in enumerate(files.items())} | |
names = {i: k for i, (k, _) in enumerate(files.items())} | |
sequences = [None for i in range(0, len(files))] | |
for (k, i) in pos.items(): | |
sequences[i] = files[k].ts | |
aligned_list = merge_all(sequences) | |
# In debug mode copy all videos into one image, downsized | |
DEBUG = True | |
display = np.zeros((l * 128, 256, 3), dtype=np.uint8) | |
if DEBUG: | |
# plot aligned sequences | |
for i in range(0, l): | |
plt.plot([aligned_list[j][i] for j in range(0, len(aligned_list))], label=names[i]) | |
plt.title('Aligned Sequences') | |
plt.xlabel('frame') | |
plt.ylabel('seconds') | |
plt.legend() | |
plt.show() | |
# Align all frames to their respective timestamps | |
i = 0 | |
n = len(aligned_list) | |
start = time.time() | |
for i in range(0, n): | |
for (k, v) in files.items(): | |
if v.current_timestamp() == aligned_list[i][pos[k]]: | |
frame = v.next_frame() | |
if DEBUG: | |
img = cv2.resize(frame, (256, 128)) | |
display[pos[k] * 128:(pos[k] + 1) * 128,:, :] = img | |
else: | |
v.nop() | |
if DEBUG: | |
cv2.imshow(k, display) | |
if cv2.waitKey(1) & 0xFF == ord('q'): | |
break | |
if i % 1000 == 0 and i > 0: | |
end = time.time() | |
print("PROCESS: {} / {} in {}[s]".format(i, n, end - start)) | |
start = time.time() | |
i += 1 | |
for (_, v) in files.items(): | |
v.writer.release() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import numpy as np | |
import matplotlib.pyplot as plt | |
import cv2 | |
import os | |
import math | |
import time | |
class AlignedVideo: | |
''' | |
Manages aligned read / write of videos | |
''' | |
def __init__(self): | |
self.video = None | |
self.writer = None | |
self.ts = [] | |
self.pos = 0 | |
self.last = None | |
def read_video(self, file, output): | |
''' | |
Open video input and output | |
file: the file to read | |
output: the output video | |
''' | |
self.video = cv2.VideoCapture(file) | |
codec = 0x00000021 # TODO Maybe change | |
width = int(self.video.get(3)) | |
height = int(self.video.get(4)) | |
fps = int(self.video.get(5)) | |
print(file, output, width, height, fps) | |
self.writer = cv2.VideoWriter(output, codec, fps, (width, height)) | |
self.last = np.zeros((height, width, 3), dtype=np.uint8) | |
def read_timestamps(self, file): | |
''' | |
Read all time stamps from CSV | |
''' | |
for line in open(file): | |
self.ts.append(float(line.strip())) | |
def current_timestamp(self): | |
''' | |
Check the current timestamp | |
''' | |
return self.ts[self.pos] | |
def nop(self): | |
''' | |
Write last image | |
''' | |
self.writer.write(self.last) | |
def next_frame(self): | |
''' | |
Read next frame from video and write it to the output. | |
Also increases pointer to timestamps | |
''' | |
_, frame = self.video.read() | |
self.writer.write(frame) | |
self.pos += 1 | |
self.last = frame | |
return frame | |
# Read all video files with their timestamps into annotated video classes | |
files = {} | |
path = 'recordings/private/session.017/video/trial1/' # Should be next to the recordings folder ... otherwise adjust path | |
for file_name in os.listdir(path): | |
file = file_name.split('.')[0].replace('_ts', "") | |
if not file in files: | |
files[file] = AlignedVideo() | |
if 'mkv' in file_name: | |
files[file].read_video(path + file_name, "{}_annotation.mp4".format(file)) | |
else: | |
files[file].read_timestamps(path + file_name) | |
# Get all possible time stamps sorted as a time stamp master | |
all_timestamps = set([]) | |
for (k, v) in files.items(): | |
for t in v.ts: | |
all_timestamps.add(t) | |
all_timestamps = list(all_timestamps) | |
all_timestamps = sorted(all_timestamps) | |
# In debug mode copy all videos into one image, downsized | |
DEBUG = False | |
l = len(files) | |
display = np.zeros((l * 128, 256, 3), dtype=np.uint8) | |
pos = {} | |
for (i, k) in enumerate(files.keys()): | |
pos[k] = i | |
# Align all frames to their respective timestamps | |
i = 0 | |
n = len(all_timestamps) | |
start = time.time() | |
for t in all_timestamps: | |
for (k, v) in files.items(): | |
if v.current_timestamp() == t: | |
frame = v.next_frame() | |
if DEBUG: | |
img = cv2.resize(frame, (256, 128)) | |
display[pos[k] * 128:(pos[k] + 1) * 128,:, :] = img | |
else: | |
v.nop() | |
if DEBUG: | |
cv2.imshow(k, display) | |
if cv2.waitKey(1) & 0xFF == ord('q'): | |
break | |
if i % 250 == 0 and i > 0: | |
end = time.time() | |
print("PROCESS: {} / {} in {}[s]".format(i, n, end - start)) | |
start = time.time() | |
i += 1 | |
for (_, v) in files.items(): | |
v.writer.release() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import numpy as np | |
import matplotlib.pyplot as plt | |
import cv2 | |
import os | |
import math | |
import time | |
class AlignedVideo: | |
''' | |
Manages aligned read / write of videos | |
''' | |
def __init__(self): | |
self.video = None | |
self.writer = None | |
self.ts = [] | |
self.pos = 0 | |
self.last = None | |
def read_video(self, file, output, n_files): | |
''' | |
Open video input and output | |
file: the file to read | |
output: the output video | |
n_files: adjust frame rate by number of files, since we add n - 1 frames per bin (since we almost never match in any video) | |
''' | |
self.video = cv2.VideoCapture(file) | |
codec = 0x00000021 # TODO Maybe change | |
width = int(self.video.get(3)) | |
height = int(self.video.get(4)) | |
fps = int(self.video.get(5)) * (n_files - 1) | |
print(file, output, width, height, fps) | |
self.writer = cv2.VideoWriter(output, codec, fps, (width, height)) | |
self.last = np.zeros((height, width, 3), dtype=np.uint8) | |
def read_timestamps(self, file): | |
''' | |
Read all time stamps from CSV | |
''' | |
for line in open(file): | |
self.ts.append(float(line.strip())) | |
def current_timestamp(self): | |
''' | |
Check the current timestamp | |
''' | |
return self.ts[self.pos] | |
def nop(self): | |
''' | |
Write last image | |
''' | |
self.writer.write(self.last) | |
def next_frame(self): | |
''' | |
Read next frame from video and write it to the output. | |
Also increases pointer to timestamps | |
''' | |
_, frame = self.video.read() | |
self.writer.write(frame) | |
self.pos += 1 | |
self.last = frame | |
return frame | |
# Read all video files with their timestamps into annotated video classes | |
files = {} | |
path = 'recordings/private/session.017/video/trial1/' # Should be next to the recordings folder ... otherwise adjust path | |
for file_name in os.listdir(path): | |
file = file_name.split('.')[0].replace('_ts', "") | |
if not file in files: | |
files[file] = AlignedVideo() | |
if 'mkv' in file_name: | |
files[file].read_video(path + file_name, "{}_annotation.mp4".format(file), 5) | |
else: | |
files[file].read_timestamps(path + file_name) | |
def merge(sorted_x, sorted_y): | |
result = [] | |
i = 0 | |
j = 0 | |
while i < len(sorted_x) and j < len(sorted_y): | |
if sorted_x[i] > sorted_y[j]: | |
result.append(sorted_y[j]) | |
i += 1 | |
elif sorted_x[i] > sorted_y[j]: | |
result.append(sorted_x[i]) | |
j += 1 | |
else: | |
result.append(sorted_x[i]) | |
i += 1 | |
j += 1 | |
for rest in range(i, len(sorted_x)): | |
result.append(sorted_x[rest]) | |
for rest in range(j, len(sorted_y)): | |
result.append(sorted_y[rest]) | |
def merge(sorted_x, sorted_y): | |
''' | |
Merge two sorted lists into a new sorted list | |
''' | |
result = [] | |
i = 0 | |
j = 0 | |
while i < len(sorted_x) and j < len(sorted_y): | |
if sorted_x[i] > sorted_y[j]: | |
result.append(sorted_y[j]) | |
j += 1 | |
elif sorted_x[i] < sorted_y[j]: | |
result.append(sorted_x[i]) | |
i += 1 | |
else: | |
result.append(sorted_x[i]) | |
i += 1 | |
j += 1 | |
for rest in range(i, len(sorted_x)): | |
result.append(sorted_x[rest]) | |
for rest in range(j, len(sorted_y)): | |
result.append(sorted_y[rest]) | |
return result | |
# Get all possible time stamps sorted as a time stamp master | |
keys = list(files.keys()) | |
all_timestamps = files[keys[0]].ts | |
for i in range(0, len(keys)): | |
all_timestamps = merge(all_timestamps, files[keys[i]].ts) | |
# In debug mode copy all videos into one image, downsized | |
DEBUG = True | |
l = len(files) | |
display = np.zeros((l * 128, 256, 3), dtype=np.uint8) | |
pos = {} | |
for (i, k) in enumerate(files.keys()): | |
pos[k] = i | |
# Align all frames to their respective timestamps | |
i = 0 | |
n = len(all_timestamps) | |
start = time.time() | |
for t in all_timestamps: | |
for (k, v) in files.items(): | |
if v.current_timestamp() == t: | |
frame = v.next_frame() | |
if DEBUG: | |
img = cv2.resize(frame, (256, 128)) | |
display[pos[k] * 128:(pos[k] + 1) * 128,:, :] = img | |
else: | |
v.nop() | |
if DEBUG: | |
cv2.imshow(k, display) | |
if cv2.waitKey(1) & 0xFF == ord('q'): | |
break | |
if i % 1000 == 0 and i > 0: | |
end = time.time() | |
print("PROCESS: {} / {} in {}[s]".format(i, n, end - start)) | |
start = time.time() | |
break | |
i += 1 | |
for (_, v) in files.items(): | |
v.writer.release() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment