How to use is_uint method in pandera

Best Python code snippet using pandera_python

MPM.py

Source:MPM.py

1# 2021.03.242# @yifan3# most probable mode for bitstream encoding4# codeword should be sorted maybe in a similar mannar like HEVC's angular prediction5# or it should be sorted based on the context? smooth -> texture6# based on statics, we got the probability 7import numpy as np8import copy9from myIO import bits2int, int2bits10from mylearner import myLearner11# encode uses idx with shape (H, W)12# if index in MPM_list, send the index in MPM_list13# else, send 3 and the index in the codeword_list which contains all other codeword expect those in MPM_list14class MPM():15    def __init__(self, n_codeword, MPM_init_list):16        self.n_MPM = len(MPM_init_list)17        self.n_codeword = n_codeword18        self.length = (int)(np.log2(n_codeword-4))+119        self.MPM_init_list = MPM_init_list20        #self.MPM_init_list.sort()21        self.MPM_list = self.MPM_init_list22        self.codeword_list = []  # code word not in current MPM_list23        self.stream = ''         # encoding bit stream <string> with 0/1 in it24        self.ct = 0              # decoding bit position indicater25                    26    def get_MPM(self, prev_idx):27        ct = 228        self.MPM_list = copy.deepcopy(self.MPM_init_list)29        for i in range(len(prev_idx)):30            if prev_idx[i] not in self.MPM_list:31                self.MPM_list[ct] = prev_idx[i]32                ct -= 1 33        self.MPM_list.sort()34        self.codeword_list = []35        for i in range(self.n_codeword):36            if i not in self.MPM_list:37                self.codeword_list.append(i)38                39    def encode_one_idx(self, prev_idx, cur_idx):40        self.get_MPM(prev_idx)41        for i in range(len(self.MPM_list)):42            if self.MPM_list[i] == cur_idx:43                return i, i44        for i in range(len(self.codeword_list)):45            if self.codeword_list[i] == cur_idx:46                return i, 347        assert (False), 'err'48    def get_prev(self, idx, i, j):49        if i < 0 or j < 0 or i >= idx.shape[0] or j >= idx.shape[1]:50            return self.MPM_init_list[0]51        else:52            return idx[i, j]53    54    def encode(self, idx):55        self.stream = ''56        idx = idx.astype('int16')57        for i in range(idx.shape[0]):58            for j in range(idx.shape[1]):59                prev_idx = [self.get_prev(idx, i-1, j-1),60                            self.get_prev(idx, i-1, j),61                            self.get_prev(idx, i, j-1)]62                val, mode = self.encode_one_idx(prev_idx, idx[i,j])63                self.stream += int2bits(mode, 2, is_uint=True, return_string=True)64                if mode == 3:                    65                    self.stream += int2bits(val, self.length, is_uint=True, return_string=True)66        return self.stream67    68    def decode_one_idx(self, prev_idx, mode):69        self.get_MPM(prev_idx)70        if mode == 3:71            idx = bits2int(self.stream[self.ct:self.ct+self.length], is_uint=True)72            self.ct += self.length73            return self.codeword_list[idx]74        else:75            return self.MPM_list[mode]76            77    def decode(self, stream, H, W):78        self.stream = stream79        idx = np.zeros((H, W))80        self.ct = 081        for i in range(idx.shape[0]):82            for j in range(idx.shape[1]):83                prev_idx = [self.get_prev(idx, i-1, j-1),84                            self.get_prev(idx, i-1, j),85                            self.get_prev(idx, i, j-1)]86                mode = bits2int(self.stream[self.ct:self.ct+2], is_uint=True)87                self.ct += 288                idx[i,j] = self.decode_one_idx(prev_idx, mode)89        return idx.astype('int16')90    91# fit uses idx with shape (K, H, W)92# encode uses idx with shape (H, W)93# apply ML to the idea,94# use previous index as feature and current index as label to train the learner95# if predict correctly, send 0, else send 1 and the corret index96# fitting time is too large97class ML_MPM(MPM):98    def __init__(self, learner, n_codeword, MPM_init_list):99        super().__init__(n_codeword, MPM_init_list)100        self.learner = myLearner(learner, n_codeword)101        self.length = (int)(np.log2(n_codeword-1))+1102        103    def fit(self, idx):104        prev, cur = [], []105        idx = idx.astype('int16')106        for k in range(idx.shape[0]):107            for i in range(idx.shape[1]):108                for j in range(idx.shape[2]):109                    prev.append( [self.get_prev(idx[k], i-1, j-1),110                                  self.get_prev(idx[k], i-1, j),111                                  self.get_prev(idx[k], i, j-1),112                                  self.get_prev(idx[k], i-1, j+1)])113                    cur.append(idx[k,i,j]) 114        prev = np.array(prev).reshape(-1, 4)115        cur = np.array(cur).reshape(-1,1)116        self.learner.fit(prev, cur)117        print('fit score:', self.learner.score(prev, cur))118        print('fit score top3',self.learner.topNscore(prev, cur, 3))119        return self120        121    def encode_one_idx(self, prev_idx, cur_idx):122        prev_idx = np.array(prev_idx).reshape(1, len(prev_idx))123        pred = self.learner.predict(prev_idx).reshape(-1)124        if pred[0] == cur_idx:125            return 0, 0126        else:127            return cur_idx, 1    128    129    def encode(self, idx):130        self.stream = ''131        idx = idx.astype('int16')132        x, y = [], []133        for i in range(idx.shape[0]):134            for j in range(idx.shape[1]):135                prev_idx = [self.get_prev(idx, i-1, j-1),136                            self.get_prev(idx, i-1, j),137                            self.get_prev(idx, i, j-1),138                            self.get_prev(idx, i-1, j+1)]139                x.append(prev_idx)140                y.append(idx[i,j])141        x, y = np.array(x), np.array(y).reshape(-1)142        px = self.learner.predict(x).reshape(-1)143        print('test score',self.learner.score(x, y))144        print('test score top3',self.learner.topNscore(x, y, 3))145        for i in range(len(px)):146            if px[i] == y[i]:147                mode = 0148            else:149                mode = 1150            self.stream += int2bits(mode, 1, is_uint=True, return_string=True)151            if mode == 1:                    152                self.stream += int2bits(y[i], self.length, is_uint=True, return_string=True)153        return self.stream154    155    def decode_one_idx(self, prev_idx):156        mode = bits2int(self.stream[self.ct:self.ct+1], is_uint=True)157        self.ct += 1158        if mode == 1:159            idx = bits2int(self.stream[self.ct:self.ct+self.length], is_uint=True)160            self.ct += self.length161            return idx162        else:163            pred = self.learner.predict(np.array(prev_idx).reshape(1,-1))164            return pred165            166    def decode(self, stream, H, W, raw_idx=None):167        self.stream = stream168        idx = np.zeros((H, W))169        self.ct = 0170        prev_idx = []171        for i in range(idx.shape[0]):172            for j in range(idx.shape[1]):173                prev_idx = [self.get_prev(idx, i-1, j-1),174                            self.get_prev(idx, i-1, j),175                            self.get_prev(idx, i, j-1),176                            self.get_prev(idx, i-1, j+1)]177                idx[i,j] = self.decode_one_idx(prev_idx)178                if raw_idx is not None:179                    assert(idx[i,j] == raw_idx[i,j]), 'Decoding Error!'180        return idx.astype('int16')181class ML_MPM3(ML_MPM):182    def __init__(self, learner, n_codeword, MPM_init_list):183        super().__init__(learner, n_codeword, MPM_init_list)184        self.length = (int)(np.log2(n_codeword-4))+1185    def encode(self, idx):186        self.stream = ''187        idx = idx.astype('int16')188        x, y = [], []189        for i in range(idx.shape[0]):190            for j in range(idx.shape[1]):191                prev_idx = [self.get_prev(idx, i-1, j-1),192                            self.get_prev(idx, i-1, j),193                            self.get_prev(idx, i, j-1),194                            self.get_prev(idx, i-1, j+1)]195                x.append(prev_idx)196                y.append(idx[i,j])197        x, y = np.array(x), np.array(y).reshape(-1)198        px = self.learner.predict_proba(x)199        idx = np.argsort(px, axis=1)200        self.MPM_list = idx[:, -3:]201        print('test score',self.learner.score(x, y))202        print('test score top3',self.learner.topNscore(x, y, 3))203        for i in range(len(self.MPM_list)):204            tmp = copy.deepcopy(self.MPM_list[i])205            tmp.sort()206            mode, c = 3, 0207            for j in range(len(tmp)):208                if tmp[j] == y[i]:209                    mode = j210                if y[i] > tmp[j]:211                    c += 1212            self.stream += int2bits(mode, 2, is_uint=True, return_string=True)213            if mode == 3:                  214                self.stream += int2bits(y[i]-c, self.length, is_uint=True, return_string=True)215        return self.stream216    217    def decode_one_idx(self, prev_idx):218        mode = bits2int(self.stream[self.ct:self.ct+2], is_uint=True)219        self.ct += 2220        prob = self.learner.predict_proba(np.array(prev_idx).reshape(1,-1)).reshape(-1)221        pred = np.argsort(prob)[-3:]222        pred.sort()223        if mode == 3:224            codeword = []225            for i in range(self.n_codeword):226                if i not in pred:227                    codeword.append(i)228            idx = bits2int(self.stream[self.ct:self.ct+self.length], is_uint=True)229            self.ct += self.length230            return codeword[idx]231        else:232            return pred[mode]233        234    def decode(self, stream, H, W, raw_idx=None):235        self.stream = stream236        idx = np.zeros((H, W))237        self.ct = 0238        for i in range(idx.shape[0]):239            for j in range(idx.shape[1]):240                prev_idx = [self.get_prev(idx, i-1, j-1),241                            self.get_prev(idx, i-1, j),242                            self.get_prev(idx, i, j-1),243                            self.get_prev(idx, i-1, j+1)]244                idx[i,j] = self.decode_one_idx(prev_idx)245                if raw_idx is not None:246                    assert(idx[i,j] == raw_idx[i,j]), 'Decoding Error!'247        return idx.astype('int16')248def Check(idx, didx):249    for i in range(idx.shape[0]):250        for j in range(idx.shape[1]):251            if idx[i,j] != didx[i,j]:252                err_list = 'idx: '+str(idx[i,j])+', become: '+str(didx[i,j])+', pos: ('+str(i)+','+str(j)+')'253                print('Error!')254                print(err_list)255                assert (False), "Not Match!"256class ML_MPMa(MPM):257    def __init__(self, learner, n_codeword, MPM_init_list):258        super().__init__(n_codeword, MPM_init_list)259        self.learner = myLearner(learner, n_codeword)260        self.length = (int)(np.log2(n_codeword-5))+1261        262    def fit(self, idx):263        prev, cur = [], []264        idx = idx.astype('int16')265        for k in range(idx.shape[0]):266            for i in range(idx.shape[1]):267                for j in range(idx.shape[2]):268                    prev.append( [self.get_prev(idx[k], i-1, j-1),269                                  self.get_prev(idx[k], i-1, j),270                                  self.get_prev(idx[k], i, j-1),271                                  self.get_prev(idx[k], i-1, j+1)])272                    cur.append(idx[k,i,j]) 273        prev = np.array(prev).reshape(-1, 4)274        cur = np.array(cur).reshape(-1,1)275        self.learner.fit(prev, cur)276        print('fit score:', self.learner.score(prev, cur))277        print('fit score top4',self.learner.topNscore(prev, cur, 4))278        return self  279        280    def encode_fail(self, sprob, cur_idx):281        a = sprob[-4:-1]282        a.sort()283        b = sprob[:-4]284        b.sort()  285        mode = 3286        for i in range(len(a)):287            if a[i] == cur_idx:288                mode = i289        self.stream += int2bits(mode, 2, is_uint=True, return_string=True)290        if mode == 3:291            for i in range(len(b)):292                if cur_idx == b[i]:293                    self.stream += int2bits(i, self.length, is_uint=True, return_string=True)294                    return295            assert (False), 'Error, missing codeword!'296        else:297            pass298    def encode(self, idx):299        self.stream = ''300        idx = idx.astype('int16')301        x, y = [], []302        for i in range(idx.shape[0]):303            for j in range(idx.shape[1]):304                prev_idx = [self.get_prev(idx, i-1, j-1),305                            self.get_prev(idx, i-1, j),306                            self.get_prev(idx, i, j-1),307                            self.get_prev(idx, i-1, j+1)]308                x.append(prev_idx)309                y.append(idx[i,j])310        x, y = np.array(x), np.array(y).reshape(-1)311        prob = self.learner.predict_proba(x)312        sprob = np.argsort(prob, axis=1)313        px = sprob[:, -1].reshape(-1)314        print('test score',self.learner.score(x, y))315        print('test score top4',self.learner.topNscore(x, y, 4))316        for i in range(len(px)):317            if px[i] == y[i]:318                mode = 0319            else:320                mode = 1321            self.stream += int2bits(mode, 1, is_uint=True, return_string=True)322            if mode == 1:                    323                self.encode_fail(sprob[i], y[i])324        return self.stream325    def decode_fail(self, idx):326        mode = bits2int(self.stream[self.ct:self.ct+2], is_uint=True)327        self.ct += 2   328        if mode < 3:329            a = idx[-4:-1]330            a.sort()331            return a[mode]332        else:333            b = idx[:-4]334            b.sort()335            val =  bits2int(self.stream[self.ct:self.ct+self.length], is_uint=True)336            self.ct += self.length337            return b[val]338    def decode_one_idx(self, prev_idx):339        mode = bits2int(self.stream[self.ct:self.ct+1], is_uint=True)340        self.ct += 1341        prob = self.learner.predict_proba(np.array(prev_idx).reshape(1,-1)).reshape(-1)342        if mode == 1:343            idx = self.decode_fail(np.argsort(prob))344            return idx345        else:346            return np.argmax(prob)347            348    def decode(self, stream, H, W, raw_idx=None):349        self.stream = stream350        idx = np.zeros((H, W)).astype('int16')351        self.ct = 0352        prev_idx = []353        for i in range(idx.shape[0]):354            for j in range(idx.shape[1]):355                prev_idx = [self.get_prev(idx, i-1, j-1),356                            self.get_prev(idx, i-1, j),357                            self.get_prev(idx, i, j-1),358                            self.get_prev(idx, i-1, j+1)]359                idx[i,j] = self.decode_one_idx(prev_idx)360                if raw_idx is not None:361                    assert((int)(idx[i,j]) == raw_idx[i,j]), 'Decoding Error!'+str(i)+'_'+str(j)+'_'+str((int)(idx[i,j]))+'_'+str(raw_idx[i,j])...

