cjm-fasthtml-job-monitor 0.0.1

A FastHTML component library for monitoring plugin job execution with progress tracking, log tailing, resource usage, and cancellation via a tabbed modal with content overlay.

cjm-fasthtml-job-monitor

Install

pip install cjm_fasthtml_job_monitor

Project Structure

nbs/
├── components/ (6)
│   ├── tabs/ (3)
│   │   ├── logs_tab.ipynb       # Log tailing view with auto-scroll.
│   │   ├── progress_tab.ipynb   # Progress bar, stage message, elapsed time, and job status badge.
│   │   └── resources_tab.ipynb  # Worker CPU, RAM, and GPU resource usage display.
│   ├── modal.ipynb    # Tabbed modal with progress, logs, and resources tabs.
│   ├── overlay.ipynb  # Semi-transparent content overlay with loading spinner.
│   └── trigger.ipynb  # Trigger button and progress button for job monitor.
├── routes/ (1)
│   └── init.ipynb  # Route factory for job trigger, progress polling, and cancellation.
├── services/ (1)
│   └── monitor.ipynb  # Service for job execution monitoring with resource telemetry.
├── html_ids.ipynb  # Prefix-based HTML ID generator for job monitor DOM elements.
└── models.ipynb    # Data types for job monitor URLs, configuration, and resource snapshots.

Total: 10 notebooks across 3 directories

Module Dependencies

graph LR
    components_modal[components.modal<br/>Modal Component]
    components_overlay[components.overlay<br/>Overlay Components]
    components_tabs_logs_tab[components.tabs.logs_tab<br/>Logs Tab]
    components_tabs_progress_tab[components.tabs.progress_tab<br/>Progress Tab]
    components_tabs_resources_tab[components.tabs.resources_tab<br/>Resources Tab]
    components_trigger[components.trigger<br/>Trigger Components]
    html_ids[html_ids<br/>HTML IDs]
    models[models<br/>Models]
    routes_init[routes.init<br/>Route Factory]
    services_monitor[services.monitor<br/>Monitor Service]

    components_modal --> components_tabs_resources_tab
    components_modal --> models
    components_modal --> components_tabs_logs_tab
    components_modal --> components_tabs_progress_tab
    components_modal --> html_ids
    components_overlay --> models
    components_overlay --> html_ids
    components_tabs_logs_tab --> html_ids
    components_tabs_progress_tab --> html_ids
    components_tabs_resources_tab --> models
    components_trigger --> models
    components_trigger --> html_ids
    routes_init --> components_overlay
    routes_init --> components_modal
    routes_init --> models
    routes_init --> components_trigger
    routes_init --> services_monitor
    routes_init --> html_ids
    services_monitor --> models

19 cross-module dependencies detected

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

HTML IDs (html_ids.ipynb)

Prefix-based HTML ID generator for job monitor DOM elements.

Import

from cjm_fasthtml_job_monitor.html_ids import (
    JobMonitorHtmlIds
)

Classes

