Network Strategy¶

def validate_data(self, df: pd.DataFrame) -> None:
    """
    Validate input data for network diagram requirements.

    Parameters
    ----------
    df : pd.DataFrame
        Input data to validate.

    Raises
    ------
    ValueError
        If DataFrame is empty, required columns missing, or mode
        configuration invalid.
    """
    logger.debug(
        f"Validating data - Shape: {df.shape}, " f"Columns: {df.columns.tolist()}"
    )

    if df.empty:
        raise ValueError("Input DataFrame is empty")

    # Mode-specific validation
    if self.mode == "bipartite":
        required_cols = [self.source_column, self.target_column]
        missing = [c for c in required_cols if c not in df.columns]
        if missing:
            raise ValueError(
                f"Missing columns for bipartite mode: {missing}. "
                f"Available: {df.columns.tolist()}"
            )

    elif self.mode == "similarity":
        if not self.group_by_column or not self.shared_column:
            raise ValueError(
                "Similarity mode requires 'group_by_column' and "
                "'shared_column' configuration."
            )
        required_cols = [self.group_by_column, self.shared_column]
        missing = [c for c in required_cols if c not in df.columns]
        if missing:
            raise ValueError(
                f"Missing columns for similarity mode: {missing}. "
                f"Available: {df.columns.tolist()}"
            )

    else:
        raise ValueError(
            f"Unknown mode: '{self.mode}'. " f"Supported: 'bipartite', 'similarity'"
        )

    logger.info(f"Data validation passed - {len(df)} records, mode='{self.mode}'")

def process_data(self, df: pd.DataFrame) -> pd.DataFrame:
    """
    Process data and create graph structure for network visualization.

    Creates edge list based on configured mode (bipartite or similarity).

    Parameters
    ----------
    df : pd.DataFrame
        Input data with required columns.

    Returns
    -------
    pd.DataFrame
        Graph data as DataFrame with source, target, and weight columns.
    """
    logger.info(f"Processing data in '{self.mode}' mode...")

    if self.mode == "bipartite":
        graph_data = self._process_bipartite(df)
    elif self.mode == "similarity":
        graph_data = self._process_similarity(df)
    else:
        raise ValueError(f"Unknown mode: {self.mode}")

    if graph_data.empty:
        raise ValueError(
            "No valid edges after processing. "
            "Check data or adjust min_edge_weight."
        )

    logger.info(f"Graph data created - {len(graph_data)} edges")

    return graph_data

def create_figure(self, processed_df: pd.DataFrame) -> go.Figure:
    """
    Create network diagram figure from processed graph data.

    Parameters
    ----------
    processed_df : pd.DataFrame
        Processed edge list data.

    Returns
    -------
    go.Figure
        Configured Plotly figure with network diagram.
    """
    logger.debug("Creating network diagram figure...")

    chart_config = self.plotly_config.get("chart", {})
    layout_config = self.plotly_config.get("layout", {})

    # Build NetworkX graph
    G = self._build_graph(processed_df)

    # Calculate node positions
    pos = self._calculate_layout(G)

    # Handle title configuration (support both string and dict)
    title_config = chart_config.get("title", {})
    if isinstance(title_config, str):
        # Backward compatibility: string title
        show_title = True
        title_text = title_config
        title_font_size = 16
    else:
        # New format: dict with show, text, font
        show_title = title_config.get("show", True)
        title_text = (
            title_config.get("text", "Network Diagram") if show_title else ""
        )
        title_font_size = title_config.get("font", {}).get("size", 16)

    # Create traces
    traces = []

    # Create edge traces
    edge_traces = self._create_edge_traces(G, pos, processed_df)
    traces.extend(edge_traces)

    # Create node traces
    if self.mode == "bipartite":
        node_traces = self._create_bipartite_node_traces(G, pos, processed_df)
    else:
        node_traces = self._create_similarity_node_traces(G, pos)

    traces.extend(node_traces)

    # Create figure
    fig = go.Figure(data=traces)

    # Layout configuration
    height = layout_config.get("height", 800)
    use_autosize = layout_config.get("autosize", False)

    # Show legend configuration
    show_legend_config = chart_config.get("show_legend", None)
    if show_legend_config is not None:
        show_legend = show_legend_config
    else:
        # Default: show legend only for bipartite mode
        show_legend = self.mode == "bipartite"

    # Get margin configuration
    margin_config = layout_config.get("margin", {})
    margin = dict(
        l=margin_config.get("l", 20),
        r=margin_config.get("r", 20),
        t=margin_config.get("t", 60),
        b=margin_config.get("b", 20),
    )

    # Get background colors
    plot_bgcolor = layout_config.get("plot_bgcolor", "white")
    paper_bgcolor = layout_config.get("paper_bgcolor", "white")

    # Get template
    template = layout_config.get("template", "simple_white")

    # Build layout update dict
    layout_update = {
        "height": height,
        "showlegend": show_legend,
        "legend": dict(x=1, y=1, bordercolor="Gainsboro", borderwidth=1),
        "hovermode": "closest",
        "xaxis": dict(showgrid=False, zeroline=False, showticklabels=False),
        "yaxis": dict(showgrid=False, zeroline=False, showticklabels=False),
        "plot_bgcolor": plot_bgcolor,
        "paper_bgcolor": paper_bgcolor,
        "template": template,
        "margin": margin,
    }

    # Add title if enabled
    if show_title and title_text:
        layout_update["title"] = dict(
            text=title_text,
            x=0.5,
            xanchor="center",
            font=dict(size=title_font_size),
        )

    # Add autosize or width
    if use_autosize:
        layout_update["autosize"] = True
    else:
        if layout_config.get("width"):
            layout_update["width"] = layout_config.get("width")

    fig.update_layout(**layout_update)

    logger.info(
        f"Network diagram created - "
        f"{G.number_of_nodes()} nodes, {G.number_of_edges()} edges, "
        f"Layout: {self.layout_algorithm}"
    )

    return fig

