(61d00a474) v0.9.7.1

This commit is contained in:
Regalis
2020-03-04 13:04:10 +01:00
parent 3c50efa5c9
commit 3c09ebe02f
5086 changed files with 786063 additions and 295871 deletions
@@ -0,0 +1,185 @@
#!/usr/bin/env python
# coding: utf-8
import numpy as np
import numpy.linalg as LA
from scipy.ndimage.filters import gaussian_filter
from scipy.sparse import csc_matrix
from scipy.sparse.linalg import inv
from MotionEST import MotionEST
"""Anandan Model"""
class Anandan(MotionEST):
"""
constructor:
cur_f: current frame
ref_f: reference frame
blk_sz: block size
beta: smooth constrain weight
k1,k2,k3: confidence coefficients
max_iter: maximum number of iterations
"""
def __init__(self, cur_f, ref_f, blk_sz, beta, k1, k2, k3, max_iter=100):
super(Anandan, self).__init__(cur_f, ref_f, blk_sz)
self.levels = int(np.log2(blk_sz))
self.intensity_hierarchy()
self.c_maxs = []
self.c_mins = []
self.e_maxs = []
self.e_mins = []
for l in xrange(self.levels + 1):
c_max, c_min, e_max, e_min = self.get_curvature(self.cur_Is[l])
self.c_maxs.append(c_max)
self.c_mins.append(c_min)
self.e_maxs.append(e_max)
self.e_mins.append(e_min)
self.beta = beta
self.k1, self.k2, self.k3 = k1, k2, k3
self.max_iter = max_iter
"""
build intensity hierarchy
"""
def intensity_hierarchy(self):
level = 0
self.cur_Is = []
self.ref_Is = []
#build each level itensity by using gaussian filters
while level <= self.levels:
cur_I = gaussian_filter(self.cur_yuv[:, :, 0], sigma=(2**level) * 0.56)
ref_I = gaussian_filter(self.ref_yuv[:, :, 0], sigma=(2**level) * 0.56)
self.ref_Is.append(ref_I)
self.cur_Is.append(cur_I)
level += 1
"""
get curvature of each block
"""
def get_curvature(self, I):
c_max = np.zeros((self.num_row, self.num_col))
c_min = np.zeros((self.num_row, self.num_col))
e_max = np.zeros((self.num_row, self.num_col, 2))
e_min = np.zeros((self.num_row, self.num_col, 2))
for r in xrange(self.num_row):
for c in xrange(self.num_col):
h11, h12, h21, h22 = 0, 0, 0, 0
for i in xrange(r * self.blk_sz, r * self.blk_sz + self.blk_sz):
for j in xrange(c * self.blk_sz, c * self.blk_sz + self.blk_sz):
if 0 <= i < self.height - 1 and 0 <= j < self.width - 1:
Ix = I[i][j + 1] - I[i][j]
Iy = I[i + 1][j] - I[i][j]
h11 += Iy * Iy
h12 += Ix * Iy
h21 += Ix * Iy
h22 += Ix * Ix
U, S, _ = LA.svd(np.array([[h11, h12], [h21, h22]]))
c_max[r, c], c_min[r, c] = S[0], S[1]
e_max[r, c] = U[:, 0]
e_min[r, c] = U[:, 1]
return c_max, c_min, e_max, e_min
"""
get ssd of motion vector:
cur_I: current intensity
ref_I: reference intensity
center: current position
mv: motion vector
"""
def get_ssd(self, cur_I, ref_I, center, mv):
ssd = 0
for r in xrange(int(center[0]), int(center[0]) + self.blk_sz):
for c in xrange(int(center[1]), int(center[1]) + self.blk_sz):
if 0 <= r < self.height and 0 <= c < self.width:
tr, tc = r + int(mv[0]), c + int(mv[1])
if 0 <= tr < self.height and 0 <= tc < self.width:
ssd += (ref_I[tr, tc] - cur_I[r, c])**2
else:
ssd += cur_I[r, c]**2
return ssd
"""
get region match of level l
l: current level
last_mvs: matchine results of last level
radius: movenment radius
"""
def region_match(self, l, last_mvs, radius):
mvs = np.zeros((self.num_row, self.num_col, 2))
min_ssds = np.zeros((self.num_row, self.num_col))
for r in xrange(self.num_row):
for c in xrange(self.num_col):
center = np.array([r * self.blk_sz, c * self.blk_sz])
#use overlap hierarchy policy
init_mvs = []
if last_mvs is None:
init_mvs = [np.array([0, 0])]
else:
for i, j in {(r, c), (r, c + 1), (r + 1, c), (r + 1, c + 1)}:
if 0 <= i < last_mvs.shape[0] and 0 <= j < last_mvs.shape[1]:
init_mvs.append(last_mvs[i, j])
#use last matching results as the start postion as current level
min_ssd = None
min_mv = None
for init_mv in init_mvs:
for i in xrange(-2, 3):
for j in xrange(-2, 3):
mv = init_mv + np.array([i, j]) * radius
ssd = self.get_ssd(self.cur_Is[l], self.ref_Is[l], center, mv)
if min_ssd is None or ssd < min_ssd:
min_ssd = ssd
min_mv = mv
min_ssds[r, c] = min_ssd
mvs[r, c] = min_mv
return mvs, min_ssds
"""
smooth motion field based on neighbor constraint
uvs: current estimation
mvs: matching results
min_ssds: minimum ssd of matching results
l: current level
"""
def smooth(self, uvs, mvs, min_ssds, l):
sm_uvs = np.zeros((self.num_row, self.num_col, 2))
c_max = self.c_maxs[l]
c_min = self.c_mins[l]
e_max = self.e_maxs[l]
e_min = self.e_mins[l]
for r in xrange(self.num_row):
for c in xrange(self.num_col):
w_max = c_max[r, c] / (
self.k1 + self.k2 * min_ssds[r, c] + self.k3 * c_max[r, c])
w_min = c_min[r, c] / (
self.k1 + self.k2 * min_ssds[r, c] + self.k3 * c_min[r, c])
w = w_max * w_min / (w_max + w_min + 1e-6)
if w < 0:
w = 0
avg_uv = np.array([0.0, 0.0])
for i, j in {(r - 1, c), (r + 1, c), (r, c - 1), (r, c + 1)}:
if 0 <= i < self.num_row and 0 <= j < self.num_col:
avg_uv += 0.25 * uvs[i, j]
sm_uvs[r, c] = (w * w * mvs[r, c] + self.beta * avg_uv) / (
self.beta + w * w)
return sm_uvs
"""
motion field estimation
"""
def motion_field_estimation(self):
last_mvs = None
for l in xrange(self.levels, -1, -1):
mvs, min_ssds = self.region_match(l, last_mvs, 2**l)
uvs = np.zeros(mvs.shape)
for _ in xrange(self.max_iter):
uvs = self.smooth(uvs, mvs, min_ssds, l)
last_mvs = uvs
for r in xrange(self.num_row):
for c in xrange(self.num_col):
self.mf[r, c] = uvs[r, c]
@@ -0,0 +1,251 @@
#!/usr/bin/env python
# coding: utf-8
import numpy as np
import numpy.linalg as LA
from Util import MSE
from MotionEST import MotionEST
"""Exhaust Search:"""
class Exhaust(MotionEST):
"""
Constructor:
cur_f: current frame
ref_f: reference frame
blk_sz: block size
wnd_size: search window size
metric: metric to compare the blocks distrotion
"""
def __init__(self, cur_f, ref_f, blk_size, wnd_size, metric=MSE):
self.name = 'exhaust'
self.wnd_sz = wnd_size
self.metric = metric
super(Exhaust, self).__init__(cur_f, ref_f, blk_size)
"""
search method:
cur_r: start row
cur_c: start column
"""
def search(self, cur_r, cur_c):
min_loss = self.block_dist(cur_r, cur_c, [0, 0], self.metric)
cur_x = cur_c * self.blk_sz
cur_y = cur_r * self.blk_sz
ref_x = cur_x
ref_y = cur_y
#search all validate positions and select the one with minimum distortion
for y in xrange(cur_y - self.wnd_sz, cur_y + self.wnd_sz):
for x in xrange(cur_x - self.wnd_sz, cur_x + self.wnd_sz):
if 0 <= x < self.width - self.blk_sz and 0 <= y < self.height - self.blk_sz:
loss = self.block_dist(cur_r, cur_c, [y - cur_y, x - cur_x],
self.metric)
if loss < min_loss:
min_loss = loss
ref_x = x
ref_y = y
return ref_x, ref_y
def motion_field_estimation(self):
for i in xrange(self.num_row):
for j in xrange(self.num_col):
ref_x, ref_y = self.search(i, j)
self.mf[i, j] = np.array(
[ref_y - i * self.blk_sz, ref_x - j * self.blk_sz])
"""Exhaust with Neighbor Constraint"""
class ExhaustNeighbor(MotionEST):
"""
Constructor:
cur_f: current frame
ref_f: reference frame
blk_sz: block size
wnd_size: search window size
beta: neigbor loss weight
metric: metric to compare the blocks distrotion
"""
def __init__(self, cur_f, ref_f, blk_size, wnd_size, beta, metric=MSE):
self.name = 'exhaust + neighbor'
self.wnd_sz = wnd_size
self.beta = beta
self.metric = metric
super(ExhaustNeighbor, self).__init__(cur_f, ref_f, blk_size)
self.assign = np.zeros((self.num_row, self.num_col), dtype=np.bool)
"""
estimate neighbor loss:
cur_r: current row
cur_c: current column
mv: current motion vector
"""
def neighborLoss(self, cur_r, cur_c, mv):
loss = 0
#accumulate difference between current block's motion vector with neighbors'
for i, j in {(-1, 0), (1, 0), (0, 1), (0, -1)}:
nb_r = cur_r + i
nb_c = cur_c + j
if 0 <= nb_r < self.num_row and 0 <= nb_c < self.num_col and self.assign[
nb_r, nb_c]:
loss += LA.norm(mv - self.mf[nb_r, nb_c])
return loss
"""
search method:
cur_r: start row
cur_c: start column
"""
def search(self, cur_r, cur_c):
dist_loss = self.block_dist(cur_r, cur_c, [0, 0], self.metric)
nb_loss = self.neighborLoss(cur_r, cur_c, np.array([0, 0]))
min_loss = dist_loss + self.beta * nb_loss
cur_x = cur_c * self.blk_sz
cur_y = cur_r * self.blk_sz
ref_x = cur_x
ref_y = cur_y
#search all validate positions and select the one with minimum distortion
# as well as weighted neighbor loss
for y in xrange(cur_y - self.wnd_sz, cur_y + self.wnd_sz):
for x in xrange(cur_x - self.wnd_sz, cur_x + self.wnd_sz):
if 0 <= x < self.width - self.blk_sz and 0 <= y < self.height - self.blk_sz:
dist_loss = self.block_dist(cur_r, cur_c, [y - cur_y, x - cur_x],
self.metric)
nb_loss = self.neighborLoss(cur_r, cur_c, [y - cur_y, x - cur_x])
loss = dist_loss + self.beta * nb_loss
if loss < min_loss:
min_loss = loss
ref_x = x
ref_y = y
return ref_x, ref_y
def motion_field_estimation(self):
for i in xrange(self.num_row):
for j in xrange(self.num_col):
ref_x, ref_y = self.search(i, j)
self.mf[i, j] = np.array(
[ref_y - i * self.blk_sz, ref_x - j * self.blk_sz])
self.assign[i, j] = True
"""Exhaust with Neighbor Constraint and Feature Score"""
class ExhaustNeighborFeatureScore(MotionEST):
"""
Constructor:
cur_f: current frame
ref_f: reference frame
blk_sz: block size
wnd_size: search window size
beta: neigbor loss weight
max_iter: maximum number of iterations
metric: metric to compare the blocks distrotion
"""
def __init__(self,
cur_f,
ref_f,
blk_size,
wnd_size,
beta=1,
max_iter=100,
metric=MSE):
self.name = 'exhaust + neighbor+feature score'
self.wnd_sz = wnd_size
self.beta = beta
self.metric = metric
self.max_iter = max_iter
super(ExhaustNeighborFeatureScore, self).__init__(cur_f, ref_f, blk_size)
self.fs = self.getFeatureScore()
"""
get feature score of each block
"""
def getFeatureScore(self):
fs = np.zeros((self.num_row, self.num_col))
for r in xrange(self.num_row):
for c in xrange(self.num_col):
IxIx = 0
IyIy = 0
IxIy = 0
#get ssd surface
for x in xrange(self.blk_sz - 1):
for y in xrange(self.blk_sz - 1):
ox = c * self.blk_sz + x
oy = r * self.blk_sz + y
Ix = self.cur_yuv[oy, ox + 1, 0] - self.cur_yuv[oy, ox, 0]
Iy = self.cur_yuv[oy + 1, ox, 0] - self.cur_yuv[oy, ox, 0]
IxIx += Ix * Ix
IyIy += Iy * Iy
IxIy += Ix * Iy
#get maximum and minimum eigenvalues
lambda_max = 0.5 * ((IxIx + IyIy) + np.sqrt(4 * IxIy * IxIy +
(IxIx - IyIy)**2))
lambda_min = 0.5 * ((IxIx + IyIy) - np.sqrt(4 * IxIy * IxIy +
(IxIx - IyIy)**2))
fs[r, c] = lambda_max * lambda_min / (1e-6 + lambda_max + lambda_min)
if fs[r, c] < 0:
fs[r, c] = 0
return fs
"""
do exhaust search
"""
def search(self, cur_r, cur_c):
min_loss = self.block_dist(cur_r, cur_c, [0, 0], self.metric)
cur_x = cur_c * self.blk_sz
cur_y = cur_r * self.blk_sz
ref_x = cur_x
ref_y = cur_y
#search all validate positions and select the one with minimum distortion
for y in xrange(cur_y - self.wnd_sz, cur_y + self.wnd_sz):
for x in xrange(cur_x - self.wnd_sz, cur_x + self.wnd_sz):
if 0 <= x < self.width - self.blk_sz and 0 <= y < self.height - self.blk_sz:
loss = self.block_dist(cur_r, cur_c, [y - cur_y, x - cur_x],
self.metric)
if loss < min_loss:
min_loss = loss
ref_x = x
ref_y = y
return ref_x, ref_y
"""
add smooth constraint
"""
def smooth(self, uvs, mvs):
sm_uvs = np.zeros(uvs.shape)
for r in xrange(self.num_row):
for c in xrange(self.num_col):
avg_uv = np.array([0.0, 0.0])
for i, j in {(r - 1, c), (r + 1, c), (r, c - 1), (r, c + 1)}:
if 0 <= i < self.num_row and 0 <= j < self.num_col:
avg_uv += uvs[i, j] / 6.0
for i, j in {(r - 1, c - 1), (r - 1, c + 1), (r + 1, c - 1),
(r + 1, c + 1)}:
if 0 <= i < self.num_row and 0 <= j < self.num_col:
avg_uv += uvs[i, j] / 12.0
sm_uvs[r, c] = (self.fs[r, c] * mvs[r, c] + self.beta * avg_uv) / (
self.beta + self.fs[r, c])
return sm_uvs
def motion_field_estimation(self):
#get matching results
mvs = np.zeros(self.mf.shape)
for r in xrange(self.num_row):
for c in xrange(self.num_col):
ref_x, ref_y = self.search(r, c)
mvs[r, c] = np.array([ref_y - r * self.blk_sz, ref_x - c * self.blk_sz])
#add smoothness constraint
uvs = np.zeros(self.mf.shape)
for _ in xrange(self.max_iter):
uvs = self.smooth(uvs, mvs)
self.mf = uvs
@@ -0,0 +1,40 @@
#!/ usr / bin / env python
#coding : utf - 8
import numpy as np
import numpy.linalg as LA
from MotionEST import MotionEST
"""Ground Truth:
Load in ground truth motion field and mask
"""
class GroundTruth(MotionEST):
"""constructor:
cur_f:current
frame ref_f:reference
frame blk_sz:block size
gt_path:ground truth motion field file path
"""
def __init__(self, cur_f, ref_f, blk_sz, gt_path, mf=None, mask=None):
self.name = 'ground truth'
super(GroundTruth, self).__init__(cur_f, ref_f, blk_sz)
self.mask = np.zeros((self.num_row, self.num_col), dtype=np.bool)
if gt_path:
with open(gt_path) as gt_file:
lines = gt_file.readlines()
for i in xrange(len(lines)):
info = lines[i].split(';')
for j in xrange(len(info)):
x, y = info[j].split(',')
#-, - stands for nothing
if x == '-' or y == '-':
self.mask[i, -j - 1] = True
continue
#the order of original file is flipped on the x axis
self.mf[i, -j - 1] = np.array([float(y), -float(x)], dtype=np.int)
else:
self.mf = mf
self.mask = mask
@@ -0,0 +1,204 @@
#!/usr/bin/env python
# coding: utf-8
import numpy as np
import numpy.linalg as LA
from scipy.ndimage.filters import gaussian_filter
from scipy.sparse import csc_matrix
from scipy.sparse.linalg import inv
from MotionEST import MotionEST
"""Horn & Schunck Model"""
class HornSchunck(MotionEST):
"""
constructor:
cur_f: current frame
ref_f: reference frame
blk_sz: block size
alpha: smooth constrain weight
sigma: gaussian blur parameter
"""
def __init__(self, cur_f, ref_f, blk_sz, alpha, sigma, max_iter=100):
super(HornSchunck, self).__init__(cur_f, ref_f, blk_sz)
self.cur_I, self.ref_I = self.getIntensity()
#perform gaussian blur to smooth the intensity
self.cur_I = gaussian_filter(self.cur_I, sigma=sigma)
self.ref_I = gaussian_filter(self.ref_I, sigma=sigma)
self.alpha = alpha
self.max_iter = max_iter
self.Ix, self.Iy, self.It = self.intensityDiff()
"""
Build Frame Intensity
"""
def getIntensity(self):
cur_I = np.zeros((self.num_row, self.num_col))
ref_I = np.zeros((self.num_row, self.num_col))
#use average intensity as block's intensity
for i in xrange(self.num_row):
for j in xrange(self.num_col):
r = i * self.blk_sz
c = j * self.blk_sz
cur_I[i, j] = np.mean(self.cur_yuv[r:r + self.blk_sz, c:c + self.blk_sz,
0])
ref_I[i, j] = np.mean(self.ref_yuv[r:r + self.blk_sz, c:c + self.blk_sz,
0])
return cur_I, ref_I
"""
Get First Order Derivative
"""
def intensityDiff(self):
Ix = np.zeros((self.num_row, self.num_col))
Iy = np.zeros((self.num_row, self.num_col))
It = np.zeros((self.num_row, self.num_col))
sz = self.blk_sz
for i in xrange(self.num_row - 1):
for j in xrange(self.num_col - 1):
"""
Ix:
(i ,j) <--- (i ,j+1)
(i+1,j) <--- (i+1,j+1)
"""
count = 0
for r, c in {(i, j + 1), (i + 1, j + 1)}:
if 0 <= r < self.num_row and 0 < c < self.num_col:
Ix[i, j] += (
self.cur_I[r, c] - self.cur_I[r, c - 1] + self.ref_I[r, c] -
self.ref_I[r, c - 1])
count += 2
Ix[i, j] /= count
"""
Iy:
(i ,j) (i ,j+1)
^ ^
| |
(i+1,j) (i+1,j+1)
"""
count = 0
for r, c in {(i + 1, j), (i + 1, j + 1)}:
if 0 < r < self.num_row and 0 <= c < self.num_col:
Iy[i, j] += (
self.cur_I[r, c] - self.cur_I[r - 1, c] + self.ref_I[r, c] -
self.ref_I[r - 1, c])
count += 2
Iy[i, j] /= count
count = 0
#It:
for r in xrange(i, i + 2):
for c in xrange(j, j + 2):
if 0 <= r < self.num_row and 0 <= c < self.num_col:
It[i, j] += (self.ref_I[r, c] - self.cur_I[r, c])
count += 1
It[i, j] /= count
return Ix, Iy, It
"""
Get weighted average of neighbor motion vectors
for evaluation of laplacian
"""
def averageMV(self):
avg = np.zeros((self.num_row, self.num_col, 2))
"""
1/12 --- 1/6 --- 1/12
| | |
1/6 --- -1/8 --- 1/6
| | |
1/12 --- 1/6 --- 1/12
"""
for i in xrange(self.num_row):
for j in xrange(self.num_col):
for r, c in {(-1, 0), (1, 0), (0, -1), (0, 1)}:
if 0 <= i + r < self.num_row and 0 <= j + c < self.num_col:
avg[i, j] += self.mf[i + r, j + c] / 6.0
for r, c in {(-1, -1), (-1, 1), (1, -1), (1, 1)}:
if 0 <= i + r < self.num_row and 0 <= j + c < self.num_col:
avg[i, j] += self.mf[i + r, j + c] / 12.0
return avg
def motion_field_estimation(self):
count = 0
"""
u_{n+1} = ~u_n - Ix(Ix.~u_n+Iy.~v+It)/(IxIx+IyIy+alpha^2)
v_{n+1} = ~v_n - Iy(Ix.~u_n+Iy.~v+It)/(IxIx+IyIy+alpha^2)
"""
denom = self.alpha**2 + np.power(self.Ix, 2) + np.power(self.Iy, 2)
while count < self.max_iter:
avg = self.averageMV()
self.mf[:, :, 1] = avg[:, :, 1] - self.Ix * (
self.Ix * avg[:, :, 1] + self.Iy * avg[:, :, 0] + self.It) / denom
self.mf[:, :, 0] = avg[:, :, 0] - self.Iy * (
self.Ix * avg[:, :, 1] + self.Iy * avg[:, :, 0] + self.It) / denom
count += 1
self.mf *= self.blk_sz
def motion_field_estimation_mat(self):
row_idx = []
col_idx = []
data = []
N = 2 * self.num_row * self.num_col
b = np.zeros((N, 1))
for i in xrange(self.num_row):
for j in xrange(self.num_col):
"""(IxIx+alpha^2)u+IxIy.v-alpha^2~u IxIy.u+(IyIy+alpha^2)v-alpha^2~v"""
u_idx = i * 2 * self.num_col + 2 * j
v_idx = u_idx + 1
b[u_idx, 0] = -self.Ix[i, j] * self.It[i, j]
b[v_idx, 0] = -self.Iy[i, j] * self.It[i, j]
#u: (IxIx+alpha^2)u
row_idx.append(u_idx)
col_idx.append(u_idx)
data.append(self.Ix[i, j] * self.Ix[i, j] + self.alpha**2)
#IxIy.v
row_idx.append(u_idx)
col_idx.append(v_idx)
data.append(self.Ix[i, j] * self.Iy[i, j])
#v: IxIy.u
row_idx.append(v_idx)
col_idx.append(u_idx)
data.append(self.Ix[i, j] * self.Iy[i, j])
#(IyIy+alpha^2)v
row_idx.append(v_idx)
col_idx.append(v_idx)
data.append(self.Iy[i, j] * self.Iy[i, j] + self.alpha**2)
#-alpha^2~u
#-alpha^2~v
for r, c in {(-1, 0), (1, 0), (0, -1), (0, 1)}:
if 0 <= i + r < self.num_row and 0 <= j + c < self.num_col:
u_nb = (i + r) * 2 * self.num_col + 2 * (j + c)
v_nb = u_nb + 1
row_idx.append(u_idx)
col_idx.append(u_nb)
data.append(-1 * self.alpha**2 / 6.0)
row_idx.append(v_idx)
col_idx.append(v_nb)
data.append(-1 * self.alpha**2 / 6.0)
for r, c in {(-1, -1), (-1, 1), (1, -1), (1, 1)}:
if 0 <= i + r < self.num_row and 0 <= j + c < self.num_col:
u_nb = (i + r) * 2 * self.num_col + 2 * (j + c)
v_nb = u_nb + 1
row_idx.append(u_idx)
col_idx.append(u_nb)
data.append(-1 * self.alpha**2 / 12.0)
row_idx.append(v_idx)
col_idx.append(v_nb)
data.append(-1 * self.alpha**2 / 12.0)
M = csc_matrix((data, (row_idx, col_idx)), shape=(N, N))
M_inv = inv(M)
uv = M_inv.dot(b)
for i in xrange(self.num_row):
for j in xrange(self.num_col):
self.mf[i, j, 0] = uv[i * 2 * self.num_col + 2 * j + 1, 0] * self.blk_sz
self.mf[i, j, 1] = uv[i * 2 * self.num_col + 2 * j, 0] * self.blk_sz
@@ -0,0 +1,109 @@
#!/ usr / bin / env python
#coding : utf - 8
import numpy as np
import numpy.linalg as LA
import matplotlib.pyplot as plt
from Util import drawMF, MSE
"""The Base Class of Estimators"""
class MotionEST(object):
"""
constructor:
cur_f: current frame
ref_f: reference frame
blk_sz: block size
"""
def __init__(self, cur_f, ref_f, blk_sz):
self.cur_f = cur_f
self.ref_f = ref_f
self.blk_sz = blk_sz
#convert RGB to YUV
self.cur_yuv = np.array(self.cur_f.convert('YCbCr'), dtype=np.int)
self.ref_yuv = np.array(self.ref_f.convert('YCbCr'), dtype=np.int)
#frame size
self.width = self.cur_f.size[0]
self.height = self.cur_f.size[1]
#motion field size
self.num_row = self.height // self.blk_sz
self.num_col = self.width // self.blk_sz
#initialize motion field
self.mf = np.zeros((self.num_row, self.num_col, 2))
"""estimation function Override by child classes"""
def motion_field_estimation(self):
pass
"""
distortion of a block:
cur_r: current row
cur_c: current column
mv: motion vector
metric: distortion metric
"""
def block_dist(self, cur_r, cur_c, mv, metric=MSE):
cur_x = cur_c * self.blk_sz
cur_y = cur_r * self.blk_sz
h = min(self.blk_sz, self.height - cur_y)
w = min(self.blk_sz, self.width - cur_x)
cur_blk = self.cur_yuv[cur_y:cur_y + h, cur_x:cur_x + w, :]
ref_x = int(cur_x + mv[1])
ref_y = int(cur_y + mv[0])
if 0 <= ref_x < self.width - w and 0 <= ref_y < self.height - h:
ref_blk = self.ref_yuv[ref_y:ref_y + h, ref_x:ref_x + w, :]
else:
ref_blk = np.zeros((h, w, 3))
return metric(cur_blk, ref_blk)
"""
distortion of motion field
"""
def distortion(self, mask=None, metric=MSE):
loss = 0
count = 0
for i in xrange(self.num_row):
for j in xrange(self.num_col):
if mask is not None and mask[i, j]:
continue
loss += self.block_dist(i, j, self.mf[i, j], metric)
count += 1
return loss / count
"""evaluation compare the difference with ground truth"""
def motion_field_evaluation(self, ground_truth):
loss = 0
count = 0
gt = ground_truth.mf
mask = ground_truth.mask
for i in xrange(self.num_row):
for j in xrange(self.num_col):
if mask is not None and mask[i][j]:
continue
loss += LA.norm(gt[i, j] - self.mf[i, j])
count += 1
return loss / count
"""render the motion field"""
def show(self, ground_truth=None, size=10):
cur_mf = drawMF(self.cur_f, self.blk_sz, self.mf)
if ground_truth is None:
n_row = 1
else:
gt_mf = drawMF(self.cur_f, self.blk_sz, ground_truth)
n_row = 2
plt.figure(figsize=(n_row * size, size * self.height / self.width))
plt.subplot(1, n_row, 1)
plt.imshow(cur_mf)
plt.title('Estimated Motion Field')
if ground_truth is not None:
plt.subplot(1, n_row, 2)
plt.imshow(gt_mf)
plt.title('Ground Truth')
plt.tight_layout()
plt.show()
@@ -0,0 +1,213 @@
#!/usr/bin/env python
# coding: utf-8
import numpy as np
import numpy.linalg as LA
from Util import MSE
from MotionEST import MotionEST
"""Search & Smooth Model with Adapt Weights"""
class SearchSmoothAdapt(MotionEST):
"""
Constructor:
cur_f: current frame
ref_f: reference frame
blk_sz: block size
wnd_size: search window size
beta: neigbor loss weight
max_iter: maximum number of iterations
metric: metric to compare the blocks distrotion
"""
def __init__(self, cur_f, ref_f, blk_size, search, max_iter=100):
self.search = search
self.max_iter = max_iter
super(SearchSmoothAdapt, self).__init__(cur_f, ref_f, blk_size)
"""
get local diffiencial of refernce
"""
def getRefLocalDiff(self, mvs):
m, n = self.num_row, self.num_col
localDiff = [[] for _ in xrange(m)]
blk_sz = self.blk_sz
for r in xrange(m):
for c in xrange(n):
I_row = 0
I_col = 0
#get ssd surface
count = 0
center = self.cur_yuv[r * blk_sz:(r + 1) * blk_sz,
c * blk_sz:(c + 1) * blk_sz, 0]
ty = np.clip(r * blk_sz + int(mvs[r, c, 0]), 0, self.height - blk_sz)
tx = np.clip(c * blk_sz + int(mvs[r, c, 1]), 0, self.width - blk_sz)
target = self.ref_yuv[ty:ty + blk_sz, tx:tx + blk_sz, 0]
for y, x in {(ty - blk_sz, tx), (ty + blk_sz, tx)}:
if 0 <= y < self.height - blk_sz and 0 <= x < self.width - blk_sz:
nb = self.ref_yuv[y:y + blk_sz, x:x + blk_sz, 0]
I_row += np.sum(np.abs(nb - center)) - np.sum(
np.abs(target - center))
count += 1
I_row //= (count * blk_sz * blk_sz)
count = 0
for y, x in {(ty, tx - blk_sz), (ty, tx + blk_sz)}:
if 0 <= y < self.height - blk_sz and 0 <= x < self.width - blk_sz:
nb = self.ref_yuv[y:y + blk_sz, x:x + blk_sz, 0]
I_col += np.sum(np.abs(nb - center)) - np.sum(
np.abs(target - center))
count += 1
I_col //= (count * blk_sz * blk_sz)
localDiff[r].append(
np.array([[I_row * I_row, I_row * I_col],
[I_col * I_row, I_col * I_col]]))
return localDiff
"""
add smooth constraint
"""
def smooth(self, uvs, mvs):
sm_uvs = np.zeros(uvs.shape)
blk_sz = self.blk_sz
for r in xrange(self.num_row):
for c in xrange(self.num_col):
nb_uv = np.array([0.0, 0.0])
for i, j in {(r - 1, c), (r + 1, c), (r, c - 1), (r, c + 1)}:
if 0 <= i < self.num_row and 0 <= j < self.num_col:
nb_uv += uvs[i, j] / 6.0
else:
nb_uv += uvs[r, c] / 6.0
for i, j in {(r - 1, c - 1), (r - 1, c + 1), (r + 1, c - 1),
(r + 1, c + 1)}:
if 0 <= i < self.num_row and 0 <= j < self.num_col:
nb_uv += uvs[i, j] / 12.0
else:
nb_uv += uvs[r, c] / 12.0
ssd_nb = self.block_dist(r, c, self.blk_sz * nb_uv)
mv = mvs[r, c]
ssd_mv = self.block_dist(r, c, mv)
alpha = (ssd_nb - ssd_mv) / (ssd_mv + 1e-6)
M = alpha * self.localDiff[r][c]
P = M + np.identity(2)
inv_P = LA.inv(P)
sm_uvs[r, c] = np.dot(inv_P, nb_uv) + np.dot(
np.matmul(inv_P, M), mv / blk_sz)
return sm_uvs
def block_matching(self):
self.search.motion_field_estimation()
def motion_field_estimation(self):
self.localDiff = self.getRefLocalDiff(self.search.mf)
#get matching results
mvs = self.search.mf
#add smoothness constraint
uvs = mvs / self.blk_sz
for _ in xrange(self.max_iter):
uvs = self.smooth(uvs, mvs)
self.mf = uvs * self.blk_sz
"""Search & Smooth Model with Fixed Weights"""
class SearchSmoothFix(MotionEST):
"""
Constructor:
cur_f: current frame
ref_f: reference frame
blk_sz: block size
wnd_size: search window size
beta: neigbor loss weight
max_iter: maximum number of iterations
metric: metric to compare the blocks distrotion
"""
def __init__(self, cur_f, ref_f, blk_size, search, beta, max_iter=100):
self.search = search
self.max_iter = max_iter
self.beta = beta
super(SearchSmoothFix, self).__init__(cur_f, ref_f, blk_size)
"""
get local diffiencial of refernce
"""
def getRefLocalDiff(self, mvs):
m, n = self.num_row, self.num_col
localDiff = [[] for _ in xrange(m)]
blk_sz = self.blk_sz
for r in xrange(m):
for c in xrange(n):
I_row = 0
I_col = 0
#get ssd surface
count = 0
center = self.cur_yuv[r * blk_sz:(r + 1) * blk_sz,
c * blk_sz:(c + 1) * blk_sz, 0]
ty = np.clip(r * blk_sz + int(mvs[r, c, 0]), 0, self.height - blk_sz)
tx = np.clip(c * blk_sz + int(mvs[r, c, 1]), 0, self.width - blk_sz)
target = self.ref_yuv[ty:ty + blk_sz, tx:tx + blk_sz, 0]
for y, x in {(ty - blk_sz, tx), (ty + blk_sz, tx)}:
if 0 <= y < self.height - blk_sz and 0 <= x < self.width - blk_sz:
nb = self.ref_yuv[y:y + blk_sz, x:x + blk_sz, 0]
I_row += np.sum(np.abs(nb - center)) - np.sum(
np.abs(target - center))
count += 1
I_row //= (count * blk_sz * blk_sz)
count = 0
for y, x in {(ty, tx - blk_sz), (ty, tx + blk_sz)}:
if 0 <= y < self.height - blk_sz and 0 <= x < self.width - blk_sz:
nb = self.ref_yuv[y:y + blk_sz, x:x + blk_sz, 0]
I_col += np.sum(np.abs(nb - center)) - np.sum(
np.abs(target - center))
count += 1
I_col //= (count * blk_sz * blk_sz)
localDiff[r].append(
np.array([[I_row * I_row, I_row * I_col],
[I_col * I_row, I_col * I_col]]))
return localDiff
"""
add smooth constraint
"""
def smooth(self, uvs, mvs):
sm_uvs = np.zeros(uvs.shape)
blk_sz = self.blk_sz
for r in xrange(self.num_row):
for c in xrange(self.num_col):
nb_uv = np.array([0.0, 0.0])
for i, j in {(r - 1, c), (r + 1, c), (r, c - 1), (r, c + 1)}:
if 0 <= i < self.num_row and 0 <= j < self.num_col:
nb_uv += uvs[i, j] / 6.0
else:
nb_uv += uvs[r, c] / 6.0
for i, j in {(r - 1, c - 1), (r - 1, c + 1), (r + 1, c - 1),
(r + 1, c + 1)}:
if 0 <= i < self.num_row and 0 <= j < self.num_col:
nb_uv += uvs[i, j] / 12.0
else:
nb_uv += uvs[r, c] / 12.0
mv = mvs[r, c] / blk_sz
M = self.localDiff[r][c]
P = M + self.beta * np.identity(2)
inv_P = LA.inv(P)
sm_uvs[r, c] = np.dot(inv_P, self.beta * nb_uv) + np.dot(
np.matmul(inv_P, M), mv)
return sm_uvs
def block_matching(self):
self.search.motion_field_estimation()
def motion_field_estimation(self):
#get local structure
self.localDiff = self.getRefLocalDiff(self.search.mf)
#get matching results
mvs = self.search.mf
#add smoothness constraint
uvs = mvs / self.blk_sz
for _ in xrange(self.max_iter):
uvs = self.smooth(uvs, mvs)
self.mf = uvs * self.blk_sz
@@ -0,0 +1,38 @@
#!/usr/bin/env python
# coding: utf-8
import numpy as np
import numpy.linalg as LA
import matplotlib.pyplot as plt
from scipy.ndimage import filters
from PIL import Image, ImageDraw
def MSE(blk1, blk2):
return np.mean(
LA.norm(
np.array(blk1, dtype=np.int) - np.array(blk2, dtype=np.int), axis=2))
def drawMF(img, blk_sz, mf):
img_rgba = img.convert('RGBA')
mf_layer = Image.new(mode='RGBA', size=img_rgba.size, color=(0, 0, 0, 0))
draw = ImageDraw.Draw(mf_layer)
width = img_rgba.size[0]
height = img_rgba.size[1]
num_row = height // blk_sz
num_col = width // blk_sz
for i in xrange(num_row):
left = (0, i * blk_sz)
right = (width, i * blk_sz)
draw.line([left, right], fill=(0, 0, 255, 255))
for j in xrange(num_col):
up = (j * blk_sz, 0)
down = (j * blk_sz, height)
draw.line([up, down], fill=(0, 0, 255, 255))
for i in xrange(num_row):
for j in xrange(num_col):
center = (j * blk_sz + 0.5 * blk_sz, i * blk_sz + 0.5 * blk_sz)
"""mf[i,j][0] is the row shift and mf[i,j][1] is the column shift In PIL coordinates, head[0] is x (column shift) and head[1] is y (row shift)."""
head = (center[0] + mf[i, j][1], center[1] + mf[i, j][0])
draw.line([center, head], fill=(255, 0, 0, 255))
return Image.alpha_composite(img_rgba, mf_layer)