@dataclass
class JobMonitorHtmlIds:
    "Prefix-based HTML ID generator for job monitor DOM elements."
    
    prefix: str  # ID prefix for this job monitor instance
    
    def modal(self) -> str:  # The modal dialog element
            """Modal dialog element."""
            return f"{self.prefix}-modal"
    
        @property
        def modal_content(self) -> str:  # Static modal body (not replaced by polling)
        "Modal dialog element."
    
    def modal_content(self) -> str:  # Static modal body (not replaced by polling)
            """Modal body container (holds tabs + footer, rendered once)."""
            return f"{self.prefix}-modal-content"
    
        @property
        def poll_anchor(self) -> str:  # Hidden div that carries HTMX polling
        "Modal body container (holds tabs + footer, rendered once)."
    
    def poll_anchor(self) -> str:  # Hidden div that carries HTMX polling
            """Hidden polling anchor (hx-get target, self-replacing)."""
            return f"{self.prefix}-poll"
    
        # --- Overlay ---
    
        @property
        def overlay(self) -> str:  # Semi-transparent content blocker
        "Hidden polling anchor (hx-get target, self-replacing)."
    
    def overlay(self) -> str:  # Semi-transparent content blocker
            """Content overlay element."""
            return f"{self.prefix}-overlay"
    
        # --- Trigger slot ---
    
        @property
        def trigger_slot(self) -> str:  # Slot for trigger/progress button
        "Content overlay element."
    
    def trigger_slot(self) -> str:  # Slot for trigger/progress button
            """Trigger button slot (swaps between trigger and progress button)."""
            return f"{self.prefix}-trigger-slot"
    
        # --- Progress ---
    
        @property
        def progress_bar(self) -> str:  # Progress bar element
        "Trigger button slot (swaps between trigger and progress button)."
    
    def progress_bar(self) -> str:  # Progress bar element
            """Progress bar element."""
            return f"{self.prefix}-progress-bar"
    
        @property
        def elapsed(self) -> str:  # Elapsed time display
        "Progress bar element."
    
    def elapsed(self) -> str:  # Elapsed time display
            """Elapsed time display element."""
            return f"{self.prefix}-elapsed"
    
        # --- Tabs ---
    
        @property
        def tabs(self) -> str:  # Tab container
        "Elapsed time display element."
    
    def tabs(self) -> str:  # Tab container
            """Tab navigation container."""
            return f"{self.prefix}-tabs"
    
        @property
        def tab_progress(self) -> str:  # Progress tab inner content
        "Tab navigation container."
    
    def tab_progress(self) -> str:  # Progress tab inner content
            """Progress tab inner content (OOB target)."""
            return f"{self.prefix}-tab-progress"
    
        @property
        def tab_logs(self) -> str:  # Logs tab inner content
        "Progress tab inner content (OOB target)."
    
    def tab_logs(self) -> str:  # Logs tab inner content
            """Logs tab inner content (OOB target)."""
            return f"{self.prefix}-tab-logs"
    
        @property
        def tab_resources(self) -> str:  # Resources tab inner content
        "Logs tab inner content (OOB target)."
    
    def tab_resources(self) -> str:  # Resources tab inner content
            """Resources tab inner content (OOB target)."""
            return f"{self.prefix}-tab-resources"
    
        # --- Footer ---
    
        @property
        def modal_footer(self) -> str:  # Modal footer (cancel button area)
        "Resources tab inner content (OOB target)."
    
    def modal_footer(self) -> str:  # Modal footer (cancel button area)
            """Modal footer (OOB target for cancel button show/hide)."""
            return f"{self.prefix}-modal-footer"
    
        # --- Logs ---
    
        @property
        def log_container(self) -> str:  # Scrollable log display
        "Modal footer (OOB target for cancel button show/hide)."
    
    def log_container(self) -> str:  # Scrollable log display
        "Log tailing container."

Route Factory (init.ipynb)

Route factory for job trigger, progress polling, and cancellation.

Import

from cjm_fasthtml_job_monitor.routes.init import (
    init_job_monitor_routes,
    check_inflight_job
)

Functions

def _get_job_data(service, job_id):
    """Extract job status fields from a Job object."""
    job = service.get_job(job_id)
    if not job
    "Extract job status fields from a Job object."

def init_job_monitor_routes(
    monitor_service: JobMonitorService,           # Service instance
    plugin_name: str,                             # Target plugin for jobs
    state_store: SQLiteWorkflowStateStore,         # For persisting job_id
    workflow_id: str,                             # Workflow ID for state access
    step_id: str,                                 # Step ID for state access
    state_key: str,                               # Key in step state for job_id
    prefix: str,                                  # URL prefix
    overlay_target_id: str,                       # ID of element to overlay
    kb_system_id: Optional[str] = None,           # Keyboard system ID to pause/resume
    on_complete: Optional[Callable] = None,       # async fn(job, request, sess) -> List[FT]
    on_cancel: Optional[Callable] = None,         # async fn(job, request, sess) -> List[FT]
    on_fail: Optional[Callable] = None,           # async fn(job, request, sess) -> List[FT]
    job_args_builder: Optional[Callable] = None,  # fn(state_store, workflow_id, session_id) -> (args, kwargs)
    config: Optional[JobMonitorConfig] = None,    # UI config
    id_prefix: str = "jm",                        # HTML ID prefix
    icon_fn: Optional[Callable] = None,           # Icon renderer fn(name, **kwargs) -> FT
) -> Tuple[APIRouter, JobMonitorUrls, JobMonitorHtmlIds]:  # (router, urls, ids)
    "Initialize job monitor routes."

