Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _process_group_dict method in autotest

Best Python code snippet using autotest_python

plot.py

Source:plot.py

...105        Fill out the plot in some way - different implementation for each subclass106        """107        raise NotImplementedError108    @abstractmethod109    def _process_group_dict(self,d:Dict)->Any:110        """111        Take a DB output dict and make a pair: <SOMETHING>,(group ID, group label)112        """113        return d114    @property115    @abstractmethod116    def kw(self) -> Set[str]:117        '''List of valid keyword arguments'''118        return {'query','title','xlab','frame','square',119                'xcols','xfunc','lcols','lfunc','gcols','gfunc'}120    #------------------------#121    # Static / Class methods #122    #------------------------#123    @staticmethod124    def pltdict()->Dict[str,Type['Plot']]:125        '''Mapping each type of plot to a unique string'''126        d = {'line':LinePlot,'bar':BarPlot,'hist':HistPlot} # type: Dict[str,Type['Plot']]127        return d128    @staticmethod129    def _make_groups(inputs : List[Dict[str,Any]],130                     gFunc  : FnArgs,131                     glFunc : FnArgs132                    ) -> List[Group]:133        """134        Take a list of processed DB outputs and sort into groups135        The groups will become different lines/bars on the final plot136        """137        groups = {} # type: Dict[Any,Group]138        counter = 0139        for x in inputs:140            g = gFunc(x)141            if g in groups:142                groups[g].add_elem(x)143            else:144                gl = glFunc(x)145                groups[g] = Group(id=counter,label=gl,rep=g,elems=[x])146                counter+=1147        return list(groups.values())148    @classmethod149    def from_file(cls,pth:str)->'Plot':150        '''Creates a Line/Bar/Hist plot from a JSON file containing KW args'''151        # Parse JSON file152        with open(pth,'r') as f:153            dic = load(f)154        # Determine which __init__ method should be used155        typ     = dic.pop('type').lower()156        plotter = cls.pltdict()[typ]157        # Create Plot object158        return plotter(**dic)159    #------------#160    # Properties #161    #------------#162    @property163    def _has_leg(self)->bool:164        return bool(self['gcols'])165    @property166    def opacity(self) -> float:167        return max(0.1,1-len(self.groups)/10)168    #-------------------#169    # Support functions #170    #-------------------#171    def _groups(self, conn : Conn, binds : list = [], funcs : Dict[str,C] = {}) -> None:172        """173        Populates: self.groups174        """175        self._init(funcs)176        assert self['query']177        results = conn.select_dict(self['query'], binds)178        self.groups = self._make_groups(results, self.gFunc, self.glFunc)179    def _data(self) -> list:180        ''' This seems to be a general enough implementation'''181        # for g in self.groups:182        return [self._draw(g.map(self._process_group_dict)) for g in self.groups]183################################################################################184class LinePlot(Plot):185    """186    Scatter or line plot - a relation between two numeric variables187    """188    def _init(self, funcs : Dict[str,C])->None:189        super()._init(funcs)190        assert 'ycols' in self, 'LinePlot requires ycols to be specified'191        # Y func, handle defaults192        if not 'yfunc' in self:193            if isinstance(self['ycols'],str):194                ycols = self['ycols'].split()195            else:196                ycols = self['ycols']197            assert len(ycols) == 1198            self['yfunc'] = identity199        self.yFunc = FnArgs(func = self['yfunc'], args = self['ycols'], funcs = funcs)200    def csv(self, pth : str) -> None:201        '''Write plot data to a csv'''202        raise NotImplementedError203    @property204    def kw(self)->Set[str]:205        return super().kw | {'ylab','ycols','yfunc','scatter'}206    def _draw(self, g : Group) -> dict:207        """process query results, then draw the lines"""208        # do aggregations, postprocessing to modify 'g', eventually209        g.sort(key=lambda x: x['x'])210        return self._add_line(g)211    def _process_group_dict(self, d : dict) -> dict:212        """213        Take a raw dictionary output from DB and make a new dictionary with:214            x,y (floats) and l (label)215        """216        return dict(x = self.xFunc(d),217                    y = self.yFunc(d),218                    l = self.lFunc(d))219    def _add_line(self, g : Group) -> dict:220        """221        Draw a line from a Group with (X,Y,LABEL) tuples as elements222        """223        leg, xyl = g.label, g.elems224        sty  = mkStyle(leg)    # get line style based on legend name225        scatr= self['scatter'] and self['scatter'][0].lower()=='t'226        mode = 'markers' if scatr else 'lines+markers'227        return dict(x       = g['x'],228                    y       = g['y'],229                    text    = g['l'],230                    mode    = mode,231                    name    = g.label,232                    marker  = dict(opacity = self.opacity,233                                   color   = sty.color),234                    line    = dict(color   = sty.color,235                                   dash    = sty.line,236                                   shape   = 'spline'))237    def _layout(self)->dict:238        return super()._layout()239################################################################################240class BarPlot(Plot):241    """242    Bar plots - relation between a real-valued variable and a categorical one243    """244    def _init(self, funcs : Dict[str,C])->None:245        super()._init(funcs)246        # SubGrouping of data points247        if 'spcols' not in self:248            # no way to subgroup if no columns provided249            self.spFunc = FnArgs(func = const(1), args = [], funcs = funcs)250            self.slFunc = FnArgs(func = const('') ,args = [], funcs = funcs)251        else:252            gcols = self['spcols']253            if 'gfunc' not in self:254                self.spFunc = FnArgs(func = joiner, args = gcols, funcs = funcs)255            else:256                self.spFunc = FnArgs(func = self['gfunc'], args = gcols, funcs = funcs)257            glcols = self['slcols' if 'slcols' in self else 'spcols']258            if 'slfunc' not in self:259                self.slFunc = FnArgs(func = joiner, args = glcols, funcs = funcs)260            else:261                self.slFunc = FnArgs(func = self['slfunc'], args = glcols, funcs = funcs)262        self.aggFunc = avg # tell typechecker that this is the real type263        if self['aggfunc']:264            self.aggFunc = mkFunc(self['aggfunc'],funcs)  # type: ignore265        self.seen = set() # type: set ### used to avoid plotting the same legend entries multiple times266    def csv(self, pth : str) -> None:267        '''Write plot data to a csv'''268        raise NotImplementedError269    def _process_group_dict(self, d : dict)->dict:270        return d # do nothing271    def _process_subgroup_dict(self,d:dict)->dict:272        return dict(val = self.xFunc(d), l = self.lFunc(d))273    def _draw(self, g : Group) -> dict:274        color = mkStyle(g.rep).color275        subgroups_ = self._make_groups(g.elems,self.spFunc,self.slFunc)276        subgroups  = [sg.map(self._process_subgroup_dict) for sg in subgroups_]277        vals = [self.aggFunc(sg['val']) for sg in subgroups]278        color = mkStyle(g.rep).color279        return dict(type = 'bar',280                    name = g.label,281                    x    = [sg.label for sg in subgroups],282                    y    = vals,283                    )#marker = dict(color=color))...

tko_rpc_utils.py

Source:tko_rpc_utils.py

...106                return rpc_utils.NULL_DATE107            if field.endswith('_time'):108                return rpc_utils.NULL_DATETIME109        return value110    def _process_group_dict(self, group_dict):111        # compute and aggregate header groups112        for i, group in enumerate(self._header_groups):113            header = tuple(self._get_field(group_dict, field)114                           for field in group)115            self._header_value_sets[i].add(header)116            group_dict.setdefault('header_values', []).append(header)117        # frontend's SelectionManager needs a unique ID118        group_values = [group_dict[field] for field in self._group_by]119        group_dict['id'] = str(group_values)120        return group_dict121    def _find_header_value_set(self, field):122        for i, group in enumerate(self._header_groups):123            if [field] == group:124                return self._header_value_sets[i]125        raise RuntimeError('Field %s not found in header groups %s' %126                           (field, self._header_groups))127    def _add_fixed_headers(self):128        for field, extra_values in self._fixed_headers.iteritems():129            header_value_set = self._find_header_value_set(field)130            for value in extra_values:131                header_value_set.add((value,))132    def _get_sorted_header_values(self):133        self._add_fixed_headers()134        sorted_header_values = [sorted(value_set)135                                for value_set in self._header_value_sets]136        # construct dicts mapping headers to their indices, for use in137        # replace_headers_with_indices()138        self._header_index_maps = []139        for value_list in sorted_header_values:140            index_map = dict((value, i) for i, value in enumerate(value_list))141            self._header_index_maps.append(index_map)142        return sorted_header_values143    def _replace_headers_with_indices(self, group_dict):144        group_dict['header_indices'] = [index_map[header_value]145                                        for index_map, header_value146                                        in zip(self._header_index_maps,147                                               group_dict['header_values'])]148        for field in self._group_by + ['header_values']:149            del group_dict[field]150    def process_group_dicts(self):151        self._fetch_data()152        if len(self._group_dicts) > self._MAX_GROUP_RESULTS:153            raise TooManyRowsError(154                'Query yielded %d rows, exceeding maximum %d' % (155                len(self._group_dicts), self._MAX_GROUP_RESULTS))156        for group_dict in self._group_dicts:157            self._process_group_dict(group_dict)158        self._header_values = self._get_sorted_header_values()159        if self._header_groups:160            for group_dict in self._group_dicts:161                self._replace_headers_with_indices(group_dict)162    def get_info_dict(self):163        return {'groups' : self._group_dicts,...

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.