common_verify.py

Source:common_verify.py

1r'''2>>> is_UInt(0)3True4>>> is_UInt(-1)5False6>>> is_UInt([])7False8>>> is_UInt(0, TypeError)9True10#doctest: +IGNORE_EXCEPTION_DETAIL11>>> is_UInt([], TypeError)12Traceback (most recent call last):13    ...14TypeError: <class 'list'> is not <class 'int'>15>>> is_pair(())16False17>>> is_pair((1, 2))18True19>>> is_Sequence([1, ''], TypeError)20True21>>> is_Sequence({})22False23>>> is_Sequence.of([1, ''], is_UInt)24False25>>> is_Sequence.of([1, 0], is_UInt)26True27>>> is_sorted_sequence([])28True29>>> is_sorted_sequence([None])30True31>>> is_sorted_sequence([1,2])32True33>>> is_sorted_sequence([1,1])34True35>>> is_sorted_sequence([1,2,3])36True37>>> is_sorted_sequence([2,1])38False39>>> is_sorted_sequence.of([1, 2], is_UInt)40True41>>> is_sorted_sequence(['', 'a'])42True43>>> is_sorted_sequence.of(['', 'a'], is_UInt)44False45>>> is_sorted_sequence.of([0, 'a'], is_UInt)46False47>>> is_sorted_sequence([0, 'a'])48Traceback (most recent call last):49    ...50TypeError: '<=' not supported between instances of 'int' and 'str'51>>> is_strict_sorted_sequence([2,2])52False53>>> is_strict_sorted_sequence([1,2])54True55>>> has_attrs([], attrs='__getitem__ index'.split())56True57>>> has_attrs([], attrs=[])58True59>>> has_attrs([], attrs=['xxxxxx'])60False61'''62__all__ = '''63    is_int64    is_UInt65    is_tuple66    is_pair67    is_str68    is_Hashable69    is_Sequence70    is_Set71    is_Mapping72    is_sorted_sequence73    is_strict_sorted_sequence74    has_attrs75    '''.split()76from .Verify import Verify77from .VerifyType import VerifyType, VerifyType__static78from .VerifyContainer import VerifyContainerType79from collections.abc import Sequence, Set, Mapping, Hashable80class is_int(VerifyType__static):81    types = int82class is_UInt(is_int):83    def _iter_verify_object_(self, obj):84        yield obj >= 0, lambda:'i < 0: {}'.format(obj)85        yield from super()._iter_verify_object_(obj)86class is_tuple(VerifyType__static):87    types = tuple88class is_pair(is_tuple):89    def _iter_verify_object_(self, obj):90        yield len(obj) == 2, lambda: 'len != 2: {}'.format(len(obj))91        yield from super()._iter_verify_object_(obj)92class is_str(VerifyType__static):93    types = str94class is_Hashable(VerifyType__static):95    types = Hashable96class is_Sequence(VerifyContainerType):97    types = Sequence98class is_Set(VerifyContainerType):99    types = Set100class is_Mapping(VerifyContainerType):101    types = Mapping102class is_sorted_sequence(is_Sequence):103    def _iter_verify_object_(self, obj):104        ls = obj105        prevs = iter(ls)106        succs = iter(ls)107        for _ in succs: break108        for i, prev, succ in zip(range(len(ls)), prevs, succs):109            yield (prev <= succ, lambda: 'not sorted sequence at {idx}: '110                'not ls[{idx}] <= ls[{idx1}]; '111                'ls[{idx}]={lsI!r}; ls[{idx1}]={lsI1!r}; ls={ls!r}'112                .format(idx=i, idx1=i+1, lsI=ls[i], lsI1=ls[i+1], ls=ls)113                )114        yield from super()._iter_verify_object_(obj)115class is_strict_sorted_sequence(is_Sequence):116    def _iter_verify_object_(self, obj):117        ls = obj118        prevs = iter(ls)119        succs = iter(ls)120        for _ in succs: break121        for i, prev, succ in zip(range(len(ls)), prevs, succs):122            yield (prev < succ, lambda: 'not sorted sequence at {idx}: '123                'not ls[{idx}] < ls[{idx1}]; '124                'ls[{idx}]={lsI!r}; ls[{idx1}]={lsI1!r}; ls={ls!r}'125                .format(idx=i, idx1=i+1, lsI=ls[i], lsI1=ls[i+1], ls=ls)126                )127        yield from super()._iter_verify_object_(obj)128class has_attrs(Verify):129    def __init__(self, *, attrs, **kwargs):130        attrs = self.attrs = tuple(sorted(set(attrs)))131        assert is_Sequence.of(attrs, is_str, TypeError)132        if not all(attr.isidentifier() for attr in attrs): raise ValueError133        super().__init__(**kwargs)134    def _iter_verify_object_(self, obj):135        for attr in self.attrs:136            yield (hasattr(obj, attr), lambda:137                'has not attr {!r}: {!r}'.format(attr, obj)138                )139        yield from super()._iter_verify_object_(obj)140if __name__ == "__main__":141    import doctest142    doctest.testmod()143    #doctest: +ELLIPSIS144    #doctest: +NORMALIZE_WHITESPACE145    #doctest: +IGNORE_EXCEPTION_DETAIL...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.