def apply_filters(
    self, df: pd.DataFrame, filters: Optional[Dict[str, Any]] = None
) -> pd.DataFrame:
    """
    Apply filters to data.

    This is a common implementation that can be overridden
    by subclasses if needed.

    Parameters
    ----------
    df : pd.DataFrame
        Data to filter.
    filters : Optional[Dict[str, Any]], default=None
        Filter specifications.

    Returns
    -------
    pd.DataFrame
        Filtered data.
    """
    import logging

    logger = logging.getLogger(__name__)

    if not filters:
        logger.debug("No filters provided, returning original data")
        return df

    logger.info(
        f"Applying filters - Input shape: {df.shape}, "
        f"Columns: {df.columns.tolist()}"
    )
    logger.info(f"Filters to apply: {filters}")

    filtered_df = df.copy()

    # Get filter configurations
    filter_configs = self.config.get("filters", [])

    for filter_config in filter_configs:
        filter_id = filter_config.get("filter_id")
        filter_type = filter_config.get("type")

        if filter_id not in filters:
            continue

        filter_value = filters[filter_id]
        data_binding = filter_config.get("data_binding", {})
        column = data_binding.get("column")

        if not column or column not in filtered_df.columns:
            logger.warning(
                f"Filter '{filter_id}': Column '{column}' not found. "
                f"Available: {filtered_df.columns.tolist()}"
            )
            continue

        # Apply range filter
        if filter_type == "range" and isinstance(filter_value, list):
            min_val, max_val = filter_value
            logger.info(
                f"Applying range filter on '{column}': " f"[{min_val}, {max_val}]"
            )
            filtered_df = filtered_df[
                (filtered_df[column] >= min_val) & (filtered_df[column] <= max_val)
            ]
            logger.info(f"After filter: {len(filtered_df)} rows remaining")

    logger.info(f"Final filtered shape: {filtered_df.shape}")
    return filtered_df

def apply_customizations(
    self, fig: go.Figure, customizations: Optional[Any] = None
) -> go.Figure:
    """
    Apply custom styling to figure.

    This is a hook for future customization features
    (FLEXIVEL and FLEXIVEL2).

    Parameters
    ----------
    fig : go.Figure
        Base figure.
    customizations : Optional[Any], default=None
        Customization specifications.

    Returns
    -------
    go.Figure
        Customized figure.
    """
    # Hook for future implementation
    return fig

def generate_plot(
    self,
    data: pd.DataFrame,
    filters: Optional[Dict[str, Any]] = None,
    customizations: Optional[Any] = None,
) -> go.Figure:
    """
    Generate complete plot (Template Method).

    This method orchestrates the entire plot generation process:
    1. Validate input data
    2. Process data
    3. Apply filters
    4. Create figure
    5. Apply customizations

    Parameters
    ----------
    data : pd.DataFrame
        Input data.
    filters : Optional[Dict[str, Any]], default=None
        Filters to apply.
    customizations : Optional[Any], default=None
        Customizations to apply.

    Returns
    -------
    go.Figure
        Complete Plotly figure.

    Raises
    ------
    ValueError
        If validation fails.
    """
    # 1. Validate
    self.validate_data(data)

    # 2. Process
    processed_df = self.process_data(data)

    # 3. Filter
    filtered_df = self.apply_filters(processed_df, filters)

    # 4. Create figure
    figure = self.create_figure(filtered_df)

    # 5. Apply customizations (hook for future)
    figure = self.apply_customizations(figure, customizations)

    return figure