Adapters

This page documents the adapter modules under jumper_extension.adapters that implement reporting, visualization, session management, and script writing. High‑level usage is described in the Public API and Jupyter API sections; the content below is generated directly from the Python code.

Data

PerformanceData

Source code in jumper_extension/adapters/data.py

class PerformanceData:
    def __init__(self, num_cpus, num_system_cpus, num_gpus):
        self.num_cpus = num_cpus
        self.num_system_cpus = num_system_cpus
        self.num_gpus = num_gpus
        self.levels = get_available_levels()
        # Minimal required base columns for loaded performance data
        self._base_columns = [
            "time",
            "memory",
            "io_read_count",
            "io_write_count",
            "io_read",
            "io_write",
            "cpu_util_avg",
            "cpu_util_min",
            "cpu_util_max",
        ]
        self.data = {
            level: self._initialize_dataframe(level) for level in self.levels
        }
        # File writers/readers mappings
        self._file_writers = {
            "json": self._write_json,
            "csv": self._write_csv,
        }
        self._file_readers = {
            "json": pd.read_json,
            "csv": pd.read_csv,
        }

    def _validate_level(self, level):
        if level not in self.levels:
            raise ValueError(
                EXTENSION_ERROR_MESSAGES[
                    ExtensionErrorCode.INVALID_LEVEL
                ].format(level=level, levels=self.levels)
            )

    def _initialize_dataframe(self, level):

        effective_num_cpus = (
            self.num_system_cpus if level == "system" else self.num_cpus
        )

        columns = self._base_columns + [f"cpu_util_{i}" for i in range(effective_num_cpus)]

        if self.num_gpus > 0:
            gpu_metrics = ["util", "band", "mem"]
            columns.extend(
                [
                    f"gpu_{metric}_{stat}"
                    for metric in gpu_metrics
                    for stat in ["avg", "min", "max"]
                ]
            )
            columns.extend(
                [
                    f"gpu_{metric}_{i}"
                    for i in range(self.num_gpus)
                    for metric in gpu_metrics
                ]
            )

        return pd.DataFrame(columns=columns)

    def _attach_cell_index(self, df, cell_history) -> pd.DataFrame:
        result = df.copy()
        result["cell_index"] = pd.NA
        times = result["time"].to_numpy()
        for row in cell_history.data.itertuples(index=False):
            mask = (times >= row.start_time) & (times <= row.end_time)
            result.loc[mask, "cell_index"] = row.cell_index
        return result

    def _write_json(self, filename: str, df: pd.DataFrame) -> None:
        with open(filename, "w") as f:
            json.dump(df.to_dict("records"), f, indent=2)

    def _write_csv(self, filename: str, df: pd.DataFrame) -> None:
        df.to_csv(filename, index=False)

    def view(self, level="process", slice_=None, cell_history=None):
        """View data for a specific level with optional slicing."""
        self._validate_level(level)
        base = (
            self.data[level]
            if slice_ is None
            else self.data[level].iloc[slice_[0] : slice_[1] + 1]
        )
        return (
            self._attach_cell_index(base, cell_history)
            if cell_history is not None
            else base
        )

    def add_sample(
        self,
        level,
        time_mark,
        cpu_util_per_core,
        memory,
        gpu_util,
        gpu_band,
        gpu_mem,
        io_counters,
    ):
        self._validate_level(level)
        effective_num_cpus = (
            self.num_system_cpus if level == "system" else self.num_cpus
        )

        last_timestamp = 0
        if len(self.data[level]):
            last_timestamp = self.data[level].loc[len(self.data[level]) - 1][
                "time"
            ]

        cumulative_metrics_ratio = time_mark - last_timestamp
        row_data = {
            "time": time_mark,
            "memory": memory,
            "io_read_count": io_counters[0] / cumulative_metrics_ratio,
            "io_write_count": io_counters[1] / cumulative_metrics_ratio,
            "io_read": io_counters[2] / cumulative_metrics_ratio,
            "io_write": io_counters[3] / cumulative_metrics_ratio,
            "cpu_util_avg": sum(cpu_util_per_core) / effective_num_cpus,
            "cpu_util_min": min(cpu_util_per_core),
            "cpu_util_max": max(cpu_util_per_core),
            **{
                f"cpu_util_{i}": cpu_util_per_core[i]
                for i in range(effective_num_cpus)
            },
        }

        if self.num_gpus > 0:
            gpu_data = {"util": gpu_util, "band": gpu_band, "mem": gpu_mem}
            for metric, values in gpu_data.items():
                row_data.update(
                    {
                        f"gpu_{metric}_avg": sum(values) / self.num_gpus,
                        f"gpu_{metric}_min": min(values),
                        f"gpu_{metric}_max": max(values),
                        **{
                            f"gpu_{metric}_{i}": values[i]
                            for i in range(self.num_gpus)
                        },
                    }
                )

        self.data[level].loc[len(self.data[level])] = row_data

    def export(
        self,
        filename="performance_data.csv",
        level="process",
        cell_history=None,
    ):
        """Export performance data to JSON or CSV."""
        self._validate_level(level)
        if len(self.data[level]) == 0:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[
                    ExtensionErrorCode.NO_PERFORMANCE_DATA
                ]
            )
            return

        # Determine format from filename extension
        _, ext = os.path.splitext(filename)
        format = ext.lower().lstrip(".")

        # Default to csv if no extension provided
        if not format:
            format = "csv"
            filename += ".csv"

        df_to_write = (
            self._attach_cell_index(self.data[level], cell_history)
            if cell_history is not None
            else self.data[level]
        )

        writer = self._file_writers.get(format)
        if writer is None:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[
                    ExtensionErrorCode.UNSUPPORTED_FORMAT
                ].format(
                    format=format,
                    supported_formats=", ".join(["json", "csv"]),
                )
            )
            return
        writer(filename, df_to_write)

        logger.info(
            EXTENSION_INFO_MESSAGES[ExtensionInfoCode.EXPORT_SUCCESS].format(
                filename=filename
            )
        )

    def load(self, filename: str) -> Optional[pd.DataFrame]:
        """Load performance data from CSV or JSON file.

        Returns:
            DataFrame if successful, None otherwise
        """
        return load_dataframe_from_file(
            filename,
            self._file_readers,
            self._base_columns,
            entity_name="performance data",
        )

export(filename='performance_data.csv', level='process', cell_history=None)

Export performance data to JSON or CSV.

Source code in jumper_extension/adapters/data.py

def export(
    self,
    filename="performance_data.csv",
    level="process",
    cell_history=None,
):
    """Export performance data to JSON or CSV."""
    self._validate_level(level)
    if len(self.data[level]) == 0:
        logger.warning(
            EXTENSION_ERROR_MESSAGES[
                ExtensionErrorCode.NO_PERFORMANCE_DATA
            ]
        )
        return

    # Determine format from filename extension
    _, ext = os.path.splitext(filename)
    format = ext.lower().lstrip(".")

    # Default to csv if no extension provided
    if not format:
        format = "csv"
        filename += ".csv"

    df_to_write = (
        self._attach_cell_index(self.data[level], cell_history)
        if cell_history is not None
        else self.data[level]
    )

    writer = self._file_writers.get(format)
    if writer is None:
        logger.warning(
            EXTENSION_ERROR_MESSAGES[
                ExtensionErrorCode.UNSUPPORTED_FORMAT
            ].format(
                format=format,
                supported_formats=", ".join(["json", "csv"]),
            )
        )
        return
    writer(filename, df_to_write)

    logger.info(
        EXTENSION_INFO_MESSAGES[ExtensionInfoCode.EXPORT_SUCCESS].format(
            filename=filename
        )
    )

load(filename)

Load performance data from CSV or JSON file.

Returns:

Type	Description
Optional[DataFrame]	DataFrame if successful, None otherwise

Source code in jumper_extension/adapters/data.py

def load(self, filename: str) -> Optional[pd.DataFrame]:
    """Load performance data from CSV or JSON file.

    Returns:
        DataFrame if successful, None otherwise
    """
    return load_dataframe_from_file(
        filename,
        self._file_readers,
        self._base_columns,
        entity_name="performance data",
    )

view(level='process', slice_=None, cell_history=None)

View data for a specific level with optional slicing.

Source code in jumper_extension/adapters/data.py

def view(self, level="process", slice_=None, cell_history=None):
    """View data for a specific level with optional slicing."""
    self._validate_level(level)
    base = (
        self.data[level]
        if slice_ is None
        else self.data[level].iloc[slice_[0] : slice_[1] + 1]
    )
    return (
        self._attach_cell_index(base, cell_history)
        if cell_history is not None
        else base
    )

Cell history and analysis

CellHistory

Source code in jumper_extension/adapters/cell_history.py

class CellHistory:
    def __init__(self):
        self._columns = [
            "cell_index",
            "raw_cell",
            "start_time",
            "end_time",
            "duration",
            "wallclock_start_time",
            "wallclock_end_time",
        ]
        self.data = pd.DataFrame(columns=self._columns)
        self.file_readers = {
            "json": pd.read_json,
            "csv": pd.read_csv,
        }
        self.current_cell = None

    def start_cell(self, raw_cell: str, cell_magics: List[str]):
        self.current_cell = {
            "cell_index": len(self.data),
            "cell_magics": cell_magics,
            "raw_cell": raw_cell,
            "start_time": time.perf_counter(),
            "end_time": None,
            "duration": None,
            "wallclock_start_time": time.time(),
            "wallclock_end_time": None,
        }

    def end_cell(self, result):
        if self.current_cell:
            self.current_cell["end_time"] = time.perf_counter()
            self.current_cell["duration"] = (
                self.current_cell["end_time"] - self.current_cell["start_time"]
            )
            self.current_cell["wallclock_end_time"] = time.time()

            new_row = pd.DataFrame([self.current_cell])
            with warnings.catch_warnings():
                warnings.simplefilter("ignore", FutureWarning)
                warnings.simplefilter("ignore", pd.errors.PerformanceWarning)
                self.data = pd.concat([self.data, new_row], ignore_index=True)
            self.current_cell = None

    def view(self, start=None, end=None):
        if start is None and end is None:
            return self.data
        return self.data.iloc[start:end]

    def print(self):
        for _, cell in self.data.iterrows():
            print(
                f"Cell #{int(cell['cell_index'])} - Duration: "
                f"{cell['duration']:.2f}s"
            )
            print("-" * 40)
            print(cell["raw_cell"])
            print("=" * 40)

    def show_itable(self):
        if self.data.empty:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[ExtensionErrorCode.NO_CELL_HISTORY]
            )
            return

        data = []
        for _, row in self.data.iterrows():
            duration = row["end_time"] - row["start_time"]
            data.append(
                {
                    "Cell index": row["cell_index"],
                    "Duration (s)": f"{duration:.2f}",
                    "Start Time": time.strftime(
                        "%H:%M:%S", time.localtime(row["start_time"])
                    ),
                    "End Time": time.strftime(
                        "%H:%M:%S", time.localtime(row["end_time"])
                    ),
                    "Code": row["raw_cell"].replace("\n", "<br>"),
                }
            )

        df = pd.DataFrame(data)

        # To avoid warnings about a non-documented 'escape' option in a notebook
        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", category=SyntaxWarning, module="itables\\.typing")
            show(
                df,
                layout={"topStart": "search", "topEnd": None},
                columnDefs=[
                    {"targets": [4], "className": "dt-left"}
                ],  # 4 - "Code" index
                escape=False,
            )

    def export(self, filename="cell_history.json"):
        if self.data.empty:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[ExtensionErrorCode.NO_CELL_HISTORY]
            )
            return

        # Determine format from filename extension
        _, ext = os.path.splitext(filename)
        format = ext.lower().lstrip(".")

        # Default to csv if no extension provided
        if not format:
            format = "csv"
            filename += ".csv"

        if format == "json":
            with open(filename, "w") as f:
                json.dump(self.data.to_dict("records"), f, indent=2)
        elif format == "csv":
            self.data.to_csv(filename, index=False)
        else:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[
                    ExtensionErrorCode.UNSUPPORTED_FORMAT
                ].format(
                    format=format,
                    supported_formats=", ".join(["json", "csv"]),
                )
            )
            return

        logger.info(
            EXTENSION_INFO_MESSAGES[ExtensionInfoCode.EXPORT_SUCCESS].format(
                filename=filename
            )
        )

    def load(self, filename: str) -> Optional[pd.DataFrame]:
        """Load cell history from CSV or JSON file.

        Returns:
            DataFrame if successful, None otherwise
        """
        return load_dataframe_from_file(
            filename,
            self.file_readers,
            self._columns,
            entity_name="cell history",
        )

    def __len__(self):
        return len(self.data)

load(filename)

Load cell history from CSV or JSON file.

Returns:

Type	Description
Optional[DataFrame]	DataFrame if successful, None otherwise

Source code in jumper_extension/adapters/cell_history.py

def load(self, filename: str) -> Optional[pd.DataFrame]:
    """Load cell history from CSV or JSON file.

    Returns:
        DataFrame if successful, None otherwise
    """
    return load_dataframe_from_file(
        filename,
        self.file_readers,
        self._columns,
        entity_name="cell history",
    )

PerformanceAnalyzer

Performance analyzer to determine workload type using relative thresholds.

Inspired by JobLabeller from: https://gitlab.hrz.tu-chemnitz.de/pika/pika-server/-/blob/ 619d62926cd85f8c20589c75aba0c6e2c51087e1/ src/post_processing/post_processing.py#L711

Source code in jumper_extension/adapters/analyzer.py

