Plot Service¶

def generate_plot(
    self,
    use_case_id: str,
    data: pd.DataFrame,
    filters: Optional[Dict[str, Any]] = None,
    customizations: Optional[Any] = None,
    force_refresh: bool = False,
) -> go.Figure:
    """
    Generate plot for given use case with caching.

    Pipeline:
    1. Load configuration from YAML
    2. Generate cache keys
    3. Check graph cache
    4. If miss: Create strategy and generate plot
    5. Cache result
    6. Return figure

    Parameters
    ----------
    use_case_id : str
        Use case identifier (e.g., "UC-2.1").
    data : pd.DataFrame
        Input data.
    filters : Optional[Dict[str, Any]], default=None
        Filters to apply (e.g., {"uc-2-1-range-slider": [10, 50]}).
    customizations : Optional[Any], default=None
        Additional customizations (future feature).
    force_refresh : bool, default=False
        Force cache refresh.

    Returns
    -------
    go.Figure
        Generated Plotly figure.

    Raises
    ------
    ValueError
        If data validation fails.
    FileNotFoundError
        If configuration file not found.

    Examples
    --------
    >>> service = PlotService()
    >>> fig = service.generate_plot("UC-2.1", df)
    >>> fig.layout.height
    500
    """
    start_time = time.time()
    logger.info(f"Generating plot for {use_case_id}")

    # 1. Load configuration (force reload if force_refresh is True)
    config = self.config_loader.load_config(use_case_id, force_reload=force_refresh)
    perf_config = config.get("performance", {})
    cache_config = perf_config.get("cache", {})

    # 2. Generate cache keys
    data_hash = self._generate_data_hash(data)
    filters_hash = self._generate_filters_hash(filters) if filters else "no_filters"

    # Generate cache key (needed for both checking and storing)
    graph_cache_key = self._get_cache_key(config, "graph", data_hash, filters_hash)

    # 3. Check if caching is enabled
    if cache_config.get("enabled", True) and not force_refresh:
        # Check graph cache (fastest)
        cached_figure = self.cache_manager.get_cached_graph(graph_cache_key)
        if cached_figure:
            cache_time = time.time() - start_time
            logger.info(
                f"Cache HIT (graph) for {use_case_id}: " f"{cache_time:.3f}s"
            )
            return cached_figure

        logger.debug(f"Cache MISS (graph) for {use_case_id}")

    # 4. Create strategy via factory
    strategy = self.factory.create_strategy(config)

    # 5. Generate plot (includes validation, processing, filtering)
    try:
        figure = strategy.generate_plot(
            data, filters=filters, customizations=customizations
        )

        # 6. Cache the figure
        if cache_config.get("enabled", True):
            ttl = self._get_cache_ttl(cache_config, "graph")
            # Note: ttl is not used by current GraphCacheManager
            # (it uses global TTL), but we pass metadata
            metadata = {"use_case_id": use_case_id, "filters": filters, "ttl": ttl}
            self.cache_manager.cache_graph(
                graph_cache_key, figure, metadata=metadata
            )
            logger.debug(f"Cached graph for {use_case_id} (TTL: {ttl}s)")

        total_time = time.time() - start_time
        logger.info(f"Plot generated for {use_case_id}: {total_time:.3f}s")

        return figure

    except Exception as e:
        logger.error(f"Error generating plot for {use_case_id}: {e}", exc_info=True)
        raise

def clear_cache(self, use_case_id: Optional[str] = None) -> None:
    """
    Clear cache for specific use case or all.

    Parameters
    ----------
    use_case_id : Optional[str], default=None
        Use case ID to clear. If None, clears all.
    """
    if use_case_id:
        # Clear specific use case (would need pattern matching)
        logger.info(f"Clearing cache for {use_case_id}")
    else:
        self.cache_manager.clear()
        logger.info("Cleared all plot caches")