KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use maybe_decorate method in Slash

Best Python code snippet using slash

linqr.py

Source:linqr.py

...179        else:180            return np.vstack([np.atleast_2d(a) for a in arrs])181182    @staticmethod183    @maybe_decorate(nb.njit(nogil=True, parallel=False, cache=True), NUMBA_LEVEL >= 4)184    def _bld_lm_blks(blocks, valid_pts3d, pt3d_idxs, pose_idxs, rr, Jrb, Jrp, Jrl, idx_dtype, pose_size):185        # TODO: debug, some problem with logic present, results are poorer than previous method186        #       - However, not sure if worthwhile to fix, seems execution times are very similar187        #       - NUMBA_LEVEL == 3, seems best execution time -wise, also ok memory use188189        blk_i = 0190        for pt_i in np.arange(len(valid_pts3d), dtype=idx_dtype):191            if not valid_pts3d[pt_i]:192                continue193194            r_idxs = np.where(pt3d_idxs == pt_i)[0].astype(idx_dtype)195            blk_pose_idxs = pose_idxs[r_idxs]196            blk_np = len(r_idxs)197            m, n_b, n_p, n_l = blk_np * 2, Jrb.shape[1], blk_np * pose_size, 3198199            if blk_np < 2:200                # too few observations of this 3d-point201                valid_pts3d[pt_i] = False202                continue203204            if len(blocks) <= blk_i:205                blocks.append(ResidualBlock(r_idxs, blk_pose_idxs, m, n_b, n_p, n_l, pose_size))206207            blk = blocks[blk_i]208            blk.reset()209            blk_i += 1210211            for i, ri in enumerate(r_idxs):212                i = idx_dtype(i)213                for j in range(n_b):214                    j = idx_dtype(j)215                    blk.S[i*2,     j] = Jrb[ri*2,     j]216                    blk.S[i*2 + 1, j] = Jrb[ri*2 + 1, j]217218                for j in range(pose_size):219                    j = idx_dtype(j)220                    blk.S[i*2,     n_b + i * pose_size + j] = Jrp[ri*2,     blk_pose_idxs[i] + j]221                    blk.S[i*2 + 1, n_b + i * pose_size + j] = Jrp[ri*2 + 1, blk_pose_idxs[i] + j]222223                for j in range(n_l):224                    j = idx_dtype(j)225                    blk.S[i*2,     n_b + n_p + j] = Jrl[ri*2,     pt_i * 3 + j]226                    blk.S[i*2 + 1, n_b + n_p + j] = Jrl[ri*2 + 1, pt_i * 3 + j]227228                blk.S[i * 2,     -1] = rr[ri * 2,     idx_dtype(0)]229                blk.S[i * 2 + 1, -1] = rr[ri * 2 + 1, idx_dtype(0)]230231    def build_landmark_blocks(self, rr, Jrb, Jrp, Jrl):232        blk_i = 0233        for i in range(len(self.problem.pts3d)):234            if not self.problem.valid_pts3d[i]:235                continue236237            r_idxs = np.where(self.problem.pt3d_idxs == i)[0].astype(self.idx_dtype)238            blk_np = len(r_idxs)239240            if blk_np < 2:241                # too few observations of this 3d-point242                self.problem.valid_pts3d[i] = False243                continue244245            # indexing sparse array faster this way instead of Jl[v_i, i:i+3]246            pose_idxs = self.problem.pose_idxs[r_idxs]247            _, _, Jrl_i, Jrl_j = self._block_indexing(r_idxs, np.ones((len(r_idxs),), dtype=self.idx_dtype) * i, 2, 3)248            rr_i, _, Jrp_i, Jrp_j = self._block_indexing(r_idxs, pose_idxs, 2, self._pose_size)249            _, _, blk_Jp_i, blk_Jp_j = self._block_indexing(np.arange(blk_np, dtype=self.idx_dtype),250                                                            np.arange(blk_np, dtype=self.idx_dtype), 2, self._pose_size)251252            # create residual vector, assign reprojection residuals253            blk_r = np.zeros((blk_np*2, 1), dtype=self.dtype)254            blk_r[:blk_np*2] = rr[rr_i]255256            # batch jacobian related to reprojection residuals257            blk_Jb = None if Jrb is None else np.array(Jrb[rr_i, :], dtype=self.dtype)    # dense matrix here258259            # create landmark jacobian matrix, assign the three columns related to this landmark260            blk_Jl = np.zeros((blk_np*2, 3), dtype=self.dtype)261            blk_Jl[:blk_np*2, :] = np.array(self._sp_index(Jrl, Jrl_i, Jrl_j).reshape((-1, 3)), dtype=self.dtype)262263            # create pose/frame jacobian matrix, assign all frames that observe this landmark264            blk_Jp = np.zeros((blk_np*2, blk_np*self._pose_size), dtype=self.dtype)265            blk_Jp[blk_Jp_i, blk_Jp_j] = self._sp_index(Jrp, Jrp_i, Jrp_j)266267            if len(self._blocks) <= blk_i:268                m, n_b, n_p, n_l = blk_r.size, blk_Jb.shape[1], blk_Jp.shape[1], blk_Jl.shape[1]269                self._blocks.append(ResidualBlock(r_idxs, pose_idxs, m, n_b, n_p, n_l, self._pose_size))270            self._blocks[blk_i].new_linearization_point(blk_r, blk_Jb, blk_Jp, blk_Jl)271            blk_i += 1272273    def _sp_index(self, A, i, j, safe=False):274        if is_own_sp_mx(A):275            B = np.empty(len(i), dtype=self.dtype)276            A.copyto((i, j), B)277            return B278        # safe is very slow, unsafe is dangerous as bugs won't be found279        return A[i, j] if safe or not sp.issparse(A) else A._get_arrayXarray(i, j)280281    @staticmethod282    def _block_indexing(row_idxs, col_idxs, block_rows, block_cols):283        row_idxs_r = InnerLinearizerQR._augment_idxs(row_idxs, block_rows)284        row_idxs_rc = np.repeat(row_idxs_r, block_cols)285        col_idxs_c = InnerLinearizerQR._augment_idxs(col_idxs, block_cols)286        col_idxs_cr = np.repeat(col_idxs_c.reshape((-1, block_cols)), block_rows, axis=0).flatten()287        return row_idxs_r, col_idxs_c, row_idxs_rc, col_idxs_cr288289    @staticmethod290    def _augment_idxs(idxs, size):291        return (np.repeat(idxs[:, None] * size, size, axis=1) + np.arange(size, dtype=idxs.dtype)[None, :]).flatten()292293    def _apply_huber(self, r, arr_J, total_nr, huber_coef, extra_weight=1.0):294        huber_coef = np.inf if huber_coef is None else huber_coef295296        residual_weight = extra_weight * (total_nr / len(r) if self.use_weighted_residuals else 1.0)297298        abs_r = np.abs(r)299        I = abs_r > huber_coef300        huber_weight = np.ones_like(r)301        huber_weight[I] *= np.sqrt(huber_coef / abs_r[I])302        dwr_dr = np.sqrt(residual_weight * huber_weight)303304        wr = dwr_dr * r305306        if arr_J is None:307            wJ = None308        else:309            wJ = []310            for i, J in enumerate(arr_J):311                if J is None:312                    wJ.append(J)313                elif sp.issparse(J):314                    wJ.append(sp.diags(dwr_dr.flatten()).dot(J))315                elif is_own_sp_mx(J):316                    J.imul_arr(dwr_dr.reshape((dwr_dr.size, -1)))317                    wJ.append(J)318                else:319                    J *= dwr_dr320                    wJ.append(J)321322        err = np.sum(0.5 * residual_weight * huber_weight * (2 - huber_weight) * r ** 2)323        return wr, wJ, err324325    def initial_cost(self):326        return self._total_error / self.residual_count()327328    def residual_count(self):329        return self.problem.pts2d.size + self.problem.meas_r.size + self.problem.meas_aa.size330331    def current_cost(self):332        nr, nx, na = self.problem.pts2d.size, self.problem.meas_r.size, self.problem.meas_aa.size333        rr, *rxra = self.problem.residual(parts=True)334335        if self.problem.inter_batch_repr_count == 0:336            _, _, err = self._apply_huber(rr, None, nr+nx+na, self.huber_coef_repr)337        else:338            c = self.problem.inter_batch_repr_count339            _, _, err1 = self._apply_huber(rr[:-c], None, nr+nx+na, self.huber_coef_repr)340            _, _, err2 = self._apply_huber(rr[-c:], None, nr+nx+na, self.huber_coef_repr,341                                           extra_weight=self.inter_batch_repr_weight)342            err = err1 + err2343        total_error = err344345        if nx > 0:346            rx = rxra[0]347            _, _, err = self._apply_huber(rx, None, nr+nx+na, self.huber_coef_loc)348            total_error += err349350        if na > 0:351            ra = rxra[0 if nx == 0 else 1]352            _, _, err = self._apply_huber(ra, None, nr+nx+na, self.huber_coef_ori)353            total_error += err354355        return total_error / (nr + nx + na)356357    def filter(self, max_repr_err):358        n_obs = len(self.problem.pts2d)359        obs_idx = self.problem.filter(max_repr_err)360        cost = self.current_cost()361        return len(obs_idx)/n_obs, cost362363    def set_pose_damping(self, _lambda):364        self._pose_damping = _lambda365366    def marginalize(self):367        assert self._state == self.STATE_LINEARIZED, 'not linearized yet'368        if NUMBA_LEVEL >= 3:369            with warnings.catch_warnings():370                # for some reason warns about non-contiguous arrays in n_p.dot, can't find any though371                warnings.filterwarnings("ignore")372                self._marginalize(nb.typed.List(self._blocks))373        else:374            self._marginalize(self._blocks)375        self._state = self.STATE_MARGINALIZED376377    @staticmethod378    @maybe_decorate(nb.njit(nogil=True, parallel=False, cache=True), NUMBA_LEVEL >= 3)379    def _marginalize(blocks):380        for blk in blocks:381            blk.marginalize()382383    def backsub_xl(self, delta_xbp):384        assert self._state == self.STATE_MARGINALIZED, 'not linearized yet'385386        l_diff = 0387        delta_xb = delta_xbp[:self.nb]388        delta_xp = delta_xbp[self.nb:].reshape((-1, self._pose_size))389        delta_xl = np.zeros((len(self._blocks), self._blocks[0].n_l), dtype=self.dtype)390        for i, blk in enumerate(self.all_blocks):391            blk_delta_xp = delta_xp[blk.pose_idxs, :].reshape((-1, 1))392            if blk.n_l > 0:393                blk_delta_xbp = np.concatenate((delta_xb, blk_delta_xp), axis=0)394                blk_delta_xl = -np.linalg.solve(blk.R1, blk.Q1T_r + blk.Q1T_Jbp.dot(blk_delta_xbp))395                delta_xl[i, :] = blk_delta_xl.T * blk.Jl_col_scale396            else:397                blk_delta_xbp = blk_delta_xp398399            # calculate linear cost diff400            blk.damp(0)401            QT_J_deltax = blk.QT_Jbp @ blk_delta_xbp402            if blk.n_l > 0:403                QT_J_deltax[:blk.n_l] += blk.R1.dot(blk_delta_xl)404            l_diff -= QT_J_deltax.T.dot(0.5 * QT_J_deltax + blk.QT_r).flatten()[0]405406        return delta_xl, l_diff / self.residual_count()407408    # def get_Q2Tr(self):409    #     return self._Q2.T.dot(self._r)410    #411    # def get_Q2TJp(self):412    #     return self._Q2.T.dot(self._Jp)413    #414    # def get_Q2TJp_premult_x(self, x_r):415    #     return x_r.T.dot(self._Q2.T).dot(self._Jp)416    #417    # def get_Q2TJp_postmult_x(self, xp):418    #     return self._Q2.T.dot(self._Jp.dot(xp))419420    def get_Q2TJbp_T_Q2TJbp_mult_x(self, xp):421        assert self._state == self.STATE_MARGINALIZED422        assert len(self.problem.cam_param_idxs) == 0, 'not implemented for cam param estimation'423        assert len(self.problem.meas_aa) == 0, 'not implemented for absolute measures'424        assert len(self.problem.meas_r) == 0, 'not implemented for absolute measures'425426        y = np.zeros((self._pose_n, self._pose_size))427        for blk in self._blocks:428            blk_xp = xp.reshape((-1, self._pose_size))[blk.pose_idxs, :].reshape((-1, 1))429            blk_y = blk.Q2T_Jbp.T.dot(blk.Q2T_Jbp.dot(blk_xp))430            y[blk.pose_idxs, :] += blk_y.reshape((-1, self._pose_size))431432        return y.reshape((-1, 1))433434    def get_Q2TJbp_T_Q2Tr(self):435        assert self._state == self.STATE_MARGINALIZED436437        brb = np.zeros((self.nb, 1), dtype=self.dtype)438        bp = np.zeros((self._pose_n, self._pose_size), dtype=self.dtype)439        for blk in self.all_blocks:440            Q2T_Jbp = blk.Q2T_Jbp441            if sp.issparse(Q2T_Jbp):442                blk_b = (blk.Q2T_r.T @ Q2T_Jbp).T443            else:444                blk_b = (blk.Q2T_r.T.dot(Q2T_Jbp)).T445            if blk.n_b > 0:446                brb[:self.nb] += blk_b[:blk.n_b]447            bp[blk.pose_idxs, :] += blk_b[blk.n_b:].reshape((-1, self._pose_size))448449        return np.concatenate((brb, bp.reshape((-1, 1))), axis=0)450451    # def get_Q2TJp_diag2(self):452    #     pass453454    def _epow2(self, arr):455        if type(arr) == np.ndarray:456            return arr ** 2457        else:458            assert sp.issparse(arr)459            arr = arr.copy()460            arr.data *= arr.data461            return arr462463    def get_Jbp_diag2(self):464        assert self._state == self.STATE_LINEARIZED465466        db2 = np.zeros((self.nb,), dtype=self.dtype)467        dp2 = np.zeros((self._pose_n, self._pose_size), dtype=self.dtype)468        for blk in self.all_blocks:469            if blk.n_b > 0:470                db2 += np.sum(self._epow2(blk.Jb), axis=0)471            dp2[blk.pose_idxs, :] += np.sum(self._epow2(blk.Jp), axis=0).reshape((-1, self._pose_size))472473        return np.concatenate((db2, dp2.flatten()))474475    def get_Q2TJbp_T_Q2TJbp_blockdiag(self):476        assert self._state == self.STATE_MARGINALIZED477478        # TODO: don't use dense Hpp479480        if 0:481            # too slow482            Hpp = self._get_Q2TJbp_T_Q2TJbp_blockdiag_sp(self.nb, self.np, self._pose_size,483                                                         self._pose_damping, self._blocks, self.dtype)484        elif NUMBA_LEVEL >= 3:485            # no extra memory used486            Hpp = DictArray2D((self.nb + self.np, self.nb + self.np), dtype=self.dtype)487            self._get_Q2TJbp_T_Q2TJbp_blockdiag_lv3(self.nb, self.np, self._pose_size,488                                                    self._pose_damping, self._blocks, Hpp)489        else:490            # uses way too much memory for large problems due to dense Hpp491            Hpp = self._get_Q2TJbp_T_Q2TJbp_blockdiag(self.nb, self.np, self._pose_size, self._pose_damping,492                                                      nb.typed.List([blk.Q2T_Jbp for blk in self._blocks]),493                                                      nb.typed.List([blk.pose_idxs for blk in self._blocks]))494495        for blk in self.non_lm_blocks:496            blk_Hpp = blk.Q2T_Jbp.T.dot(blk.Q2T_Jbp)497            tmp = np.arange(len(blk.pose_idxs))498            _, _, bi, bj = self._block_indexing(tmp, tmp, self._pose_size, self._pose_size)499            _, _, i, j = self._block_indexing(blk.pose_idxs, blk.pose_idxs, self._pose_size, self._pose_size)500            if is_own_sp_mx(Hpp):501                Hpp.idx_isum_arr(i + self.nb, j + self.nb, np.array(blk_Hpp[bi, bj]).flatten())502            else:503                Hpp[i + self.nb, j + self.nb] += np.array(blk_Hpp[bi, bj]).flatten()504505        if sp.issparse(Hpp) and sp.isspmatrix_csc(Hpp):506            return Hpp507        if sp.issparse(Hpp):508            return Hpp.tocsc()509        if is_own_sp_mx(Hpp):510            return own_sp_mx_to_coo(Hpp).tocsc()511        return sp.csc_matrix(Hpp)512513    @staticmethod514    def _get_Q2TJbp_T_Q2TJbp_blockdiag_sp(n_b, n_p, psize, pose_damping, blocks, dtype):515        H = sp.dok_matrix((n_b + n_p, n_b + n_p), dtype=dtype)516517        bj, bi = np.meshgrid(np.arange(n_b), np.arange(n_b))518        pj, pi = np.meshgrid(np.arange(psize), np.arange(psize))519        bj, bi, pj, pi = map(lambda x: x.flatten(), (bj, bi, pj, pi))520521        for blk in blocks:522            if n_b > 0:523                A = blk.Q2T_Jbp[:, 0:n_b]  # should copy or call np.ascontiguousarray?524                blk_Hbb = A.T.dot(A).copy()525                H[bi, bj] += blk_Hbb.flatten()526527            for j, idx in enumerate(blk.pose_idxs):528                g0, b0 = n_b + idx * psize, n_b + j * psize529                A = blk.Q2T_Jbp[:, b0:b0+psize].copy()530                blk_Hpp = A.T.dot(A)531                H[g0 + pi, g0 + pj] += blk_Hpp.flatten()532533        if pose_damping is not None:534            i = np.arange(n_b + n_p, dtype=np.int32)535            H[i, i] += dtype.type(pose_damping)536537        return H538539    @staticmethod540    @maybe_decorate(nb.njit(nogil=True, parallel=False, cache=True), NUMBA_LEVEL >= 3)541    def _get_Q2TJbp_T_Q2TJbp_blockdiag_lv3(_nb, _np, psize, pose_damping, blocks, H):542        for k in range(len(blocks)):543            blk = blocks[k]544            if _nb > 0:545                A = blk.Q2T_Jbp[:, 0:_nb].copy()546                blk_Hbb = A.T.dot(A)547                for i in range(_nb):548                    for j in range(_nb):549                        H[i, j] += blk_Hbb[i, j]550551            for j, idx in enumerate(blk.pose_idxs):552                g0, b0 = _nb + idx * psize, _nb + j * psize553                g1, b1 = g0 + psize, b0 + psize554                A = blk.Q2T_Jbp[:, b0:b1].copy()555                blk_Hpp = A.T.dot(A)556                for i, gi in enumerate(range(g0, g1)):557                    for j, gj in enumerate(range(g0, g1)):558                        H[gi, gj] += blk_Hpp[i, j]559560        if pose_damping is not None:561            val = np.float32(pose_damping)562            for i in range(_nb + _np):563                H[i, i] += val564565    @staticmethod566    @maybe_decorate(nb.njit(nogil=True, parallel=False, cache=True), NUMBA_LEVEL >= 1)567    def _get_Q2TJbp_T_Q2TJbp_blockdiag(_nb, _np, psize, pose_damping, arr_Q2T_Jbp, arr_pose_idxs):568        H = np.zeros((_nb + _np, _nb + _np), dtype=arr_Q2T_Jbp[0].dtype)569        for i, Q2T_Jbp in enumerate(arr_Q2T_Jbp):570            if _nb > 0:571                A = Q2T_Jbp[:, 0:_nb].copy()572                blk_Hbb = A.T.dot(A)573                H[:_nb, :_nb] += blk_Hbb574575            for j, idx in enumerate(arr_pose_idxs[i]):576                g0, b0 = _nb + idx * psize, _nb + j * psize577                g1, b1 = g0 + psize, b0 + psize578                A = Q2T_Jbp[:, b0:b1].copy()579                blk_Hpp = A.T.dot(A)580                H[g0:g1, g0:g1] += blk_Hpp581582        if pose_damping is not None:583            H += pose_damping * np.eye(H.shape[0], dtype=H.dtype)584585        return H586587    def get_Jbp_T_Jbp_blockdiag(self):588        assert self._state == self.STATE_LINEARIZED, 'can only use if state == LINEARIZED'589        assert len(self.problem.cam_param_idxs) == 0, 'not implemented for cam param estimation'590591        Hpp = np.zeros((self.np, self.np), dtype=self.dtype)592        for blk in self.all_blocks:593            for i, idx in enumerate(blk.pose_idxs):594                g0, b0, br0 = blk.pose_idxs[i] * self._pose_size, i * self._pose_size, i * 2595                g1, b1, br1 = g0 + self._pose_size, b0 + self._pose_size, br0 + 2596                blk_Hpp = blk.Jp[br0:br1, b0:b1].T.dot(blk.Jp[br0:br1, b0:b1])597                Hpp[g0:g1, g0:g1] += blk_Hpp598599        if self._pose_damping is not None:600            Hpp += self._pose_damping * np.eye(Hpp.shape[0], dtype=self.dtype)601602        return sp.csc_matrix(Hpp)603604    def scale_Jl_cols(self):605        assert self._state == self.STATE_LINEARIZED, 'can only use if state == LINEARIZED'606        for blk in self._blocks:607            blk.Jl_col_scale = 1 / (self.jacobi_scaling_eps + np.linalg.norm(blk.Jl, axis=0, keepdims=True))608            blk.Jl *= blk.Jl_col_scale609610    def scale_Jbp_cols(self, jac_scaling):611        assert self._state == self.STATE_MARGINALIZED, 'can only use if state == MARGINALIZED'612        jac_rb_scaling = jac_scaling[None, :self.nb]613        jac_rp_scaling = jac_scaling[self.nb:].reshape((-1, self._pose_size))614        for blk in self.all_blocks:615            assert not blk.is_damped() or blk.n_l == 0, 'apply scaling before damping'616            blk_rp = jac_rp_scaling[blk.pose_idxs, :].reshape((1, -1))617            blk_jsc = np.concatenate((jac_rb_scaling, blk_rp), axis=1) if blk.n_l > 0 else blk_rp618            if sp.issparse(blk.QT_Jbp):619                blk.QT_Jbp = blk.QT_Jbp * sp.diags(blk_jsc.flatten())620            else:621                blk_QT_Jbp = blk.QT_Jbp  # direct blk.QT_Jbp inplace multiplication will fail because of @property decorator622                blk_QT_Jbp *= blk_jsc    # same as blk.QT_Jbp = blk.QT_Jbp.dot(n_p.diag(blk_jsc))623624    def set_landmark_damping(self, _lambda):625        assert self._state == self.STATE_MARGINALIZED, 'can only use if state == MARGINALIZED'626        for blk in self._blocks:627            blk.damp(_lambda)628629    def get_stage1(self, precond_mx):630        assert False, 'not implemented'631        # TODO: would need to restructure so that all block level operations implemented at ResidualBlock632633    def get_stage2(self, _lambda, jac_scaling_D, precond_mx):634        assert False, 'not implemented'635636    def right_multiply(self, xp):637        return self.get_Q2TJbp_T_Q2TJbp_mult_x(xp)638639    def nb(self) -> int:640        return len(self.problem.xb)641642643nb_type, idx_type, np_type = nb.float32, nb.int32, np.float32644array_type = nb_type[:, :]645646@maybe_decorate(647    nb.experimental.jitclass([648        ('r_idxs', idx_type[:]),649        ('pose_idxs', idx_type[:]),650        ('m', nb.intp),651        ('n_b', nb.intp),652        ('n_p', nb.intp),653        ('n_l', nb.intp),654        ('psize', nb.intp),655        ('pnum', nb.intp),656        ('S', array_type),657        ('Jl_col_scale', nb.optional(array_type)),658        ('_marginalized', nb.boolean),659        ('_damping_rots', nb.optional(nb.types.ListType(array_type))),660    ]), NUMBA_LEVEL >= 3)661class ResidualBlock:662    def __init__(self, r_idxs, pose_idxs, m, n_b, n_p, n_l, psize=6):663        self.r_idxs = r_idxs664        self.pose_idxs = pose_idxs665        self.psize = psize666        self.m, self.n_b, self.n_p, self.n_l = m, n_b, n_p, n_l667        self.pnum = self.n_p // self.psize668669        self.S = np.empty((self.m + self.n_l, self.n_b + self.n_p + self.n_l + 1), dtype=np_type)670        self.Jl_col_scale = None671        self._marginalized = False672        self._damping_rots = None673674    def reset(self):675        self._marginalized = False676        self._damping_rots = None677        self.S[:] = self.S.dtype.type(0.0)678679    def new_linearization_point(self, r, Jb, Jp, Jl):680        self.reset()681        self.r, self.Jb, self.Jp, self.Jl = r, Jb, Jp, Jl682683    def marginalize(self):684        assert not self.is_marginalized(), 'already marginalized'685        self._marginalize(self.S, self.m, self.n_b, self.n_p, self.n_l)686        self._marginalized = True687688    @staticmethod689    @maybe_decorate(nb.njit(nogil=True, parallel=False, cache=True), NUMBA_LEVEL == 2)690    def _marginalize(_S, m, n_b, n_p, n_l):691        Q, R = np.linalg.qr(_S[:m, n_b + n_p:-1], mode='complete')  # NOTE: numba 0.51.1 doesn't support mode argument,692                                                                    #       this uses own qr function at algo/linalg.py693        if 0:694            # S[:m, ...] is not contiguous, need to ignore warnings in non jitted code695            _S[:m, :n_b + n_p] = Q.T.dot(_S[:m, :n_b + n_p])   # Q.T.dot(Jbp)696            _S[:m, n_b + n_p:-1] = R                           # Q.T.dot(Jl)697            _S[:m, -1:] = Q.T.dot(_S[:m, -1:])                 # Q.T.dot(r)698        else:699            # similar speed as the alternative above, no warnings, maybe needs a bit more memory?700            A = np.ascontiguousarray(_S[:m, :])701            _S[:m, :] = Q.T.dot(A)702703    def damp(self, _lambda):704        assert self.is_marginalized(), 'not yet marginalized'705        assert _lambda >= 0, 'lambda must be >= 0'706707        if self.n_l == 0:708            return709710        if self.is_damped():711            self.undamp()712713        if _lambda == 0:714            pass  # by default lambda already zero if no damping715        else:716            # make and apply givens rotations717            self.S[-self.n_l:, self.n_b + self.n_p:self.n] = np.eye(self.n_l, dtype=self.S.dtype) * np.sqrt(_lambda)718            self._damping_rots = self._damp(self.S, self.n_b + self.n_p, self.n_l)719720    @staticmethod721    @maybe_decorate(nb.njit(nogil=True, parallel=False, cache=True), 3 > NUMBA_LEVEL >= 1)722    def _damp(S, l_idx, k):723        damping_rots = nb.typed.List.empty_list(array_type)724        for n in range(k):725            for m in range(n + 1):726                G = tools.make_givens(S[n, l_idx + n], S[-k + n - m, l_idx + n], dtype=S.dtype)727                tools.apply_givens(S, G, S.shape[0] - k + n - m, n)728                damping_rots.append(G)729        return damping_rots730731    def undamp(self):732        assert self.is_damped(), 'not damped yet'733        self._undamp(self._damping_rots, self.S, self.n_l)734        self._damping_rots = None735736    @staticmethod737    @maybe_decorate(nb.njit(nogil=True, parallel=False, cache=True), 3 > NUMBA_LEVEL >= 1)738    def _undamp(damping_rots, S, k):739        for n in range(k-1, -1, -1):740            for m in range(n, -1, -1):741                G = damping_rots.pop()742                tools.apply_givens(S, G.T, S.shape[0] - k + n - m, n)    # TODO: contiguous743744    def is_marginalized(self):745        return self._marginalized746747    def is_damped(self):748        return self._damping_rots is not None or self.n_l == 0749750    @property751    def r(self):
...