class PerformanceAnalyzer:
    """
    Performance analyzer to determine workload type using relative thresholds.

    Inspired by `JobLabeller` from:
    https://gitlab.hrz.tu-chemnitz.de/pika/pika-server/-/blob/
    619d62926cd85f8c20589c75aba0c6e2c51087e1/
    src/post_processing/post_processing.py#L711
    """
    # Default thresholds
    DEFAULT_THRESHOLDS = {
        'memory_ratio': 0.8,  # memory limit 0.80
        'cpu_ratio': 0.7,  # CPU capacity 0.70
        'gpu_util_ratio': 0.8,  # GPU utilization
        'gpu_memory_ratio': 0.8,  # GPU memory

        # --- thresholds for "GPU idle" detection
        # minimum memory usage required to treat GPU as allocated
        'gpu_alloc_min_mem_gb': 0.1,
        # minimum GPU utilization to treat GPU in idle state
        'gpu_util_idle_threshold': 0.05,
        # minimum overall usage fraction required to trigger the tag
        'gpu_alloc_min_fraction': 0.5,

    }

    def __init__(
        self,
        thresholds: Optional[Dict[str, float]] = None,
    ):
        """
        Initialize analyzer with relative thresholds

        Args:
            thresholds: Custom threshold values (uses defaults if None)
        """
        self.thresholds = {**self.DEFAULT_THRESHOLDS, **(thresholds or {})}

    def analyze_cell_performance(
        self,
        perfdata: "pd.DataFrame",
        memory_limit: float,
        gpu_memory_limit: Optional[float] = None,
    ) -> List[TagScore]:
        """
        Analyze cell performance and determine tags

        Args:
            perfdata: DataFrame with performance data
            memory_limit: System memory limit in GB
            gpu_memory_limit: GPU memory limit in GB (if available)

        Returns:
            List[TagScore]: Ranked performance tags for the cell
        """

        logger.debug(f"{memory_limit = }")
        logger.debug(f"{gpu_memory_limit = }")

        # Compute normalized metrics
        metrics = self._compute_metrics(perfdata, gpu_memory_limit)

        # Calculate resource utilization ratios
        ratios = self._calculate_utilization_ratios(metrics, memory_limit, gpu_memory_limit)

        # Create the ranked tags list
        ranked_tags = self._create_ranked_tags(ratios)

        # Detect "GPU allocated but not used" and prepend if applicable
        gpu_unused_tag = self._detect_gpu_allocated_but_not_used(perfdata, gpu_memory_limit)
        if gpu_unused_tag is not None:
            # Prepend the GPU allocated but not used as this is the most important tag
            ranked_tags = [gpu_unused_tag] +  ranked_tags

        logger.debug(f"{ranked_tags = }")

        return ranked_tags if ranked_tags else [TagScore(PerformanceTag.NORMAL, 0.0)]

    @staticmethod
    def _compute_metrics(
            perfdata: "pd.DataFrame",
            gpu_memory_limit: Optional[float],
    ) -> Dict[str, float]:
        """Compute raw performance metrics"""
        metrics = {}

        # CPU metrics
        if 'cpu_util_avg' in perfdata.columns:
            metrics['cpu_avg'] = perfdata['cpu_util_avg'].mean()

        # Memory metrics
        if 'memory' in perfdata.columns:
            metrics['memory_avg_gb'] = perfdata['memory'].mean()

        # GPU metrics
        if 'gpu_util_avg' in perfdata.columns:
            metrics['gpu_util_avg'] = perfdata['gpu_util_avg'].mean()

        if 'gpu_mem_avg' in perfdata.columns and gpu_memory_limit:
            metrics['gpu_memory_avg_gb'] = perfdata['gpu_mem_avg'].mean()

        return metrics

    def _calculate_utilization_ratios(self, metrics: Dict[str, float],
                                      memory_limit: float,
                                      gpu_memory_limit: Optional[float]) -> Dict[str, float]:
        """Calculate utilization ratios relative to system limits"""
        ratios = {}

        # Memory ratio (current usage / limit)
        memory_avg = metrics.get('memory_avg_gb', 0)
        ratios['memory'] = self._safe_ratio(memory_avg, memory_limit)

        # CPU ratio (utilization / 100%)
        cpu_avg = metrics.get('cpu_avg', 0)
        ratios['cpu'] = self._safe_ratio(cpu_avg, 100.0)

        # GPU utilization ratio
        gpu_util = metrics.get('gpu_util_avg', 0)
        ratios['gpu_util'] = self._safe_ratio(gpu_util, 100.0)

        # GPU memory ratio
        if gpu_memory_limit and gpu_memory_limit > 0:
            gpu_memory = metrics.get('gpu_memory_avg_gb', 0)
            ratios['gpu_memory'] = self._safe_ratio(gpu_memory, gpu_memory_limit)
        else:
            ratios['gpu_memory'] = 0.0

        logger.debug(f"ratios: {ratios}")

        return ratios

    @staticmethod
    def _safe_ratio(measured: float, maximum: float) -> float:
        """Safely calculate ratio with error handling"""
        try:
            if maximum is None or maximum <= 0 or measured is None:
                return 0.0
            return min(1.0, max(0.0, measured / maximum))
        except (TypeError, ZeroDivisionError):
            return 0.0

    def _create_ranked_tags(self, ratios: Dict[str, float]) -> List[TagScore]:
        """Create the ranked list of tags based on ratios (0.0-1.0 scale)"""

        # Sort by descending ratios
        sorted_ratios = sorted(ratios.items(), key=lambda x: x[1], reverse=True)

        # Create ranked tags for resources that exceed the minimum threshold
        tag_mapping = {
            'cpu': PerformanceTag.CPU_BOUND,
            'memory': PerformanceTag.MEMORY_BOUND,
            'gpu_util': PerformanceTag.GPU_UTIL_BOUND,
            'gpu_memory': PerformanceTag.GPU_MEMORY_BOUND,
        }

        ranked_tags = []

        for resource, ratio in sorted_ratios:
            threshold_key = f'{resource}_ratio'
            threshold = self.thresholds.get(threshold_key, 0.0)
            if ratio >= threshold:
                tag = tag_mapping.get(resource)
                if tag:
                    ranked_tags.append(TagScore(tag, ratio))

        return ranked_tags

    def _detect_gpu_allocated_but_not_used(
        self,
        perfdata: "pd.DataFrame",
        gpu_memory_limit: Optional[float],
    ) -> Optional[TagScore]:
        """
        Detect case when GPU memory is allocated but GPU compute utilization stays idle
        for a significant fraction of measurement time.
        """
        # must have GPU columns and a GPU present
        if gpu_memory_limit is None:
            return None
        if 'gpu_mem_avg' not in perfdata.columns or 'gpu_util_avg' not in perfdata.columns:
            return None
        if perfdata.empty:
            return None

        memory_threshold_gb = max(float(self.thresholds.get('gpu_alloc_min_mem_gb', 0.1)), 0.0)
        utilization_idle_threshold = float(self.thresholds.get('gpu_util_idle_threshold', 0.05))  # 0..1
        min_fraction = float(self.thresholds.get('gpu_alloc_min_fraction', 0.5))        # 0..1

        # allocation considered if memory usage exceeds memory_threshold_gb
        mask_allocated = perfdata['gpu_mem_avg'] > memory_threshold_gb
        if mask_allocated.sum() == 0:
            return None

        # idle if util ≤ util_idle_thr * 100 (%)
        mask_idle = perfdata['gpu_util_avg'] <= (utilization_idle_threshold * 100.0)

        mask_allocated_and_idle = mask_allocated & mask_idle
        frac = float(mask_allocated_and_idle.mean())

        logger.debug(f"GPU idle check:")
        logger.debug(f"gpu_mem_avg:\n{perfdata['gpu_mem_avg']}")
        logger.debug(f"mask_allocated:\n{mask_allocated}\n")
        logger.debug(f"gpu_util_avg:\n{perfdata['gpu_util_avg']}")
        logger.debug(f"mask_idle:\n{mask_idle}\n")
        logger.debug(f"GPU not used {min_fraction = }")
        logger.debug(f"GPU not used {frac = }")

        if frac >= min_fraction:
            return TagScore(PerformanceTag.GPU_ALLOCATED_BUT_NOT_USED, frac)
        return None

__init__(thresholds=None)

Initialize analyzer with relative thresholds

Parameters:

Name	Type	Description	Default
thresholds	Optional[Dict[str, float]]	Custom threshold values (uses defaults if None)	None

Source code in jumper_extension/adapters/analyzer.py

def __init__(
    self,
    thresholds: Optional[Dict[str, float]] = None,
):
    """
    Initialize analyzer with relative thresholds

    Args:
        thresholds: Custom threshold values (uses defaults if None)
    """
    self.thresholds = {**self.DEFAULT_THRESHOLDS, **(thresholds or {})}

analyze_cell_performance(perfdata, memory_limit, gpu_memory_limit=None)

Analyze cell performance and determine tags

Parameters:

Name	Type	Description	Default
perfdata	DataFrame	DataFrame with performance data	required
memory_limit	float	System memory limit in GB	required
gpu_memory_limit	Optional[float]	GPU memory limit in GB (if available)	None

Returns:

Type	Description
List[TagScore]	List[TagScore]: Ranked performance tags for the cell

Source code in jumper_extension/adapters/analyzer.py

def analyze_cell_performance(
    self,
    perfdata: "pd.DataFrame",
    memory_limit: float,
    gpu_memory_limit: Optional[float] = None,
) -> List[TagScore]:
    """
    Analyze cell performance and determine tags

    Args:
        perfdata: DataFrame with performance data
        memory_limit: System memory limit in GB
        gpu_memory_limit: GPU memory limit in GB (if available)

    Returns:
        List[TagScore]: Ranked performance tags for the cell
    """

    logger.debug(f"{memory_limit = }")
    logger.debug(f"{gpu_memory_limit = }")

    # Compute normalized metrics
    metrics = self._compute_metrics(perfdata, gpu_memory_limit)

    # Calculate resource utilization ratios
    ratios = self._calculate_utilization_ratios(metrics, memory_limit, gpu_memory_limit)

    # Create the ranked tags list
    ranked_tags = self._create_ranked_tags(ratios)

    # Detect "GPU allocated but not used" and prepend if applicable
    gpu_unused_tag = self._detect_gpu_allocated_but_not_used(perfdata, gpu_memory_limit)
    if gpu_unused_tag is not None:
        # Prepend the GPU allocated but not used as this is the most important tag
        ranked_tags = [gpu_unused_tag] +  ranked_tags

    logger.debug(f"{ranked_tags = }")

    return ranked_tags if ranked_tags else [TagScore(PerformanceTag.NORMAL, 0.0)]

PerformanceTag

Bases: Enum

Performance tags for classifying cells

Source code in jumper_extension/adapters/analyzer.py

class PerformanceTag(Enum):
    """Performance tags for classifying cells"""
    NORMAL = "normal"
    CPU_BOUND = "cpu-bound"
    MEMORY_BOUND = "memory-bound"
    GPU_UTIL_BOUND = "gpu-util-bound"
    GPU_MEMORY_BOUND = "gpu-memory-bound"
    GPU_ALLOCATED_BUT_NOT_USED = "gpu-allocated-but-not-used"

    def __str__(self):
        return self.value

TagScore dataclass

Tag with its score for ranking

Source code in jumper_extension/adapters/analyzer.py

@dataclass
class TagScore:
    """Tag with its score for ranking"""
    tag: PerformanceTag
    score: float

Reporting and visualization

PerformanceReporter

Adapter class for performance reporting

Source code in jumper_extension/adapters/reporter.py

class PerformanceReporter:
    """Adapter class for performance reporting"""
    def __init__(
        self,
        printer: ReportPrinter,
        displayer: ReportDisplayerProtocol
    ):
        self.printer = printer
        self.displayer = displayer

    def attach(
        self,
        monitor: MonitorProtocol,
    ):
        """Attach started PerformanceMonitor"""
        # Attach to printer
        self.printer.monitor = monitor
        self.printer.min_duration = monitor.interval
        # Attach to displayer
        self.displayer.monitor = monitor
        self.displayer.min_duration = monitor.interval

    def print(self, cell_range=None, level="process"):
        """Print performance report"""
        self.printer.print(cell_range, level)

    def display(self, cell_range=None, level="process"):
        """Display performance report"""
        self.displayer.display(cell_range, level)

attach(monitor)

Attach started PerformanceMonitor

Source code in jumper_extension/adapters/reporter.py

def attach(
    self,
    monitor: MonitorProtocol,
):
    """Attach started PerformanceMonitor"""
    # Attach to printer
    self.printer.monitor = monitor
    self.printer.min_duration = monitor.interval
    # Attach to displayer
    self.displayer.monitor = monitor
    self.displayer.min_duration = monitor.interval

display(cell_range=None, level='process')

Display performance report

Source code in jumper_extension/adapters/reporter.py

def display(self, cell_range=None, level="process"):
    """Display performance report"""
    self.displayer.display(cell_range, level)

print(cell_range=None, level='process')

Print performance report

Source code in jumper_extension/adapters/reporter.py

def print(self, cell_range=None, level="process"):
    """Print performance report"""
    self.printer.print(cell_range, level)

ReportBuilder

Base class for report builders

Source code in jumper_extension/adapters/reporter.py

class ReportBuilder:
    """Base class for report builders"""
    def __init__(
        self,
        monitor: MonitorProtocol,
        cell_history: CellHistory,
        analyzer: PerformanceAnalyzer,
    ):
        self.monitor = monitor
        self.cell_history = cell_history
        self.min_duration = None
        self.analyzer = analyzer

    def _prepare_report_data(self, cell_range, level):
        """Prepare all necessary data for performance reporting.

        Returns:
            dict: Dictionary containing filtered_cells, perfdata, ranked_tags,
                  total_duration, and other data needed for display methods.
                  Returns None if data preparation fails.
        """

        cell_range = self._resolve_cell_range(cell_range)

        if cell_range is None:
            return

        # Filter cell history data first using cell_range
        start_idx, end_idx = cell_range
        filtered_cells = self.cell_history.view(start_idx, end_idx + 1)

        perfdata = self.monitor.data.view(level=level)
        perfdata = filter_perfdata(
            filtered_cells, perfdata, compress_idle=False
        )

        # Check if non-empty, otherwise print results
        if perfdata.empty:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[
                    ExtensionErrorCode.NO_PERFORMANCE_DATA
                ]
            )
            return

        # Analyze cell performance
        memory_limit = self.monitor.memory_limits[level]
        gpu_memory_limit = self.monitor.gpu_memory if self.monitor.num_gpus > 0 else None

        tags_model = self.analyzer.analyze_cell_performance(
            perfdata,
            memory_limit,
            gpu_memory_limit
        )

        # Calculate the total duration of selected cells
        total_duration = filtered_cells["duration"].sum()

        return {
            'cell_range': cell_range,
            'filtered_cells': filtered_cells,
            'perfdata': perfdata,
            'tags_model': tags_model,
            'total_duration': total_duration,
        }

    def _resolve_cell_range(self, cell_range) -> Union[Tuple[int, int], None]:
        """
        Resolve cell range for performance reporting.

        Behavior:
        - If cell_range is None, selects the last cell whose duration is not "short"
         and returns it as a singleton range (idx, idx).
        - Returns None if:
          - no active monitor is attached,
          - the history has no cells,
          - there is no cell with a non-short duration.
        """

        if not self.monitor:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[ExtensionErrorCode.NO_ACTIVE_MONITOR]
            )
            return None

        if cell_range is None:
            valid_cells = self.cell_history.view()

            if len(valid_cells) > 0:
                # Filter for non-short cells
                min_duration = (
                    self.min_duration if self.min_duration is not None else 0
                )
                non_short_cells = valid_cells[
                    valid_cells["duration"] >= min_duration
                    ]

                if len(non_short_cells) > 0:
                    # Get the last non-short cell index
                    last_valid_cell_idx = int(
                        non_short_cells.iloc[-1]["cell_index"]
                    )
                    cell_range = (last_valid_cell_idx, last_valid_cell_idx)
                    return cell_range
                else:
                    logger.warning(
                        EXTENSION_ERROR_MESSAGES[
                            ExtensionErrorCode.NO_PERFORMANCE_DATA
                        ]
                    )
                    return None
            else:
                return None

        return cell_range

    @staticmethod
    def _format_performance_tags(ranked_tags: List[TagScore]):
        """Format ranked performance tags for display"""
        if not ranked_tags:
            return [{"name": "UNKNOWN", "slug": "unknown"}]

        # If the only classification is NORMAL, do not display any tag
        if len(ranked_tags) == 1 and ranked_tags[0].tag == PerformanceTag.NORMAL:
            return []

        # Format all tags with their scores/ratios
        tag_displays = []
        for tag_score in ranked_tags:
            # Create slug for CSS hooks and uppercase name for display
            tag_slug = str(tag_score.tag)
            tag_name = tag_slug.upper()
            tag_displays.append({
                "name": tag_name,
                "slug": tag_slug,
            })
        return tag_displays

ReportDisplayer

Bases: ReportBuilder

Source code in jumper_extension/adapters/reporter.py

class ReportDisplayer(ReportBuilder):
    def __init__(
        self,
        monitor: MonitorProtocol,
        cell_history: CellHistory,
        analyzer: PerformanceAnalyzer,
        templates_dir=None
    ):
        super().__init__(monitor, cell_history, analyzer)
        self.templates_dir = Path(templates_dir) if templates_dir else Path(__file__).parent.parent / "templates"

    def display(self, cell_range=None, level="process"):
        """Print performance report"""

        data = self._prepare_report_data(cell_range, level)
        if data is None:
            return

        filtered_cells = data['filtered_cells']
        perfdata = data['perfdata']
        tags_model = data['tags_model']
        total_duration = data['total_duration']

        tags_display = self._format_performance_tags(tags_model)

        # Build report
        metrics_spec = [
            (f"CPU Util (Across {self.monitor.num_cpus} CPUs)", "cpu_util_avg", "-"),
            ("Memory (GB)", "memory", f"{self.monitor.memory_limits[level]:.2f}" if hasattr(self.monitor, "memory_limits") else "-"),
            (f"GPU Util (Across {getattr(self.monitor, 'num_gpus', 0)} GPUs)", "gpu_util_avg", "-"),
            ("GPU Memory (GB)", "gpu_mem_avg", f"{getattr(self.monitor, 'gpu_memory', 0.0):.2f}"),
        ]
        metrics_rows = []
        for name, col, total in metrics_spec:
            if col in perfdata.columns:
                metrics_rows.append({
                    "name": name,
                    "avg": float(perfdata[col].mean()),
                    "min": float(perfdata[col].min()),
                    "max": float(perfdata[col].max()),
                    "total": total,
                })

        # Render Jinja2 HTML from external files
        env = Environment(
            loader=FileSystemLoader(str(self.templates_dir)),
            autoescape=select_autoescape(["html", "xml"])
        )
        report_html_path = Path("report") / "report.html"
        template = env.get_template(report_html_path.as_posix())
        # Read external stylesheet and inline it for notebook rendering
        try:
            styles_path = self.templates_dir / "report" / "styles.css"
            inline_styles = styles_path.read_text(encoding="utf-8") if styles_path.exists() else ""
        except Exception:
            inline_styles = ""

        html = template.render(
            duration=total_duration,
            n_cells=len(filtered_cells) if filtered_cells is not None else 1,
            metrics=metrics_rows,
            tags=tags_display,
            inline_styles=inline_styles,
        )
        display(HTML(html))

