cjm-system-monitor-nvidia

An NVIDIA GPU monitoring plugin for the cjm-plugin-system that provides real-time hardware telemetry via nvitop.

Install

pip install cjm_system_monitor_nvidia

Project Structure

nbs/
├── meta.ipynb   # Metadata introspection for the NVIDIA monitor plugin used by cjm-ctl to generate the registration manifest
└── plugin.ipynb # Plugin implementation for NVIDIA GPU monitoring using nvitop

Total: 2 notebooks

Module Dependencies

graph LR
    meta[meta<br/>meta]
    plugin[plugin<br/>plugin]

No cross-module dependencies detected.

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

meta (meta.ipynb)

Metadata introspection for the NVIDIA monitor plugin used by cjm-ctl to generate the registration manifest

Import

from cjm_system_monitor_nvidia.meta import (
    get_plugin_metadata
)

Functions

def get_plugin_metadata() -> Dict[str, Any]:  # Plugin metadata for manifest generation
    """Return metadata required to register this plugin with the PluginManager."""
    # Fallback base path (current behavior for backward compatibility)
    base_path = os.path.dirname(os.path.dirname(sys.executable))
    
    # Use CJM config if available, else fallback to env-relative paths
    cjm_data_dir = os.environ.get("CJM_DATA_DIR")
    
    # Plugin data directory
    plugin_name = "cjm-system-monitor-nvidia"
    if cjm_data_dir
    "Return metadata required to register this plugin with the PluginManager."

plugin (plugin.ipynb)

Plugin implementation for NVIDIA GPU monitoring using nvitop

Import

from cjm_system_monitor_nvidia.plugin import (
    NvidiaMonitorPlugin
)

Classes

class NvidiaMonitorPlugin:
    def __init__(self):
        """Initialize the NVIDIA monitor plugin."""
        self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
        self.config = {}

    @property
    def name(self) -> str:  # Plugin identifier
    "NVIDIA System Monitor using nvitop."
    
    def __init__(self):
            """Initialize the NVIDIA monitor plugin."""
            self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
            self.config = {}
    
        @property
        def name(self) -> str:  # Plugin identifier
        "Initialize the NVIDIA monitor plugin."
    
    def name(self) -> str:  # Plugin identifier
            """Plugin name."""
            return "sys-mon-nvidia"
        
        @property
        def version(self) -> str:  # Plugin version
        "Plugin name."
    
    def version(self) -> str:  # Plugin version
            """Plugin version."""
            return "1.0.0"
    
        def initialize(
            self,
            config: Optional[Dict[str, Any]] = None  # Configuration dictionary
        ) -> None
        "Plugin version."
    
    def initialize(
            self,
            config: Optional[Dict[str, Any]] = None  # Configuration dictionary
        ) -> None
        "Initialize or reconfigure the plugin."
    
    def get_config_schema(self) -> Dict[str, Any]:  # JSON Schema
            """Return JSON Schema for configuration."""
            return {}  # No config needed for monitoring
    
        def get_current_config(self) -> Dict[str, Any]:  # Current config
        "Return JSON Schema for configuration."
    
    def get_current_config(self) -> Dict[str, Any]:  # Current config
            """Return current configuration."""
            return self.config
    
        def cleanup(self) -> None
        "Return current configuration."
    
    def cleanup(self) -> None:
            """Clean up resources."""
            pass
    
        def _get_gpu_info_internal(self) -> Dict[str, Any]:  # Raw GPU data
        "Clean up resources."
    
    def execute(
            self,
            command: str = "get_system_status",  # Command to execute
            **kwargs
        ) -> Dict[str, Any]:  # SystemStats as dictionary
        "Collect stats and return standardized SystemStats dictionary."