test_plugin_dependency.py

Source:test_plugin_dependency.py

...7@pytest.mark.parametrize('needs_decorate_method', [True, False])8@pytest.mark.parametrize('provides_decorate_method', [True, False])9def test_needs_provides_plugin_name(needs_decorate_method, provides_decorate_method, checkpoint1, checkpoint2):10    @slash.plugins.active  # pylint: disable=abstract-method, unused-variable11    @maybe_decorate(slash.plugins.needs('p'), not needs_decorate_method)12    @autoname13    class NeedsPlugin(PluginInterface):14        @maybe_decorate(slash.plugins.needs('p'), needs_decorate_method)15        def session_start(self):16            checkpoint2()17    @slash.plugins.active  # pylint: disable=abstract-method, unused-variable18    @maybe_decorate(slash.plugins.provides('p'), not provides_decorate_method)19    @autoname20    class ProvidesPlugin(PluginInterface):21        @maybe_decorate(slash.plugins.provides('p'), provides_decorate_method)22        def session_start(self):23            checkpoint1()24    slash.hooks.session_start()  # pylint: disable=no-member25    assert checkpoint1.timestamp < checkpoint2.timestamp26def test_provides_globally_needs_globally(checkpoint1, checkpoint2):27    '''28    Plugin A: Provides x at class level29    Plugin B: Needs x at class level30    '''31    @slash.plugins.provides('x')32    class PluginA(slash.plugins.interface.PluginInterface):33        def get_name(self):34            return 'plugin a'35        def session_start(self):36            checkpoint1()37        def test_start(self):38            pass39    @slash.plugins.needs('x')40    class PluginB(slash.plugins.interface.PluginInterface):41        def get_name(self):42            return 'plugin b'43        def session_start(self):44            checkpoint2()45        def error_added(self, result, error): # pylint: disable=unused-argument46            pass47    for plugin_cls in [PluginA, PluginB]:48        slash.plugins.manager.install(plugin_cls(), activate_later=True)49    slash.plugins.manager.activate_pending_plugins()50    slash.hooks.session_start()  # pylint: disable=no-member51    assert checkpoint1.timestamp < checkpoint2.timestamp52    slash.plugins.manager.deactivate('plugin a')53    with pytest.raises(CannotResolveDependencies) as caught:54        slash.hooks.session_start()  # pylint: disable=no-member55    assert caught.value.unmet_dependencies == set(['x'])56def test_provides_globally_needs_specific_hook(checkpoint1, checkpoint2):57    '''58    Plugin A: Provides x at class level59    Plugin B: Needs x for specific hook60    '''61    @slash.plugins.provides('x')62    class PluginA(slash.plugins.interface.PluginInterface):63        def get_name(self):64            return 'plugin a'65        def session_start(self):66            checkpoint1()67        def test_start(self):68            pass69    class PluginB(slash.plugins.interface.PluginInterface):70        def get_name(self):71            return 'plugin b'72        @slash.plugins.needs('x')73        def session_start(self):74            checkpoint2()75        def error_added(self, result, error): # pylint: disable=unused-argument76            pass77    for plugin_cls in [PluginA, PluginB]:78        slash.plugins.manager.install(plugin_cls(), activate_later=True)79    slash.plugins.manager.activate_pending_plugins()80    slash.hooks.session_start()  # pylint: disable=no-member81    assert checkpoint1.timestamp < checkpoint2.timestamp82    slash.plugins.manager.deactivate('plugin a')83    with pytest.raises(CannotResolveDependencies) as caught:84        slash.hooks.session_start()  # pylint: disable=no-member85    assert caught.value.unmet_dependencies == set(['x'])86def test_provides_globally_needs_specific_hook_which_does_not_exist_at_a(checkpoint2):87    '''88    Plugin A: Provides x at class level89    Plugin B: Needs x for specific hook, this hook does not definied in A90    Expectations:91    Should work in the empty sense92    all non-needing hooks should work, even when missing from A, the specific hook needs to happen in A before B.93    '''94    @slash.plugins.provides('x')95    class PluginA(slash.plugins.interface.PluginInterface):96        def get_name(self):97            return 'plugin a'98        def test_start(self):99            pass100    class PluginB(slash.plugins.interface.PluginInterface):101        def get_name(self):102            return 'plugin b'103        @slash.plugins.needs('x')104        def session_start(self):105            checkpoint2()106        def error_added(self, result, error): # pylint: disable=unused-argument107            pass108    for plugin_cls in [PluginA, PluginB]:109        slash.plugins.manager.install(plugin_cls(), activate_later=True)110    slash.plugins.manager.activate_pending_plugins()111    slash.hooks.session_start()  # pylint: disable=no-member112    assert checkpoint2.called113    slash.plugins.manager.deactivate('plugin a')114    with pytest.raises(CannotResolveDependencies) as caught:115        slash.hooks.session_start()  # pylint: disable=no-member116    assert caught.value.unmet_dependencies == set(['x'])117def test_provides_specific_hook_needs_globally(checkpoint1, checkpoint2):118    '''119    Plugin A: Provides x on a specific hook120    Plugin B: Needs x at class level121    Expectations:122    This case should fail, because logically the other hooks don't have anyone to provide X for them123    '''124    class PluginA(slash.plugins.interface.PluginInterface):125        def get_name(self):126            return 'plugin a'127        @slash.plugins.provides('x')128        def session_start(self):129            checkpoint1()130        def test_start(self):131            pass132    @slash.plugins.needs('x')133    class PluginB(slash.plugins.interface.PluginInterface):134        def get_name(self):135            return 'plugin b'136        def session_start(self):137            checkpoint2()138        def error_added(self, result, error): # pylint: disable=unused-argument139            pass140    for plugin_cls in [PluginA, PluginB]:141        slash.plugins.manager.install(plugin_cls(), activate_later=True)142    slash.plugins.manager.activate_pending_plugins()143    slash.hooks.session_start()  # pylint: disable=no-member144    with pytest.raises(CannotResolveDependencies) as caught:145        slash.hooks.error_added(result=None, error=None)  # pylint: disable=no-member146    assert caught.value.unmet_dependencies == set(['x'])147def test_provides_specific_hook_needs_globally_with_this_hook_only(checkpoint1, checkpoint2):148    '''149    Plugin A: Provides x on a specific hook150    Plugin B: Needs x at class level, but only has one hook (the one provided by A)151    '''152    class PluginA(slash.plugins.interface.PluginInterface):153        def get_name(self):154            return 'plugin a'155        @slash.plugins.provides('x')156        def session_start(self):157            checkpoint1()158        def test_start(self):159            pass160    @slash.plugins.needs('x')161    class PluginB(slash.plugins.interface.PluginInterface):162        def get_name(self):163            return 'plugin b'164        def session_start(self):165            checkpoint2()166    for plugin_cls in [PluginA, PluginB]:167        slash.plugins.manager.install(plugin_cls(), activate_later=True)168    slash.plugins.manager.activate_pending_plugins()169    slash.hooks.session_start()  # pylint: disable=no-member170    assert checkpoint1.timestamp < checkpoint2.timestamp171    slash.plugins.manager.deactivate('plugin a')172    with pytest.raises(CannotResolveDependencies) as caught:173        slash.hooks.session_start()  # pylint: disable=no-member174    assert caught.value.unmet_dependencies == set(['x'])175@pytest.mark.parametrize('needs_parent_level', [True, False])176@pytest.mark.parametrize('provides_parent_level', [True, False])177def test_provides_needs_with_inheritence_on_class_level(checkpoint, checkpoint1, checkpoint2, needs_parent_level, provides_parent_level):178    '''179    Plugin A: Provides x in class level (by it self or by inheritence)180    Plugin b: Needs x in class level (by it self or by inheritence)181    '''182    # pylint: disable=abstract-method183    @maybe_decorate(slash.plugins.provides('x'), provides_parent_level)184    class PluginAParent(slash.plugins.interface.PluginInterface):185        def test_start(self):186            pass187    @maybe_decorate(slash.plugins.provides('x'), not provides_parent_level)188    class PluginA(PluginAParent):189        def get_name(self):190            return 'plugin a'191        def session_start(self):192            checkpoint1()193    @maybe_decorate(slash.plugins.needs('x'), needs_parent_level)194    class PluginBParent(slash.plugins.interface.PluginInterface):195        def error_added(self, result, error): # pylint: disable=unused-argument196            checkpoint()197    @maybe_decorate(slash.plugins.needs('x'), not needs_parent_level)198    class PluginB(PluginBParent):199        def get_name(self):200            return 'plugin b'201        def session_start(self):202            checkpoint2()203    for plugin_cls in [PluginA, PluginB]:204        slash.plugins.manager.install(plugin_cls(), activate_later=True)205    slash.plugins.manager.activate_pending_plugins()206    # session_start hook should be provided the PluginA.session_start method207    slash.hooks.session_start()  # pylint: disable=no-member208    assert checkpoint1.timestamp < checkpoint2.timestamp209    # error_added hook should be provided by empty registration of pluginA210    slash.hooks.error_added(result=None, error=None)  # pylint: disable=no-member211    assert checkpoint.called...