display(cell_range=None, level='process')

Print performance report

Source code in jumper_extension/adapters/reporter.py

def display(self, cell_range=None, level="process"):
    """Print performance report"""

    data = self._prepare_report_data(cell_range, level)
    if data is None:
        return

    filtered_cells = data['filtered_cells']
    perfdata = data['perfdata']
    tags_model = data['tags_model']
    total_duration = data['total_duration']

    tags_display = self._format_performance_tags(tags_model)

    # Build report
    metrics_spec = [
        (f"CPU Util (Across {self.monitor.num_cpus} CPUs)", "cpu_util_avg", "-"),
        ("Memory (GB)", "memory", f"{self.monitor.memory_limits[level]:.2f}" if hasattr(self.monitor, "memory_limits") else "-"),
        (f"GPU Util (Across {getattr(self.monitor, 'num_gpus', 0)} GPUs)", "gpu_util_avg", "-"),
        ("GPU Memory (GB)", "gpu_mem_avg", f"{getattr(self.monitor, 'gpu_memory', 0.0):.2f}"),
    ]
    metrics_rows = []
    for name, col, total in metrics_spec:
        if col in perfdata.columns:
            metrics_rows.append({
                "name": name,
                "avg": float(perfdata[col].mean()),
                "min": float(perfdata[col].min()),
                "max": float(perfdata[col].max()),
                "total": total,
            })

    # Render Jinja2 HTML from external files
    env = Environment(
        loader=FileSystemLoader(str(self.templates_dir)),
        autoescape=select_autoescape(["html", "xml"])
    )
    report_html_path = Path("report") / "report.html"
    template = env.get_template(report_html_path.as_posix())
    # Read external stylesheet and inline it for notebook rendering
    try:
        styles_path = self.templates_dir / "report" / "styles.css"
        inline_styles = styles_path.read_text(encoding="utf-8") if styles_path.exists() else ""
    except Exception:
        inline_styles = ""

    html = template.render(
        duration=total_duration,
        n_cells=len(filtered_cells) if filtered_cells is not None else 1,
        metrics=metrics_rows,
        tags=tags_display,
        inline_styles=inline_styles,
    )
    display(HTML(html))

ReportDisplayerProtocol

Bases: Protocol

Structural protocol for HTML/text report displayers.

Source code in jumper_extension/adapters/reporter.py

@runtime_checkable
class ReportDisplayerProtocol(Protocol):
    """Structural protocol for HTML/text report displayers."""
    def display(self, cell_range=None, level: str = "process") -> None: ...

ReportPrinter

Bases: ReportBuilder

Source code in jumper_extension/adapters/reporter.py

class ReportPrinter(ReportBuilder):
    def __init__(
        self,
        monitor: MonitorProtocol,
        cell_history: CellHistory,
        analyzer: PerformanceAnalyzer,
    ):
        super().__init__(monitor, cell_history, analyzer)

    def print(self, cell_range=None, level="process"):
        """Print performance report"""
        data = self._prepare_report_data(cell_range, level)
        if data is None:
            return

        filtered_cells = data['filtered_cells']
        perfdata = data['perfdata']
        tags_model = data['tags_model']
        total_duration = data['total_duration']

        print("-" * 40)
        print("JUmPER Performance Report")
        print("-" * 40)
        n_cells = len(filtered_cells)
        print(
            f"Duration: {total_duration:.2f}s "
            f"({n_cells} cell{'s' if n_cells != 1 else ''})"
        )
        print("-" * 40)

        # Output performance tags
        tags_display = self._format_performance_tags(tags_model)
        if tags_display:
            print("Signature(s):")
            tags_line = " | ".join(tag["name"] for tag in tags_display)
            print(tags_line)

            print("-" * 40)

        # Report table
        metrics = [
            (
                f"CPU Util (Across {self.monitor.num_cpus} CPUs)",
                "cpu_util_avg",
                "-",
            ),
            (
                "Memory (GB)",
                "memory",
                f"{self.monitor.memory_limits[level]:.2f}",
            ),
            (
                f"GPU Util (Across {self.monitor.num_gpus} GPUs)",
                "gpu_util_avg",
                "-",
            ),
            (
                "GPU Memory (GB)",
                "gpu_mem_avg",
                f"{self.monitor.gpu_memory:.2f}",
            ),
        ]

        print(f"{'Metric':<25} {'AVG':<8} {'MIN':<8} {'MAX':<8} {'TOTAL':<8}")
        print("-" * 65)
        for name, col, total in metrics:
            if col in perfdata.columns:
                print(
                    f"{name:<25} {perfdata[col].mean():<8.2f} "
                    f"{perfdata[col].min():<8.2f} {perfdata[col].max():<8.2f} "
                    f"{total:<8}"
                )

print(cell_range=None, level='process')

Print performance report

Source code in jumper_extension/adapters/reporter.py

def print(self, cell_range=None, level="process"):
    """Print performance report"""
    data = self._prepare_report_data(cell_range, level)
    if data is None:
        return

    filtered_cells = data['filtered_cells']
    perfdata = data['perfdata']
    tags_model = data['tags_model']
    total_duration = data['total_duration']

    print("-" * 40)
    print("JUmPER Performance Report")
    print("-" * 40)
    n_cells = len(filtered_cells)
    print(
        f"Duration: {total_duration:.2f}s "
        f"({n_cells} cell{'s' if n_cells != 1 else ''})"
    )
    print("-" * 40)

    # Output performance tags
    tags_display = self._format_performance_tags(tags_model)
    if tags_display:
        print("Signature(s):")
        tags_line = " | ".join(tag["name"] for tag in tags_display)
        print(tags_line)

        print("-" * 40)

    # Report table
    metrics = [
        (
            f"CPU Util (Across {self.monitor.num_cpus} CPUs)",
            "cpu_util_avg",
            "-",
        ),
        (
            "Memory (GB)",
            "memory",
            f"{self.monitor.memory_limits[level]:.2f}",
        ),
        (
            f"GPU Util (Across {self.monitor.num_gpus} GPUs)",
            "gpu_util_avg",
            "-",
        ),
        (
            "GPU Memory (GB)",
            "gpu_mem_avg",
            f"{self.monitor.gpu_memory:.2f}",
        ),
    ]

    print(f"{'Metric':<25} {'AVG':<8} {'MIN':<8} {'MAX':<8} {'TOTAL':<8}")
    print("-" * 65)
    for name, col, total in metrics:
        if col in perfdata.columns:
            print(
                f"{name:<25} {perfdata[col].mean():<8.2f} "
                f"{perfdata[col].min():<8.2f} {perfdata[col].max():<8.2f} "
                f"{total:<8}"
            )

UnavailableReportDisplayer

Source code in jumper_extension/adapters/reporter.py

class UnavailableReportDisplayer:
    def __init__(self, reason="Display not available."):
        self._reason = reason

    def display(self, cell_range=None, level="process"):
        """non-opt display"""
        logger.info(
            EXTENSION_INFO_MESSAGES[
                ExtensionInfoCode.HTML_REPORTS_NOT_AVAILABLE
            ].format(reason=self._reason))

display(cell_range=None, level='process')

non-opt display

Source code in jumper_extension/adapters/reporter.py

def display(self, cell_range=None, level="process"):
    """non-opt display"""
    logger.info(
        EXTENSION_INFO_MESSAGES[
            ExtensionInfoCode.HTML_REPORTS_NOT_AVAILABLE
        ].format(reason=self._reason))

build_performance_reporter(cell_history, templates_dir=None, display_disabled=False, display_disabled_reason='Display not available.', thresholds=None)

Build PerformanceReporter object. Allows building a reporter without displaying.

Source code in jumper_extension/adapters/reporter.py

def build_performance_reporter(
    cell_history: CellHistory,
    templates_dir=None,
    display_disabled: bool = False,
    display_disabled_reason="Display not available.",
    thresholds=None,
):
    """
    Build PerformanceReporter object.
    Allows building a reporter without displaying.
    """
    monitor = UnavailablePerformanceMonitor(
        reason="Monitor has not been started yet."
    )
    analyzer = PerformanceAnalyzer(thresholds=thresholds)
    printer = ReportPrinter(monitor, cell_history, analyzer)
    if display_disabled:
        displayer = UnavailableReportDisplayer(
            reason=display_disabled_reason
        )
    else:
        displayer = ReportDisplayer(
            monitor,
            cell_history,
            analyzer,
            templates_dir
        )
    return PerformanceReporter(printer, displayer)

InteractivePlotWrapper

Interactive plotter with dropdown selection and reusable matplotlib axes.

Source code in jumper_extension/adapters/visualizer.py

class InteractivePlotWrapper:
    """Interactive plotter with dropdown selection and reusable matplotlib
    axes."""

    def __init__(
        self,
        plot_callback,
        metrics: List[str],
        labeled_options,
        perfdata_by_level,
        cell_range=None,
        show_idle=False,
        figsize=None,
    ):
        self.plot_callback, self.perfdata_by_level, self.metrics = (
            plot_callback,
            perfdata_by_level,
            metrics,
        )
        self.labeled_options = labeled_options
        self.cell_range, self.show_idle, self.figsize = (
            cell_range,
            show_idle,
            figsize,
        )
        self.shown_metrics, self.panel_count, self.max_panels = (
            set(),
            0,
            len(metrics) * 4,
        )
        # Store plot panels for updates
        self.plot_panels = []

        self.output_container = widgets.HBox(
            layout=Layout(
                display="flex",
                flex_flow="row wrap",
                align_items="center",
                justify_content="space-between",
                width="100%",
            )
        )
        self.add_panel_button = widgets.Button(
            description="Add Plot Panel",
            layout=Layout(margin="0 auto 20px auto"),
        )
        self.add_panel_button.on_click(self._on_add_panel_clicked)

    def display_ui(self):
        """Display the Add button and all interactive panels."""
        display(widgets.VBox([self.add_panel_button, self.output_container]))
        self._on_add_panel_clicked(None)

    def _on_add_panel_clicked(self, _):
        """Add a new plot panel with dropdown and persistent matplotlib
        axis."""
        if self.panel_count >= self.max_panels:
            self.add_panel_button.disabled = True
            self.output_container.children += (
                widgets.HTML("<b>All panels have been added.</b>"),
            )
            return

        self.output_container.children += (
            widgets.HBox(
                [
                    self._create_dropdown_plot_panel(),
                    self._create_dropdown_plot_panel(),
                ],
            ),
        )
        self.panel_count += 2

        if self.panel_count >= self.max_panels:
            self.add_panel_button.disabled = True

    def _create_dropdown_plot_panel(self):
        """Create metric and level dropdown + matplotlib figure panel with
        persistent Axes."""
        metric_dropdown = widgets.Dropdown(
            options=self.labeled_options,
            value=self._get_next_metric(),
            description="Metric:",
        )
        level_dropdown = widgets.Dropdown(
            options=get_available_levels(),
            value="process",
            description="Level:",
        )
        fig, ax = plt.subplots(figsize=self.figsize, constrained_layout=True)
        if not is_ipympl_backend():
            plt.close(fig)
        output = widgets.Output()

        def update_plot():
            metric = metric_dropdown.value
            level = level_dropdown.value
            df = self.perfdata_by_level.get(level)
            if not is_ipympl_backend():
                output.clear_output(wait=True)
            with output:
                ax.clear()
                if df is not None and not df.empty:
                    self.plot_callback(
                        df, metric, self.cell_range, self.show_idle, ax, level
                    )
                fig.canvas.draw_idle()
                if not is_ipympl_backend():
                    display(fig)

        def on_dropdown_change(change):
            if change["type"] == "change" and change["name"] == "value":
                update_plot()

        metric_dropdown.observe(on_dropdown_change)
        level_dropdown.observe(on_dropdown_change)

        # Store panel data for updates
        panel_data = {
            "metric_dropdown": metric_dropdown,
            "level_dropdown": level_dropdown,
            "figure": fig,
            "axes": ax,
            "output": output,
            "update_plot": update_plot,
        }
        self.plot_panels.append(panel_data)

        # Initial plot
        update_plot()
        if is_ipympl_backend():
            with output:
                plt.show()

        return widgets.VBox(
            [widgets.HBox([metric_dropdown, level_dropdown]), output]
        )

    def _get_next_metric(self):
        for metric in self.metrics:
            if metric not in self.shown_metrics:
                self.shown_metrics.add(metric)
                return metric
        return None

    def update_data(self, perfdata_by_level, cell_range, show_idle):
        self.perfdata_by_level = perfdata_by_level
        self.cell_range = cell_range
        self.show_idle = show_idle
        for panel in self.plot_panels:
            panel["output"].clear_output(wait=True)
            panel["update_plot"]()

display_ui()

Display the Add button and all interactive panels.

Source code in jumper_extension/adapters/visualizer.py

def display_ui(self):
    """Display the Add button and all interactive panels."""
    display(widgets.VBox([self.add_panel_button, self.output_container]))
    self._on_add_panel_clicked(None)

InteractivePlotlyWrapper

Interactive plotter that renders controls and figures via pure HTML/JS.

All metric × level × show-idle combinations are pre-computed server-side and embedded as Plotly JSON in a single self-contained HTML block. The browser handles dropdown changes and the show-idle toggle without any Python round-trips.

Source code in jumper_extension/adapters/visualizer.py

