# cjm-fasthtml-resources
<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->
## Install
``` bash
pip install cjm_fasthtml_resources
```
## Project Structure
nbs/
├── core/ (4)
│ ├── management_config.ipynb # Configuration schema for resource management settings including GPU thresholds
│ ├── manager.ipynb # Track worker processes and detect resource conflicts for GPU/CPU usage
│ ├── monitoring_config.ipynb # Configuration schema for resource monitoring refresh intervals and SSE settings
│ └── validation.ipynb # Validate resource availability before job execution and determine appropriate actions
└── utils/ (2)
├── plugin_utils.ipynb # Utilities for analyzing plugin configurations and resource requirements
└── route_helpers.ipynb # Helper utilities for resource monitoring route handlers
Total: 6 notebooks across 2 directories
## Module Dependencies
``` mermaid
graph LR
core_management_config[core.management_config<br/>Management Configuration]
core_manager[core.manager<br/>Resource Manager]
core_monitoring_config[core.monitoring_config<br/>Monitoring Configuration]
core_validation[core.validation<br/>Resource Validation]
utils_plugin_utils[utils.plugin_utils<br/>Plugin Resource Utilities]
utils_route_helpers[utils.route_helpers<br/>Route Helpers]
core_validation --> core_manager
utils_plugin_utils --> core_manager
```
*2 cross-module dependencies detected*
## CLI Reference
No CLI commands found in this project.
## Module Overview
Detailed documentation for each module in the project:
### Management Configuration (`management_config.ipynb`)
> Configuration schema for resource management settings including GPU
> thresholds
#### Import
``` python
from cjm_fasthtml_resources.core.management_config import (
RESOURCE_MANAGEMENT_SCHEMA
)
```
#### Variables
``` python
RESOURCE_MANAGEMENT_SCHEMA = {7 items}
```
### Resource Manager (`manager.ipynb`)
> Track worker processes and detect resource conflicts for GPU/CPU usage
#### Import
``` python
from cjm_fasthtml_resources.core.manager import (
PLUGIN_RESOURCE_CONFIG_KEYS,
ResourceType,
ResourceStatus,
ResourceConflict,
WorkerState,
ResourceManager
)
```
#### Classes
``` python
class ResourceType(Enum):
"Types of resources to monitor."
```
``` python
class ResourceStatus(Enum):
"Status of resource availability."
```
``` python
@dataclass
class ResourceConflict:
"Information about a resource conflict."
resource_type: ResourceType
status: ResourceStatus
app_pids: List[int] # PIDs from this application using the resource
external_pids: List[int] # External PIDs using the resource
app_processes: List[Dict[str, Any]] # Detailed info about app processes
external_processes: List[Dict[str, Any]] # Detailed info about external processes
```
``` python
@dataclass
class WorkerState:
"""
State of a worker process.
Enhanced to support lifecycle-aware and cloud-aware plugins from cjm-fasthtml-plugins.
"""
pid: int
worker_type: str # e.g., "transcription", "llm"
job_id: Optional[str]
plugin_id: Optional[str]
plugin_name: Optional[str]
loaded_plugin_resource: Optional[str] # The plugin resource identifier currently loaded
config: Optional[Dict[str, Any]] # Current plugin configuration
status: str = 'idle' # idle, running, busy
execution_mode: Optional[str] # Execution mode (in_process, subprocess, cloud_gpu, etc.)
child_pids: List[int] = field(...) # PIDs of child processes
container_id: Optional[str] # Docker container ID if applicable
conda_env: Optional[str] # Conda environment name if applicable
is_remote: bool = False # Whether this worker uses remote/cloud resources
remote_resource: Optional[Dict[str, Any]] # Remote resource info (serialized RemoteResourceInfo)
```
``` python
class ResourceManager:
def __init__(
self,
gpu_memory_threshold_percent:float=45.0 # GPU memory usage threshold; external processes using more than this percentage are considered conflicts
)
"Manages resource tracking and conflict detection for the application. Tracks PIDs associated with application workers (transcription, LLM, etc.) and provides methods to check resource availability and conflicts. Enhanced to support lifecycle-aware and cloud-aware plugins from cjm-fasthtml-plugins."
def __init__(
self,
gpu_memory_threshold_percent:float=45.0 # GPU memory usage threshold; external processes using more than this percentage are considered conflicts
)
"Initialize the resource manager."
def register_worker(
self,
pid:int, # Process ID of the worker
worker_type:str, # Type of worker (e.g., "transcription", "llm")
job_id:Optional[str]=None, # Optional job ID if worker is processing a job
plugin_id:Optional[str]=None, # Optional plugin unique ID
plugin_name:Optional[str]=None, # Optional plugin name
loaded_plugin_resource:Optional[str]=None, # Optional identifier of the loaded plugin resource
config:Optional[Dict[str, Any]]=None, # Optional plugin configuration
plugin_instance:Optional[Any]=None # Optional plugin instance for lifecycle/cloud protocol detection
)
"Register a worker process with the resource manager."
def get_all_related_pids(
self,
parent_pid:int # Parent worker PID
) -> List[int]: # List of all PIDs (parent + children)
"Get parent PID and all child PIDs managed by this worker."
def update_worker_state(
self,
pid:int, # Process ID of the worker
job_id:Optional[str]=None, # Optional job ID to update
plugin_id:Optional[str]=None, # Optional plugin ID to update
plugin_name:Optional[str]=None, # Optional plugin name to update
loaded_plugin_resource:Optional[str]=None, # Optional loaded plugin resource to update
config:Optional[Dict[str, Any]]=None, # Optional config to update
status:Optional[str]=None # Optional status to update
)
"Update the state of a registered worker."
def unregister_worker(
self,
pid:int # Process ID of the worker to unregister
)
"Unregister a worker process."
def get_worker_by_pid(
self,
pid:int # Process ID
) -> Optional[WorkerState]: # Worker state or None
"Get worker state by PID."
def get_worker_by_job(
self,
job_id:str # Job ID
) -> Optional[WorkerState]: # Worker state or None
"Get worker state by job ID."
def get_all_workers(self) -> List[WorkerState]: # List of all registered workers
"""Get all registered workers."""
return list(self._worker_states.values())
def get_app_pids(self) -> Set[int]: # Set of all PIDs managed by this application (parents only)
"Get all registered workers."
def get_app_pids(self) -> Set[int]: # Set of all PIDs managed by this application (parents only)
"""Get all PIDs managed by this application (parents only)."""
return set(self._worker_states.keys())
def get_all_app_pids_including_children(self) -> Set[int]: # Set of all PIDs (parents and children)
"Get all PIDs managed by this application (parents only)."
def get_all_app_pids_including_children(self) -> Set[int]: # Set of all PIDs (parents and children)
"""Get all PIDs managed by this application including child processes."""
all_pids = set(self._worker_states.keys())
for worker in self._worker_states.values()
"Get all PIDs managed by this application including child processes."
def get_workers_by_type(
self,
worker_type:str # Type of worker (e.g., "transcription", "llm", "ollama")
) -> List[WorkerState]: # List of workers matching the type
"Get all workers of a specific type."
def get_active_worker_types(self) -> Set[str]: # Set of worker type strings
"""Get set of all active worker types."""
return {w.worker_type for w in self._worker_states.values()}
def has_worker_type(
self,
worker_type:str # Type of worker to check
) -> bool: # True if at least one worker of this type exists
"Get set of all active worker types."
def has_worker_type(
self,
worker_type:str # Type of worker to check
) -> bool: # True if at least one worker of this type exists
"Check if a worker of the specified type exists."
def get_cloud_workers(self) -> List[WorkerState]: # List of workers with is_remote=True
"""Get all workers using cloud/remote resources."""
return [w for w in self._worker_states.values() if w.is_remote]
def estimate_total_cloud_cost(
self,
duration_hours:float=1.0 # Duration to estimate for
) -> float: # Total estimated cost in USD
"Get all workers using cloud/remote resources."
def estimate_total_cloud_cost(
self,
duration_hours:float=1.0 # Duration to estimate for
) -> float: # Total estimated cost in USD
"Estimate total cost of all running cloud resources."
def check_gpu_availability(self) -> ResourceConflict: # ResourceConflict with details about GPU usage
"""Check GPU availability and identify conflicts. Uses configurable GPU memory threshold to determine if external processes are using significant GPU resources. Enhanced to detect child processes from lifecycle-aware plugins."""
try
"Check GPU availability and identify conflicts. Uses configurable GPU memory threshold to determine if external processes are using significant GPU resources. Enhanced to detect child processes from lifecycle-aware plugins."
def check_memory_availability(
self,
threshold_percent:float=90.0 # Memory usage threshold to consider as conflict
) -> ResourceConflict: # ResourceConflict with details about memory usage
"Check system memory availability."
```
#### Variables
``` python
PLUGIN_RESOURCE_CONFIG_KEYS = [5 items]
```
### Monitoring Configuration (`monitoring_config.ipynb`)
> Configuration schema for resource monitoring refresh intervals and SSE
> settings
#### Import
``` python
from cjm_fasthtml_resources.core.monitoring_config import (
RESOURCE_MONITOR_SCHEMA,
LAST_UPDATE_TIMES,
SSE_CONFIG
)
```
#### Variables
``` python
RESOURCE_MONITOR_SCHEMA = {7 items}
LAST_UPDATE_TIMES = {7 items}
SSE_CONFIG = {3 items}
```
### Plugin Resource Utilities (`plugin_utils.ipynb`)
> Utilities for analyzing plugin configurations and resource
> requirements
#### Import
``` python
from cjm_fasthtml_resources.utils.plugin_utils import (
is_local_plugin,
uses_gpu_device,
get_plugin_resource_identifier,
compare_plugin_resources,
get_plugin_resource_requirements
)
```
#### Functions
``` python
def is_local_plugin(
plugin_meta # Plugin metadata with config_schema attribute
) -> bool: # True if plugin is local, False if API-based
"Check if a plugin is local (vs API-based)."
```
``` python
def uses_gpu_device(
"Check if a plugin is configured to use GPU."
```
``` python
def get_plugin_resource_identifier(
"Extract the plugin resource identifier from plugin configuration. Checks common plugin resource configuration keys like 'resource_id', 'model_id', 'model', 'model_name', etc."
```
``` python
def compare_plugin_resources(
config1:Dict[str, Any], # First plugin configuration
config2:Dict[str, Any] # Second plugin configuration
) -> bool: # True if both configs specify the same plugin resource, False otherwise
"Compare two plugin configurations to see if they use the same plugin resource."
```
``` python
def get_plugin_resource_requirements(
plugin_id:str, # Unique plugin ID
plugin_registry, # Plugin registry instance with get_plugin, load_plugin_config methods
plugin_config:Optional[Dict[str, Any]]=None # Optional plugin configuration
) -> Dict[str, Any]: # Dictionary with resource requirement information (is_local, uses_gpu, plugin_resource, device)
"Get resource requirements for a plugin."
```
### Route Helpers (`route_helpers.ipynb`)
> Helper utilities for resource monitoring route handlers
#### Import
``` python
from cjm_fasthtml_resources.utils.route_helpers import (
wrap_card_in_container,
create_card_update
)
```
#### Functions
``` python
def wrap_card_in_container(
content, # Card content to wrap
html_id, # HTML ID for the container
card_cls=None, # Card class (optional, can be provided via DaisyUI)
bg_cls=None, # Background class (optional, can be provided via DaisyUI)
shadow_cls=None, # Shadow class (optional, can be provided via Tailwind)
**kwargs # Additional attributes for the Div
): # Wrapped card container (Div)
"Wrap card content in a Div container with standard styling. This consolidates the common pattern of wrapping monitoring cards in styled containers."
```
``` python
def create_card_update(
render_fn:Callable, # Function to render the card
info:Dict[str, Any], # Info dictionary to pass to render function
target_id:str, # Target HTML ID for the swap
swap_type:str="outerHTML" # Type of swap
): # OOB swap element
"Create an OOB swap update for a card. This consolidates the pattern of creating OOB swaps for card updates in SSE streaming."
```
### Resource Validation (`validation.ipynb`)
> Validate resource availability before job execution and determine
> appropriate actions
#### Import
``` python
from cjm_fasthtml_resources.core.validation import (
ValidationAction,
ValidationResult,
validate_resources_for_job
)
```
#### Functions
``` python
def validate_resources_for_job(
resource_manager, # ResourceManager instance
plugin_registry, # Plugin registry protocol (has get_plugin, load_plugin_config methods)
get_plugin_resource_requirements, # Function: (plugin_id, config) -> Dict with requirements
compare_plugin_resources, # Function: (config1, config2) -> bool (same resource?)
get_plugin_resource_identifier, # Function: (config) -> str (resource ID)
plugin_id:str, # Unique plugin ID
plugin_config:Optional[Dict[str, Any]]=None, # Plugin configuration (will load if not provided)
worker_pid:Optional[int]=None, # PID of the worker that will run the job (if known)
worker_type:str="transcription", # Type of worker (e.g., "transcription", "llm", "ollama")
verbose:bool=False # Whether to print verbose logging
) -> ValidationResult: # ValidationResult with action to take
"Validate if resources are available to run a job with the specified plugin. This function is dependency-injected with helper functions to avoid tight coupling with specific plugin registry implementations."
```
#### Classes
``` python
class ValidationAction(Enum):
"Actions that can be taken based on validation results."
```
``` python
@dataclass
class ValidationResult:
"Result of resource validation."
action: ValidationAction
can_proceed: bool
message: str
conflict: Optional[ResourceConflict]
current_worker: Optional[WorkerState]
plugin_name_to_reload: Optional[str] # Plugin name to reload
new_config: Optional[Dict[str, Any]] # Config for reload
```
Raw data
{
"_id": null,
"home_page": "https://github.com/cj-mills/cjm-fasthtml-resources",
"name": "cjm-fasthtml-resources",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": null,
"keywords": "nbdev jupyter notebook python",
"author": "Christian J. Mills",
"author_email": "9126128+cj-mills@users.noreply.github.com",
"download_url": "https://files.pythonhosted.org/packages/9b/f2/ba29f71e4327292e59808208de6d2929d30ebab51db75eb47a35e8c0f34b/cjm_fasthtml_resources-0.0.3.tar.gz",
"platform": null,
"description": "# cjm-fasthtml-resources\n\n\n<!-- WARNING: THIS FILE WAS AUTOGENERATED! DO NOT EDIT! -->\n\n## Install\n\n``` bash\npip install cjm_fasthtml_resources\n```\n\n## Project Structure\n\n nbs/\n \u251c\u2500\u2500 core/ (4)\n \u2502 \u251c\u2500\u2500 management_config.ipynb # Configuration schema for resource management settings including GPU thresholds\n \u2502 \u251c\u2500\u2500 manager.ipynb # Track worker processes and detect resource conflicts for GPU/CPU usage\n \u2502 \u251c\u2500\u2500 monitoring_config.ipynb # Configuration schema for resource monitoring refresh intervals and SSE settings\n \u2502 \u2514\u2500\u2500 validation.ipynb # Validate resource availability before job execution and determine appropriate actions\n \u2514\u2500\u2500 utils/ (2)\n \u251c\u2500\u2500 plugin_utils.ipynb # Utilities for analyzing plugin configurations and resource requirements\n \u2514\u2500\u2500 route_helpers.ipynb # Helper utilities for resource monitoring route handlers\n\nTotal: 6 notebooks across 2 directories\n\n## Module Dependencies\n\n``` mermaid\ngraph LR\n core_management_config[core.management_config<br/>Management Configuration]\n core_manager[core.manager<br/>Resource Manager]\n core_monitoring_config[core.monitoring_config<br/>Monitoring Configuration]\n core_validation[core.validation<br/>Resource Validation]\n utils_plugin_utils[utils.plugin_utils<br/>Plugin Resource Utilities]\n utils_route_helpers[utils.route_helpers<br/>Route Helpers]\n\n core_validation --> core_manager\n utils_plugin_utils --> core_manager\n```\n\n*2 cross-module dependencies detected*\n\n## CLI Reference\n\nNo CLI commands found in this project.\n\n## Module Overview\n\nDetailed documentation for each module in the project:\n\n### Management Configuration (`management_config.ipynb`)\n\n> Configuration schema for resource management settings including GPU\n> thresholds\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_resources.core.management_config import (\n RESOURCE_MANAGEMENT_SCHEMA\n)\n```\n\n#### Variables\n\n``` python\nRESOURCE_MANAGEMENT_SCHEMA = {7 items}\n```\n\n### Resource Manager (`manager.ipynb`)\n\n> Track worker processes and detect resource conflicts for GPU/CPU usage\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_resources.core.manager import (\n PLUGIN_RESOURCE_CONFIG_KEYS,\n ResourceType,\n ResourceStatus,\n ResourceConflict,\n WorkerState,\n ResourceManager\n)\n```\n\n#### Classes\n\n``` python\nclass ResourceType(Enum):\n \"Types of resources to monitor.\"\n```\n\n``` python\nclass ResourceStatus(Enum):\n \"Status of resource availability.\"\n```\n\n``` python\n@dataclass\nclass ResourceConflict:\n \"Information about a resource conflict.\"\n \n resource_type: ResourceType\n status: ResourceStatus\n app_pids: List[int] # PIDs from this application using the resource\n external_pids: List[int] # External PIDs using the resource\n app_processes: List[Dict[str, Any]] # Detailed info about app processes\n external_processes: List[Dict[str, Any]] # Detailed info about external processes\n```\n\n``` python\n@dataclass\nclass WorkerState:\n \"\"\"\n State of a worker process.\n \n Enhanced to support lifecycle-aware and cloud-aware plugins from cjm-fasthtml-plugins.\n \"\"\"\n \n pid: int\n worker_type: str # e.g., \"transcription\", \"llm\"\n job_id: Optional[str]\n plugin_id: Optional[str]\n plugin_name: Optional[str]\n loaded_plugin_resource: Optional[str] # The plugin resource identifier currently loaded\n config: Optional[Dict[str, Any]] # Current plugin configuration\n status: str = 'idle' # idle, running, busy\n execution_mode: Optional[str] # Execution mode (in_process, subprocess, cloud_gpu, etc.)\n child_pids: List[int] = field(...) # PIDs of child processes\n container_id: Optional[str] # Docker container ID if applicable\n conda_env: Optional[str] # Conda environment name if applicable\n is_remote: bool = False # Whether this worker uses remote/cloud resources\n remote_resource: Optional[Dict[str, Any]] # Remote resource info (serialized RemoteResourceInfo)\n```\n\n``` python\nclass ResourceManager:\n def __init__(\n self,\n gpu_memory_threshold_percent:float=45.0 # GPU memory usage threshold; external processes using more than this percentage are considered conflicts\n )\n \"Manages resource tracking and conflict detection for the application. Tracks PIDs associated with application workers (transcription, LLM, etc.) and provides methods to check resource availability and conflicts. Enhanced to support lifecycle-aware and cloud-aware plugins from cjm-fasthtml-plugins.\"\n \n def __init__(\n self,\n gpu_memory_threshold_percent:float=45.0 # GPU memory usage threshold; external processes using more than this percentage are considered conflicts\n )\n \"Initialize the resource manager.\"\n \n def register_worker(\n self,\n pid:int, # Process ID of the worker\n worker_type:str, # Type of worker (e.g., \"transcription\", \"llm\")\n job_id:Optional[str]=None, # Optional job ID if worker is processing a job\n plugin_id:Optional[str]=None, # Optional plugin unique ID\n plugin_name:Optional[str]=None, # Optional plugin name\n loaded_plugin_resource:Optional[str]=None, # Optional identifier of the loaded plugin resource\n config:Optional[Dict[str, Any]]=None, # Optional plugin configuration\n plugin_instance:Optional[Any]=None # Optional plugin instance for lifecycle/cloud protocol detection\n )\n \"Register a worker process with the resource manager.\"\n \n def get_all_related_pids(\n self,\n parent_pid:int # Parent worker PID\n ) -> List[int]: # List of all PIDs (parent + children)\n \"Get parent PID and all child PIDs managed by this worker.\"\n \n def update_worker_state(\n self,\n pid:int, # Process ID of the worker\n job_id:Optional[str]=None, # Optional job ID to update\n plugin_id:Optional[str]=None, # Optional plugin ID to update\n plugin_name:Optional[str]=None, # Optional plugin name to update\n loaded_plugin_resource:Optional[str]=None, # Optional loaded plugin resource to update\n config:Optional[Dict[str, Any]]=None, # Optional config to update\n status:Optional[str]=None # Optional status to update\n )\n \"Update the state of a registered worker.\"\n \n def unregister_worker(\n self,\n pid:int # Process ID of the worker to unregister\n )\n \"Unregister a worker process.\"\n \n def get_worker_by_pid(\n self,\n pid:int # Process ID\n ) -> Optional[WorkerState]: # Worker state or None\n \"Get worker state by PID.\"\n \n def get_worker_by_job(\n self,\n job_id:str # Job ID\n ) -> Optional[WorkerState]: # Worker state or None\n \"Get worker state by job ID.\"\n \n def get_all_workers(self) -> List[WorkerState]: # List of all registered workers\n \"\"\"Get all registered workers.\"\"\"\n return list(self._worker_states.values())\n \n def get_app_pids(self) -> Set[int]: # Set of all PIDs managed by this application (parents only)\n \"Get all registered workers.\"\n \n def get_app_pids(self) -> Set[int]: # Set of all PIDs managed by this application (parents only)\n \"\"\"Get all PIDs managed by this application (parents only).\"\"\"\n return set(self._worker_states.keys())\n \n def get_all_app_pids_including_children(self) -> Set[int]: # Set of all PIDs (parents and children)\n \"Get all PIDs managed by this application (parents only).\"\n \n def get_all_app_pids_including_children(self) -> Set[int]: # Set of all PIDs (parents and children)\n \"\"\"Get all PIDs managed by this application including child processes.\"\"\"\n all_pids = set(self._worker_states.keys())\n for worker in self._worker_states.values()\n \"Get all PIDs managed by this application including child processes.\"\n \n def get_workers_by_type(\n self,\n worker_type:str # Type of worker (e.g., \"transcription\", \"llm\", \"ollama\")\n ) -> List[WorkerState]: # List of workers matching the type\n \"Get all workers of a specific type.\"\n \n def get_active_worker_types(self) -> Set[str]: # Set of worker type strings\n \"\"\"Get set of all active worker types.\"\"\"\n return {w.worker_type for w in self._worker_states.values()}\n \n def has_worker_type(\n self,\n worker_type:str # Type of worker to check\n ) -> bool: # True if at least one worker of this type exists\n \"Get set of all active worker types.\"\n \n def has_worker_type(\n self,\n worker_type:str # Type of worker to check\n ) -> bool: # True if at least one worker of this type exists\n \"Check if a worker of the specified type exists.\"\n \n def get_cloud_workers(self) -> List[WorkerState]: # List of workers with is_remote=True\n \"\"\"Get all workers using cloud/remote resources.\"\"\"\n return [w for w in self._worker_states.values() if w.is_remote]\n \n def estimate_total_cloud_cost(\n self,\n duration_hours:float=1.0 # Duration to estimate for\n ) -> float: # Total estimated cost in USD\n \"Get all workers using cloud/remote resources.\"\n \n def estimate_total_cloud_cost(\n self,\n duration_hours:float=1.0 # Duration to estimate for\n ) -> float: # Total estimated cost in USD\n \"Estimate total cost of all running cloud resources.\"\n \n def check_gpu_availability(self) -> ResourceConflict: # ResourceConflict with details about GPU usage\n \"\"\"Check GPU availability and identify conflicts. Uses configurable GPU memory threshold to determine if external processes are using significant GPU resources. Enhanced to detect child processes from lifecycle-aware plugins.\"\"\"\n try\n \"Check GPU availability and identify conflicts. Uses configurable GPU memory threshold to determine if external processes are using significant GPU resources. Enhanced to detect child processes from lifecycle-aware plugins.\"\n \n def check_memory_availability(\n self,\n threshold_percent:float=90.0 # Memory usage threshold to consider as conflict\n ) -> ResourceConflict: # ResourceConflict with details about memory usage\n \"Check system memory availability.\"\n```\n\n#### Variables\n\n``` python\nPLUGIN_RESOURCE_CONFIG_KEYS = [5 items]\n```\n\n### Monitoring Configuration (`monitoring_config.ipynb`)\n\n> Configuration schema for resource monitoring refresh intervals and SSE\n> settings\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_resources.core.monitoring_config import (\n RESOURCE_MONITOR_SCHEMA,\n LAST_UPDATE_TIMES,\n SSE_CONFIG\n)\n```\n\n#### Variables\n\n``` python\nRESOURCE_MONITOR_SCHEMA = {7 items}\nLAST_UPDATE_TIMES = {7 items}\nSSE_CONFIG = {3 items}\n```\n\n### Plugin Resource Utilities (`plugin_utils.ipynb`)\n\n> Utilities for analyzing plugin configurations and resource\n> requirements\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_resources.utils.plugin_utils import (\n is_local_plugin,\n uses_gpu_device,\n get_plugin_resource_identifier,\n compare_plugin_resources,\n get_plugin_resource_requirements\n)\n```\n\n#### Functions\n\n``` python\ndef is_local_plugin(\n plugin_meta # Plugin metadata with config_schema attribute\n) -> bool: # True if plugin is local, False if API-based\n \"Check if a plugin is local (vs API-based).\"\n```\n\n``` python\ndef uses_gpu_device(\n \"Check if a plugin is configured to use GPU.\"\n```\n\n``` python\ndef get_plugin_resource_identifier(\n \"Extract the plugin resource identifier from plugin configuration. Checks common plugin resource configuration keys like 'resource_id', 'model_id', 'model', 'model_name', etc.\"\n```\n\n``` python\ndef compare_plugin_resources(\n config1:Dict[str, Any], # First plugin configuration\n config2:Dict[str, Any] # Second plugin configuration\n) -> bool: # True if both configs specify the same plugin resource, False otherwise\n \"Compare two plugin configurations to see if they use the same plugin resource.\"\n```\n\n``` python\ndef get_plugin_resource_requirements(\n plugin_id:str, # Unique plugin ID\n plugin_registry, # Plugin registry instance with get_plugin, load_plugin_config methods\n plugin_config:Optional[Dict[str, Any]]=None # Optional plugin configuration\n) -> Dict[str, Any]: # Dictionary with resource requirement information (is_local, uses_gpu, plugin_resource, device)\n \"Get resource requirements for a plugin.\"\n```\n\n### Route Helpers (`route_helpers.ipynb`)\n\n> Helper utilities for resource monitoring route handlers\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_resources.utils.route_helpers import (\n wrap_card_in_container,\n create_card_update\n)\n```\n\n#### Functions\n\n``` python\ndef wrap_card_in_container(\n content, # Card content to wrap\n html_id, # HTML ID for the container\n card_cls=None, # Card class (optional, can be provided via DaisyUI)\n bg_cls=None, # Background class (optional, can be provided via DaisyUI)\n shadow_cls=None, # Shadow class (optional, can be provided via Tailwind)\n **kwargs # Additional attributes for the Div\n): # Wrapped card container (Div)\n \"Wrap card content in a Div container with standard styling. This consolidates the common pattern of wrapping monitoring cards in styled containers.\"\n```\n\n``` python\ndef create_card_update(\n render_fn:Callable, # Function to render the card\n info:Dict[str, Any], # Info dictionary to pass to render function\n target_id:str, # Target HTML ID for the swap\n swap_type:str=\"outerHTML\" # Type of swap\n): # OOB swap element\n \"Create an OOB swap update for a card. This consolidates the pattern of creating OOB swaps for card updates in SSE streaming.\"\n```\n\n### Resource Validation (`validation.ipynb`)\n\n> Validate resource availability before job execution and determine\n> appropriate actions\n\n#### Import\n\n``` python\nfrom cjm_fasthtml_resources.core.validation import (\n ValidationAction,\n ValidationResult,\n validate_resources_for_job\n)\n```\n\n#### Functions\n\n``` python\ndef validate_resources_for_job(\n resource_manager, # ResourceManager instance\n plugin_registry, # Plugin registry protocol (has get_plugin, load_plugin_config methods)\n get_plugin_resource_requirements, # Function: (plugin_id, config) -> Dict with requirements\n compare_plugin_resources, # Function: (config1, config2) -> bool (same resource?)\n get_plugin_resource_identifier, # Function: (config) -> str (resource ID)\n plugin_id:str, # Unique plugin ID\n plugin_config:Optional[Dict[str, Any]]=None, # Plugin configuration (will load if not provided)\n worker_pid:Optional[int]=None, # PID of the worker that will run the job (if known)\n worker_type:str=\"transcription\", # Type of worker (e.g., \"transcription\", \"llm\", \"ollama\")\n verbose:bool=False # Whether to print verbose logging\n) -> ValidationResult: # ValidationResult with action to take\n \"Validate if resources are available to run a job with the specified plugin. This function is dependency-injected with helper functions to avoid tight coupling with specific plugin registry implementations.\"\n```\n\n#### Classes\n\n``` python\nclass ValidationAction(Enum):\n \"Actions that can be taken based on validation results.\"\n```\n\n``` python\n@dataclass\nclass ValidationResult:\n \"Result of resource validation.\"\n \n action: ValidationAction\n can_proceed: bool\n message: str\n conflict: Optional[ResourceConflict]\n current_worker: Optional[WorkerState]\n plugin_name_to_reload: Optional[str] # Plugin name to reload\n new_config: Optional[Dict[str, Any]] # Config for reload\n```\n",
"bugtrack_url": null,
"license": "Apache Software License 2.0",
"summary": "Resource monitoring and management system for tracking GPU/CPU usage and worker processes in FastHTML applications.",
"version": "0.0.3",
"project_urls": {
"Homepage": "https://github.com/cj-mills/cjm-fasthtml-resources"
},
"split_keywords": [
"nbdev",
"jupyter",
"notebook",
"python"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "e481b1bd1691d476652d6282330fd7ad0b6efde023b15fa0a00f2ad539ab4774",
"md5": "dddc3c088a361bddaede4500519fbe51",
"sha256": "7d1ee72d0d821f928de9f0d3aec6f0364d27fccdc4c570a435bdcd167fd7eb5b"
},
"downloads": -1,
"filename": "cjm_fasthtml_resources-0.0.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "dddc3c088a361bddaede4500519fbe51",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 24525,
"upload_time": "2025-10-24T00:28:29",
"upload_time_iso_8601": "2025-10-24T00:28:29.189811Z",
"url": "https://files.pythonhosted.org/packages/e4/81/b1bd1691d476652d6282330fd7ad0b6efde023b15fa0a00f2ad539ab4774/cjm_fasthtml_resources-0.0.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "9bf2ba29f71e4327292e59808208de6d2929d30ebab51db75eb47a35e8c0f34b",
"md5": "e85d2ad40e98cd426e2ab1eaa7c19343",
"sha256": "d250971fba72ebcf1a93a4ecab7082558ad52c69f9435f21a11ac15f9cd2ebbb"
},
"downloads": -1,
"filename": "cjm_fasthtml_resources-0.0.3.tar.gz",
"has_sig": false,
"md5_digest": "e85d2ad40e98cd426e2ab1eaa7c19343",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 25881,
"upload_time": "2025-10-24T00:28:30",
"upload_time_iso_8601": "2025-10-24T00:28:30.137854Z",
"url": "https://files.pythonhosted.org/packages/9b/f2/ba29f71e4327292e59808208de6d2929d30ebab51db75eb47a35e8c0f34b/cjm_fasthtml_resources-0.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-24 00:28:30",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cj-mills",
"github_project": "cjm-fasthtml-resources",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "cjm-fasthtml-resources"
}
JSON