test_plugins.py

Source:test_plugins.py

...21def test_registers_on_kwargs(class_level_needs, class_level_provides):22    needs_decorator = plugins.needs('other_requirement')23    provides_decorator = plugins.provides('another_provided_requirement')24    @slash.plugins.active  # pylint: disable=unused-variable25    @maybe_decorate(needs_decorator, class_level_needs)26    @maybe_decorate(provides_decorator, class_level_provides)27    class SamplePlugin(PluginInterface):28        def get_name(self):29            return 'sample'30        @plugins.registers_on('some.hook', provides=['provided_requirement'], needs=['some_requirement'], tags=['tag'])31        @maybe_decorate(needs_decorator, not class_level_needs)32        @maybe_decorate(provides_decorator, not class_level_provides)33        def plugin_method(self):34            pass35    @gossip.register('some.hook', provides=['some_requirement', 'other_requirement'])36    def _unused():37        pass38    gossip.trigger('some.hook')39    hook = gossip.get_hook('some.hook')40    [registration] = [reg for reg in hook.get_registrations() if reg.func.__name__ == 'plugin_method']41    assert registration.tags == {'tag'}42    assert registration.needs == frozenset(['some_requirement', 'other_requirement'])43    assert registration.provides == frozenset(['provided_requirement', 'another_provided_requirement'])44def test_registers_on_with_private_methods(restore_plugins_on_cleanup, checkpoint):45    @slash.plugins.active  # pylint: disable=unused-variable46    class SamplePlugin(PluginInterface):...

util.py

Source:util.py

...77    if hasattr(request.body, "read"):78        return request.body.read()79    return request.body80def auto_decorate(decorator, predicate=lambda name, value: isinstance(value, types.FunctionType)):81    def maybe_decorate(attribute, value):82        if predicate(attribute, value):83            value = decorator(value)84        return value85    class DecorateAll(type):86        def __setattr__(cls, attribute, value):87            return super(DecorateAll, cls).__setattr__(attribute, maybe_decorate(attribute, value))88        def __new__(cls, name, bases, attributes_dict):89            new_attributes_dict = {90                attribute: maybe_decorate(attribute, value) for attribute, value in attributes_dict.items()91            }92            return super(DecorateAll, cls).__new__(cls, name, bases, new_attributes_dict)...

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.