class InteractivePlotlyWrapper:
    """Interactive plotter that renders controls and figures via pure HTML/JS.

    All metric × level × show-idle combinations are pre-computed server-side
    and embedded as Plotly JSON in a single self-contained HTML block.
    The browser handles dropdown changes and the show-idle toggle without any
    Python round-trips.
    """

    _TEMPLATES_DIR = (
        Path(__file__).parent.parent / "templates" / "plotly_visualizer"
    )

    def __init__(
        self,
        plot_callback,
        metrics: List[str],
        labeled_options,
        perfdata_by_level,
        cell_range=None,
        show_idle=False,
        _precomputed_figures=None,
    ):
        self.plot_callback    = plot_callback
        self.perfdata_by_level = perfdata_by_level
        self.metrics          = metrics
        self.labeled_options  = labeled_options
        self.cell_range       = cell_range
        self.show_idle        = show_idle
        self.max_panels       = len(metrics) * 4
        # Pre-computed figures supplied by PlotlyPerformanceVisualizer; if
        # None we fall back to computing lazily from plot_callback.
        self._precomputed_figures = _precomputed_figures
        self._display_handle  = None
        self._container_id    = f"jump-vis-{uuid.uuid4().hex[:8]}"

    # ------------------------------------------------------------------ #
    # Public API (kept stable with the old ipywidgets implementation)     #
    # ------------------------------------------------------------------ #

    def display_ui(self):
        html = self._render_html()
        self._display_handle = display(HTML(html), display_id=True)

    def update_data(self, perfdata_by_level, cell_range, show_idle):
        self.perfdata_by_level    = perfdata_by_level
        self.cell_range           = cell_range
        self.show_idle            = show_idle
        self._precomputed_figures = None  # stale; recompute on render
        if self._display_handle is not None:
            self._display_handle.update(HTML(self._render_html()))
        else:
            self.display_ui()

    # ------------------------------------------------------------------ #
    # Internals                                                           #
    # ------------------------------------------------------------------ #

    def _compute_figures_from_callback(self):
        """Fallback: compute figures for the current show_idle state only.

        Both show_idle variants are attempted but the 'false' variant relies
        on ``_compressed_cell_boundaries`` being set correctly on the
        visualizer instance.  When called from ``update_data`` this may not
        hold, so only the current variant is guaranteed to be accurate.
        """
        result = {}
        levels = list(self.perfdata_by_level.keys())
        for _label, metric in self.labeled_options:
            result[metric] = {}
            for level in levels:
                df = self.perfdata_by_level.get(level)
                result[metric][level] = {}
                for key in ("true", "false"):
                    if df is None or df.empty:
                        result[metric][level][key] = None
                        continue
                    try:
                        fig = self.plot_callback(
                            df,
                            metric,
                            self.cell_range,
                            key == "true",
                            level,
                        )
                        result[metric][level][key] = (
                            json.loads(fig.to_json()) if fig is not None else None
                        )
                    except Exception:
                        result[metric][level][key] = None
        return result

    # CSS and JS components are loaded in this order.
    _CSS_COMPONENTS = [
        "toolbar",
        "show_idle_checkbox",
        "cell_range_slider",
        "add_panel_button",
        "panel",
    ]
    _JS_COMPONENTS = [
        "show_idle_checkbox",
        "cell_range_slider",
        "add_panel_button",
        "panel",
    ]

    def _read_component_file(self, component: str, ext: str) -> str:
        path = self._TEMPLATES_DIR / component / f"{component}.{ext}"
        try:
            return path.read_text(encoding="utf-8") if path.exists() else ""
        except Exception:
            return ""

    def _render_html(self):
        pre = (
            self._precomputed_figures
            if self._precomputed_figures is not None
            else self._compute_figures_from_callback()
        )

        # Unpack rich pre-computed structure or fall back to plain figures dict
        if isinstance(pre, dict) and "figures" in pre:
            figures            = pre["figures"]
            ylims              = pre.get("ylims", {})
            boundaries_false   = pre.get("boundaries_false", [])
            boundaries_true    = pre.get("boundaries_true", [])
            min_cell_index     = pre.get("min_cell_index", 0)
            max_cell_index     = pre.get("max_cell_index", 0)
            initial_cell_range = pre.get(
                "initial_cell_range", [min_cell_index, max_cell_index]
            )
        else:
            figures            = pre
            ylims              = {}
            boundaries_false   = []
            boundaries_true    = []
            min_cell_index     = 0
            max_cell_index     = 0
            initial_cell_range = [0, 0]

        levels  = get_available_levels()
        init_lo = initial_cell_range[0] if initial_cell_range else min_cell_index
        init_hi = (
            initial_cell_range[1] if len(initial_cell_range) > 1 else max_cell_index
        )

        # ── 1. Collect component CSS ──────────────────────────────────────
        css_parts = [
            self._read_component_file(c, "css") for c in self._CSS_COMPONENTS
        ]

        # ── 2. Render HTML via Jinja2 (visualizer.html includes sub-templates)
        env = Environment(
            loader=FileSystemLoader(str(self._TEMPLATES_DIR)),
            autoescape=False,
        )
        env.filters["tojson"] = json.dumps
        html_ctx = dict(
            container_id=self._container_id,
            logo_src=logo_image,
            initial_show_idle=self.show_idle,
            min_cell_index=min_cell_index,
            max_cell_index=max_cell_index,
            init_lo=init_lo,
            init_hi=init_hi,
        )
        body_html = env.get_template("visualizer.html").render(**html_ctx)

        # ── 3. Collect component JS then main orchestration script ────────
        js_parts = [
            self._read_component_file(c, "js") for c in self._JS_COMPONENTS
        ]
        main_js_path = self._TEMPLATES_DIR / "main.js"
        try:
            js_parts.append(
                main_js_path.read_text(encoding="utf-8")
                if main_js_path.exists()
                else ""
            )
        except Exception:
            pass

        # ── 4. Embedded Python data (must precede component + main JS) ────
        data_js = "\n".join([
            f"var CID      = {json.dumps(self._container_id)};",
            f"var FIGS     = {json.dumps(figures)};",
            f"var YLIMS    = {json.dumps(ylims)};",
            f"var BND_F    = {json.dumps(boundaries_false)};",
            f"var BND_T    = {json.dumps(boundaries_true)};",
            f"var OPTS     = {json.dumps(self.labeled_options)};",
            f"var LEVS     = {json.dumps(levels)};",
            f"var MAX      = {self.max_panels};",
            f"var MIN_CELL = {min_cell_index};",
            f"var MAX_CELL = {max_cell_index};",
            f"var INIT_RNG = {json.dumps(initial_cell_range)};",
        ])

        # ── 5. Plotly CDN loader (injected once per output block) ─────────
        plotly_loader = (
            "<script>\n"
            "(function(){\n"
            "  if(typeof window.Plotly!=='undefined')return;\n"
            "  var s=document.createElement('script');\n"
            "  s.src='https://cdn.plot.ly/plotly-2.35.2.min.js';\n"
            "  s.charset='utf-8';\n"
            "  document.head.appendChild(s);\n"
            "})();\n"
            "</script>"
        )

        # ── 6. Assemble final HTML ────────────────────────────────────────
        return "\n".join([
            plotly_loader,
            "<style>\n" + "\n".join(css_parts) + "\n</style>",
            body_html,
            "<script>\n" + data_js + "\n\n" + "\n\n".join(js_parts) + "\n</script>",
        ])

MatplotlibPerformanceVisualizer

Bases: PerformanceVisualizer

Matplotlib backend visualizer.

Source code in jumper_extension/adapters/visualizer.py

class MatplotlibPerformanceVisualizer(PerformanceVisualizer):
    """Matplotlib backend visualizer."""

    def _plot_metric(
        self,
        df,
        metric,
        cell_range=None,
        show_idle=False,
        ax: plt.Axes = None,
        level="process",
    ):
        """Plot a single metric using its configuration."""
        config = next(
            (
                subset[metric]
                for subset in self.subsets.values()
                if metric in subset
            ),
            None,
        )
        if not config or not isinstance(config, dict):
            return

        plot_type = config.get("type")
        if plot_type == "single_series":
            column = config.get("column")
            title = config.get("title", "")
            ylim = config.get("ylim")
            if metric == "memory" and ylim is None:
                ylim = (0, self.monitor.memory_limits[level])
            if not column or column not in df.columns:
                return
        elif plot_type == "multi_series":
            prefix = config.get("prefix", "")
            title = config.get("title", "")
            ylim = config.get("ylim")
            series_cols = [
                col
                for col in df.columns
                if prefix
                and col.startswith(prefix)
                and not col.endswith("avg")
            ]
            avg_column = f"{prefix}avg" if prefix else None
            if (
                avg_column is None or avg_column not in df.columns
            ) and not series_cols:
                return
        elif plot_type == "summary_series":
            columns = config.get("columns", [])
            title = config.get("title", "")
            ylim = config.get("ylim")
            if level == "system":
                title = re.sub(
                    r"\d+", str(self.monitor.num_system_cpus), title
                )
            available_cols = [col for col in columns if col in df.columns]
            if not available_cols:
                return
        else:
            return

        if ax is None:
            _, ax = plt.subplots(figsize=self.figsize)

        if plot_type == "single_series":
            series = df[column]
            if metric in (
                "io_read",
                "io_write",
                "io_read_count",
                "io_write_count",
            ):
                diffs = df[column].astype(float).diff().clip(lower=0)
                if metric in ("io_read", "io_write"):
                    diffs = diffs / (1024**2)
                series = diffs.fillna(0.0)
                if self._io_window > 1:
                    series = series.rolling(
                        window=self._io_window, min_periods=1
                    ).mean()
            ax.plot(df["time"], series, color="blue", linewidth=2)
        elif plot_type == "summary_series":
            line_styles, alpha_vals = ["dotted", "-", "--"], [0.35, 1.0, 0.35]
            for i, (col, label) in enumerate(
                zip(columns, ["Min", "Average", "Max"])
            ):
                if col in df.columns:
                    ax.plot(
                        df["time"],
                        df[col],
                        color="blue",
                        linestyle=line_styles[i % len(line_styles)],
                        linewidth=2,
                        alpha=alpha_vals[i % len(alpha_vals)],
                        label=label,
                    )
            ax.legend()
        elif plot_type == "multi_series":
            for col in series_cols:
                ax.plot(df["time"], df[col], "-", alpha=0.5, label=col)
            if avg_column in df.columns:
                ax.plot(
                    df["time"], df[avg_column], "b-", linewidth=2, label="Mean"
                )
            ax.legend()

        ax.set_title(title + (" (No Idle)" if not show_idle else ""))
        ax.set_xlabel("Time (seconds)")
        ax.grid(True)
        if ylim:
            ax.set_ylim(ylim)
        self._draw_cell_boundaries(ax, cell_range, show_idle)

    def _draw_cell_boundaries(self, ax, cell_range=None, show_idle=False):
        """Draw cell boundaries as colored rectangles with cell indices."""
        colors = jumper_colors
        y_min, y_max = ax.get_ylim()
        x_max, height = ax.get_xlim()[1], y_max - y_min
        min_duration = self.min_duration or 0

        def draw_cell_rect(start_time, duration, cell_num, alpha):
            if (
                duration < min_duration
                or start_time > x_max
                or start_time + duration < 0
            ):
                return
            color = colors[cell_num % len(colors)]
            ax.add_patch(
                plt.Rectangle(
                    (start_time, y_min),
                    duration,
                    height,
                    facecolor=color,
                    alpha=alpha,
                    edgecolor="black",
                    linestyle="--",
                    linewidth=1,
                    zorder=0,
                )
            )
            ax.text(
                start_time + duration / 2,
                y_max - height * 0.1,
                f"#{cell_num}",
                ha="center",
                va="center",
                fontsize=10,
                fontweight="bold",
                zorder=1,
                bbox=dict(
                    boxstyle="round,pad=0.3", facecolor="white", alpha=0.8
                ),
            )

        if not show_idle and hasattr(self, "_compressed_cell_boundaries"):
            for cell in self._compressed_cell_boundaries:
                draw_cell_rect(
                    cell["start_time"],
                    cell["duration"],
                    int(cell["cell_index"]),
                    0.4,
                )
        else:
            filtered_cells = self.cell_history.view()
            if cell_range:
                try:
                    mask = (filtered_cells["cell_index"] >= cell_range[0]) & (
                        filtered_cells["cell_index"] <= cell_range[1]
                    )
                    cells = filtered_cells[mask]
                except Exception:
                    cells = filtered_cells
            else:
                cells = filtered_cells
            for _, cell in cells.iterrows():
                start_time = cell["start_time"] - self.monitor.start_time
                draw_cell_rect(
                    start_time, cell["duration"], int(cell["cell_index"]), 0.5
                )

    def _create_interactive_wrapper(
        self,
        metrics,
        labeled_options,
        processed_perfdata,
        current_cell_range,
        current_show_idle,
    ):
        return InteractivePlotWrapper(
            self._plot_metric,
            metrics,
            labeled_options,
            processed_perfdata,
            current_cell_range,
            current_show_idle,
            self.figsize,
        )

    def _render_direct_plot(
        self,
        processed_data,
        metrics,
        cell_range,
        show_idle,
        level,
        save_jpeg=None,
        pickle_file=None,
        metric_subsets=None,
    ):
        n_metrics = len(metrics)
        fig, axes = plt.subplots(
            n_metrics,
            1,
            figsize=(10, 3 * n_metrics),
            constrained_layout=True,
        )
        if n_metrics == 1:
            axes = [axes]

        for i, metric in enumerate(metrics):
            self._plot_metric(
                processed_data, metric, cell_range, show_idle, axes[i], level
            )

        if save_jpeg:
            if not save_jpeg.endswith(".jpg") and not save_jpeg.endswith(
                ".jpeg"
            ):
                save_jpeg += ".jpg"
            fig.savefig(
                save_jpeg, format="jpeg", dpi=300, bbox_inches="tight"
            )
            print(f"Plot saved as JPEG: {save_jpeg}")

        if pickle_file:
            if not pickle_file.endswith(".pkl"):
                pickle_file += ".pkl"
            plot_data = {
                "figure": fig,
                "axes": axes,
                "metrics": metrics,
                "processed_data": processed_data,
                "cell_range": cell_range,
                "level": level,
                "show_idle": show_idle,
                "metric_subsets": metric_subsets,
            }
            with open(pickle_file, "wb") as f:
                pickle.dump(plot_data, f)

            print(f"Plot objects serialized to: {pickle_file}")
            print("\n# Python code to reload and display the plot:")
            print("import pickle")
            print("import matplotlib.pyplot as plt")
            print("")
            print(f"# Load the pickled plot data")
            print(f"with open('{pickle_file}', 'rb') as f:")
            print("    plot_data = pickle.load(f)")
            print("")
            print("# Extract the figure and display")
            print("fig = plot_data['figure']")
            print("plt.show()")
            print("")
            print("# Access other data:")
            print("# axes = plot_data['axes']")
            print("# metrics = plot_data['metrics']")
            print("# processed_data = plot_data['processed_data']")
            print("# cell_range = plot_data['cell_range']")
            print("# level = plot_data['level']")

        plt.show()

PerformanceVisualizer

Visualizes performance metrics collected by PerformanceMonitor.

Supports multiple levels: 'user', 'process' (default), 'system', and 'slurm' (if available)

Source code in jumper_extension/adapters/visualizer.py