def check_inflight_job(
    monitor_service: JobMonitorService,       # Service instance
    plugin_name: str,                         # Target plugin
    state_store: SQLiteWorkflowStateStore,    # State store
    workflow_id: str,                         # Workflow ID
    session_id: str,                          # Session ID
    step_id: str,                             # Step ID
    state_key: str,                           # State key for job_id
    config: JobMonitorConfig,                 # Display config
    ids: JobMonitorHtmlIds,                   # Element IDs
    urls: JobMonitorUrls,                     # Route URLs
    icon_fn: Optional[Callable] = None,       # Icon renderer
) -> Tuple[Optional[FT], Optional[FT], bool]:  # (trigger_or_progress_btn, overlay_or_placeholder, is_running)
    """
    Check for in-flight job and return appropriate UI state.
    
    Returns:
        - Button element (trigger or progress button)
        - Overlay element (active overlay or empty placeholder)
        - Whether a job is currently running
    """

Logs Tab (logs_tab.ipynb)

Log tailing view with auto-scroll.

Import

from cjm_fasthtml_job_monitor.components.tabs.logs_tab import (
    render_logs_tab
)

Functions

def render_logs_tab(
    ids: JobMonitorHtmlIds,  # Element IDs
    logs: str = '',          # Log text content
) -> FT:  # Logs tab content
    "Render logs tab with auto-scroll to bottom."

Modal Component (modal.ipynb)

Tabbed modal with progress, logs, and resources tabs.

Import

from cjm_fasthtml_job_monitor.components.modal import (
    render_poll_anchor,
    render_tab_content_oob,
    render_footer_oob,
    render_poll_response,
    render_job_modal
)

Functions

def render_poll_anchor(
    ids: JobMonitorHtmlIds,       # Element IDs
    urls: JobMonitorUrls,         # Route URLs
    config: JobMonitorConfig,     # Display config
    is_active: bool = True,       # Whether polling should continue
) -> FT:  # Hidden poll anchor div
    """
    Render the hidden polling anchor element.
    
    When `is_active`, includes hx-get/hx-trigger for continued polling.
    When not active, renders as an inert hidden div (stops polling).
    """

def render_tab_content_oob(
    ids: JobMonitorHtmlIds,                      # Element IDs
    status: str = 'pending',                     # Job status
    progress_value: float = 0.0,                 # 0.0 to 1.0
    status_message: str = '',                    # Stage message
    started_at: Optional[float] = None,          # Unix timestamp
    completed_at: Optional[float] = None,        # Unix timestamp
    logs: str = '',                              # Log text
    resources: Optional[ResourceSnapshot] = None, # Resource data
) -> tuple:  # (progress_div, logs_div, resources_div) with OOB attrs
    "Render the three tab inner content divs as OOB swap targets."

def render_footer_oob(
    ids: JobMonitorHtmlIds,    # Element IDs
    urls: JobMonitorUrls,      # Route URLs
    is_active: bool = True,    # Whether job is active (show cancel)
) -> FT:  # Footer div with OOB attr
    "Render the modal footer with cancel button as OOB swap."

def render_poll_response(
    config: JobMonitorConfig,                    # Display config
    ids: JobMonitorHtmlIds,                      # Element IDs
    urls: JobMonitorUrls,                        # Route URLs
    status: str = 'pending',                     # Job status
    progress_value: float = 0.0,                 # 0.0 to 1.0
    status_message: str = '',                    # Stage message
    started_at: Optional[float] = None,          # Unix timestamp
    completed_at: Optional[float] = None,        # Unix timestamp
    logs: str = '',                              # Log text
    resources: Optional[ResourceSnapshot] = None, # Resource data
) -> tuple:  # (poll_anchor, progress_oob, logs_oob, resources_oob, footer_oob)
    """
    Render the poll response: updated poll anchor + OOB tab content updates.
    
    This is the primary response for the progress polling route.
    The poll anchor is the 'primary' swap target (outerHTML on itself).
    The three tab content divs and footer are OOB swaps.
    """

