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

NodeDataStore

Single source of truth for per-node hardware metadata and time-series data.

register_node(info) stores both the NodeInfo and the corresponding PerformanceData container under the same node key.

Access patterns

store.hardware – Dict[str, NodeInfo] (metadata) store.view(level) – aggregate DataFrame across all nodes store.view(level, node=n) – single-node DataFrame store.add_sample(node, level, row) – append one flat-dict sample

Source code in jumper_extension/adapters/data/node.py

class NodeDataStore:
    """Single source of truth for per-node hardware metadata and time-series data.

    ``register_node(info)`` stores both the ``NodeInfo`` and the corresponding
    ``PerformanceData`` container under the same node key.

    Access patterns
    ---------------
    ``store.hardware``                     – Dict[str, NodeInfo] (metadata)
    ``store.view(level)``                  – aggregate DataFrame across all nodes
    ``store.view(level, node=n)``          – single-node DataFrame
    ``store.add_sample(node, level, row)`` – append one flat-dict sample
    """

    def __init__(self) -> None:
        self._info: Dict[str, NodeInfo] = {}
        self._nodes: Dict[str, PerformanceData] = {}

    # ------------------------------------------------------------------ #
    # Registration                                                        #
    # ------------------------------------------------------------------ #

    def register_node(self, info: NodeInfo) -> None:
        self._info[info.node] = info
        self._nodes[info.node] = PerformanceData()

    @property
    def hardware(self) -> Dict[str, NodeInfo]:
        return self._info

    def node_names(self) -> List[str]:
        return list(self._info.keys())

    @property
    def levels(self) -> List[str]:
        if not self._nodes:
            return get_available_levels()
        return list(next(iter(self._nodes.values())).levels)

    # ------------------------------------------------------------------ #
    # Writing                                                             #
    # ------------------------------------------------------------------ #

    def add_sample(self, node: str, level: str, row: dict) -> None:
        perf_data = self._nodes.get(node)
        if perf_data is None:
            return
        perf_data.add_sample(level, row)

    def init_node_schema(
        self, node: str, columns_by_level: Dict[str, List[str]]
    ) -> None:
        """Store per-level column lists so view() can return a correctly shaped
        empty DataFrame before the first sample arrives."""
        perf_data = self._nodes.get(node)
        if perf_data is None:
            return
        perf_data._schema_columns = dict(columns_by_level)

    def load_frames(self, node: str, frames: Dict[str, pd.DataFrame]) -> None:
        """Inject pre-loaded DataFrames into a registered node's data container.

        Used by offline (imported) monitors to populate data without going
        through the live add_sample() path.
        """
        perf_data = self._nodes.get(node)
        if perf_data is None:
            return
        for level, df in frames.items():
            perf_data._rows[level] = df.to_dict("records")

    # ------------------------------------------------------------------ #
    # Reading                                                             #
    # ------------------------------------------------------------------ #

    def view(
        self,
        level: str = "process",
        node: Optional[str] = None,
        slice_=None,
        cell_history=None,
    ) -> pd.DataFrame:
        if not self._nodes:
            return pd.DataFrame()

        if node is not None:
            perf_data = self._nodes.get(node)
            if perf_data is None:
                return pd.DataFrame()
            return perf_data.view(level=level, slice_=slice_, cell_history=cell_history)

        if len(self._nodes) == 1:
            return next(iter(self._nodes.values())).view(
                level=level, slice_=slice_, cell_history=cell_history
            )

        return self._aggregate(level, cell_history)

    def _aggregate(self, level: str, cell_history=None) -> pd.DataFrame:
        node_dfs: Dict[str, pd.DataFrame] = {}
        for n, perf in self._nodes.items():
            df = perf.view(level)
            if not df.empty:
                node_dfs[n] = df

        if not node_dfs:
            return pd.DataFrame()
        if len(node_dfs) == 1:
            df = next(iter(node_dfs.values()))
            return self._attach_cell_index(df, cell_history) if cell_history else df

        min_len = min(len(df) for df in node_dfs.values())
        if min_len == 0:
            return pd.DataFrame()

        frames = [
            df.iloc[:min_len].reset_index(drop=True)
            for df in node_dfs.values()
        ]
        result = frames[0].copy()

        self._aggregate_memory(frames, result)
        self._aggregate_io(frames, result)
        self._aggregate_cpu(frames, result)
        self._aggregate_gpu(frames, result)

        if cell_history is not None:
            result = self._attach_cell_index(result, cell_history)
        return result

    def _aggregate_memory(
        self, frames: List[pd.DataFrame], result: pd.DataFrame
    ) -> None:
        if all("memory" in f.columns for f in frames):
            result["memory"] = sum(f["memory"] for f in frames)

    def _aggregate_io(
        self, frames: List[pd.DataFrame], result: pd.DataFrame
    ) -> None:
        for col in ("io_read", "io_write", "io_read_count", "io_write_count"):
            if all(col in f.columns for f in frames):
                result[col] = sum(f[col] for f in frames)

    def _aggregate_cpu(
        self, frames: List[pd.DataFrame], result: pd.DataFrame
    ) -> None:
        if all("cpu_util_avg" in f.columns for f in frames):
            result["cpu_util_avg"] = sum(f["cpu_util_avg"] for f in frames) / len(frames)
        if all("cpu_util_min" in f.columns for f in frames):
            result["cpu_util_min"] = pd.concat(
                [f["cpu_util_min"] for f in frames], axis=1
            ).min(axis=1)
        if all("cpu_util_max" in f.columns for f in frames):
            result["cpu_util_max"] = pd.concat(
                [f["cpu_util_max"] for f in frames], axis=1
            ).max(axis=1)
        drop = [c for c in result.columns if re.match(r"cpu_util_\d+$", c)]
        result.drop(columns=drop, errors="ignore", inplace=True)

    def _aggregate_gpu(
        self, frames: List[pd.DataFrame], result: pd.DataFrame
    ) -> None:
        for metric in ("util", "band", "mem"):
            avg_col = f"gpu_{metric}_avg"
            vals = [f[avg_col] for f in frames if avg_col in f.columns]
            if vals:
                result[avg_col] = sum(vals) / len(vals)

            min_col = f"gpu_{metric}_min"
            vals = [f[min_col] for f in frames if min_col in f.columns]
            if vals:
                result[min_col] = pd.concat(vals, axis=1).min(axis=1)

            max_col = f"gpu_{metric}_max"
            vals = [f[max_col] for f in frames if max_col in f.columns]
            if vals:
                result[max_col] = pd.concat(vals, axis=1).max(axis=1)

        drop = [c for c in result.columns if re.match(r"gpu_(util|band|mem)_\d+$", c)]
        result.drop(columns=drop, errors="ignore", inplace=True)

    def _attach_cell_index(self, df: pd.DataFrame, 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

    # ------------------------------------------------------------------ #
    # Export / load (delegate to primary node)                           #
    # ------------------------------------------------------------------ #

    def export(
        self,
        filename: str = "performance_data.csv",
        level: str = "process",
        cell_history=None,
    ) -> None:
        if not self._nodes:
            return
        df = self.view(level=level, cell_history=cell_history)
        if df.empty:
            return
        first = next(iter(self._nodes.values()))
        _, ext = os.path.splitext(filename)
        format = ext.lower().lstrip(".") or "csv"
        if not format:
            format = "csv"
            filename += ".csv"
        writer = first._file_writers.get(format)
        if writer:
            writer(filename, df)

    def load(self, filename: str) -> Optional[pd.DataFrame]:
        if not self._nodes:
            return None
        return next(iter(self._nodes.values())).load(filename)

init_node_schema(node, columns_by_level)

Store per-level column lists so view() can return a correctly shaped empty DataFrame before the first sample arrives.

Source code in jumper_extension/adapters/data/node.py

def init_node_schema(
    self, node: str, columns_by_level: Dict[str, List[str]]
) -> None:
    """Store per-level column lists so view() can return a correctly shaped
    empty DataFrame before the first sample arrives."""
    perf_data = self._nodes.get(node)
    if perf_data is None:
        return
    perf_data._schema_columns = dict(columns_by_level)

load_frames(node, frames)

Inject pre-loaded DataFrames into a registered node's data container.

Used by offline (imported) monitors to populate data without going through the live add_sample() path.

Source code in jumper_extension/adapters/data/node.py

def load_frames(self, node: str, frames: Dict[str, pd.DataFrame]) -> None:
    """Inject pre-loaded DataFrames into a registered node's data container.

    Used by offline (imported) monitors to populate data without going
    through the live add_sample() path.
    """
    perf_data = self._nodes.get(node)
    if perf_data is None:
        return
    for level, df in frames.items():
        perf_data._rows[level] = df.to_dict("records")

aggregate_node_info(hardware)

Return a synthetic NodeInfo aggregating all nodes in hardware.

CPUs/GPUs are summed, gpu_memory takes the max, memory_limits are summed per level, gpu_name is taken from the first node that has one. Used by reporter/service/session to get a single summary view.

Source code in jumper_extension/adapters/data/node.py

def aggregate_node_info(hardware: Dict[str, NodeInfo]) -> NodeInfo:
    """Return a synthetic NodeInfo aggregating all nodes in *hardware*.

    CPUs/GPUs are summed, gpu_memory takes the max, memory_limits are
    summed per level, gpu_name is taken from the first node that has one.
    Used by reporter/service/session to get a single summary view.
    """
    nodes = list(hardware.values())
    if not nodes:
        return NodeInfo(
            node="aggregate", num_cpus=0, num_system_cpus=0, num_gpus=0,
            gpu_memory=0.0, gpu_name="", memory_limits={},
        )
    all_levels = {lvl for n in nodes for lvl in n.memory_limits}
    return NodeInfo(
        node="aggregate",
        num_cpus=sum(n.num_cpus for n in nodes),
        num_system_cpus=sum(n.num_system_cpus for n in nodes),
        num_gpus=sum(n.num_gpus for n in nodes),
        gpu_memory=max(n.gpu_memory for n in nodes),
        gpu_name=next((n.gpu_name for n in nodes if n.gpu_name), ""),
        memory_limits={
            lvl: sum(n.memory_limits.get(lvl, 0.0) for n in nodes)
            for lvl in all_levels
        },
        cpu_handles=[h for n in nodes for h in n.cpu_handles],
    )

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.nodes.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
        hardware = aggregate_node_info(self.monitor.nodes.hardware)
        memory_limit = hardware.memory_limits.get(level, 0.0)
        gpu_memory_limit = hardware.gpu_memory if hardware.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
        hardware = aggregate_node_info(self.monitor.nodes.hardware)
        metrics_spec = [
            (f"CPU Util (Across {hardware.num_cpus} CPUs)", "cpu_util_avg", "-"),
            ("Memory (GB)", "memory", f"{hardware.memory_limits.get(level, 0.0):.2f}"),
            (f"GPU Util (Across {hardware.num_gpus} GPUs)", "gpu_util_avg", "-"),
            ("GPU Memory (GB)", "gpu_mem_avg", f"{hardware.gpu_memory:.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
    hardware = aggregate_node_info(self.monitor.nodes.hardware)
    metrics_spec = [
        (f"CPU Util (Across {hardware.num_cpus} CPUs)", "cpu_util_avg", "-"),
        ("Memory (GB)", "memory", f"{hardware.memory_limits.get(level, 0.0):.2f}"),
        (f"GPU Util (Across {hardware.num_gpus} GPUs)", "gpu_util_avg", "-"),
        ("GPU Memory (GB)", "gpu_mem_avg", f"{hardware.gpu_memory:.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
        hardware = aggregate_node_info(self.monitor.nodes.hardware)
        metrics = [
            (
                f"CPU Util (Across {hardware.num_cpus} CPUs)",
                "cpu_util_avg",
                "-",
            ),
            (
                "Memory (GB)",
                "memory",
                f"{hardware.memory_limits.get(level, 0.0):.2f}",
            ),
            (
                f"GPU Util (Across {hardware.num_gpus} GPUs)",
                "gpu_util_avg",
                "-",
            ),
            (
                "GPU Memory (GB)",
                "gpu_mem_avg",
                f"{hardware.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
    hardware = aggregate_node_info(self.monitor.nodes.hardware)
    metrics = [
        (
            f"CPU Util (Across {hardware.num_cpus} CPUs)",
            "cpu_util_avg",
            "-",
        ),
        (
            "Memory (GB)",
            "memory",
            f"{hardware.memory_limits.get(level, 0.0):.2f}",
        ),
        (
            f"GPU Util (Across {hardware.num_gpus} GPUs)",
            "gpu_util_avg",
            "-",
        ),
        (
            "GPU Memory (GB)",
            "gpu_mem_avg",
            f"{hardware.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)

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 in self.monitor.nodes.levels:
            try:
                df_out = self.monitor.nodes.view(level=level, cell_history=self.cell_history)
            except Exception:
                df_out = pd.DataFrame()
            if not df_out.empty:
                schemas_perf[level] = list(df_out.columns)
                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
        """
        hardware = aggregate_node_info(self.monitor.nodes.hardware)
        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": hardware.num_cpus,
                "num_system_cpus": hardware.num_system_cpus,
                "num_gpus": hardware.num_gpus,
                "gpu_memory": hardware.gpu_memory,
                "gpu_name": hardware.gpu_name,
                "memory_limits": hardware.memory_limits,
                "cpu_handles": hardware.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": self.monitor.nodes.levels,
            "schemas": {
                "perf": schemas_perf,
                "cell_history": list(ch_df.columns),
            },
            "visualizer": {
                "default_metric_subsets": list(
                    getattr(self.visualizer, "default_subsets", ("cpu", "mem", "io"))
                ) + (["gpu", "gpu_all"] if hardware.num_gpus 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