class PerformanceVisualizer:
    """Visualizes performance metrics collected by PerformanceMonitor.

    Supports multiple levels: 'user', 'process' (default), 'system', and
    'slurm' (if available)
    """

    def __init__(self, cell_history: CellHistory):
        self.monitor = UnavailablePerformanceMonitor(
            reason="Monitor has not been started yet."
        )
        self.cell_history = cell_history
        self.figsize = (5, 3)
        self.min_duration = None
        self._io_window = None
        self.subsets = {}

    def attach(
        self,
        monitor: MonitorProtocol,
    ):
        """Attach started PerformanceMonitor."""
        self.monitor = monitor
        self.min_duration = self.monitor.interval
        # Smooth IO with ~1s rolling window based on sampling interval
        try:
            self._io_window = max(
                1, int(round(1.0 / (self.monitor.interval or 1.0)))
            )
        except Exception:
            self._io_window = 1
        self._build_subsets()

    def _build_subsets(self):
        """Build a dictionary of metric subsets based on the provided
        configuration"""
        # Compressed metrics configuration (dict-based entries for clarity)
        self.subsets = {
            "cpu_all": {
                "cpu": {
                    "type": "multi_series",
                    "prefix": "cpu_util_",
                    "title": "CPU Utilization (%) - Across Cores",
                    "ylim": (0, 100),
                    "label": "CPU Utilization (All Cores)",
                }
            },
            "gpu_all": {
                "gpu_util": {
                    "type": "multi_series",
                    "prefix": "gpu_util_",
                    "title": "GPU Utilization (%) - Across GPUs",
                    "ylim": (0, 100),
                    "label": "GPU Utilization (All GPUs)",
                },
                "gpu_band": {
                    "type": "multi_series",
                    "prefix": "gpu_band_",
                    "title": "GPU Bandwidth Usage (%) - Across GPUs",
                    "ylim": (0, 100),
                    "label": "GPU Bandwidth (All GPUs)",
                },
                "gpu_mem": {
                    "type": "multi_series",
                    "prefix": "gpu_mem_",
                    "title": "GPU Memory Usage (GB) - Across GPUs",
                    "ylim": (0, self.monitor.gpu_memory),
                    "label": "GPU Memory (All GPUs)",
                },
            },
            "cpu": {
                "cpu_summary": {
                    "type": "summary_series",
                    "columns": [
                        "cpu_util_min",
                        "cpu_util_avg",
                        "cpu_util_max",
                    ],
                    "title": (
                        "CPU Utilization (%) - "
                        f"{self.monitor.num_cpus} CPUs"
                    ),
                    "ylim": (0, 100),
                    "label": "CPU Utilization Summary",
                }
            },
            "gpu": {
                "gpu_util_summary": {
                    "type": "summary_series",
                    "columns": [
                        "gpu_util_min",
                        "gpu_util_avg",
                        "gpu_util_max",
                    ],
                    "title": (
                        "GPU Utilization (%) - "
                        f"{self.monitor.num_gpus} GPUs"
                    ),
                    "ylim": (0, 100),
                    "label": "GPU Utilization Summary",
                },
                "gpu_band_summary": {
                    "type": "summary_series",
                    "columns": [
                        "gpu_band_min",
                        "gpu_band_avg",
                        "gpu_band_max",
                    ],
                    "title": (
                        "GPU Bandwidth Usage (%) - "
                        f"{self.monitor.num_gpus} GPUs"
                    ),
                    "ylim": (0, 100),
                    "label": "GPU Bandwidth Summary",
                },
                "gpu_mem_summary": {
                    "type": "summary_series",
                    "columns": ["gpu_mem_min", "gpu_mem_avg", "gpu_mem_max"],
                    "title": (
                        "GPU Memory Usage (GB) - "
                        f"{self.monitor.num_gpus} GPUs"
                    ),
                    "ylim": (0, self.monitor.gpu_memory),
                    "label": "GPU Memory Summary",
                },
            },
            "mem": {
                "memory": {
                    "type": "single_series",
                    "column": "memory",
                    "title": "Memory Usage (GB)",
                    "ylim": None,  # Will be set dynamically based on level
                    "label": "Memory Usage",
                }
            },
            "io": {
                "io_read": {
                    "type": "single_series",
                    "column": "io_read",
                    "title": "I/O Read (MB/s)",
                    "ylim": None,
                    "label": "IO Read MB/s",
                },
                "io_write": {
                    "type": "single_series",
                    "column": "io_write",
                    "title": "I/O Write (MB/s)",
                    "ylim": None,
                    "label": "IO Write MB/s",
                },
                "io_read_count": {
                    "type": "single_series",
                    "column": "io_read_count",
                    "title": "I/O Read Operations (ops/s)",
                    "ylim": None,
                    "label": "IO Read Ops",
                },
                "io_write_count": {
                    "type": "single_series",
                    "column": "io_write_count",
                    "title": "I/O Write Operations (ops/s)",
                    "ylim": None,
                    "label": "IO Write Ops",
                },
            },
        }

    def _resolve_metric_subsets(
        self,
        metrics: Optional[List[str]]
    ) -> Tuple[str, ...]:
        """Map user-specified metrics or subsets to visualizer subset keys."""
        if not metrics:
            return ("cpu", "mem", "io")

        resolved: List[str] = []
        metric_list = (
            [metrics]
            if isinstance(metrics, str)
            else list(metrics)
        )
        for metric in metric_list:
            if not metric:
                continue
            metric_key = str(metric).strip()
            if metric_key in self.subsets:
                resolved.append(metric_key)
                continue
            found_subset = next(
                (
                    subset
                    for subset, cfg in self.subsets.items()
                    if metric_key in cfg
                ),
                None,
            )
            if found_subset:
                resolved.append(found_subset)
            else:
                logger.warning(
                    EXTENSION_ERROR_MESSAGES[
                        ExtensionErrorCode.INVALID_METRIC_SUBSET
                    ].format(
                        subset=metric_key,
                        supported_subsets=", ".join(self.subsets.keys()),
                    )
                )

        # Remove duplicates while preserving order; fall back to defaults
        deduped = tuple(dict.fromkeys(resolved))
        return deduped or ("cpu", "mem", "io")

    def _compress_time_axis(self, perfdata, cell_range):
        """Compress time axis by removing idle periods between cells"""
        if perfdata.empty:
            return perfdata, []

        start_idx, end_idx = cell_range
        cell_data = self.cell_history.view(start_idx, end_idx + 1)
        compressed_perfdata, cell_boundaries, current_time = (
            perfdata.copy(),
            [],
            0,
        )

        for idx, cell in cell_data.iterrows():
            cell_mask = (perfdata["time"] >= cell["start_time"]) & (
                perfdata["time"] <= cell["end_time"]
            )
            cell_perfdata = perfdata[cell_mask]

            if not cell_perfdata.empty:
                original_start, cell_duration = (
                    cell["start_time"],
                    cell["end_time"] - cell["start_time"],
                )
                compressed_perfdata.loc[cell_mask, "time"] = current_time + (
                    cell_perfdata["time"].values - original_start
                )
                cell_boundaries.append(
                    {
                        "cell_index": cell["cell_index"],
                        "start_time": current_time,
                        "end_time": current_time + cell_duration,
                        "duration": cell_duration,
                    }
                )
                current_time += cell_duration

        return compressed_perfdata, cell_boundaries

    def _collect_metric_options(self, metric_subsets):
        metrics = []
        labeled_options = []
        for subset in metric_subsets:
            if subset in self.subsets:
                for metric_key, cfg in self.subsets[subset].items():
                    metrics.append(metric_key)
                    label = (
                        cfg.get("label")
                        if isinstance(cfg, dict)
                        else metric_key
                    )
                    labeled_options.append((label or metric_key, metric_key))
            else:
                logger.warning(
                    EXTENSION_ERROR_MESSAGES[
                        ExtensionErrorCode.INVALID_METRIC_SUBSET
                    ].format(
                        subset=subset,
                        supported_subsets=", ".join(self.subsets.keys()),
                    )
                )
        return metrics, labeled_options

    def _prepare_processed_data_for_level(self, cell_range, show_idle, level):
        start_idx, end_idx = cell_range
        filtered_cells = self.cell_history.view(start_idx, end_idx + 1)
        perfdata = filter_perfdata(
            filtered_cells,
            self.monitor.data.view(level=level),
            not show_idle,
        )

        if perfdata.empty:
            return None

        if not show_idle:
            processed_data, self._compressed_cell_boundaries = (
                self._compress_time_axis(perfdata, cell_range)
            )
        else:
            processed_data = perfdata.copy()
            processed_data["time"] -= self.monitor.start_time
        return processed_data

    def _prepare_processed_data_for_interactive(
        self,
        current_cell_range,
        current_show_idle,
    ):
        start_idx, end_idx = current_cell_range
        cells_all = self.cell_history.view()
        try:
            mask = (cells_all["cell_index"] >= start_idx) & (
                cells_all["cell_index"] <= end_idx
            )
            filtered_cells = cells_all[mask]
        except Exception:
            filtered_cells = cells_all

        perfdata_by_level = {}
        for available_level in get_available_levels():
            perfdata_by_level[available_level] = filter_perfdata(
                filtered_cells,
                self.monitor.data.view(level=available_level),
                not current_show_idle,
            )

        if all(df.empty for df in perfdata_by_level.values()):
            return None

        processed_perfdata = {}
        for level_key, perfdata in perfdata_by_level.items():
            if not perfdata.empty:
                if not current_show_idle:
                    processed_data, self._compressed_cell_boundaries = (
                        self._compress_time_axis(perfdata, current_cell_range)
                    )
                    processed_perfdata[level_key] = processed_data
                else:
                    processed_data = perfdata.copy()
                    processed_data["time"] -= self.monitor.start_time
                    processed_perfdata[level_key] = processed_data
            else:
                processed_perfdata[level_key] = perfdata

        return processed_perfdata

    def _create_interactive_wrapper(
        self,
        metrics,
        labeled_options,
        processed_perfdata,
        current_cell_range,
        current_show_idle,
    ):
        raise NotImplementedError

    def _render_direct_plot(
        self,
        processed_data,
        metrics,
        cell_range,
        show_idle,
        level,
        save_jpeg=None,
        pickle_file=None,
        metric_subsets=None,
    ):
        raise NotImplementedError

    def _plot_direct(
        self,
        metric_subsets,
        cell_range,
        show_idle,
        level,
        save_jpeg=None,
        pickle_file=None,
    ):
        processed_data = self._prepare_processed_data_for_level(
            cell_range, show_idle, level
        )
        if processed_data is None:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[
                    ExtensionErrorCode.NO_PERFORMANCE_DATA
                ]
            )
            return

        metrics, _ = self._collect_metric_options(metric_subsets)
        if not metrics:
            logger.warning("No valid metrics found to plot")
            return

        self._render_direct_plot(
            processed_data=processed_data,
            metrics=metrics,
            cell_range=cell_range,
            show_idle=show_idle,
            level=level,
            save_jpeg=save_jpeg,
            pickle_file=pickle_file,
            metric_subsets=metric_subsets,
        )

    def plot(
        self,
        metric_subsets=("cpu", "mem", "io"),
        cell_range=None,
        show_idle=False,
        level=None,
        save_jpeg=None,
        pickle_file=None,
    ):
        metrics_missing = not metric_subsets
        if metrics_missing:
            metric_subsets = ("cpu", "mem", "io")
            if self.monitor.num_gpus:
                metric_subsets += (
                    "gpu",
                    "gpu_all",
                )

        """Plot performance metrics with interactive widgets for
        configuration."""
        valid_cells = self.cell_history.view()
        if len(valid_cells) == 0:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[ExtensionErrorCode.NO_CELL_HISTORY]
            )
            return

        # Default to all cells if no range specified
        try:
            min_cell_idx = int(valid_cells["cell_index"].min())
            max_cell_idx = int(valid_cells["cell_index"].max())
        except Exception:
            min_cell_idx, max_cell_idx = 0, len(valid_cells) - 1
        if cell_range is None:
            cell_start_index = 0
            for cell_idx in range(len(valid_cells) - 1, -1, -1):
                if valid_cells.iloc[cell_idx]["duration"] > self.min_duration:
                    cell_start_index = cell_idx
                    break
            start = int(valid_cells.iloc[cell_start_index]["cell_index"])
            end = int(valid_cells["cell_index"].max())
            if start > end:
                start, end = end, start
            cell_range = (start, end)

        # If level is specified, plot directly without widgets
        if level is not None:
            metric_subsets = self._resolve_metric_subsets(metric_subsets)
            return self._plot_direct(metric_subsets, cell_range, show_idle,
                                     level, save_jpeg, pickle_file)

        # Create interactive widgets
        style = {"description_width": "initial"}
        show_idle_checkbox = widgets.Checkbox(
            value=show_idle, description="Show idle periods"
        )
        # Sanitize slider value within bounds and ordered
        try:
            s0, s1 = cell_range
            if s0 > s1:
                s0, s1 = s1, s0
            s0 = max(min_cell_idx, min(s0, max_cell_idx))
            s1 = max(min_cell_idx, min(s1, max_cell_idx))
            slider_value = (s0, s1)
        except Exception:
            slider_value = (min_cell_idx, max_cell_idx)
        cell_range_slider = widgets.IntRangeSlider(
            value=slider_value,
            min=min_cell_idx,
            max=max_cell_idx,
            step=1,
            description="Cell range:",
            style=style,
        )

        logo_widget = widgets.HTML(
            value=f"<img src="
            f'"{logo_image}"'
            f'alt="JUmPER Logo" style="height: auto; width: 100px;">'
        )

        box_layout = Layout(
            display="flex",
            flex_flow="row wrap",
            align_items="center",
            justify_content="space-between",
            width="100%",
        )

        config_widgets = widgets.HBox(
            [
                widgets.HTML("<b>Plot Configuration:</b>"),
                show_idle_checkbox,
                cell_range_slider,
                logo_widget,
            ],
            layout=box_layout,
        )
        plot_output = widgets.Output()
        plot_wrapper = None

        def update_plots():
            nonlocal plot_wrapper
            current_cell_range, current_show_idle = (
                cell_range_slider.value,
                show_idle_checkbox.value,
            )
            processed_perfdata = self._prepare_processed_data_for_interactive(
                current_cell_range, current_show_idle
            )
            if processed_perfdata is None:
                with plot_output:
                    plot_output.clear_output()
                    logger.warning(
                        EXTENSION_ERROR_MESSAGES[
                            ExtensionErrorCode.NO_PERFORMANCE_DATA
                        ]
                    )
                    plot_wrapper = None
                return

            metrics, labeled_options = self._collect_metric_options(
                metric_subsets
            )
            if not metrics:
                with plot_output:
                    plot_output.clear_output()
                    logger.warning("No valid metrics found to plot")
                    plot_wrapper = None
                return

            with plot_output:
                if plot_wrapper is None:
                    plot_output.clear_output()
                    plot_wrapper = self._create_interactive_wrapper(
                        metrics,
                        labeled_options,
                        processed_perfdata,
                        current_cell_range,
                        current_show_idle,
                    )
                    plot_wrapper.display_ui()
                else:
                    plot_wrapper.update_data(
                        processed_perfdata,
                        current_cell_range,
                        current_show_idle,
                    )

        for widget in [show_idle_checkbox, cell_range_slider]:
            widget.observe(lambda change: update_plots(), names="value")

        display(widgets.VBox([config_widgets, plot_output]))
        update_plots()

attach(monitor)

Attach started PerformanceMonitor.

Source code in jumper_extension/adapters/visualizer.py

def attach(
    self,
    monitor: MonitorProtocol,
):
    """Attach started PerformanceMonitor."""
    self.monitor = monitor
    self.min_duration = self.monitor.interval
    # Smooth IO with ~1s rolling window based on sampling interval
    try:
        self._io_window = max(
            1, int(round(1.0 / (self.monitor.interval or 1.0)))
        )
    except Exception:
        self._io_window = 1
    self._build_subsets()

PlotlyPerformanceVisualizer

Bases: PerformanceVisualizer

Plotly-based visualizer compatible with VisualizerProtocol.

Source code in jumper_extension/adapters/visualizer.py

