init commit

Author: yifan-fanyi

BinaryTree.py

@@ -0,0 +1,70 @@
+import copy
+import numpy as np
+
+class BinaryTree:
+    @staticmethod
+    def saver(idx):
+        def idxConvert(idx):
+            for i in range(1, len(idx)):
+                r = idx[i].shape[1] // idx[i-1].shape[1]
+                for k in range(idx[i].shape[0]):
+                    for ii in range(idx[i].shape[1]):
+                        for jj in range(idx[i].shape[2]):
+                            if idx[i-1][k,ii//r,jj//r,0] < 0.5:
+                                idx[i][k, ii, jj, 0] = -2 
+            return idx
+        stream = []
+        idx = idxConvert(idx)
+        for i in range(0, len(idx)):
+            for k in range(idx[i].shape[0]):
+                for ii in range(idx[i].shape[1]):
+                    for jj in range(idx[i].shape[2]):
+                        if idx[i][k,ii,jj,0] > -1:
+                            stream.append(idx[i][k,ii,jj,0])
+        return stream
+    @staticmethod
+    def loader(stream, S):
+        idx, ct = [], 0
+        for i in range(len(S)):
+            idx.append(np.zeros(S[i]).astype('int16'))
+        for k in range(idx[0].shape[0]):
+            for ii in range(idx[0].shape[1]):
+                for jj in range(idx[0].shape[2]):
+                    idx[0][k,ii,jj,0] = stream[ct]
+                    ct += 1
+        for i in range(1, len(idx)):
+            r = idx[i].shape[1] // idx[i-1].shape[1]
+            for k in range(idx[i].shape[0]):
+                for ii in range(idx[i].shape[1]):
+                    for jj in range(idx[i].shape[2]):
+                        if idx[i-1][k,ii//r,jj//r,0] == 0:
+                            idx[i][k,ii,jj,0] = 0
+                        else:
+                            idx[i][k,ii,jj,0] = stream[ct]
+                            ct += 1
+        return idx
+    @staticmethod
+    def checker(idx, ref_idx):
+        for i in range(len(idx)):
+            s = np.sum(np.abs(idx[i]-ref_idx[i]))
+            assert(s == 0), "Error!"
+        
+    @staticmethod        
+    def UnitTest():
+        def gen_data(S):
+            idx = [np.random.randint(2, size=S[0])]
+            for i in range(1,len(S)):
+                r = S[i][1] // S[i-1][1]
+                idx.append(np.random.randint(2, size=S[i]))
+                for k in range(idx[i].shape[0]):
+                    for ii in range(idx[i].shape[1]):
+                        for jj in range(idx[i].shape[2]):
+                            if idx[i-1][k,ii//r,jj//r, 0] == 0:
+                                idx[i][k,ii,jj,0] = 0
+            return idx
+        S = [(5,4,4,1), (5,8,8,1),(5,16,16,1),(5,64,64,1)]
+        idx = gen_data(S)
+        stream = BinaryTree.saver(copy.deepcopy(idx))
+        idxd = BinaryTree.loader(stream, S)
+        BinaryTree.checker(idxd, idx)
+        print("UnitTest Pass")

EntropyCoding.py

@@ -0,0 +1,90 @@
+import numpy as np
+import copy 
+from Huffman import Huffman
+from BinaryTree import BinaryTree
+
+class EntropyCoding():
+    def __init__(self, par):
+        self.par = par
+        self.EC_list = {}
+
+    def idx_select(self, label, idx=None):
+        if idx is None:
+            return label
+        tmp = []
+        r = label.shape[1] // idx.shape[1]
+        for i in range(label.shape[0]):
+            for j in range(label.shape[1]):
+                for k in range(label.shape[2]):
+                    if idx[i, j//r, k//r] == 1:
+                        tmp.append(label[i,j,k, 0])
+        return np.array(tmp).astype('int32')
+
+    def inverse_idx_select(self, tmp, idx, S):
+        label = np.zeros(S)
+        r = label.shape[1] // idx.shape[1]
+        ct = 0
+        for i in range(label.shape[0]):
+            for j in range(label.shape[1]):
+                for k in range(label.shape[2]):
+                    if idx[i, j//r, k//r] == 1:
+                        label[i,j,k,0] = tmp[ct]
+                        ct += 1
+        return label
+
+    def fit(self, save):
+        label_list, idx_list = save['label'], save['idx']
+        for i in range(1, self.par['n_hop']+1):
+            label, idx = label_list['hop'+str(i)], idx_list['hop'+str(i)]
+            tmp_list = [Huffman().fit(label[0])]
+            for j in range(1, len(self.par['win']['hop'+str(i)])):
+                tmp_list.append(Huffman().fit(self.idx_select(label[j], idx[j-1])))
+            self.EC_list['hop'+str(i)] = tmp_list
+        return self
+
+    def refit(self, save, par):
+        label_list, idx_list = save['label'], save['idx']
+        for i in range(1, self.par['n_hop']+1):
+            label, idx = label_list['hop'+str(i)], idx_list['hop'+str(i)]
+            if par['hop'+str(i)][0][0] == True:
+                self.EC_list['hop'+str(i)][0] = Huffman().fit(label[0])
+            for j in range(1, len(self.par['win']['hop'+str(i)])):
+                if par['hop'+str(i)][j][0] == True:
+                    self.EC_list['hop'+str(i)][j] = Huffman().fit(self.idx_select(label[j], idx[j-1]))
+        return self
+
+    def encode(self, save, S=1024):
+        stream = {'idx':{}, 'label':{}}
+        label_list, idx_list = save['label'], save['idx']
+        n_img = label_list['hop1'][0].shape[0]
+        for i in range(1, self.par['n_hop']+1):
+            label, idx = label_list['hop'+str(i)], idx_list['hop'+str(i)]
+            st = BinaryTree.saver(idx)
+            stream['idx']['hop'+str(i)] = st
+            print('Hop-%d'%(i))
+            print('   save tree:%1.8f'%(len(st)/(n_img*S**2)))
+            tmp_st = [self.EC_list['hop'+str(i)][0].encode(label[0])]
+            print('  Level-0 Huffman %1.8f bpp'%((len(tmp_st[-1])) / (n_img*S**2)))
+            for j in range(1, len(self.par['win']['hop'+str(i)])):
+                st = self.EC_list['hop'+str(i)][j].encode(self.idx_select(label[j], idx[j-1]))
+                tmp_st.append(st)
+                print('  Level-%d Huffman %1.8f bpp'%(j, len(st)/(n_img*S**2)))
+            stream['label']['hop'+str(i)] = tmp_st
+        return stream
+
+    def decode(self, stream):
+        save = {'label':{}, 'idx':{}}
+        for i in range(1, self.par['n_hop']+1):
+            if len(self.par['shape']['hop'+str(i)]) > 1:
+                save['idx']['hop'+str(i)] = BinaryTree.loader(stream['idx']['hop'+str(i)], self.par['shape']['hop'+str(i)][:-1])
+            else:
+                save['idx']['hop'+str(i)] = []
+            tmp, _ = self.EC_list['hop'+str(i)][0].decode(stream['label']['hop'+str(i)][0])
+            tmp = tmp.reshape(self.par['shape']['hop'+str(i)][0])
+            label = [tmp]
+            for j in range(1, len(self.par['win']['hop'+str(i)])):
+                tmp, _ = self.EC_list['hop'+str(i)][j].decode(stream['label']['hop'+str(i)][j])
+                tmp = self.inverse_idx_select(tmp, save['idx']['hop'+str(i)][j-1], self.par['shape']['hop'+str(i)][j])
+                label.append(tmp)
+            save['label']['hop'+str(i)] = label
+        return save

GIC.py

@@ -0,0 +1,66 @@
+import numpy as np
+import os
+import pickle
+import copy 
+import json
+import warnings
+warnings.filterwarnings("ignore")
+os.environ["CUDA_VISIBLE_DEVICES"]="6,7"
+from load_img import Load_from_Folder, Load_Images
+from evaluate import Time, MSE, PSNR
+from MBMBVQ import MBMBVQ
+from EntropyCoding import EntropyCoding
+
+class GIC():
+    def __init__(self, par):
+        self.MBMBVQ = MBMBVQ(par)
+        self.EC = EntropyCoding(par)
+    def change_n_img(self, n_img):
+        for i in range(1, self.EC.par['n_hop']+1):
+            for j in range(len(self.EC.par['shape']['hop'+str(i)])):
+                self.EC.par['shape']['hop'+str(i)][j][0] = n_img
+    @Time
+    def fit(self, Y):
+        self.change_n_img(Y.shape[0])
+        self.MBMBVQ.fit(copy.deepcopy(Y))
+        save = self.MBMBVQ.encode(copy.deepcopy(Y))
+        self.EC.fit(save)
+        return self
+    @Time
+    def refit(self, Y, par):
+        self.change_n_img(Y.shape[0])
+        self.MBMBVQ.refit(copy.deepcopy(Y), par)
+        save = self.MBMBVQ.encode(copy.deepcopy(Y))
+        self.EC.refit(save, par)
+        return self
+    @Time
+    def encode(self, Y):
+        self.change_n_img(Y.shape[0])
+        save = self.MBMBVQ.encode(Y)
+        stream = self.EC.encode(save, S=Y.shape[1])
+        return stream, save['DC'], save
+    
+    @Time
+    def decode(self, stream, DC):
+        save = self.EC.decode(stream)
+        save['DC'] = DC
+        iY = self.MBMBVQ.decode(save)
+        return iY
+
+if __name__ == "__main__":
+    with open('./test_data/test_par1.json', 'r') as f:
+        par = json.load(f)
+    gic = GIC_Y(par)
+
+    Y_list = Load_from_Folder(folder='./test_data/', color='YUV', ct=-1)
+    Y = np.array(Y_list)[:,:,:,:1]
+
+    gic.fit(Y)
+
+    stream, dc = gic.encode(Y)
+
+    iY = gic.decode(stream, dc)
+    print('MSE=%5.3f, PSNR=%3.5f'%(MSE(Y, iY), PSNR(Y, iY)))
+    print('------------------')
+    print(" * Ref result: "+'MSE=129.342, PSNR=27.01340')
+ 

Huffman.py

@@ -0,0 +1,54 @@
+# @yifan
+#
+import numpy as np
+import huffman
+
+class Huffman():
+    def __init__(self):
+        self.hist = None
+        self.dict = {}
+        self.inv_dict = {}
+        self.version = '2021.05.20'
+
+    def Hist(self, x, bins=64):
+        x = x.reshape(-1).astype('int32')
+        res = np.zeros(bins)
+        for i in range(bins):
+            res[i] = len(x[x == i])
+        return res   
+
+    def make_dict(self):
+        tmp = []
+        for i in range(len(self.hist)):
+            tmp.append((str(i), self.hist[i]))
+        self.dict = huffman.codebook(tmp)
+        self.inv_dict = {v: k for k, v in self.dict.items()}
+        
+    def fit(self, X):
+        bins = len(np.unique(X.astype('int16')))
+        self.hist = self.Hist(X, bins=bins)
+        self.make_dict()
+        return self
+        
+    def encode(self, X):
+        X = X.reshape(-1).astype('int32')
+        stream = ''
+        for i in range(len(X)):
+            stream += self.dict[str(X[i])]
+        return stream
+
+    def decode(self, stream, start=0, size=-1):
+        if size < 0:
+            size = len(stream)
+        dX, last, ct = [], start, 0
+        for i in range(start, len(stream)):
+            if stream[last:i] in self.inv_dict:
+                dX.append((int)(self.inv_dict[stream[last:i]]))
+                last = i
+                ct += 1
+            if ct >= size:
+                break
+        if ct < size:
+            dX.append((int)(self.inv_dict[stream[last:]]))
+            last = len(stream)
+        return np.array(dX), last

LANCZOS.py

@@ -0,0 +1,32 @@
+import numpy as np
+from PIL import Image
+
+class LANCZOS:
+    @staticmethod
+    def split(X, ratio=2):
+        DC_L = LANCZOS.resample(X, ratio)
+        X_inv_L = LANCZOS.resample(DC_L, 1/ratio)
+        X_inv_L = np.round(X_inv_L)
+        AC = X - X_inv_L
+        return DC_L, AC
+    @staticmethod
+    def inv_split(DC, AC, ratio=2):
+        DC = LANCZOS.resample(DC, 1/ratio)
+        return DC+AC
+    def inv_split_moreRef(DC, AC, k, i, j, win, ratio=2):
+        DC = LANCZOS.resample(DC, 1/ratio)
+        DC = Shrink(DC, win=win)
+        return DC[k, i, j].reshape(-1)+AC.reshape(-1)
+    @staticmethod
+    def resample(X, ratio=2):
+        image_list = []
+        for i in range(X.shape[0]):
+            tmp = []
+            for k in range(X.shape[-1]):
+                size1 = int(X.shape[1]/ratio)
+                size2 = int(X.shape[2]/ratio)
+                image_tmp = Image.fromarray(X[i,:,:,k]).resize(size=(size1, size2), resample=Image.LANCZOS)
+                tmp.append(np.array(image_tmp).reshape(1, size1, size2, 1))
+            image_list.append(np.concatenate(tmp, axis=-1))
+        output = np.concatenate(image_list, axis=0)
+        return output

LLSR.py

@@ -0,0 +1,34 @@
+# 2020.01.28
+# @yifan
+#
+import numpy as np
+
+class LLSR():
+    def __init__(self, onehot=True, normalize=False):
+        self.onehot = onehot
+        self.normalize = normalize
+        self.weight = []
+
+    def fit(self, X, Y):
+        if self.onehot == True:
+            Y = np.eye(len(np.unique(Y)))[Y.reshape(-1)]
+        A = np.ones((X.shape[0], 1))
+        X = np.concatenate((X, A), axis=1)
+        self.weight, _, _, _ = np.linalg.lstsq(X, Y, rcond=None)
+        return self
+
+    def predict(self, X):
+        pred = self.predict_proba(X)
+        return np.argmax(pred, axis=1)
+
+    def predict_proba(self, X):
+        A = np.ones((X.shape[0], 1))
+        X = np.concatenate((X, A), axis=1)
+        pred = np.matmul(X, self.weight)
+        if self.normalize == True:
+            pred = (pred - np.min(pred, axis=1, keepdims=True))/ np.sum((pred - np.min(pred, axis=1, keepdims=True) + 1e-15), axis=1, keepdims=True)
+        return pred
+
+    def score(self, X, Y):
+        pred = self.predict(X)
+        return accuracy_score(Y, pred)