def render_job_modal(
    config: JobMonitorConfig,                    # Display config
    ids: JobMonitorHtmlIds,                      # Element IDs
    urls: JobMonitorUrls,                        # Route URLs
    status: str = 'pending',                     # Job status
    progress_value: float = 0.0,                 # 0.0 to 1.0
    status_message: str = '',                    # Stage message
    started_at: Optional[float] = None,          # Unix timestamp
    completed_at: Optional[float] = None,        # Unix timestamp
    logs: str = '',                              # Log text
    resources: Optional[ResourceSnapshot] = None, # Resource data
    open_on_render: bool = False,                # Auto-open via JS
) -> FT:  # Dialog element
    """
    Render the full tabbed modal dialog.
    
    The tab structure (radio inputs + tab-content wrappers) is static.
    Each tab-content wrapper contains an inner div with a stable ID
    that gets OOB-swapped by the progress route. This prevents the
    selected tab from resetting on each poll cycle.
    
    Closable via: Escape key, X button (top-right), or clicking backdrop.
    """

Variables

_TERMINAL_STATUSES

Models (models.ipynb)

Data types for job monitor URLs, configuration, and resource snapshots.

Import

from cjm_fasthtml_job_monitor.models import (
    JobMonitorUrls,
    ResourceSnapshot,
    JobMonitorConfig
)

Classes

@dataclass
class JobMonitorUrls:
    "URL endpoints for the job monitor routes."
    
    trigger: str  # POST -- submit job
    progress: str  # GET -- poll progress
    cancel: str  # POST -- cancel job

@dataclass
class ResourceSnapshot:
    "Point-in-time resource usage for a worker."
    
    worker_pid: int  # Worker process ID
    cpu_percent: float  # CPU utilization %
    memory_rss_mb: float  # Process tree RSS in MB
    gpu_memory_mb: Optional[float]  # Per-process GPU memory in MB
    gpu_index: Optional[int]  # Which GPU device
    gpu_name: Optional[str]  # GPU device name
    gpu_total_mb: Optional[float]  # Total GPU memory in MB
    gpu_load_percent: Optional[float]  # GPU compute utilization %

@dataclass
class JobMonitorConfig:
    "Configuration for a job monitor instance."
    
    modal_title: str = 'Job Execution'  # Modal header title
    trigger_label: str = 'Run'  # Trigger button label
    trigger_icon: Optional[str]  # Lucide icon name for trigger button
    progress_label: str = 'View Progress'  # Progress button label (when modal closed)
    poll_interval_ms: int = 1000  # HTMX polling interval in milliseconds
    log_lines: int = 50  # Number of log lines to show
    overlay_z_index: int = 10  # Overlay z-index

Monitor Service (monitor.ipynb)

Service for job execution monitoring with resource telemetry.

Import

from cjm_fasthtml_job_monitor.services.monitor import (
    JobMonitorService
)

Functions

async def _submit_job(
    self,
    plugin_name: str,       # Target plugin
    *args,
    priority: int = 0,
    **kwargs
) -> str:  # job_id
    "Submit a job to the queue."

def _get_job(self, job_id: str) -> Optional[Job]:  # Job or None
    """Get job by ID."""
    return self._queue.get_job(job_id)

async def _cancel_job(self, job_id: str) -> bool:  # True if cancelled
    "Get job by ID."

async def _cancel_job(self, job_id: str) -> bool:  # True if cancelled
    "Cancel a job."

def _get_logs(
    self,
    plugin_name: str,              # Plugin whose logs to read
    lines: int = 50,               # Max lines to return
    current_session_only: bool = True,  # Filter to current session
) -> str:  # Log text
    "Get plugin logs, optionally filtered to current session."

def _filter_current_session(
    raw: str,        # Full log text
    max_lines: int,  # Max lines to return
) -> str:  # Filtered log text
    "Extract logs from the most recent session (after last '--- Starting' marker)."

def _get_resource_snapshot(
    self,
    plugin_name: str,  # Plugin whose worker to query
) -> Optional[ResourceSnapshot]:  # Snapshot or None
    "Get current resource usage for a plugin's worker."

def _enrich_gpu_stats(
    self,
    snapshot: ResourceSnapshot,  # Snapshot to enrich in place
) -> None
    "Add per-process GPU stats from system monitor plugin."

Classes