class PlotlyPerformanceVisualizer(PerformanceVisualizer):
    """Plotly-based visualizer compatible with VisualizerProtocol."""

    def _build_metric_plot(self, df, metric, show_idle=False, level="process"):
        config = next(
            (
                subset[metric]
                for subset in self.subsets.values()
                if metric in subset
            ),
            None,
        )
        if not config or not isinstance(config, dict):
            return None

        traces = []
        y_values = []
        plot_type = config.get("type")
        title = config.get("title", "")
        ylim = config.get("ylim")

        if plot_type == "single_series":
            column = config.get("column")
            if not column or column not in df.columns:
                return None

            series = df[column]
            if metric in (
                "io_read",
                "io_write",
                "io_read_count",
                "io_write_count",
            ):
                diffs = df[column].astype(float).diff().clip(lower=0)
                if metric in ("io_read", "io_write"):
                    diffs = diffs / (1024**2)
                series = diffs.fillna(0.0)
                if self._io_window and self._io_window > 1:
                    series = series.rolling(
                        window=self._io_window, min_periods=1
                    ).mean()

            trace = go.Scatter(
                x=df["time"].tolist(),
                y=series.tolist(),
                mode="lines",
                line=dict(color="blue", width=2),
                name=config.get("label", column),
            )
            traces.append(trace)
            y_values.extend(series.tolist())
            if metric == "memory" and ylim is None:
                ylim = (0, self.monitor.memory_limits[level])

        elif plot_type == "summary_series":
            columns = config.get("columns", [])
            if level == "system":
                title = re.sub(
                    r"\d+", str(self.monitor.num_system_cpus), title
                )
            line_styles = ["dot", "solid", "dash"]
            alpha_vals = [0.35, 1.0, 0.35]
            labels = ["Min", "Average", "Max"]

            for i, col in enumerate(columns):
                if col not in df.columns:
                    continue
                y_series = df[col]
                trace = go.Scatter(
                    x=df["time"].tolist(),
                    y=y_series.tolist(),
                    mode="lines",
                    line=dict(
                        color="blue",
                        dash=line_styles[i % len(line_styles)],
                        width=2,
                    ),
                    opacity=alpha_vals[i % len(alpha_vals)],
                    name=labels[i % len(labels)],
                )
                traces.append(trace)
                y_values.extend(y_series.tolist())

        elif plot_type == "multi_series":
            prefix = config.get("prefix", "")
            series_cols = [
                col
                for col in df.columns
                if prefix
                and col.startswith(prefix)
                and not col.endswith("avg")
            ]
            avg_column = f"{prefix}avg" if prefix else None
            if (
                avg_column is None or avg_column not in df.columns
            ) and not series_cols:
                return None

            for col in series_cols:
                y_series = df[col]
                traces.append(
                    go.Scatter(
                        x=df["time"].tolist(),
                        y=y_series.tolist(),
                        mode="lines",
                        line=dict(width=1),
                        opacity=0.5,
                        name=col,
                    )
                )
                y_values.extend(y_series.tolist())

            if avg_column in df.columns:
                avg_series = df[avg_column]
                traces.append(
                    go.Scatter(
                        x=df["time"].tolist(),
                        y=avg_series.tolist(),
                        mode="lines",
                        line=dict(color="blue", width=2),
                        name="Mean",
                    )
                )
                y_values.extend(avg_series.tolist())
        else:
            return None

        clean_values = []
        for value in y_values:
            try:
                val = float(value)
            except (TypeError, ValueError):
                continue
            if val != val:
                continue
            if val == float("inf") or val == float("-inf"):
                continue
            clean_values.append(val)

        if ylim is None:
            if clean_values:
                y_min = min(clean_values)
                y_max = max(clean_values)
                if y_min == y_max:
                    pad = abs(y_min) * 0.05 or 1.0
                else:
                    pad = (y_max - y_min) * 0.05
                ylim = (y_min - pad, y_max + pad)
            else:
                ylim = (0, 1)

        title = title + (" (No Idle)" if not show_idle else "")
        return {"traces": traces, "title": title, "ylim": ylim}

    def _get_plotly_cell_boundaries(self, cell_range=None, show_idle=False):
        min_duration = self.min_duration or 0
        boundaries = []

        if not show_idle and hasattr(self, "_compressed_cell_boundaries"):
            for cell in self._compressed_cell_boundaries:
                duration = cell.get("duration", 0)
                if duration < min_duration:
                    continue
                boundaries.append(
                    {
                        "start_time": float(cell["start_time"]),
                        "duration": float(duration),
                        "cell_index": int(cell["cell_index"]),
                    }
                )
            return boundaries

        filtered_cells = self.cell_history.view()
        if cell_range:
            try:
                mask = (filtered_cells["cell_index"] >= cell_range[0]) & (
                    filtered_cells["cell_index"] <= cell_range[1]
                )
                filtered_cells = filtered_cells[mask]
            except Exception:
                pass

        monitor_start = self.monitor.start_time or 0.0
        for _, cell in filtered_cells.iterrows():
            try:
                duration = float(cell["duration"])
                if duration < min_duration:
                    continue
                boundaries.append(
                    {
                        "start_time": float(cell["start_time"]) - monitor_start,
                        "duration": duration,
                        "cell_index": int(cell["cell_index"]),
                    }
                )
            except Exception:
                continue
        return boundaries

    def _draw_cell_boundaries_plotly(
        self,
        fig,
        row,
        ylim,
        cell_range=None,
        show_idle=False,
    ):
        y_min, y_max = ylim
        height = (y_max - y_min) or 1.0
        axis_suffix = "" if row == 1 else str(row)
        xref = f"x{axis_suffix}"
        yref = f"y{axis_suffix}"

        for cell in self._get_plotly_cell_boundaries(cell_range, show_idle):
            start_time = cell["start_time"]
            duration = cell["duration"]
            cell_num = cell["cell_index"]
            color = jumper_colors[cell_num % len(jumper_colors)]
            fig.add_shape(
                type="rect",
                x0=start_time,
                x1=start_time + duration,
                y0=y_min,
                y1=y_max,
                xref=xref,
                yref=yref,
                fillcolor=color,
                opacity=0.4,
                line=dict(color="black", dash="dash", width=1),
                layer="below",
            )
            fig.add_annotation(
                x=start_time + duration / 2,
                y=y_max - height * 0.1,
                xref=xref,
                yref=yref,
                text=f"#{cell_num}",
                showarrow=False,
                font=dict(size=10),
                bgcolor="rgba(255,255,255,0.8)",
            )

    def _compute_cell_boundaries_json(self, cell_range, show_idle):
        """Return cell boundary data as a list of plain dicts for JS consumption."""
        result = []
        for cell in self._get_plotly_cell_boundaries(cell_range, show_idle):
            cell_num = int(cell["cell_index"])
            result.append(
                {
                    "cell_index": cell_num,
                    "x0": float(cell["start_time"]),
                    "x1": float(cell["start_time"] + cell["duration"]),
                    "color": jumper_colors[cell_num % len(jumper_colors)],
                }
            )
        return result

    def _build_single_metric_figure(
        self,
        df,
        metric,
        cell_range=None,
        show_idle=False,
        level="process",
        include_boundaries=True,
    ):
        metric_plot = self._build_metric_plot(
            df, metric, show_idle=show_idle, level=level
        )
        if not metric_plot:
            return None

        fig = go.Figure()
        for trace in metric_plot["traces"]:
            fig.add_trace(trace)

        fig.update_layout(
            title=metric_plot["title"],
            xaxis_title="Time (seconds)",
            template="plotly_white",
            legend=dict(
                orientation="h",
                yanchor="top",
                y=0.99,
                xanchor="center",
                x=0.5,
                bgcolor="rgba(255,255,255,0.8)",
            ),
            margin=dict(l=24, r=8, t=45, b=35),
            # Keep width container-driven, but use a compact height close to
            # former matplotlib proportions.
            height=max(220, int(self.figsize[1] * 105)),
            autosize=True,
        )
        fig.update_xaxes(showgrid=True)
        fig.update_yaxes(showgrid=True, range=list(metric_plot["ylim"]))
        if include_boundaries:
            self._draw_cell_boundaries_plotly(
                fig,
                row=1,
                ylim=metric_plot["ylim"],
                cell_range=cell_range,
                show_idle=show_idle,
            )
        return fig

    def _render_direct_plot(
        self,
        processed_data,
        metrics,
        cell_range,
        show_idle,
        level,
        save_jpeg=None,
        pickle_file=None,
        metric_subsets=None,
    ):
        prepared = []
        for metric in metrics:
            metric_plot = self._build_metric_plot(
                processed_data, metric, show_idle=show_idle, level=level
            )
            if metric_plot:
                prepared.append((metric, metric_plot))

        if not prepared:
            logger.warning("No valid metrics found to plot")
            return

        fig = make_subplots(
            rows=len(prepared),
            cols=1,
            shared_xaxes=True,
            vertical_spacing=0.08,
            subplot_titles=[item[1]["title"] for item in prepared],
        )

        for row, (metric, metric_plot) in enumerate(prepared, start=1):
            for trace in metric_plot["traces"]:
                fig.add_trace(trace, row=row, col=1)
            fig.update_yaxes(
                range=list(metric_plot["ylim"]),
                showgrid=True,
                row=row,
                col=1,
            )
            fig.update_xaxes(showgrid=True, row=row, col=1)
            self._draw_cell_boundaries_plotly(
                fig,
                row=row,
                ylim=metric_plot["ylim"],
                cell_range=cell_range,
                show_idle=show_idle,
            )

        fig.update_xaxes(
            title_text="Time (seconds)",
            row=len(prepared),
            col=1,
        )
        fig.update_layout(
            template="plotly_white",
            showlegend=True,
            legend=dict(
                orientation="h",
                yanchor="top",
                y=0.99,
                xanchor="center",
                x=0.5,
                bgcolor="rgba(255,255,255,0.8)",
            ),
            # Roughly match legacy matplotlib subplot density while staying
            # responsive in width.
            height=max(260, int(270 * len(prepared))),
            margin=dict(l=24, r=8, t=40, b=35),
            autosize=True,
        )

        if save_jpeg:
            if not save_jpeg.endswith(".jpg") and not save_jpeg.endswith(
                ".jpeg"
            ):
                save_jpeg += ".jpg"
            fig.write_image(save_jpeg, format="jpeg", scale=2)
            print(f"Plot saved as JPEG: {save_jpeg}")

        if pickle_file:
            if not pickle_file.endswith(".pkl"):
                pickle_file += ".pkl"
            plot_data = {
                "figure_dict": fig.to_dict(),
                "metrics": [item[0] for item in prepared],
                "processed_data": processed_data,
                "cell_range": cell_range,
                "level": level,
                "show_idle": show_idle,
                "metric_subsets": metric_subsets,
            }
            with open(pickle_file, "wb") as f:
                pickle.dump(plot_data, f)

            print(f"Plot objects serialized to: {pickle_file}")
            print("\n# Python code to reload and display the plot:")
            print("import pickle")
            print("import plotly.graph_objects as go")
            print("")
            print(f"with open('{pickle_file}', 'rb') as f:")
            print("    plot_data = pickle.load(f)")
            print("")
            print("fig = go.Figure(plot_data['figure_dict'])")
            print("fig.show()")

        fig.show(config={"responsive": True})

    def _precompute_figures_for_wrapper(
        self,
        metrics,
        perfdata_no_idle,
        perfdata_with_idle,
        full_range,
    ):
        """Pre-compute all metric × level × show_idle figure dicts.

        Builds figures without embedded cell boundaries (JS draws them from the
        separately stored ``boundaries_false`` / ``boundaries_true`` lists).

        ``perfdata_no_idle`` must already be prepared before this call so that
        ``self._compressed_cell_boundaries`` is set to the correct full-range
        boundaries.  The with-idle figures are built second; they never read
        ``_compressed_cell_boundaries`` (gated on the ``show_idle`` flag inside
        ``_get_plotly_cell_boundaries``).
        """
        figures = {}  # metric → level → key → dict|None
        ylims   = {}  # metric → level → key → [ymin, ymax]
        all_levels = set(
            list(perfdata_no_idle or {}) + list(perfdata_with_idle or {})
        )

        for metric in metrics:
            figures[metric] = {}
            ylims[metric]   = {}
            for level in all_levels:
                figures[metric][level] = {}
                ylims[metric][level]   = {}
                # Build no-idle first (so _compressed_cell_boundaries is used)
                for key, df in [
                    ("false", (perfdata_no_idle or {}).get(level)),
                    ("true",  (perfdata_with_idle or {}).get(level)),
                ]:
                    if df is None or df.empty:
                        figures[metric][level][key] = None
                        ylims[metric][level][key]   = [0, 1]
                        continue
                    try:
                        fig = self._build_single_metric_figure(
                            df, metric, full_range,
                            show_idle=(key == "true"),
                            level=level,
                            include_boundaries=False,
                        )
                        if fig is not None:
                            fd = json.loads(fig.to_json())
                            figures[metric][level][key] = fd
                            try:
                                ylims[metric][level][key] = (
                                    fd["layout"]["yaxis"]["range"]
                                )
                            except (KeyError, TypeError):
                                ylims[metric][level][key] = [0, 1]
                        else:
                            figures[metric][level][key] = None
                            ylims[metric][level][key]   = [0, 1]
                    except Exception:
                        figures[metric][level][key] = None
                        ylims[metric][level][key]   = [0, 1]

        boundaries_false = self._compute_cell_boundaries_json(
            full_range, show_idle=False
        )
        boundaries_true = self._compute_cell_boundaries_json(
            full_range, show_idle=True
        )

        return {
            "figures":          figures,
            "ylims":            ylims,
            "boundaries_false": boundaries_false,
            "boundaries_true":  boundaries_true,
        }

    def _create_interactive_wrapper(
        self,
        metrics,
        labeled_options,
        processed_perfdata,
        current_cell_range,
        current_show_idle,
    ):
        # Kept for compatibility with the parent class contract; the Plotly
        # version delegates entirely to plot() and does not use this path.
        raise NotImplementedError(
            "_create_interactive_wrapper is not used by PlotlyPerformanceVisualizer; "
            "see plot() override instead."
        )

    def plot(
        self,
        metric_subsets=("cpu", "mem", "io"),
        cell_range=None,
        show_idle=False,
        level=None,
        save_jpeg=None,
        pickle_file=None,
    ):
        """Plot performance metrics using a self-contained pure HTML/JS output."""
        metrics_missing = not metric_subsets
        if metrics_missing:
            metric_subsets = ("cpu", "mem", "io")
            if self.monitor.num_gpus:
                metric_subsets += ("gpu", "gpu_all")

        valid_cells = self.cell_history.view()
        if len(valid_cells) == 0:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[ExtensionErrorCode.NO_CELL_HISTORY]
            )
            return

        try:
            min_cell_idx = int(valid_cells["cell_index"].min())
            max_cell_idx = int(valid_cells["cell_index"].max())
        except Exception:
            min_cell_idx, max_cell_idx = 0, len(valid_cells) - 1

        # Determine the initial slider position (default: last significant cell)
        if cell_range is None:
            cell_start_index = 0
            for cell_idx in range(len(valid_cells) - 1, -1, -1):
                if valid_cells.iloc[cell_idx]["duration"] > self.min_duration:
                    cell_start_index = cell_idx
                    break
            start = int(valid_cells.iloc[cell_start_index]["cell_index"])
            end   = int(valid_cells["cell_index"].max())
            if start > end:
                start, end = end, start
            initial_cell_range = (start, end)
        else:
            initial_cell_range = (
                max(min_cell_idx, min(cell_range[0], max_cell_idx)),
                max(min_cell_idx, min(cell_range[1], max_cell_idx)),
            )

        # When a specific level is requested, use the direct (non-interactive)
        # path which supports save_jpeg and pickle_file.
        if level is not None:
            metric_subsets = self._resolve_metric_subsets(metric_subsets)
            return self._plot_direct(
                metric_subsets, initial_cell_range, show_idle, level,
                save_jpeg, pickle_file,
            )

        metric_subsets = self._resolve_metric_subsets(metric_subsets)
        metrics, labeled_options = self._collect_metric_options(metric_subsets)
        if not metrics:
            logger.warning("No valid metrics found to plot")
            return

        # Use the FULL cell range so the slider can range across all cells.
        full_range = (min_cell_idx, max_cell_idx)

        # Prepare no-idle data first — this sets _compressed_cell_boundaries
        # for the full range, which is required before building no-idle figures
        # and before calling _compute_cell_boundaries_json(show_idle=False).
        processed_no_idle = self._prepare_processed_data_for_interactive(
            full_range, False
        )
        if processed_no_idle is None:
            logger.warning(
                EXTENSION_ERROR_MESSAGES[ExtensionErrorCode.NO_PERFORMANCE_DATA]
            )
            return

        # Prepare with-idle data (does not modify _compressed_cell_boundaries)
        processed_with_idle = self._prepare_processed_data_for_interactive(
            full_range, True
        )

        precomputed = self._precompute_figures_for_wrapper(
            metrics,
            processed_no_idle,
            processed_with_idle,
            full_range,
        )
        precomputed["min_cell_index"]    = min_cell_idx
        precomputed["max_cell_index"]    = max_cell_idx
        precomputed["initial_cell_range"] = list(initial_cell_range)

        perfdata_for_fallback = (
            processed_no_idle if not show_idle else processed_with_idle
        ) or {}

        wrapper = InteractivePlotlyWrapper(
            self._build_single_metric_figure,
            metrics,
            labeled_options,
            perfdata_for_fallback,
            initial_cell_range,
            show_idle,
            _precomputed_figures=precomputed,
        )
        wrapper.display_ui()

plot(metric_subsets=('cpu', 'mem', 'io'), cell_range=None, show_idle=False, level=None, save_jpeg=None, pickle_file=None)

Plot performance metrics using a self-contained pure HTML/JS output.

Source code in jumper_extension/adapters/visualizer.py

