def _build_chart_sub_tables()

in azure/Kqlmagic/column_guesser.py [0:0]

• 95 lines of code
• 56 McCabe index (conditional complexity)

    def _build_chart_sub_tables(self, properties:dict, name=None, x_type="first") -> list:
        self.chart_sub_tables = []
        self._build_columns(name, without_data=True)

        #
        # discover column X index
        #
        x_col_idx = None
        x_col_name = properties.get(VisualizationKeys.X_COLUMN)
        if x_col_name is not None:
            for idx, c in enumerate(self.columns):
                if c.name == x_col_name:
                    x_col_idx = idx
                    break
        elif x_type == "first":
            x_col_idx = 0
        elif x_type == "quantity":
            for idx, c in enumerate(self.columns):
                if c.is_quantity:
                    if not c.is_datetime: 
                        x_col_idx = idx
                        break
                    elif x_col_idx is None:
                        x_col_idx = idx
        elif x_type == "datetime":
            # find first of type datetime
            for idx, c in enumerate(self.columns):
                if c.is_datetime:
                    x_col_idx = idx
                    break
        if x_col_idx is None:
            # this print is not for debug
            print("No valid xcolumn")
            return []

        rows = self

        #
        # discover x direction, and always sort ascending
        #
        is_descending_sorted = None
        if self.columns[x_col_idx].is_quantity and len(rows) >= 2:
            if properties.get(VisualizationKeys.IS_QUERY_SORTED) is True:
                previous_col_value = rows[0][x_col_idx]
                is_descending_sorted = True
                for r in rows[1:]:
                    current_col_value = r[x_col_idx]
                    if previous_col_value < current_col_value:
                        is_descending_sorted = False
                        break
                    previous_col_value = current_col_value

            rows = list(reversed(list(self))) if is_descending_sorted else sorted(self, key=lambda row: row[x_col_idx])

        #
        # create a new unique list of col_x values (keep same order)
        #
        col_x = Column(col=self.columns[x_col_idx])
        for row in rows:
            if row[x_col_idx] not in col_x:
                col_x.append(row[x_col_idx])

        #
        # discover series columns (each combination of values in this columns, is a serie)
        #
        specified_series_columns = properties.get(VisualizationKeys.SERIES) or []
        series_columns = [c for idx, c in enumerate(self.columns) if c.name in specified_series_columns or (idx != x_col_idx and not c.is_quantity)]

        #
        # discover y columns
        #
        y_cols_name = properties.get(VisualizationKeys.Y_COLUMNS)
        if y_cols_name is not None:
            quantity_columns = [c for c in self.columns if c.name in y_cols_name and c.is_quantity]
        else:
            quantity_columns = [c for idx, c in enumerate(self.columns) if idx != x_col_idx and c.is_quantity and c.name not in [s.name for s in series_columns]] 
        if len(quantity_columns) < 1:
            # this print is not for debug
            print("No valid ycolumns")
            return []

        #
        # create chart sub-tables
        # a sub-table for each serie X y-col
        #
        chart_sub_tables_dict = {}
        for row in rows:
            for qcol in quantity_columns:
                if len(series_columns) > 0:
                    sub_table_name = f'{":".join([str(row[col.idx]) for col in series_columns])}:{qcol.name}'
                else:
                    sub_table_name = qcol.name
                chart_sub_table = chart_sub_tables_dict.get(sub_table_name)
                if chart_sub_table is None:
                    chart_sub_table = chart_sub_tables_dict[sub_table_name] = ChartSubTable(
                        name=sub_table_name,
                        col_x=Column(col=self.columns[x_col_idx]),
                        col_y=Column(col=qcol),
                        mapping=dict(zip(col_x, [None for i in range(len(col_x))])),
                        is_descending_sorted=is_descending_sorted,
                    )
                chart_sub_table[row[x_col_idx]] = datetime_to_linear_ticks(row[qcol.idx]) if qcol.is_datetime else row[qcol.idx]
        self.chart_sub_tables = list(chart_sub_tables_dict.values())

        col_y_min = properties.get(VisualizationKeys.Y_MIN)
        col_y_min = col_y_min if is_quantity(col_y_min) else None

        col_y_max = properties.get(VisualizationKeys.Y_MAX)
        col_y_max = col_y_max if is_quantity(col_y_max) else None

        for tab in self.chart_sub_tables:
            tab.col_y_min = col_y_min
            tab.col_y_max = col_y_max
            if tab.col_y_min is None and tab.col_y_max is None:
                pass
            elif tab.col_y_min is not None and tab.col_y_max is not None:
                pass
            elif tab.col_y_min is None:
                try:
                    tab.col_y_min = min(min(filter(lambda x: x is not None, tab.values())), 0) * 1.1
                except:
                    tab.col_y_min = None
            elif tab.col_y_max is None:
                try:
                    tab.col_y_max = max(filter(lambda x: x is not None, tab.values())) * 1.1
                except:
                    tab.col_y_max = None

        return self.chart_sub_tables