class JobMonitorService:
    def __init__(
        self,
        queue: JobQueue,                          # Job queue instance
        manager: PluginManager,                   # For worker stats, logs, sysmon
        sysmon_plugin_name: Optional[str] = None, # System monitor plugin name (e.g., 'cjm-system-monitor-nvidia')
    )
    "Service for job execution monitoring with resource telemetry."
    
    def __init__(
            self,
            queue: JobQueue,                          # Job queue instance
            manager: PluginManager,                   # For worker stats, logs, sysmon
            sysmon_plugin_name: Optional[str] = None, # System monitor plugin name (e.g., 'cjm-system-monitor-nvidia')
        )

Overlay Components (overlay.ipynb)

Semi-transparent content overlay with loading spinner.

Import

from cjm_fasthtml_job_monitor.components.overlay import (
    render_job_overlay,
    render_job_overlay_placeholder
)

Functions

def render_job_overlay(
    ids: JobMonitorHtmlIds,       # Element IDs
    config: JobMonitorConfig,     # Display config (for z-index)
) -> FT:  # Overlay div with centered spinner
    "Render semi-transparent overlay with loading spinner."

def render_job_overlay_placeholder(
    ids: JobMonitorHtmlIds,  # Element IDs
) -> FT:  # Empty div with overlay ID
    "Render empty placeholder (swapping this in removes the overlay)."

Progress Tab (progress_tab.ipynb)

Progress bar, stage message, elapsed time, and job status badge.

Import

from cjm_fasthtml_job_monitor.components.tabs.progress_tab import (
    render_progress_tab
)

Functions

def _render_status_badge(
    status: str,  # Job status string
) -> FT:  # Badge element
    "Render a colored badge for job status."

def _format_elapsed(
    started_at: Optional[float],   # Unix timestamp when job started
    completed_at: Optional[float], # Unix timestamp when job completed
) -> str:  # Formatted elapsed time string
    "Format elapsed time as M:SS."

def _elapsed_timer_script(
    ids: JobMonitorHtmlIds,      # Element IDs
    started_at: Optional[float], # Unix timestamp
) -> FT:  # Script element for client-side timer
    "Generate JS for client-side elapsed time updates."

def render_progress_tab(
    ids: JobMonitorHtmlIds,                # Element IDs
    status: str = 'pending',               # Job status
    progress_value: float = 0.0,           # 0.0 to 1.0
    status_message: str = '',              # Stage message
    started_at: Optional[float] = None,    # Unix timestamp
    completed_at: Optional[float] = None,  # Unix timestamp
) -> FT:  # Progress tab content
    "Render progress tab content."

Variables

_STATUS_BADGE_COLORS = {5 items}

Resources Tab (resources_tab.ipynb)

Worker CPU, RAM, and GPU resource usage display.

Import

from cjm_fasthtml_job_monitor.components.tabs.resources_tab import (
    render_resources_tab
)

Functions

def _render_stat_row(
    label: str,          # Stat label (e.g., 'CPU')
    value_text: str,     # Formatted value (e.g., '45.2%')
    bar_pct: Optional[int] = None,   # Progress bar percentage (0-100)
    bar_color: str = '',             # DaisyUI progress color class
) -> FT:  # Stat row element
    "Render a single stat row with label, value, and optional progress bar."

def render_resources_tab(
    resources: Optional[ResourceSnapshot] = None,  # Resource data
) -> FT:  # Resources tab content
    "Render resources tab content."

Trigger Components (trigger.ipynb)

Trigger button and progress button for job monitor.

Import

from cjm_fasthtml_job_monitor.components.trigger import (
    render_job_trigger,
    render_job_progress_button
)

Functions

def render_job_trigger(
    config: JobMonitorConfig,     # Display config
    ids: JobMonitorHtmlIds,       # Element IDs
    urls: JobMonitorUrls,         # Route URLs
    disabled: bool = False,       # Disable button
    icon_fn: Optional[callable] = None,  # Icon renderer fn(name, **kwargs) -> FT
) -> FT:  # Trigger button wrapped in slot div
    "Render the initial trigger button."

def render_job_progress_button(
    config: JobMonitorConfig,   # Display config
    ids: JobMonitorHtmlIds,     # Element IDs
) -> FT:  # Progress button wrapped in slot div
    "Render 'View Progress' button with spinner."