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

How to use init_data method in hypothesis

Best Python code snippet using hypothesis

MOMspot.py

Source:MOMspot.py

1# -*- coding: utf-8 -*-2import numpy as np3import pandas as pd4import matplotlib.pyplot as plt5from math import log,floor6import tqdm7from scipy.optimize import minimize8# colors for plot9deep_saffron = '#FF9933'10air_force_blue = '#5D8AA8'11def backMean(X,d):12    M = []13    w = X[:d].sum()14    M.append(w/d)15    for i in range(d,len(X)):16        w = w - X[i-d] + X[i]17        M.append(w/d)18    return np.array(M)19class momSPOT:20    """21    This class allows to run biSPOT algorithm on univariate dataset (upper and lower bounds)22    23    Attributes24    ----------25    proba : float26        Detection level (risk), chosen by the user27        28    extreme_quantile : float29        current threshold (bound between normal and abnormal events)30        31    data : numpy.array32        stream33    34    init_data : numpy.array35        initial batch of observations (for the calibration/initialization step)36    37    init_threshold : float38        initial threshold computed during the calibration step39    40    peaks : numpy.array41        array of peaks (excesses above the initial threshold)42    43    n : int44        number of observed values45    46    Nt : int47        number of observed peaks48    """49    def __init__(self, q = 1e-4):50        """51        Constructor52        Parameters53        ----------54        q55            Detection level (risk)56    57        Returns58        ----------59        biSPOT object60        """61        self.proba = q62        self.data = None63        self.init_data = None64        self.n = 065        nonedict =  {'up':None,'down':None}66        67        self.extreme_quantile = dict.copy(nonedict)68        self.init_threshold = dict.copy(nonedict)69        self.peaks = dict.copy(nonedict)70        self.gamma = dict.copy(nonedict)71        self.sigma = dict.copy(nonedict)72        self.Nt = {'up':0,'down':0}73        74        75    def __str__(self):76        s = ''77        s += 'Streaming Peaks-Over-Threshold Object\n'78        s += 'Detection level q = %s\n' % self.proba79        if self.data is not None:80            s += 'Data imported : Yes\n'81            s += '\t initialization  : %s values\n' % self.init_data.size82            s += '\t stream : %s values\n' % self.data.size83        else:84            s += 'Data imported : No\n'85            return s86            87        if self.n == 0:88            s += 'Algorithm initialized : No\n'89        else:90            s += 'Algorithm initialized : Yes\n'91            s += '\t initial threshold : %s\n' % self.init_threshold92            93            r = self.n-self.init_data.size94            if r > 0:95                s += 'Algorithm run : Yes\n'96                s += '\t number of observations : %s (%.2f %%)\n' % (r,100*r/self.n)97                s += '\t triggered alarms : %s (%.2f %%)\n' % (len(self.alarm),100*len(self.alarm)/self.n)98            else:99                s += '\t number of peaks  : %s\n' % self.Nt100                s += '\t upper extreme quantile : %s\n' % self.extreme_quantile['up']101                s += '\t lower extreme quantile : %s\n' % self.extreme_quantile['down']102                s += 'Algorithm run : No\n'103        return s104    105    106    def fit(self,init_data,data):107        """108        Import data to biSPOT object109        110        Parameters111        ----------112        init_data : list, numpy.array or pandas.Series113            initial batch to calibrate the algorithm ()114            115        data : numpy.array116            data for the run (list, np.array or pd.series)117    118        """119        if isinstance(data,list):120            self.data = np.array(data)121        elif isinstance(data,np.ndarray):122            self.data = data123        elif isinstance(data,pd.Series):124            self.data = data.values125        else:126            print('This data format (%s) is not supported' % type(data))127            return128            129        if isinstance(init_data,list):130            self.init_data = np.array(init_data)131        elif isinstance(init_data,np.ndarray):132            self.init_data = init_data133        elif isinstance(init_data,pd.Series):134            self.init_data = init_data.values135        elif isinstance(init_data,int):136            self.init_data = self.data[:init_data]137            self.data = self.data[init_data:]138        elif isinstance(init_data,float) & (init_data<1) & (init_data>0):139            r = int(init_data*data.size)140            self.init_data = self.data[:r]141            self.data = self.data[r:]142        else:143            print('The initial data cannot be set')144            return145        146    def add(self,data):147        """148        This function allows to append data to the already fitted data149        150        Parameters151        ----------152        data : list, numpy.array, pandas.Series153            data to append154        """155        if isinstance(data,list):156            data = np.array(data)157        elif isinstance(data,np.ndarray):158            data = data159        elif isinstance(data,pd.Series):160            data = data.values161        else:162            print('This data format (%s) is not supported' % type(data))163            return164        165        self.data = np.append(self.data,data)166        return167    def initialize(self, verbose = True):168        """169        Run the calibration (initialization) step170        171        Parameters172        ----------173        verbose : bool174            (default = True) If True, gives details about the batch initialization175        """176        n_init = self.init_data.size177        178        S = np.sort(self.init_data)     # we sort X to get the empirical quantile179        self.init_threshold['up'] = S[int(0.98*n_init)] # t is fixed for the whole algorithm180        self.init_threshold['down'] = S[int(0.02*n_init)] # t is fixed for the whole algorithm181        # initial peaks182        self.peaks['up'] = self.init_data[self.init_data>self.init_threshold['up']]-self.init_threshold['up']183        self.peaks['down'] = -(self.init_data[self.init_data<self.init_threshold['down']]-self.init_threshold['down'])184        self.Nt['up'] = self.peaks['up'].size185        self.Nt['down'] = self.peaks['down'].size186        self.n = n_init187        188        if verbose:189            print('Initial threshold : %s' % self.init_threshold)190            print('Number of peaks : %s' % self.Nt)191            #print('Grimshaw maximum log-likelihood estimation ... ', end = '')192            193        l = {'up':None,'down':None}194        for side in ['up','down']:195            g,s,l[side] = self._MOM(side)196            self.extreme_quantile[side] = self._quantile(side,g,s)197            self.gamma[side] = g198            self.sigma[side] = s199        200        ltab = 20201        form = ('\t'+'%20s' + '%20.2f' + '%20.2f')202        '''203        if verbose:204            print('[done]')205            print('\t' + 'Parameters'.rjust(ltab) + 'Upper'.rjust(ltab) + 'Lower'.rjust(ltab))206            print('\t' + '-'*ltab*3)207            print(form % (chr(0x03B3),self.gamma['up'],self.gamma['down']))208            print(form % (chr(0x03C3),self.sigma['up'],self.sigma['down']))209            print(form % ('likelihood',l['up'],l['down']))210            print(form % ('Extreme quantile',self.extreme_quantile['up'],self.extreme_quantile['down']))211            print('\t' + '-'*ltab*3)212        '''213        return 214    215    def _MOM(self,side,epsilon = 1e-8, n_points = 10):216        Yi = self.peaks[side]217        avg = np.mean(Yi)218        var = np.var(Yi)219        sigma = 0.5*avg*(avg**2/var + 1)220        gamma = 0.5*(avg**2/var - 1)221        print gamma, sigma222        return gamma,sigma,100223    224    def _quantile(self,side,gamma,sigma):225        """226        Compute the quantile at level 1-q for a given side227        228        Parameters229        ----------230        side : str231            'up' or 'down'232        gamma : float233            GPD parameter234        sigma : float235            GPD parameter236        Returns237        ----------238        float239            quantile at level 1-q for the GPD(Î³,Ï,Î¼=0)240        """241        if side == 'up':242            r = self.n * self.proba / self.Nt[side]243            if gamma != 0:244                return self.init_threshold['up'] + (sigma/gamma)*(pow(r,-gamma)-1)245            else:246                return self.init_threshold['up'] - sigma*log(r)247        elif side == 'down':248            r = self.n * self.proba / self.Nt[side]249            if gamma != 0:250                return self.init_threshold['down'] - (sigma/gamma)*(pow(r,-gamma)-1)251            else:252                return self.init_threshold['down'] + sigma*log(r)253        else:254            print('error : the side is not right')255        256    def run(self, with_alarm = True):257        """258        Run biSPOT on the stream259        260        Parameters261        ----------262        with_alarm : bool263            (default = True) If False, SPOT will adapt the threshold assuming \264            there is no abnormal values265        Returns266        ----------267        dict268            keys : 'upper_thresholds', 'lower_thresholds' and 'alarms'269            270            '***-thresholds' contains the extreme quantiles and 'alarms' contains \271            the indexes of the values which have triggered alarms272            273        """274        if (self.n>self.init_data.size):275            print('Warning : the algorithm seems to have already been run, you \276            should initialize before running again')277            return {}278        279        # list of the thresholds280        thup = []281        thdown = []282        alarm = []283        # Loop over the stream284        for i in tqdm.tqdm(range(self.data.size)):285    286            # If the observed value exceeds the current threshold (alarm case)287            if self.data[i]>self.extreme_quantile['up'] :288                # if we want to alarm, we put it in the alarm list289                if with_alarm:290                    alarm.append(i)291                # otherwise we add it in the peaks292                else:293                    self.peaks['up'] = np.append(self.peaks['up'],self.data[i]-self.init_threshold['up'])294                    self.Nt['up'] += 1295                    self.n += 1296                    # and we update the thresholds297                    g,s,l = self._MOM('up')298                    self.extreme_quantile['up'] = self._quantile('up',g,s)299            # case where the value exceeds the initial threshold but not the alarm ones300            elif self.data[i]>self.init_threshold['up']:301                    # we add it in the peaks302                    self.peaks['up'] = np.append(self.peaks['up'],self.data[i]-self.init_threshold['up'])303                    self.Nt['up'] += 1304                    self.n += 1305                    # and we update the thresholds306                    g,s,l = self._MOM('up')307                    self.extreme_quantile['up'] = self._quantile('up',g,s)308                    309            elif self.data[i]<self.extreme_quantile['down'] :310                # if we want to alarm, we put it in the alarm list311                if with_alarm:312                    alarm.append(i)313                # otherwise we add it in the peaks314                else:315                    self.peaks['down'] = np.append(self.peaks['down'],-(self.data[i]-self.init_threshold['down']))316                    self.Nt['down'] += 1317                    self.n += 1318                    # and we update the thresholds319                    g,s,l = self._MOM('down')320                    self.extreme_quantile['down'] = self._quantile('down',g,s)321            # case where the value exceeds the initial threshold but not the alarm ones322            elif self.data[i]<self.init_threshold['down']:323                    # we add it in the peaks324                    self.peaks['down'] = np.append(self.peaks['down'],-(self.data[i]-self.init_threshold['down']))325                    self.Nt['down'] += 1326                    self.n += 1327                    # and we update the thresholds328                    g,s,l = self._MOM('down')329                    self.extreme_quantile['down'] = self._quantile('down',g,s)330            else:331                self.n += 1332                333            thup.append(self.extreme_quantile['up']) # thresholds record334            thdown.append(self.extreme_quantile['down']) # thresholds record335        336        return {'upper_thresholds' : thup,'lower_thresholds' : thdown, 'alarms': alarm}337    338    def plot(self,run_results,with_alarm = True):339        """340        Plot the results of given by the run341        342        Parameters343        ----------344        run_results : dict345            results given by the 'run' method346        with_alarm : bool347            (default = True) If True, alarms are plotted.348        Returns349        ----------350        list351            list of the plots352            353        """354        x = range(self.data.size)355        K = run_results.keys()356        357        ts_fig, = plt.plot(x,self.data,color=air_force_blue)358        fig = [ts_fig]359        360        if 'upper_thresholds' in K:361            thup = run_results['upper_thresholds']362            uth_fig, = plt.plot(x,thup,color=deep_saffron,lw=2,ls='dashed')363            fig.append(uth_fig)364            365        if 'lower_thresholds' in K:366            thdown = run_results['lower_thresholds']367            lth_fig, = plt.plot(x,thdown,color=deep_saffron,lw=2,ls='dashed')368            fig.append(lth_fig)369        370        if with_alarm and ('alarms' in K):371            alarm = run_results['alarms']372            al_fig = plt.scatter(alarm,self.data[alarm],color='red')373            fig.append(al_fig)374            375        plt.xlim((0,self.data.size))376        plt.show()377        ...