def plot(
    self,
    metric_subsets=("cpu", "mem", "io"),
    cell_range=None,
    show_idle=False,
    level=None,
    save_jpeg=None,
    pickle_file=None,
):
    """Plot performance metrics using a self-contained pure HTML/JS output."""
    metrics_missing = not metric_subsets
    if metrics_missing:
        metric_subsets = ("cpu", "mem", "io")
        if self.monitor.num_gpus:
            metric_subsets += ("gpu", "gpu_all")

    valid_cells = self.cell_history.view()
    if len(valid_cells) == 0:
        logger.warning(
            EXTENSION_ERROR_MESSAGES[ExtensionErrorCode.NO_CELL_HISTORY]
        )
        return

    try:
        min_cell_idx = int(valid_cells["cell_index"].min())
        max_cell_idx = int(valid_cells["cell_index"].max())
    except Exception:
        min_cell_idx, max_cell_idx = 0, len(valid_cells) - 1

    # Determine the initial slider position (default: last significant cell)
    if cell_range is None:
        cell_start_index = 0
        for cell_idx in range(len(valid_cells) - 1, -1, -1):
            if valid_cells.iloc[cell_idx]["duration"] > self.min_duration:
                cell_start_index = cell_idx
                break
        start = int(valid_cells.iloc[cell_start_index]["cell_index"])
        end   = int(valid_cells["cell_index"].max())
        if start > end:
            start, end = end, start
        initial_cell_range = (start, end)
    else:
        initial_cell_range = (
            max(min_cell_idx, min(cell_range[0], max_cell_idx)),
            max(min_cell_idx, min(cell_range[1], max_cell_idx)),
        )

    # When a specific level is requested, use the direct (non-interactive)
    # path which supports save_jpeg and pickle_file.
    if level is not None:
        metric_subsets = self._resolve_metric_subsets(metric_subsets)
        return self._plot_direct(
            metric_subsets, initial_cell_range, show_idle, level,
            save_jpeg, pickle_file,
        )

    metric_subsets = self._resolve_metric_subsets(metric_subsets)
    metrics, labeled_options = self._collect_metric_options(metric_subsets)
    if not metrics:
        logger.warning("No valid metrics found to plot")
        return

    # Use the FULL cell range so the slider can range across all cells.
    full_range = (min_cell_idx, max_cell_idx)

    # Prepare no-idle data first — this sets _compressed_cell_boundaries
    # for the full range, which is required before building no-idle figures
    # and before calling _compute_cell_boundaries_json(show_idle=False).
    processed_no_idle = self._prepare_processed_data_for_interactive(
        full_range, False
    )
    if processed_no_idle is None:
        logger.warning(
            EXTENSION_ERROR_MESSAGES[ExtensionErrorCode.NO_PERFORMANCE_DATA]
        )
        return

    # Prepare with-idle data (does not modify _compressed_cell_boundaries)
    processed_with_idle = self._prepare_processed_data_for_interactive(
        full_range, True
    )

    precomputed = self._precompute_figures_for_wrapper(
        metrics,
        processed_no_idle,
        processed_with_idle,
        full_range,
    )
    precomputed["min_cell_index"]    = min_cell_idx
    precomputed["max_cell_index"]    = max_cell_idx
    precomputed["initial_cell_range"] = list(initial_cell_range)

    perfdata_for_fallback = (
        processed_no_idle if not show_idle else processed_with_idle
    ) or {}

    wrapper = InteractivePlotlyWrapper(
        self._build_single_metric_figure,
        metrics,
        labeled_options,
        perfdata_for_fallback,
        initial_cell_range,
        show_idle,
        _precomputed_figures=precomputed,
    )
    wrapper.display_ui()

UnavailableVisualizer

A stub that type-checks against VisualizerProtocol but only logs that visualization is unavailable.

Source code in jumper_extension/adapters/visualizer.py

class UnavailableVisualizer:
    """
    A stub that type-checks against VisualizerProtocol but
    only logs that visualization is unavailable.
    """
    def __init__(self, reason: str = "Plotting not available."):
        self._reason = reason

    def attach(self, monitor: MonitorProtocol) -> None: ...

    def plot(
        self,
        metric_subsets=("cpu", "mem", "io"),
        cell_range=None,
        show_idle=False,
    ) -> None:
        logger.info(
            EXTENSION_INFO_MESSAGES[ExtensionInfoCode.PLOTS_NOT_AVAILABLE].format(
                reason=self._reason
            )
        )

VisualizerProtocol

Bases: Protocol

Structural protocol for visualizers used by the service.

Source code in jumper_extension/adapters/visualizer.py

@runtime_checkable
class VisualizerProtocol(Protocol):
    """Structural protocol for visualizers used by the service."""
    def attach(self, monitor: MonitorProtocol) -> None: ...
    def plot(
        self,
        metric_subsets=("cpu", "mem", "io"),
        cell_range=None,
        show_idle=False,
        level=None,
        save_jpeg=None,
        pickle_file=None,
    ) -> None: ...

build_performance_visualizer(cell_history, plots_disabled=False, plots_disabled_reason='Plotting not available.', backend='matplotlib')

Build visualizer object with selected backend.

Supported backends: - matplotlib (default) - plotly

Source code in jumper_extension/adapters/visualizer.py

def build_performance_visualizer(
    cell_history: CellHistory,
    plots_disabled: bool = False,
    plots_disabled_reason: str = "Plotting not available.",
    backend: str = "matplotlib",
) -> VisualizerProtocol:
    """
    Build visualizer object with selected backend.

    Supported backends:
    - matplotlib (default)
    - plotly
    """
    if plots_disabled:
        return UnavailableVisualizer(reason=plots_disabled_reason)

    backend_name = (backend or "matplotlib").strip().lower()
    if backend_name == "plotly":
        return PlotlyPerformanceVisualizer(cell_history)
    if backend_name != "matplotlib":
        logger.warning(
            f"Unknown visualizer backend '{backend}'. "
            "Falling back to matplotlib."
        )
    return MatplotlibPerformanceVisualizer(cell_history)

Sessions and scripts

SessionExporter

Handles exporting a monitoring session to directory/ZIP.

Source code in jumper_extension/adapters/session.py

class SessionExporter:
    """Handles exporting a monitoring session to directory/ZIP."""

    def __init__(self, monitor, cell_history, visualizer, reporter, logger):
        self.monitor = monitor
        self.cell_history = cell_history
        self.visualizer = visualizer
        self.reporter = reporter
        self.logger = logger

    def export(self, path: Optional[str] = None) -> str:
        """Export session to a directory or, if path ends with .zip, to a zip archive.

        Returns the path to the exported directory or created archive.
        """
        export_dir, zip_target = self._determine_export_paths(path)
        os.makedirs(export_dir, exist_ok=True)

        schemas_perf = self._export_performance_data(export_dir)
        ch_df = self._export_cell_history(export_dir)
        manifest = self._build_manifest(schemas_perf, ch_df)
        self._write_manifest(export_dir, manifest)

        if zip_target:
            return self._create_zip_archive(export_dir, zip_target)

        self.logger.info(
            EXTENSION_INFO_MESSAGES[ExtensionInfoCode.EXPORT_SUCCESS].format(
                filename=export_dir
            )
        )
        return export_dir

    def _determine_export_paths(self, path: Optional[str]) -> tuple:
        """Determine the export directory and optional ZIP target path.

        Returns:
            tuple: (export_dir, zip_target) where zip_target is None if not creating a ZIP
        """
        zip_target = None

        if path and path.lower().endswith(".zip"):
            export_dir = tempfile.mkdtemp(prefix="jumper-session-")
            zip_target = path
        else:
            export_dir = os.path.abspath(path or self._default_session_dirname())

        return export_dir, zip_target

    def _export_performance_data(self, export_dir: str) -> Dict[str, List[str]]:
        """Export performance data CSVs for each monitoring level.

        Args:
            export_dir: Directory to write CSV files to

        Returns:
            Dict mapping level names to their column schemas
        """
        schemas_perf: Dict[str, List[str]] = {}
        level_filenames = {
            "process": "perf_process.csv",
            "user": "perf_user.csv",
            "system": "perf_system.csv",
            "slurm": "perf_slurm.csv",
        }

        for level, df in self.monitor.data.data.items():
            try:
                df_out = self.monitor.data.view(level=level, cell_history=self.cell_history)
            except Exception:
                df_out = df
            schemas_perf[level] = list(df_out.columns)
            if not df_out.empty:
                fname = level_filenames.get(level, f"perf_{level}.csv")
                df_out.to_csv(os.path.join(export_dir, fname), index=False)

        return schemas_perf

    def _export_cell_history(self, export_dir: str) -> pd.DataFrame:
        """Export cell history to CSV.

        Args:
            export_dir: Directory to write the cell history CSV to

        Returns:
            DataFrame containing the cell history
        """
        ch_df = self.cell_history.view()
        if not ch_df.empty:
            ch_df.to_csv(os.path.join(export_dir, "cell_history.csv"), index=False)
        return ch_df

    def _build_manifest(self, schemas_perf: Dict[str, List[str]], ch_df: pd.DataFrame) -> dict:
        """Build the manifest dictionary containing session metadata.

        Args:
            schemas_perf: Performance data schemas by level
            ch_df: Cell history DataFrame

        Returns:
            Manifest dictionary
        """
        return {
            "version": "1.0",
            "app": {"name": "JUmPER", "version": self._app_version()},
            "monitor": {
                "interval": getattr(self.monitor, "interval", 1.0),
                "start_time": getattr(self.monitor, "start_time", None),
                "stop_time": getattr(self.monitor, "stop_time", None),
                "wallclock_start_time": getattr(self.monitor, "wallclock_start_time", None),
                "wallclock_stop_time": getattr(self.monitor, "wallclock_stop_time", None),
                "num_cpus": getattr(self.monitor, "num_cpus", 0),
                "num_system_cpus": getattr(self.monitor, "num_system_cpus", 0),
                "num_gpus": getattr(self.monitor, "num_gpus", 0),
                "gpu_memory": getattr(self.monitor, "gpu_memory", 0.0),
                "gpu_name": getattr(self.monitor, "gpu_name", ""),
                "memory_limits": getattr(self.monitor, "memory_limits", {}),
                "cpu_handles": getattr(self.monitor, "cpu_handles", []),
                "pid": getattr(self.monitor, "pid", None),
                "uid": getattr(self.monitor, "uid", None),
                "slurm_job": getattr(self.monitor, "slurm_job", None),
                "os": os.name,
                "python": sys.version.split(" ")[0],
            },
            "levels": list(self.monitor.data.data.keys()),
            "schemas": {
                "perf": schemas_perf,
                "cell_history": list(ch_df.columns),
            },
            "visualizer": {
                "default_metric_subsets": [
                    "cpu",
                    "mem",
                    "io",
                ] + (["gpu", "gpu_all"] if getattr(self.monitor, "num_gpus", 0) else []),
                "figsize": list(getattr(self.visualizer, "figsize", (5, 3))),
                "io_window": getattr(self.visualizer, "_io_window", None),
                "last_state": {},
            },
            "reporter": {
                "level": getattr(self.reporter, "level", "process") if hasattr(self.reporter, "level") else "process",
                "format": "text",
                "thresholds": getattr(self.reporter.printer.analyzer, "thresholds", {}),
            },
            "time_origin": "perf_counter",
            "timezone": time.tzname[0] if time.tzname else "",
        }

    def _write_manifest(self, export_dir: str, manifest: dict) -> None:
        """Write the manifest JSON file to the export directory.

        Args:
            export_dir: Directory to write the manifest to
            manifest: Manifest dictionary to serialize
        """
        with open(os.path.join(export_dir, "manifest.json"), "w", encoding="utf-8") as f:
            json.dump(manifest, f, indent=2)

    def _create_zip_archive(self, export_dir: str, zip_target: str) -> str:
        """Create a ZIP archive from the export directory and clean up.

        Args:
            export_dir: Directory containing exported files
            zip_target: Path to the ZIP file to create

        Returns:
            Path to the created ZIP file
        """
        with zipfile.ZipFile(zip_target, "w", compression=zipfile.ZIP_DEFLATED) as zf:
            for root, _, files in os.walk(export_dir):
                for name in files:
                    ap = os.path.join(root, name)
                    rel = os.path.relpath(ap, export_dir)
                    zf.write(ap, rel)

        self.logger.info(
            EXTENSION_INFO_MESSAGES[ExtensionInfoCode.EXPORT_SUCCESS].format(
                filename=zip_target
            )
        )

        # Clean up temp dir
        try:
            shutil.rmtree(export_dir)
        except Exception:
            pass

        return zip_target

    def _default_session_dirname(self) -> str:
        ts = datetime.now().strftime("%Y%m%d-%H%M%S")
        return f"jumper-session-{ts}"

    def _app_version(self) -> str:
        try:
            here = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
            pyproject = os.path.join(here, "..", "pyproject.toml")
            pyproject = os.path.normpath(pyproject)
            if os.path.exists(pyproject):
                with open(pyproject, "r", encoding="utf-8") as f:
                    for line in f:
                        s = line.strip()
                        if s.startswith("version") and "=" in s:
                            val = s.split("=", 1)[1].strip().strip('"')
                            if val:
                                return val
        except Exception:
            pass
        return "unknown"

export(path=None)

Export session to a directory or, if path ends with .zip, to a zip archive.

Returns the path to the exported directory or created archive.

Source code in jumper_extension/adapters/session.py

def export(self, path: Optional[str] = None) -> str:
    """Export session to a directory or, if path ends with .zip, to a zip archive.

    Returns the path to the exported directory or created archive.
    """
    export_dir, zip_target = self._determine_export_paths(path)
    os.makedirs(export_dir, exist_ok=True)

    schemas_perf = self._export_performance_data(export_dir)
    ch_df = self._export_cell_history(export_dir)
    manifest = self._build_manifest(schemas_perf, ch_df)
    self._write_manifest(export_dir, manifest)

    if zip_target:
        return self._create_zip_archive(export_dir, zip_target)

    self.logger.info(
        EXTENSION_INFO_MESSAGES[ExtensionInfoCode.EXPORT_SUCCESS].format(
            filename=export_dir
        )
    )
    return export_dir

SessionImporter

Handles importing a monitoring session from directory/ZIP.

Source code in jumper_extension/adapters/session.py

class SessionImporter:
    """Handles importing a monitoring session from directory/ZIP."""

    def __init__(self, logger):
        self.logger = logger

    def import_(self, path: str, service) -> bool:
        """Import a session into the given service. Returns True on success."""
        if not path:
            return False

        work_dir, cleanup_dir = self._prepare_work_directory(path)

        try:
            manifest = self._load_manifest(work_dir)
            self._load_cell_history(work_dir, service)
            perf_dfs = self._load_performance_data(work_dir)

            self._setup_offline_monitor(manifest, perf_dfs, service, source=path)
            self._setup_reporter(manifest, service)
            self._apply_visualizer_settings(manifest, service)

            return True
        finally:
            if cleanup_dir and work_dir and os.path.isdir(work_dir):
                try:
                    shutil.rmtree(work_dir)
                except Exception:
                    pass

    def _prepare_work_directory(self, path: str) -> tuple:
        """Prepare the work directory from ZIP or direct path.

        Args:
            path: Path to ZIP file or directory

        Returns:
            tuple: (work_dir, cleanup_dir) where cleanup_dir indicates if temp dir should be cleaned
        """
        if path.lower().endswith(".zip"):
            work_dir = tempfile.mkdtemp(prefix="jumper-session-import-")
            with zipfile.ZipFile(path, "r") as zf:
                zf.extractall(work_dir)
            return work_dir, True
        else:
            return path, False

    def _load_manifest(self, work_dir: str) -> dict:
        """Load the manifest JSON file from the work directory.

        Args:
            work_dir: Directory containing the manifest file

        Returns:
            Manifest dictionary
        """
        manifest_path = os.path.join(work_dir, "manifest.json")
        with open(manifest_path, "r", encoding="utf-8") as f:
            return json.load(f)

    def _load_cell_history(self, work_dir: str, service) -> None:
        """Load cell history data from CSV into the service.

        Args:
            work_dir: Directory containing the cell history CSV
            service: Service object to load data into
        """
        ch_csv = os.path.join(work_dir, "cell_history.csv")
        if os.path.exists(ch_csv):
            try:
                service.cell_history.data = pd.read_csv(ch_csv)
            except Exception:
                pass

    def _load_performance_data(self, work_dir: str) -> Dict[str, pd.DataFrame]:
        """Load performance data CSVs from the work directory.

        Args:
            work_dir: Directory containing performance CSV files

        Returns:
            Dict mapping level names to their DataFrames
        """
        level_files = {
            "process": "perf_process.csv",
            "user": "perf_user.csv",
            "system": "perf_system.csv",
            "slurm": "perf_slurm.csv",
        }
        perf_dfs: Dict[str, pd.DataFrame] = {}

        for level, fname in level_files.items():
            fpath = os.path.join(work_dir, fname)
            if os.path.exists(fpath):
                try:
                    perf_dfs[level] = pd.read_csv(fpath)
                except Exception:
                    continue

        return perf_dfs

    def _setup_offline_monitor(self, manifest: dict, perf_dfs: Dict[str, pd.DataFrame], service, source: Optional[str]) -> None:
        """Create and attach an offline performance monitor to the service.

        Args:
            manifest: Manifest dictionary with monitor configuration
            perf_dfs: Performance data DataFrames by level
            service: Service object to attach monitor to
        """
        offline = OfflinePerformanceMonitor(
            manifest=manifest,
            perf_dfs=perf_dfs,
            source=source,
        )
        service.monitor = offline
        service.visualizer.attach(service.monitor)

    def _setup_reporter(self, manifest: dict, service) -> None:
        """Rebuild and attach the performance reporter with thresholds from manifest.

        Args:
            manifest: Manifest dictionary with reporter configuration
            service: Service object to attach reporter to
        """
        thresholds = None
        try:
            thresholds = manifest.get("reporter", {}).get("thresholds")
        except Exception:
            thresholds = None

        from jumper_extension.adapters.reporter import build_performance_reporter

        service.reporter = build_performance_reporter(
            service.cell_history,
            display_disabled=False,
            display_disabled_reason="Display not available.",
            thresholds=thresholds,
        )
        service.reporter.attach(service.monitor)

    def _apply_visualizer_settings(self, manifest: dict, service) -> None:
        """Apply visualizer settings from the manifest to the service.

        Args:
            manifest: Manifest dictionary with visualizer configuration
            service: Service object with visualizer to configure
        """
        try:
            viz = manifest.get("visualizer", {})
            if isinstance(viz.get("figsize"), list) and len(viz.get("figsize")) == 2:
                service.visualizer.figsize = (viz["figsize"][0], viz["figsize"][1])
            if viz.get("io_window"):
                try:
                    service.visualizer._io_window = int(viz.get("io_window"))
                except Exception:
                    pass
        except Exception:
            pass

