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

How to use _sort_dfs method in lisa

Best Python code snippet using lisa_python

transformer.py

Source:transformer.py

...71        all[transformer.name] = transformer72    # build new sorted results73    for transformer in transformers:74        if transformer.name not in visited:75            _sort_dfs(all, transformer, visited, sorted_transformers)76    # sort by phase: init, expanded.77    init_transformers: List[schema.Transformer] = []78    expanded_transformers: List[schema.Transformer] = []79    cleanup_transformers: List[schema.Transformer] = []80    for transformer in sorted_transformers:81        if transformer.phase == constants.TRANSFORMER_PHASE_INIT:82            init_transformers.append(transformer)83        elif transformer.phase == constants.TRANSFORMER_PHASE_EXPANDED:84            expanded_transformers.append(transformer)85        elif transformer.phase == constants.TRANSFORMER_PHASE_CLEANUP:86            cleanup_transformers.append(transformer)87        else:88            raise LisaException(f"unknown transformer phase: {transformer.phase}")89    sorted_transformers = (90        init_transformers + expanded_transformers + cleanup_transformers91    )92    # check cycle reference93    referenced: Set[str] = set()94    for transformer in sorted_transformers:95        for item in transformer.depends_on:96            if item not in referenced:97                raise LisaException(98                    f"found cycle dependent transformers: "99                    f"'{transformer.name}' and '{item}'"100                )101        referenced.add(transformer.name)102    return sorted_transformers103def _sort_dfs(104    transformers: Dict[str, schema.Transformer],105    transformer: schema.Transformer,106    visited: Set[str],107    sorted_transformers: List[schema.Transformer],108) -> None:109    visited.add(transformer.name)110    for item in transformer.depends_on:111        if item not in visited:112            dependent = transformers.get(item, None)113            if not dependent:114                raise LisaException(115                    f"transformer '{transformer.name}' "116                    f"depends on non-existing transformer "117                    f"'{item}'"118                )119            _sort_dfs(transformers, dependent, visited, sorted_transformers)120    sorted_transformers.append(transformer)121def _load_transformers(122    runbook_builder: RunbookBuilder,123    variables: Optional[Dict[str, VariableEntry]] = None,124) -> Dict[str, schema.Transformer]:125    transformers_data = runbook_builder.partial_resolve(126        partial_name=constants.TRANSFORMER, variables=variables127    )128    transformers = schema.load_by_type_many(schema.Transformer, transformers_data)129    return {x.name: x for x in transformers}130def _run_transformers(131    runbook_builder: RunbookBuilder,132    phase: str = constants.TRANSFORMER_PHASE_INIT,133) -> Dict[str, VariableEntry]:...

sma.py

Source:sma.py

...21            count_df = date_df.groupby(date_df.dt.floor("d")).size().reset_index(name=p)22            count_df[p] = count_df[p].rolling(window=self.window).mean()23            count_df.set_index("date", inplace=True)24            dfs.append(count_df)25        dfs = self._sort_dfs(dfs)26        charts = [self._plot_sma_all(dfs)]27        mid = int(len(participants) / 2)28        if self.filter_top:29            charts.append(self._plot_sma_top(dfs, mid))30        if self.filter_bottom:31            charts.append(self._plot_sma_bottom(dfs, mid))32        return charts33    def _plot_sma_all(self, dfs: List[pd.DataFrame]) -> Chart:34        ax = dfs[0].plot(35            title=f"Moving Avg of messages sent per day on a {self.window} day window for all chatters"36        )37        for i in range(1, len(dfs)):38            dfs[i].plot(ax=ax)39        return Chart(title="sma_activity_all", figure=ax.get_figure())40    def _plot_sma_top(self, dfs: List[pd.DataFrame], n: int) -> Chart:41        dfs = self._copy_dfs(dfs[:n])42        ax = dfs[0].plot(43            title=f"Moving Avg of messages sent per day on a {self.window} day window for top chatters"44        )45        for i in range(1, len(dfs)):46            dfs[i].plot(ax=ax)47        return Chart(title="sma_activity_top", figure=ax.get_figure())48    def _plot_sma_bottom(self, dfs: List[pd.DataFrame], n: int) -> Chart:49        dfs = self._copy_dfs(dfs[n:])50        ax = dfs[0].plot(51            title=f"Moving Avg of messages sent per day on a {self.window} day window for bottom chatters"52        )53        for i in range(1, len(dfs)):54            dfs[i].plot(ax=ax)55        return Chart(title="sma_activity_bottom", figure=ax.get_figure())56    def _copy_dfs(self, dfs: List[pd.DataFrame]) -> List[pd.DataFrame]:57        return [df.copy(deep=True) for df in dfs]58    def _sort_dfs(self, dfs: List[pd.DataFrame]) -> List[pd.DataFrame]:59        sender_counts = []60        for df in dfs:61            name = [c for c in df.columns][0]62            sender_counts.append((df[name].sum(), df))63        sender_counts.sort(key=lambda x: x[0], reverse=True)...

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.