data.py

Source:data.py

...24        self.object_class = object_class25        self.target_visible = target_visible26        self.object_ids = object_ids27        self.multiobj_mode = multiobj_mode28        self.init_data = self._construct_init_data(init_data)29        self._ensure_start_frame()30    def _ensure_start_frame(self):31        # Ensure start frame is 032        start_frame = min(list(self.init_data.keys()))33        if start_frame > 0:34            self.frames = self.frames[start_frame:]35            if self.ground_truth_rect is not None:36                if isinstance(self.ground_truth_rect, (dict, OrderedDict)):37                    for obj_id, gt in self.ground_truth_rect.items():38                        self.ground_truth_rect[obj_id] = gt[start_frame:,:]39                else:40                    self.ground_truth_rect = self.ground_truth_rect[start_frame:,:]41            if self.ground_truth_seg is not None:42                self.ground_truth_seg = self.ground_truth_seg[start_frame:]43                assert len(self.frames) == len(self.ground_truth_seg)44            if self.target_visible is not None:45                self.target_visible = self.target_visible[start_frame:]46            self.init_data = {frame-start_frame: val for frame, val in self.init_data.items()}47    def _construct_init_data(self, init_data):48        if init_data is not None:49            if not self.multiobj_mode:50                assert self.object_ids is None or len(self.object_ids) == 151                for frame, init_val in init_data.items():52                    if 'bbox' in init_val and isinstance(init_val['bbox'], (dict, OrderedDict)):53                        init_val['bbox'] = init_val['bbox'][self.object_ids[0]]54            # convert to list55            for frame, init_val in init_data.items():56                if 'bbox' in init_val:57                    if isinstance(init_val['bbox'], (dict, OrderedDict)):58                        init_val['bbox'] = OrderedDict({obj_id: list(init) for obj_id, init in init_val['bbox'].items()})59                    else:60                        init_val['bbox'] = list(init_val['bbox'])61        else:62            init_data = {0: dict()}     # Assume start from frame 063            if self.object_ids is not None:64                init_data[0]['object_ids'] = self.object_ids65            if self.ground_truth_rect is not None:66                if self.multiobj_mode:67                    assert isinstance(self.ground_truth_rect, (dict, OrderedDict))68                    init_data[0]['bbox'] = OrderedDict({obj_id: list(gt[0,:]) for obj_id, gt in self.ground_truth_rect.items()})69                else:70                    assert self.object_ids is None or len(self.object_ids) == 171                    if isinstance(self.ground_truth_rect, (dict, OrderedDict)):72                        init_data[0]['bbox'] = list(self.ground_truth_rect[self.object_ids[0]][0, :])73                    else:74                        init_data[0]['bbox'] = list(self.ground_truth_rect[0,:])75            if self.ground_truth_seg is not None:76                init_data[0]['mask'] = self.ground_truth_seg[0]77        return init_data78    def init_info(self):79        info = self.frame_info(frame_num=0)80        return info81    def frame_info(self, frame_num):82        info = self.object_init_data(frame_num=frame_num)83        return info84    def init_bbox(self, frame_num=0):85        return self.object_init_data(frame_num=frame_num).get('init_bbox')86    def init_mask(self, frame_num=0):87        return self.object_init_data(frame_num=frame_num).get('init_mask')88    def get_info(self, keys, frame_num=None):89        info = dict()90        for k in keys:91            val = self.get(k, frame_num=frame_num)92            if val is not None:93                info[k] = val94        return info95    def object_init_data(self, frame_num=None) -> dict:96        if frame_num is None:97            frame_num = 098        if frame_num not in self.init_data:99            return dict()100        init_data = dict()101        for key, val in self.init_data[frame_num].items():102            if val is None:103                continue104            init_data['init_'+key] = val105        if 'init_mask' in init_data and init_data['init_mask'] is not None:106            anno = imread_indexed(init_data['init_mask'])107            if not self.multiobj_mode and self.object_ids is not None:108                assert len(self.object_ids) == 1109                anno = (anno == int(self.object_ids[0])).astype(np.uint8)...

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.