import_(path, service)

Import a session into the given service. Returns True on success.

Source code in jumper_extension/adapters/session.py

def import_(self, path: str, service) -> bool:
    """Import a session into the given service. Returns True on success."""
    if not path:
        return False

    work_dir, cleanup_dir = self._prepare_work_directory(path)

    try:
        manifest = self._load_manifest(work_dir)
        self._load_cell_history(work_dir, service)
        perf_dfs = self._load_performance_data(work_dir)

        self._setup_offline_monitor(manifest, perf_dfs, service, source=path)
        self._setup_reporter(manifest, service)
        self._apply_visualizer_settings(manifest, service)

        return True
    finally:
        if cleanup_dir and work_dir and os.path.isdir(work_dir):
            try:
                shutil.rmtree(work_dir)
            except Exception:
                pass

NotebookScriptWriter

Class for writing notebook content to a Python script.

Collects code from cells and saves it to a Python file with optional metadata about execution time and cell numbers.

Source code in jumper_extension/adapters/script_writer.py

class NotebookScriptWriter:
    """
    Class for writing notebook content to a Python script.

    Collects code from cells and saves it to a Python file with optional
    metadata about execution time and cell numbers.
    """

    def __init__(self, cell_history: CellHistory):
        self.cell_history = cell_history
        self.output_path = None
        # recording state
        self._recording = False
        self._start_time = None
        self._start_cell_index: Optional[int] = None
        self._settings_state = Settings()
        # names of magics that start/stop script writing (to exclude their cells)
        self._control_magics = {"start_write_script", "end_write_script"}

    def is_recording_active(self) -> bool:
        """Check if cell is being recorded."""
        return self._recording

    def start_recording(self, settings_state: Settings, output_path: Optional[str] = None):
        """
        Start recording code from cells.

        Args:
            settings_state: Extension settings at the time of recording started
            output_path: Path to the output file (overrides value from __init__)
        """
        self._settings_state = settings_state
        if output_path:
            self.output_path = output_path
        else:
            # Generate default filename
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            self.output_path = f"notebook_script_{timestamp}.py"

        # mark the current tail of CellHistory; everything after this point is "to be written"
        self._recording = True
        self._start_time = datetime.now()
        # exclude the cell that triggered the start magic itself
        self._start_cell_index = len(self.cell_history)

    def stop_recording(self) -> Optional[str]:
        """
        Stop recording and save accumulated code to file.

        Returns:
            Path to the created file or None on error
        """
        if not self._recording:
            logger.warning("[JUmPER]: Recording was not started")
            return None

        # collect cells recorded since start, excluding start/end control magic cells
        try:
            history = self.cell_history.view()
        except Exception as e:
            logger.error(f"[JUmPER]: Failed to access CellHistory: {e}")
            return None

        if history is None or history.empty:
            logger.warning("[JUmPER]: No cells in CellHistory")
            return None

        selected = []
        for _, row in history.iterrows():
            try:
                idx = int(row.get("cell_index"))
            except Exception:
                continue
            if self._start_cell_index is not None and idx < self._start_cell_index:
                continue
            if self.is_control_cell(row.get("cell_magics")):
                continue
            selected.append(
                {
                    "index": idx,
                    "timestamp": datetime.fromtimestamp(row["start_time"])
                    if isinstance(row.get("start_time"), (int, float))
                    else self._start_time or datetime.now(),
                    "raw_cell": row.get("raw_cell", ""),
                    "cell_magics": row.get("cell_magics") or [],
                }
            )

        if not selected:
            logger.warning("[JUmPER]: No recorded cells to save")
            # reset state
            self._recording = False
            self._start_cell_index = None
            return None

        try:
            self._write_to_file(selected)
            logger.info(
                f"[JUmPER]: Recorded {len(selected)} cells "
                f"to file '{self.output_path}'"
            )
            return self.output_path
        except Exception as e:
            logger.error(
                f"[JUmPER]: Error writing file: {e}"
            )
            return None
        finally:
            # reset state
            self._recording = False
            self._start_cell_index = None

    def is_control_cell(self, cell_magics):
        """Select cells with index >= start and exclude cells that contain control magics"""
        if cell_magics is None:
            return False
        try:
            for m in cell_magics:
                # m may be like "%perfmonitor_start ..." or "perfmonitor_start ..."
                name = m.lstrip("%")
                name = name.split(maxsplit=1)[0]
                if name in self._control_magics:
                    return True
        except Exception:
            pass
        return False
    def _write_to_file(self, recorded_cells: List[dict]):
        """
        Write accumulated cells to Python file.
        """
        output_path = Path(self.output_path)
        output_path.parent.mkdir(parents=True, exist_ok=True)

        with open(output_path, "w", encoding="utf-8") as f:
            # File header
            header = dedent(f"""\
                #!/usr/bin/env python3
                \"\"\"
                Auto-generated script from Jupyter notebook
                Generated: {datetime.now():%Y-%m-%d %H:%M:%S}
                Recording started: {self._start_time:%Y-%m-%d %H:%M:%S} if self._start_time else ""
                Total cells: {len(recorded_cells)}
                \"\"\"

                from jumper_extension.core.service import build_perfmonitor_magic_adapter
                magic_adapter = build_perfmonitor_magic_adapter(
                    plots_disabled=True,
                    plots_disabled_reason="Plotting disabled in generated script.",
                    display_disabled=True,
                    display_disabled_reason="Display disabled in generated script."
                )

                {self._restore_perfmonitor()}
            """)
            f.write(header)

            # Write code from each recorded cell
            for cell in recorded_cells:
                f.write(f"# Cell {cell['index']}\n")
                ts = cell['timestamp']
                f.write(f"# Recorded at: {ts.strftime('%H:%M:%S') if isinstance(ts, datetime) else ts}\n")
                raw_cell = cell.get("raw_cell", "")
                f.write("# --- Cell print ---\n")
                f.write(f"raw_cell = {raw_cell!r}\n")
                f.write(f"print('-' * 40)\n")
                f.write(f"print('Cell {cell['index']}')\n")
                f.write(f"print('-' * 40)\n")
                f.write(f"print(raw_cell)\n")
                f.write("print('-' * 13 + ' Cell output ' + '-' * 14)\n")
                cell_magics = cell.get("cell_magics") or []
                # compute should_skip_report: True if cell contains only line magics (non-empty lines start with '%')
                non_empty_lines = [ln for ln in raw_cell.splitlines() if ln.strip()]
                is_pure_magic = bool(non_empty_lines) and all(ln.lstrip().startswith("%") for ln in non_empty_lines)
                f.write(
                    "magic_adapter.on_pre_run_cell("
                    f"raw_cell, "
                    f"{cell_magics!r}, "
                    f"{is_pure_magic!r}"
                    ")\n"
                )
                f.write("# --- Cell content ---\n")
                transformed = self._transform_cell_code(
                    raw_cell,
                    cell_magics
                )
                f.write(f"{transformed}\n")
                f.write("# --- Cell End -------\n")
                f.write("magic_adapter.on_post_run_cell('')\n")
                f.write("\n")
            base_name = output_path.stem
            perf_csv = f"{base_name}_perfdata.csv"
            cell_csv = f"{base_name}_cell_history.csv"
            footer = dedent(
                f"""\
                # --- Export results to CSV ---
                # Performance data by level (default level from settings)
                magic_adapter.perfmonitor_export_perfdata("--file {perf_csv}")
                # Cell execution history
                magic_adapter.perfmonitor_export_cell_history("--file {cell_csv}")
                """
            )
            f.write(footer)

    def _restore_perfmonitor(self) -> str:
        if self._settings_state.monitoring.running:
            settings = self._settings_state

            # Determine interval to restore
            interval = settings.monitoring.user_interval
            if not interval:
                interval = settings.monitoring.default_interval

            # If auto-reports were enabled, a single enable call will both start monitoring
            # (if needed) and configure reports consistently with original settings.
            if settings.perfreports.enabled:
                level = settings.perfreports.level
                args = f"--level {level} --interval {interval}"
                if settings.perfreports.text:
                    args += " --text"
                return f"magic_adapter.perfmonitor_enable_perfreports({args!r})\n"

            # Otherwise just restore monitor start with the same interval
            return f"magic_adapter.perfmonitor_start({str(interval)!r})\n"

        return ""

    def _transform_cell_code(self, raw_cell: str, cell_magics: List[str]) -> str:
        """
        Replace captured magic commands with magic_adapter calls while keeping
        the rest of the code intact.
        """
        if not raw_cell:
            return ""

        # Build a lookup from magic line prefix to magic_adapter call string
        # We rely on CellHistory.cell_magics entries having the original magic lines (e.g. "%perfmonitor_start 1.0")
        replacements = {}
        for magic in cell_magics:
            # Normalize leading '%'
            stripped_no_pct = magic[1:] if magic.startswith("%") else magic
            parts = stripped_no_pct.split(maxsplit=1)
            cmd = parts[0]
            args = parts[1] if len(parts) > 1 else ""
            # construct a Python call to the magic_adapter method
            # prefer passing the whole "line" string of arguments
            if args:
                call = f"magic_adapter.{cmd}({args!r})"
            else:
                # Methods generally accept a single 'line' argument; pass empty string for uniformity
                call = f'magic_adapter.{cmd}("")'
            # map original magic literal (with or without %) to replacement
            replacements[magic] = call
            # also allow matching without the leading '%', just in case
            replacements[stripped_no_pct] = call

        # Now transform the cell line by line
        out_lines: List[str] = []
        for line in raw_cell.splitlines():
            lstrip = line.lstrip()
            # Only attempt replacement if line starts with a magic marker
            if lstrip.startswith("%"):
                # Exact match by full line (common for bare magic lines)
                rep = replacements.get(lstrip)
                if rep is None:
                    # Try by the first token
                    key = lstrip.split("#", 1)[0].strip()  # drop trailing inline comments if any
                    rep = replacements.get(key)
                if rep is None:
                    # As a fallback, try to parse and replace if it's one of captured commands
                    token = lstrip[1:].split(maxsplit=1)[0]
                    for k, v in replacements.items():
                        if k.lstrip("%").split(maxsplit=1)[0] == token:
                            rep = v
                            break
                if rep is not None:
                    # keep original indentation
                    indent = line[: len(line) - len(lstrip)]
                    out_lines.append(f"{indent}{rep}")
                    continue
            out_lines.append(line)
        return "\n".join(out_lines)

is_control_cell(cell_magics)

Select cells with index >= start and exclude cells that contain control magics

Source code in jumper_extension/adapters/script_writer.py

def is_control_cell(self, cell_magics):
    """Select cells with index >= start and exclude cells that contain control magics"""
    if cell_magics is None:
        return False
    try:
        for m in cell_magics:
            # m may be like "%perfmonitor_start ..." or "perfmonitor_start ..."
            name = m.lstrip("%")
            name = name.split(maxsplit=1)[0]
            if name in self._control_magics:
                return True
    except Exception:
        pass
    return False

is_recording_active()

Check if cell is being recorded.

Source code in jumper_extension/adapters/script_writer.py

def is_recording_active(self) -> bool:
    """Check if cell is being recorded."""
    return self._recording

start_recording(settings_state, output_path=None)

Start recording code from cells.

Parameters:

Name	Type	Description	Default
settings_state	Settings	Extension settings at the time of recording started	required
output_path	Optional[str]	Path to the output file (overrides value from init)	None

Source code in jumper_extension/adapters/script_writer.py

def start_recording(self, settings_state: Settings, output_path: Optional[str] = None):
    """
    Start recording code from cells.

    Args:
        settings_state: Extension settings at the time of recording started
        output_path: Path to the output file (overrides value from __init__)
    """
    self._settings_state = settings_state
    if output_path:
        self.output_path = output_path
    else:
        # Generate default filename
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.output_path = f"notebook_script_{timestamp}.py"

    # mark the current tail of CellHistory; everything after this point is "to be written"
    self._recording = True
    self._start_time = datetime.now()
    # exclude the cell that triggered the start magic itself
    self._start_cell_index = len(self.cell_history)

stop_recording()

Stop recording and save accumulated code to file.

Returns:

Type	Description
Optional[str]	Path to the created file or None on error

Source code in jumper_extension/adapters/script_writer.py

def stop_recording(self) -> Optional[str]:
    """
    Stop recording and save accumulated code to file.

    Returns:
        Path to the created file or None on error
    """
    if not self._recording:
        logger.warning("[JUmPER]: Recording was not started")
        return None

    # collect cells recorded since start, excluding start/end control magic cells
    try:
        history = self.cell_history.view()
    except Exception as e:
        logger.error(f"[JUmPER]: Failed to access CellHistory: {e}")
        return None

    if history is None or history.empty:
        logger.warning("[JUmPER]: No cells in CellHistory")
        return None

    selected = []
    for _, row in history.iterrows():
        try:
            idx = int(row.get("cell_index"))
        except Exception:
            continue
        if self._start_cell_index is not None and idx < self._start_cell_index:
            continue
        if self.is_control_cell(row.get("cell_magics")):
            continue
        selected.append(
            {
                "index": idx,
                "timestamp": datetime.fromtimestamp(row["start_time"])
                if isinstance(row.get("start_time"), (int, float))
                else self._start_time or datetime.now(),
                "raw_cell": row.get("raw_cell", ""),
                "cell_magics": row.get("cell_magics") or [],
            }
        )

    if not selected:
        logger.warning("[JUmPER]: No recorded cells to save")
        # reset state
        self._recording = False
        self._start_cell_index = None
        return None

    try:
        self._write_to_file(selected)
        logger.info(
            f"[JUmPER]: Recorded {len(selected)} cells "
            f"to file '{self.output_path}'"
        )
        return self.output_path
    except Exception as e:
        logger.error(
            f"[JUmPER]: Error writing file: {e}"
        )
        return None
    finally:
        # reset state
        self._recording = False
        self._start_cell_index = None